feat: Add experimental Grace Hash Join operator with spill-to-disk support [research, will not merge]

[andygrove]

Summary

Add a Grace Hash Join (GHJ) operator that replaces Spark's ShuffledHashJoinExec with a spill-capable native implementation. Instead of loading the entire build side into a single hash table (which OOMs on large builds), GHJ hash-partitions both sides into N buckets and joins them independently, spilling to disk under memory pressure.

This is an experimental feature, disabled by default (spark.comet.exec.graceHashJoin.enabled=false).

How it works

Phase 1 — Build partitioning: Hash-partition build input into 16 buckets using a prefix-sum algorithm. When memory is tight, spill the largest partition to disk (Arrow IPC with LZ4 compression).

Phase 2 — Probe partitioning: Same hash-partitioning on probe input. On first memory pressure, aggressively spill ALL non-spilled partitions (both sides) to prevent thrashing with concurrent operators sharing the memory pool.

Phase 3 — Per-partition joins: Join each partition sequentially via HashJoinExec. Adjacent partitions are merged when builds are small (target 32 MB per group) to reduce overhead. Oversized build partitions are recursively repartitioned (up to 3 levels = 4096 effective partitions). A fast path streams probe data directly for tiny builds (< 10 MB / executor cores).

Key design decisions

• All join types supported: Inner, Left, Right, Full, LeftSemi, LeftAnti, LeftMark, RightSemi, RightAnti, RightMark
• Aggressive spill strategy: On first probe-side memory pressure, spill ALL partitions atomically rather than one-at-a-time, avoiding repeated spill/grow cycles with the fair memory pool
• Spilled probe streaming: Probe data streams from spill files via SpillReaderExec instead of loading into memory, keeping peak memory at ~1/N of the full dataset
• Partition merging: After partitioning, adjacent small partitions are merged to reduce the number of HashJoinExec calls (e.g., 16 partitions with 48 MB total build → 2 merged groups)
• Batch splitting fix: DataFusion's DataSourceExec wraps output with BatchSplitStream, which slices large batches into 8192-row chunks. Arrow's zero-copy slicing shares buffers, causing get_record_batch_memory_size() to report the full buffer size per slice. This inflated memory accounting 85x in collect_left_input, causing phantom OOM. Fixed by setting batch_size to the actual row count in the TaskContext passed to Phase 3 joins.

Configuration

Config	Default	Description
spark.comet.exec.graceHashJoin.enabled	false	Enable Grace Hash Join
spark.comet.exec.graceHashJoin.numPartitions	16	Number of hash partitions
spark.comet.exec.graceHashJoin.fastPathThreshold	10485760 (10 MB)	Memory budget for fast-path hash tables (divided by executor cores). Set to 0 to disable.

Files changed

• native/core/src/execution/operators/grace_hash_join.rs — GHJ operator implementation (~2600 lines)
• native/core/src/execution/planner.rs — Wire GHJ into the physical planner
• common/.../CometConf.scala — Configuration entries
• spark/.../operators.scala — CometGraceHashJoinExec Spark operator and metrics
• spark/.../CometMetricNode.scala — Metric mapping for GHJ
• docs/.../grace-hash-join-design.md — Design document

[andygrove]

GHJ is slightly slower than HJ currently.

[andygrove]

q72 completes without error, but performance is 💩 (755 seconds vs Spark's 90 seconds)

[andygrove]

q72 is now 19 seconds! 🍾

[andygrove]

I have to give Claude Code a shoutout for tracking down this bug. It took many iterations on debugging.

  Root cause found and fixed: DataFusion's DataSourceExec always wraps output with BatchSplitStream, which slices batches larger than batch_size (default 8192 rows). In GHJ Phase 3:                     
                                                                                                                                                                                                          
  1. A build batch with 696,344 rows / 22 MB is passed through DataSourceExec
  2. BatchSplitStream slices it into 696,344 / 8192 = 85 slices
  3. Each slice shares the original Arrow buffers via zero-copy batch.slice()
  4. get_record_batch_memory_size() reports the full buffer size (~22 MB) for each slice
  5. collect_left_input calls try_grow(22 MB) 85 times → 1.87 GB phantom reservation
  6. The actual memory is only ~22 MB → 85x over-counting → spurious OOM

  Fix: Created context_without_batch_splitting() that produces a TaskContext with batch_size = usize::MAX, preventing BatchSplitStream from slicing. Applied to all 3 Phase 3 code paths:
  - Fast path (join_partition_recursive via fast path)
  - Recursive path (join_partition_recursive)
  - Spilled probe path (join_with_spilled_probe)

  You can verify by running TPC-DS q72 again. The build batch will now pass through as a single batch, and collect_left_input will correctly account for ~22 MB (1 try_grow) instead of 1.87 GB (85
  try_grows).