Merge branch 'develop_930' of github.com:ModelEngine-Group/data-platform into develop_930

Author: chenghh-9609

backend/pom.xml

@@ -106,9 +106,11 @@
             </dependency>
             <!-- MyBatis version alignment -->
             <dependency>
-                <groupId>com.baomidou</groupId>
-                <artifactId>mybatis-plus-spring-boot3-starter</artifactId>
-                <version>${mybatis-plus.version}</version>
+              <groupId>com.baomidou</groupId>
+              <artifactId>mybatis-plus-bom</artifactId>
+              <version>${mybatis-plus.version}</version>
+              <type>pom</type>
+              <scope>import</scope>
             </dependency>
 
             <dependency>
@@ -167,6 +169,10 @@
             <artifactId>mybatis-plus-spring-boot3-starter</artifactId>
             <version>${mybatis-plus.version}</version>
         </dependency>
+      <dependency>
+        <groupId>com.baomidou</groupId>
+        <artifactId>mybatis-plus-jsqlparser</artifactId>
+      </dependency>
 
         <!-- Log4j2 API -->
         <dependency>

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

@@ -12,7 +12,6 @@
 import com.dataengine.cleaning.interfaces.dto.UpdateCleaningTemplateRequest;
 import lombok.RequiredArgsConstructor;
 import org.apache.commons.lang3.StringUtils;
