[EPIC] Spilling coverage gaps vs Spark and DuckDB: window, cross join, and pool-driven reclaim

[comphead]

[EPIC] Spilling coverage gaps

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 feat: sort-merge fallback for partitioned hash joins under memory pressure #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 [EPIC] Spilling Hash Join — run any join in a bounded memory budget #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.

• Support spilling for WindowAggExec #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
• [EPIC] Spilling Hash Join — run any join in a bounded memory budget #24768 [EPIC] Spilling Hash Join (in progress, tracked separately)
• Support spilling in TopK queries #15538 Support spilling in TopK queries
• Enable external operator reclaim / spill hooks for external memory managers #21422 Enable external operator reclaim / spill hooks for external memory managers — the pool-driven-callback direction
• Unify the spilling aggregate streams behind one spill-replay driver #25537 Unify the spilling aggregate streams behind one spill-replay driver
• Fix Memory Backpressure/Spilling Coordination #20715 Fix Memory Backpressure/Spilling Coordination
• FairSpillPool penalises the active operator in a pipeline of blocking spillable operators #22036 FairSpillPool penalises the active operator in a pipeline of blocking spillable operators
• Clarify memory-limit guarantees for external-sort spilling #25183 Clarify memory-limit guarantees for external-sort spilling

Related epics

Existing spilling epics, for context. This ticket does not replace any of them.

• [EPIC] Spilling Hash Join — run any join in a bounded memory budget #24768 [EPIC] Spilling Hash Join — run any join in a bounded memory budget (active)
• [EPIC] Additional improvements to larger than memory / spilling sorts #17593 [EPIC] Additional improvements to larger than memory / spilling sorts
• [EPIC] Improved Externalized / Spilling / Large than Memory Hash Aggregation #13123 [EPIC] Improved Externalized / Spilling / Large than Memory Hash Aggregation
• [Epic]: Google Summer of Code 2025 Improving Spilling Execution #16065 [Epic] Google Summer of Code 2025 Improving Spilling Execution
• [EPIC] A collection of tickets for improving sorting larger than memory datasets / spilling sorts #15271 [EPIC] Improving sorting larger than memory datasets (closed, superseded by [EPIC] Additional improvements to larger than memory / spilling sorts #17593)
• [EPIC] Use blocked / chunked memory management in hash aggregation #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.