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

[peter-toth]

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 TableCapability.SCAN_MERGING capability. Making it a table capability (rather than a marker on Scan) lets Spark decide before build(), and lets a V1-fallback source — whose Scan Spark wraps in an internal V1ScanWrapper — still opt in. 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 relations' tables declare the SCAN_MERGING capability;
• both scans are mergeable: each came out of the plain column-pruning + filter-pushdown path, carrying only pushdowns a rebuilt scan can reproduce. A non-reproducible pushdown — a pushed aggregate, join, variant extraction, limit, offset, sort/top-N, or table sample — makes a scan non-mergeable. This is tracked by a default-safe mergeableScan flag on DataSourceV2ScanRelation: it defaults to false and is granted only on the plain-scan path, so any other current or future build site is non-mergeable unless it explicitly opts in;
• neither reports key-grouped partitioning or an ordering — the rebuilt scan doesn't reconstruct these, so the merge is declined rather than silently dropping them (preserving them is a follow-up);
• each scan reads a subset of its relation's columns at the same type — a column read at a narrower type via nested-schema pruning carries extractor ordinals against that narrow layout, so merging it (the rebuilt scan reads the full type) would read the wrong field; such a pair is declined (widening to the union of nested fields is a follow-up);
• their fully-pushed (strict) filters are equal (differing-strict merge is a follow-up).

The two scans may still differ in projected columns (the merged scan reads the union) and in best-effort / post-scan filters (OR-widened above the merged scan by symmetric filter propagation).

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 and that the rebuilt scan is itself mergeable. Reusing the production pushdown end to end means the merged scan goes through the same filter translation, column pruning and iterative PartitionPredicate second pass. When the merged scan sits under a Filter, the build is deferred to that Filter so it's built once with the strict filters plus the best-effort OR pruning together. Mechanically, a small MergeContext threads down through the recursion, and TryMergeResult gains a dsv2DeferredScan field carrying the pending scan's union columns and strict filters back up to that Filter. A separate dsv2Merged fact propagates the "a DSv2 merge happened below" signal all the way up (past the innermost Filter that builds the scan), so a stacked outer Filter — e.g. a DPP or runtime-bloom Filter that PartitionPruning / InjectRuntimeFilter insert above the scan-adjacent one after pushdown — still recognizes the merge and can OR-widen.

Because a DSv2 merge's strict filters are equal on both sides, the best-effort OR-widen (which only broadens the enclosing Filter condition offered to the source for data skipping — the Filter above still guarantees correctness) is allowed under a new DSv2-specific config spark.sql.optimizer.mergeSubplans.filterPropagation.dsv2SymmetricFilterPropagation.enabled even when the general symmetric propagation is off. It ships default false for now; it can be flipped on in a later release once more sources have exercised the path.

To make the merge sound, this PR also completes DataSourceV2ScanRelation.pushedFilters, which previously dropped fully-pushed filters referencing a column pruned out of the scan output (e.g. an unselected partition column the source enforces internally); the merge compares and re-enforces this set, so it must be complete. This widens the field's effect slightly beyond the merge: pushedFilters participates in canonicalized, which backs sameResult, so two scans differing only in a filter over a pruned-out column are no longer sameResult. Physical reuse is unaffected (BatchScanExec.doCanonicalize doesn't consult the field), but CacheManager.lookupCachedData uses sameResult, so such a query may correctly stop hitting the cache — the previous behavior treated the two as equivalent, which was a wrong hit. DataSourceV2Suite pins this canonicalization inequality. Non-deterministic filters cannot reach the merge: they aren't pushed to V2 sources (SPARK-58207).

This is the generic alternative to #56264, which added scan merging per file format; here any V2 source gets it with a single capability.

Note on structure: the first commit is a pure move of MergeSubplans/PlanMerger (and suites) from sql/catalyst to a new sql/core execution.planmerging package — required because the merge now calls sql/core-only pushdown — and can be split into its own PR. The move does not change the rule's batch position (it was and is registered only in SparkOptimizer); because the rule's fully-qualified class name changed, a 4.2→4.3 migration-guide note is added (mirroring the DynamicJoinSelection rename), since spark.sql.optimizer.excludedRules matches by that name. The remaining commits are the feature, the build-deferral, the TableCapability opt-in, the outer-Filter dsv2Merged fact, and several self-review passes.

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 behavior change for existing sources: the merge is opt-in through the new TableCapability.SCAN_MERGING capability, which no built-in source declares, so plans and results are unchanged. This PR adds one new SQL config (spark.sql.optimizer.mergeSubplans.filterPropagation.dsv2SymmetricFilterPropagation.enabled, default false), which only affects sources that opt into SCAN_MERGING.

