Add support for spilling in hash joins #2545

Activity

1. andygrove

   added this to the 1.0.0 milestone on Oct 10, 2025
2. andygrove

   added

   enhancementNew feature or requestNew feature or request

   on Oct 10, 2025
3. andygrove

   removed this from the 1.0.0 milestone on Apr 18, 2026
4. dwsmith1983 commented on Sep 28, 2026

   

   dwsmith1983

   on Sep 28, 2026

   Contributor

   More actions

   I'll take the Comet side of this. The operator work is upstream now: jayzhan211's epic (apache/datafusion#24768), with a sort-merge fallback under memory pressure as its first step (apache/datafusion#25217), and #6309 points the roadmap there, so nothing here should duplicate it.

   Two Comet pieces:

   1. A size guard in RewriteJoin. Today spark.comet.exec.forceShuffledHashJoin rewrites every eligible sort-merge join, and getOptimalBuildSide reads sizeInBytes only to pick the build side. Rewriting only when the build side's estimate is under a threshold, the way Spark's JoinSelection gates shuffled hash join with canBuildLocalHashMapBySize, keeps the hash join for the joins that fit, which is where the TPC-H gain is, and leaves the rest on sort-merge join until spilling lands. Small change, no native code.

   2. Once feat: sort-merge fallback for partitioned hash joins under memory pressure datafusion#25217 lands and Comet takes that DataFusion release: the Comet side of the fallback (its metrics, the fair_unified interaction, and a memory-limited join matrix on TPC-H and TPC-DS like feat(benchmarks): Record the join failure matrix under a fixed memory budget datafusion#24779), so the hash join can go on by default.

   @andygrove is that the split you want, or should Comet wait for the fallback?
5. dwsmith1983 commented on Sep 29, 2026

   

   dwsmith1983

   on Sep 29, 2026

   Contributor

   More actions

   The first piece, the size guard in RewriteJoin, is up as #6384. With spark.comet.exec.forceShuffledHashJoin on, a sort-merge join is rewritten only when the build side's estimate is under a limit, by default Spark's own canBuildLocalHashMapBySize rule (autoBroadcastJoinThreshold times the shuffle partition count). Larger joins keep the sort-merge join, still native, with the reason in the plan info. spark.comet.exec.forceShuffledHashJoin.maxBuildSize sets a fixed limit or turns it off, and the two hash join benchmark configs set it to -1 so they keep measuring every join.

   The second piece waits for apache/datafusion#25217.