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

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

@@ -294,6 +294,39 @@ object CometConf extends ShimCometConf {
   val COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED: ConfigEntry[Boolean] =
     createExecEnabledConfig("localTableScan", defaultValue = false)
 
+  val COMET_EXEC_GRACE_HASH_JOIN_ENABLED: ConfigEntry[Boolean] =
+    conf(s"$COMET_EXEC_CONFIG_PREFIX.graceHashJoin.enabled")
+      .category(CATEGORY_EXEC)
+      .doc(
+        "Whether to enable Grace Hash Join. When enabled, Comet will use a Grace Hash Join " +
+          "operator that partitions both sides into buckets and can spill to disk when memory " +
+          "is tight. Supports all join types. This is an experimental feature.")
+      .booleanConf
+      .createWithDefault(false)
+
+  val COMET_EXEC_GRACE_HASH_JOIN_NUM_PARTITIONS: ConfigEntry[Int] =
+    conf(s"$COMET_EXEC_CONFIG_PREFIX.graceHashJoin.numPartitions")
+      .category(CATEGORY_EXEC)
+      .doc("The number of partitions (buckets) to use for Grace Hash Join. A higher number " +
+        "reduces the size of each partition but increases overhead.")
+      .intConf
+      .checkValue(v => v > 0, "The number of partitions must be positive.")
+      .createWithDefault(16)
+
+  val COMET_EXEC_GRACE_HASH_JOIN_FAST_PATH_THRESHOLD: ConfigEntry[Int] =
+    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. " +
+          "Set to 0 to disable the fast path. Larger values risk OOM because HashJoinExec " +
+          "creates non-spillable hash tables.")
+      .intConf
+      .checkValue(v => v >= 0, "The fast path threshold must be non-negative.")
+      .createWithDefault(10 * 1024 * 1024) // 10 MB
+
   val COMET_NATIVE_COLUMNAR_TO_ROW_ENABLED: ConfigEntry[Boolean] =
     conf(s"$COMET_EXEC_CONFIG_PREFIX.columnarToRow.native.enabled")
       .category(CATEGORY_EXEC)

native/Cargo.toml

@@ -34,7 +34,7 @@ edition = "2021"
 rust-version = "1.88"
 
 [workspace.dependencies]
-arrow = { version = "57.3.0", features = ["prettyprint", "ffi", "chrono-tz"] }
+arrow = { version = "57.3.0", features = ["prettyprint", "ffi", "chrono-tz", "ipc_compression"] }
 async-trait = { version = "0.1" }
 bytes = { version = "1.11.1" }
 parquet = { version = "57.2.0", default-features = false, features = ["experimental"] }

native/core/Cargo.toml

@@ -35,6 +35,7 @@ include = [
 publish = false
 
 [dependencies]
+ahash = "0.8"
 arrow = { workspace = true }
 parquet = { workspace = true, default-features = false, features = ["experimental", "arrow"] }
 futures = { workspace = true }

native/core/src/execution/jni_api.rs

@@ -173,6 +173,8 @@ struct ExecutionContext {
     pub memory_pool_config: MemoryPoolConfig,
     /// Whether to log memory usage on each call to execute_plan
     pub tracing_enabled: bool,
+    /// Spark configuration map for comet-specific settings
+    pub spark_conf: HashMap<String, String>,
 }
 
 /// Accept serialized query plan and return the address of the native query plan.
@@ -320,6 +322,7 @@ pub unsafe extern "system" fn Java_org_apache_comet_Native_createPlan(
                 explain_native,
                 memory_pool_config,
                 tracing_enabled,
+                spark_conf: spark_config,
             });
 
             Ok(Box::into_raw(exec_context) as i64)
@@ -531,7 +534,8 @@ pub unsafe extern "system" fn Java_org_apache_comet_Native_executePlan(
                 let start = Instant::now();
                 let planner =
                     PhysicalPlanner::new(Arc::clone(&exec_context.session_ctx), partition)
-                        .with_exec_id(exec_context_id);
+                        .with_exec_id(exec_context_id)
+                        .with_spark_conf(exec_context.spark_conf.clone());
                 let (scans, root_op) = planner.create_plan(
                     &exec_context.spark_plan,
                     &mut exec_context.input_sources.clone(),