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

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 `org.apache.spark.sql.execution.adaptive.DynamicJoinSelection` in `spark.sql.adaptive.optimizer.excludedRules`, that name no longer matches any rule (unknown names are silently ignored); set `spark.sql.adaptive.convertSortMergeJoinToShuffledHashJoin.enabled` to `false` instead.
+- 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.catalyst.optimizer.MergeScalarSubqueries`, in `spark.sql.optimizer.excludedRules`, that name no longer matches any rule (unknown names are silently ignored); use `org.apache.spark.sql.execution.planmerging.MergeSubplans` instead.
 - 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 `spark.sql.execution.combineAdjacentAggregation` to `false`.
 - 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 result these functions may return a value that differs from earlier releases in the last ULP. `percentile_disc` and `percentile_approx` are unaffected.

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
 }

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)
+}

sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala

@@ -160,19 +160,36 @@ 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,
     scan: Scan,
     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))
     )
   }
 }

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 " +

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()
+}

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(