[fix][dingo-exec] Interrupt with infinite regionSplit exception

dingodb · Nov 4, 2024 · 5a5e0c8 · 5a5e0c8
1 parent df6ab2f
commit 5a5e0c8
Show file tree

Hide file tree

Showing 3 changed files with 26 additions and 1 deletion.
diff --git a/dingo-calcite/src/main/java/io/dingodb/calcite/executor/ShowIndexFromTableExecutor.java b/dingo-calcite/src/main/java/io/dingodb/calcite/executor/ShowIndexFromTableExecutor.java
@@ -21,6 +21,7 @@
 import io.dingodb.common.meta.SchemaState;
 import io.dingodb.meta.DdlService;
 import io.dingodb.meta.entity.Column;
+import io.dingodb.meta.entity.IndexTable;
 import io.dingodb.meta.entity.InfoSchema;
 import io.dingodb.meta.entity.Table;
 
@@ -95,7 +96,11 @@ public Object[] getIndexCol(Table index, String columnName, int seqIndex, boolea
         Object[] val = new Object[16];
         val[0] = tableName;
         val[1] = unique ? "1" : "0";
-        val[2] = index.getName();
+        if (index instanceof IndexTable) {
+            val[2] = index.getName();
+        } else {
+            val[2] = "PRIMARY";
+        }
         val[3] = seqIndex;
         val[4] = columnName;
         val[5] = 'A';

diff --git a/dingo-exec/src/main/java/io/dingodb/exec/operator/NewCalcDistributionOperator.java b/dingo-exec/src/main/java/io/dingodb/exec/operator/NewCalcDistributionOperator.java
@@ -164,6 +164,21 @@ private static CompletableFuture<Boolean> push(
             .exceptionally(ex -> {
                 if (ex != null) {
                     if (ex.getCause() instanceof RegionSplitException) {
+                        int retry;
+                        if (param.getSplitRetry().containsKey(distribution.getId())) {
+                            int retryCnt = param.getSplitRetry().get(distribution.getId());
+                            retry = retryCnt + 1;
+                        } else {
+                            retry = 1;
+                        }
+                        if (retry > 50) {
+                            MetaService.root().invalidateDistribution(param.getTd().getTableId());
+                        }
+                        if (retry > 100) {
+                            LogUtils.error(log, ex.getMessage(), ex);
+                            throw new RuntimeException("The number of split retries exceeds the maximum limit");
+                        }
+                        param.getSplitRetry().put(distribution.getId(), retry);
                         NavigableMap<ByteArrayUtils.ComparableByteArray, RangeDistribution> tmpDistribution =
                             MetaService.root().getRangeDistribution(param.getTd().getTableId());
                         DistributionSourceParam copyParam = param.copy(

diff --git a/dingo-exec/src/main/java/io/dingodb/exec/operator/params/DistributionSourceParam.java b/dingo-exec/src/main/java/io/dingodb/exec/operator/params/DistributionSourceParam.java
@@ -19,6 +19,7 @@
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.annotation.JsonPropertyOrder;
 import com.fasterxml.jackson.annotation.JsonTypeName;
+import io.dingodb.common.CommonId;
 import io.dingodb.common.partition.RangeDistribution;
 import io.dingodb.common.util.ByteArrayUtils;
 import io.dingodb.common.util.Optional;
@@ -30,7 +31,9 @@
 import lombok.Getter;
 import lombok.Setter;
 
+import java.util.Map;
 import java.util.NavigableMap;
+import java.util.concurrent.ConcurrentHashMap;
 
 @Getter
 @JsonTypeName("distributionSource")
@@ -62,6 +65,8 @@ public class DistributionSourceParam extends SourceParam {
     private int keepOrder;
     @JsonProperty("concurrencyLevel")
     private final int concurrencyLevel;
+    @Setter
+    private Map<CommonId, Integer> splitRetry = new ConcurrentHashMap<>();
 
     public DistributionSourceParam(
         Table td,