[SPARK-40259][SQL] Support merging equivalent DataSource V2 scans in MergeSubplans#57360
[SPARK-40259][SQL] Support merging equivalent DataSource V2 scans in MergeSubplans#57360peter-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.
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).
d91b200 to
dfab083
Compare
LuciferYang
left a comment
LuciferYang
left a comment
There was a problem hiding this comment.
This framework is indeed more generic. Will the implementation for the built-in FileScan be handled as a follow-up?
| "the merged scan must keep the shared strict filter on a") | ||
| } | ||
|
|
||
| test("SPARK-40259: a pushed limit blocks merging (hasBlockingPushdown classification)") { |
There was a problem hiding this comment.
hasBlockingPushdown lists five merge-blocking pushdowns: limit, offset, sample, top-N, and the non-deterministic filter. The tests only exercise limit (and only as a flag assertion) plus the non-deterministic filter. Nothing covers pushedOffset, pushedSample, or sortOrders. If one of them ever falls out of the denylist, two scans that pushed a sample or top-N would fuse into one, the result would no longer be reproducible, and no test would fail. Could you add a flag assertion for offset/sample/top-N to match the limit one, and for at least one of them also assert that two such scans don't fuse? That way the whole flag-to-decline path is covered.
| case s: DataSourceV2ScanRelation => s | ||
| } | ||
|
|
||
| test("SPARK-40259: merge two DSv2 scans whose filter is strict only via the second pass") { |
There was a problem hiding this comment.
The Phase 2 OR-widen pruning (rePushMergedScan) only has the plan-shape assertions in MergeSubplansSuite. Both end-to-end tests in DSv2PlanMergingSuite use strict partition filters, so neither one goes through the path where the two post-scan filters differ and the scan gets OR-widened. I think this path is already fairly safe: what gets pushed is Or(np, cp), which is a superset, the Filter above re-checks exactness, and the non-deterministic case is caught by :2262. So this is a nice-to-have, not a blocker. If you want the end-to-end suite to cover it too, you could add a checkAnswer case in DSv2PlanMergingSuite with different post-scan filters (say c1 > 1 / c2 > 1), and throw in a rand() on one side while you're at it.
| s"expected the identical filter a > 1 to be re-pushed to the merged scan, got: $pushed") | ||
| } | ||
|
|
||
| test("SPARK-40259: merge three DSv2 scans with differing filters re-pushes the full OR") { |
There was a problem hiding this comment.
this test uses three distinct filters, so it only hits the new-alias branch of the tagged path. The reuse branch, where the third filter matches one of the earlier sides (existingNPFilter = Some, along with the rePushMergedScan it drives), never runs against a DSv2 scan. The general reuse test at :818 uses a LocalRelation, which doesn't drive rePushMergedScan. It'd be worth adding a 3-way DSv2 case where the third side's filter reuses one of the earlier ones, so the conditions collection in the reuse branch gets exercised too.
| // sides, so keep it unchanged. But if it sits above a merged DSv2 scan, recover the | ||
| // row-group pruning for its condition on that scan (Phase 2). | ||
| val prunedChild = rePushMergedScan(mergedChild, Some(cp.condition)) | ||
| val mergedPlan = Filter(cp.condition, prunedChild) |
There was a problem hiding this comment.
Regarding the MERGED_FILTER_TAG branch below(:349), a non-Project child throws IllegalStateException (this line was migrated in, not added by this PR, but the PR is what first lets a DSv2 scan reach it). Right now the tag is only ever set on a filter whose child is a Project, so the branch is unreachable. Still, since the merge is meant to be best-effort and should fall back to not merging on anything unexpected, throwing here instead of returning None is a bit fragile. Might be worth turning it into a fallback while you're in here.
| private def samePushedFilters( | ||
| np: DataSourceV2ScanRelation, | ||
| cp: DataSourceV2ScanRelation): Boolean = { | ||
| val cpOutputByName = cp.relation.output.map(a => a.name -> a).toMap |
There was a problem hiding this comment.
The output.map(a => a.name -> a).toMap idiom shows up three times, in samePushedFilters, tryMergeScanRelations, and buildMergedScan(:529、:583、:617）. A one-line private helper (something like byName(attrs)) would tidy it up. Pure cleanup.
2 participants
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)