[Bug] Fix bug that can not repair replica in DECOMMISSION state

.github/ISSUE_TEMPLATE/bug_report.yml

@@ -25,7 +25,7 @@ body:
       value: |
         Thank you very much for submitting feedback to Doris to help Doris develop better.
 
-        If it is a problem with Doris, please go to the [Discussion](https://github.com/apache/incubator-doris/discussions)
+        If it is an idea or help wanted, please go to the [Discussion](https://github.com/apache/incubator-doris/discussions)
         or [Dev Mail List](mailto:dev@doris.apache.org)
 
   - type: checkboxes

fe/fe-core/src/main/java/org/apache/doris/catalog/Catalog.java

@@ -29,6 +29,7 @@
 import org.apache.doris.analysis.AddRollupClause;
 import org.apache.doris.analysis.AdminCheckTabletsStmt;
 import org.apache.doris.analysis.AdminCheckTabletsStmt.CheckType;
+import org.apache.doris.analysis.AdminCleanTrashStmt;
 import org.apache.doris.analysis.AdminSetConfigStmt;
 import org.apache.doris.analysis.AdminSetReplicaStatusStmt;
 import org.apache.doris.analysis.AlterClause;
@@ -43,7 +44,6 @@
 import org.apache.doris.analysis.CancelAlterSystemStmt;
 import org.apache.doris.analysis.CancelAlterTableStmt;
 import org.apache.doris.analysis.CancelBackupStmt;
-import org.apache.doris.analysis.AdminCleanTrashStmt;
 import org.apache.doris.analysis.ColumnRenameClause;
 import org.apache.doris.analysis.CreateClusterStmt;
 import org.apache.doris.analysis.CreateDbStmt;
@@ -105,9 +105,9 @@
 import org.apache.doris.cluster.Cluster;
 import org.apache.doris.cluster.ClusterNamespace;
 import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.ClientPool;
 import org.apache.doris.common.Config;
 import org.apache.doris.common.ConfigBase;
-import org.apache.doris.common.ClientPool;
 import org.apache.doris.common.DdlException;
 import org.apache.doris.common.ErrorCode;
 import org.apache.doris.common.ErrorReport;
@@ -224,12 +224,12 @@
 import org.apache.doris.task.DropReplicaTask;
 import org.apache.doris.task.MasterTaskExecutor;
 import org.apache.doris.thrift.BackendService;
+import org.apache.doris.thrift.TNetworkAddress;
 import org.apache.doris.thrift.TStorageFormat;
 import org.apache.doris.thrift.TStorageMedium;
 import org.apache.doris.thrift.TStorageType;
 import org.apache.doris.thrift.TTabletType;
 import org.apache.doris.thrift.TTaskType;
-import org.apache.doris.thrift.TNetworkAddress;
 import org.apache.doris.transaction.DbUsedDataQuotaInfoCollector;
 import org.apache.doris.transaction.GlobalTransactionMgr;
 import org.apache.doris.transaction.PublishVersionDaemon;
@@ -243,10 +243,14 @@
 import com.google.common.collect.Multimap;
 import com.google.common.collect.Queues;
 import com.google.common.collect.Sets;
+import com.sleepycat.je.rep.InsufficientLogException;
+import com.sleepycat.je.rep.NetworkRestore;
+import com.sleepycat.je.rep.NetworkRestoreConfig;
 
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.codehaus.jackson.map.ObjectMapper;
 
 import java.io.BufferedReader;
 import java.io.DataInputStream;
@@ -273,11 +277,6 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 
-import com.sleepycat.je.rep.InsufficientLogException;
-import com.sleepycat.je.rep.NetworkRestore;
-import com.sleepycat.je.rep.NetworkRestoreConfig;
-import org.codehaus.jackson.map.ObjectMapper;
-
 public class Catalog {
     private static final Logger LOG = LogManager.getLogger(Catalog.class);
     // 0 ~ 9999 used for qe
@@ -4080,7 +4079,7 @@ public static void getDdlStmt(DdlStmt ddlStmt, String dbName, Table table, List<
             sb.append(Joiner.on(", ").join(keysColumnNames)).append(")");
 
             if (!Strings.isNullOrEmpty(table.getComment())) {
-                sb.append("\nCOMMENT \"").append(table.getComment()).append("\"");
+                sb.append("\nCOMMENT \"").append(table.getComment(true)).append("\"");
             }
 
             // partition
@@ -4207,7 +4206,7 @@ public static void getDdlStmt(DdlStmt ddlStmt, String dbName, Table table, List<
         } else if (table.getType() == TableType.MYSQL) {
             MysqlTable mysqlTable = (MysqlTable) table;
             if (!Strings.isNullOrEmpty(table.getComment())) {
-                sb.append("\nCOMMENT \"").append(table.getComment()).append("\"");
+                sb.append("\nCOMMENT \"").append(table.getComment(true)).append("\"");
             }
             // properties
             sb.append("\nPROPERTIES (\n");
@@ -4225,7 +4224,7 @@ public static void getDdlStmt(DdlStmt ddlStmt, String dbName, Table table, List<
         } else if (table.getType() == TableType.ODBC) {
             OdbcTable odbcTable = (OdbcTable) table;
             if (!Strings.isNullOrEmpty(table.getComment())) {
-                sb.append("\nCOMMENT \"").append(table.getComment()).append("\"");
+                sb.append("\nCOMMENT \"").append(table.getComment(true)).append("\"");
             }
             // properties
             sb.append("\nPROPERTIES (\n");
@@ -4245,7 +4244,7 @@ public static void getDdlStmt(DdlStmt ddlStmt, String dbName, Table table, List<
         } else if (table.getType() == TableType.BROKER) {
             BrokerTable brokerTable = (BrokerTable) table;
             if (!Strings.isNullOrEmpty(table.getComment())) {
-                sb.append("\nCOMMENT \"").append(table.getComment()).append("\"");
+                sb.append("\nCOMMENT \"").append(table.getComment(true)).append("\"");
             }
             // properties
             sb.append("\nPROPERTIES (\n");
@@ -4263,7 +4262,7 @@ public static void getDdlStmt(DdlStmt ddlStmt, String dbName, Table table, List<
         } else if (table.getType() == TableType.ELASTICSEARCH) {
             EsTable esTable = (EsTable) table;
             if (!Strings.isNullOrEmpty(table.getComment())) {
-                sb.append("\nCOMMENT \"").append(table.getComment()).append("\"");
+                sb.append("\nCOMMENT \"").append(table.getComment(true)).append("\"");
             }
 
             // partition
@@ -4299,7 +4298,7 @@ public static void getDdlStmt(DdlStmt ddlStmt, String dbName, Table table, List<
         } else if (table.getType() == TableType.HIVE) {
             HiveTable hiveTable = (HiveTable) table;
             if (!Strings.isNullOrEmpty(table.getComment())) {
-                sb.append("\nCOMMENT \"").append(table.getComment()).append("\"");
+                sb.append("\nCOMMENT \"").append(table.getComment(true)).append("\"");
             }
             // properties
             sb.append("\nPROPERTIES (\n");

fe/fe-core/src/main/java/org/apache/doris/catalog/Column.java

@@ -27,6 +27,7 @@
 import org.apache.doris.common.FeMetaVersion;
 import org.apache.doris.common.io.Text;
 import org.apache.doris.common.io.Writable;
+import org.apache.doris.common.util.SqlUtils;
 import org.apache.doris.persist.gson.GsonUtils;
 import org.apache.doris.thrift.TColumn;
 import org.apache.doris.thrift.TColumnType;
@@ -298,7 +299,14 @@ public void setComment(String comment) {
     }
 
     public String getComment() {
-        return comment;
+        return getComment(false);
+    }
+
+    public String getComment(boolean escapeQuota) {
+        if (!escapeQuota) {
+            return comment;
+        }
+        return SqlUtils.escapeQuota(comment);
     }
 
     public int getOlapColumnIndexSize() {
@@ -488,7 +496,7 @@ public String toSql(boolean isUniqueTable) {
         if (defaultValue != null && getDataType() != PrimitiveType.HLL && getDataType() != PrimitiveType.BITMAP) {
             sb.append("DEFAULT \"").append(defaultValue).append("\" ");
         }
-        sb.append("COMMENT \"").append(comment).append("\"");
+        sb.append("COMMENT \"").append(getComment(true)).append("\"");
 
         return sb.toString();
     }

fe/fe-core/src/main/java/org/apache/doris/catalog/Table.java

@@ -21,18 +21,19 @@
 import org.apache.doris.common.FeMetaVersion;
 import org.apache.doris.common.io.Text;
 import org.apache.doris.common.io.Writable;
+import org.apache.doris.common.util.SqlUtils;
 import org.apache.doris.common.util.Util;
 import org.apache.doris.thrift.TTableDescriptor;
 
-import org.apache.commons.lang.NotImplementedException;
-import org.apache.logging.log4j.LogManager;
-import org.apache.logging.log4j.Logger;
-
 import com.google.common.base.Preconditions;
 import com.google.common.base.Strings;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 
+import org.apache.commons.lang.NotImplementedException;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
 import java.io.DataInput;
 import java.io.DataOutput;
 import java.io.IOException;
@@ -396,8 +397,15 @@ public String getMysqlType() {
     }
 
     public String getComment() {
+        return getComment(false);
+    }
+
+    public String getComment(boolean escapeQuota) {
         if (!Strings.isNullOrEmpty(comment)) {
-            return comment;
+            if (!escapeQuota) {
+                return comment;
+            }
+            return SqlUtils.escapeQuota(comment);
         }
         return type.name();
     }

fe/fe-core/src/main/java/org/apache/doris/catalog/Tablet.java

@@ -201,8 +201,7 @@ public Multimap<Long, Long> getNormalReplicaBackendPathMap() {
             }
 
             ReplicaState state = replica.getState();
-            if (infoService.checkBackendAlive(replica.getBackendId())
-                    && (state == ReplicaState.NORMAL || state == ReplicaState.ALTER)) {
+            if (infoService.checkBackendAlive(replica.getBackendId()) && state.canLoad()) {
                 map.put(replica.getBackendId(), replica.getPathHash());
             }
         }

fe/fe-core/src/main/java/org/apache/doris/clone/ColocateTableCheckerAndBalancer.java

@@ -382,7 +382,8 @@ private boolean relocateAndBalance(GroupId groupId, Set<Long> unavailableBeIds,
                     if (!backendsSet.contains(destBeId) && !hostsSet.contains(destBe.getHost())) {
                         Preconditions.checkState(backendsSet.contains(srcBeId), srcBeId);
                         flatBackendsPerBucketSeq.set(seqIndex, destBeId);
-                        LOG.info("replace backend {} with backend {} in colocate group {}", srcBeId, destBeId, groupId);
+                        LOG.info("replace backend {} with backend {} in colocate group {}, idx: {}",
+                                srcBeId, destBeId, groupId, seqIndex);
                         // just replace one backend at a time, src and dest BE id should be recalculated because
                         // flatBackendsPerBucketSeq is changed.
                         isChanged = true;

fe/fe-core/src/main/java/org/apache/doris/clone/TabletSchedCtx.java

@@ -592,12 +592,31 @@ public void chooseDestReplicaForVersionIncomplete(Map<Long, PathSlot> backendsWo
         if (slot == null) {
             throw new SchedException(Status.SCHEDULE_FAILED, "backend of dest replica is missing");
         }
-        
+
         long destPathHash = slot.takeSlot(chosenReplica.getPathHash());
         if (destPathHash == -1) {
             throw new SchedException(Status.SCHEDULE_FAILED, "unable to take slot of dest path");
         }
-
+
+        if (chosenReplica.getState() == ReplicaState.DECOMMISSION) {
+            // Since this replica is selected as the repair object of VERSION_INCOMPLETE,
+            // it means that this replica needs to be able to accept loading data.
+            // So if this replica was previously set to DECOMMISSION, this state needs to be reset to NORMAL.
+            // It may happen as follows:
+            // 1. A tablet of colocation table is in COLOCATION_REDUNDANT state
+            // 2. The tablet is being scheduled and set one of replica as DECOMMISSION in TabletScheduler.deleteReplicaInternal()
+            // 3. The tablet will then be scheduled again
+            // 4. But at that time, the BE node of the replica that was
+            //    set to the DECOMMISSION state in step 2 is returned to the colocation group.
+            //    So the tablet's health status becomes VERSION_INCOMPLETE.
+            //
+            // If we do not reset this replica state to NORMAL, the tablet's health status will be in VERSION_INCOMPLETE
+            // forever, because the replica in the DECOMMISSION state will not receive the load task.
+            chosenReplica.setWatermarkTxnId(-1);
+            chosenReplica.setState(ReplicaState.NORMAL);
+            LOG.info("choose replica {} on backend {} of tablet {} as dest replica for version incomplete," +
+                    " and change state from DECOMMISSION to NORMAL", chosenReplica.getId(), chosenReplica.getBackendId(), tabletId);
+        }
         setDest(chosenReplica.getBackendId(), chosenReplica.getPathHash());
     }
 

fe/fe-core/src/main/java/org/apache/doris/clone/TabletScheduler.java

@@ -559,22 +559,22 @@ private void handleTabletByTypeAndStatus(TabletStatus status, TabletSchedCtx tab
             throws SchedException {
         if (tabletCtx.getType() == Type.REPAIR) {
             switch (status) {
-            case REPLICA_MISSING:
-                handleReplicaMissing(tabletCtx, batchTask);
-                break;
-            case VERSION_INCOMPLETE:
-            case NEED_FURTHER_REPAIR: // same as version incomplete, it prefer to the dest replica which need further repair
-                handleReplicaVersionIncomplete(tabletCtx, batchTask);
-                break;
-            case REPLICA_RELOCATING:
-                handleReplicaRelocating(tabletCtx, batchTask);
-                break;
-            case REDUNDANT:
-                handleRedundantReplica(tabletCtx, false);
-                break;
-            case FORCE_REDUNDANT:
-                handleRedundantReplica(tabletCtx, true);
-                break;
+                case REPLICA_MISSING:
+                    handleReplicaMissing(tabletCtx, batchTask);
+                    break;
+                case VERSION_INCOMPLETE:
+                case NEED_FURTHER_REPAIR: // same as version incomplete, it prefer to the dest replica which need further repair
+                    handleReplicaVersionIncomplete(tabletCtx, batchTask);
+                    break;
+                case REPLICA_RELOCATING:
+                    handleReplicaRelocating(tabletCtx, batchTask);
+                    break;
+                case REDUNDANT:
+                    handleRedundantReplica(tabletCtx, false);
+                    break;
+                case FORCE_REDUNDANT:
+                    handleRedundantReplica(tabletCtx, true);
+                    break;
                 case REPLICA_MISSING_IN_CLUSTER:
                     handleReplicaClusterMigration(tabletCtx, batchTask);
                     break;

fe/fe-core/src/main/java/org/apache/doris/common/util/SqlUtils.java

@@ -17,6 +17,8 @@
 
 package org.apache.doris.common.util;
 
+import org.apache.parquet.Strings;
+
 public class SqlUtils {
     public static String escapeUnquote(String ident) {
         return ident.replaceAll("``", "`");
@@ -35,4 +37,11 @@ public static String getIdentSql(String ident) {
         sb.append('`');
         return sb.toString();
     }
+
+    public static String escapeQuota(String str) {
+        if (Strings.isNullOrEmpty(str)) {
+            return str;
+        }
+        return str.replaceAll("\"", "\\\\\"");
+    }
 }