fix: keep unconverted DPP filters in a scan's canonical form

[dwsmith1983]

Which issue does this PR close?

Closes #6264. Part of #6133 (the q64 row loss).

Rationale for this change

CometScanUtils.filterUnusedDynamicPruningExpressions dropped a DPP filter from a scan's canonical form when its subquery was still the adaptive placeholder, not only when it had become TrueLiteral. AQE canonicalizes a query stage from its exchange as it was before the stage optimizer rules ran, which is before CometPlanAdaptiveDynamicPruningFilters converts the placeholder. So an exchange above that stage saw a scan with no DPP filter. Two such exchanges over scans of the same table with different DPP filters compared equal, and AQE reused the first for both. One branch then read the other's rows. TPC-DS q64 builds the same store_sales join for 1999 and 2000, which is where #6133 loses its rows.

The extra stripping came with #4112 so that otherwise identical scans could share a stage while their DPP was still unconverted. That reuse does not need it. The quick stageCache lookup uses the exchange before optimization, but createQueryStages checks the cache again with newStage.plan.canonicalized after the stage rules have run, and by then the unused filter is TrueLiteral and is dropped as in Spark.

What changes are included in this PR?

• filterUnusedDynamicPruningExpressions drops only DynamicPruningExpression(TrueLiteral), matching FileSourceScanExec. It is shared by CometNativeScanExec, CometScanExec and CometIcebergNativeScanExec.
• CometNativeScanExec.doCanonicalize drops every DPP filter from originalPlan. No DPP rule rewrites originalPlan, so its filters are a stale copy. The scan's DPP identity is in its top-level partitionFilters. Without this, the SPARK-32509 test ("unused DPP filter and exchange reuse") stops reusing, because the stale placeholder in originalPlan keeps the scan apart from its twin.

One reuse goes away, as in Spark: a parent exchange over two scan stages whose DPP later becomes TrueLiteral is no longer shared, because its key was fixed while the placeholder was still there. On TPC-DS at SF1 with AQE (and AQEPropagateEmptyRelation excluded, so queries that return no rows on this data keep their plan shape), the final plans of q14a, q14b, q23a, q23b, q24a and q24b have the same number of ReusedExchange and ReusedSubquery nodes before and after this change. q64 has one more ReusedExchange and keeps both store_sales DPP scans, where main drops one of them.

How are these changes tested?

New tests in CometExecSuite run a UNION ALL of one CTE joined to two different dimension filters, with AQE on and coalescing off, and check the answer against Spark. On Spark 3.5 and later, where these scans run AQE DPP in Comet, they also check that each branch keeps its own DPP scan pruned to 1 and 3 partitions and that no ReusedExchange sits over a DPP scan:

• a broadcast over a sort-merge join of the DPP scan's shuffle stage, the q64 shape. Main returns 100 rows where Spark returns 400.
• a shuffle over the DPP scan's aggregate stage. Main returns 7 rows on Spark 3.5 and 21 on 4.1, where Spark returns 28.
• the dimension join in the scan's own stage, which passes on main and guards the scan stage path.

CometIcebergNativeSuite gets the q64 shape for the Iceberg native scan. Main returns 100 rows where Spark returns 400 on Spark 3.5 and 4.1, and a wrong answer on 3.4. On 3.4 the Iceberg scan stays in Comet with Spark's DPP rule planning its filter, so this path was exposed there too.

With the change, the new tests pass on Spark 3.4, 3.5, 4.0 and 4.1. On 3.5, CometExecSuite, the DPP fallback suites, the TPC-DS plan stability suites and CometIcebergNativeSuite pass. Reverting the helper change fails the new tests. Reverting the originalPlan change fails the SPARK-32509 test.

[sunchao]

Summary

