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

[peter-toth]

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 (combineRequiredOrdering).

  Both compare reported expressions by transform semantics (TransformExpression.isSameFunction, the way a storage-partitioned join compares its two sides) rather than by canonical equality. V2ExpressionUtils binds a transform function afresh on every derivation, and canonicalName is a documented obligation (its default returns a random UUID to force an override) while equals on BoundFunction is required nowhere -- so for a connector meeting only the documented contract, canonical equality calls two identical bucket(4, id) reports different and declines the merge. Connectors that do define equals semantically are unaffected, but that is a courtesy rather than a contract.

  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 check runs at the build's callers, and per build attempt, so a report degradation stays distinguishable from a strict filter that cannot be re-enforced. 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.keyGroupedPartitioningDegradation.enabled
• spark.sql.optimizer.mergeSubplans.dsv2ScanMerge.orderingDegradation.enabled

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 ...Degradation.enabled=true, both for a single reporting input (the degradation is found after the rebuild) and for two incompatible ones (declined before it);
  • two independently derived bucket(4, a) reports merge and the rebuilt scan keeps the report -- the only test that reaches the transform comparison. It uses a FunctionCatalog fixture that binds a fresh BoundFunction per call, the way a real connector does; Spark's own UnboundBucketFunction returns a singleton, so a fixture built on that would pass with or without the fix;
  • the merge proceeds when the rebuilt scan re-derives the required ordering, and declines when it re-derives less than the two inputs' combined (stronger) ordering;
  • the deferred (under-Filter) build path enforces the required report too: decline by default, merge with the config on;
  • the deferred build checks the report on EACH attempt, so a source that reports only over the unpruned file set still merges on the strict-only attempt -- which is what the leaf builds when no Filter is above the scan, so checking only the first attempt would leave the deferred path weaker than the leaf path;
  • 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. It also asserts that the preserved partitioning reaches the physical plan, i.e. the merged BatchScanExec reports KeyedPartitioning on the partition column.

build/sbt 'sql/testOnly *MergeSubplansSuite *DSv2PlanMergingSuite *PlanMergingSuite' -- 99 tests pass. dev/lint-scala clean.

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

Generated-by: Claude Code (Opus 5)

[peter-toth]

cc @LuciferYang

[LuciferYang]

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]

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]

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]

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]

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]

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]

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.

[LuciferYang]

Sorry for the delayed reply.

[LuciferYang]

The not-worse check catches a report weaker than an input's, not one that appears when neither input had one.

• on a (c1,c2) table with the sides reading {c1,c3} and {c2,c3}, the merged scan gains Some([c1,c2]) and starts taking the KeyedPartitioning branch of replanWithRuntimeFilters: pruned splits get padded with None and launch empty tasks, and three SparkException invariants now apply.
• when the inputs conflict and the config is on, line 730 empties the requirement, so that dimension's post-rebuild check goes too.

[peter-toth]

Both halves are answered in the summary comment — short version: the gain is real and is now documented as deliberate, and emptying the requirement when the inputs conflict is a no-op, because that dimension's config being on already short-circuits mergeDegradesReporting.

[LuciferYang]

mergeDegradesReporting compares the expression list only, but SPJ also needs numPartitions and partitionKeys to match. once the .orElse at 831 drops the best-effort filter the split count grows, so the expression list survives while the key set changes. The check still returns false, an SPJ join no longer lines up, and the shuffle comes back.

[peter-toth]

Agreed, and documented on the method: only the expressions are compared because that is all DataSourceV2ScanRelation carries. Worth noting it isn't specific to the .orElse either — the first build can prune differently from either input too.

[LuciferYang]

The ordering dimension has kGP's root cause: orderingSatisfies goes to SortOrder.satisfies, then semanticEquals, which is canonicalized ==. for a source reporting sort(bucket(4, id)) the two binds differ, so both sides report the same ordering and the pair is still declined at the leaf.

[peter-toth]

Fixed, and this was the most valuable comment of the round. Both dimensions now compare via isSameFunction, recursively, so a nested transform's children are compared the same way. Note BoundFunction.canonicalName()'s default returns a random UUID, so this works exactly for connectors that override it — the same set for which SPJ works at all.

[LuciferYang]

The report-related declines moved from the read-only gate to behind rebuildScan, so each attempt costs one or two full rebuilds, and merge walks the whole cache per subplan.

• connector that must list files to answer numPartitions() pays file planning at optimizer time.
• the .orElse cannot tell the two None causes apart. A structural degradation fails again; a source reporting per pruned file set makes the retry succeed instead.

