From 8b48dd918f0ebf0de3a5287098081d02f96e23ea Mon Sep 17 00:00:00 2001 From: deardeng Date: Mon, 13 Jul 2026 16:07:32 +0800 Subject: [PATCH] [fix](fe) Fix colocate replica allocation edit log (#65457) Related PR: #49569 Problem Summary: Related PR #49569 changed the colocate replica allocation helper boolean from needEditLog to isReplay, but kept the old call-site values. As a result, normal ALTER COLOCATE GROUP replica allocation changes were treated as replay and skipped logColocateModifyRepliaAlloc, so the new replica allocation was not persisted for follower replay or master failover. The replay path was treated as a normal operation and could append another edit log while loading journal entries. This patch flips the two call-site values: leader ALTER uses isReplay = false and writes the edit log, while replay uses isReplay = true and does not append a new edit log. It also adds FE unit tests covering both behaviors. --- .../doris/catalog/ColocateTableIndex.java | 4 +- .../doris/catalog/ColocateTableTest.java | 95 +++++++++++++++++++ 2 files changed, 97 insertions(+), 2 deletions(-) diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/ColocateTableIndex.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/ColocateTableIndex.java index 9fa53d062100a3..6bf1cf2e19b0c7 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/catalog/ColocateTableIndex.java +++ b/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()); } diff --git a/fe/fe-core/src/test/java/org/apache/doris/catalog/ColocateTableTest.java b/fe/fe-core/src/test/java/org/apache/doris/catalog/ColocateTableTest.java index 3f0079fb9451be..ef96012d64ef5d 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/catalog/ColocateTableTest.java +++ b/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>> copyBackendsPerBucketSeq( + Map>> backendsPerBucketSeq) { + Map>> copied = new HashMap<>(); + for (Map.Entry>> entry : backendsPerBucketSeq.entrySet()) { + List> copiedBuckets = new ArrayList<>(); + for (List 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"