This is an automated email from the ASF dual-hosted git repository.
luwei16 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 ded08aebefd [fix](binlog) Commit table stream offsets for empty
inserts (#67126)
ded08aebefd is described below
commit ded08aebefdb76b167c1f5fa164feaa1b4732205
Author: Luwei <[email protected]>
AuthorDate: Mon Aug 31 17:04:14 2026 +0800
[fix](binlog) Commit table stream offsets for empty inserts (#67126)
### What problem does this PR solve?
Issue Number: close #66273
Related PR: None
Problem Summary: When the optimizer reduces an INSERT INTO SELECT from a
Table Stream to an empty relation, the previous empty-insert shortcut
bypasses the transaction lifecycle. The target receives zero rows, but
the stream offset is not committed, so the same stream state can be read
repeatedly. Keep empty Table Stream reads in the transaction lifecycle,
skip coordinator data execution, and commit the offset atomically.
Ordinary empty inserts continue to use the existing fast-return path.
### Release note
Table Stream offsets now advance when INSERT INTO SELECT succeeds with
zero rows.
### Check List (For Author)
- Test: Unit Test
- InsertIntoTableCommandTableStreamTest and OlapInsertExecutorTest (21
tests)
- ./build.sh --fe -j 192
- Behavior changed: Yes (successful zero-row Table Stream inserts now
commit offsets; ordinary inserts are unchanged)
- Does this need documentation: No
---
.../commands/insert/AbstractInsertExecutor.java | 12 +++-
.../commands/insert/InsertIntoTableCommand.java | 6 +-
.../InsertIntoTableCommandTableStreamTest.java | 77 ++++++++++++++++++++++
.../commands/insert/OlapInsertExecutorTest.java | 42 +++++++++++-
4 files changed, 132 insertions(+), 5 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/AbstractInsertExecutor.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/AbstractInsertExecutor.java
index 4ea4fad8dd2..52d596315c6 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/AbstractInsertExecutor.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/AbstractInsertExecutor.java
@@ -268,7 +268,9 @@ public abstract class AbstractInsertExecutor {
// Pre-execution work may register external resources, so it must
share the transaction cleanup scope.
beforeExec();
executor.updateProfile(false);
- execImpl(executor);
+ if (!emptyInsert) {
+ execImpl(executor);
+ }
checkStrictModeAndFilterRatio();
for (InsertExecutorListener listener : listeners) {
listener.beforeComplete(this, executor, jobId);
@@ -300,6 +302,14 @@ public abstract class AbstractInsertExecutor {
return emptyInsert;
}
+ /**
+ * Return whether this insert needs its transaction lifecycle. A Table
Stream offset update
+ * must be committed even when optimization proves that the target
receives no rows.
+ */
+ public boolean requiresTransaction() {
+ return !emptyInsert || !streamUpdateInfos.isEmpty();
+ }
+
public void setStreamUpdateInfos(List<TableStreamUpdateInfo>
streamUpdateInfos) {
this.streamUpdateInfos = streamUpdateInfos;
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java
index 4a1440872f0..b7bb29a62f1 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java
@@ -335,7 +335,7 @@ public class InsertIntoTableCommand extends Command
implements NeedAuditEncrypti
newestTargetTableIf.readUnlock();
continue;
}
- if (!insertExecutor.isEmptyInsert()) {
+ if (insertExecutor.requiresTransaction()) {
insertExecutor.beginTransaction();
insertExecutor.finalizeSink(
buildResult.planner.getFragments().get(0),
buildResult.dataSink,
@@ -688,8 +688,8 @@ public class InsertIntoTableCommand extends Command
implements NeedAuditEncrypti
private void runInternal(ConnectContext ctx, StmtExecutor executor) throws
Exception {
AbstractInsertExecutor insertExecutor = initPlan(ctx, executor);
- // if the insert stmt data source is empty, directly return, no need
to be executed.
- if (insertExecutor.isEmptyInsert()) {
+ // An empty Table Stream read still needs to commit its offset update
atomically.
+ if (!insertExecutor.requiresTransaction()) {
return;
}
if (insertExecutorListener != null) {
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommandTableStreamTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommandTableStreamTest.java
index 0d989e9e98a..f46ec641663 100755
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommandTableStreamTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommandTableStreamTest.java
@@ -46,6 +46,7 @@ import
org.apache.doris.nereids.trees.plans.logical.LogicalOlapTableStreamScan;
import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
import org.apache.doris.nereids.util.PlanChecker;
import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.QueryState.MysqlStateType;
import org.apache.doris.qe.StmtExecutor;
import org.apache.doris.system.Backend;
import org.apache.doris.system.SystemInfoService;
@@ -190,6 +191,82 @@ public class InsertIntoTableCommandTableStreamTest extends
TestWithFeService {
Assertions.assertEquals(update.getNext(), ((OlapTableStreamUpdate)
txnUpdate).getNext());
}
+ @Test
+ public void testEmptyInsertStartsTransactionForStreamOffsetUpdate() throws
Exception {
+ String sql = "insert into test_stream.tbl_target "
+ + "select * from test_stream.s1 where false";
+ InsertIntoTableCommand command = (InsertIntoTableCommand)
parser.parseSingle(sql);
+
+ resetQueryContext();
+ AbstractInsertExecutor insertExecutor = command.initPlan(
+ connectContext, new StmtExecutor(connectContext, sql), true);
+ try {
+ Assertions.assertTrue(insertExecutor.isEmptyInsert());
+
Assertions.assertFalse(insertExecutor.getStreamUpdateInfos().isEmpty());
+ Assertions.assertNotEquals(AbstractInsertExecutor.INVALID_TXN_ID,
insertExecutor.getTxnId());
+ } finally {
+ insertExecutor.onFail(new RuntimeException("test cleanup"));
+ resetQueryContext();
+ }
+ }
+
+ @Test
+ public void testOrdinaryEmptyInsertStillSkipsTransaction() throws
Exception {
+ String sql = "insert into test_stream.tbl_target "
+ + "select * from test_stream.tbl_stream_base where false";
+ InsertIntoTableCommand command = (InsertIntoTableCommand)
parser.parseSingle(sql);
+
+ resetQueryContext();
+ AbstractInsertExecutor insertExecutor = command.initPlan(
+ connectContext, new StmtExecutor(connectContext, sql), true);
+
+ Assertions.assertTrue(insertExecutor.isEmptyInsert());
+ Assertions.assertTrue(insertExecutor.getStreamUpdateInfos().isEmpty());
+ Assertions.assertFalse(insertExecutor.requiresTransaction());
+ Assertions.assertEquals(AbstractInsertExecutor.INVALID_TXN_ID,
insertExecutor.getTxnId());
+ }
+
+ @Test
+ public void testEmptyInsertCommitsStreamOffsetUpdate() throws Exception {
+ Database db = (Database)
Env.getCurrentInternalCatalog().getDbOrMetaException("test_stream");
+ OlapTable baseTable = (OlapTable)
db.getTableOrMetaException("tbl_stream_base");
+ createTable("create stream if not exists
test_stream.s_empty_insert_commit "
+ + "on table test_stream.tbl_stream_base
properties('show_initial_rows' = 'false')");
+ OlapTableStream stream = (OlapTableStream)
db.getTableOrMetaException("s_empty_insert_commit");
+ Map<Long, Long> originalTso = new HashMap<>();
+ Map<Long, Long> expectedTso = new HashMap<>();
+
+ for (Partition partition : baseTable.getPartitions()) {
+ originalTso.put(partition.getId(), partition.getTso());
+ long nextTso = partition.getTso() + 1000;
+ expectedTso.put(partition.getId(), nextTso);
+ partition.setVisibleVersionAndTime(
+ partition.getVisibleVersion(),
partition.getVisibleVersionTime(), nextTso);
+ }
+
+ String sql = "insert into test_stream.tbl_target "
+ + "select * from test_stream.s_empty_insert_commit where
false";
+ InsertIntoTableCommand command = (InsertIntoTableCommand)
parser.parseSingle(sql);
+ resetQueryContext();
+ connectContext.getState().reset();
+ connectContext.resetReturnRows();
+ try {
+ command.run(connectContext, new StmtExecutor(connectContext, sql));
+
+ Assertions.assertEquals(MysqlStateType.OK,
connectContext.getState().getStateType());
+ Assertions.assertEquals(0L, connectContext.getReturnRows());
+ for (Map.Entry<Long, Long> entry : expectedTso.entrySet()) {
+ Assertions.assertEquals(entry.getValue(),
stream.getStreamUpdate(entry.getKey()).first);
+ }
+ } finally {
+ for (Partition partition : baseTable.getPartitions()) {
+
partition.setVisibleVersionAndTime(partition.getVisibleVersion(),
+ partition.getVisibleVersionTime(),
originalTso.get(partition.getId()));
+ }
+ resetQueryContext();
+ }
+ }
+
@Test
public void testInitPlanCollectsUpdatesForTwoStreams() throws Exception {
Database db = (Database)
Env.getCurrentInternalCatalog().getDbOrMetaException("test_stream");
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 96fb43ffbae..3d7c9b1aa0f 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
@@ -22,6 +22,7 @@ import org.apache.doris.catalog.Database;
import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.EnvFactory;
import org.apache.doris.catalog.OlapTable;
+import org.apache.doris.catalog.stream.TableStreamUpdateInfo;
import org.apache.doris.common.Status;
import org.apache.doris.common.profile.ExecutionProfile;
import org.apache.doris.common.profile.Profile;
@@ -53,6 +54,7 @@ import org.junit.jupiter.api.Test;
import org.mockito.MockedStatic;
import org.mockito.Mockito;
+import java.util.List;
import java.util.Optional;
/**
@@ -208,6 +210,40 @@ class OlapInsertExecutorTest {
}
}
+ @Test
+ void testEmptyStreamInsertCommitsWithoutCoordinatorExecution() throws
Exception {
+ ConnectContext ctx = createExecutorContext();
+ Coordinator coordinator = createCoordinator();
+ GlobalTransactionMgrIface txnMgr =
Mockito.mock(GlobalTransactionMgrIface.class);
+ TransactionState txnState = Mockito.mock(TransactionState.class);
+ LoadManager loadManager = Mockito.mock(LoadManager.class);
+ Env currentEnv = createCurrentEnv(loadManager);
+ StmtExecutor stmtExecutor = createStmtExecutor();
+ List<TableStreamUpdateInfo> streamUpdateInfos =
List.of(Mockito.mock(TableStreamUpdateInfo.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.commitAndPublishTransaction(
+ Mockito.any(), Mockito.anyList(), Mockito.anyLong(),
Mockito.anyList(), Mockito.anyLong(),
+ Mockito.isNull(),
Mockito.eq(streamUpdateInfos))).thenReturn(true);
+
+ OlapInsertExecutor executor = createExecutor(ctx, true);
+ executor.txnId = 10005L;
+ executor.setStreamUpdateInfos(streamUpdateInfos);
+ executor.executeSingleInsert(stmtExecutor);
+
+ Mockito.verify(coordinator, Mockito.never()).exec();
+ Mockito.verify(txnMgr).commitAndPublishTransaction(
+ Mockito.eq(executor.getDatabase()), Mockito.anyList(),
Mockito.eq(10005L),
+ Mockito.argThat(List::isEmpty), Mockito.anyLong(),
Mockito.isNull(),
+ Mockito.eq(streamUpdateInfos));
+ Assertions.assertEquals(TransactionStatus.VISIBLE,
executor.txnStatus);
+ Assertions.assertEquals(0L, ctx.getReturnRows());
+ }
+ }
+
// Build a fresh context per case so insertResult and QueryState do not
leak between tests.
private ConnectContext createExecutorContext() {
ConnectContext ctx = new ConnectContext();
@@ -269,6 +305,10 @@ class OlapInsertExecutorTest {
// Create an executor with mocked table metadata because this test only
validates timeout result handling.
private OlapInsertExecutor createExecutor(ConnectContext ctx) {
+ return createExecutor(ctx, false);
+ }
+
+ private OlapInsertExecutor createExecutor(ConnectContext ctx, boolean
emptyInsert) {
Database database = Mockito.mock(Database.class);
Mockito.when(database.getFullName()).thenReturn("test_db");
Mockito.when(database.getId()).thenReturn(1L);
@@ -280,7 +320,7 @@ class OlapInsertExecutorTest {
Mockito.when(table.getId()).thenReturn(2L);
return new OlapInsertExecutor(ctx, table, "label_test",
Mockito.mock(NereidsPlanner.class),
- Optional.empty(), false, 0L);
+ Optional.empty(), emptyInsert, 0L);
}
private OlapInsertExecutor
createExecutorWithBeforeExecFailure(ConnectContext ctx) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