-import org.apache.ibatis.annotations.Param;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
@@ -30,16 +29,11 @@ public class CleaningTemplateService {
 
     private final OperatorInstanceMapper operatorInstanceMapper;
 
-    public List<CleaningTemplate> getTemplates(String keywords, Integer page, Integer size) {
+    public List<CleaningTemplate> getTemplates(String keywords) {
         List<OperatorResponse> allOperators = cleaningTemplateMapper.findAllOperators();
         Map<String, OperatorResponse> operatorsMap = allOperators.stream()
                 .collect(Collectors.toMap(OperatorResponse::getId, Function.identity()));
-
-        Integer offset = null;
-        if (page != null && size != null) {
-            offset = page * size;
-        }
-        List<TemplateWithInstance> allTemplates = cleaningTemplateMapper.findAllTemplates(keywords, size, offset);
+        List<TemplateWithInstance> allTemplates = cleaningTemplateMapper.findAllTemplates(keywords);
         Map<String, List<TemplateWithInstance>> templatesMap = allTemplates.stream()
                 .collect(Collectors.groupingBy(TemplateWithInstance::getId));
         return templatesMap.entrySet().stream().map(twi -> {
@@ -63,10 +57,6 @@ public List<CleaningTemplate> getTemplates(String keywords, Integer page, Intege
         }).toList();
     }
 
-    public int countTemplates(String keywords) {
-        return cleaningTemplateMapper.findAllTemplates(keywords, null, null).size();
-    }
-
     @Transactional
     public CleaningTemplate createTemplate(CreateCleaningTemplateRequest request) {
         CleaningTemplate template = new CleaningTemplate();

backend/services/data-cleaning-service/src/main/java/com/dataengine/cleaning/infrastructure/persistence/mapper/CleaningTemplateMapper.java

@@ -11,8 +11,7 @@
 @Mapper
 public interface CleaningTemplateMapper {
 
-    List<TemplateWithInstance> findAllTemplates(@Param("keywords") String keywords,
-                                                @Param("size") Integer size, @Param("offset") Integer offset);
+    List<TemplateWithInstance> findAllTemplates(@Param("keywords") String keywords);
 
     List<OperatorResponse> findAllOperators();
 

backend/services/data-cleaning-service/src/main/java/com/dataengine/cleaning/interfaces/api/CleaningTemplateController.java

@@ -18,6 +18,7 @@
 import org.springframework.web.bind.annotation.RequestParam;
 import org.springframework.web.bind.annotation.RestController;
 
+import java.util.Comparator;
 import java.util.List;
 
 
@@ -32,14 +33,17 @@ public ResponseEntity<Response<PagedResponse<CleaningTemplate>>> cleaningTemplat
             @RequestParam(value = "page", required = false) Integer page,
             @RequestParam(value = "size", required = false) Integer size,
             @RequestParam(value = "keywords", required = false) String keyword) {
-        List<CleaningTemplate> templates = cleaningTemplateService.getTemplates(keyword, page, size);
-        if (page != null && size != null) {
-            int count  = cleaningTemplateService.countTemplates(keyword);
-            int totalPages = (count + size + 1) / size;
-            return ResponseEntity.ok(Response.ok(PagedResponse.of(templates, page, count, totalPages)));
-        } else {
+        List<CleaningTemplate> templates = cleaningTemplateService.getTemplates(keyword);
+        if (page == null || size == null) {
             return ResponseEntity.ok(Response.ok(PagedResponse.of(templates)));
         }
+        int count = templates.size();
+        int totalPages = (count + size + 1) / size;
+        List<CleaningTemplate> limitTemplates = templates.stream()
+                .sorted(Comparator.comparing(CleaningTemplate::getCreatedAt))
+                .skip((long) page * size)
+                .limit(size).toList();
+        return ResponseEntity.ok(Response.ok(PagedResponse.of(limitTemplates, page, count, totalPages)));
     }
 
     @PostMapping

backend/services/data-cleaning-service/src/main/resources/mappers/CleaningTemplateMapper.xml

@@ -11,9 +11,6 @@
             </if>
         </where>
         ORDER BY created_at
-        <if test="size != null and offset != null">
-            LIMIT ${size} OFFSET ${offset}
-        </if>
     </select>
 
     <select id="findAllOperators" resultType="com.dataengine.cleaning.interfaces.dto.OperatorResponse">

backend/services/data-collection-service/pom.xml

@@ -164,6 +164,37 @@
                     <classifier>exec</classifier>
                 </configuration>
             </plugin>
+            <plugin>
+              <groupId>org.apache.maven.plugins</groupId>
+              <artifactId>maven-compiler-plugin</artifactId>
+              <version>3.11.0</version>
+              <configuration>
+                <source>${maven.compiler.source}</source>
+                <target>${maven.compiler.target}</target>
+                <annotationProcessorPaths>
+                  <!-- 顺序很重要 -->
+                  <path>
+                    <groupId>org.projectlombok</groupId>
+                    <artifactId>lombok</artifactId>
+                    <version>${lombok.version}</version>
+                  </path>
+                  <path>
+                    <groupId>org.projectlombok</groupId>
+                    <artifactId>lombok-mapstruct-binding</artifactId>
+                    <version>${lombok-mapstruct-binding.version}</version>
+                  </path>
+                  <path>
+                    <groupId>org.mapstruct</groupId>
+                    <artifactId>mapstruct-processor</artifactId>
+                    <version>${mapstruct.version}</version>
+                  </path>
+                </annotationProcessorPaths>
+                <compilerArgs>
+                  <arg>-parameters</arg>
+                  <arg>-Amapstruct.defaultComponentModel=spring</arg>
+                </compilerArgs>
+              </configuration>
+            </plugin>
         </plugins>
     </build>
 </project>

backend/services/data-collection-service/src/main/java/com/dataengine/collection/application/service/CollectionTaskService.java

@@ -13,7 +13,10 @@
 import org.springframework.transaction.annotation.Transactional;
 
 import java.time.LocalDateTime;
-import java.util.*;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
 
 @Slf4j
 @Service
@@ -29,13 +32,17 @@ public CollectionTask create(CollectionTask task) {
         task.setCreatedAt(LocalDateTime.now());
         task.setUpdatedAt(LocalDateTime.now());
         taskMapper.insert(task);
+        executeTaskNow(task);
+        return task;
+    }
+
+    private void executeTaskNow(CollectionTask task) {
         if (Objects.equals(task.getSyncMode(), SyncMode.ONCE.getValue())) {
             TaskExecution exec = dataxExecutionService.createExecution(task);
             int timeout = task.getTimeoutSeconds() == null ? 3600 : task.getTimeoutSeconds();
             dataxExecutionService.runAsync(task, exec.getId(), timeout);
             log.info("Triggered DataX execution for task {} at {}, execId={}", task.getId(), LocalDateTime.now(), exec.getId());
         }
-        return task;
     }
 
     @Transactional
@@ -63,18 +70,7 @@ public List<CollectionTask> list(Integer page, Integer size, String status, Stri
 
     @Transactional
     public TaskExecution startExecution(CollectionTask task) {
-        TaskExecution exec = new TaskExecution();
-        exec.setId(UUID.randomUUID().toString());
-        exec.setTaskId(task.getId());
-        exec.setTaskName(task.getName());
-        exec.setStatus(TaskStatus.RUNNING);
-        exec.setProgress(0.0);
-        exec.setStartedAt(LocalDateTime.now());
-        exec.setCreatedAt(LocalDateTime.now());
-        executionMapper.insert(exec);
-        taskMapper.updateLastExecution(task.getId(), exec.getId());
-        taskMapper.updateStatus(task.getId(), TaskStatus.RUNNING.name());
-        return exec;
+        return dataxExecutionService.createExecution(task);
     }
 
     // ---- Template related merged methods ----

backend/services/data-collection-service/src/main/java/com/dataengine/collection/application/service/DataxExecutionService.java

@@ -28,22 +28,15 @@ public class DataxExecutionService {
 
     private final DataxJobBuilder jobBuilder;
     private final DataxProcessRunner processRunner;
-    private final DataxProperties props;
     private final TaskExecutionMapper executionMapper;
     private final CollectionTaskMapper taskMapper;
 
 
     @Transactional
     public TaskExecution createExecution(CollectionTask task) {
-
-        TaskExecution exec = new TaskExecution();
-        exec.setId(UUID.randomUUID().toString());
+        TaskExecution exec = TaskExecution.initTaskExecution();
         exec.setTaskId(task.getId());
         exec.setTaskName(task.getName());
-        exec.setStatus(TaskStatus.RUNNING);
-        exec.setProgress(0.0);
-        exec.setStartedAt(LocalDateTime.now());
-        exec.setCreatedAt(LocalDateTime.now());
         executionMapper.insert(exec);
         taskMapper.updateLastExecution(task.getId(), exec.getId());
         taskMapper.updateStatus(task.getId(), TaskStatus.RUNNING.name());

backend/services/data-collection-service/src/main/java/com/dataengine/collection/domain/model/CollectionTask.java

@@ -1,8 +1,13 @@
 package com.dataengine.collection.domain.model;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import lombok.Data;
 
 import java.time.LocalDateTime;
+import java.util.Collections;
+import java.util.Map;
 
 @Data
 public class CollectionTask {
@@ -22,4 +27,19 @@ public class CollectionTask {
     private LocalDateTime updatedAt;
     private String createdBy;
     private String updatedBy;
+
+    public void addPath() {
+        try {
+            ObjectMapper objectMapper = new ObjectMapper();
+            Map<String, Object> parameter = objectMapper.readValue(
+                config,
+                new TypeReference<>() {}
+            );
+            parameter.put("destPath", "/dataset/local/" + id);
+            parameter.put("filePaths", Collections.singletonList(parameter.get("destPath")));
+            config = objectMapper.writeValueAsString(parameter);
+        } catch (JsonProcessingException e) {
+            throw new RuntimeException(e);
+        }
+    }
 }

backend/services/data-collection-service/src/main/java/com/dataengine/collection/domain/model/TaskExecution.java

@@ -3,6 +3,7 @@
 import lombok.Data;
 
 import java.time.LocalDateTime;
+import java.util.UUID;
 
 @Data
 public class TaskExecution {
@@ -25,4 +26,14 @@ public class TaskExecution {
     private String config;
     private String result;
     private LocalDateTime createdAt;
+
+    public static TaskExecution initTaskExecution() {
+        TaskExecution exec = new TaskExecution();
+        exec.setId(UUID.randomUUID().toString());
+        exec.setStatus(TaskStatus.RUNNING);
+        exec.setProgress(0.0);
+        exec.setStartedAt(LocalDateTime.now());
+        exec.setCreatedAt(LocalDateTime.now());
+        return exec;
+    }
 }