fix: update join metrics tests for GraceHashJoinExec

- HashJoin test now matches CometGraceHashJoinExec and checks GHJ
  metrics (no build_mem_used, adds spill_count)
- BroadcastHashJoin test removes build_mem_used assertion since
  the native side does not report this metric
- Remove dead CometHashJoinExec case class (createExec already
  produces CometGraceHashJoinExec)

Author: andygrove

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

@@ -1740,61 +1740,6 @@ object CometHashJoinExec extends CometOperatorSerde[HashJoin] with CometHashJoin
   }
 }
 
-case class CometHashJoinExec(
-    override val nativeOp: Operator,
-    override val originalPlan: SparkPlan,
-    override val output: Seq[Attribute],
-    override val outputOrdering: Seq[SortOrder],
-    leftKeys: Seq[Expression],
-    rightKeys: Seq[Expression],
-    joinType: JoinType,
-    condition: Option[Expression],
-    buildSide: BuildSide,
-    override val left: SparkPlan,
-    override val right: SparkPlan,
-    override val serializedPlanOpt: SerializedPlan)
-    extends CometBinaryExec {
-
-  override def outputPartitioning: Partitioning = joinType match {
-    case _: InnerLike =>
-      PartitioningCollection(Seq(left.outputPartitioning, right.outputPartitioning))
-    case LeftOuter => left.outputPartitioning
-    case RightOuter => right.outputPartitioning
-    case FullOuter => UnknownPartitioning(left.outputPartitioning.numPartitions)
-    case LeftExistence(_) => left.outputPartitioning
-    case x =>
-      throw new IllegalArgumentException(s"ShuffledJoin should not take $x as the JoinType")
-  }
-
-  override def withNewChildrenInternal(newLeft: SparkPlan, newRight: SparkPlan): SparkPlan =
-    this.copy(left = newLeft, right = newRight)
-
-  override def stringArgs: Iterator[Any] =
-    Iterator(leftKeys, rightKeys, joinType, buildSide, condition, left, right)
-
-  override def equals(obj: Any): Boolean = {
-    obj match {
-      case other: CometHashJoinExec =>
-        this.output == other.output &&
-        this.leftKeys == other.leftKeys &&
-        this.rightKeys == other.rightKeys &&
-        this.condition == other.condition &&
-        this.buildSide == other.buildSide &&
-        this.left == other.left &&
-        this.right == other.right &&
-        this.serializedPlanOpt == other.serializedPlanOpt
-      case _ =>
-        false
-    }
-  }
-
-  override def hashCode(): Int =
-    Objects.hashCode(output, leftKeys, rightKeys, condition, buildSide, left, right)
-
-  override lazy val metrics: Map[String, SQLMetric] =
-    CometMetricNode.hashJoinMetrics(sparkContext)
-}
-
 case class CometGraceHashJoinExec(
     override val nativeOp: Operator,
     override val originalPlan: SparkPlan,

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

@@ -692,16 +692,14 @@ class CometExecSuite extends CometTestBase {
         df.collect()
 
         val metrics = find(df.queryExecution.executedPlan) {
-          case _: CometHashJoinExec => true
+          case _: CometGraceHashJoinExec => true
           case _ => false
         }.map(_.metrics).get
 
         assert(metrics.contains("build_time"))
         assert(metrics("build_time").value > 1L)
         assert(metrics.contains("build_input_batches"))
         assert(metrics("build_input_batches").value == 5L)
-        assert(metrics.contains("build_mem_used"))
-        assert(metrics("build_mem_used").value > 1L)
         assert(metrics.contains("build_input_rows"))
         assert(metrics("build_input_rows").value == 5L)
         assert(metrics.contains("input_batches"))
@@ -714,6 +712,8 @@ class CometExecSuite extends CometTestBase {
         assert(metrics("output_rows").value == 5L)
         assert(metrics.contains("join_time"))
         assert(metrics("join_time").value > 1L)
+        assert(metrics.contains("spill_count"))
+        assert(metrics("spill_count").value == 0)
       }
     }
   }
@@ -733,8 +733,6 @@ class CometExecSuite extends CometTestBase {
         assert(metrics("build_time").value > 1L)
         assert(metrics.contains("build_input_batches"))
         assert(metrics("build_input_batches").value == 25L)
-        assert(metrics.contains("build_mem_used"))
-        assert(metrics("build_mem_used").value > 1L)
         assert(metrics.contains("build_input_rows"))
         assert(metrics("build_input_rows").value == 25L)
         assert(metrics.contains("input_batches"))