Skip to content

[SPARK-40259][SQL] Support merging equivalent DataSource V2 scans in MergeSubplans#57360

Draft
peter-toth wants to merge 2 commits into
apache:masterfrom
peter-toth:SPARK-40259-dsv2-planmerger-support
Draft

[SPARK-40259][SQL] Support merging equivalent DataSource V2 scans in MergeSubplans#57360
peter-toth wants to merge 2 commits into
apache:masterfrom
peter-toth:SPARK-40259-dsv2-planmerger-support

Conversation

@peter-toth

@peter-toth peter-toth commented Jul 19, 2026

Copy link
Copy Markdown
Contributor

What changes were proposed in this pull request?

This adds a generic, Spark-side merge of equivalent DataSource V2 scans in the MergeSubplans rule. When two subplans read the same V2 table, their scans can be fused into a single scan. V1 file sources already get this (via FileSourceStrategy), 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. PlanMerger fuses two DataSourceV2ScanRelations only when all of the following hold:

  • they read the same relation — same table, catalog, identifier and options (compared by the relation's canonical form);
  • both scans opt in via SupportsScanMerging;
  • neither carries a pushdown that a rebuilt scan cannot reproduce — a pushed aggregate, join, variant extraction, limit, offset, sort/top-N, or table sample, or a fully-pushed non-deterministic filter (tracked by the new hasMergeBlockingPushdown flag on DataSourceV2ScanRelation);
  • neither reports key-grouped partitioning (storage-partitioned join) or an ordering — the rebuilt scan doesn't reconstruct these, so the merge is declined rather than silently dropping them (preserving them across a merge is left as a follow-up);
  • their fully-pushed (strict) filters are equal (merging scans whose strict filters differ is left as a follow-up).

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, PlanMerger rebuilds the merged scan by driving the real V2ScanRelationPushDown (through a new rebuildScan entry 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 iterative PartitionPredicate second 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 built Scan — 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) from sql/catalyst to a new sql/core execution.planmerging package — 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_sales collapse 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 SupportsScanMerging marker, which no built-in source implements, so plans and results are unchanged for existing sources. The pushedFilters completeness 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 -- no SupportsScanMerging, 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 different pushedFilters must not canonicalize equal).
  • DSv2PlanMergingSuite (end-to-end with a real session): two scalar subqueries over a SupportsScanMerging source whose partition filter is strict only via the iterative second pass fuse into one scan reading the union of columns (checkAnswer verifies 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)

… 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.
…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
peter-toth force-pushed the SPARK-40259-dsv2-planmerger-support branch from d91b200 to dfab083 Compare July 20, 2026 16:55
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant