chore: [branch-17] backport #4812

docs/source/user-guide/latest/compatibility/expressions/spark-3.4/aggregate.md

@@ -44,4 +44,5 @@ The following differences from Spark are always present and do not require any a
 The following differences from Spark are always present and do not require any additional configuration:
 
 - This function is not deterministic. Results may not match Spark.
+
 <!--END:EXPR_COMPAT-->

docs/source/user-guide/latest/compatibility/expressions/spark-3.4/array.md

@@ -70,4 +70,5 @@ The following incompatibilities cause `SortArray` to fall back to Spark by defau
 The following cases are not supported by Comet:
 
 - Nested arrays with `Struct` or `Null` child values are not supported natively and will fall back to Spark.
+
 <!--END:EXPR_COMPAT-->

docs/source/user-guide/latest/compatibility/expressions/spark-3.4/datetime.md

@@ -136,4 +136,5 @@ The following cases are not supported by Comet:
 The following cases are not supported by Comet:
 
 - Only `TimestampType` and `DateType` inputs are supported. `TimestampNTZType` is not supported because Comet incorrectly applies timezone conversion to TimestampNTZ values.
+
 <!--END:EXPR_COMPAT-->

docs/source/user-guide/latest/compatibility/expressions/spark-3.4/map.md

@@ -38,4 +38,5 @@ By default, Comet accelerates `MapFromEntries` using JVM codegen dispatch, which
 
 - `BinaryType` is not supported as a map key in `map_from_entries`
 - `BinaryType` is not supported as a map value in `map_from_entries`
+
 <!--END:EXPR_COMPAT-->

docs/source/user-guide/latest/compatibility/expressions/spark-3.4/math.md

@@ -38,4 +38,5 @@ The following cases are not supported by Comet:
 The following cases are not supported by Comet:
 
 - The `seed` argument must be a literal value
+
 <!--END:EXPR_COMPAT-->

docs/source/user-guide/latest/compatibility/expressions/spark-3.4/misc.md

@@ -50,4 +50,5 @@ The following cases are not supported by Comet:
 The following incompatibilities cause `SortOrder` to fall back to Spark by default. Set `spark.comet.expression.SortOrder.allowIncompatible=true` to enable Comet acceleration despite these differences.
 
 - When `spark.comet.exec.strictFloatingPoint=true`, sorting on floating-point types is not 100% compatible with Spark
+
 <!--END:EXPR_COMPAT-->

docs/source/user-guide/latest/compatibility/expressions/spark-3.4/string.md

@@ -94,4 +94,5 @@ The following differences from Spark are always present and do not require any a
 The following incompatibilities cause `StringTranslate` to fall back to Spark by default. Set `spark.comet.expression.StringTranslate.allowIncompatible=true` to enable Comet acceleration despite these differences.
 
 - DataFusion's translate iterates over Unicode graphemes (Spark uses code points) and substitutes U+0000 instead of treating it as a deletion sentinel
+
 <!--END:EXPR_COMPAT-->

docs/source/user-guide/latest/compatibility/expressions/spark-3.4/struct.md

@@ -36,4 +36,5 @@ The following incompatibilities cause `StructsToCsv` to fall back to Spark by de
 The following cases are not supported by Comet:
 
 - Complex types (arrays, maps, structs) in the schema are not supported
+
 <!--END:EXPR_COMPAT-->

docs/source/user-guide/latest/compatibility/expressions/spark-3.5/aggregate.md

@@ -44,4 +44,5 @@ The following differences from Spark are always present and do not require any a
 The following differences from Spark are always present and do not require any additional configuration:
 
 - This function is not deterministic. Results may not match Spark.
+
 <!--END:EXPR_COMPAT-->

docs/source/user-guide/latest/compatibility/expressions/spark-3.5/array.md

@@ -70,4 +70,5 @@ The following incompatibilities cause `SortArray` to fall back to Spark by defau
 The following cases are not supported by Comet:
 
 - Nested arrays with `Struct` or `Null` child values are not supported natively and will fall back to Spark.
+
 <!--END:EXPR_COMPAT-->

docs/source/user-guide/latest/compatibility/expressions/spark-3.5/datetime.md

@@ -136,4 +136,5 @@ The following cases are not supported by Comet:
 The following cases are not supported by Comet:
 
 - Only `TimestampType` and `DateType` inputs are supported. `TimestampNTZType` is not supported because Comet incorrectly applies timezone conversion to TimestampNTZ values.
+
 <!--END:EXPR_COMPAT-->

docs/source/user-guide/latest/compatibility/expressions/spark-3.5/map.md

