This is an automated email from the ASF dual-hosted git repository. yiguolei pushed a commit to branch branch-4.2 in repository https://gitbox.apache.org/repos/asf/doris.git
commit 1a30544d6e39bce919ed3ffcd4fe1b2012dcc062 Author: 924060929 <[email protected]> AuthorDate: Mon Oct 12 13:58:24 2026 +0800 branch-4.1: [fix](fe) Skip insert after losing concurrent CTAS table creation #68849 (#68881) cherry pick from #68849 --- .../cloud/datasource/CloudInternalCatalog.java | 5 ++ .../apache/doris/datasource/InternalCatalog.java | 11 +++ .../org/apache/doris/catalog/CreateTableTest.java | 94 ++++++++++++++++++++++ .../cloud/datasource/CloudInternalCatalogTest.java | 72 +++++++++++++++++ 4 files changed, 182 insertions(+) diff --git a/fe/fe-core/src/main/java/org/apache/doris/cloud/datasource/CloudInternalCatalog.java b/fe/fe-core/src/main/java/org/apache/doris/cloud/datasource/CloudInternalCatalog.java index e00c1ee86cf..5cd384a9ad4 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/cloud/datasource/CloudInternalCatalog.java +++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/datasource/CloudInternalCatalog.java @@ -1037,6 +1037,11 @@ public class CloudInternalCatalog extends InternalCatalog { } } + @Override + public void onCreateTableConflict(long dbId, OlapTable table) throws DdlException { + dropMaterializedIndex(dbId, table.getId(), table.getIndexIdList(), true); + } + public void dropMaterializedIndex(long dbId, long tableId, List<Long> indexIds, boolean dropTable) throws DdlException { if (Config.enable_check_compatibility_mode) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/InternalCatalog.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/InternalCatalog.java index f6575ac3d45..b862973bd00 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/InternalCatalog.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/InternalCatalog.java @@ -2306,6 +2306,10 @@ public class InternalCatalog implements CatalogIf<Database> { return partition; } + /** Release resources allocated for a table that lost the atomic registration race. */ + public void onCreateTableConflict(long dbId, OlapTable table) throws DdlException { + } + public void beforeCreatePartitions(long dbId, long tableId, List<Long> partitionIds, List<Long> indexIds, boolean isCreateTable) throws DdlException { @@ -3314,6 +3318,8 @@ public class InternalCatalog implements CatalogIf<Database> { } // register table, write create table edit log result = db.createTableWithoutLock(olapTable, false, createTableInfo.isIfNotExists()); + // CTAS must skip INSERT when another creator registered the table first. + tableHasExist = result.second; if (!result.second) { olapTable.writeLock(); holdTableLock = true; @@ -3322,6 +3328,11 @@ public class InternalCatalog implements CatalogIf<Database> { db.writeUnlock(); } try { + if (result.second) { + // Cloud indexes have already been committed. Recycle them outside the DB lock, + // including when a CREATE without IF NOT EXISTS must report a duplicate error. + onCreateTableConflict(db.getId(), olapTable); + } if (!result.first) { ErrorReport.reportDdlException(ErrorCode.ERR_TABLE_EXISTS_ERROR, tableShowName); } diff --git a/fe/fe-core/src/test/java/org/apache/doris/catalog/CreateTableTest.java b/fe/fe-core/src/test/java/org/apache/doris/catalog/CreateTableTest.java index fa885be8c2e..af5b461339a 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/catalog/CreateTableTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/catalog/CreateTableTest.java @@ -24,14 +24,22 @@ import org.apache.doris.common.DdlException; import org.apache.doris.common.ExceptionChecker; import org.apache.doris.common.FeConstants; import org.apache.doris.common.UserException; +import org.apache.doris.datasource.InternalCatalog; import org.apache.doris.nereids.exceptions.AnalysisException; +import org.apache.doris.nereids.parser.NereidsParser; +import org.apache.doris.nereids.trees.plans.commands.CreateTableCommand; +import org.apache.doris.nereids.trees.plans.commands.info.CreateTableInfo; import org.apache.doris.resource.Tag; import org.apache.doris.utframe.TestWithFeService; import com.google.common.collect.Maps; import org.junit.Assert; +import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; +import org.mockito.Mockito; +import java.util.Collections; +import java.util.HashSet; import java.util.Map; import java.util.Set; import java.util.UUID; @@ -51,6 +59,92 @@ public class CreateTableTest extends TestWithFeService { createDatabase("test"); } + @Test + public void testCreateTableReturnsExistingAfterRegistrationRace() throws Exception { + String sql = "CREATE TABLE IF NOT EXISTS test.registration_race (k INT) " + + "DISTRIBUTED BY HASH(k) BUCKETS 1 PROPERTIES('replication_num'='1')"; + InternalCatalog catalog = Env.getCurrentInternalCatalog(); + CreateTableInfo winner = analyzeCreateTable(sql); + Assertions.assertFalse(catalog.createTable(winner)); + + Database db = catalog.getDbOrDdlException("test"); + Table table = db.getTableOrDdlException("registration_race"); + Set<Long> tabletIds = new HashSet<>(Env.getCurrentInvertedIndex().getTabletMetaMap().keySet()); + + // Both creators can observe absence before either registers its table. Make only the + // loser's initial probe stale; atomic registration must still see the winner's real table. + Database racingDb = Mockito.spy(db); + Mockito.doReturn(false).doCallRealMethod().when(racingDb).isTableExist("registration_race"); + InternalCatalog racingCatalog = Mockito.spy(catalog); + Mockito.doReturn(racingDb).when(racingCatalog).getDbOrDdlException("test"); + Mockito.doAnswer(invocation -> { + Assertions.assertFalse(db.isWriteLockHeldByCurrentThread()); + OlapTable loser = invocation.getArgument(1); + Assertions.assertTrue(Collections.disjoint(((OlapTable) table).getIndexIdList(), + loser.getIndexIdList())); + return invocation.callRealMethod(); + }).when(racingCatalog).onCreateTableConflict(Mockito.eq(db.getId()), Mockito.any(OlapTable.class)); + Assertions.assertTrue(racingCatalog.createTable(analyzeCreateTable(sql))); + Mockito.verify(racingCatalog).onCreateTableConflict(Mockito.eq(db.getId()), + Mockito.argThat(loser -> loser.getId() != table.getId())); + Assertions.assertSame(table, db.getTableOrDdlException("registration_race")); + Assertions.assertEquals(tabletIds, Env.getCurrentInvertedIndex().getTabletMetaMap().keySet()); + + // A sequential IF NOT EXISTS remains a no-op as well. + Assertions.assertTrue(catalog.createTable(analyzeCreateTable(sql))); + } + + @Test + public void testCreateTableRegistrationRaceWithoutIfNotExists() throws Exception { + String sql = "CREATE TABLE test.registration_race_error (k INT) " + + "DISTRIBUTED BY HASH(k) BUCKETS 1 PROPERTIES('replication_num'='1')"; + createTable(sql); + InternalCatalog catalog = Env.getCurrentInternalCatalog(); + Database db = catalog.getDbOrDdlException("test"); + Table table = db.getTableOrDdlException("registration_race_error"); + Set<Long> tabletIds = new HashSet<>(Env.getCurrentInvertedIndex().getTabletMetaMap().keySet()); + Database racingDb = Mockito.spy(db); + Mockito.doReturn(false).doCallRealMethod().when(racingDb).isTableExist("registration_race_error"); + InternalCatalog racingCatalog = Mockito.spy(catalog); + Mockito.doReturn(racingDb).when(racingCatalog).getDbOrDdlException("test"); + + Assertions.assertThrows(DdlException.class, () -> racingCatalog.createTable(analyzeCreateTable(sql))); + Mockito.verify(racingCatalog).onCreateTableConflict(Mockito.eq(db.getId()), + Mockito.argThat(loser -> loser.getId() != table.getId())); + Assertions.assertSame(table, db.getTableOrDdlException("registration_race_error")); + Assertions.assertEquals(tabletIds, Env.getCurrentInvertedIndex().getTabletMetaMap().keySet()); + } + + @Test + public void testCreateTableRegistrationRaceCleanupFailure() throws Exception { + String sql = "CREATE TABLE IF NOT EXISTS test.registration_race_cleanup (k INT) " + + "DISTRIBUTED BY HASH(k) BUCKETS 1 PROPERTIES('replication_num'='1')"; + createTable(sql); + InternalCatalog catalog = Env.getCurrentInternalCatalog(); + Database db = catalog.getDbOrDdlException("test"); + Table table = db.getTableOrDdlException("registration_race_cleanup"); + Set<Long> tabletIds = new HashSet<>(Env.getCurrentInvertedIndex().getTabletMetaMap().keySet()); + Database racingDb = Mockito.spy(db); + Mockito.doReturn(false).doCallRealMethod().when(racingDb).isTableExist("registration_race_cleanup"); + InternalCatalog racingCatalog = Mockito.spy(catalog); + Mockito.doReturn(racingDb).when(racingCatalog).getDbOrDdlException("test"); + Mockito.doThrow(new DdlException("failed to recycle losing indexes")) + .when(racingCatalog).onCreateTableConflict(Mockito.eq(db.getId()), Mockito.any(OlapTable.class)); + + DdlException error = Assertions.assertThrows(DdlException.class, + () -> racingCatalog.createTable(analyzeCreateTable(sql))); + Assertions.assertTrue(error.getMessage().contains("failed to recycle losing indexes")); + Assertions.assertSame(table, db.getTableOrDdlException("registration_race_cleanup")); + Assertions.assertEquals(tabletIds, Env.getCurrentInvertedIndex().getTabletMetaMap().keySet()); + } + + private CreateTableInfo analyzeCreateTable(String sql) { + CreateTableCommand command = (CreateTableCommand) new NereidsParser().parseSingle(sql); + CreateTableInfo info = command.getCreateTableInfo(); + info.validate(connectContext); + return info; + } + @Test public void testDuplicateCreateTable() throws Exception { // test diff --git a/fe/fe-core/src/test/java/org/apache/doris/cloud/datasource/CloudInternalCatalogTest.java b/fe/fe-core/src/test/java/org/apache/doris/cloud/datasource/CloudInternalCatalogTest.java new file mode 100644 index 00000000000..e241e84b135 --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/cloud/datasource/CloudInternalCatalogTest.java @@ -0,0 +1,72 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.doris.cloud.datasource; + +import org.apache.doris.catalog.OlapTable; +import org.apache.doris.cloud.proto.Cloud; +import org.apache.doris.cloud.rpc.MetaServiceProxy; +import org.apache.doris.common.DdlException; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import org.mockito.MockedStatic; +import org.mockito.Mockito; + +import java.util.Arrays; +import java.util.List; + +public class CloudInternalCatalogTest { + @Test + public void testCreateTableConflictDropsOnlyLosingIndexes() throws Exception { + testCreateTableConflictCleanup(Cloud.MetaServiceCode.OK); + } + + @Test + public void testCreateTableConflictReportsCleanupFailure() throws Exception { + testCreateTableConflictCleanup(Cloud.MetaServiceCode.INVALID_ARGUMENT); + } + + private void testCreateTableConflictCleanup(Cloud.MetaServiceCode code) throws Exception { + OlapTable loser = Mockito.mock(OlapTable.class); + List<Long> indexIds = Arrays.asList(31L, 32L); + Mockito.when(loser.getId()).thenReturn(21L); + Mockito.when(loser.getIndexIdList()).thenReturn(indexIds); + MetaServiceProxy proxy = Mockito.mock(MetaServiceProxy.class); + Mockito.when(proxy.dropIndex(Mockito.any())).thenReturn(Cloud.IndexResponse.newBuilder() + .setStatus(Cloud.MetaServiceResponseStatus.newBuilder().setCode(code).setMsg("cleanup failed")) + .build()); + try (MockedStatic<MetaServiceProxy> mockedProxy = Mockito.mockStatic(MetaServiceProxy.class)) { + mockedProxy.when(MetaServiceProxy::getInstance).thenReturn(proxy); + CloudInternalCatalog catalog = new CloudInternalCatalog(); + if (code == Cloud.MetaServiceCode.OK) { + catalog.onCreateTableConflict(11L, loser); + } else { + DdlException error = Assertions.assertThrows(DdlException.class, + () -> catalog.onCreateTableConflict(11L, loser)); + Assertions.assertTrue(error.getMessage().contains("cleanup failed")); + } + ArgumentCaptor<Cloud.IndexRequest> request = ArgumentCaptor.forClass(Cloud.IndexRequest.class); + Mockito.verify(proxy).dropIndex(request.capture()); + Assertions.assertEquals(11L, request.getValue().getDbId()); + Assertions.assertEquals(21L, request.getValue().getTableId()); + Assertions.assertEquals(indexIds, request.getValue().getIndexIdsList()); + } + } + +} --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