• Prior state and problem: Canonicalization discarded unresolved dynamic partition pruning filters, allowing AQE to reuse exchanges across differently pruned scans and lose rows.
• Design approach: Preserve unresolved DPP in the shared scan helper and remove stale DPP copies only from CometNativeScanExec.originalPlan during canonicalization.
• Correctness / compatibility analysis: The helper matches Spark’s V1 and V2 scan behavior in 3.4.3, 3.5.9, 4.0.4, 4.1.3 and 4.2.0. AQE retains pre-optimization stage identities and separately checks optimized plans for reuse, supporting both parts of this fix.
• Key design decisions: Top-level partitionFilters retain the active pruning identity. originalPlan still contributes schema, projection and other scan identity. The change simplifies the shared helper without adding an abstraction.
• Implementation sketch: Two production changes plus three Parquet regression cases and one Iceberg case cover parent broadcast, parent shuffle and same-stage joins.
• Behavioral changes worth calling out: Differently pruned branches retain separate exchange identities. Some parent exchanges remain separate even when pruning later becomes unused, matching Spark’s conservative behavior. No runtime benchmark was performed.
• Suggested improvements: None meeting the P1/P2 reporting threshold. No introduced P1/P2 issues found within this review.

Reviewed the complete four-file diff from 605051ad239ef704f5f25d67910a446a6b6d7c70 to ae530c268d6a7617f851308143a1a917bec6764c. The PR remains non-draft. Snapshot and live discussions contain no reviews, issue comments or review threads. Routed skill: .ai/skills/review-comet-pr/SKILL.md. No specialized sibling skill applies.

Validation: Compiled the exact-head helper and Comet adaptive placeholder class against Spark 4.1.3. A standalone probe confirmed distinct scan and parent-exchange keys for different DPP filters, reproduced their collapse with the base helper, and verified reuse after DPP becomes TrueLiteral. Spark baseline executions of the three added Parquet query shapes returned 400, 28 and 400 rows. These checks do not execute Comet’s native pipeline.

Exact-head CI: Comet CI, CodeQL and the title workflow report action_required, with no executed checks. Only labeling passed. This checkout lacks built Comet native artifacts, so full Comet, Spark SQL and Iceberg suites were not run locally. Their integration verdict remains outstanding.

[dwsmith1983]

@andygrove could you approve a CI run on 48708be48? CI has not run on this PR yet.

[dwsmith1983]

@sunchao could you approve a CI run on b3f4f058b? CI has not run here yet.

[Neuw84]

We tested this PR together with #6268 on TPC-DS at SF1000: Parquet on S3, a Spark 4.1 cluster on EKS, 8 executors x 13 cores on m5.4xlarge, one availability zone, AQE on with 300 shuffle partitions.

The build was main b58b2f3a with #6268 (588c029f) and #6270 (b3f4f058) merged on top (HEAD 0f22d064), with the native library built for x86-64-v3. Comet ran with native scan, exec and shuffle enabled.

