Repository: tapdata/tapdata (language: python)

Issue:
Fix: Fixed not found measurements if task completed and closed simult…
…aneous occurrence.


Discussion / hints:

Fix patch:
diff --git a/iengine/iengine-app/src/main/java/io/tapdata/Runnable/LoadSchemaRunner.java b/iengine/iengine-app/src/main/java/io/tapdata/Runnable/LoadSchemaRunner.java
index 2a4c4b67d4..26e0972e77 100644
--- a/iengine/iengine-app/src/main/java/io/tapdata/Runnable/LoadSchemaRunner.java
+++ b/iengine/iengine-app/src/main/java/io/tapdata/Runnable/LoadSchemaRunner.java
@@ -1,5 +1,6 @@
 package io.tapdata.Runnable;
 
+import com.mongodb.client.result.UpdateResult;
 import com.tapdata.constant.ConnectionUtil;
 import com.tapdata.constant.ConnectorConstant;
 import com.tapdata.constant.Log4jUtil;
@@ -10,6 +11,7 @@
 import com.tapdata.entity.RelateDataBaseTable;
 import com.tapdata.entity.Schema;
 import com.tapdata.mongo.ClientMongoOperator;
+import com.tapdata.tm.commons.schema.DataSourceConnectionDto;
 import io.tapdata.TapInterface;
 import io.tapdata.common.ConverterUtil;
 import io.tapdata.common.TapInterfaceUtil;
@@ -40,10 +42,7 @@
 import org.springframework.data.mongodb.core.query.Query;
 import org.springframework.data.mongodb.core.query.Update;
 
-import java.util.ArrayList;
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Consumer;
 import java.util.function.Predicate;
