morningman opened a new pull request, #69996:
URL: https://github.com/apache/doris/pull/69996
### What problem does this PR solve?
Issue Number: None
Related PR: #68630, #69945
Problem Summary:
**Stacked on #69945.** The first two commits are #69945 (the BE-local half
of this port and its review fix); this PR adds the last commit. Once #69945 is
merged, this branch is rebased onto master and only the last commit remains.
#### 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
- A Flight SQL query leaves its result on the BEs. With parallel result sink
(the default), each BE that runs the result fragment holds one result buffer
and `GetFlightInfo` returns one endpoint for it; the client reads each endpoint
with its own `DoGet` stream.
- Since #67504 the FE closes the coordinator as soon as `GetFlightInfo` has
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.
- A `DoGet` is served by `ArrowFlightBatchLocalReader` when the result
buffer is on the same BE, or by `ArrowFlightBatchRemoteReader`, which pulls the
blocks from the result BE with the `fetch_arrow_data` BRPC.
- #69945 makes a reader that is closed, cancelled, fails or is destroyed
before EOF cancel the query on its own BE when the buffer is local, and lets an
aborting `cancel_plan_fragment` free the query's Flight buffers on the BE that
receives it.
#### The problem, and what it cost
- **An abort on one endpoint stays on that BE.** A single-fragment query has
no exchange between its BEs, so cancelling it on one BE does not reach the
others, and the coordinator that could have sent them a cancel is already
closed. When the client aborts one endpoint (a client-side `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.
- **A remote reader cannot cancel anything.** #69945 interrupts its pending
fetch, but the query on the result BE keeps running.
- **A local reader whose query context has expired** frees its own buffers
but cannot tell the query's other BEs.
#### How this PR fixes it
- **A route per Flight query on the FE.** `FlightSqlQueryCancellation` maps
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()`, overridden by `NereidsCoordinator`
from its per-BE tasks). `StmtExecutor` registers it when the query leaves its
result on the BEs (`!isReturnResultFromLocal()`), before the coordinator is
closed or deferred, through the session's `FlightSqlChannel`. The channel drops
its routes in `reset()`, which runs when the session's next request begins
(`FlightProtocolAdapter.beginRequest()`) and when the session is torn down
(`releaseSession()` closes the channel). A route also expires
`execution_timeout + 5 s` after it was registered, and is removed once a
cancellation through it succeeds.
- **`cancelFlightQuery(result_id)`, a new FrontendService RPC.** It answers
NOT_FOUND when the FE has no route for the id. Otherwise it sends the existing
`cancel_plan_fragment` to 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_fragment` is
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.
- **The BE reports every abort.** After the local cleanup of #69945,
`ArrowFlightBatchReaderBase::close()` calls `cancelFlightQuery` on each running
FE in turn. 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. A stream read to EOF never calls the FE.
- The thrift method is appended to `FrontendService` with the same name,
signature and comment as on branch-4.1 and branch-4.2.
| Situation | Before (master with #69945) | After |
|---|---|---|
| Client aborts one endpoint of a query whose results are on several BEs |
Only the BE holding that endpoint cancels the query; the other BEs run to the
end and keep their unread results until the timeout or the delayed cleanup |
The query is cancelled on every BE of the query |
| Client aborts a stream served by a remote reader | The pending fetch is
interrupted; the query keeps running on the result BE | The query is cancelled
on the result BE and on all other BEs |
| Client aborts a local stream after its query context expired | Only the
local buffers are freed | The query's other BEs are cancelled too |
| FE that did not plan the query is asked | — | Answers NOT_FOUND; the BE
asks the next FE |
| Stream read to EOF, then closed | Nothing cancelled | Nothing cancelled;
the FE is not called |
**Upgrades and mixed versions.** Both the FE that planned the query and the
BE that serves the stream need this PR.
| FE | Result BE | Effect of an abort |
|---|---|---|
| Without this PR | any | The BE gets an unknown-method error from
`cancelFlightQuery`, logs a warning for that FE, and only the local cleanup of
#69945 applies |
| With this PR | With #69945 | The query is cancelled and its Flight buffers
are freed on every BE |
| With this PR | Without #69945 | The query's fragments are cancelled; that
BE keeps its result buffer until its usual timeout |
**Cost.** The FE calls run on the `DoGet` thread, including from the
reader's destructor. With an unreachable FE, an early close waits up to two
attempts with a 2 s RPC timeout for that FE before it moves on, so one abort
can take a few seconds. On the FE, `cancelFlightQuery` holds one thrift thread
for at most 1 s. The route keeps only ids and addresses, not the coordinator,
its scan state or its query queue slot.
Differences from branch-4.x: the route is registered in master's
`!isReturnResultFromLocal()` branch (#67900) instead of the `ARROW_FLIGHT_SQL`
connect type branch, before the `mustOutliveDispatch()` deferral, so deferred
queries are routed as well; master's `FlightProtocolAdapter` already resets and
closes the channel at the right points, so it needs no change; the BE reader
includes `runtime/frontend_info.h` itself, because `exec_env.h` no longer
includes it (#66400); the BE test reads and restores `ExecEnv::_frontends`
through its `unique_ptr`; the FE test is JUnit 5 and adds one test for master's
two cleanup points; the regression suite keeps its table at the end, as
AGENTS.md asks; and the two branches of the BE's FE loop got comments.
#### Results
- FE UT `FlightSqlQueryCancellationTest` (5 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
expire `execution_timeout + 5 s` after registration; resetting a channel
removes its routes; the session's next request (`beginRequest()`) and its
teardown (`releaseSession()`) remove them, without any cancellation RPC.
- `StmtExecutorTest#testCloudReplanDispatchProfileLifecycle` now 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.
- BE UT `ArrowFlightRemoteCancelTest` gets the FE half of branch-4.x's
fixture back: an in-process thrift FE (`FlightCancelCoordinator`) that answers
`cancelFlightQuery` by sending `cancel_plan_fragment` to an in-process BRPC
result BE. The 6 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.
`CancellationInterruptsPendingBrpcFetch` and
`FetchAfterCancellationStillCompletesRpc` are back to branch-4.x's form, which
needs the FE route (#69945 carried variants without it).
- 17 Flight session and coordinator FE test classes (177 tests) pass, with
Checkstyle. The 4 BE suites (`ArrowFlightCancelTest`,
`ArrowFlightRemoteCancelTest`, `ResultBufferMgrTest`,
`ArrowResultBlockBufferTest`, 28 tests) pass 10 times in a row in one ASAN
process.
- Mutations, each red in its target tests: on the FE, the channel keeps its
routes; a failed cancellation drops the route; a successful one keeps it;
routes never expire; `StmtExecutor` does not register. 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.
- On a local single-BE cluster,
`arrow_flight_sql_p0/test_flight_cancel_cleanup` (new; it aborts one endpoint
after the first batch, once by closing it and once by an explicit cancel, and
requires the query's pipelines to disappear from every BE within 20 s) passes,
and the BE log shows the `cancel_plan_fragment` that the FE sent through the
route for both aborts. The other 20 `arrow_flight_sql_p0` suites and the 14
`external_table_p0/remote_doris` suites pass, and no failed cancellation
attempt is logged. A single BE cannot show the cross-BE effect: there the BE
serving the stream already cancels the query itself (#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.
### 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. This needs both the FE and the BEs to be upgraded.
### Check List (For Author)
- Test: Unit Test / Regression test
- FE UT: `FlightSqlQueryCancellationTest` (5, new), `StmtExecutorTest`
(55) and 15 other Flight session / coordinator classes: 177 tests pass; fe-core
Checkstyle 0 violations
- BE UT: `ArrowFlightCancelTest` (9), `ArrowFlightRemoteCancelTest` (9,
6 new), `ResultBufferMgrTest` (6), `ArrowResultBlockBufferTest` (4): 28 tests
pass 10 times in a row (ASAN, macOS); 7 mutations (5 FE, 2 BE) each caught
- `build-support/check-build-hygiene.sh` passes; clang-format 16 clean;
clang-tidy reports nothing on the changed lines
- Regression: `arrow_flight_sql_p0` (21 suites, including the new
`test_flight_cancel_cleanup`) and `external_table_p0/remote_doris` (14 suites)
pass on a local single-BE cluster; the cross-BE path is only covered by the UTs
locally
- Behavior changed: Yes. A Flight result stream closed, cancelled, failed or
destroyed before EOF now cancels its query on every BE of the query, through
the FE that planned it.
- Does this need documentation: No
Original author: @Gabriel39
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