[peter-toth]

Fixed: the report check moved to the callers of tryBuildMergedDSv2Scan and runs per attempt, so a degradation no longer looks like a filter-enforcement failure. Separately, you're right that the retry itself has no test — deleting the .orElse leaves everything green. That predates this PR, so I've noted it as a follow-up rather than folding it in here.

[LuciferYang]

No test makes the merged scan carry a non-empty ordering, so the check at 877 never returns false with a non-empty requirement. replacing merged.ordering.getOrElse(Nil) with Nil stays green, and that switches merging off for every SupportsReportOrdering source. making combineRequiredOrdering pick the weaker side is also green, and would combine [a] and [a, b] into [a].

[peter-toth]

Confirmed by measurement — Nil in place of merged.ordering.getOrElse(Nil) left all 82 tests green. Two new tests fix that. On the weaker-side variant: that is green only for two non-empty orderings; flipping the empty-side branches along with it fails the existing single-side decline test.

[LuciferYang]

The comment above combineRequiredKeyGroupedPartitioning says kGP "must be equal (bucketing is a table property)", and on that reasoning else None is dead code. Line 712 says the two sides "usually agree ... but need not" and lists the reasons.

[peter-toth]

Reworded. Your deeper point stands too: differing pruning explains differing split counts and partition values, not differing key expressions, and only the expressions are compared — so the comment now points at the ordering dimension, where a per-scan report really can arise.

[LuciferYang]

The test is still named do not merge DSv2 scans that report key-grouped partitioning or ordering, and this PR removes that rule; the comment underneath now says the merge would degrade what the input reported.

[peter-toth]

Renamed to "do not merge DSv2 scans when the merge would degrade a reported key-grouped partitioning or ordering". Kept the SPARK-40259 prefix, since the test predates this PR.

[LuciferYang]

nit: The withSQLConf(V2_BUCKETING_ENABLED -> "true") here does nothing: it is true by default, only the physical outputPartitioning reads it, and every assertion is on optimizedPlan.

[peter-toth]

Correct. Rather than drop it I made it matter: the test now also asserts the merged scan reports KeyedPartitioning on c1 in the physical plan, with AQE off so executedPlan is walkable.

[LuciferYang]

These two open a new mergeSubplans.dsv2ScanMerge.* namespace and are the only booleans in the family without an .enabled suffix; the existing four sit under filterPropagation and all end in .enabled.

[peter-toth]

Fixed. I dropped allow at the same time — with it, the key came to 94 chars, one over the line limit, which would have made it the only key in SQLConf split across two literals and so unsearchable from a log.

[LuciferYang]

nit: InMemoryScanMergingReportingCatalog.createTable is the third copy of this body (the parent plus the sibling in this file), and capabilities() is copied verbatim too.

[peter-toth]

Fixed: a newScanMergingTable hook on the catalog, and InMemoryScanMergingPartitionFilterTable now extends the reporting table, so createTable and capabilities() each exist once.

[peter-toth]

No problem at all, and thanks — this review was greatly appreciated. It found real gaps, including a couple I'd have shipped. Everything is in 97d5247.

What changed

• Transform comparison (:858). This was the big one. TransformExpression is a plain case class, so canonicalized == compares the BoundFunction instance, and V2ExpressionUtils binds afresh on every derivation — so two identical bucket(4, id) reports compared unequal and every transform-partitioned source declined the merge. Both dimensions now compare with isSameFunction, recursively, so a nested transform's children don't fall back to the instance comparison either. One fixture trap worth flagging: Spark's UnboundBucketFunction.bind returns a singleton, so an in-repo bucket test passes with or without the fix — the new test uses a FunctionCatalog that binds a fresh instance per call, like a real connector, and it fails without the fix.
• Report checked per build attempt (:831, :806). The check moved out of tryBuildMergedDSv2Scan to its callers, and now runs per attempt. A degradation is no longer indistinguishable from a filter-enforcement failure, so the build's None again means only "strict filters not re-enforceable" — which restores the tryBuildFilterDSv2ScanChild contract you flagged — and a source that reports per scan still gets its chance from the strict-only attempt.
• Coverage (:877, :2622). You were right that the ordering check never fired with a non-empty requirement; I measured it — replacing merged.ordering.getOrElse(Nil) with Nil left all 82 tests green. TestV2Table can now report an ordering or a bucket partitioning, and there are four new tests: the two ordering ones, the bucket-transform one, and the incompatible-inputs-with-the-config-on case, which was indeed uncovered.
• Configs (:7297, :7298). Both renamed to end in .enabled, and both docs now mention the decline that happens before any rebuild. I dropped the redundant allow while renaming, because allowKeyGroupedPartitioningDegradation.enabled came to 94 chars — one over the line limit, so it would have been the only key in SQLConf split across two literals, and not greppable from a log.
• Comments and names (:724, :841, :2603). All fixed, plus two more the same pass turned up: the (Filter, Filter) call site still claimed strict filters were the only reason to decline, and DSv2DeferredScan's @params named the configs by their old keys.
• Nits (:179, :104). The withSQLConf was decorative as you say — rather than drop it I made it matter, by also asserting the merged scan reports KeyedPartitioning on the partition column in the physical plan (AQE off so the plan is walkable). The fixture is down to one createTable and one capabilities().

