[SPARK-40259][SQL] Support merging equivalent DataSource V2 scans in MergeSubplans#57360
Draft
peter-toth wants to merge 2 commits into
Draft
[SPARK-40259][SQL] Support merging equivalent DataSource V2 scans in MergeSubplans#57360peter-toth wants to merge 2 commits into
peter-toth wants to merge 2 commits into
Conversation
… package Relocate the MergeSubplans optimizer rule and its PlanMerger helper from sql/catalyst (org.apache.spark.sql.catalyst.optimizer) to a new sql/core package org.apache.spark.sql.execution.planmerging. The ScalarSubqueryReference and NonGroupingAggregateReference placeholder expressions stay in catalyst (MergeSubplansReferences.scala) because BloomFilterMightContain references them. Also move the existing sql/core PlanMergeSuite into the new package as PlanMergingSuite, and move MergeSubplansSuite alongside the rule.
peter-toth
force-pushed
the
SPARK-40259-dsv2-planmerger-support
branch
from
July 20, 2026 10:32
2129bdf to
d91b200
Compare
…MergeSubplans Add a generic, Spark-side merge of two DataSource V2 scans of the same table that differ only in projected columns and/or pushed filters, so subplans reading the same V2 table are fused into one scan (as already happens for V1 file sources). A source opts in with the new SupportsScanMerging marker; PlanMerger rebuilds the merged scan by driving the real V2ScanRelationPushDown, so the iterative PartitionPredicate pass runs and the (equal) strict filters come back fully enforced. DataSourceV2ScanRelation.pushedFilters now records the complete set of fully-pushed filters (referencing the relation's pre-pruning output), fixing a pre-existing bug where a filter on a pruned-out column silently vanished from the field. hasBlockingPushdown blocks the merge whenever a pushdown cannot be reproduced by rebuilding the scan: aggregate/join/limit/offset/top-N/sample, and now a fully-pushed non-deterministic filter (dropped from the deterministic-only pushedFilters, so it cannot be re-applied on the rebuilt scan).
peter-toth
force-pushed
the
SPARK-40259-dsv2-planmerger-support
branch
from
July 20, 2026 16:55
d91b200 to
dfab083
Compare
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.
What changes were proposed in this pull request?
This adds a generic, Spark-side merge of equivalent DataSource V2 scans in the
MergeSubplansrule. When two subplans read the same V2 table, their scans can be fused into a single scan. V1 file sources already get this (viaFileSourceStrategy), but every V2 scan was built independently, so the same table could be read many times.A source opts in with a new zero-method marker interface
SupportsScanMerging.PlanMergerfuses twoDataSourceV2ScanRelations only when all of the following hold:SupportsScanMerging;hasMergeBlockingPushdownflag onDataSourceV2ScanRelation);The two scans may still differ in their projected columns (the merged scan reads the union) and in best-effort / post-scan filters (those are OR-widened above the merged scan by the existing symmetric filter propagation in
MergeSubplans).When these hold,
PlanMergerrebuilds the merged scan by driving the realV2ScanRelationPushDown(through a newrebuildScanentry point) over the union of columns and the (equal) strict filters, and verifies each strict filter comes back fully enforced. Reusing the production pushdown end to end means the merged scan goes through the same filter translation, column pruning and iterativePartitionPredicatesecond pass, instead of reimplementing any of it.To make the merge sound, this PR also completes
DataSourceV2ScanRelation.pushedFilters. Previously it dropped fully-pushed filters that referenced a column pruned out of the scan output (e.g. an unselected partition column the source enforces internally). That was harmless before — the field's only active consumer was plan canonicalization, which already distinguishes scans by their builtScan— but the merge compares and re-enforces this set, so it needs to be complete.This is the generic alternative to #56264, which added scan merging per file format; here any V2 source gets it with a single trait.
Note on structure: the first commit is a pure move of
MergeSubplans/PlanMerger(and their suites) fromsql/catalystto a newsql/coreexecution.planmergingpackage — required because the merge now calls sql/core-only pushdown. The second commit is the feature. The first commit can be split into its own PR if that makes review easier.Why are the changes needed?
Under DataSource V2, subplans that read the same table each build their own scan, so the table is scanned once per subplan. The motivating case is TPC-DS q9, whose 15 scalar subqueries over
store_salescollapse to a single scan under V1 but read the table 15 times under V2. This closes that gap generically for any V2 source.Does this PR introduce any user-facing change?
No. The merge is opt-in through the new
SupportsScanMergingmarker, which no built-in source implements, so plans and results are unchanged for existing sources. ThepushedFilterscompleteness change is internal (it feeds plan canonicalization and the merge decision).How was this patch tested?
New tests:
MergeSubplansSuite(plan tests, comparing the full optimized plan): column-union merge, differing-filter OR-widen (2- and 3-way), identical-filter re-push, equal-strict re-push (2- and 3-way), and mixed equal-strict + differing-post-scan; plus the decline gates -- noSupportsScanMerging, a merge-blocking pushdown, reported key-grouped partitioning/ordering, a strict filter not re-enforced on rebuild, a fully-pushed non-deterministic filter, and a pushed limit. Also a negative canonicalization check (two scans with differentpushedFiltersmust not canonicalize equal).DSv2PlanMergingSuite(end-to-end with a real session): two scalar subqueries over aSupportsScanMergingsource whose partition filter is strict only via the iterative second pass fuse into one scan reading the union of columns (checkAnswerverifies correctness); and two subqueries with different partition filters are not fused.Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8)