From f1a13caee9e9ddcf879c8ac42cba7bffe02ad8fe Mon Sep 17 00:00:00 2001 From: Raghav Date: Thu, 3 Sep 2026 20:43:31 +0530 Subject: [PATCH 1/3] [fix](arrow-flight) Bound idle Arrow Flight SQL sessions separately from wait_timeout A Flight SQL session idles under the same wait_timeout as a MySQL connection (28800s by default). Since the coordinator of a BE-served Flight query is kept alive across GetFlightInfo -> DoGet until the session's next query or its close, an abandoned session - a client that opens a session per query and never sends CloseSession - keeps that query's workload-group queue slot for the whole wait_timeout. Eight such sessions fill a max_concurrency=8 group and every later query in it fails with "query queue timeout". Add a mutable FE config, arrow_flight_session_idle_timeout_second (default 3600), applied by the existing connection timeout checker to ARROW_FLIGHT_SQL contexts only as min(wait_timeout, max(config, exec timeout)). The exec-timeout floor matters: a Flight session is COM_SLEEP while the client drains the result via DoGet and its idle clock runs from the query's start, so a bound below query_timeout would kill a long result stream before the query's own timeout could. 0 disables the bound. MySQL-protocol connections are unchanged. The kill log line now reports the effective idle timeout. Signed-off-by: Raghvendra Singh Co-Authored-By: Claude Fable 5 --- .../java/org/apache/doris/common/Config.java | 9 ++ .../org/apache/doris/qe/ConnectContext.java | 15 +++- .../sessions/FlightSqlConnectContext.java | 23 +++++ .../FlightSqlSessionIdleTimeoutTest.java | 86 +++++++++++++++++++ 4 files changed, 130 insertions(+), 3 deletions(-) create mode 100644 fe/fe-core/src/test/java/org/apache/doris/service/arrowflight/sessions/FlightSqlSessionIdleTimeoutTest.java diff --git a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java index c0f2a1b76be716..cd2d3986c091cd 100644 --- a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java +++ b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java @@ -2644,6 +2644,15 @@ public class Config extends ConfigBase { @ConfField(description = "Maximum number of connections for the Arrow Flight Server per FE.") public static int arrow_flight_max_connections = 4096; + @ConfField(mutable = true, masterOnly = false, description = "Idle timeout for Arrow Flight SQL sessions, " + + "in seconds. A Flight session that has been sleeping longer than this is killed by the " + + "connection timeout checker, exactly as a MySQL connection is killed after wait_timeout, but " + + "with its own, shorter bound: a Flight query keeps its coordinator and its workload-group " + + "queue slot alive until the session's next query or its close, so an abandoned session holds " + + "a slot for as long as it lives. The effective idle bound is min(wait_timeout, this value). " + + "0 disables the Flight-specific bound (wait_timeout alone applies).") + public static int arrow_flight_session_idle_timeout_second = 3600; + @ConfField(mutable = true, masterOnly = true, description = "In auto bucketing, the number of buckets is " + "estimated based on the partition size. For storage " + "and computing integration, a partition size of 5GB " + "is estimated as one bucket, but for cloud, a " diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectContext.java b/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectContext.java index 428f86c957e92d..c467fa9136dda6 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectContext.java +++ b/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectContext.java @@ -1268,8 +1268,8 @@ public void kill(boolean killConnection) { private void killByTimeout(boolean killConnection) { if (killConnection) { LOG.warn("kill wait timeout connection, connection type: {}, connectionId: {}, remote: {}, " - + "wait timeout: {}", - getConnectType(), connectionId, getRemoteHostPortString(), sessionVariable.getWaitTimeoutS()); + + "idle timeout: {}", + getConnectType(), connectionId, getRemoteHostPortString(), getIdleTimeoutS()); killConnection(); } // Now, cancel running query. @@ -1292,6 +1292,15 @@ public void cancelQuery(Status cancelReason) { } } + /** + * How long this connection may sleep (COM_SLEEP) before the timeout checker kills it. + * The MySQL protocol bound is the session's wait_timeout; protocol-specific contexts may + * tighten it (never widen it) — see FlightSqlConnectContext. + */ + public long getIdleTimeoutS() { + return sessionVariable.getWaitTimeoutS(); + } + public void checkTimeout(long now) { if (startTime <= 0) { return; @@ -1301,7 +1310,7 @@ public void checkTimeout(long now) { boolean killFlag = false; boolean killConnection = false; if (command == MysqlCommand.COM_SLEEP) { - if (delta > sessionVariable.getWaitTimeoutS() * 1000L) { + if (delta > getIdleTimeoutS() * 1000L) { // Need kill this connection. killFlag = true; killConnection = true; diff --git a/fe/fe-core/src/main/java/org/apache/doris/service/arrowflight/sessions/FlightSqlConnectContext.java b/fe/fe-core/src/main/java/org/apache/doris/service/arrowflight/sessions/FlightSqlConnectContext.java index ceddfaa563988a..18a7043bed5219 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/service/arrowflight/sessions/FlightSqlConnectContext.java +++ b/fe/fe-core/src/main/java/org/apache/doris/service/arrowflight/sessions/FlightSqlConnectContext.java @@ -17,6 +17,7 @@ package org.apache.doris.service.arrowflight.sessions; +import org.apache.doris.common.Config; import org.apache.doris.common.Status; import org.apache.doris.mysql.MysqlChannel; import org.apache.doris.qe.ConnectContext; @@ -50,6 +51,28 @@ public FlightSqlChannel getFlightSqlChannel() { return flightSqlChannel; } + /** + * Flight sessions get their own idle bound (Config.arrow_flight_session_idle_timeout_second), + * tighter than the MySQL wait_timeout: an abandoned Flight session keeps its last query's + * coordinator — and that query's workload-group queue slot — alive until it is closed, so a + * client that opens a session per query and never closes it would otherwise pin a slot for + * the whole wait_timeout (8h by default). + * + * A Flight session is COM_SLEEP while the client drains the result from the BE (DoGet), and + * its idle clock runs from the query's START — so the bound is floored at the query's own + * execution timeout: a long result drain is never killed before query_timeout would kill the + * query itself. Effective bound = min(wait_timeout, max(config, exec timeout)); 0 disables it. + */ + @Override + public long getIdleTimeoutS() { + long waitTimeoutS = super.getIdleTimeoutS(); + int flightIdleS = Config.arrow_flight_session_idle_timeout_second; + if (flightIdleS <= 0) { + return waitTimeoutS; + } + return Math.min(waitTimeoutS, Math.max((long) flightIdleS, (long) getExecTimeoutS())); + } + @Override public MysqlChannel getMysqlChannel() { throw new RuntimeException("getMysqlChannel not in mysql connection"); diff --git a/fe/fe-core/src/test/java/org/apache/doris/service/arrowflight/sessions/FlightSqlSessionIdleTimeoutTest.java b/fe/fe-core/src/test/java/org/apache/doris/service/arrowflight/sessions/FlightSqlSessionIdleTimeoutTest.java new file mode 100644 index 00000000000000..bd37696d10d853 --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/service/arrowflight/sessions/FlightSqlSessionIdleTimeoutTest.java @@ -0,0 +1,86 @@ +// 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.service.arrowflight.sessions; + +import org.apache.doris.common.Config; +import org.apache.doris.common.FeConstants; +import org.apache.doris.qe.ConnectContext; + +import org.junit.Assert; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; + +/** + * The Flight-only idle bound: a Flight session's idle timeout is + * min(wait_timeout, arrow_flight_session_idle_timeout_second); it never widens wait_timeout, + * 0 disables it, and a MySQL-protocol context is untouched. + */ +public class FlightSqlSessionIdleTimeoutTest { + + @BeforeAll + public static void setUp() { + // ConnectContext.init() registers the session with Env unless running as a unit test. + FeConstants.runningUnitTest = true; + } + + @Test + public void testFlightIdleBoundOnlyTightensWaitTimeout() { + int saved = Config.arrow_flight_session_idle_timeout_second; + try { + FlightSqlConnectContext ctx = new FlightSqlConnectContext("test-peer-identity"); + long waitTimeoutS = ctx.getSessionVariable().getWaitTimeoutS(); + Assert.assertTrue(waitTimeoutS > 0); + // the bound is floored at the query's exec timeout; pin it low so the bound is visible + ctx.getSessionVariable().setQueryTimeoutS(5); + + // tighter than wait_timeout and above the exec timeout -> the Flight bound wins + Config.arrow_flight_session_idle_timeout_second = 7; + Assert.assertEquals(7L, ctx.getIdleTimeoutS()); + + // below the exec timeout -> the exec timeout floors it (a result drain is never + // killed before query_timeout would kill the query) + Config.arrow_flight_session_idle_timeout_second = 3; + Assert.assertEquals(5L, ctx.getIdleTimeoutS()); + ctx.getSessionVariable().setQueryTimeoutS(20); + Assert.assertEquals(20L, ctx.getIdleTimeoutS()); + ctx.getSessionVariable().setQueryTimeoutS(5); + + // looser than wait_timeout -> wait_timeout still applies (the bound never widens it) + Config.arrow_flight_session_idle_timeout_second = (int) Math.min(Integer.MAX_VALUE, waitTimeoutS + 1000); + Assert.assertEquals(waitTimeoutS, ctx.getIdleTimeoutS()); + + // 0 disables the Flight-specific bound + Config.arrow_flight_session_idle_timeout_second = 0; + Assert.assertEquals(waitTimeoutS, ctx.getIdleTimeoutS()); + } finally { + Config.arrow_flight_session_idle_timeout_second = saved; + } + } + + @Test + public void testMysqlContextKeepsWaitTimeout() { + int saved = Config.arrow_flight_session_idle_timeout_second; + try { + Config.arrow_flight_session_idle_timeout_second = 7; + ConnectContext ctx = new ConnectContext(); + Assert.assertEquals(ctx.getSessionVariable().getWaitTimeoutS(), ctx.getIdleTimeoutS()); + } finally { + Config.arrow_flight_session_idle_timeout_second = saved; + } + } +} From c662d85d1634bf1af5f37ca8cc265c1d42ab0777 Mon Sep 17 00:00:00 2001 From: morningman Date: Sat, 5 Sep 2026 15:32:22 +0800 Subject: [PATCH 2/3] [fix](arrow-flight) Release deferred Flight queries of idle sessions instead of killing the session Follow-up to the first revision of #67504 after review: - Narrow the #64799 deferral gate: only a coordinator that still hands out splits to the BE (an external-table scan in batch mode, see the new ScanNode/Coordinator.hasBatchSplitSource) outlives GetFlightInfo. Every other Arrow Flight query closes its coordinator at the end of GetFlightInfo again, releasing the workload-group queue slot and the active_queries entry right away. Finalizing the FE side does not cancel BE execution, so DoGet is unaffected. - Replace arrow_flight_session_idle_timeout_second by arrow_flight_deferred_query_idle_timeout_second: the connection timeout checker now finalizes the deferred executors of a sleeping Flight session and leaves the session alive; wait_timeout still governs the session. A killed session would have made the client's next call fail with "UserSession expire after access". - Freeze the execution timeout when the executor is deferred, so a SET_VAR query_timeout hint (reverted at the end of execute()) still floors the bound. - Tests: FlightSqlDeferredQueryIdleTimeoutTest drives checkTimeout, ArrowFlightDeferralGateTest covers the predicates, StmtExecutorTest covers the frozen timeout; regression cases for the internal-table release (arrow_flight_sql_p0) and the idle reaper on a batch-mode Iceberg scan. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01ApxvNkMK8TiNycv44P8Ezk --- .../java/org/apache/doris/common/Config.java | 18 +- .../org/apache/doris/planner/ScanNode.java | 11 ++ .../org/apache/doris/qe/ConnectContext.java | 60 +++++-- .../java/org/apache/doris/qe/Coordinator.java | 19 +++ .../org/apache/doris/qe/StmtExecutor.java | 51 ++++-- .../FlightSqlConnectProcessor.java | 11 +- .../sessions/FlightSqlConnectContext.java | 23 --- .../doris/qe/ArrowFlightDeferralGateTest.java | 69 ++++++++ .../org/apache/doris/qe/StmtExecutorTest.java | 33 ++++ ...FlightSqlDeferredQueryIdleTimeoutTest.java | 156 ++++++++++++++++++ .../FlightSqlSessionIdleTimeoutTest.java | 86 ---------- .../test_arrow_flight_query_release.groovy | 88 ++++++++++ ...t_iceberg_arrow_flight_split_source.groovy | 38 +++++ 13 files changed, 511 insertions(+), 152 deletions(-) create mode 100644 fe/fe-core/src/test/java/org/apache/doris/qe/ArrowFlightDeferralGateTest.java create mode 100644 fe/fe-core/src/test/java/org/apache/doris/service/arrowflight/sessions/FlightSqlDeferredQueryIdleTimeoutTest.java delete mode 100644 fe/fe-core/src/test/java/org/apache/doris/service/arrowflight/sessions/FlightSqlSessionIdleTimeoutTest.java create mode 100644 regression-test/suites/arrow_flight_sql_p0/test_arrow_flight_query_release.groovy diff --git a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java index cd2d3986c091cd..a56912a77cdd28 100644 --- a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java +++ b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java @@ -2644,14 +2644,16 @@ public class Config extends ConfigBase { @ConfField(description = "Maximum number of connections for the Arrow Flight Server per FE.") public static int arrow_flight_max_connections = 4096; - @ConfField(mutable = true, masterOnly = false, description = "Idle timeout for Arrow Flight SQL sessions, " - + "in seconds. A Flight session that has been sleeping longer than this is killed by the " - + "connection timeout checker, exactly as a MySQL connection is killed after wait_timeout, but " - + "with its own, shorter bound: a Flight query keeps its coordinator and its workload-group " - + "queue slot alive until the session's next query or its close, so an abandoned session holds " - + "a slot for as long as it lives. The effective idle bound is min(wait_timeout, this value). " - + "0 disables the Flight-specific bound (wait_timeout alone applies).") - public static int arrow_flight_session_idle_timeout_second = 3600; + @ConfField(mutable = true, description = "Arrow Flight SQL only. A query that scans an external table in " + + "batch mode keeps its FE coordinator alive after GetFlightInfo, so the BE can keep fetching splits " + + "while the client pulls the results (DoGet); that coordinator is normally released when the " + + "session runs its next query or is closed. Most Flight clients never close a session, so the " + + "coordinator, and with it the query's workload group queue slot and its active_queries entry, " + + "would otherwise stay held until wait_timeout. If the session stays idle for longer than this " + + "many seconds after the query started, the coordinator is released anyway. The bound is never " + + "shorter than the query's own execution timeout, and the session itself is not killed " + + "(wait_timeout still governs that). 0 disables the bound.") + public static int arrow_flight_deferred_query_idle_timeout_second = 3600; @ConfField(mutable = true, masterOnly = true, description = "In auto bucketing, the number of buckets is " + "estimated based on the partition size. For storage " diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/ScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/planner/ScanNode.java index efa5d7e5406228..98d9056e1af08c 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/planner/ScanNode.java +++ b/fe/fe-core/src/main/java/org/apache/doris/planner/ScanNode.java @@ -134,6 +134,17 @@ public TupleDescriptor getTupleDesc() { return desc; } + /** + * Whether this scan hands out its splits lazily through a batch {@link SplitSource} that the + * BE fetches from the FE while it is scanning (external-table batch mode, see + * {@link SplitGenerator#isBatchMode()}). Such a scan needs its coordinator alive until the BE + * has finished scanning, even after the FE is done dispatching the query: closing the + * coordinator releases the split source ({@link #stop()}) and the BE's next split fetch fails. + */ + public boolean hasBatchSplitSource() { + return splitAssignment != null; + } + protected abstract void createScanRangeLocations() throws UserException; /** diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectContext.java b/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectContext.java index c467fa9136dda6..80ac4cb1170480 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectContext.java +++ b/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectContext.java @@ -1006,7 +1006,9 @@ public void clear() { // held by the coordinator's scan nodes), so closing the coordinator at the end of // GetFlightInfo would release the SplitSource too early and make the BE's fetchSplitBatch fail // with "Split source X is released". These executors are finalized when the next query starts - // on this connection, or when the connection is torn down. See #62259. + // on this connection, when the connection is torn down, or by the idle reaper in checkTimeout + // once the connection has been sleeping for arrow_flight_deferred_query_idle_timeout_second. + // See #62259 and #67503. private final List flightSqlDeferredExecutors = new ArrayList<>(); public void addFlightSqlDeferredExecutor(StmtExecutor executor) { @@ -1033,6 +1035,45 @@ public void closeFlightSqlDeferredExecutors() { } } + /** + * How long, in seconds, a sleeping connection may keep its deferred Arrow Flight executors + * before the timeout checker finalizes them without killing the connection + * (Config.arrow_flight_deferred_query_idle_timeout_second). A Flight client that opens a + * session per query and never closes it would otherwise pin each deferred query's query queue + * slot and query registration until wait_timeout (8h by default). The bound is never shorter + * than the execution timeout the deferred query was run with: the client may still be pulling + * that query's results from the BE, which still needs the batch split source the coordinator + * holds. Returns -1 when the bound is disabled or nothing is deferred. + */ + public long getFlightSqlDeferredExecutorsIdleTimeoutS() { + int configTimeoutS = Config.arrow_flight_deferred_query_idle_timeout_second; + if (configTimeoutS <= 0) { + return -1; + } + long execTimeoutS = -1; + synchronized (flightSqlDeferredExecutors) { + if (flightSqlDeferredExecutors.isEmpty()) { + return -1; + } + for (StmtExecutor deferredExecutor : flightSqlDeferredExecutors) { + execTimeoutS = Math.max(execTimeoutS, deferredExecutor.getDeferredExecTimeoutS()); + } + } + return Math.max(configTimeoutS, execTimeoutS); + } + + // Called by the timeout checker for a sleeping connection that is not past wait_timeout yet. + private void reapIdleFlightSqlDeferredExecutors(long idleMs) { + long timeoutS = getFlightSqlDeferredExecutorsIdleTimeoutS(); + if (timeoutS < 0 || idleMs <= timeoutS * 1000L) { + return; + } + LOG.warn("release deferred arrow flight query of idle connection, connectionId: {}, remote: {}, " + + "idle: {}ms, idle timeout: {}s", + connectionId, getRemoteHostPortString(), idleMs, timeoutS); + closeFlightSqlDeferredExecutors(); + } + /** * This method is idempotent. */ @@ -1268,8 +1309,8 @@ public void kill(boolean killConnection) { private void killByTimeout(boolean killConnection) { if (killConnection) { LOG.warn("kill wait timeout connection, connection type: {}, connectionId: {}, remote: {}, " - + "idle timeout: {}", - getConnectType(), connectionId, getRemoteHostPortString(), getIdleTimeoutS()); + + "wait timeout: {}", + getConnectType(), connectionId, getRemoteHostPortString(), sessionVariable.getWaitTimeoutS()); killConnection(); } // Now, cancel running query. @@ -1292,15 +1333,6 @@ public void cancelQuery(Status cancelReason) { } } - /** - * How long this connection may sleep (COM_SLEEP) before the timeout checker kills it. - * The MySQL protocol bound is the session's wait_timeout; protocol-specific contexts may - * tighten it (never widen it) — see FlightSqlConnectContext. - */ - public long getIdleTimeoutS() { - return sessionVariable.getWaitTimeoutS(); - } - public void checkTimeout(long now) { if (startTime <= 0) { return; @@ -1310,10 +1342,12 @@ public void checkTimeout(long now) { boolean killFlag = false; boolean killConnection = false; if (command == MysqlCommand.COM_SLEEP) { - if (delta > getIdleTimeoutS() * 1000L) { + if (delta > sessionVariable.getWaitTimeoutS() * 1000L) { // Need kill this connection. killFlag = true; killConnection = true; + } else { + reapIdleFlightSqlDeferredExecutors(delta); } } else { String timeoutTag = "query"; diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java b/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java index 4a9e58949ca870..7502e4a57534b3 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java +++ b/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java @@ -810,6 +810,25 @@ public void exec() throws Exception { execInternal(); } + /** + * Whether the BE keeps calling back into this coordinator after {@link #exec()} returned: an + * external-table scan in batch mode fetches its splits lazily from the split source that its + * scan node holds, so the coordinator must not be closed until the BE has finished scanning. + * Arrow Flight SQL uses this to decide whether a query's coordinator has to outlive + * GetFlightInfo, the client pulling the results from the BE later in DoGet. See #62259. + */ + public boolean hasBatchSplitSource() { + if (scanNodes == null) { + return false; + } + for (ScanNode scanNode : scanNodes) { + if (scanNode.hasBatchSplitSource()) { + return true; + } + } + return false; + } + @Override public void close() { // NOTE: all close method should be no exception diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java b/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java index 113985e1cd198b..bc0faf28758845 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java +++ b/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java @@ -208,6 +208,10 @@ public class StmtExecutor { // is finalized later by ConnectContext (see #62259), so the eager close in executeAndSendResult // is skipped. private volatile boolean deferredForArrowFlight = false; + // The execution timeout in effect when the coordinator was deferred. Captured at that moment + // because per-statement SET_VAR values are reverted at the end of execute(), so reading + // ConnectContext.getExecTimeoutS() later would report the session value instead. + private volatile int deferredExecTimeoutS = -1; private MasterOpExecutor masterOpExecutor = null; // Optional forward target for cancellations issued on this executor: statements that // spawn a nested internal executor with its own query id (e.g. IVM dry-run delta @@ -1084,6 +1088,21 @@ public boolean isDeferredForArrowFlight() { return deferredForArrowFlight; } + // Execution timeout (seconds) the deferred query was run with; -1 when the query is not deferred. + public int getDeferredExecTimeoutS() { + return deferredExecTimeoutS; + } + + // Keep this query's coordinator alive past GetFlightInfo (see the gate in executeAndSendResult) + // and hand it to the ConnectContext, which finalizes it later. Records the execution timeout in + // effect right now: it floors the idle reaper's bound and must be the value the query actually + // ran with, not the session value left behind after SET_VAR hints are reverted. + void deferForArrowFlight() { + deferredForArrowFlight = true; + deferredExecTimeoutS = context.getExecTimeoutS(); + context.addFlightSqlDeferredExecutor(this); + } + // Finalize an Arrow Flight query whose coordinator was kept alive across the // GetFlightInfo -> DoGet phases: close the coordinator (releasing external-table batch // SplitSources and the query queue slot) and then unregister the query. See #62259. @@ -1563,23 +1582,21 @@ public void executeAndSendResult(boolean isOutfileQuery, boolean isSendFields, if (context.getConnectType().equals(ConnectType.ARROW_FLIGHT_SQL)) { Preconditions.checkState(!context.isReturnResultFromLocal()); profile.getSummaryProfile().setTempStartTime(); - // Defer closing the coordinator to ConnectContext (closed on the next query or - // connection teardown) instead of in the finally block below. This gate covers - // every Arrow Flight query whose results are produced on the BE (coordBase == - // coord) -- internal-table and external, batch or not. It is REQUIRED only for an - // external-table scan in batch mode, where the BE lazily fetches splits from the FE - // during the later DoGet phase, so closing the coordinator here would release its - // batch SplitSource too early and break DoGet. Other remote-result queries do not - // need deferral (the BE buffers their result independently) but are captured by the - // same gate; the trade-off is their coordinator, query queue slot and query - // registration stay held until the next query / teardown instead of being released - // at the end of GetFlightInfo. A short-circuit point query is the one case with a - // different coordBase, and it can no longer reach here: it has no Arrow result on - // either side, so LogicalResultSinkToShortCircuitPointQuery keeps Arrow Flight SQL - // on the normal execution path. See #62259 and #67368. - if (coordBase == coord) { - deferredForArrowFlight = true; - context.addFlightSqlDeferredExecutor(this); + // The client pulls the results from the BE later (DoGet). Only an external-table + // scan in batch mode still needs the coordinator after this point: the BE fetches + // its splits lazily from the split source the coordinator holds, so closing the + // coordinator here would release that source too early and break DoGet (#62259). + // Such a coordinator is closed later by ConnectContext: on the session's next + // query, on teardown, or by the idle reaper in checkTimeout. The trade-off is that + // its query queue slot and query registration stay held until then. Every other + // query closes its coordinator in the finally block below and releases both right + // away, the BE buffering its results independently of the coordinator (#67503). + // A short-circuit point query is the one case with a different coordBase, and it + // can no longer reach here: it has no Arrow result on either side, so + // LogicalResultSinkToShortCircuitPointQuery keeps Arrow Flight SQL on the normal + // execution path (#67368). + if (coordBase == coord && coord.hasBatchSplitSource()) { + deferForArrowFlight(); } return; } diff --git a/fe/fe-core/src/main/java/org/apache/doris/service/arrowflight/FlightSqlConnectProcessor.java b/fe/fe-core/src/main/java/org/apache/doris/service/arrowflight/FlightSqlConnectProcessor.java index 24e30c3942f20f..2296986c535dd3 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/service/arrowflight/FlightSqlConnectProcessor.java +++ b/fe/fe-core/src/main/java/org/apache/doris/service/arrowflight/FlightSqlConnectProcessor.java @@ -196,11 +196,12 @@ public void fetchArrowFlightSchema(int timeoutMs) { @Override public void close() throws Exception { ctx.setCommand(MysqlCommand.COM_SLEEP); - // Executors whose results are pulled from the BE keep their coordinator alive past - // GetFlightInfo (registered as deferred executors on the ConnectContext) so the BE can - // still fetch external-table splits during DoGet. Do NOT finalize those here; they are - // finalized when the next query starts or the connection is torn down. Executors that are - // not deferred (local results, or a query that already failed) are finalized now. See #62259. + // An external-table scan in batch mode keeps its coordinator alive past GetFlightInfo + // (registered as a deferred executor on the ConnectContext) so the BE can still fetch its + // splits during DoGet. Do NOT finalize those here; they are finalized when the next query + // starts, when the connection is torn down, or by the idle reaper in + // ConnectContext.checkTimeout. Every other executor (local results, results the BE buffers + // on its own, or a query that already failed) is finalized now. See #62259 and #67503. for (StmtExecutor asynExecutor : returnResultFromRemoteExecutor) { if (!asynExecutor.isDeferredForArrowFlight()) { asynExecutor.finalizeQuery(); diff --git a/fe/fe-core/src/main/java/org/apache/doris/service/arrowflight/sessions/FlightSqlConnectContext.java b/fe/fe-core/src/main/java/org/apache/doris/service/arrowflight/sessions/FlightSqlConnectContext.java index 18a7043bed5219..ceddfaa563988a 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/service/arrowflight/sessions/FlightSqlConnectContext.java +++ b/fe/fe-core/src/main/java/org/apache/doris/service/arrowflight/sessions/FlightSqlConnectContext.java @@ -17,7 +17,6 @@ package org.apache.doris.service.arrowflight.sessions; -import org.apache.doris.common.Config; import org.apache.doris.common.Status; import org.apache.doris.mysql.MysqlChannel; import org.apache.doris.qe.ConnectContext; @@ -51,28 +50,6 @@ public FlightSqlChannel getFlightSqlChannel() { return flightSqlChannel; } - /** - * Flight sessions get their own idle bound (Config.arrow_flight_session_idle_timeout_second), - * tighter than the MySQL wait_timeout: an abandoned Flight session keeps its last query's - * coordinator — and that query's workload-group queue slot — alive until it is closed, so a - * client that opens a session per query and never closes it would otherwise pin a slot for - * the whole wait_timeout (8h by default). - * - * A Flight session is COM_SLEEP while the client drains the result from the BE (DoGet), and - * its idle clock runs from the query's START — so the bound is floored at the query's own - * execution timeout: a long result drain is never killed before query_timeout would kill the - * query itself. Effective bound = min(wait_timeout, max(config, exec timeout)); 0 disables it. - */ - @Override - public long getIdleTimeoutS() { - long waitTimeoutS = super.getIdleTimeoutS(); - int flightIdleS = Config.arrow_flight_session_idle_timeout_second; - if (flightIdleS <= 0) { - return waitTimeoutS; - } - return Math.min(waitTimeoutS, Math.max((long) flightIdleS, (long) getExecTimeoutS())); - } - @Override public MysqlChannel getMysqlChannel() { throw new RuntimeException("getMysqlChannel not in mysql connection"); diff --git a/fe/fe-core/src/test/java/org/apache/doris/qe/ArrowFlightDeferralGateTest.java b/fe/fe-core/src/test/java/org/apache/doris/qe/ArrowFlightDeferralGateTest.java new file mode 100644 index 00000000000000..a02de20f14f635 --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/qe/ArrowFlightDeferralGateTest.java @@ -0,0 +1,69 @@ +// 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.analysis.DescriptorTable; +import org.apache.doris.datasource.split.SplitAssignment; +import org.apache.doris.planner.PlanFragment; +import org.apache.doris.planner.ScanNode; +import org.apache.doris.thrift.TUniqueId; + +import com.google.common.collect.Lists; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; + +import java.lang.reflect.Field; +import java.util.List; + +/** + * The predicate behind the Arrow Flight deferral gate in StmtExecutor.executeAndSendResult (#67503): + * a coordinator has to outlive GetFlightInfo only when one of its scans still hands out splits to + * the BE lazily, i.e. an external-table scan in batch mode holding a batch split source (#62259). + */ +public class ArrowFlightDeferralGateTest { + + private static ScanNode scanNode(boolean batchSplitSource) throws Exception { + ScanNode node = Mockito.mock(ScanNode.class, Mockito.CALLS_REAL_METHODS); + if (batchSplitSource) { + // FileQueryScanNode.createScanRangeLocations sets this only in batch mode. + Field field = ScanNode.class.getDeclaredField("splitAssignment"); + field.setAccessible(true); + field.set(node, Mockito.mock(SplitAssignment.class)); + } + return node; + } + + private static Coordinator coordinator(List scanNodes) { + return new Coordinator(1L, new TUniqueId(1L, 2L), new DescriptorTable(), Lists.newArrayList(), + scanNodes, "UTC", false, false); + } + + @Test + public void testScanNodeHasBatchSplitSourceOnlyWhenSplitsAreHandedOutLazily() throws Exception { + Assertions.assertFalse(scanNode(false).hasBatchSplitSource()); + Assertions.assertTrue(scanNode(true).hasBatchSplitSource()); + } + + @Test + public void testCoordinatorHasBatchSplitSourceIfAnyScanDoes() throws Exception { + Assertions.assertFalse(coordinator(Lists.newArrayList()).hasBatchSplitSource()); + Assertions.assertFalse(coordinator(Lists.newArrayList(scanNode(false), scanNode(false))).hasBatchSplitSource()); + Assertions.assertTrue(coordinator(Lists.newArrayList(scanNode(false), scanNode(true))).hasBatchSplitSource()); + } +} diff --git a/fe/fe-core/src/test/java/org/apache/doris/qe/StmtExecutorTest.java b/fe/fe-core/src/test/java/org/apache/doris/qe/StmtExecutorTest.java index fd40fb68dad421..dac844f33b00a1 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/qe/StmtExecutorTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/qe/StmtExecutorTest.java @@ -78,6 +78,39 @@ public void testShowNull() throws Exception { Assertions.assertEquals(QueryState.MysqlStateType.OK, connectContext.getState().getStateType()); } + // The deferral gate (#67503): a coordinator is kept alive past GetFlightInfo only when the BE + // still fetches splits from it (Coordinator.hasBatchSplitSource), and the execution timeout it + // ran with is frozen at that moment. SET_VAR hint values are reverted when execute() ends, so + // the idle reaper must not read the session value later. + @Test + public void testDeferForArrowFlightFreezesExecTimeoutInEffect() throws Exception { + int savedQueryTimeout = connectContext.getSessionVariable().getQueryTimeoutS(); + int savedIdleTimeout = Config.arrow_flight_deferred_query_idle_timeout_second; + connectContext.setQueryId(new TUniqueId(0x67503L, 0x1L)); + try { + Config.arrow_flight_deferred_query_idle_timeout_second = 1; + connectContext.getSessionVariable().setQueryTimeoutS(1234); + StmtExecutor stmtExecutor = new StmtExecutor(connectContext, ""); + Assertions.assertFalse(stmtExecutor.isDeferredForArrowFlight()); + Assertions.assertEquals(-1, stmtExecutor.getDeferredExecTimeoutS()); + + stmtExecutor.deferForArrowFlight(); + + Assertions.assertTrue(stmtExecutor.isDeferredForArrowFlight()); + Assertions.assertEquals(1234, stmtExecutor.getDeferredExecTimeoutS()); + // the reaper's bound is floored at the frozen value ... + Assertions.assertEquals(1234L, connectContext.getFlightSqlDeferredExecutorsIdleTimeoutS()); + // ... even after the session value moved on, as it does when a SET_VAR hint is reverted + connectContext.getSessionVariable().setQueryTimeoutS(5); + Assertions.assertEquals(1234, stmtExecutor.getDeferredExecTimeoutS()); + Assertions.assertEquals(1234L, connectContext.getFlightSqlDeferredExecutorsIdleTimeoutS()); + } finally { + connectContext.closeFlightSqlDeferredExecutors(); + connectContext.getSessionVariable().setQueryTimeoutS(savedQueryTimeout); + Config.arrow_flight_deferred_query_idle_timeout_second = savedIdleTimeout; + } + } + // Arrow Flight SQL keeps a query's coordinator alive across GetFlightInfo -> DoGet (see #62259); // it is released later by finalizeArrowFlightQuery(), which closes the coordinator and then // unregisters the query. The close and the unregister must be independent: if coord.close() diff --git a/fe/fe-core/src/test/java/org/apache/doris/service/arrowflight/sessions/FlightSqlDeferredQueryIdleTimeoutTest.java b/fe/fe-core/src/test/java/org/apache/doris/service/arrowflight/sessions/FlightSqlDeferredQueryIdleTimeoutTest.java new file mode 100644 index 00000000000000..211f2ab984908a --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/service/arrowflight/sessions/FlightSqlDeferredQueryIdleTimeoutTest.java @@ -0,0 +1,156 @@ +// 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.service.arrowflight.sessions; + +import org.apache.doris.common.Config; +import org.apache.doris.common.FeConstants; +import org.apache.doris.mysql.MysqlCommand; +import org.apache.doris.qe.ConnectContext; +import org.apache.doris.qe.StmtExecutor; + +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; + +/** + * The idle reaper for deferred Arrow Flight queries (#67503). A sleeping Flight session whose last + * query kept its coordinator alive (an external-table scan in batch mode, see #62259) gets that + * coordinator finalized by the connection timeout checker once the session has been idle for + * arrow_flight_deferred_query_idle_timeout_second, floored at the execution timeout the query ran + * with. The session itself is not killed, wait_timeout still governs that, and a MySQL session is + * untouched. + */ +public class FlightSqlDeferredQueryIdleTimeoutTest { + private int savedIdleTimeout; + private boolean savedRunningUnitTest; + + @BeforeEach + public void setUp() { + savedIdleTimeout = Config.arrow_flight_deferred_query_idle_timeout_second; + savedRunningUnitTest = FeConstants.runningUnitTest; + // ConnectContext.init() registers the session with Env unless running as a unit test. + FeConstants.runningUnitTest = true; + } + + @AfterEach + public void tearDown() { + Config.arrow_flight_deferred_query_idle_timeout_second = savedIdleTimeout; + FeConstants.runningUnitTest = savedRunningUnitTest; + } + + private static StmtExecutor deferredExecutor(int execTimeoutS) { + StmtExecutor executor = Mockito.mock(StmtExecutor.class); + Mockito.when(executor.getDeferredExecTimeoutS()).thenReturn(execTimeoutS); + return executor; + } + + // A Flight session that ran a query and has been sleeping since; the client never closed it. + private static FlightSqlConnectContext sleepingFlightSession(StmtExecutor... deferred) { + FlightSqlConnectContext ctx = new FlightSqlConnectContext("test-peer-identity"); + ctx.setCommand(MysqlCommand.COM_SLEEP); + ctx.setStartTime(); + for (StmtExecutor executor : deferred) { + ctx.addFlightSqlDeferredExecutor(executor); + } + return ctx; + } + + @Test + public void testIdleSessionReleasesDeferredQueryButIsNotKilled() { + Config.arrow_flight_deferred_query_idle_timeout_second = 7; + StmtExecutor deferred = deferredExecutor(5); + FlightSqlConnectContext ctx = sleepingFlightSession(deferred); + long start = ctx.getStartTime(); + Assertions.assertEquals(7L, ctx.getFlightSqlDeferredExecutorsIdleTimeoutS()); + + // not idle for long enough yet + ctx.checkTimeout(start + 7_000L); + Mockito.verify(deferred, Mockito.never()).finalizeArrowFlightQuery(); + Assertions.assertFalse(ctx.isKilled()); + + // past the bound: the deferred coordinator is finalized and the session survives + ctx.checkTimeout(start + 7_001L); + Mockito.verify(deferred).finalizeArrowFlightQuery(); + Assertions.assertFalse(ctx.isKilled()); + Assertions.assertEquals(-1L, ctx.getFlightSqlDeferredExecutorsIdleTimeoutS()); + + // a later tick has nothing left to release + ctx.checkTimeout(start + 60_000L); + Mockito.verify(deferred, Mockito.times(1)).finalizeArrowFlightQuery(); + Assertions.assertFalse(ctx.isKilled()); + } + + @Test + public void testBoundIsFlooredAtTheExecTimeoutTheDeferredQueryRanWith() { + Config.arrow_flight_deferred_query_idle_timeout_second = 3; + StmtExecutor shortQuery = deferredExecutor(5); + StmtExecutor longQuery = deferredExecutor(20); + FlightSqlConnectContext ctx = sleepingFlightSession(shortQuery, longQuery); + long start = ctx.getStartTime(); + // the longest deferred query wins: a client may still be pulling its results from the BE + Assertions.assertEquals(20L, ctx.getFlightSqlDeferredExecutorsIdleTimeoutS()); + + ctx.checkTimeout(start + 19_999L); + Mockito.verify(shortQuery, Mockito.never()).finalizeArrowFlightQuery(); + Mockito.verify(longQuery, Mockito.never()).finalizeArrowFlightQuery(); + + ctx.checkTimeout(start + 20_001L); + Mockito.verify(shortQuery).finalizeArrowFlightQuery(); + Mockito.verify(longQuery).finalizeArrowFlightQuery(); + Assertions.assertFalse(ctx.isKilled()); + } + + @Test + public void testZeroDisablesTheReaper() { + Config.arrow_flight_deferred_query_idle_timeout_second = 0; + StmtExecutor deferred = deferredExecutor(5); + FlightSqlConnectContext ctx = sleepingFlightSession(deferred); + Assertions.assertEquals(-1L, ctx.getFlightSqlDeferredExecutorsIdleTimeoutS()); + + // idle for almost the whole wait_timeout: nothing is released and the session is alive + long waitTimeoutMs = ctx.getSessionVariable().getWaitTimeoutS() * 1000L; + ctx.checkTimeout(ctx.getStartTime() + waitTimeoutMs - 1); + Mockito.verify(deferred, Mockito.never()).finalizeArrowFlightQuery(); + Assertions.assertFalse(ctx.isKilled()); + } + + @Test + public void testNothingDeferredMeansNoBound() { + Config.arrow_flight_deferred_query_idle_timeout_second = 7; + FlightSqlConnectContext ctx = sleepingFlightSession(); + Assertions.assertEquals(-1L, ctx.getFlightSqlDeferredExecutorsIdleTimeoutS()); + + ctx.checkTimeout(ctx.getStartTime() + 3_600_000L); + Assertions.assertFalse(ctx.isKilled()); + } + + @Test + public void testMysqlSessionIsUntouched() { + Config.arrow_flight_deferred_query_idle_timeout_second = 1; + ConnectContext ctx = new ConnectContext(); + ctx.setCommand(MysqlCommand.COM_SLEEP); + ctx.setStartTime(); + Assertions.assertEquals(-1L, ctx.getFlightSqlDeferredExecutorsIdleTimeoutS()); + + // idle far beyond the Flight bound but within wait_timeout: still alive + ctx.checkTimeout(ctx.getStartTime() + 3_600_000L); + Assertions.assertFalse(ctx.isKilled()); + } +} diff --git a/fe/fe-core/src/test/java/org/apache/doris/service/arrowflight/sessions/FlightSqlSessionIdleTimeoutTest.java b/fe/fe-core/src/test/java/org/apache/doris/service/arrowflight/sessions/FlightSqlSessionIdleTimeoutTest.java deleted file mode 100644 index bd37696d10d853..00000000000000 --- a/fe/fe-core/src/test/java/org/apache/doris/service/arrowflight/sessions/FlightSqlSessionIdleTimeoutTest.java +++ /dev/null @@ -1,86 +0,0 @@ -// 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.service.arrowflight.sessions; - -import org.apache.doris.common.Config; -import org.apache.doris.common.FeConstants; -import org.apache.doris.qe.ConnectContext; - -import org.junit.Assert; -import org.junit.jupiter.api.BeforeAll; -import org.junit.jupiter.api.Test; - -/** - * The Flight-only idle bound: a Flight session's idle timeout is - * min(wait_timeout, arrow_flight_session_idle_timeout_second); it never widens wait_timeout, - * 0 disables it, and a MySQL-protocol context is untouched. - */ -public class FlightSqlSessionIdleTimeoutTest { - - @BeforeAll - public static void setUp() { - // ConnectContext.init() registers the session with Env unless running as a unit test. - FeConstants.runningUnitTest = true; - } - - @Test - public void testFlightIdleBoundOnlyTightensWaitTimeout() { - int saved = Config.arrow_flight_session_idle_timeout_second; - try { - FlightSqlConnectContext ctx = new FlightSqlConnectContext("test-peer-identity"); - long waitTimeoutS = ctx.getSessionVariable().getWaitTimeoutS(); - Assert.assertTrue(waitTimeoutS > 0); - // the bound is floored at the query's exec timeout; pin it low so the bound is visible - ctx.getSessionVariable().setQueryTimeoutS(5); - - // tighter than wait_timeout and above the exec timeout -> the Flight bound wins - Config.arrow_flight_session_idle_timeout_second = 7; - Assert.assertEquals(7L, ctx.getIdleTimeoutS()); - - // below the exec timeout -> the exec timeout floors it (a result drain is never - // killed before query_timeout would kill the query) - Config.arrow_flight_session_idle_timeout_second = 3; - Assert.assertEquals(5L, ctx.getIdleTimeoutS()); - ctx.getSessionVariable().setQueryTimeoutS(20); - Assert.assertEquals(20L, ctx.getIdleTimeoutS()); - ctx.getSessionVariable().setQueryTimeoutS(5); - - // looser than wait_timeout -> wait_timeout still applies (the bound never widens it) - Config.arrow_flight_session_idle_timeout_second = (int) Math.min(Integer.MAX_VALUE, waitTimeoutS + 1000); - Assert.assertEquals(waitTimeoutS, ctx.getIdleTimeoutS()); - - // 0 disables the Flight-specific bound - Config.arrow_flight_session_idle_timeout_second = 0; - Assert.assertEquals(waitTimeoutS, ctx.getIdleTimeoutS()); - } finally { - Config.arrow_flight_session_idle_timeout_second = saved; - } - } - - @Test - public void testMysqlContextKeepsWaitTimeout() { - int saved = Config.arrow_flight_session_idle_timeout_second; - try { - Config.arrow_flight_session_idle_timeout_second = 7; - ConnectContext ctx = new ConnectContext(); - Assert.assertEquals(ctx.getSessionVariable().getWaitTimeoutS(), ctx.getIdleTimeoutS()); - } finally { - Config.arrow_flight_session_idle_timeout_second = saved; - } - } -} diff --git a/regression-test/suites/arrow_flight_sql_p0/test_arrow_flight_query_release.groovy b/regression-test/suites/arrow_flight_sql_p0/test_arrow_flight_query_release.groovy new file mode 100644 index 00000000000000..0913164a1eb933 --- /dev/null +++ b/regression-test/suites/arrow_flight_sql_p0/test_arrow_flight_query_release.groovy @@ -0,0 +1,88 @@ +// 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. + +// Regression for https://github.com/apache/doris/issues/67503 +// +// Over Arrow Flight SQL a query runs in two phases: GetFlightInfo (plan and start it on the BE) +// and DoGet (the client pulls the results from the BE). Only an external-table scan in batch mode +// needs its FE coordinator after GetFlightInfo (#62259). Every other query has to release its +// coordinator, and with it the workload group queue slot and the active_queries entry, at the end +// of GetFlightInfo: most Flight clients never close their session, so a coordinator that waited +// for the session's next query kept one queue slot per finished query until wait_timeout. +// +// The framework's Flight session behaves like such a client: it is reused across statements and +// never closed. +suite("test_arrow_flight_query_release", "arrow_flight_sql") { + def tableName = "test_arrow_flight_query_release_tbl" + def wgName = "test_arrow_flight_query_release_wg" + + def forComputeGroupStr = "" + if (isCloudMode()) { + def clusters = sql " SHOW CLUSTERS; " + assertTrue(!clusters.isEmpty()) + forComputeGroupStr = " for ${clusters[0][0]} " + } + + sql "DROP TABLE IF EXISTS ${tableName}" + sql """ + CREATE TABLE ${tableName} (id int, name varchar(20)) + DUPLICATE KEY(id) DISTRIBUTED BY HASH(id) BUCKETS 1 + PROPERTIES ("replication_num" = "1") + """ + sql "INSERT INTO ${tableName} VALUES (1, 'a'), (2, 'b'), (3, 'c')" + + sql "ADMIN SET FRONTEND CONFIG ('enable_workload_group' = 'true')" + sql "DROP WORKLOAD GROUP IF EXISTS ${wgName} ${forComputeGroupStr}" + // One running query at a time and no waiting queue: while a query still holds the slot, the + // next scanning query in the group fails at once with "query waiting queue is full". + sql """ + CREATE WORKLOAD GROUP ${wgName} ${forComputeGroupStr} + PROPERTIES ('max_concurrency' = '1', 'max_queue_size' = '0', 'queue_timeout' = '0') + """ + try { + // The Flight session is a session of its own, so it is bound to the group separately. + sql "SET workload_group = '${wgName}'" + arrow_flight_sql "SET workload_group = '${wgName}'" + + // A scanning query over Arrow Flight SQL. The session stays open afterwards. + def flightRows = arrow_flight_sql "SELECT id, name FROM ${tableName} ORDER BY id" + assertEquals(3, flightRows.size()) + + // Its coordinator was released at the end of GetFlightInfo, so the query is gone from + // active_queries. The LIKE pattern is assembled with CONCAT so that this statement's own + // text does not match it. + def registered = sql """ + SELECT QUERY_ID, SQL FROM information_schema.active_queries + WHERE SQL LIKE CONCAT('%FROM ${tableName}', ' ORDER BY id%') + """ + assertTrue(registered.isEmpty(), "finished Arrow Flight query is still registered: ${registered}") + + // ... and its queue slot is free again: a scanning query in the same group runs instead + // of failing with "query waiting queue is full". + def mysqlRows = sql "SELECT id FROM ${tableName} ORDER BY id" + assertEquals(3, mysqlRows.size()) + } finally { + sql "SET workload_group = 'normal'" + try { + arrow_flight_sql "SET workload_group = 'normal'" + } catch (Throwable ignore) { + // best effort: the Flight session must not keep pointing at the dropped group + } + sql "DROP WORKLOAD GROUP IF EXISTS ${wgName} ${forComputeGroupStr}" + sql "DROP TABLE IF EXISTS ${tableName}" + } +} diff --git a/regression-test/suites/external_table_p0/iceberg/test_iceberg_arrow_flight_split_source.groovy b/regression-test/suites/external_table_p0/iceberg/test_iceberg_arrow_flight_split_source.groovy index e721d0e9d8eab6..93aaf0098ac9d2 100644 --- a/regression-test/suites/external_table_p0/iceberg/test_iceberg_arrow_flight_split_source.groovy +++ b/regression-test/suites/external_table_p0/iceberg/test_iceberg_arrow_flight_split_source.groovy @@ -81,6 +81,12 @@ suite("test_iceberg_arrow_flight_split_source", "p0,external") { "s3.region" = "us-east-1" );""" + // #67503: the idle reaper for a deferred batch-mode scan (see below). Set the bound low, and + // restore the FE's original value afterwards. + int idleTimeoutS = 10 + def origIdleTimeout = sql """ ADMIN SHOW FRONTEND CONFIG LIKE 'arrow_flight_deferred_query_idle_timeout_second' """ + assert origIdleTimeout.size() == 1 : "arrow_flight_deferred_query_idle_timeout_second not found in FE config" + Connection flightConn = null try { // Baseline over the MySQL protocol (works regardless of the bug). @@ -120,7 +126,39 @@ suite("test_iceberg_arrow_flight_split_source", "p0,external") { // deferred coordinator when the next query starts. def flightLimited = flightSql """ select * from ${table} limit 10 """ assert flightLimited.size() > 0 && flightLimited.size() <= 10 : "unexpected row count: ${flightLimited.size()}" + + // #67503: a batch-mode scan keeps its coordinator (and with it the query's workload group + // queue slot and its active_queries entry) alive after GetFlightInfo, until the session + // runs its next query or is closed. A client that does neither would hold them until + // wait_timeout, so the FE releases the coordinator once the session has been idle for + // arrow_flight_deferred_query_idle_timeout_second, never before the query's own execution + // timeout, and without killing the session. + sql """ ADMIN SET FRONTEND CONFIG ('arrow_flight_deferred_query_idle_timeout_second' = '${idleTimeoutS}') """ + flightSql """ set query_timeout = ${idleTimeoutS} """ + def flightReap = flightSql """ select * from ${table} limit 13 """ + assertEquals(13, flightReap.size()) + + // The LIKE pattern is assembled with CONCAT so that this statement's own text does not + // match it. + def deferredQuery = { -> + sql """ select QUERY_ID from information_schema.active_queries + where SQL like CONCAT('%from ${table} limit', ' 13%') """ + } + // Right after the scan the query is still registered: its coordinator is deferred. + assert deferredQuery().size() == 1 : "expected the batch-mode Flight query to stay registered until the idle reaper releases it" + + // Once the session has been idle for the bound, the reaper releases it ... + long deadline = System.currentTimeMillis() + 60_000L + while (!deferredQuery().isEmpty() && System.currentTimeMillis() < deadline) { + Thread.sleep(1000) + } + assert deferredQuery().isEmpty() : "the idle reaper did not release the deferred Flight query within 60s" + + // ... and the session survives: it still runs queries. + def afterReap = flightSql """ select * from ${table} limit 1 """ + assertEquals(1, afterReap.size()) } finally { + sql """ ADMIN SET FRONTEND CONFIG ('arrow_flight_deferred_query_idle_timeout_second' = '${origIdleTimeout[0][1]}') """ // Close our own connection (best effort) so a dead endpoint cannot mask the real failure, // then drop the catalog over the reliable MySQL connection. if (flightConn != null) { From 888313142ef112a5dd52d4e536259832b16e3852 Mon Sep 17 00:00:00 2001 From: morningman Date: Sat, 5 Sep 2026 23:50:55 +0800 Subject: [PATCH 3/3] [test](arrow-flight) Cover the non-batch external scan on the eager release path Follow-up to #67504 after review: the deferral gate this PR narrows moves exactly one case - an external-table scan that is NOT in batch mode - from "coordinator deferred past GetFlightInfo" to "coordinator closed at the end of GetFlightInfo". That case had no Arrow Flight coverage anywhere: the only external-catalog Flight suite forces batch mode on its session and asserts it, so all four of its data queries exercise the deferred side, and the new arrow_flight_sql_p0 case covers the eager side only for an internal table, where neither a split source nor a connector read session exists. Add the missing cell to test_iceberg_arrow_flight_split_source, reusing the catalog and Flight connection it already sets up: - A negative control mirroring the existing batch assertion: "(approximate)" is emitted only when isBatchMode(), so its absence proves the scan really is on the synchronous split path. Without it the block could silently run in batch mode and pass. enable_external_table_batch_mode=false is a reliable off switch here - IcebergScanPlanProvider.streamingSplitEstimate returns -1 when it is unset, and Iceberg does not override supportsBatchScan (SPI default false), so the partition-count flavor cannot route around it. - A full scan that must return every row: the FE has closed the coordinator by then, and the BE buffers the result independently of it. - A check that the release really was eager - the query is gone from active_queries right after the client has the rows. No polling is needed because finalizeQuery() runs inside GetFlightInfo. A distinct limit keeps the query text apart from the other scans and the LIKE pattern is built with CONCAT so the probe cannot match itself. Batch mode is restored afterwards, since the idle-reaper assertions below need a deferred coordinator to release. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01WGqBt7EdtV7rc7tD4nmEqx --- ...t_iceberg_arrow_flight_split_source.groovy | 35 +++++++++++++++++++ 1 file changed, 35 insertions(+) diff --git a/regression-test/suites/external_table_p0/iceberg/test_iceberg_arrow_flight_split_source.groovy b/regression-test/suites/external_table_p0/iceberg/test_iceberg_arrow_flight_split_source.groovy index 93aaf0098ac9d2..67b9a431b825be 100644 --- a/regression-test/suites/external_table_p0/iceberg/test_iceberg_arrow_flight_split_source.groovy +++ b/regression-test/suites/external_table_p0/iceberg/test_iceberg_arrow_flight_split_source.groovy @@ -127,6 +127,41 @@ suite("test_iceberg_arrow_flight_split_source", "p0,external") { def flightLimited = flightSql """ select * from ${table} limit 10 """ assert flightLimited.size() > 0 && flightLimited.size() <= 10 : "unexpected row count: ${flightLimited.size()}" + // #67503, the other side of the deferral gate: the SAME external table scanned WITHOUT + // batch mode is not deferred. Its coordinator, and with it the query's workload group queue + // slot and its active_queries entry, is released at the end of GetFlightInfo, before the + // client pulls anything. That is the case the gate actually moved, so it needs its own + // coverage here: every other Flight query in this suite runs in batch mode. + flightSql """ set enable_external_table_batch_mode = false """ + + // Negative control, mirroring the batch assertion above: "(approximate)" is emitted only + // when isBatchMode(), so its absence proves this really is the synchronous split path and + // the assertions below cannot silently pass on the batch path. + def explainNonBatch = flightSql """ explain select * from ${table} """ + boolean stillBatch = explainNonBatch.any { row -> + row.any { cell -> cell != null && cell.toString().contains("approximate") } + } + assert !stillBatch : "expected the non-batch split path in the Arrow Flight plan, got: ${explainNonBatch}" + + // The scan must still be complete: the FE closed the coordinator at the end of + // GetFlightInfo, and the BE buffers the result independently of it. + def flightNonBatch = flightSql """ select * from ${table} """ + assertEquals(expectedRows, (flightNonBatch.size() as long)) + + // ... and the release really was eager, unlike the batch-mode scan below. A distinct limit + // keeps this query's text apart from the other scans, and the LIKE pattern is assembled + // with CONCAT so that the probe statement's own text does not match it. No polling is + // needed: finalizeQuery() runs inside GetFlightInfo, so it has already happened by the time + // the client has the rows. + def flightNonBatchLimited = flightSql """ select * from ${table} limit 17 """ + assertEquals(17, flightNonBatchLimited.size()) + def nonBatchRegistered = sql """ select QUERY_ID from information_schema.active_queries + where SQL like CONCAT('%from ${table} limit', ' 17%') """ + assert nonBatchRegistered.isEmpty() : "a non-batch Flight query must release its coordinator at the end of GetFlightInfo, still registered: ${nonBatchRegistered}" + + // Back to batch mode: the idle reaper below needs a deferred coordinator to release. + flightSql """ set enable_external_table_batch_mode = true """ + // #67503: a batch-mode scan keeps its coordinator (and with it the query's workload group // queue slot and its active_queries entry) alive after GetFlightInfo, until the session // runs its next query or is closed. A client that does neither would hold them until