This is an automated email from the ASF dual-hosted git repository.
morningman 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 33003c6a74d [fix](fe) Finish a failed prepared EXECUTE like a failed
COM_QUERY (#69566)
33003c6a74d is described below
commit 33003c6a74d2c244e7bdbe6e3bf0dfeff3481aa4
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Sun Oct 25 08:06:07 2026 +0000
[fix](fe) Finish a failed prepared EXECUTE like a failed COM_QUERY (#69566)
### What problem does this PR solve?
Issue Number: None
Related PR: #66473, #66530
Problem Summary:
**In short.** A prepared statement's `COM_STMT_EXECUTE` that fails after
its scans were planned leaves its query-finish callbacks registered for
good, keeps its connector scope open until the statement's next EXECUTE
(or forever), and never stops the scan nodes left for the end of the
statement. This PR ends every EXECUTE the way a COM_QUERY statement
ends, which is what branch-4.x (#66473) does in `handleExecute`.
**Background.** A plugin scan registers a query-finish callback while it
plans its splits (`PluginDrivenScanNode.getSplits` →
`QueryFinishCleanup`, keyed by the query id). The callback releases the
connector's per-query read transaction (Hive full-ACID and insert-only
reads open a metastore read transaction with a shared read lock during
`planScan`) and closes the statement's connector scope, which holds
connector metadata and any connector write transaction the executor has
not finished. It runs when the query is unregistered:
`StmtExecutor.finalizeQuery` → `QeProcessorImpl.unregisterQuery` →
`QueryFinishCallbackRegistry.runAndClear`. A COM_QUERY statement also
closes its `StatementContext` in `ConnectProcessor`'s per-statement
finally, which closes the scope of statements that never registered a
callback and stops the scan nodes registered to stop at the statement
end (a remote Doris scan sends its CloseSession there).
**The problem, and what it cost.** `MysqlConnectProcessor.handleExecute`
has neither. An execution that fails after planning never reaches
`finalizeQuery`: a scan-level SQL block rule (`checkBlockRulesByScan`
runs after `planner.plan` and before `handleQueryWithRetry`), or a
prepared `INSERT ... SELECT` whose transaction fails to begin.
| What the failed EXECUTE leaves behind | Master today | After this PR |
|---|---|---|
| Its `QueryFinishCleanup` in the process-wide registry | kept for good:
the next EXECUTE has a new query id, and COM_STMT_CLOSE and a connection
reset do not touch the registry | run and removed when the EXECUTE ends
|
| A connector read transaction / shared read lock opened while planning
| not released by Doris | released when the EXECUTE ends |
| Its connector scope (metadata, a connector write transaction never
committed) | closed at the statement's next EXECUTE
(`PreparedStatementContext.nextStatementContext`), never if there is
none | closed when the EXECUTE ends |
| Scan nodes registered to stop at the statement end | never stopped:
`StatementContext.close()` never runs for an EXECUTE | stopped when the
EXECUTE ends |
**How this PR fixes it.** `handleExecute` runs the execution in a `try`
whose `finally` unregisters the execution's query id and then closes the
statement context that `ExecuteCommand` ran it with, which is the
prepared statement's current context
(`PreparedStatementContext.nextStatementContext`). Unregistering runs
and drops the callbacks still registered for the query id, so they run
in the same order as for a finished query: read-transaction release
first, then the scope. Both calls are no-ops for an execution that
finished normally: `finalizeQuery` already unregistered its query id
(`runAndClear` finds nothing), and the scope's `closeAll` is close-once.
The master-side forwarded path already closes the statement context
after `handleExecute`, and closing it twice is a no-op. A
binary-protocol execution has sent its whole result when `execute`
returns (Arrow Flight does not use this path), so nothing still running
is closed.
**Results.** `MysqlConnectProcessorExecuteCleanupTest` (new) drives
`handleExecute` with an executor that does what planning leaves behind:
- an EXECUTE that registers a query-finish callback and is then rejected
(a SQL block rule) runs that callback once and closes its scope;
- an EXECUTE that fails before any scan registered closes its scope;
- a successful EXECUTE, whose query is unregistered by `finalizeQuery`,
runs its callback once and closes its scope value once.
With `handleExecute` of master, the first two fail (`expected: <1> but
was: <0>` for the callback and for the scope close) and the third
passes. `MysqlConnectProcessorLongDataTest` and
`MysqlConnectProcessorCursorFetchTest` mock the executor so that it
never executes; their connections now carry the query id the statement's
PREPARE left, as a real connection does. These 3 classes and
`ExecuteCommandTest`, `ConnectProcessorForwardProtocolTest`,
`ConnectProcessorRetryTest`, `StatementContextTest`,
`QueryFinishCallbackRegistryTest`,
`PluginDrivenScanNodeReadTxnReleaseTest`, `AuditLogWorkloadGroupTest`
and `TaskProcessorTest` pass (77 tests).
---
.../org/apache/doris/qe/MysqlConnectProcessor.java | 27 +++-
.../qe/MysqlConnectProcessorCursorFetchTest.java | 4 +
.../MysqlConnectProcessorExecuteCleanupTest.java | 161 +++++++++++++++++++++
.../qe/MysqlConnectProcessorLongDataTest.java | 4 +
4 files changed, 192 insertions(+), 4 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/qe/MysqlConnectProcessor.java
b/fe/fe-core/src/main/java/org/apache/doris/qe/MysqlConnectProcessor.java
index 79e57b2316f..c5a0931df1e 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/MysqlConnectProcessor.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/MysqlConnectProcessor.java
@@ -255,10 +255,14 @@ public class MysqlConnectProcessor extends
ConnectProcessor {
stmt.setOrigStmt(prepareCommand.getOriginalStmt());
executor = new StmtExecutor(ctx, stmt, true);
ctx.setExecutor(executor);
- if (null != queryId) {
- executor.execute(queryId);
- } else {
- executor.execute();
+ try {
+ if (null != queryId) {
+ executor.execute(queryId);
+ } else {
+ executor.execute();
+ }
+ } finally {
+ finishExecution(prepCtx);
}
if (ctx.getSessionVariable().isEnablePreparedStmtAuditLog()) {
stmtStr = executeStmt.toSql();
@@ -281,6 +285,21 @@ public class MysqlConnectProcessor extends
ConnectProcessor {
}
}
+ /**
+ * COM_STMT_EXECUTE has no per-statement close like the one COM_QUERY runs
in its finally. An execution that
+ * fails after its scans were planned, such as one a scan-level SQL block
rule rejects or a prepared
+ * INSERT ... SELECT whose transaction fails to begin, never reaches
unregisterQuery: the query-finish callbacks
+ * its scans registered (releasing a connector read transaction, closing
the connector scope) would stay in the
+ * registry for good, its connector scope would stay open until the
statement's next EXECUTE, and scan nodes left
+ * for the statement end would never stop. The context ExecuteCommand ran
this execution with is the prepared
+ * statement's current one (see
PreparedStatementContext#nextStatementContext). Both calls are no-ops for an
+ * execution that finished normally, which already unregistered its query
and closed its scope.
+ */
+ private void finishExecution(PreparedStatementContext prepCtx) {
+ QeProcessorImpl.INSTANCE.unregisterQuery(ctx.queryId());
+ prepCtx.getStatementContext().close();
+ }
+
// process COM_EXECUTE, parse binary row data
//
https://dev.mysql.com/doc/dev/mysql-server/latest/page_protocol_com_stmt_execute.html
private void handleExecute() {
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/qe/MysqlConnectProcessorCursorFetchTest.java
b/fe/fe-core/src/test/java/org/apache/doris/qe/MysqlConnectProcessorCursorFetchTest.java
index 1694ccb3185..03632297903 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/qe/MysqlConnectProcessorCursorFetchTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/qe/MysqlConnectProcessorCursorFetchTest.java
@@ -20,6 +20,7 @@ package org.apache.doris.qe;
import org.apache.doris.mysql.MysqlCommand;
import org.apache.doris.nereids.StatementContext;
import org.apache.doris.nereids.trees.plans.commands.PrepareCommand;
+import org.apache.doris.thrift.TUniqueId;
import com.google.common.collect.ImmutableMap;
import org.junit.jupiter.api.Assertions;
@@ -46,6 +47,9 @@ public class MysqlConnectProcessorCursorFetchTest {
StatementContext statementContext = new StatementContext();
PreparedStatementContext prepared = new PreparedStatementContext(
command, context, statementContext, "select 1");
+ // The query id the COM_STMT_PREPARE of the statement left on the
connection; the mocked executor
+ // never executes, so it does not set one.
+ context.setQueryId(new TUniqueId(7L, 7L));
ByteBuffer packet = ByteBuffer.allocate(9);
packet.position(packet.limit()); // COM_STMT_EXECUTE header
consumed, no parameter payload.
try (MockedConstruction<StmtExecutor> executors =
Mockito.mockConstruction(StmtExecutor.class);
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/qe/MysqlConnectProcessorExecuteCleanupTest.java
b/fe/fe-core/src/test/java/org/apache/doris/qe/MysqlConnectProcessorExecuteCleanupTest.java
new file mode 100644
index 00000000000..fc57877b9d9
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/qe/MysqlConnectProcessorExecuteCleanupTest.java
@@ -0,0 +1,161 @@
+// 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.qe;
+
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.util.DebugUtil;
+import org.apache.doris.connector.spi.ConnectorStatementScope;
+import org.apache.doris.mysql.MysqlCommand;
+import org.apache.doris.nereids.StatementContext;
+import org.apache.doris.nereids.trees.plans.commands.PrepareCommand;
+import org.apache.doris.qe.QueryState.MysqlStateType;
+import org.apache.doris.thrift.TUniqueId;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedConstruction;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+import java.nio.ByteBuffer;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * A COM_STMT_EXECUTE ends with what COM_QUERY's per-statement finally does:
the query-finish callbacks of the
+ * execution run and its connector scope closes, also when the execution fails
after planning.
+ */
+public class MysqlConnectProcessorExecuteCleanupTest {
+
+ @Test
+ public void testExecuteRejectedAfterPlanningRunsItsQueryFinishCallbacks()
throws Exception {
+ // PluginDrivenScanNode.getSplits registers a query-finish callback
while planning; a scan-level SQL block
+ // rule then rejects the plan, so the execution never reaches
unregisterQuery.
+ ConnectContext context = newContext();
+ PreparedStatementContext prepared = newPrepared(context);
+ TUniqueId queryId = new TUniqueId(31L, 1L);
+ AtomicInteger callbackRuns = new AtomicInteger();
+ CloseCounter scopeValue = new CloseCounter();
+ try (MockedConstruction<StmtExecutor> executors =
Mockito.mockConstruction(StmtExecutor.class,
+ (executor, init) -> Mockito.doAnswer(invocation -> {
+ ConnectorStatementScope scope = startExecution(context,
prepared, queryId, scopeValue);
+
QeProcessorImpl.INSTANCE.registerQueryFinishCallback(DebugUtil.printId(queryId),
() -> {
+ callbackRuns.incrementAndGet();
+ scope.closeAll();
+ });
+ throw new AnalysisException("sql hits sql block rule");
+ }).when(executor).execute());
+ MockedStatic<AuditLogHelper> audit =
Mockito.mockStatic(AuditLogHelper.class)) {
+ execute(context, prepared);
+
+ Assertions.assertEquals(MysqlStateType.ERR,
context.getState().getStateType());
+ Assertions.assertEquals(1, callbackRuns.get(), "the failed
execution's query-finish callback must run");
+ Assertions.assertEquals(1, scopeValue.closes.get(), "the failed
execution's connector scope must close");
+ } finally {
+ // Keep the process-wide registry clean even when an assertion
above failed.
+ QeProcessorImpl.INSTANCE.unregisterQuery(queryId);
+ }
+ }
+
+ @Test
+ public void testExecuteFailingBeforeAnyScanClosesItsConnectorScope()
throws Exception {
+ // A connector opened statement-scoped state during analysis and the
execution failed before a scan
+ // registered a query-finish callback: only the end of the execution
can close the scope.
+ ConnectContext context = newContext();
+ PreparedStatementContext prepared = newPrepared(context);
+ TUniqueId queryId = new TUniqueId(31L, 2L);
+ CloseCounter scopeValue = new CloseCounter();
+ try (MockedConstruction<StmtExecutor> executors =
Mockito.mockConstruction(StmtExecutor.class,
+ (executor, init) -> Mockito.doAnswer(invocation -> {
+ startExecution(context, prepared, queryId, scopeValue);
+ throw new AnalysisException("analysis failed");
+ }).when(executor).execute());
+ MockedStatic<AuditLogHelper> audit =
Mockito.mockStatic(AuditLogHelper.class)) {
+ execute(context, prepared);
+
+ Assertions.assertEquals(1, scopeValue.closes.get(), "the failed
execution's connector scope must close");
+ }
+ }
+
+ @Test
+ public void testSuccessfulExecuteClosesItsScopeOnce() throws Exception {
+ // finalizeQuery unregisters a finished execution, which runs its
callbacks and closes its scope; the end of
+ // the execution must then neither run the callbacks again nor close a
value twice.
+ ConnectContext context = newContext();
+ PreparedStatementContext prepared = newPrepared(context);
+ TUniqueId queryId = new TUniqueId(31L, 3L);
+ AtomicInteger callbackRuns = new AtomicInteger();
+ CloseCounter scopeValue = new CloseCounter();
+ try (MockedConstruction<StmtExecutor> executors =
Mockito.mockConstruction(StmtExecutor.class,
+ (executor, init) -> Mockito.doAnswer(invocation -> {
+ ConnectorStatementScope scope = startExecution(context,
prepared, queryId, scopeValue);
+
QeProcessorImpl.INSTANCE.registerQueryFinishCallback(DebugUtil.printId(queryId),
() -> {
+ callbackRuns.incrementAndGet();
+ scope.closeAll();
+ });
+ QeProcessorImpl.INSTANCE.unregisterQuery(queryId);
+ return null;
+ }).when(executor).execute());
+ MockedStatic<AuditLogHelper> audit =
Mockito.mockStatic(AuditLogHelper.class)) {
+ execute(context, prepared);
+
+ Assertions.assertEquals(1, callbackRuns.get());
+ Assertions.assertEquals(1, scopeValue.closes.get());
+ } finally {
+ QeProcessorImpl.INSTANCE.unregisterQuery(queryId);
+ }
+ }
+
+ private static ConnectContext newContext() {
+ ConnectContext context = new ConnectContext(null, false);
+ context.setCommand(MysqlCommand.COM_STMT_EXECUTE);
+ return context;
+ }
+
+ private static PreparedStatementContext newPrepared(ConnectContext
context) {
+ PrepareCommand command = Mockito.mock(PrepareCommand.class);
+ Mockito.when(command.getOriginalStmt()).thenReturn(new
OriginStatement("select 1", 0));
+ return new PreparedStatementContext(
+ command, context, new StatementContext(context, new
OriginStatement("select 1", 0)), "select 1");
+ }
+
+ private static void execute(ConnectContext context,
PreparedStatementContext prepared) {
+ ByteBuffer packet = ByteBuffer.allocate(9);
+ packet.position(packet.limit()); // COM_STMT_EXECUTE header consumed,
no parameter payload.
+ new MysqlConnectProcessor(context).handleExecute(prepared.command, 7,
prepared, packet, null);
+ }
+
+ /** What ExecuteCommand.run and StmtExecutor set up for one execution,
with a closeable value in its scope. */
+ private static ConnectorStatementScope startExecution(ConnectContext
context, PreparedStatementContext prepared,
+ TUniqueId queryId, CloseCounter scopeValue) {
+ StatementContext execution = prepared.nextStatementContext();
+ context.setStatementContext(execution);
+ context.setQueryId(queryId);
+ ConnectorStatementScope scope =
execution.getOrCreateConnectorStatementScope();
+ scope.computeIfAbsent("metadata", () -> scopeValue);
+ return scope;
+ }
+
+ private static final class CloseCounter implements AutoCloseable {
+ private final AtomicInteger closes = new AtomicInteger();
+
+ @Override
+ public void close() {
+ closes.incrementAndGet();
+ }
+ }
+}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/qe/MysqlConnectProcessorLongDataTest.java
b/fe/fe-core/src/test/java/org/apache/doris/qe/MysqlConnectProcessorLongDataTest.java
index d649f469a7d..9c214a76d78 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/qe/MysqlConnectProcessorLongDataTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/qe/MysqlConnectProcessorLongDataTest.java
@@ -34,6 +34,7 @@ import org.apache.doris.nereids.trees.plans.PlaceholderId;
import org.apache.doris.nereids.trees.plans.PrepareCommandPlanner;
import org.apache.doris.nereids.trees.plans.commands.PrepareCommand;
import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.thrift.TUniqueId;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -312,6 +313,9 @@ public class MysqlConnectProcessorLongDataTest {
new OriginStatement("select ?", 0));
prepared = new PreparedStatementContext(command, context, new
StatementContext(), "select ?");
context.addPreparedStatementContext("7", prepared);
+ // The query id the COM_STMT_PREPARE of the statement left on the
connection; the mocked executors
+ // below never execute, so they do not set one.
+ context.setQueryId(new TUniqueId(7L, 7L));
processor = new MysqlConnectProcessor(context) {
@Override
protected void resolveWorkloadGroupName() {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
