Skip to content

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

Merged
mbutrovich merged 6 commits into
apache:mainfrom
mbutrovich:iceberg_reuse_fix
Jul 3, 2026
Merged

fix: prevent wrong results from Iceberg native scan exchange reuse with different pushed filters#4812
mbutrovich merged 6 commits into
apache:mainfrom
mbutrovich:iceberg_reuse_fix

Conversation

@mbutrovich

Copy link
Copy Markdown
Contributor

Which issue does this PR close?

Closes #4774.

Rationale for this change

Comet silently produces wrong results when the same Iceberg table is scanned more than once with different pushed-down filters (for example a FULL OUTER JOIN or UNION ALL of two differently-filtered reads). Spark's non-AQE ReuseExchangeAndSubquery keys exchange reuse off Exchange.canonicalized, which delegates to the scan's canonical form. CometIcebergNativeScanExec.equals/hashCode/doCanonicalize identified a scan by metadataLocation, output, serializedPlanOpt, and runtimeFilters only. None of these carry the pushed static filters, which live in the @transient originalPlan that canonicalization nulls out. Two scans of the same table+snapshot with different filters therefore canonicalize identically, and reuse collapses them into one, so one branch reads the other branch's data.

Vanilla Spark avoids this because BatchScanExec.equals compares scan.toBatch, and Iceberg's SparkBatch.equals compares table.name() plus SparkScan.hashCode(), which folds in pushed filters, snapshot, branch, and read schema. Spark also keeps the scan reference alive through doCanonicalize, so the fingerprint survives. Comet's V1 Parquet scan (CometNativeScanExec) is not affected because its filters are top-level partitionFilters/dataFilters fields that are already in equality and preserved by canonicalization.

What changes are included in this PR?

  • Add a non-transient scanHashCode: Int field to CometIcebergNativeScanExec, captured from scanExec.scan.hashCode() in the apply factory while the transient scan is still available.
  • Include scanHashCode in equals and hashCode, and carry it through doCanonicalize (it is the only distinguishing field left once originalPlan is nulled) and convertBlock.

How are these changes tested?

New test in CometIcebergNativeSuite: "exchange reuse must not collapse scans with different pushed filters (#4774)". It runs a UNION ALL of two aggregations over the same partitioned Iceberg table with different partition filters under non-AQE with spark.sql.exchange.reuse=true. It fails on main (branch B reuses branch A's exchange, yielding A twice and dropping B) and passes with this change. Beyond checkSparkAnswerAndOperator, it asserts both CometIcebergNativeScanExec nodes survive and that no ReusedExchangeExec collapsed the branches. The existing AQE DPP broadcast-reuse tests continue to pass, confirming legitimate reuse (identical scans still share an exchange) is preserved.

@mbutrovich mbutrovich changed the title fix: include scanHashCode in CometIcebergNativeScan identity to prevent incorrect exchange reuse fix: prevent wrong results from Iceberg native scan exchange reuse with different pushed filters Jul 3, 2026
@mbutrovich

Copy link
Copy Markdown
Contributor Author

@sandugood are you able to test with this branch? Thanks again for reporting the issue!

@sandugood

Copy link
Copy Markdown
Contributor

Going to test it right now, thanks for quick response!

@mbutrovich
mbutrovich requested a review from andygrove July 3, 2026 14:27
@mbutrovich mbutrovich self-assigned this Jul 3, 2026
@mbutrovich mbutrovich added this to the 1.0.0 milestone Jul 3, 2026
Comment thread spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala Outdated

@andygrove andygrove left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

LGTM pending CI. Thanks @mbutrovich

@sandugood

Copy link
Copy Markdown
Contributor

Tested against this branch on the same pipeline. Got same exact results for both default Spark and Spark+Comet. Thanks for a great fix, @mbutrovich

@mbutrovich

Copy link
Copy Markdown
Contributor Author

Tested against this branch on the same pipeline. Got same exact results for both default Spark and Spark+Comet. Thanks for a great fix, @mbutrovich

Great to hear! Thank you for confirming and the quick correspondence while working these issues.

@mbutrovich
mbutrovich merged commit 920f295 into apache:main Jul 3, 2026
69 checks passed
@mbutrovich
mbutrovich deleted the iceberg_reuse_fix branch July 3, 2026 16:52
mbutrovich added a commit that referenced this pull request Jul 6, 2026
* fix: prevent wrong results from Iceberg native scan exchange reuse with different pushed filters (#4812)

(cherry picked from commit 920f295)

* run prettier
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

Comet silently produces wrong results while skipping shuffle

3 participants