-
Notifications
You must be signed in to change notification settings - Fork 284
feat: CometNativeScan per-partition plan data, add DPP [iceberg] #3446
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Open
mbutrovich
wants to merge
35
commits into
apache:main
Choose a base branch
from
mbutrovich:cometnativescan-dpp
base: main
Could not load branches
Branch not found: {{ refName }}
Loading
Could not load tags
Nothing to show
Loading
Are you sure you want to change the base?
Some commits from the old base branch may be removed from the timeline,
and old review comments may become outdated.
Conversation
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
# Conflicts: # spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala
….sql.execution.SubqueryExec
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.
Claude helped me with the PR description. After fact-checking a couple of hallucinations about tests that it thought were added, I think this is accurate now:
Which issue does this PR close?
Closes #3442.
Rationale for this change
Dynamic Partition Pruning (DPP) is an important optimization for star schema queries. Previously,
native_datafusionscans fell back to Spark when DPP was present. This PR adds full DPP support by deferring partition serialization to execution time, after DPP subqueries resolve.Absent of DPP optimizations, this also reduces the serialization overhead of scanning large amounts of Parquet data. Previously every Spark partition received every other partition's metadata about Parquet files to read.
What changes are included in this PR?
Architecture:
CometNativeScanExecnow defers partition serialization to execution time via lazyserializedPartitionDataCometNativeScan.convert()creates a placeholder operator with just ascan_idserializePartitions()resolves DPP subqueries and serializes the filtered partitionsoriginalPlan.partitionFiltersinstead ofpartitionFiltersbecause AQE'sPlanDynamicPruningFilterstransformsInSubqueryExec→Literal.TrueLiteralviamakeCopy, butoriginalPlanis not in the active plan tree and retains the original filtersConfig:
spark.comet.scan.dpp.enabled(default: true) replacesspark.comet.dppFallback.enabledCOMET_DPP_ENABLEDnative_datafusionCometNativeScanExecwith DPP (this PR)native_datafusionCometIcebergNativeScanExecwith DPP (#3349)autonative_iceberg_compatShims:
getDppFilteredFilePartitions()andgetDppFilteredBucketedFilePartitions()toShimCometScanExecfor Spark 3.4/3.5/4.0resolveSubqueryAdaptiveBroadcast()toShimSubqueryBroadcastfor DPP subquery resolutionOther:
equals/hashCodefromCometNativeScanExecin favor of case class defaults to prevent incorrect AQE exchange reuse between scans with different projectionsHow are these changes tested?
Comet tests:
New tests in
CometExecSuite:DPP with native_datafusion scan - join with dynamic partition pruning- verifies basic DPP with partition pruningDPP with native_datafusion scan - multiple partition columns- verifies DPP with two partition columnsDPP with native_datafusion scan - SubqueryExec (non-broadcast DPP)- verifies DPP works with non-broadcast subqueriesDPP with native_datafusion scan - ReusedSubqueryExec (subquery reuse)- verifies DPP works with reused subqueriesNew test in
CometIcebergNativeSuite:runtime filtering - DPP with non-broadcast join- verifies Iceberg DPP works with SubqueryExecSpark SQL tests:
After implementing, we had 24 Spark SQL test failures related to DPP. Updated Spark 3.5.8 diff now looks for CometNativeScanExec and properly pass the tests now.
We also had 22 test failures related to bucket scans with DPP. I modified the partitioning logic and tests pass, so I am confident that we're getting good test coverage out of Spark SQL tests.