fix: prevent wrong results from Iceberg native scan exchange reuse with different pushed filters

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

@@ -60,6 +60,7 @@ case class CometIcebergNativeScanExec(
     @transient override val originalPlan: BatchScanExec,
     override val serializedPlanOpt: SerializedPlan,
     metadataLocation: String,
+    scanHashCode: Int,
     @transient nativeIcebergScanMetadata: CometIcebergNativeScanMetadata)
     extends CometLeafExec {
 
@@ -258,6 +259,7 @@ case class CometIcebergNativeScanExec(
       originalPlan,
       newSerializedPlan,
       metadataLocation,
+      scanHashCode,
       nativeIcebergScanMetadata)
   }
 
@@ -271,6 +273,10 @@ case class CometIcebergNativeScanExec(
       null, // Don't need originalPlan for canonicalization
       SerializedPlan(None),
       metadataLocation,
+      // originalPlan is nulled here, so scanHashCode is the only field left that distinguishes
+      // scans differing solely in pushed-down filters. Dropping it lets ReuseExchange collapse
+      // them (see #4774).
+      scanHashCode,
       null
     ) // Don't need metadata for canonicalization
   }
@@ -296,21 +302,36 @@ case class CometIcebergNativeScanExec(
    * `originalPlan` (`@transient`) and `nativeIcebergScanMetadata` (`@transient`) are
    * intentionally omitted: they're recoverable from `metadataLocation` + the serialized plan and
    * including them would over-constrain equality across re-planning.
+   *
+   * `scanHashCode` (Iceberg's `SparkScan.hashCode()`, folding in pushed filters, snapshot,
+   * branch, and read schema) distinguishes scans that differ only in static pushed-down filters,
+   * so that ReuseExchange does not collapse two scans that read different data (see #4774). It is
+   * an `Int` and so carries a theoretical hash-collision risk, but `metadataLocation` (table +
+   * snapshot path) is compared alongside it, matching how Iceberg's `SparkBatch.equals` pairs the
+   * hash with `table.name()`.
    */
   override def equals(obj: Any): Boolean = {
     obj match {
       case other: CometIcebergNativeScanExec =>
         this.metadataLocation == other.metadataLocation &&
         this.output == other.output &&
         this.serializedPlanOpt == other.serializedPlanOpt &&
-        this.runtimeFilters == other.runtimeFilters
+        this.runtimeFilters == other.runtimeFilters &&
+        this.scanHashCode == other.scanHashCode
       case _ =>
         false
     }
   }
 
   override def hashCode(): Int =
-    Objects.hashCode(metadataLocation, output.asJava, serializedPlanOpt, runtimeFilters)
+    Objects.hashCode(
+      metadataLocation,
+      output.asJava,
+      serializedPlanOpt,
+      runtimeFilters,
+      // Ascribe java.lang.Integer: Scala 2.12 rejects the implicit Int -> Object conversion into
+      // the Object... varargs of Guava's Objects.hashCode.
+      scanHashCode: java.lang.Integer)
 }
 
 object CometIcebergNativeScanExec {
@@ -330,6 +351,9 @@ object CometIcebergNativeScanExec {
       scanExec,
       SerializedPlan(None),
       metadataLocation,
+      // Capture Iceberg's scan hash now, while the transient scan is still available; it is
+      // needed for equality after canonicalization nulls originalPlan (see #4774).
+      scanExec.scan.hashCode(),
       nativeIcebergScanMetadata)
 
     scanExec.logicalLink.foreach(exec.setLogicalLink)

spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala

@@ -3055,6 +3055,68 @@ class CometIcebergNativeSuite
     }
   }
 
+  test("exchange reuse must not collapse scans with different pushed filters (#4774)") {
+    assume(icebergAvailable, "Iceberg not available")
+
+    withTempIcebergDir { warehouseDir =>
+      withSQLConf(
+        // Non-AQE: ReuseExchangeAndSubquery keys reuse off Exchange.canonicalized, which is
+        // where a scan's canonical form losing its pushed filters causes an incorrect collapse.
+        SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
+        SQLConf.EXCHANGE_REUSE_ENABLED.key -> "true",
+        "spark.sql.catalog.reuse_cat" -> "org.apache.iceberg.spark.SparkCatalog",
+        "spark.sql.catalog.reuse_cat.type" -> "hadoop",
+        "spark.sql.catalog.reuse_cat.warehouse" -> warehouseDir.getAbsolutePath,
+        CometConf.COMET_ENABLED.key -> "true",
+        CometConf.COMET_EXEC_ENABLED.key -> "true",
+        CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") {
+
+        spark.sql("""
+          CREATE TABLE reuse_cat.db.reuse_table (
+            id INT,
+            category STRING,
+            value DOUBLE
+          ) USING iceberg
+          PARTITIONED BY (category)
+        """)
+
+        spark.sql("""
+          INSERT INTO reuse_cat.db.reuse_table VALUES
+          (1, 'A', 10.0), (2, 'A', 20.0), (3, 'A', 30.0),
+          (4, 'B', 100.0), (5, 'B', 200.0), (6, 'B', 300.0)
+        """)
+
+        // Each branch filters on a different partition. Iceberg pushes the predicate fully into
+        // the scan, so nothing distinguishes the two branches in the plan tree above the scans.
+        // If the scans canonicalize identically, reuse points branch B at branch A's exchange,
+        // yielding A twice and dropping B.
+        val query =
+          """
+            SELECT category, SUM(value) AS total
+            FROM reuse_cat.db.reuse_table WHERE category = 'A' GROUP BY category
+            UNION ALL
+            SELECT category, SUM(value) AS total
+            FROM reuse_cat.db.reuse_table WHERE category = 'B' GROUP BY category
+          """
+
+        val (_, cometPlan) = checkSparkAnswerAndOperator(query)
+
+        // Guard against a silently-correct future regression: the two branches read different
+        // partitions, so their exchanges must not be collapsed. Assert both scans survive and
+        // that no ReusedExchangeExec merged the branches.
+        assert(
+          collectIcebergNativeScans(cometPlan).length == 2,
+          "Expected 2 CometIcebergNativeScanExec (one per branch) but found " +
+            s"${collectIcebergNativeScans(cometPlan).length}. Plan:\n$cometPlan")
+        assert(
+          collect(cometPlan) { case r: ReusedExchangeExec => r }.isEmpty,
+          s"Scans with different pushed filters must not share an exchange:\n$cometPlan")
+
+        spark.sql("DROP TABLE reuse_cat.db.reuse_table")
+      }
+    }
+  }
+
   // ---- AQE DPP broadcast reuse tests ----
 
   private def collectIcebergDPPSubqueries(plan: SparkPlan): Seq[SparkPlan] = {