Repository navigation
Conversation
andygrove
added a commit
to andygrove/datafusion-ballista
that referenced
this pull request
Oct 30, 2026
The user guide page is now apache#2713.
milenkovicm
reviewed
Oct 30, 2026
| or when building the session: | ||
|
|
||
| ```rust | ||
| let config = SessionConfig::new_with_ballista() |
Contributor
There was a problem hiding this comment.
Maybe we could add new_with_ballista_pipelined()?
Member
Author
There was a problem hiding this comment.
Good idea! I'd lean toward a typed setter rather than a new constructor, to match how the other settings are exposed on SessionConfigExt, something like SessionConfig::new_with_ballista().with_ballista_execution_mode(ExecutionMode::Pipelined). Since this PR is docs only, I filed #2729 for it and will update this snippet once it lands.
github-actions
Bot
added
the
documentation
milenkovicm
pushed a commit
to milenkovicm/datafusion-ballista
that referenced
this pull request
Oct 31, 2026
…front, streaming exchange) (apache#2657) * feat: let a standalone cluster run push-staged Add ballista.standalone.task_scheduling_policy (default pull-staged). With push-staged, the standalone scheduler pushes tasks and the standalone executor registers on a free gRPC port to receive them. The standalone scheduler now binds its listener before it is built and advertises its real address. Push-staged executors report task status to the endpoint each launch names, which was hardcoded to localhost:50050. * feat: add pipelined exchange operators Add PipelinedShuffleWriterExec, PipelinedShuffleReaderExec and the per-executor PipelinedRegistry that connects them, with their protobuf messages, serde, the FetchPipelinedPartition Flight action and the ballista.execution.mode and ballista.pipelined.* settings. Nothing plans these operators yet. The scheduler rejects jobs with ballista.execution.mode = pipelined, and executors answer pipelined fetches with unimplemented, until later changes wire them in. * refactor: simplify the pipelined exchange and reuse shared client helpers - Lease pooled connections for the life of a remote pipelined stream, and build unpooled clients with new_ballista_client, as the shuffle reader does. Encode Flight tickets with encode_protobuf. - Compute the writer's routing once, drop unused builder and getter methods, and let JoinSet's drop abort the remaining inputs. - Track unclaimed queues with a counter instead of rescanning every partition on each open, share one bounded set for closed jobs and stages, and stop cloning per batch in the queue stream. - Keep a broadcast reader whole when restricting partitions, forget a consumer's waiting slot when it times out, and document the minimums of the pipelined settings. * fix: share one connection per producer executor in the pipelined reader Each remote stream checked out its own connection, so a wide pipelined job opened one per producer and consumer task pair and overflowed the executors' listen queues. A reader partition now multiplexes its streams to an executor onto one connection, capped at 64 streams per connection. * fix: listen with a 4096 backlog on the executor's gRPC and Flight ports tonic's `serve` binds through the standard library, which listens with a backlog of 128. A pipelined job connects from every reader task on every executor when it starts, so an executor can get thousands of connections within a second and overflow both its accept and SYN queues. Overflowed connections wait for a SYN retransmit, and once the kernel falls back to SYN cookies, a client whose first segments were dropped is reset as soon as it sends more. The reader then sees `ConnectionReset` on a stream it never opened, with nothing logged on either side. Bind the Flight server through `create_grpc_server_incoming`, which the executor's gRPC server already uses, and have that helper listen with `GrpcServerConfig::listen_backlog`, 4096 by default. The kernel caps the value at `net.core.somaxconn`, which is 4096 on Linux 5.4 and later. * fix: retry a pipelined stream open whose connection failed before the server answered A wide pipelined job opens its remote streams in one burst, and a connection the producer's executor could not accept in time can be reset by the kernel as the reader starts sending. The reader saw that as a `ConnectionReset` on a stream it had not opened, and since pipelined jobs do not retry tasks, one such reset failed the job. `fetch_pipelined_partition` now reports a failure of the connection itself, before the server answered, as `GrpcConnectionError`, and keeps `GrpcActionError` for anything the server answered. The reader retries the former on a new connection, `io_retries_times` times with `io_retry_wait_time_ms` between attempts, the settings the blocking shuffle reader already uses. A failed connection is dropped from its slot, so the other streams of the partition that share it connect again rather than fail on it. An error the server answered, including the producer's own failure, is still final: the producer may already have handed the partition to the failed stream. Tests cover a stream open that is reset on its first connection and served on its second, and a server-side failure that is not retried. * feat: share one connection per executor pair for pipelined reads A pipelined reader opened a connection per reader partition to each producer executor, so a job opened executors x tasks x stage boundaries connections to every executor at once, enough to overflow its listen queue however large the queue is. The executor's client pool now keeps one shared connection per (host, port, config), which the streams of every task on the executor multiplex over, so an executor holds one connection to each other executor. A connection that fails before the server answers is discarded from the pool, and the open is retried on a new one. Sharing is safe because of how the windows are set. A stream nobody is reading holds up to its stream window of data on the connection, and if stalled streams could use up the connection window they would starve the streams something is waiting on. The shared connection therefore advertises the largest connection window HTTP/2 allows and a small stream window, `ballista.pipelined.stream_window_bytes` (512 KiB by default), so stalled streams hold at most one window each and a connection carries over 4000 streams before any could starve the rest. The executor's server is explicitly left without a limit on concurrent streams, since a stream the client had to queue behind a limit would leave its producer blocked on a full queue. `BallistaClientPool` gains a `shared` method with a default implementation, so existing pools keep compiling. * fix: close unclaimed pipelined queues on cleanup and simplify the exchange Address review feedback on the pipelined exchange operators. Fixes: - `PipelinedRegistry::remove_job` and `remove_stage` now close the queues no consumer has claimed. The writer keeps its `ProducerOutputs` alive for the life of the task, so dropping the registry's reference alone left a producer blocked on a full queue until the claim timeout. The test now holds the outputs the way the writer does. - A producer stays registered until its stage or job is removed, instead of being forgotten once every queue is claimed. A second open of a queue now fails at once with "already being read" rather than waiting out the registration timeout for a producer that will never register. - A pipelined stream open is retried only when the connection failed before the server answered. Once the response headers have arrived the executor may have handed the partition to that stream, so a failure on the first message is reported as a server failure and not retried. - `is_transport_failure` relies on the status's source chain alone. A reset arrives as `Cancelled` and a refused connection as `Unavailable`, both with the transport error as source; the string match on "transport error" and the bare `Unavailable` check were redundant. Simplifications: - `ProducerKey` is replaced by the existing `TaskKey`, which gains a `Display` impl. - The writer threads one `Outputs` struct through its input tasks in place of an `Arc<HashMap>` and an `Arc<AtomicUsize>`; outputs are dense slots, so hash routing indexes them directly, and the hash coalescers live beside them rather than as an `Option` on every output. `Routing::Hash` holds the `Partitioning` instead of rebuilding it per input. `finish_outputs` becomes `Outputs::finish`. - `open_remote` reuses the blocking reader's `with_retry`. - `BallistaClient::do_get_stream` holds the `do_get` and schema-read sequence that `execute_do_get` and `fetch_pipelined_partition` both repeated. - `serialize_hash_partitioning` serves both the sort shuffle writer and the pipelined writer. - The producer's failure is stored once, on `ProducerOutputs`; a broadcast partition records only whether it finished or failed. - `try_new_broadcast` becomes `new_broadcast`, since it cannot fail. `ProducerOutputs::schema` and `producer_count` are no longer public. - Reader metrics are `local_partitions` and `remote_partitions`, and the writer's send time is `write_time`, matching the blocking operators. - `DefaultBallistaClientPool::shared` returns an open connection without taking the connecting lock. - The `io-util` tokio feature moves to dev-dependencies. - Registry tests share a `queue` helper, and the remote-read tests keep their listeners open instead of reusing a port a dropped listener had. Docs note where the claim timeout is measured from and the plan shapes that relies on, that `max_concurrent_streams(None)` is tonic's default, and that the stream window also bounds a stream's throughput per round trip. * refactor: simplify pipelined registry, writer outputs and shared connections - Mark a shared client discarded with a flag; the pool skips discarded connections and evicts their slots, so no discard closure or generation counter is needed - Hold a producer's registry slot and broadcast replay state in tokio::sync::watch channels instead of waiter lists and Notify - Store the producer failure in a OnceLock - Remember closed jobs and stages in one bounded map keyed by job - Keep each output's coalescer inside the output and derive whether all consumers are gone instead of counting - Take the registry from the session config only, dropping with_registry from the pipelined writer and reader * feat: run pipelined jobs end to end Wire the pipelined exchange operators into the executor, scheduler, client and Flight SQL, and remove the guards that rejected pipelined jobs until now. - Executor: own a PipelinedRegistry, serve FetchPipelinedPartition from do_get, run PipelinedShuffleWriterExec stage roots, share one memory pool among a job's tasks, and drop a job's registry entries on cleanup. - Scheduler: plan pipelined jobs as a PipelinedExecutionGraph and place all of their tasks at once, reserving the widest stage's vcores on each executor. Choose join strategies by estimated build size. - Client: stream every final-stage partition while the job runs, cancel the job when its results are dropped, and report its failure. - Flight SQL: serve a pipelined result through one merged endpoint. - Standalone: run push-staged when the session asks for pipelined execution. - Tests, a TPC-H SF10 pipelined CI leg, and a user guide page. * fix: keep the standalone executor's gRPC port bound until it serves Pass the bound listener to a new crate-private executor_server::startup_with_incoming instead of releasing the port and binding it again, which let another port-0 bind in the process take it. Correct the keep-alive comment about the stop channel, and make the two-executor test check the scheduler rejects poll_work, so it fails if the push-staged setting is ignored. * docs: move the pipelined execution page to its own PR The user guide page is now apache#2713. * docs: restore the toctree without the pipelined page * fix: build caches blocking from pipelined sessions and drop lost cache data A pipelined job streams its output and leaves no shuffle files, so a cache built from a pipelined session recorded an empty entry. Cache builds now run blocking whatever the session's mode; reads keep it. A pipelined job that loses cache data, to a lost executor or a failed fetch from a live one, now fails with CACHE_DATA_LOST so the build is dropped and a retry rebuilds it. * fix: release pipelined reservations on executor loss and failed saves Losing an executor now drops its share of a pipelined job's reservation, so an executor that registers again under the same id does not get those vcores on top of its fresh capacity. succeed_job now releases a pipelined job's reservation before persisting its terminal state, as abort_job does, so a failed save cannot strand the vcores. Also drop the configs.md link to the pipelined execution guide, which lands in a later PR. * fix: import ExecutionPlan in execution_graph tests
andygrove
force-pushed
the
pipelined-docs
branch
from
October 31, 2026 22:54
239e3c4 to
227598d
Compare
andygrove
marked this pull request as ready for review
October 31, 2026 22:54
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
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.
Which issue does this PR close?
Part of #2652. This doesn't close it.
Rationale for this change
Pipelined execution (#2652) needs a user guide page explaining when to use the mode, how a job runs, how much memory it takes and which settings control it. The page was part of #2657, which turns the mode on. It was split out so the docs could be reviewed separately from the scheduler code. #2657 has now merged, so the mode is available on
main.What changes are included in this PR?
docs/source/user-guide/pipelined-execution.md. It covers:docs/source/index.rst, after the tuning guide.configs.md).Are there any user-facing changes?
Yes, a new documentation page. There are no code changes.