[fix][dingo-calcite] Modifying column names leads to data loss

Author: guojn1

dingo-calcite/src/main/java/io/dingodb/calcite/DingoDdlExecutor.java

@@ -1998,6 +1998,9 @@ public static int getReplica(int targetReplica, int type) {
 
     private static void validateAddColumn(ColumnDefinition newColumn) {
         DingoType type = newColumn.getType();
+        if ("NULL".equalsIgnoreCase(newColumn.getDefaultValue())) {
+            newColumn.setDefaultValue(null);
+        }
         if (newColumn.getDefaultValue() == null) {
             if (!newColumn.isNullable()) {
                 if (type instanceof StringType) {

dingo-executor/src/main/java/io/dingodb/server/executor/ddl/IndexAddFiller.java

@@ -362,17 +362,6 @@ public boolean commitSecond() {
                 param.setPartId(newPartId);
                 param.setTableId(tableId);
             }
-            //String key = Base64.getEncoder().encodeToString(keyValue.getKey());
-            //if (!caches.containsKey(key)) {
-            //    caches.put(key, keyValue);
-            //}
-            //if (caches.size() % max_pre_write_count == 0) {
-            //    try {
-            //        commitSecondData(caches.values());
-            //    } finally {
-            //        caches.clear();
-            //    }
-            //}
         }
         if (!param.getKeys().isEmpty()) {
             boolean result = txnCommit(param, txnId, param.getTableId(), param.getPartId());
@@ -518,56 +507,6 @@ private Iterator<KeyValue> getLocalIterator() {
         return cache.scan(prefix);
     }
 
-    public void commitSecondData(Collection<KeyValue> secondData) {
-        CommitParam param = new CommitParam(dingoType, isolationLevel, txnId.seq,
-            commitTs, primaryKey, TransactionType.OPTIMISTIC);
-        param.init(null);
-        for (KeyValue keyValue : secondData) {
-            CommonId tableId = indexTable.tableId;
-            int from = 1;
-            //Arrays.copyOfRange(keyValue.getKey(), from, from += CommonId.LEN);
-            from += CommonId.LEN;
-            CommonId newPartId = CommonId.decode(Arrays.copyOfRange(keyValue.getKey(), from, from += CommonId.LEN));
-            byte[] key = new byte[keyValue.getKey().length - from];
-            System.arraycopy(keyValue.getKey(), from , key, 0, key.length);
-            CommonId partId = param.getPartId();
-            if (partId == null) {
-                partId = newPartId;
-                param.setPartId(partId);
-                param.setTableId(tableId);
-                param.addKey(key);
-            } else if (partId.equals(newPartId)) {
-                param.addKey(key);
-                if (param.getKeys().size() == max_pre_write_count) {
-                    boolean result = txnCommit(param, txnId, tableId, partId);
-                    if (!result) {
-                        throw new RuntimeException(txnId + " " + partId + ",txnCommit false,PrimaryKey:"
-                            + Arrays.toString(param.getPrimaryKey()));
-                    }
-                    param.getKeys().clear();
-                    param.setPartId(null);
-                }
-            } else {
-                boolean result = txnCommit(param, txnId, param.getTableId(), partId);
-                if (!result) {
-                    throw new RuntimeException(txnId + " " + partId + ",txnCommit false,PrimaryKey:"
-                        + Arrays.toString(param.getPrimaryKey()));
-                }
-                param.getKeys().clear();
-                param.addKey(key);
-                param.setPartId(newPartId);
-                param.setTableId(tableId);
-            }
-        }
-        if (!param.getKeys().isEmpty()) {
-            boolean result = txnCommit(param, txnId, param.getTableId(), param.getPartId());
-            if (!result) {
-                throw new RuntimeException(txnId + " " + param.getPartId()
-                    + ",txnCommit false,PrimaryKey:" + Arrays.toString(param.getPrimaryKey()));
-            }
-        }
-    }
-
     public TxnLocalData getTxnLocalData(Object[] tuplesTmp) {
         KeyValue keyValue = wrap(indexCodec::encode).apply(tuplesTmp);
         NavigableMap<ByteArrayUtils.ComparableByteArray, RangeDistribution> ranges =

dingo-executor/src/main/java/io/dingodb/server/executor/ddl/ModifyColumnFiller.java

@@ -47,6 +47,7 @@
 import java.util.Iterator;
 import java.util.NavigableMap;
 import java.util.stream.Collectors;
+import java.util.stream.IntStream;
 
 import static io.dingodb.common.CommonId.CommonType.FILL_BACK;
 import static io.dingodb.common.util.NoBreakFunctions.wrap;
@@ -84,6 +85,9 @@ public boolean preWritePrimary(ReorgBackFillTask task) {
         columnIndices = table.getColumnIndices(indexTable.columns.stream()
             .map(Column::getName)
             .collect(Collectors.toList()));
+        if (columnIndices.contains(-1)) {
+            columnIndices = IntStream.range(0, table.getColumns().size()).boxed().collect(Collectors.toList());
+        }
         colLen = columnIndices.size();
         indexCodec = CodecService.getDefault()
             .createKeyValueCodec(indexTable.getCodecVersion(), indexTable.version, indexTable.tupleType(),