How was this patch tested?

New tests:

• MergeSubplansSuite (plan tests, comparePlans where the merge changes the plan): column-union merge; differing-filter OR-widen (2- and 3-way, including a reuse-branch case); a stacked outer-Filter pair fusing via the propagated dsv2Merged fact; identical-filter re-push; equal-strict re-push (2- and 3-way); mixed equal-strict + differing-post-scan; the DSv2 symmetric config gate (merges when it is on and the general one is off, not when both are off); the decline gates (no SCAN_MERGING capability, a merge-blocking pushdown, a column read at a nested-pruned type, reported key-grouped partitioning/ordering, a strict filter not re-enforced on rebuild); a mergeableScan classification check driving a pushed limit/offset/top-N/sample through the real pushdown; a non-deterministic pruning-predicate drop; a negative canonicalization check; and an invariant assertion that no placeholder DataSourceV2Relation survives a merge.
• DSv2PlanMergingSuite (end-to-end with a real session): two scalar subqueries whose partition filter is strict only via the iterative second pass fuse into one scan (checkAnswer verifies correctness); two subqueries with different partition filters are not fused; and two subqueries with differing best-effort filters OR-widen into one scan.

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

Generated-by: Claude Code (Opus 4.8)

[LuciferYang]

This framework is indeed more generic. Will the implementation for the built-in FileScan be handled as a follow-up?

[LuciferYang]

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.

[peter-toth]

Fixed in eaa81ce.

[LuciferYang]

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.

[peter-toth]

Fixed in eaa81ce.

[LuciferYang]

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.

[peter-toth]

Fixed in eaa81ce.

[LuciferYang]

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.

[peter-toth]

Fixed in eaa81ce.

[LuciferYang]

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.

[peter-toth]

Fixed in eaa81ce.

[dongjoon-hyun]

Just a question: Are you targeting Apache Spark 4.3.0 whose Feature Freeze is scheduled on 1st August, @peter-toth ?

[peter-toth]

Thanks for the reminder about the cut. I will try to conclude it Monday or Tuesday so maybe it might make it into the release, but not a must-have for 4.3.

[peter-toth]

Yes, built-in FileScan is a follow-up. The main thing needed is for the built-in file scans (ParquetScan, OrcScan, ...) to opt into SupportsScanMerging — the rebuild path already handles their pushdown: FileScanBuilder uses SupportsPushDownCatalystFilters, which the shared V2ScanRelationPushDown / PushDownUtils rebuild this PR reuses already supports. So it's mainly the opt-in, plus verifying the reproducibility contract holds for file sources. This lines up with your #56264, so happy to coordinate on the file-source side.

More broadly, DSv2 file sources have lagged V1 for a while — they still fall back to V1 by default (spark.sql.sources.useV1SourceList), and features like partition pruning were dropped in the early migration. So it makes sense to land the generic path first, which every real V2 connector benefits from, and treat the built-in file sources as a follow-up.

Somewhat related: I remember a comment on some PR listing the features DSv2 file sources still lack to reach V1 parity, but I can't find it again — do you happen to remember it or have a pointer? It'd help scope the file-source follow-up.

Also, I'm keeping this a draft for now — I'm not fully happy with the merge logic yet, so I'll iterate and come back to it in a day or two.

[peter-toth]

@LuciferYang this is ready for review now.

cc @cloud-fan as well.

[LuciferYang]

It’s late here. I’ll come back with feedback tomorrow.

[LuciferYang]

