[fix][dingo-executor] Add back up safe point ignore heartbeat transaction

Author: githubgxll

dingo-common/src/main/java/io/dingodb/common/mysql/scope/ScopeVariables.java

@@ -244,6 +244,16 @@ public static int lookupBatchSize() {
         }
     }
 
+    // for DTS
+    public static boolean ignoreHeartBeatTxn() {
+        try {
+            String autocommitSwitch = executorProp.getOrDefault("ignore_heartbeat_txn", "true").toString();
+            return "true".equalsIgnoreCase(autocommitSwitch);
+        } catch (Exception e) {
+            return false;
+        }
+    }
+
     public static synchronized void setExecutorProp(String key, String val) {
         if ("rpc_batch_size".equalsIgnoreCase(key)) {
             int rpcBatchSize = Integer.parseInt(val);

dingo-exec/src/main/java/io/dingodb/exec/transaction/impl/TransactionManager.java

@@ -31,7 +31,9 @@
 import lombok.extern.slf4j.Slf4j;
 import org.checkerframework.checker.nullness.qual.NonNull;
 
+import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
 
@@ -206,4 +208,21 @@ public static Object getIndex(CommonId txnId, CommonId indexId) {
     public static long getMinTs() {
         return trans.keySet().stream().mapToLong(txnId -> txnId.seq).min().orElse(Long.MAX_VALUE);
     }
+
+    public static long getIgnoreHeartBeatMinTs() {
+        long minTs = Long.MAX_VALUE;
+        Set<CommonId> txnIds = trans.keySet();
+        for (CommonId txnId : txnIds) {
+            if (txnId.seq < minTs) {
+                ITransaction iTransaction = trans.get(txnId);
+                List<String> sqlList = iTransaction.getSqlList();
+                if (!sqlList.isEmpty() && sqlList.stream().allMatch("select 1"::equals)) {
+                    LogUtils.info(log, "ignore heartbeat sql, txnId:{}", txnId);
+                    continue;
+                }
+                minTs = txnId.seq;
+            }
+        }
+        return minTs;
+    }
 }

dingo-store-proxy/src/main/java/io/dingodb/store/proxy/common/Gc.java

@@ -21,6 +21,7 @@
 import io.dingodb.common.config.DingoConfiguration;
 import io.dingodb.common.log.LogUtils;
 import io.dingodb.common.meta.Tenant;
+import io.dingodb.common.mysql.scope.ScopeVariables;
 import io.dingodb.common.session.Session;
 import io.dingodb.common.session.SessionUtil;
 import io.dingodb.common.tenant.TenantConstant;
@@ -110,7 +111,7 @@ public static Pair<String, Long> safePointUpdate() {
             LogUtils.info(log, "Run safe point update task.");
             Set<Location> coordinators = coordinatorSet();
             long reqTs = tso();
-            long safeTs = safeTs(getTxnDurationSafeTs(reqTs));
+            long safeTs = safeTs(getTxnDurationSafeTs(reqTs), false);
             List<Region> regions = getRegions(coordinators, reqTs);
             LogUtils.info(log, "Run safe point update task, current ts: {}, safe ts: {}", reqTs, safeTs);
             for (Region region : regions) {
@@ -228,7 +229,7 @@ public static GcObj startBackUpSafeByPoint(long point, long latestTso) {
         try {
             LogUtils.info(log, "Run back up safe point update task.");
             Set<Location> coordinators = coordinatorSet();
-            long safeTs = safeTs(point);
+            long safeTs = safeTs(point, ScopeVariables.ignoreHeartBeatTxn());
             List<Region> regions = getRegions(coordinators, latestTso);
             LogUtils.info(log, "Run back up safe point update task, current ts: {}, safe ts: {}",
                 latestTso, safeTs);
@@ -450,7 +451,7 @@ private static long getTxnDurationSafeTs(long requestId) {
         return requestId - (TimeUnit.SECONDS.toMillis(duration) << GcApi.PHYSICAL_SHIFT);
     }
 
-    private static long safeTs(long safeTs) {
+    private static long safeTs(long safeTs, boolean ignoreHeartBeat) {
         Integer retry = Optional.mapOrGet(
             DingoConfiguration.instance().find("retry", int.class),
             __ -> __,
@@ -469,7 +470,7 @@ private static long safeTs(long safeTs) {
                 LogUtils.warn(log, "Cross node get remote min start ts failed, retry times: {}", retry, e);
             }
         }
-        long localMinTs = TransactionManager.getMinTs();
+        long localMinTs = ignoreHeartBeat ? TransactionManager.getIgnoreHeartBeatMinTs() : TransactionManager.getMinTs();
         long minTxnTs = Math.min(remoteMinStartTs, localMinTs);
 
         return Math.min(minTxnTs, safeTs);