[SPARK-58549][SQL] Preserve key-grouped partitioning and ordering across a DSv2 scan merge by peter-toth · Pull Request #57753 · apache/spark

peter-toth · 2026-08-04T09:00:35Z

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)

…oss a DSv2 scan merge Follow-up to SPARK-40259. Instead of declining a DSv2 scan merge whenever either input reports key-grouped partitioning or ordering, allow the merge and re-derive the merged scan's own report, declining only if that would degrade what the inputs reported. - Drop the kGP/ordering conjuncts from the mergeable gate. - At the leaf, combine the two inputs' reports (remapped into the merged relation's attribute space) into the single report the merge must preserve: kGP must be equal, ordering is the stronger of the two. If the inputs are incompatible (differing non-empty kGP, or neither ordering satisfies the other) no rebuilt scan could keep both not-worse, so decline right there -- before rebuilding -- unless the matching config accepts degrading that dimension. - After rebuildScan, run V2ScanPartitioningAndOrdering on the single merged scan node to re-derive its partitioning/ordering, 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 (carried through DSv2DeferredScan for the deferred build). Gated per dimension by two new configs, default false: spark.sql.optimizer.mergeSubplans.dsv2ScanMerge.allowKeyGroupedPartitioningDegradation and ...allowOrderingDegradation. With defaults it is a pure improvement (merge when not worse, decline on degradation). - Tests: default-decline-on-degradation, default-decline-on-incompatible-inputs, and config-allows-degradation in MergeSubplansSuite; a preserves-kGP end-to-end test in DSv2PlanMergingSuite with a new reporting SCAN_MERGING fixture.

peter-toth · 2026-08-04T15:01:28Z

cc @LuciferYang

LuciferYang · 2026-08-04T15:47:28Z

I’ll take a look at it tomorrow.

dongjoon-hyun

Although I understand your intention, for new test cases, SPARK-58549 test prefix should be used according to this PR's JIRA instead of the previous one, @peter-toth .

dongjoon-hyun · 2026-08-04T16:49:34Z

@@ -169,4 +172,39 @@ class DSv2PlanMergingSuite extends QueryTest with SharedSparkSession
      }
    }
  }
