andygrove opened a new pull request, #2695: URL: https://github.com/apache/datafusion-ballista/pull/2695
# Which issue does this PR close? Part of #2652. # Rationale for this change #2657 adds an opt-in pipelined execution mode and is too large to review in one piece, so it is being split into smaller PRs. #2670 added the pipelined exchange operators (`PipelinedShuffleWriterExec`, `PipelinedShuffleReaderExec`, `PipelinedRegistry`) and #2685 added join selection for pipelined jobs. This PR adds the executor side: running pipelined tasks and serving their output to consumers over Arrow Flight. Nothing here runs yet. The scheduler still rejects pipelined jobs in `submit_job`, and #2657 removes that guard. # What changes are included in this PR? - **Registry.** Each `Executor` owns a `PipelinedRegistry` and attaches it to every task's session config, which is how a task's pipelined writer and readers find it. Job cleanup, both poll-based and push-based, drops the job's registry entries, or just the listed stages. - **Flight service.** `BallistaFlightService::with_pipelined_registry` serves `FetchPipelinedPartition` tickets from the registry. The response opens right away and waits for the producer inside the stream, so a consumer that arrives before its producer is not timed out by its own request deadline. Dictionary columns stay dictionary-encoded so the stream matches the schema the reader declares. A service without a registry still refuses these tickets with `Unimplemented`. Both the executor process and the standalone executor wire in the registry. - **Task execution.** `DefaultExecutionEngine` recognizes `PipelinedShuffleWriterExec` at the root of a task plan and runs it through a new `PipelinedQueryStageExec`. It also gives `PipelinedShuffleReaderExec` the executor's Flight client pool. - **Memory.** All of a pipelined job's tasks on one executor run at the same time on one vcore reservation. They share one runtime, and so one memory pool, sized by that reservation (`Executor::produce_runtime_for_pipelined_task`). Separate per-task pools would add up to more than the executor's memory. The runtime is held weakly and goes away when the job's last task on the executor finishes. Tests cover streaming a partition through `do_get` when the fetch arrives before the producer starts, keeping dictionary columns dictionary-encoded, refusing fetches when there is no registry, and sharing the runtime across a job's tasks. # Are there any user-facing changes? No. The changes are inert until the scheduler accepts pipelined jobs. The public API only gains new items: `PipelinedQueryStageExec`, `BallistaFlightService::with_pipelined_registry`, and the `Executor` methods `pipelined_registry`, `with_pipelined_registry`, `remove_pipelined_job_data` and `produce_runtime_for_pipelined_task`. -- 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]
