fix: resolve CI failures for grace hash join PR

- Rename CometGraceHashJoinExec back to CometHashJoinExec to avoid
  breaking spark-sql test diffs that reference CometHashJoinExec
- Add output_batches and join_time metrics to GraceHashJoinMetrics
  (both fast and slow paths)
- Fix clippy type_complexity warning on create_fast_path_stream

Author: andygrove

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

@@ -52,7 +52,7 @@ The configured threshold is the total budget across all concurrent tasks on the
 SortMergeJoinExec
     -> RewriteJoin converts to ShuffledHashJoinExec (removes input sorts)
     -> CometExecRule wraps as CometHashJoinExec
-    -> CometHashJoinExec.createExec() creates CometGraceHashJoinExec
+    -> CometHashJoinExec.createExec() creates CometHashJoinExec
     -> Serialized to protobuf via JNI
     -> PhysicalPlanner (Rust) creates GraceHashJoinExec
 ```

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

@@ -429,6 +429,10 @@ struct GraceHashJoinMetrics {
     input_rows: Count,
     /// Number of probe-side input batches
     input_batches: Count,
+    /// Number of output batches
+    output_batches: Count,
+    /// Time spent in per-partition joins
+    join_time: Time,
 }
 
 impl GraceHashJoinMetrics {
@@ -444,6 +448,8 @@ impl GraceHashJoinMetrics {
                 .counter("build_input_batches", partition),
             input_rows: MetricBuilder::new(metrics).counter("input_rows", partition),
             input_batches: MetricBuilder::new(metrics).counter("input_batches", partition),
+            output_batches: MetricBuilder::new(metrics).counter("output_batches", partition),
+            join_time: MetricBuilder::new(metrics).subset_time("join_time", partition),
         }
     }
 }
@@ -821,8 +827,12 @@ async fn execute_grace_hash_join(
                 )?;
 
                 let output_metrics = metrics.baseline.clone();
+                let output_batch_count = metrics.output_batches.clone();
+                let join_time = metrics.join_time.clone();
                 let result_stream = stream.inspect_ok(move |batch| {
+                    let _timer = join_time.timer();
                     output_metrics.record_output(batch.num_rows());
+                    output_batch_count.add(1);
                 });
 
                 return Ok(result_stream.boxed());
@@ -1072,7 +1082,7 @@ fn partition_from_buffer(
 }
 
 /// Create the fast-path HashJoinExec stream (no partitioning, no spilling).
-#[allow(clippy::too_many_arguments)]
+#[allow(clippy::too_many_arguments, clippy::type_complexity)]
 fn create_fast_path_stream(
     build_data: Vec<RecordBatch>,
     probe_stream: SendableRecordBatchStream,
@@ -1303,6 +1313,8 @@ async fn execute_slow_path(
     drop(tx);
 
     let output_metrics = metrics.baseline.clone();
+    let output_batch_count = metrics.output_batches.clone();
+    let join_time = metrics.join_time.clone();
     let output_row_count = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
     let counter = Arc::clone(&output_row_count);
     let jt = join_type;
@@ -1311,7 +1323,9 @@ async fn execute_slow_path(
         rx.recv().await.map(|batch| (batch, rx))
     })
     .inspect_ok(move |batch| {
+        let _timer = join_time.timer();
         output_metrics.record_output(batch.num_rows());
+        output_batch_count.add(1);
         let prev = counter.fetch_add(batch.num_rows(), std::sync::atomic::Ordering::Relaxed);
         let new_total = prev + batch.num_rows();
         // Log every ~1M rows to detect exploding joins

spark/src/main/scala/org/apache/spark/sql/comet/operators.scala

@@ -1724,7 +1724,7 @@ object CometHashJoinExec extends CometOperatorSerde[HashJoin] with CometHashJoin
     doConvert(join, builder, childOp: _*)
 
   override def createExec(nativeOp: Operator, op: HashJoin): CometNativeExec = {
-    CometGraceHashJoinExec(
+    CometHashJoinExec(
       nativeOp,
       op,
       op.output,
@@ -1740,7 +1740,7 @@ object CometHashJoinExec extends CometOperatorSerde[HashJoin] with CometHashJoin
   }
 }
 
-case class CometGraceHashJoinExec(
+case class CometHashJoinExec(
     override val nativeOp: Operator,
     override val originalPlan: SparkPlan,
     override val output: Seq[Attribute],
@@ -1774,7 +1774,7 @@ case class CometGraceHashJoinExec(
 
   override def equals(obj: Any): Boolean = {
     obj match {
-      case other: CometGraceHashJoinExec =>
+      case other: CometHashJoinExec =>
         this.output == other.output &&
         this.leftKeys == other.leftKeys &&
         this.rightKeys == other.rightKeys &&

spark/src/test/scala/org/apache/comet/exec/CometExecSuite.scala

@@ -692,7 +692,7 @@ class CometExecSuite extends CometTestBase {
         df.collect()
 
         val metrics = find(df.queryExecution.executedPlan) {
-          case _: CometGraceHashJoinExec => true
+          case _: CometHashJoinExec => true
           case _ => false
         }.map(_.metrics).get
 

spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala

@@ -27,7 +27,7 @@ import org.scalatest.Tag
 import org.apache.spark.sql.CometTestBase
 import org.apache.spark.sql.catalyst.TableIdentifier
 import org.apache.spark.sql.catalyst.analysis.UnresolvedRelation
-import org.apache.spark.sql.comet.{CometBroadcastExchangeExec, CometBroadcastHashJoinExec, CometGraceHashJoinExec}
+import org.apache.spark.sql.comet.{CometBroadcastExchangeExec, CometBroadcastHashJoinExec, CometHashJoinExec}
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types.{DataTypes, Decimal, StructField, StructType}
 
@@ -592,13 +592,13 @@ class CometJoinSuite extends CometTestBase {
     }
   }
 
-  test("Grace HashJoin - plan shows CometGraceHashJoinExec") {
+  test("Grace HashJoin - plan shows CometHashJoinExec") {
     withSQLConf(graceHashJoinConf: _*) {
       withParquetTable((0 until 50).map(i => (i, i % 5)), "tbl_a") {
         withParquetTable((0 until 50).map(i => (i % 10, i + 2)), "tbl_b") {
           val df = sql(
             "SELECT /*+ SHUFFLE_HASH(tbl_a) */ * FROM tbl_a JOIN tbl_b ON tbl_a._2 = tbl_b._1")
-          checkSparkAnswerAndOperator(df, Seq(classOf[CometGraceHashJoinExec]))
+          checkSparkAnswerAndOperator(df, Seq(classOf[CometHashJoinExec]))
         }
       }
     }