This is an automated email from the ASF dual-hosted git repository.
924060929 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 631ccd57525 [fix](fe) Skip insert after losing concurrent CTAS table
creation (#68849)
631ccd57525 is described below
commit 631ccd575252a1ef1a9cbcf7809ea00b397b6dcc
Author: 924060929 <[email protected]>
AuthorDate: Sat Oct 10 17:20:13 2026 +0800
[fix](fe) Skip insert after losing concurrent CTAS table creation (#68849)
### What problem does this PR solve?
Issue Number: N/A
Related PR: #35210, #53486
Problem Summary:
Two concurrent `CREATE TABLE IF NOT EXISTS ... AS SELECT` statements can
both pass the initial table-existence check. Only one registers its
table; the other receives the atomic registration result indicating that
the table already exists. However, `InternalCatalog.createOlapTable`
leaves its returned `tableHasExist` flag false, so both CTAS statements
continue into INSERT and append their rows to the same table. For
example, two concurrent `CREATE TABLE IF NOT EXISTS t AS SELECT 1 AS k`
statements can leave two rows instead of one.
History: #35210 introduced the existing-table return contract and the
CTAS INSERT skip, but initialized `tableHasExist` to false without
propagating the duplicate-registration result in the OLAP path. This
fixed the sequential existing-table case while leaving the concurrent
registration case unhandled. #53486 later carried the same omission into
the `CreateTableInfo` path used by the current implementation.
Propagate the existing-table result from
`Database.createTableWithoutLock` while holding the existing database
write lock. The losing creator releases its resources, then returns true
so the existing CTAS early-return branch skips INSERT. In cloud mode,
`afterCreatePartitions` has already committed the losing indexes and
removed their PREPARED recycle markers before registration. FE-only
tablet/colocate cleanup cannot recycle these committed indexes. Add a
catalog lifecycle hook after the database lock is released;
`CloudInternalCatalog` uses it to call `dropMaterializedIndex` for the
losing table and all its index IDs, including row-binlog indexes. Invoke
this for both IF NOT EXISTS and duplicate-error paths. Propagate cleanup
failures instead of reporting a successful no-op; existing FE cleanup
still runs and the winning table is preserved.
Scope: this PR changes FE behavior only and uses the existing
MetaService drop-index RPC. The follow-up table-version KV cleanup
extension has been reverted; there are no changes under `cloud/` or to
`gensrc/proto/cloud.proto` in the final PR diff. Orphaned table-version
KV discovery in the cloud recycler is a separate existing issue outside
this FE repair.
The new tests deterministically simulate a stale initial existence probe
against a real database, exercise the actual atomic registration and
cleanup, and check that the winner's table and tablet metadata survive.
They also cover sequential IF NOT EXISTS, the duplicate-registration
error without IF NOT EXISTS, cleanup outside the database lock,
preservation of the winner on cleanup failure, and the cloud drop RPC
IDs and error propagation.
### Release note
Fix duplicate rows when concurrent CREATE TABLE IF NOT EXISTS AS SELECT
statements target the same internal table, and recycle committed cloud
indexes allocated by the losing creator.
### Check List (For Author)
- Test
- [x] Unit Test
- [ ] Regression test
- [x] Manual test (add detailed scripts or steps below)
- Behavior changed:
- [ ] No.
- [x] Yes. A CTAS statement that loses concurrent table registration
skips INSERT as required by IF NOT EXISTS.
- Does this need documentation?
- [x] No.
- [ ] Yes.
Validation (JDK 17):
- After reverting the cloud table-version extension, the tracked source
tree exactly matches `0a85b29c053`. Re-ran all three listed FE test
classes on restored head `f24ad6c5c69`: 27 tests passed, 0
failures/errors/skips. FE Checkstyle and `git diff --check` also passed.
The final PR diff contains only four files under `fe/`, with no
MetaService, recycler, or protobuf changes.
- Cloud cleanup follow-up: `bash run-fe-ut.sh --run
org.apache.doris.catalog.CreateTableTest,org.apache.doris.nereids.trees.plans.commands.CreateTableCommandTest,org.apache.doris.cloud.datasource.CloudInternalCatalogTest`:
27 tests passed (0 failures/errors/skips). Tests cover the real
registration race, cleanup outside the database lock, both
duplicate-result paths, winner preservation and FE cleanup on failure,
and the exact cloud drop RPC IDs and error propagation.
- Cloud cleanup follow-up: FE Checkstyle passed with 0 violations; `git
diff --check` passed. The RPC boundary is mocked; a live
MetaService/recycler deployment has not been tested.
- `bash run-fe-ut.sh --run
'org.apache.doris.catalog.CreateTableTest,org.apache.doris.nereids.trees.plans.commands.CreateTableCommandTest'`:
23 tests passed before the cloud cleanup follow-up.
- Negative control: removing only the production fix makes
`testCreateTableReturnsExistingAfterRegistrationRace` fail with
`expected: <true> but was: <false>`; restoring the fix passes.
- `cd fe && mvn checkstyle:check -pl fe-core`: passed.
- `git diff --check`: passed.
Deployed validation of the original CTAS return-value fix on 2026-10-10
(before the cloud cleanup follow-up): apply the same production fix to
the exact affected 4.1 candidate source
(`7fda449b7afb09c266ce2c4fc738a4d5e46fc458`), build FE with `build.sh
--fe --clean`, and deploy it against the same four unmodified candidate
BEs.
- Fresh unpatched control: all 8 concurrent `AS SELECT 1` trials
reproduce duplicate inserts (2 callers leave 2 rows; 8 callers leave 8
rows).
- Patched FE: all 80 concurrent trials pass (400 CTAS requests across
2/8 callers and default/local-shuffle-disabled settings). Exactly one
caller inserts one row per table; all other callers affect zero rows.
- All 24 trials of the four original queries pass, including full result
multiset comparison, across default, local-shuffle-disabled and
single-pipeline settings. Only one CTAS caller inserts each query's
38/32/26/25 rows. Snapshot GLOBAL variables are applied with SET GLOBAL;
final replay has no setup or query errors.
- All 80 sequential existing-table CTAS checks remain no-ops.
- All 10 concurrent trials without IF NOT EXISTS produce one successful
creator and one duplicate-table error, preserve the winning data, and
allow subsequent ordinary INSERTs.
- Existing regression suites `test_create_table_if_not_exists_as_select`
and `nereids_test_ctas` pass through `run-regression-test.sh` (0 failed
suites, 0 fatal scripts).
- FE audit/registration logs confirm the losing creators skip INSERT.
The final FE and all four BEs are healthy.
This is a local shared-nothing cluster; x86 BEs run under QEMU on an
arm64 host. The original S3/cloud deployment, long-running stress, and
the full regression suite have not been retested. Please backport to
branch-4.1.
---
.../cloud/datasource/CloudInternalCatalog.java | 5 ++
.../apache/doris/datasource/InternalCatalog.java | 11 +++
.../org/apache/doris/catalog/CreateTableTest.java | 93 ++++++++++++++++++++++
.../cloud/datasource/CloudInternalCatalogTest.java | 45 +++++++++++
4 files changed, 154 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 283a53b455b..95228b01c14 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
@@ -1300,6 +1300,11 @@ public class CloudInternalCatalog extends
InternalCatalog {
}
}
+ @Override
+ public void onCreateTableConflict(long dbId, OlapTable table) throws
DdlException {
+ dropMaterializedIndex(dbId, table.getId(), table.getIndexIdList(true),
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 7165ae6b505..da9dbeeed0b 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
@@ -2421,6 +2421,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 {
@@ -3396,6 +3400,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;
@@ -3404,6 +3410,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 d0c05a31ccb..eb61be84189 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,7 +24,11 @@ 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.thrift.TInvertedIndexFileStorageFormat;
import org.apache.doris.utframe.TestWithFeService;
@@ -32,7 +36,10 @@ import org.apache.doris.utframe.TestWithFeService;
import com.google.common.collect.Maps;
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;
@@ -80,6 +87,92 @@ public class CreateTableTest extends TestWithFeService {
}
}
+ @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(true),
+ loser.getIndexIdList(true)));
+ 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
index 0725fe07049..6f124c7fad9 100644
---
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
@@ -19,9 +19,13 @@ package org.apache.doris.cloud.datasource;
import org.apache.doris.analysis.DataSortInfo;
import org.apache.doris.catalog.KeysType;
+import org.apache.doris.catalog.OlapTable;
import org.apache.doris.catalog.Replica;
import org.apache.doris.catalog.Tablet;
+import org.apache.doris.cloud.proto.Cloud;
+import org.apache.doris.cloud.rpc.MetaServiceProxy;
import org.apache.doris.common.Config;
+import org.apache.doris.common.DdlException;
import org.apache.doris.proto.OlapFile;
import org.apache.doris.thrift.TCompressionType;
import org.apache.doris.thrift.TInvertedIndexFileStorageFormat;
@@ -31,9 +35,13 @@ import org.apache.doris.thrift.TTabletType;
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.Collections;
+import java.util.List;
public class CloudInternalCatalogTest {
@Test
@@ -69,6 +77,43 @@ 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(true)).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());
+ }
+ }
+
private OlapFile.TabletMetaCloudPB createTabletMeta(Tablet tablet) throws
Exception {
return new CloudInternalCatalog().createTabletMetaBuilder(
1L, 2L, 3L, tablet, TTabletType.TABLET_TYPE_DISK, 23,
KeysType.DUP_KEYS, (short) 1,
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
