Remove nativeIcebergScanMetadata from hashCode and equals in CometIcebergNativeScanExec since it's redundant to the wrapped scan object and introduces a ton of serde overhead.

Author: mbutrovich

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

@@ -41,7 +41,7 @@ import org.apache.comet.iceberg.CometIcebergNativeScanMetadata
 case class CometBatchScanExec(
     wrapped: BatchScanExec,
     runtimeFilters: Seq[Expression],
-    nativeIcebergScanMetadata: Option[CometIcebergNativeScanMetadata] = None)
+    @transient nativeIcebergScanMetadata: Option[CometIcebergNativeScanMetadata] = None)
     extends DataSourceV2ScanExecBase
     with CometPlan {
   def ordering: Option[Seq[SortOrder]] = wrapped.ordering
@@ -99,26 +99,23 @@ case class CometBatchScanExec(
   override def equals(other: Any): Boolean = other match {
     case other: CometBatchScanExec =>
       // `wrapped` in `this` and `other` could reference to the same `BatchScanExec` object,
-      // check `runtimeFilters` and `nativeIcebergScanMetadata` equality too.
-      this.wrappedScan == other.wrappedScan && this.runtimeFilters == other.runtimeFilters &&
-      this.nativeIcebergScanMetadata == other.nativeIcebergScanMetadata
+      // check `runtimeFilters` equality too.
+      this.wrappedScan == other.wrappedScan && this.runtimeFilters == other.runtimeFilters
     case _ =>
       false
   }
 
   override def hashCode(): Int = {
-    Objects.hashCode(
-      wrappedScan,
-      runtimeFilters,
-      Integer.valueOf(nativeIcebergScanMetadata.map(_.hashCode()).getOrElse(0)))
+    Objects.hashCode(wrappedScan, runtimeFilters)
   }
 
   override def doCanonicalize(): CometBatchScanExec = {
     this.copy(
       wrapped = wrappedScan.doCanonicalize(),
       runtimeFilters = QueryPlan.normalizePredicates(
         runtimeFilters.filterNot(_ == DynamicPruningExpression(Literal.TrueLiteral)),
-        output))
+        output),
+      nativeIcebergScanMetadata = None)
   }
 
   override def nodeName: String = {

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

@@ -166,8 +166,7 @@ case class CometIcebergNativeScanExec(
         this.metadataLocation == other.metadataLocation &&
         this.output == other.output &&
         this.serializedPlanOpt == other.serializedPlanOpt &&
-        this.numPartitions == other.numPartitions &&
-        this.nativeIcebergScanMetadata == other.nativeIcebergScanMetadata
+        this.numPartitions == other.numPartitions
       case _ =>
         false
     }
@@ -178,8 +177,7 @@ case class CometIcebergNativeScanExec(
       metadataLocation,
       output.asJava,
       serializedPlanOpt,
-      numPartitions: java.lang.Integer,
-      nativeIcebergScanMetadata)
+      numPartitions: java.lang.Integer)
 }
 
 object CometIcebergNativeScanExec {