From e147ee8225a2d1d65866d42869d63c111a49960f Mon Sep 17 00:00:00 2001 From: Matt Butrovich Date: Fri, 3 Jul 2026 10:13:20 -0400 Subject: [PATCH 1/4] add scanHashCode to equals for CometIcebergNativeScan for proper reuse --- .../comet/CometIcebergNativeScanExec.scala | 26 +++++++- .../comet/CometIcebergNativeSuite.scala | 64 +++++++++++++++++++ 2 files changed, 88 insertions(+), 2 deletions(-) diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/CometIcebergNativeScanExec.scala b/spark/src/main/scala/org/apache/spark/sql/comet/CometIcebergNativeScanExec.scala index e08cd5b29d..062542c1cb 100644 --- a/spark/src/main/scala/org/apache/spark/sql/comet/CometIcebergNativeScanExec.scala +++ b/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,6 +302,13 @@ 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 { @@ -303,14 +316,20 @@ case class 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, + scanHashCode) } object CometIcebergNativeScanExec { @@ -330,6 +349,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) diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala index a12c4ee137..46796dbf2d 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala @@ -3055,6 +3055,70 @@ class CometIcebergNativeSuite } } + // ---- non-AQE exchange reuse tests ---- + + 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, + s"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] = { From c1c16ea35df4d11fc704da6e3c162463d32ca1d0 Mon Sep 17 00:00:00 2001 From: Matt Butrovich Date: Fri, 3 Jul 2026 10:25:05 -0400 Subject: [PATCH 2/4] fix format issue --- .../test/scala/org/apache/comet/CometIcebergNativeSuite.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala index 46796dbf2d..039b039128 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala @@ -3108,7 +3108,7 @@ class CometIcebergNativeSuite // that no ReusedExchangeExec merged the branches. assert( collectIcebergNativeScans(cometPlan).length == 2, - s"Expected 2 CometIcebergNativeScanExec (one per branch) but found " + + "Expected 2 CometIcebergNativeScanExec (one per branch) but found " + s"${collectIcebergNativeScans(cometPlan).length}. Plan:\n$cometPlan") assert( collect(cometPlan) { case r: ReusedExchangeExec => r }.isEmpty, From c29106242551caadb850b4b688530120b57e960d Mon Sep 17 00:00:00 2001 From: Matt Butrovich Date: Fri, 3 Jul 2026 10:36:32 -0400 Subject: [PATCH 3/4] address PR feedback --- .../test/scala/org/apache/comet/CometIcebergNativeSuite.scala | 2 -- 1 file changed, 2 deletions(-) diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala index 039b039128..7a33f40eea 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala @@ -3055,8 +3055,6 @@ class CometIcebergNativeSuite } } - // ---- non-AQE exchange reuse tests ---- - test("exchange reuse must not collapse scans with different pushed filters (#4774)") { assume(icebergAvailable, "Iceberg not available") From ced1c1977f27df5ec2a7c7553494e19b1579db27 Mon Sep 17 00:00:00 2001 From: Matt Butrovich Date: Fri, 3 Jul 2026 10:49:23 -0400 Subject: [PATCH 4/4] fix scala 2.12 --- .../apache/spark/sql/comet/CometIcebergNativeScanExec.scala | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/CometIcebergNativeScanExec.scala b/spark/src/main/scala/org/apache/spark/sql/comet/CometIcebergNativeScanExec.scala index 062542c1cb..b262494833 100644 --- a/spark/src/main/scala/org/apache/spark/sql/comet/CometIcebergNativeScanExec.scala +++ b/spark/src/main/scala/org/apache/spark/sql/comet/CometIcebergNativeScanExec.scala @@ -329,7 +329,9 @@ case class CometIcebergNativeScanExec( output.asJava, serializedPlanOpt, runtimeFilters, - scanHashCode) + // 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 {