buildMergedScan builds projectList as columns.flatMap(a => relOutByName.get(a.name)), looking each column up by name in relation.output, which hands back the pre-pruning type. Fine for flat columns. For nested ones columns carries the pruned type like struct<b>, the projection substitutes struct<a,b,c>, and identifyRootFields uses att.dataType directly for a bare attribute, so the rebuilt scan reads the whole struct back and nested pruning stops working on the merge path. The worse part is the ordinal: the GetStructField above the scan was computed against the pruned schema, mapAttributes swaps the attribute but leaves the ordinal alone, and ordinal 0 of struct<a,b,c> is a where it used to mean b. So the read returns the wrong field. The cp side is not remapped by design (the comment on line 677), and the struct<b> type it holds no longer matches what the scan produces either. The post-check on lines 733-735 compares only the count and the names of scan.output, never dataType, so the widening passes silently. Two attributes sharing one exprId with different types should have been caught by hasUniqueExprIdsForOutput, but that walks each node's output and the mismatched reference sits in an intermediate Project. Both fixes I reached for first are too weak. Swapping the post-check for DataType.equalsStructurally misses the mixed case: cp selects s whole, np selects only s.b, columns is cp's struct<a,b,c>, the rebuilt type matches, the check passes, and np's ordinal 0 still reads a. Passing columns through as the projection does not work either, because unionAttrs is unioned by name (lines 652-655), so s.b against s.c keeps only cp's struct<b> and np then reads b. The smallest change I can defend is a pre-check in mergeable: decline the merge when two same-named attributes differ in dataType, or when any scan output type differs from relation.output, which is the signal that nested pruning is in play. Keeping the optimization would mean running StructType.merge over the union and re-deriving both sides' GetStructField ordinals against the merged schema, a much larger change. Could you also add a nested-column fixture? TestV2Table and InMemoryScanMergingPartitionFilterTable are both flat, so this path has no test coverage.

[peter-toth]

Declined via the new subset-at-same-type gate in mergeable, which rejects exactly this nested-pruned case (66eeda37ef5); I also removed the by-name projection so the merged scan projects relation attributes directly (1d8548db39c). Added the do not merge DSv2 scans when a column is read at a nested-pruned type fixture.

[LuciferYang]

mergeable checks hasMergeBlockingPushdown carefully on both input scans, but buildMergedScan validates only the strict filters, the column count and the column names on what comes back from the rebuild. It never re-checks the flag. Meanwhile rebuildScan runs the full apply, where all three buildScanWithPushed* rules set the flag to true and replace the exprIds, relying on an outer Project to map them back, and collectFirst throws exactly that wrapper away. Nothing reaches this today: the synthetic projection is bare attributes, the variant rule returns early because its only requested field targets VariantType and gets dropped, and ParquetScanBuilder, the single implementation of variant pushdown, produces a scan that does not implement SupportsScanMerging. It is still a quiet trap. Any future pushdown that fires on Project(attrs, Filter(cond, relation)) yields an accepted but wrong merged scan, and because the cp side is deliberately not remapped, the parent references dangle. Adding !scan.hasMergeBlockingPushdown to the filter on line 733 covers it.

[peter-toth]

Done in fef130de72b — added scan.mergeableScan to the post-rebuild filter in tryBuildMergedDSv2Scan.

[LuciferYang]

hasBlockingPushdown covers only the pruneColumns path. The aggregate, join and variant rules each hardcode true, and since the parameter defaults to false, forgetting it does not break the build. The tests happen to miss all three: MergeSubplansSuite.scala:2574 drives limit, offset, top-N and sample through the real pushdown, which is the helper path, and MergeSubplansSuite.scala:2051 fakes the flag with .copy() to exercise the consuming side. I deleted line 891 and everything still compiled. After that, two subqueries with a pushed aggregate fuse, the rebuild does not re-push the aggregate, and the plan expects aggregate columns while the scan hands back raw rows. The holder predicates those three rules match on are all readable from the helper, so folding the four terms into hasBlockingPushdown and calling it from every site makes the omission impossible. Separately, GroupBasedRowLevelOperationScanPlanning.scala:89 is a fourth production construction site that pushed the command condition and set neither field. It cannot reach the merge path today, since the same scan instance takes the identical branch first and row-level rewrites do not produce a scalar subquery or a non-grouping aggregate, but its builder comes from operation.newScanBuilder while rebuildScan calls table.asReadable.newScanBuilder, so it is not rebuildable by construction. Setting true there beats leaving the default.

[peter-toth]

Flipped the field to default-safe mergeableScan (default false, granted only on the plain-scan path) instead of folding the terms, so aggregate/join/variant, GroupBasedRowLevelOperationScanPlanning, and any future site are non-mergeable by default (d83d2c01eaa). Added positive and negative classification tests.

[LuciferYang]

