Repository: tapdata/tapdata (language: python)

Issue:
TAP-5062:fix queryRecords net out huge



Discussion / hints:

Fix patch:
diff --git a/RELEASE b/RELEASE
index f40a0d9ac7..ea5ca51129 100644
--- a/RELEASE
+++ b/RELEASE
@@ -1,2 +1,2 @@
-LST: v3.14.0
-DEV: v3.15.0-dev
\ No newline at end of file
+LST: v3.15.0
+DEV: v3.16.0-dev
\ No newline at end of file
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 f8e8d5ef96..9f18b34d1e 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,5 +1,6 @@
 package io.tapdata.flow.engine.V2.filter;
 
+import com.tapdata.tm.commons.util.NoPrimaryKeyVirtualField;
 import io.tapdata.entity.event.TapEvent;
 import io.tapdata.entity.schema.TapTable;
 import io.tapdata.flow.engine.V2.util.TapEventUtil;
@@ -16,7 +17,7 @@ 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) return null;
-        data.entrySet().removeIf(entry -> !fieldNames.contains(entry.getKey()));
+        data.entrySet().removeIf(entry -> !NoPrimaryKeyVirtualField.FIELD_NAME.equalsIgnoreCase(entry.getKey()) && !fieldNames.contains(entry.getKey()));
         return data;
     }
 
