Repository navigation
[fix](arrow-flight) Cancel a Flight query on all its BEs when a result stream closes early - #69996
morningman wants to merge 7 commits into
Conversation
|
Thank you for your contribution to Apache Doris. Please clearly describe your PR:
|
|
run buildall |
|
/review |
Cloud UT Coverage ReportIncrement line coverage Increment coverage report
|
There was a problem hiding this comment.
Static exact-head review of PR 69996 completed after one review round. I traced the result-buffer query/index lifecycle, local and remote Arrow Flight reader cancellation, BRPC callback ownership, FE route registration and retries, Coordinator/Nereids endpoint identities, status filtering, fetch-RPC completion, and the appended thrift method across supported branch snapshots. One P1 finding is included: session reset or teardown can unregister a remote query's cancellation route while its DoGet is still active, so the later abort receives NOT_FOUND and leaves remote pipelines and result buffers running until timeout. The mixed-version older-BE behavior is documented by the PR and its test (fragment cancellation with normal buffer timeout), so it was not reported as a defect. No existing P0/P1 inline findings were present, and the supplied review-focus file contains no additional focus. This was a static review; no builds or tests were executed.
Review converged after 1 round(s).
|
|
||
| public synchronized void reset() { | ||
| resultCache.invalidateAll(); | ||
| FlightSqlQueryCancellation.INSTANCE.unregister(remoteResultIds); |
There was a problem hiding this comment.
[P1] Keep the cancellation route alive while a DoGet is still active. Flight calls are not transport-serialized, so a client can start the next request or close the session while the previous stream is still reading; beginRequest()/teardown reaches this reset() and removes remoteResultIds. When that stream later closes, the BE's cancelFlightQuery gets NOT_FOUND and sends no cancelPipelineXPlanFragmentAsync to the remote result BEs, leaving their pipelines and result buffers running until the timeout. Track active DoGets or cancel before unregistering so reset cannot invalidate an in-flight stream's route.
There was a problem hiding this comment.
Fixed in ba0c1d9926d. Agreed, and the local run shows the case is wider than overlapping streams: the BE reports an abort from its DoGet thread, so a client that closes a stream and goes straight to its next request can start that request before the report reaches the FE. In test_flight_cancel_cleanup, before this change, the report of the first abort came after the suite's health-check request and found no route.
A route now belongs to no session. FlightSqlChannel is back to its master form, and StmtExecutor registers through FlightSqlQueryCancellation.register(ConnectContext, Coordinator), which keeps nothing of the session. A route goes when its cancellation succeeds, when its query's execution timeout plus 5 s has passed, or, oldest first and as a whole, when the registry would hold more than 100,000 result ids. Without the session as a bound, that cap keeps the registry from growing with the query rate times the timeout. The registry is an insertion-ordered map instead of the Caffeine cache, whose size eviction can reject a new entry instead of evicting the oldest one.
Tests: routesOutliveTheSessionsNextRequestAndTeardown registers from a Flight session, runs beginRequest() and releaseProtocolSession(), and still cancels; boundDropsWholeRoutesOldestFirst covers the cap; test_flight_cancel_cleanup gains a mode that runs the session's next request before the abort. On a local cluster, the FE cancels all three aborts of that suite through the route; before the change it cancelled one of them.
TPC-H: Total hot run time: 29312 ms |
TPC-DS: Total hot run time: 151636 ms |
ClickBench: Total hot run time: 25.13 s |
BE UT Coverage ReportIncrement line coverage Increment coverage report
|
|
Addressed review round 1 in
The description is updated to match: the route's lifecycle, the differences from branch-4.x, and the results. |
|
run buildall |
|
/review |
TPC-H: Total hot run time: 28707 ms |
There was a problem hiding this comment.
Static exact-head review of PR 69996 completed after one review round at ba0c1d9. I reviewed all 18 changed paths, including the FE route registry and Coordinator/Nereids backend identity publication, BE local and remote Arrow Flight reader cancellation and BRPC callback lifetime, query-wide result-buffer bookkeeping and mixed-version cancel handling, the appended FrontendService thrift method across branch-3.1/4.0/4.1/4.2 snapshots, and the added unit and regression tests. The prior P1 inline finding 4443420978 was independently revalidated as fixed at this head: FlightSqlChannel no longer owns or unregisters the route, and registration is session-independent; it is not reposted and is omitted from existing_blocking_comment_ids. No additional actionable defect was found. The supplied review-focus file has no additional focus. This was a static review; no builds or tests were executed (tests were inspected only).
Review converged after 1 round(s).
Review converged after 1 round(s).
TPC-DS: Total hot run time: 151724 ms |
ClickBench: Total hot run time: 24.43 s |
BE UT Coverage ReportIncrement line coverage Increment coverage report
|
BE Regression && UT Coverage ReportIncrement line coverage Increment coverage report
|
ba0c1d9 to
8c4c7d0
Compare
|
run buildall |
|
/review |
There was a problem hiding this comment.
Static review of head 8c4c7d0: two new P2 findings concern synchronous FE transport waits during stream teardown and the cancellation registry's unbudgeted backend-address snapshots. No new P0/P1 finding was confirmed.
The existing P1 thread 4443420978 is fixed on this head: route ownership is independent of FlightSqlChannel/session cleanup, and the added test covers cancellation after both the next request and session teardown. Its ID is not retained as blocking, and that finding is not reposted. No additional user focus was supplied.
End-to-end flow inspected: GetFlightInfo dispatches BE fragments -> snapshots result IDs, query ID and all BE addresses -> closes or defers the coordinator -> publishes endpoint tickets -> an early DoGet close reports the result ID to the FE -> the FE sends the existing query-level cancel RPC to every captured BE -> the BE light work pool cancels query pipelines and Flight buffers. EOF suppresses the report.
Critical checkpoint conclusions:
- Goal and proof: query-wide cancellation is wired before endpoint publication, including remote readers and completed local fragments. Added FE/BE unit tests and regression modes cover the intended behavior; the two inline findings identify remaining resource-handling defects.
- Scope and clarity: changes are focused on cancellation routing, two coordinator address accessors, dispatch registration, the service method and associated tests; existing cancellation RPCs are reused.
- Concurrency and locks: registry map operations share one monitor; immutable snapshots and conditional removal protect concurrent cancellation/replacement. RPC dispatch/waits occur outside that monitor. Buffer locks are released before query cancellation, and the light work pool avoids a Flight-pool callback cycle. The synchronous transport wait is reported inline.
- Lifecycle and initialization: routes retain IDs/addresses rather than sessions, coordinators, scan state or queue slots; normal cancellation removes whole routes. No new production cross-translation-unit initialization dependency. Registry retention cost is reported inline.
- Configuration: no settings added; execution timeout is captured per route. Existing Thrift timeout/retry defaults interact with the new synchronous caller as described inline.
- Compatibility: no storage format or existing function signature changed. Old FEs can reject the additive method and retain baseline local cleanup; older BEs understand query cancellation but retain their historical buffer-cleanup behavior.
- Thrift/protobuf: HEAD and live branch-3.1, 4.0, 4.1 and 4.2 definitions inspected. The method is absent on 3.1/4.0 and has the same signature on 4.1/4.2; referenced TUniqueId/TStatus field IDs and types match. No enum values or required struct fields added.
- Parallel paths: old and Nereids coordinators capture all query BEs; CloudCoordinator inheritance, parallel/query-ID and serial/instance-ID tickets, deferred scans, local/remote readers and expired local contexts were traced. Flight disables the separate short-circuit point-query path.
- Conditional checks and errors: NOT_FOUND advances FE discovery; failed cancellation preserves the route for retry; EOF avoids cancelling successful streams. Reopening discards incomplete-response transports. No additional confirmed failure-propagation defect.
- Test coverage: FE tests cover fan-out, application errors, expiry, session independence and whole-route FIFO eviction. BE tests cover remote cleanup, rejected Flight work, retry, old BEs, expired contexts, pending fetch and EOF. Regression checks every live BE for each query in close/cancel/next-request modes. Tests omit inherited transport timeouts/unreachable FEs and multi-BE route-weight accounting, the scenarios raised inline.
- Test results: no expected-output artifacts changed; assertions and mocked responses inspected. No builds, tests, benchmarks, formatting checks or runtime validation were executed in this static-only review. Author-reported runs were not independently reproduced; a single-BE run cannot demonstrate cross-BE cleanup.
- Observability: failures carry query/result IDs, FE/backend addresses or status; existing backend cancellation logging supports tracing. No additional correctness issue in logging was confirmed.
- Transactions and persistence: no EditLog, visibility-version, MoW or durable metadata change. Routes are ephemeral; FE restart loses them and existing BE timeouts remain the fallback.
- Data writes and crash behavior: production changes cancel query execution and buffers, without changing data-write atomicity. Regression writes fixture data only. No new lock leak found; the retained-heap risk is reported inline.
- FE/BE values: the new service passes the ticket's existing result ID; the route translates it to the actual query ID used by the existing backend cancellation RPC in both sink modes.
- Performance and memory: registration snapshots result IDs and all participating BE addresses; eviction removes whole routes. Response waits share one deadline after dispatch. The two inline comments cover the verified timeout and heap-budget problems.
- Other issues and final recheck: multi-statement rejection, retry ID replacement, schema-fetch failure, EOF, mixed-version and cancellation callback paths were rechecked. FIFO expiry behind a live longer-timeout route does not permit expired cancellation because lookup rechecks its deadline; no separate finding was established.
Review converged in 2 rounds. All 11 changed files, assigned checkpoint scopes, existing threads, candidate dispositions and risk focuses were covered; all latest-round reviewers returned NO_NEW_VALUABLE_FINDINGS on the frozen payload. No unresolved static-review scope remains. Runtime validation and workload/heap measurements remain outside this review's authorized scope.
Review converged after 2 round(s).
| // One retry: the FE keeps the route when an attempt fails (for example a BE missed the | ||
| // FE's one-second deadline), and a cached connection to the FE may have gone stale. | ||
| for (int attempt = 0; attempt < 2; ++attempt) { | ||
| FrontendServiceConnection client(env->frontend_client_cache(), address, 2000, &status); |
There was a problem hiding this comment.
[P2] Bound the actual FE call on stream teardown
The 2000 ms argument is applied only when ClientCacheHelper::_create_client creates a transport; a cache hit keeps its previous socket timeout. The shared frontend_client_cache is also used by profile reporting with the default 60000 ms timeout (runtime_query_statistics_mgr.cpp), so an aborted DoGet can inherit a 60-second wait on an unresponsive FE before this loop reaches the FE that owns the route. ClientConnection also defaults to three acquisition attempts with 1+2+4-second backoffs inside each outer attempt, adding at least 14 seconds even on immediate connection refusal. These waits now run in Close() and the reader destructor, occupying the stream thread and delaying cancellation of the remote producers/buffers. Give this path an actual connection/RPC deadline, including reused transports and acquisition retries, or perform the reporting outside stream teardown; cover a long-timeout cached connection and an unreachable FE in the tests.
There was a problem hiding this comment.
Fixed in 152c56c38cc. Agreed, and the cache worked against this call in both directions. A reused transport keeps whatever its opener set: profile and status reports open theirs with thrift_rpc_timeout_ms (60 s), and the information_schema scanners open theirs to the master FE with the query's execution timeout (900 s by default). The other way round, this call's 2 s transports, and the 1 s ones that reopen(1000) made after any thrift error (an older FE's unknown-method reply included), went back to the shared cache, where fetchSplitBatch, whose FE side holds the call for up to fetch_splits_max_wait_time (1 s by default), or a status report could take them. An FE that has stopped also stays on every BE's list: the master sends it with process UUID 0, which get_running_frontends() counts as running, so when it sorts before the FE that planned the query, every early close on every BE waits for it.
The cancellation now keeps FE connections of its own (ExecEnv::flight_cancel_frontend_client_cache, at most two kept per FE), as MasterServerClient does, so each transport it uses was opened with its 2 s timeout and none reaches another caller. Acquiring a connection does not back off (max_retries = 0), a transport is reopened with the same 2 s, and an FE that cannot be connected to gets no second attempt, since neither reason for the retry applies to it. An FE ahead of the planning one now costs an early close about 0.1 s when it refuses connections, 3.1 s when it does not answer the connect (thrift_connect_timeout_seconds), and 4 s when it accepts the connection and does not answer. The local cleanup of #69945 still runs first. The Cost paragraph of the description understated this wait and is corrected.
Tests: CloseIgnoresLongTimeoutSharedTransport leaves a 60 s transport to an FE whose first answer stalls for 5 s in the shared cache and requires the early close to finish within 4 s, through the retry; CloseMovesPastUnreachableFrontend puts an FE that refuses connections first and requires the close to reach the planning FE within 2 s. Before this change they took 5.0 s and 14.7 s.
| long now = nanoClock.getAsLong(); | ||
| while (!results.isEmpty()) { | ||
| Route eldest = results.values().iterator().next(); | ||
| if (results.size() <= capacity && !eldest.isExpired(now)) { |
There was a problem hiding this comment.
[P2] Include backend address snapshots in the registry budget
results.size() counts distinct result IDs, but in the default parallel-result mode every endpoint of a query uses the same query ID, so registration consumes one capacity unit while the Route deep-copies every participating BE address. Normal EOF never unregisters the route. For example, on a 1000-BE cluster, about 111 completed distributed Flight queries/sec during the default 900-second timeout can leave roughly 100000 routes retaining 100 million address objects plus their list references. That can consume gigabytes of FE heap despite the claimed tens-of-MB bound and despite all streams having finished. A distributed aggregate with a single result endpoint has the same mismatch. Account for retained backend addresses in the eviction weight, or otherwise bound/share those snapshots, while preserving whole-route eviction and the session-independent lifetime.
There was a problem hiding this comment.
Fixed in 1c9a89481ed. Agreed. The result ids of a query already share one route and its address list, but the bound counted only the ids, and in parallel result mode a query has one. The registry now also bounds the BE addresses that its routes hold, at 1,000,000 (about 30 MB), and drops whole routes, oldest first, while either bound is exceeded; a route's addresses are released when its last result id leaves, also when a newer registration takes the id over. Under the old bound, a route held about 3 KB with 100 BEs and 29 KB with 1,000, so a full registry held about 0.3 GB and 2.9 GB; now it stays within about 55 MB whatever the cluster size. Large clusters give up retention instead: with 1,000 BEs the registry keeps the routes of the last 1,000 queries, about 9 s of them at 110 queries/s, and a query whose route was pushed out ends at its timeout, as before this PR. Dropping a route when its streams reach EOF would need the BEs to report every EOF to the FE as well, one more FE call per endpoint and query; I left that out.
Tests: boundCountsRetainedBackends requires a registration whose addresses exceed the bound to drop the oldest route as a whole, and reregisteredResultReleasesOldRouteBackends requires a route whose result id another registration took over to release its addresses once.
TPC-H: Total hot run time: 28824 ms |
TPC-DS: Total hot run time: 151469 ms |
ClickBench: Total hot run time: 24.25 s |
BE Regression && UT Coverage ReportIncrement line coverage Increment coverage report
|
…t stream closes early ### What problem does this PR solve? Issue Number: None Related PR: apache#68630, apache#69945 Problem Summary: Port the FE route of branch-4.x apache#68630 to master, on top of apache#69945 (its BE-local half). With apache#69945 alone, a DoGet stream closed before EOF cancels its query only on the BE that holds the stream's results. A query whose results are spread over several BEs (parallel result sink, one endpoint per BE) kept running on the other BEs: master closes the coordinator when GetFlightInfo returns (apache#67504), so nothing sends them a cancel, and a single-fragment query has no exchange that would carry the cancellation between BEs. Their pipelines ran until the query finished or timed out. A remote reader (a DoGet served by a BE other than the result BE) could not cancel the result BE's query at all. The FE now keeps a route for each Flight query that leaves its results on the BEs: from each result id (the fragment instance id of an endpoint) to the query id and the BRPC addresses of all its BEs (FlightSqlQueryCancellation, registered by StmtExecutor before the coordinator is closed or deferred). The route belongs to the session's FlightSqlChannel and is dropped when the session's next request begins (FlightProtocolAdapter.beginRequest resets the channel), when the session is torn down (releaseSession closes the channel), when a cancellation succeeds, and at the latest execution_timeout + 5 s after registration. The new FrontendService RPC cancelFlightQuery(result_id) sends the existing cancel_plan_fragment, which older BEs understand and which does not use the Flight read pool, to every BE of the query and waits at most 1 s for all of them; when any BE fails, the route stays so that the caller can retry. A reader closed before EOF now asks the running FEs in turn: an FE without the route answers NOT_FOUND and the BE moves on to the next one, a failed attempt is retried once, and the transport is reopened after a thrift error. This runs on the DoGet thread, so with an unreachable FE one early close can take a few seconds (two attempts with a 2 s RPC timeout per FE). Compatibility: cancelFlightQuery has the same name and signature as on branch-4.1 and branch-4.2. An FE without it fails the call; the BE logs a warning for that FE and only the local cleanup of apache#69945 takes effect. A result BE without apache#69945 cancels the query but keeps its result buffer until its usual timeout. Differences from branch-4.x: the route is registered in master's `!isReturnResultFromLocal()` branch (apache#67900), before the `mustOutliveDispatch()` deferral, so deferred queries are routed as well; FlightProtocolAdapter needs no change because the route stays in FlightSqlChannel; the BE reader includes runtime/frontend_info.h itself (exec_env.h no longer includes it, apache#66400); the regression suite keeps its table at the end (AGENTS.md). ### Release note Arrow Flight SQL: cancelling or closing a result stream before its end now cancels the query on every BE that runs it, not only on the BE that serves the stream. The FE and the BEs must both be upgraded. ### Check List (For Author) - Test: Unit Test / Regression test - FE UT: FlightSqlQueryCancellationTest (5, new), StmtExecutorTest and 15 other Flight session / coordinator test classes - BE UT: ArrowFlightCancelTest (9), ArrowFlightRemoteCancelTest (9, 6 new), ResultBufferMgrTest (6), ArrowResultBlockBufferTest (4) - Regression: arrow_flight_sql_p0/test_flight_cancel_cleanup (new) and the rest of arrow_flight_sql_p0 on a local single-BE cluster; the cross-BE path is covered by the BE UT's in-process FE and remote BE - Behavior changed: Yes. A Flight result stream closed before EOF cancels its query on every BE of the query. - Does this need documentation: No Co-authored-by: Gabriel <liwenqiang@selectdb.com> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
8c4c7d0 to
1c9a894
Compare
|
run buildall |
|
/review |
|
Codex automated review failed and did not complete. Error: Selected model is at capacity. Please try a different model. Please inspect the workflow logs and rerun the review after the underlying issue is resolved. |
TPC-H: Total hot run time: 28574 ms |
TPC-DS: Total hot run time: 151957 ms |
ClickBench: Total hot run time: 24.21 s |
|
/review |
There was a problem hiding this comment.
Static exact-head review of PR 69996 (head 1c9a894) found three verified issues. The new Flight cancellation route is registered before GetFlightInfo finishes fetching the BE schema; schema or prepared-schema failures cancel the coordinator but leave the route in the global registry until expiry, allowing failed requests to evict live routes and later DoGet closes to miss cancellation. The BE reader loop binds an unused info element under -Wunused/-Werror, which blocks compilation. The reader also probes every running FE serially with no overall deadline; multiple process_uuid==0 stale FEs can accumulate per-FE connection/RPC timeouts before the owning FE is reached, delaying stream teardown and remote cancellation. The registry accounting, expiry, replacement, partial-failure retry, local/remote cleanup, cache ownership, and Thrift compatibility paths were rechecked. Existing threads r4463117156, r4463117060, and r4443420978 were read and are addressed by this head, so they are not reposted. The required user review focus file contains no additional focus points. No builds or tests were run because this invocation is static-only.
Review converged after 2 round(s).
| } | ||
| // A result endpoint can outlive its local fragment. Find the owning FE by buffer ID, | ||
| // then use its query-wide cancellation route, including for older result BEs. | ||
| for (const auto& [address, info] : env->get_running_frontends()) { |
There was a problem hiding this comment.
[P1] Avoid the unused structured-binding element here. info is never referenced, while BE Clang builds enable -Wunused under -Werror, so this new loop can fail compilation with an unused-variable diagnostic. Iterate the map keys or use the repository's intentionally ignored binding convention.
There was a problem hiding this comment.
This does not break the build. Clang's -Wunused-variable reports a structured binding declaration only when none of its bindings is referenced, and address is used in the loop. CI built this file at that head with Clang 20.1.8 and the same flags, -Wall -Wextra -Werror and -Wunused among them: TeamCity BE UT build 1092350 compiled arrow_flight_batch_reader.cpp.o without a diagnostic and stopped later at master's own vparquet_writer_test.cpp break (fixed by #70209), and the COMPILE build passed. branch-4.1 and branch-4.2 have had the same line since #68630, and master has many loops that bind an element they never read, for example vtablet_writer_v2.cpp:484 and exchange_source_operator.h:125. With 35aece4, the change for the other comment on this loop, info is read now: the FEs are ordered by its process UUID.
| // A BE whose DoGet stream the client aborts reports it through cancelFlightQuery; | ||
| // keep the query's backends so that call can cancel the query on every one of | ||
| // them after the coordinator below is closed. | ||
| FlightSqlQueryCancellation.INSTANCE.register(context, coord); |
There was a problem hiding this comment.
[P2] Remove the cancellation route when GetFlightInfo fails. register() runs at this line before DorisFlightSqlProducer.fetchArrowFlightSchema(5000) and before the returned FlightInfo/ticket exists. If that schema RPC times out or errors (or a later prepared-schema check rejects the result), the producer catch only cancels/finalizes the coordinator; it never removes the route. The result IDs and BE snapshots then occupy the global registry until timeout+5s, so repeated failed requests can evict routes for successful queries and make their later DoGet cancellation return NOT_FOUND. Register after schema admission or explicitly unregister this failed query path.
There was a problem hiding this comment.
Fixed in 0193881, and c528cf8 goes a step further. Agreed that such a route can never be used: the request hands out no ticket, so no BE can report an early close of its results. A GetFlightInfo that fails after it began, at the schema fetch or because a prepared statement's result schema no longer matches, now drops the routes of its endpoints before it answers; while routes belonged to the session, its next request dropped them. Only the failing request's routes go: a request that fails before it begins (an empty query) finds the previous request's endpoints on the session and leaves their routes, since those streams may still be open.
Such a failure also left the query running when it was not deferred: its coordinator was closed already, so nothing cancelled it before the BE's result buffer timer (execution_timeout + 5 s). c528cf8 sends the cancellation through the route to every BE of such a query before the route goes. A deferred query is still cancelled through its coordinator, and not a second time; the prepared-statement branch has cancelled its query through the executor already, so it only drops the route. The failure path does not wait for the BEs' answers: the route goes whatever they are, and the client is waiting for the error.
The effect of the stale route was smaller than described, though. A query read to its end keeps its route until it expires as well, since a BE does not report EOF to the FE, so a failed request took no more room than a successful one; the bound fills only at about 110 registrations a second sustained (default timeout, up to about 10 BEs a query), failed or not, and a query whose route is pushed out ends at its timeout, as before this PR. I left the registration where it is instead of moving it after the schema check: that is where the coordinator is closed or deferred, and the statement and prepared-statement paths would each need a registration of their own.
Tests: DorisFlightSqlProducerTest#testGetFlightInfoCancelsQueryAndDropsRouteWhenSchemaFetchFails (both BEs of the query get the cancellation and the route goes), #testGetFlightInfoLeavesDeferredQueryToItsCoordinator, #testFailedRequestKeepsPreviousRequestRoute (an empty query leaves the previous request's route and cancels nothing), DorisFlightSqlSchemaTest#executionSchemaDriftExpiresHandleAndCleansResults (the prepared-statement case), and FlightSqlQueryCancellationTest#unregisterDropsWholeRoutes and #failedRequestCancelsWithoutWaitingAndDropsRoutes.
| // An FE that sorts before the one that planned the query delays every early close on | ||
| // every BE. So the connection comes from a cache that holds only transports opened | ||
| // with this 2 s timeout, and acquiring it does not back off (max_retries = 0). | ||
| FrontendServiceConnection client(env->flight_cancel_frontend_client_cache(), address, |
There was a problem hiding this comment.
[P2] Bound the total FE probe during stream teardown. close() asks every running FE in address order, and process_uuid == 0 entries are intentionally treated as alive; with several stale FEs before the planner, each can consume its connection/RPC retry budget before the loop reaches the route owner. The dedicated cache fixes per-FE reuse/backoff but leaves cumulative delay proportional to the number of stale FEs, blocking the DoGet/destructor thread and postponing remote cancellation. Use one overall deadline, remember/prioritize the planner FE, or probe the FEs concurrently.
There was a problem hiding this comment.
Fixed in 35aece4. The FEs the master reaches are now asked first, and those it reports with process UUID 0 (unreachable, and kept on a BE's list until they are dropped) after them. The FE that planned the query is reachable unless it has lost the master's heartbeat since, and then it is still asked, last. So a stopped FE delays an early close only when no reachable FE holds the route: when the route has expired or been pushed out, or when the planning FE is itself the unreachable one. An FE that hangs while the master still reaches it costs up to 4 s until the master's heartbeat to it fails.
I did not add an overall deadline: with this order it would only cut those cases short, and in the last one the early close would lose its cancellation instead of waiting for it. Probing the FEs concurrently would take a thread per FE for every early close, since these thrift clients are synchronous. Sending the cancellation only to the planning FE needs the ticket to name that FE, which is left to a follow-up.
Test: ArrowFlightRemoteCancelTest#CloseAsksReachableFrontendsFirst puts an FE that accepts connections and never answers, reported with process UUID 0, ahead of the planning FE, and requires the close to finish within 1 s with one call to the planning FE. It takes 14 ms; with the FEs asked in address order it took 4.1 s.
… fails ### What problem does this PR solve? Issue Number: None Related PR: apache#69945 Problem Summary: A Flight query whose results stay on the BEs registers its cancellation route in `StmtExecutor`, before GetFlightInfo fetches the Arrow schema from the result BE and builds the tickets. When the request fails after that, at the schema fetch or because a prepared statement's result schema no longer matches, no ticket names the query's results, so no BE can report an early close of them, yet the route stayed in `FlightSqlQueryCancellation` until its query's timeout plus 5 s, or until newer routes pushed it out. While routes belonged to the session, its next request dropped them; since routes outlive the session, nothing did. A GetFlightInfo that fails after it began now drops the routes of its endpoints, each as a whole, and so does the prepared-statement schema check before it resets the session. A request that fails before it begins, such as an empty query, finds the previous request's endpoints on the session and leaves their routes: a client may still be pulling those results. ### Release note None ### Check List (For Author) - Test: Unit Test - `DorisFlightSqlProducerTest#testGetFlightInfoDropsRouteWhenSchemaFetchFails` (a failed schema fetch drops the route its query registered), `#testFailedRequestKeepsPreviousRequestRoute` (an empty query leaves the previous request's route), and `FlightSqlQueryCancellationTest#unregisterDropsWholeRoutes` (a route goes as a whole, with its addresses). `DorisFlightSqlSchemaTest#executionSchemaDriftExpiresHandleAndCleansResults` now also requires the route of a prepared statement whose result schema changed to go, with no second cancellation. - Behavior changed: No - Does this need documentation: No Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
… their routes ### What problem does this PR solve? Issue Number: None Related PR: apache#69945 Problem Summary: When a GetFlightInfo fails after its query ran, at the Arrow schema fetch for instance, no client can pull the query's results. A deferred query is then cancelled through its coordinator, which is still open. Any other query has closed its coordinator already, and nothing cancelled it: its fragments and result buffers ran on until the BE's result buffer timer ended them, execution_timeout plus 5 s later. The query's cancellation route holds what a cancellation needs. The failure path now sends the cancellation through it to every BE of each query that no deferred coordinator cancelled, and drops the routes as before. A deferred query whose coordinator fails to cancel it falls back to its route. The failure path does not wait for the BEs' answers: the routes go whatever they are, so no retry follows, the client is waiting for the request's error, and a deferred coordinator's cancellation does not wait for them either. ### Release note None ### Check List (For Author) - Test: Unit Test - `DorisFlightSqlProducerTest#testGetFlightInfoCancelsQueryAndDropsRouteWhenSchemaFetchFails` (a query that was not deferred is cancelled on both its BEs and its route goes), `#testGetFlightInfoLeavesDeferredQueryToItsCoordinator` (a deferred query is cancelled once, through its coordinator), and `FlightSqlQueryCancellationTest#failedRequestCancelsWithoutWaitingAndDropsRoutes` (BEs that never answer do not hold the request up). - Behavior changed: No - Does this need documentation: No Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
### What problem does this PR solve? Issue Number: None Related PR: apache#69945 Problem Summary: A Flight result stream closed before EOF asks the running FEs, in address order, to cancel its query until one answers OK. Only the FE that planned the query holds the route, and the list includes the FEs the master reports with process UUID 0: those it cannot reach, which stay on a BE's list until they are dropped. Every such FE that sorted before the planning FE delayed every early close on every BE by its connection budget: about 0.1 s when it refuses connections, 3.1 s when it does not answer the connect, and 4 s when it accepts connections and never answers. The planning FE is one the master reaches, unless it has lost the master's heartbeat since. The FEs with a non-zero process UUID are now asked first and the others last, so a stopped FE delays an early close only when no reachable FE holds the route: when the route has expired or been pushed out, or when the planning FE is itself the unreachable one. An FE that hangs while the master still reaches it costs up to 4 s as before, until the master's heartbeat to it fails. No overall deadline is added: with this order it would only cut those cases short, and in the last one the early close would lose its cancellation. ### Release note None ### Check List (For Author) - Test: Unit Test - `ArrowFlightRemoteCancelTest#CloseAsksReachableFrontendsFirst`: an FE that accepts connections and never answers, reported with process UUID 0 and sorting before the planning FE, no longer delays the close (14 ms; about 4 s with the FEs asked in address order). - Behavior changed: No - Does this need documentation: No Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
|
run buildall |
|
/review |
TPC-H: Total hot run time: 28304 ms |
TPC-DS: Total hot run time: 151549 ms |
ClickBench: Total hot run time: 25.09 s |
There was a problem hiding this comment.
Static review of apache/doris PR #69996 at head 35aece4 completed in one round and converged. No new inline finding is accepted. The existing threads were read first: the session-owned route bug, shared FE-client timeout, registry address bound, schema-failure route cleanup, unused-binding concern, and cumulative FE-probe concern are either fixed at this head or explicitly addressed in their replies, so none is reposted and no existing P0/P1 remains applicable.
The task goal is to cancel an Arrow Flight query across every participating BE when a result stream closes early, including remote readers, while preserving result cleanup for older nodes. The FE route is registered after remote execution has endpoint and backend information and before the coordinator is closed or deferred; the BE reader asks heartbeat-listed FEs in reachable-first order, uses a cancellation-only client cache with bounded connection attempts, and routes the result id to every captured BE. Local and remote reader paths, normal EOF, query-level cancellation, and old-BE handling were traced end to end.
Concurrency and lifecycle checkpoints are satisfied by the synchronized route registry and identity-aware whole-route removal, immutable route snapshots during asynchronous cancellation, the request/deferred-executor ownership rules, and the dedicated BE client-cache lifetime. Replacement, expiry, capacity eviction, partial backend dispatch, retry retention, session next-request, teardown, schema-fetch failure, and prepared-schema mismatch were checked. No lock-order, use-after-reset, or counter-drift issue was substantiated.
The change adds no configuration, persistence, transaction, storage-format, or user-authored network endpoint. The new FrontendService method is appended with an optional argument; the same signature is present on branch-4.1 and branch-4.2, while older branches do not reuse an id. Unknown-method responses from older FEs are caught by the BE path and it continues probing. The FE/BE cancellation status conversion and the existing cancel_plan_fragment compatibility path were inspected.
The changed unit and regression tests cover parallel and nonparallel result endpoints, local and remote readers, EOF versus abort, retry and stale transports, reachable and unknown FEs, old result BEs, route bounds and replacement, failed GetFlightInfo cleanup, prepared-schema drift, next-request and session teardown, and a subsequent healthy query. These tests were inspected but not executed because the review contract is static-only and prohibits builds and test runs. No user-provided review focus was supplied.
Review convergence: true. Coverage includes every changed path and all assigned subagent scopes; each subagent returned NO_NEW_VALUABLE_FINDINGS after its final recheck.
Review converged after 1 round(s).
FE UT Coverage ReportIncrement line coverage |
BE UT Coverage ReportIncrement line coverage Increment coverage report
|
BE Regression && UT Coverage ReportIncrement line coverage Increment coverage report
|
|
run p0 |
|
run nonConcurrent |
BE Regression && UT Coverage ReportIncrement line coverage Increment coverage report
|
1 similar comment
BE Regression && UT Coverage ReportIncrement line coverage Increment coverage report
|
BE Regression && UT Coverage ReportIncrement line coverage Increment coverage report
|
BE Regression && UT Coverage ReportIncrement line coverage Increment coverage report
|
BE Regression && UT Coverage ReportIncrement line coverage Increment coverage report
|
What problem does this PR solve?
Issue Number: None
Related PR: #68630, #69945
Problem Summary:
Builds on #69945, now merged. This branch is rebased onto master
0eb5131db62and holds seven commits: the port; its first review fix, which keeps a route past the session's next request; two fixes from the second review, which bound the BE's FE calls on stream teardown and count the BE addresses of the routes in the registry bound; and three from the third review, which drop the routes of a GetFlightInfo that fails, cancel through those routes its queries that no deferred coordinator cancels, and have the BE ask the FEs the master reaches first.In short
This ports the FE half of branch-4.x's #68630 to master. With #69945, an Arrow Flight SQL stream that the client closes or cancels before its end cancels its query only on the BE that serves the stream, and only when that BE also holds the stream's results. A query whose results are spread over several BEs keeps running on all the other BEs until it finishes or times out. With this PR the FE remembers, for each Flight query that leaves its results on the BEs, which BEs run it; the BE that sees the aborted stream asks the FE, through the new
FrontendService.cancelFlightQuery, to cancel the query on every one of them.Background
GetFlightInforeturns one endpoint for it; the client reads each endpoint with its ownDoGetstream.GetFlightInfohas returned (a query that still needs it, a batch-mode external-table scan or a remote Doris scan, is kept until the session's next request). After that, nothing on the FE cancels the query on behalf of an open stream.DoGetis served byArrowFlightBatchLocalReaderwhen the result buffer is on the same BE, or byArrowFlightBatchRemoteReader, which pulls the blocks from the result BE with thefetch_arrow_dataBRPC.cancel_plan_fragmentfree the query's Flight buffers on the BE that receives it.The problem, and what it cost
LIMIT, a cancelled cursor, a dropped connection), the other BEs run their pipelines to the end and keep their unread results until the query's timeout or the delayed buffer cleanup; a result that does not fit in a buffer back-pressures the result sink until the query times out.How this PR fixes it
FlightSqlQueryCancellationmaps each result id (the fragment instance id of an endpoint) to the query id and the BRPC addresses of all the query's BEs (Coordinator.getBackendBrpcAddresses(), from its per-BE tasks).StmtExecutorregisters it when the query leaves its result on the BEs (!isReturnResultFromLocal()), before the coordinator is closed or deferred (FlightSqlQueryCancellation.register(ConnectContext, Coordinator)). The route belongs to no session. The client reads the results from the BEs, so it can still hold a stream of the query open when the session's next request begins or the session is torn down, and close it early afterwards. The BE also reports an abort from itsDoGetthread, so even a client that closes a stream before it sends its next request can be overtaken by that request. A route goes when a cancellation through it succeeds, when the GetFlightInfo that registered it fails before answering with tickets,execution_timeout + 5 safter it was registered, or, oldest first and as a whole, when the registry would hold more than 100,000 result ids or 1,000,000 BE addresses. A route costs about 270 bytes plus about 30 bytes per BE address, and the result ids of a query share it, so the registry stays within about 55 MB however many BEs a query runs on. Without the session as a bound, these caps keep the registry from growing with the query rate times the timeout; a query whose route was pushed out ends at its timeout, as before this PR. The registry is an insertion-ordered map under a lock, and the cancellation RPCs are sent outside it.cancelFlightQuery(result_id), a new FrontendService RPC. It answers NOT_FOUND when the FE has no route for the id. Otherwise it sends the existingcancel_plan_fragmentto every BE of the query, waits at most 1 s in total for their answers, and returns OK only when all of them succeeded; on any failure the route stays, so that the caller can retry.cancel_plan_fragmentis understood by older BEs and runs on the light work pool, not on the Flight read pool, so it also gets through when the Flight pool is saturated or shut down.execution_timeout + 5 s): its route now sends the cancellation to every BE of the query, without waiting for the answers, and the request's routes go. The prepared-statement check cancels its query through the executor and only drops the route. A request that fails before it begins, such as an empty query, finds the previous request's endpoints on the session and leaves their routes.ArrowFlightBatchReaderBase::close()callscancelFlightQueryon each running FE in turn, first the FEs the master reaches and then those it reports with process UUID 0. OK ends the loop; NOT_FOUND (an FE that did not plan the query) moves on to the next FE; any other failure is retried once, and the transport is reopened after a thrift error. These calls use FE connections of their own (ExecEnv::flight_cancel_frontend_client_cache, at most two kept per FE), all opened with a 2 s timeout, and acquire them without backing off; an FE that cannot be connected to is not retried. A stream read to EOF never calls the FE.FrontendServicewith the same name, signature and comment as on branch-4.1 and branch-4.2.Upgrades and mixed versions. Both the FE that planned the query and the BE that serves the stream need this PR.
cancelFlightQuery, logs a warning for that FE, and only the local cleanup of #69945 appliesCost. The FE calls run on the
DoGetthread, including from the reader's destructor, after the local cleanup of #69945. A stopped FE stays on the BEs' list until it is dropped, reported with process UUID 0, and the FEs the master reaches are asked first, so a stopped FE delays an early close only when no reachable FE holds the route (the route expired or was pushed out, or the planning FE is itself unreachable): by about 0.1 s when it refuses connections, 3.1 s when it does not answer the connect (thrift_connect_timeout_seconds), and 4 s when it accepts the connection and does not answer. An FE that hangs while the master still reaches it costs up to 4 s until the master's heartbeat to it fails. A GetFlightInfo that fails sends its cancellations without waiting for them. On the FE,cancelFlightQueryholds one thrift thread for at most 1 s, and each idle connection a BE keeps for these calls holds another. The route keeps only ids and addresses, not the coordinator, its scan state or its query queue slot, and the registry holds at most 100,000 result ids and 1,000,000 BE addresses, about 55 MB.Differences from branch-4.x: the route is registered in master's
!isReturnResultFromLocal()branch (#67900) instead of theARROW_FLIGHT_SQLconnect type branch, before themustOutliveDispatch()deferral, so deferred queries are routed as well; a route does not belong to the session: branch-4.x drops it when the session's next request begins and when the session is torn down (FlightSqlChannel.reset()), so an abort reported after either finds no route and the other BEs run until the timeout, while here a route stays until its cancellation succeeds or it expires, and the registry is bounded instead; the BE reader includesruntime/frontend_info.hitself, becauseexec_env.hno longer includes it (#66400); the BE test reads and restoresExecEnv::_frontendsthrough itsunique_ptr; the FE test is JUnit 5 and tests that a route outlives the session's next request and its teardown, and the bound; the regression suite keeps its table at the end, as AGENTS.md asks; the two branches of the BE's FE loop got comments; the BE asks the FEs the master reaches first, where branch-4.x asks them in address order; and a GetFlightInfo that fails cancels through their routes its queries that were not deferred, where branch-4.x only finalizes the deferred executors and cancels nothing.Results
FlightSqlQueryCancellationTest(9 tests, new): a cancellation reaches every BE of the route after the coordinator and the selected endpoint's fragment are gone, and a successful one removes the route for all of its result ids; a BE that answers an error keeps the route, and the retry succeeds; routes expireexecution_timeout + 5 safter registration; a route registered from a session survives the session's next request (beginRequest()) and its teardown (releaseProtocolSession()) and still cancels; the bound drops whole routes, oldest first, also when the BE addresses exceed it; a route whose result id a newer registration took over releases its addresses; an expired route leaves at the next registration; a request that fails before answering with tickets drops its routes as a whole, and cancels the queries no deferred coordinator cancelled on all their BEs without waiting for them.DorisFlightSqlProducerTest: a GetFlightInfo whose schema fetch fails cancels its query on both its BEs and drops its route; a deferred query is cancelled once, through its coordinator; an empty query leaves the previous request's route and cancels nothing.DorisFlightSqlSchemaTest#executionSchemaDriftExpiresHandleAndCleansResultsnow also requires the route of a prepared statement whose result schema changed to go.StmtExecutorTest#testCloudReplanDispatchProfileLifecyclenow checks, for its two Flight outcomes (closed and deferred coordinator), that the dispatched query registered its route; its coordinator mock returns the query id and one BE address, and itsexec()adds one endpoint, as the coordinator does.ArrowFlightRemoteCancelTestgets the FE half of branch-4.x's fixture back: an in-process thrift FE (FlightCancelCoordinator) that answerscancelFlightQueryby sendingcancel_plan_fragmentto an in-process BRPC result BE. The 9 new tests: a remote reader's close reaches the result BE; the cancellation also works when the Flight work pool rejects work; an FE error is retried once; an older result BE that only understands query cancellation gets it and keeps its buffer; a local reader whose query context has expired still reports the abort; a stream read to EOF does not call the FE; a close is not held up by a 60 s transport that another caller left in the shared FE cache to an FE whose first answer stalls, moves past an FE that refuses connections, and asks an FE that accepts connections and never answers, reported with process UUID 0 and sorting first, only after the planning FE (14 ms; 4.1 s in address order).CancellationInterruptsPendingBrpcFetchandFetchAfterCancellationStillCompletesRpcare back to branch-4.x's form, which needs the FE route ([fix](be) Cancel Arrow Flight queries on the result BE when a result stream closes early #69945 carried variants without it).ArrowFlightCancelTest,ArrowFlightRemoteCancelTest,ResultBufferMgrTest,ArrowResultBlockBufferTest, 30 tests) pass 10 times in a row in one ASAN process.StmtExecutordoes not register; and, for the review fix, the session drops its routes again at its next request and teardown; the bound evicts the route just registered instead of the oldest one, or single result ids instead of whole routes; an expired route can still be cancelled; registration keeps expired routes. On the BE, the reader stops calling the FEs (the 7 tests that need the route turn red, the other 11 stay green); the reader does not retry. For the second review's fixes: on the BE, taking the connections from the shared cache and acquiring them with the default backoff (one round) turn the two new tests red, each its own (the close took 5.0 s and 7.4 s); with the reader as it was before these fixes they took 5.0 s and 14.7 s. On the FE, not releasing a route's addresses, releasing them per result id instead of with the last one, leaving the address bound out of the eviction, and keeping the addresses of the route an id leaves when a newer registration takes it over each turn their target test red. That an FE which cannot be connected to gets no second attempt has no test of its own: against an FE that refuses connections the second attempt costs about 0.1 s, and the difference only shows with an FE that does not answer the connect (3.1 s instead of 6.2 s). For the third review's fixes: on the FE, dropping the routes whether or not the request has begun, dropping single result ids instead of whole routes, cancelling deferred queries through their routes too, dropping the routes without cancelling, not handling the failure at all, waiting for the BEs' answers, and leaving the prepared-statement branch's route each turn their target tests red; on the BE, asking the FEs in address order turns the new test red (4.1 s).arrow_flight_sql_p0/test_flight_cancel_cleanup(new; it aborts one endpoint after the first batch, by closing it, by an explicit cancel, and by closing it after the session ran its next request, and each time requires the query's pipelines to disappear from every BE within 20 s) passes, and the BE log shows thecancel_plan_fragmentthat the FE sent through the route for all three aborts. Before the review fix it showed it for one of the three: the third abort found no route, and so did the first, whose report came after the suite's next request (its first attempt met a stale connection after an FE restart; the retry reopened it and found no route). The other 20arrow_flight_sql_p0suites and the 14external_table_p0/remote_dorissuites pass. A single BE cannot show the cross-BE effect: there the BE serving the stream already cancels the query itself ([fix](be) Cancel Arrow Flight queries on the result BE when a result stream closes early #69945), so the suite passes without this PR too; on several BEs it needs this PR. Locally the cross-BE path is covered by the BE UT's in-process FE and remote result BE.0eb5131db62, after [refactor](coordinator) Unify query execution in Coordinator #69830 removedNereidsCoordinatorand made its implementation theCoordinator:getBackendBrpcAddresses()is now a method of that class, with the body of the formerNereidsCoordinatoroverride (the legacy implementation went with the old class).git range-diffshows no other change to the first commit and none to the second. The FE and BE tests above ran on this base; master'sdoris_be_testdoes not build since [feature](be) Add session controls for Parquet page indexes #69971 (vparquet_writer_test.cpp), so the BE UT here ran with that test's one-line fix ([fix](test) Build vparquet_writer_test against the location-based Iceberg WriteInfo #70209) applied locally. The regression suites were not rerun on this base, including thearrow_flight_sql_p0suites master added since1adeb3a6801.Release note
Arrow Flight SQL: cancelling or closing a result stream before its end now cancels the query on every BE that runs it, not only on the BE that holds the stream's results, and a GetFlightInfo that fails after its query ran cancels that query instead of leaving it to run until its timeout. This needs both the FE and the BEs to be upgraded.
Check List (For Author)
FlightSqlQueryCancellationTest(7, new),StmtExecutorTest(55) and 26 other Flight session / coordinator classes: 297 tests pass; fe-core Checkstyle 0 violations. Third review:FlightSqlQueryCancellationTest(9),DorisFlightSqlProducerTest(15),DorisFlightSqlSchemaTest(39) and 6 other Flight / coordinator classes, 149 tests pass on35aece4ee7f, fe-core Checkstyle 0 violationsArrowFlightCancelTest(9),ArrowFlightRemoteCancelTest(11, 8 new),ResultBufferMgrTest(6),ArrowResultBlockBufferTest(4): 30 tests pass 10 times in a row (ASAN, macOS); 7 mutations (5 FE, 2 BE) each caught, 5 more on the FE for the first review fix, 6 more (2 BE, 4 FE) for the second, and 8 more (7 FE, 1 BE) for the thirdbuild-support/check-build-hygiene.shpasses; clang-format 16 clean; clang-tidy reports nothing on the changed linesarrow_flight_sql_p0(21 suites, including the newtest_flight_cancel_cleanup) andexternal_table_p0/remote_doris(14 suites) pass on a local single-BE cluster, also after the review fix; the cross-BE path is only covered by the UTs locally0eb5131db62: the FE and BE UT above; the regression suites were not rerun on this baseArrowFlightCancelTest,ArrowFlightRemoteCancelTest, now 12 with 9 new) 10 times in a row, on35aece4ee7f; clang-format, clang-tidy and the build hygiene check clean; the regression suites were not rerunOriginal author: @Gabriel39