This is an automated email from the ASF dual-hosted git repository.
sollhui 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 a1853f20e79 [fix](mow) Retry delete bitmap calculation failures for
INSERT (#68342)
a1853f20e79 is described below
commit a1853f20e7960ba1ddd15de786d83f345adb2fad
Author: hui lai <[email protected]>
AuthorDate: Wed Sep 23 14:26:15 2026 +0800
[fix](mow) Retry delete bitmap calculation failures for INSERT (#68342)
### What problem does this PR solve?
Cloud MOW INSERT stopped retrying delete bitmap lock errors and
calculation timeouts after the insert executor switched to the
attachment-aware commit overload. That overload performs one attempt
because BE-driven loads already retry at their caller. Ordinary INSERT
is affected even when its attachment is null.
Add an explicit FE-owned retry entry and reuse the existing cloud retry
loop. Nereids INSERT now uses this entry, preserving the transaction
attachment and table stream updates on every attempt. Keep BE-driven
commit entry points single-attempt, the existing attempt limit and
signature lifecycle, and shared-nothing publish-timeout behavior.
---
.../transaction/CloudGlobalTransactionMgr.java | 11 +-
.../plans/commands/insert/OlapInsertExecutor.java | 2 +-
.../transaction/GlobalTransactionMgrIface.java | 10 ++
.../CloudTransactionCommitRetryTest.java | 162 +++++++++++++++++++++
.../commands/insert/OlapInsertExecutorTest.java | 39 ++++-
5 files changed, 216 insertions(+), 8 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgr.java
b/fe/fe-core/src/main/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgr.java
index 53fad5a02ed..8268c71925f 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgr.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgr.java
@@ -1636,6 +1636,15 @@ public class CloudGlobalTransactionMgr implements
GlobalTransactionMgrIface {
public boolean commitAndPublishTransaction(DatabaseIf db, List<Table>
tableList, long transactionId,
List<TabletCommitInfo>
tabletCommitInfos, long timeoutMillis)
throws UserException {
+ return commitAndPublishTransactionWithRetry(db, tableList,
transactionId, tabletCommitInfos, timeoutMillis,
+ null, Collections.emptyList());
+ }
+
+ @Override
+ public boolean commitAndPublishTransactionWithRetry(DatabaseIf db,
List<Table> tableList, long transactionId,
+ List<TabletCommitInfo> tabletCommitInfos, long timeoutMillis,
+ TxnCommitAttachment txnCommitAttachment,
List<TableStreamUpdateInfo> streamUpdateInfos)
+ throws UserException {
StopWatch stopWatch = new StopWatch();
stopWatch.start();
int retryTimes = 0;
@@ -1644,7 +1653,7 @@ public class CloudGlobalTransactionMgr implements
GlobalTransactionMgrIface {
while (true) {
try {
res = commitAndPublishTransaction(db, tableList,
transactionId, tabletCommitInfos, timeoutMillis,
- null);
+ txnCommitAttachment, streamUpdateInfos);
break;
} catch (UserException e) {
LOG.warn("failed to commit txn,
txnId={},retryTimes={},exception={}",
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/OlapInsertExecutor.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/OlapInsertExecutor.java
index 7dc65ebce29..29b80a91aad 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/OlapInsertExecutor.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/OlapInsertExecutor.java
@@ -233,7 +233,7 @@ public class OlapInsertExecutor extends
AbstractInsertExecutor {
} catch (Exception abortTxnException) {
LOG.warn("errors when abort txn. {}",
ctx.getQueryIdentifier(), abortTxnException);
}
- } else if
(Env.getCurrentGlobalTransactionMgr().commitAndPublishTransaction(
+ } else if
(Env.getCurrentGlobalTransactionMgr().commitAndPublishTransactionWithRetry(
database, Lists.newArrayList((Table) table),
txnId,
TabletCommitInfo.fromThrift(coordinator.getCommitInfos()),
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/transaction/GlobalTransactionMgrIface.java
b/fe/fe-core/src/main/java/org/apache/doris/transaction/GlobalTransactionMgrIface.java
index e241ff37982..1e9a388e141 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/transaction/GlobalTransactionMgrIface.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/transaction/GlobalTransactionMgrIface.java
@@ -124,6 +124,16 @@ public interface GlobalTransactionMgrIface extends
Writable {
txnCommitAttachment);
}
+ // FE-driven inserts own the cloud delete bitmap retries. BE-driven loads
retry at the caller
+ // and must continue to use the single-attempt overloads that accept an
attachment.
+ default boolean commitAndPublishTransactionWithRetry(DatabaseIf db,
List<Table> tableList, long transactionId,
+ List<TabletCommitInfo> tabletCommitInfos, long timeoutMillis,
+ TxnCommitAttachment txnCommitAttachment,
List<TableStreamUpdateInfo> streamUpdateInfos)
+ throws UserException {
+ return commitAndPublishTransaction(db, tableList, transactionId,
tabletCommitInfos, timeoutMillis,
+ txnCommitAttachment, streamUpdateInfos);
+ }
+
public void commitTransaction2PC(Database db, List<Table> tableList, long
transactionId, long timeoutMillis)
throws UserException;
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/cloud/transaction/CloudTransactionCommitRetryTest.java
b/fe/fe-core/src/test/java/org/apache/doris/cloud/transaction/CloudTransactionCommitRetryTest.java
new file mode 100644
index 00000000000..38aa9bda213
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/cloud/transaction/CloudTransactionCommitRetryTest.java
@@ -0,0 +1,162 @@
+// 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.transaction;
+
+import org.apache.doris.catalog.Database;
+import org.apache.doris.catalog.Table;
+import org.apache.doris.catalog.stream.TableStreamUpdateInfo;
+import org.apache.doris.common.Config;
+import org.apache.doris.common.InternalErrorCode;
+import org.apache.doris.common.UserException;
+import org.apache.doris.common.jmockit.Deencapsulation;
+import
org.apache.doris.job.extensions.insert.streaming.StreamingTaskTxnCommitAttachment;
+import org.apache.doris.transaction.GlobalTransactionMgrIface;
+import org.apache.doris.transaction.TabletCommitInfo;
+import org.apache.doris.transaction.TxnCommitAttachment;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+
+class CloudTransactionCommitRetryTest {
+ private static final long TXN_ID = 123L;
+ private static final long TIMEOUT_MS = 1000L;
+ private final Database db = new Database(1L, "retry_test");
+ private final List<Table> tables = Collections.emptyList();
+ private final List<TabletCommitInfo> commitInfos = Collections.emptyList();
+ private final List<TableStreamUpdateInfo> streamUpdates =
Collections.singletonList(
+ Mockito.mock(TableStreamUpdateInfo.class));
+ private final TxnCommitAttachment attachment = new
StreamingTaskTxnCommitAttachment();
+ private CloudGlobalTransactionMgr manager;
+ private Map<Long, Long> signatures;
+ private int previousRetryTimes;
+
+ @BeforeEach
+ void setUp() {
+ previousRetryTimes = Config.mow_calculate_delete_bitmap_retry_times;
+ Config.mow_calculate_delete_bitmap_retry_times = 3;
+ manager = Mockito.spy(new CloudGlobalTransactionMgr());
+ signatures = Deencapsulation.getField(manager, "txnLastSignatureMap");
+ }
+
+ @AfterEach
+ void tearDown() {
+ Config.mow_calculate_delete_bitmap_retry_times = previousRetryTimes;
+ }
+
+ @Test
+ void testRetryPreservesAttachmentStreamUpdatesAndSignature() throws
Exception {
+ AtomicInteger attempts = new AtomicInteger();
+ Mockito.doAnswer(invocation -> {
+ Assertions.assertSame(attachment, invocation.getArgument(5));
+ Assertions.assertSame(streamUpdates, invocation.getArgument(6));
+ if (attempts.incrementAndGet() == 1) {
+ signatures.put(TXN_ID, 456L);
+ throw retryableError();
+ }
+ Assertions.assertEquals(Long.valueOf(456L),
signatures.get(TXN_ID));
+ return true;
+ }).when(manager).commitAndPublishTransaction(db, tables, TXN_ID,
commitInfos, TIMEOUT_MS,
+ attachment, streamUpdates);
+
+ Assertions.assertTrue(manager.commitAndPublishTransactionWithRetry(db,
tables, TXN_ID,
+ commitInfos, TIMEOUT_MS, attachment, streamUpdates));
+ Assertions.assertEquals(2, attempts.get());
+ Assertions.assertFalse(signatures.containsKey(TXN_ID));
+ }
+
+ @Test
+ void testRetryExhaustionPreservesAttemptLimitAndCleansSignature() throws
Exception {
+ UserException failure = retryableError();
+ signatures.put(TXN_ID, 456L);
+ Mockito.doThrow(failure).when(manager).commitAndPublishTransaction(db,
tables, TXN_ID,
+ commitInfos, TIMEOUT_MS, attachment, streamUpdates);
+
+ Assertions.assertSame(failure,
Assertions.assertThrows(UserException.class,
+ () -> manager.commitAndPublishTransactionWithRetry(db, tables,
TXN_ID,
+ commitInfos, TIMEOUT_MS, attachment, streamUpdates)));
+ Mockito.verify(manager,
Mockito.times(3)).commitAndPublishTransaction(db, tables, TXN_ID,
+ commitInfos, TIMEOUT_MS, attachment, streamUpdates);
+ Assertions.assertFalse(signatures.containsKey(TXN_ID));
+ }
+
+ @Test
+ void testNonRetryableErrorIsPropagatedImmediately() throws Exception {
+ UserException failure = new UserException("injected non-retryable
error");
+ signatures.put(TXN_ID, 456L);
+ Mockito.doThrow(failure).when(manager).commitAndPublishTransaction(db,
tables, TXN_ID,
+ commitInfos, TIMEOUT_MS, attachment, streamUpdates);
+
+ Assertions.assertSame(failure,
Assertions.assertThrows(UserException.class,
+ () -> manager.commitAndPublishTransactionWithRetry(db, tables,
TXN_ID,
+ commitInfos, TIMEOUT_MS, attachment, streamUpdates)));
+ Mockito.verify(manager).commitAndPublishTransaction(db, tables, TXN_ID,
+ commitInfos, TIMEOUT_MS, attachment, streamUpdates);
+ Assertions.assertFalse(signatures.containsKey(TXN_ID));
+ }
+
+ @Test
+ void testFiveArgumentEntryStillRetriesWithoutAttachment() throws Exception
{
+
Mockito.doThrow(retryableError()).doReturn(true).when(manager).commitAndPublishTransaction(
+ db, tables, TXN_ID, commitInfos, TIMEOUT_MS, null,
Collections.emptyList());
+
+ Assertions.assertTrue(manager.commitAndPublishTransaction(db, tables,
TXN_ID, commitInfos, TIMEOUT_MS));
+ Mockito.verify(manager,
Mockito.times(2)).commitAndPublishTransaction(db, tables, TXN_ID,
+ commitInfos, TIMEOUT_MS, null, Collections.emptyList());
+ }
+
+ @Test
+ void testBeCommitEntryLeavesRetryAndSignatureToCaller() throws Exception {
+ UserException failure = retryableError();
+ signatures.put(TXN_ID, 456L);
+ Mockito.doThrow(failure).when(manager).commitAndPublishTransaction(db,
tables, TXN_ID,
+ commitInfos, TIMEOUT_MS, attachment, Collections.emptyList());
+
+ Assertions.assertSame(failure,
Assertions.assertThrows(UserException.class,
+ () -> manager.commitAndPublishTransaction(db, tables, TXN_ID,
+ commitInfos, TIMEOUT_MS, attachment)));
+ Mockito.verify(manager).commitAndPublishTransaction(db, tables, TXN_ID,
+ commitInfos, TIMEOUT_MS, attachment, Collections.emptyList());
+ Assertions.assertEquals(Long.valueOf(456L), signatures.get(TXN_ID));
+ }
+
+ @Test
+ void testSharedNothingDefaultPreservesPublishTimeoutResult() throws
Exception {
+ GlobalTransactionMgrIface localManager =
Mockito.mock(GlobalTransactionMgrIface.class,
+ Mockito.CALLS_REAL_METHODS);
+ Mockito.when(localManager.commitAndPublishTransaction(db, tables,
TXN_ID,
+ commitInfos, TIMEOUT_MS, attachment)).thenReturn(false);
+
+
Assertions.assertFalse(localManager.commitAndPublishTransactionWithRetry(db,
tables, TXN_ID,
+ commitInfos, TIMEOUT_MS, attachment, streamUpdates));
+ Mockito.verify(localManager).commitAndPublishTransaction(db, tables,
TXN_ID,
+ commitInfos, TIMEOUT_MS, attachment);
+ }
+
+ private UserException retryableError() {
+ return new UserException(InternalErrorCode.DELETE_BITMAP_LOCK_ERR,
+ "Failed to calculate delete bitmap. Timeout.");
+ }
+}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/OlapInsertExecutorTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/OlapInsertExecutorTest.java
index d45cc155fea..d2d792541dc 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/OlapInsertExecutorTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/OlapInsertExecutorTest.java
@@ -86,9 +86,9 @@ class OlapInsertExecutorTest {
prepareFactoryMocks(envFactoryMock, envMock, coordinator, txnMgr,
txnState, currentEnv);
ctx.setEnv(currentEnv);
- Mockito.when(txnMgr.commitAndPublishTransaction(
+ Mockito.when(txnMgr.commitAndPublishTransactionWithRetry(
Mockito.any(), Mockito.anyList(), Mockito.anyLong(),
Mockito.anyList(), Mockito.anyLong(),
- Mockito.isNull())).thenReturn(false);
+ Mockito.isNull(), Mockito.anyList())).thenReturn(false);
OlapInsertExecutor executor = createExecutor(ctx);
executor.txnId = 10001L;
@@ -131,9 +131,9 @@ class OlapInsertExecutorTest {
prepareFactoryMocks(envFactoryMock, envMock, coordinator, txnMgr,
txnState, currentEnv);
ctx.setEnv(currentEnv);
- Mockito.when(txnMgr.commitAndPublishTransaction(
+ Mockito.when(txnMgr.commitAndPublishTransactionWithRetry(
Mockito.any(), Mockito.anyList(), Mockito.anyLong(),
Mockito.anyList(), Mockito.anyLong(),
- Mockito.isNull())).thenReturn(false);
+ Mockito.isNull(), Mockito.anyList())).thenReturn(false);
OlapInsertExecutor executor = createExecutor(ctx);
executor.txnId = 10002L;
@@ -155,6 +155,33 @@ class OlapInsertExecutorTest {
}
}
+ @Test
+ void testOrdinaryInsertUsesRetryEntry() throws Exception {
+ ConnectContext ctx = createExecutorContext();
+ Coordinator coordinator = createCoordinator();
+ GlobalTransactionMgrIface txnMgr =
Mockito.mock(GlobalTransactionMgrIface.class);
+ TransactionState txnState = Mockito.mock(TransactionState.class);
+ Env currentEnv = createCurrentEnv(Mockito.mock(LoadManager.class));
+
+ try (MockedStatic<EnvFactory> envFactoryMock =
Mockito.mockStatic(EnvFactory.class);
+ MockedStatic<Env> envMock = Mockito.mockStatic(Env.class)) {
+ prepareFactoryMocks(envFactoryMock, envMock, coordinator, txnMgr,
txnState, currentEnv);
+ ctx.setEnv(currentEnv);
+ Mockito.when(txnMgr.commitAndPublishTransactionWithRetry(
+ Mockito.any(), Mockito.anyList(), Mockito.anyLong(),
Mockito.anyList(), Mockito.anyLong(),
+ Mockito.isNull(), Mockito.anyList())).thenReturn(true);
+
+ OlapInsertExecutor executor = createExecutor(ctx);
+ executor.txnId = 10006L;
+ executor.onComplete();
+
+ Mockito.verify(txnMgr).commitAndPublishTransactionWithRetry(
+ Mockito.eq(executor.getDatabase()), Mockito.anyList(),
Mockito.eq(10006L),
+ Mockito.anyList(), Mockito.anyLong(), Mockito.isNull(),
Mockito.anyList());
+ Assertions.assertEquals(TransactionStatus.VISIBLE,
executor.txnStatus);
+ }
+ }
+
@Test
void testOnFailAbortsUncommittedTransaction() throws Exception {
ConnectContext ctx = createExecutorContext();
@@ -226,7 +253,7 @@ class OlapInsertExecutorTest {
MockedStatic<Env> envMock = Mockito.mockStatic(Env.class)) {
prepareFactoryMocks(envFactoryMock, envMock, coordinator, txnMgr,
txnState, currentEnv);
ctx.setEnv(currentEnv);
- Mockito.when(txnMgr.commitAndPublishTransaction(
+ Mockito.when(txnMgr.commitAndPublishTransactionWithRetry(
Mockito.any(), Mockito.anyList(), Mockito.anyLong(),
Mockito.anyList(), Mockito.anyLong(),
Mockito.isNull(),
Mockito.eq(streamUpdateInfos))).thenReturn(true);
@@ -236,7 +263,7 @@ class OlapInsertExecutorTest {
executor.executeSingleInsert(stmtExecutor);
Mockito.verify(coordinator, Mockito.never()).exec();
- Mockito.verify(txnMgr).commitAndPublishTransaction(
+ Mockito.verify(txnMgr).commitAndPublishTransactionWithRetry(
Mockito.eq(executor.getDatabase()), Mockito.anyList(),
Mockito.eq(10005L),
Mockito.argThat(List::isEmpty), Mockito.anyLong(),
Mockito.isNull(),
Mockito.eq(streamUpdateInfos));
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
