comphead opened a new issue, #25637:
URL: https://github.com/apache/datafusion/issues/25637
# [EPIC] Spilling coverage gaps vs Spark and DuckDB: window, cross join, and
pool-driven reclaim
**TL;DR.** An operator-by-operator audit of DataFusion 55.1.0 against Spark
4.x and DuckDB `main`. DataFusion has good coverage of sort, aggregate,
sort-merge join, nested loop join and repartition. Three gaps remain where at
least one other engine does better: **window functions**, **cross join**, and
**hash join** (already covered by #24768). Underneath them is a structural
difference in *where* each engine implements spilling, which is why the gaps
cluster the way they do.
This epic collects the gaps and links the existing work rather than
duplicating it.
## Where DataFusion spills today
Ground truth is the set of `MemoryConsumer::with_can_spill(true)`
registrations plus `SpillManager` users in `datafusion/physical-plan/src`.
| Operator | Spills | Reference |
| --- | --- | --- |
| `SortExec` / `ExternalSorter` | ✅ | `sorts/sort.rs` |
| `SortPreservingMergeExec` | ✅ | `sorts/multi_level_merge.rs` |
| `AggregateExec`, all four modes | ✅ |
`aggregates/{hash,single,ordered_*,grouped_hash,partial_reduce}_stream.rs` |
| `SortMergeJoinExec` | ✅ | `joins/sort_merge_join/exec.rs` |
| `NestedLoopJoinExec` | ✅ (since 54) | `joins/nested_loop_join.rs` |
| `RepartitionExec` | ✅ | `repartition/mod.rs` |
| `HashJoinExec` | ❌ | `joins/hash_join/exec.rs`, `collect_left_input` calls
`try_grow` and propagates |
| `WindowAggExec` / `BoundedWindowAggExec` | ❌ | no spill code |
| `CrossJoinExec` | ❌ | no spill code |
| `TopK` / `grouped_topk_stream` | ❌ | bounded by `k`, but a large `k` still
fails |
| `SymmetricHashJoinExec`, `UnnestExec`, `RecursiveQueryExec` | ❌ | no spill
code |
## How Spark and DuckDB compare
| | DataFusion 55 | Spark 4.x | DuckDB `main` |
| --- | --- | --- | --- |
| Sort | ✅ | ✅ | ✅ |
| Hash aggregate | ✅ all modes | ✅ sort-based fallback | ✅ radix partitioned
|
| Sort-merge join | ✅ | ✅ | ✅ |
| **Hash join** | ❌ query fails | ❌ but planner avoids it | ✅ hybrid
external |
| **Window** | ❌ | ✅ | ✅ |
| **Cross join** | ❌ | ✅ | passive |
| Nested loop join | ✅ | ❌ broadcast only | passive |
| Shuffle / repartition | ✅ | ✅ | n/a |
| Top-N | ❌ | bounded by `k` | bounded by `k` |
Evidence for the two comparison columns:
- **Spark window** spills through `ExternalAppendOnlyUnsafeRowArray`, tuned
by `spark.sql.windowExec.buffer.spill.threshold` and
`spark.sql.windowExec.buffer.spill.size.threshold`. Same array backs session
windows and pandas-UDF windows.
- **Spark cross join** spills through the same array, tuned by
`spark.sql.cartesianProductExec.buffer.spill.threshold`.
- **Spark hash join does not spill either.** `HashedRelation.spill()` is
hardcoded to return `0L`, and `ShuffledHashJoinExec.scala` contains no spill
path. Spark avoids the problem in the planner instead:
`spark.sql.join.preferSortMergeJoin` defaults to `true`. That is essentially
the approach in #25217.
- **DuckDB hash join** has an explicit external mode with `ProbeSpill` and a
`TemporaryMemoryState` reservation
(`src/execution/operator/join/physical_hash_join.cpp`). That is essentially the
approach described in #24768.
- **DuckDB window** is out-of-core via `ColumnDataCollection` plus
`src/common/sort/hashed_sort.cpp`. DuckDB documents larger-than-memory support
for all four blocking operators (`GROUP BY`, `JOIN`, `ORDER BY`, `OVER`).
## Why the gaps cluster where they do
The three engines implement spilling at different layers.
- **DuckDB spills at the storage layer.** `ColumnDataAllocator` and
`TupleDataAllocator` allocate through the `BufferManager`, so any operator
buffering into a `ColumnDataCollection` or `TupleDataCollection` gets eviction
to the temp directory for free. Operator-level work such as the external hash
join exists to bound the *working set*, not to make spilling possible. This is
why DuckDB's nested loop join, cross product and IEJoin degrade instead of
failing despite having no external algorithm of their own.
- **Spark spills at the task memory manager layer.** Every `MemoryConsumer`
exposes a `spill()` callback that `TaskMemoryManager` invokes under pressure.
Generic mechanism, opt-in per data structure, and two important structures
decline (`BytesToBytesMap`, `HashedRelation`).
- **DataFusion spills per operator.** `with_can_spill(true)` is a flag the
pool reads, not a callback it can invoke. There is no `spill()` for
`MemoryPool` to call. Each operator polls `try_grow`, catches
`ResourcesExhausted`, and drives `SpillManager` itself.
Consequence: DataFusion's supported list is exactly the set of operators
someone wrote spill code for, an unconverted operator fails hard rather than
degrading, and every new operator repeats the same `try_grow` / catch / spill /
replay pattern. #25537 is a step toward factoring that out for aggregates.
#21422 is the closest existing discussion of a reclaim hook, raised from the
Comet side because Spark's `TaskMemoryManager` has no way to ask a DataFusion
operator to spill.
## Tasks
Highest value first.
- [ ] #22946 Support spilling for `WindowAggExec` — the clearest gap, both
other engines handle it
- [ ] Support spilling for `CrossJoinExec` — no issue yet, will file if
there is interest
- [ ] #24768 `[EPIC]` Spilling Hash Join (in progress, tracked separately)
- [ ] #15538 Support spilling in `TopK` queries
- [ ] #21422 Enable external operator reclaim / spill hooks for external
memory managers — the pool-driven-callback direction
- [ ] #25537 Unify the spilling aggregate streams behind one spill-replay
driver
- [ ] #20715 Fix Memory Backpressure/Spilling Coordination
- [ ] #22036 `FairSpillPool` penalises the active operator in a pipeline of
blocking spillable operators
- [ ] #25183 Clarify memory-limit guarantees for external-sort spilling
## Related epics
Existing spilling epics, for context. This ticket does not replace any of
them.
- #24768 `[EPIC]` Spilling Hash Join — run any join in a bounded memory
budget (active)
- #17593 `[EPIC]` Additional improvements to larger than memory / spilling
sorts
- #13123 `[EPIC]` Improved Externalized / Spilling / Large than Memory Hash
Aggregation
- #16065 `[Epic]` Google Summer of Code 2025 Improving Spilling Execution
- #15271 `[EPIC]` Improving sorting larger than memory datasets (closed,
superseded by #17593)
- #24704 `[EPIC]` Use blocked / chunked memory management in hash
aggregation (adjacent)
## Notes
Versions audited: DataFusion 55.1.0, `apache/spark@master`,
`duckdb/duckdb@main` with the `preview` docs. The "passive" entries for
DuckDB's nested loop join, cross product and IEJoin are inferred from the
allocator type rather than from an explicit external code path in those
operators.
--
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]
