This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 93c706cc15d branch-4.1: [fix](fe) Skip insert after losing concurrent
CTAS table creation #68849 (#68881)
93c706cc15d is described below
commit 93c706cc15d58d040f6918b25e0770e1369df69e
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 c3dd64b3071..a9a7451a75f 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
@@ -1033,6 +1033,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]
