Dandandan opened a new issue, #23303: URL: https://github.com/apache/datafusion/issues/23303
### Is your feature request related to a problem or challenge? DataFusion's physical plans are written and tested assuming everything runs in one process: one `Arc<dyn ExecutionPlan>` instance, state shared via `Arc`, and all partitions of an operator polled together by cooperative threads on a shared runtime. Distributed engines built on DataFusion (Ballista, [datafusion-distributed](https://github.com/datafusion-contrib/datafusion-distributed)) violate every one of those assumptions: - the plan is cut into **stages** at shuffle boundaries, and each stage is **serialized** (both projects use `datafusion-proto`) and shipped elsewhere; - each task runs **one partition** of a stage, on its **own decoded plan instance**, with **no shared execution state** with sibling tasks; - data crosses stage boundaries as **serialized batches over a transport**, never passed by reference. There is currently no test surface in DataFusion that exercises these assumptions. A change that quietly depends on the in-process model compiles, passes CI, and only breaks once it reaches a downstream engine — usually discovered much later. ### Describe the solution you'd like Add a small, opt-in, in-tree crate that models the distributed execution style **in-process** — no network, no gRPC, no separate processes — serving two purposes: 1. A **reference implementation / example** of how distributed execution works on top of DataFusion: breaking a query into stages, serializing plans, and executing tasks in isolation. Much smaller to read than a full engine. 2. A **regression guard**: once a subset of its tests runs in CI, it catches the class of change that breaks isolated execution before it ships to downstream projects. The core pieces (as prototyped in #23282): - **Stage splitting**: walk the physical-optimized plan and cut a new stage at each `RepartitionExec(Hash)`, `CoalescePartitionsExec`, and `SortPreservingMergeExec`, producing a `StageGraph`. - **Serialization + isolation** (the core invariant): each stage plan is encoded once via `datafusion-proto`, and every task decodes its own fresh plan instance against its own `SessionContext`/`TaskContext` — no shared `Arc` execution state between tasks. - **Streaming in-memory exchange**: `ExchangeSinkExec` hash-partitions and pushes Arrow-IPC-encoded frames into bounded channels; `ExchangeSourceExec` merges and decodes them (the in-process analogue of a network endpoint). - **Executor**: spawn-all, no-barrier — tasks are wired together by bounded channels (backpressure, not a barrier), mirroring datafusion-distributed's pipelined streaming. - **Differential harness**: `run_distributed(ctx, plan, config)` plus `assert_distributed_eq`, asserting the distributed result equals single-node `collect(plan)`. The crate would be excluded from the workspace so it has zero impact on normal builds and published crates; it builds only when explicitly targeted. This approach already caught a real regression on its first run: a 4-file CSV scan returning 4× the correct aggregate under isolated per-partition execution, traced to file work being shared across a plan instance's partitions and drained correctly only when all partitions are polled together. ### Describe alternatives you've considered - **Rely on downstream projects' CI** (Ballista, datafusion-distributed): this is the status quo — regressions are found long after merge, and the feedback loop crosses repo boundaries. - **Build the tests directly against an existing distributed engine**: brings in transport dependencies (gRPC, Arrow Flight) and a much larger surface than needed to exercise the stage/serialize/isolate contract. - **Not doing this**: distributed engines continue to be broken by in-process assumptions with no early signal. ### Additional context - Draft PR with a working prototype: #23282 - Follow-up: run a subset of sqllogictests through the distributed harness for much broader differential coverage (separate issue). -- 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]
