Repository navigation
Conversation
…6270) * fix: keep unconverted DPP filters in a scan's canonical form AQE canonicalizes a query stage from its exchange before the stage rules convert DPP placeholders. Comet dropped a DPP filter from the scan's canonical form while it was still a placeholder, so exchanges over scans with different DPP filters compared equal and AQE reused one for both, losing the other branch's rows. The shared helper now drops only DynamicPruningExpression(TrueLiteral), as FileSourceScanExec does. CometNativeScanExec.doCanonicalize drops every DPP filter from originalPlan, a stale copy no DPP rule rewrites, so an unused DPP scan still matches its twin. Closes apache#6264. Part of apache#6133. * docs: describe how a native scan keeps its DPP identity when canonicalized The plan identity section still said CometNativeScanExec canonicalizes originalPlan while removing unused dynamic pruning filters. It now strips every DPP filter from that stale copy, and the scan's DPP identity comes from the top-level partitionFilters, where only DynamicPruningExpression(TrueLiteral) is dropped. Say both, and point a new scan at the second rule with the reason it matters for AQE reuse. * fix: strip every DPP filter from the Delta scan's originalPlan when canonicalizing CometDeltaNativeScanExec kept a stale DPP placeholder in its originalPlan copy once filterUnusedDynamicPruningExpressions stopped dropping placeholders, so after AQE turned the live filter into TrueLiteral the scan no longer matched its twin and lost SPARK-32509 exchange reuse. Drop every DPP filter from originalPlan, as CometNativeScanExec does; runtimeFilters still carry the scan's DPP identity. Adds a Delta version of the SPARK-32509 reuse test. * test: check that DPP scan stages on a Delta table are not reused across branches Ports the apache#6264 UNION ALL test to the native Delta scan, so a Delta scan whose live DPP filter differs from its twin's keeps its own shuffle stage and its own rows. * refactor: share the scan originalPlan canonicalization and DPP reuse check CometNativeScanExec and CometDeltaNativeScanExec now call CometScanUtils.canonicalizeOriginalPlan instead of each stripping DPP filters from originalPlan inline, and the operator guide points new scans at it. The no-reuse-over-a-DPP-scan assertion lives once in CometTestBase and the Comet, Iceberg and Delta suites use it. The CometExecSuite test that passed on main is replaced by one where both branches use the same DPP filter: the shuffles must still be reused as in Spark, which fails if the canonical form keeps too much. * test: share the reused exchange helpers in CometTestBase assertExchangeReuseOver and assertNoReusedExchangeOverDppScan now call one reusedExchangesOver, and assertSameExchangeReuseAsSpark replaces the reuse count comparison written out in CometExecSuite and CometDeltaNativeScanSuite. --------- Co-authored-by: Andy Grove <agrove@apache.org> (cherry picked from commit e242944) Adapted for branch-1.1: the Delta contrib (CometDeltaNativeScanExec and its suite) is not on this branch, so those two files are dropped, and so is the contributor guide change. CometScanUtils keeps only the FileSourceScanExec import, since apache#6753's FilePartition and NativeConfig code is not here.
github-actions
Bot
added
bug
Member
Author
|
@laxmanmamididd you may want this one too Is it helpful for us to pubish a 1.1.x patch release or are you doing your own builds? |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Which issue does this PR close?
Backport of #6270 to
branch-1.1. #6270 closed #6264 onmain; the issue stays closed.branch-1.2already has the fix, because #6270 merged before that branch was cut, so nobranch-1.2backport is needed (git merge-base --is-ancestor e242944ec4 apache/branch-1.2is true).Rationale for this change
With AQE on, two exchanges that differ only in the dynamic partition pruning filter of a scan below them compare equal, and AQE replaces the second with a
ReusedExchangeof the first. The consumer of the second exchange then reads the other branch's rows, and the rows it should have read are silently lost.branch-1.1has the bug:CometScanUtils.filterUnusedDynamicPruningExpressionsdrops any DPP filter whose subquery is still the adaptive placeholder. It does not affect only an opt-in path, since it needs only AQE, DPP and a native scan.This is not a 1.1.0 regression. The bug is in every release that has the filter helper. It is a silent wrong-result bug with
priority:critical, so it is a candidate under the backporting guide's "wrong results" rule.What changes are included in this PR?
Cherry-pick of e242944 with
-x. The shared helper now drops onlyDynamicPruningExpression(TrueLiteral), asFileSourceScanExecdoes, andCometNativeScanExec.doCanonicalizedrops every DPP filter from the staleoriginalPlancopy so an unused DPP scan still matches its twin.Adaptations, all recorded in the commit message:
contrib/delta-sparkis not onbranch-1.1, so the changes toCometDeltaNativeScanExecandCometDeltaNativeScanSuiteare dropped.adding_a_new_operator.mdchange is dropped. Onlydocs/source/user-guide/latest/is published from a release branch.CometScanUtilskeeps only theFileSourceScanExecimport. TheFilePartitionandNativeConfigimports in the source commit belong to fix: read each native scan partition through the object store its files live in #6753, which is not on this branch.The rest of the diff (
CometNativeScanExec,CometTestBase,CometExecSuite,CometIcebergNativeSuite) applied without changes.How are these changes tested?
The source PR's tests: two
AQE DPP ... not reused across branches (#6264)tests inCometExecSuiteand one inCometIcebergNativeSuite, plus a control inCometExecSuitethat scans with the same DPP filter are still reused.Run locally on this branch with Spark 3.5 and a
libcometbuilt from this branch (CometExecSuite,CometIcebergNativeSuite): 263 passed, 0 failed, 10 canceled. The canceled tests are unrelated version-gated ones (variant, TIME type, V3 deletion vectors).With
spark/src/mainreverted tobranch-1.1, the threenot reused across branchestests fail and the control passes. So the branch has the bug and the tests catch it.spotless:checkis clean.