perf: optimistic fast path for grace hash join

Skip build-side hash partitioning when the fast path threshold is set.
Instead of always computing hashes and splitting every build batch into
N partitions (only to collect them back together for the fast path),
buffer the build side directly. When the build fits in memory and is
under the threshold, feed it straight to HashJoinExec with zero
partitioning overhead.

Falls back to the partitioned slow path on memory pressure or when the
build exceeds the threshold.

Also fix CometConf fastPathThreshold type from intConf to longConf to
support values > 2 GB without integer overflow, and remove a duplicate
config line in the benchmark TOML.

~4% improvement on both TPC-H and TPC-DS benchmarks.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: andygrove

benchmarks/tpc/engines/comet-hashjoin.toml

@@ -32,7 +32,6 @@ 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" = "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

@@ -314,7 +314,7 @@ object CometConf extends ShimCometConf {
       .checkValue(v => v > 0, "The number of partitions must be positive.")
       .createWithDefault(16)
 
-  val COMET_EXEC_GRACE_HASH_JOIN_FAST_PATH_THRESHOLD: ConfigEntry[Int] =
+  val COMET_EXEC_GRACE_HASH_JOIN_FAST_PATH_THRESHOLD: ConfigEntry[Long] =
     conf(s"$COMET_EXEC_CONFIG_PREFIX.graceHashJoin.fastPathThreshold")
       .category(CATEGORY_EXEC)
       .doc(
@@ -324,9 +324,9 @@ object CometConf extends ShimCometConf {
           "threshold, the join executes as a single HashJoinExec without spilling. " +
           "Set to 0 to disable the fast path. Larger values risk OOM because HashJoinExec " +
           "creates non-spillable hash tables.")
-      .intConf
+      .longConf
       .checkValue(v => v >= 0, "The fast path threshold must be non-negative.")
-      .createWithDefault(10 * 1024 * 1024) // 10 MB
+      .createWithDefault(10L * 1024 * 1024) // 10 MB
 
   val COMET_NATIVE_COLUMNAR_TO_ROW_ENABLED: ConfigEntry[Boolean] =
     conf(s"$COMET_EXEC_CONFIG_PREFIX.columnarToRow.native.enabled")