andygrove opened a new pull request, #2696:
URL: https://github.com/apache/datafusion-ballista/pull/2696
# Which issue does this PR close?
Part of #2652.
# Rationale for this change
#2657 adds an opt-in pipelined execution mode and is being split into
smaller PRs. #2670 added the exchange operators, #2685 added join selection,
and #2695 lets executors serve pipelined partitions over Arrow Flight. This PR
covers the reading side: the client and the Flight SQL frontend.
A pipelined job's final stage streams its output while the job runs, and it
can't finish until that output has been read. So a client can't wait for the
job to succeed and then fetch partitions one by one as it does for a blocking
job. It has to start reading once the job is running, and read every partition
at the same time.
Nothing here runs yet. The scheduler still rejects pipelined jobs and never
fills in the new fields. #2657 removes that guard and reports the locations.
# What changes are included in this PR?
- **Proto.** `RunningJob` and `SuccessfulJob` gain
`pipelined_output_locations`, which lists where the final stage serves each
output partition. `SuccessfulJob` carries them too, so a client that polls can
still read a small result that finished before it saw the job running. The
blocking execution graphs set them to empty.
- **Client** (`distributed_query.rs`). When a pull-mode or push-mode status
reports pipelined locations, the client streams every final-stage partition at
once. It opens one connection per executor (or proxy) and multiplexes that
executor's partitions over it. The result stream also:
- watches the job's status, and ends with the job's own error if the job
fails. A fetch error waits a short grace period for that error, because a
failed job's cleanup usually breaks the fetch first, with a less useful message.
- cancels the job if it is dropped before it ends, which frees the job's
vcores at once.
- waits for the final stage to start for as long as
`ballista.pipelined.registration_timeout_seconds` allows, read from the
submitted settings.
- **Flight SQL frontend.** `QueryResult` gains `pipelined_partitions` and
`pipelined_registration_timeout`. The scheduler's backend returns as soon as a
pipelined job is running. The service serves all of the job's partitions
through a single endpoint, using a new `StatementHandle::Pipelined` ticket.
`DoGet` opens every partition before reading any, so a client that reads
endpoints one at a time can't stall the job. Dictionary columns stay
dictionary-encoded to match the advertised schema.
Unit tests cover the result stream (passing data through, reporting the
job's failure in preference to a fetch error, falling back to the fetch error,
cancelling on an early drop), reading the timeout from the settings, the ticket
encoding and its truncation checks, and the frontend handing out one endpoint
with a fetch action per partition. End-to-end tests through a real cluster stay
in #2657, because they need the scheduler.
# Are there any user-facing changes?
There's no behavior change until the scheduler runs pipelined jobs.
This PR does make small breaking changes to the public API, so it has the
`api-change` label:
- `ballista_flight_sql::backend::QueryResult` has two new public fields. A
custom `QueryBackend` that builds a `QueryResult` must set
`pipelined_partitions: vec![]` and `pipelined_registration_timeout:
Duration::default()`.
- The generated `RunningJob` and `SuccessfulJob` protobuf structs have a new
`pipelined_output_locations` field. Code that builds them with struct literals
must set it, or use `..Default::default()`.
--
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]