This moves MergeSubplans from catalyst.optimizer to execution.planmerging, and Rule.ruleName is the fully qualified class name that the exclusion check compares as a string. The rule is in neither nonExcludableRules list. So anyone who turned the optimization off with spark.sql.optimizer.excludedRules=org.apache.spark.sql.catalyst.optimizer.MergeSubplans, possible since 4.2.0, finds the name no longer matches after upgrading. Unknown names are ignored without warning, so the optimization comes back on quietly. The repo already has a precedent for this exact situation: the DynamicJoinSelection rename at docs/sql-migration-guide.md:32, which even spells out "unknown names are silently ignored". Following that format in the 4.2 to 4.3 section should be enough. It is also worth noting in the PR description that the batch position did not change, since I checked and the rule was registered only in SparkOptimizer before the move as well. It reads like an ordering change otherwise.

[peter-toth]

Added a 4.2→4.3 migration-guide note (mirroring DynamicJoinSelection, noting unknown excludedRules names are silently ignored) (f7d4414f412); also noted in the PR description that the batch position is unchanged (registered only in SparkOptimizer before and after the move).

[LuciferYang]

The config doc says equal strict filters mean both sides read the same base rows, but the strict set comes out of filterNot(postScanFilterSet.contains), so a predicate the source hands back for a post-scan re-check never enters it, and line 115 of the same file says that overlap is normal (the parquet row group filter). A source that prunes on p and also returns p therefore reports an empty strict set on both sides, the sets compare equal, and the criterion holds while the property the doc claims does not. I first read this as IO amplification, but following it through, most cases recover: once Or(p='a', p='b') is rejected in the first pass it reaches createPartitionPredicates in the second, gets wrapped as a PartitionPredicate on the partition column and pushed, and the pruning comes back. What actually degrades is a source that is opt-in, partially pushes, cannot express OR, and has no second pass, which is a bounded regression. So the doc is the part that needs work: replace "read the same base rows" with what does hold, that the strict filters are equal and the pruning predicate is best effort with the enclosing Filter guaranteeing correctness. Whether the default should flip to false is your call; I would keep it true and fix the wording. I also withdraw the idea of requiring the post-scan filters to match. After early pushdown the Filter above the scan is the post-scan set, so that amounts to requiring the two Filter conditions to be equal, which lands in the identical branch and removes the feature.

[peter-toth]

Reworded the doc (dropped "read the same base rows" → the strict filters are equal and the pruning is best-effort, with the enclosing Filter guaranteeing correctness) (d8e600ec16e). On the default I went the other way from your lean and shipped it false for the freeze (14995973b04); we can flip it on in a later release. Thanks for withdrawing the post-scan-match idea.

[LuciferYang]

The reasoning in the invariant block, that stacked Filters are combined by CombineFilters before this rule runs, does not hold. CombineFilters itself does run earlier, inside operatorOptimizationRuleSet as part of super.defaultBatches. But PartitionPruning (SparkOptimizer.scala:62) and InjectRuntimeFilter (SparkOptimizer.scala:68) then each insert a Filter on top of an existing one, and both sit ahead of MergeSubplans (SparkOptimizer.scala:70), while the PushDownPredicates pass that would re-combine them waits for the next batch at SparkOptimizer.scala:73. So any query eligible for DPP or a bloom filter arrives here as Filter(DPP, Filter(cond, scan)). The mechanism is still safe (the inner arm consumes the deferral and returns deferredScan = None), just not for the reason the comment gives, and reading the recursion against that comment will mislead you. It also means the outer arm sees deferred = None, so the DSv2 exemption never fires and the merge is abandoned after a scan was already built, which makes DSv2 merging depend on how the Filters happen to stack. If that is intended, it is worth saying so in the comment.

[peter-toth]

Good catch on both. Fixed the comment (d8e600ec16e) — the stacked Filters here are DPP / InjectRuntimeFilter runtime filters inserted after pushdown, deferral consumed at the innermost pair. And I fixed the behavior you flagged (the outer arm seeing deferred = None so the exemption never fired): a new dsv2Merged fact on TryMergeResult propagates the merge signal past the innermost pair, so a stacked outer (Filter, Filter) pair recognizes it and OR-widens too, gated on the DSv2-symmetric config (5fb710d3a95). Added a synthetic stacked-Filter test.

[LuciferYang]