+
+  test("SPARK-40259: a scan merge preserves the sources' reported key-grouped partitioning") {


Suggested change

test("SPARK-40259: a scan merge preserves the sources' reported key-grouped partitioning") {

test("SPARK-58549: a scan merge preserves the sources' reported key-grouped partitioning") {

dongjoon-hyun · 2026-08-04T16:49:49Z

@@ -2618,6 +2619,109 @@ class MergeSubplansSuite extends PlanTest {
    assertDeclines(s => s.copy(ordering = Some(Seq(SortOrder(s.output.head, Ascending)))))
  }

+  test("SPARK-40259: do not merge DSv2 scans reporting incompatible kGP/ordering") {


Suggested change

test("SPARK-40259: do not merge DSv2 scans reporting incompatible kGP/ordering") {

test("SPARK-58549: do not merge DSv2 scans reporting incompatible kGP/ordering") {

dongjoon-hyun · 2026-08-04T16:50:00Z

+    assertDeclines(s => s.copy(ordering = Some(Seq(SortOrder(s.output.head, Ascending)))))
+  }
+
+  test("SPARK-40259: merge DSv2 scans reporting kGP/ordering when the degradation config allows") {


Suggested change

test("SPARK-40259: merge DSv2 scans reporting kGP/ordering when the degradation config allows") {

test("SPARK-58549: merge DSv2 scans reporting kGP/ordering when the degradation config allows") {

dongjoon-hyun · 2026-08-04T16:50:10Z

+      SQLConf.MERGE_SUBPLANS_DSV2_ALLOW_ORDERING_DEGRADATION.key)
+  }
+
+  test("SPARK-40259: enforce the required report on the deferred under-Filter scan build") {


Suggested change

test("SPARK-40259: enforce the required report on the deferred under-Filter scan build") {

test("SPARK-58549: enforce the required report on the deferred under-Filter scan build") {

dongjoon-hyun · 2026-08-04T17:00:28Z

+    // stronger, which satisfies both). None from combine* means the inputs are INCOMPATIBLE -- no
+    // rebuilt scan could keep both not-worse -- so decline HERE, before rebuilding, unless the
+    // matching config accepts degrading that dimension.
+    val requiredKeyGroupedPartitioning = combineRequiredKeyGroupedPartitioning(


requiredKeyGroupedPartitioning / requiredOrdering seem to carry a double meaning here. These Options use .isEmpty to mean the inputs' reports are incompatible, while the same names are reused for the Seq parameters of tryBuildMergedDSv2Scan / mergeDegradesReporting, where empty means no requirement. Renaming the Options to combined* (and deriving the expected* Seqs from them) makes the early-decline condition read correctly at the use site. Maybe

requiredKeyGroupedPartitioning -> combinedKeyGroupedPartitioning

requiredOrdering -> combinedOrdering

dongjoon-hyun

+1 for the logic. LGTM with two minor issues in the above.

peter-toth · 2026-08-04T18:01:37Z

Thanks @dongjoon-hyun! Both applied — the SPARK-58549 test prefixes in 4016674, and requiredKeyGroupedPartitioning/requiredOrdering renamed to combined* at the tryMergeDSv2 use site. Good catch on the double meaning: the Options now read as "the combined report, None if incompatible", and the Seq parameters keep required* where empty means no requirement.

uros-b approved these changes Aug 4, 2026

View reviewed changes

dongjoon-hyun requested changes Aug 4, 2026

View reviewed changes

dongjoon-hyun reviewed Aug 4, 2026

View reviewed changes

Use the SPARK-58549 prefix for the new test cases

4016674

dongjoon-hyun reviewed Aug 4, 2026

View reviewed changes

Rename the combined kGP/ordering Options to combined* at the use site

02f5dbd

dongjoon-hyun approved these changes Aug 4, 2026

View reviewed changes

Navigation Menu

Uh oh!

[SPARK-58549][SQL] Preserve key-grouped partitioning and ordering across a DSv2 scan merge - #57753

[SPARK-58549][SQL] Preserve key-grouped partitioning and ordering across a DSv2 scan merge#57753
peter-toth wants to merge 3 commits into
apache:masterfrom
peter-toth:SPARK-58549-preserve-kgp-ordering-dsv2-scan-merge

peter-toth commented Aug 4, 2026

Uh oh!

peter-toth commented Aug 4, 2026

Uh oh!

LuciferYang commented Aug 4, 2026

Uh oh!

dongjoon-hyun left a comment

Uh oh!

dongjoon-hyun Aug 4, 2026

Uh oh!

dongjoon-hyun Aug 4, 2026

Uh oh!

dongjoon-hyun Aug 4, 2026

Uh oh!

dongjoon-hyun Aug 4, 2026

Uh oh!

dongjoon-hyun Aug 4, 2026

Uh oh!

dongjoon-hyun left a comment

Uh oh!

peter-toth commented Aug 4, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

	test("SPARK-40259: a scan merge preserves the sources' reported key-grouped partitioning") {
	test("SPARK-58549: a scan merge preserves the sources' reported key-grouped partitioning") {

	test("SPARK-40259: do not merge DSv2 scans reporting incompatible kGP/ordering") {
	test("SPARK-58549: do not merge DSv2 scans reporting incompatible kGP/ordering") {

	test("SPARK-40259: merge DSv2 scans reporting kGP/ordering when the degradation config allows") {
	test("SPARK-58549: merge DSv2 scans reporting kGP/ordering when the degradation config allows") {

	test("SPARK-40259: enforce the required report on the deferred under-Filter scan build") {
	test("SPARK-58549: enforce the required report on the deferred under-Filter scan build") {

Uh oh!

Conversation

peter-toth commented Aug 4, 2026

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Was this patch authored or co-authored using generative AI tooling?

Uh oh!

peter-toth commented Aug 4, 2026

Uh oh!

LuciferYang commented Aug 4, 2026

Uh oh!

dongjoon-hyun left a comment

Choose a reason for hiding this comment

Uh oh!

dongjoon-hyun Aug 4, 2026

Choose a reason for hiding this comment

Uh oh!

dongjoon-hyun Aug 4, 2026

Choose a reason for hiding this comment

Uh oh!

dongjoon-hyun Aug 4, 2026

Choose a reason for hiding this comment

Uh oh!

dongjoon-hyun Aug 4, 2026

Choose a reason for hiding this comment

Uh oh!

dongjoon-hyun Aug 4, 2026

Choose a reason for hiding this comment

Uh oh!

dongjoon-hyun left a comment

Choose a reason for hiding this comment

Uh oh!

peter-toth commented Aug 4, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants