fix: improve grace hash join config defaults

- Change fastPathThreshold to per-task (64 MB) instead of per-executor
  divided by cores (was 10 MB total)
- Change maxBuildSize default from -1 (no limit) to 100 MB to keep SMJ
  for large build sides where streaming merge outperforms hash join
- Remove benchmark config overrides that are now covered by defaults

Author: andygrove

benchmarks/tpc/engines/comet-hashjoin.toml

@@ -32,6 +32,4 @@ driver_class_path = ["$COMET_JAR"]
 "spark.shuffle.manager" = "org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager"
 "spark.comet.scan.impl" = "native_datafusion"
 "spark.comet.exec.replaceSortMergeJoin" = "true"
-"spark.comet.exec.replaceSortMergeJoin.maxBuildSize" = "104857600"
-"spark.comet.exec.graceHashJoin.fastPathThreshold" = "34359738368"
 "spark.comet.expression.Cast.allowIncompatible" = "true"

common/src/main/scala/org/apache/comet/CometConf.scala

@@ -318,15 +318,14 @@ object CometConf extends ShimCometConf {
     conf(s"$COMET_EXEC_CONFIG_PREFIX.graceHashJoin.fastPathThreshold")
       .category(CATEGORY_EXEC)
       .doc(
-        "Total memory budget in bytes for Grace Hash Join fast-path hash tables across " +
-          "all concurrent tasks. This is divided by spark.executor.cores to get the per-task " +
-          "threshold. When a build side fits in memory and is smaller than the per-task " +
-          "threshold, the join executes as a single HashJoinExec without spilling. " +
+        "Per-task memory budget in bytes for Grace Hash Join fast-path hash tables. " +
+          "When a build side fits in memory and is smaller than this threshold, " +
+          "the join executes as a single HashJoinExec without partitioning or spilling. " +
           "Set to 0 to disable the fast path. Larger values risk OOM because HashJoinExec " +
           "creates non-spillable hash tables.")
       .longConf
       .checkValue(v => v >= 0, "The fast path threshold must be non-negative.")
-      .createWithDefault(10L * 1024 * 1024) // 10 MB
+      .createWithDefault(64L * 1024 * 1024) // 64 MB
 
   val COMET_NATIVE_COLUMNAR_TO_ROW_ENABLED: ConfigEntry[Boolean] =
     conf(s"$COMET_EXEC_CONFIG_PREFIX.columnarToRow.native.enabled")
@@ -414,7 +413,7 @@ object CometConf extends ShimCometConf {
           "on pre-sorted data outperforms hash join's per-task hash table construction " +
           "for large build sides. Set to -1 to disable this check and always replace.")
       .longConf
-      .createWithDefault(-1L)
+      .createWithDefault(100L * 1024 * 1024) // 100 MB
 
   val COMET_EXEC_SHUFFLE_WITH_HASH_PARTITIONING_ENABLED: ConfigEntry[Boolean] =
     conf("spark.comet.native.shuffle.partitioning.hash.enabled")

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

@@ -32,9 +32,9 @@ Supports all join types: Inner, Left, Right, Full, LeftSemi, LeftAnti, LeftMark,
 | Config Key                                           | Type    | Default    | Description                                                |
 | ---------------------------------------------------- | ------- | ---------- | ---------------------------------------------------------- |
 | `spark.comet.exec.replaceSortMergeJoin`              | boolean | `false`    | Replace SortMergeJoin with ShuffledHashJoin (enables GHJ)  |
-| `spark.comet.exec.replaceSortMergeJoin.maxBuildSize` | long    | `-1`       | 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     | `10485760` | Total fast-path budget in bytes, divided by executor cores |
+| `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                         |
 
 ### SMJ Replacement Guard
 

native/core/src/execution/planner.rs

@@ -1576,20 +1576,15 @@ impl PhysicalPlanner {
 
                 use crate::execution::spark_config::{
                     SparkConfig, COMET_GRACE_HASH_JOIN_FAST_PATH_THRESHOLD,
-                    COMET_GRACE_HASH_JOIN_NUM_PARTITIONS, SPARK_EXECUTOR_CORES,
+                    COMET_GRACE_HASH_JOIN_NUM_PARTITIONS,
                 };
 
                 let num_partitions = self
                     .spark_conf
                     .get_usize(COMET_GRACE_HASH_JOIN_NUM_PARTITIONS, 16);
-                let executor_cores = self.spark_conf.get_usize(SPARK_EXECUTOR_CORES, 1).max(1);
-                // The configured threshold is the total budget across all
-                // concurrent tasks. Divide by executor cores so each task's
-                // fast-path hash table stays within its fair share.
                 let fast_path_threshold = self
                     .spark_conf
-                    .get_usize(COMET_GRACE_HASH_JOIN_FAST_PATH_THRESHOLD, 10 * 1024 * 1024)
-                    / executor_cores;
+                    .get_usize(COMET_GRACE_HASH_JOIN_FAST_PATH_THRESHOLD, 64 * 1024 * 1024);
 
                 let build_left = join.build_side == BuildSide::BuildLeft as i32;