diff --git a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/HazelcastBaseNode.java b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/HazelcastBaseNode.java
index 858f67a830..f8e887c4c5 100644
--- a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/HazelcastBaseNode.java
+++ b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/HazelcastBaseNode.java
@@ -221,7 +221,12 @@ public final void init(@NotNull Processor.Context context) throws Exception {
 		}
 	}
 	protected void buildLogListener(){
-		processorBaseContext.getTapTableMap().logListener(new TapLogger.LogListener() {
+		processorBaseContext.getTapTableMap().logListener(logListener());
+	}
+
+	@NotNull
+	protected TapLogger.LogListener logListener() {
+		return new TapLogger.LogListener() {
 			@Override
 			public void debug(String log) {
 				obsLogger.debug(log);
@@ -245,7 +250,7 @@ public void fatal(String log) {
 			@Override
 			public void memory(String memoryLog) {
 			}
-		});
+		};
 	}
 
 	protected MonitorManager initMonitor() {
diff --git a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastPdkBaseNode.java b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastPdkBaseNode.java
index 6ffc6eb735..870e7d495d 100644
--- a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastPdkBaseNode.java
+++ b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastPdkBaseNode.java
@@ -6,7 +6,6 @@
 import com.tapdata.constant.Log4jUtil;
 import com.tapdata.constant.MapUtil;
 import com.tapdata.entity.DatabaseTypeEnum;
-import com.tapdata.entity.dataflow.SyncProgress;
 import com.tapdata.entity.task.context.DataProcessorContext;
 import com.tapdata.tm.commons.dag.DmlPolicy;
 import com.tapdata.tm.commons.dag.DmlPolicyEnum;
@@ -89,6 +88,7 @@ public abstract class HazelcastPdkBaseNode extends HazelcastDataBaseNode {
 
 	public static final String FUNCTION_RETRY_STATUS = "functionRetryStatus";
 	private static final String DOUBLE_ACTIVE = "doubleActive";
+	private static final String WRITE_THREAD_SIZE = "writeThreadSize";
 	protected TapRecordSkipDetector skipDetector;
 	private PdkStateMap pdkStateMap;
 
@@ -235,6 +235,11 @@ protected Map<String, Object> generateNodeConfig(Node<?> node, TaskDto taskDto)
 		if (null == nodeConfig) {
 			nodeConfig = new HashMap<>();
 		}
+        if (node instanceof DataParentNode) {
+            Integer initialConcurrentWriteNum = Optional.ofNullable(((DataParentNode<?>) node).getInitialConcurrentWriteNum()).orElse(1);
+			Integer cdcConcurrentWriteNum = Optional.ofNullable(((DataParentNode<?>) node).getCdcConcurrentWriteNum()).orElse(1);
+            nodeConfig.put(WRITE_THREAD_SIZE, Math.max(initialConcurrentWriteNum, cdcConcurrentWriteNum));
+        }
 		nodeConfig.put(DOUBLE_ACTIVE, taskDto.getDoubleActive());
 		Boolean oldVersionTimezone = taskDto.getOldVersionTimezone();
 		oldVersionTimezone = CommonUtils.getPropertyBool(OLD_VERSION_TIME_ZONE_PROP_KEY, oldVersionTimezone);
diff --git a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastSourcePdkBaseNode.java b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastSourcePdkBaseNode.java
index f47e210fc5..70e8fa7586 100644
--- a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastSourcePdkBaseNode.java
+++ b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastSourcePdkBaseNode.java
@@ -26,12 +26,14 @@
 import com.tapdata.tm.commons.dag.process.JoinProcessorNode;
 import com.tapdata.tm.commons.dag.process.MergeTableNode;
 import com.tapdata.tm.commons.dag.process.UnionProcessorNode;
+import com.tapdata.tm.commons.exception.NoPrimaryKeyException;
 import com.tapdata.tm.commons.schema.MetadataInstancesDto;
 import com.tapdata.tm.commons.schema.TransformerWsMessageDto;
 import com.tapdata.tm.commons.task.dto.Message;
 import com.tapdata.tm.commons.task.dto.TaskDto;
 import com.tapdata.tm.commons.util.ConnHeartbeatUtils;
 import com.tapdata.tm.commons.util.NoPrimaryKeyTableSelectType;
+import com.tapdata.tm.commons.util.NoPrimaryKeyVirtualField;
 import io.tapdata.Runnable.LoadSchemaRunner;
 import io.tapdata.aspect.SourceCDCDelayAspect;
 import io.tapdata.aspect.SourceDynamicTableAspect;
@@ -183,6 +185,7 @@ public abstract class HazelcastSourcePdkBaseNode extends HazelcastPdkBaseNode {
 	private SimpleConcurrentProcessorImpl<List<TapdataEvent>, List<TapdataEvent>> toTapValueConcurrentProcessor;
 	private int drainSize;
 	private int toTapValueBatchSize;
+    protected final NoPrimaryKeyVirtualField noPrimaryKeyVirtualField = new NoPrimaryKeyVirtualField();
 
 	public HazelcastSourcePdkBaseNode(DataProcessorContext dataProcessorContext) {
 		super(dataProcessorContext);
@@ -211,6 +214,7 @@ private boolean needCdcDelay() {
 
 	@Override
 	protected void doInit(@NotNull Context context) throws TapCodeException {
+        noPrimaryKeyVirtualField.init(getNode().getGraph());
         AutoRecovery.setEnqueueConsumer(getNode().getTaskId(), this::enqueue);
 		ConcurrentHashSet<TaskNodeInfo> taskNodeInfos = taskResourceSupervisorManager.getTaskNodeInfos();
 		ThreadGroup connectorOnTaskThreadGroup = getReuseOrNewThreadGroup(taskNodeInfos);
@@ -231,6 +235,7 @@ protected void doInit(@NotNull Context context) throws TapCodeException {
 			} catch (Throwable e) {
 				throw new NodeException(e).context(getProcessorBaseContext());
 			}
+            dataProcessorContext.getTapTableMap().forEach((id, table) -> noPrimaryKeyVirtualField.add(table));
 			initSourceReadBatchSize();
 			initSourceEventQueue();
 			initSyncProgress();
@@ -915,6 +920,7 @@ protected boolean handleNewTables(List<String> addList) {
 					}
 
 					tapdataEvents.add(tapdataEvent);
+                    noPrimaryKeyVirtualField.add(addTapTable);
 				}
 				if (!isRunning()) {
 					return true;
@@ -1065,20 +1071,32 @@ protected TapdataEvent wrapSingleTapdataEvent(TapEvent tapEvent, SyncStage syncS
 		tapdataEvent.setTapEvent(tapEvent);
 		tapdataEvent.setSyncStage(syncStage);
 		if (tapEvent instanceof TapRecordEvent) {
+            TapRecordEvent recordEvent = (TapRecordEvent) tapEvent;
 			if (SyncStage.INITIAL_SYNC == syncStage) {
 				if (isLast && !StringUtils.equalsAnyIgnoreCase(dataProcessorContext.getTaskDto().getSyncType(),
 						TaskDto.SYNC_TYPE_DEDUCE_SCHEMA, TaskDto.SYNC_TYPE_TEST_RUN)) {
-					tapdataEvent.setBatchOffset(BatchOffsetUtil.getTableOffsetInfo(syncProgress, ((TapRecordEvent) tapEvent).getTableId()));
+					tapdataEvent.setBatchOffset(BatchOffsetUtil.getTableOffsetInfo(syncProgress, recordEvent.getTableId()));
 					tapdataEvent.setStreamOffset(syncProgress.getStreamOffsetObj());
 					tapdataEvent.setSourceTime(syncProgress.getSourceTime());
 				}
 			} else if (SyncStage.CDC == syncStage) {
 				tapdataEvent.setStreamOffset(offsetObj);
-				if (null == ((TapRecordEvent) tapEvent).getReferenceTime())
+				if (null == recordEvent.getReferenceTime())
 					throw new RuntimeException("Tap CDC event's reference time is null");
-				tapdataEvent.setSourceTime(((TapRecordEvent) tapEvent).getReferenceTime());
+				tapdataEvent.setSourceTime(recordEvent.getReferenceTime());
 			}
-		} else if (tapEvent instanceof HeartbeatEvent) {
+            try {
+                if (!noPrimaryKeyVirtualField.addHashValue(recordEvent)) return null;
+            } catch (NoPrimaryKeyException e) {
+                if (NoPrimaryKeyException.CODE_INCOMPLETE_FIELDS == e.getCode()) {
+                    obsLogger.warn("Table '{}' user lacks complete before information, and subsequent update and delete events will be ignored: {}", recordEvent.getTableId(), e.getMessage());
+                    logger.warn("Table '{}' user lacks complete before information, and subsequent update and delete events will be ignored", recordEvent.getTableId(), e);
+                } else {
+                    obsLogger.warn("Table '{}' add hash filed failed, and subsequent update and delete events will be ignored: {}", recordEvent.getTableId(), e.getMessage());
+                    logger.error("Table '{}' add hash filed failed, and subsequent update and delete events will be ignored", recordEvent.getTableId(), e);
+                }
+            }
+        } else if (tapEvent instanceof HeartbeatEvent) {
 			tapdataEvent = TapdataHeartbeatEvent.create(((HeartbeatEvent) tapEvent).getReferenceTime(), offsetObj);
 		} else if (tapEvent instanceof TapDDLEvent) {
 			obsLogger.info("Source node received an ddl event: " + tapEvent);
@@ -1253,27 +1271,9 @@ protected void doCount(List<String> tableList) {
 		}
 
 		if (dataProcessorContext.getTapTableMap().keySet().size() > ASYNCLY_COUNT_SNAPSHOT_ROW_SIZE_TABLE_THRESHOLD) {
-			logger.info("Start to asynchronously count the size of rows for the source table(s)");
-			AtomicReference<TaskDto> task = new AtomicReference<>(dataProcessorContext.getTaskDto());
-			AtomicReference<Node<?>> node = new AtomicReference<>(dataProcessorContext.getNode());
-			snapshotRowSizeThreadPool = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.SECONDS, new SynchronousQueue<>());
-			CompletableFuture.runAsync(() -> {
-						String name = String.format("Snapshot-Row-Size-Query-Thread-%s(%s)-%s(%s)",
-								task.get().getName(), task.get().getId().toHexString(), node.get().getName(), node.get().getId());
-						Thread.currentThread().setName(name);
-
-						doCountSynchronously(batchCountFunction, tableList);
-					}, snapshotRowSizeThreadPool)
-					.whenComplete((v, e) -> {
-						if (null != e) {
-							obsLogger.warn("Query snapshot row size failed: " + e.getMessage() + "\n" + Log4jUtil.getStackString(e));
-						} else {
-							obsLogger.info("Query snapshot row size completed: " + node.get().getName() + "(" + node.get().getId() + ")");
-						}
-						ExecutorUtil.shutdown(this.snapshotRowSizeThreadPool, 10L, TimeUnit.SECONDS);
-					});
+			asyncCountTable(batchCountFunction, tableList);
 		} else {
-			doCountSynchronously(batchCountFunction, tableList);
+			doCountSynchronously(batchCountFunction, tableList, false);
 		}
 	}
 
@@ -1287,7 +1287,7 @@ protected void setDefaultRowSizeMap() {
 	}
 
 	@SneakyThrows
-	protected void doCountSynchronously(BatchCountFunction batchCountFunction, List<String> tableList) {
+	protected void doCountSynchronously(BatchCountFunction batchCountFunction, List<String> tableList, boolean displayTableListFirst) {
 		if (null == batchCountFunction) {
 			setDefaultRowSizeMap();
 			obsLogger.warn("PDK node does not support table batch count: " + dataProcessorContext.getDatabaseType());
@@ -1307,13 +1307,16 @@ protected void doCountSynchronously(BatchCountFunction batchCountFunction, List<
 							createPdkMethodInvoker().runnable(
 									() -> {
 										try {
-											long count = batchCountFunction.count(getConnectorNode().getConnectorContext(), table);
-
-											if (null == snapshotRowSizeMap) {
-												snapshotRowSizeMap = new HashMap<>();
+											long count;
+											if (displayTableListFirst) {
+												count = -1;
+											} else {
+												count = batchCountFunction.count(getConnectorNode().getConnectorContext(), table);
+												if (null == snapshotRowSizeMap) {
+													snapshotRowSizeMap = new HashMap<>();
+												}
+												snapshotRowSizeMap.putIfAbsent(tableName, count);
 											}
-											snapshotRowSizeMap.putIfAbsent(tableName, count);
-
 											if (null != tableCountFuncAspect) {
 												AspectUtils.accept(tableCountFuncAspect.state(TableCountFuncAspect.STATE_COUNTING).getTableCountConsumerList(), table.getName(), count);
 											}
@@ -1327,6 +1330,35 @@ protected void doCountSynchronously(BatchCountFunction batchCountFunction, List<
 		}
 	}
 
+	@SneakyThrows
+	protected void doTableNameSynchronously(BatchCountFunction batchCountFunction, List<String> tableList) {
+		doCountSynchronously(batchCountFunction, tableList, true);
+		asyncCountTable(batchCountFunction, tableList);
+	}
+
+	protected void asyncCountTable(BatchCountFunction batchCountFunction, List<String> tableList) {
+		logger.info("Start to asynchronously count the size of rows for the source table(s)");
+		AtomicReference<TaskDto> task = new AtomicReference<>(dataProcessorContext.getTaskDto());
+		AtomicReference<Node<?>> node = new AtomicReference<>(dataProcessorContext.getNode());
+		snapshotRowSizeThreadPool = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.SECONDS, new SynchronousQueue<>());
+		CompletableFuture.runAsync(() -> {
+					String name = String.format("Snapshot-Row-Size-Query-Thread-%s(%s)-%s(%s)",
+							task.get().getName(), task.get().getId().toHexString(), node.get().getName(), node.get().getId());
+					Thread.currentThread().setName(name);
+
+					doCountSynchronously(batchCountFunction, tableList, false);
+				}, snapshotRowSizeThreadPool)
+				.whenComplete((v, e) -> {
+					if (null != e) {
+						obsLogger.warn("Query snapshot row size failed: " + e.getMessage() + "\n" + Log4jUtil.getStackString(e));
+					} else {
+						obsLogger.info("Query snapshot row size completed: " + node.get().getName() + "(" + node.get().getId() + ")");
+					}
+					ExecutorUtil.shutdown(this.snapshotRowSizeThreadPool, 10L, TimeUnit.SECONDS);
+				});
+	}
+
+
 	protected Long doBatchCountFunction(BatchCountFunction batchCountFunction, TapTable table) {
 		AtomicReference<Long> counts = new AtomicReference<>();
 
diff --git a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastSourcePdkDataNode.java b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastSourcePdkDataNode.java
index a6d39c08af..3885af3c8a 100644
--- a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastSourcePdkDataNode.java
+++ b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastSourcePdkDataNode.java
@@ -363,6 +363,7 @@ protected DoSnapshotFunctions checkFunctions(List<String> tableList) {
 				.tables(tableList)
 		);
 		syncProgress.setSyncStage(SyncStage.INITIAL_SYNC.name());
+		doCount(tableList);
 		ConnectorNode connectorNode = getConnectorNode();
 		ConnectorFunctions connectorFunctions = connectorNode.getConnectorFunctions();
 		BatchCountFunction batchCountFunction = connectorFunctions.getBatchCountFunction();
@@ -371,6 +372,7 @@ protected DoSnapshotFunctions checkFunctions(List<String> tableList) {
 			setDefaultRowSizeMap();
 			obsLogger.warn("PDK node does not support table batch count: {}", databaseType);
 		}
+		doTableNameSynchronously(batchCountFunction, tableList);
 		BatchReadFunction batchReadFunction = connectorFunctions.getBatchReadFunction();
 		QueryByAdvanceFilterFunction queryByAdvanceFilterFunction = connectorFunctions.getQueryByAdvanceFilterFunction();
 		ExecuteCommandFunction executeCommandFunction = connectorFunctions.getExecuteCommandFunction();
@@ -394,7 +396,7 @@ protected void doSnapshotInvoke(String tableName, DoSnapshotFunctions functions,
 		ExecuteCommandFunction executeCommandFunction = functions.getExecuteCommandFunction();
 
 		PDKMethodInvoker pdkMethodInvoker = createPdkMethodInvoker();
-		try (AutoCloseable ignoreTableCountCloseable = doAsyncTableCount(batchCountFunction, tableName)) {
+		try {
 			executeDataFuncAspect(
 					BatchReadFuncAspect.class, () -> new BatchReadFuncAspect()
 							.eventBatchSize(readBatchSize)
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 f488295017..879df2872b 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
@@ -24,6 +24,7 @@
 import com.tapdata.tm.commons.task.dto.MergeTableProperties;
 import com.tapdata.tm.commons.task.dto.TaskDto;
 import com.tapdata.tm.commons.util.JsonUtil;
+import com.tapdata.tm.commons.util.NoPrimaryKeyVirtualField;
 import com.tapdata.tm.shareCdcTableMetrics.ShareCdcTableMetricsDto;
 import io.tapdata.aspect.CreateTableFuncAspect;
 import io.tapdata.aspect.NewFieldFuncAspect;
@@ -41,10 +42,7 @@
 import io.tapdata.entity.event.dml.TapRecordEvent;
 import io.tapdata.entity.event.dml.TapUpdateRecordEvent;
 import io.tapdata.entity.schema.TapTable;
-import io.tapdata.entity.schema.value.DateTime;
-import io.tapdata.entity.schema.value.TapDateTimeValue;
-import io.tapdata.entity.schema.value.TapDateValue;
-import io.tapdata.entity.schema.value.TapMapValue;
+import io.tapdata.entity.schema.value.*;
 import io.tapdata.entity.utils.InstanceFactory;
 import io.tapdata.error.TapEventException;
 import io.tapdata.error.TapdataEventException;
@@ -1080,17 +1078,30 @@ protected void replaceIllegalDateWithNullIfNeed(TapRecordEvent event){
 	}
 
 	protected void replaceIllegalDate(Map<String, Object> data){
-		for(Map.Entry<String, Object> entry : data.entrySet()){
-			if(entry.getValue() instanceof TapDateTimeValue){
+		for (Map.Entry<String, Object> entry : data.entrySet()) {
+			if (entry.getValue() instanceof TapDateTimeValue) {
 				TapDateTimeValue tapDateTimeValue = (TapDateTimeValue) entry.getValue();
-				DateTime dateTime = tapDateTimeValue.getValue();
-				if (dateTime.isContainsIllegal()){
-					entry.setValue(null);
-				}
+				replaceIllegalDateTime2Null(tapDateTimeValue.getValue(), entry);
+			} else if (entry.getValue() instanceof TapDateValue) {
+				TapDateValue tapDateValue = (TapDateValue) entry.getValue();
+				replaceIllegalDateTime2Null(tapDateValue.getValue(), entry);
+			}else if (entry.getValue() instanceof TapTimeValue){
+				TapTimeValue tapTimeValue = (TapTimeValue) entry.getValue();
+				replaceIllegalDateTime2Null(tapTimeValue.getValue(),entry);
+			} else if (entry.getValue() instanceof TapYearValue) {
+				TapYearValue tapYearValue = (TapYearValue) entry.getValue();
+				replaceIllegalDateTime2Null(tapYearValue.getValue(),entry);
 			}
 		}
 	}
 
+	protected void replaceIllegalDateTime2Null(DateTime tapDateValue, Map.Entry<String, Object> entry) {
+		DateTime dateTime = tapDateValue;
+		if (dateTime.isContainsIllegal()) {
+			entry.setValue(null);
+		}
+	}
+
 
 	private boolean handleExactlyOnceWriteCacheIfNeed(TapdataEvent tapdataEvent, List<TapRecordEvent> exactlyOnceWriteCache) {
 		if (!tableEnableExactlyOnceWrite(tapdataEvent.getSyncStage(), getTgtTableNameFromTapEvent(tapdataEvent.getTapEvent()))) {
@@ -1201,7 +1212,7 @@ protected void handleTapTablePrimaryKeys(TapTable tapTable) {
 			} else {
 				Collection<String> logicUniqueKey = tapTable.primaryKeys(true);
 				if (CollectionUtils.isEmpty(logicUniqueKey)) {
-					tapTable.setLogicPrimaries(Collections.emptyList());
+					tapTable.setLogicPrimaries(NoPrimaryKeyVirtualField.getVirtualHashFieldNames(tapTable));
 				}
 			}
 		} else if (writeStrategy.equals(com.tapdata.tm.commons.task.dto.MergeTableProperties.MergeType.appendWrite.name())) {
diff --git a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/concurrent/PartitionConcurrentProcessor.java b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/concurrent/PartitionConcurrentProcessor.java
index 2685256979..b5523e6998 100644
--- a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/concurrent/PartitionConcurrentProcessor.java
+++ b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/concurrent/PartitionConcurrentProcessor.java
@@ -237,6 +237,7 @@ public void process(List<TapdataEvent> tapdataEvents, boolean async) {
 		try {
 			AtomicBoolean singleMode = new AtomicBoolean(false);
             TapdataEvent offsetEvent = null;
+			List<TapdataEvent> controlOffsetEvents = new ArrayList<>();
 			for (TapdataEvent tapdataEvent : tapdataEvents) {
 				if (!isRunning()) {
 					break;
@@ -252,14 +253,19 @@ public void process(List<TapdataEvent> tapdataEvents, boolean async) {
 						processSignalWithWait(tapdataEvent);
 					}
 				}
-                if (null != tapdataEvent.getBatchOffset() || null != tapdataEvent.getStreamOffset()) {
-                    offsetEvent = tapdataEvent;
-                }
+				TapdataEvent tempOffsetEvent = getOffsetEvent(tapdataEvent, controlOffsetEvents);
+				offsetEvent = tempOffsetEvent != null ? tempOffsetEvent : offsetEvent;
 			}
 			if (null != offsetEvent) {
 				generateWatermarkEvent(offsetEvent);
 			}
 
+			if (CollectionUtils.isNotEmpty(controlOffsetEvents)) {
+				for (TapdataEvent controlOffsetEvent : controlOffsetEvents) {
+					generateWatermarkEvent(controlOffsetEvent);
+				}
+			}
+
 			if (!async) {
 				waitingForProcessToCurrent();
 			}
@@ -268,6 +274,22 @@ public void process(List<TapdataEvent> tapdataEvents, boolean async) {
 		}
 	}
 
+	protected TapdataEvent getOffsetEvent(TapdataEvent tapdataEvent, List<TapdataEvent> controlOffsetEvent) {
+		if (null != tapdataEvent.getBatchOffset() || null != tapdataEvent.getStreamOffset()) {
+			if (tapdataEvent.isDML() || tapdataEvent.isDDL()) {
+				return tapdataEvent;
+			} else {
+				if (tapdataEvent.isConcurrentWrite()) {
+					return tapdataEvent;
+				}else{
+					controlOffsetEvent.add(tapdataEvent);
+				}
+			}
+		}
+		return null;
+	}
+
+
 	protected void processDML(TapdataEvent tapdataEvent, AtomicBoolean singleMode) throws InterruptedException {
 		String tableName = "";
 		try {
diff --git a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/processor/HazelcastProcessorNode.java b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/processor/HazelcastProcessorNode.java
index 16f5b95363..7542a51a93 100644
--- a/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/processor/HazelcastProcessorNode.java
+++ b/iengine/iengine-app/src/main/java/io/tapdata/flow/engine/V2/node/hazelcast/processor/HazelcastProcessorNode.java
@@ -17,7 +17,11 @@
 import io.tapdata.entity.codec.ToTapValueCodec;
 import io.tapdata.entity.event.TapEvent;
 import io.tapdata.entity.event.control.HeartbeatEvent;
+import io.tapdata.entity.event.dml.TapDeleteRecordEvent;
+import io.tapdata.entity.event.dml.TapInsertRecordEvent;
 import io.tapdata.entity.event.dml.TapRecordEvent;
+import io.tapdata.entity.event.dml.TapUpdateRecordEvent;
+import io.tapdata.entity.logger.TapLogger;
 import io.tapdata.entity.schema.TapTable;
 import io.tapdata.entity.schema.type.TapType;
 import io.tapdata.entity.schema.value.TapValue;
@@ -80,7 +84,7 @@ protected void updateNodeConfig(TapdataEvent tapdataEvent) {
 		}
 	}
 
-	private void initDataFlowProcessor() throws TapCodeException {
+	protected void initDataFlowProcessor() throws TapCodeException {
 		final Stage stage = HazelcastUtil.node2CommonStage(processorBaseContext.getNode());
 		dataFlowProcessor = createDataFlowProcessor(processorBaseContext.getNode(), stage);
 		Job job = new Job();
@@ -105,6 +109,7 @@ private void initDataFlowProcessor() throws TapCodeException {
 		);
 		try {
 			dataFlowProcessor.initialize(processorContext, stage);
+			dataFlowProcessor.logListener(logListener());
 		} catch (Exception e) {
 			throw new TapCodeException(TaskProcessorExCode_11.INIT_DATA_FLOW_PROCESSOR_FAILED, "Init data flow processor failed", e);
 		}
@@ -139,6 +144,15 @@ protected void tryProcess(TapdataEvent tapdataEvent, BiConsumer<TapdataEvent, Pr
 				} else {
 					TapEventUtil.setBefore(tapRecordEvent, processedMessage.getBefore());
 					TapEventUtil.setAfter(tapRecordEvent, processedMessage.getAfter());
+					List<String> removedFields = null;
+					if (tapRecordEvent instanceof TapUpdateRecordEvent) {
+						removedFields = ((TapUpdateRecordEvent) tapRecordEvent).getRemovedFields();
+					}
+					tapRecordEvent = message2TapEvent(processedMessage);
+					if (null != removedFields) {
+						TapEventUtil.setRemoveFields(tapRecordEvent, removedFields);
+					}
+					tapdataEvent.setTapEvent(tapRecordEvent);
 				}
 				handleRemoveFields(tapdataEvent);
 				consumer.accept(tapdataEvent, getProcessResult(processedMessage.getTableName()));
@@ -202,7 +216,7 @@ private void updateTapType(Map<String, Object> map, Map<String, TapType> tapType
 
 	}
 
-	private DataFlowProcessor createDataFlowProcessor(Node node, Stage stage) {
+	protected DataFlowProcessor createDataFlowProcessor(Node node, Stage stage) {
 		NodeTypeEnum nodeType = NodeTypeEnum.get(node.getType());
 		DataFlowProcessor dataFlowProcessor = null;
 		switch (nodeType) {
diff --git a/iengine/iengine-app/src/main/java/io/tapdata/websocket/ManagementWebsocketHandler.java b/iengine/iengine-app/src/main/java/io/tapdata/websocket/ManagementWebsocketHandler.java
index 39be6f205e..56968346d5 100644
--- a/iengine/iengine-app/src/main/java/io/tapdata/websocket/ManagementWebsocketHandler.java
+++ b/iengine/iengine-app/src/main/java/io/tapdata/websocket/ManagementWebsocketHandler.java
@@ -225,7 +225,7 @@ protected void connect(String baseURL) {
 					+ "?agentId={agentId}&access_token={access_token}");
 
 
-			WebSocketClient client = new StandardWebSocketClient();
+			WebSocketClient client = ManagementWebsocketHandler.createWebSocketClient();
 
 			currentWsUrl = UriComponentsBuilder.fromUriString(currentWsUrl)
 					.buildAndExpand(agentId, configCenter.getConfig(ConfigurationCenter.TOKEN)).encode().toUri().toString();
@@ -247,9 +247,12 @@ protected void connect(String baseURL) {
 
 			session.setSession(listenableFuture.get());
 			logger.info("Connect to web socket server success, url {}", currentWsUrl);
-		} catch (Exception e) {
-			logger.error("Create web socket by url {} connection failed {}", currentWsUrl, e.getMessage(), e);
+		} catch (InterruptedException interruptedException) {
 			Thread.currentThread().interrupt();
+			logger.warn("Connect to web socket Thread interrupted,Thread name:{}", Thread.currentThread().getName());
+		}
+		catch (Exception e) {
+			logger.error("Create web socket by url {} connection failed {}", currentWsUrl, e.getMessage(), e);
 		}
 	}
 
@@ -387,6 +390,10 @@ public void sendMessage(TextMessage textMessage) throws RuntimeException {
 		session.sendMessage(textMessage);
 	}
 
+	public static WebSocketClient createWebSocketClient() {
+		return new StandardWebSocketClient();
+	}
+
 	protected class SessionOption implements AutoCloseable {
 		private WebSocketSession session;
 
diff --git a/iengine/iengine-common/src/main/java/com/tapdata/mongo/HttpClientMongoOperator.java b/iengine/iengine-common/src/main/java/com/tapdata/mongo/HttpClientMongoOperator.java
index 74ebf1bf73..9d165df294 100644
--- a/iengine/iengine-common/src/main/java/com/tapdata/mongo/HttpClientMongoOperator.java
+++ b/iengine/iengine-common/src/main/java/com/tapdata/mongo/HttpClientMongoOperator.java
@@ -34,6 +34,10 @@
 import java.io.File;
 import java.lang.reflect.Field;
 import java.lang.reflect.Modifier;
+import java.time.ZoneId;
+import java.time.ZoneOffset;
+import java.time.ZonedDateTime;
+import java.time.format.DateTimeFormatter;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Date;
@@ -750,7 +754,7 @@ private <T> T findAndModifyByCollection(String collection, Query query, Update u
 		return result;
 	}
 
-	private void validateToken() {
+	protected void validateToken() {
 
 		LoginResp loginResp = (LoginResp) configCenter.getConfig(ConfigurationCenter.LOGIN_INFO);
 
@@ -758,13 +762,15 @@ private void validateToken() {
 			refreshToken();
 		} else {
 			long expiredTimestamp = loginResp.getExpiredTimestamp();
-			if (expiredTimestamp - System.currentTimeMillis() <= 86400 * 1000) {
+			Long ttl = loginResp.getTtl();
+			long bufferTime = ttl > 0 ? ttl / 10 : 0;
+			if (expiredTimestamp - System.currentTimeMillis() < bufferTime) {
 				refreshToken();
 			}
 		}
 	}
 
-	private void refreshToken() {
+	protected void refreshToken() {
 		String accessCode = (String) configCenter.getConfig(ConfigurationCenter.ACCESS_CODE);
 
 		LoginResp loginResp = null;
@@ -776,8 +782,13 @@ private void refreshToken() {
 			try {
 				loginResp = restTemplateOperator.postOne(params, "users/generatetoken", LoginResp.class);
 				if (loginResp != null) {
-					Date date = (Date) DateUtil.parse(loginResp.getCreated());
-					long expiredTimestamp = date.getTime() + (loginResp.getTtl() * 1000);
+					String parsePattern = DateUtil.determineDateFormat(loginResp.getCreated());
+					DateTimeFormatter formatter = DateTimeFormatter.ofPattern(parsePattern);
+					ZonedDateTime utcZonedDateTime = ZonedDateTime.parse(loginResp.getCreated(), formatter.withZone(ZoneOffset.UTC));
+					ZonedDateTime systemZonedDateTime = utcZonedDateTime.withZoneSameInstant(ZoneId.systemDefault());
+					Date date = Date.from(systemZonedDateTime.toInstant());
+
+					long expiredTimestamp = date.getTime() + loginResp.getTtl() * 1000;
 					loginResp.setExpiredTimestamp(expiredTimestamp);
 
 					configCenter.putConfig(ConfigurationCenter.TOKEN, loginResp.getId());
@@ -797,6 +808,7 @@ private void refreshToken() {
 					}
 				}
 			} catch (Exception e) {
+				retryCount++;
 				logger.error("refresh token failed {}", e.getMessage(), e);
 			}
 		}
diff --git a/iengine/iengine-common/src/main/java/com/tapdata/processor/dataflow/DataFlowProcessor.java b/iengine/iengine-common/src/main/java/com/tapdata/processor/dataflow/DataFlowProcessor.java
index f271cccf77..9f61972589 100644
--- a/iengine/iengine-common/src/main/java/com/tapdata/processor/dataflow/DataFlowProcessor.java
+++ b/iengine/iengine-common/src/main/java/com/tapdata/processor/dataflow/DataFlowProcessor.java
@@ -2,6 +2,7 @@
 
 import com.tapdata.entity.dataflow.Stage;
 import com.tapdata.processor.Processor;
+import io.tapdata.entity.logger.TapLogger;
 
 /**
  * @author jackin
@@ -11,5 +12,6 @@ public interface DataFlowProcessor extends Processor {
 	void initialize(ProcessorContext context, Stage stage) throws Exception;
 
 	Stage getStage();
-
+	default void logListener(TapLogger.LogListener logListener) {
+	}
 }
diff --git a/iengine/iengine-common/src/main/java/com/tapdata/processor/dataflow/FieldDataFlowProcessor.java b/iengine/iengine-common/src/main/java/com/tapdata/processor/dataflow/FieldDataFlowProcessor.java
index 27b81a4d7d..0bbd0bcc34 100644
--- a/iengine/iengine-common/src/main/java/com/tapdata/processor/dataflow/FieldDataFlowProcessor.java
+++ b/iengine/iengine-common/src/main/java/com/tapdata/processor/dataflow/FieldDataFlowProcessor.java
@@ -155,10 +155,10 @@ record = new HashMap<>();
 					FieldProcessUtil.fieldProcess(after, fieldProcesses, rollbackRemoveFields, deleteAllFields);
 					message.setAfter(after);
 				}
-				if (null != before){
+				if (MapUtils.isNotEmpty(before)){
 					fieldScript(message, before, "before");
 				}
-				if (null != after){
+				if (MapUtils.isNotEmpty(after)){
 					fieldScript(message, after, "after");
 				}
 			} else if (CollectionUtils.isNotEmpty(cloneFieldProcesses)) {
diff --git a/iengine/iengine-common/src/main/java/com/tapdata/processor/dataflow/RowFilterProcessor.java b/iengine/iengine-common/src/main/java/com/tapdata/processor/dataflow/RowFilterProcessor.java
index d5a292cd2d..818958a325 100644
--- a/iengine/iengine-common/src/main/java/com/tapdata/processor/dataflow/RowFilterProcessor.java
+++ b/iengine/iengine-common/src/main/java/com/tapdata/processor/dataflow/RowFilterProcessor.java
@@ -9,6 +9,7 @@
 import com.tapdata.processor.ScriptConnection;
 import com.tapdata.processor.ScriptUtil;
 import com.tapdata.processor.error.RowFilterProcessorExCode_24;
+import io.tapdata.entity.logger.TapLogger;
 import io.tapdata.exception.TapCodeException;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.collections.MapUtils;
@@ -43,6 +44,7 @@ public class RowFilterProcessor implements DataFlowProcessor {
 	private FilterAction action;
 
 	private Map<String, Object> processContext;
+	private boolean withoutBeforeLog;
 
 	@Override
 	public void initialize(ProcessorContext context, Stage stage) throws Exception {
@@ -71,6 +73,12 @@ public void initialize(ProcessorContext context, Stage stage) throws Exception {
 		}
 
 		processContext = new ConcurrentHashMap<>();
+		withoutBeforeLog = false;
+	}
+
+	private TapLogger.LogListener logListener;
+	public void logListener(TapLogger.LogListener logListener){
+		this.logListener = logListener;
 	}
 
 	public MessageEntity process(MessageEntity message) {
@@ -87,11 +95,34 @@ public MessageEntity process(MessageEntity message) {
 			Map<String, Object> record = MapUtils.isNotEmpty(message.getAfter()) ? message.getAfter() : message.getBefore();
 			try {
 				if (MapUtils.isNotEmpty(record)) {
-					Object o = ScriptUtil.invokeScript(engine, SCRIPT_FUNCTION_NAME, message, context.getSourceConn(), context.getTargetConn(), context.getJob(), processContext, logger, null);
-					if (o instanceof Boolean) {
-						Boolean result = (Boolean) o;
+					if ("u".equals(message.getOp())) {
+						if (MapUtils.isEmpty(message.getBefore()) && !withoutBeforeLog && null != logListener) {
+							logListener.warn("current database does not support record before value when update, the data will insert or delete record if after value meets condition");
+							withoutBeforeLog = true;
+						}
+						Object before = ScriptUtil.invokeScript(engine, SCRIPT_FUNCTION_NAME, message, context.getSourceConn(), context.getTargetConn(), context.getJob(), processContext, logger, "before");
+						Object after = ScriptUtil.invokeScript(engine, SCRIPT_FUNCTION_NAME, message, context.getSourceConn(), context.getTargetConn(), context.getJob(), processContext, logger, "after");
+						if (isTrue(before) && !isTrue(after)) { // before满足，after不满足
+							beforeMeetMessage(message);
+						} else if (!isTrue(before) && isTrue(after)) { // before不满足，after满足
+							afterMeetMessage(message);
+						} else if (isTrue(before) && isTrue(after)) {
+							if (MapUtils.isEmpty(message.getBefore())) {
+								afterMeetMessage(message);
+							} else {
+								message = FilterAction.DISCARD == action ? null : message;
+							}
+						} else { // 不满足
+							if (MapUtils.isEmpty(message.getBefore())) {
+								beforeMeetMessage(message);
+							}else {
+								message = FilterAction.DISCARD == action ? message : null;
+							}
+						}
+					} else {
+						Object o = ScriptUtil.invokeScript(engine, SCRIPT_FUNCTION_NAME, message, context.getSourceConn(), context.getTargetConn(), context.getJob(), processContext, logger, null);
 						// 满足条件处理
-						if (result != null && result) {
+						if (isTrue(o)) {
 							message = FilterAction.DISCARD == action ? null : message;
 						}
 						// 不满足条件处理
@@ -99,10 +130,6 @@ public MessageEntity process(MessageEntity message) {
 							message = FilterAction.DISCARD == action ? message : null;
 						}
 					}
-					// 不满足条件处理
-					else {
-						message = FilterAction.DISCARD == action ? message : null;
-					}
 				}
 			} catch (Exception e) {
 				TapCodeException tapCodeException = new TapCodeException(RowFilterProcessorExCode_24.JAVA_SCRIPT_ERROR, e);
@@ -113,6 +140,39 @@ public MessageEntity process(MessageEntity message) {
 		return message;
 	}
 
+	private void beforeMeetMessage(MessageEntity message) {
+		if (FilterAction.DISCARD == action) {
+			message.setOp("i");
+			message.setBefore(null);
+		} else {
+			message.setOp("d");
+			if (MapUtils.isEmpty(message.getBefore())) {
+				message.setBefore(message.getAfter());
+			}
+			message.setAfter(null);
+		}
+	}
+
+	private void afterMeetMessage(MessageEntity message) {
+		if (FilterAction.DISCARD == action) {
+			message.setOp("d");
+			if (MapUtils.isEmpty(message.getBefore())) {
+				message.setBefore(message.getAfter());
+			}
+			message.setAfter(null);
+		} else {
+			message.setOp("i");
+			message.setBefore(null);
+		}
+	}
+
+	private boolean isTrue(Object o) {
+		if (o instanceof Boolean) {
+			return (Boolean) o;
+		}
+		return false;
+	}
+
 	@Override
 	public List<MessageEntity> process(List<MessageEntity> batch) {
 		if (CollectionUtils.isNotEmpty(batch)) {
diff --git a/iengine/modules/observable-module/src/main/java/io/tapdata/observable/metric/handler/DataNodeSampleHandler.java b/iengine/modules/observable-module/src/main/java/io/tapdata/observable/metric/handler/DataNodeSampleHandler.java
index 85be3274ce..22b1e21847 100644
--- a/iengine/modules/observable-module/src/main/java/io/tapdata/observable/metric/handler/DataNodeSampleHandler.java
+++ b/iengine/modules/observable-module/src/main/java/io/tapdata/observable/metric/handler/DataNodeSampleHandler.java
@@ -403,7 +403,7 @@ public void handleWriteRecordAccept(Long acceptTime, WriteListResult<TapRecordEv
 	AtomicBoolean firstTableCount = new AtomicBoolean(true);
 
 	public void handleTableCountAccept(String table ,long count) {
-		currentSnapshotTableRowTotalMap.putIfAbsent(table, count);
+		currentSnapshotTableRowTotalMap.put(table, count);
 		Optional.ofNullable(snapshotRowCounter).ifPresent(counter -> counter.inc(count));
 	}
 
diff --git a/iengine/modules/observable-module/src/main/java/io/tapdata/observable/metric/handler/RandomSampleEventHandler.java b/iengine/modules/observable-module/src/main/java/io/tapdata/observable/metric/handler/RandomSampleEventHandler.java
index a3e7bef418..1e296ecc84 100644
--- a/iengine/modules/observable-module/src/main/java/io/tapdata/observable/metric/handler/RandomSampleEventHandler.java
+++ b/iengine/modules/observable-module/src/main/java/io/tapdata/observable/metric/handler/RandomSampleEventHandler.java
@@ -43,12 +43,12 @@ public long sampleMemoryTapEvent(List<?> events, HandleEvent handle) {
         if (null == events) return 0L;
         List<Object> samples = Collections.singletonList(events.get(0));
         totalSize = events.size();
-		long sizeOfSampleListByte = 0L;
+		long sizeOfSampleByte = 0L;
         for (Object item : samples) {
             TapEvent tapEvent = handle.handel(item);
-            sizeOfSampleListByte += sizeOfTapEvent(tapEvent);
+            sizeOfSampleByte += sizeOfTapEvent(tapEvent);
         }
-        return sizeOfSampleListByte * getTotalSize();
+        return sizeOfSampleByte;
     }
 
     protected long sizeOfTapEvent(TapEvent tapEvent) {
@@ -83,7 +83,7 @@ protected List<Object> randomSampleList(List<?> events, Double sampleRate) {
 
     protected void unitConversion(HandlerUtil.EventTypeRecorder recorder, long sizeOfSampleListByte) {
         long size = sizeOfSampleListByte * getTotalSize();
-        recorder.setMemorySize(size / 10);
+        recorder.setMemorySize(size / 4);
         recorder.setMemoryUtil("B");
     }
 
diff --git a/iengine/modules/observable-module/src/main/java/io/tapdata/observable/metric/handler/TaskSampleHandler.java b/iengine/modules/observable-module/src/main/java/io/tapdata/observable/metric/handler/TaskSampleHandler.java
index 122c3fcc3d..128e3d9966 100644
--- a/iengine/modules/observable-module/src/main/java/io/tapdata/observable/metric/handler/TaskSampleHandler.java
+++ b/iengine/modules/observable-module/src/main/java/io/tapdata/observable/metric/handler/TaskSampleHandler.java
@@ -270,7 +270,7 @@ public void addTable(String... tables) {
 
     public void handleTableCountAccept(String table, long count) {
         snapshotRowTotal.inc(count);
-        currentSnapshotTableRowTotalMap.putIfAbsent(table, count);
+        currentSnapshotTableRowTotalMap.put(table, count);
     }
 
     public void handleCreateTableEnd() {
diff --git a/manager/tm-api/src/main/java/com/tapdata/tm/alarm/service/AlarmService.java b/manager/tm-api/src/main/java/com/tapdata/tm/alarm/service/AlarmService.java
index ce13e80778..597964a20c 100644
--- a/manager/tm-api/src/main/java/com/tapdata/tm/alarm/service/AlarmService.java
+++ b/manager/tm-api/src/main/java/com/tapdata/tm/alarm/service/AlarmService.java
@@ -10,6 +10,7 @@
 import com.tapdata.tm.commons.task.constant.NotifyEnum;
 import com.tapdata.tm.commons.task.dto.TaskDto;
 import com.tapdata.tm.commons.task.dto.alarm.AlarmRuleDto;
+import com.tapdata.tm.commons.task.dto.alarm.AlarmSettingDto;
 import com.tapdata.tm.commons.task.dto.alarm.AlarmSettingVO;
 import com.tapdata.tm.commons.task.dto.alarm.AlarmVO;
 import com.tapdata.tm.config.security.UserDetail;
@@ -53,6 +54,8 @@ Page<AlarmListInfoVo> list(String status,
 
     List<AlarmChannelDto> getAvailableChannels();
 
+    boolean checkOpen(TaskDto taskDto, String nodeId, AlarmKeyEnum key, NotifyEnum type, List<AlarmSettingDto> settingDtos);
+
     boolean checkOpen(TaskDto taskDto, String nodeId, AlarmKeyEnum key, NotifyEnum type, UserDetail userDetail);
 
 	boolean checkOpen(List<AlarmSettingVO> alarmSettingVOS, AlarmKeyEnum key, NotifyEnum type, UserDetail userDetail);
diff --git a/manager/tm-api/src/main/java/com/tapdata/tm/base/aop/MeasureAOP.java b/manager/tm-api/src/main/java/com/tapdata/tm/base/aop/MeasureAOP.java
index 42fb786294..d5a40b6b22 100644
--- a/manager/tm-api/src/main/java/com/tapdata/tm/base/aop/MeasureAOP.java
+++ b/manager/tm-api/src/main/java/com/tapdata/tm/base/aop/MeasureAOP.java
@@ -3,6 +3,7 @@
 import cn.hutool.core.date.DateUnit;
 import cn.hutool.core.date.DateUtil;
 import com.google.common.collect.Maps;
+import com.tapdata.tm.Settings.service.AlarmSettingService;
 import com.tapdata.tm.alarm.constant.AlarmComponentEnum;
 import com.tapdata.tm.alarm.constant.AlarmStatusEnum;
 import com.tapdata.tm.alarm.constant.AlarmTypeEnum;
@@ -13,12 +14,14 @@
 import com.tapdata.tm.commons.task.constant.AlarmKeyEnum;
 import com.tapdata.tm.commons.task.dto.TaskDto;
 import com.tapdata.tm.commons.task.dto.alarm.AlarmRuleDto;
+import com.tapdata.tm.commons.task.dto.alarm.AlarmSettingDto;
 import com.tapdata.tm.config.security.UserDetail;
 import com.tapdata.tm.message.constant.Level;
 import com.tapdata.tm.task.service.TaskService;
 import com.tapdata.tm.user.service.UserService;
 import com.tapdata.tm.utils.MongoUtils;
 import io.tapdata.common.sample.request.SampleRequest;
+import org.apache.commons.collections4.CollectionUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.aspectj.lang.JoinPoint;
 import org.aspectj.lang.annotation.AfterReturning;
@@ -28,10 +31,7 @@
 import org.springframework.data.mongodb.core.query.Update;
 import org.springframework.stereotype.Component;
 
-import java.util.List;
-import java.util.Map;
-import java.util.Objects;
-import java.util.Optional;
+import java.util.*;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Function;
 import java.util.stream.Collectors;
@@ -46,34 +46,55 @@ public class MeasureAOP {
     private final TaskService taskService;
     private final AlarmService alarmService;
     private final UserService userService;
+    private final AlarmSettingService alarmSettingService;
 //    private final InspectService inspectService;
     private final Map<String, Map<String, AtomicInteger>> obsMap = Maps.newConcurrentMap();
 
-    public MeasureAOP(TaskService taskService, AlarmService alarmService, UserService userService) {
+    private final String TASK_ID = "taskId";
+
+    private final String CURRENT_EVENT_TIMESTAMP = "currentEventTimestamp";
+
+    public MeasureAOP(TaskService taskService, AlarmService alarmService, UserService userService,AlarmSettingService alarmSettingService) {
         this.taskService = taskService;
         this.alarmService = alarmService;
         this.userService = userService;
+        this.alarmSettingService = alarmSettingService;
     }
 
 
     @AfterReturning("execution(* com.tapdata.tm.monitor.service.MeasurementServiceV2.addAgentMeasurement(..))")
     public void addAgentMeasurement(JoinPoint joinPoint) {
         List<SampleRequest> samples = (List<SampleRequest>) joinPoint.getArgs()[0];
-
+        Set<String> taskIds = samples.stream().filter(sampleRequest -> sampleRequest.getTags().containsKey(TASK_ID)).map(sampleRequest -> sampleRequest.getTags().get(TASK_ID)).collect(Collectors.toSet());
+        Map<String, TaskDto> taskDtoMap = new HashMap<>();
+        Map<String, UserDetail> userDetailMap = new HashMap<>();
+        Map<String, List<AlarmSettingDto>> alarmSettingMap = new HashMap<>();
+        if(CollectionUtils.isNotEmpty(taskIds)){
+            taskIds.forEach(taskId -> {
+                TaskDto taskDto = taskService.findByTaskId(MongoUtils.toObjectId(taskId),"_id","dag","user_id","agentId","name",CURRENT_EVENT_TIMESTAMP,"alarmSettings");
+                taskDtoMap.put(taskId, taskDto);
+            });
+            taskDtoMap.values().forEach(taskDto -> {
+                UserDetail userDetail = userService.loadUserById(MongoUtils.toObjectId(taskDto.getUserId()));
+                List<AlarmSettingDto> alarmSettingDtos = alarmSettingService.findAllAlarmSetting(userDetail);
+                userDetailMap.put(taskDto.getUserId(), userDetail);
+                alarmSettingMap.put(userDetail.getUserId(), alarmSettingDtos);
+            });
+        }
         samples.forEach(sampleRequest -> {
             String type = sampleRequest.getTags().get("type");
-            String taskId = sampleRequest.getTags().get("taskId");
+            String taskId = sampleRequest.getTags().get(TASK_ID);
             Map<String, Number> vs = sampleRequest.getSample().getVs();
 
             if (StringUtils.isNotBlank(taskId)) {
-                TaskDto taskDto = taskService.findByTaskId(MongoUtils.toObjectId(taskId));
+                TaskDto taskDto = taskDtoMap.get(taskId);
                 // task alarm
                 Map<String, List<AlarmRuleDto>> ruleMap = alarmService.getAlarmRuleDtos(taskDto);
                 if (null != ruleMap && !ruleMap.isEmpty()) {
-                    UserDetail userDetail = userService.loadUserById(MongoUtils.toObjectId(taskDto.getUserId()));
+                    UserDetail userDetail = userDetailMap.get(taskDto.getUserId());
                     if ("task".equals(type)) {
 
-                        boolean checkOpen = alarmService.checkOpen(taskDto, null, AlarmKeyEnum.TASK_INCREMENT_DELAY, null, userDetail);
+                        boolean checkOpen =alarmService.checkOpen(taskDto, null, AlarmKeyEnum.TASK_INCREMENT_DELAY, null, alarmSettingMap.get(userDetail.getUserId()));
                         if (checkOpen) {
                             Optional.ofNullable(ruleMap.get(taskId)).ifPresent(rules -> {
                                 Map<AlarmKeyEnum, AlarmRuleDto> collect = rules.stream().collect(Collectors.toMap(AlarmRuleDto::getKey, Function.identity(), (e1, e2) -> e1));
@@ -95,13 +116,13 @@ public void addAgentMeasurement(JoinPoint joinPoint) {
                                 String nodeName = dag.getNode(nodeId).getName();
                                 boolean checkOpen;
                                 if (sourceNode.isPresent()) {
-                                    checkOpen = alarmService.checkOpen(taskDto, nodeId, AlarmKeyEnum.DATANODE_AVERAGE_HANDLE_CONSUME, null, userDetail);
+                                    checkOpen = alarmService.checkOpen(taskDto, nodeId, AlarmKeyEnum.DATANODE_AVERAGE_HANDLE_CONSUME, null, alarmSettingMap.get(userDetail.getUserId()));
                                     template = getTemplate(AlarmKeyEnum.DATANODE_AVERAGE_HANDLE_CONSUME, true);
                                 } else if (targetNode.isPresent()) {
-                                    checkOpen = alarmService.checkOpen(taskDto, nodeId, AlarmKeyEnum.DATANODE_AVERAGE_HANDLE_CONSUME, null, userDetail);
+                                    checkOpen = alarmService.checkOpen(taskDto, nodeId, AlarmKeyEnum.DATANODE_AVERAGE_HANDLE_CONSUME, null, alarmSettingMap.get(userDetail.getUserId()));
                                     template = getTemplate(AlarmKeyEnum.DATANODE_AVERAGE_HANDLE_CONSUME, false);
                                 } else {
-                                    checkOpen = alarmService.checkOpen(taskDto, nodeId, AlarmKeyEnum.PROCESSNODE_AVERAGE_HANDLE_CONSUME, null, userDetail);
+                                    checkOpen = alarmService.checkOpen(taskDto, nodeId, AlarmKeyEnum.PROCESSNODE_AVERAGE_HANDLE_CONSUME, null, alarmSettingMap.get(userDetail.getUserId()));
                                     template = getTemplate(AlarmKeyEnum.PROCESSNODE_AVERAGE_HANDLE_CONSUME, null);
                                 }
 
@@ -151,7 +172,7 @@ private void setTaskSnapshotDate(Map<String, Number> vs, String taskId, TaskDto
 
         }
 
-        Number currentEventTimestamp = vs.get("currentEventTimestamp");
+        Number currentEventTimestamp = vs.get(CURRENT_EVENT_TIMESTAMP);
         boolean checkCdcOpen = alarmService.checkOpen(taskDto, null, AlarmKeyEnum.TASK_INCREMENT_START, null, userDetail);
         if (checkCdcOpen && Objects.isNull(taskDto.getCurrentEventTimestamp()) && Objects.nonNull(currentEventTimestamp) && currentEventTimestamp.longValue() > 0L  && TaskDto.STATUS_RUNNING.equals(taskDto.getStatus())) {
             Map<String, Object> param = Maps.newHashMap();
@@ -175,7 +196,7 @@ private void setTaskSnapshotDate(Map<String, Number> vs, String taskId, TaskDto
         }
         if (Objects.nonNull(currentEventTimestamp) && !currentEventTimestamp.equals(taskDto.getCurrentEventTimestamp())
                 && !currentEventTimestamp.equals(0)) {
-            update.set("currentEventTimestamp", currentEventTimestamp);
+            update.set(CURRENT_EVENT_TIMESTAMP, currentEventTimestamp);
         }
 
         if (update.getUpdateObject().size() > 0) {
@@ -284,7 +305,7 @@ protected void taskIncrementDelayAlarm(TaskDto task, String taskId, Number repli
         }
     }
 
-    private void supplmentDelayAvg(TaskDto task, String taskId, Number number, AlarmRuleDto alarmRuleDto, String nodeId, String nodeName, String[] template) {
+    protected void supplmentDelayAvg(TaskDto task, String taskId, Number number, AlarmRuleDto alarmRuleDto, String nodeId, String nodeName, String[] template) {
         if (Objects.isNull(number) || Objects.isNull(alarmRuleDto)) {
             return;
         }
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 63ac5557c5..ab817989d2 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
@@ -494,9 +494,9 @@ protected static TestResponseDto sendEmailForProxy(MailAccountDto parms, List<St
 
         Properties properties = new Properties();
         properties.put("mail.smtp.host", parms.getHost());
-        properties.put("mail.smtp.port", parms.getPort());
+        properties.put("mail.smtp.port", String.valueOf(parms.getPort()));
         properties.put("mail.smtp.auth", "true");
-        if ("SSL".equals(parms.getProtocol())){
+        if ("SSL".equals(parms.getProtocol())) {
             properties.put("mail.smtp.ssl.enable", "true");
         } else if ("TLS".equals(parms.getProtocol())) {
             properties.put("mail.smtp.starttls.enable", "true");
diff --git a/manager/tm-common/src/main/java/com/tapdata/tm/commons/dag/Node.java b/manager/tm-common/src/main/java/com/tapdata/tm/commons/dag/Node.java
index ad4198f039..e5b3e1991a 100644
--- a/manager/tm-common/src/main/java/com/tapdata/tm/commons/dag/Node.java
+++ b/manager/tm-common/src/main/java/com/tapdata/tm/commons/dag/Node.java
@@ -11,6 +11,7 @@
 import com.tapdata.tm.commons.task.dto.alarm.AlarmRuleVO;
 import com.tapdata.tm.commons.task.dto.alarm.AlarmSettingVO;
 import com.tapdata.tm.commons.util.CapitalizedEnum;
+import com.tapdata.tm.commons.util.NoPrimaryKeyVirtualField;
 import io.github.openlg.graphlib.Graph;
 import io.tapdata.entity.event.ddl.TapDDLEvent;
 import io.tapdata.entity.event.ddl.entity.ValueChange;
@@ -248,6 +249,7 @@ public void transformSchema(DAG.Options options) {
         }
 
         if (this.outputSchema != null) {
+            NoPrimaryKeyVirtualField.addVirtualField(outputSchema, this);
             S changedSchema = outputSchema;//filterChangedSchema(this.outputSchema, options);  // 过滤出修改过的模型
             if (changedSchema != null) {
                 String taskId = service.getTaskId().toHexString();
diff --git a/manager/tm-common/src/main/java/com/tapdata/tm/commons/dag/nodes/DataParentNode.java b/manager/tm-common/src/main/java/com/tapdata/tm/commons/dag/nodes/DataParentNode.java
index 5cb5453e5f..e7dcca120d 100644
--- a/manager/tm-common/src/main/java/com/tapdata/tm/commons/dag/nodes/DataParentNode.java
+++ b/manager/tm-common/src/main/java/com/tapdata/tm/commons/dag/nodes/DataParentNode.java
@@ -14,6 +14,7 @@
 import com.tapdata.tm.commons.task.dto.MergeTableProperties;
 import com.tapdata.tm.commons.task.dto.TaskDto;
 import com.tapdata.tm.commons.util.MetaDataBuilderUtils;
+import com.tapdata.tm.commons.util.NoPrimaryKeySyncMode;
 import lombok.EqualsAndHashCode;
 import lombok.Getter;
 import lombok.Setter;
@@ -101,6 +102,9 @@ public abstract class DataParentNode<S> extends Node<S> {
 
     /** 根据更新条件创建唯一索引或普通索引，默认为 true，创建唯一索引*/
     private Boolean uniqueIndexEnable;
+
+    /** 无主键表同步模式， */
+    private String noPkSyncMode;
     /**
      * constructor for node
      *
@@ -267,6 +271,10 @@ public void setReadPartitionOptions(ReadPartitionOptions readPartitionOptions) {
         this.readPartitionOptions = readPartitionOptions;
     }
 
+    public NoPrimaryKeySyncMode getNoPrimaryKeySyncMode() {
+        return NoPrimaryKeySyncMode.fromValue(getNoPkSyncMode());
+    }
+
     protected void removeAllPrimaryKeys(Schema schema) {
         if(null == schema) return;
         List<Field> fields = schema.getFields();
diff --git a/manager/tm-common/src/main/java/com/tapdata/tm/commons/exception/NoPrimaryKeyException.java b/manager/tm-common/src/main/java/com/tapdata/tm/commons/exception/NoPrimaryKeyException.java
new file mode 100644
index 0000000000..2b68902a7c
--- /dev/null
+++ b/manager/tm-common/src/main/java/com/tapdata/tm/commons/exception/NoPrimaryKeyException.java
@@ -0,0 +1,41 @@
+package com.tapdata.tm.commons.exception;
+
+/**
+ * 无主键处理异常
+ *
+ * @author <a href="mailto:harsen_lin@163.com">Harsen</a>
+ * @version v1.0 2024/10/17 16:07 Create
+ */
+public class NoPrimaryKeyException extends RuntimeException {
+    public static final int CODE_INCOMPLETE_FIELDS = 1;
+    public static final int CODE_NOTFOUND_HASH_ALGORITHM = 10;
+    public static final int CODE_OTHER = 99;
+
+    private final int code;
+
+    public NoPrimaryKeyException(String message, int code) {
+        super(message);
+        this.code = code;
+    }
+
+    public NoPrimaryKeyException(String message, Throwable cause, int code) {
+        super(message, cause);
+        this.code = code;
+    }
+
+    public int getCode() {
+        return code;
+    }
+
+    public static NoPrimaryKeyException incompleteFields(String field) {
+        return new NoPrimaryKeyException(String.format("not exist field '%s'", field), CODE_INCOMPLETE_FIELDS);
+    }
+
+    public static NoPrimaryKeyException notfoundHashAlgorithm(String type, Throwable e) {
+        return new NoPrimaryKeyException(String.format("%s algorithm not found", type), e, CODE_NOTFOUND_HASH_ALGORITHM);
+    }
+
+    public static NoPrimaryKeyException otherFailed(Throwable e) {
+        return new NoPrimaryKeyException("add hash filed value failed", e, CODE_OTHER);
+    }
+}
diff --git a/manager/tm-common/src/main/java/com/tapdata/tm/commons/schema/Field.java b/manager/tm-common/src/main/java/com/tapdata/tm/commons/schema/Field.java
index 183298bf8c..977f4f98aa 100644
--- a/manager/tm-common/src/main/java/com/tapdata/tm/commons/schema/Field.java
+++ b/manager/tm-common/src/main/java/com/tapdata/tm/commons/schema/Field.java
@@ -14,6 +14,7 @@ public class Field implements Serializable {
     public static final String SOURCE_JOB_ANALYZE = "job_analyze";
     public static final String SOURCE_AUTO = "auto";
     public static final String SOURCE_MANUAL = "manual";
+    public static final String SOURCE_VIRTUAL_HASH = "virtual_hash";
 
     private Object autoincrement;
 
@@ -165,7 +166,7 @@ public String getId() {
     private Boolean fixed;
 
     private Boolean dataTypeSupport;
-    private String createSource; // manual/auto/job_analyze
+    private String createSource; // manual/auto/job_analyze/virtual_pk
     private String changeRuleId; // 如果命中字段变更规则则存在值
 
     private boolean useDefaultValue = true;
diff --git a/manager/tm-common/src/main/java/com/tapdata/tm/commons/util/NoPrimaryKeySyncMode.java b/manager/tm-common/src/main/java/com/tapdata/tm/commons/util/NoPrimaryKeySyncMode.java
new file mode 100644
index 0000000000..37192e6900
--- /dev/null
+++ b/manager/tm-common/src/main/java/com/tapdata/tm/commons/util/NoPrimaryKeySyncMode.java
@@ -0,0 +1,26 @@
+package com.tapdata.tm.commons.util;
+
+/**
+ * 无主键表处理模式
+ *
+ * @author <a href="mailto:harsen_lin@163.com">Harsen</a>
+ * @version v1.0 2024/10/25 17:03 Create
+ */
+public enum NoPrimaryKeySyncMode {
+
+    ALL_COLUMNS,
+    ADD_HASH,
+    ;
+
+    public static NoPrimaryKeySyncMode fromValue(String value) {
+        // value 为空时代表旧任务配置，使用 ALL_COLUMNS 模式
+        if (null == value || value.trim().isEmpty()) return ALL_COLUMNS;
+
+        for (NoPrimaryKeySyncMode item : NoPrimaryKeySyncMode.values()) {
+            if (item.name().equalsIgnoreCase(value)) {
+                return item;
+            }
+        }
+        return ADD_HASH;
+    }
+}
diff --git a/manager/tm-common/src/main/java/com/tapdata/tm/commons/util/NoPrimaryKeyVirtualField.java b/manager/tm-common/src/main/java/com/tapdata/tm/commons/util/NoPrimaryKeyVirtualField.java
new file mode 100644
index 0000000000..7ea15dec1e
--- /dev/null
+++ b/manager/tm-common/src/main/java/com/tapdata/tm/commons/util/NoPrimaryKeyVirtualField.java
@@ -0,0 +1,374 @@
+package com.tapdata.tm.commons.util;
+
+import com.tapdata.tm.commons.dag.Element;
+import com.tapdata.tm.commons.dag.Node;
+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.JoinProcessorNode;
+import com.tapdata.tm.commons.dag.process.MergeTableNode;
+import com.tapdata.tm.commons.exception.NoPrimaryKeyException;
+import com.tapdata.tm.commons.schema.Field;
+import com.tapdata.tm.commons.schema.MetadataInstancesDto;
+import com.tapdata.tm.commons.schema.Schema;
+import io.github.openlg.graphlib.Graph;
+import io.tapdata.entity.event.dml.TapDeleteRecordEvent;
+import io.tapdata.entity.event.dml.TapInsertRecordEvent;
+import io.tapdata.entity.event.dml.TapRecordEvent;
+import io.tapdata.entity.event.dml.TapUpdateRecordEvent;
+import io.tapdata.entity.schema.TapField;
+import io.tapdata.entity.schema.TapIndex;
+import io.tapdata.entity.schema.TapTable;
+import io.tapdata.entity.schema.type.TapString;
+import io.tapdata.entity.schema.type.TapType;
+import io.tapdata.entity.utils.InstanceFactory;
+import io.tapdata.entity.utils.JsonParser;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.*;
+import java.util.function.Consumer;
+import java.util.function.Predicate;
+
+/**
+ * 无主键虚拟列
+ *
+ * <ul>
+ *     <h3>已知问题：</h3>
+ *     <li>虚拟列名为固定值，可能会与用户数据模型产生冲突</li>
+ *     <li>产品现阶段，还无法在目标模型获取到来自哪个节点的哪个表，源节点会对无主键表统一添加 hash 列</li>
+ *     <li>存在多条字段及值完全一致的数据，删除和更新会将目标的所有数据进行变更，目标可能存在多余或丢失数据情况</li>
+ *     <li>不支持字段 DDL</li>
+ * </ul>
+ * <ul>
+ *     <h3>关键需求：</h3>
+ *     <li>在源节点生成 hash 列数据，可以避免处理节点修改数据的场景下，对同个数据多次修改导致 before 中 hash 列错误问题</li>
+ *     <li>在目标节点补充 hash 列模型，是因为需要根据目标节点配置来判断是否添加</li>
+ * </ul>
+ *
+ * @author <a href="mailto:harsen_lin@163.com">Harsen</a>
+ * @version v1.0 2024/10/15 17:19 Create
+ */
+public class NoPrimaryKeyVirtualField {
+
+    public static final String HASH_ALGORITHM = "MD5";
+    public static final String FIELD_NAME = "_no_pk_hash";
+    public static final int FIELD_LENGTH = 32;
+    public static final TapType FIELD_TAP_TYPE = new TapString((long) FIELD_LENGTH, false);
+    public static final byte SPLIT_CHAR = ',';
+
+    protected Consumer<TapTable> addTable;
+    protected Predicate<TapRecordEvent> addHashValue;
+    protected final Map<String, HashValueAppender> noPkTables = new HashMap<>();
+
+    public void init(Graph<? extends Element, ? extends Element> graph) {
+        if (isEnable(graph)) {
+            this.noPkTables.clear();
+            this.addTable = table -> {
+                // 过滤：有主键、有唯一索引的表
+                if (Optional.ofNullable(table.primaryKeys()).map(keys -> !keys.isEmpty()).orElse(false)) return;
+                if (Optional.ofNullable(table.getIndexList()).map(list -> {
+                    for (TapIndex index : list) {
+                        if (Boolean.TRUE.equals(index.getPrimary())) return true;
+                        if (Boolean.TRUE.equals(index.isUnique())) return true;
+                    }
+                    return false;
+                }).orElse(false)) return;
+
+                // 添加 hash 列到数据上
+                String tableName = table.getName();
+                List<String> keys = new ArrayList<>(table.getNameFieldMap().keySet());
+                noPkTables.put(tableName, new HashValueAppender(tableName, keys));
+            };
+            this.addHashValue = event -> {
+                HashValueAppender handle = noPkTables.get(event.getTableId());
+                if (null != handle) {
+                    return handle.apply(event);
+                }
+                return true;
+            };
+        } else {
+            this.noPkTables.clear();
+            this.addTable = table -> {
+            };
+            this.addHashValue = table -> true;
+        }
+    }
+
+    public void add(TapTable table) {
+        this.addTable.accept(table);
+    }
+
+    public boolean addHashValue(TapRecordEvent event) {
+        return this.addHashValue.test(event);
+    }
+
+    // ---------- 以下为静态函数 ----------
+
+    public static void addVirtualField(Object schemaObj, Node<?> node) {
+        if (isEnable(node.getGraph())) {
+            // 没有前置节点，跳过
+            if (node.predecessors().isEmpty()) return;
+            checkAndAddVirtualField(schemaObj, node);
+        }
+    }
+
+    public static Collection<String> getVirtualHashFieldNames(TapTable table) {
+        Collection<String> virtualHashFields = new HashSet<>();
+        Optional.ofNullable(table.childItems()).ifPresent(fields -> {
+            for (TapField field : fields) {
+                if (Field.SOURCE_VIRTUAL_HASH.equals(field.getCreateSource())) {
+                    virtualHashFields.add(field.getName());
+                }
+            }
+        });
+        return virtualHashFields;
+    }
+
+    public static List<String> getVirtualHashFieldNames(MetadataInstancesDto metadata) {
+        List<String> virtualHashFields = new ArrayList<>();
+        Optional.ofNullable(metadata.getFields()).ifPresent(fields -> {
+            for (Field field : fields) {
+                if (Field.SOURCE_VIRTUAL_HASH.equals(field.getCreateSource())) {
+                    virtualHashFields.add(field.getFieldName());
+                }
+            }
+        });
+        return virtualHashFields;
+    }
+
+    public static boolean isEnable(Graph<? extends Element, ? extends Element> graph) {
+        for (String id : graph.getNodes()) {
+            // 存在表合并，不启用
+            Element graphNode = graph.getNode(id);
+            if (graphNode instanceof MergeTableNode) return false;
+            // 存在表关联，不启用
+            if (graphNode instanceof JoinProcessorNode) return false;
+            // 过滤：追加写入模式
+            if (graphNode instanceof DataParentNode) {
+                // 没有前置节点，跳过
+                DataParentNode<?> dataParentNode = (DataParentNode<?>) graphNode;
+                if (!dataParentNode.predecessors().isEmpty() && NoPrimaryKeySyncMode.ADD_HASH == dataParentNode.getNoPrimaryKeySyncMode()) {
+                    return true;
+                }
+            }
+        }
+        return false;
+    }
+
+    // ---------- 以下为内部函数 ----------
+
+    protected static void checkAndAddVirtualField(Object schemaObj, Node<?> node) {
+        if (schemaObj instanceof Schema) {
+            Schema schema = (Schema) schemaObj;
+            if (isNeed2AddField(schema, node)) {
+                addVirtualField2Schema(schema, node);
+            }
+        } else if (schemaObj instanceof List) {
+            for (Object schema : (List<?>) schemaObj) {
+                checkAndAddVirtualField(schema, node);
+            }
+        }
+    }
+
+    protected static boolean isNeed2AddField(Schema schema, Node<?> node) {
+        // 过滤：主键，唯一索引
+        if (hasPrimaryOrUniqueOrKeys(schema)) return false;
+
+        // 过滤：指定关联字段
+        if (node instanceof TableNode) {
+            TableNode tableNode = (TableNode) node;
+            List<String> conditionFields = tableNode.getUpdateConditionFields();
+            return hasVirtualField(conditionFields);
+        } else if (node instanceof DatabaseNode) {
+            DatabaseNode databaseNode = (DatabaseNode) node;
+            Map<String, List<String>> conditionFieldMap = databaseNode.getUpdateConditionFieldMap();
+            List<String> conditionFields = null == conditionFieldMap ? null : conditionFieldMap.get(schema.getName());
+            return hasVirtualField(conditionFields);
+        }
+        return false;
+    }
+
+    protected static boolean hasVirtualField(List<String> conditionFields) {
+        if (null != conditionFields && !conditionFields.isEmpty()) {
+            for (String field : conditionFields) {
+                if (FIELD_NAME.equalsIgnoreCase(field)) return true;
+            }
+            return false;
+        }
+        return true;
+    }
+
+    protected static void addVirtualField2Schema(Schema schema, Node<?> node) {
+        List<Field> fields = schema.getFields();
+        // 字段已存在，不添加
+        for (Field field : fields) {
+            if (FIELD_NAME.equalsIgnoreCase(field.getFieldName())) return;
+        }
+        // 字段不存在，添加
+        Field field = createVirtualField(schema, node);
+        fields.add(field);
+    }
+
+    protected static Field createVirtualField(Schema schema, Node<?> node) {
+        Field field = new Field();
+        field.setTableName(schema.getName());
+        field.setFieldName(FIELD_NAME);
+        field.setOriginalFieldName(FIELD_NAME);
+        field.setSource(Field.SOURCE_VIRTUAL_HASH);
+        field.setCreateSource(Field.SOURCE_VIRTUAL_HASH);
+        field.setColumnSize(FIELD_LENGTH);
+        field.setOriPrecision(FIELD_LENGTH);
+        field.setJavaType("String");
+        field.setJavaType1("String");
+        field.setTapType(InstanceFactory.instance(JsonParser.class).toJson(FIELD_TAP_TYPE));
+        field.setColumnPosition(schema.getFields().size() + 1);
+        field.setPrimaryKey(false);
+        field.setId(MetaDataBuilderUtils.generateFieldId(node.getId(), field.getTableName(), field.getFieldName()));
+        return field;
+    }
+
+    protected static boolean hasPrimaryOrUniqueOrKeys(Schema schema) {
+        if (schema.isHasPrimaryKey() || schema.isHasUnionIndex()) return true;
+
+        List<Field> fields = schema.getFields();
+        if (null != fields) {
+            for (Field field : fields) {
+                if (Boolean.TRUE.equals(field.getPrimaryKey())) return true;
+                if (field.isUnique()) return true;
+            }
+        }
+        return false;
+    }
+
+    protected static class HashValueAppender {
+
+        final String table;
+        final List<String> keys;
+        private Predicate<TapInsertRecordEvent> insertRecordEventPredicate;
+        private Predicate<TapUpdateRecordEvent> updateRecordEventPredicate;
+        private Predicate<TapDeleteRecordEvent> deleteRecordEventPredicate;
+
+        HashValueAppender(String table, List<String> keys) {
+            this.table = table;
+            this.keys = keys;
+            keys.removeIf(key -> key.equalsIgnoreCase(FIELD_NAME));
+            insertRecordEventPredicate = event -> {
+                Optional.ofNullable(event.getAfter())
+                    .ifPresent(data -> data.put(FIELD_NAME, toHash(keys, data, false)));
+                return true;
+            };
+            updateRecordEventPredicate = event -> {
+                Map<String, Object> data = event.getBefore();
+                data.put(FIELD_NAME, toHash(keys, data, true));
+                data = event.getAfter();
+                data.put(FIELD_NAME, toHash(keys, data, true));
+                return true;
+            };
+            deleteRecordEventPredicate = event -> {
+                Map<String, Object> data = event.getBefore();
+                data.put(FIELD_NAME, toHash(keys, data, true));
+                return true;
+            };
+        }
+
+        boolean apply(TapRecordEvent event) {
+            if (event instanceof TapInsertRecordEvent) {
+                return insertRecordEventPredicate.test((TapInsertRecordEvent) event);
+            } else if (event instanceof TapUpdateRecordEvent) {
+                return updateRecordEventPredicate.test((TapUpdateRecordEvent) event);
+            } else if (event instanceof TapDeleteRecordEvent) {
+                return deleteRecordEventPredicate.test((TapDeleteRecordEvent) event);
+            }
+            return true;
+        }
+
+        protected String toHash(List<String> keys, Map<String, Object> data, boolean isStrictMode) {
+            try {
+                MessageDigest md = MessageDigest.getInstance(HASH_ALGORITHM);
+                try (ByteArrayOutputStream baos = new ByteArrayOutputStream()) {
+                    boolean isFirst = true;
+                    baos.write('[');
+                    for (String key : keys) {
+                        if (isStrictMode && !data.containsKey(key)) {
+                            throw NoPrimaryKeyException.incompleteFields(key);
+                        }
+                        if (isFirst) {
+                            isFirst = false;
+                        } else {
+                            baos.write(SPLIT_CHAR);
+                        }
+
+                        Object val = data.get(key);
+                        byte[] bytes = toBytes(val);
+                        baos.write(bytes);
+                    }
+                    baos.write(']');
+
+                    byte[] hashBytes = md.digest(baos.toByteArray());
+                    StringBuilder hashHex = new StringBuilder();
+                    for (byte b : hashBytes) {
+                        hashHex.append(String.format("%02x", b));
+                    }
+                    return hashHex.toString(); // 返回 128 位（32 个字符）的哈希值
+                }
+            } catch (NoPrimaryKeyException e) {
+                updateRecordEventPredicate = event -> false;
+                deleteRecordEventPredicate = event -> false;
+                throw e;
+            } catch (NoSuchAlgorithmException e) {
+                updateRecordEventPredicate = event -> false;
+                deleteRecordEventPredicate = event -> false;
+                insertRecordEventPredicate = event -> false;
+                throw NoPrimaryKeyException.notfoundHashAlgorithm(HASH_ALGORITHM, e);
+            } catch (Exception e) {
+                updateRecordEventPredicate = event -> false;
+                deleteRecordEventPredicate = event -> false;
+                insertRecordEventPredicate = event -> false;
+                throw NoPrimaryKeyException.otherFailed(e);
+            }
+        }
+
+        protected byte[] toBytes(Object data) throws IOException {
+            if (null == data) return new byte[0];
+            if (data instanceof byte[]) return (byte[]) data;
+            if (data.getClass().isArray()) return arrayToBytes(Arrays.asList((Object[]) data));
+            if (data instanceof Collection) return arrayToBytes((Collection<?>) data);
+            if (data instanceof Map) return mapToBytes((Map<?, ?>) data);
+            return data.toString().getBytes();
+        }
+
+        protected byte[] arrayToBytes(Collection<?> collection) throws IOException {
+            try (ByteArrayOutputStream baos = new ByteArrayOutputStream()) {
+                boolean isFirst = true;
+                baos.write('[');
+                for (Object o : collection) {
+                    if (isFirst) {
+                        isFirst = false;
+                    } else {
+                        baos.write(SPLIT_CHAR);
+                    }
+                    baos.write(toBytes(o));
+                }
+                baos.write(']');
+                return baos.toByteArray();
+            }
+        }
+
+        protected byte[] mapToBytes(Map<?, ?> map) throws IOException {
+            try (ByteArrayOutputStream baos = new ByteArrayOutputStream()) {
+                baos.write('{');
+                for (Map.Entry<?, ?> en : map.entrySet()) {
+                    baos.write(toBytes(en.getKey()));
+                    baos.write(':');
+                    baos.write(toBytes(en.getValue()));
+                }
+                baos.write('}');
+                return baos.toByteArray();
+            }
+        }
+
+    }
+}
diff --git a/manager/tm-oss/src/main/java/com/tapdata/tm/alarm/service/impl/AlarmServiceImpl.java b/manager/tm-oss/src/main/java/com/tapdata/tm/alarm/service/impl/AlarmServiceImpl.java
index 0d00987780..07aaba1c21 100644
--- a/manager/tm-oss/src/main/java/com/tapdata/tm/alarm/service/impl/AlarmServiceImpl.java
+++ b/manager/tm-oss/src/main/java/com/tapdata/tm/alarm/service/impl/AlarmServiceImpl.java
@@ -146,6 +146,11 @@ public List<AlarmChannelDto> getAvailableChannels() {
         return new ArrayList<>();
     }
 
+    @Override
+    public boolean checkOpen(TaskDto taskDto, String nodeId, AlarmKeyEnum key, NotifyEnum type, List<AlarmSettingDto> settingDtos) {
+        throw new BizException(ERROR);
+    }
+
     @Override
     public boolean checkOpen(TaskDto taskDto, String nodeId, AlarmKeyEnum key, NotifyEnum type, UserDetail userDetail) {
         throw new BizException(ERROR);
diff --git a/manager/tm/src/main/java/com/tapdata/tm/Settings/service/SettingsServiceImpl.java b/manager/tm/src/main/java/com/tapdata/tm/Settings/service/SettingsServiceImpl.java
index 22cc5d7c56..a36d168aff 100644
--- a/manager/tm/src/main/java/com/tapdata/tm/Settings/service/SettingsServiceImpl.java
+++ b/manager/tm/src/main/java/com/tapdata/tm/Settings/service/SettingsServiceImpl.java
@@ -207,7 +207,7 @@ public List<SettingsDto> findALl(String decode, Filter filter) {
             //}).collect(Collectors.toList());
         } else {
             settingsList.stream().filter(settings -> {
-                if ("smtp.server.password".equals(settings.getKey()) || "ad.bind.password".equals(settings.getKey()))
+                if ("smtp.server.password".equals(settings.getKey()) || "ldap.bind.password".equals(settings.getKey()))
                     settings.setValue("*****");
                 return true;
             }).collect(Collectors.toList());
diff --git a/manager/tm/src/main/java/com/tapdata/tm/cluster/controller/ClusterStateController.java b/manager/tm/src/main/java/com/tapdata/tm/cluster/controller/ClusterStateController.java
index c07d1465b9..e7935cca1f 100644
--- a/manager/tm/src/main/java/com/tapdata/tm/cluster/controller/ClusterStateController.java
+++ b/manager/tm/src/main/java/com/tapdata/tm/cluster/controller/ClusterStateController.java
@@ -5,6 +5,7 @@
 import com.tapdata.tm.cluster.dto.*;
 import com.tapdata.tm.cluster.service.ClusterStateService;
 import com.tapdata.tm.utils.MongoUtils;
+import com.tapdata.tm.worker.service.WorkerService;
 import io.swagger.v3.oas.annotations.Operation;
 import io.swagger.v3.oas.annotations.Parameter;
 import io.swagger.v3.oas.annotations.enums.ParameterIn;
@@ -157,7 +158,7 @@ public ResponseMessage<ClusterStateDto> replaceById2(@PathVariable("id") String
     @Operation(summary = "Delete a model instance by {{id}} from the data source")
     @DeleteMapping("{id}")
     public ResponseMessage<Boolean> delete(@PathVariable("id") String id) {
-        return success(clusterStateService.deleteById(MongoUtils.toObjectId(id)));
+        return success(clusterStateService.deleteCluster(MongoUtils.toObjectId(id), getLoginUser()));
     }
 
     /**
diff --git a/manager/tm/src/main/java/com/tapdata/tm/cluster/service/ClusterStateService.java b/manager/tm/src/main/java/com/tapdata/tm/cluster/service/ClusterStateService.java
index f870194746..400efd1e14 100644
--- a/manager/tm/src/main/java/com/tapdata/tm/cluster/service/ClusterStateService.java
+++ b/manager/tm/src/main/java/com/tapdata/tm/cluster/service/ClusterStateService.java
@@ -11,6 +11,7 @@
 import com.tapdata.tm.Settings.constant.SettingsEnum;
 import com.tapdata.tm.Settings.service.SettingsService;
 import com.tapdata.tm.agent.service.AgentGroupService;
+import com.tapdata.tm.base.dto.Field;
 import com.tapdata.tm.base.dto.Filter;
 import com.tapdata.tm.base.dto.Page;
 import com.tapdata.tm.base.exception.BizException;
@@ -27,6 +28,7 @@
 import com.tapdata.tm.config.security.UserDetail;
 import com.tapdata.tm.message.dto.MessageDto;
 import com.tapdata.tm.message.service.MessageService;
+import com.tapdata.tm.worker.dto.WorkerDto;
 import com.tapdata.tm.worker.entity.Worker;
 import com.tapdata.tm.worker.service.WorkerService;
 import lombok.NonNull;
@@ -421,4 +423,25 @@ public Page<ClusterStateDto> getAll(Filter filter) {
         });
         return page;
     }
+
+    public boolean deleteCluster(ObjectId id, UserDetail user) {
+        Field field = new Field();
+        field.put("systemInfo", 1);
+        ClusterStateDto clusterStateDto = findById(id, field);
+        boolean unBind = true;
+        if (null != clusterStateDto && null != clusterStateDto.getSystemInfo()) {
+            String processId = clusterStateDto.getSystemInfo().getProcess_id();
+            if (null == processId) {
+                unBind = false;
+            }
+            WorkerDto worker = workerService.findByProcessId(processId, user, "licenseBind");
+            if (null != worker && worker.getLicenseBind()) {
+                unBind = workerService.unbindByProcessId(processId);
+            }
+        }
+        if (unBind) {
+            return deleteById(id);
+        }
+        return false;
+    }
 }
diff --git a/manager/tm/src/main/java/com/tapdata/tm/metadatadefinition/service/MetadataDefinitionService.java b/manager/tm/src/main/java/com/tapdata/tm/metadatadefinition/service/MetadataDefinitionService.java
index 102ab86daa..f747dd9c1f 100644
--- a/manager/tm/src/main/java/com/tapdata/tm/metadatadefinition/service/MetadataDefinitionService.java
+++ b/manager/tm/src/main/java/com/tapdata/tm/metadatadefinition/service/MetadataDefinitionService.java
@@ -7,6 +7,7 @@
 import com.tapdata.tm.base.dto.Field;
 import com.tapdata.tm.base.dto.Filter;
 import com.tapdata.tm.base.dto.Page;
+import com.tapdata.tm.base.dto.Where;
 import com.tapdata.tm.base.exception.BizException;
 import com.tapdata.tm.base.service.BaseService;
 import com.tapdata.tm.commons.base.dto.BaseDto;
@@ -38,13 +39,7 @@
 import org.springframework.data.mongodb.core.query.Update;
 import org.springframework.stereotype.Service;
 
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Comparator;
-import java.util.List;
-import java.util.Map;
-import java.util.Objects;
-import java.util.Set;
+import java.util.*;
 import java.util.stream.Collectors;
 
 /**
@@ -56,6 +51,7 @@
 @Slf4j
 public class MetadataDefinitionService extends BaseService<MetadataDefinitionDto, MetadataDefinitionEntity, ObjectId, MetadataDefinitionRepository> {
 
+    public static final String ITEM_TYPE = "item_type";
     @Autowired
     MongoTemplate mongoTemplate;
 
@@ -320,7 +316,16 @@ public Page<MetadataDefinitionDto> findAndChildAccount(Filter filter, boolean se
 
     @Override
     public Page<MetadataDefinitionDto> find(Filter filter, UserDetail user) {
-        Page<MetadataDefinitionDto> dtoPage = settingsService.isCloud()? super.find(filter, user) : super.find(filter);
+        Page<MetadataDefinitionDto> dtoPage;
+        Where where = new Where();
+        Map<String, String> itemMap = new HashMap<>();
+        itemMap.put(ITEM_TYPE, "dataflow");
+        where.put("or", com.tapdata.tm.utils.Lists.of(itemMap));
+        if (null != filter && where.equals(filter.getWhere())) {
+            dtoPage = settingsService.isCloud() ? super.find(filter, user) : super.find(filter);
+        } else {
+            dtoPage = super.find(filter, user);
+        }
         if (filter.getOrder() == null) {
             dtoPage.getItems().sort(Comparator.comparing(MetadataDefinitionDto::getValue));
             dtoPage.getItems().sort(Comparator.comparing(s -> {
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 7bca3e289e..b198ae58de 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
@@ -1031,7 +1031,9 @@ public Page<TableSyncStaticVo> querySyncStatic(TableSyncStaticDto dto, UserDetai
 
             Number snapshotSyncRate = vs.get("snapshotSyncRate");
             BigDecimal syncRate;
-            if (Objects.nonNull(snapshotSyncRate)) {
+            if (snapshotRowTotal.get() == -1L && snapshotInsertRowTotal.get() > 0L) {
+                syncRate = new BigDecimal(-1);
+            } else if (Objects.nonNull(snapshotSyncRate)) {
                 syncRate = BigDecimal.valueOf(snapshotSyncRate.doubleValue());
             } else if (snapshotRowTotal.get() != 0L) {
                 syncRate = new BigDecimal(snapshotInsertRowTotal.get()).divide(new BigDecimal(snapshotRowTotal.get()), 2, RoundingMode.HALF_UP);
@@ -1045,6 +1047,8 @@ public Page<TableSyncStaticVo> querySyncStatic(TableSyncStaticDto dto, UserDetai
                 syncRate = BigDecimal.ONE;
             } else if (syncRate.compareTo(BigDecimal.ZERO) == 0) {
                 fullSyncStatus = "NOT_START";
+            } else if (syncRate.intValue() == -1) {
+                fullSyncStatus = "COUNTING";
             } else {
                 fullSyncStatus = "ING";
             }
diff --git a/manager/tm/src/main/java/com/tapdata/tm/task/service/impl/TaskRecordServiceImpl.java b/manager/tm/src/main/java/com/tapdata/tm/task/service/impl/TaskRecordServiceImpl.java
index 23b26bd949..e1bb3669db 100644
--- a/manager/tm/src/main/java/com/tapdata/tm/task/service/impl/TaskRecordServiceImpl.java
+++ b/manager/tm/src/main/java/com/tapdata/tm/task/service/impl/TaskRecordServiceImpl.java
@@ -40,6 +40,7 @@ public class TaskRecordServiceImpl implements TaskRecordService {
     private MongoTemplate mongoTemplate;
     private MeasurementServiceV2 measurementServiceV2;
     private UserService userService;
+    private static final String TASK_ID = "taskId";
 
     @Override
     public void createRecord(TaskRecord taskRecord) {
@@ -64,7 +65,7 @@ public Page<TaskRecordListVo> queryRecords(TaskRecordDto dto) {
         Integer page = dto.getPage();
         Integer size = dto.getSize();
 
-        Query query = new Query(Criteria.where("taskId").is(taskId));
+        Query query = new Query(Criteria.where(TASK_ID).is(taskId));
         long count = mongoTemplate.count(query, TaskRecord.class);
         if (count == 0) {
             return new Page<>(0, Collections.emptyList());
@@ -77,6 +78,7 @@ public Page<TaskRecordListVo> queryRecords(TaskRecordDto dto) {
 
         query.with(pageable);
         query.with(Sort.by("createTime").descending());
+        query.fields().include("id",TASK_ID,"user_id","taskSnapshot.status", "inputTotal", "outputTotal", "statusStack");
 
         List<TaskRecord> taskRecords = mongoTemplate.find(query, TaskRecord.class);
 
@@ -174,7 +176,7 @@ public TaskDto queryTask(String taskRecordId, String userId) {
 
     @Override
     public void cleanTaskRecord(String taskId) {
-        Query query = new Query(Criteria.where("taskId").is(taskId));
+        Query query = new Query(Criteria.where(TASK_ID).is(taskId));
         mongoTemplate.remove(query, TaskRecord.class);
     }
 }
diff --git a/manager/tm/src/main/java/com/tapdata/tm/task/service/impl/TaskSaveServiceImpl.java b/manager/tm/src/main/java/com/tapdata/tm/task/service/impl/TaskSaveServiceImpl.java
index cfa886575d..c5359137ab 100644
--- a/manager/tm/src/main/java/com/tapdata/tm/task/service/impl/TaskSaveServiceImpl.java
+++ b/manager/tm/src/main/java/com/tapdata/tm/task/service/impl/TaskSaveServiceImpl.java
@@ -19,6 +19,7 @@
 import com.tapdata.tm.commons.task.dto.alarm.AlarmRuleVO;
 import com.tapdata.tm.commons.task.dto.alarm.AlarmSettingDto;
 import com.tapdata.tm.commons.task.dto.alarm.AlarmSettingVO;
+import com.tapdata.tm.commons.util.NoPrimaryKeyVirtualField;
 import com.tapdata.tm.config.security.UserDetail;
 import com.tapdata.tm.metadatainstance.service.MetadataInstancesService;
 import com.tapdata.tm.task.service.TaskSaveService;
@@ -149,6 +150,11 @@ public void syncTaskSetting(TaskDto taskDto, UserDetail userDetail) {
                                                     .collect(Collectors.toList());
                                             if (CollectionUtils.isNotEmpty(columnList)) {
                                                 databaseNode.getUpdateConditionFieldMap().put(schema.getName(), columnList);
+                                            } else {
+                                                columnList = NoPrimaryKeyVirtualField.getVirtualHashFieldNames(schema);
+                                                if (CollectionUtils.isNotEmpty(columnList)) {
+                                                    databaseNode.getUpdateConditionFieldMap().put(schema.getName(), columnList);
+                                                }
                                             }
                                         }
                                     });


Test patch:
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 d9cdd58853..7f7f062d2d 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
@@ -1,5 +1,6 @@
 package io.tapdata.flow.engine.V2.filter;
 
+import com.tapdata.tm.commons.util.NoPrimaryKeyVirtualField;
 import org.junit.Assert;
 import org.junit.Test;
 
@@ -8,7 +9,7 @@
 import java.util.Map;
 import java.util.Set;
 
-import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.*;
 
 public class FilterUtilTest {
 
@@ -149,4 +150,19 @@ public void testProcessTableFields2() {
         assertEquals("张三", finalData.get("name"));
         assertEquals(18, finalData.get("age"));
     }
+
+    @Test
+    public void testProcessTableFieldsOfNoPrimaryKeyHash() {
+        String expectedHash = "hash";
+        String expectedName = "张三";
+        Map<String, Object> data = new HashMap<>();
+        data.put(NoPrimaryKeyVirtualField.FIELD_NAME, expectedHash);
+        data.put("name", expectedName);
+        Set<String> fieldNames = new HashSet<>();
+        fieldNames.add("name");
+        Map<String, Object> finalData = FilterUtil.processTableFields(data, fieldNames);
+        assertNotNull(finalData);
+        assertEquals(expectedHash, finalData.get(NoPrimaryKeyVirtualField.FIELD_NAME));
+        assertEquals(expectedName, finalData.get("name"));
+    }
 }
diff --git a/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/HazelcastBaseNodeTest.java b/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/HazelcastBaseNodeTest.java
index 85e4f10d73..9a84757590 100644
--- a/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/HazelcastBaseNodeTest.java
+++ b/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/HazelcastBaseNodeTest.java
@@ -301,6 +301,7 @@ void testBuildLogListener(){
 			ObsLogger obsLogger = mock(ObsLogger.class);
 			ReflectionTestUtils.setField(mockHazelcastBaseNode,"obsLogger",obsLogger);
 			doCallRealMethod().when(mockHazelcastBaseNode).buildLogListener();
+			doCallRealMethod().when(mockHazelcastBaseNode).logListener();
 			TapTableMap tapTableMap = TapTableMap.create("111");
 			when(processorBaseContext.getTapTableMap()).thenReturn(tapTableMap);
 			mockHazelcastBaseNode.buildLogListener();
diff --git a/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastPdkBaseNodeTest.java b/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastPdkBaseNodeTest.java
index 383589cca5..4b03ce31c7 100644
--- a/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastPdkBaseNodeTest.java
+++ b/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastPdkBaseNodeTest.java
@@ -479,9 +479,9 @@ void testTableNode() {
 			TableNode node2 = new TableNode();
 			node2.setNodeConfig(nodeConfig);
 			taskDto.setDoubleActive(true);
-			Assertions.assertEquals(2, hazelcastPdkBaseNode.generateNodeConfig(node1, taskDto).size());
+			Assertions.assertEquals(3, hazelcastPdkBaseNode.generateNodeConfig(node1, taskDto).size());
 			Assertions.assertTrue(hazelcastPdkBaseNode.generateNodeConfig(node1, taskDto).containsKey("doubleActive"));
-			Assertions.assertEquals(3, hazelcastPdkBaseNode.generateNodeConfig(node2, taskDto).size());
+			Assertions.assertEquals(4, hazelcastPdkBaseNode.generateNodeConfig(node2, taskDto).size());
 		}
 
 		@Test
@@ -493,9 +493,9 @@ void testDatabaseNode() {
 			DatabaseNode node2 = new DatabaseNode();
 			node2.setNodeConfig(nodeConfig);
 			taskDto.setDoubleActive(true);
-			Assertions.assertEquals(2, hazelcastPdkBaseNode.generateNodeConfig(node1, taskDto).size());
+			Assertions.assertEquals(3, hazelcastPdkBaseNode.generateNodeConfig(node1, taskDto).size());
 			Assertions.assertTrue(hazelcastPdkBaseNode.generateNodeConfig(node1, taskDto).containsKey("doubleActive"));
-			Assertions.assertEquals(3, hazelcastPdkBaseNode.generateNodeConfig(node2, taskDto).size());
+			Assertions.assertEquals(4, hazelcastPdkBaseNode.generateNodeConfig(node2, taskDto).size());
 		}
 
 		@Test
diff --git a/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastSourcePdkBaseNodeTest.java b/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastSourcePdkBaseNodeTest.java
index b359723042..27df71bace 100644
--- a/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastSourcePdkBaseNodeTest.java
+++ b/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastSourcePdkBaseNodeTest.java
@@ -22,10 +22,13 @@
 import com.tapdata.tm.commons.schema.TransformerWsMessageDto;
 import com.tapdata.tm.commons.task.dto.TaskDto;
 import com.tapdata.tm.commons.util.ConnHeartbeatUtils;
+import com.tapdata.tm.commons.util.NoPrimaryKeyVirtualField;
 import io.tapdata.aspect.StreamReadFuncAspect;
 import io.tapdata.aspect.TableCountFuncAspect;
+import io.tapdata.aspect.utils.AspectUtils;
 import io.tapdata.common.concurrent.SimpleConcurrentProcessorImpl;
 import io.tapdata.common.concurrent.TapExecutors;
+import io.tapdata.entity.aspect.Aspect;
 import io.tapdata.entity.aspect.AspectManager;
 import io.tapdata.entity.aspect.AspectObserver;
 import io.tapdata.entity.event.TapEvent;
@@ -61,6 +64,7 @@
 import io.tapdata.schema.TapTableMap;
 import io.tapdata.supervisor.TaskResourceSupervisorManager;
 import lombok.SneakyThrows;
+import org.apache.logging.log4j.Logger;
 import org.bson.types.ObjectId;
 import org.junit.Assert;
 import org.junit.jupiter.api.*;
@@ -68,14 +72,8 @@
 import org.mockito.Mockito;
 import org.springframework.test.util.ReflectionTestUtils;
 
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.ScheduledExecutorService;
+import java.util.*;
+import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.BiConsumer;
@@ -119,6 +117,7 @@ class DoInitTest {
 		void beforeEach() {
 			TaskResourceSupervisorManager taskResourceSupervisorManager = new TaskResourceSupervisorManager();
 			ReflectionTestUtils.setField(mockInstance, "taskResourceSupervisorManager", taskResourceSupervisorManager);
+			ReflectionTestUtils.setField(mockInstance, "noPrimaryKeyVirtualField", new NoPrimaryKeyVirtualField());
 			doCallRealMethod().when(mockInstance).doInit(jetContext);
 		}
 
@@ -1245,7 +1244,7 @@ void test1() {
 			tapDDLUnknownEvent.setOriginDDL("alter table xxx add new_field number(8,0)");
 			TapdataEvent tapdataEvent = instance.wrapSingleTapdataEvent(tapDDLUnknownEvent, SyncStage.CDC, null, true);
 			assertNull(tapdataEvent);
-			verify(mockObsLogger).warn(any());
+			verify(mockObsLogger, atLeastOnce()).warn(any());
 		}
 	}
 
@@ -1413,4 +1412,141 @@ void test1() {
 			assertFalse(thread.isAlive());
 		}
 	}
+
+
+	@Nested
+	class doTableNameSynchronouslyTest{
+		private BatchCountFunction batchCountFunction;
+		private List<String> tableList;
+		@BeforeEach
+		void setUp() {
+			tableList = Arrays.asList("table1", "table2");
+			TapTableMap map = TapTableMap.create("nodeId");
+			map.putNew("table1", new TapTable("table1"), "qualifiedName1");
+			map.putNew("table2", new TapTable("table2"), "qualifiedName2");
+			when(dataProcessorContext.getTapTableMap()).thenReturn(map);
+			batchCountFunction = mock(BatchCountFunction.class);
+		}
+
+		@Test
+		void testDoTableNameSynchronously_IsRunningFalse() {
+			when(mockInstance.isRunning()).thenReturn(false);
+			doCallRealMethod().when(mockInstance).doTableNameSynchronously(batchCountFunction, tableList);
+			mockInstance.doTableNameSynchronously(batchCountFunction, tableList);
+			verify(mockInstance, never()).executeDataFuncAspect(any(), any(), any());
+		}
+
+		@Test
+		void testDoTableNameSynchronously_NormalExecution() {
+			when(mockInstance.isRunning()).thenReturn(true);
+			try (MockedStatic<PDKInvocationMonitor> mb = Mockito
+					.mockStatic(PDKInvocationMonitor.class)) {
+				mb.when(()->PDKInvocationMonitor.invoke(any(),any(),any())).thenAnswer(invocationOnMock -> {
+					return null;
+				});
+				doCallRealMethod().when(mockInstance).doTableNameSynchronously(batchCountFunction, tableList);
+				mockInstance.doTableNameSynchronously(batchCountFunction, tableList);
+				verify(mockInstance, times(1)).asyncCountTable(batchCountFunction, tableList);
+			}
+		}
+	}
+
+	@Nested
+	class asyncCountTableTest{
+		private BatchCountFunction batchCountFunction;
+		protected ObsLogger obsLogger;
+		private List<String> tableList;
+		private TaskDto taskDto;
+		private Node node;
+		private ExecutorService snapshotRowSizeThreadPool;
+		private Logger logger = mock(Logger.class);
+		@BeforeEach
+		void setUp() {
+			batchCountFunction = mock(BatchCountFunction.class);
+			tableList = Arrays.asList("table1", "table2");
+
+			taskDto = mock(TaskDto.class);
+			node = mock(Node.class);
+
+			when(dataProcessorContext.getTaskDto()).thenReturn(taskDto);
+			when(dataProcessorContext.getNode()).thenReturn(node);
+			when(taskDto.getName()).thenReturn("TestTask");
+			when(taskDto.getId()).thenReturn(new ObjectId("507f1f77bcf86cd799439011"));
+			when(node.getName()).thenReturn("TestNode");
+			when(node.getId()).thenReturn("12345");
+			snapshotRowSizeThreadPool = mock(ExecutorService.class);
+			ReflectionTestUtils.setField(mockInstance, "snapshotRowSizeThreadPool", snapshotRowSizeThreadPool);
+			obsLogger = mock(ObsLogger.class);
+			ReflectionTestUtils.setField(mockInstance, "obsLogger", obsLogger);
+			logger = mock(Logger.class);
+			ReflectionTestUtils.setField(mockInstance, "logger", logger);
+			doCallRealMethod().when(mockInstance).asyncCountTable(batchCountFunction, tableList);
+
+		}
+
+		@Test
+		void testAsyncCountTable_Success() throws InterruptedException {
+			doNothing().when(mockInstance).doCountSynchronously(batchCountFunction, tableList, false);
+			doNothing().when(snapshotRowSizeThreadPool).shutdown();
+
+			mockInstance.asyncCountTable(batchCountFunction, tableList);
+
+			verify(logger).info(contains("Start to asynchronously count the size of rows for the source table(s)"));
+
+			Thread.sleep(100); // Wait for async execution
+
+			verify(mockInstance).doCountSynchronously(batchCountFunction, tableList, false);
+			verify(obsLogger).info(contains("Query snapshot row size completed"));
+		}
+
+		@Test
+		void testAsyncCountTable_ExceptionHandling() throws InterruptedException {
+			doThrow(new RuntimeException("Count error")).when(mockInstance).doCountSynchronously(any(), any(), eq(false));
+			mockInstance.asyncCountTable(batchCountFunction, tableList);
+			Thread.sleep(100);
+			verify(obsLogger).warn(contains("Query snapshot row size failed"));
+		}
+	}
+	@Nested
+	class doCountSynchronouslyTest{
+		private BatchCountFunction batchCountFunction;
+		private DataProcessorContext dataProcessorContext;
+		private ObsLogger obsLogger;
+
+		@BeforeEach
+		void setUp() {
+			batchCountFunction = mock(BatchCountFunction.class);
+			dataProcessorContext = mock(DataProcessorContext.class);
+			obsLogger = mock(ObsLogger.class);
+			ReflectionTestUtils.setField(mockInstance,"dataProcessorContext",dataProcessorContext);
+			ReflectionTestUtils.setField(mockInstance,"obsLogger",obsLogger);
+		}
+
+		@Test
+		void testDoCountSynchronously_withBatchCountFunctionNull() {
+			doCallRealMethod().when(mockInstance).doCountSynchronously(null, Arrays.asList("table1"), false);
+			mockInstance.doCountSynchronously(null, Arrays.asList("table1"), false);
+
+			verify(obsLogger, times(1)).warn(anyString());
+		}
+
+		@Test
+		void testDoCountSynchronously_withException() throws Throwable {
+			String tableName = "table1";
+			TapTable tapTable = mock(TapTable.class);
+			TapTableMap map = TapTableMap.create("nodeId");
+			map.putNew("table1",tapTable,"qualifiedName1");
+			when(dataProcessorContext.getTapTableMap()).thenReturn(map);
+
+			when(batchCountFunction.count(any(), any())).thenThrow(new RuntimeException("Test Exception"));
+
+			try {
+				doCallRealMethod().when(mockInstance).doCountSynchronously(batchCountFunction, Arrays.asList(tableName), false);
+				mockInstance.doCountSynchronously(batchCountFunction, Arrays.asList(tableName), false);
+			} catch (Exception e) {
+				fail("Exception should be handled");
+			}
+			verify(obsLogger, never()).warn(anyString());
+		}
+	}
 }
diff --git a/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastSourcePdkDataNodeTest.java b/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastSourcePdkDataNodeTest.java
index ecd7777457..9aec3bc53d 100644
--- a/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastSourcePdkDataNodeTest.java
+++ b/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/HazelcastSourcePdkDataNodeTest.java
@@ -507,7 +507,7 @@ void verifyAssert(VerifyDifferent v) throws Exception {
 				verify(obsLogger, times(v.obsLoggerInfo2())).info("Table {} is detected that it has been removed, the snapshot read will be skipped", "id");
 				verify(instance, times(v.createPdkMethodInvoker())).createPdkMethodInvoker();
 				verify(instance, times(v.doAsyncTableCount())).doAsyncTableCount(batchCountFunction, tableId);
-				verify(ignoreTableCountCloseable, times(v.close())).close();
+//				verify(ignoreTableCountCloseable, times(v.close())).close();
 				verify(instance,times(v.executeDataFuncAspect())).executeDataFuncAspect(any(Class.class), any(Callable.class), any(CommonUtils.AnyErrorConsumer.class));
 				verify(obsLogger, times(v.obsLoggerInfo4())).info("Table [{}] has been completed batch read, will skip batch read on the next run", "id");
 				verify(instance, times(v.removePdkMethodInvoker())).removePdkMethodInvoker(pdkMethodInvoker);
@@ -861,7 +861,7 @@ void testNormal() throws Exception {
 						.removeTablesContains(1)
 						.removeTablesRemove(0)
 						.createPdkMethodInvoker(1)
-						.doAsyncTableCount(1)
+//						.doAsyncTableCount(1)
 						.close(1)
 						.executeDataFuncAspect(1)
 						.updateBatchOffset(1)
@@ -892,7 +892,7 @@ void testLastIsRunningIsFalse() throws Exception {
 						.lockBySourceRunnerLock(2).unlock(2)
 						.removeTablesContains(1)
 						.createPdkMethodInvoker(1)
-						.doAsyncTableCount(1)
+//						.doAsyncTableCount(1)
 						.close(1)
 						.executeDataFuncAspect(1)
 						.updateBatchOffset(1)
@@ -923,7 +923,7 @@ void testNewTablesIsEmpty() throws Exception {
 						.lockBySourceRunnerLock(2).unlock(2)
 						.removeTablesContains(1)
 						.createPdkMethodInvoker(1)
-						.doAsyncTableCount(1)
+//						.doAsyncTableCount(1)
 						.close(1)
 						.executeDataFuncAspect(1)
 						.updateBatchOffset(1)
@@ -959,7 +959,7 @@ void testThrowableWhenExecuteDataFuncAspect() throws Exception {
 						.removeTablesContains(1)
 						.removeTablesRemove(0)
 						.createPdkMethodInvoker(1)
-						.doAsyncTableCount(1)
+//						.doAsyncTableCount(1)
 						.close(1)
 						.executeDataFuncAspect(1)
 						.removePdkMethodInvoker(1)
@@ -1022,7 +1022,7 @@ void testTapCodeExceptionWhenExecuteDataFuncAspect() throws Exception {
 						.removeTablesContains(1)
 						.removeTablesRemove(0)
 						.createPdkMethodInvoker(1)
-						.doAsyncTableCount(1)
+//						.doAsyncTableCount(1)
 						.close(1)
 						.executeDataFuncAspect(1)
 						.removePdkMethodInvoker(1)
@@ -1083,7 +1083,7 @@ void testRemoveTablesIsNull() throws Exception {
 						.snapshotReadTableBeginAspect(1)
 						.lockBySourceRunnerLock(2).unlock(2)
 						.createPdkMethodInvoker(1)
-						.doAsyncTableCount(1)
+//						.doAsyncTableCount(1)
 						.close(1)
 						.executeDataFuncAspect(1)
 						.updateBatchOffset(1)
@@ -1198,7 +1198,7 @@ void testSourceRunnerFirstTimeGetIsFalse() throws Exception {
 						.removeTablesContains(1)
 						.removeTablesRemove(0)
 						.createPdkMethodInvoker(1)
-						.doAsyncTableCount(1)
+//						.doAsyncTableCount(1)
 						.close(1)
 						.executeDataFuncAspect(1)
 						.updateBatchOffset(1)
@@ -1238,7 +1238,7 @@ void butchCountFunctionIsNull() throws Exception {
 						.removeTablesContains(1)
 						.removeTablesRemove(0)
 						.createPdkMethodInvoker(1)
-						.doAsyncTableCount(1)
+//						.doAsyncTableCount(1)
 						.close(1)
 						.executeDataFuncAspect(1)
 						.updateBatchOffset(1)
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 86cce4bd7f..a97e73be69 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
@@ -32,8 +32,7 @@
 import io.tapdata.entity.event.dml.TapUpdateRecordEvent;
 import io.tapdata.entity.schema.TapField;
 import io.tapdata.entity.schema.TapTable;
-import io.tapdata.entity.schema.value.DateTime;
-import io.tapdata.entity.schema.value.TapDateTimeValue;
+import io.tapdata.entity.schema.value.*;
 import io.tapdata.error.TapdataEventException;
 import io.tapdata.error.TaskTargetProcessorExCode_15;
 import io.tapdata.exception.TapCodeException;
@@ -720,6 +719,7 @@ void test2() {
 			event.setContainsIllegalDate(true);
 			doCallRealMethod().when(hazelcastTargetPdkBaseNode).replaceIllegalDateWithNullIfNeed(event);
 			doCallRealMethod().when(hazelcastTargetPdkBaseNode).replaceIllegalDate(any());
+			doCallRealMethod().when(hazelcastTargetPdkBaseNode).replaceIllegalDateTime2Null(any(),any());
 			hazelcastTargetPdkBaseNode.replaceIllegalDateWithNullIfNeed(event);
 			assertNull(((TapInsertRecordEvent) event).getAfter().get("date"));
 			assertNull(((TapInsertRecordEvent) event).getAfter().get("last_date"));
@@ -742,6 +742,7 @@ void test3() {
 			event.setContainsIllegalDate(true);
 			doCallRealMethod().when(hazelcastTargetPdkBaseNode).replaceIllegalDateWithNullIfNeed(event);
 			doCallRealMethod().when(hazelcastTargetPdkBaseNode).replaceIllegalDate(any());
+			doCallRealMethod().when(hazelcastTargetPdkBaseNode).replaceIllegalDateTime2Null(any(),any());
 			hazelcastTargetPdkBaseNode.replaceIllegalDateWithNullIfNeed(event);
 			assertNull(((TapUpdateRecordEvent) event).getBefore().get("date"));
 			assertNull(((TapUpdateRecordEvent) event).getAfter().get("date"));
@@ -819,6 +820,52 @@ void testNotIllegalDateAcceptable() {
 			assertFalse(hazelcastTargetPdkBaseNode.illegalDateAcceptable);
 		}
 	}
+	@Nested
+	class replaceIllegalDateTest{
+		private HashMap<String,Object> after;
+		@BeforeEach
+		void init() {
+			doCallRealMethod().when(hazelcastTargetPdkBaseNode).replaceIllegalDate(anyMap());
+			doCallRealMethod().when(hazelcastTargetPdkBaseNode).replaceIllegalDateTime2Null(any(), any());
+			after=new HashMap<>();
+		}
+		@DisplayName("test replace tapDateValue")
+		@Test
+		void test1() {
+			TapDateValue tapDateValue = new TapDateValue();
+			DateTime dateTime = new DateTime("0000-00-00", DateTime.DATE_TYPE);
+			tapDateValue.setValue(dateTime);
+			after.put("testDate",tapDateValue);
+			hazelcastTargetPdkBaseNode.replaceIllegalDate(after);
+		}
+		@DisplayName("test replace tapYearValue")
+		@Test
+		void test2(){
+			TapYearValue tapYearValue = new TapYearValue();
+			DateTime dateTime = new DateTime("0000",DateTime.YEAR_TYPE);
+			tapYearValue.setValue(dateTime);
+			after.put("testYear",tapYearValue);
+			hazelcastTargetPdkBaseNode.replaceIllegalDate(after);
+		}
+		@DisplayName("test replace tapTimeValue")
+		@Test
+		void test3(){
+			TapTimeValue tapTimeValue = new TapTimeValue();
+			DateTime dateTime = new DateTime("00:00:00",DateTime.TIME_TYPE);
+			tapTimeValue.setValue(dateTime);
+			after.put("testTime",tapTimeValue);
+			hazelcastTargetPdkBaseNode.replaceIllegalDate(after);
+		}
+		@DisplayName("test replace TapDateTimeValue")
+		@Test
+		void test4(){
+			TapDateTimeValue tapDateTimeValue = new TapDateTimeValue();
+			DateTime dateTime = new DateTime("0000-00-00 00:00:00",DateTime.DATETIME_TYPE);
+			tapDateTimeValue.setValue(dateTime);
+			after.put("testDateTimeValue",tapDateTimeValue);
+			hazelcastTargetPdkBaseNode.replaceIllegalDate(after);
+		}
+	}
 
 	@Nested
 	@DisplayName("Method doInit test")
@@ -927,6 +974,7 @@ void beforeEach() {
 			tapTable.setId("test");
 			LinkedHashMap<String, TapField> nameFieldMap = new LinkedHashMap<>();
 			TapField primary = new TapField();
+			primary.setPrimaryKeyPos(1);
 			primary.setPrimaryKey(true);
 			nameFieldMap.put("primary", primary);
 			tapTable.setNameFieldMap(nameFieldMap);
diff --git a/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/concurrent/PartitionConcurrentProcessorTest.java b/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/concurrent/PartitionConcurrentProcessorTest.java
index 40fb35c76b..105d1140a4 100644
--- a/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/concurrent/PartitionConcurrentProcessorTest.java
+++ b/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/data/pdk/concurrent/PartitionConcurrentProcessorTest.java
@@ -105,7 +105,9 @@ void testEventOrders() throws Exception {
         list.add(generateDDLEvent(new TapField("ddl3", "string"))); // need after ddl1,ddl2,1,2,3,4,5,6
         list.add(generateInsertEvent("i", 7, null));
         list.add(new TapdataCompleteSnapshotEvent());
-        list.add(new TapdataCompleteTableSnapshotEvent());
+        TapdataCompleteTableSnapshotEvent tapdataCompleteTableSnapshotEvent = new TapdataCompleteTableSnapshotEvent();
+        tapdataCompleteTableSnapshotEvent.setBatchOffset("OVER");
+        list.add(tapdataCompleteTableSnapshotEvent);
         list.add(new TapdataStartingCdcEvent());
         list.add(TapdataStartedCdcEvent.create());
         list.add(generateInsertEvent("u", 8, null));
@@ -637,6 +639,7 @@ void testProcessOffset() {
         PartitionConcurrentProcessor processor = mock(PartitionConcurrentProcessor.class);
         when(processor.isRunning()).thenReturn(true);
         doCallRealMethod().when(processor).process(anyList(), anyBoolean());
+        doCallRealMethod().when(processor).getOffsetEvent(any(),anyList());
         doAnswer(invocationOnMock -> {
             offsetEvent.set(invocationOnMock.getArgument(0));
             return null;
diff --git a/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/processor/HazelcastProcessorNodeTest.java b/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/processor/HazelcastProcessorNodeTest.java
index 33ef909180..1a01c8ed48 100644
--- a/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/processor/HazelcastProcessorNodeTest.java
+++ b/iengine/iengine-app/src/test/java/io/tapdata/flow/engine/V2/node/hazelcast/processor/HazelcastProcessorNodeTest.java
@@ -3,20 +3,34 @@
 import base.BaseTaskTest;
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.hazelcast.jet.core.Processor;
+import com.tapdata.constant.HazelcastUtil;
 import com.tapdata.entity.TapdataEvent;
 import com.tapdata.entity.TransformToTapValueResult;
 import com.tapdata.entity.dataflow.Capitalized;
+import com.tapdata.entity.dataflow.Stage;
 import com.tapdata.entity.task.context.DataProcessorContext;
+import com.tapdata.entity.task.context.ProcessorBaseContext;
+import com.tapdata.processor.dataflow.DataFlowProcessor;
+import com.tapdata.processor.dataflow.RowFilterProcessor;
 import com.tapdata.tm.commons.dag.Node;
 import com.tapdata.tm.commons.dag.process.FieldProcessorNode;
 import com.tapdata.tm.commons.dag.process.FieldRenameProcessorNode;
+import com.tapdata.tm.commons.dag.process.RowFilterProcessorNode;
+import com.tapdata.tm.commons.task.dto.TaskDto;
 import io.tapdata.MockTaskUtil;
 import io.tapdata.entity.event.dml.TapInsertRecordEvent;
+import io.tapdata.entity.event.dml.TapUpdateRecordEvent;
+import io.tapdata.entity.logger.TapLogger;
+import io.tapdata.observable.logging.ObsLogger;
 import lombok.SneakyThrows;
+import org.bson.types.ObjectId;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.DisplayName;
 import org.junit.jupiter.api.Nested;
 import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+import org.mockito.internal.verification.Times;
 import org.springframework.test.util.ReflectionTestUtils;
 
 import java.io.File;
@@ -68,6 +82,22 @@ void test1() {
 
 			hazelcastProcessorNode.tryProcess(tapdataEvent, (event, processResult) -> assertEquals(data.get(1), (((TapInsertRecordEvent) event.getTapEvent()).getAfter())));
 		}
+
+		@Test
+		@DisplayName("test update event contains remove field")
+		void test2() {
+			TapdataEvent tapdataEvent = new TapdataEvent();
+			TapUpdateRecordEvent tapUpdateRecordEvent = TapUpdateRecordEvent.create().init();
+			ArrayList<String> removedFields = new ArrayList<>();
+			tapUpdateRecordEvent.setRemovedFields(removedFields);
+			List<?> data = json2Pojo(String.join(File.separator, "task", "json", TAG, "tryProcessTest1_data.json"), new TypeReference<List<?>>() {
+			});
+			tapUpdateRecordEvent.setAfter((Map<String, Object>) data.get(0));
+			tapdataEvent.setTapEvent(tapUpdateRecordEvent);
+
+			hazelcastProcessorNode.tryProcess(tapdataEvent, (event, processResult) -> assertEquals(data.get(1), (((TapUpdateRecordEvent) event.getTapEvent()).getAfter())));
+			assertEquals(removedFields, ((TapUpdateRecordEvent) tapdataEvent.getTapEvent()).getRemovedFields());
+		}
 	}
 
 	@Nested
@@ -323,4 +353,32 @@ void test3() {
 			assertTrue(result.contains("FIELD2"));
 		}
 	}
+	@Nested
+	class initDataFlowProcessorTest{
+		@Test
+		void testInitDataFlowProcessorSimple() {
+			hazelcastProcessorNode = mock(HazelcastProcessorNode.class);
+			processorBaseContext = mock(DataProcessorContext.class);
+			mockObsLogger = mock(ObsLogger.class);
+			ReflectionTestUtils.setField(hazelcastProcessorNode, "processorBaseContext", processorBaseContext);
+			ReflectionTestUtils.setField(hazelcastProcessorNode, "clientMongoOperator", mockClientMongoOperator);
+			ReflectionTestUtils.setField(hazelcastProcessorNode, "obsLogger", mockObsLogger);
+			Node node = mock(RowFilterProcessorNode.class);
+			when(processorBaseContext.getNode()).thenReturn(node);
+			when(node.getType()).thenReturn("row_filter_processor");
+			Stage stage = mock(Stage.class);
+			DataFlowProcessor dataFlowProcessor = mock(RowFilterProcessor.class);
+			try (MockedStatic<HazelcastUtil> mb = Mockito
+					.mockStatic(HazelcastUtil.class)) {
+				mb.when(()->HazelcastUtil.node2CommonStage(node)).thenReturn(stage);
+				when(hazelcastProcessorNode.createDataFlowProcessor(node, stage)).thenReturn(dataFlowProcessor);
+				TaskDto task = mock(TaskDto.class);
+				when(processorBaseContext.getTaskDto()).thenReturn(task);
+				when(task.getId()).thenReturn(mock(ObjectId.class));
+				doCallRealMethod().when(hazelcastProcessorNode).initDataFlowProcessor();
+				hazelcastProcessorNode.initDataFlowProcessor();
+				verify(dataFlowProcessor, new Times(1)).logListener(any());
+			}
+		}
+	}
 }
\ No newline at end of file
diff --git a/iengine/iengine-app/src/test/java/io/tapdata/websocket/ManagementWebsocketHandlerTest.java b/iengine/iengine-app/src/test/java/io/tapdata/websocket/ManagementWebsocketHandlerTest.java
index 3045edfc2a..86226b1d10 100644
--- a/iengine/iengine-app/src/test/java/io/tapdata/websocket/ManagementWebsocketHandlerTest.java
+++ b/iengine/iengine-app/src/test/java/io/tapdata/websocket/ManagementWebsocketHandlerTest.java
@@ -3,6 +3,7 @@
 import com.tapdata.constant.ConfigurationCenter;
 import com.tapdata.tm.sdk.util.Version;
 import com.tapdata.tm.worker.WorkerSingletonLock;
+import org.apache.logging.log4j.Logger;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Nested;
@@ -13,10 +14,13 @@
 import org.springframework.util.concurrent.ListenableFuture;
 import org.springframework.web.socket.TextMessage;
 import org.springframework.web.socket.WebSocketSession;
+import org.springframework.web.socket.client.standard.StandardWebSocketClient;
 
+import java.net.URI;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import java.util.concurrent.ExecutionException;
 
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.Mockito.*;
@@ -126,5 +130,30 @@ void SessionOptionTest_Error(){
             TextMessage textMessage = new TextMessage("test");
             Assertions.assertThrows(RuntimeException.class,()->sessionOption.sendMessage(textMessage));
         }
+
+        @Test
+        void SessionOptionTest_Interrupt_Exception() throws ExecutionException, InterruptedException {
+            ListenableFuture mockListenableFuture = mock(ListenableFuture.class);
+            ReflectionTestUtils.setField(managementWebsocketHandlerTest, "listenableFuture", mockListenableFuture);
+            doThrow(new InterruptedException()).when(mockListenableFuture).get();
+            StandardWebSocketClient mockClient = mock(StandardWebSocketClient.class);
+            when(mockClient.doHandshake(any(), any(), any(URI.class))).thenReturn(mockListenableFuture);
+            Logger mockLogger = mock(Logger.class);
+            ReflectionTestUtils.setField(managementWebsocketHandlerTest, "logger", mockLogger);
+
+            ManagementWebsocketHandler spyManagementWebsocketHandlerTest = spy(managementWebsocketHandlerTest);
+
+            try (MockedStatic<WorkerSingletonLock> mockedStatic = Mockito.mockStatic(WorkerSingletonLock.class);
+                 MockedStatic<Version> versionMockedStatic = Mockito.mockStatic(Version.class);
+                 MockedStatic<ManagementWebsocketHandler> managementWebsocketHandlerMockedStatic = mockStatic(ManagementWebsocketHandler.class);) {
+                mockedStatic.when(() -> WorkerSingletonLock.addTag2WsUrl(anyString())).thenReturn("ws://test:8080/ws/agent?agentId=test&access_token=test");
+                versionMockedStatic.when(Version::get).thenReturn("test");
+                managementWebsocketHandlerMockedStatic.when(() -> ManagementWebsocketHandler.createWebSocketClient()).thenReturn(mockClient);
+                String baseStr = "http://test:8080/api/";
+                spyManagementWebsocketHandlerTest.connect(baseStr);
+                verify(mockLogger, times(1)).warn("Connect to web socket Thread interrupted,Thread name:{}", Thread.currentThread().getName());
+            }
+        }
+
     }
 }
diff --git a/iengine/iengine-common/src/test/java/com/tapdata/mongo/HttpClientMongoOperatorTest.java b/iengine/iengine-common/src/test/java/com/tapdata/mongo/HttpClientMongoOperatorTest.java
new file mode 100644
index 0000000000..82d836b0ab
--- /dev/null
+++ b/iengine/iengine-common/src/test/java/com/tapdata/mongo/HttpClientMongoOperatorTest.java
@@ -0,0 +1,111 @@
+package com.tapdata.mongo;
+
+import com.tapdata.constant.ConfigurationCenter;
+import com.tapdata.entity.LoginResp;
+import com.tapdata.entity.User;
+import org.apache.logging.log4j.Logger;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.DisplayName;
+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 static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.*;
+
+public class HttpClientMongoOperatorTest {
+    @Nested
+    class ValidateTokenTest{
+        HttpClientMongoOperator httpClientMongoOperator;
+        ConfigurationCenter configurationCenter;
+        @BeforeEach
+        void init(){
+            httpClientMongoOperator = mock(HttpClientMongoOperator.class);
+            configurationCenter = new ConfigurationCenter();
+            ReflectionTestUtils.setField(httpClientMongoOperator, "configCenter", configurationCenter);
+        }
+        @DisplayName("test validateToken when ExpiredTimestamp less than refreshBufferTime")
+        @Test
+        void test1(){
+            doCallRealMethod().when(httpClientMongoOperator).validateToken();
+            long ttl = 300;
+            LoginResp loginResp = new LoginResp();
+            loginResp.setExpiredTimestamp(System.currentTimeMillis() + ttl * 1000);
+            loginResp.setTtl(ttl);
+
+            configurationCenter.putConfig(ConfigurationCenter.LOGIN_INFO,loginResp);
+            httpClientMongoOperator.validateToken();
+            verify(httpClientMongoOperator,times(0)).refreshToken();
+        }
+
+        @DisplayName("test validateToken when ExpiredTimestamp Greater than refreshBufferTime")
+        @Test
+        void test2(){
+            doCallRealMethod().when(httpClientMongoOperator).validateToken();
+            long ttl = 300;
+            LoginResp loginResp = new LoginResp();
+            loginResp.setTtl(ttl);
+            loginResp.setExpiredTimestamp(0);
+            configurationCenter.putConfig(ConfigurationCenter.LOGIN_INFO, loginResp);
+            httpClientMongoOperator.validateToken();
+            verify(httpClientMongoOperator,times(1)).refreshToken();
+        }
+        @DisplayName("test validateToken when login info is null")
+        @Test
+        void test3(){
+            doCallRealMethod().when(httpClientMongoOperator).validateToken();
+            httpClientMongoOperator.validateToken();
+            verify(httpClientMongoOperator,times(1)).refreshToken();
+        }
+    }
+    @Nested
+    class RefreshTokenClass{
+        HttpClientMongoOperator httpClientMongoOperator;
+        ConfigurationCenter configurationCenter;
+        String accessCode = "accessCode";
+        RestTemplateOperator restTemplateOperator;
+        Logger logger;
+
+        @BeforeEach
+        void init() {
+            httpClientMongoOperator = mock(HttpClientMongoOperator.class);
+            restTemplateOperator = mock(RestTemplateOperator.class);
+            logger = mock(Logger.class);
+            configurationCenter = new ConfigurationCenter();
+            ReflectionTestUtils.setField(httpClientMongoOperator, "configCenter", configurationCenter);
+            configurationCenter.putConfig(ConfigurationCenter.ACCESS_CODE, accessCode);
+            ReflectionTestUtils.setField(httpClientMongoOperator, "restTemplateOperator", restTemplateOperator);
+            ReflectionTestUtils.setField(httpClientMongoOperator, "logger", logger);
+        }
+
+        @DisplayName("test refresh success")
+        @Test
+        void test1() {
+            String token = "testAccessToken";
+            LoginResp loginResp = new LoginResp();
+            loginResp.setCreated("2024-10-17T08:00:00Z");
+            loginResp.setTtl(2000L);
+            loginResp.setId(token);
+            loginResp.setUserId("testUserId");
+            when(restTemplateOperator.postOne(anyMap(), anyString(), any())).thenReturn(loginResp);
+            User user = new User();
+            user.setId("userId");
+            when(httpClientMongoOperator.findOne(any(Query.class), anyString(), any())).thenReturn(user);
+            doCallRealMethod().when(httpClientMongoOperator).refreshToken();
+            httpClientMongoOperator.refreshToken();
+            assertEquals(token, configurationCenter.getConfig(ConfigurationCenter.TOKEN));
+        }
+        @DisplayName("test refresh token error")
+        @Test
+        void test2() {
+            when(restTemplateOperator.postOne(anyMap(), anyString(), any())).thenThrow(new RuntimeException("can not get token"));
+            User user = new User();
+            user.setId("userId");
+            when(httpClientMongoOperator.findOne(any(Query.class), anyString(), any())).thenReturn(user);
+            doCallRealMethod().when(httpClientMongoOperator).refreshToken();
+            httpClientMongoOperator.refreshToken();
+            verify(logger,times(6)).error(anyString(),anyString(),any(Exception.class));
+        }
+    }
+}
diff --git a/iengine/iengine-common/src/test/java/com/tapdata/processor/dataflow/RowFilterProcessorTest.java b/iengine/iengine-common/src/test/java/com/tapdata/processor/dataflow/RowFilterProcessorTest.java
index 24d3481bae..e3daab5160 100644
--- a/iengine/iengine-common/src/test/java/com/tapdata/processor/dataflow/RowFilterProcessorTest.java
+++ b/iengine/iengine-common/src/test/java/com/tapdata/processor/dataflow/RowFilterProcessorTest.java
@@ -17,6 +17,8 @@
 import java.util.HashMap;
 import java.util.Map;
 
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.mockito.Mockito.*;
 
 public class RowFilterProcessorTest {
@@ -32,10 +34,12 @@ class processTest{
         private Map<String, Object> processContext;
         private Invocable engine;
         private Logger logger;
+        private Map<String, Object> before;
+        private Map<String, Object> after;
         @BeforeEach
         @SneakyThrows
         void beforeEach() {
-            message = mock(MessageEntity.class);
+            message = new MessageEntity();
             engine = spy(ScriptUtil.getScriptEngine(
                     JSEngineEnum.GRAALVM_JS.getEngineName(),
                     "",
@@ -49,27 +53,151 @@ void beforeEach() {
             ReflectionTestUtils.setField(rowFilterProcessor,"engine",engine);
             context = mock(ProcessorContext.class);
             ReflectionTestUtils.setField(rowFilterProcessor,"context",context);
-            processContext = mock(HashMap.class);
+            processContext = new HashMap<>();
             ReflectionTestUtils.setField(rowFilterProcessor,"processContext",processContext);
             logger = mock(Logger.class);
             ReflectionTestUtils.setField(rowFilterProcessor,"logger",logger);
+            before = new HashMap<>();
+            after = new HashMap<>();
+            before.put("before","test");
+            after.put("after","test");
         }
         @Test
-        @SneakyThrows
-        void testSimple(){
+        void testUpdateEventSimple(){
             try (MockedStatic<ScriptUtil> mb = Mockito
                     .mockStatic(ScriptUtil.class)) {
                 mb.when(()->ScriptUtil.invokeScript(engine, "filter", message, context.getSourceConn(),
-                        context.getTargetConn(), context.getJob(), processContext, logger, null)).thenReturn(null);
-                when(message.getOp()).thenReturn("u");
-                Map<String, Object> before = mock(HashMap.class);
-                Map<String, Object> after = mock(HashMap.class);
-                when(message.getBefore()).thenReturn(before);
-                when(message.getAfter()).thenReturn(after);
+                        context.getTargetConn(), context.getJob(), processContext, logger, "before")).thenReturn(true);
+                mb.when(()->ScriptUtil.invokeScript(engine, "filter", message, context.getSourceConn(),
+                        context.getTargetConn(), context.getJob(), processContext, logger, "after")).thenReturn(true);
+                message.setOp("u");
+                message.setBefore(before);
+                message.setAfter(after);
+                doCallRealMethod().when(rowFilterProcessor).process(message);
+                MessageEntity actual = rowFilterProcessor.process(message);
+                assertEquals(message, actual);
+                mb.verify(() -> ScriptUtil.invokeScript(engine, "filter", message, context.getSourceConn(),
+                        context.getTargetConn(), context.getJob(), processContext, logger, "before"),new Times(1));
+                mb.verify(() -> ScriptUtil.invokeScript(engine, "filter", message, context.getSourceConn(),
+                        context.getTargetConn(), context.getJob(), processContext, logger, "after"),new Times(1));
+
+            }
+        }
+        @Test
+        void testUpdateEventReturnNull(){
+            try (MockedStatic<ScriptUtil> mb = Mockito
+                    .mockStatic(ScriptUtil.class)) {
+                mb.when(()->ScriptUtil.invokeScript(engine, "filter", message, context.getSourceConn(),
+                        context.getTargetConn(), context.getJob(), processContext, logger, "before")).thenReturn(false);
+                mb.when(()->ScriptUtil.invokeScript(engine, "filter", message, context.getSourceConn(),
+                        context.getTargetConn(), context.getJob(), processContext, logger, "after")).thenReturn(false);
+                message.setOp("u");
+                message.setBefore(before);
+                message.setAfter(after);
                 doCallRealMethod().when(rowFilterProcessor).process(message);
-                rowFilterProcessor.process(message);
+                MessageEntity actual = rowFilterProcessor.process(message);
+                assertNull(actual);
                 mb.verify(() -> ScriptUtil.invokeScript(engine, "filter", message, context.getSourceConn(),
-                        context.getTargetConn(), context.getJob(), processContext, logger, null),new Times(1));
+                        context.getTargetConn(), context.getJob(), processContext, logger, "before"),new Times(1));
+                mb.verify(() -> ScriptUtil.invokeScript(engine, "filter", message, context.getSourceConn(),
+                        context.getTargetConn(), context.getJob(), processContext, logger, "after"),new Times(1));
+
+            }
+        }
+        @Test
+        void testUpdateEvent2DeleteEvent() {
+            try (MockedStatic<ScriptUtil> mb = Mockito
+                    .mockStatic(ScriptUtil.class)) {
+                mb.when(()->ScriptUtil.invokeScript(engine, "filter", message, context.getSourceConn(),
+                        context.getTargetConn(), context.getJob(), processContext, logger, "before")).thenReturn(true);
+                mb.when(()->ScriptUtil.invokeScript(engine, "filter", message, context.getSourceConn(),
+                        context.getTargetConn(), context.getJob(), processContext, logger, "after")).thenReturn(false);
+                message.setOp("u");
+                message.setBefore(before);
+                message.setAfter(after);
+                doCallRealMethod().when(rowFilterProcessor).process(message);
+                MessageEntity actual = rowFilterProcessor.process(message);
+                assertEquals("d", actual.getOp());
+                assertNull(actual.getAfter());
+            }
+        }
+        @Test
+        void testUpdateEvent2InsertEvent() {
+            try (MockedStatic<ScriptUtil> mb = Mockito
+                    .mockStatic(ScriptUtil.class)) {
+                mb.when(()->ScriptUtil.invokeScript(engine, "filter", message, context.getSourceConn(),
+                        context.getTargetConn(), context.getJob(), processContext, logger, "before")).thenReturn(false);
+                mb.when(()->ScriptUtil.invokeScript(engine, "filter", message, context.getSourceConn(),
+                        context.getTargetConn(), context.getJob(), processContext, logger, "after")).thenReturn(true);
+                message.setOp("u");
+                message.setBefore(before);
+                message.setAfter(after);
+                doCallRealMethod().when(rowFilterProcessor).process(message);
+                MessageEntity actual = rowFilterProcessor.process(message);
+                assertEquals("i", actual.getOp());
+                assertNull(actual.getBefore());
+            }
+        }
+        @Test
+        void testDeleteEventSimple(){
+            try (MockedStatic<ScriptUtil> mb = Mockito
+                    .mockStatic(ScriptUtil.class)) {
+                mb.when(()->ScriptUtil.invokeScript(engine, "filter", message, context.getSourceConn(),
+                        context.getTargetConn(), context.getJob(), processContext, logger, null)).thenReturn(true);
+                message.setOp("d");
+                message.setBefore(before);
+                doCallRealMethod().when(rowFilterProcessor).process(message);
+                MessageEntity actual = rowFilterProcessor.process(message);
+                assertEquals(message, actual);
+            }
+        }
+        @Test
+        void testDeleteEventReturnNull(){
+            try (MockedStatic<ScriptUtil> mb = Mockito
+                    .mockStatic(ScriptUtil.class)) {
+                mb.when(()->ScriptUtil.invokeScript(engine, "filter", message, context.getSourceConn(),
+                        context.getTargetConn(), context.getJob(), processContext, logger, null)).thenReturn(null);
+                message.setOp("d");
+                message.setBefore(before);
+                doCallRealMethod().when(rowFilterProcessor).process(message);
+                MessageEntity actual = rowFilterProcessor.process(message);
+                assertNull(actual);
+            }
+        }
+        @Test
+        void testUpdateEventWithoutBefore2InsertEvent() {
+            ReflectionTestUtils.setField(rowFilterProcessor, "action", RowFilterProcessor.FilterAction.DISCARD);
+            try (MockedStatic<ScriptUtil> mb = Mockito
+                    .mockStatic(ScriptUtil.class)) {
+                mb.when(()->ScriptUtil.invokeScript(engine, "filter", message, context.getSourceConn(),
+                        context.getTargetConn(), context.getJob(), processContext, logger, "before")).thenReturn(false);
+                mb.when(()->ScriptUtil.invokeScript(engine, "filter", message, context.getSourceConn(),
+                        context.getTargetConn(), context.getJob(), processContext, logger, "after")).thenReturn(false);
+                message.setOp("u");
+                message.setBefore(null);
+                message.setAfter(after);
+                doCallRealMethod().when(rowFilterProcessor).process(message);
+                MessageEntity actual = rowFilterProcessor.process(message);
+                assertEquals("i", actual.getOp());
+                assertNull(actual.getBefore());
+            }
+        }
+        @Test
+        void testUpdateEventWithoutBefore2DeleteEvent() {
+            ReflectionTestUtils.setField(rowFilterProcessor, "action", RowFilterProcessor.FilterAction.DISCARD);
+            try (MockedStatic<ScriptUtil> mb = Mockito
+                    .mockStatic(ScriptUtil.class)) {
+                mb.when(()->ScriptUtil.invokeScript(engine, "filter", message, context.getSourceConn(),
+                        context.getTargetConn(), context.getJob(), processContext, logger, "before")).thenReturn(true);
+                mb.when(()->ScriptUtil.invokeScript(engine, "filter", message, context.getSourceConn(),
+                        context.getTargetConn(), context.getJob(), processContext, logger, "after")).thenReturn(true);
+                message.setOp("u");
+                message.setBefore(null);
+                message.setAfter(after);
+                doCallRealMethod().when(rowFilterProcessor).process(message);
+                MessageEntity actual = rowFilterProcessor.process(message);
+                assertEquals("d", actual.getOp());
+                assertNull(actual.getAfter());
             }
         }
     }
diff --git a/iengine/modules/observable-module/src/test/java/io/tapdata/observable/metric/handler/DataNodeSampleHandlerTest.java b/iengine/modules/observable-module/src/test/java/io/tapdata/observable/metric/handler/DataNodeSampleHandlerTest.java
index 413164255e..cd29bc9434 100644
--- a/iengine/modules/observable-module/src/test/java/io/tapdata/observable/metric/handler/DataNodeSampleHandlerTest.java
+++ b/iengine/modules/observable-module/src/test/java/io/tapdata/observable/metric/handler/DataNodeSampleHandlerTest.java
@@ -9,9 +9,13 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Nested;
 import org.junit.jupiter.api.Test;
+import org.springframework.test.util.ReflectionTestUtils;
 
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.anyLong;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.*;
 import static org.mockito.Mockito.doCallRealMethod;
 import static org.mockito.Mockito.doNothing;
 import static org.mockito.Mockito.mock;
@@ -154,4 +158,17 @@ void testHandleWriteRecordAccept() {
             verify(handler.outputSizeSpeed, times(1)).add(recorder.getMemorySize());
         }
     }
+    @Nested
+    class handleTableCountAcceptTest{
+        @Test
+        void testHandleTableCountAcceptSample() {
+            Map<String, Long> currentSnapshotTableRowTotalMap = new HashMap<>();
+            ReflectionTestUtils.setField(handler, "currentSnapshotTableRowTotalMap", currentSnapshotTableRowTotalMap);
+            String table = "table1";
+            doCallRealMethod().when(handler).handleTableCountAccept(anyString(), anyLong());
+            handler.handleTableCountAccept(table, -1L);
+            handler.handleTableCountAccept(table, 10L);
+            assertEquals(10L, currentSnapshotTableRowTotalMap.get(table));
+        }
+    }
 }
diff --git a/iengine/modules/observable-module/src/test/java/io/tapdata/observable/metric/handler/RandomSampleEventHandlerTest.java b/iengine/modules/observable-module/src/test/java/io/tapdata/observable/metric/handler/RandomSampleEventHandlerTest.java
index db9a120d2c..379b73343f 100644
--- a/iengine/modules/observable-module/src/test/java/io/tapdata/observable/metric/handler/RandomSampleEventHandlerTest.java
+++ b/iengine/modules/observable-module/src/test/java/io/tapdata/observable/metric/handler/RandomSampleEventHandlerTest.java
@@ -3,6 +3,7 @@
 import com.tapdata.entity.TapdataHeartbeatEvent;
 import io.tapdata.entity.event.TapEvent;
 import io.tapdata.entity.event.ddl.index.TapCreateIndexEvent;
+import io.tapdata.entity.event.dml.TapInsertRecordEvent;
 import io.tapdata.entity.event.dml.TapUpdateRecordEvent;
 import org.apache.commons.lang3.RandomUtils;
 import org.junit.Assert;
@@ -18,6 +19,9 @@
 import java.util.UUID;
 import java.util.function.Function;
 
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.*;
+
 public class RandomSampleEventHandlerTest {
 
     @Nested
@@ -25,18 +29,18 @@ class RandomSampleEventHandlerInitTest {
         @Test
         void rangeLessThenZero() {
             RandomSampleEventHandler r = new RandomSampleEventHandler(-1);
-            Assertions.assertEquals(1, r.sampleRate);
+            assertEquals(1, r.sampleRate);
         }
         @Test
         void rangeLagerThenOne() {
             RandomSampleEventHandler r = new RandomSampleEventHandler(21);
-            Assertions.assertEquals(1, r.sampleRate);
+            assertEquals(1, r.sampleRate);
         }
 
         @Test
         void rangeBetweenZeroAndOne() {
             RandomSampleEventHandler r = new RandomSampleEventHandler(0.8);
-            Assertions.assertEquals(0.8, r.sampleRate);
+            assertEquals(0.8, r.sampleRate);
         }
     }
 
@@ -263,7 +267,7 @@ public void testSimpleMemoryNotRecordTapEvent() {
             events.add(event);
             RandomSampleEventHandler.HandleEvent handle = (e) -> (TapEvent) e;
             handler.sampleMemoryTapEvent(recorder, events, handle);
-            Assertions.assertEquals( 0, recorder.getMemorySize());
+            assertEquals( 0, recorder.getMemorySize());
         }
     }
 
@@ -301,4 +305,20 @@ public static Integer simpleInteger(Object obj) {
         return RandomUtils.nextInt();
     }
 
+    @Nested
+    class sampleMemoryTapEventTest {
+        @Test
+        void testSampleMemoryTapEvent() {
+            RandomSampleEventHandler handler = mock(RandomSampleEventHandler.class);
+            List events = new ArrayList<>();
+            TapInsertRecordEvent tapEvent = mock(TapInsertRecordEvent.class);
+            events.add(tapEvent);
+            RandomSampleEventHandler.HandleEvent handle = mock(RandomSampleEventHandler.HandleEvent.class);
+            when(handle.handel(events.get(0))).thenReturn(tapEvent);
+            when(handler.sizeOfTapEvent(any())).thenReturn(100L);
+            doCallRealMethod().when(handler).sampleMemoryTapEvent(events, handle);
+            long actual = handler.sampleMemoryTapEvent(events, handle);
+            assertEquals(100L, actual);
+        }
+    }
 }
diff --git a/iengine/modules/observable-module/src/test/java/io/tapdata/observable/metric/handler/TaskSampleHandlerTest.java b/iengine/modules/observable-module/src/test/java/io/tapdata/observable/metric/handler/TaskSampleHandlerTest.java
index 55e5ac50cc..409dd8195a 100644
--- a/iengine/modules/observable-module/src/test/java/io/tapdata/observable/metric/handler/TaskSampleHandlerTest.java
+++ b/iengine/modules/observable-module/src/test/java/io/tapdata/observable/metric/handler/TaskSampleHandlerTest.java
@@ -1,9 +1,19 @@
 package io.tapdata.observable.metric.handler;
 
+import io.tapdata.common.sample.sampler.CounterSampler;
 import org.junit.Assert;
 import org.junit.jupiter.api.Test;
+import org.springframework.test.util.ReflectionTestUtils;
 
 import java.lang.reflect.Field;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doCallRealMethod;
+import static org.mockito.Mockito.mock;
 
 public class TaskSampleHandlerTest {
 
@@ -37,4 +47,17 @@ private void testStaticFields(String fieldName, String value) {
         Assert.assertNotNull(v);
         Assert.assertEquals(value, v);
     }
+    @Test
+    void handleTableCountAcceptTest() {
+        TaskSampleHandler handler = mock(TaskSampleHandler.class);
+        Map<String, Long> currentSnapshotTableRowTotalMap = new HashMap<>();
+        ReflectionTestUtils.setField(handler, "currentSnapshotTableRowTotalMap", currentSnapshotTableRowTotalMap);
+        CounterSampler snapshotRowTotal = mock(CounterSampler.class);
+        ReflectionTestUtils.setField(handler, "snapshotRowTotal", snapshotRowTotal);
+        String table = "table1";
+        doCallRealMethod().when(handler).handleTableCountAccept(anyString(), anyLong());
+        handler.handleTableCountAccept(table, -1L);
+        handler.handleTableCountAccept(table, 10L);
+        assertEquals(10L, currentSnapshotTableRowTotalMap.get(table));
+    }
 }
diff --git a/manager/tm-api/src/test/java/com/tapdata/tm/base/aop/MeasureAOPTest.java b/manager/tm-api/src/test/java/com/tapdata/tm/base/aop/MeasureAOPTest.java
index 9975bb85a9..6d63e5c9b0 100644
--- a/manager/tm-api/src/test/java/com/tapdata/tm/base/aop/MeasureAOPTest.java
+++ b/manager/tm-api/src/test/java/com/tapdata/tm/base/aop/MeasureAOPTest.java
@@ -1,11 +1,15 @@
 package com.tapdata.tm.base.aop;
 
 import com.google.common.collect.Maps;
+import com.tapdata.tm.Settings.service.AlarmSettingService;
 import com.tapdata.tm.alarm.constant.AlarmComponentEnum;
 import com.tapdata.tm.alarm.constant.AlarmStatusEnum;
 import com.tapdata.tm.alarm.constant.AlarmTypeEnum;
 import com.tapdata.tm.alarm.entity.AlarmInfo;
 import com.tapdata.tm.alarm.service.AlarmService;
+import com.tapdata.tm.commons.dag.DAG;
+import com.tapdata.tm.commons.dag.Node;
+import com.tapdata.tm.commons.dag.nodes.TableNode;
 import com.tapdata.tm.commons.task.constant.AlarmKeyEnum;
 import com.tapdata.tm.commons.task.dto.TaskDto;
 import com.tapdata.tm.commons.task.dto.alarm.AlarmRuleDto;
@@ -38,6 +42,7 @@ class TaskIncrementDelayAlarmTest{
         private TaskService taskService;
         private AlarmService alarmService;
         private UserService userService;
+        private AlarmSettingService alarmSettingService;
         Map<String, Map<String, AtomicInteger>> obsMap = Maps.newConcurrentMap();
         String taskId = "66837fb973828322f83f7074";
         List<AlarmInfo> alarmInfos=new ArrayList<>();
@@ -47,6 +52,7 @@ void setUp() {
             taskService = mock(TaskService.class);
             alarmService = mock(AlarmService.class);
             userService = mock(UserService.class);
+            alarmSettingService = mock(AlarmSettingService.class);
             Map<String,AtomicInteger> infoMap=new HashMap<>();
             infoMap.put(taskId+"-replicateLag",new AtomicInteger(50000));
             obsMap.put(taskId,infoMap);
@@ -61,7 +67,7 @@ void test1(){
             TaskDto taskDto = new TaskDto();
             taskDto.setId(new ObjectId(taskId));
             taskDto.setCurrentEventTimestamp(System.currentTimeMillis());
-            MeasureAOP measureAOP = new MeasureAOP(taskService, alarmService, userService);
+            MeasureAOP measureAOP = new MeasureAOP(taskService, alarmService, userService,alarmSettingService);
             ReflectionTestUtils.setField(measureAOP, "obsMap", obsMap);
             doAnswer(invocationOnMock -> {
                 AlarmInfo alarmInfo = (AlarmInfo)invocationOnMock.getArgument(0);
@@ -80,7 +86,7 @@ void test2(){
             TaskDto taskDto = new TaskDto();
             taskDto.setId(new ObjectId(taskId));
             taskDto.setCurrentEventTimestamp(System.currentTimeMillis());
-            MeasureAOP measureAOP = new MeasureAOP(taskService, alarmService, userService);
+            MeasureAOP measureAOP = new MeasureAOP(taskService, alarmService, userService,alarmSettingService);
             ReflectionTestUtils.setField(measureAOP, "obsMap", obsMap);
             doAnswer(invocationOnMock -> {
                 AlarmInfo alarmInfo = (AlarmInfo)invocationOnMock.getArgument(0);
@@ -100,7 +106,7 @@ void test3(){
             TaskDto taskDto = new TaskDto();
             taskDto.setId(new ObjectId(taskId));
             taskDto.setCurrentEventTimestamp(System.currentTimeMillis());
-            MeasureAOP measureAOP = new MeasureAOP(taskService, alarmService, userService);
+            MeasureAOP measureAOP = new MeasureAOP(taskService, alarmService, userService,alarmSettingService);
             doAnswer(invocationOnMock -> {
                 AlarmInfo alarmInfo = (AlarmInfo)invocationOnMock.getArgument(0);
                 assertEquals(Level.RECOVERY,alarmInfo.getLevel());
@@ -118,7 +124,7 @@ void test4(){
             TaskDto taskDto = new TaskDto();
             taskDto.setId(new ObjectId(taskId));
             taskDto.setCurrentEventTimestamp(System.currentTimeMillis());
-            MeasureAOP measureAOP = new MeasureAOP(taskService, alarmService, userService);
+            MeasureAOP measureAOP = new MeasureAOP(taskService, alarmService, userService,alarmSettingService);
             ReflectionTestUtils.setField(measureAOP, "obsMap", obsMap);
             doAnswer(invocationOnMock -> {
                 AlarmInfo alarmInfo = (AlarmInfo)invocationOnMock.getArgument(0);
@@ -141,7 +147,7 @@ void test5(){
             TaskDto taskDto = new TaskDto();
             taskDto.setId(new ObjectId(taskId));
             taskDto.setCurrentEventTimestamp(System.currentTimeMillis());
-            MeasureAOP measureAOP = new MeasureAOP(taskService, alarmService, userService);
+            MeasureAOP measureAOP = new MeasureAOP(taskService, alarmService, userService,alarmSettingService);
             ReflectionTestUtils.setField(measureAOP, "obsMap", obsMap);
             doAnswer(invocationOnMock -> {
                 AlarmInfo alarmInfo = (AlarmInfo)invocationOnMock.getArgument(0);
@@ -161,6 +167,7 @@ class AddAgentMeasurementTest{
         private TaskService taskService;
         private AlarmService alarmService;
         private UserService userService;
+        private AlarmSettingService alarmSettingService;
         private MeasureAOP measureAOP;
         private TaskDto taskDto;
         UserDetail userDetail;
@@ -177,20 +184,23 @@ void setUp() {
             alarmService = mock(AlarmService.class);
             userService = mock(UserService.class);
             measureAOP = mock(MeasureAOP.class);
+            alarmSettingService = mock(AlarmSettingService.class);
             ReflectionTestUtils.setField(measureAOP, "taskService", taskService);
             ReflectionTestUtils.setField(measureAOP, "alarmService", alarmService);
             ReflectionTestUtils.setField(measureAOP, "userService", userService);
+            ReflectionTestUtils.setField(measureAOP, "alarmSettingService", alarmSettingService);
 
             taskDto = new TaskDto();
             taskDto.setId(MongoUtils.toObjectId(taskId));
+            taskDto.setCurrentEventTimestamp(System.currentTimeMillis());
             when(taskService.findByTaskId(any())).thenReturn(taskDto);
-
             alarmRuleDto = getAlarmRuleDto();
             List<AlarmRuleDto> alarmRuleDtoList=new ArrayList<>();
             alarmRuleDtoList.add(alarmRuleDto);
 
             Map<String, List<AlarmRuleDto>> ruleMap=new HashMap<>();
             ruleMap.put(taskId,alarmRuleDtoList);
+            ruleMap.put("test", alarmRuleDtoList);
             when(alarmService.getAlarmRuleDtos(any())).thenReturn(ruleMap);
 
             userDetail = mock(UserDetail.class);
@@ -201,7 +211,8 @@ void setUp() {
         @DisplayName("test addAgentMeasurement when sample type is task and have alarmRule is TASK_INCREMENT_DELAY")
         @Test
         void test1(){
-            when(alarmService.checkOpen(taskDto,null, AlarmKeyEnum.TASK_INCREMENT_DELAY, null, userDetail)).thenReturn(true);
+            when(alarmService.checkOpen(any(),any(),any(),any(),anyList())).thenReturn(true);
+            when(taskService.findByTaskId(MongoUtils.toObjectId(taskId),"_id","dag","user_id","agentId","name","currentEventTimestamp","alarmSettings")).thenReturn(taskDto);
             List<SampleRequest> list = new ArrayList<>();
             SampleRequest sampleRequest=new SampleRequest();
 
@@ -227,6 +238,126 @@ void test1(){
             measureAOP.addAgentMeasurement(joinPoint);
             verify(measureAOP,times(1)).taskIncrementDelayAlarm(taskDto,taskId, 20000, alarmRuleDto);
         }
+
+        @DisplayName("test addAgentMeasurement when sample type is node")
+        @Test
+        void test2(){
+            DAG dag = mock(DAG.class);
+            List<Node> nodes = new ArrayList<>();
+            TableNode node = new TableNode();
+            node.setId("test");
+            node.setName("table");
+            nodes.add(node);
+            when(dag.getSources()).thenReturn(nodes);
+            when(dag.getNode("test")).thenReturn((Node)node);
+            taskDto.setDag(dag);
+            when(alarmService.checkOpen(any(),any(),any(),any(),anyList())).thenReturn(true);
+            when(taskService.findByTaskId(MongoUtils.toObjectId(taskId),"_id","dag","user_id","agentId","name","currentEventTimestamp","alarmSettings")).thenReturn(taskDto);
+            List<SampleRequest> list = new ArrayList<>();
+            SampleRequest sampleRequest=new SampleRequest();
+
+            Map<String, String> tags =new HashMap<>();
+            tags.put("type","node");
+            tags.put("taskId",taskId);
+            tags.put("nodeId","test");
+            sampleRequest.setTags(tags);
+
+            Map<String,Number> vs=new HashMap<>();
+            vs.put("replicateLag",20000);
+            Sample sample=new Sample();
+            sample.setVs(vs);
+            sampleRequest.setSample(sample);
+
+            list.add(sampleRequest);
+
+            JoinPoint joinPoint = mock(JoinPoint.class);
+
+            Object[] objects = new Object[10];
+            objects[0]=list;
+            when(joinPoint.getArgs()).thenReturn(objects);
+            doCallRealMethod().when(measureAOP).addAgentMeasurement(joinPoint);
+            measureAOP.addAgentMeasurement(joinPoint);
+            verify(measureAOP,times(1)).supplmentDelayAvg(any(),any(),any(),any(),any(),any(),any());
+        }
+
+        @DisplayName("test addAgentMeasurement when sample type is node")
+        @Test
+        void test3(){
+            DAG dag = mock(DAG.class);
+            List<Node> nodes = new ArrayList<>();
+            TableNode node = new TableNode();
+            node.setId("test");
+            node.setName("table");
+            nodes.add(node);
+            when(dag.getTargets()).thenReturn(nodes);
+            when(dag.getNode("test")).thenReturn((Node)node);
+            taskDto.setDag(dag);
+            when(alarmService.checkOpen(any(),any(),any(),any(),anyList())).thenReturn(true);
+            when(taskService.findByTaskId(MongoUtils.toObjectId(taskId),"_id","dag","user_id","agentId","name","currentEventTimestamp","alarmSettings")).thenReturn(taskDto);
+            List<SampleRequest> list = new ArrayList<>();
+            SampleRequest sampleRequest=new SampleRequest();
+
+            Map<String, String> tags =new HashMap<>();
+            tags.put("type","node");
+            tags.put("taskId",taskId);
+            tags.put("nodeId","test");
+            sampleRequest.setTags(tags);
+
+            Map<String,Number> vs=new HashMap<>();
+            vs.put("replicateLag",20000);
+            Sample sample=new Sample();
+            sample.setVs(vs);
+            sampleRequest.setSample(sample);
+
+            list.add(sampleRequest);
+
+            JoinPoint joinPoint = mock(JoinPoint.class);
+
+            Object[] objects = new Object[10];
+            objects[0]=list;
+            when(joinPoint.getArgs()).thenReturn(objects);
+            doCallRealMethod().when(measureAOP).addAgentMeasurement(joinPoint);
+            measureAOP.addAgentMeasurement(joinPoint);
+            verify(measureAOP,times(1)).supplmentDelayAvg(any(),any(),any(),any(),any(),any(),any());
+        }
+
+        @DisplayName("test addAgentMeasurement when sample type is node")
+        @Test
+        void test4(){
+            DAG dag = mock(DAG.class);
+            TableNode node = new TableNode();
+            node.setId("test");
+            node.setName("table");
+            when(dag.getNode("test")).thenReturn((Node)node);
+            taskDto.setDag(dag);
+            when(alarmService.checkOpen(any(),any(),any(),any(),anyList())).thenReturn(true);
+            when(taskService.findByTaskId(MongoUtils.toObjectId(taskId),"_id","dag","user_id","agentId","name","currentEventTimestamp","alarmSettings")).thenReturn(taskDto);
+            List<SampleRequest> list = new ArrayList<>();
+            SampleRequest sampleRequest=new SampleRequest();
+
+            Map<String, String> tags =new HashMap<>();
+            tags.put("type","node");
+            tags.put("taskId",taskId);
+            tags.put("nodeId","test");
+            sampleRequest.setTags(tags);
+
+            Map<String,Number> vs=new HashMap<>();
+            vs.put("replicateLag",20000);
+            Sample sample=new Sample();
+            sample.setVs(vs);
+            sampleRequest.setSample(sample);
+
+            list.add(sampleRequest);
+
+            JoinPoint joinPoint = mock(JoinPoint.class);
+
+            Object[] objects = new Object[10];
+            objects[0]=list;
+            when(joinPoint.getArgs()).thenReturn(objects);
+            doCallRealMethod().when(measureAOP).addAgentMeasurement(joinPoint);
+            measureAOP.addAgentMeasurement(joinPoint);
+            verify(measureAOP,times(1)).supplmentDelayAvg(any(),any(),any(),any(),any(),any(),any());
+        }
     }
 
     @NotNull
diff --git a/manager/tm-common/src/test/java/com/tapdata/tm/commons/util/NoPrimaryKeySyncModeTest.java b/manager/tm-common/src/test/java/com/tapdata/tm/commons/util/NoPrimaryKeySyncModeTest.java
new file mode 100644
index 0000000000..789a35f27d
--- /dev/null
+++ b/manager/tm-common/src/test/java/com/tapdata/tm/commons/util/NoPrimaryKeySyncModeTest.java
@@ -0,0 +1,32 @@
+package com.tapdata.tm.commons.util;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+
+/**
+ * @author <a href="mailto:harsen_lin@163.com">Harsen</a>
+ * @version v1.0 2024/10/28 10:33 Create
+ */
+public class NoPrimaryKeySyncModeTest {
+
+    @Nested
+    class fromValueTest {
+        @Test
+        void testBlank() {
+            Assertions.assertEquals(NoPrimaryKeySyncMode.ALL_COLUMNS, NoPrimaryKeySyncMode.fromValue(null));
+            Assertions.assertEquals(NoPrimaryKeySyncMode.ALL_COLUMNS, NoPrimaryKeySyncMode.fromValue(""));
+        }
+
+        @Test
+        void testAllColumns() {
+            Assertions.assertEquals(NoPrimaryKeySyncMode.ALL_COLUMNS, NoPrimaryKeySyncMode.fromValue("ALL_COLUMNS"));
+        }
+
+        @Test
+        void testAddHash() {
+            Assertions.assertEquals(NoPrimaryKeySyncMode.ADD_HASH, NoPrimaryKeySyncMode.fromValue("ADD_HASH"));
+            Assertions.assertEquals(NoPrimaryKeySyncMode.ADD_HASH, NoPrimaryKeySyncMode.fromValue("others"));
+        }
+    }
+}
diff --git a/manager/tm-common/src/test/java/com/tapdata/tm/commons/util/NoPrimaryKeyVirtualFieldTest.java b/manager/tm-common/src/test/java/com/tapdata/tm/commons/util/NoPrimaryKeyVirtualFieldTest.java
new file mode 100644
index 0000000000..3d72f80a8b
--- /dev/null
+++ b/manager/tm-common/src/test/java/com/tapdata/tm/commons/util/NoPrimaryKeyVirtualFieldTest.java
@@ -0,0 +1,749 @@
+package com.tapdata.tm.commons.util;
+
+import com.tapdata.tm.commons.dag.Element;
+import com.tapdata.tm.commons.dag.Node;
+import com.tapdata.tm.commons.dag.nodes.CacheNode;
+import com.tapdata.tm.commons.dag.nodes.DatabaseNode;
+import com.tapdata.tm.commons.dag.nodes.TableNode;
+import com.tapdata.tm.commons.dag.process.JoinProcessorNode;
+import com.tapdata.tm.commons.dag.process.MergeTableNode;
+import com.tapdata.tm.commons.exception.NoPrimaryKeyException;
+import com.tapdata.tm.commons.schema.*;
+import io.github.openlg.graphlib.Graph;
+import io.tapdata.entity.event.dml.TapDeleteRecordEvent;
+import io.tapdata.entity.event.dml.TapInsertRecordEvent;
+import io.tapdata.entity.event.dml.TapUnknownRecordEvent;
+import io.tapdata.entity.event.dml.TapUpdateRecordEvent;
+import io.tapdata.entity.schema.TapTable;
+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.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.*;
+
+/**
+ * @author <a href="mailto:harsen_lin@163.com">Harsen</a>
+ * @version v1.0 2024/10/18 10:12 Create
+ */
+public class NoPrimaryKeyVirtualFieldTest {
+
+    Field createVirualField(String tableName) {
+        Field field = new Field();
+        field.setTableName(tableName);
+        field.setFieldName(NoPrimaryKeyVirtualField.FIELD_NAME);
+        field.setOriginalFieldName(NoPrimaryKeyVirtualField.FIELD_NAME);
+        field.setSource(Field.SOURCE_VIRTUAL_HASH);
+        field.setCreateSource(Field.SOURCE_VIRTUAL_HASH);
+        return field;
+    }
+
+    Field createNormalField(String tableName, String fieldName) {
+        return createNormalField(tableName, fieldName, 0);
+    }
+
+    Field createNormalField(String tableName, String fieldName, Integer pkPos) {
+        Field field = new Field();
+        field.setTableName(tableName);
+        field.setFieldName(fieldName);
+        field.setOriginalFieldName(fieldName);
+        field.setSource(Field.SOURCE_AUTO);
+        field.setCreateSource(Field.SOURCE_AUTO);
+        if (null != pkPos && pkPos > 0) {
+            field.setPrimaryKey(true);
+            field.setPrimaryKeyPosition(pkPos);
+        }
+        return field;
+    }
+
+    @Nested
+    class init {
+        String tableNameNoPk = "no_pk_tab";
+        NoPrimaryKeyVirtualField instance;
+
+        @BeforeEach
+        void setUp() {
+            instance = new NoPrimaryKeyVirtualField();
+        }
+
+        @Test
+        void testTrue() {
+            try (MockedStatic<NoPrimaryKeyVirtualField> mockedStatic = Mockito.mockStatic(NoPrimaryKeyVirtualField.class)) {
+                mockedStatic.when(() -> NoPrimaryKeyVirtualField.isEnable(Mockito.any())).thenReturn(true);
+                instance.init(null);
+
+                // add no pk table
+                instance.add(Optional.of(new Schema()).map(schema -> {
+                    schema.setOriginalName(tableNameNoPk);
+                    schema.setFields(new ArrayList<>());
+                    schema.getFields().add(createNormalField(schema.getName(), "title"));
+                    schema.getFields().add(createNormalField(schema.getName(), "created"));
+                    return PdkSchemaConvert.toPdk(schema);
+                }).get());
+                Assertions.assertTrue(instance.noPkTables.containsKey(tableNameNoPk));
+                Assertions.assertTrue(instance.addHashValue(TapInsertRecordEvent
+                    .create()
+                    .table(tableNameNoPk)
+                    .after(Optional.of(new LinkedHashMap<String, Object>()).map(data -> {
+                        data.put("title", "init");
+                        data.put("created", System.currentTimeMillis());
+                        return data;
+                    }).get())
+                ));
+            }
+        }
+
+        @Test
+        void testFalse() {
+            try (MockedStatic<NoPrimaryKeyVirtualField> mockedStatic = Mockito.mockStatic(NoPrimaryKeyVirtualField.class)) {
+                mockedStatic.when(() -> NoPrimaryKeyVirtualField.isEnable(Mockito.any())).thenReturn(false);
+                instance.init(null);
+
+                // add no pk table
+                instance.add(Optional.of(new Schema()).map(schema -> {
+                    schema.setOriginalName(tableNameNoPk);
+                    schema.setFields(new ArrayList<>());
+                    schema.getFields().add(createNormalField(schema.getName(), "title"));
+                    schema.getFields().add(createNormalField(schema.getName(), "created"));
+                    return PdkSchemaConvert.toPdk(schema);
+                }).get());
+                Assertions.assertTrue(instance.noPkTables.isEmpty());
+                Assertions.assertTrue(instance.addHashValue(TapInsertRecordEvent
+                    .create()
+                    .table(tableNameNoPk)
+                    .after(Optional.of(new LinkedHashMap<String, Object>()).map(data -> {
+                        data.put("title", "init");
+                        data.put("created", System.currentTimeMillis());
+                        return data;
+                    }).get())
+                ));
+            }
+        }
+    }
+
+    @Nested
+    class addVirtualField {
+        String tableName;
+        List<Field> fields;
+        Schema schema;
+        Graph<Element, Element> graph;
+
+        @BeforeEach
+        void setUp() {
+            tableName = "no_pk_test";
+
+            fields = new ArrayList<>();
+            fields.add(createNormalField(tableName, "title"));
+            fields.add(createNormalField(tableName, "created"));
+
+            schema = new Schema();
+            schema.setOriginalName(tableName);
+            schema.setFields(fields);
+
+            graph = new Graph<>();
+        }
+
+        @Test
+        void testIgnoreWithoutPk() {
+            TableNode sourceNode = new TableNode();
+            sourceNode.setId(UUID.randomUUID().toString());
+            TableNode targetNode = new TableNode();
+            targetNode.setId(UUID.randomUUID().toString());
+
+            graph.setEdge(sourceNode.getId(), targetNode.getId());
+            targetNode.setGraph(graph);
+
+            schema.setHasPrimaryKey(true);
+            NoPrimaryKeyVirtualField.addVirtualField((Object) schema, (Node<?>) targetNode);
+            Assertions.assertEquals(0, fields.stream().filter(f -> NoPrimaryKeyVirtualField.FIELD_NAME.equals(f.getFieldName())).count());
+        }
+
+        @Test
+        void testIgnoreWithoutNotDataParentNode() {
+            TableNode sourceNode = new TableNode();
+            sourceNode.setId(UUID.randomUUID().toString());
+            CacheNode targetNode = new CacheNode();
+            targetNode.setId(UUID.randomUUID().toString());
+
+            graph.setEdge(sourceNode.getId(), targetNode.getId());
+            targetNode.setGraph(graph);
+
+            NoPrimaryKeyVirtualField.addVirtualField((Object) schema, (Node<?>) targetNode);
+            Assertions.assertEquals(0, fields.stream().filter(f -> NoPrimaryKeyVirtualField.FIELD_NAME.equals(f.getFieldName())).count());
+        }
+
+        @Test
+        void testAdd2TableNode() {
+            TableNode sourceNode = new TableNode();
+            sourceNode.setId(UUID.randomUUID().toString());
+            TableNode targetNode = new TableNode();
+            targetNode.setId(UUID.randomUUID().toString());
+            targetNode.setNoPkSyncMode(NoPrimaryKeySyncMode.ADD_HASH.name());
+
+            graph.setEdge(sourceNode.getId(), targetNode.getId());
+            graph.setNode(sourceNode.getId(), sourceNode);
+            graph.setNode(targetNode.getId(), targetNode);
+            sourceNode.setGraph(graph);
+            targetNode.setGraph(graph);
+
+            // not add to source
+            NoPrimaryKeyVirtualField.addVirtualField((Object) schema, (Node<?>) sourceNode);
+            Assertions.assertEquals(0, fields.stream().filter(f -> NoPrimaryKeyVirtualField.FIELD_NAME.equals(f.getFieldName())).count());
+            // not add with appoint conditions
+            targetNode.setUpdateConditionFields(Collections.singletonList("title"));
+            NoPrimaryKeyVirtualField.addVirtualField((Object) schema, (Node<?>) targetNode);
+            Assertions.assertEquals(0, fields.stream().filter(f -> NoPrimaryKeyVirtualField.FIELD_NAME.equals(f.getFieldName())).count());
+            targetNode.setUpdateConditionFields(new ArrayList<>());
+            // add to target
+            NoPrimaryKeyVirtualField.addVirtualField((Object) schema, (Node<?>) targetNode);
+            Assertions.assertEquals(1, fields.stream().filter(f -> NoPrimaryKeyVirtualField.FIELD_NAME.equals(f.getFieldName())).count());
+            // virtual filed exists
+            targetNode.setUpdateConditionFields(Collections.singletonList(NoPrimaryKeyVirtualField.FIELD_NAME));
+            NoPrimaryKeyVirtualField.addVirtualField((Object) schema, (Node<?>) targetNode);
+            Assertions.assertEquals(1, fields.stream().filter(f -> NoPrimaryKeyVirtualField.FIELD_NAME.equals(f.getFieldName())).count());
+        }
+
+        @Test
+        void testAdd2DatabaseNode() {
+            List<List<Schema>> schemas = Collections.singletonList(Collections.singletonList(schema));
+
+            DatabaseNode sourceNode = new DatabaseNode();
+            sourceNode.setId(UUID.randomUUID().toString());
+            DatabaseNode targetNode = new DatabaseNode();
+            targetNode.setId(UUID.randomUUID().toString());
+
+            graph.setEdge(sourceNode.getId(), targetNode.getId());
+            graph.setNode(sourceNode.getId(), sourceNode);
+            graph.setNode(targetNode.getId(), targetNode);
+            targetNode.setGraph(graph);
+            sourceNode.setGraph(graph);
+
+            // not add to source
+            NoPrimaryKeyVirtualField.addVirtualField(schemas, sourceNode);
+            Assertions.assertEquals(0, fields.stream().filter(f -> NoPrimaryKeyVirtualField.FIELD_NAME.equals(f.getFieldName())).count());
+            // not add with appendWrite
+            targetNode.setNoPkSyncMode(NoPrimaryKeySyncMode.ALL_COLUMNS.name());
+            NoPrimaryKeyVirtualField.addVirtualField(schemas, targetNode);
+            Assertions.assertEquals(0, fields.stream().filter(f -> NoPrimaryKeyVirtualField.FIELD_NAME.equals(f.getFieldName())).count());
+            targetNode.setNoPkSyncMode(NoPrimaryKeySyncMode.ADD_HASH.name());
+            // not add with appoint conditions
+            targetNode.setUpdateConditionFieldMap(new HashMap<>());
+            targetNode.getUpdateConditionFieldMap().put(schema.getName(), Collections.singletonList("title"));
+            NoPrimaryKeyVirtualField.addVirtualField(schemas, targetNode);
+            Assertions.assertEquals(0, fields.stream().filter(f -> NoPrimaryKeyVirtualField.FIELD_NAME.equals(f.getFieldName())).count());
+            targetNode.setUpdateConditionFieldMap(new HashMap<>());
+            // add to target
+            NoPrimaryKeyVirtualField.addVirtualField(schemas, targetNode);
+            Assertions.assertEquals(1, fields.stream().filter(f -> NoPrimaryKeyVirtualField.FIELD_NAME.equals(f.getFieldName())).count());
+            // virtual filed exists
+            NoPrimaryKeyVirtualField.addVirtualField(schemas, targetNode);
+            Assertions.assertEquals(1, fields.stream().filter(f -> NoPrimaryKeyVirtualField.FIELD_NAME.equals(f.getFieldName())).count());
+        }
+    }
+
+    @Nested
+    class hasPrimaryOrUniqueOrKeys {
+        Schema schema;
+
+        @BeforeEach
+        void setUp() {
+            schema = new Schema();
+            schema.setFields(new ArrayList<>());
+            schema.getFields().add(createNormalField("no_pk_test", "title"));
+        }
+
+        @Test
+        void testFieldNull() {
+            schema.setFields(null);
+            Assertions.assertFalse(NoPrimaryKeyVirtualField.hasPrimaryOrUniqueOrKeys(schema));
+        }
+
+        @Test
+        void testFalse() {
+            Assertions.assertFalse(NoPrimaryKeyVirtualField.hasPrimaryOrUniqueOrKeys(schema));
+        }
+
+        @Test
+        void testExistUk() {
+            Field filed = createNormalField("no_pk_test", "uk");
+            filed.setUnique(true);
+            schema.getFields().add(filed);
+            Assertions.assertTrue(NoPrimaryKeyVirtualField.hasPrimaryOrUniqueOrKeys(schema));
+
+            schema.setHasUnionIndex(true);
+            Assertions.assertTrue(NoPrimaryKeyVirtualField.hasPrimaryOrUniqueOrKeys(schema));
+        }
+
+        @Test
+        void testExistPk() {
+            Field filed = createNormalField("no_pk_test", "id", 1);
+            schema.getFields().add(filed);
+            Assertions.assertTrue(NoPrimaryKeyVirtualField.hasPrimaryOrUniqueOrKeys(schema));
+
+            schema.setHasPrimaryKey(true);
+            Assertions.assertTrue(NoPrimaryKeyVirtualField.hasPrimaryOrUniqueOrKeys(schema));
+        }
+    }
+
+    @Nested
+    class getVirtualHashFieldNames {
+        String schemaName;
+        MetadataInstancesDto metadata;
+
+        @BeforeEach
+        void setUp() {
+            schemaName = "no_pk_test";
+
+            metadata = new MetadataInstancesDto();
+            metadata.setOriginalName(schemaName);
+            metadata.setFields(new ArrayList<>());
+        }
+
+        @Test
+        void testFieldExistsWithMetadata() {
+            metadata.getFields().add(createVirualField(metadata.getName()));
+
+            List<String> fieldNames = NoPrimaryKeyVirtualField.getVirtualHashFieldNames(metadata);
+            Assertions.assertNotNull(fieldNames);
+            Assertions.assertEquals(1, fieldNames.size());
+            Assertions.assertTrue(fieldNames.contains(NoPrimaryKeyVirtualField.FIELD_NAME));
+        }
+
+        @Test
+        void testFieldNotExistsWithMetadata() {
+            String fieldName = NoPrimaryKeyVirtualField.FIELD_NAME + "_no";
+            metadata.getFields().add(createNormalField(metadata.getName(), fieldName));
+
+            List<String> fieldNames = NoPrimaryKeyVirtualField.getVirtualHashFieldNames(metadata);
+            Assertions.assertNotNull(fieldNames);
+            Assertions.assertTrue(fieldNames.isEmpty());
+        }
+
+        @Test
+        void testFieldExistsWithTapTable() {
+            metadata.getFields().add(createVirualField(metadata.getName()));
+
+            TapTable tapTable = PdkSchemaConvert.toPdk(metadata);
+            Collection<String> fieldNames = NoPrimaryKeyVirtualField.getVirtualHashFieldNames(tapTable);
+            Assertions.assertNotNull(fieldNames);
+            Assertions.assertEquals(1, fieldNames.size());
+            Assertions.assertTrue(fieldNames.contains(NoPrimaryKeyVirtualField.FIELD_NAME));
+        }
+
+        @Test
+        void testFieldNotExistsWithTapTable() {
+            String fieldName = NoPrimaryKeyVirtualField.FIELD_NAME + "_no";
+            metadata.getFields().add(createNormalField(metadata.getName(), fieldName));
+
+            TapTable tapTable = PdkSchemaConvert.toPdk(metadata);
+            Collection<String> fieldNames = NoPrimaryKeyVirtualField.getVirtualHashFieldNames(tapTable);
+            Assertions.assertNotNull(fieldNames);
+            Assertions.assertTrue(fieldNames.isEmpty());
+        }
+    }
+
+    @Nested
+    class instance {
+        String tableNameHashPk = "has_pk_table";
+        String tableNameHashPi = "has_pi_table";
+        String tableNameHashUk = "has_uk_table";
+        String tableNameNoPk = "no_pk_table";
+        NoPrimaryKeyVirtualField instance;
+
+        @BeforeEach
+        void setUp() {
+            instance = new NoPrimaryKeyVirtualField();
+            try (MockedStatic<NoPrimaryKeyVirtualField> mockedStatic = Mockito.mockStatic(NoPrimaryKeyVirtualField.class)) {
+                mockedStatic.when(() -> NoPrimaryKeyVirtualField.isEnable(Mockito.any())).thenReturn(true);
+                instance.init(null);
+            }
+
+            // add primary key table
+            instance.add(Optional.of(new Schema()).map(schema -> {
+                schema.setOriginalName(tableNameHashPk);
+                schema.setFields(new ArrayList<>());
+                schema.getFields().add(createNormalField(schema.getName(), "id", 1));
+                schema.getFields().add(createNormalField(schema.getName(), "title"));
+                return PdkSchemaConvert.toPdk(schema);
+            }).get());
+
+            // add primary index table
+            instance.add(Optional.of(new Schema()).map(schema -> {
+                schema.setOriginalName(tableNameHashPi);
+                schema.setIndices(new ArrayList<>());
+                schema.getIndices().add(Optional.of(new TableIndex()).map(index -> {
+                    TableIndexColumn tableIndexColumn = new TableIndexColumn();
+                    tableIndexColumn.setColumnName("title");
+
+                    index.setIndexName("hash_title");
+                    index.setColumns(new ArrayList<>());
+                    index.getColumns().add(tableIndexColumn);
+                    return index;
+                }).get());
+                schema.getIndices().add(Optional.of(new TableIndex()).map(index -> {
+                    TableIndexColumn tableIndexColumn = new TableIndexColumn();
+                    tableIndexColumn.setColumnName("id");
+
+                    index.setIndexName("pk_id");
+                    index.setColumns(new ArrayList<>());
+                    index.getColumns().add(tableIndexColumn);
+                    index.setPrimaryKey("true");
+                    return index;
+                }).get());
+                schema.setFields(new ArrayList<>());
+                schema.getFields().add(Optional.of(createNormalField(schema.getName(), "id")).map(field -> {
+                    field.setUnique(true);
+                    return field;
+                }).get());
+                schema.getFields().add(createNormalField(schema.getName(), "title"));
+                return PdkSchemaConvert.toPdk(schema);
+            }).get());
+
+            // add union key table
+            instance.add(Optional.of(new Schema()).map(schema -> {
+                schema.setOriginalName(tableNameHashUk);
+                schema.setHasUnionIndex(true);
+                schema.setIndices(new ArrayList<>());
+                schema.getIndices().add(Optional.of(new TableIndex()).map(index -> {
+                    TableIndexColumn tableIndexColumn = new TableIndexColumn();
+                    tableIndexColumn.setColumnName("title");
+
+                    index.setIndexName("hash_title");
+                    index.setColumns(new ArrayList<>());
+                    index.getColumns().add(tableIndexColumn);
+                    return index;
+                }).get());
+                schema.getIndices().add(Optional.of(new TableIndex()).map(index -> {
+                    TableIndexColumn tableIndexColumn = new TableIndexColumn();
+                    tableIndexColumn.setColumnName("id");
+
+                    index.setIndexName("uk_id");
+                    index.setColumns(new ArrayList<>());
+                    index.getColumns().add(tableIndexColumn);
+                    index.setUnique(true);
+                    return index;
+                }).get());
+                schema.setFields(new ArrayList<>());
+                schema.getFields().add(Optional.of(createNormalField(schema.getName(), "id")).map(field -> {
+                    field.setUnique(true);
+                    return field;
+                }).get());
+                schema.getFields().add(createNormalField(schema.getName(), "title"));
+                return PdkSchemaConvert.toPdk(schema);
+            }).get());
+
+            // add no pk table
+            instance.add(Optional.of(new Schema()).map(schema -> {
+                schema.setOriginalName(tableNameNoPk);
+                schema.setFields(new ArrayList<>());
+                schema.getFields().add(createNormalField(schema.getName(), "title"));
+                schema.setIndices(new ArrayList<>());
+                schema.getIndices().add(Optional.of(new TableIndex()).map(index -> {
+                    TableIndexColumn tableIndexColumn = new TableIndexColumn();
+                    tableIndexColumn.setColumnName("title");
+
+                    index.setIndexName("hash_title");
+                    index.setColumns(new ArrayList<>());
+                    index.getColumns().add(tableIndexColumn);
+                    return index;
+                }).get());
+                return PdkSchemaConvert.toPdk(schema);
+            }).get());
+        }
+
+        @Test
+        void testAdd() {
+            Assertions.assertFalse(instance.noPkTables.containsKey(tableNameHashPk));
+            Assertions.assertFalse(instance.noPkTables.containsKey(tableNameHashPi));
+            Assertions.assertFalse(instance.noPkTables.containsKey(tableNameHashUk));
+            Assertions.assertTrue(instance.noPkTables.containsKey(tableNameNoPk));
+        }
+
+        @Test
+        void testErrorIncompleteFields() {
+            TapUpdateRecordEvent recordEvent = TapUpdateRecordEvent.create()
+                .table(tableNameNoPk)
+                .before(Optional.of(new LinkedHashMap<String, Object>()).map(data -> {
+                    data.put("created", System.currentTimeMillis());
+                    return data;
+                }).get())
+                .after(Optional.of(new LinkedHashMap<String, Object>()).map(data -> {
+                    data.put("created", System.currentTimeMillis());
+                    return data;
+                }).get());
+            Assertions.assertThrows(NoPrimaryKeyException.class, () -> instance.addHashValue(recordEvent));
+            Assertions.assertFalse(instance.addHashValue(recordEvent));
+            Assertions.assertFalse(instance.addHashValue(TapDeleteRecordEvent.create()
+                .table(tableNameNoPk)
+                .before(Optional.of(new LinkedHashMap<String, Object>()).map(data -> {
+                    data.put("created", System.currentTimeMillis());
+                    return data;
+                }).get())
+            ));
+        }
+
+        @Test
+        void testErrorNotfoundHashAlgorithm() {
+            TapUpdateRecordEvent recordEvent = TapUpdateRecordEvent.create()
+                .table(tableNameNoPk)
+                .before(Optional.of(new LinkedHashMap<String, Object>()).map(data -> {
+                    data.put("created", System.currentTimeMillis());
+                    return data;
+                }).get())
+                .after(Optional.of(new LinkedHashMap<String, Object>()).map(data -> {
+                    data.put("created", System.currentTimeMillis());
+                    return data;
+                }).get());
+
+            try (MockedStatic<MessageDigest> mockedStatic = Mockito.mockStatic(MessageDigest.class)) {
+                mockedStatic.when(() -> MessageDigest.getInstance(Mockito.anyString())).thenThrow(new NoSuchAlgorithmException("XXX not found"));
+                Assertions.assertThrows(NoPrimaryKeyException.class, () -> instance.addHashValue(recordEvent));
+                Assertions.assertFalse(instance.addHashValue(recordEvent));
+                Assertions.assertFalse(instance.addHashValue(TapInsertRecordEvent.create()
+                    .table(tableNameNoPk)
+                    .after(Optional.of(new LinkedHashMap<String, Object>()).map(data -> {
+                        data.put("created", System.currentTimeMillis());
+                        return data;
+                    }).get())
+                ));
+                Assertions.assertFalse(instance.addHashValue(TapDeleteRecordEvent.create()
+                    .table(tableNameNoPk)
+                    .before(Optional.of(new LinkedHashMap<String, Object>()).map(data -> {
+                        data.put("created", System.currentTimeMillis());
+                        return data;
+                    }).get())
+                ));
+            }
+        }
+
+        @Test
+        void testOtherError() {
+            TapUpdateRecordEvent recordEvent = TapUpdateRecordEvent.create()
+                .table(tableNameNoPk)
+                .before(Optional.of(new LinkedHashMap<String, Object>()).map(data -> {
+                    data.put("created", System.currentTimeMillis());
+                    return data;
+                }).get())
+                .after(Optional.of(new LinkedHashMap<String, Object>()).map(data -> {
+                    data.put("created", System.currentTimeMillis());
+                    return data;
+                }).get());
+
+            try (MockedStatic<MessageDigest> mockedStatic = Mockito.mockStatic(MessageDigest.class)) {
+                mockedStatic.when(() -> MessageDigest.getInstance(Mockito.anyString())).thenThrow(new RuntimeException("others"));
+                Assertions.assertThrows(NoPrimaryKeyException.class, () -> instance.addHashValue(recordEvent));
+                Assertions.assertFalse(instance.addHashValue(recordEvent));
+                Assertions.assertFalse(instance.addHashValue(TapInsertRecordEvent.create()
+                    .table(tableNameNoPk)
+                    .after(Optional.of(new LinkedHashMap<String, Object>()).map(data -> {
+                        data.put("created", System.currentTimeMillis());
+                        return data;
+                    }).get())
+                ));
+                Assertions.assertFalse(instance.addHashValue(TapDeleteRecordEvent.create()
+                    .table(tableNameNoPk)
+                    .before(Optional.of(new LinkedHashMap<String, Object>()).map(data -> {
+                        data.put("created", System.currentTimeMillis());
+                        return data;
+                    }).get())
+                ));
+            }
+        }
+
+        @Nested
+        class addHashValue {
+
+            @Test
+            void testSkipTable() {
+                TapInsertRecordEvent recordEvent = TapInsertRecordEvent.create();
+                recordEvent.table(tableNameHashPk);
+                Assertions.assertTrue(instance.addHashValue(recordEvent));
+            }
+
+            @Test
+            void testInsertRecordEvent() {
+                TapInsertRecordEvent recordEvent = TapInsertRecordEvent.create();
+                recordEvent.table(tableNameNoPk);
+                recordEvent.setAfter(Optional.of(new LinkedHashMap<String, Object>()).map(data -> {
+                    data.put("title", "1");
+                    return data;
+                }).get());
+                Assertions.assertTrue(instance.addHashValue(recordEvent));
+            }
+
+            @Test
+            void testUpdateRecordEvent() {
+                TapUpdateRecordEvent recordEvent = TapUpdateRecordEvent.create();
+                recordEvent.table(tableNameNoPk);
+                recordEvent.setBefore(Optional.of(new LinkedHashMap<String, Object>()).map(data -> {
+                    data.put("title", "2");
+                    return data;
+                }).get());
+                recordEvent.setAfter(Optional.of(new LinkedHashMap<String, Object>()).map(data -> {
+                    data.put("title", "1");
+                    return data;
+                }).get());
+                Assertions.assertTrue(instance.addHashValue(recordEvent));
+            }
+
+            @Test
+            void testDeleteRecordEvent() {
+                TapDeleteRecordEvent recordEvent = TapDeleteRecordEvent.create();
+                recordEvent.table(tableNameNoPk);
+                recordEvent.setBefore(Optional.of(new LinkedHashMap<String, Object>()).map(data -> {
+                    data.put("title", "2");
+                    return data;
+                }).get());
+                Assertions.assertTrue(instance.addHashValue(recordEvent));
+            }
+
+            @Test
+            void testUnknownRecordEvent() {
+                TapUnknownRecordEvent recordEvent = TapUnknownRecordEvent.create();
+                recordEvent.setTableId(tableNameNoPk);
+                Assertions.assertTrue(instance.addHashValue(recordEvent));
+            }
+        }
+    }
+
+    @Nested
+    class allTypes {
+        String tableName = "no_pk_table";
+        NoPrimaryKeyVirtualField instance;
+        TapInsertRecordEvent recordEvent;
+
+        @BeforeEach
+        void setUp() {
+            instance = new NoPrimaryKeyVirtualField();
+            try (MockedStatic<NoPrimaryKeyVirtualField> mockedStatic = Mockito.mockStatic(NoPrimaryKeyVirtualField.class)) {
+                mockedStatic.when(() -> NoPrimaryKeyVirtualField.isEnable(Mockito.any())).thenReturn(true);
+                instance.init(null);
+            }
+            instance.add(Optional.of(new Schema()).map(schema -> {
+                schema.setOriginalName(tableName);
+                schema.setFields(new ArrayList<>());
+                schema.getFields().add(createNormalField(schema.getName(), "key"));
+                schema.getFields().add(createNormalField(schema.getName(), "val"));
+                return PdkSchemaConvert.toPdk(schema);
+            }).get());
+            recordEvent = TapInsertRecordEvent.create().table(tableName).after(new LinkedHashMap<>());
+        }
+
+        void assertType(String key, Object val) {
+            recordEvent.getAfter().put("key", key);
+            recordEvent.getAfter().put("val", val);
+            Assertions.assertTrue(instance.addHashValue(recordEvent));
+            Assertions.assertNotNull(recordEvent.getAfter().get(NoPrimaryKeyVirtualField.FIELD_NAME));
+        }
+
+        @Test
+        void testTypes() {
+            assertType("non", null);
+        }
+
+        @Test
+        void testBytes() {
+            assertType("bytes", "hello".getBytes());
+        }
+
+        @Test
+        void testString() {
+            assertType("string", "hello");
+        }
+
+        @Test
+        void testInteger() {
+            assertType("integer", 1);
+        }
+
+        @Test
+        void testFloat() {
+            assertType("float", 1.1f);
+        }
+
+        @Test
+        void testDouble() {
+            assertType("double", 1.1d);
+        }
+
+        @Test
+        void testArray() {
+            assertType("array", new String[]{"hello", "world"});
+        }
+
+        @Test
+        void testList() {
+            assertType("list", Arrays.asList("hello", "world"));
+        }
+
+        @Test
+        void testMap() {
+            assertType("map", new HashMap<String, Object>() {{
+                put("hello", "world");
+            }});
+        }
+    }
+
+    @Nested
+    class isEnable {
+        TableNode sourceNode;
+        TableNode targetNode;
+        Graph<Element, Element> graph;
+
+        @BeforeEach
+        void setUp() {
+            graph = new Graph<>();
+
+            sourceNode = new TableNode();
+            sourceNode.setId(UUID.randomUUID().toString());
+            targetNode = new TableNode();
+            targetNode.setId(UUID.randomUUID().toString());
+            targetNode.setNoPkSyncMode(NoPrimaryKeySyncMode.ADD_HASH.name());
+        }
+
+        void addNodes(Element... elements) {
+            String preId = null;
+            for (Element e : elements) {
+                if (null != preId) {
+                    graph.setEdge(preId, e.getId());
+                }
+                graph.setNode(e.getId(), e);
+                e.setGraph(graph);
+                preId = e.getId();
+            }
+        }
+
+        @Test
+        void testTrue() {
+            addNodes(sourceNode, targetNode);
+            Assertions.assertTrue(NoPrimaryKeyVirtualField.isEnable(graph));
+        }
+
+        @Test
+        void testFalseWithMode() {
+            targetNode.setNoPkSyncMode(NoPrimaryKeySyncMode.ALL_COLUMNS.name());
+            addNodes(sourceNode, targetNode);
+            Assertions.assertFalse(NoPrimaryKeyVirtualField.isEnable(graph));
+        }
+
+        @Test
+        void testFalseWithMergeNode() {
+            MergeTableNode mergeNode = new MergeTableNode();
+            mergeNode.setId(UUID.randomUUID().toString());
+            addNodes(sourceNode, mergeNode, targetNode);
+            Assertions.assertFalse(NoPrimaryKeyVirtualField.isEnable(graph));
+        }
+
+        @Test
+        void testFalseWithJoinNode() {
+            JoinProcessorNode joinNode = new JoinProcessorNode();
+            joinNode.setId(UUID.randomUUID().toString());
+            addNodes(sourceNode, joinNode, targetNode);
+            Assertions.assertFalse(NoPrimaryKeyVirtualField.isEnable(graph));
+        }
+    }
+}
diff --git a/manager/tm-oss/src/test/java/com/tapdata/tm/alarm/service/AlarmServiceImplTest.java b/manager/tm-oss/src/test/java/com/tapdata/tm/alarm/service/AlarmServiceImplTest.java
index 8da116754b..a1c844c9eb 100644
--- a/manager/tm-oss/src/test/java/com/tapdata/tm/alarm/service/AlarmServiceImplTest.java
+++ b/manager/tm-oss/src/test/java/com/tapdata/tm/alarm/service/AlarmServiceImplTest.java
@@ -10,6 +10,8 @@
 import org.mockito.MockedStatic;
 import org.mockito.Mockito;
 
+import java.util.ArrayList;
+
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.Mockito.mock;
 
@@ -54,10 +56,17 @@ void testAdd(){
     }
 
     @Test
-    void testCheckOpen(){
+    void testCheckOpen1(){
+        try(MockedStatic<MessageUtil> messageUtilMockedStatic = Mockito.mockStatic(MessageUtil.class)){
+            messageUtilMockedStatic.when(()->MessageUtil.getMessage(anyString())).thenReturn("error");
+            Assertions.assertThrows(BizException.class,()->alarmService.checkOpen(null,null,null,null,mock(UserDetail.class)));
+        }
+    }
+    @Test
+    void testCheckOpen3(){
         try(MockedStatic<MessageUtil> messageUtilMockedStatic = Mockito.mockStatic(MessageUtil.class)){
             messageUtilMockedStatic.when(()->MessageUtil.getMessage(anyString())).thenReturn("error");
-            Assertions.assertThrows(BizException.class,()->alarmService.checkOpen(null,null,null,null,null));
+            Assertions.assertThrows(BizException.class,()->alarmService.checkOpen(null,null,null,null,new ArrayList<>()));
         }
     }
 
diff --git a/manager/tm/src/test/java/com/tapdata/tm/Settings/service/SettingsServiceTest.java b/manager/tm/src/test/java/com/tapdata/tm/Settings/service/SettingsServiceTest.java
index 50fe3eebd5..8cfdb73972 100644
--- a/manager/tm/src/test/java/com/tapdata/tm/Settings/service/SettingsServiceTest.java
+++ b/manager/tm/src/test/java/com/tapdata/tm/Settings/service/SettingsServiceTest.java
@@ -189,7 +189,7 @@ void testFindALlWithPwd() {
             settings1.setKey("smtp.server.password");
             settings1.setValue("123456");
             Settings settings2 = new Settings();
-            settings2.setKey("ad.bind.password");
+            settings2.setKey("ldap.bind.password");
             settings2.setValue("12345");
             List<Settings> list = new ArrayList<>();
             list.add(settings1);
diff --git a/manager/tm/src/test/java/com/tapdata/tm/cluster/service/ClusterStateServiceTest.java b/manager/tm/src/test/java/com/tapdata/tm/cluster/service/ClusterStateServiceTest.java
index d429049473..cd8d426f2f 100644
--- a/manager/tm/src/test/java/com/tapdata/tm/cluster/service/ClusterStateServiceTest.java
+++ b/manager/tm/src/test/java/com/tapdata/tm/cluster/service/ClusterStateServiceTest.java
@@ -6,14 +6,17 @@
 import com.tapdata.tm.Settings.constant.SettingUtil;
 import com.tapdata.tm.Settings.service.SettingsService;
 import com.tapdata.tm.agent.service.AgentGroupService;
+import com.tapdata.tm.base.dto.Field;
 import com.tapdata.tm.cluster.dto.ClusterStateDto;
 import com.tapdata.tm.cluster.dto.SystemInfo;
 import com.tapdata.tm.clusterOperation.service.ClusterOperationService;
 import com.tapdata.tm.config.security.UserDetail;
 import com.tapdata.tm.message.service.MessageService;
 import com.tapdata.tm.worker.dto.TcmInfo;
+import com.tapdata.tm.worker.dto.WorkerDto;
 import com.tapdata.tm.worker.entity.Worker;
 import com.tapdata.tm.worker.service.WorkerService;
+import org.bson.types.ObjectId;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Nested;
@@ -28,14 +31,12 @@
 import java.util.List;
 import java.util.UUID;
 
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyInt;
 import static org.mockito.ArgumentMatchers.anyList;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.mockStatic;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.*;
 
 class ClusterStateServiceTest {
     ClusterStateService clusterStateService;
@@ -190,4 +191,37 @@ void testPingTimeMoreThan30Sec() {
                     1);
         }
     }
+    @Nested
+    class deleteClusterTest{
+        @Test
+        public void testDeleteCluster() {
+            Field field = new Field();
+            field.put("systemInfo", 1);
+            WorkerDto worker = new WorkerDto();
+            worker.setLicenseBind(true);
+            when(workerService.findByProcessId(anyString(), any(UserDetail.class), anyString())).thenReturn(worker);
+            ObjectId id = new ObjectId();
+            UserDetail user = mock(UserDetail.class);
+            when(clusterStateService.deleteById(id)).thenReturn(true);
+            doCallRealMethod().when(clusterStateService).deleteCluster(id, user);
+            boolean result = clusterStateService.deleteCluster(id, user);
+            assertTrue(result);
+
+            when(clusterStateService.findById(id, field)).thenReturn(null);
+            result = clusterStateService.deleteCluster(id, user);
+            assertTrue(result);
+
+            ClusterStateDto clusterStateDto = new ClusterStateDto();
+            SystemInfo systemInfo = mock(SystemInfo.class);
+            clusterStateDto.setSystemInfo(systemInfo);
+            when(systemInfo.getProcess_id()).thenReturn("123");
+            when(clusterStateService.findById(id, field)).thenReturn(clusterStateDto);
+            result = clusterStateService.deleteCluster(id, user);
+            assertFalse(result);
+
+            when(workerService.unbindByProcessId("123")).thenReturn(true);
+            result = clusterStateService.deleteCluster(id, user);
+            assertTrue(result);
+        }
+    }
 }
\ No newline at end of file
diff --git a/manager/tm/src/test/java/com/tapdata/tm/measurementServiceV2/service/MeasurementServiceV2ImplTest.java b/manager/tm/src/test/java/com/tapdata/tm/measurementServiceV2/service/MeasurementServiceV2ImplTest.java
index f32b8e715b..21f9bca4dc 100644
--- a/manager/tm/src/test/java/com/tapdata/tm/measurementServiceV2/service/MeasurementServiceV2ImplTest.java
+++ b/manager/tm/src/test/java/com/tapdata/tm/measurementServiceV2/service/MeasurementServiceV2ImplTest.java
@@ -4,10 +4,7 @@
 import com.tapdata.tm.monitor.dto.TableSyncStaticDto;
 import com.tapdata.tm.monitor.service.MeasurementServiceV2Impl;
 import com.tapdata.tm.task.service.TaskService;
-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.junit.jupiter.api.*;
 import org.springframework.data.mongodb.core.MongoTemplate;
 import org.springframework.data.mongodb.core.query.Criteria;
 import org.springframework.data.mongodb.core.query.Query;
@@ -40,29 +37,4 @@ void testTableNameIsNull(){
         }
 
     }
-    @Nested
-    class QuerySyncStatic{
-        MeasurementServiceV2Impl measurementServiceV2 = mock(MeasurementServiceV2Impl.class);
-        MongoTemplate mongoTemplate;
-        TaskService taskService;
-        @BeforeEach
-        void before(){
-            mongoTemplate = mock(MongoTemplate.class);
-            taskService = mock(TaskService.class);
-            ReflectionTestUtils.setField(measurementServiceV2,"mongoOperations",mongoTemplate);
-            ReflectionTestUtils.setField(measurementServiceV2,"taskService",taskService);
-        }
-        @Test
-        void test(){
-            TableSyncStaticDto tableSyncStaticDto = new TableSyncStaticDto("test",1,20,"test");
-            when(mongoTemplate.count(any(Query.class),anyString())).thenAnswer(invocationOnMock -> {
-                Query query = invocationOnMock.getArgument(0);
-                Assertions.assertTrue(query.getQueryObject().containsKey("tags.table"));
-                return 0L;
-            });
-            doCallRealMethod().when(measurementServiceV2).querySyncStatic(any(),any());
-            measurementServiceV2.querySyncStatic(tableSyncStaticDto,mock(UserDetail.class));
-        }
-
-    }
 }
diff --git a/manager/tm/src/test/java/com/tapdata/tm/metadatadefinition/service/MetadataDefinitionServiceTest.java b/manager/tm/src/test/java/com/tapdata/tm/metadatadefinition/service/MetadataDefinitionServiceTest.java
index 01e65c56cf..13995299b1 100644
--- a/manager/tm/src/test/java/com/tapdata/tm/metadatadefinition/service/MetadataDefinitionServiceTest.java
+++ b/manager/tm/src/test/java/com/tapdata/tm/metadatadefinition/service/MetadataDefinitionServiceTest.java
@@ -4,6 +4,7 @@
 import com.tapdata.tm.Settings.service.SettingsService;
 import com.tapdata.tm.agent.dto.GroupDto;
 import com.tapdata.tm.base.dto.Filter;
+import com.tapdata.tm.base.dto.Where;
 import com.tapdata.tm.config.security.UserDetail;
 import com.tapdata.tm.metadatadefinition.param.BatchUpdateParam;
 import com.tapdata.tm.metadatadefinition.repository.MetadataDefinitionRepository;
@@ -16,6 +17,7 @@
 import org.springframework.data.mongodb.core.query.Update;
 import org.springframework.test.util.ReflectionTestUtils;
 
+import java.util.HashMap;
 import java.util.List;
 
 import static org.mockito.ArgumentMatchers.any;
@@ -99,7 +101,31 @@ void testDass() {
             when(settingsService.isCloud()).thenReturn(false);
             Filter filter = new Filter();
             metadataDefinitionService.find(filter,mock(UserDetail.class));
+            verify(metadataDefinitionRepository,times(1)).findAll(any(Filter.class), any(UserDetail.class));
+        }
+        @Test
+        void testDassTag() {
+            when(settingsService.isCloud()).thenReturn(false);
+            Filter filter = new Filter();
+            Where where = new Where();
+            HashMap itemMap = new HashMap();
+            itemMap.put("item_type", "dataflow");
+            where.put("or", com.tapdata.tm.utils.Lists.of(itemMap));
+            filter.setWhere(where);
+            metadataDefinitionService.find(filter,mock(UserDetail.class));
             verify(metadataDefinitionRepository,times(1)).findAll(any(Filter.class));
         }
+        @Test
+        void testCloudTag() {
+            when(settingsService.isCloud()).thenReturn(true);
+            Filter filter = new Filter();
+            Where where = new Where();
+            HashMap itemMap = new HashMap();
+            itemMap.put("item_type", "dataflow");
+            where.put("or", com.tapdata.tm.utils.Lists.of(itemMap));
+            filter.setWhere(where);
+            metadataDefinitionService.find(filter,mock(UserDetail.class));
+            verify(metadataDefinitionRepository,times(1)).findAll(any(Filter.class), any(UserDetail.class));
+        }
     }
 }
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 f775b96b4c..c55ab2585d 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
@@ -1,8 +1,13 @@
 package com.tapdata.tm.monitor.service;
 
 
+import com.tapdata.tm.base.dto.Page;
+import com.tapdata.tm.commons.dag.DAG;
+import com.tapdata.tm.commons.task.dto.Dag;
 import com.tapdata.tm.commons.task.dto.TaskDto;
+import com.tapdata.tm.config.security.UserDetail;
 import com.tapdata.tm.monitor.constant.KeyWords;
+import com.tapdata.tm.monitor.dto.TableSyncStaticDto;
 import com.tapdata.tm.monitor.entity.MeasurementEntity;
 import com.tapdata.tm.monitor.vo.TableSyncStaticVo;
 import com.tapdata.tm.task.bean.TableStatusInfoDto;
@@ -23,8 +28,7 @@
 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.junit.jupiter.api.Assertions.*;
 import static org.mockito.ArgumentMatchers.*;
 import static org.mockito.Mockito.doCallRealMethod;
 import static org.mockito.Mockito.doNothing;
@@ -410,4 +414,207 @@ void testOneSampleAsc() {
         }
 
     }
+
+    @Nested
+    class QuerySyncStatic{
+        MeasurementServiceV2Impl measurementServiceV2 = mock(MeasurementServiceV2Impl.class);
+        MongoTemplate mongoTemplate;
+        TaskService taskService;
+        TableSyncStaticDto tableSyncStaticDto;
+        @BeforeEach
+        void before(){
+            mongoTemplate = mock(MongoTemplate.class);
+            taskService = mock(TaskService.class);
+            ReflectionTestUtils.setField(measurementServiceV2,"mongoOperations",mongoTemplate);
+            ReflectionTestUtils.setField(measurementServiceV2,"taskService",taskService);
+            tableSyncStaticDto = new TableSyncStaticDto("test",1,20,"test");
+        }
+        @Test
+        void test(){
+            when(mongoTemplate.count(any(Query.class),anyString())).thenAnswer(invocationOnMock -> {
+                Query query = invocationOnMock.getArgument(0);
+                Assertions.assertTrue(query.getQueryObject().containsKey("tags.table"));
+                return 0L;
+            });
+            doCallRealMethod().when(measurementServiceV2).querySyncStatic(any(),any());
+            measurementServiceV2.querySyncStatic(tableSyncStaticDto,mock(UserDetail.class));
+        }
+
+        @Test
+        void testQuerySyncStatic_SyncRate() {
+            TaskDto taskDto = new TaskDto();
+            taskDto.setId(new ObjectId("615c9f48f1d842b8b78bf9c8"));
+            taskDto.setDag(mock(DAG.class));
+
+            when(taskService.findOne(any(Query.class), any(UserDetail.class))).thenReturn(taskDto);
+            when(mongoTemplate.count(any(Query.class), eq(MeasurementEntity.COLLECTION_NAME))).thenReturn(5L);
+            MeasurementEntity measurementEntity = new MeasurementEntity();
+            Map<String, String> tags = new HashMap<>();
+            tags.put("table","table1");
+            measurementEntity.setTags(tags);
+            List<Sample> samples = new ArrayList<>();
+            Sample sample = mock(Sample.class);
+            samples.add(sample);
+            Map<String, Number> vs = new HashMap<>();
+            vs.put("snapshotRowTotal", -1);
+            vs.put("snapshotInsertRowTotal", 100);
+            when(sample.getVs()).thenReturn(vs);
+            measurementEntity.setSamples(samples);
+            when(mongoTemplate.find(any(Query.class), eq(MeasurementEntity.class), eq(MeasurementEntity.COLLECTION_NAME)))
+                    .thenReturn(Collections.singletonList(measurementEntity));
+
+            doCallRealMethod().when(measurementServiceV2).querySyncStatic(any(),any());
+            Page<TableSyncStaticVo> result = measurementServiceV2.querySyncStatic(tableSyncStaticDto,mock(UserDetail.class));
+
+            assertEquals(5, result.getTotal());
+            verify(taskService, times(1)).findOne(any(Query.class), any(UserDetail.class));
+            verify(mongoTemplate, times(1)).count(any(Query.class), eq(MeasurementEntity.COLLECTION_NAME));
+            verify(mongoTemplate, times(1)).find(any(Query.class), eq(MeasurementEntity.class), eq(MeasurementEntity.COLLECTION_NAME));
+        }
+
+        @Test
+        void testQuerySyncStatic_SyncRateIsZero() {
+            TaskDto taskDto = new TaskDto();
+            taskDto.setId(new ObjectId("615c9f48f1d842b8b78bf9c8"));
+            taskDto.setDag(mock(DAG.class));
+
+            when(taskService.findOne(any(Query.class), any(UserDetail.class))).thenReturn(taskDto);
+            when(mongoTemplate.count(any(Query.class), eq(MeasurementEntity.COLLECTION_NAME))).thenReturn(5L);
+            MeasurementEntity measurementEntity = new MeasurementEntity();
+            Map<String, String> tags = new HashMap<>();
+            tags.put("table","table1");
+            measurementEntity.setTags(tags);
+            List<Sample> samples = new ArrayList<>();
+            Sample sample = mock(Sample.class);
+            samples.add(sample);
+            Map<String, Number> vs = new HashMap<>();
+            vs.put("snapshotRowTotal", 1);
+            vs.put("snapshotInsertRowTotal", 0);
+            when(sample.getVs()).thenReturn(vs);
+            measurementEntity.setSamples(samples);
+            when(mongoTemplate.find(any(Query.class), eq(MeasurementEntity.class), eq(MeasurementEntity.COLLECTION_NAME)))
+                    .thenReturn(Collections.singletonList(measurementEntity));
+
+            doCallRealMethod().when(measurementServiceV2).querySyncStatic(any(),any());
+            Page<TableSyncStaticVo> result = measurementServiceV2.querySyncStatic(tableSyncStaticDto,mock(UserDetail.class));
+
+            assertEquals(5, result.getTotal());
+        }
+
+        @Test
+        void testQuerySyncStatic_SyncRateIsOne() {
+            TaskDto taskDto = new TaskDto();
+            taskDto.setId(new ObjectId("615c9f48f1d842b8b78bf9c8"));
+            taskDto.setDag(mock(DAG.class));
+
+            when(taskService.findOne(any(Query.class), any(UserDetail.class))).thenReturn(taskDto);
+            when(mongoTemplate.count(any(Query.class), eq(MeasurementEntity.COLLECTION_NAME))).thenReturn(5L);
+            MeasurementEntity measurementEntity = new MeasurementEntity();
+            Map<String, String> tags = new HashMap<>();
+            tags.put("table","table1");
+            measurementEntity.setTags(tags);
+            List<Sample> samples = new ArrayList<>();
+            Sample sample = mock(Sample.class);
+            samples.add(sample);
+            Map<String, Number> vs = new HashMap<>();
+            vs.put("snapshotRowTotal", 1);
+            vs.put("snapshotInsertRowTotal", 1);
+            when(sample.getVs()).thenReturn(vs);
+            measurementEntity.setSamples(samples);
+            when(mongoTemplate.find(any(Query.class), eq(MeasurementEntity.class), eq(MeasurementEntity.COLLECTION_NAME)))
+                    .thenReturn(Collections.singletonList(measurementEntity));
+
+            doCallRealMethod().when(measurementServiceV2).querySyncStatic(any(),any());
+            Page<TableSyncStaticVo> result = measurementServiceV2.querySyncStatic(tableSyncStaticDto,mock(UserDetail.class));
+
+            assertEquals(5, result.getTotal());
+        }
+
+        @Test
+        void testQuerySyncStatic_Sync() {
+            TaskDto taskDto = new TaskDto();
+            taskDto.setId(new ObjectId("615c9f48f1d842b8b78bf9c8"));
+            taskDto.setDag(mock(DAG.class));
+
+            when(taskService.findOne(any(Query.class), any(UserDetail.class))).thenReturn(taskDto);
+            when(mongoTemplate.count(any(Query.class), eq(MeasurementEntity.COLLECTION_NAME))).thenReturn(5L);
+            MeasurementEntity measurementEntity = new MeasurementEntity();
+            Map<String, String> tags = new HashMap<>();
+            tags.put("table","table1");
+            measurementEntity.setTags(tags);
+            List<Sample> samples = new ArrayList<>();
+            Sample sample = mock(Sample.class);
+            samples.add(sample);
+            Map<String, Number> vs = new HashMap<>();
+            vs.put("snapshotSyncRate", 0.5);
+            vs.put("snapshotRowTotal", 1);
+            vs.put("snapshotInsertRowTotal", 1);
+            when(sample.getVs()).thenReturn(vs);
+            measurementEntity.setSamples(samples);
+            when(mongoTemplate.find(any(Query.class), eq(MeasurementEntity.class), eq(MeasurementEntity.COLLECTION_NAME)))
+                    .thenReturn(Collections.singletonList(measurementEntity));
+
+            doCallRealMethod().when(measurementServiceV2).querySyncStatic(any(),any());
+            Page<TableSyncStaticVo> result = measurementServiceV2.querySyncStatic(tableSyncStaticDto,mock(UserDetail.class));
+
+            assertEquals(5, result.getTotal());
+        }
+        @Test
+        void testQuerySyncStatic() {
+            TaskDto taskDto = new TaskDto();
+            taskDto.setId(new ObjectId("615c9f48f1d842b8b78bf9c8"));
+            taskDto.setDag(mock(DAG.class));
+
+            when(taskService.findOne(any(Query.class), any(UserDetail.class))).thenReturn(taskDto);
+            when(mongoTemplate.count(any(Query.class), eq(MeasurementEntity.COLLECTION_NAME))).thenReturn(5L);
+            MeasurementEntity measurementEntity = new MeasurementEntity();
+            Map<String, String> tags = new HashMap<>();
+            tags.put("table","table1");
+            measurementEntity.setTags(tags);
+            List<Sample> samples = new ArrayList<>();
+            Sample sample = mock(Sample.class);
+            samples.add(sample);
+            Map<String, Number> vs = new HashMap<>();
+            vs.put("snapshotRowTotal", 0);
+            vs.put("snapshotInsertRowTotal", 1);
+            when(sample.getVs()).thenReturn(vs);
+            measurementEntity.setSamples(samples);
+            when(mongoTemplate.find(any(Query.class), eq(MeasurementEntity.class), eq(MeasurementEntity.COLLECTION_NAME)))
+                    .thenReturn(Collections.singletonList(measurementEntity));
+
+            doCallRealMethod().when(measurementServiceV2).querySyncStatic(any(),any());
+            Page<TableSyncStaticVo> result = measurementServiceV2.querySyncStatic(tableSyncStaticDto,mock(UserDetail.class));
+
+            assertEquals(5, result.getTotal());
+        }
+        @Test
+        void testQuerySyncStatic_snapshotRowTotal() {
+            TaskDto taskDto = new TaskDto();
+            taskDto.setId(new ObjectId("615c9f48f1d842b8b78bf9c8"));
+            taskDto.setDag(mock(DAG.class));
+
+            when(taskService.findOne(any(Query.class), any(UserDetail.class))).thenReturn(taskDto);
+            when(mongoTemplate.count(any(Query.class), eq(MeasurementEntity.COLLECTION_NAME))).thenReturn(5L);
+            MeasurementEntity measurementEntity = new MeasurementEntity();
+            Map<String, String> tags = new HashMap<>();
+            tags.put("table","table1");
+            measurementEntity.setTags(tags);
+            List<Sample> samples = new ArrayList<>();
+            Sample sample = mock(Sample.class);
+            samples.add(sample);
+            Map<String, Number> vs = new HashMap<>();
+            vs.put("snapshotRowTotal", -1);
+            vs.put("snapshotInsertRowTotal", 0);
+            when(sample.getVs()).thenReturn(vs);
+            measurementEntity.setSamples(samples);
+            when(mongoTemplate.find(any(Query.class), eq(MeasurementEntity.class), eq(MeasurementEntity.COLLECTION_NAME)))
+                    .thenReturn(Collections.singletonList(measurementEntity));
+
+            doCallRealMethod().when(measurementServiceV2).querySyncStatic(any(),any());
+            Page<TableSyncStaticVo> result = measurementServiceV2.querySyncStatic(tableSyncStaticDto,mock(UserDetail.class));
+
+            assertEquals(5, result.getTotal());
+        }
+    }
+
 }
diff --git a/manager/tm/src/test/java/com/tapdata/tm/task/service/impl/TaskRecordServiceImplTest.java b/manager/tm/src/test/java/com/tapdata/tm/task/service/impl/TaskRecordServiceImplTest.java
index 114c251cd8..02da6cd5d5 100644
--- a/manager/tm/src/test/java/com/tapdata/tm/task/service/impl/TaskRecordServiceImplTest.java
+++ b/manager/tm/src/test/java/com/tapdata/tm/task/service/impl/TaskRecordServiceImplTest.java
@@ -1,6 +1,11 @@
 package com.tapdata.tm.task.service.impl;
 
+import com.tapdata.tm.base.dto.Page;
+import com.tapdata.tm.task.bean.TaskRecordDto;
 import com.tapdata.tm.task.entity.TaskRecord;
+import com.tapdata.tm.task.vo.TaskRecordListVo;
+import com.tapdata.tm.user.service.UserService;
+import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Nested;
 import org.junit.jupiter.api.Test;
@@ -9,19 +14,26 @@
 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.Mockito.*;
 
 public class TaskRecordServiceImplTest {
+    TaskRecordServiceImpl taskRecordService;
+    MongoTemplate mongoTemplate;
+    UserService userService;
+    @BeforeEach
+    void init(){
+        mongoTemplate = mock(MongoTemplate.class);
+        taskRecordService = new TaskRecordServiceImpl();
+        ReflectionTestUtils.setField(taskRecordService,"mongoTemplate",mongoTemplate);
+        userService = mock(UserService.class);
+        ReflectionTestUtils.setField(taskRecordService,"userService",userService);
+    }
+
     @Nested
     class CleanTaskRecordTest{
-        TaskRecordServiceImpl taskRecordService;
-        MongoTemplate mongoTemplate;
-        @BeforeEach
-        void init(){
-            mongoTemplate = mock(MongoTemplate.class);
-            taskRecordService = new TaskRecordServiceImpl();
-            ReflectionTestUtils.setField(taskRecordService,"mongoTemplate",mongoTemplate);
-        }
 
         @Test
         void test_main(){
@@ -30,4 +42,22 @@ void test_main(){
             verify(mongoTemplate,times(1)).remove(eq(query), eq(TaskRecord.class));
         }
     }
+
+    @Nested
+    class QueryRecordsTest{
+
+        @Test
+        void test_main(){
+            TaskRecordDto dto = new TaskRecordDto();
+            dto.setTaskId("test");
+            dto.setPage(1);
+            dto.setSize(10);
+            when(mongoTemplate.count(eq(new Query(Criteria.where("taskId").is("test"))),eq(TaskRecord.class))).thenReturn(2L);
+            List<TaskRecord> taskRecords = new ArrayList<>();
+            when(mongoTemplate.find(any(),eq(TaskRecord.class))).thenReturn(taskRecords);
+            Page<TaskRecordListVo> result =  taskRecordService.queryRecords(dto);
+            Assertions.assertEquals(2L,result.getTotal());
+        }
+
+    }
 }