refactor: update dataset handling and clean up unused parameters

Author: hhhhsc701

backend/services/data-cleaning-service/src/main/java/com/dataengine/cleaning/application/httpclient/DatasetClient.java

@@ -37,7 +37,7 @@ public class DatasetClient {
     public static DatasetResponse createDataset(String name, String type) {
         CreateDatasetRequest createDatasetRequest = new CreateDatasetRequest();
         createDatasetRequest.setName(name);
-        createDatasetRequest.setType(type);
+        createDatasetRequest.setDatasetType(type);
 
 
         String jsonPayload;

backend/services/data-cleaning-service/src/main/java/com/dataengine/cleaning/application/service/CleaningTaskService.java

@@ -29,7 +29,6 @@
 import java.io.File;
 import java.io.FileWriter;
 import java.io.IOException;
-import java.time.ZoneId;
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
@@ -56,7 +55,8 @@ public List<CleaningTask> getTasks(String status, String keywords, Integer page,
 
     @Transactional
     public CleaningTask createTask(CreateCleaningTaskRequest request) {
-        DatasetResponse datasetResponse = DatasetClient.createDataset(request.getDestDatasetName(), request.getDestDatasetType());
+        DatasetResponse datasetResponse = DatasetClient.createDataset(request.getDestDatasetName(),
+                request.getDestDatasetType());
 
         CleaningTask task = new CleaningTask();
         task.setName(request.getName());
@@ -74,7 +74,7 @@ public CleaningTask createTask(CreateCleaningTaskRequest request) {
                 .map(OperatorInstanceConverter.INSTANCE::operatorToDo).toList();
         operatorInstanceMapper.insertInstance(taskId, instancePos);
 
-        taskExecutor.submit(() -> executeTask(task, request, datasetResponse.getId()));
+        taskExecutor.submit(() -> executeTask(task, request));
         return task;
     }
 
@@ -87,7 +87,7 @@ public void deleteTask(String taskId) {
         cleaningTaskMapper.deleteTask(taskId);
     }
 
-    private void executeTask(CleaningTask task, CreateCleaningTaskRequest request, String destDatasetId) {
+    private void executeTask(CleaningTask task, CreateCleaningTaskRequest request) {
         task.setStatus(CleaningTask.StatusEnum.RUNNING);
         cleaningTaskMapper.updateTaskStatus(task);
         prepareTask(task, request.getInstance());
@@ -139,9 +139,7 @@ private void scanDataset(String taskId, String srcDatasetId) {
                             "fileSize", content.getSize() + "B",
                             "filePath", content.getFilePath(),
                             "fileType", content.getFileType(),
-                            "fileId", content.getId(),
-                            "sourceFileModifyTime",
-                            content.getLastAccessTime().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli()))
+                            "fileId", content.getId()))
                     .toList();
             writeListMapToJsonlFile(files, FLOW_PATH + "/" + taskId + "/dataset.jsonl");
             pageNumber += 1;

backend/services/data-cleaning-service/src/main/java/com/dataengine/cleaning/domain/model/CreateDatasetRequest.java

@@ -12,15 +12,16 @@
 @Setter
 @NoArgsConstructor
 public class CreateDatasetRequest {
-
+    /** 数据集名称 */
     private String name;
-
+    /** 数据集描述 */
     private String description;
-
-    private String type;
-
-    @Valid
+    /** 数据集类型 */
+    private String datasetType;
+    /** 标签列表 */
     private List<String> tags;
-
+    /** 数据源 */
     private String dataSource;
+    /** 目标位置 */
+    private String targetLocation;
 }

backend/services/data-cleaning-service/src/main/java/com/dataengine/cleaning/domain/model/DatasetResponse.java

@@ -6,6 +6,7 @@
 import lombok.Setter;
 
 import java.time.LocalDateTime;
+import java.util.List;
 
 /**
  * 数据集实体（与数据库表 t_dm_datasets 对齐）
@@ -22,7 +23,7 @@ public class DatasetResponse {
     /** 数据集描述 */
     private String description;
     /** 数据集类型 */
-    private DatasetTypeResponse type;
+    private String datasetType;
     /** 数据集状态 */
     private String status;
     /** 数据源 */

deployment/helm/ray/ray-cluster/values.yaml

@@ -117,7 +117,7 @@ head:
   volumeMounts:
     - mountPath: /tmp/ray
       name: log-volume
-      subPath: ray
+      subPath: ray/head
     - mountPath: /dataset
       name: dataset-volume
     - mountPath: /flow
@@ -137,6 +137,9 @@ head:
       ports:
         - containerPort: 8080
       volumeMounts:
+        - mountPath: /tmp/ray
+          name: log-volume
+          subPath: ray/head
         - mountPath: /var/log/data-platform
           name: log-volume
         - mountPath: /dataset
@@ -237,7 +240,7 @@ worker:
   volumeMounts:
     - mountPath: /tmp/ray
       name: log-volume
-      subPath: ray
+      subPath: ray/worker
     - mountPath: /dataset
       name: dataset-volume
     - mountPath: /flow

runtime/python-executor/data_platform/sqlite_manager/persistence_atction.py

@@ -41,7 +41,6 @@ def persistence_task_info(self, sample: Dict[str, Any]):
         file_type = str(sample.get("fileType"))
         file_name = str(sample.get("fileName"))
         file_path = str(sample.get("filePath"))
-        source_file_modify_time = int(sample.get("sourceFileModifyTime") if sample.get("sourceFileModifyTime") else "0")
         status = int(sample.get("execute_status"))
         failed_reason = sample.get("failed_reason")
         operator_id = str(failed_reason.get("op_name")) if failed_reason else ""
@@ -50,8 +49,8 @@ def persistence_task_info(self, sample: Dict[str, Any]):
         child_id = sample.get("childId")
         slice_num = sample.get('slice_num', 0)
         insert_data = [instance_id, meta_file_name, meta_file_type, meta_file_id, meta_file_size, file_id, file_size,
-                       file_type, file_name, file_path, source_file_modify_time, status, operator_id, error_code,
-                       incremental, child_id, slice_num]
+                       file_type, file_name, file_path, status, operator_id, error_code, incremental, child_id,
+                       slice_num]
         self.insert_clean_result(insert_data, instance_id)
 
     def insert_clean_result(self, insert_data, instance_id):

runtime/python-executor/data_platform/sqlite_manager/sql/sql_config.json

@@ -1,7 +1,7 @@
 {
   "query_sql": "SELECT * FROM task_instance_info WHERE instance_id = ?",
   "db_path": "/flow/sqlite.db",
-  "insert_sql": "INSERT INTO task_instance_info (instance_id, meta_file_name, meta_file_type, meta_file_id, meta_file_size, file_id, file_size, file_type, file_name, file_path, source_file_modify_time, status, operator_id, error_code, incremental, childId, slice_num) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
-  "create_tables_sql": "CREATE TABLE IF NOT EXISTS task_instance_info (instance_id TEXT, meta_file_name TEXT, meta_file_type TEXT, meta_file_id TEXT, meta_file_size TEXT, file_id TEXT, file_size TEXT, file_type TEXT, file_name TEXT, file_path TEXT, source_file_modify_time INTEGER, status INTEGER, operator_id TEXT, error_code TEXT, incremental TEXT, childId INTEGER, slice_num INTEGER DEFAULT 0)",
+  "insert_sql": "INSERT INTO task_instance_info (instance_id, meta_file_name, meta_file_type, meta_file_id, meta_file_size, file_id, file_size, file_type, file_name, file_path, status, operator_id, error_code, incremental, childId, slice_num) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+  "create_tables_sql": "CREATE TABLE IF NOT EXISTS task_instance_info (instance_id TEXT, meta_file_name TEXT, meta_file_type TEXT, meta_file_id TEXT, meta_file_size TEXT, file_id TEXT, file_size TEXT, file_type TEXT, file_name TEXT, file_path TEXT, status INTEGER, operator_id TEXT, error_code TEXT, incremental TEXT, childId INTEGER, slice_num INTEGER DEFAULT 0)",
   "delete_task_instance_sql": "DELETE FROM task_instance_info WHERE instance_id = ?"
 }

runtime/python-executor/data_platform/wrappers/data_platform_executor.py

@@ -55,8 +55,6 @@ def load_meta(line):
             meta["extraFilePath"] = None
         if not meta.get("extraFileType"):
             meta["extraFileType"] = None
-        if not meta.get("sourceFileModifyTime"):
-            meta["sourceFileModifyTime"] = None
         return meta
 
     def run(self):