andygrove opened a new pull request, #2670:
URL: https://github.com/apache/datafusion-ballista/pull/2670
# Which issue does this PR close?
Part of #2652. This is the second of a series of smaller PRs split out of
#2657 (pipelined execution). It is independent of #2669.
# Rationale for this change
In pipelined execution, every task of a job runs at once and stages stream
record batches to each other instead of exchanging shuffle files. This PR adds
the exchange itself: the operators at either end of a pipelined stage boundary,
and the per-executor registry that connects them. Nothing plans these operators
yet, so this PR can't change how existing jobs run.
#2657 describes the whole feature, its design notes and SF1000 results.
# What changes are included in this PR?
- **`execution_plans/pipelined/`**
- `PipelinedRegistry` holds a bounded queue for each output partition of
each running producer task, keyed by job, stage and task.
- Broadcast outputs, which every consumer task reads in full, are kept
in replay buffers instead.
- It remembers recently removed jobs and stages, so a leftover task
fails at once instead of waiting out its timeout.
- `PipelinedShuffleWriterExec` is a stage root. It drives every input
partition and routes batches into the queues: hash-partitioned, passthrough or
broadcast.
- Each queue ends with an explicit end marker, so a consumer can tell a
finished partition from a failed producer.
- The writer stops reading its input once every consumer has gone away.
- It fails if no consumer opens a full queue within
`ballista.pipelined.registration_timeout_seconds`.
- `PipelinedShuffleReaderExec` is a stage leaf. It drains all of its
producers concurrently: directly from the registry when the producer is on the
same executor, and over Flight otherwise. Reading concurrently is what keeps
bounded queues free of deadlock for tree-shaped plans.
- `tests.rs` runs writers against readers in-process.
- **Protobuf and serde:** `PipelinedShuffleWriterExecNode`,
`PipelinedShuffleReaderExecNode`, `PipelinedPartitionLocation`, and the
`FetchPipelinedPartition` action, plus
`BallistaClient::fetch_pipelined_partition`. The Flight proxy in `serving.rs`
forwards the new action.
- **Settings:** `ballista.execution.mode` (`blocking` or `pipelined`),
`ballista.pipelined.queue_capacity` and
`ballista.pipelined.registration_timeout_seconds`. They are documented in a new
section of the config reference.
- **Guards until the rest lands:**
- The scheduler rejects a job with `ballista.execution.mode = pipelined`
("not supported yet").
- The executor's Flight service answers `FetchPipelinedPartition` with
`unimplemented`.
- Later PRs in the series replace both guards.
# How are these changes tested?
- Unit tests for the registry, covering open and claim, timeouts, replay,
cancellation and removed jobs.
- Unit tests for the writer and reader, run against each other in-process,
covering hash, passthrough and broadcast outputs, end markers, producer
failure, and early consumer drop.
- Serde round trips and config parsing.
- A scheduler test checks that a pipelined job is rejected and no graph is
created.
# Are there any user-facing changes?
There are three new session settings, documented in the config reference.
Setting `ballista.execution.mode = pipelined` currently makes the job fail with
a "not supported yet" error. The default behavior is unchanged.
--
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]