@@ -119,9 +118,7 @@ public void run() {
 		Thread.currentThread().setName(String.format(THREAD_NAME, connections.getName()));
 
 		logger.info("Starting load schema fields, connection name: {}", connections.getName());
-
-		Update update = new Update().set(ConnectorConstant.LOAD_FIELDS, ConnectorConstant.LOAD_FIELD_STATUS_LOADING);
-		updateConnections(update);
+        if (updateConnections2Loading(connections)) return;
 
 		try {
 				if (StringUtils.isBlank(connections.getPdkType())) {
@@ -299,16 +296,44 @@ private void updateSchema(String loadFieldsStatus, Throwable error) {
 
 		if (needUpdate) {
 			update.set(ConnectorConstant.LOAD_FIELDS, loadFieldsStatus)
-					.set("schemaVersion", this.schemaVersion)
-					.set("lastUpdate", this.lastUpdate)
-					.set("everLoadSchema", true);
+					.set(DataSourceConnectionDto.FIELD_SCHEMA_VERSION, this.schemaVersion)
+					.set(DataSourceConnectionDto.FIELD_LAST_UPDATE, this.lastUpdate)
+					.set(DataSourceConnectionDto.FIELD_EVER_LOAD_SCHEMA, true);
 			updateConnections(update);
 			schema.getTables().clear();
 			schema.getTapTables().clear();
 		}
 	}
 
-	private void updateConnections(Update update) {
+    // return true is not to load schema
+    protected boolean updateConnections2Loading(Connections connections) {
+        String partialUpdateWithSchemaVersion = connections.getPartialUpdateWithSchemaVersion();
+        if (null == partialUpdateWithSchemaVersion) {
+            Update update = new Update()
+                .set(DataSourceConnectionDto.FIELD_LOAD_FIELDS_STATUS, DataSourceConnectionDto.LOAD_FIELD_STATUS_LOADING)
+                .set(DataSourceConnectionDto.FIELD_SCHEMA_VERSION, schemaVersion);
+            updateConnections(update);
+        } else {
+            connections.setTable_filter(connections.getPartialUpdateFilter());
+            if (null != connections.getPartialUpdateFilter()) {
+                loadSchemaProgress.setTableCount(connections.getPartialUpdateFilter().split(",").length);
+            }
+
+            Query query = new Query(Criteria.where("_id").is(connections.getId())
+                .and(DataSourceConnectionDto.FIELD_SCHEMA_VERSION).is(partialUpdateWithSchemaVersion));
+            Update update = new Update()
+                .set(DataSourceConnectionDto.FIELD_LOAD_FIELDS_STATUS, DataSourceConnectionDto.LOAD_FIELD_STATUS_LOADING)
+                .set(DataSourceConnectionDto.FIELD_PARTIAL_UPDATE_FILTER, connections.getPartialUpdateFilter())
+                .set(DataSourceConnectionDto.FIELD_LAST_UPDATE, lastUpdate)
+                .set(DataSourceConnectionDto.FIELD_SCHEMA_VERSION, schemaVersion);
+
+            UpdateResult updateResult = clientMongoOperator.update(query, update, ConnectorConstant.CONNECTION_COLLECTION + "/update-partial-schema");
+            return updateResult.getMatchedCount() <= 0;
+        }
+        return false;
+    }
+
+	protected void updateConnections(Update update) {
 		Query query = new Query(Criteria.where("_id").is(connections.getId()));
 		if (update == null) {
 			return;
diff --git a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/filter/FilterUtil.java b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/filter/FilterUtil.java
index 0e09801dac..f8e8d5ef96 100644
--- a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/filter/FilterUtil.java
+++ b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/filter/FilterUtil.java
@@ -1,11 +1,9 @@
 package io.tapdata.flow.engine.V2.filter;
 
-import io.tapdata.entity.error.CoreException;
 import io.tapdata.entity.event.TapEvent;
 import io.tapdata.entity.schema.TapTable;
 import io.tapdata.flow.engine.V2.util.TapEventUtil;
 
-import java.util.HashMap;
 import java.util.Map;
 import java.util.Set;
 
@@ -17,14 +15,9 @@ public class FilterUtil {
      * */
     public static Map<String, Object> processTableFields(Map<String, Object> data, Set<String> fieldNames) {
         if (null == fieldNames || fieldNames.isEmpty()) return data;
-        if (null == data) data = new HashMap<>();
-        Map<String, Object> finalData = new HashMap<>();
-        for (String fieldName : fieldNames) {
-            if (data.containsKey(fieldName)) {
-                finalData.put(fieldName, data.get(fieldName));
-            }
-        }
-        return finalData;
+        if (null == data) return null;
+        data.entrySet().removeIf(entry -> !fieldNames.contains(entry.getKey()));
+        return data;
     }
 
     public static void filterEventData(TapTable tapTable, TapEvent e) {
diff --git a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastTargetPdkBaseNode.java b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastTargetPdkBaseNode.java
index 585b0a5626..1b6e285b25 100644
--- a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastTargetPdkBaseNode.java
+++ b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastTargetPdkBaseNode.java
@@ -311,6 +311,7 @@ protected boolean createTable(TapTable tapTable, AtomicBoolean succeed,boolean i
 				// only execute start function aspect so that it would be cheated as input
 				AspectUtils.executeAspect(new CreateTableFuncAspect()
 						.createTableEvent(tapCreateTableEvent.get())
+						.setInit(init)
 						.connectorContext(getConnectorNode().getConnectorContext())
 						.dataProcessorContext(dataProcessorContext).state(NewFieldFuncAspect.STATE_START));
 				clientMongoOperator.insertOne(Collections.singletonList(tapTable),
diff --git a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastTargetPdkDataNode.java b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastTargetPdkDataNode.java
index 9da71d8601..752501afb1 100644
--- a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastTargetPdkDataNode.java
+++ b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastTargetPdkDataNode.java
@@ -475,6 +475,7 @@ protected void dropTable(ExistsDataProcessEnum existsDataProcessEnum, String tab
 				// only execute start function aspect so that it would be cheated as input
 				AspectUtils.executeAspect(new DropTableFuncAspect()
 						.dropTableEvent(tapDropTableEvent.get())
+						.setInit(init)
 						.connectorContext(getConnectorNode().getConnectorContext())
 						.dataProcessorContext(dataProcessorContext).state(NewFieldFuncAspect.STATE_START));
 			}
diff --git a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/processor/HazelcastMergeNode.java b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/processor/HazelcastMergeNode.java
index 61ee21ac04..f016a13f58 100644
--- a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/processor/HazelcastMergeNode.java
+++ b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/processor/HazelcastMergeNode.java
@@ -1000,10 +1000,14 @@ protected boolean needLookup(TapdataEvent tapdataEvent) {
 		SyncStage syncStage = tapdataEvent.getSyncStage();
 		if (isInvalidOperation(tapdataEvent)) return false;
 		String op = getOp(tapdataEvent);
-		if (op.equals(OperationType.DELETE.getOp()) || op.equals(OperationType.UPDATE.getOp())) {
+		if (op.equals(OperationType.DELETE.getOp())) {
 			return false;
 		}
 		String preNodeId = getPreNodeId(tapdataEvent);
+		EnableUpdateJoinKey enableUpdateJoinKey = enableUpdateJoinKeyMap.get(preNodeId);
+		if (op.equals(OperationType.UPDATE.getOp()) && Boolean.FALSE.equals(enableUpdateJoinKey.isEnableChildren())) {
+			return false;
+		}
 		boolean existsInLookupMap = this.lookupMap.containsKey(preNodeId);
 		if (existsInLookupMap && SyncStage.INITIAL_SYNC.equals(syncStage)
 				&& (isMainTableFirstMode() || (isSubTableFirstMode() && !isFirstMergeLevel(preNodeId)))) {
@@ -1966,7 +1970,6 @@ private void recursiveCompareChildrenJoinKeyValue(TapdataEvent tapdataEvent, Lis
 			}
 			List<JoinKeyReference> childJoinKeyReferences = joinKeyReference.getChildJoinKeyReferences();
 			if (CollectionUtils.isNotEmpty(childJoinKeyReferences)) {
-				List<Map<String, String>> joinKeys = mergeTableProperties.getJoinKeys();
 				Map<String, Object> beforeJoinKey = buildParentBeforeJoinKeyValueMap(mergeTableProperties, before);
 				if (MapUtils.isNotEmpty(parentBeforeJoinKey)) {
 					beforeJoinKey.putAll(parentBeforeJoinKey);
diff --git a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/schedule/TapdataTaskScheduler.java b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/schedule/TapdataTaskScheduler.java
index b25043792c..83280cf52f 100644
--- a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/schedule/TapdataTaskScheduler.java
+++ b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/schedule/TapdataTaskScheduler.java
@@ -8,8 +8,6 @@
 import com.tapdata.tm.commons.task.dto.TaskDto;
 import com.tapdata.tm.commons.task.dto.TaskOpRespDto;
 import com.tapdata.tm.sdk.available.TmStatusService;
-import io.tapdata.aspect.TaskStopAspect;
-import io.tapdata.aspect.utils.AspectUtils;
 import io.tapdata.common.SettingService;
 import io.tapdata.dao.MessageDao;
 import io.tapdata.entity.memory.MemoryFetcher;
@@ -412,8 +410,11 @@ public void errorOrStopTask() {
 						if (TerminalMode.STOP_GRACEFUL == terminalMode) {
 							stopTaskResource = StopTaskResource.STOPPED;
 						} else if (TerminalMode.COMPLETE == terminalMode) {
-							stopTaskResource = StopTaskResource.COMPLETE;
-						} else if (TerminalMode.INTERNAL_STOP == terminalMode) {
+                            // 确保任务启动周期大于 5 秒，让指标上报一次
+                            if (System.currentTimeMillis() - taskClient.getCreateTime() < 5000) return;
+
+                            stopTaskResource = StopTaskResource.COMPLETE;
+                        } else if (TerminalMode.INTERNAL_STOP == terminalMode) {
 							if (taskClient.stop()) {
 								clearTaskCacheAfterStopped(taskClient);
 								clearTaskRetryCache(taskId);
@@ -578,29 +579,27 @@ public boolean stopTaskCallAssignApi(TaskClient<TaskDto> taskClient, StopTaskRes
 		}
 		final boolean stop = taskClient.stop();
 		if (stop) {
-			final String taskId = taskClient.getTask().getId().toHexString();
+			final TaskDto task = taskClient.getTask();
+			final String taskName = task.getName();
+			final String taskId = task.getId().toHexString();
 			String resource = ConnectorConstant.TASK_COLLECTION + "/" + stopTaskResource.getResource();
 			try {
-				try {
-					logger.info("Call {} api to modify task [{}] status", resource, taskClient.getTask().getName());
-					TaskOpRespDto taskOpRespDto = clientMongoOperator.updateById(new Update(), resource, taskId, TaskOpRespDto.class);
-					if(CollectionUtils.isEmpty(taskOpRespDto.getSuccessIds())){
-						return false;
-					}
-				} catch (Exception e) {
-					if (StringUtils.isNotBlank(e.getMessage()) && e.getMessage().contains("Transition.Not.Supported")) {
-						// 违反TM状态机，不再进行修改任务状态的重试
-						logger.warn("Call api to stop task status to " + resource + " failed, will set task to error, message: " + e.getMessage(), e);
-						clientMongoOperator.updateById(new Update(), ConnectorConstant.TASK_COLLECTION + "/" + StopTaskResource.RUN_ERROR.getResource(), taskId, TaskDto.class);
-					} else {
-						throw new RuntimeException(String.format("Call stop task api failed, api uri: %s, task: %s[%s]",
-								resource, taskClient.getTask().getName(), taskClient.getTask().getId()), e);
-					}
+				logger.info("Call {} api to modify task [{}] status", resource, taskName);
+				TaskOpRespDto taskOpRespDto = clientMongoOperator.updateById(new Update(), resource, taskId, TaskOpRespDto.class);
+				if(CollectionUtils.isEmpty(taskOpRespDto.getSuccessIds())){
+					return false;
 				}
+				return true;
 			} catch (Exception e) {
-				logger.warn(e.getMessage(), e);
+				if (StringUtils.isNotBlank(e.getMessage()) && e.getMessage().contains("Transition.Not.Supported")) {
+					// 违反TM状态机，不再进行修改任务状态的重试
+					logger.warn("Call api to stop task status to {} failed, will set task to error, message: {}", resource, e.getMessage(), e);
+					clientMongoOperator.updateById(new Update(), ConnectorConstant.TASK_COLLECTION + "/" + StopTaskResource.RUN_ERROR.getResource(), taskId, TaskDto.class);
+					return true;
+				} else {
+					logger.warn("Call stop task api failed, api uri: {}, task: {}[{}]", resource, taskName, taskId, e);
+				}
 			}
-			return true;
 		}
 		return false;
 	}
@@ -685,7 +684,7 @@ public DataMap memory(String keyRegex, String memoryLevel) {
 		return dataMap;
 	}
 
-	private enum StopTaskResource {
+	protected enum StopTaskResource {
 		STOPPED("stopped"),
 		RUN_ERROR("runError"),
 		COMPLETE("complete"),
diff --git a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/task/TaskClient.java b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/task/TaskClient.java
index 8901374b86..f14ecf77cf 100644
--- a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/task/TaskClient.java
+++ b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/task/TaskClient.java
@@ -62,6 +62,8 @@ default int getRetryTime() {
 		return 0;
 	}
 
+	long getCreateTime();
+
 	default boolean resume() {
 		throw new UnsupportedOperationException();
 	}
diff --git a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/task/impl/HazelcastTaskClient.java b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/task/impl/HazelcastTaskClient.java
index 8274ee8c1f..f42b256a3a 100644
--- a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/task/impl/HazelcastTaskClient.java
+++ b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/task/impl/HazelcastTaskClient.java
@@ -61,6 +61,7 @@ public class HazelcastTaskClient implements TaskClient<TaskDto> {
 	private final AtomicInteger retryCounter;
 	private AtomicBoolean retrying;
 	private final AutoRecovery autoRecovery;
+    private final long createTime = System.currentTimeMillis();
 
 	public static HazelcastTaskClient create(TaskDto taskDto, ClientMongoOperator clientMongoOperator, ConfigurationCenter configurationCenter, HazelcastInstance hazelcastInstance) {
 		return new HazelcastTaskClient(null, taskDto, clientMongoOperator, configurationCenter, hazelcastInstance);
@@ -276,7 +277,12 @@ public int getRetryTime() {
 		return retryCounter.get();
 	}
 
-	public void setJob(Job job) {
+    @Override
+    public long getCreateTime() {
+        return createTime;
+    }
+
+    public void setJob(Job job) {
 		this.job = job;
 	}
 }
diff --git a/iengine/iengine-app/src/main/java/io/tapdata/websocket/handler/DataInspectEventHandler.java b/iengine/iengine-app/src/main/java/io/tapdata/websocket/handler/DataInspectEventHandler.java
index a9b7dc9e2c..3c946a02f8 100644
--- a/iengine/iengine-app/src/main/java/io/tapdata/websocket/handler/DataInspectEventHandler.java
+++ b/iengine/iengine-app/src/main/java/io/tapdata/websocket/handler/DataInspectEventHandler.java
@@ -44,7 +44,6 @@ public Object handle(Map event) {
 				return WebSocketEventResult.handleFailed(WebSocketEventResult.Type.EXECUTE_DATA_INSPECT_RESULT, "Inspect message can not be empty.");
 			}
 		}catch (Exception e){
-			logger.info("abc");
 			return WebSocketEventResult.handleFailed(WebSocketEventResult.Type.EXECUTE_DATA_INSPECT_RESULT, "Inspect message can not be empty.");
 		}
 
diff --git a/iengine/iengine-common/src/main/java/com/tapdata/entity/Connections.java b/iengine/iengine-common/src/main/java/com/tapdata/entity/Connections.java
index 483e11663b..a092959e76 100644
--- a/iengine/iengine-common/src/main/java/com/tapdata/entity/Connections.java
+++ b/iengine/iengine-common/src/main/java/com/tapdata/entity/Connections.java
@@ -154,6 +154,8 @@ public class Connections implements Serializable {
 	private String table_filter;
 	private String tableExcludeFilter;
 	private Boolean openTableExcludeFilter;
+	private String partialUpdateWithSchemaVersion;
+	private String partialUpdateFilter;
 
 	/**
 	 * byte
@@ -1177,7 +1179,23 @@ public void setOpenTableExcludeFilter(Boolean openTableExcludeFilter) {
 		this.openTableExcludeFilter = openTableExcludeFilter;
 	}
 
-	public String getSheet_start() {
+    public String getPartialUpdateWithSchemaVersion() {
+        return partialUpdateWithSchemaVersion;
+    }
+
+    public void setPartialUpdateWithSchemaVersion(String partialUpdateWithSchemaVersion) {
+        this.partialUpdateWithSchemaVersion = partialUpdateWithSchemaVersion;
+    }
+
+    public String getPartialUpdateFilter() {
+        return partialUpdateFilter;
+    }
+
+    public void setPartialUpdateFilter(String partialUpdateFilter) {
+        this.partialUpdateFilter = partialUpdateFilter;
+    }
+
+    public String getSheet_start() {
 		return sheet_start;
 	}
 
diff --git a/manager/tm-api/src/main/java/com/tapdata/tm/ds/entity/DataSourceEntity.java b/manager/tm-api/src/main/java/com/tapdata/tm/ds/entity/DataSourceEntity.java
index 429172a47d..b4722c35b6 100644
--- a/manager/tm-api/src/main/java/com/tapdata/tm/ds/entity/DataSourceEntity.java
+++ b/manager/tm-api/src/main/java/com/tapdata/tm/ds/entity/DataSourceEntity.java
@@ -162,6 +162,7 @@ public class DataSourceEntity extends BaseEntity implements IDataPermissionEntit
     private String search_databaseType;
     private String connectionUrl;
     private String lastUpdateTime;
+    private Long lastUpdate;
     private String errorMsg;
     private Integer increamentalTps;
     private Integer initialReadSize;
@@ -333,6 +334,7 @@ public class DataSourceEntity extends BaseEntity implements IDataPermissionEntit
 
     private String multiConnectionInstanceId;
 	private List<String> namespace;
+    private Long timeDifference;
 
     /**
      * 后续 开放可以多选 flow engine 的话，这里一定要删除
diff --git a/manager/tm-api/src/main/java/com/tapdata/tm/ds/service/impl/DataSourceService.java b/manager/tm-api/src/main/java/com/tapdata/tm/ds/service/impl/DataSourceService.java
index b5523f5c73..0f982e3d09 100644
--- a/manager/tm-api/src/main/java/com/tapdata/tm/ds/service/impl/DataSourceService.java
+++ b/manager/tm-api/src/main/java/com/tapdata/tm/ds/service/impl/DataSourceService.java
@@ -64,6 +64,8 @@ public DataSourceService(@NonNull DataSourceRepository repository) {
 
     public abstract long upsertByWhere(Where where, Document update, DataSourceConnectionDto connectionDto, UserDetail user);
 
+    public abstract long updatePartialSchema(String id, String loadFieldsStatus, Long lastUpdate, String fromSchemaVersion, String toSchemaVersion, String filters, UserDetail user);
+
     public abstract void loadSchema(UserDetail user, List<TapTable> tables, DataSourceConnectionDto oldConnectionDto, String expression, String databaseId, Boolean loadSchemaField,Boolean partLoad);
 
     public abstract void updateAfter(UserDetail user, DataSourceConnectionDto connectionDto, String oldName, Boolean submit);
diff --git a/manager/tm-api/src/main/java/com/tapdata/tm/task/service/TaskService.java b/manager/tm-api/src/main/java/com/tapdata/tm/task/service/TaskService.java
index 1fe41640c7..42a5093f35 100644
--- a/manager/tm-api/src/main/java/com/tapdata/tm/task/service/TaskService.java
+++ b/manager/tm-api/src/main/java/com/tapdata/tm/task/service/TaskService.java
@@ -1,7 +1,5 @@
 package com.tapdata.tm.task.service;
 
-import cn.hutool.http.server.HttpServerRequest;
-import cn.hutool.http.server.HttpServerResponse;
 import com.mongodb.client.result.UpdateResult;
 import com.tapdata.tm.base.dto.*;
 import com.tapdata.tm.base.service.BaseService;
@@ -294,6 +292,8 @@ public <T extends BaseDto> T convertToDto(TaskEntity entity, Class<T> dtoClass,
 
     public abstract boolean checkCloudTaskLimit(ObjectId taskId, UserDetail user, boolean checkCurrentTask);
     public abstract void updateDelayTime(ObjectId taskId, long delayTime);
+    public abstract void refreshSchemas(TaskDto taskDto, String nodeIds, String keys, UserDetail userDetail);
+    public abstract void checkSourceTimeDifference(TaskDto taskDto,UserDetail userDetail);
 
     @Data
     @AllArgsConstructor
@@ -316,4 +316,5 @@ public static class Char2Group {
         private long totalDelete = 0;
         private long totalDeleteSize = 0;
     }
-}
\ No newline at end of file
+}
+
diff --git a/manager/tm-api/src/main/java/com/tapdata/tm/utils/MailUtils.java b/manager/tm-api/src/main/java/com/tapdata/tm/utils/MailUtils.java
index c8fb2f995e..ae25a67637 100644
--- a/manager/tm-api/src/main/java/com/tapdata/tm/utils/MailUtils.java
+++ b/manager/tm-api/src/main/java/com/tapdata/tm/utils/MailUtils.java
@@ -102,6 +102,10 @@ public SendStatus sendHtmlMail(String subject, List<String> toList, String usern
         Document doc = Jsoup.parse(html);
         doc.getElementById(USER_NAME).html(username);
 
+        if (StringUtils.isEmpty(maiContent) || StringUtils.isEmpty(maiContent.trim())) {
+            sendStatus.setErrorMessage("mailContent 为空");
+            return sendStatus;
+        }
         if (StringUtils.isEmpty(agentName)) {
             sendStatus.setErrorMessage("agentName 为空");
             return sendStatus;
diff --git a/manager/tm-common/src/main/java/com/tapdata/tm/commons/cdcdelay/CdcDelay.java b/manager/tm-common/src/main/java/com/tapdata/tm/commons/cdcdelay/CdcDelay.java
index fe408e8005..d83a299c0c 100644
--- a/manager/tm-common/src/main/java/com/tapdata/tm/commons/cdcdelay/CdcDelay.java
+++ b/manager/tm-common/src/main/java/com/tapdata/tm/commons/cdcdelay/CdcDelay.java
@@ -14,11 +14,13 @@
 import java.text.ParseException;
 import java.text.SimpleDateFormat;
 import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.ZoneOffset;
 import java.util.Date;
 import java.util.List;
 import java.util.Map;
 import java.util.TimeZone;
-import java.util.function.Consumer;
+import java.util.function.LongConsumer;
 import java.util.function.LongConsumer;
 
 /**
@@ -82,7 +84,7 @@ public TapEvent filterAndCalcDelay(TapEvent tapEvent, @NonNull LongConsumer dela
      * @param recordEvent heartbeat record event
      * @return heartbeat event times
      */
-    private static Long parseTs(@NonNull TapRecordEvent recordEvent) {
+    protected static Long parseTs(@NonNull TapRecordEvent recordEvent) {
         Object ts = null;
         if (recordEvent instanceof TapInsertRecordEvent) {
             Map<String, Object> after = ((TapInsertRecordEvent) recordEvent).getAfter();
@@ -103,6 +105,8 @@ private static Long parseTs(@NonNull TapRecordEvent recordEvent) {
             return ((Instant) ts).toEpochMilli();
         } else if (ts instanceof Long) {
             return (Long) ts;
+        } else if (ts instanceof LocalDateTime) {
+            return ((LocalDateTime) ts).toInstant((ZoneOffset.UTC)).toEpochMilli();
         } else if (ts instanceof String) {
             try {
                 SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'");
diff --git a/manager/tm-common/src/main/java/com/tapdata/tm/commons/schema/DataSourceConnectionDto.java b/manager/tm-common/src/main/java/com/tapdata/tm/commons/schema/DataSourceConnectionDto.java
index 195fa09e12..09038b6c1d 100644
--- a/manager/tm-common/src/main/java/com/tapdata/tm/commons/schema/DataSourceConnectionDto.java
+++ b/manager/tm-common/src/main/java/com/tapdata/tm/commons/schema/DataSourceConnectionDto.java
@@ -27,6 +27,16 @@ public class DataSourceConnectionDto extends BaseDto implements IDataPermissionD
         public static final String STATUS_READY = "ready";
         public static final String STATUS_TESTING = "testing";
 
+    public static final String FIELD_LAST_UPDATE = "lastUpdate";
+    public static final String FIELD_SCHEMA_VERSION = "schemaVersion";
+    public static final String FIELD_EVER_LOAD_SCHEMA = "everLoadSchema";
+    public static final String FIELD_PARTIAL_UPDATE_FILTER = "partialUpdateFilter";
+    public static final String FIELD_PARTIAL_UPDATE_WITH_SCHEMA_VERSION = "partialUpdateWithSchemaVersion";
+    public static final String FIELD_LOAD_FIELDS_STATUS = "loadFieldsStatus";
+    public static final String LOAD_FIELD_STATUS_LOADING = "loading";
+    public static final String LOAD_FIELD_STATUS_FINISHED = "finished";
+    public static final String LOAD_FIELD_STATUS_ERROR = "error";
+
         /** 数据源连接名称 */
         private String name;
         /** 数据源的配置信息 符合connectiondefinition的jsonschema的json  */
@@ -146,6 +156,7 @@ public class DataSourceConnectionDto extends BaseDto implements IDataPermissionD
         private String search_databaseType;
         private String connectionUrl;
         private String lastUpdateTime;
+        private Long lastUpdate;
         private String errorMsg;
         private Integer increamentalTps;
         private Integer initialReadSize;
diff --git a/manager/tm-common/src/main/java/com/tapdata/tm/commons/task/dto/ParentTaskDto.java b/manager/tm-common/src/main/java/com/tapdata/tm/commons/task/dto/ParentTaskDto.java
index 4f6a8db8d6..bfac93f9f4 100644
--- a/manager/tm-common/src/main/java/com/tapdata/tm/commons/task/dto/ParentTaskDto.java
+++ b/manager/tm-common/src/main/java/com/tapdata/tm/commons/task/dto/ParentTaskDto.java
@@ -22,6 +22,7 @@ public class ParentTaskDto extends SchedulableDto {
     public final static String TYPE_INITIAL_SYNC = "initial_sync";
     public final static String TYPE_INITIAL_SYNC_CDC = "initial_sync+cdc";
     public final static String TYPE_CDC = "cdc";
+    public static final String FIELD_TRANSFORM_STATUS = "transformStatus";
 
     /** 扩展参数*/
     private Map<String, Object> attrs;
diff --git a/manager/tm-common/src/main/java/com/tapdata/tm/commons/task/dto/TaskDto.java b/manager/tm-common/src/main/java/com/tapdata/tm/commons/task/dto/TaskDto.java
index d8f678ff2e..b1f3574b45 100644
--- a/manager/tm-common/src/main/java/com/tapdata/tm/commons/task/dto/TaskDto.java
+++ b/manager/tm-common/src/main/java/com/tapdata/tm/commons/task/dto/TaskDto.java
@@ -209,6 +209,10 @@ public boolean isAutoInspect() {
      * */
     private Map<String, Object> agentGroupInfo;
     private Boolean oldVersionTimezone = true;
+    /**
+     * 源端数据源与引擎最大时间差
+     */
+    private Long timeDifference;
 
     public DAG getDag() {
         if (dag != null) {
diff --git a/manager/tm/src/main/java/com/tapdata/tm/ds/controller/DataSourceController.java b/manager/tm/src/main/java/com/tapdata/tm/ds/controller/DataSourceController.java
index e5f3b64c6b..c4570b1e2c 100644
--- a/manager/tm/src/main/java/com/tapdata/tm/ds/controller/DataSourceController.java
+++ b/manager/tm/src/main/java/com/tapdata/tm/ds/controller/DataSourceController.java
@@ -664,4 +664,44 @@ public ResponseMessage<String> addConnection(@RequestBody DataSourceConnectionDt
         return success(result);
     }
 
+    @Operation(summary = "Update partial schema with schemaVersion")
+    @PostMapping("update-partial-schema/update")
+    public ResponseMessage<Map<String, Object>> updatePartialSchema(@RequestParam("where") String whereJson, @RequestBody String reqBody) {
+        String resultKey = "count";
+        Map<String, Object> result = new HashMap<>();
+        result.put(resultKey, 0);
+
+        Where where = parseWhere(whereJson);
+        if (null == where) return success(result);
+
+        String id = (String) where.get("_id");
+        if (StringUtils.isBlank(id)) return success(result);
+        String fromSchemaVersion = (String) where.get(DataSourceConnectionDto.FIELD_SCHEMA_VERSION);
+        if (StringUtils.isBlank(fromSchemaVersion)) return success(result);
+
+        Document updateDoc = Optional.ofNullable(reqBody)
+            .map(jsonStr -> InstanceFactory.instance(JsonParser.class).fromJson(reqBody, Document.class))
+            .map(doc -> {
+                Object setObj = doc.get("$set");
+                if (null != setObj) {
+                    String setJson = JsonUtil.toJsonUseJackson(setObj);
+                    return JsonUtil.parseJsonUseJackson(setJson, new TypeReference<Document>() {
+                    });
+                }
+                return null;
+            }).orElse(null);
+        if (null == updateDoc) return success(result);
+
+        String filters = updateDoc.getString(DataSourceConnectionDto.FIELD_PARTIAL_UPDATE_FILTER);
+        String loadFieldsStatus = updateDoc.getString(DataSourceConnectionDto.FIELD_LOAD_FIELDS_STATUS);
+        if (StringUtils.isBlank(loadFieldsStatus)) return success(result);
+        String toSchemaVersion = updateDoc.getString(DataSourceConnectionDto.FIELD_SCHEMA_VERSION);
+        if (StringUtils.isBlank(toSchemaVersion)) return success(result);
+        Long lastUpdate = updateDoc.getLong(DataSourceConnectionDto.FIELD_LAST_UPDATE);
+        if (null == lastUpdate) return success(result);
+
+        UserDetail user = getLoginUser();
+        result.put(resultKey, dataSourceService.updatePartialSchema(id, loadFieldsStatus, lastUpdate, fromSchemaVersion, toSchemaVersion, filters, user));
+        return success(result);
+    }
 }
diff --git a/manager/tm/src/main/java/com/tapdata/tm/ds/service/impl/DataSourceServiceImpl.java b/manager/tm/src/main/java/com/tapdata/tm/ds/service/impl/DataSourceServiceImpl.java
index c488b50b77..5719093770 100644
--- a/manager/tm/src/main/java/com/tapdata/tm/ds/service/impl/DataSourceServiceImpl.java
+++ b/manager/tm/src/main/java/com/tapdata/tm/ds/service/impl/DataSourceServiceImpl.java
@@ -116,8 +116,6 @@
 public class DataSourceServiceImpl extends DataSourceService{
 
     private final static String connectNameReg = "^([\u4e00-\u9fa5]|[A-Za-z])[\\s\\S]*$";
-    private final static String LAST_UPDATE= "lastUpdate";
-    public static final String LOAD_FIELD_STATUS_FINISHED = "finished";
     @Value("${gateway.secret:}")
     private String gatewaySecret;
     @Value("#{'${spring.profiles.include:idaas}'.split(',')}")
@@ -1225,11 +1223,11 @@ private void checkName(String name) {
     private void deleteModels(String loadFieldsStatus, String datasourceId, Long schemaVersion, UserDetail user) {
         log.debug("delete model, loadFieldsStatus = {}, datasourceId = {}, schemaVersion = {}",
                 loadFieldsStatus, datasourceId, schemaVersion);
-        if (LOAD_FIELD_STATUS_FINISHED.equals(loadFieldsStatus) && schemaVersion != null) {
+        if (DataSourceConnectionDto.LOAD_FIELD_STATUS_FINISHED.equals(loadFieldsStatus) && schemaVersion != null) {
             log.debug("loadFieldsStatus is finished, update model delete flag");
             // handle delete model, not match schemaVersion will update is_deleted to true
             Criteria criteria = Criteria.where("is_deleted").ne(true).and("source._id").is(datasourceId)
-                    .and(LAST_UPDATE).lt(schemaVersion).and("taskId").exists(false).and("meta_type").ne("database");
+                    .and(DataSourceConnectionDto.FIELD_LAST_UPDATE).lt(schemaVersion).and("taskId").exists(false).and("meta_type").ne("database");
             log.info("Delete metadata update filter: {}", criteria);
             Query query = new Query(criteria);
             LiveDataPlatformDto liveDataPlatformDto = liveDataPlatformService.findOne(new Query(), user);
@@ -1257,6 +1255,10 @@ private void deleteModels(String loadFieldsStatus, String datasourceId, Long sch
 
 
     public void sendTestConnection(DataSourceConnectionDto connectionDto, boolean updateSchema, Boolean submit, UserDetail user) {
+        sendTestConnection(connectionDto, updateSchema, submit, null, user);
+    }
+
+    public void sendTestConnection(DataSourceConnectionDto connectionDto, boolean updateSchema, Boolean submit, String partialUpdateFilter, UserDetail user) {
         log.info("send test connection, connection = {}, updateSchema = {}， submit = {}", connectionDto.getName(), updateSchema, submit);
 
         submit = submit != null && submit;
@@ -1302,6 +1304,10 @@ public void sendTestConnection(DataSourceConnectionDto connectionDto, boolean up
         data.put("type", "testConnection");
         data.put("editTest", false);
         data.put("updateSchema", updateSchema);
+        if (null != partialUpdateFilter) {
+            data.put(DataSourceConnectionDto.FIELD_PARTIAL_UPDATE_WITH_SCHEMA_VERSION, connectionDto.getSchemaVersion());
+            data.put(DataSourceConnectionDto.FIELD_PARTIAL_UPDATE_FILTER, partialUpdateFilter);
+        }
         MessageQueueDto queueDto = new MessageQueueDto();
         queueDto.setReceiver(processId);
         queueDto.setData(data);
@@ -1455,19 +1461,25 @@ public long upsertByWhere(Where where, Document update, DataSourceConnectionDto
                 inValues.add("view");
                 Criteria criteria2 = Criteria.where("source._id").is(connectionId).and("meta_type").in(inValues);
                 metadataInstancesService.update(new Query(criteria2), Update.update("databaseId", databaseId), user);
-                flushDatabaseMetadataInstanceLastUpdate((String) set.get("loadFieldsStatus"), connectionId, (Long) set.get(LAST_UPDATE), user);
+
+                Long lastUpdate = null;
+                String loadFieldsStatus = null;
+                if (null != set) {
+                    lastUpdate = (Long) set.get(DataSourceConnectionDto.FIELD_LAST_UPDATE);
+                    loadFieldsStatus = (String) set.get(DataSourceConnectionDto.FIELD_LOAD_FIELDS_STATUS);
+                    flushDatabaseMetadataInstanceLastUpdate(loadFieldsStatus, connectionId, lastUpdate, user);
+                }
                 if (hasSchema) {
                     if (CollectionUtils.isNotEmpty(tables)) {
-                        Long schemaVersion = (Long) set.get(LAST_UPDATE);
-                        String loadFieldsStatus = (String) set.get("loadFieldsStatus");
-                        Boolean loadSchemaField = set.get("loadSchemaField") != null ? ((Boolean) set.get("loadSchemaField")) : true;
+                        Long schemaVersion = (Long) set.get(DataSourceConnectionDto.FIELD_LAST_UPDATE);
+                        Boolean loadSchemaField = !Boolean.FALSE.equals(set.get("loadSchemaField"));
                         loadSchema(user, tables, oldConnectionDto, definitionDto.getExpression(), databaseId, loadSchemaField,false);
                         deleteModels(loadFieldsStatus, oldConnectionDto.getId().toHexString(), schemaVersion, user);
                         update.put("loadSchemaTime", new Date());
                     }
                 } else {
-                    if (set != null && set.get(LAST_UPDATE) != null) {
-                        deleteModels(LOAD_FIELD_STATUS_FINISHED, connectionId, (Long) set.get(LAST_UPDATE), user);
+                    if (null != lastUpdate) {
+                        deleteModels(DataSourceConnectionDto.LOAD_FIELD_STATUS_FINISHED, connectionId, lastUpdate, user);
                     }
                 }
             }
@@ -1506,7 +1518,41 @@ public long upsertByWhere(Where where, Document update, DataSourceConnectionDto
         return 0L;
     }
 
-    public void loadSchema(UserDetail user, List<TapTable> tables, DataSourceConnectionDto oldConnectionDto, String expression, String databaseId, Boolean loadSchemaField,Boolean partLoad) {
+    @Override
+    public long updatePartialSchema(String id, String loadFieldsStatus, Long lastUpdate, String fromSchemaVersion, String toSchemaVersion, String filters, UserDetail user) {
+        long result = 0;
+        // 更新连接加载状态
+        DataSourceEntity entity = repository.findAndModify(
+            Query.query(Criteria.where("_id").is(id)
+                .and(DataSourceConnectionDto.FIELD_SCHEMA_VERSION).is(fromSchemaVersion))
+            , Update.update(DataSourceConnectionDto.FIELD_SCHEMA_VERSION, toSchemaVersion)
+                .set(DataSourceConnectionDto.FIELD_LAST_UPDATE, lastUpdate)
+                .set(DataSourceConnectionDto.FIELD_LOAD_FIELDS_STATUS, loadFieldsStatus)
+            , user);
+
+        // 更新连接对应模型的版本（部分更新，需要将没更新的模型版本进行设置）
+        if (null != entity) {
+            result = 1;
+            Criteria where = Criteria.where("source._id").is(id)
+                .and("is_deleted").ne(true)
+                .and("meta_type").is(MetaType.table)
+                .and("sourceType").is(SourceTypeEnum.SOURCE)
+                .and("source." + DataSourceConnectionDto.FIELD_SCHEMA_VERSION).is(fromSchemaVersion);
+            if(null != filters) {
+                where = where.and("name").nin(Arrays.asList(filters.split(",")));
+            }
+            UpdateResult updateResult = metadataInstancesService.updateMany(Query.query(where)
+                , Update.update("source." + DataSourceConnectionDto.FIELD_SCHEMA_VERSION, toSchemaVersion)
+                    .set(DataSourceConnectionDto.FIELD_LAST_UPDATE, lastUpdate)
+            );
+            result += updateResult.getModifiedCount();
+        }
+
+        // 返回数量大于 0，才需要进行模型加载
+        return result;
+    }
+
+    public void loadSchema(UserDetail user, List<TapTable> tables, DataSourceConnectionDto oldConnectionDto, String expression, String databaseId, Boolean loadSchemaField, Boolean partLoad) {
         for (TapTable table : tables) {
             PdkSchemaConvert.getTableFieldTypesGenerator().autoFill(table.getNameFieldMap() == null ? new LinkedHashMap<>() : table.getNameFieldMap(), DefaultExpressionMatchingMap.map(expression));
         }
@@ -1889,6 +1935,9 @@ public void updateConnectionOptions(ObjectId id, ConnectionOptions options, User
         if (null != options.getDbVersion()){
             update.set("db_version",options.getDbVersion());
         }
+        if (null != options.getTimeDifference()){
+            update.set("timeDifference",options.getTimeDifference());
+        }
 
         updateByIdNotChangeLast(id, update, user);
 
@@ -2073,7 +2122,7 @@ public DataSourceConnectionDto findById(ObjectId id, String... fields) {
 
     @Override
     public void flushDatabaseMetadataInstanceLastUpdate(String loadFieldsStatus, String connectionId, Long lastUpdate, UserDetail userDetail) {
-        if (LOAD_FIELD_STATUS_FINISHED.equals(loadFieldsStatus) && lastUpdate != null && StringUtils.isNotBlank(connectionId)) {
+        if (DataSourceConnectionDto.LOAD_FIELD_STATUS_FINISHED.equals(loadFieldsStatus) && lastUpdate != null && StringUtils.isNotBlank(connectionId)) {
             DataSourceConnectionDto connectionDto = findById(toObjectId(connectionId), userDetail);
             if (connectionDto == null) {
                 return;
@@ -2081,7 +2130,7 @@ public void flushDatabaseMetadataInstanceLastUpdate(String loadFieldsStatus, Str
             String qualifiedName = MetaDataBuilderUtils.generateQualifiedName(MetaType.database.name(), connectionDto, null);
             Criteria criteria = Criteria.where("qualified_name").is(qualifiedName);
             Query query = new Query(criteria);
-            metadataInstancesService.update(query,Update.update(LAST_UPDATE,lastUpdate),userDetail);
+            metadataInstancesService.update(query, Update.update(DataSourceConnectionDto.FIELD_LAST_UPDATE, lastUpdate), userDetail);
         }
 
     }
diff --git a/manager/tm/src/main/java/com/tapdata/tm/init/patches/daas/V3_10_1_Task_Last_Updated_Index.java b/manager/tm/src/main/java/com/tapdata/tm/init/patches/daas/V3_10_1_Task_Last_Updated_Index.java
new file mode 100644
index 0000000000..49dacd8afd
--- /dev/null
+++ b/manager/tm/src/main/java/com/tapdata/tm/init/patches/daas/V3_10_1_Task_Last_Updated_Index.java
@@ -0,0 +1,33 @@
+package com.tapdata.tm.init.patches.daas;
+
+import com.tapdata.tm.init.PatchType;
+import com.tapdata.tm.init.PatchVersion;
+import com.tapdata.tm.init.patches.AbsPatch;
+import com.tapdata.tm.init.patches.PatchAnnotation;
+import com.tapdata.tm.utils.SpringContextHelper;
+import io.tapdata.utils.AppType;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.springframework.core.env.Environment;
+import org.springframework.data.domain.Sort;
+import org.springframework.data.mongodb.core.MongoTemplate;
+import org.springframework.data.mongodb.core.index.Index;
+
+@PatchAnnotation(appType = AppType.DAAS, version = "3.10-1")
+public class V3_10_1_Task_Last_Updated_Index extends AbsPatch {
+    private static final Logger logger = LogManager.getLogger(V3_10_1_Task_Last_Updated_Index.class);
+
+    public V3_10_1_Task_Last_Updated_Index(PatchType type, PatchVersion version) {
+        super(type, version);
+    }
+
+    @Override
+    public void run() {
+        logger.info("Execute java patch: {}...", getClass().getName());
+        String collectionName = "Task";
+        MongoTemplate mongoTemplate = SpringContextHelper.getBean(MongoTemplate.class);
+
+        Index index = new Index().named("index_task_last_updated").on("last_updated", Sort.Direction.DESC).background();
+        mongoTemplate.indexOps(collectionName).ensureIndex(index);
+    }
+}
diff --git a/manager/tm/src/main/java/com/tapdata/tm/init/patches/dfs/V3_10_2_Task_Last_Updated_Index.java b/manager/tm/src/main/java/com/tapdata/tm/init/patches/dfs/V3_10_2_Task_Last_Updated_Index.java
new file mode 100644
index 0000000000..40298ec149
--- /dev/null
+++ b/manager/tm/src/main/java/com/tapdata/tm/init/patches/dfs/V3_10_2_Task_Last_Updated_Index.java
@@ -0,0 +1,32 @@
+package com.tapdata.tm.init.patches.dfs;
+
+import com.tapdata.tm.init.PatchType;
+import com.tapdata.tm.init.PatchVersion;
+import com.tapdata.tm.init.patches.AbsPatch;
+import com.tapdata.tm.init.patches.PatchAnnotation;
+import com.tapdata.tm.utils.SpringContextHelper;
+import io.tapdata.utils.AppType;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.springframework.data.domain.Sort;
+import org.springframework.data.mongodb.core.MongoTemplate;
+import org.springframework.data.mongodb.core.index.Index;
+
+@PatchAnnotation(appType = AppType.DFS, version = "3.10-2")
+public class V3_10_2_Task_Last_Updated_Index extends AbsPatch {
+    private static final Logger logger = LogManager.getLogger(V3_10_2_Task_Last_Updated_Index.class);
+
+    public V3_10_2_Task_Last_Updated_Index(PatchType type, PatchVersion version) {
+        super(type, version);
+    }
+
+    @Override
+    public void run() {
+        logger.info("Execute java patch: {}...", getClass().getName());
+        String collectionName = "Task";
+        MongoTemplate mongoTemplate = SpringContextHelper.getBean(MongoTemplate.class);
+
+        Index index = new Index().named("index_task_last_updated").on("last_updated", Sort.Direction.DESC).background();
+        mongoTemplate.indexOps(collectionName).ensureIndex(index);
+    }
+}
diff --git a/manager/tm/src/main/java/com/tapdata/tm/lineage/analyzer/impl/TableAnalyzerV1.java b/manager/tm/src/main/java/com/tapdata/tm/lineage/analyzer/impl/TableAnalyzerV1.java
index 389f832ad0..28ad3930ee 100644
--- a/manager/tm/src/main/java/com/tapdata/tm/lineage/analyzer/impl/TableAnalyzerV1.java
+++ b/manager/tm/src/main/java/com/tapdata/tm/lineage/analyzer/impl/TableAnalyzerV1.java
@@ -523,9 +523,9 @@ private static LineageTask wrapLineageTask(TaskEntity task, Node node) {
 		);
 	}
 
-	private List<ModulesEntity> findModules(String connectionId, String table) {
+	protected List<ModulesEntity> findModules(String connectionId, String table) {
 		Criteria criteria = new Criteria("datasource").is(connectionId)
-				.and("tableName").is(table);
+				.and("tableName").is(table).and("is_deleted").ne(true);
 		Query query = Query.query(criteria);
 		query.fields().include(MODULES_INCLUDE_FIELDS);
 		return modulesRepository.findAll(query);
diff --git a/manager/tm/src/main/java/com/tapdata/tm/message/service/MessageServiceImpl.java b/manager/tm/src/main/java/com/tapdata/tm/message/service/MessageServiceImpl.java
index 99e7ed2bfb..b7f260365f 100644
--- a/manager/tm/src/main/java/com/tapdata/tm/message/service/MessageServiceImpl.java
+++ b/manager/tm/src/main/java/com/tapdata/tm/message/service/MessageServiceImpl.java
@@ -99,7 +99,7 @@ public class MessageServiceImpl extends MessageService{
 
     private ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
 
-    private final static String FEISHU_ADDRESS = "http://34.96.213.48:30008/send_to/feishu/group";
+    private final static String FEISHU_ADDRESS = CommonUtils.getProperty("FEISHU_ADDRESS","https://lark-api.tapdata.net/send_to/feishu/group");
 
 
 
@@ -872,8 +872,8 @@ private void informUser(MessageDto messageDto) {
             MsgTypeEnum msgTypeEnum = MsgTypeEnum.getEnumByValue(msgType);
             String clickHref = mailUtils.getAgentClick(metadataName, msgTypeEnum);
             if(checkSending(userDetail)){
-                SendStatus sendStatus = mailUtils.sendHtmlMail(subject + MAIL_SUBJECT, getMailAddressList(userDetail), username, metadataName, clickHref, emailTip);
-                eventsService.recordEvents(subject + MAIL_SUBJECT, MAIL_CONTENT, userDetail.getEmail(), messageDto, sendStatus, retry, Type.NOTICE_MAIL);
+                SendStatus sendStatus = mailUtils.sendHtmlMail(MAIL_SUBJECT + subject, getMailAddressList(userDetail), username, metadataName, clickHref, emailTip);
+                eventsService.recordEvents(MAIL_SUBJECT + subject, MAIL_CONTENT, userDetail.getEmail(), messageDto, sendStatus, retry, Type.NOTICE_MAIL);
             }
         }
 
diff --git a/manager/tm/src/main/java/com/tapdata/tm/monitor/service/MeasurementServiceV2Impl.java b/manager/tm/src/main/java/com/tapdata/tm/monitor/service/MeasurementServiceV2Impl.java
index 44847e2cbd..31d4dd62e1 100644
--- a/manager/tm/src/main/java/com/tapdata/tm/monitor/service/MeasurementServiceV2Impl.java
+++ b/manager/tm/src/main/java/com/tapdata/tm/monitor/service/MeasurementServiceV2Impl.java
@@ -400,21 +400,7 @@ public Document toDocument(AggregationOperationContext context) {
         AggregationResults<MeasurementEntity> results = mongoOperations.aggregate(aggregation, MeasurementEntity.COLLECTION_NAME, MeasurementEntity.class);
         List<MeasurementEntity> entities = results.getMappedResults();
 
-        for (MeasurementEntity entity : entities) {
-            String hash = hashTag(entity.getTags());
-            for(Sample sample : entity.getSamples()) {
-                if (!data.containsKey(hash)) {
-                    data.put(hash, sample);
-                    continue;
-                }
-
-                long oldInterval = Math.abs(data.get(hash).getDate().getTime() - time);
-                long newInterval = Math.abs(sample.getDate().getTime() - time);
-                if (newInterval < oldInterval) {
-                    data.put(hash, sample);
-                }
-            }
-        }
+        parse2SampleData(entities, data, time);
 
         for (String hash : data.keySet()) {
             Sample sample = data.get(hash);
@@ -445,9 +431,32 @@ public Document toDocument(AggregationOperationContext context) {
         return data;
     }
 
-    private Double getTaskLastFiveMinutesQps(Map<String, String> tags) {
+    protected void parse2SampleData(List<MeasurementEntity> entities, Map<String, Sample> data, long time) {
+        Sample sample;
+        List<Sample> samples;
+        for (MeasurementEntity entity : entities) {
+            String hash = hashTag(entity.getTags());
+
+            // 获取最新点的指标（倒序是因为任务在1秒内完成时 init 和 completed 指标同时间一样）
+            samples = entity.getSamples();
+            for(int i = samples.size() - 1; i >= 0; i --) {
+                sample = samples.get(i);
+                if (!data.containsKey(hash)) {
+                    data.put(hash, sample);
+                } else {
+                    long oldInterval = Math.abs(data.get(hash).getDate().getTime() - time);
+                    long newInterval = Math.abs(sample.getDate().getTime() - time);
+                    if (newInterval < oldInterval) {
+                        data.put(hash, sample);
+                    }
+                }
+            }
+        }
+    }
+
+    protected Double getTaskLastFiveMinutesQps(Map<String, String> tags) {
         Criteria criteria = new Criteria();
-        tags.forEach((k,v)->{
+        tags.forEach((k, v) -> {
             String format = String.format(TAG_FORMAT, k);
             criteria.and(format).is(v);
         });
@@ -462,12 +471,13 @@ private Double getTaskLastFiveMinutesQps(Map<String, String> tags) {
         List<Map> mappedResults = mongoOperations.aggregate(aggregation, MeasurementEntity.COLLECTION_NAME, Map.class).getMappedResults();
         if (CollectionUtils.isNotEmpty(mappedResults)) {
             Map map = mappedResults.get(0);
-            if (map.containsKey("qps")) {
+            return Optional.ofNullable(map.get("qps")).map(v -> {
                 try {
-                    return Double.parseDouble(map.get("qps").toString());
+                    return Double.parseDouble(v.toString());
                 } catch (NumberFormatException ignored) {
+                    return null;
                 }
-            }
+            }).orElse(0D);
         }
         return 0D;
     }
diff --git a/manager/tm/src/main/java/com/tapdata/tm/schedule/util/LoadSchemaScheduleUtil.java b/manager/tm/src/main/java/com/tapdata/tm/schedule/util/LoadSchemaScheduleUtil.java
index acb1e8a661..3d3485d152 100644
--- a/manager/tm/src/main/java/com/tapdata/tm/schedule/util/LoadSchemaScheduleUtil.java
+++ b/manager/tm/src/main/java/com/tapdata/tm/schedule/util/LoadSchemaScheduleUtil.java
@@ -90,7 +90,7 @@ protected void loadSchemaOnce(DataSourceConnectionDto dataSource, long count, in
         if (null != loadSchemaTime) {
             lastLoadTime = loadSchemaTime.getTime();
         }
-        if (time == hour && now() - lastLoadTime >= intervalDay * 24 * 60 * 60 * 1000) {
+        if (time == hour && now() - lastLoadTime >= intervalDay * 24 * 60 * 60 * 1000L) {
             service.sendTestConnection(dataSource, true, true, user);
             //比较大的表，需要sleep一下
             sleep(sleepTime);
diff --git a/manager/tm/src/main/java/com/tapdata/tm/task/controller/TaskController.java b/manager/tm/src/main/java/com/tapdata/tm/task/controller/TaskController.java
index 4757fe3726..615dc54b95 100644
--- a/manager/tm/src/main/java/com/tapdata/tm/task/controller/TaskController.java
+++ b/manager/tm/src/main/java/com/tapdata/tm/task/controller/TaskController.java
@@ -4,6 +4,7 @@
 import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
+import io.swagger.annotations.ApiParam;
 import org.springframework.core.io.InputStreamResource;
 import org.springframework.http.ResponseEntity;
 
@@ -392,6 +393,7 @@ public ResponseMessage<TaskDto> findById(
 				if (StringUtils.isNotBlank(taskDto.getCrontabScheduleMsg())) {
 					taskDto.setCrontabScheduleMsg(MessageUtil.getMessage(taskDto.getCrontabScheduleMsg()));
 				}
+                taskService.checkSourceTimeDifference(taskDto,user);
 			}
 			return success(taskDto);
 		}
@@ -1433,4 +1435,22 @@ public ResponseMessage<List<String>> calculatedTimeRange(@RequestBody QueryOpera
         List<String> range = TimeTransFormationUtil.calculatedTimeRange(workerLocalTime,queryOperator,offsetHours);
         return success(range);
     }
+
+    @Operation(summary = "Refresh task schemas")
+    @PutMapping("/{id}/re-schemas")
+    public ResponseMessage<Object> refreshSchemas(HttpServletRequest request
+        , @PathVariable String id
+        , @ApiParam(name = "nodeIds", value = "Refresh the specified node schemas when present, split multiple using ','"
+    ) @RequestParam(required = false) String nodeIds
+        , @ApiParam(name = "keys", value = "Refresh the specified table schemas when present, split multiple using ','"
+    ) @RequestParam(required = false) String keys) {
+        UserDetail user = getLoginUser();
+        ObjectId objectId = MongoUtils.toObjectId(id);
+        TaskDto resultTask = dataPermissionCheckOfId(request, user, objectId, DataPermissionActionEnums.Edit,
+            () -> taskService.findById(objectId, new Field())
+        );
+
+        taskService.refreshSchemas(resultTask, nodeIds, keys, user);
+        return success();
+    }
 }
diff --git a/manager/tm/src/main/java/com/tapdata/tm/task/service/TaskServiceImpl.java b/manager/tm/src/main/java/com/tapdata/tm/task/service/TaskServiceImpl.java
index 932e285db2..b402a73b49 100644
--- a/manager/tm/src/main/java/com/tapdata/tm/task/service/TaskServiceImpl.java
+++ b/manager/tm/src/main/java/com/tapdata/tm/task/service/TaskServiceImpl.java
@@ -29,6 +29,7 @@
 import com.tapdata.tm.commons.dag.logCollector.LogCollectorNode;
 import com.tapdata.tm.commons.schema.bean.ResponseBody;
 import com.tapdata.tm.commons.schema.bean.ValidateDetail;
+import com.tapdata.tm.ds.entity.DataSourceEntity;
 import com.tapdata.tm.monitor.service.BatchService;
 import com.tapdata.tm.shareCdcTableMapping.service.ShareCdcTableMappingService;
 import com.tapdata.tm.task.bean.*;
@@ -272,6 +273,7 @@ public class TaskServiceImpl extends TaskService{
     public static final String EDGE_MILESTONES = "edgeMilestones";
     public static final String SYNC_PROGRESS = "syncProgress";
     public static final String TASK_NOT_FOUND = "Task.NotFound";
+    public static final String EX_TASK_NOT_FOUND_DS = "Task.NotfoundDatasource";
     public static final String STATUS = "status";
     public static final String CREATE_TIME = "createTime";
     public static final String DAG_NODES_CONNECTION_ID = "dag.nodes.connectionId";
@@ -4454,6 +4456,23 @@ public void updateDelayTime(ObjectId taskId, long delayTime) {
         update(new Query(criteria), update);
     }
 
+    @Override
+    public void checkSourceTimeDifference(TaskDto taskDto,UserDetail userDetail) {
+        DAG dag = taskDto.getDag();
+        if(null != dag){
+           Set<String> connectionIds = dag.getSourceNodes().stream().filter(node -> node instanceof DataParentNode)
+                   .map(node -> ((DataParentNode<?>) node).getConnectionId()).collect(Collectors.toSet());
+           if(CollectionUtils.isNotEmpty(connectionIds)){
+               Query query = new Query(Criteria.where("_id").in(connectionIds));
+               query.fields().include("timeDifference");
+               List<DataSourceEntity> dataSourceEntities = dataSourceService.findAll(query, userDetail);
+               Long timeDifference = dataSourceEntities.stream().map(DataSourceEntity::getTimeDifference).filter(Objects::nonNull).max(Comparator.comparing(x -> x)).orElse(null);
+               taskDto.setTimeDifference(timeDifference);
+           }
+
+        }
+    }
+
 
     public void increaseClear(ObjectId taskId, String srcNode, String tgtNode, UserDetail user) {
         //清理只需要清楚syncProgress数据就行
@@ -4990,4 +5009,98 @@ protected void checkUnwindProcess(DAG dag){
         }
     }
 
+    @Override
+    public void refreshSchemas(TaskDto taskDto, String nodeIds, String keys, UserDetail userDetail) {
+        // 更新任务加载状态，使前端能自动刷新界面模型
+        updateById(taskDto.getId()
+            , Update.update(TaskDto.FIELD_TRANSFORM_STATUS, MetadataTransformerDto.StatusEnum.running.name())
+            , userDetail
+        );
+
+        String taskId = taskDto.getId().toHexString();
+        boolean isRefreshTaskSchema = false;
+        int timeout = 30 * 1000;
+        long beginTime = System.currentTimeMillis();
+        Set<String> connIdList = new LinkedHashSet<>();
+        Set<String> nodeIdList = new LinkedHashSet<>();
+        if (null == nodeIds) {
+            isRefreshTaskSchema = true;
+            keys = Optional.ofNullable(taskDto.getDag()).map(dag -> {
+                Set<String> tableNames = new LinkedHashSet<>();
+                for (Node<?> node : dag.getNodes()) {
+                    if (node instanceof DatabaseNode) {
+                        DatabaseNode databaseNode = (DatabaseNode) node;
+                        connIdList.add(databaseNode.getConnectionId());
+                    } else if (node instanceof TableNode) {
+                        TableNode tableNode = (TableNode) node;
+                        connIdList.add(tableNode.getConnectionId());
+                        String tableName = tableNode.getTableName();
+                        if (StringUtils.isNotBlank(tableName)) {
+                            tableNames.add(tableNode.getTableName());
+                        }
+                    }
+                }
+                if (tableNames.isEmpty()) return null;
+                return String.join(",", tableNames);
+            }).orElse(null);
+        } else {
+            nodeIdList.addAll(Arrays.asList(nodeIds.split(",")));
+            Optional.ofNullable(taskDto.getDag()).ifPresent(dag -> {
+                for (String nodeId : nodeIdList) {
+                    Node<?> node = dag.getNode(nodeId);
+                    if (node instanceof DataParentNode) {
+                        connIdList.add(((DataParentNode<?>) node).getConnectionId());
+                    }
+                }
+            });
+        }
+        if (connIdList.isEmpty()) return; // 没有连接不刷新模型
+
+        // 刷新连接模型
+        for (String connId : connIdList) {
+            DataSourceConnectionDto connDto = dataSourceService.findById(MongoUtils.toObjectId(connId));
+            if (null == connDto) throw new BizException(EX_TASK_NOT_FOUND_DS, connId);
+            dataSourceService.sendTestConnection(connDto, true, true, keys, userDetail);
+        }
+
+        // 等待连接加载模型完成
+        for (String connId : connIdList) {
+            wait2ConnectionsLoadFinished(taskId, MongoUtils.toObjectId(connId), beginTime, timeout, userDetail);
+        }
+
+        // 清理逻辑模型
+        if (isRefreshTaskSchema) {
+            metadataInstancesService.deleteTaskMetadata(taskId, userDetail);
+        } else {
+            for (String nodeId : nodeIdList) {
+                metadataInstancesService.deleteLogicModel(taskId, nodeId);
+            }
+        }
+
+        // 执行推演
+        transformSchemaService.transformSchema(taskDto, userDetail);
+    }
+
+    protected void wait2ConnectionsLoadFinished(String taskId, ObjectId connId, long beginTime, int timeout, UserDetail userDetail) {
+        Field fields = new Field();
+        fields.put(DataSourceConnectionDto.FIELD_LOAD_FIELDS_STATUS, 1);
+        fields.put(DataSourceConnectionDto.FIELD_LAST_UPDATE, 1);
+        DataSourceConnectionDto newConn;
+        do {
+            newConn = dataSourceService.getById(connId, fields, null, userDetail);
+            if (null != newConn.getLastUpdate() && beginTime < newConn.getLastUpdate()) {
+                if (!DataSourceConnectionDto.LOAD_FIELD_STATUS_FINISHED.equals(newConn.getLoadFieldsStatus())) {
+                    log.info("Task '{}' connection '{}' refresh failed, status={}", taskId, connId, newConn.getStatus());
+                }
+                return;
+            }
+            try {
+                Thread.sleep(1000);
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                break;
+            }
+        } while (System.currentTimeMillis() - beginTime < timeout); // 30秒超时
+        log.info("Task '{}' connection '{}' refresh timeout", taskId, connId);
+    }
 }
diff --git a/manager/tm/src/main/resources/init/dfs/version b/manager/tm/src/main/resources/init/dfs/version
index bd42c8eba0..c433f75f73 100644
--- a/manager/tm/src/main/resources/init/dfs/version
+++ b/manager/tm/src/main/resources/init/dfs/version
@@ -1,1 +1,1 @@
-3.8-2
\ No newline at end of file
+3.10-2
\ No newline at end of file
diff --git a/manager/tm/src/main/resources/init/idaas/version b/manager/tm/src/main/resources/init/idaas/version
index 52c2864109..669c6da390 100644
--- a/manager/tm/src/main/resources/init/idaas/version
+++ b/manager/tm/src/main/resources/init/idaas/version
@@ -1,1 +1,1 @@
-3.8-1
\ No newline at end of file
+3.10-1
\ No newline at end of file
diff --git a/manager/tm/src/main/resources/mailTemplate.html b/manager/tm/src/main/resources/mailTemplate.html
index 17cb487798..8ee0483c15 100644
--- a/manager/tm/src/main/resources/mailTemplate.html
+++ b/manager/tm/src/main/resources/mailTemplate.html
@@ -7,7 +7,7 @@
 <body>
 <div style="margin: 50px auto 0; font-size: 12px; color: #000; max-width: 700px">
   <div style="overflow: hidden; line-height: 30px">
-    <img src="https://auth.tapdata.net/assets/img/logo2.png" alt="" style="float: left; height: 30px" />
+    <img src="https://20778419.s21i.faiusr.com/4/2/ABUIABAEGAAg6p_HtQYo0ImmswQw1go4gAY.png" alt="" style="float: left; height: 30px" />
     <div style="float: right; color: #76808e">Make your data as accessible as tap water</div>
   </div>
   <div style="margin-top: 17px; background: #fff; box-shadow: -2px 0px 8px 0px rgb(0 0 0 / 10%)">
diff --git a/manager/tm/src/main/resources/messages.properties b/manager/tm/src/main/resources/messages.properties
index 09bc94cadb..c30441eb5c 100644
--- a/manager/tm/src/main/resources/messages.properties
+++ b/manager/tm/src/main/resources/messages.properties
@@ -50,6 +50,7 @@ Task.ScheduleLimit=The number of tasks that the engine can be called exceeds the
 Task.ManuallyScheduleLimit=The number of tasks that the engine is manually specified by the user can be called exceeds the limit. Please specify another engine {0}
 Task.ResetAgentNotFound=The reset task did not find an available engine, please start/restart the engine and try again
 Task.ResetStatusInvalid=The current status of the task that needs to be reset has been updated, please refresh the page and try again
+Task.NotfoundDatasource=Not found datasource '{0}'
 LogCollect.ConfigUpdateError=Current conditions do not allow modification
 LogCollect.ExternalStorageUpdateError=Current conditions do not allow to modify external storage
 LogCollect.UriInvalid=Mongodb uri filling is illegal
@@ -270,6 +271,6 @@ webhook.history.page.size.error=The paging parameter [pageSize] is invalid, the
 insufficient.permissions=Insufficient permissions: {0}, current operation requires permission: {1}
 inspect.result.not.exists=The verification record for the current verification task no longer exists: {0}
 
-source.setting.check.cdc=[cdc check] The configuration of the data source: {0} does not support incremental functionality, and incremental tasks cannot be configured. 
+source.setting.check.cdc=[cdc check] The configuration of the data source: {0} does not support incremental functionality, and incremental tasks cannot be configured.
 
-export.inspectDetails.failed = Export inspectDetails failed :{0}
\ No newline at end of file
+export.inspectDetails.failed = Export inspectDetails failed :{0}
diff --git a/manager/tm/src/main/resources/messages_zh_CN.properties b/manager/tm/src/main/resources/messages_zh_CN.properties
index 742ba2d208..aec104cfc3 100644
--- a/manager/tm/src/main/resources/messages_zh_CN.properties
+++ b/manager/tm/src/main/resources/messages_zh_CN.properties
@@ -26,14 +26,14 @@ Inspect.Name.Exist=\u540D\u79F0\u5DF2\u5B58\u5728\uFF0C\u8BF7\u4FEE\u6539
 Inspect.ProcessId.NotFound= \u627E\u4E0D\u5230\u5BF9\u5E94\u7684agent
 Inspect.agentTag.null=tag\u4E0D\u80FD\u4E3A\u7A7A
 Inspect.task.null=task\u4E0D\u80FD\u4E3A\u7A7A
-Inspect.Recovery.NotFieldMethod=\u975e\u5b57\u6bb5\u6821\u9a8c\u4efb\u52a1
-Inspect.Recovery.StatusNotDone=\u6821\u9a8c\u4efb\u52a1\u4e0d\u662f\u5df2\u5b8c\u6210\u72b6\u6001
-Inspect.Recovery.ResultNotFound=\u6821\u9a8c\u4efb\u52a1\u6ca1\u6709\u5dee\u5f02\u7ed3\u679c
-Inspect.Recovery.IsNotWithTask=\u6821\u9a8c\u4e0d\u662f\u57fa\u4e8e\u540c\u6b65\u4efb\u52a1\u521b\u5efa
-Inspect.Recovery.TaskNotFound=\u540c\u6b65\u4efb\u52a1\u4e0d\u5b58\u5728
-Inspect.Recovery.TaskTypeIsNotCDC=\u540c\u6b65\u4efb\u52a1\u4e0d\u5305\u542b\u589e\u91cf
-Inspect.Recovery.TaskIsNotRunning=\u540c\u6b65\u4efb\u52a1\u975e\u8fd0\u884c\u72b6\u6001
-Inspect.Recovery.TaskSyncStatusIsNotCDC=\u540c\u6b65\u4efb\u52a1\u4e0d\u5728\u589e\u91cf\u9636\u6bb5
+Inspect.Recovery.NotFieldMethod=\u975E\u5B57\u6BB5\u6821\u9A8C\u4EFB\u52A1
+Inspect.Recovery.StatusNotDone=\u6821\u9A8C\u4EFB\u52A1\u4E0D\u662F\u5DF2\u5B8C\u6210\u72B6\u6001
+Inspect.Recovery.ResultNotFound=\u6821\u9A8C\u4EFB\u52A1\u6CA1\u6709\u5DEE\u5F02\u7ED3\u679C
+Inspect.Recovery.IsNotWithTask=\u6821\u9A8C\u4E0D\u662F\u57FA\u4E8E\u540C\u6B65\u4EFB\u52A1\u521B\u5EFA
+Inspect.Recovery.TaskNotFound=\u540C\u6B65\u4EFB\u52A1\u4E0D\u5B58\u5728
+Inspect.Recovery.TaskTypeIsNotCDC=\u540C\u6B65\u4EFB\u52A1\u4E0D\u5305\u542B\u589E\u91CF
+Inspect.Recovery.TaskIsNotRunning=\u540C\u6B65\u4EFB\u52A1\u975E\u8FD0\u884C\u72B6\u6001
+Inspect.Recovery.TaskSyncStatusIsNotCDC=\u540C\u6B65\u4EFB\u52A1\u4E0D\u5728\u589E\u91CF\u9636\u6BB5
 
 
 
@@ -74,6 +74,7 @@ Task.ScheduleLimit=\u5F15\u64CE\u53EF\u4EE5\u88AB\u8C03\u7528\u7684\u4EFB\u52A1\
 Task.ManuallyScheduleLimit=\u7528\u6237\u6307\u5B9A\u5F15\u64CE\u53EF\u4EE5\u88AB\u8C03\u7528\u7684\u4EFB\u52A1\u8D85\u8FC7\u4E86\u9650\u5236\u6570\uFF0C\u8BF7\u66FF\u6362\u5176\u4ED6\u5F15\u64CE {0}
 Task.ResetAgentNotFound=\u91CD\u7F6E\u4EFB\u52A1\u6CA1\u6709\u627E\u5230\u53EF\u7528\u7684\u5F15\u64CE\uFF0C\u8BF7\u542F\u52A8/\u91CD\u542F\u5F15\u64CE\u540E\u518D\u8BD5
 Task.ResetStatusInvalid=\u5F53\u524D\u9700\u8981\u91CD\u7F6E\u7684\u4EFB\u52A1\u72B6\u6001\u88AB\u66F4\u65B0\uFF0C\u8BF7\u5237\u65B0\u9875\u9762\u518D\u8BD5
+Task.NotfoundDatasource=\u627E\u4E0D\u5230\u6570\u636E\u6E90 '{0}'
 LogCollect.ConfigUpdateError=\u5F53\u524D\u6761\u4EF6\u4E0D\u5141\u8BB8\u4FEE\u6539
 LogCollect.ExternalStorageUpdateError=\u5F53\u524D\u6761\u4EF6\u4E0D\u5141\u8BB8\u4FEE\u6539\u5916\u5B58\u914D\u7F6E
 LogCollect.UriInvalid=Mongodb uri \u586B\u5199\u4E0D\u5408\u6CD5
@@ -121,8 +122,8 @@ Sid.is.wrong=sid\u9519\u8BEF
 
 Modules.BasePathAndVersion.Existed=\u8BE5\u7248\u672C\u4E0Bbasepath\u540D\u79F0\u5DF2\u5B58\u5728
 Modules.Connection.Null=API\u0020\u6570\u636E\u5E93\u8FDE\u63A5\u4E0D\u80FD\u4E3A\u7A7A
-Modules.Permission.Scope.Null=\u6743\u9650\u8303\u56f4\u4e0d\u80fd\u4e3a\u7a7a
-Modules.Tags.Null = \u6240\u5c5e\u5e94\u7528\u4e0d\u80fd\u4e3a\u7a7a
+Modules.Permission.Scope.Null=\u6743\u9650\u8303\u56F4\u4E0D\u80FD\u4E3A\u7A7A
+Modules.Tags.Null = \u6240\u5C5E\u5E94\u7528\u4E0D\u80FD\u4E3A\u7A7A
 
 AccessCode.Is.Null=AccessCode\u0020\u4E3A\u7A7A
 AccessCode.No.User=AccessCode\u0020\u627E\u4E0D\u5230\u5BF9\u5E94\u7684\u7528\u6237
@@ -299,6 +300,8 @@ webhook.history.page.size.error=\u5206\u9875\u53C2\u6570[pageSize]\u65E0\u6548\u
 insufficient.permissions=\u6743\u9650\u4E0D\u8DB3: {0}\uFF0C\u5F53\u524D\u64CD\u4F5C\u9700\u8981\u6743\u9650\uFF1A{1}
 inspect.result.not.exists=\u5F53\u524D\u6821\u9A8C\u4EFB\u52A1\u7684\u6821\u9A8C\u8BB0\u5F55\u5DF2\u7ECF\u4E0D\u5B58\u5728\uFF1A{0}
 
-source.setting.check.cdc= \u589e\u91cf\u68c0\u67e5\u3011\u0020\u6570\u636e\u6e90\u003a\u007b\u0030\u007d\u7684\u589e\u91cf\u529f\u80fd\u4e0d\u652f\u6301\uff0c\u4e0d\u80fd\u914d\u7f6e\u589e\u91cf\u4efb\u52a1\u000d\u000a
+source.setting.check.cdc= \u589E\u91CF\u68C0\u67E5\u3011\u0020\u6570\u636E\u6E90:{0}\u7684\u589E\u91CF\u529F\u80FD\u4E0D\u652F\u6301\uFF0C\u4E0D\u80FD\u914D\u7F6E\u589E\u91CF\u4EFB\u52A1\u000d\u000a
 
-export.inspectDetails.failed =\u4e0b\u8f7d\u6821\u9a8c\u8be6\u60c5\u5931\u8d25 :{0}
\ No newline at end of file
+export.inspectDetails.failed =\u4E0B\u8F7D\u6821\u9A8C\u8BE6\u60C5\u5931\u8D25 :{0}
+
+export.inspectDetails.failed =\u4E0B\u8F7D\u6821\u9A8C\u8BE6\u60C5\u5931\u8D25 :{0}
\ No newline at end of file


Test patch:
diff --git a/iengine/iengine-app/src/test/java/io/tapdata/Runnable/LoadSchemaRunnerTest.java b/iengine/iengine-app/src/test/java/io/tapdata/Runnable/LoadSchemaRunnerTest.java
new file mode 100644
index 0000000000..b75622abc7
--- /dev/null
+++ b/iengine/iengine-app/src/test/java/io/tapdata/Runnable/LoadSchemaRunnerTest.java
@@ -0,0 +1,116 @@
+package io.tapdata.Runnable;
+
+import com.mongodb.client.result.UpdateResult;
+import com.tapdata.constant.UUIDGenerator;
+import com.tapdata.entity.Connections;
+import com.tapdata.entity.LoadSchemaProgress;
+import com.tapdata.mongo.ClientMongoOperator;
+import com.tapdata.tm.commons.schema.DataSourceConnectionDto;
+import io.tapdata.utils.UnitTestUtils;
+import org.bson.Document;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+import org.springframework.data.mongodb.core.query.Query;
+import org.springframework.data.mongodb.core.query.Update;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+/**
+ * @author <a href="mailto:harsen_lin@163.com">Harsen</a>
+ * @version v1.0 2024/7/24 10:45 Create
+ */
+class LoadSchemaRunnerTest {
+    LoadSchemaRunner runner;
+    long lastUpdate;
+    String schemaVersion;
+    LoadSchemaProgress loadSchemaProgress;
+    Connections connections;
+    ClientMongoOperator clientMongoOperator;
+
+    @BeforeEach
+    void setUp() {
+        runner = mock(LoadSchemaRunner.class);
+
+        lastUpdate = System.currentTimeMillis();
+        UnitTestUtils.injectField(LoadSchemaRunner.class, runner, "lastUpdate", lastUpdate);
+        schemaVersion = UUIDGenerator.uuid();
+        UnitTestUtils.injectField(LoadSchemaRunner.class, runner, "schemaVersion", schemaVersion);
+        loadSchemaProgress = mock(LoadSchemaProgress.class);
+        UnitTestUtils.injectField(LoadSchemaRunner.class, runner, "loadSchemaProgress", loadSchemaProgress);
+        connections = mock(Connections.class);
+        UnitTestUtils.injectField(LoadSchemaRunner.class, runner, "connections", connections);
+        clientMongoOperator = mock(ClientMongoOperator.class);
+        UnitTestUtils.injectField(LoadSchemaRunner.class, runner, "clientMongoOperator", clientMongoOperator);
+    }
+
+    @Nested
+    class UpdateConnections2LoadingTest {
+
+        @Test
+        void testFullUpdate() {
+            doCallRealMethod().when(runner).updateConnections2Loading(connections);
+
+            assertFalse(runner.updateConnections2Loading(connections));
+            verify(runner, times(1)).updateConnections(any());
+        }
+
+        @Test
+        void testPartialUpdateAndReturnFalse() {
+            String connId = "test-conn-id";
+            String oldSchemaVersion = "test-old-version";
+            String partialUpdateFilter = "test-partial-update-filter";
+            String loadFieldsStatus = DataSourceConnectionDto.LOAD_FIELD_STATUS_LOADING;
+
+            UpdateResult updateResult = mock(UpdateResult.class);
+            when(updateResult.getMatchedCount()).thenReturn(1L);
+            when(connections.getId()).thenReturn(connId);
+            when(connections.getPartialUpdateWithSchemaVersion()).thenReturn(oldSchemaVersion);
+            when(connections.getPartialUpdateFilter()).thenReturn(partialUpdateFilter);
+            doAnswer(invocation -> {
+                Document doc = invocation.<Query>getArgument(0).getQueryObject();
+                assertEquals(connId, doc.getString("_id"));
+                assertEquals(oldSchemaVersion, doc.get(DataSourceConnectionDto.FIELD_SCHEMA_VERSION));
+
+                doc = invocation.<Update>getArgument(1).getUpdateObject().get("$set", Document.class);
+                assertEquals(loadFieldsStatus, doc.getString(DataSourceConnectionDto.FIELD_LOAD_FIELDS_STATUS));
+                assertEquals(partialUpdateFilter, doc.getString(DataSourceConnectionDto.FIELD_PARTIAL_UPDATE_FILTER));
+                assertEquals(lastUpdate, doc.getLong(DataSourceConnectionDto.FIELD_LAST_UPDATE));
+                assertEquals(schemaVersion, doc.getString(DataSourceConnectionDto.FIELD_SCHEMA_VERSION));
+                return updateResult;
+            }).when(clientMongoOperator).update(any(Query.class), any(Update.class), anyString());
+            doCallRealMethod().when(runner).updateConnections2Loading(eq(connections));
+
+            assertFalse(runner.updateConnections2Loading(connections));
+            verify(clientMongoOperator, times(1)).update(any(Query.class), any(Update.class), anyString());
+        }
+
+        @Test
+        void testPartialUpdateAndNullFilterAndReturnFalse() {
+            String connId = "test-conn-id";
+            String oldSchemaVersion = "test-old-version";
+            String loadFieldsStatus = DataSourceConnectionDto.LOAD_FIELD_STATUS_LOADING;
+
+            UpdateResult updateResult = mock(UpdateResult.class);
+            when(connections.getId()).thenReturn(connId);
+            when(connections.getPartialUpdateWithSchemaVersion()).thenReturn(oldSchemaVersion);
+            doAnswer(invocation -> {
+                Document doc = invocation.<Query>getArgument(0).getQueryObject();
+                assertEquals(connId, doc.getString("_id"));
+                assertEquals(oldSchemaVersion, doc.get(DataSourceConnectionDto.FIELD_SCHEMA_VERSION));
+
+                doc = invocation.<Update>getArgument(1).getUpdateObject().get("$set", Document.class);
+                assertEquals(loadFieldsStatus, doc.getString(DataSourceConnectionDto.FIELD_LOAD_FIELDS_STATUS));
+                assertNull(doc.getString(DataSourceConnectionDto.FIELD_PARTIAL_UPDATE_FILTER));
+                assertEquals(lastUpdate, doc.getLong(DataSourceConnectionDto.FIELD_LAST_UPDATE));
+                assertEquals(schemaVersion, doc.getString(DataSourceConnectionDto.FIELD_SCHEMA_VERSION));
+                return updateResult;
+            }).when(clientMongoOperator).update(any(Query.class), any(Update.class), anyString());
+            doCallRealMethod().when(runner).updateConnections2Loading(eq(connections));
+
+            assertTrue(runner.updateConnections2Loading(connections));
+            verify(clientMongoOperator, times(1)).update(any(Query.class), any(Update.class), anyString());
+        }
+    }
+}
diff --git a/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/filter/FilterUtilTest.java b/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/filter/FilterUtilTest.java
index 9b0063f576..d9cdd58853 100644
--- a/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/filter/FilterUtilTest.java
+++ b/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/filter/FilterUtilTest.java
@@ -67,8 +67,7 @@ public void processTableFieldsOfNullEventData() {
         Set<String> fields = new HashSet<>();
         fields.add(key1);
         eventData = FilterUtil.processTableFields(eventData, fields);
-        Assert.assertTrue(fields.size() > eventData.size());
-        Assert.assertFalse(eventData.containsKey(key1));
+        Assert.assertNull(eventData);
 
     }
 
diff --git a/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastTargetPdkBaseNodeTest.java b/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastTargetPdkBaseNodeTest.java
index 6b73359219..ce852b71d2 100644
--- a/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastTargetPdkBaseNodeTest.java
+++ b/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastTargetPdkBaseNodeTest.java
@@ -19,8 +19,10 @@
 import com.tapdata.tm.commons.dag.process.MergeTableNode;
 import com.tapdata.tm.commons.dag.process.UnwindProcessNode;
 import com.tapdata.tm.commons.task.dto.TaskDto;
+import com.tapdata.tm.commons.util.JsonUtil;
 import io.tapdata.aspect.CreateTableFuncAspect;
 import io.tapdata.aspect.DropTableFuncAspect;
+import io.tapdata.aspect.utils.AspectUtils;
 import io.tapdata.entity.codec.filter.TapCodecsFilterManager;
 import io.tapdata.entity.event.TapEvent;
 import io.tapdata.entity.event.ddl.table.TapCreateTableEvent;
@@ -57,6 +59,7 @@
 import org.mockito.MockedStatic;
 import org.mockito.Mockito;
 import org.mockito.internal.verification.Times;
+import org.springframework.beans.BeanUtils;
 import org.springframework.test.util.ReflectionTestUtils;
 
 import java.util.*;
@@ -366,6 +369,34 @@ void createTableTestForInit(){
 			hazelcastTargetPdkBaseNode.createTable(tapTable, succeed,true);
 		}
 
+		@Test
+		void createTableFalseTestForInit() {
+			ReflectionTestUtils.setField(hazelcastTargetPdkBaseNode, "unwindProcess", false);
+			TapTable tapTable = new TapTable();
+			tapTable.setId("test");
+			AtomicBoolean succeed = new AtomicBoolean(true);
+			Node node = mock(Node.class);
+			when(hazelcastTargetPdkBaseNode.getNode()).thenReturn(node);
+			when(node.disabledNode()).thenReturn(false);
+			ConnectorNode connectorNode = mock(ConnectorNode.class);
+			when(hazelcastTargetPdkBaseNode.getConnectorNode()).thenReturn(connectorNode);
+			ConnectorFunctions functions = mock(ConnectorFunctions.class);
+			when(connectorNode.getConnectorFunctions()).thenReturn(functions);
+			when(functions.getCreateTableFunction()).thenReturn(null);
+			when(dataProcessorContext.getTargetConn()).thenReturn(mock(Connections.class));
+			try (MockedStatic<AspectUtils> aspectUtilsMockedStatic = Mockito.mockStatic(AspectUtils.class)) {
+				aspectUtilsMockedStatic.when(() -> AspectUtils.executeAspect(any())).then(a -> {
+					CreateTableFuncAspect createTableFuncAspect = a.getArgument(0);
+					Assertions.assertTrue(createTableFuncAspect.isInit());
+					return null;
+				});
+
+				doCallRealMethod().when(hazelcastTargetPdkBaseNode).createTable(tapTable, succeed, true);
+				hazelcastTargetPdkBaseNode.createTable(tapTable, succeed, true);
+
+			}
+		}
+
 		@Test
 		void dropTableTestForInit(){
 			TaskDto taskDto = new TaskDto();
@@ -400,6 +431,42 @@ void dropTableTestForInit(){
 			doCallRealMethod().when(hazelcastTargetPdkDataNode).dropTable(existsDataProcessEnum,"test",true);
 			hazelcastTargetPdkDataNode.dropTable(existsDataProcessEnum, "test",true);
 		}
+
+		@Test
+		void dropTableFalseTestForInit(){
+			TaskDto taskDto = new TaskDto();
+			taskDto.setType("initial_sync");
+			when(dataProcessorContext.getTaskDto()).thenReturn(taskDto);
+			HazelcastTargetPdkDataNode	hazelcastTargetPdkDataNode = mock(HazelcastTargetPdkDataNode.class);
+			ReflectionTestUtils.setField(hazelcastTargetPdkDataNode, "dataProcessorContext", dataProcessorContext);
+			ReflectionTestUtils.setField(hazelcastTargetPdkDataNode, "clientMongoOperator", mockClientMongoOperator);
+			TapTable tapTable = new TapTable();
+			tapTable.setId("test");
+			Node node = mock(Node.class);
+			when(hazelcastTargetPdkDataNode.getNode()).thenReturn(node);
+			when(node.disabledNode()).thenReturn(false);
+			ConnectorNode connectorNode = mock(ConnectorNode.class);
+			when(hazelcastTargetPdkDataNode.getConnectorNode()).thenReturn(connectorNode);
+			ConnectorFunctions functions = mock(ConnectorFunctions.class);
+			when(connectorNode.getConnectorFunctions()).thenReturn(functions);
+			when(functions.getDropTableFunction()).thenReturn(null);
+			when(dataProcessorContext.getTargetConn()).thenReturn(mock(Connections.class));
+
+
+			try (MockedStatic<AspectUtils> aspectUtilsMockedStatic = Mockito.mockStatic(AspectUtils.class)) {
+				aspectUtilsMockedStatic.when(() -> AspectUtils.executeAspect(any())).then(a -> {
+					DropTableFuncAspect actualData = a.getArgument(0);
+					Assertions.assertTrue(actualData.isInit());
+					return null;
+				});
+
+				ExistsDataProcessEnum existsDataProcessEnum = ExistsDataProcessEnum.DROP_TABLE;
+				doCallRealMethod().when(hazelcastTargetPdkDataNode).dropTable(existsDataProcessEnum,"test",true);
+				hazelcastTargetPdkDataNode.dropTable(existsDataProcessEnum, "test",true);
+
+			}
+
+		}
 	}
 
 	@Nested
diff --git a/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/processor/HazelcastMergeNodeTest.java b/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/processor/HazelcastMergeNodeTest.java
index baa70f51ad..351a623a4b 100644
--- a/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/processor/HazelcastMergeNodeTest.java
+++ b/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/processor/HazelcastMergeNodeTest.java
@@ -1687,6 +1687,7 @@ class NeedLookUpTest {
 		List<String> needCacheList = new ArrayList<>();
 		private Map<String, List<MergeTableProperties>> lookupMap = new HashMap<>();
 		private Set<String> firstLevelMergeNodeIds = new HashSet<>();
+		private Map<String, HazelcastMergeNode.EnableUpdateJoinKey> enableUpdateJoinKeyMap;
 
 		@BeforeEach
 		void setUp() {
@@ -1697,6 +1698,8 @@ void setUp() {
 			lookupMap.put("123", mergeTableProperties);
 			ReflectionTestUtils.setField(mockHazelcastMergeNode, "lookupMap", lookupMap);
 			ReflectionTestUtils.setField(mockHazelcastMergeNode, "firstLevelMergeNodeIds", firstLevelMergeNodeIds);
+			enableUpdateJoinKeyMap = new HashMap<>();
+			ReflectionTestUtils.setField(mockHazelcastMergeNode, "enableUpdateJoinKeyMap", enableUpdateJoinKeyMap);
 		}
 
 		@DisplayName("test task is initalSync and mergeMode is mainTableFirst")
@@ -1733,8 +1736,10 @@ void test3() {
 			processorBaseContext.getTaskDto().setType(SyncTypeEnum.CDC.getSyncType());
 			doReturn(false).when(mockHazelcastMergeNode).isSubTableFirstMode();
 			TapdataEvent tapdataEvent = new TapdataEvent();
+			tapdataEvent.addNodeId("123");
 			TapUpdateRecordEvent tapUpdateRecordEvent = TapUpdateRecordEvent.create().init();
 			tapdataEvent.setTapEvent(tapUpdateRecordEvent);
+			enableUpdateJoinKeyMap.computeIfAbsent("123", k -> new HazelcastMergeNode.EnableUpdateJoinKey());
 			assertFalse(mockHazelcastMergeNode.needLookup(tapdataEvent));
 
 			TapDeleteRecordEvent tapDeleteRecordEvent = TapDeleteRecordEvent.create().init();
@@ -1742,6 +1747,24 @@ void test3() {
 			tapdataEvent.setTapEvent(tapDeleteRecordEvent);
 			assertFalse(mockHazelcastMergeNode.needLookup(tapdataEvent));
 		}
+
+		@Test
+		@DisplayName("test cdc update event, and enable update join key is true, expect return true")
+		void test4() {
+			processorBaseContext.getTaskDto().setType(SyncTypeEnum.CDC.getSyncType());
+			doReturn(false).when(mockHazelcastMergeNode).isSubTableFirstMode();
+			TapdataEvent tapdataEvent = new TapdataEvent();
+			tapdataEvent.addNodeId("123");
+			TapUpdateRecordEvent tapUpdateRecordEvent = TapUpdateRecordEvent.create().init();
+			tapdataEvent.setTapEvent(tapUpdateRecordEvent);
+			enableUpdateJoinKeyMap.computeIfAbsent("123", k -> {
+				HazelcastMergeNode.EnableUpdateJoinKey enableUpdateJoinKey = new HazelcastMergeNode.EnableUpdateJoinKey();
+				enableUpdateJoinKey.enableChildren();
+				return enableUpdateJoinKey;
+			});
+			tapdataEvent.setSyncStage(SyncStage.CDC);
+			assertTrue(mockHazelcastMergeNode.needLookup(tapdataEvent));
+		}
 	}
 
 	@Nested
diff --git a/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/schedule/TapdataTaskSchedulerTest.java b/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/schedule/TapdataTaskSchedulerTest.java
index 443b49f81a..5d72d44955 100644
--- a/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/schedule/TapdataTaskSchedulerTest.java
+++ b/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/schedule/TapdataTaskSchedulerTest.java
@@ -5,11 +5,11 @@
 import com.tapdata.entity.ResponseBody;
 import com.tapdata.mongo.ClientMongoOperator;
 import com.tapdata.tm.commons.task.dto.TaskDto;
+import com.tapdata.tm.commons.task.dto.TaskOpRespDto;
 import com.tapdata.tm.sdk.available.TmStatusService;
 import io.tapdata.entity.utils.DataMap;
 import io.tapdata.exception.RestDoNotRetryException;
 import io.tapdata.exception.TmUnavailableException;
-import io.tapdata.flow.engine.V2.task.OpType;
 import io.tapdata.flow.engine.V2.task.TaskClient;
 import io.tapdata.flow.engine.V2.task.impl.HazelcastTaskClient;
 import io.tapdata.flow.engine.V2.task.operation.StartTaskOperation;
@@ -23,6 +23,7 @@
 import io.tapdata.utils.UnitTestUtils;
 import org.apache.logging.log4j.Logger;
 import org.bson.types.ObjectId;
+import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.DisplayName;
 import org.junit.jupiter.api.Nested;
@@ -432,4 +433,171 @@ void testHandleTaskOperation() {
 		TapdataTaskScheduler instance = new TapdataTaskScheduler();
 		instance.handleTaskOperation(taskOperation);
 	}
+
+	@Nested
+	class StopTaskCallAssignApiTest {
+		TapdataTaskScheduler scheduler;
+		Logger logger;
+		TaskClient<TaskDto> taskClient;
+		TapdataTaskScheduler.StopTaskResource stopTaskResource;
+		ClientMongoOperator clientMongoOperator;
+
+		TaskDto taskDto;
+		TaskOpRespDto taskOpRespDto;
+		@BeforeEach
+		void init() {
+			scheduler = mock(TapdataTaskScheduler.class);
+			logger = mock(Logger.class);
+			taskClient = mock(TaskClient.class);
+			stopTaskResource = mock(TapdataTaskScheduler.StopTaskResource.class);
+			clientMongoOperator = mock(ClientMongoOperator.class);
+			ReflectionTestUtils.setField(scheduler, "logger", logger);
+			ReflectionTestUtils.setField(scheduler, "clientMongoOperator", clientMongoOperator);
+
+			taskDto = mock(TaskDto.class);
+			taskOpRespDto = mock(TaskOpRespDto.class);
+
+			when(taskDto.getId()).thenReturn(new ObjectId());
+			when(taskDto.getName()).thenReturn("name");
+			when(taskClient.getTask()).thenReturn(taskDto);
+			when(taskClient.stop()).thenReturn(true);
+			when(stopTaskResource.getResource()).thenReturn("source");
+			doNothing().when(logger).info(anyString(), anyString(), anyString());
+			doNothing().when(logger).warn(anyString(), anyString(), anyString(), any(Exception.class));
+			doNothing().when(logger).warn(anyString(), anyString(), anyString(), anyString(), any(Exception.class));
+			when(clientMongoOperator.updateById(any(Update.class), anyString(), anyString(), any(Class.class))).thenReturn(taskOpRespDto);
+			when(taskOpRespDto.getSuccessIds()).thenReturn(new ArrayList<>());
+
+			when(scheduler.stopTaskCallAssignApi(taskClient, stopTaskResource)).thenCallRealMethod();
+		}
+
+		@Test
+		void taskClientIsNull() {
+			when(scheduler.stopTaskCallAssignApi(null, stopTaskResource)).thenCallRealMethod();
+			Assertions.assertTrue(scheduler.stopTaskCallAssignApi(null, stopTaskResource));
+		}
+
+		@Test
+		void taskIsNull() {
+			when(taskClient.getTask()).thenReturn(null);
+			Assertions.assertTrue(scheduler.stopTaskCallAssignApi(taskClient, stopTaskResource));
+		}
+
+		@Test
+		void taskIdIsNull() {
+			ObjectId is = mock(ObjectId.class);
+			when(is.toHexString()).thenReturn("");
+			when(taskDto.getId()).thenReturn(is);
+			Assertions.assertTrue(scheduler.stopTaskCallAssignApi(taskClient, stopTaskResource));
+		}
+
+		@Test
+		void testNormal() {
+			Assertions.assertFalse(scheduler.stopTaskCallAssignApi(taskClient, stopTaskResource));
+			verify(taskClient, times(3)).getTask();
+			verify(taskClient, times(1)).stop();
+			verify(taskDto, times(1)).getName();
+			verify(taskDto, times(2)).getId();
+			verify(clientMongoOperator, times(1)).updateById(any(Update.class), anyString(), anyString(), any(Class.class));
+			verify(logger, times(1)).info(anyString(), anyString(), anyString());
+			verify(logger, times(0)).warn(anyString(), any(Exception.class));
+			verify(taskOpRespDto, times(1)).getSuccessIds();
+		}
+		@Test
+		void testNotStop() {
+			when(taskClient.stop()).thenReturn(false);
+			Assertions.assertFalse(scheduler.stopTaskCallAssignApi(taskClient, stopTaskResource));
+			verify(taskClient, times(2)).getTask();
+			verify(taskClient, times(1)).stop();
+			verify(taskDto, times(0)).getName();
+			verify(taskDto, times(1)).getId();
+			verify(clientMongoOperator, times(0)).updateById(any(Update.class), anyString(), anyString(), any(Class.class));
+			verify(logger, times(0)).info(anyString(), anyString(), anyString());
+			verify(logger, times(0)).warn(anyString(), anyString(), anyString(), any(Exception.class));
+			verify(taskOpRespDto, times(0)).getSuccessIds();
+		}
+		@Test
+		void testNotEmpty() {
+			List<String> objects = new ArrayList<>();
+			objects.add("");
+			when(taskOpRespDto.getSuccessIds()).thenReturn(objects);
+			Assertions.assertTrue(scheduler.stopTaskCallAssignApi(taskClient, stopTaskResource));
+			verify(taskClient, times(3)).getTask();
+			verify(taskClient, times(1)).stop();
+			verify(taskDto, times(1)).getName();
+			verify(taskDto, times(2)).getId();
+			verify(clientMongoOperator, times(1)).updateById(any(Update.class), anyString(), anyString(), any(Class.class));
+			verify(logger, times(1)).info(anyString(), anyString(), anyString());
+			verify(logger, times(0)).warn(anyString(), anyString(), anyString(), any(Exception.class));
+			verify(taskOpRespDto, times(1)).getSuccessIds();
+		}
+		@Test
+		void testThrow() {
+			when(clientMongoOperator.updateById(any(Update.class), anyString(), anyString(), any(Class.class))).thenAnswer(a -> {
+				throw new Exception("");
+			});
+			Assertions.assertFalse(scheduler.stopTaskCallAssignApi(taskClient, stopTaskResource));
+			verify(taskClient, times(3)).getTask();
+			verify(taskClient, times(1)).stop();
+			verify(taskDto, times(1)).getName();
+			verify(taskDto, times(2)).getId();
+			verify(clientMongoOperator, times(1)).updateById(any(Update.class), anyString(), anyString(), any(Class.class));
+			verify(logger, times(1)).info(anyString(), anyString(), anyString());
+			verify(logger, times(0)).warn(anyString(), anyString(), anyString(), any(Exception.class));
+			verify(taskOpRespDto, times(0)).getSuccessIds();
+		}
+		@Test
+		void testThrow2() {
+			when(clientMongoOperator.updateById(any(Update.class), anyString(), anyString(), any(Class.class))).thenAnswer(a -> {
+				throw new Exception("222");
+			});
+			Assertions.assertFalse(scheduler.stopTaskCallAssignApi(taskClient, stopTaskResource));
+			verify(taskClient, times(3)).getTask();
+			verify(taskClient, times(1)).stop();
+			verify(taskDto, times(1)).getName();
+			verify(taskDto, times(2)).getId();
+			verify(clientMongoOperator, times(1)).updateById(any(Update.class), anyString(), anyString(), any(Class.class));
+			verify(logger, times(1)).info(anyString(), anyString(), anyString());
+			verify(logger, times(0)).warn(anyString(), anyString(), anyString(), any(Exception.class));
+			verify(taskOpRespDto, times(0)).getSuccessIds();
+		}
+		@Test
+		void testThrowTransitionNoSupport() {
+			when(clientMongoOperator.updateById(any(Update.class), anyString(), anyString(), any(Class.class))).thenAnswer(a -> {
+				throw new Exception("Transition.Not.Supported");
+			});
+			Assertions.assertThrows(Exception.class, () -> scheduler.stopTaskCallAssignApi(taskClient, stopTaskResource));
+			verify(taskClient, times(3)).getTask();
+			verify(taskClient, times(1)).stop();
+			verify(taskDto, times(1)).getName();
+			verify(taskDto, times(2)).getId();
+			verify(clientMongoOperator, times(2)).updateById(any(Update.class), anyString(), anyString(), any(Class.class));
+			verify(logger, times(1)).info(anyString(), anyString(), anyString());
+			verify(logger, times(1)).warn(anyString(), anyString(), anyString(), any(Exception.class));
+			verify(taskOpRespDto, times(0)).getSuccessIds();
+		}
+		@Test
+		void testThrowTransitionNoSupport2() {
+			final AtomicBoolean isFirst = new AtomicBoolean(true);
+			when(clientMongoOperator.updateById(any(Update.class), anyString(), anyString(), any(Class.class))).thenAnswer(a -> {
+				synchronized (isFirst) {
+					if (isFirst.get()) {
+						isFirst.set(false);
+						throw new Exception("Transition.Not.Supported");
+					} else {
+						return null;
+					}
+				}
+			});
+			Assertions.assertTrue(scheduler.stopTaskCallAssignApi(taskClient, stopTaskResource));
+			verify(taskClient, times(3)).getTask();
+			verify(taskClient, times(1)).stop();
+			verify(taskDto, times(1)).getName();
+			verify(taskDto, times(2)).getId();
+			verify(clientMongoOperator, times(2)).updateById(any(Update.class), anyString(), anyString(), any(Class.class));
+			verify(logger, times(1)).info(anyString(), anyString(), anyString());
+			verify(logger, times(1)).warn(anyString(), anyString(), anyString(), any(Exception.class));
+			verify(taskOpRespDto, times(0)).getSuccessIds();
+		}
+	}
 }
diff --git a/manager/tm-api/src/test/java/com/tapdata/tm/utils/MailUtilsTest.java b/manager/tm-api/src/test/java/com/tapdata/tm/utils/MailUtilsTest.java
index 4f087eaa08..7bcafc89ab 100644
--- a/manager/tm-api/src/test/java/com/tapdata/tm/utils/MailUtilsTest.java
+++ b/manager/tm-api/src/test/java/com/tapdata/tm/utils/MailUtilsTest.java
@@ -90,6 +90,26 @@ void testSendHtmlMail_subjectNotNull(){
             throw new RuntimeException(e);
         }
     }
+    @Test
+    void testSendHtmlMail_subjectNotNull_mailContentIsNull(){
+        String mailContent = null;
+        when(blacklistService.inBlacklist(anyString())).thenReturn(false);
+        List<String> addressList = new ArrayList<>();
+        addressList.add("test@qq.com");
+        SendStatus result = mailUtils.sendHtmlMail("",addressList,"test","test","test",mailContent);
+        Assertions.assertEquals("false",result.getStatus());
+        Assertions.assertTrue(result.getErrorMessage().contains("mailContent"));
+    }
+    @Test
+    void testSendHtmlMail_subjectNotNull_mailContentIsEmpty(){
+        String mailContent = "";
+        when(blacklistService.inBlacklist(anyString())).thenReturn(false);
+        List<String> addressList = new ArrayList<>();
+        addressList.add("test@qq.com");
+        SendStatus result = mailUtils.sendHtmlMail("",addressList,"test","test","test",mailContent);
+        Assertions.assertEquals("false",result.getStatus());
+        Assertions.assertTrue(result.getErrorMessage().contains("mailContent"));
+    }
 
     @Test
     void testSendHtmlMail_subjectNotNull_notInBlacklistAddressIsNull(){
diff --git a/manager/tm-api/src/test/resources/mailTemplate.html b/manager/tm-api/src/test/resources/mailTemplate.html
index 344eed6339..f18d1c70eb 100644
--- a/manager/tm-api/src/test/resources/mailTemplate.html
+++ b/manager/tm-api/src/test/resources/mailTemplate.html
@@ -7,7 +7,7 @@
 <body>
 <div style="margin: 50px auto 0; font-size: 12px; color: #000; max-width: 700px">
   <div style="overflow: hidden; line-height: 30px">
-    <img src="https://cloud.tapdata.net/assets/img/logo2.png" alt="" style="float: left; height: 30px" />
+    <img src="https://20778419.s21i.faiusr.com/4/2/ABUIABAEGAAg6p_HtQYo0ImmswQw1go4gAY.png" alt="" style="float: left; height: 30px" />
     <div style="float: right; color: #76808e">让数据随时可用</div>
   </div>
   <div style="margin-top: 17px; background: #fff; box-shadow: -2px 0px 8px 0px rgb(0 0 0 / 10%)">
diff --git a/manager/tm-common/src/test/java/com/tapdata/tm/commons/cdcdelay/CdcDelayTest.java b/manager/tm-common/src/test/java/com/tapdata/tm/commons/cdcdelay/CdcDelayTest.java
index 7d1bfaebeb..34f022ad76 100644
--- a/manager/tm-common/src/test/java/com/tapdata/tm/commons/cdcdelay/CdcDelayTest.java
+++ b/manager/tm-common/src/test/java/com/tapdata/tm/commons/cdcdelay/CdcDelayTest.java
@@ -4,14 +4,19 @@
 import com.tapdata.tm.commons.util.ConnHeartbeatUtils;
 import io.tapdata.entity.event.TapEvent;
 import io.tapdata.entity.event.control.HeartbeatEvent;
+import io.tapdata.entity.event.dml.TapInsertRecordEvent;
 import io.tapdata.entity.event.dml.TapUpdateRecordEvent;
+import org.bson.Document;
+import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.DisplayName;
 import org.junit.jupiter.api.Test;
 import org.springframework.test.util.ReflectionTestUtils;
 
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.ZoneOffset;
 import java.util.*;
-import java.util.function.Consumer;
 import java.util.function.LongConsumer;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -72,4 +77,50 @@ void test3(){
         TapEvent tapEvent = cdcDelay.filterAndCalcDelay(tapUpdateRecordEvent, consumer, TaskDto.SYNC_TYPE_LOG_COLLECTOR);
         assertEquals(tapUpdateRecordEvent,tapEvent);
     }
+
+    @Test
+    void testLocalDateTime() {
+        LocalDateTime now = LocalDateTime.now();
+        long l = now.toInstant(ZoneOffset.UTC).toEpochMilli();
+        TapInsertRecordEvent insertRecordEvent = TapInsertRecordEvent.create().init()
+                .after(new Document("id",1).append("ts", now));
+        Long aLong = CdcDelay.parseTs(insertRecordEvent);
+        Assertions.assertNotNull(aLong);
+        Assertions.assertEquals(l, aLong);
+    }
+    @Test
+    void testDate() {
+        Date now = new Date();
+        TapInsertRecordEvent insertRecordEvent = TapInsertRecordEvent.create().init()
+                .after(new Document("id",1).append("ts", now));
+        Long aLong = CdcDelay.parseTs(insertRecordEvent);
+        Assertions.assertNotNull(aLong);
+        Assertions.assertEquals(now.getTime(), aLong);
+    }
+    @Test
+    void testInstant() {
+        Instant now = new Date().toInstant();
+        TapInsertRecordEvent insertRecordEvent = TapInsertRecordEvent.create().init()
+                .after(new Document("id",1).append("ts", now));
+        Long aLong = CdcDelay.parseTs(insertRecordEvent);
+        Assertions.assertNotNull(aLong);
+        Assertions.assertEquals(now.toEpochMilli(), aLong);
+    }
+    @Test
+    void testLong() {
+        Date now = new Date();
+        TapInsertRecordEvent insertRecordEvent = TapInsertRecordEvent.create().init()
+                .after(new Document("id",1).append("ts", now.getTime()));
+        Long aLong = CdcDelay.parseTs(insertRecordEvent);
+        Assertions.assertNotNull(aLong);
+        Assertions.assertEquals(now.getTime(), aLong);
+    }
+    @Test
+    void testString() {
+        TapInsertRecordEvent insertRecordEvent = TapInsertRecordEvent.create().init()
+                .after(new Document("id",1).append("ts", "2024-07-25 10:50:46.001")).referenceTime(0L);
+        Long aLong = CdcDelay.parseTs(insertRecordEvent);
+        Assertions.assertNotNull(aLong);
+        Assertions.assertEquals(0L, aLong);
+    }
 }
diff --git a/manager/tm/src/test/java/com/tapdata/tm/ds/service/impl/DataSourceServiceImplTest.java b/manager/tm/src/test/java/com/tapdata/tm/ds/service/impl/DataSourceServiceImplTest.java
index 0153d9174b..f990422269 100644
--- a/manager/tm/src/test/java/com/tapdata/tm/ds/service/impl/DataSourceServiceImplTest.java
+++ b/manager/tm/src/test/java/com/tapdata/tm/ds/service/impl/DataSourceServiceImplTest.java
@@ -1,6 +1,7 @@
 package com.tapdata.tm.ds.service.impl;
 
 
+import com.mongodb.client.result.UpdateResult;
 import com.tapdata.tm.Settings.constant.SettingUtil;
 import com.tapdata.tm.agent.service.AgentGroupService;
 import com.tapdata.tm.base.exception.BizException;
@@ -15,10 +16,12 @@
 import com.tapdata.tm.messagequeue.dto.MessageQueueDto;
 import com.tapdata.tm.messagequeue.service.MessageQueueService;
 import com.tapdata.tm.metadatainstance.service.MetadataInstancesService;
+import com.tapdata.tm.metadatainstance.vo.SourceTypeEnum;
 import com.tapdata.tm.permissions.DataPermissionHelper;
 import com.tapdata.tm.report.service.UserDataReportService;
 import com.tapdata.tm.worker.entity.Worker;
 import com.tapdata.tm.worker.service.WorkerService;
+import org.bson.Document;
 import org.bson.types.ObjectId;
 import org.junit.jupiter.api.*;
 import org.mockito.Answers;
@@ -36,6 +39,7 @@
 import java.util.Optional;
 
 import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.Mockito.*;
@@ -466,4 +470,130 @@ void test_MANUALLY_SPECIFIED_BY_THE_USER_AGENT_GROUP_PriorityProcessIdIsNull(){
         }
 
     }
-}
\ No newline at end of file
+
+
+    @Nested
+    class UpdatePartialSchemaTest {
+        DataSourceRepository dataSourceRepository;
+        MetadataInstancesService metadataInstancesService;
+        UserDetail userDetail;
+
+        Long lastUpdate;
+        String connId = "test-conn-id";
+        String loadFieldsStatus = "finished";
+        String fromSchemaVersion = "test-from-version";
+        String toSchemaVersion = "test-to-version";
+
+        @BeforeEach
+        void setUp() {
+            dataSourceRepository = mock(DataSourceRepository.class);
+            metadataInstancesService = mock(MetadataInstancesService.class);
+            dataSourceService = spy(new DataSourceServiceImpl(dataSourceRepository));
+            ReflectionTestUtils.setField(dataSourceService, "metadataInstancesService", metadataInstancesService);
+            userDetail = mock(UserDetail.class);
+            lastUpdate = System.currentTimeMillis();
+        }
+
+        @Test
+        void testUnChangeConnection() {
+            String filters = null;
+            doAnswer(invocation -> {
+                Document doc = invocation.<Query>getArgument(0).getQueryObject();
+                assertEquals(connId, doc.getString("_id"));
+                assertEquals(fromSchemaVersion, doc.getString(DataSourceConnectionDto.FIELD_SCHEMA_VERSION));
+
+                doc = invocation.<Update>getArgument(1).getUpdateObject().get("$set", Document.class);
+                assertEquals(toSchemaVersion, doc.getString(DataSourceConnectionDto.FIELD_SCHEMA_VERSION));
+                assertEquals(lastUpdate, doc.getLong(DataSourceConnectionDto.FIELD_LAST_UPDATE));
+                assertEquals(loadFieldsStatus, doc.getString(DataSourceConnectionDto.FIELD_LOAD_FIELDS_STATUS));
+                return null;
+            }).when(dataSourceRepository).findAndModify(any(Query.class), any(Update.class), eq(userDetail));
+
+            // 没更新到连接时，不更新模型状态
+            long count = dataSourceService.updatePartialSchema(connId, loadFieldsStatus, lastUpdate, fromSchemaVersion, toSchemaVersion, filters, userDetail);
+            assertEquals(0, count);
+        }
+
+        @Test
+        void testChangeConnectionAndNullFilter() {
+            String filters = null;
+
+            DataSourceEntity dataSourceEntity = mock(DataSourceEntity.class);
+            when(dataSourceRepository.findAndModify(any(Query.class), any(Update.class), eq(userDetail))).thenReturn(dataSourceEntity);
+
+            UpdateResult updateResult = mock(UpdateResult.class);
+            doAnswer(invocation -> {
+                Document doc = invocation.<Query>getArgument(0).getQueryObject();
+                assertEquals(connId, doc.getString("source._id"));
+                assertEquals("{\"$ne\": true}", doc.get("is_deleted", Document.class).toJson());
+                assertEquals(MetaType.table, doc.get("meta_type"));
+                assertEquals(SourceTypeEnum.SOURCE, doc.get("sourceType"));
+                assertEquals(fromSchemaVersion, doc.getString("source." + DataSourceConnectionDto.FIELD_SCHEMA_VERSION));
+
+                doc = invocation.<Update>getArgument(1).getUpdateObject().get("$set", Document.class);
+                assertEquals(toSchemaVersion, doc.getString("source." + DataSourceConnectionDto.FIELD_SCHEMA_VERSION));
+                assertEquals(lastUpdate, doc.getLong(DataSourceConnectionDto.FIELD_LAST_UPDATE));
+                return updateResult;
+            }).when(metadataInstancesService).updateMany(any(Query.class), any(Update.class));
+
+            long count = dataSourceService.updatePartialSchema(connId, loadFieldsStatus, lastUpdate, fromSchemaVersion, toSchemaVersion, filters, userDetail);
+            assertEquals(1, count);
+        }
+
+        @Test
+        void testChangeConnectionAndNotNullFilter() {
+            String filters = "test-filter";
+
+            DataSourceEntity dataSourceEntity = mock(DataSourceEntity.class);
+            when(dataSourceRepository.findAndModify(any(Query.class), any(Update.class), eq(userDetail))).thenReturn(dataSourceEntity);
+
+            UpdateResult updateResult = mock(UpdateResult.class);
+            when(updateResult.getModifiedCount()).thenReturn(1L);
+            doAnswer(invocation -> {
+                Document doc = invocation.<Query>getArgument(0).getQueryObject();
+                assertEquals(connId, doc.getString("source._id"));
+                assertEquals("{\"$ne\": true}", doc.get("is_deleted", Document.class).toJson());
+                assertEquals(MetaType.table, doc.get("meta_type"));
+                assertEquals(SourceTypeEnum.SOURCE, doc.get("sourceType"));
+                assertEquals(fromSchemaVersion, doc.getString("source." + DataSourceConnectionDto.FIELD_SCHEMA_VERSION));
+                assertEquals("{\"$nin\": [\"" + filters + "\"]}", doc.get("name", Document.class).toJson());
+
+                doc = invocation.<Update>getArgument(1).getUpdateObject().get("$set", Document.class);
+                assertEquals(toSchemaVersion, doc.getString("source." + DataSourceConnectionDto.FIELD_SCHEMA_VERSION));
+                assertEquals(lastUpdate, doc.getLong(DataSourceConnectionDto.FIELD_LAST_UPDATE));
+                return updateResult;
+            }).when(metadataInstancesService).updateMany(any(Query.class), any(Update.class));
+
+            long count = dataSourceService.updatePartialSchema(connId, loadFieldsStatus, lastUpdate, fromSchemaVersion, toSchemaVersion, filters, userDetail);
+            assertEquals(2, count);
+        }
+
+        @Test
+        void testChangeConnectionAndNotMultiFilter() {
+            String filters = "test-filter1,test-filter2";
+
+            DataSourceEntity dataSourceEntity = mock(DataSourceEntity.class);
+            when(dataSourceRepository.findAndModify(any(Query.class), any(Update.class), eq(userDetail))).thenReturn(dataSourceEntity);
+
+            UpdateResult updateResult = mock(UpdateResult.class);
+            when(updateResult.getModifiedCount()).thenReturn(1L);
+            doAnswer(invocation -> {
+                Document doc = invocation.<Query>getArgument(0).getQueryObject();
+                assertEquals(connId, doc.getString("source._id"));
+                assertEquals("{\"$ne\": true}", doc.get("is_deleted", Document.class).toJson());
+                assertEquals(MetaType.table, doc.get("meta_type"));
+                assertEquals(SourceTypeEnum.SOURCE, doc.get("sourceType"));
+                assertEquals(fromSchemaVersion, doc.getString("source." + DataSourceConnectionDto.FIELD_SCHEMA_VERSION));
+                assertEquals("{\"$nin\": [\"" + String.join("\", \"", filters.split(",")) + "\"]}", doc.get("name", Document.class).toJson());
+
+                doc = invocation.<Update>getArgument(1).getUpdateObject().get("$set", Document.class);
+                assertEquals(toSchemaVersion, doc.getString("source." + DataSourceConnectionDto.FIELD_SCHEMA_VERSION));
+                assertEquals(lastUpdate, doc.getLong(DataSourceConnectionDto.FIELD_LAST_UPDATE));
+                return updateResult;
+            }).when(metadataInstancesService).updateMany(any(Query.class), any(Update.class));
+
+            long count = dataSourceService.updatePartialSchema(connId, loadFieldsStatus, lastUpdate, fromSchemaVersion, toSchemaVersion, filters, userDetail);
+            assertEquals(2, count);
+        }
+    }
+}
diff --git a/manager/tm/src/test/java/com/tapdata/tm/ds/service/impl/DataSourceServiceTest.java b/manager/tm/src/test/java/com/tapdata/tm/ds/service/impl/DataSourceServiceTest.java
index fbcca993a9..1c9b9d266a 100644
--- a/manager/tm/src/test/java/com/tapdata/tm/ds/service/impl/DataSourceServiceTest.java
+++ b/manager/tm/src/test/java/com/tapdata/tm/ds/service/impl/DataSourceServiceTest.java
@@ -50,6 +50,7 @@ void testUpdateConnectionOptions() {
         final ObjectId id = new ObjectId(new GregorianCalendar(2020, Calendar.JANUARY, 1).getTime(), 0);
         final ConnectionOptions options = new ConnectionOptions();
         options.setDbVersion("dbVersion");
+        options.setTimeDifference(1000L);
         final UserDetail user = new UserDetail("userId", "customerId", "username", "password", "customerType",
                 "accessCode", false, false, false, false, Arrays.asList(new SimpleGrantedAuthority("role")));
         Criteria criteria = Criteria.where("_id").is(id);
@@ -62,6 +63,7 @@ void testUpdateConnectionOptions() {
         definitionDto.setCapabilities(Arrays.asList(new Capability("id")));
         Update expect = new Update();
         expect.set("db_version","dbVersion");
+        expect.set("timeDifference",1000L);
         expect.set("capabilities",definitionDto.getCapabilities());
         try (MockedStatic<DataPermissionService> serviceMockedStatic = Mockito.mockStatic(DataPermissionService.class)){
             serviceMockedStatic.when(DataPermissionService::isCloud).thenReturn(true);
diff --git a/manager/tm/src/test/java/com/tapdata/tm/lineage/analyzer/impl/TableAnalyzerV1Test.java b/manager/tm/src/test/java/com/tapdata/tm/lineage/analyzer/impl/TableAnalyzerV1Test.java
new file mode 100644
index 0000000000..ed39739d02
--- /dev/null
+++ b/manager/tm/src/test/java/com/tapdata/tm/lineage/analyzer/impl/TableAnalyzerV1Test.java
@@ -0,0 +1,47 @@
+package com.tapdata.tm.lineage.analyzer.impl;
+
+import com.tapdata.tm.modules.entity.ModulesEntity;
+import com.tapdata.tm.modules.repository.ModulesRepository;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+import org.springframework.data.mongodb.core.query.Query;
+import org.springframework.test.util.ReflectionTestUtils;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class TableAnalyzerV1Test {
+    @Nested
+    class FindModulesTest{
+        TableAnalyzerV1 tableAnalyzerV1;
+        ModulesRepository modulesRepository;
+        @BeforeEach
+        void init(){
+            modulesRepository = mock(ModulesRepository.class);
+            tableAnalyzerV1 = new TableAnalyzerV1();
+            ReflectionTestUtils.setField(tableAnalyzerV1,"modulesRepository",modulesRepository);
+        }
+
+        @Test
+        void main_test(){
+            List<ModulesEntity> modulesEntities = new ArrayList<>();
+            ModulesEntity modules = new ModulesEntity();
+            modulesEntities.add(modules);
+            when(modulesRepository.findAll(any(Query.class))).thenAnswer(invocationOnMock -> {
+                Query query = invocationOnMock.getArgument(0);
+                Assertions.assertEquals(query.getQueryObject().get("datasource"),"test");
+                Assertions.assertEquals(query.getQueryObject().get("tableName"),"test");
+                System.out.println(query);
+                return modulesEntities;
+            });
+            List<ModulesEntity> result = tableAnalyzerV1.findModules("test","test");
+            Assertions.assertEquals(modulesEntities.size(),result.size());
+        }
+    }
+}
diff --git a/manager/tm/src/test/java/com/tapdata/tm/message/service/MessageServiceTest.java b/manager/tm/src/test/java/com/tapdata/tm/message/service/MessageServiceTest.java
index a0902d10bc..82208dff92 100644
--- a/manager/tm/src/test/java/com/tapdata/tm/message/service/MessageServiceTest.java
+++ b/manager/tm/src/test/java/com/tapdata/tm/message/service/MessageServiceTest.java
@@ -453,7 +453,7 @@ void testInformUser_SendingEmailByMessageDto() throws InvocationTargetException,
             messageDto.setSourceModule("agent");
             when(mockUserService.loadUserById(new ObjectId("62bc5008d4958d013d97c7a6"))).thenReturn(userDetail);
             informUser2.invoke(messageServiceUnderTest,messageDto);
-            verify(mockMailUtils,times(1)).sendHtmlMail("Your deployed Agent name is online【Tapdata】",Arrays.asList("test@test.com"), "Hi, : ", "name",null,"Agent online");
+            verify(mockMailUtils,times(1)).sendHtmlMail("【Tapdata】Your deployed Agent name is online",Arrays.asList("test@test.com"), "Hi, : ", "name",null,"Agent online");
         }
     }
 
@@ -525,7 +525,7 @@ void testInformUser_SendingEmailOnAgentExpiring() throws InvocationTargetExcepti
 
             informUser2.invoke(messageServiceUnderTest,messageDto);
             verify(mockMailUtils,times(1)).sendHtmlMail(
-                    "The instance (agent-xxxxxx) you subscribed to will expire in 7 days to avoid affecting the running of your tasks. Please renew in time.【Tapdata】",
+                    "【Tapdata】The instance (agent-xxxxxx) you subscribed to will expire in 7 days to avoid affecting the running of your tasks. Please renew in time.",
                     Arrays.asList("test@test.com"),
                     "Hi, : ",
                     "agent-xxxxxx",null,
@@ -535,7 +535,7 @@ void testInformUser_SendingEmailOnAgentExpiring() throws InvocationTargetExcepti
             messageDto.setMessageMetadata(JsonUtil.toJson(metadata));
             informUser2.invoke(messageServiceUnderTest,messageDto);
             verify(mockMailUtils,times(1)).sendHtmlMail(
-                    "The instance (agent-xxxxxx) you subscribed to will expire in 1 days to avoid affecting the running of your tasks. Please renew in time.【Tapdata】",
+                    "【Tapdata】The instance (agent-xxxxxx) you subscribed to will expire in 1 days to avoid affecting the running of your tasks. Please renew in time.",
                     Arrays.asList("test@test.com"),
                     "Hi, : ",
                     "agent-xxxxxx",null,
@@ -570,7 +570,7 @@ void testInformUser_SendingEmailOnAgentExpired() throws InvocationTargetExceptio
 
             informUser2.invoke(messageServiceUnderTest,messageDto);
             verify(mockMailUtils,times(1)).sendHtmlMail(
-                    "The instance (agent-xxxxxx) you subscribed to has expired and will be released in 1 day.【Tapdata】",
+                    "【Tapdata】The instance (agent-xxxxxx) you subscribed to has expired and will be released in 1 day.",
                     Arrays.asList("test@test.com"),
                     "Hi, : ",
                     "agent-xxxxxx",null,
@@ -581,7 +581,7 @@ void testInformUser_SendingEmailOnAgentExpired() throws InvocationTargetExceptio
 
             informUser2.invoke(messageServiceUnderTest,messageDto);
             verify(mockMailUtils,times(1)).sendHtmlMail(
-                    "The instance (agent-xxxxxx) you subscribed to has expired. Please keep your auto-renewal account with sufficient balance and the system will automatically renew it for you. If the renewal is not completed, please contact customer service for assistance.【Tapdata】",
+                    "【Tapdata】The instance (agent-xxxxxx) you subscribed to has expired. Please keep your auto-renewal account with sufficient balance and the system will automatically renew it for you. If the renewal is not completed, please contact customer service for assistance.",
                     Arrays.asList("test@test.com"),
                     "Hi, : ",
                     "agent-xxxxxx",null,
diff --git a/manager/tm/src/test/java/com/tapdata/tm/monitor/service/MeasurementServiceV2ImplTest.java b/manager/tm/src/test/java/com/tapdata/tm/monitor/service/MeasurementServiceV2ImplTest.java
index 538ad8a58f..18f4dc62ba 100644
--- a/manager/tm/src/test/java/com/tapdata/tm/monitor/service/MeasurementServiceV2ImplTest.java
+++ b/manager/tm/src/test/java/com/tapdata/tm/monitor/service/MeasurementServiceV2ImplTest.java
@@ -14,13 +14,17 @@
 import org.mockito.internal.verification.Times;
 import org.springframework.data.mongodb.core.BulkOperations;
 import org.springframework.data.mongodb.core.MongoTemplate;
+import org.springframework.data.mongodb.core.aggregation.Aggregation;
+import org.springframework.data.mongodb.core.aggregation.AggregationResults;
 import org.springframework.data.mongodb.core.query.Query;
 import org.springframework.test.util.ReflectionTestUtils;
 
 import java.math.BigDecimal;
 import java.util.*;
+import java.util.function.Consumer;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.*;
 import static org.mockito.Mockito.doCallRealMethod;
 import static org.mockito.Mockito.doNothing;
@@ -269,4 +273,131 @@ void testQueryTableMeasurement(){
             assertEquals(new Date(1717413352L),tableStatusInfoDto.getLastDataChangeTime());
         }
     }
-}
\ No newline at end of file
+
+    @Nested
+    class GetTaskLastFiveMinutesQpsTest {
+        String type = "task";
+        String taskId = "test-task-id";
+        String taskRecordId = "test-task-record-id";
+        Map<String, String> tags;
+        MongoTemplate mongoOperations;
+
+        @BeforeEach
+        void beforeEach() {
+            tags = new HashMap<>();
+            tags.put("type", type);
+            tags.put("taskId", taskId);
+            tags.put("taskRecordId", taskRecordId);
+
+            mongoOperations = mock(MongoTemplate.class);
+            ReflectionTestUtils.setField(measurementServiceV2, "mongoOperations", mongoOperations);
+        }
+
+        @Test
+        void testNotfound() {
+            AggregationResults<Map> results = mock(AggregationResults.class);
+            when(mongoOperations.aggregate(any(Aggregation.class), eq(MeasurementEntity.COLLECTION_NAME), eq(Map.class))).thenReturn(results);
+            doCallRealMethod().when(measurementServiceV2).getTaskLastFiveMinutesQps(any());
+
+            assertEquals(0D, measurementServiceV2.getTaskLastFiveMinutesQps(tags));
+        }
+
+        @Test
+        void testNotParse() {
+            Map<String, String> values = new HashMap<>();
+            values.put("qps", "error");
+            AggregationResults<Map> results = mock(AggregationResults.class);
+            when(results.getMappedResults()).thenReturn(Arrays.asList(values));
+            when(mongoOperations.aggregate(any(Aggregation.class), eq(MeasurementEntity.COLLECTION_NAME), eq(Map.class))).thenReturn(results);
+            doCallRealMethod().when(measurementServiceV2).getTaskLastFiveMinutesQps(any());
+
+            assertEquals(0D, measurementServiceV2.getTaskLastFiveMinutesQps(tags));
+        }
+
+        @Test
+        void testSuccess() {
+            Map<String, String> values = new HashMap<>();
+            values.put("qps", "100");
+            AggregationResults<Map> results = mock(AggregationResults.class);
+            when(results.getMappedResults()).thenReturn(Arrays.asList(values));
+            when(mongoOperations.aggregate(any(Aggregation.class), eq(MeasurementEntity.COLLECTION_NAME), eq(Map.class))).thenReturn(results);
+            doCallRealMethod().when(measurementServiceV2).getTaskLastFiveMinutesQps(any());
+
+            assertEquals(100, measurementServiceV2.getTaskLastFiveMinutesQps(tags));
+        }
+    }
+
+    @Nested
+    class Parse2SampleDataTest {
+        String type = "task";
+        String taskId = "test-task-id";
+        String taskRecordId = "test-task-record-id";
+        List<MeasurementEntity> entities;
+        Map<String, Sample> data;
+        long time;
+
+        @BeforeEach
+        void beforeEach() {
+            entities = new ArrayList<>();
+            data = new HashMap<>();
+            time = System.currentTimeMillis();
+        }
+
+        private Sample addSample(List<Sample> samples, Consumer<Sample> consumer) {
+            Sample sample = new Sample();
+            sample.setDate(new Date());
+            consumer.accept(sample);
+            samples.add(sample);
+            return sample;
+        }
+
+        private MeasurementEntity mockMeasurementEntity(String type, String taskId, String taskRecordId) {
+            Map<String, String> tags = new HashMap<>();
+            tags.put("type", type);
+            tags.put("taskId", taskId);
+            tags.put("taskRecordId", taskRecordId);
+            MeasurementEntity entity = new MeasurementEntity();
+            entity.setTags(tags);
+            entity.setSamples(new ArrayList<>());
+            return entity;
+        }
+
+        @Test
+        void testDoubleSampleSameDate() {
+            Date nowTime = new Date();
+            MeasurementEntity entity = mockMeasurementEntity(type, taskId, taskRecordId);
+            addSample(entity.getSamples(), sample -> sample.setDate(nowTime));
+            Sample expected = addSample(entity.getSamples(), sample -> sample.setDate(nowTime));
+            entities.add(entity);
+
+            doCallRealMethod().when(measurementServiceV2).parse2SampleData(any(), eq(data), eq(time));
+            measurementServiceV2.parse2SampleData(entities, data, time);
+            assertTrue(data.containsValue(expected));
+        }
+
+        @Test
+        void testDoubleSampleUseLast() {
+            long nowTime = System.currentTimeMillis();
+            MeasurementEntity entity = mockMeasurementEntity(type, taskId, taskRecordId);
+            Sample expected = addSample(entity.getSamples(), sample -> sample.setDate(new Date(nowTime)));
+            addSample(entity.getSamples(), sample -> sample.setDate(new Date(nowTime + 1000)));
+            entities.add(entity);
+
+            doCallRealMethod().when(measurementServiceV2).parse2SampleData(any(), eq(data), eq(time));
+            measurementServiceV2.parse2SampleData(entities, data, time);
+            assertTrue(data.containsValue(expected));
+        }
+
+        @Test
+        void testOneSample() {
+            MeasurementEntity entity = mockMeasurementEntity(type, taskId, taskRecordId);
+            Sample expected = addSample(entity.getSamples(), sample -> {});
+            entities.add(entity);
+
+            doCallRealMethod().when(measurementServiceV2).parse2SampleData(any(), eq(data), eq(time));
+            measurementServiceV2.parse2SampleData(entities, data, time);
+            assertTrue(data.containsValue(expected));
+        }
+
+    }
+}
diff --git a/manager/tm/src/test/java/com/tapdata/tm/task/service/TaskServiceImplTest.java b/manager/tm/src/test/java/com/tapdata/tm/task/service/TaskServiceImplTest.java
index 6b82a02abd..7a54b7401d 100644
--- a/manager/tm/src/test/java/com/tapdata/tm/task/service/TaskServiceImplTest.java
+++ b/manager/tm/src/test/java/com/tapdata/tm/task/service/TaskServiceImplTest.java
@@ -16,7 +16,10 @@
 import com.tapdata.tm.commons.dag.*;
 import com.tapdata.tm.commons.dag.logCollector.LogCollecotrConnConfig;
 import com.tapdata.tm.commons.dag.logCollector.LogCollectorNode;
-import com.tapdata.tm.commons.dag.nodes.*;
+import com.tapdata.tm.commons.dag.nodes.CacheNode;
+import com.tapdata.tm.commons.dag.nodes.DataParentNode;
+import com.tapdata.tm.commons.dag.nodes.DatabaseNode;
+import com.tapdata.tm.commons.dag.nodes.TableNode;
 import com.tapdata.tm.commons.dag.process.*;
 import com.tapdata.tm.commons.dag.vo.SyncObjects;
 import com.tapdata.tm.commons.externalStorage.ExternalStorageDto;
@@ -34,6 +37,7 @@
 import com.tapdata.tm.dataflowinsight.dto.DataFlowInsightStatisticsDto;
 import com.tapdata.tm.disruptor.constants.DisruptorTopicEnum;
 import com.tapdata.tm.disruptor.service.DisruptorService;
+import com.tapdata.tm.ds.entity.DataSourceEntity;
 import com.tapdata.tm.ds.service.impl.DataSourceService;
 import com.tapdata.tm.ds.service.impl.DataSourceServiceImpl;
 import com.tapdata.tm.externalStorage.service.ExternalStorageService;
@@ -51,6 +55,7 @@
 import com.tapdata.tm.messagequeue.dto.MessageQueueDto;
 import com.tapdata.tm.messagequeue.service.MessageQueueServiceImpl;
 import com.tapdata.tm.metadatainstance.service.MetaDataHistoryService;
+import com.tapdata.tm.metadatainstance.service.MetadataInstancesService;
 import com.tapdata.tm.metadatainstance.service.MetadataInstancesServiceImpl;
 import com.tapdata.tm.monitor.entity.MeasurementEntity;
 import com.tapdata.tm.monitor.param.IdParam;
@@ -94,8 +99,8 @@
 import io.tapdata.exception.TapCodeException;
 import io.tapdata.pdk.apis.entity.Capability;
 import io.tapdata.pdk.apis.entity.TestItem;
+import io.tapdata.utils.UnitTestUtils;
 import lombok.SneakyThrows;
-import org.apache.commons.lang3.StringUtils;
 import org.bson.BsonValue;
 import org.bson.Document;
 import org.bson.types.ObjectId;
@@ -133,7 +138,6 @@
 
 import static com.tapdata.tm.task.service.TaskServiceImpl.AGENT_ID;
 import static org.junit.jupiter.api.Assertions.*;
-import static org.junit.jupiter.api.Assertions.assertNotEquals;
 import static org.mockito.Mockito.*;
 import static org.springframework.beans.BeanUtils.copyProperties;
 
@@ -148,6 +152,7 @@ class TaskServiceImplTest {
     TaskDto taskDto;
     UserDetail user;
     ChartViewService chartViewService;
+    DataSourceService dataSourceService;
     @BeforeEach
     void init() {
         taskService = mock(TaskServiceImpl.class);
@@ -162,6 +167,8 @@ void init() {
         ReflectionTestUtils.setField(taskService, "chartViewService", chartViewService);
         taskDto = mock(TaskDto.class);
         user = mock(UserDetail.class);
+        dataSourceService = mock(DataSourceServiceImpl.class);
+        ReflectionTestUtils.setField(taskService, "dataSourceService", dataSourceService);
     }
 
     @Nested
@@ -5120,4 +5127,282 @@ void test3(){
         }
 
     }
-}
\ No newline at end of file
+    @Nested
+    class CheckSourceTimeDifferenceTest{
+
+        @Test
+        void test_main(){
+            TaskDto taskDto = new TaskDto();
+            UserDetail userDetail = mock(UserDetail.class);
+            DAG dag = mock(DAG.class);
+            taskDto.setDag(dag);
+            List<Node> nodes = new ArrayList<>();
+            DatabaseNode databaseNode = new DatabaseNode();
+            databaseNode.setConnectionId("test");
+            nodes.add(databaseNode);
+            when(dag.getSourceNodes()).thenReturn(nodes);
+            doCallRealMethod().when(taskService).checkSourceTimeDifference(taskDto,userDetail);
+            List<DataSourceEntity> dataSourceEntities = new ArrayList<>();
+            DataSourceEntity dataSourceEntity = new DataSourceEntity();
+            dataSourceEntity.setTimeDifference(1000L);
+            dataSourceEntities.add(dataSourceEntity);
+            when(dataSourceService.findAll(any(Query.class),any(UserDetail.class))).thenReturn(dataSourceEntities);
+            taskService.checkSourceTimeDifference(taskDto,userDetail);
+            Assertions.assertEquals(1000L,taskDto.getTimeDifference());
+        }
+        @Test
+        void test_tableNodes(){
+            TaskDto taskDto = new TaskDto();
+            UserDetail userDetail = mock(UserDetail.class);
+            DAG dag = mock(DAG.class);
+            taskDto.setDag(dag);
+            List<Node> nodes = new ArrayList<>();
+            TableNode node1 = new TableNode();
+            node1.setConnectionId("test1");
+            TableNode node2 = new TableNode();
+            node2.setConnectionId("test2");
+            TableNode node3 = new TableNode();
+            node3.setConnectionId("test1");
+            nodes.add(node1);
+            nodes.add(node2);
+            nodes.add(node3);
+            when(dag.getSourceNodes()).thenReturn(nodes);
+            doCallRealMethod().when(taskService).checkSourceTimeDifference(taskDto,userDetail);
+            List<DataSourceEntity> dataSourceEntities = new ArrayList<>();
+            DataSourceEntity dataSourceEntity1 = new DataSourceEntity();
+            dataSourceEntity1.setTimeDifference(1000L);
+            DataSourceEntity dataSourceEntity2 = new DataSourceEntity();
+            dataSourceEntity2.setTimeDifference(2000L);
+            dataSourceEntities.add(dataSourceEntity1);
+            dataSourceEntities.add(dataSourceEntity2);
+            Set<String> connectionIds = new HashSet<>();
+            connectionIds.add("test1");
+            connectionIds.add("test2");
+            Query query = new Query(Criteria.where("_id").in(connectionIds));
+            query.fields().include("timeDifference");
+            when(dataSourceService.findAll(eq(query),eq(userDetail))).thenReturn(dataSourceEntities);
+            taskService.checkSourceTimeDifference(taskDto,userDetail);
+            Assertions.assertEquals(2000L,taskDto.getTimeDifference());
+        }
+        @Test
+        void test_DagIsNull(){
+            TaskDto taskDto = new TaskDto();
+            UserDetail userDetail = mock(UserDetail.class);
+            doCallRealMethod().when(taskService).checkSourceTimeDifference(taskDto,userDetail);
+            taskService.checkSourceTimeDifference(taskDto,userDetail);
+            Assertions.assertNull(taskDto.getTimeDifference());
+        }
+
+        @Test
+        void test_connectionIdsIsEmpty(){
+            TaskDto taskDto = new TaskDto();
+            UserDetail userDetail = mock(UserDetail.class);
+            DAG dag = mock(DAG.class);
+            taskDto.setDag(dag);
+            List<Node> nodes = new ArrayList<>();
+            nodes.add(new MigrateJsProcessorNode());
+            when(dag.getSourceNodes()).thenReturn(nodes);
+            doCallRealMethod().when(taskService).checkSourceTimeDifference(taskDto,userDetail);
+            taskService.checkSourceTimeDifference(taskDto,userDetail);
+            Assertions.assertNull(taskDto.getTimeDifference());
+        }
+    }
+
+    @Nested
+    class RefreshSchemasTest {
+
+        DAG dag;
+        TaskDto taskDto;
+        DataSourceService dataSourceService;
+        MetadataInstancesService metadataInstancesService;
+        TransformSchemaService transformSchemaService;
+
+        @BeforeEach
+        void init() {
+            taskDto = new TaskDto();
+            taskDto.setId(new ObjectId());
+
+            dag = mock(DAG.class);
+            taskDto.setDag(dag);
+
+            dataSourceService = mock(DataSourceServiceImpl.class);
+            UnitTestUtils.injectField(TaskServiceImpl.class, taskService, "dataSourceService", dataSourceService);
+            metadataInstancesService = mock(MetadataInstancesServiceImpl.class);
+            UnitTestUtils.injectField(TaskServiceImpl.class, taskService, "metadataInstancesService", metadataInstancesService);
+            transformSchemaService = mock(TransformSchemaService.class);
+            UnitTestUtils.injectField(TaskServiceImpl.class, taskService, "transformSchemaService", transformSchemaService);
+        }
+
+        @Test
+        void testEmptyNodes() {
+            String nodeIds = null;
+            String keys = null;
+
+            // 检查任务是否修改为推演中，实现前端自动刷新
+            doAnswer(invocation -> {
+                Document doc = invocation.<Update>getArgument(1).getUpdateObject().get("$set", Document.class);
+                assertEquals(MetadataTransformerDto.StatusEnum.running.name(), doc.getString(TaskDto.FIELD_TRANSFORM_STATUS));
+                return null;
+            }).when(taskService).updateById(eq(taskDto.getId()), any(Update.class), eq(user));
+            doCallRealMethod().when(taskService).refreshSchemas(taskDto, nodeIds, keys, user);
+
+            // 无节点不刷新模型
+            taskService.refreshSchemas(taskDto, nodeIds, keys, user);
+        }
+
+        @Test
+        void testNotfoundDatasource() {
+            String connId = "test-conn-id";
+            String nodeIds = null;
+            String keys = null;
+
+            List<Node> nodes = new ArrayList<>();
+            nodes.add(Optional.of(new TableNode()).map(n -> {
+                n.setConnectionId(connId);
+                return n;
+            }).get());
+
+            // test not tableNode and DatabaseNode
+            nodes.add(Optional.of(new FieldRenameProcessorNode()).get());
+
+            when(dag.getNodes()).thenReturn(nodes);
+            doCallRealMethod().when(taskService).refreshSchemas(taskDto, nodeIds, keys, user);
+            assertThrows(BizException.class, () -> taskService.refreshSchemas(taskDto, nodeIds, keys, user));
+        }
+
+        @Test
+        void testRefreshTaskTableNode() {
+            String connId = "test-conn-id";
+            String tableName = "test-table-name";
+            String nodeIds = null;
+            String keys = null;
+
+            DataSourceConnectionDto connDto = mock(DataSourceConnectionDto.class);
+            when(dataSourceService.findById(any())).thenReturn(connDto);
+
+            List<Node> nodes = new ArrayList<>();
+            nodes.add(Optional.of(new TableNode()).map(n -> {
+                n.setConnectionId(connId);
+                n.setTableName(tableName);
+                return n;
+            }).get());
+            when(dag.getNodes()).thenReturn(nodes);
+            doCallRealMethod().when(taskService).refreshSchemas(taskDto, nodeIds, keys, user);
+
+            // 无节点不刷新模型
+            taskService.refreshSchemas(taskDto, nodeIds, keys, user);
+
+            verify(taskService, times(1)).wait2ConnectionsLoadFinished(anyString(), any(), anyLong(), anyInt(), eq(user));
+            verify(metadataInstancesService, times(1)).deleteTaskMetadata(anyString(), any());
+            verify(transformSchemaService, times(1)).transformSchema(eq(taskDto), eq(user));
+        }
+
+        @Test
+        void testRefreshTaskDatabaseNode() {
+            String connId = "test-conn-id";
+            String nodeIds = null;
+            String keys = null;
+
+            DataSourceConnectionDto connDto = mock(DataSourceConnectionDto.class);
+            when(dataSourceService.findById(any())).thenReturn(connDto);
+
+            List<Node> nodes = new ArrayList<>();
+            nodes.add(Optional.of(new DatabaseNode()).map(n -> {
+                n.setConnectionId(connId);
+                return n;
+            }).get());
+            when(dag.getNodes()).thenReturn(nodes);
+            doCallRealMethod().when(taskService).refreshSchemas(taskDto, nodeIds, keys, user);
+
+            // 无节点不刷新模型
+            taskService.refreshSchemas(taskDto, nodeIds, keys, user);
+
+            verify(taskService, times(1)).wait2ConnectionsLoadFinished(anyString(), any(), anyLong(), anyInt(), eq(user));
+            verify(metadataInstancesService, times(1)).deleteTaskMetadata(anyString(), any());
+            verify(transformSchemaService, times(1)).transformSchema(eq(taskDto), eq(user));
+        }
+
+        @Test
+        void testRefreshTaskNode() {
+            String connId = "test-conn-id";
+            String nodeIds = "test-node-id";
+            String keys = null;
+
+            DataSourceConnectionDto connDto = mock(DataSourceConnectionDto.class);
+            when(dataSourceService.findById(any())).thenReturn(connDto);
+
+            TableNode tableNode = Optional.of(new TableNode()).map(n -> {
+                n.setId(nodeIds);
+                n.setConnectionId(connId);
+                return n;
+            }).get();
+            when(dag.getNode(nodeIds)).thenReturn((Node) tableNode);
+            doCallRealMethod().when(taskService).refreshSchemas(taskDto, nodeIds, keys, user);
+
+            // 无节点不刷新模型
+            taskService.refreshSchemas(taskDto, nodeIds, keys, user);
+
+            verify(taskService, times(1)).wait2ConnectionsLoadFinished(anyString(), any(), anyLong(), anyInt(), eq(user));
+            verify(metadataInstancesService, times(1)).deleteLogicModel(anyString(), eq(nodeIds));
+            verify(transformSchemaService, times(1)).transformSchema(eq(taskDto), eq(user));
+        }
+    }
+
+    @Nested
+    class Wait2ConnectionsLoadFinishedTest {
+        String taskId;
+        ObjectId connId;
+        long beginTime;
+        int timeout;
+        DataSourceService dataSourceService;
+
+        @BeforeEach
+        void init() {
+            taskDto = new TaskDto();
+            taskDto.setId(new ObjectId());
+            taskId = taskDto.getId().toHexString();
+            connId = ObjectId.get();
+            beginTime = System.currentTimeMillis();
+            timeout = 3 * 1000;
+
+            dataSourceService = mock(DataSourceServiceImpl.class);
+            UnitTestUtils.injectField(TaskServiceImpl.class, taskService, "dataSourceService", dataSourceService);
+        }
+
+        @Test
+        void testTimeout() {
+            timeout = 1000;
+
+            DataSourceConnectionDto connDto = mock(DataSourceConnectionDto.class);
+            when(dataSourceService.getById(eq(connId), any(), any(), eq(user))).thenReturn(connDto);
+
+            doCallRealMethod().when(taskService).wait2ConnectionsLoadFinished(eq(taskId), eq(connId), eq(beginTime), eq(timeout), eq(user));
+
+            taskService.wait2ConnectionsLoadFinished(taskId, connId, beginTime, timeout, user);
+        }
+
+        @Test
+        void testFinished() {
+            DataSourceConnectionDto connDto = mock(DataSourceConnectionDto.class);
+            when(connDto.getLastUpdate()).thenReturn(beginTime + 1);
+            when(connDto.getLoadFieldsStatus()).thenReturn(DataSourceConnectionDto.LOAD_FIELD_STATUS_FINISHED);
+            when(dataSourceService.getById(eq(connId), any(), any(), eq(user))).thenReturn(connDto);
+
+            doCallRealMethod().when(taskService).wait2ConnectionsLoadFinished(eq(taskId), eq(connId), eq(beginTime), eq(timeout), eq(user));
+
+            taskService.wait2ConnectionsLoadFinished(taskId, connId, beginTime, timeout, user);
+        }
+
+        @Test
+        void testError() {
+            DataSourceConnectionDto connDto = mock(DataSourceConnectionDto.class);
+            when(connDto.getLastUpdate()).thenReturn(beginTime + 1);
+            when(connDto.getLoadFieldsStatus()).thenReturn(DataSourceConnectionDto.LOAD_FIELD_STATUS_ERROR);
+            when(dataSourceService.getById(eq(connId), any(), any(), eq(user))).thenReturn(connDto);
+
+            doCallRealMethod().when(taskService).wait2ConnectionsLoadFinished(eq(taskId), eq(connId), eq(beginTime), eq(timeout), eq(user));
+
+            taskService.wait2ConnectionsLoadFinished(taskId, connId, beginTime, timeout, user);
+        }
+    }
+}