[fix](arrow-flight) Keep coordinator alive across GetFlightInfo/DoGet for external table scan - #64799
Merged
morningman merged 6 commits intoJul 18, 2026
Conversation
Contributor
|
Thank you for your contribution to Apache Doris. Please clearly describe your PR:
|
Contributor
Author
|
run buildall |
Contributor
TPC-H: Total hot run time: 28681 ms |
Contributor
TPC-DS: Total hot run time: 172304 ms |
Contributor
ClickBench: Total hot run time: 25.16 s |
Contributor
FE Regression Coverage ReportIncrement line coverage |
Contributor
Author
|
run buildall |
Contributor
TPC-H: Total hot run time: 29147 ms |
Contributor
TPC-DS: Total hot run time: 173041 ms |
Contributor
ClickBench: Total hot run time: 25.31 s |
Contributor
FE Regression Coverage ReportIncrement line coverage |
Contributor
Author
|
run buildall |
Contributor
TPC-H: Total hot run time: 29708 ms |
Contributor
TPC-DS: Total hot run time: 171061 ms |
Contributor
ClickBench: Total hot run time: 25.03 s |
morningman
force-pushed
the
fix-62259-arrow-flight-split-source-lifecycle
branch
from
June 25, 2026 11:09
eecedcb to
cd50205
Compare
Contributor
Author
|
run buildall |
Contributor
FE UT Coverage ReportIncrement line coverage |
Contributor
TPC-H: Total hot run time: 29023 ms |
Contributor
TPC-DS: Total hot run time: 172555 ms |
Contributor
ClickBench: Total hot run time: 25.23 s |
Contributor
FE Regression Coverage ReportIncrement line coverage |
Contributor
FE Regression Coverage ReportIncrement line coverage |
Contributor
Author
|
/review |
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
I found two correctness issues in the Arrow Flight remote-result lifecycle changes.
Critical checkpoints: the PR targets the right split-source lifetime problem, but the current cleanup trigger is still not safe for concurrent/interleaved Flight RPCs and it misses a failed-GetFlightInfo cleanup path. The change is small and focused, and it does not add configs, persistence, storage-format changes, or FE/BE protocol fields. It does involve non-trivial lifecycle and concurrency: deferred coordinators hold query queue slots and external-table split sources after GetFlightInfo, while DoGet runs independently against BE endpoints. The tests cover normal cleanup and one external-table scan, but they do not cover an overlapping second statement or schema-fetch failure after deferral.
Validation: no existing inline review threads were present; git diff --check over the GitHub changed-file list was clean. I did not run FE tests because thirdparty/installed/bin/protoc is missing in this checkout.
User focus: no additional user-provided review focus was supplied.
Subagent conclusions: OPT-1 became MAIN-1 and is submitted as an inline comment; TSC-1 became MAIN-2 and is submitted as an inline comment. No candidates were dismissed or merged as duplicates. Convergence round 1 ended with both optimizer-rewrite and tests-session-config replying NO_NEW_VALUABLE_FINDINGS for the same current ledger/comment set.
| // deferred (Arrow Flight keeps it alive across GetFlightInfo -> DoGet so the BE can | ||
| // fetch external-table splits during DoGet). By now the previous DoGet is done. #62259 | ||
| connectContext.closeFlightSqlDeferredExecutors(); | ||
| // After the previous query was executed, there was no getStreamStatement to take away the result. |
Contributor
There was a problem hiding this comment.
The new cleanup assumes that starting the next statement means the previous DoGet has finished, but Flight RPCs are independent and this method has no completion signal from the BE result stream. A client can get FlightInfo for q1, keep reading q1's BE endpoint, and issue GetFlightInfo for q2 on the same session; this call will close q1's deferred coordinator. Closing the coordinator stops the scan nodes and removes the batch split sources, while q1's BE scan can still call fetchSplitBatch, which then fails with Split source X is released. This reintroduces the issue for pipelined/interleaved clients. Please keep the deferred coordinator until the result stream is actually complete or the session is closed, or reject/serialize a new statement while a remote Arrow Flight result is still live.
Contributor
Author
There was a problem hiding this comment.
I looked into this and don't think it's a regression this PR introduces, nor something that can be properly fixed within this PR's scope. Details:
Not a new failure mode / not a regression. Before this PR, every external-table batch-split query over Arrow Flight already failed with Split source X is released, because the coordinator was closed at the end of GetFlightInfo — sequential queries included (that is exactly #62259). After this PR the sequential case works; the interleaved case you describe fails as it did before. So nothing is "reintroduced" — interleaving never worked.
Interleaving two queries on one session is not a supported usage. A bearer token maps to a single shared ConnectContext (FlightSessionsWithTokenManager), and that context is single-query-by-construction: each GetFlightInfo resets the shared FlightSqlChannel, clears the single flightSqlEndpointsLocations list, and reuses one queryId. Two concurrent/interleaved queries on one session corrupt all of that, not just the deferred coordinator — the same way sharing one JDBC Connection across threads is unsafe. The supported pattern (drain the result, then issue the next query) makes "the previous DoGet is done by the time the next query starts" hold, since the BE has consumed the SplitSource by the time the client reaches end-of-stream.
The suggested fix needs a signal the FE does not have. For an external-table scan the result DoGet endpoint points directly at the BE; the FE is not in that data path and gets no completion signal when the BE-side stream finishes. "Keep the coordinator until the result stream completes" or "serialize/reject a new statement while a result is live" both require a new BE -> FE completion notification (or session-level serialization keyed on it), which is a separate architectural change out of scope here.
Leaks are already bounded. The "or the session is closed" fallback is implemented: FlightSqlConnectPoolMgr.unregisterConnection() -> closeFlightSqlDeferredExecutors() covers idle/query timeout, bearer-token expiry and explicit CloseSession. And the BE-side hardening in #64797 makes a stale fetchSplitBatch fail gracefully instead of crashing the BE.
If first-class concurrent Flight clients are ever needed, the right follow-up is a BE -> FE completion signal plus session serialization; I can track that separately.
| asynExecutor.finalizeQuery(); | ||
| if (!asynExecutor.isDeferredForArrowFlight()) { | ||
| asynExecutor.finalizeQuery(); | ||
| } |
Contributor
There was a problem hiding this comment.
This guard also skips cleanup when GetFlightInfo fails after a remote executor has already been marked deferred. In that flow, StmtExecutor.executeAndSendResult() adds the executor to the deferred list after coordBase.exec(), then executeQueryStatement() still has to fetch the Arrow schema. If that schema fetch times out, returns non-OK/empty/mismatched schema, or hits an RPC error, try-with-resources calls this close(), but both deferred executors are skipped and the outer catch only rethrows. No FlightInfo is returned, so no DoGet can need the coordinator, but the query registration, query queue slot, and split sources stay alive until a later query or session/token teardown. Please explicitly close the deferred executors on this error path, and add a test that fails schema fetch after deferral.
Contributor
Author
There was a problem hiding this comment.
Confirmed — this is a real (bounded) leak. If the Arrow schema fetch in executeQueryStatement fails after the coordinator was already deferred during planning, FlightSqlConnectProcessor.close() skips the deferred executor and the outer catch only rethrows, so the coordinator (its external-table batch SplitSource, the query queue slot and the query registration) stays alive until the next query starts or the connection is torn down — even though no DoGet will ever pull this query's results.
Fixed in 41a79a6 by finalizing the deferred coordinator on the GetFlightInfo error path: executeQueryStatement's catch now calls connectContext.closeFlightSqlDeferredExecutors(). At that point the list holds only this failed query's coordinator (the previous one was already finalized at the top of the method), and it covers every post-deferral failure, not just the schema fetch.
Added DorisFlightSqlProducerTest.testGetFlightInfoFinalizesDeferredExecutorWhenSchemaFetchFails, which defers a coordinator then fails the schema fetch and asserts the deferred executor is finalized (verified to fail without the fix).
Contributor
Author
|
run buildall |
Contributor
FE UT Coverage ReportIncrement line coverage |
Contributor
TPC-H: Total hot run time: 29096 ms |
Contributor
TPC-DS: Total hot run time: 172666 ms |
Contributor
TPC-H: Total hot run time: 29628 ms |
Contributor
TPC-DS: Total hot run time: 177704 ms |
Contributor
ClickBench: Total hot run time: 25.06 s |
Contributor
Author
|
/review |
Contributor
There was a problem hiding this comment.
Review summary
Requesting changes for four P1 lifecycle and cancellation defects in the new deferred-coordinator ownership.
Critical checkpoint conclusions
- Goal and proof: The change does keep legacy and Nereids external-batch
SplitSources alive across sequentialGetFlightInfo/DoGet, and the Iceberg regression plus schema-fetch-failure unit test exercise the main success and error paths. The lifecycle is not complete because of the four inline findings. - Scope and focus: The plumbing is localized, but the
coordBase == coordcondition is broader than the actual batch-file-scan requirement and retains ordinary distributed queries too. - Concurrency: Flight request threads, token eviction,
CloseSession, administrative/REST cancellation, and workload-policy threads can act independently on the shared context. The list monitor keeps heavy close work outside the lock, but it lacks a terminal add/drain handshake, immutable executor identity, and a cancellation lookup. - Lifecycle and static initialization: Deferred ownership spans separate Flight RPCs and the BE result stream. Release on next request/error/teardown is not sufficient for terminal publication, supported EOF, or cancellation. No static-initialization or circular-reference issue applies.
- Configuration: No configuration is added. Existing defaults (
wait_timeout28,800 seconds and long token lifetime) amplify retained-resource impact. - Compatibility: No storage, wire, function-symbol, or rolling-upgrade format changes were found.
- Parallel paths: Statement and prepared execution share the producer path; both legacy and Nereids distributed coordinators defer. Point queries and local FE results retain their separate cleanup paths. No omitted parallel implementation produced another finding.
- Special conditions: The changed condition is documented but not minimal: only batch file scans need the live
SplitSource, while all distributed BE-result queries enter it. - Tests and results: Added tests cover sequential drain, idempotence, failure continuation, schema-fetch failure, pool unregister, real batch scanning, and next-query cleanup. Missing latch/barrier, two-session queue-release, stable-ID, and post-processor-close cancellation tests map directly to the four inline issues. No changed expected-result defect was found.
- Observability: Finalizer failures are logged and cleanup continues, but query-ID KILL can report success after a null-executor no-op; that is covered by the cancellation finding. No separate observability-only issue was substantiated.
- Transactions and persistence: No
EditLogor persisted metadata changes apply. Delayed Qe unregister can delay Hive transaction deregistration, already included in the resource-retention finding. - Data writes: No data-write or transaction-atomicity path is modified.
- FE-BE propagation: No new FE-BE variable or serialization field is introduced.
- Performance and resources: Broad deferral retains coordinator, scan-node, workload queue, Qe instance, and sometimes Hive ownership; terminal drain-before-add can make that retention unbounded.
- Other issues and deduplication: The two existing threads cover premature cleanup for unsupported interleaved streams and the now-fixed schema-fetch error path. None of the four comments duplicates them or another accepted finding. A proposed catalog-cleanup concern was dismissed because the repository rule is table-specific and neighboring Iceberg suites routinely clean temporary catalogs.
- User focus:
review_focus.txtcontained no additional focus point; the full PR was reviewed. - Validation and completion: Static review only, as required by the task; no local build or test command was run. The checkout also lacks
.worktree_initialized,thirdparty/installed, and the bundledprotoc. Live CI currently has compile, CheckStyle, external, P0, nonconcurrent, performance, and cloud checks passing; FE unit tests are pending and FE coverage is failing. Review convergence completed in three rounds, with all three final reviewers returningNO_NEW_VALUABLE_FINDINGSagainst this exact four-comment set and every candidate accepted, deduplicated, or dismissed.
|
|
||
| public void addFlightSqlDeferredExecutor(StmtExecutor executor) { | ||
| synchronized (flightSqlDeferredExecutors) { | ||
| flightSqlDeferredExecutors.add(executor); |
Contributor
There was a problem hiding this comment.
[P1] Make terminal teardown atomic with deferred registration
Token eviction or CloseSession can call unregisterConnection() on another thread while this query is still inside coordBase.exec(). If teardown drains the currently empty list and removes the context/token mapping first, the query then returns, adds itself here, and FlightSqlConnectProcessor.close() skips it because it is deferred. No later request or timeout can retrieve that removed context, so its Qe registration/instance accounting, queue token, coordinator, and batch SplitSources remain indefinitely. The monitor prevents list corruption but does not close this drain-then-add window. Please mark terminal teardown under the same synchronization and make a late add finalize/reject the executor, with a latch test for this ordering.
| // registration stay held until the next query / teardown instead of being released | ||
| // at the end of GetFlightInfo. Point queries use a different coordBase (not | ||
| // deferred). See #62259. | ||
| if (coordBase == coord) { |
Contributor
There was a problem hiding this comment.
[P1] Do not leave every BE-result query attached to the Flight session
This gate also defers internal-table and non-batch queries even though only an external batch scan needs a live SplitSource. Skipping Coordinator.close() retains the workload-group token, and delaying unregisterQuery() retains per-user instance accounting (and Hive transaction deregistration). After the client fully drains DoGet, the FE receives no completion callback. The Arrow Java 19 driver used by the new regression sends CloseSession from close() only when an optional catalog is present (source); this URL has none, so without another statement the resources remain until the default 28,800-second idle timeout. With workload-group max concurrency 1, one completed q1 therefore leaves q2 on another session queued for up to eight hours. Please restrict deferral to plans that actually own batch SplitSources and provide completion-driven cleanup/decoupled ownership for those plans; add a two-session queue-release test.
| coord.close(); | ||
| } | ||
| } finally { | ||
| finalizeQuery(); |
Contributor
There was a problem hiding this comment.
[P1] Unregister the deferred executor's captured query ID
This calls finalizeQuery(), which removes context.queryId() rather than the ID owned by this executor. closeFlightSqlDeferredExecutors() clears q1 from the list and finalizes it outside the monitor; if token teardown races an already-admitted next request, q2 can see the empty list and handleQuery() can reset/set the shared ID before q1 reaches this line. Q1 then remains registered (or the null removal throws), while q2 may be unregistered and lose status/profile/cancel routing; instance accounting and Hive transaction deregistration use the same wrong key. Please capture the immutable query ID when this executor is registered/deferred and finalize that exact ID, with a blocked-q1-close/q2-ID latch test.
| } | ||
| returnResultFromRemoteExecutor.clear(); | ||
| executor.finalizeQuery(); | ||
| if (executor != null && !executor.isDeferredForArrowFlight()) { |
Contributor
There was a problem hiding this comment.
[P1] Keep deferred queries reachable by cancellation
This branch keeps q1 active for the later DoGet, but ctx.clear() immediately nulls ConnectContext.executor. During DoGet, query-ID lookup still finds this context; however, KILL QUERY calls FlightSqlConnectContext.kill(false) -> cancelQuery(), which now sees no executor and does nothing, after which KILL reports success. REST and workload-policy cancellation use the same no-op path, and none falls back to the retained Qe coordinator/private deferred list, so the BE fragments continue. Please keep deferred executors addressable by their captured query ID (also aligning with stable ownership in the other finding) and route cancellation to the matching executor/coordinator; add a test that closes the processor, cancels q1 during DoGet, and verifies coordinator cancellation plus later single finalization.
Contributor
FE UT Coverage ReportIncrement line coverage |
Gabriel39
approved these changes
Jul 18, 2026
github-actions
Bot
added
the
approved
Contributor
|
PR approved by at least one committer and no changes requested. |
Contributor
|
PR approved by anyone and no changes requested. |
yiguolei
pushed a commit
to yiguolei/incubator-doris
that referenced
this pull request
Aug 5, 2026
… for external table scan (apache#64799) Issue Number: close apache#62259 Related PR: apache#64797 Problem Summary: Arrow Flight SQL queries against Iceberg (and other external) tables in batch split mode crashed the BE / failed with `Split source X is released`. Arrow Flight executes a query in two phases: `GetFlightInfo` (plan + submit to BE) and `DoGet` (the client pulls results from the BE). For an external table scan in batch split mode, the BE keeps scanning during `DoGet` and lazily fetches file splits from the FE via the `fetchSplitBatch` RPC, using an async `SplitSource` that the FE coordinator holds (through its scan nodes). The FE closed the coordinator at the end of `GetFlightInfo` (`StmtExecutor.executeAndSendResult`'s `finally` → `Coordinator.close()` → `ScanNode.stop()` → `SplitSourceManager.removeSplitSource()`) and also unregistered it (`FlightSqlConnectProcessor.close()` → `StmtExecutor.finalizeQuery()`). So by the time the BE called `fetchSplitBatch` during `DoGet`, the `SplitSource` was already gone. The MySQL protocol is unaffected because plan + execute share one request, so the coordinator stays alive until all results are consumed. This PR keeps the coordinator (and its `SplitSource`) alive across the two phases and cleans it up reliably: - **StmtExecutor**: for an Arrow Flight query that produces results on the BE (`coordBase == coord`), mark it deferred, register the executor on the `ConnectContext`, and skip the eager `Coordinator.close()` in the `finally`. A failed query (whose `exec()` threw) is not deferred and is closed as before. - **ConnectContext**: hold the deferred executors and add `closeFlightSqlDeferredExecutors()`, which closes their coordinators (releasing the `SplitSource` and the query queue slot) and unregisters the queries. - **FlightSqlConnectProcessor.close()**: do not finalize deferred executors. - **DorisFlightSqlProducer**: finalize the previous query's deferred coordinator when the next query starts on the connection. - **FlightSqlConnectPoolMgr.unregisterConnection()**: finalize deferred coordinators when the connection is torn down. All teardown paths (idle/query timeout, bearer token expiry, explicit `CloseSession`) reach here, so an abandoned connection cannot leak the coordinator. Non-Arrow-Flight paths (MySQL, internal tables, point queries) are unchanged: `deferredForArrowFlight` can only become true for `ARROW_FLIGHT_SQL`. The BE-side error-path hardening (so any `fetchSplitBatch` failure fails gracefully instead of crashing the BE) is handled separately in apache#64797.
16 tasks
morningman
pushed a commit
that referenced
this pull request
Aug 5, 2026
wyxxxcat
pushed a commit
to wyxxxcat/doris
that referenced
this pull request
Aug 17, 2026
… for external table scan (apache#64799) ### What problem does this PR solve? Issue Number: close apache#62259 Related PR: apache#64797 Problem Summary: Arrow Flight SQL queries against Iceberg (and other external) tables in batch split mode crashed the BE / failed with `Split source X is released`. Arrow Flight executes a query in two phases: `GetFlightInfo` (plan + submit to BE) and `DoGet` (the client pulls results from the BE). For an external table scan in batch split mode, the BE keeps scanning during `DoGet` and lazily fetches file splits from the FE via the `fetchSplitBatch` RPC, using an async `SplitSource` that the FE coordinator holds (through its scan nodes). The FE closed the coordinator at the end of `GetFlightInfo` (`StmtExecutor.executeAndSendResult`'s `finally` → `Coordinator.close()` → `ScanNode.stop()` → `SplitSourceManager.removeSplitSource()`) and also unregistered it (`FlightSqlConnectProcessor.close()` → `StmtExecutor.finalizeQuery()`). So by the time the BE called `fetchSplitBatch` during `DoGet`, the `SplitSource` was already gone. The MySQL protocol is unaffected because plan + execute share one request, so the coordinator stays alive until all results are consumed. This PR keeps the coordinator (and its `SplitSource`) alive across the two phases and cleans it up reliably: - **StmtExecutor**: for an Arrow Flight query that produces results on the BE (`coordBase == coord`), mark it deferred, register the executor on the `ConnectContext`, and skip the eager `Coordinator.close()` in the `finally`. A failed query (whose `exec()` threw) is not deferred and is closed as before. - **ConnectContext**: hold the deferred executors and add `closeFlightSqlDeferredExecutors()`, which closes their coordinators (releasing the `SplitSource` and the query queue slot) and unregisters the queries. - **FlightSqlConnectProcessor.close()**: do not finalize deferred executors. - **DorisFlightSqlProducer**: finalize the previous query's deferred coordinator when the next query starts on the connection. - **FlightSqlConnectPoolMgr.unregisterConnection()**: finalize deferred coordinators when the connection is torn down. All teardown paths (idle/query timeout, bearer token expiry, explicit `CloseSession`) reach here, so an abandoned connection cannot leak the coordinator. Non-Arrow-Flight paths (MySQL, internal tables, point queries) are unchanged: `deferredForArrowFlight` can only become true for `ARROW_FLIGHT_SQL`. The BE-side error-path hardening (so any `fetchSplitBatch` failure fails gracefully instead of crashing the BE) is handled separately in apache#64797.
yiguolei
added a commit
to yiguolei/incubator-doris
that referenced
this pull request
Aug 18, 2026
) ### What problem does this PR solve? pick apache#64799 apache#66437
16 tasks
yiguolei
added a commit
to yiguolei/incubator-doris
that referenced
this pull request
Aug 19, 2026
) ### What problem does this PR solve? pick apache#64799 apache#66437
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
4 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What problem does this PR solve?
Issue Number: close #62259
Related PR: #64797
Problem Summary:
Arrow Flight SQL queries against Iceberg (and other external) tables in batch split mode crashed the BE / failed with
Split source X is released.Arrow Flight executes a query in two phases:
GetFlightInfo(plan + submit to BE) andDoGet(the client pulls results from the BE). For an external table scan in batch split mode, the BE keeps scanning duringDoGetand lazily fetches file splits from the FE via thefetchSplitBatchRPC, using an asyncSplitSourcethat the FE coordinator holds (through its scan nodes).The FE closed the coordinator at the end of
GetFlightInfo(StmtExecutor.executeAndSendResult'sfinally→Coordinator.close()→ScanNode.stop()→SplitSourceManager.removeSplitSource()) and also unregistered it (FlightSqlConnectProcessor.close()→StmtExecutor.finalizeQuery()). So by the time the BE calledfetchSplitBatchduringDoGet, theSplitSourcewas already gone. The MySQL protocol is unaffected because plan + execute share one request, so the coordinator stays alive until all results are consumed.This PR keeps the coordinator (and its
SplitSource) alive across the two phases and cleans it up reliably:coordBase == coord), mark it deferred, register the executor on theConnectContext, and skip the eagerCoordinator.close()in thefinally. A failed query (whoseexec()threw) is not deferred and is closed as before.closeFlightSqlDeferredExecutors(), which closes their coordinators (releasing theSplitSourceand the query queue slot) and unregisters the queries.CloseSession) reach here, so an abandoned connection cannot leak the coordinator.Non-Arrow-Flight paths (MySQL, internal tables, point queries) are unchanged:
deferredForArrowFlightcan only become true forARROW_FLIGHT_SQL.The BE-side error-path hardening (so any
fetchSplitBatchfailure fails gracefully instead of crashing the BE) is handled separately in #64797.Release note
Fix Arrow Flight SQL queries against external tables (e.g. Iceberg) failing with
Split source X is releasedor crashing the BE in batch split mode.Check List (For Author)
Added
regression-test/suites/external_table_p0/iceberg/test_iceberg_arrow_flight_split_source.groovy. It forces batch split mode on the Arrow Flight session (num_files_in_batch_mode=1), asserts viaexplainthat the scan really uses the batchSplitSourcepath (approximate) so it cannot silently pass on the non-batch path, then scansformat_v2.sample_cow_orcover Arrow Flight and checks all rows come back. The test runs in the external (docker) pipeline and is skipped when the Iceberg env or the Arrow Flight endpoint is not configured.Behavior changed:
SplitSource) is now kept alive until the next query starts on the connection or the connection is torn down, instead of being closed at the end ofGetFlightInfo.Does this need documentation?
🤖 Generated with Claude Code