The invariant comment on MergeContext is genuinely informative, and walking it point by point I agree nothing is reachable today: the Aggregate, Join and single-sided Filter arms all reset filterAboveScan to false, and by the time (Filter, Filter) returns the deferral has been consumed, so those _ discards can only be None. The problem is that the whole arrangement rests on a comment with no assertion behind it (and one of its three claims does not hold, see the other note), in a file that has grown from 647 to 932 lines with a single 294-line tryMergePlans. If someone later adds an arm that forwards context unchanged, which is exactly what the three Project arms do, the resulting plan keeps a placeholder bare DataSourceV2Relation; the read path in DataSourceV2Strategy only matches DataSourceV2ScanRelation, so planning fails outright. And since merge()'s backstop is to decline silently, a regression shows up as merges quietly not happening, which no test that does not assert scan counts will notice. One line in the existing DSv2 tests, assert(plan.collectFirst { case _: DataSourceV2Relation => }.isEmpty), is cheap insurance. The more thorough option is pulling the DSv2 merge into its own file or class; it touches the generic recursion in eight places and it is the part that drags sql/core imports into what was a pure catalyst utility, which is what forced the cross-module move in the first place.

[peter-toth]

Added the cheap insurance — assertNoPlaceholderRelation (asserts no bare DataSourceV2Relation survives a merge) across the DSv2 tests (32f024130e2). Kept the DSv2 merge in the same file for now; the split into its own file is a reasonable follow-up if you feel strongly.

[LuciferYang]

The impact of no longer remapping pushedFilters in pruneColumns is a bit wider than "internal" suggests. canonicalized is what sameResult compares, so two scans differing only in a filter that references a pruned-out column were equivalent before and are not now. Physical reuse is unaffected, since ReuseExchangeAndSubquery compares SparkPlan.canonicalized and BatchScanExec.doCanonicalize never touches this field. But CacheManager.lookupCachedData uses plan.sameResult(cd.plan), so a query that used to hit the cache may stop hitting it. The flip side is that the previous behavior treated two scans with different p values as equivalent, which was a wrong hit, so the direction is right. DataSourceV2Suite.scala:1340 already pins the canonicalization inequality; what I would add is a note in the PR description covering this layer rather than calling it internal. Whether the cache side needs an end-to-end test is your call. I originally wanted an assertion that two subqueries are no longer sameResult, but that pins the behavior at the wrong layer since physical reuse never consulted the field, so I withdraw it.

[peter-toth]

Agreed — no code change here; I covered the CacheManager.sameResult layer in the PR description rather than calling it "internal" (physical reuse is unaffected, and the old behavior was a wrong hit, so the direction is right). DataSourceV2Suite:1340 already pins the canonicalization inequality. Thanks for withdrawing the sameResult assertion.

[LuciferYang]

mergeable gates on the reported keyGroupedPartitioning and ordering from both sides, on the grounds that the rebuild does not reconstruct them. I followed the rebuild path and confirmed rebuildScan runs only V2ScanRelationPushDown; V2ScanPartitioningAndOrdering is a separate entry in earlyScanPushDownRules and does not run again, so both fields are always None on the merged result. I suspected there was a case where both sides pruned the partition key, the union brought it back, and a reportable scan was lost, but I could not construct it: when the union lacks the key the reporting rule's own subsetOf(outputSet) check rejects it anyway, and when it has the key that side was already declined at the gate. So there is no defect here. I would just add a line to the tryMergeScanRelations comment saying rebuilt scans never carry reported partitioning or ordering, since the current wording reads as if this were only an input-side restriction.

[peter-toth]

