You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by pt...@apache.org on 2026/07/31 20:16:01 UTC
(spark) branch master updated: [SPARK-40259][SQL] Support merging equivalent DataSource V2 scans in MergeSubplans
This is an automated email from the ASF dual-hosted git repository.
peter-toth pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 4dcdd4dba544 [SPARK-40259][SQL] Support merging equivalent DataSource V2 scans in MergeSubplans
4dcdd4dba544 is described below
commit 4dcdd4dba544210898180a410b8b30e0b85157cb
Author: Peter Toth <pe...@gmail.com>
AuthorDate: Fri Jul 31 22:15:46 2026 +0200
[SPARK-40259][SQL] Support merging equivalent DataSource V2 scans in MergeSubplans
### 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 `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 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 fut [...]
- 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 m [...]
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 [...]
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 [...]
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 `Dyna [...]
### 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 `SC [...]
- `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)
Closes #57360 from peter-toth/SPARK-40259-dsv2-planmerger-support.
Authored-by: Peter Toth <pe...@gmail.com>
Signed-off-by: Peter Toth <pe...@gmail.com>
---
docs/sql-migration-guide.md | 1 +
.../sql/connector/catalog/TableCapability.java | 30 +-
.../optimizer/MergeSubplansReferences.scala | 65 ++
.../datasources/v2/DataSourceV2Relation.scala | 35 +-
.../org/apache/spark/sql/internal/SQLConf.scala | 15 +
.../InMemoryScanMergingPartitionFilterTable.scala | 92 +++
.../spark/sql/execution/SparkOptimizer.scala | 1 +
.../datasources/v2/V2ScanRelationPushDown.scala | 82 +-
.../sql/execution/planmerging}/MergeSubplans.scala | 45 +-
.../sql/execution/planmerging}/PlanMerger.scala | 484 ++++++++---
.../spark/sql/InjectRuntimeFilterSuite.scala | 2 +-
.../spark/sql/connector/DataSourceV2Suite.scala | 69 +-
.../planmerging/DSv2PlanMergingSuite.scala | 172 ++++
.../planmerging}/MergeSubplansSuite.scala | 883 ++++++++++++++++++++-
.../planmerging/PlanMergingSuite.scala} | 5 +-
15 files changed, 1815 insertions(+), 166 deletions(-)
diff --git a/docs/sql-migration-guide.md b/docs/sql-migration-guide.md
index 6efed7224c0d..5595cc09ab86 100644
--- a/docs/sql-migration-guide.md
+++ b/docs/sql-migration-guide.md
@@ -30,6 +30,7 @@ license: |
- Since Spark 4.3, the configuration key `spark.sql.sources.v2.bucketing.allowJoinKeysSubsetOfPartitionKeys.enabled` has been renamed to `spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled` to reflect that it now applies to storage-partitioned joins, aggregates, and windows. The old key continues to work as an alias.
- Since Spark 4.3, the Spark Thrift Server rejects setting JVM system properties through the `set:system:` session configuration overlay (for example, in a JDBC connection string). To restore the previous behavior, set `spark.sql.legacy.hive.thriftServer.allowSettingSystemProperties` to `true`.
- Since Spark 4.3, the adaptive execution rule `org.apache.spark.sql.execution.adaptive.DynamicJoinSelection` has been renamed to `DemoteBroadcastHashJoin`, which now only demotes broadcast hash joins (emitting `NO_BROADCAST_HASH`). Its selection of shuffled hash join over sort merge join has moved to a new physical rule gated by `spark.sql.adaptive.convertSortMergeJoinToShuffledHashJoin.enabled` (default `true`). If you previously disabled the shuffled-hash-join preference by listing `o [...]
+- Since Spark 4.3, the optimizer rule `MergeSubplans` has moved from package `org.apache.spark.sql.catalyst.optimizer` to `org.apache.spark.sql.execution.planmerging`, so its fully qualified name is now `org.apache.spark.sql.execution.planmerging.MergeSubplans` (the rule was itself renamed from `MergeScalarSubqueries` to `MergeSubplans` in Spark 4.2). If you previously disabled it by listing `org.apache.spark.sql.catalyst.optimizer.MergeSubplans`, or the older `org.apache.spark.sql.catal [...]
- Since Spark 4.3, `spark.sql.execution.replaceHashWithSortAgg` defaults to `true`. Spark now replaces a hash-based aggregate with a sort aggregate when the aggregate's child is already sorted on the grouping keys. To restore the previous behavior, set `spark.sql.execution.replaceHashWithSortAgg` to `false`.
- Since Spark 4.3, `spark.sql.execution.combineAdjacentAggregation` defaults to `true`. Spark now merges an adjacent partial/final aggregate pair (with no shuffle between them) into a single complete-mode aggregate. This setting is independent of `spark.sql.execution.replaceHashWithSortAgg`, so disabling only `replaceHashWithSortAgg` still leaves adjacent aggregation combined; to fully restore the previous partial/final staging, set both `spark.sql.execution.replaceHashWithSortAgg` and ` [...]
- Since Spark 4.3, the exact `percentile`, `percentile_cont`, and `median` aggregate functions (including their `WITHIN GROUP (ORDER BY ...)` forms) compute the linear interpolation between two neighboring values as `lower + fraction * (higher - lower)` instead of `(1 - fraction) * lower + fraction * higher`. The two are equal in exact arithmetic, but the new form is monotonically non-decreasing in the requested percentage and avoids a rounding error the old form could introduce. As a re [...]
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableCapability.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableCapability.java
index 6a7c9b704e42..599e801f2a07 100644
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableCapability.java
+++ b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableCapability.java
@@ -126,5 +126,33 @@ public enum TableCapability {
*
* @since 4.3.0
*/
- GENERATE_COLUMN_VALUES_ON_WRITE
+ GENERATE_COLUMN_VALUES_ON_WRITE,
+
+ /**
+ * Signals that Spark may fuse two batch scans of this table that differ only in their projected
+ * columns and/or pushed filters into a single scan (Spark-side scan merging).
+ * <p>
+ * By returning this capability a table declares a determinism contract: holding the scan options
+ * constant, the rows and columns a scan reads are fully determined by the filters pushed via
+ * {@link org.apache.spark.sql.connector.read.SupportsPushDownV2Filters} and the columns pruned
+ * via {@link org.apache.spark.sql.connector.read.SupportsPushDownRequiredColumns}. Equivalently,
+ * obtaining a fresh {@link org.apache.spark.sql.connector.read.ScanBuilder} with the same options
+ * and re-applying the same pushed filters and pruned columns yields an equivalent scan.
+ * <p>
+ * Given that contract, Spark builds the merged scan itself: it prunes a fresh ScanBuilder to the
+ * union of both read schemas, re-pushes the (possibly OR-widened) filters, and builds. The merged
+ * scan reads the union of the two scans' columns and a superset of their rows; each original
+ * scan's result is recovered by a projection and filter applied above it. The connector supplies
+ * no merge logic of its own.
+ * <p>
+ * This capability lives on the table rather than on the scan so that a source using the V1 scan
+ * fallback (whose scan Spark wraps in an internal wrapper) can still opt in. A table need not
+ * reason about pushdowns that are not reproducible this way (a pushed aggregate, join, variant
+ * extraction, limit, offset, top-N, or table sample): Spark tracks those on its own side while
+ * building the scan and never merges a scan that carries one, whether or not the table returns
+ * this capability.
+ *
+ * @since 4.3.0
+ */
+ SCAN_MERGING
}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/MergeSubplansReferences.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/MergeSubplansReferences.scala
new file mode 100644
index 000000000000..9ec2227f969c
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/MergeSubplansReferences.scala
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.optimizer
+
+import org.apache.spark.sql.catalyst.expressions.{Attribute, ExprId, LeafExpression, Unevaluable}
+import org.apache.spark.sql.catalyst.plans.logical.LeafNode
+import org.apache.spark.sql.catalyst.trees.TreePattern.{NO_GROUPING_AGGREGATE_REFERENCE, SCALAR_SUBQUERY_REFERENCE, TreePattern}
+import org.apache.spark.sql.types.DataType
+
+// The temporary reference placeholders below are produced by the `MergeSubplans` rule (now in
+// sql/core) but must remain in catalyst: `ScalarSubqueryReference` is referenced by the catalyst
+// expression `BloomFilterMightContain`, and catalyst cannot depend on sql/core.
+
+/**
+ * Temporary reference to a subquery which is added to a `PlanMerger`.
+ *
+ * @param level The level of the replaced subquery. It defines the `PlanMerger` instance into which
+ * the subquery is merged.
+ * @param mergedPlanIndex The index of the merged plan in the `PlanMerger`.
+ * @param outputIndex The index of the output attribute of the merged plan.
+ * @param dataType The data type of the original scalar subquery.
+ * @param exprId The expression id of the original scalar subquery.
+ */
+case class ScalarSubqueryReference(
+ level: Int,
+ mergedPlanIndex: Int,
+ outputIndex: Int,
+ override val dataType: DataType,
+ exprId: ExprId) extends LeafExpression with Unevaluable {
+ override def nullable: Boolean = true
+
+ final override val nodePatterns: Seq[TreePattern] = Seq(SCALAR_SUBQUERY_REFERENCE)
+}
+
+/**
+ * Temporary reference to a non-grouping aggregate which is added to a `PlanMerger`.
+ *
+ * @param level The level of the replaced aggregate. It defines the `PlanMerger` instance into which
+ * the aggregate is merged.
+ * @param mergedPlanIndex The index of the merged plan in the `PlanMerger`.
+ * @param outputIndices The indices of the output attributes of the merged plan.
+ * @param output The output of the original aggregate.
+ */
+case class NonGroupingAggregateReference(
+ level: Int,
+ mergedPlanIndex: Int,
+ outputIndices: Seq[Int],
+ override val output: Seq[Attribute]) extends LeafNode {
+ final override val nodePatterns: Seq[TreePattern] = Seq(NO_GROUPING_AGGREGATE_REFERENCE)
+}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala
index 06f9b21d7225..52c2674adc2c 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala
@@ -160,8 +160,22 @@ case class DataSourceV2Relation(
* @param keyGroupedPartitioning if set, the partitioning expressions that are used to split the
* rows in the scan across different partitions
* @param ordering if set, the ordering provided by the scan
- * @param pushedFilters Catalyst expressions for filters that were fully pushed to the data
- * source and do not appear as post-scan filters
+ * @param pushedFilters Catalyst expressions for filters that were fully pushed to the data source
+ * and do not appear as post-scan filters. These reference the relation's
+ * (pre-pruning) output, so they may reference columns pruned out of `output`
+ * (e.g. an unselected partition column the source enforces internally). This
+ * complete set is what lets `PlanMerger` soundly compare and re-enforce a
+ * scan's filters when fusing two scans via a Spark-side scan merge
+ * (`TableCapability.SCAN_MERGING`).
+ * @param mergeableScan whether this scan may be fused with an equivalent scan by a Spark-side scan
+ * merge (see `TableCapability.SCAN_MERGING`).
+ * Default false (not mergeable): only the plain column-pruning + filter
+ * pushdown path in `V2ScanRelationPushDown` sets this true, and only when the
+ * scan carries nothing a rebuilt scan cannot reproduce. A scan with a
+ * non-reproducible pushdown (aggregate, join, variant extraction, limit,
+ * offset, top-N, sample) or by any other rule stays not-mergeable by default,
+ * so merging is safe by construction -- a new scan-relation build site need
+ * not opt out.
*/
case class DataSourceV2ScanRelation(
relation: DataSourceV2Relation,
@@ -169,10 +183,13 @@ case class DataSourceV2ScanRelation(
output: Seq[AttributeReference],
keyGroupedPartitioning: Option[Seq[Expression]] = None,
ordering: Option[Seq[SortOrder]] = None,
- pushedFilters: Seq[Expression] = Seq.empty) extends LeafNode with NamedRelation {
+ pushedFilters: Seq[Expression] = Seq.empty,
+ mergeableScan: Boolean = false) extends LeafNode with NamedRelation {
// TODO: Override validConstraints to return ExpressionSet(pushedFilters) so that pushed
// filters participate in constraint propagation (InferFiltersFromConstraints, PruneFilters).
+ // Note: pushedFilters may reference columns pruned out of `output`, so constraint use must first
+ // intersect with `outputSet` (a constraint has to reference the node's output).
// This changes which filters InferFiltersFromConstraints adds or removes (e.g., it may
// skip adding IsNotNull when the scan already implies it, or infer new filters across
// joins), so plan stability testing is needed first.
@@ -191,6 +208,14 @@ case class DataSourceV2ScanRelation(
override def name: String = relation.name
+ // A leaf relation references no upstream attributes. `pushedFilters` (and, for that matter,
+ // partitioning/ordering) are scan metadata, not references to resolve, and `pushedFilters` may
+ // reference columns pruned out of `output` (e.g. an unselected partition column). Without this
+ // override those would surface as `missingInput`, which the optimizer's plan-change validation
+ // flags as dangling references. `mapExpressions`/`transformExpressions` still rewrite the
+ // metadata expressions -- they iterate the product directly, independent of `references`.
+ override def references: AttributeSet = AttributeSet.empty
+
override def simpleString(maxFields: Int): String = {
val outputString = truncatedString(output, "[", ", ", "]", maxFields)
val nameWithTimeTravelSpec = relation.timeTravelSpec match {
@@ -239,7 +264,9 @@ case class DataSourceV2ScanRelation(
ordering = ordering.map(
_.map(o => o.copy(child = QueryPlan.normalizeExpressions(o.child, output)))
),
- pushedFilters = pushedFilters.map(QueryPlan.normalizeExpressions(_, output))
+ // pushedFilters may reference columns pruned out of `output` (see the field doc), so they are
+ // normalized against the relation's full output rather than `output`.
+ pushedFilters = pushedFilters.map(QueryPlan.normalizeExpressions(_, relation.output))
)
}
}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
index ee9d06683360..45908f299bf6 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
@@ -7263,6 +7263,21 @@ object SQLConf {
.booleanConf
.createWithDefault(false)
+ val MERGE_SUBPLANS_DSV2_SYMMETRIC_FILTER_PROPAGATION_ENABLED = buildConf(
+ "spark.sql.optimizer.mergeSubplans.filterPropagation.dsv2SymmetricFilterPropagation.enabled")
+ .doc("When true, two DataSource V2 scan subplans that pushed the same strict filters but " +
+ "carry different best-effort (post-scan) filters can merge into one scan even when " +
+ s"${MERGE_SUBPLANS_SYMMETRIC_FILTER_PROPAGATION_ENABLED.key} is false. The strict filters " +
+ "are equal (re-enforced on the rebuilt scan) and the differing filters are pushed only as " +
+ "best-effort row-group/partition pruning (OR-widened), with the enclosing Filter " +
+ "re-checking exactness above the scan. Unlike the general symmetric case, OR-widening " +
+ "cannot change the strict (enforced) row set. Has no effect when " +
+ s"${MERGE_SUBPLANS_FILTER_PROPAGATION_ENABLED.key} is false.")
+ .version("4.3.0")
+ .withBindingPolicy(ConfigBindingPolicy.SESSION)
+ .booleanConf
+ .createWithDefault(false)
+
val MERGE_SUBPLANS_FILTER_PROPAGATION_THROUGH_JOIN_ENABLED =
buildConf("spark.sql.optimizer.mergeSubplans.filterPropagation.throughJoin.enabled")
.doc("When set to true, filter attributes can propagate through Join nodes during subplan " +
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryScanMergingPartitionFilterTable.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryScanMergingPartitionFilterTable.scala
new file mode 100644
index 000000000000..37de3b560b5c
--- /dev/null
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryScanMergingPartitionFilterTable.scala
@@ -0,0 +1,92 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connector.catalog
+
+import java.util
+
+import org.apache.spark.sql.catalyst.analysis.TableAlreadyExistsException
+import org.apache.spark.sql.connector.expressions.Transform
+import org.apache.spark.sql.connector.read.{Batch, Scan, ScanBuilder}
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+/**
+ * Catalog that hands out [[InMemoryScanMergingPartitionFilterTable]]s, an iterative-pushdown source
+ * that additionally opts in to Spark-side scan merging. Used to exercise merging two DSv2 scans
+ * whose (equal) filter is strict only via the iterative PartitionPredicate second pass.
+ */
+class InMemoryScanMergingPartitionFilterCatalog
+ extends InMemoryTableEnhancedPartitionFilterCatalog {
+ import CatalogV2Implicits._
+
+ override def createTable(
+ ident: Identifier,
+ columns: Array[Column],
+ partitions: Array[Transform],
+ properties: util.Map[String, String]): Table = {
+ if (tables.containsKey(ident)) {
+ throw new TableAlreadyExistsException(ident.asMultipartIdentifier)
+ }
+ InMemoryTableCatalog.maybeSimulateFailedTableCreation(properties)
+ val tableName = s"$name.${ident.quoted}"
+ val table =
+ new InMemoryScanMergingPartitionFilterTable(tableName, columns, partitions, properties)
+ tables.put(ident, table)
+ namespaces.putIfAbsent(ident.namespace.toList, Map())
+ table
+ }
+}
+
+/**
+ * An [[InMemoryEnhancedPartitionFilterTable]] that returns the `TableCapability.SCAN_MERGING`
+ * capability, so [[org.apache.spark.sql.execution.planmerging.PlanMerger]] may fuse two scans of
+ * this table. Its scan is wrapped in a thin [[NonReportingScan]] so a partitioned table does not
+ * set the scan relation's `keyGroupedPartitioning` (whose preservation across a merge is a separate
+ * follow-up); this keeps the fixture focused on the iterative-pushdown behavior under test.
+ */
+class InMemoryScanMergingPartitionFilterTable(
+ name: String,
+ columns: Array[Column],
+ partitioning: Array[Transform],
+ properties: util.Map[String, String])
+ extends InMemoryEnhancedPartitionFilterTable(name, columns, partitioning, properties) {
+
+ override def capabilities(): util.Set[TableCapability] = {
+ val caps = new util.HashSet[TableCapability](super.capabilities())
+ caps.add(TableCapability.SCAN_MERGING)
+ caps
+ }
+
+ override def newScanBuilder(options: CaseInsensitiveStringMap): ScanBuilder =
+ new InMemoryEnhancedPartitionFilterScanBuilder(schema()) {
+ override def build(): Scan = NonReportingScan(super.build())
+ }
+}
+
+/**
+ * Thin scan decorator that exposes only `readSchema`, `toBatch` and `description`, dropping the
+ * base scan's `SupportsReportPartitioning`/`SupportsReportStatistics`. So the scan relation carries
+ * no reported partitioning/ordering/statistics -- for a partitioned table this keeps
+ * `keyGroupedPartitioning` unset, which the scan merge requires (preserving reported partitioning
+ * across a merge is a separate follow-up).
+ */
+case class NonReportingScan(inner: Scan) extends Scan {
+ override def readSchema(): StructType = inner.readSchema()
+ override def toBatch: Batch = inner.toBatch
+ override def description(): String = inner.description()
+}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkOptimizer.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkOptimizer.scala
index 0994220afe0c..85723550821b 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkOptimizer.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkOptimizer.scala
@@ -26,6 +26,7 @@ import org.apache.spark.sql.connector.catalog.CatalogManager
import org.apache.spark.sql.execution.datasources.{MarkSingleTaskExecution, PruneFileSourcePartitions, PullOutVariantExtractions, PushVariantIntoScan, SchemaPruning, V1Writes}
import org.apache.spark.sql.execution.datasources.v2.{GroupBasedRowLevelOperationScanPlanning, OptimizeMetadataOnlyDeleteFromTable, V2ScanPartitioningAndOrdering, V2ScanRelationPushDown, V2Writes}
import org.apache.spark.sql.execution.dynamicpruning.{CleanupDynamicPruningFilters, PartitionPruning, RowLevelOperationRuntimeGroupFiltering}
+import org.apache.spark.sql.execution.planmerging.MergeSubplans
import org.apache.spark.sql.execution.python.{ExtractGroupingPythonUDFFromAggregate, ExtractPythonUDFFromAggregate, ExtractPythonUDFs, ExtractPythonUDTFs}
class SparkOptimizer(
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala
index d7c518d0113e..1bd43e07afb2 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala
@@ -69,6 +69,24 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper {
}
}
+ /**
+ * Rebuilds a single scan for a Spark-side scan merge (see
+ * `TableCapability.SCAN_MERGING`): runs this rule on a synthetic
+ * `Project(projectList, Filter(conditions, relation))` and returns the resulting
+ * [[DataSourceV2ScanRelation]]. This lets `PlanMerger` fuse two scans of the same table by
+ * expressing what the merged scan should project and filter, while the pushdown lifecycle (filter
+ * translation, column pruning, the iterative PartitionPredicate second pass) stays owned here.
+ * `conditions` are pushed as a single `Filter` directly above the relation; the caller decides
+ * which of them must come back fully enforced.
+ */
+ def rebuildScan(
+ relation: DataSourceV2Relation,
+ projectList: Seq[NamedExpression],
+ conditions: Seq[Expression]): Option[DataSourceV2ScanRelation] = {
+ val child = conditions.reduceOption(And).map(Filter(_, relation)).getOrElse(relation)
+ apply(Project(projectList, child)).collectFirst { case s: DataSourceV2ScanRelation => s }
+ }
+
private def collapseGroupedSumOfCount(plan: LogicalPlan): LogicalPlan = {
val excludedRules = SQLConf.get.optimizerExcludedRules.toSeq.flatMap(Utils.stringToSeq)
if (excludedRules.contains(CollapseGroupedSumOfCount.ruleName)) {
@@ -111,8 +129,9 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper {
val postScanFilters = postScanFiltersWithoutSubquery ++ normalizedFiltersWithSubquery
// Compute the pushed filter expressions: the normalized filters that were fully pushed
- // down (i.e., not in postScanFilters). These are stored on the scan relation for
- // potential future use in constraint propagation.
+ // down (i.e., not in postScanFilters). These are stored on the scan relation for potential
+ // future use in constraint propagation, and are read by a Spark-side scan merge (see
+ // TableCapability.SCAN_MERGING) to compare and re-enforce a scan's filters.
val postScanFilterSet = ExpressionSet(postScanFiltersWithoutSubquery)
sHolder.pushedFilterExpressions = normalizedFiltersWithoutSubquery
.filterNot(postScanFilterSet.contains)
@@ -826,6 +845,37 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper {
}
}
+ /**
+ * Whether the plain-scan path ([[pruneColumns]]) carries a pushdown that a Spark-side scan merge
+ * (see `TableCapability.SCAN_MERGING`) cannot reproduce by
+ * rebuilding the scan from a fresh `ScanBuilder` and re-applying the pushed filters and pruned
+ * columns. Used only to decide the plain-scan `mergeableScan` flag (`!hasBlockingPushdown`).
+ *
+ * Scan merging is default-safe: [[DataSourceV2ScanRelation.mergeableScan]] defaults to false and
+ * this is the ONLY site that ever sets it true. Every other scan-relation build path -- the
+ * dedicated aggregate/join/variant rules ([[buildScanWithPushedAggregate]] /
+ * [[buildScanWithPushedJoin]] / [[buildScanWithPushedVariants]]), row-level-operation planning,
+ * and anything added later -- leaves the scan not-mergeable by default, so those sites need no
+ * change here.
+ *
+ * The plain-scan non-reproducible pushdowns are:
+ * - `pushedLimit` -- pushed LIMIT
+ * - `pushedOffset` -- pushed OFFSET
+ * - `pushedSample` -- pushed table sample
+ * - `sortOrders` -- pushed sort order (top-N / ordering)
+ *
+ * Reproducible (re-applied by the merge), so NOT blocking: `output` (column pruning, via
+ * `SupportsPushDownRequiredColumns`) and `pushedPredicates` / `pushedFilterExpressions`
+ * (deterministic filters, re-pushed via `SupportsPushDownV2Filters`). A pushed aggregate, join or
+ * variant never reaches here -- their build rules run first. When a new pushdown capability adds
+ * plain-scan state to [[ScanBuilderHolder]], list it above so it keeps the scan not-mergeable.
+ */
+ private def hasBlockingPushdown(holder: ScanBuilderHolder): Boolean =
+ holder.pushedLimit.isDefined ||
+ holder.pushedOffset.isDefined ||
+ holder.pushedSample.isDefined ||
+ holder.sortOrders.nonEmpty
+
def buildScanWithPushedAggregate(plan: LogicalPlan): LogicalPlan = plan.transform {
case holder: ScanBuilderHolder if holder.pushedAggregate.isDefined =>
// No need to do column pruning because only the aggregate columns are used as
@@ -838,6 +888,8 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper {
val wrappedScan = getWrappedScan(scan, holder)
// Note: holder.pushedFilterExpressions is not propagated here because the output schema
// changes to aggregate columns. When validConstraints is wired up, this needs revisiting.
+ // A pushed aggregate cannot be reproduced by rebuilding the scan, so this scan stays
+ // not-mergeable (the default `mergeableScan = false`).
val scanRelation = DataSourceV2ScanRelation(holder.relation, wrappedScan, realOutput)
val projectList = realOutput.zip(holder.output).map { case (a1, a2) =>
// The data source may return columns with arbitrary data types and it's safer to cast them
@@ -857,6 +909,8 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper {
val wrappedScan = getWrappedScan(scan, holder)
// Note: holder.pushedFilterExpressions is not propagated here because the output schema
// changes with pushed join. When validConstraints is wired up, this needs revisiting.
+ // A pushed join cannot be reproduced by rebuilding the scan, so this scan stays not-mergeable
+ // (the default `mergeableScan = false`).
val scanRelation = DataSourceV2ScanRelation(holder.relation, wrappedScan, realOutput)
// When join is pushed down, the real output is going to be, for example,
@@ -885,6 +939,8 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper {
val wrappedScan = getWrappedScan(scan, holder)
// Note: holder.pushedFilterExpressions is not propagated here because the output schema
// changes with variant extraction. When validConstraints is wired up, this needs revisiting.
+ // Pushed variant extraction cannot be reproduced by rebuilding the scan, so this scan stays
+ // not-mergeable (the default `mergeableScan = false`).
val scanRelation = DataSourceV2ScanRelation(holder.relation, wrappedScan, realOutput)
// Create projection to map real output to expected output (with transformed types)
@@ -941,10 +997,11 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper {
case projectionOverSchema(newExpr) => newExpr
}
- // Remap pushed filter attributes to the pruned output schema and drop filters
- // whose references are no longer in the pruned output. Catch FIELD_NOT_FOUND
- // because ProjectionOverSchema throws when a pushed filter references a nested
- // struct field that was pruned from the schema.
+ // Remap pushed filter attributes to the pruned output schema and drop filters whose
+ // references are no longer in the pruned output. Catch FIELD_NOT_FOUND because
+ // ProjectionOverSchema throws when a pushed filter references a nested struct field that was
+ // pruned from the schema. This feeds only the Spark post-pushdown adjustment below; the scan
+ // relation's own pushedFilters keep the complete set (see the next comment).
val remappedPushedFilters = sHolder.pushedFilterExpressions.flatMap { filter =>
try Some(projectionFunc(filter))
catch {
@@ -952,8 +1009,19 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper {
None
}
}.filter(_.references.subsetOf(AttributeSet(output)))
+
+ // Record the fully-pushed filter expressions on the scan relation, keeping their references
+ // to the relation's (pre-pruning) output. These include filters on columns that were pruned
+ // out of the scan output -- e.g. an unselected partition column the source still enforces
+ // internally. PlanMerger needs this complete set to compare two scans' filters and to
+ // re-enforce them when it rebuilds a merged scan; remapping to the pruned output (as the
+ // post-scan filters below are) would silently drop pruned-out filter columns and make the
+ // merge unsound. See DataSourceV2ScanRelation.pushedFilters.
val scanRelation = DataSourceV2ScanRelation(sHolder.relation, wrappedScan, output,
- pushedFilters = remappedPushedFilters)
+ pushedFilters = sHolder.pushedFilterExpressions,
+ // The one site that grants mergeability: a plain scan carrying only reproducible pushdowns
+ // (column pruning + deterministic filters) may be fused. See hasBlockingPushdown.
+ mergeableScan = !hasBlockingPushdown(sHolder))
val finalFilters = normalizedFilters.map(projectionFunc)
// bottom-most filters are put in the left of the list.
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/MergeSubplans.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/MergeSubplans.scala
similarity index 89%
rename from sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/MergeSubplans.scala
rename to sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/MergeSubplans.scala
index 037abc207298..45c06f86bd41 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/MergeSubplans.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/MergeSubplans.scala
@@ -15,16 +15,16 @@
* limitations under the License.
*/
-package org.apache.spark.sql.catalyst.optimizer
+package org.apache.spark.sql.execution.planmerging
import scala.collection.mutable.ArrayBuffer
import org.apache.spark.sql.catalyst.expressions._
-import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, CTERelationDef, CTERelationRef, LeafNode, LogicalPlan, OneRowRelation, Project, Subquery, WithCTE}
+import org.apache.spark.sql.catalyst.optimizer.{NonGroupingAggregateReference, ScalarSubqueryReference}
+import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, CTERelationDef, CTERelationRef, LogicalPlan, OneRowRelation, Project, Subquery, WithCTE}
import org.apache.spark.sql.catalyst.rules.Rule
-import org.apache.spark.sql.catalyst.trees.TreePattern.{AGGREGATE, CTE, NO_GROUPING_AGGREGATE_REFERENCE, SCALAR_SUBQUERY, SCALAR_SUBQUERY_REFERENCE, TreePattern}
+import org.apache.spark.sql.catalyst.trees.TreePattern.{AGGREGATE, CTE, NO_GROUPING_AGGREGATE_REFERENCE, SCALAR_SUBQUERY, SCALAR_SUBQUERY_REFERENCE}
import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.types.DataType
/**
* This rule tries to merge multiple subplans that have one row result. This can be either the plan
@@ -340,40 +340,3 @@ object MergeSubplans extends Rule[LogicalPlan] {
}
}
-/**
- * Temporal reference to a subquery which is added to a `PlanMerger`.
- *
- * @param level The level of the replaced subquery. It defines the `PlanMerger` instance into which
- * the subquery is merged.
- * @param mergedPlanIndex The index of the merged plan in the `PlanMerger`.
- * @param outputIndex The index of the output attribute of the merged plan.
- * @param dataType The dataType of original scalar subquery.
- * @param exprId The expression id of the original scalar subquery.
- */
-case class ScalarSubqueryReference(
- level: Int,
- mergedPlanIndex: Int,
- outputIndex: Int,
- override val dataType: DataType,
- exprId: ExprId) extends LeafExpression with Unevaluable {
- override def nullable: Boolean = true
-
- final override val nodePatterns: Seq[TreePattern] = Seq(SCALAR_SUBQUERY_REFERENCE)
-}
-
-/**
- * Temporal reference to a non-grouping aggregate which is added to a `PlanMerger`.
- *
- * @param level The level of the replaced aggregate. It defines the `PlanMerger` instance into which
- * the aggregate is merged.
- * @param mergedPlanIndex The index of the merged plan in the `PlanMerger`.
- * @param outputIndices The indices of the output attributes of the merged plan.
- * @param output The output of original aggregate.
- */
-case class NonGroupingAggregateReference(
- level: Int,
- mergedPlanIndex: Int,
- outputIndices: Seq[Int],
- override val output: Seq[Attribute]) extends LeafNode {
- final override val nodePatterns: Seq[TreePattern] = Seq(NO_GROUPING_AGGREGATE_REFERENCE)
-}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PlanMerger.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala
similarity index 54%
rename from sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PlanMerger.scala
rename to sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala
index 1c43f91cee9d..df74bbe0036d 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PlanMerger.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala
@@ -15,15 +15,17 @@
* limitations under the License.
*/
-package org.apache.spark.sql.catalyst.optimizer
+package org.apache.spark.sql.execution.planmerging
import scala.collection.mutable
-import org.apache.spark.sql.catalyst.expressions.{Alias, And, Attribute, AttributeMap, Expression, If, Literal, NamedExpression, Or}
+import org.apache.spark.sql.catalyst.expressions.{Alias, And, Attribute, AttributeMap, AttributeSet, Expression, ExpressionSet, If, Literal, NamedExpression, Or}
import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression
import org.apache.spark.sql.catalyst.plans.{Cross, Inner, JoinType, LeftAnti, LeftOuter, LeftSemi, RightOuter}
import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, Filter, Join, LogicalPlan, Project}
import org.apache.spark.sql.catalyst.trees.TreeNodeTag
+import org.apache.spark.sql.connector.catalog.TableCapability
+import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, DataSourceV2ScanRelation, V2ScanRelationPushDown}
import org.apache.spark.sql.internal.SQLConf
/**
@@ -62,8 +64,8 @@ object PlanMerger {
// Global counter for generating unique names for propagated filter attributes across all
// PlanMerger instances.
- private[optimizer] val curId = new java.util.concurrent.atomic.AtomicLong()
- private[optimizer] def newId: Long = curId.getAndIncrement()
+ private[planmerging] val curId = new java.util.concurrent.atomic.AtomicLong()
+ private[planmerging] def newId: Long = curId.getAndIncrement()
}
/**
@@ -135,7 +137,9 @@ class PlanMerger(
symmetricFilterPropagationEnabled: Boolean =
SQLConf.get.getConf(SQLConf.MERGE_SUBPLANS_SYMMETRIC_FILTER_PROPAGATION_ENABLED),
filterPropagationThroughJoinEnabled: Boolean =
- SQLConf.get.getConf(SQLConf.MERGE_SUBPLANS_FILTER_PROPAGATION_THROUGH_JOIN_ENABLED)) {
+ SQLConf.get.getConf(SQLConf.MERGE_SUBPLANS_FILTER_PROPAGATION_THROUGH_JOIN_ENABLED),
+ dsv2SymmetricFilterPropagationEnabled: Boolean =
+ SQLConf.get.getConf(SQLConf.MERGE_SUBPLANS_DSV2_SYMMETRIC_FILTER_PROPAGATION_ENABLED)) {
val cache = mutable.ArrayBuffer.empty[MergedPlan]
/**
@@ -166,8 +170,8 @@ class PlanMerger(
val outputMap = AttributeMap(plan.output.zipWithIndex)
MergeResult(newMergedPlan, i, outputMap)
}.orElse {
- tryMergePlans(plan, mp.plan, false).collect {
- case TryMergeResult(mergedPlan, npMapping, None, None) =>
+ tryMergePlans(plan, mp.plan, MergeContext(filterPropagationSupported = false)).collect {
+ case TryMergeResult(mergedPlan, npMapping, None, None, None, _) =>
val newMergedPlan = MergedPlan(mergedPlan, true)
cache(i) = newMergedPlan
val outputMap = AttributeMap(npMapping.iterator.map { case (origAttr, mergedAttr) =>
@@ -220,12 +224,64 @@ class PlanMerger(
* merged plan. `None` when no differing filter was propagated.
* @param cachedPlanFilter Like `newPlanFilter` but for the cached plan's side. Always a freshly
* created alias when present, so no `isNew` flag is needed.
+ * @param dsv2Merged Whether an (equal-strict) DSv2 scan merge occurred anywhere in this merged
+ * subtree. Unlike `dsv2DeferredScan` (consumed at the enclosing Filter that
+ * builds the scan), this fact is propagated all the way up: it lets a Filter
+ * pair that is NOT the innermost one still recognize the merge below it and
+ * apply the DSv2-symmetric exemption. It only ever gates behaviour when
+ * `dsv2SymmetricFilterPropagationEnabled` is on.
*/
case class TryMergeResult(
mergedPlan: LogicalPlan,
newPlanMapping: AttributeMap[Attribute],
newPlanFilter: Option[(Attribute, Boolean)] = None,
- cachedPlanFilter: Option[Attribute] = None)
+ cachedPlanFilter: Option[Attribute] = None,
+ dsv2DeferredScan: Option[DSv2DeferredScan] = None,
+ dsv2Merged: Boolean = false)
+
+ /**
+ * Carries a DSv2 scan whose build has been deferred from the leaf up to the enclosing [[Filter]],
+ * so the merged scan is built exactly once per merge round (strict + best-effort filters
+ * together) rather than once strict-only at the leaf and then rebuilt at the Filter.
+ *
+ * The relation to rebuild from is not carried here: the deferring leaf leaves it in the plan as
+ * the placeholder [[TryMergeResult.mergedPlan]] (the sole bare [[DataSourceV2Relation]] in the
+ * subtree), and `tryBuildFilterDSv2ScanChild` recovers it from there. Only what the tree does NOT
+ * hold is carried: the projected `unionAttrs` and the `strictFilters` to re-enforce.
+ *
+ * @param unionAttrs The union of both sides' projected columns the merged scan must produce.
+ * @param strictFilters The strict pushed filters that must be re-enforced by the rebuilt scan.
+ */
+ case class DSv2DeferredScan(
+ unionAttrs: Seq[Attribute],
+ strictFilters: Seq[Expression])
+
+ /**
+ * Context threaded DOWN through [[tryMergePlans]] recursion.
+ *
+ * Invariants:
+ * - `filterAboveScan` is set true ONLY by the `(Filter, Filter)` arm; the leaf defers building
+ * the merged DSv2 scan ONLY when it is true. A deferred result flows leaf -> (pass-through
+ * Project arms, which propagate `dsv2DeferredScan` unchanged) -> the enclosing `(Filter,
+ * Filter)` arm, which builds it once (via `tryBuildFilterDSv2ScanChild`) and returns
+ * `dsv2DeferredScan = None`.
+ * - A `(Filter, Filter)`'s children CAN themselves be Filters: `PartitionPruning` and
+ * `InjectRuntimeFilter` insert a Filter above an existing one after `CombineFilters` has run,
+ * and the `PushDownPredicates` pass that would re-combine them runs only in a later batch. The
+ * deferral is consumed at the INNERMOST `(Filter, Filter)` pair (which builds the scan); an
+ * enclosing pair sees `dsv2DeferredScan = None`, but `dsv2Merged` still marks the merge below
+ * it, so it can apply the DSv2-symmetric exemption too. This is sound because
+ * `V2ScanRelationPushDown` only pushes the innermost (scan-adjacent) Filter chain to the scan:
+ * an outer stacked Filter was never scan pruning (a runtime filter inserted after pushdown, or
+ * one separated from the scan by an operator that blocks pushdown), so OR-widening it above the
+ * built scan drops no pruning.
+ * - The `merge()` pattern requiring `dsv2DeferredScan = None` is the fail-safe backstop: if a
+ * deferred scan somehow reached `merge()` unbuilt, it declines the merge rather than emitting a
+ * plan with a placeholder relation.
+ * - Eligibility gating stays at the leaf (read-only, inspects only the two input scans). Only the
+ * build moves up.
+ */
+ case class MergeContext(filterPropagationSupported: Boolean, filterAboveScan: Boolean = false)
/**
* Recursively attempts to merge two plans by traversing their tree structures.
@@ -249,40 +305,62 @@ class PlanMerger(
private def tryMergePlans(
newPlan: LogicalPlan,
cachedPlan: LogicalPlan,
- filterPropagationSupported: Boolean): Option[TryMergeResult] = {
- checkIdenticalPlans(newPlan, cachedPlan).map(TryMergeResult(cachedPlan, _)).orElse(
+ context: MergeContext): Option[TryMergeResult] = {
+ // The plain "reuse the cached plan as-is" result, shared by every branch below. Lazy because
+ // the DSv2 merge path under a Filter does not need it when the merge itself succeeds.
+ lazy val identical = checkIdenticalPlans(newPlan, cachedPlan).map(TryMergeResult(cachedPlan, _))
+ // A DSv2 scan pair is handled here in one place, rather than split between this leading check
+ // and the structural match below. Under a Filter, merging DEFERS the scan build so the
+ // enclosing Filter's row-group pruning is pushed in a single rebuild -- so try the merge first
+ // and fall back to plain reuse if the scans cannot merge. Trying the merge first even when the
+ // two scans are identical is deliberate: the deferred rebuild re-pushes the enclosing Filter's
+ // condition as a best-effort filter, which plain reuse would leave as a post-scan Filter, so
+ // reusing identical scans here would forfeit that pruning. Without a Filter there is no pruning
+ // to recover, so reuse an identical scan as-is (no rebuild) and only merge scans that differ
+ // (projected columns / strict filters). Any other plan pair uses the general reuse.
+ val earlyResult = (newPlan, cachedPlan) match {
+ case (np: DataSourceV2ScanRelation, cp: DataSourceV2ScanRelation) =>
+ if (context.filterAboveScan) {
+ tryMergeScanRelations(np, cp, context).orElse(identical)
+ } else {
+ identical.orElse(tryMergeScanRelations(np, cp, context))
+ }
+ case _ => identical
+ }
+ earlyResult.orElse(
(newPlan, cachedPlan) match {
case (np: Project, cp: Project) =>
- tryMergePlans(np.child, cp.child, filterPropagationSupported).map {
- case TryMergeResult(mergedChild, npMapping, npFilter, cpFilter) =>
+ tryMergePlans(np.child, cp.child, context).map {
+ case TryMergeResult(mergedChild, npMapping, npFilter, cpFilter, deferred, dsv2Merged) =>
val (mergedProjectList, newNPMapping) =
mergeNamedExpressions(np.projectList, cp.projectList, npMapping, npFilter, cpFilter)
TryMergeResult(Project(mergedProjectList, mergedChild), newNPMapping, npFilter,
- cpFilter)
+ cpFilter, deferred, dsv2Merged)
}
case (np, cp: Project) =>
- tryMergePlans(np, cp.child, filterPropagationSupported).map {
- case TryMergeResult(mergedChild, npMapping, npFilter, cpFilter) =>
+ tryMergePlans(np, cp.child, context).map {
+ case TryMergeResult(mergedChild, npMapping, npFilter, cpFilter, deferred, dsv2Merged) =>
val (mergedProjectList, newNPMapping) =
mergeNamedExpressions(np.output, cp.projectList, npMapping, npFilter, cpFilter)
TryMergeResult(Project(mergedProjectList, mergedChild), newNPMapping, npFilter,
- cpFilter)
+ cpFilter, deferred, dsv2Merged)
}
case (np: Project, cp) =>
- tryMergePlans(np.child, cp, filterPropagationSupported).map {
- case TryMergeResult(mergedChild, npMapping, npFilter, cpFilter) =>
+ tryMergePlans(np.child, cp, context).map {
+ case TryMergeResult(mergedChild, npMapping, npFilter, cpFilter, deferred, dsv2Merged) =>
val (mergedProjectList, newNPMapping) =
mergeNamedExpressions(np.projectList, cp.output, npMapping, npFilter, cpFilter)
TryMergeResult(Project(mergedProjectList, mergedChild), newNPMapping, npFilter,
- cpFilter)
+ cpFilter, deferred, dsv2Merged)
}
case (np: Aggregate, cp: Aggregate) if supportedAggregateMerge(np, cp) =>
// Filter propagation into the aggregate is only safe when there is no grouping.
val childFilterPropagationSupported = filterPropagationEnabled &&
np.groupingExpressions.isEmpty && cp.groupingExpressions.isEmpty
- tryMergePlans(np.child, cp.child, childFilterPropagationSupported).flatMap {
- case TryMergeResult(mergedChild, npMapping, None, None) =>
+ tryMergePlans(np.child, cp.child,
+ MergeContext(childFilterPropagationSupported, filterAboveScan = false)).flatMap {
+ case TryMergeResult(mergedChild, npMapping, None, None, _, dsv2Merged) =>
val mappedNPGroupingExpression =
np.groupingExpressions.map(mapAttributes(_, npMapping))
// Order of grouping expression does matter as merging different grouping orders can
@@ -294,11 +372,11 @@ class PlanMerger(
mergeNamedExpressions(np.aggregateExpressions, cp.aggregateExpressions, npMapping)
val mergedPlan =
Aggregate(cp.groupingExpressions, mergedAggregateExpressions, mergedChild)
- Some(TryMergeResult(mergedPlan, newNPMapping))
+ Some(TryMergeResult(mergedPlan, newNPMapping, dsv2Merged = dsv2Merged))
} else {
None
}
- case TryMergeResult(mergedChild, npMapping, npFilterOpt, cpFilterOpt) =>
+ case TryMergeResult(mergedChild, npMapping, npFilterOpt, cpFilterOpt, _, dsv2Merged) =>
// childFilterPropagationSupported guarantees both aggregates have no grouping, so
// the grouping-match check is skipped.
assert(childFilterPropagationSupported)
@@ -317,55 +395,87 @@ class PlanMerger(
mergeNamedExpressions(filteredNPAggregateExpressions,
filteredCPAggregateExpressions, npMapping)
val mergedPlan = Aggregate(Seq.empty, mergedAggregateExpressions, mergedChild)
- Some(TryMergeResult(mergedPlan, newNPMapping))
+ Some(TryMergeResult(mergedPlan, newNPMapping, dsv2Merged = dsv2Merged))
}
case (np: Filter, cp: Filter) =>
- tryMergePlans(np.child, cp.child, filterPropagationSupported).flatMap {
- case TryMergeResult(mergedChild, npMapping, npFilter, cpFilter) =>
+ tryMergePlans(np.child, cp.child, context.copy(filterAboveScan = true)).flatMap {
+ case TryMergeResult(mergedChild, npMapping, npFilter, cpFilter, deferred, dsv2Merged) =>
val mappedNPCondition = mapAttributes(np.condition, npMapping)
// Comparing the canonicalized form is required to ignore different forms of the same
// expression.
if (mappedNPCondition.canonicalized == cp.condition.canonicalized) {
- // Identical conditions: the filter node itself adds no new discrimination between
- // the two sides, so we keep it unchanged and pass the child's mappings up.
- val mergedPlan = Filter(cp.condition, mergedChild)
- Some(TryMergeResult(mergedPlan, npMapping, npFilter, cpFilter))
- } else if (filterPropagationSupported && symmetricFilterPropagationEnabled) {
+ // Identical conditions: the filter node adds no new discrimination between the two
+ // sides, so keep it unchanged. If it sits above a deferred merged DSv2 scan, build
+ // that scan once here with this condition as a best-effort filter. If the build
+ // cannot re-enforce the strict filters, decline the whole merge (the leaf's
+ // strict-only build would have failed identically).
+ tryBuildFilterDSv2ScanChild(mergedChild, deferred, Some(cp.condition))
+ .map { prunedChild =>
+ val mergedPlan = Filter(cp.condition, prunedChild)
+ TryMergeResult(
+ mergedPlan, npMapping, npFilter, cpFilter, dsv2Merged = dsv2Merged)
+ }
+ // Symmetric propagation broadens the merged scan to OR(f1, f2); it is off by default
+ // because the two sides may read disjoint data. A DSv2 scan merge is exempt under its
+ // own config: `dsv2Merged` marks an (equal-strict) DSv2 merge below -- whose equal
+ // strict filters mean both sides read the same base set, so the OR only weakens the
+ // best-effort filter. Unlike `deferred`, `dsv2Merged` survives past the innermost
+ // Filter that built the scan, so a stacked outer Filter pair recognizes it too.
+ } else if (context.filterPropagationSupported &&
+ (symmetricFilterPropagationEnabled ||
+ (dsv2SymmetricFilterPropagationEnabled && dsv2Merged))) {
if (cp.getTagValue(PlanMerger.MERGED_FILTER_TAG).isDefined) {
// cp Filter is already a merged filter from a previous round: its condition
// is OR(f0, f1, ...) and its child Project already contains aliases for those
// attributes. Only create a new alias for the np side, and extend the OR
- // condition.
- val newNPCondition = npFilter.fold(mappedNPCondition) {
- case (f, _) => And(f, mappedNPCondition)
- }
- val childProject = mergedChild match {
- case p: Project => p
- case other => throw new IllegalStateException(
- "Expected Project child under MERGED_FILTER_TAG filter, got " +
- s"${other.getClass.getSimpleName}")
- }
- // If newNPCondition is already aliased in the child Project (e.g. a third
- // subplan whose filter matches one from a previous merge round), reuse the
- // existing attribute instead of creating a redundant alias.
- val existingNPFilter = childProject.projectList.collectFirst {
- case a: Alias if a.child.canonicalized == newNPCondition.canonicalized =>
- a.toAttribute
- }
- existingNPFilter match {
- case Some(reusedFilter) =>
- val newFilter = cp.withNewChildren(Seq(mergedChild))
- Some(TryMergeResult(newFilter, npMapping, Some((reusedFilter, false)), None))
- case None =>
- val newNPFilterAlias =
- Alias(newNPCondition, s"propagatedFilter_${PlanMerger.newId}")()
- val newNPFilter = newNPFilterAlias.toAttribute
- val newProject = childProject.copy(
- projectList = childProject.projectList ++ Seq(newNPFilterAlias))
- val newFilter = Filter(Or(cp.condition, newNPFilter), newProject)
- newFilter.copyTagsFrom(cp)
- Some(TryMergeResult(newFilter, npMapping, Some((newNPFilter, true)), None))
+ // condition. A tagged filter is always built with a Project child (see the
+ // first-time branch below), so a non-Project child should not happen; decline the
+ // merge rather than fail, keeping the merge best-effort.
+ mergedChild match {
+ case childProject: Project =>
+ val newNPCondition = npFilter.fold(mappedNPCondition) {
+ case (f, _) => And(f, mappedNPCondition)
+ }
+ // If newNPCondition is already aliased in the child Project (e.g. a third
+ // subplan whose filter matches one from a previous merge round), reuse the
+ // existing attribute instead of creating a redundant alias.
+ val existingNPFilter = childProject.projectList.collectFirst {
+ case a: Alias if a.child.canonicalized == newNPCondition.canonicalized =>
+ a.toAttribute
+ }
+ val (newProjectList, newCondition, newNPFilterOut) =
+ existingNPFilter match {
+ case Some(reusedFilter) =>
+ // np matches an existing side: no new alias, OR condition unchanged.
+ (childProject.projectList, cp.condition, (reusedFilter, false))
+ case None =>
+ val newNPFilterAlias =
+ Alias(newNPCondition, s"propagatedFilter_${PlanMerger.newId}")()
+ (childProject.projectList :+ newNPFilterAlias,
+ Or(cp.condition, newNPFilterAlias.toAttribute): Expression,
+ (newNPFilterAlias.toAttribute, true))
+ }
+ // Phase 2: the leaf re-merge rebuilt the scan with strict filters only,
+ // dropping the OR best-effort filter established in earlier rounds, so
+ // re-establish it here from ALL propagated conditions, not just the new
+ // side's. Only the aliases the OR condition references are filter sides;
+ // other aliases in the Project are computed columns, not filters.
+ val conditions = newProjectList.collect {
+ case a: Alias if newCondition.references.contains(a.toAttribute) => a.child
+ }
+ tryBuildFilterDSv2ScanChild(
+ childProject.child, deferred, conditions.reduceOption(Or))
+ .map { prunedChild =>
+ val newProject = childProject.copy(
+ projectList = newProjectList, child = prunedChild)
+ val newFilter = Filter(newCondition, newProject)
+ newFilter.copyTagsFrom(cp)
+ TryMergeResult(newFilter, npMapping, Some(newNPFilterOut), None,
+ dsv2Merged = dsv2Merged)
+ }
+ case _ =>
+ None
}
} else {
// First-time filter propagation: alias both sides' conditions as boolean
@@ -377,31 +487,40 @@ class PlanMerger(
val newNPCondition =
npFilter.fold(mappedNPCondition) { case (f, _) => And(f, mappedNPCondition) }
val newCPCondition = cpFilter.fold(cp.condition)(And(_, cp.condition))
- val newNPFilterAlias =
- Alias(newNPCondition, s"propagatedFilter_${PlanMerger.newId}")()
- val newCPFilterAlias =
- Alias(newCPCondition, s"propagatedFilter_${PlanMerger.newId}")()
- val newNPFilter = newNPFilterAlias.toAttribute
- val newCPFilter = newCPFilterAlias.toAttribute
- val project = Project(
- mergedChild.output.toList ++ Seq(newNPFilterAlias, newCPFilterAlias),
- mergedChild)
- val newFilter = Filter(Or(newNPFilter, newCPFilter), project)
- newFilter.copyTagsFrom(cp)
- newFilter.setTagValue(PlanMerger.MERGED_FILTER_TAG, ())
- Some(TryMergeResult(newFilter, npMapping, Some((newNPFilter, true)),
- Some(newCPFilter)))
+ // The OR-widen moves both conditions into a boolean Project above the merged
+ // scan, so the scan itself would read the full table. Build the deferred scan
+ // here with OR(np condition, cp condition) as the best-effort filter (the Filter
+ // above still enforces exactness). The best-effort filter is derived from the
+ // scan-level conditions, not the propagated filter attributes in newNP/newCP.
+ tryBuildFilterDSv2ScanChild(
+ mergedChild, deferred, Some(Or(mappedNPCondition, cp.condition)))
+ .map { prunedChild =>
+ val newNPFilterAlias =
+ Alias(newNPCondition, s"propagatedFilter_${PlanMerger.newId}")()
+ val newCPFilterAlias =
+ Alias(newCPCondition, s"propagatedFilter_${PlanMerger.newId}")()
+ val newNPFilter = newNPFilterAlias.toAttribute
+ val newCPFilter = newCPFilterAlias.toAttribute
+ val project = Project(
+ prunedChild.output.toList ++ Seq(newNPFilterAlias, newCPFilterAlias),
+ prunedChild)
+ val newFilter = Filter(Or(newNPFilter, newCPFilter), project)
+ newFilter.copyTagsFrom(cp)
+ newFilter.setTagValue(PlanMerger.MERGED_FILTER_TAG, ())
+ TryMergeResult(newFilter, npMapping, Some((newNPFilter, true)),
+ Some(newCPFilter), dsv2Merged = dsv2Merged)
+ }
}
} else {
None
}
}
- case (np: Filter, cp) if filterPropagationSupported =>
- tryMergePlans(np.child, cp, filterPropagationSupported).collect {
+ case (np: Filter, cp) if context.filterPropagationSupported =>
+ tryMergePlans(np.child, cp, context.copy(filterAboveScan = false)).collect {
// If the cp side already propagated a filter from deeper recursion, the merge is
// effectively symmetric (both sides have a filter condition). Abort unless
// symmetricFilterPropagationEnabled.
- case TryMergeResult(mergedChild, npMapping, npFilter, cpFilter)
+ case TryMergeResult(mergedChild, npMapping, npFilter, cpFilter, _, dsv2Merged)
if cpFilter.isEmpty || symmetricFilterPropagationEnabled =>
val mappedNPCondition = mapAttributes(np.condition, npMapping)
val newNPCondition = npFilter.fold(mappedNPCondition) {
@@ -413,14 +532,15 @@ class PlanMerger(
val project = Project(
mergedChild.output.toList :+ newNPFilterAlias,
mergedChild)
- TryMergeResult(project, npMapping, Some((newNPFilter, true)), cpFilter)
+ TryMergeResult(project, npMapping, Some((newNPFilter, true)), cpFilter,
+ dsv2Merged = dsv2Merged)
}
- case (np, cp: Filter) if filterPropagationSupported =>
- tryMergePlans(np, cp.child, filterPropagationSupported).collect {
+ case (np, cp: Filter) if context.filterPropagationSupported =>
+ tryMergePlans(np, cp.child, context.copy(filterAboveScan = false)).collect {
// If the np side already propagated a filter from deeper recursion, the merge is
// effectively symmetric (both sides have a filter condition). Abort unless
// symmetricFilterPropagationEnabled.
- case TryMergeResult(mergedChild, npMapping, npFilter, cpFilter)
+ case TryMergeResult(mergedChild, npMapping, npFilter, cpFilter, _, dsv2Merged)
if npFilter.isEmpty || symmetricFilterPropagationEnabled =>
if (cp.getTagValue(PlanMerger.MERGED_FILTER_TAG).isDefined) {
// cp is a previously-merged Filter: its condition is `OR(pf_0, pf_1, ...)` and cp's
@@ -430,7 +550,7 @@ class PlanMerger(
// upstream, which simplifies to `FILTER pf_i` -- wasted work and plan bloat.
// Drop cp's Filter and let the recursion's result flow up with cpFilter = None so
// cp's aggregates are left untouched.
- TryMergeResult(mergedChild, npMapping, npFilter, None)
+ TryMergeResult(mergedChild, npMapping, npFilter, None, dsv2Merged = dsv2Merged)
} else {
val newCPCondition = cpFilter.fold(cp.condition)(And(_, cp.condition))
val newCPFilterAlias =
@@ -439,15 +559,18 @@ class PlanMerger(
val project = Project(
mergedChild.output.toList :+ newCPFilterAlias,
mergedChild)
- TryMergeResult(project, npMapping, npFilter, Some(newCPFilter))
+ TryMergeResult(project, npMapping, npFilter, Some(newCPFilter),
+ dsv2Merged = dsv2Merged)
}
}
case (np: Join, cp: Join) if np.joinType == cp.joinType && np.hint == cp.hint =>
- tryMergePlans(np.left, cp.left, filterPropagationSupported).flatMap {
- case TryMergeResult(mergedLeft, leftNPMapping, leftNPFilter, leftCPFilter) =>
- tryMergePlans(np.right, cp.right, filterPropagationSupported).flatMap {
- case TryMergeResult(mergedRight, rightNPMapping, rightNPFilter, rightCPFilter)
+ tryMergePlans(np.left, cp.left, context.copy(filterAboveScan = false)).flatMap {
+ case TryMergeResult(mergedLeft, leftNPMapping, leftNPFilter, leftCPFilter, _,
+ leftDsv2Merged) =>
+ tryMergePlans(np.right, cp.right, context.copy(filterAboveScan = false)).flatMap {
+ case TryMergeResult(mergedRight, rightNPMapping, rightNPFilter, rightCPFilter, _,
+ rightDsv2Merged)
// If both children independently propagate filter attributes we would need to
// AND them into a new alias above the join, which is not yet supported.
if !(leftNPFilter.isDefined && rightNPFilter.isDefined) &&
@@ -474,7 +597,7 @@ class PlanMerger(
val npFilter = leftNPFilter.orElse(rightNPFilter)
val cpFilter = leftCPFilter.orElse(rightCPFilter)
Some(TryMergeResult(cp.withNewChildren(Seq(mergedLeft, mergedRight)), npMapping,
- npFilter, cpFilter))
+ npFilter, cpFilter, dsv2Merged = leftDsv2Merged || rightDsv2Merged))
} else {
None
}
@@ -488,6 +611,187 @@ class PlanMerger(
})
}
+ /**
+ * The DSv2 scan merge: fuse two scans of the same table that differ only in projected columns
+ * (and carry the same strict pushed filters) into a single scan reading the union of their
+ * columns. The connector opts in via
+ * `TableCapability.SCAN_MERGING`; Spark runs the real DSv2 pushdown
+ * ([[V2ScanRelationPushDown]]) on a synthetic `Filter` over the relation, extracts the merged
+ * scan, and verifies the (equal) strict filters remain fully enforced. Eligibility gating
+ * (`mergeable`) is read-only, inspecting only the two input scans.
+ *
+ * When this scan sits under a [[Filter]] (`context.filterAboveScan`), the build is DEFERRED: the
+ * merged scan is not built here but carried up as a [[DSv2DeferredScan]] and built exactly once
+ * at the enclosing Filter (via `tryBuildFilterDSv2ScanChild`), where the strict filters and the
+ * Filter's best-effort row-group pruning are known and can be pushed together in a single
+ * rebuild. The placeholder plan returned in that case is the bare relation (its output is a
+ * superset of the union columns); the Filter arm splices the built scan in by reference identity.
+ * Otherwise (no enclosing Filter) the scan is built here, strict-only.
+ *
+ * A differing post-scan filter is handled by filter propagation at the Filter, not here. There is
+ * no fallback, so any anomaly (a strict filter the rebuilt scan does not fully enforce, an
+ * unexpected output schema) results in `None` (no merge): it must be correct on its own.
+ */
+ private def tryMergeScanRelations(
+ np: DataSourceV2ScanRelation,
+ cp: DataSourceV2ScanRelation,
+ context: MergeContext): Option[TryMergeResult] = {
+ // np's relation attributes paired with cp's by position, reused below. Safe because the
+ // `mergeable` gate requires the two relations to be canonically equal, so both list the table's
+ // columns in the same order; lazy so it is only built once that check has passed. A scan's
+ // (pruned) `output` is a subset of its `relation.output` -- even where nested field pruning
+ // narrows a column's type -- so this maps each scan's output, and its pushed filters (which
+ // reference the relation's full output), through to the other scan's relation.
+ lazy val npRelationMapping = AttributeMap[Attribute](np.relation.output.zip(cp.relation.output))
+
+ // Each side must read a subset of its relation's columns AT THE SAME type. Column pruning may
+ // drop columns, but a struct/array/map column read at a narrower type via nested schema pruning
+ // leaves its extractors (e.g. GetStructField ordinals) resolved against the narrow layout; the
+ // merge rebuilds the column at the relation's full type, so those ordinals would read the wrong
+ // field. This subset property also guarantees every output attribute maps through
+ // npRelationMapping (and, in tryBuildMergedDSv2Scan, back to the relation). Widening the
+ // merged scan to the union of nested fields and remapping the ordinals is a possible follow-up.
+ def readsSubsetOfRelation(scan: DataSourceV2ScanRelation): Boolean = {
+ val relTypes = AttributeMap(scan.relation.output.map(a => a -> a.dataType))
+ scan.output.forall(a => relTypes.get(a).contains(a.dataType))
+ }
+
+ val mergeable =
+ // Same table, options, catalog and identifier: the relation's canonical form covers all of
+ // these (options compares by content via `CaseInsensitiveStringMap.equals`).
+ np.relation.canonicalized == cp.relation.canonicalized &&
+ // Both scans are mergeable: each came out of the plain column-pruning + filter pushdown
+ // path carrying only pushdowns a rebuilt scan can reproduce. A scan with a non-reproducible
+ // pushdown (aggregate, join, variant, limit, offset, top-N, sample) or built by any other
+ // rule is not mergeable by default.
+ np.mergeableScan && cp.mergeableScan &&
+ // Reported partitioning/ordering (e.g. storage-partitioned join, reported sort) is not
+ // reconstructed by the rebuilt scan -- it never carries reported partitioning or ordering,
+ // as V2ScanPartitioningAndOrdering is a separate early rule that rebuildScan does not run.
+ // So decline the merge when either input reports a NON-EMPTY one rather than silently
+ // dropping it; merging these can be added as a follow-up. A source that implements
+ // SupportsReportOrdering/SupportsReportPartitioning but reports nothing yields Some(Nil),
+ // so test the inner Seq (forall) rather than the Option (isEmpty), which would decline it.
+ np.keyGroupedPartitioning.forall(_.isEmpty) &&
+ cp.keyGroupedPartitioning.forall(_.isEmpty) &&
+ np.ordering.forall(_.isEmpty) && cp.ordering.forall(_.isEmpty) &&
+ // The table opts in to Spark-side merging (a table capability, so a V1-fallback source
+ // whose scan Spark wraps can still opt in). Both relations are the same table (canonically
+ // equal, checked above), but check each to be safe.
+ np.relation.table.capabilities().contains(TableCapability.SCAN_MERGING) &&
+ cp.relation.table.capabilities().contains(TableCapability.SCAN_MERGING) &&
+ // Each side reads a subset of its relation's columns at the same type (see above).
+ readsSubsetOfRelation(np) && readsSubsetOfRelation(cp) &&
+ // Both pushed the same strict filters, so re-pushing reproduces both sides' row sets. Remap
+ // np's pushed filters onto cp's via npRelationMapping (the full relation-to-relation
+ // mapping, since a pushed filter may reference a column pruned out of np.output) before
+ // comparing as sets.
+ ExpressionSet(np.pushedFilters.map(mapAttributes(_, npRelationMapping))) ==
+ ExpressionSet(cp.pushedFilters)
+
+ // Eligibility is settled above (a read-only gate); everything below constructs the merge.
+ if (!mergeable) {
+ return None
+ }
+
+ // np's columns expressed in cp's relation space; the subset check above guarantees each maps.
+ val npMapping = AttributeMap(np.output.map(a => a -> npRelationMapping(a)))
+ // cp's columns are already cp.relation attributes; append the np-only columns, in np.output
+ // order (npMapping.values would be exprId-hash-ordered).
+ val unionAttrs = cp.output ++ np.output.map(npMapping).filterNot(cp.outputSet.contains)
+
+ if (context.filterAboveScan) {
+ // Defer the build to the enclosing Filter so the scan is built once with strict +
+ // best-effort filters. The placeholder mergedPlan is the bare relation (its output is a
+ // superset of unionAttrs). rebuildScan reuses the relation's attributes, so mapping
+ // np.output to cp's relation attributes is consistent with the eventual built scan.
+ Some(TryMergeResult(cp.relation, npMapping,
+ dsv2DeferredScan = Some(DSv2DeferredScan(unionAttrs, cp.pushedFilters)), dsv2Merged = true))
+ } else {
+ // No enclosing Filter: build the merged scan here enforcing the (equal) strict filters over
+ // the union of columns, with no best-effort filter (no post-scan Filter to prune on).
+ tryBuildMergedDSv2Scan(cp.relation, unionAttrs, cp.pushedFilters, bestEffortFilter = None)
+ .map(TryMergeResult(_, npMapping, dsv2Merged = true))
+ }
+ }
+
+ /**
+ * Rebuilds the merged DSv2 scan via [[V2ScanRelationPushDown.rebuildScan]], projecting
+ * `unionAttrs` and filtering by `strictFilters` (plus the `bestEffortFilter`). This reuses
+ * the production pushdown end to end -- the same filter translation, column pruning,
+ * determinism/subquery handling and iterative PartitionPredicate second pass -- rather than
+ * reimplementing a slice of it here.
+ *
+ * `strictFilters` must come back fully enforced (present in the rebuilt scan's `pushedFilters`);
+ * otherwise `None`, because nothing above the leaf re-checks it. The `bestEffortFilter` is
+ * offered to the source only when sound: it is dropped unless it is deterministic (a
+ * non-deterministic predicate the source prunes on would drop rows the enclosing Filter cannot
+ * recover) and references only the relation's own columns (propagated boolean filter attributes
+ * are not columns of the relation).
+ */
+ private def tryBuildMergedDSv2Scan(
+ relation: DataSourceV2Relation,
+ unionAttrs: Seq[Attribute],
+ strictFilters: Seq[Expression],
+ bestEffortFilter: Option[Expression]): Option[DataSourceV2ScanRelation] = {
+ val relationOut = relation.outputSet
+ // Defensive: strict filters come from `pushedFilters`, which reference only relation columns,
+ // so this holds today. If a future caller offers a filter over non-relation attributes, decline
+ // the merge rather than build an unsound scan (there is no fallback above the leaf).
+ if (!strictFilters.forall(_.references.subsetOf(relationOut))) {
+ return None
+ }
+ // `unionAttrs` are the relation's own attributes (the caller builds the union in the relation's
+ // space), so they are the projection directly.
+ // strictFilters are enforced; the bestEffortFilter is offered too, but only when it is
+ // expressible over the relation -- a condition referencing propagated boolean filter aliases
+ // rather than relation columns is dropped. A non-deterministic predicate needs no handling
+ // here: SPARK-58207 keeps non-deterministic filters from being pushed to a V2 source, so an
+ // offered one is simply not pushed (and dropped when the scan is extracted).
+ val conds = strictFilters ++ bestEffortFilter.filter(_.references.subsetOf(relationOut))
+ V2ScanRelationPushDown.rebuildScan(relation, unionAttrs, conds).filter { scan =>
+ // The rebuilt scan must itself be mergeable. rebuildScan re-runs the full pushdown, so any
+ // non-reproducible pushdown it introduces would make the merged scan unsound. Today's
+ // Project-over-Filter input only triggers the plain path (always mergeable), so this is
+ // defensive: it re-validates the rebuild's output against the same gate applied to its
+ // inputs, rather than trusting the rebuild if its input plan ever broadens.
+ scan.mergeableScan &&
+ // Every intended-strict filter must be fully enforced by the rebuilt scan (nothing above
+ // re-checks it), and the scan must produce exactly the requested union of columns.
+ strictFilters.forall(ExpressionSet(scan.pushedFilters).contains) &&
+ scan.outputSet == AttributeSet(unionAttrs)
+ }
+ }
+
+ /**
+ * Threads the deferred DSv2 scan build through a [[Filter]] arm. If the merged child carries a
+ * [[DSv2DeferredScan]], build the scan once here -- at the enclosing Filter, with the strict
+ * filters plus the Filter's `bestEffortFilter` -- and splice it in place of the placeholder
+ * relation. Tries strict + best-effort first, then strict-only (the best-effort filter is
+ * droppable); returns `None` only if the strict filters cannot be re-enforced at all, in which
+ * case the caller must decline the merge (the leaf's strict-only build would have failed
+ * identically). If there is no deferred scan (a non-DSv2 child, or a scan not under a Filter),
+ * there is nothing to build and the child is returned unchanged.
+ */
+ private def tryBuildFilterDSv2ScanChild(
+ child: LogicalPlan,
+ dsv2DeferredScan: Option[DSv2DeferredScan],
+ bestEffortFilter: Option[Expression]): Option[LogicalPlan] = dsv2DeferredScan match {
+ case None => Some(child)
+ case Some(d) =>
+ // The deferring leaf left the relation to rebuild from in the plan as the placeholder
+ // mergedPlan. It is the sole bare DataSourceV2Relation in the subtree (early pushdown has
+ // turned every other relation into a DataSourceV2ScanRelation), so recover it by type here
+ // rather than carrying it on DSv2DeferredScan.
+ child.collectFirst { case r: DataSourceV2Relation => r }.flatMap { relation =>
+ tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, bestEffortFilter)
+ .orElse(tryBuildMergedDSv2Scan(relation, d.unionAttrs, d.strictFilters, None))
+ .map { built =>
+ child.transformUp { case r: DataSourceV2Relation if r eq relation => built }
+ }
+ }
+ }
+
// Returns true when a filter attribute originating from `fromLeft` child of a join with
// `joinType` can be safely propagated through that join to a parent Aggregate.
//
@@ -517,7 +821,7 @@ class PlanMerger(
}
}
- private def mapAttributes[T <: Expression](expr: T, outputMap: AttributeMap[Attribute]) = {
+ private def mapAttributes[T <: Expression](expr: T, outputMap: AttributeMap[_ <: Attribute]) = {
expr.transform {
case a: Attribute => outputMap.getOrElse(a, a)
}.asInstanceOf[T]
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala
index f045a92b6dc9..f5db5373299a 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala
@@ -21,10 +21,10 @@ import java.io.File
import org.apache.spark.sql.catalyst.expressions.{Alias, BloomFilterMightContain, Literal}
import org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression, BloomFilterAggregate}
-import org.apache.spark.sql.catalyst.optimizer.MergeSubplans
import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, Filter, LogicalPlan}
import org.apache.spark.sql.execution.{ReusedSubqueryExec, SubqueryExec}
import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanHelper, AQEPropagateEmptyRelation}
+import org.apache.spark.sql.execution.planmerging.MergeSubplans
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.types.{IntegerType, StructType}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2Suite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2Suite.scala
index 846600ca5b74..1ae10fbf9679 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2Suite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2Suite.scala
@@ -1368,8 +1368,11 @@ class DataSourceV2Suite extends SharedSparkSession with AdaptiveSparkPlanHelper
"column i should be pruned from the scan output")
assert(!scan.scan.asInstanceOf[SupportsReportStatistics].reflectsFullyPushedDownFilters(),
"fake connector should request Spark post-pushdown adjustments")
- assert(scan.pushedFilters.isEmpty,
- "pushedFilters should drop filters whose references were pruned")
+ // SPARK-40259 made pushedFilters the complete set: it keeps a fully-pushed filter even when
+ // the filter's column is pruned from the scan output (the adjustment below uses a separate,
+ // pruned-output-remapped set, so it still does not re-add i > 3).
+ assert(scan.pushedFilters.exists(hasIGt3),
+ "pushedFilters keeps i > 3 even though column i was pruned (complete-set semantics)")
val optimizedPlan = q.queryExecution.optimizedPlan
assert(!optimizedPlan.collect { case f: LogicalFilter => f }.exists(f =>
@@ -1509,33 +1512,34 @@ class DataSourceV2Suite extends SharedSparkSession with AdaptiveSparkPlanHelper
"pushedFilters should not contain the unsupported filter on column j")
}
- test("pushedFilters are remapped by ProjectionOverSchema after nested schema pruning") {
+ test("pushedFilters keep the relation-level struct type after nested schema pruning") {
val df = spark.read.format(classOf[NestedSchemaDataSourceV2].getName).load()
// NestedSchemaScanBuilder pushes GreaterThan on "s.a".
- // Selecting only s.a triggers nested schema pruning: s goes from struct<a,b> to struct<a>.
+ // Selecting only s.a triggers nested schema pruning: the scan output narrows s to struct<a>,
+ // but pushedFilters records the fully-pushed filter against the relation's pre-pruning schema,
+ // so its struct column keeps the full type struct<a, b>.
val q = df.select($"s.a").filter($"s.a" > 3)
checkAnswer(q, (4 until 10).map(i => Row(i)))
val scanRelation = getScanRelation(q)
assert(scanRelation.pushedFilters.nonEmpty,
"pushedFilters should be non-empty")
- // Find the struct attribute referenced by the pushed filter.
- // Before remapping it would have type struct<a,b>; after remapping, struct<a>.
val structAttrs = scanRelation.pushedFilters
.flatMap(_.collect { case a: AttributeReference if a.name == "s" => a })
assert(structAttrs.nonEmpty, "pushed filter should reference struct column s")
- val prunedStructType = structAttrs.head.dataType.asInstanceOf[StructType]
- assert(prunedStructType.fieldNames.toSeq == Seq("a"),
- s"struct column in pushed filter should be pruned to struct<a> but was $prunedStructType")
+ val structType = structAttrs.head.dataType.asInstanceOf[StructType]
+ assert(structType.fieldNames.toSeq == Seq("a", "b"),
+ s"struct column in pushed filter should keep type struct<a, b> but was $structType")
}
- test("pushedFilters drops filters referencing pruned nested struct fields") {
+ test("pushedFilters keep filters referencing pruned nested struct fields") {
// Disable constraint propagation so IsNotNull(s.a) is not added as a post-scan
// filter (it would keep field a alive in the struct).
withSQLConf(SQLConf.CONSTRAINT_PROPAGATION_ENABLED.key -> "false") {
val df = spark.read.format(classOf[NestedSchemaDataSourceV2].getName).load()
- // Filter on s.a but select only s.b. Column pruning narrows s to struct<b>,
- // so the pushed filter on s.a can't be remapped and should be dropped.
+ // Filter on s.a but select only s.b. Column pruning narrows the scan output's s to struct<b>,
+ // but pushedFilters keeps the fully-pushed filter on s.a (against the relation schema) so a
+ // later scan merge can re-enforce it.
val q = df.filter($"s.a" > 3).select($"s.b")
checkAnswer(q, (4 until 10).map(i => Row(-i)))
@@ -1544,8 +1548,8 @@ class DataSourceV2Suite extends SharedSparkSession with AdaptiveSparkPlanHelper
filter.collect { case a: AttributeReference if a.name == "s" => a }
.flatMap(_.dataType.asInstanceOf[StructType].fieldNames)
}
- assert(!referencedStructFields.contains("a"),
- "pushedFilters should not reference pruned nested field a")
+ assert(referencedStructFields.contains("a"),
+ "pushedFilters should keep the filter referencing nested field a")
}
}
@@ -1574,6 +1578,32 @@ class DataSourceV2Suite extends SharedSparkSession with AdaptiveSparkPlanHelper
"Canonicalized instances with equivalent pushedFilters should be equal")
}
+ test("scan canonicalization distinguishes different pushedFilters") {
+ // Negative counterpart to the test above: two scans of the same relation whose pushedFilters
+ // differ must NOT canonicalize equal. This inequality is what lets MergeSubplans tell the two
+ // scans apart; if they compared equal it could treat them as identical and apply one side's
+ // filter to both -- a wrong-answer bug.
+ val table = new SimpleDataSourceV2().getTable(CaseInsensitiveStringMap.empty())
+
+ val relation1 = DataSourceV2Relation.create(
+ table, None, None, CaseInsensitiveStringMap.empty())
+ val relation2 = DataSourceV2Relation.create(
+ table, None, None, CaseInsensitiveStringMap.empty())
+ val scan1 = relation1.table.asReadable.newScanBuilder(relation1.options).build()
+ val scan2 = relation2.table.asReadable.newScanBuilder(relation2.options).build()
+
+ val filter1 = CatalystGreaterThan(relation1.output.head, CatalystLiteral(3))
+ val filter2 = CatalystGreaterThan(relation2.output.head, CatalystLiteral(5))
+
+ val scanRelation1 = DataSourceV2ScanRelation(relation1, scan1, relation1.output,
+ pushedFilters = Seq(filter1))
+ val scanRelation2 = DataSourceV2ScanRelation(relation2, scan2, relation2.output,
+ pushedFilters = Seq(filter2))
+
+ assert(scanRelation1.canonicalized != scanRelation2.canonicalized,
+ "Canonicalized instances with different pushedFilters must not be equal")
+ }
+
test("pushedFilters excludes non-deterministic filters") {
val df = spark.read.format(classOf[AdvancedDataSourceV2].getName).load()
// i > 3 is pushable and deterministic; rand() > 0.5 is non-deterministic and not pushable.
@@ -1637,20 +1667,23 @@ class DataSourceV2Suite extends SharedSparkSession with AdaptiveSparkPlanHelper
"non-deterministic filter should be retained as a post-scan Filter")
}
- test("pushedFilters drops filters referencing pruned columns") {
+ test("pushedFilters keep filters referencing pruned columns") {
// Disable constraint propagation so IsNotNull(i) is not added (it would keep
// column i in the scan output). This simulates a connector that pushes IsNotNull.
withSQLConf(SQLConf.CONSTRAINT_PROPAGATION_ENABLED.key -> "false") {
val df = spark.read.format(classOf[AdvancedDataSourceV2].getName).load()
- // i > 3 is fully pushed; selecting only j causes column pruning to drop i.
+ // i > 3 is fully pushed; selecting only j causes column pruning to drop i from the scan
+ // output, but pushedFilters keeps the fully-pushed filter on i (against the relation schema)
+ // so a later scan merge can re-enforce it.
val q = df.filter($"i" > 3).select($"j")
checkAnswer(q, (4 until 10).map(i => Row(-i)))
val scanRelation = getScanRelation(q)
assert(!scanRelation.output.exists(_.name == "i"),
"column i should be pruned from scan output")
- assert(scanRelation.pushedFilters.isEmpty,
- "pushedFilters should drop filters referencing pruned columns")
+ val referencedCols = scanRelation.pushedFilters.flatMap(_.references.map(_.name)).toSet
+ assert(referencedCols.contains("i"),
+ "pushedFilters should keep the filter referencing the pruned column i")
}
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/DSv2PlanMergingSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/DSv2PlanMergingSuite.scala
new file mode 100644
index 000000000000..ad40f8c9e1a9
--- /dev/null
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/DSv2PlanMergingSuite.scala
@@ -0,0 +1,172 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.planmerging
+
+import org.scalatest.BeforeAndAfter
+
+import org.apache.spark.sql.{DataFrame, QueryTest, Row}
+import org.apache.spark.sql.connector.FakeV2ProviderWithCustomSchema
+import org.apache.spark.sql.connector.catalog.InMemoryScanMergingPartitionFilterCatalog
+import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, DataSourceV2ScanRelation}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+
+/**
+ * End-to-end test for the DSv2 scan-merge gap this change closes: a source whose (equal) filter is
+ * strict only via the iterative PartitionPredicate second pass. When [[PlanMerger]] rebuilds the
+ * merged scan it drives the real
+ * [[org.apache.spark.sql.execution.datasources.v2.V2ScanRelationPushDown]] (Approach 2), so the
+ * second pass runs and the filter comes back strict -- the merge proceeds. A hand-rolled single
+ * push-predicates call would only run the first pass, see the filter come back post-scan,
+ * mis-classify it as non-strict and decline the merge (leaving two scans).
+ */
+class DSv2PlanMergingSuite extends QueryTest with SharedSparkSession
+ with BeforeAndAfter {
+
+ private val v2Source = classOf[FakeV2ProviderWithCustomSchema].getName
+ private val tbl = "scanmerge.t"
+
+ before {
+ spark.conf.set("spark.sql.catalog.scanmerge",
+ classOf[InMemoryScanMergingPartitionFilterCatalog].getName)
+ }
+
+ after {
+ spark.sessionState.catalogManager.reset()
+ spark.conf.unset("spark.sql.catalog.scanmerge")
+ }
+
+ private def v2Scans(df: DataFrame): Seq[DataSourceV2ScanRelation] =
+ df.queryExecution.optimizedPlan.collectWithSubqueries {
+ case s: DataSourceV2ScanRelation => s
+ }
+
+ // A successful DSv2 merge builds the scan and leaves NO bare DataSourceV2Relation in the plan.
+ // A leaked deferred scan (e.g. if a future recursion arm forwarded `deferredScan` without
+ // building it) would surface as an unbuilt placeholder relation the read path cannot plan --
+ // cheap guard so a regression fails loudly here, not as a silently-declined merge (see
+ // MergeContext).
+ private def assertNoPlaceholderRelation(df: DataFrame): Unit =
+ assert(
+ df.queryExecution.optimizedPlan.collectWithSubqueries {
+ case r: DataSourceV2Relation => r
+ }.isEmpty,
+ s"unbuilt placeholder DataSourceV2Relation left in plan:\n${df.queryExecution.optimizedPlan}")
+
+ test("SPARK-40259: merge two DSv2 scans whose filter is strict only via the second pass") {
+ withTable(tbl) {
+ sql(s"CREATE TABLE $tbl (part_col string, c1 int, c2 int) USING $v2Source " +
+ "PARTITIONED BY (part_col)")
+ sql(s"INSERT INTO $tbl VALUES ('a', 1, 10), ('a', 2, 20), ('b', 3, 30)")
+
+ // `part_col IN ('a')` is untranslatable/returned in the first pass and only accepted (strict)
+ // in the iterative PartitionPredicate second pass. The two scalar subqueries differ only in
+ // their projected data column, so PlanMerger fuses them into a single scan reading {c1, c2}.
+ val df = sql(
+ s"""
+ |SELECT
+ | (SELECT max(c1) FROM $tbl WHERE part_col IN ('a')) AS m1,
+ | (SELECT max(c2) FROM $tbl WHERE part_col IN ('a')) AS m2
+ |""".stripMargin)
+
+ // Correctness: the merged scan must still enforce the partition filter. Reading all
+ // partitions would give (3, 30) instead of the filtered (2, 20).
+ checkAnswer(df, Row(2, 20))
+
+ // The merged subquery is referenced once per scalar subquery, so the logical plan duplicates
+ // it (physical planning reuses it). Dedupe by canonical form: a successful merge leaves a
+ // single distinct scan reading the union of both columns; declining would leave two distinct
+ // scans, one per column.
+ val scans = v2Scans(df)
+ assert(scans.nonEmpty, s"expected a DSv2 scan:\n${df.queryExecution.optimizedPlan}")
+ assert(scans.map(_.canonicalized).distinct.length == 1,
+ s"the two scans should be fused into one:\n${df.queryExecution.optimizedPlan}")
+ val scan = scans.head
+ assert(scan.output.map(_.name).toSet == Set("c1", "c2"),
+ s"the merged scan should read the union of both columns; got ${scan.output}")
+ // The filter must be re-enforced strictly on the merged scan (present in pushedFilters),
+ // which only happens because the rebuild runs the iterative second pass.
+ assert(scan.pushedFilters.exists(_.references.exists(_.name == "part_col")),
+ s"the part_col filter should be re-pushed strict onto the merged scan; " +
+ s"got pushedFilters=${scan.pushedFilters.mkString("[", ", ", "]")}")
+ assertNoPlaceholderRelation(df)
+ }
+ }
+
+ test("SPARK-40259: do not merge DSv2 scans with different strict partition filters") {
+ withTable(tbl) {
+ sql(s"CREATE TABLE $tbl (part_col string, c1 int, c2 int) USING $v2Source " +
+ "PARTITIONED BY (part_col)")
+ sql(s"INSERT INTO $tbl VALUES ('a', 1, 10), ('a', 2, 20), ('b', 3, 30)")
+
+ // Same table and shape, but different (strict, fully-enforced) partition filters. The strict
+ // filters are unequal, so the two scans must NOT fuse -- otherwise one side's partition would
+ // be read for both, giving a wrong answer.
+ val df = sql(
+ s"""
+ |SELECT
+ | (SELECT max(c1) FROM $tbl WHERE part_col IN ('a')) AS m1,
+ | (SELECT max(c2) FROM $tbl WHERE part_col IN ('b')) AS m2
+ |""".stripMargin)
+
+ checkAnswer(df, Row(2, 30))
+
+ val scans = v2Scans(df)
+ assert(scans.map(_.canonicalized).distinct.length == 2,
+ s"scans with different partition filters must not be fused:\n" +
+ df.queryExecution.optimizedPlan)
+ }
+ }
+
+ test("SPARK-40259: merge two DSv2 scans with differing best-effort filters (OR-widen)") {
+ withTable(tbl) {
+ sql(s"CREATE TABLE $tbl (part_col string, c1 int, c2 int) USING $v2Source " +
+ "PARTITIONED BY (part_col)")
+ sql(s"INSERT INTO $tbl VALUES ('a', 1, 10), ('a', 2, 20), ('b', 3, 30)")
+
+ // c1 > 1 / c2 > 1 are data-column filters this source does not enforce (best-effort,
+ // post-scan), so the two scans carry no strict filter and fuse into one reading {c1, c2}.
+ // The differing post-scan filters are OR-widened above the merged scan (Phase 2) and each
+ // aggregate keeps its own filter. This runs the differing-filter path end to end, which the
+ // plan-shape tests in MergeSubplansSuite cover only structurally. (A rand() on one side is
+ // not added here: it would make the scalar subquery non-deterministic, so MergeSubplans would
+ // not extract it at all; the non-deterministic pruning drop is unit-covered in
+ // MergeSubplansSuite.)
+ withSQLConf(
+ SQLConf.MERGE_SUBPLANS_FILTER_PROPAGATION_ENABLED.key -> "true",
+ SQLConf.MERGE_SUBPLANS_SYMMETRIC_FILTER_PROPAGATION_ENABLED.key -> "true") {
+ val df = sql(
+ s"""
+ |SELECT
+ | (SELECT max(c1) FROM $tbl WHERE c1 > 1) AS m1,
+ | (SELECT max(c2) FROM $tbl WHERE c2 > 1) AS m2
+ |""".stripMargin)
+
+ // c1 > 1 -> {2, 3} -> max 3; c2 > 1 -> {10, 20, 30} -> max 30.
+ checkAnswer(df, Row(3, 30))
+
+ val scans = v2Scans(df)
+ assert(scans.map(_.canonicalized).distinct.length == 1,
+ s"the two scans should be fused into one:\n${df.queryExecution.optimizedPlan}")
+ assert(scans.head.output.map(_.name).toSet == Set("c1", "c2"),
+ s"the merged scan should read the union of both columns; got ${scans.head.output}")
+ assertNoPlaceholderRelation(df)
+ }
+ }
+ }
+}
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/MergeSubplansSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/MergeSubplansSuite.scala
similarity index 64%
rename from sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/MergeSubplansSuite.scala
rename to sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/MergeSubplansSuite.scala
index dfa17f926c5a..5e4b7fef314b 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/MergeSubplansSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/MergeSubplansSuite.scala
@@ -15,15 +15,23 @@
* limitations under the License.
*/
-package org.apache.spark.sql.catalyst.optimizer
+package org.apache.spark.sql.execution.planmerging
import org.apache.spark.sql.catalyst.dsl.expressions._
import org.apache.spark.sql.catalyst.dsl.plans._
-import org.apache.spark.sql.catalyst.expressions.{Alias, And, Attribute, CreateNamedStruct, GetStructField, If, Literal, Or, ScalarSubquery}
+import org.apache.spark.sql.catalyst.expressions.{Alias, And, Ascending, Attribute, AttributeReference, CreateNamedStruct, ExprId, GetStructField, If, Literal, Or, ScalarSubquery, SortOrder}
import org.apache.spark.sql.catalyst.plans._
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.catalyst.rules._
+import org.apache.spark.sql.catalyst.types.DataTypeUtils.toAttributes
+import org.apache.spark.sql.connector.catalog.{SupportsRead, Table, TableCapability}
+import org.apache.spark.sql.connector.expressions.{SortOrder => V2SortOrder}
+import org.apache.spark.sql.connector.expressions.filter.Predicate
+import org.apache.spark.sql.connector.read.{Scan, ScanBuilder, SupportsPushDownLimit, SupportsPushDownOffset, SupportsPushDownRequiredColumns, SupportsPushDownTableSample, SupportsPushDownTopN, SupportsPushDownV2Filters}
+import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, DataSourceV2ScanRelation, V2ScanRelationPushDown}
import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
class MergeSubplansSuite extends PlanTest {
@@ -1966,4 +1974,875 @@ class MergeSubplansSuite extends PlanTest {
comparePlans(Optimize.execute(originalQuery.analyze), originalQuery.analyze)
}
}
+
+ // ---- SPARK-40259: generic DSv2 scan merge ----
+
+ private val v2Table = new TestV2Table(StructType(Seq(
+ StructField("a", IntegerType), StructField("b", IntegerType), StructField("c", StringType))))
+
+ /** Same shape as [[v2Table]] but does NOT declare the `SCAN_MERGING` capability. */
+ private val v2TableNoMerge = new TestV2Table(StructType(Seq(
+ StructField("a", IntegerType), StructField("b", IntegerType), StructField("c", StringType))),
+ supportsScanMerging = false)
+
+ /** A `DataSourceV2ScanRelation` over `table` projecting only the given columns. */
+ private def v2ScanReadingOn(table: TestV2Table, cols: Seq[String]): DataSourceV2ScanRelation = {
+ val fullOutput = toAttributes(table.schema())
+ val relation =
+ DataSourceV2Relation(table, fullOutput, None, None, CaseInsensitiveStringMap.empty())
+ val output = cols.map(c => fullOutput.find(_.name == c).get)
+ val scan = TestV2Scan(StructType(output.map(a => StructField(a.name, a.dataType, a.nullable))))
+ // These stand in for a plain scan produced by the column-pruning path, which is mergeable.
+ DataSourceV2ScanRelation(relation, scan, output, mergeableScan = true)
+ }
+
+ /** A `DataSourceV2ScanRelation` over [[v2Table]] projecting only the given columns. */
+ private def v2ScanReading(cols: String*): DataSourceV2ScanRelation =
+ v2ScanReadingOn(v2Table, cols)
+
+ /** A `DataSourceV2ScanRelation` over a table without the `SCAN_MERGING` capability. */
+ private def v2ScanReadingNoMerge(cols: String*): DataSourceV2ScanRelation =
+ v2ScanReadingOn(v2TableNoMerge, cols)
+
+ private def v2Scans(plan: LogicalPlan): Seq[DataSourceV2ScanRelation] =
+ plan.collectWithSubqueries { case s: DataSourceV2ScanRelation => s }
+
+ /**
+ * Normalizes a merged plan so `comparePlans` can match it: (1) drops each DSv2 scan's
+ * dynamically-built Scan to a schema-only placeholder (the pushed describe() strings are asserted
+ * separately), and (2) resets the nested DataSourceV2Relation's output exprIds to a positional
+ * scheme. The reset is needed because `DataSourceV2ScanRelation` is a `LeafNode`, so its
+ * `relation` is a constructor arg rather than a child -- PlanTest's `normalizeExprIds` never
+ * recurses into it, and those exprIds otherwise differ between the expected and actual plans.
+ */
+ private def normalizeScans(plan: LogicalPlan): LogicalPlan = plan.transformWithSubqueries {
+ case s: DataSourceV2ScanRelation =>
+ val fixedRelation = s.relation.copy(
+ output = s.relation.output.zipWithIndex.map { case (a, i) => a.withExprId(ExprId(i)) })
+ s.copy(relation = fixedRelation, scan = TestV2Scan(s.scan.readSchema()))
+ }
+
+ // Normalize merged DSv2 scans before the standard comparison (a no-op on plans without them),
+ // so tests can call plain comparePlans. See normalizeScans for why this is needed.
+ override protected def comparePlans(
+ plan1: LogicalPlan,
+ plan2: LogicalPlan,
+ checkAnalysis: Boolean = true): Unit =
+ super.comparePlans(normalizeScans(plan1), normalizeScans(plan2), checkAnalysis)
+
+ test("SPARK-40259: merge DSv2 scans that differ only in projected columns") {
+ val sub1 = ScalarSubquery(v2ScanReading("a").groupBy()(sum($"a").as("sum_a")))
+ val sub2 = ScalarSubquery(v2ScanReading("b").groupBy()(sum($"b").as("sum_b")))
+ val originalQuery = testRelation.select(sub1, sub2)
+
+ // Expected: the two scans fuse into one reading {a, b}, feeding a single aggregate. No filters,
+ // so no OR-widen propagation.
+ val mergedScan = v2ScanReadingOn(v2Table, Seq("a", "b"))
+ val mergedSubquery = mergedScan
+ .groupBy()(sum($"a").as("sum_a"), sum($"b").as("sum_b"))
+ .select(CreateNamedStruct(Seq(
+ Literal("sum_a"), $"sum_a",
+ Literal("sum_b"), $"sum_b")).as("mergedValue"))
+ val analyzedMergedSubquery = mergedSubquery.analyze
+ val correctAnswer = WithCTE(
+ testRelation.select(
+ extractorExpression(0, analyzedMergedSubquery.output, 0),
+ extractorExpression(0, analyzedMergedSubquery.output, 1)),
+ Seq(definitionNode(analyzedMergedSubquery, 0)))
+ comparePlans(Optimize.execute(originalQuery.analyze), correctAnswer.analyze)
+ }
+
+ test("SPARK-40259: do not merge DSv2 scans when a pushdown is merge-blocking") {
+ val sub1 = ScalarSubquery(v2ScanReading("a").groupBy()(sum($"a").as("sum_a")))
+ val blockingScan = v2ScanReading("b").copy(mergeableScan = false)
+ val sub2 = ScalarSubquery(blockingScan.groupBy()(sum($"b").as("sum_b")))
+ val originalQuery = testRelation.select(sub1, sub2)
+
+ // A merge-blocking pushdown declines the merge: the plan is left unchanged.
+ comparePlans(Optimize.execute(originalQuery.analyze), originalQuery.analyze)
+ }
+
+ test("SPARK-40259: do not merge DSv2 scans when a column is read at a nested-pruned type") {
+ // A struct column read at a narrower (nested-pruned) type carries GetStructField ordinals
+ // against that narrow layout. The merged scan reads the full struct, so remapping those
+ // ordinals onto it would read the wrong field -- the merge must decline. np reads s pruned to
+ // <b>, cp reads a plain column x; without the guard they fuse into a scan of {s:<a,b>, x}.
+ val schema = StructType(Seq(
+ StructField("s", StructType(Seq(
+ StructField("a", IntegerType), StructField("b", IntegerType)))),
+ StructField("x", IntegerType)))
+ val relation = DataSourceV2Relation(
+ new TestV2Table(schema), toAttributes(schema), None, None, CaseInsensitiveStringMap.empty())
+ val sFull = relation.output.find(_.name == "s").get
+ val xFull = relation.output.find(_.name == "x").get
+
+ // np reads s at struct<b> (nested-pruned); cp reads x. Both opt in and are otherwise mergeable.
+ val sPruned = AttributeReference(
+ "s", StructType(Seq(StructField("b", IntegerType))))(sFull.exprId)
+ val npScan = DataSourceV2ScanRelation(relation,
+ TestV2Scan(StructType(Seq(StructField("s", sPruned.dataType)))),
+ Seq(sPruned), mergeableScan = true)
+ val cpScan = DataSourceV2ScanRelation(relation,
+ TestV2Scan(StructType(Seq(StructField("x", IntegerType)))),
+ Seq(xFull), mergeableScan = true)
+
+ val sub1 = ScalarSubquery(npScan.groupBy()(sum(GetStructField(sPruned, 0)).as("sb")))
+ val sub2 = ScalarSubquery(cpScan.groupBy()(sum(xFull).as("sx")))
+ val originalQuery = testRelation.select(sub1, sub2)
+
+ // The nested-pruned column declines the merge: the plan is left unchanged.
+ comparePlans(Optimize.execute(originalQuery.analyze), originalQuery.analyze)
+ }
+
+ test("SPARK-40259: merge DSv2 scans with different filters via OR-widen propagation") {
+ val sub1 = ScalarSubquery(
+ v2ScanReading("a").where($"a" > 1).groupBy()(sum($"a").as("sum_a"))) // cp
+ val sub2 = ScalarSubquery(
+ v2ScanReading("b").where($"b" > 2).groupBy()(sum($"b").as("sum_b"))) // np
+ val originalQuery = testRelation.select(sub1, sub2)
+
+ // Expected: the two scans fuse into one reading {a, b}; the differing filters become
+ // propagatedFilter aliases OR-widened in a Filter, and each side's aggregate carries its
+ // filter as a FILTER clause (np = sub2 gets id 0, cp = sub1 gets id 1).
+ val mergedScan = v2ScanReadingOn(v2Table, Seq("a", "b"))
+ val npFilterAlias = Alias($"b" > 2, "propagatedFilter_0")()
+ val cpFilterAlias = Alias($"a" > 1, "propagatedFilter_1")()
+ val npFilter = npFilterAlias.toAttribute
+ val cpFilter = cpFilterAlias.toAttribute
+ val mergedSubquery = mergedScan
+ .select(mergedScan.output ++ Seq(npFilterAlias, cpFilterAlias): _*)
+ .where(Or(npFilter, cpFilter))
+ .groupBy()(
+ sum($"a", Some(cpFilter)).as("sum_a"),
+ sum($"b", Some(npFilter)).as("sum_b"))
+ .select(CreateNamedStruct(Seq(
+ Literal("sum_a"), $"sum_a",
+ Literal("sum_b"), $"sum_b")).as("mergedValue"))
+ val analyzedMergedSubquery = mergedSubquery.analyze
+ val correctAnswer = WithCTE(
+ testRelation.select(
+ extractorExpression(0, analyzedMergedSubquery.output, 0),
+ extractorExpression(0, analyzedMergedSubquery.output, 1)),
+ Seq(definitionNode(analyzedMergedSubquery, 0)))
+
+ withSQLConf(
+ SQLConf.MERGE_SUBPLANS_FILTER_PROPAGATION_ENABLED.key -> "true",
+ SQLConf.MERGE_SUBPLANS_SYMMETRIC_FILTER_PROPAGATION_ENABLED.key -> "true") {
+ val optimized = Optimize.execute(originalQuery.analyze)
+ comparePlans(optimized, correctAnswer.analyze)
+ // Additional: the merged scan re-pushes OR(a > 1, b > 2) for row-group pruning, which the
+ // structural comparison above normalizes out.
+ val pushed = v2Scans(optimized).head.scan.asInstanceOf[TestV2Scan].pushed
+ assert(pushed.nonEmpty && pushed.mkString.contains("a") && pushed.mkString.contains("b"),
+ s"expected an OR pruning predicate over a and b pushed to the merged scan, got: $pushed")
+ }
+ }
+
+ test("SPARK-40259: DSv2 symmetric merge honors its own config when the general one is off") {
+ // Two DSv2 scans with equal (empty) strict filters but differing best-effort filters. The
+ // general symmetric propagation is off, so a non-DSv2 pair would not merge; the DSv2-specific
+ // config allows it, since the equal strict filters mean both sides read the same base set and
+ // only the best-effort OR pruning is broadened.
+ val sub1 = ScalarSubquery(
+ v2ScanReading("a").where($"a" > 1).groupBy()(sum($"a").as("sum_a"))) // cp
+ val sub2 = ScalarSubquery(
+ v2ScanReading("b").where($"b" > 2).groupBy()(sum($"b").as("sum_b"))) // np
+ val originalQuery = testRelation.select(sub1, sub2)
+
+ // Expected when the DSv2 config allows it: the two scans fuse into one reading {a, b}; the
+ // differing filters become propagatedFilter aliases OR-widened in a Filter, and each side's
+ // aggregate carries its filter as a FILTER clause (np = sub2 gets id 0, cp = sub1 gets id 1).
+ val mergedScan = v2ScanReadingOn(v2Table, Seq("a", "b"))
+ val npFilterAlias = Alias($"b" > 2, "propagatedFilter_0")()
+ val cpFilterAlias = Alias($"a" > 1, "propagatedFilter_1")()
+ val npFilter = npFilterAlias.toAttribute
+ val cpFilter = cpFilterAlias.toAttribute
+ val mergedSubquery = mergedScan
+ .select(mergedScan.output ++ Seq(npFilterAlias, cpFilterAlias): _*)
+ .where(Or(npFilter, cpFilter))
+ .groupBy()(
+ sum($"a", Some(cpFilter)).as("sum_a"),
+ sum($"b", Some(npFilter)).as("sum_b"))
+ .select(CreateNamedStruct(Seq(
+ Literal("sum_a"), $"sum_a",
+ Literal("sum_b"), $"sum_b")).as("mergedValue"))
+ val analyzedMergedSubquery = mergedSubquery.analyze
+ val correctAnswer = WithCTE(
+ testRelation.select(
+ extractorExpression(0, analyzedMergedSubquery.output, 0),
+ extractorExpression(0, analyzedMergedSubquery.output, 1)),
+ Seq(definitionNode(analyzedMergedSubquery, 0)))
+
+ withSQLConf(
+ SQLConf.MERGE_SUBPLANS_FILTER_PROPAGATION_ENABLED.key -> "true",
+ SQLConf.MERGE_SUBPLANS_SYMMETRIC_FILTER_PROPAGATION_ENABLED.key -> "false",
+ SQLConf.MERGE_SUBPLANS_DSV2_SYMMETRIC_FILTER_PROPAGATION_ENABLED.key -> "true") {
+ comparePlans(Optimize.execute(originalQuery.analyze), correctAnswer.analyze)
+ }
+
+ // With the DSv2 config also off, the differing-filter pair must NOT merge (plan unchanged).
+ withSQLConf(
+ SQLConf.MERGE_SUBPLANS_FILTER_PROPAGATION_ENABLED.key -> "true",
+ SQLConf.MERGE_SUBPLANS_SYMMETRIC_FILTER_PROPAGATION_ENABLED.key -> "false",
+ SQLConf.MERGE_SUBPLANS_DSV2_SYMMETRIC_FILTER_PROPAGATION_ENABLED.key -> "false") {
+ comparePlans(Optimize.execute(originalQuery.analyze), originalQuery.analyze)
+ }
+ }
+
+ test("SPARK-40259: DSv2 merge fact propagates to an outer stacked Filter pair") {
+ // Stacked Filters over each DSv2 scan (cp = sub1: inner a>1, outer a<100; np = sub2: inner b>2,
+ // outer b<200). The leaf scans merge and the deferral is consumed at the INNER (Filter, Filter)
+ // pair; the OUTER pair sees dsv2DeferredScan = None. The merge only fuses if the "a DSv2 merge
+ // happened below" fact (TryMergeResult.dsv2Merged) still reaches the outer pair so it applies
+ // the DSv2-symmetric exemption -- dsv2DeferredScan alone (consumed at the inner pair) would
+ // not. Optimize runs only MergeSubplans, so the stacked Filters are not combined before the
+ // rule sees them.
+ val sub1 = ScalarSubquery(
+ v2ScanReading("a").where($"a" > 1).where($"a" < 100).groupBy()(sum($"a").as("sum_a")))
+ val sub2 = ScalarSubquery(
+ v2ScanReading("b").where($"b" > 2).where($"b" < 200).groupBy()(sum($"b").as("sum_b")))
+ val originalQuery = testRelation.select(sub1, sub2)
+
+ // Expected when the DSv2 config is on: the scans fuse into one reading {a, b}. Merge traversal
+ // (inner-to-outer):
+ // Inner pair (np: b>2, cp: a>1) OR-widens:
+ // f0 = Alias(b > 2, "propagatedFilter_0") -- np / sum_b
+ // f1 = Alias(a > 1, "propagatedFilter_1") -- cp / sum_a
+ // Outer pair (np: b<200, cp: a<100) OR-widens, AND-combining each side's inner filter:
+ // f2 = Alias(AND(f0, b < 200), "propagatedFilter_2") -- np
+ // f3 = Alias(AND(f1, a < 100), "propagatedFilter_3") -- cp
+ // Aggregate consumes f2/f3 as FILTER clauses: sum_a FILTER f3, sum_b FILTER f2 -- i.e. each
+ // side's per-side AND(inner, outer).
+ val mergedScan = v2ScanReadingOn(v2Table, Seq("a", "b"))
+ val f0Alias = Alias($"b" > 2, "propagatedFilter_0")()
+ val f1Alias = Alias($"a" > 1, "propagatedFilter_1")()
+ val f0 = f0Alias.toAttribute
+ val f1 = f1Alias.toAttribute
+ val innerFilter = mergedScan
+ .select(mergedScan.output ++ Seq(f0Alias, f1Alias): _*)
+ .where(Or(f0, f1))
+ val f2Alias = Alias(And(f0, $"b" < 200), "propagatedFilter_2")()
+ val f3Alias = Alias(And(f1, $"a" < 100), "propagatedFilter_3")()
+ val f2 = f2Alias.toAttribute
+ val f3 = f3Alias.toAttribute
+ val mergedSubquery = innerFilter
+ .select(innerFilter.output ++ Seq(f2Alias, f3Alias): _*)
+ .where(Or(f2, f3))
+ .groupBy()(
+ sum($"a", Some(f3)).as("sum_a"),
+ sum($"b", Some(f2)).as("sum_b"))
+ .select(CreateNamedStruct(Seq(
+ Literal("sum_a"), $"sum_a",
+ Literal("sum_b"), $"sum_b")).as("mergedValue"))
+ val analyzedMergedSubquery = mergedSubquery.analyze
+ val correctAnswer = WithCTE(
+ testRelation.select(
+ extractorExpression(0, analyzedMergedSubquery.output, 0),
+ extractorExpression(0, analyzedMergedSubquery.output, 1)),
+ Seq(definitionNode(analyzedMergedSubquery, 0)))
+
+ // General symmetric off; the DSv2-specific config on. The flag lets BOTH the inner and outer
+ // Filter pair OR-widen, so the two scans fuse into one and each aggregate FILTER is the
+ // per-side AND(inner, outer).
+ withSQLConf(
+ SQLConf.MERGE_SUBPLANS_FILTER_PROPAGATION_ENABLED.key -> "true",
+ SQLConf.MERGE_SUBPLANS_SYMMETRIC_FILTER_PROPAGATION_ENABLED.key -> "false",
+ SQLConf.MERGE_SUBPLANS_DSV2_SYMMETRIC_FILTER_PROPAGATION_ENABLED.key -> "true") {
+ comparePlans(Optimize.execute(originalQuery.analyze), correctAnswer.analyze)
+ }
+
+ // With the DSv2 config off the inner pair already declines, so nothing fuses (plan unchanged).
+ withSQLConf(
+ SQLConf.MERGE_SUBPLANS_FILTER_PROPAGATION_ENABLED.key -> "true",
+ SQLConf.MERGE_SUBPLANS_SYMMETRIC_FILTER_PROPAGATION_ENABLED.key -> "false",
+ SQLConf.MERGE_SUBPLANS_DSV2_SYMMETRIC_FILTER_PROPAGATION_ENABLED.key -> "false") {
+ comparePlans(Optimize.execute(originalQuery.analyze), originalQuery.analyze)
+ }
+ }
+
+ test("SPARK-40259: merge proceeds without pruning when the source rejects the pruning") {
+ val rejecting = new TestV2Table(
+ StructType(Seq(StructField("a", IntegerType), StructField("b", IntegerType))),
+ acceptsFilters = false)
+ val s1 = v2ScanReadingOn(rejecting, Seq("a"))
+ val s2 = v2ScanReadingOn(rejecting, Seq("b"))
+ val sub1 = ScalarSubquery(s1.where(s1.output.head > 1).groupBy()(sum($"a").as("sum_a")))
+ val sub2 = ScalarSubquery(s2.where(s2.output.head > 2).groupBy()(sum($"b").as("sum_b")))
+ val originalQuery = testRelation.select(sub1, sub2)
+
+ // Expected: the same OR-widen merge as the accepting case -- only the pushed pruning differs
+ // (a rejecting source records none). np = sub2 -> id 0, cp = sub1 -> id 1.
+ val mergedScan = v2ScanReadingOn(rejecting, Seq("a", "b"))
+ val npFilterAlias = Alias($"b" > 2, "propagatedFilter_0")()
+ val cpFilterAlias = Alias($"a" > 1, "propagatedFilter_1")()
+ val npFilter = npFilterAlias.toAttribute
+ val cpFilter = cpFilterAlias.toAttribute
+ val mergedSubquery = mergedScan
+ .select(mergedScan.output ++ Seq(npFilterAlias, cpFilterAlias): _*)
+ .where(Or(npFilter, cpFilter))
+ .groupBy()(
+ sum($"a", Some(cpFilter)).as("sum_a"),
+ sum($"b", Some(npFilter)).as("sum_b"))
+ .select(CreateNamedStruct(Seq(
+ Literal("sum_a"), $"sum_a",
+ Literal("sum_b"), $"sum_b")).as("mergedValue"))
+ val analyzedMergedSubquery = mergedSubquery.analyze
+ val correctAnswer = WithCTE(
+ testRelation.select(
+ extractorExpression(0, analyzedMergedSubquery.output, 0),
+ extractorExpression(0, analyzedMergedSubquery.output, 1)),
+ Seq(definitionNode(analyzedMergedSubquery, 0)))
+
+ withSQLConf(
+ SQLConf.MERGE_SUBPLANS_FILTER_PROPAGATION_ENABLED.key -> "true",
+ SQLConf.MERGE_SUBPLANS_SYMMETRIC_FILTER_PROPAGATION_ENABLED.key -> "true") {
+ val optimized = Optimize.execute(originalQuery.analyze)
+ comparePlans(optimized, correctAnswer.analyze)
+ // Phase 2 degrades gracefully: the merge happens; a rejecting source records no pruning.
+ assert(v2Scans(optimized).head.scan.asInstanceOf[TestV2Scan].pushed.isEmpty,
+ "a rejecting source must not record any pushed pruning predicate")
+ }
+ }
+
+ test("SPARK-40259: do not merge when a strict pushed filter is not re-enforced on rebuild") {
+ // The scans claim a strict filter on "a" (in pushedFilters), but the source enforces nothing
+ // strictly (empty strictColumns -> everything is best-effort). Re-pushing the "strict" filter
+ // would leave it merely best-effort with nothing above to re-check it, so the merge must abort.
+ val bestEffort = new TestV2Table(StructType(Seq(
+ StructField("a", IntegerType), StructField("b", IntegerType), StructField("c", StringType))))
+ val s1 = v2ScanReadingOn(bestEffort, Seq("a", "b"))
+ val s2 = v2ScanReadingOn(bestEffort, Seq("a", "c"))
+ val sub1 = ScalarSubquery(
+ s1.copy(pushedFilters = Seq(s1.output.find(_.name == "a").get > 0))
+ .groupBy()(sum($"b").as("sum_b")))
+ val sub2 = ScalarSubquery(
+ s2.copy(pushedFilters = Seq(s2.output.find(_.name == "a").get > 0))
+ .groupBy()(sum($"c").as("max_c")))
+ val originalQuery = testRelation.select(sub1, sub2)
+
+ // A strict filter the rebuilt scan cannot re-enforce declines the merge: plan left unchanged.
+ comparePlans(Optimize.execute(originalQuery.analyze), originalQuery.analyze)
+ }
+
+ test("SPARK-40259: do not merge DSv2 scans of a table without the SCAN_MERGING capability") {
+ val sub1 = ScalarSubquery(v2ScanReadingNoMerge("a").groupBy()(sum($"a").as("sum_a")))
+ val sub2 = ScalarSubquery(v2ScanReadingNoMerge("b").groupBy()(sum($"b").as("sum_b")))
+ val originalQuery = testRelation.select(sub1, sub2)
+
+ // A table without the SCAN_MERGING capability declines the merge: plan left unchanged.
+ comparePlans(Optimize.execute(originalQuery.analyze), originalQuery.analyze)
+ }
+
+ test("SPARK-40259: merge DSv2 scans with identical filters re-pushes the pruning") {
+ // Identical post-scan filter (a > 1), differing columns: the two scans fuse and the identical
+ // condition is re-pushed to the merged scan for row-group pruning (Phase 2, single condition).
+ val sub1 = ScalarSubquery(
+ v2ScanReading("a", "b").where($"a" > 1).groupBy()(sum($"b").as("sum_b")))
+ val sub2 = ScalarSubquery(
+ v2ScanReading("a", "c").where($"a" > 1).groupBy()(sum($"c").as("sum_c")))
+ val originalQuery = testRelation.select(sub1, sub2)
+
+ // Expected: the identical filter a > 1 kept as a single Filter above the merged scan {a, b, c};
+ // no OR-widen (the conditions match). The aggregates read b and c.
+ val mergedScan = v2ScanReadingOn(v2Table, Seq("a", "b", "c"))
+ val mergedSubquery = mergedScan.where($"a" > 1)
+ .groupBy()(sum($"b").as("sum_b"), sum($"c").as("sum_c"))
+ .select(CreateNamedStruct(Seq(
+ Literal("sum_b"), $"sum_b",
+ Literal("sum_c"), $"sum_c")).as("mergedValue"))
+ val analyzedMergedSubquery = mergedSubquery.analyze
+ val correctAnswer = WithCTE(
+ testRelation.select(
+ extractorExpression(0, analyzedMergedSubquery.output, 0),
+ extractorExpression(0, analyzedMergedSubquery.output, 1)),
+ Seq(definitionNode(analyzedMergedSubquery, 0)))
+ val optimized = Optimize.execute(originalQuery.analyze)
+ comparePlans(optimized, correctAnswer.analyze)
+ // Additional: the identical a > 1 is re-pushed to the merged scan for pruning.
+ val pushed = v2Scans(optimized).head.scan.asInstanceOf[TestV2Scan].pushed
+ assert(pushed.nonEmpty && pushed.mkString.contains("a"),
+ s"expected the identical filter a > 1 to be re-pushed to the merged scan, got: $pushed")
+ }
+
+ test("SPARK-40259: merge three DSv2 scans with differing filters re-pushes the full OR") {
+ // Three-way merge exercises the tagged (MERGED_FILTER_TAG) branch: the leaf re-merge rebuilds
+ // the scan strict-only each round, so the tagged branch must re-establish the full 3-way OR.
+ val t = new TestV2Table(StructType(Seq(StructField("a", IntegerType),
+ StructField("b", IntegerType), StructField("d", IntegerType))))
+ val sub1 = ScalarSubquery(
+ v2ScanReadingOn(t, Seq("a")).where($"a" > 1).groupBy()(sum($"a").as("s1")))
+ val sub2 = ScalarSubquery(
+ v2ScanReadingOn(t, Seq("b")).where($"b" > 2).groupBy()(sum($"b").as("s2")))
+ val sub3 = ScalarSubquery(
+ v2ScanReadingOn(t, Seq("d")).where($"d" > 3).groupBy()(sum($"d").as("s3")))
+ val originalQuery = testRelation.select(sub1, sub2, sub3)
+
+ // Expected: one merged scan reading {a, b, d}; step 1 merges sub1 (cp) + sub2 (np) ->
+ // propagatedFilter_0 = b > 2 (np), _1 = a > 1 (cp); step 2 merges sub3 (np) -> _2 = d > 3,
+ // extending the OR. Each aggregate carries its side's FILTER.
+ val mergedScan = v2ScanReadingOn(t, Seq("a", "b", "d"))
+ val npFilter0Alias = Alias($"b" > 2, "propagatedFilter_0")()
+ val cpFilter0Alias = Alias($"a" > 1, "propagatedFilter_1")()
+ val npFilter1Alias = Alias($"d" > 3, "propagatedFilter_2")()
+ val npFilter0 = npFilter0Alias.toAttribute
+ val cpFilter0 = cpFilter0Alias.toAttribute
+ val npFilter1 = npFilter1Alias.toAttribute
+ val mergedSubquery = mergedScan
+ .select(mergedScan.output ++ Seq(npFilter0Alias, cpFilter0Alias, npFilter1Alias): _*)
+ .where(Or(Or(npFilter0, cpFilter0), npFilter1))
+ .groupBy()(
+ sum($"a", Some(cpFilter0)).as("s1"),
+ sum($"b", Some(npFilter0)).as("s2"),
+ sum($"d", Some(npFilter1)).as("s3"))
+ .select(CreateNamedStruct(Seq(
+ Literal("s1"), $"s1",
+ Literal("s2"), $"s2",
+ Literal("s3"), $"s3")).as("mergedValue"))
+ val analyzedMergedSubquery = mergedSubquery.analyze
+ val correctAnswer = WithCTE(
+ testRelation.select(
+ extractorExpression(0, analyzedMergedSubquery.output, 0),
+ extractorExpression(0, analyzedMergedSubquery.output, 1),
+ extractorExpression(0, analyzedMergedSubquery.output, 2)),
+ Seq(definitionNode(analyzedMergedSubquery, 0)))
+
+ withSQLConf(
+ SQLConf.MERGE_SUBPLANS_FILTER_PROPAGATION_ENABLED.key -> "true",
+ SQLConf.MERGE_SUBPLANS_SYMMETRIC_FILTER_PROPAGATION_ENABLED.key -> "true") {
+ val optimized = Optimize.execute(originalQuery.analyze)
+ comparePlans(optimized, correctAnswer.analyze)
+ // Additional: the merged scan carries the full 3-way OR pruning over a, b and d.
+ val pushed = v2Scans(optimized).head.scan.asInstanceOf[TestV2Scan].pushed.mkString
+ assert(pushed.contains("a") && pushed.contains("b") && pushed.contains("d"),
+ s"expected an OR pruning over a, b and d pushed to the merged scan, got: $pushed")
+ }
+ }
+
+ test("SPARK-40259: three-way merge reuses an existing propagated filter (reuse branch)") {
+ // sub3's filter (a > 1) matches sub1's, so the tagged-filter merge takes the reuse branch:
+ // no new propagatedFilter alias is created and the OR stays two-way; sub3's aggregate reuses
+ // the existing filter attribute. This exercises the reuse branch against a DSv2 scan, which
+ // the all-distinct-filters three-way test above does not.
+ val t = new TestV2Table(StructType(Seq(StructField("a", IntegerType),
+ StructField("b", IntegerType), StructField("c", IntegerType))))
+ val sub1 = ScalarSubquery(
+ v2ScanReadingOn(t, Seq("a")).where($"a" > 1).groupBy()(sum($"a").as("s1")))
+ val sub2 = ScalarSubquery(
+ v2ScanReadingOn(t, Seq("b")).where($"b" > 2).groupBy()(sum($"b").as("s2")))
+ val sub3 = ScalarSubquery(
+ v2ScanReadingOn(t, Seq("a", "c")).where($"a" > 1).groupBy()(sum($"c").as("s3")))
+ val originalQuery = testRelation.select(sub1, sub2, sub3)
+
+ // Expected: one merged scan reading {a, b, c}; step 1 merges sub1 (cp) + sub2 (np) ->
+ // propagatedFilter_0 = b > 2 (np), _1 = a > 1 (cp). Step 2 merges sub3 (np, a > 1): its
+ // condition matches _1, so no new alias is created and the OR stays two-way; sub3 reuses _1.
+ val mergedScan = v2ScanReadingOn(t, Seq("a", "b", "c"))
+ val npFilterAlias = Alias($"b" > 2, "propagatedFilter_0")()
+ val cpFilterAlias = Alias($"a" > 1, "propagatedFilter_1")()
+ val npFilter = npFilterAlias.toAttribute
+ val cpFilter = cpFilterAlias.toAttribute
+ val mergedSubquery = mergedScan
+ .select(mergedScan.output ++ Seq(npFilterAlias, cpFilterAlias): _*)
+ .where(Or(npFilter, cpFilter))
+ .groupBy()(
+ sum($"a", Some(cpFilter)).as("s1"),
+ sum($"b", Some(npFilter)).as("s2"),
+ sum($"c", Some(cpFilter)).as("s3"))
+ .select(CreateNamedStruct(Seq(
+ Literal("s1"), $"s1",
+ Literal("s2"), $"s2",
+ Literal("s3"), $"s3")).as("mergedValue"))
+ val analyzedMergedSubquery = mergedSubquery.analyze
+ val correctAnswer = WithCTE(
+ testRelation.select(
+ extractorExpression(0, analyzedMergedSubquery.output, 0),
+ extractorExpression(0, analyzedMergedSubquery.output, 1),
+ extractorExpression(0, analyzedMergedSubquery.output, 2)),
+ Seq(definitionNode(analyzedMergedSubquery, 0)))
+
+ withSQLConf(
+ SQLConf.MERGE_SUBPLANS_FILTER_PROPAGATION_ENABLED.key -> "true",
+ SQLConf.MERGE_SUBPLANS_SYMMETRIC_FILTER_PROPAGATION_ENABLED.key -> "true") {
+ val optimized = Optimize.execute(originalQuery.analyze)
+ comparePlans(optimized, correctAnswer.analyze)
+ // The OR references only the two distinct conditions (a > 1, b > 2); c is not a filter.
+ val pushed = v2Scans(optimized).head.scan.asInstanceOf[TestV2Scan].pushed.mkString
+ assert(pushed.contains("a") && pushed.contains("b"),
+ s"expected an OR pruning over a and b pushed to the merged scan, got: $pushed")
+ }
+ }
+
+ test("SPARK-40259: a non-deterministic filter conjunct is not pushed as a pruning predicate") {
+ withSQLConf(
+ SQLConf.MERGE_SUBPLANS_FILTER_PROPAGATION_ENABLED.key -> "true",
+ SQLConf.MERGE_SUBPLANS_SYMMETRIC_FILTER_PROPAGATION_ENABLED.key -> "true") {
+ // Bare non-grouping aggregates (joined) go through the Aggregate extraction path, which --
+ // unlike the ScalarSubquery path -- does not gate on determinism, so the OR-widen pruning
+ // predicate carries a non-deterministic conjunct into the scan-merge Phase 2.
+ val agg1 = v2ScanReading("a").where(($"a" > 1) && (rand(0) < Literal(0.5)))
+ .groupBy()(sum($"a").as("s1"))
+ val agg2 = v2ScanReading("b").where($"b" > 2).groupBy()(sum($"b").as("s2"))
+ val originalQuery = agg1.join(agg2)
+
+ val scans = v2Scans(Optimize.execute(originalQuery.analyze))
+ assert(scans.length == 1, "the scans should still be fused")
+ val pushed = scans.head.scan.asInstanceOf[TestV2Scan].pushed.mkString
+ // Pruning predicate `(a > 1 AND rand() < 0.5) OR (b > 2)` is non-deterministic as a whole.
+ // A source that prunes on it uses its own rand() draw, while the enclosing Filter re-checks
+ // exactness with a different draw, so pruned rows would be lost. Best-effort pruning degrades
+ // gracefully: the predicate is dropped rather than weakened, so nothing is pushed and no
+ // rand() reaches the source. The merge itself is unaffected.
+ assert(!pushed.contains("RAND"),
+ s"the non-deterministic rand() conjunct must not be pushed as pruning, got: $pushed")
+ assert(pushed.isEmpty,
+ s"a non-deterministic pruning predicate must be dropped wholesale, got: $pushed")
+ }
+ }
+
+ test("SPARK-40259: merge DSv2 scans that pushed the same strict filters (re-push path)") {
+ // A source that fully enforces (strict) predicates on column "a".
+ val strictOnA = new TestV2Table(
+ StructType(Seq(StructField("a", IntegerType), StructField("b", IntegerType),
+ StructField("c", StringType))),
+ strictColumns = Set("a"))
+ val s1 = v2ScanReadingOn(strictOnA, Seq("a", "b"))
+ val s2 = v2ScanReadingOn(strictOnA, Seq("a", "c"))
+ val sub1 = ScalarSubquery(
+ s1.copy(pushedFilters = Seq(s1.output.find(_.name == "a").get > 0))
+ .groupBy()(sum($"b").as("sum_b")))
+ val sub2 = ScalarSubquery(
+ s2.copy(pushedFilters = Seq(s2.output.find(_.name == "a").get > 0))
+ .groupBy()(sum($"c").as("max_c")))
+ val originalQuery = testRelation.select(sub1, sub2)
+
+ // Expected: one merged scan reading {a, b, c} that keeps the strict a > 0 in pushedFilters; the
+ // two aggregates read b and c. No post-scan filter, so no OR-widen.
+ val mergedScan0 = v2ScanReadingOn(strictOnA, Seq("a", "b", "c"))
+ val mergedScan = mergedScan0.copy(
+ pushedFilters = Seq(mergedScan0.output.find(_.name == "a").get > 0))
+ val mergedSubquery = mergedScan
+ .groupBy()(sum($"b").as("sum_b"), sum($"c").as("max_c"))
+ .select(CreateNamedStruct(Seq(
+ Literal("sum_b"), $"sum_b",
+ Literal("max_c"), $"max_c")).as("mergedValue"))
+ val analyzedMergedSubquery = mergedSubquery.analyze
+ val correctAnswer = WithCTE(
+ testRelation.select(
+ extractorExpression(0, analyzedMergedSubquery.output, 0),
+ extractorExpression(0, analyzedMergedSubquery.output, 1)),
+ Seq(definitionNode(analyzedMergedSubquery, 0)))
+ val optimized = Optimize.execute(originalQuery.analyze)
+ comparePlans(optimized, correctAnswer.analyze)
+ // Additional: the strict a > 0 is re-enforced on the merged scan (normalized out above).
+ assert(v2Scans(optimized).head.pushedFilters.nonEmpty,
+ "the merged relation must carry the strict pushed filters")
+ }
+
+ test("SPARK-40259: merge scans with equal strict filters and differing post-scan filters") {
+ // Mixed source: "p" is fully enforced (strict), "a"/"b" are best-effort. Both sides push
+ // the same strict filter p > 0 and carry a differing post-scan filter (a > 1 / b > 2). The
+ // scans fuse on the equal strict p; the differing post-scan filters OR-widen above the
+ // merged scan, so the rebuild re-pushes p > 0 (strict) and (a > 1 OR b > 2) (pruning) at
+ // once -- the only path that drives buildMergedScan with both non-empty.
+ val mixed = new TestV2Table(
+ StructType(Seq(StructField("p", IntegerType), StructField("a", IntegerType),
+ StructField("b", IntegerType))),
+ strictColumns = Set("p"))
+ val s1 = v2ScanReadingOn(mixed, Seq("p", "a"))
+ val s2 = v2ScanReadingOn(mixed, Seq("p", "b"))
+ val sub1 = ScalarSubquery(
+ s1.copy(pushedFilters = Seq(s1.output.find(_.name == "p").get > 0))
+ .where($"a" > 1).groupBy()(sum($"a").as("s1")))
+ val sub2 = ScalarSubquery(
+ s2.copy(pushedFilters = Seq(s2.output.find(_.name == "p").get > 0))
+ .where($"b" > 2).groupBy()(sum($"b").as("s2")))
+ val originalQuery = testRelation.select(sub1, sub2)
+
+ // Expected: one merged scan reading {p, a, b} that keeps p > 0 strict (pushedFilters), with
+ // the differing post-scan filters OR-widened above it (np = sub2 -> id 0, cp = sub1 -> id 1).
+ val mergedScan0 = v2ScanReadingOn(mixed, Seq("p", "a", "b"))
+ val mergedScan = mergedScan0.copy(
+ pushedFilters = Seq(mergedScan0.output.find(_.name == "p").get > 0))
+ val npFilterAlias = Alias($"b" > 2, "propagatedFilter_0")()
+ val cpFilterAlias = Alias($"a" > 1, "propagatedFilter_1")()
+ val npFilter = npFilterAlias.toAttribute
+ val cpFilter = cpFilterAlias.toAttribute
+ val mergedSubquery = mergedScan
+ .select(mergedScan.output ++ Seq(npFilterAlias, cpFilterAlias): _*)
+ .where(Or(npFilter, cpFilter))
+ .groupBy()(
+ sum($"a", Some(cpFilter)).as("s1"),
+ sum($"b", Some(npFilter)).as("s2"))
+ .select(CreateNamedStruct(Seq(
+ Literal("s1"), $"s1",
+ Literal("s2"), $"s2")).as("mergedValue"))
+ val analyzedMergedSubquery = mergedSubquery.analyze
+ val correctAnswer = WithCTE(
+ testRelation.select(
+ extractorExpression(0, analyzedMergedSubquery.output, 0),
+ extractorExpression(0, analyzedMergedSubquery.output, 1)),
+ Seq(definitionNode(analyzedMergedSubquery, 0)))
+
+ withSQLConf(
+ SQLConf.MERGE_SUBPLANS_FILTER_PROPAGATION_ENABLED.key -> "true",
+ SQLConf.MERGE_SUBPLANS_SYMMETRIC_FILTER_PROPAGATION_ENABLED.key -> "true") {
+ val optimized = Optimize.execute(originalQuery.analyze)
+ comparePlans(optimized, correctAnswer.analyze)
+ // Additional: p stays strict (in pushedFilters) and the OR (a, b) is re-pushed as pruning,
+ // both normalized out of the structural comparison above.
+ val mergedResult = v2Scans(optimized).head
+ assert(mergedResult.pushedFilters.exists(_.references.exists(_.name == "p")),
+ s"the strict filter on p must stay enforced; got ${mergedResult.pushedFilters}")
+ val pushed = mergedResult.scan.asInstanceOf[TestV2Scan].pushed.mkString
+ assert(pushed.contains("a") && pushed.contains("b"),
+ s"the differing post-scan filters must be re-pushed as OR pruning, got: $pushed")
+ }
+ }
+
+ test("SPARK-40259: do not merge DSv2 scans that report key-grouped partitioning or ordering") {
+ // The rebuilt merged scan does not reconstruct reported partitioning/ordering, so a scan
+ // reporting either declines the merge (checked on both the np and cp side) -- the plan is left
+ // unchanged -- rather than silently dropping it. Preserving them across a merge is a deferred
+ // follow-up. (The plain-scan merge is already covered by the projected-columns test above.)
+ def assertDeclines(withField: DataSourceV2ScanRelation => DataSourceV2ScanRelation): Unit =
+ Seq(
+ (withField(v2ScanReading("a")), v2ScanReading("b")),
+ (v2ScanReading("a"), withField(v2ScanReading("b")))).foreach { case (npScan, cpScan) =>
+ val q = testRelation.select(
+ ScalarSubquery(npScan.groupBy()(sum($"a").as("sa"))),
+ ScalarSubquery(cpScan.groupBy()(sum($"b").as("sb"))))
+ comparePlans(Optimize.execute(q.analyze), q.analyze)
+ }
+
+ assertDeclines(s => s.copy(keyGroupedPartitioning = Some(Seq(s.output.head))))
+ assertDeclines(s => s.copy(ordering = Some(Seq(SortOrder(s.output.head, Ascending)))))
+ }
+
+ test("SPARK-40259: merge DSv2 scans that report empty key-grouped partitioning or ordering") {
+ // A source implementing SupportsReportPartitioning/SupportsReportOrdering but reporting nothing
+ // yields Some(Nil), not None (V2ScanPartitioningAndOrdering sets the field unconditionally). An
+ // empty report carries no partitioning/ordering to drop, so the merge should still proceed --
+ // the gate tests the inner Seq, not the Option. The fused plan is the plain column union.
+ def assertMerges(withEmptyField: DataSourceV2ScanRelation => DataSourceV2ScanRelation): Unit = {
+ val sub1 = ScalarSubquery(withEmptyField(v2ScanReading("a")).groupBy()(sum($"a").as("sum_a")))
+ val sub2 = ScalarSubquery(withEmptyField(v2ScanReading("b")).groupBy()(sum($"b").as("sum_b")))
+ val originalQuery = testRelation.select(sub1, sub2)
+
+ val mergedScan = v2ScanReadingOn(v2Table, Seq("a", "b"))
+ val mergedSubquery = mergedScan
+ .groupBy()(sum($"a").as("sum_a"), sum($"b").as("sum_b"))
+ .select(CreateNamedStruct(Seq(
+ Literal("sum_a"), $"sum_a",
+ Literal("sum_b"), $"sum_b")).as("mergedValue"))
+ val analyzedMergedSubquery = mergedSubquery.analyze
+ val correctAnswer = WithCTE(
+ testRelation.select(
+ extractorExpression(0, analyzedMergedSubquery.output, 0),
+ extractorExpression(0, analyzedMergedSubquery.output, 1)),
+ Seq(definitionNode(analyzedMergedSubquery, 0)))
+ comparePlans(Optimize.execute(originalQuery.analyze), correctAnswer.analyze)
+ }
+
+ assertMerges(_.copy(keyGroupedPartitioning = Some(Nil)))
+ assertMerges(_.copy(ordering = Some(Nil)))
+ }
+
+ test("SPARK-40259: merge DSv2 scans reading identical columns but differing filters") {
+ // Both scans read only "a", so the column union adds nothing (the np-only set is empty); the
+ // merge is driven purely by the differing filters, and the OR is still re-pushed for pruning.
+ // (sub1 is cp, sub2 is np.)
+ val sub1 = ScalarSubquery(v2ScanReading("a").where($"a" > 1).groupBy()(sum($"a").as("s1")))
+ val sub2 = ScalarSubquery(v2ScanReading("a").where($"a" > 2).groupBy()(sum($"a").as("s2")))
+ val originalQuery = testRelation.select(sub1, sub2)
+
+ // Expected: one merged scan reading just {a} (np-only set empty), the differing filters
+ // OR-widened above it (np = sub2 -> id 0, cp = sub1 -> id 1).
+ val mergedScan = v2ScanReadingOn(v2Table, Seq("a"))
+ val npFilterAlias = Alias($"a" > 2, "propagatedFilter_0")()
+ val cpFilterAlias = Alias($"a" > 1, "propagatedFilter_1")()
+ val npFilter = npFilterAlias.toAttribute
+ val cpFilter = cpFilterAlias.toAttribute
+ val mergedSubquery = mergedScan
+ .select(mergedScan.output ++ Seq(npFilterAlias, cpFilterAlias): _*)
+ .where(Or(npFilter, cpFilter))
+ .groupBy()(
+ sum($"a", Some(cpFilter)).as("s1"),
+ sum($"a", Some(npFilter)).as("s2"))
+ .select(CreateNamedStruct(Seq(
+ Literal("s1"), $"s1",
+ Literal("s2"), $"s2")).as("mergedValue"))
+ val analyzedMergedSubquery = mergedSubquery.analyze
+ val correctAnswer = WithCTE(
+ testRelation.select(
+ extractorExpression(0, analyzedMergedSubquery.output, 0),
+ extractorExpression(0, analyzedMergedSubquery.output, 1)),
+ Seq(definitionNode(analyzedMergedSubquery, 0)))
+
+ withSQLConf(
+ SQLConf.MERGE_SUBPLANS_FILTER_PROPAGATION_ENABLED.key -> "true",
+ SQLConf.MERGE_SUBPLANS_SYMMETRIC_FILTER_PROPAGATION_ENABLED.key -> "true") {
+ val optimized = Optimize.execute(originalQuery.analyze)
+ comparePlans(optimized, correctAnswer.analyze)
+ val pushed = v2Scans(optimized).head.scan.asInstanceOf[TestV2Scan].pushed.mkString
+ assert(pushed.contains("a"), s"the OR over a should be re-pushed as pruning, got: $pushed")
+ }
+ }
+
+ test("SPARK-40259: merge three DSv2 scans that pushed the same strict filter") {
+ // Round-2 merge feeds the already-merged relation back through samePushedFilters. All three
+ // scans carry the same strict a > 0, so they fuse into a single scan reading a, b, c, d.
+ val strictOnA = new TestV2Table(
+ StructType(Seq(StructField("a", IntegerType), StructField("b", IntegerType),
+ StructField("c", IntegerType), StructField("d", IntegerType))),
+ strictColumns = Set("a"))
+ def strictSub(col: String): ScalarSubquery = {
+ val s = v2ScanReadingOn(strictOnA, Seq("a", col))
+ ScalarSubquery(
+ s.copy(pushedFilters = Seq(s.output.find(_.name == "a").get > 0))
+ .groupBy()(sum(s.output.find(_.name == col).get).as(s"agg_$col")))
+ }
+ val originalQuery = testRelation.select(strictSub("b"), strictSub("c"), strictSub("d"))
+
+ // Expected: one merged scan reading {a, b, c, d} keeping the shared strict a > 0; three
+ // aggregates read b, c and d. No post-scan filter, so no OR-widen.
+ val mergedScan0 = v2ScanReadingOn(strictOnA, Seq("a", "b", "c", "d"))
+ val mergedScan = mergedScan0.copy(
+ pushedFilters = Seq(mergedScan0.output.find(_.name == "a").get > 0))
+ val mergedSubquery = mergedScan
+ .groupBy()(sum($"b").as("agg_b"), sum($"c").as("agg_c"), sum($"d").as("agg_d"))
+ .select(CreateNamedStruct(Seq(
+ Literal("agg_b"), $"agg_b",
+ Literal("agg_c"), $"agg_c",
+ Literal("agg_d"), $"agg_d")).as("mergedValue"))
+ val analyzedMergedSubquery = mergedSubquery.analyze
+ val correctAnswer = WithCTE(
+ testRelation.select(
+ extractorExpression(0, analyzedMergedSubquery.output, 0),
+ extractorExpression(0, analyzedMergedSubquery.output, 1),
+ extractorExpression(0, analyzedMergedSubquery.output, 2)),
+ Seq(definitionNode(analyzedMergedSubquery, 0)))
+ val optimized = Optimize.execute(originalQuery.analyze)
+ comparePlans(optimized, correctAnswer.analyze)
+ assert(v2Scans(optimized).head.pushedFilters.exists(_.references.exists(_.name == "a")),
+ "the merged scan must keep the shared strict filter on a")
+ }
+
+ test("SPARK-40259: pushed limit/offset/top-N/sample block merging " +
+ "(hasBlockingPushdown classification)") {
+ // hasBlockingPushdown flags every pushdown a rebuilt scan cannot reproduce. Drive each term
+ // through the real pushdown (the stub opts into all of them) so that a term silently dropped
+ // from the denylist is caught. Limit, offset and sample each set only their own holder field,
+ // so they are isolated; top-N sets sortOrders AND pushedLimit, so it exercises the top-N path
+ // but does not isolate the sortOrders term (no plan pushes a sort order without a limit).
+ def pushDown(op: DataSourceV2Relation => LogicalPlan): DataSourceV2ScanRelation = {
+ val relation = DataSourceV2Relation(
+ v2Table, toAttributes(v2Table.schema()), None, None, CaseInsensitiveStringMap.empty())
+ V2ScanRelationPushDown(op(relation))
+ .collectFirst { case s: DataSourceV2ScanRelation => s }.get
+ }
+
+ // Default-safe: only the plain column-pruning path grants mergeability, so a plain scan with
+ // no non-reproducible pushdown must come out mergeable (guards the one granting site).
+ assert(pushDown(r => Project(Seq(r.output.head), r)).mergeableScan,
+ "a plain scan with only reproducible pushdowns must be mergeable")
+
+ assert(!pushDown(r => Limit(Literal(1), r)).mergeableScan,
+ "a pushed limit must block merging")
+ assert(!pushDown(r => Offset(Literal(1), r)).mergeableScan,
+ "a pushed offset must block merging")
+ assert(
+ !pushDown(r => Limit(Literal(1),
+ Sort(Seq(SortOrder(r.output.head, Ascending)), global = true, r)))
+ .mergeableScan,
+ "a pushed top-N must block merging")
+ assert(!pushDown(r => Sample(0.0, 0.5, withReplacement = false, Some(0L), r))
+ .mergeableScan,
+ "a pushed sample must block merging")
+
+ // End-to-end: a scan that pushed a sample does not fuse with another (plain) scan of the same
+ // table -- the classification declines the scan merge, so the plan is left unchanged. The two
+ // read different columns, so they are not identical subplans and could only fuse via the
+ // (blocked) scan merge, not the identical-plan reuse short-circuit.
+ val sampled = pushDown(r => Sample(0.0, 0.5, withReplacement = false, Some(0L), r))
+ val plain = v2ScanReading("a")
+ val q = testRelation.select(
+ ScalarSubquery(plain.groupBy()(sum($"a").as("s1"))),
+ ScalarSubquery(sampled.groupBy()(sum(sampled.output(1)).as("s2"))))
+ comparePlans(Optimize.execute(q.analyze), q.analyze)
+ }
+}
+
+/**
+ * Minimal readable DSv2 table whose scan opts into merging, for scan-merge tests.
+ *
+ * A pushed predicate that references only columns in `strictColumns` is fully enforced (strict):
+ * accepted by the source and NOT returned as a post-scan filter. Any other predicate is
+ * best-effort (stats): accepted for row-group pruning but returned so it is re-checked above the
+ * scan. When `acceptsFilters` is false the source rejects everything: nothing is accepted and
+ * every predicate is returned (used to exercise the "no pruning recovery" path).
+ */
+private class TestV2Table(
+ tableSchema: StructType,
+ acceptsFilters: Boolean = true,
+ strictColumns: Set[String] = Set.empty,
+ supportsScanMerging: Boolean = true)
+ extends Table with SupportsRead {
+ override def name(): String = "test_v2_table"
+ override def schema(): StructType = tableSchema
+ override def capabilities(): java.util.Set[TableCapability] = {
+ val caps = java.util.EnumSet.of(TableCapability.BATCH_READ)
+ if (supportsScanMerging) caps.add(TableCapability.SCAN_MERGING)
+ caps
+ }
+ override def newScanBuilder(options: CaseInsensitiveStringMap): ScanBuilder =
+ new TestV2ScanBuilder(tableSchema, acceptsFilters, strictColumns)
+}
+
+private class TestV2ScanBuilder(
+ tableSchema: StructType,
+ acceptsFilters: Boolean,
+ strictColumns: Set[String])
+ extends ScanBuilder with SupportsPushDownRequiredColumns with SupportsPushDownV2Filters
+ with SupportsPushDownLimit with SupportsPushDownOffset with SupportsPushDownTopN
+ with SupportsPushDownTableSample {
+ private var prunedSchema: StructType = tableSchema
+ private var accepted: Array[Predicate] = Array.empty // strict UNION stats (pushedPredicates)
+
+ override def pruneColumns(requiredSchema: StructType): Unit = prunedSchema = requiredSchema
+
+ // Opts into limit pushdown so V2ScanRelationPushDown records a merge-blocking pushedLimit.
+ override def pushLimit(limit: Int): Boolean = true
+
+ // Opts into offset/top-N/sample pushdown too, so V2ScanRelationPushDown records the matching
+ // merge-blocking pushdown for each (used by the hasBlockingPushdown classification test).
+ override def pushOffset(offset: Int): Boolean = true
+ override def pushTopN(orders: Array[V2SortOrder], limit: Int): Boolean = true
+ override def isPartiallyPushed(): Boolean = false
+ override def pushTableSample(
+ lowerBound: Double,
+ upperBound: Double,
+ withReplacement: Boolean,
+ seed: Long): Boolean = true
+
+ override def pushPredicates(predicates: Array[Predicate]): Array[Predicate] = {
+ if (!acceptsFilters) {
+ predicates // reject everything: nothing accepted, all returned as post-scan
+ } else {
+ accepted = predicates
+ predicates.filterNot(isStrict) // stats predicates are re-checked above the scan
+ }
+ }
+
+ private def isStrict(p: Predicate): Boolean =
+ p.references().nonEmpty &&
+ p.references().forall(r => strictColumns.contains(r.fieldNames().mkString(".")))
+
+ override def pushedPredicates(): Array[Predicate] = accepted
+ override def build(): Scan = TestV2Scan(prunedSchema, accepted.map(_.describe()).toSeq)
+}
+
+/** `pushed` records the `describe()` of the predicates pushed to the scan, for test assertions. */
+private case class TestV2Scan(schema: StructType, pushed: Seq[String] = Seq.empty)
+ extends Scan {
+ override def readSchema(): StructType = schema
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/PlanMergeSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/PlanMergingSuite.scala
similarity index 99%
rename from sql/core/src/test/scala/org/apache/spark/sql/PlanMergeSuite.scala
rename to sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/PlanMergingSuite.scala
index e1109f20e604..7440c5badd67 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/PlanMergeSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/PlanMergingSuite.scala
@@ -15,14 +15,15 @@
* limitations under the License.
*/
-package org.apache.spark.sql
+package org.apache.spark.sql.execution.planmerging
+import org.apache.spark.sql.Row
import org.apache.spark.sql.execution._
import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanHelper}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
-class PlanMergeSuite extends SharedSparkSession
+class PlanMergingSuite extends SharedSparkSession
with AdaptiveSparkPlanHelper {
import testImplicits._
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org