[fix](arrow-flight) Release a finished Flight query's coordinator instead of holding it until wait_timeout

fe/fe-common/src/main/java/org/apache/doris/common/Config.java

@@ -2644,6 +2644,17 @@ 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, 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 "
             + "and computing integration, a partition size of 5GB " + "is estimated as one bucket, but for cloud, a "

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;
 
     /**

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<StmtExecutor> 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.
      */
@@ -1305,6 +1346,8 @@ public void checkTimeout(long now) {
                 // Need kill this connection.
                 killFlag = true;
                 killConnection = true;
+            } else {
+                reapIdleFlightSqlDeferredExecutors(delta);
             }
         } else {
             String timeoutTag = "query";

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

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;
             }

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();

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<ScanNode> scanNodes) {
+        return new Coordinator(1L, new TUniqueId(1L, 2L), new DescriptorTable(), Lists.<PlanFragment>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());
+    }
+}

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()