peter-toth opened a new pull request, #57753:
URL: https://github.com/apache/spark/pull/57753
### What changes were proposed in this pull request?
Follow-up to SPARK-40259 (subquery plan merge for DataSource V2 scans).
Today the DSv2 scan merge declines whenever either input scan reports
key-grouped partitioning or ordering. The rebuilt merged scan carries neither
on its own, so fusing the two scans could drop a partitioning/ordering the
original plan relied on and force an extra shuffle or sort. Declining is safe
but leaves the merge unused for any partitioned/ordered source.
This PR lets the merge proceed and re-derives the merged scan's own report
instead of declining up front:
- Drop the `keyGroupedPartitioning`/`ordering` conjuncts from the
`mergeable` gate in `PlanMerger.tryMergeScanRelations`.
- At the leaf, combine the two inputs' reports (np's remapped into cp's
relation space) into the single report the merge must preserve:
- key-grouped partitioning: the two must be equal
(`combineRequiredKeyGroupedPartitioning`);
- ordering: the stronger of the two, i.e. the one that satisfies the other
via `SortOrder.orderingSatisfies` (`combineRequiredOrdering`).
If the inputs are incompatible (differing non-empty kGP, or neither
ordering satisfies the other), no rebuilt scan can keep both not-worse, so the
merge is declined right there -- before rebuilding -- saving a scan rebuild.
- After `V2ScanRelationPushDown.rebuildScan`, re-derive the merged scan's
partitioning/ordering by running `V2ScanPartitioningAndOrdering` on the single
merged scan node, then check it against the combined required report
(`mergeDegradesReporting`): decline if the merged scan's kGP does not match, or
its ordering does not satisfy, what was required. The required report is
carried through `DSv2DeferredScan` for the deferred (under-`Filter`) build.
Two new opt-in configs gate accepting a degradation instead of declining
(both default `false`), so with the defaults this is a pure improvement --
merge when not worse, decline on loss:
-
`spark.sql.optimizer.mergeSubplans.dsv2ScanMerge.allowKeyGroupedPartitioningDegradation`
- `spark.sql.optimizer.mergeSubplans.dsv2ScanMerge.allowOrderingDegradation`
Only sources that declare the `SCAN_MERGING` table capability are affected.
### Why are the changes needed?
The parent feature (SPARK-40259) conservatively declines a merge whenever an
input reports key-grouped partitioning or ordering, to avoid forcing a
shuffle/sort that the original plan avoided. That is correct but overly broad:
when the rebuilt merged scan re-derives the same (or a stronger) report, fusing
the scans loses nothing and still removes a duplicate scan -- which matters for
partitioned/ordered sources, e.g. an SPJ join inside the merged subplan. This
PR keeps the safety (decline on real degradation) while capturing the merge
when it is genuinely not worse.
### Does this PR introduce _any_ user-facing change?
No. The two new configs are opt-in and default to `false`, and no built-in
source declares `SCAN_MERGING`, so there is no behavior change for existing
sources.
### How was this patch tested?
New and existing unit tests, all under `sql/core`:
- `MergeSubplansSuite`:
- default-decline when a single input reports kGP/ordering the rebuilt
scan drops (degradation);
- default-decline when the two inputs report incompatible kGP/ordering
(declined at the leaf, before rebuild);
- config-allows-degradation: the merge proceeds to the plain column union
under each `allow...Degradation=true`;
- the deferred (under-`Filter`) build path enforces the required report
too: decline by default, merge with the config on;
- the pre-existing empty-report case still merges.
- `DSv2PlanMergingSuite`: an end-to-end test that a scan merge preserves the
sources' reported key-grouped partitioning, using a new `SCAN_MERGING` fixture
(`InMemoryScanMergingReportingCatalog`/`InMemoryScanMergingReportingTable`)
that keeps its reported partitioning (no `NonReportingScan` wrapper), an
identity-partitioned table under `spark.sql.sources.v2.bucketing.enabled=true`,
and two scalar subqueries reading the partition column.
`build/sbt 'sql/testOnly *MergeSubplansSuite *DSv2PlanMergingSuite
*PlanMergingSuite'` -- 95 tests pass. `dev/lint-scala` clean.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
