chore: add CometGraceHashJoinSuite to CI, run prettier and cargo fmt

- Add CometGraceHashJoinSuite to the 'exec' suite list in both
  pr_build_linux.yml and pr_build_macos.yml.
- Update the Grace Hash Join design doc to list the new
  maxConcurrentPartitions config and to describe fastPathThreshold
  as an executor-wide budget (divided by spark.executor.cores).
- cargo fmt + prettier formatting.

Author: andygrove

.github/workflows/pr_build_linux.yml

@@ -354,6 +354,7 @@ jobs:
               org.apache.comet.exec.CometGenerateExecSuite
               org.apache.comet.exec.CometWindowExecSuite
               org.apache.comet.exec.CometJoinSuite
+              org.apache.comet.exec.CometGraceHashJoinSuite
               org.apache.comet.CometNativeSuite
               org.apache.comet.CometSparkSessionExtensionsSuite
               org.apache.spark.CometPluginsSuite

.github/workflows/pr_build_macos.yml

@@ -193,6 +193,7 @@ jobs:
               org.apache.comet.exec.CometGenerateExecSuite
               org.apache.comet.exec.CometWindowExecSuite
               org.apache.comet.exec.CometJoinSuite
+              org.apache.comet.exec.CometGraceHashJoinSuite
               org.apache.comet.CometNativeSuite
               org.apache.comet.CometSparkSessionExtensionsSuite
               org.apache.spark.CometPluginsSuite

docs/source/contributor-guide/grace-hash-join-design.md

@@ -29,12 +29,13 @@ Supports all join types: Inner, Left, Right, Full, LeftSemi, LeftAnti, LeftMark,
 
 ## Configuration
 
-| Config Key                                           | Type    | Default    | Description                                                |
-| ---------------------------------------------------- | ------- | ---------- | ---------------------------------------------------------- |
-| `spark.comet.exec.replaceSortMergeJoin`              | boolean | `false`    | Replace SortMergeJoin with ShuffledHashJoin (enables GHJ)  |
-| `spark.comet.exec.replaceSortMergeJoin.maxBuildSize` | long    | `104857600`  | Max build-side bytes for SMJ replacement. `-1` = no limit  |
-| `spark.comet.exec.graceHashJoin.numPartitions`       | int     | `16`         | Number of hash partitions (buckets)                        |
-| `spark.comet.exec.graceHashJoin.fastPathThreshold`   | int     | `67108864`   | Per-task fast-path budget in bytes                         |
+| Config Key                                               | Type    | Default     | Description                                                |
+| -------------------------------------------------------- | ------- | ----------- | ---------------------------------------------------------- |
+| `spark.comet.exec.replaceSortMergeJoin`                  | boolean | `false`     | Replace SortMergeJoin with ShuffledHashJoin (enables GHJ)  |
+| `spark.comet.exec.replaceSortMergeJoin.maxBuildSize`     | long    | `104857600` | Max build-side bytes for SMJ replacement. `-1` = no limit  |
+| `spark.comet.exec.graceHashJoin.numPartitions`           | int     | `16`        | Number of hash partitions (buckets)                        |
+| `spark.comet.exec.graceHashJoin.fastPathThreshold`       | long    | `67108864`  | Executor-wide fast-path budget in bytes (divided by cores) |
+| `spark.comet.exec.graceHashJoin.maxConcurrentPartitions` | int     | `2`         | Max partitions joined in parallel during Phase 3           |
 
 ### SMJ Replacement Guard
 

native/core/src/execution/operators/grace_hash_join/exec.rs

@@ -44,9 +44,9 @@ use tokio::sync::mpsc;
 
 use super::metrics::GraceHashJoinMetrics;
 use super::partition::{
-    buffer_build_optimistic, finish_spill_writers, merge_finished_partitions,
-    partition_build_side, partition_from_buffer, partition_probe_side, sub_partition_batches,
-    BuildBufferResult, FinishedPartition, HashPartition, ScratchSpace,
+    buffer_build_optimistic, finish_spill_writers, merge_finished_partitions, partition_build_side,
+    partition_from_buffer, partition_probe_side, sub_partition_batches, BuildBufferResult,
+    FinishedPartition, HashPartition, ScratchSpace,
 };
 use super::spill::{read_spilled_batches, SpillReaderExec, StreamSourceExec};
 use super::{JoinOnRef, DEFAULT_NUM_PARTITIONS, MAX_RECURSION_DEPTH, TARGET_PARTITION_BUILD_SIZE};

native/core/src/execution/operators/grace_hash_join/spill.rs

@@ -253,10 +253,8 @@ impl ExecutionPlan for SpillReaderExec {
                 ) {
                     Ok(r) => r,
                     Err(e) => {
-                        let _ = tx.blocking_send(Err(DataFusionError::ArrowError(
-                            Box::new(e),
-                            None,
-                        )));
+                        let _ =
+                            tx.blocking_send(Err(DataFusionError::ArrowError(Box::new(e), None)));
                         return;
                     }
                 };
@@ -265,10 +263,8 @@ impl ExecutionPlan for SpillReaderExec {
                     let batch = match batch_result {
                         Ok(b) => b,
                         Err(e) => {
-                            let _ = tx.blocking_send(Err(DataFusionError::ArrowError(
-                                Box::new(e),
-                                None,
-                            )));
+                            let _ = tx
+                                .blocking_send(Err(DataFusionError::ArrowError(Box::new(e), None)));
                             return;
                         }
                     };
@@ -394,9 +390,7 @@ impl ExecutionPlan for StreamSourceExec {
 // ---------------------------------------------------------------------------
 
 /// Read record batches from a finished spill file.
-pub(super) fn read_spilled_batches(
-    spill_file: &RefCountedTempFile,
-) -> DFResult<Vec<RecordBatch>> {
+pub(super) fn read_spilled_batches(spill_file: &RefCountedTempFile) -> DFResult<Vec<RecordBatch>> {
     let file = File::open(spill_file.path())
         .map_err(|e| DataFusionError::Execution(format!("Failed to open spill file: {e}")))?;
     let reader = BufReader::with_capacity(SPILL_IO_BUFFER_SIZE, file);

native/core/src/execution/planner.rs

@@ -1712,8 +1712,7 @@ impl PhysicalPlanner {
                     // concurrent tasks. Divide it by `spark.executor.cores` so each
                     // task's fast-path hash table stays within its fair share and N
                     // concurrent tasks don't collectively exceed the configured budget.
-                    let executor_cores =
-                        self.spark_conf.get_usize(SPARK_EXECUTOR_CORES, 1).max(1);
+                    let executor_cores = self.spark_conf.get_usize(SPARK_EXECUTOR_CORES, 1).max(1);
                     let total_fast_path_threshold = self
                         .spark_conf
                         .get_usize(COMET_GRACE_HASH_JOIN_FAST_PATH_THRESHOLD, 64 * 1024 * 1024);