Added a line to the tryMergeScanRelations comment noting the rebuilt scan never carries reported partitioning or ordering (so it's not only an input-side restriction) (d8e600ec16e). Thanks for confirming there's no defect.

[LuciferYang]

The [[org.apache.spark.sql.execution.planmerging. / PlanMerger]] link in the pushedFilters doc seems split across two lines, so scaladoc reads it as target ...planmerging. with label PlanMerger and cannot resolve it. CI will not go red, since SparkBuild.scala:1673 excludes this file from sbt unidoc; the effect is limited to IDE navigation and rendering. Only two other places in all of src/main are written this way, so it is not an established habit either. Putting [[...PlanMerger]] on one line, reflowing the preceding sentence if needed, fixes it. Alternatively use the short name in backticks the way MergeSubplansReferences.scala:25-26 does.

[peter-toth]

Fixed — put [[...PlanMerger]] on one line using the short backtick name, the way MergeSubplansReferences does (d8e600ec16e).

[peter-toth]

Thanks for the very thorough second pass, @LuciferYang — this was really helpful. I've pushed an update (through b4b5207) addressing all of it and refreshed the PR description. Summary, calling out where I went a different way than the literal suggestion:

Correctness

• Nested-pruned column (the struct<b> ordinal bug): declined via a new up-front gate in mergeable — each scan must read a subset of its relation's columns at the same type, which rejects exactly this case. I also removed the by-name projection entirely (see below), so the merged scan projects relation attributes directly. Added a nested-column decline fixture.
• Rebuilt scan not re-checked: added scan.mergeableScan to the post-rebuild filter.
• hasBlockingPushdown default-false footgun: rather than fold the four terms into it, I flipped the field to be default-safe — renamed mergeableScan, defaulting to false, granted only on the plain column-pruning path. Agg/join/variant, GroupBasedRowLevelOperationScanPlanning, and any future site are non-mergeable by default, so a forgotten site can't produce a wrong merge. Added positive+negative classification tests.
• byName duplicate-name mapping: removed name matching altogether. The np↔cp mapping is now positional (AttributeMap(np.relation.output.zip(cp.relation.output)), sound because the gate requires canonically-equal relations) keyed by exprId — no by-name lookup left to fool, and the samePushedFilters concern is covered too.

Opt-in placement: moved from the SupportsScanMerging Scan marker to a TableCapability.SCAN_MERGING. Fixes the V1-fallback problem you flagged, lets Spark decide before build(), and removes the wrapper-just-to-attach-the-marker in the fixture. Better to fix before the API ships — thanks.

Stacked Filters: fixed the comment (these are DPP / InjectRuntimeFilter runtime filters inserted after pushdown, not something CombineFilters would merge). And I fixed the behavior you flagged — the outer arm seeing deferred = None so the exemption never fired: a new dsv2Merged fact on TryMergeResult propagates past the innermost pair, so a stacked outer (Filter, Filter) recognizes the merge and OR-widens (gated on the DSv2 config). Added a stacked-Filter test. Invariant: added the cheap assertNoPlaceholderRelation insurance; kept the merge in-file for now (the split is a fair follow-up).

Config default: reworded the doc (dropped "read the same base rows"). On the default I went the other way from your lean and shipped it false for now — conservative for the freeze; can flip in a later release. Thanks for withdrawing the post-scan-match idea.

Docs/notes: added the 4.2→4.3 migration-guide note (batch position unchanged, noted in the description); covered the CacheManager.sameResult layer in the description rather than calling it "internal"; added the "rebuilt scans never carry reported partitioning/ordering" line; fixed the split scaladoc link.

The round-1 test-coverage items are in as well. PTAL when you get a chance.

[cloud-fan]

0 blocking, 0 non-blocking, 2 nits.
The scan-merging safety gates and rebuild path are coherent and well covered; only two documentation nits remain.

Nits: 2 minor items (see inline comments).

Verification

I traced the opt-in from TableCapability.SCAN_MERGING through plain-scan mergeability classification, the MergeSubplans recursion, deferred filter handling, and the production pushdown rebuild. The rebuild is accepted only when it remains mergeable, fully re-enforces every strict filter, and returns exactly the requested union output; tests cover both config modes and the principal decline gates.

[cloud-fan]

These placeholders are temporary, not temporal (time-related). Please use temporary here and in the two class Scaladocs below.

[peter-toth]

Thanks @cloud-fan! Fixed in 466b6b69b12.

[cloud-fan]

Please phrase this as “The data type of the original scalar subquery.” The output parameter below has the same missing-article issue: “The output of the original aggregate.”

[peter-toth]

Fixed in 466b6b69b12.

[peter-toth]

@LuciferYang, I wonder if you could do another round of review today. If you agree with the changes this feature can make it into 4.3. No worries if not, I will adjust the new configs to 5.0 then.

[LuciferYang]

@LuciferYang, I wonder if you could do another round of review today. If you agree with the changes this feature can make it into 4.3. No worries if not, I will adjust the new configs to 5.0 then.

I will take a look at it today.