Where I'd push back a little

• :872, second half — emptying the requirement at line 730 is a no-op rather than a lost check: reaching getOrElse(Nil) with a None implies that dimension's config is on, and mergeDegradesReporting already short-circuits on it.
• :872, first half — a gained report is real (the partitioning pass is reference-subset guarded, so {c1,c3} + {c2,c3} does give the merged scan Some([c1,c2])), but I read it as the win rather than a hazard: the source reported that partitioning for the table, and an input dropped it only because a pruned column left it inexpressible. I've documented it so it reads as deliberate.
• :867 — agreed the check sees only the expressions, and that limitation is now written down. keyGroupedPartitioning is Option[Seq[Expression]], so split counts and partition values aren't visible at that layer, and planning input partitions in the optimizer to compare them seems worse than the gap. It isn't specific to the .orElse either — the first build can prune differently from either input too.
• :802 — a zero-key KeyGroupedPartitioning does now reach the merged scan as Some(Nil). But the two unmerged scans would each have carried the same Some(Nil) and collapsed the same way, so it's not worse than not merging — the old None was accidentally better than the original plan. Commented rather than changed.
• :2622, first half — I couldn't reproduce this one: setting the equality to false, and deleting the np-side mapAttributes, both fail DSv2PlanMergingSuite's new end-to-end test. The two subquery relations carry different exprIds, so the remap is what makes the two reports comparable at all. Both do stay green if MergeSubplansSuite runs alone.
• :877, second half — picking the weaker side is green only for two non-empty orderings; flipping the empty-side branches along with it fails the existing single-side decline test. The new test covers the non-empty case.

Two follow-ups I'd rather not fold in here

• TransformExpression's equality is the root cause and it reaches past this rule: PartitioningPreservingUnaryExecNode.projectKeyedPartitionings puts kGP expressions in an ExpressionSet (so dedup silently fails and the cross-product can hit aliasCandidateLimit), BatchScanExec.equals compares keyGroupedPartitioning directly, and DataSourceV2ScanRelation's canonical form does too, so PlanMerger.checkIdenticalPlans can miss two identical subqueries. Overriding equals/hashCode in terms of canonicalName() would fix all of them and let me delete the helper this PR adds. I'll file a JIRA.
• The strict-only retry in tryBuildFilterDSv2ScanChild is untested — deleting the .orElse leaves everything green — because the fixture is all-or-nothing on filters, so both attempts fail together. That predates this PR (SPARK-40259); it needs a source that pushes a batch or nothing.

---

Correction (2026-08-08). Nothing to change in the PR — the fix is right as it stands. But I dug deeper into the transform-comparison bullet above, and my justification for it was too strong. "Every transform-partitioned source declined the merge" isn't true: TransformExpression's equality delegates to BoundFunction.equals, and Iceberg's BaseScalarFunction overrides equals/hashCode in terms of canonicalName (apache/iceberg#9873), so Iceberg sources were never affected — for them this change is a no-op.

What's left is narrower but real. canonicalName is a documented obligation, its default a random UUID to force an override; equals is required nowhere. So a connector meeting only the documented contract does compare two identical reports unequal — Iceberg itself was in that state until 2024. The fix stands, with the reason being "don't depend on an undocumented courtesy" rather than "everything is broken today". The same assumption exists outside this rule — PartitioningPreservingUnaryExecNode's ExpressionSet dedup, BatchScanExec.equals, DataSourceV2ScanRelation's canonical form — which is what the TransformExpression equality JIRA will cover.

[LuciferYang]

LGTM , Thank you @peter-toth

[peter-toth]

Merge Summary:

• merged into master 8645e2c
• merged into branch-4.x c060e1a

Posted by merge_spark_pr.py

[peter-toth]

Thank you @uros-b, @dongjoon-hyun and @LuciferYang for the review!