@@ -38,4 +38,5 @@ By default, Comet accelerates `MapFromEntries` using JVM codegen dispatch, which
 
 - `BinaryType` is not supported as a map key in `map_from_entries`
 - `BinaryType` is not supported as a map value in `map_from_entries`
+
 <!--END:EXPR_COMPAT-->

docs/source/user-guide/latest/compatibility/expressions/spark-3.5/math.md

@@ -38,4 +38,5 @@ The following cases are not supported by Comet:
 The following cases are not supported by Comet:
 
 - The `seed` argument must be a literal value
+
 <!--END:EXPR_COMPAT-->

docs/source/user-guide/latest/compatibility/expressions/spark-3.5/misc.md

@@ -56,4 +56,5 @@ The following incompatibilities cause `SortOrder` to fall back to Spark by defau
 The following cases are not supported by Comet:
 
 - Falls back to Spark when the input type cannot be cast to string.
+
 <!--END:EXPR_COMPAT-->

docs/source/user-guide/latest/compatibility/expressions/spark-3.5/string.md

@@ -94,4 +94,5 @@ The following differences from Spark are always present and do not require any a
 The following incompatibilities cause `StringTranslate` to fall back to Spark by default. Set `spark.comet.expression.StringTranslate.allowIncompatible=true` to enable Comet acceleration despite these differences.
 
 - DataFusion's translate iterates over Unicode graphemes (Spark uses code points) and substitutes U+0000 instead of treating it as a deletion sentinel
+
 <!--END:EXPR_COMPAT-->

docs/source/user-guide/latest/compatibility/expressions/spark-3.5/struct.md

@@ -36,4 +36,5 @@ The following incompatibilities cause `StructsToCsv` to fall back to Spark by de
 The following cases are not supported by Comet:
 
 - Complex types (arrays, maps, structs) in the schema are not supported
+
 <!--END:EXPR_COMPAT-->

docs/source/user-guide/latest/compatibility/expressions/spark-4.0/aggregate.md

@@ -44,4 +44,5 @@ The following differences from Spark are always present and do not require any a
 The following differences from Spark are always present and do not require any additional configuration:
 
 - This function is not deterministic. Results may not match Spark.
+
 <!--END:EXPR_COMPAT-->

docs/source/user-guide/latest/compatibility/expressions/spark-4.0/array.md

@@ -70,4 +70,5 @@ The following incompatibilities cause `SortArray` to fall back to Spark by defau
 The following cases are not supported by Comet:
 
 - Nested arrays with `Struct` or `Null` child values are not supported natively and will fall back to Spark.
+
 <!--END:EXPR_COMPAT-->

docs/source/user-guide/latest/compatibility/expressions/spark-4.0/datetime.md

@@ -136,4 +136,5 @@ The following cases are not supported by Comet:
 The following cases are not supported by Comet:
 
 - Only `TimestampType` and `DateType` inputs are supported. `TimestampNTZType` is not supported because Comet incorrectly applies timezone conversion to TimestampNTZ values.
+
 <!--END:EXPR_COMPAT-->

docs/source/user-guide/latest/compatibility/expressions/spark-4.0/map.md

@@ -48,4 +48,5 @@ The following incompatibilities cause `MapSort` to fall back to Spark by default
 The following cases are not supported by Comet:
 
 - MapSort is unsupported for non-scalar key types (struct, array, map, etc.).
+
 <!--END:EXPR_COMPAT-->

docs/source/user-guide/latest/compatibility/expressions/spark-4.0/math.md

@@ -38,4 +38,5 @@ The following cases are not supported by Comet:
 The following cases are not supported by Comet:
 
 - The `seed` argument must be a literal value
+
 <!--END:EXPR_COMPAT-->

docs/source/user-guide/latest/compatibility/expressions/spark-4.0/misc.md

@@ -56,4 +56,5 @@ The following incompatibilities cause `SortOrder` to fall back to Spark by defau
 The following cases are not supported by Comet:
 
 - Falls back to Spark when the input type cannot be cast to string.
+
 <!--END:EXPR_COMPAT-->

docs/source/user-guide/latest/compatibility/expressions/spark-4.0/string.md

@@ -94,4 +94,5 @@ The following differences from Spark are always present and do not require any a
 The following incompatibilities cause `StringTranslate` to fall back to Spark by default. Set `spark.comet.expression.StringTranslate.allowIncompatible=true` to enable Comet acceleration despite these differences.
 
 - DataFusion's translate iterates over Unicode graphemes (Spark uses code points) and substitutes U+0000 instead of treating it as a deletion sentinel
