This is an automated email from the ASF dual-hosted git repository.
deardeng 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 e94ad2421b3 [fix](partition) handle rollback failure when adding
partition (#64650)
e94ad2421b3 is described below
commit e94ad2421b389e3692b38e75ab1711c0b5241b44
Author: deardeng <[email protected]>
AuthorDate: Fri Aug 7 10:28:18 2026 +0800
[fix](partition) handle rollback failure when adding partition (#64650)
When adding a partition fails after metadata has been partially updated,
the rollback path must clean up the newly created partition state
consistently. Otherwise the catalog and meta-service state can diverge,
leaving later add-partition or recovery operations with stale partition
metadata.
---
.../org/apache/doris/catalog/PartitionInfo.java | 2 +
.../apache/doris/datasource/InternalCatalog.java | 68 +++++-
.../doris/datasource/InternalCatalogTest.java | 232 +++++++++++++++++++++
3 files changed, 299 insertions(+), 3 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/catalog/PartitionInfo.java
b/fe/fe-core/src/main/java/org/apache/doris/catalog/PartitionInfo.java
index 5cec453baac..f23954280f3 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/PartitionInfo.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/PartitionInfo.java
@@ -359,8 +359,10 @@ public class PartitionInfo {
public void dropPartition(long partitionId) {
idToDataProperty.remove(partitionId);
+ idToStoragePolicy.remove(partitionId);
idToReplicaAllocation.remove(partitionId);
idToInMemory.remove(partitionId);
+ idToTabletType.remove(partitionId);
idToItem.remove(partitionId);
idToTempItem.remove(partitionId);
}
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 dbc222ac1c4..0545d8671c0 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
@@ -1717,8 +1717,12 @@ public class InternalCatalog implements
CatalogIf<Database> {
}
};
Throwable creationThrowable = null;
+ boolean partitionInfoUpdated = false;
+ boolean editLogWritten = false;
+ boolean partitionPublished = false;
+ long partitionId = -1L;
try {
- long partitionId = Config.isCloudMode() &&
!FeConstants.runningUnitTest && isCreateTable
+ partitionId = Config.isCloudMode() && !FeConstants.runningUnitTest
&& isCreateTable
? generatedPartitionId : idGeneratorBuffer.getNextId();
List<Long> partitionIds = Lists.newArrayList(partitionId);
List<Long> indexIds = new ArrayList<>(indexIdToMeta.keySet());
@@ -1801,6 +1805,7 @@ public class InternalCatalog implements
CatalogIf<Database> {
// update partition info
partitionInfo.handleNewSinglePartitionDesc(singlePartitionDesc, partitionId,
isTempPartition);
+ partitionInfoUpdated = true;
// log
PartitionPersistInfo info = null;
@@ -1822,30 +1827,45 @@ public class InternalCatalog implements
CatalogIf<Database> {
}
if (!isCreateTable) {
- afterCreatePartitions(db.getId(), olapTable.getId(),
partitionIds, indexIds, isCreateTable,
- false /* isBatchCommit */, olapTable);
+ try {
+ afterCreatePartitions(db.getId(), olapTable.getId(),
partitionIds, indexIds, isCreateTable,
+ false /* isBatchCommit */, olapTable);
+ } catch (Throwable t) {
+ if (rollbackPartitionInfoWithTableLock(db, tableName,
partitionId, partitionInfoUpdated,
+ editLogWritten, partitionPublished,
olapTable)) {
+ partitionInfoUpdated = false;
+ }
+ throw t;
+ }
}
if (writeEditLog) {
Env.getCurrentEnv().getEditLog().logAddPartition(info);
+ editLogWritten = true;
if (isTempPartition) {
olapTable.addTempPartition(partition);
} else {
olapTable.addPartition(partition);
}
+ partitionPublished = true;
LOG.info("succeed in creating partition[{}], temp: {}",
partitionId, isTempPartition);
} else {
batchPartitions.add(Pair.of(info, partition));
+ partitionPublished = true;
LOG.info("postpone creating partition[{}], temp: {}",
partitionId, isTempPartition);
}
} finally {
olapTable.writeUnlock();
}
} catch (DdlException e) {
+ rollbackPartitionInfoOnCreatePartitionFailure(db, tableName,
partitionId, partitionInfoUpdated,
+ editLogWritten, partitionPublished);
failedCleanCallback.run();
creationThrowable = e;
throw e;
} catch (Throwable t) {
// Ensure cleanup and propagate unexpected errors as well
+ rollbackPartitionInfoOnCreatePartitionFailure(db, tableName,
partitionId, partitionInfoUpdated,
+ editLogWritten, partitionPublished);
failedCleanCallback.run();
creationThrowable = t;
throw t;
@@ -1868,6 +1888,48 @@ public class InternalCatalog implements
CatalogIf<Database> {
}
}
+ private void rollbackPartitionInfoOnCreatePartitionFailure(Database db,
String tableName, long partitionId,
+ boolean partitionInfoUpdated, boolean editLogWritten, boolean
partitionPublished) {
+ if (!partitionInfoUpdated || partitionId < 0 || editLogWritten ||
partitionPublished) {
+ return;
+ }
+
+ try {
+ OlapTable olapTable = db.getOlapTableOrDdlException(tableName);
+ olapTable.writeLockOrDdlException();
+ try {
+ PartitionInfo partitionInfo = olapTable.getPartitionInfo();
+ partitionInfo.dropPartition(partitionId);
+ LOG.info("rollback partition info after failed partition
creation, db={}, table={}, partition_id={}",
+ db.getFullName(), tableName, partitionId);
+ } finally {
+ olapTable.writeUnlock();
+ }
+ } catch (Throwable t) {
+ LOG.warn("failed to rollback partition info after failed partition
creation, db={}, table={}, "
+ + "partition_id={}", db.getFullName(), tableName,
partitionId, t);
+ }
+ }
+
+ private boolean rollbackPartitionInfoWithTableLock(Database db, String
tableName, long partitionId,
+ boolean partitionInfoUpdated, boolean editLogWritten, boolean
partitionPublished, OlapTable olapTable) {
+ if (!partitionInfoUpdated || partitionId < 0 || editLogWritten ||
partitionPublished) {
+ return false;
+ }
+
+ try {
+ PartitionInfo partitionInfo = olapTable.getPartitionInfo();
+ partitionInfo.dropPartition(partitionId);
+ LOG.info("rollback partition info after failed partition creation,
db={}, table={}, partition_id={}",
+ db.getFullName(), tableName, partitionId);
+ return true;
+ } catch (Throwable t) {
+ LOG.warn("failed to rollback partition info after failed partition
creation, db={}, table={}, "
+ + "partition_id={}", db.getFullName(), tableName,
partitionId, t);
+ return false;
+ }
+ }
+
public void addMultiPartitions(Database db, String tableName,
AlterMultiPartitionOp multiPartitionOp)
throws DdlException {
List<SinglePartitionDesc> singlePartitionDescs;
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/InternalCatalogTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/InternalCatalogTest.java
new file mode 100644
index 00000000000..5a10553b0ae
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/InternalCatalogTest.java
@@ -0,0 +1,232 @@
+// 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.datasource;
+
+import org.apache.doris.analysis.PartitionKeyDesc;
+import org.apache.doris.analysis.PartitionValue;
+import org.apache.doris.analysis.SinglePartitionDesc;
+import org.apache.doris.catalog.BinlogConfig;
+import org.apache.doris.catalog.CatalogTestUtil;
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.DataProperty;
+import org.apache.doris.catalog.Database;
+import org.apache.doris.catalog.DiskInfo;
+import org.apache.doris.catalog.DistributionInfo;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.FakeEnv;
+import org.apache.doris.catalog.HashDistributionInfo;
+import org.apache.doris.catalog.KeysType;
+import org.apache.doris.catalog.LocalReplica;
+import org.apache.doris.catalog.LocalTablet;
+import org.apache.doris.catalog.MaterializedIndex;
+import org.apache.doris.catalog.MaterializedIndex.IndexState;
+import org.apache.doris.catalog.MaterializedIndexMeta;
+import org.apache.doris.catalog.MetaIdGenerator.IdGeneratorBuffer;
+import org.apache.doris.catalog.OlapTable;
+import org.apache.doris.catalog.Partition;
+import org.apache.doris.catalog.PartitionInfo;
+import org.apache.doris.catalog.PartitionItem;
+import org.apache.doris.catalog.PartitionKey;
+import org.apache.doris.catalog.RangePartitionInfo;
+import org.apache.doris.catalog.RangePartitionItem;
+import org.apache.doris.catalog.Replica.ReplicaState;
+import org.apache.doris.catalog.ReplicaAllocation;
+import org.apache.doris.catalog.ScalarType;
+import org.apache.doris.catalog.TableProperty;
+import org.apache.doris.catalog.TabletMeta;
+import org.apache.doris.common.DdlException;
+import org.apache.doris.nereids.trees.plans.commands.info.AddPartitionOp;
+import org.apache.doris.system.Backend;
+import org.apache.doris.system.SystemInfoService;
+import org.apache.doris.system.SystemInfoService.HostInfo;
+import org.apache.doris.thrift.TStorageMedium;
+import org.apache.doris.thrift.TStorageType;
+import org.apache.doris.thrift.TTabletType;
+
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Range;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+public class InternalCatalogTest {
+ private static final long TABLE_ID = 1000L;
+ private static final long INDEX_ID = 1000L;
+ private static final long PARTITION_ID = 1001L;
+ private static final long TABLET_ID = 1002L;
+ private static final long BACKEND_ID = 1003L;
+ private static final long REPLICA_ID = 1004L;
+ private static final long BACKEND_ID_2 = 1005L;
+ private static final long BACKEND_ID_3 = 1006L;
+ private static final String TABLE_NAME = "range_table";
+ private static final String PARTITION_NAME = "p0";
+ private static final String NEW_PARTITION_NAME = "p_commit_failed";
+
+ private Database db;
+ private FailingCommitInternalCatalog catalog;
+ private FakeEnv fakeEnv;
+
+ @Before
+ public void setUp() throws Exception {
+ fakeEnv = new FakeEnv();
+ Env env = new TestingEnv();
+ FakeEnv.setEnv(env);
+ FakeEnv.setSystemInfo(createSystemInfoService());
+ db = createRangePartitionDb();
+ catalog = new FailingCommitInternalCatalog();
+ }
+
+ @After
+ public void tearDown() {
+ fakeEnv.close();
+ }
+
+ @Test
+ public void testAddPartitionRollbackPartitionInfoOnCommitFailure() throws
Exception {
+ AddPartitionOp addPartitionOp = createAddPartitionOp();
+ DdlException exception = Assert.assertThrows(DdlException.class,
+ () -> catalog.addPartition(db, TABLE_NAME, addPartitionOp,
false, 0, true, null));
+ Assert.assertTrue(exception.getMessage().contains("injected commit
failure"));
+ long newPartitionId = catalog.getCommittedPartitionId();
+
+ OlapTable table = (OlapTable) db.getTableOrDdlException(TABLE_NAME);
+ Assert.assertNull(table.getPartition(NEW_PARTITION_NAME));
+ Assert.assertNull(table.getPartition(newPartitionId));
+
+ PartitionInfo partitionInfo = table.getPartitionInfo();
+ Assert.assertNull(partitionInfo.getItem(newPartitionId));
+ Assert.assertNull(partitionInfo.getDataProperty(newPartitionId));
+ Assert.assertEquals(ReplicaAllocation.DEFAULT_ALLOCATION,
partitionInfo.getReplicaAllocation(newPartitionId));
+ }
+
+ private AddPartitionOp createAddPartitionOp() {
+ SinglePartitionDesc singlePartitionDesc = new
SinglePartitionDesc(false, NEW_PARTITION_NAME,
+ PartitionKeyDesc.createLessThan(Lists.newArrayList(new
PartitionValue("20"))), Maps.newHashMap());
+ return new AddPartitionOp(singlePartitionDesc, null,
Maps.newHashMap(), false);
+ }
+
+ private SystemInfoService createSystemInfoService() {
+ SystemInfoService systemInfoService = new SystemInfoService();
+ systemInfoService.addBackend(createBackend(BACKEND_ID, "host1"));
+ systemInfoService.addBackend(createBackend(BACKEND_ID_2, "host2"));
+ systemInfoService.addBackend(createBackend(BACKEND_ID_3, "host3"));
+ return systemInfoService;
+ }
+
+ private Backend createBackend(long backendId, String host) {
+ Backend backend = new Backend(backendId, host, 9050);
+ DiskInfo diskInfo = new DiskInfo("/path/to/disk1/");
+ diskInfo.setAvailableCapacityB(2L << 40);
+ diskInfo.setTotalCapacityB(2L << 40);
+ backend.setDisks(ImmutableMap.of("disk1", diskInfo));
+ backend.setAlive(true);
+ return backend;
+ }
+
+ private Database createRangePartitionDb() throws Exception {
+ Column keyColumn = new Column("k1", ScalarType.INT);
+ keyColumn.setIsKey(true);
+ Column valueColumn = new Column("v1", ScalarType.INT);
+ List<Column> columns = Lists.newArrayList(keyColumn, valueColumn);
+
+ MaterializedIndex baseIndex = new MaterializedIndex(INDEX_ID,
IndexState.NORMAL);
+ LocalTablet tablet = new LocalTablet(TABLET_ID);
+ TabletMeta tabletMeta = new TabletMeta(CatalogTestUtil.testDbId1,
TABLE_ID, PARTITION_ID, INDEX_ID, 0,
+ TStorageMedium.HDD);
+ baseIndex.addTablet(tablet, tabletMeta);
+ tablet.addReplica(new LocalReplica(REPLICA_ID, BACKEND_ID, 0,
ReplicaState.NORMAL));
+
+ DistributionInfo distributionInfo = new HashDistributionInfo(1,
Lists.newArrayList(keyColumn));
+ Partition partition = new Partition(PARTITION_ID, PARTITION_NAME,
baseIndex, distributionInfo);
+ partition.updateVisibleVersion(1L);
+
+ RangePartitionInfo partitionInfo = new
RangePartitionInfo(Lists.newArrayList(keyColumn));
+ PartitionKey lower =
PartitionKey.createInfinityPartitionKey(Lists.newArrayList(keyColumn), false);
+ PartitionKey upper =
PartitionKey.createPartitionKey(Lists.newArrayList(new PartitionValue("10")),
+ Lists.newArrayList(keyColumn));
+ PartitionItem partitionItem = new
RangePartitionItem(Range.closedOpen(lower, upper));
+ partitionInfo.setItem(PARTITION_ID, false, partitionItem);
+ partitionInfo.setDataProperty(PARTITION_ID, new
DataProperty(TStorageMedium.HDD));
+ partitionInfo.setReplicaAllocation(PARTITION_ID,
ReplicaAllocation.DEFAULT_ALLOCATION);
+
+ OlapTable table = new OlapTable(TABLE_ID, TABLE_NAME, columns,
KeysType.DUP_KEYS, partitionInfo,
+ distributionInfo);
+ table.setTableProperty(new TableProperty(Maps.newHashMap()));
+ table.setIndexMeta(INDEX_ID, TABLE_NAME, columns, 0, 0, (short) 1,
TStorageType.COLUMN, KeysType.DUP_KEYS);
+ table.setBaseIndexId(INDEX_ID);
+ table.addPartition(partition);
+
+ Database database = new Database(CatalogTestUtil.testDbId1,
CatalogTestUtil.testDb1);
+ database.registerTable(table);
+ return database;
+ }
+
+ private static class FailingCommitInternalCatalog extends InternalCatalog {
+ private long committedPartitionId;
+
+ @Override
+ protected Partition createPartitionWithIndices(long dbId, OlapTable
tbl, long partitionId,
+ String partitionName, Map<Long, MaterializedIndexMeta>
indexIdToMeta,
+ DistributionInfo distributionInfo, DataProperty dataProperty,
+ ReplicaAllocation replicaAlloc, Long versionInfo, Set<String>
bfColumns, Set<Long> tabletIdSet,
+ boolean isInMemory, TTabletType tabletType, String
storagePolicy,
+ IdGeneratorBuffer idGeneratorBuffer, BinlogConfig
binlogConfig, boolean isStorageMediumSpecified)
+ throws DdlException {
+ MaterializedIndex baseIndex = new
MaterializedIndex(tbl.getBaseIndexId(), IndexState.NORMAL);
+ Partition partition = new Partition(partitionId, partitionName,
baseIndex, distributionInfo);
+ if (versionInfo != null) {
+ partition.updateVisibleVersion(versionInfo);
+ partition.setNextVersion(versionInfo + 1);
+ }
+ return partition;
+ }
+
+ @Override
+ public void afterCreatePartitions(long dbId, long tableId, List<Long>
partitionIds, List<Long> indexIds,
+ boolean isCreateTable, boolean isBatchCommit, OlapTable
olapTable) throws DdlException {
+ Assert.assertEquals(TABLE_ID, tableId);
+ Assert.assertEquals(1, partitionIds.size());
+ committedPartitionId = partitionIds.get(0);
+ throw new DdlException("injected commit failure");
+ }
+
+ public long getCommittedPartitionId() {
+ return committedPartitionId;
+ }
+ }
+
+ private static class TestingEnv extends Env {
+ private final HostInfo selfNode = new HostInfo("127.0.0.1", 9010);
+
+ private TestingEnv() throws Exception {
+ super(false);
+ }
+
+ @Override
+ public HostInfo getSelfNode() {
+ return selfNode;
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
