branch-4.0: [fix](fe) Fix colocate replica allocation edit log #65457

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

@@ -664,7 +664,7 @@ public void replayModifyReplicaAlloc(ColocatePersistInfo info) throws UserExcept
         writeLock();
         try {
             modifyColocateGroupReplicaAllocation(info.getGroupId(), info.getReplicaAlloc(),
-                    info.getBackendsPerBucketSeq(), false);
+                    info.getBackendsPerBucketSeq(), true /* isReplay */);
         } finally {
             writeUnlock();
         }
@@ -868,7 +868,7 @@ public void alterColocateGroup(AlterColocateGroupCommand command) throws UserExc
                 backendsPerBucketSeq = newBackendsPerBucketSeq;
                 Preconditions.checkState(backendsPerBucketSeq.size() == replicaAlloc.getAllocMap().size());
                 modifyColocateGroupReplicaAllocation(groupSchema.getGroupId(), replicaAlloc,
-                        backendsPerBucketSeq, true);
+                        backendsPerBucketSeq, false /* isReplay */);
             } else {
                 throw new DdlException("Unknown colocate group property: " + properties.keySet());
             }

fe/fe-core/src/test/java/org/apache/doris/catalog/ColocateTableTest.java

@@ -21,12 +21,15 @@
 import org.apache.doris.common.DdlException;
 import org.apache.doris.common.jmockit.Deencapsulation;
 import org.apache.doris.nereids.parser.NereidsParser;
+import org.apache.doris.nereids.trees.plans.commands.AlterColocateGroupCommand;
 import org.apache.doris.nereids.trees.plans.commands.AlterTableCommand;
 import org.apache.doris.nereids.trees.plans.commands.CreateDatabaseCommand;
 import org.apache.doris.nereids.trees.plans.commands.CreateTableCommand;
 import org.apache.doris.nereids.trees.plans.commands.DropDatabaseCommand;
 import org.apache.doris.nereids.trees.plans.commands.info.DropDatabaseInfo;
 import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.persist.ColocatePersistInfo;
+import org.apache.doris.persist.EditLog;
 import org.apache.doris.qe.ConnectContext;
 import org.apache.doris.qe.StmtExecutor;
 import org.apache.doris.resource.Tag;
@@ -42,8 +45,11 @@
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.ExpectedException;
+import org.mockito.Mockito;
 
 import java.io.File;
+import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
@@ -118,6 +124,44 @@ private static void alterTable(String sql) throws Exception {
         }
     }
 
+    private static void alterColocateGroup(String sql) throws Exception {
+        NereidsParser nereidsParser = new NereidsParser();
+        LogicalPlan parsed = nereidsParser.parseSingle(sql);
+        StmtExecutor stmtExecutor = new StmtExecutor(connectContext, sql);
+        if (parsed instanceof AlterColocateGroupCommand) {
+            ((AlterColocateGroupCommand) parsed).run(connectContext, stmtExecutor);
+        } else {
+            Assert.fail("Expected AlterColocateGroupCommand, but parsed: " + parsed.getClass().getSimpleName());
+        }
+    }
+
+    private static void createSingleReplicaColocateTable(String tableName) throws Exception {
+        createTable("create table " + dbName + "." + tableName + " (\n"
+                + " `k1` int NULL COMMENT \"\",\n"
+                + " `k2` varchar(10) NULL COMMENT \"\"\n"
+                + ") ENGINE=OLAP\n"
+                + "DUPLICATE KEY(`k1`, `k2`)\n"
+                + "COMMENT \"OLAP\"\n"
+                + "DISTRIBUTED BY HASH(`k1`, `k2`) BUCKETS 1\n"
+                + "PROPERTIES (\n"
+                + " \"replication_num\" = \"1\",\n"
+                + " \"colocate_with\" = \"" + groupName + "\"\n"
+                + ");");
+    }
+
+    private static Map<Tag, List<List<Long>>> copyBackendsPerBucketSeq(
+            Map<Tag, List<List<Long>>> backendsPerBucketSeq) {
+        Map<Tag, List<List<Long>>> copied = new HashMap<>();
+        for (Map.Entry<Tag, List<List<Long>>> entry : backendsPerBucketSeq.entrySet()) {
+            List<List<Long>> copiedBuckets = new ArrayList<>();
+            for (List<Long> backends : entry.getValue()) {
+                copiedBuckets.add(new ArrayList<>(backends));
+            }
+            copied.put(entry.getKey(), copiedBuckets);
+        }
+        return copied;
+    }
+
     @Test
     public void testCreateOneTable() throws Exception {
         createTable("create table " + dbName + "." + tableName1 + " (\n"
@@ -161,6 +205,57 @@ public void testCreateOneTable() throws Exception {
         Assert.assertEquals((short) 1, groupSchema.getReplicaAlloc().getTotalReplicaNum());
     }
 
+    @Test
+    public void testAlterColocateGroupReplicaAllocationLogsEditLog() throws Exception {
+        createSingleReplicaColocateTable(tableName1);
+
+        Env env = Env.getCurrentEnv();
+        EditLog originalEditLog = env.getEditLog();
+        EditLog mockEditLog = Mockito.mock(EditLog.class);
+        env.setEditLog(mockEditLog);
+        try {
+            alterColocateGroup("ALTER COLOCATE GROUP " + dbName + "." + groupName
+                    + " SET (\"replication_num\" = \"1\")");
+
+            Mockito.verify(mockEditLog, Mockito.times(1))
+                    .logColocateModifyRepliaAlloc(Mockito.any(ColocatePersistInfo.class));
+
+            ColocateTableIndex index = Env.getCurrentColocateIndex();
+            Database db = Env.getCurrentInternalCatalog().getDbOrMetaException(fullDbName);
+            String fullGroupName = GroupId.getFullGroupName(db.getId(), groupName);
+            Assert.assertEquals((short) 1,
+                    index.getGroupSchema(fullGroupName).getReplicaAlloc().getTotalReplicaNum());
+        } finally {
+            env.setEditLog(originalEditLog);
+        }
+    }
+
+    @Test
+    public void testReplayModifyReplicaAllocationDoesNotLogEditLog() throws Exception {
+        createSingleReplicaColocateTable(tableName1);
+
+        ColocateTableIndex index = Env.getCurrentColocateIndex();
+        Database db = Env.getCurrentInternalCatalog().getDbOrMetaException(fullDbName);
+        long tableId = db.getTableOrMetaException(tableName1).getId();
+        GroupId groupId = index.getGroup(tableId);
+        ColocatePersistInfo info = ColocatePersistInfo.createForModifyReplicaAlloc(
+                groupId, new ReplicaAllocation((short) 1),
+                copyBackendsPerBucketSeq(index.getBackendsPerBucketSeq(groupId)));
+
+        Env env = Env.getCurrentEnv();
+        EditLog originalEditLog = env.getEditLog();
+        EditLog mockEditLog = Mockito.mock(EditLog.class);
+        env.setEditLog(mockEditLog);
+        try {
+            index.replayModifyReplicaAlloc(info);
+
+            Mockito.verify(mockEditLog, Mockito.never())
+                    .logColocateModifyRepliaAlloc(Mockito.any(ColocatePersistInfo.class));
+        } finally {
+            env.setEditLog(originalEditLog);
+        }
+    }
+
     @Test
     public void testCreateTwoTableWithSameGroup() throws Exception {
         createTable("create table " + dbName + "." + tableName1 + " (\n"