+
 <!--END:EXPR_COMPAT-->

docs/source/user-guide/latest/compatibility/expressions/spark-4.0/struct.md

@@ -36,4 +36,5 @@ The following incompatibilities cause `StructsToCsv` to fall back to Spark by de
 The following cases are not supported by Comet:
 
 - Complex types (arrays, maps, structs) in the schema are not supported
+
 <!--END:EXPR_COMPAT-->

docs/source/user-guide/latest/compatibility/expressions/spark-4.1/aggregate.md

@@ -44,4 +44,5 @@ The following differences from Spark are always present and do not require any a
 The following differences from Spark are always present and do not require any additional configuration:
 
 - This function is not deterministic. Results may not match Spark.
+
 <!--END:EXPR_COMPAT-->

docs/source/user-guide/latest/compatibility/expressions/spark-4.1/array.md

@@ -70,4 +70,5 @@ The following incompatibilities cause `SortArray` to fall back to Spark by defau
 The following cases are not supported by Comet:
 
 - Nested arrays with `Struct` or `Null` child values are not supported natively and will fall back to Spark.
+
 <!--END:EXPR_COMPAT-->

docs/source/user-guide/latest/compatibility/expressions/spark-4.1/datetime.md

@@ -136,4 +136,5 @@ The following cases are not supported by Comet:
 The following cases are not supported by Comet:
 
 - Only `TimestampType` and `DateType` inputs are supported. `TimestampNTZType` is not supported because Comet incorrectly applies timezone conversion to TimestampNTZ values.
+
 <!--END:EXPR_COMPAT-->

docs/source/user-guide/latest/compatibility/expressions/spark-4.1/map.md

@@ -48,4 +48,5 @@ The following incompatibilities cause `MapSort` to fall back to Spark by default
 The following cases are not supported by Comet:
 
 - MapSort is unsupported for non-scalar key types (struct, array, map, etc.).
+
 <!--END:EXPR_COMPAT-->

docs/source/user-guide/latest/compatibility/expressions/spark-4.1/math.md

@@ -38,4 +38,5 @@ The following cases are not supported by Comet:
 The following cases are not supported by Comet:
 
 - The `seed` argument must be a literal value
+
 <!--END:EXPR_COMPAT-->

docs/source/user-guide/latest/compatibility/expressions/spark-4.1/misc.md

@@ -56,4 +56,5 @@ The following incompatibilities cause `SortOrder` to fall back to Spark by defau
 The following cases are not supported by Comet:
 
 - Falls back to Spark when the input type cannot be cast to string.
+
 <!--END:EXPR_COMPAT-->

docs/source/user-guide/latest/compatibility/expressions/spark-4.1/string.md

@@ -94,4 +94,5 @@ The following differences from Spark are always present and do not require any a
 The following incompatibilities cause `StringTranslate` to fall back to Spark by default. Set `spark.comet.expression.StringTranslate.allowIncompatible=true` to enable Comet acceleration despite these differences.
 
 - DataFusion's translate iterates over Unicode graphemes (Spark uses code points) and substitutes U+0000 instead of treating it as a deletion sentinel
+
 <!--END:EXPR_COMPAT-->

docs/source/user-guide/latest/compatibility/expressions/spark-4.1/struct.md

@@ -36,4 +36,5 @@ The following incompatibilities cause `StructsToCsv` to fall back to Spark by de
 The following cases are not supported by Comet:
 
 - Complex types (arrays, maps, structs) in the schema are not supported
+
 <!--END:EXPR_COMPAT-->

docs/source/user-guide/latest/installation.md

@@ -92,6 +92,7 @@ Here are the direct links for downloading the Comet $COMET_VERSION jar file.
 - [Comet plugin for Spark 3.5 / Scala 2.13](https://repo1.maven.org/maven2/org/apache/datafusion/comet-spark-spark3.5_2.13/$COMET_VERSION/comet-spark-spark3.5_2.13-$COMET_VERSION.jar)
 - [Comet plugin for Spark 4.0 / Scala 2.13](https://repo1.maven.org/maven2/org/apache/datafusion/comet-spark-spark4.0_2.13/$COMET_VERSION/comet-spark-spark4.0_2.13-$COMET_VERSION.jar)
 - [Comet plugin for Spark 4.1 / Scala 2.13](https://repo1.maven.org/maven2/org/apache/datafusion/comet-spark-spark4.1_2.13/$COMET_VERSION/comet-spark-spark4.1_2.13-$COMET_VERSION.jar)
+
 <!-- ENDIF -->
 
 ## Building from source

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

@@ -2933,6 +2933,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] = {