peter-toth opened a new pull request, #57360: URL: https://github.com/apache/spark/pull/57360
### 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 `DataSourceV2ScanRelation`s 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): column-union merge, differing-filter OR-widen, strict-filter re-push, the non-deterministic-filter-blocks-merge guard, and the negative gates (no opt-in, merge-blocking pushdown). - `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 are fused into one scan reading the union of columns, with `checkAnswer` verifying correctness. Regression: `DataSourceV2Suite`, `InjectRuntimeFilterSuite`, and the moved `PlanMergingSuite`. scalastyle 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]