q64 now returns every row. It gives 12,185 rows, the same as vanilla Spark, with the same checksum (bce69949). It ran in 63.8 s with 249/249 operators on Comet. Before this fix, Comet lost rows on q64 at this scale (#6264).

All 103 queries complete, and row counts match vanilla Spark on every query. Checksums match on all but q65, which has ties in its ORDER BY and differs the same way for every engine we compare.

Thanks for the fix.

[dwsmith1983]

Thanks @Neuw84 for running this with #6268 at SF1000. Good to see q64 back to 12,185 rows and the full run matching Spark.

[Neuw84]

If curious the details are here:

https://vecruntime.github.io/vecruntime/benchmarks/tpcds-1tb.html

The plan is also have a more hybrid approach where I can customize operators, but I need more time on that.

Other than that seems like the arrowflight columnar shuffle is quite performant, maybe can be also be used as inspiration for comet ( on vecruntime I can also plug comet columnar shuffle). I know that there is some discussions on a new shuffle!

[andygrove]

I would like to do a deeper review of this PR before it is merged.

[sunchao]

Summary

• Prior state and problem: Canonicalization discarded unresolved dynamic partition pruning filters, allowing AQE to reuse exchanges across differently pruned scans and lose rows.
• Design approach: Preserve unresolved DPP in the shared helper. Remove stale DPP copies from CometNativeScanExec.originalPlan, retaining active pruning identity in top-level partitionFilters.
• Correctness / compatibility analysis: This matches Spark’s V1/V2 scan canonicalization in 3.4.3, 3.5.9, 4.0.4, 4.1.3 and 4.2.0. Their AQE implementations retain pre-optimization stage identities and separately check optimized plans for reuse.
• Key design decisions: originalPlan continues to distinguish schema, projection and static filters. The change simplifies the helper without introducing another abstraction or changing runtime filter evaluation.
• Implementation sketch: Two production changes plus three Parquet regression cases and one Iceberg case cover parent broadcasts, parent shuffles and same-stage joins. Tests compare results against Spark and inspect pruning/reuse.
• Behavioral changes worth calling out: Against branch-1.1, this intentionally fixes incorrect exchange reuse. Some parent exchanges remain separate when DPP later becomes unused, matching Spark’s conservative behavior. No runtime performance benchmark was performed.
• Suggested improvements: None meeting the reporting threshold. No introduced P1/P2 issues found within this review.

Reviewed the complete four-file diff from fef94f6cd78b18151dff57b7a936798385356de5 to d07859be8cd18a02a331f2333a968ec9c9b28c5a. The PR remains non-draft. Read the existing review, conversation and thread metadata. No substantiated unresolved P1/P2 review concerns were identified. Routed skill: .ai/skills/review-comet-pr/SKILL.md. No specialized sibling skill applies.

Validation: Compiled the exact-head helper and adaptive placeholder class against Spark 4.1.3. All 16 bounded canonicalization checks passed, reproducing the base helper’s collapse of distinct scan/parent-exchange keys and confirming distinct keys after the fix, equivalent-filter reuse, unused-filter removal and preservation of static filters.

Exact-head CI: Comet CI and CodeQL report action_required. Comet CI has no executed jobs. Only labeling passed.

Validation limits: The probe does not execute Comet’s native pipeline. This checkout has no built native artifacts, so the Comet, Spark SQL and Iceberg integration suites were not run locally. Their verdict remains outstanding. The discussion’s successful SF1000 run used an older combined build.

[andygrove]

The "Plan Identity and Exchange Reuse" section of docs/source/contributor-guide/adding_a_new_operator.md (lines 373-375, added in #6352 after this PR was opened) still says CometNativeScanExec.doCanonicalize canonicalizes originalPlan while removing unused dynamic pruning filters. After this change it strips every DPP filter from originalPlan, and the scan's DPP identity comes from the top-level partitionFilters, where only DynamicPruningExpression(TrueLiteral) is dropped. That second part is the rule someone writing a new scan needs to copy, since dropping an unconverted placeholder is what caused #6264. Could you update that paragraph to describe both?

Otherwise this matches how Spark's FileSourceScanExec and BatchScanExec canonicalize in 3.4 through 4.2. I ran the new tests against main (100 vs 400 rows and 7 vs 28 on 4.1, and 100 vs 400 for the Iceberg test on 3.4) and Spark's DynamicPartitionPruningSuite with Comet on 4.1. I also tried a few extra shapes: the same filter in both branches, DPP that ends up unused, a branch with no DPP, and AQE off. Answers and reuse counts matched vanilla Spark in all of them.

[sunchao]

Summary

• Prior state and problem: Canonicalization discarded unresolved dynamic partition pruning filters, allowing AQE to reuse exchanges across differently pruned scans and return incorrect rows.
• Design approach: Preserve unresolved DPP in the shared scan helper. Strip stale DPP copies from CometNativeScanExec.originalPlan, retaining active pruning identity in top-level partitionFilters.
• Correctness / compatibility analysis: The helper matches Spark’s V1/V2 scan canonicalization in 3.4.3, 3.5.9, 4.0.4, 4.1.3 and 4.2.0. Their AQE implementations preserve pre-optimization stage identities and separately check optimized plans for reuse.
• Key design decisions: originalPlan continues to distinguish schema, projection and static filters. Removing the placeholder-specific exceptions simplifies the implementation without adding an abstraction or changing runtime filter evaluation.
• Implementation sketch: Two production methods change. Three Parquet regression cases and one Iceberg case cover parent broadcasts, parent shuffles and same-stage joins. The documentation now explains both canonicalization rules, addressing the existing review concern.
• Behavioral changes worth calling out: Against branch-1.1, this intentionally prevents incorrect exchange reuse. Some parent exchanges remain separate when DPP later becomes unused, matching Spark’s conservative behavior. Runtime performance was not benchmarked.
• Suggested improvements: None meeting the reporting threshold. No introduced P1/P2 issues found within this review.

Reviewed the entire five-file diff from 965c8bbe289ff850b614c8c3833ed7802134115b to da0a37921c4acf7e3f02228530871aa4a87ccce1. The PR remains non-draft. Read all existing reviews and issue comments, and confirmed there are no inline comments or review threads. No substantiated unresolved P1/P2 concerns remain. Routed skill: .ai/skills/review-comet-pr/SKILL.md. No sibling skill applies.

Validation: Freshly compiled the exact-head helper and adaptive-placeholder class against Spark 4.1.3. All 16 bounded canonicalization checks passed, reproducing the base helper’s collapse of distinct scan/parent-exchange identities and confirming separation at head, equivalent-filter identity, unused-filter removal and static-filter preservation. Verified the relevant Spark source files against upstream tags for all five supported profiles.

Exact-head CI: Comet CI and CodeQL report action_required. Comet CI has no executed jobs. Only labeling passed.

Validation limits: The probe does not execute Comet’s native pipeline. This checkout has no built native artifacts, so full Comet, Spark SQL and Iceberg suites were not run locally. Their exact-head integration verdict remains outstanding. The successful integration and benchmark runs reported in the discussion predate this SHA.

[dwsmith1983]

Could you update that paragraph to describe both?

Updated in da0a379. The paragraph now says doCanonicalize strips every DPP filter from the stale originalPlan copy, and that the scan's DPP identity comes from the top-level partitionFilters, where filterUnusedDynamicPruningExpressions drops only DynamicPruningExpression(TrueLiteral), as FileSourceScanExec does. It points a new scan at that second rule and says why: a filter that still holds the adaptive broadcast placeholder has to stay, because AQE canonicalizes a query stage as its exchange was before the placeholder was converted.

[sunchao]

Summary

• Prior state and problem: Canonicalization discarded unresolved dynamic partition pruning filters, allowing AQE to reuse exchanges across differently pruned scans and return incorrect rows.
• Design approach: Preserve unresolved DPP in the shared scan helper. Remove stale DPP copies from CometNativeScanExec.originalPlan, retaining active pruning identity in top-level partitionFilters.
• Correctness / compatibility analysis: The filtering rule matches Spark’s V1/V2 scans in 3.4.3, 3.5.9, 4.0.4, 4.1.3 and 4.2.0. Their AQE implementations retain pre-optimization stage identities and separately check optimized plans for reuse, supporting both changes.
• Key design decisions: originalPlan retains schema, projection and static-filter identity. The helper becomes simpler, with no new abstraction or change to runtime filter evaluation.
• Implementation sketch: Two production methods change. Three Parquet regression cases and one Iceberg case compare results against Spark and inspect pruning/reuse. The documentation update addresses the existing review concern.
• Behavioral changes worth calling out: Compared with branch-1.1, this intentionally prevents incorrect exchange reuse. Some parent exchanges remain separate when DPP later becomes unused, matching Spark’s conservative behavior. Runtime performance was not benchmarked.
• Suggested improvements: None meeting the reporting threshold. No introduced P1/P2 issues found within this review.

Reviewed the entire five-file diff from 965c8bbe289ff850b614c8c3833ed7802134115b to da0a37921c4acf7e3f02228530871aa4a87ccce1. Confirmed the PR remains non-draft. Read all existing reviews and issue comments, and confirmed there are no inline comments or review threads. No substantiated unresolved P1/P2 concerns remain. Routed skill: .ai/skills/review-comet-pr/SKILL.md. No specialized sibling skill applies.

Validation: Freshly compiled the exact-head helper and adaptive-placeholder class against Spark 4.1.3. All 16 bounded canonicalization checks passed, reproducing the base helper’s collapse of distinct scan/parent-exchange identities and confirming separation at head, equivalent-filter reuse, unused-filter removal and static-filter preservation. Verified relevant Spark sources against upstream tags for all five supported profiles.

Exact-head CI: Comet CI and CodeQL report action_required. Comet CI has no executed jobs. Only labeling passed.

Validation limits: The probe does not execute Comet’s native pipeline. This checkout has no built native artifacts, so Comet, Spark SQL and Iceberg integration suites were not run locally. Their exact-head verdict remains outstanding. Successful integration and benchmark runs reported in the discussion predate this SHA.

[sunchao]

Summary

• Prior state and problem: Canonicalization discarded unresolved dynamic partition pruning filters, allowing AQE to reuse exchanges across differently pruned scans and return incorrect rows.
• Design approach: Preserve unresolved DPP in the shared scan helper. Remove stale DPP copies from CometNativeScanExec.originalPlan, retaining active pruning identity in top-level partitionFilters.
• Correctness / compatibility analysis: The filtering rule matches Spark’s V1/V2 scans in 3.4.3, 3.5.9, 4.0.4, 4.1.3 and 4.2.0. Their AQE implementations preserve pre-optimization stage identities and separately check optimized plans for reuse, supporting both changes.
• Key design decisions: originalPlan retains schema, projection and static-filter identity. Removing placeholder-specific exceptions simplifies the helper without adding an abstraction or changing runtime filter evaluation.
• Implementation sketch: Two production methods change. Three Parquet regression cases and one Iceberg case compare results against Spark and inspect pruning/reuse. The documentation now explains both canonicalization rules, addressing the existing review concern.
• Behavioral changes worth calling out: Against branch-1.1, this intentionally prevents incorrect exchange reuse. Some parent exchanges remain separate when DPP later becomes unused, matching Spark’s conservative behavior. Runtime performance was not benchmarked.
• Suggested improvements: None meeting the reporting threshold. No introduced P1/P2 issues found within this review.

Reviewed the entire five-file diff from 4ea367aab4af430fce5dafa84351bedd288ec074 to 08470610e0cf9edb719626fbea57d8069d4afa1a. Confirmed the PR remains non-draft. Read existing reviews and issue comments. The snapshot contains no inline comments or review threads. No substantiated unresolved P1/P2 concerns remain. Routed skill: .ai/skills/review-comet-pr/SKILL.md. No specialized sibling skill applies.

Validation: Freshly compiled the exact-head helper and adaptive-placeholder class against Spark 4.1.3. All 16 bounded canonicalization checks passed, reproducing the base helper’s collapse of distinct scan/parent-exchange identities and confirming separation at head, equivalent-filter identity, unused-filter removal and static-filter preservation. Verified relevant Spark source files against upstream tags for all five supported profiles.

Exact-head CI: Comet CI and CodeQL report action_required. Comet CI has no executed jobs. Only labeling passed.

Validation limits: The probe does not execute Comet’s native pipeline. This checkout has no built native artifacts, so full Comet, Spark SQL and Iceberg suites were not run locally. Their exact-head integration verdict remains outstanding. Successful integration and benchmark runs reported in the discussion predate this SHA.

[sunchao]

Summary

• Prior state and problem: Dropping unresolved dynamic partition pruning filters during canonicalization allowed AQE to reuse exchanges across differently pruned scans, returning incorrect rows.
• Design approach: Preserve unresolved DPP in the shared scan helper. Remove stale DPP copies from CometNativeScanExec.originalPlan, with active pruning identity retained in top-level partitionFilters.
• Correctness / compatibility analysis: The filtering rule matches Spark’s V1/V2 scans in 3.4.3, 3.5.9, 4.0.4, 4.1.3 and 4.2.0. AQE preserves pre-optimization stage identities and separately checks optimized plans for reuse, supporting both changes.
• Key design decisions: originalPlan retains schema, projection and static-filter identity. Removing placeholder-specific exceptions simplifies the helper without adding abstractions or per-row work.
• Implementation sketch: Two production methods change. Three Parquet regression cases and one Iceberg case compare answers against Spark and inspect pruning/reuse. The documentation update addresses the existing review concern.
• Behavioral changes worth calling out: Compared with branch-1.1, this intentionally prevents incorrect exchange reuse. Some parent exchanges remain separate when DPP later becomes unused, matching Spark’s conservative behavior. Runtime performance was not benchmarked.
• Suggested improvements: None meeting the reporting threshold. No introduced P1/P2 issues found within this review.

Reviewed the entire five-file diff from 9dc8c3ca89962506078831f7064a25c75da83883 to 20d7ef5f4a50fa9eac9b5297cca4764feb27129e. Confirmed the PR remains non-draft. Read existing reviews and issue comments. The snapshot contains no inline comments or review threads. No substantiated unresolved P1/P2 concerns remain. Routed skill: .ai/skills/review-comet-pr/SKILL.md. No specialized sibling applies.

Validation: Freshly compiled the current helper and adaptive-placeholder class against Spark 4.1.3. All 16 bounded canonicalization checks passed, reproducing the base helper’s collapse of distinct scan/parent-exchange identities and confirming separation at head, equivalent-filter identity, unused-filter removal and static-filter preservation. Verified relevant Spark sources against upstream tags for all five supported profiles.

Exact-head CI: Comet CI and CodeQL report action_required. Comet CI has no executed jobs. Only labeling passed.

Validation limits: The probe does not execute Comet’s native pipeline. This checkout lacks built native artifacts, so Comet, Spark SQL and Iceberg integration suites were not run locally. Their exact-head verdict remains outstanding. Successful integration and benchmark runs reported in the discussion predate this SHA.

[sunchao]

Summary

• Prior state and problem: Dropping unresolved dynamic partition pruning filters during canonicalization allowed AQE to reuse exchanges across differently pruned scans and return incorrect rows.
• Design approach: Preserve unresolved DPP in the shared scan helper. Strip stale DPP copies from CometNativeScanExec.originalPlan, keeping active pruning identity in top-level partitionFilters.
• Correctness / compatibility analysis: The filtering rule matches Spark’s V1/V2 scans in 3.4.3, 3.5.9, 4.0.4, 4.1.3 and 4.2.0. AQE preserves pre-optimization stage identities and separately checks optimized plans for reuse, supporting both changes.
• Key design decisions: originalPlan retains schema, projection and static-filter identity. Removing placeholder-specific exceptions simplifies the implementation without adding abstractions or per-row work.
• Implementation sketch: Two production methods change. Three Parquet regression cases and one Iceberg case compare answers against Spark and inspect pruning/reuse. The documentation now explains both canonicalization rules, addressing the existing review concern.
• Behavioral changes worth calling out: Compared with branch-1.1, this intentionally prevents incorrect exchange reuse. Some parent exchanges remain separate when DPP later becomes unused, matching Spark’s conservative behavior. Runtime performance was not benchmarked.
• Suggested improvements: None meeting the reporting threshold. No introduced P1/P2 issues found within this review.

Reviewed the entire five-file diff from 8c783aa88104616dcf0f7876b4f7a9ba71e2bf31 to 81a3421bfa32f4ae606017c3981d7ec79cf05242. Confirmed the PR remains non-draft. Read existing reviews and issue comments and confirmed there are no inline comments or review threads. No substantiated unresolved P1/P2 concerns remain. Routed skill: .ai/skills/review-comet-pr/SKILL.md. No specialized sibling skill applies.

Validation: Freshly compiled the exact-head helper and adaptive-placeholder class against Spark 4.1.3. All 16 bounded canonicalization checks passed, reproducing the base helper’s collapse of distinct scan/parent-exchange identities and confirming separation at head, equivalent-filter identity, unused-filter removal and static-filter preservation. Verified relevant Spark sources against upstream tags for all five supported profiles.

Exact-head CI: Comet CI and CodeQL report action_required. Comet CI has no executed jobs. Only labeling passed.

Validation limits: The probe does not execute Comet’s native pipeline. This checkout lacks built native artifacts, so Comet, Spark SQL and Iceberg integration suites were not run locally. Their exact-head verdict remains outstanding. Successful integration and benchmark runs reported in the discussion predate this SHA.