[SPARK-57205][SQL] Add SupportsScanMerging to merge equivalent V2 file scans

sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/SupportsScanMerging.java

@@ -0,0 +1,72 @@
+/*
+ * 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.read;
+
+import java.util.Optional;
+
+import org.apache.spark.annotation.Evolving;
+import org.apache.spark.sql.connector.catalog.SupportsRead;
+
+/**
+ * A mix-in interface for {@link Scan}. Data sources can implement this interface to allow
+ * Spark's optimizer to fuse two {@link Scan}s of the same {@link SupportsRead} table into a
+ * single {@link Scan} covering both: it reads the union of their projected columns and a
+ * superset of their rows, from which each original scan's result can be recovered by a
+ * projection and filter applied above the merged scan.
+ * <p>
+ * Spark calls {@link #mergeWith(SupportsScanMerging, SupportsRead)} when it detects that two
+ * scans target the same table and read structurally compatible data (e.g. same file
+ * index, schema, partitioning) but differ in pushed predicates, projections, or other
+ * scan-level state. Implementations decide whether the two scans can be safely combined
+ * and, if so, return the merged scan via {@link Optional#of}. Returning {@link Optional#empty}
+ * declines the merge.
+ * <p>
+ * <b>Contract for implementations:</b>
+ * <ul>
+ *   <li>{@link #mergeWith} must be commutative: {@code a.mergeWith(b, t)} and
+ *       {@code b.mergeWith(a, t)} must produce semantically equivalent merged scans (or both
+ *       decline).</li>
+ *   <li>The merged scan must produce a superset of the rows produced by either input scan.
+ *       Read-side filtering can then be applied above the scan to recover the original
+ *       per-scan rows, e.g. via {@code FILTER (WHERE ...)} clauses on aggregates.</li>
+ *   <li>If the two scans carry state that cannot be safely combined -- for example a pushed
+ *       aggregate, or incompatible partitioning -- the implementation must decline by
+ *       returning {@link Optional#empty}.</li>
+ *   <li>The {@code table} argument is the {@link SupportsRead} that owns both scans;
+ *       implementations typically use it to obtain a fresh
+ *       {@link ScanBuilder} via {@link SupportsRead#newScanBuilder} for constructing the
+ *       merged scan.</li>
+ * </ul>
+ *
+ * @since 4.3.0
+ */
+@Evolving
+public interface SupportsScanMerging extends Scan {
+
+  /**
+   * Attempts to merge this scan with {@code other} into a single equivalent scan.
+   *
+   * @param other the other scan to merge with; guaranteed to also implement
+   *              {@link SupportsScanMerging} and to be a scan of the same {@code table}
+   * @param table the {@link SupportsRead} table that owns both scans, used to obtain
+   *              a {@link ScanBuilder} for constructing the merged scan
+   * @return {@link Optional#of} the merged scan if merging is supported, or
+   *         {@link Optional#empty} to decline
+   */
+  Optional<SupportsScanMerging> mergeWith(SupportsScanMerging other, SupportsRead table);
+}

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PlanMerger.scala

@@ -19,11 +19,15 @@ package org.apache.spark.sql.catalyst.optimizer
 
 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, AttributeReference, 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.catalyst.types.DataTypeUtils.toAttributes
+import org.apache.spark.sql.connector.catalog.SupportsRead
+import org.apache.spark.sql.connector.read.SupportsScanMerging
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2ScanRelation
 import org.apache.spark.sql.internal.SQLConf
 
 /**
@@ -483,11 +487,99 @@ class PlanMerger(
             case _ => None
           }
 
+        case (np: DataSourceV2ScanRelation, cp: DataSourceV2ScanRelation) =>
+          tryMergeV2ScanRelations(np, cp)
+
         // Otherwise merging is not possible.
         case _ => None
       })
   }
 
+  /** Attempt to merge two V2 scan relations via the [[SupportsScanMerging]] connector
+   *  interface. Both scans must reference the same underlying relation and have
+   *  canonicalized-identical relation-level `pushedFilters` (see the gate below). The
+   *  merged scan's output is the union of both sides' read schemas; cp's attribute IDs
+   *  are preserved for shared columns and np's attribute IDs are mapped to the
+   *  corresponding merged attribute.
+   *
+   *  How this reaches V1 parity for filter differences: for sources whose pushdown is
+   *  best-effort (e.g. Parquet row-group filters), the exact predicate remains as a
+   *  post-scan [[Filter]] node ABOVE the scan, and the relation-level `pushedFilters`
+   *  field is empty. So two branches that differ only in their WHERE clause look like
+   *  `Filter(p1) over scan` vs `Filter(p2) over scan` -- the existing `(Filter, Filter)`
+   *  symmetric-propagation cases OR-widen the exact filter and emit per-side aggregate
+   *  `FILTER (WHERE ...)` clauses, exactly as for V1. This leaf case only merges the
+   *  scans underneath, OR-widening their best-effort pushed filter (via the connector's
+   *  `mergeWith`) so the merged scan reads a superset of either side's rows. The
+   *  per-side correctness is then restored by the Filter nodes above.
+   */
+  private def tryMergeV2ScanRelations(
+      np: DataSourceV2ScanRelation,
+      cp: DataSourceV2ScanRelation): Option[TryMergeResult] = {
+    // Both scans must reference the same underlying relation.
+    if (np.relation.canonicalized != cp.relation.canonicalized) return None
+
+    // Relation-level pushedFilters must match. For best-effort sources these are empty
+    // (the exact predicate stays as a post-scan Filter, handled by the (Filter, Filter)
+    // cases above). For sources that report EXACT pushdown (no residual Filter node),
+    // differing pushedFilters here would have no Filter above to re-apply per-side
+    // predicates, so OR-widening the scan would change results -- this gate
+    // conservatively declines that case.
+    if (ExpressionSet(np.pushedFilters) != ExpressionSet(cp.pushedFilters)) return None
+
+    // Both scans must implement SupportsScanMerging.
+    val (npm, cpm) = (np.scan, cp.scan) match {
+      case (a: SupportsScanMerging, b: SupportsScanMerging) => (a, b)
+      case _ => return None
+    }
+
+    // The underlying table must be SupportsRead so we can call newScanBuilder.
+    val table = np.relation.table match {
+      case sr: SupportsRead => sr
+      case _ => return None
+    }
+
+    val mergedOpt = npm.mergeWith(cpm, table)
+    if (!mergedOpt.isPresent) return None
+    val mergedScan = mergedOpt.get()
+
+    // Build merged output: preserve cp's attribute IDs for shared columns, allocate
+    // fresh attribute IDs for np-exclusive columns.
+    val realOutput = toAttributes(mergedScan.readSchema())
+    val cpByName = cp.output.map(a => a.name -> a).toMap
+    val mergedOutput = realOutput.map { ra =>
+      cpByName.get(ra.name) match {
+        case Some(cpa) =>
+          AttributeReference(ra.name, ra.dataType, ra.nullable, ra.metadata)(
+            cpa.exprId, cpa.qualifier)
+        case None => ra
+      }
+    }
+
+    // np.output -> mergedOutput by name. If any np-output column is missing from
+    // the merged scan, the merge is not usable. AttributeMap is invariant in its
+    // value type, so widen the pairs to (Attribute, Attribute) explicitly.
+    val mergedByName = mergedOutput.map(a => a.name -> a).toMap
+    val mappingOpts: Seq[Option[(Attribute, Attribute)]] =
+      np.output.map(npa => mergedByName.get(npa.name).map(npa -> _))
+    if (mappingOpts.exists(_.isEmpty)) return None
+
+    val npMapping = AttributeMap(mappingOpts.flatten)
+
+    // Preserve the original pushedFilters (np.pushedFilters == cp.pushedFilters by
+    // the check above). Partitioning and ordering hints from V2ScanPartitioningAndOrdering
+    // are intentionally dropped: a merged scan with a wider read schema may have a
+    // different output partitioning than either input, and re-running pushdown via
+    // the connector's mergeWith does not preserve those hints. Consumers that need
+    // them can be re-fixed by a later pass of V2ScanPartitioningAndOrdering.
+    val mergedRelation = DataSourceV2ScanRelation(
+      np.relation,
+      mergedScan,
+      mergedOutput,
+      pushedFilters = cp.pushedFilters)
+    Some(TryMergeResult(mergedRelation, npMapping))
+  }
+
   // 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.
   //

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -1633,6 +1633,18 @@ object SQLConf {
     .booleanConf
     .createWithDefault(false)
 
+  val FILE_SCAN_MERGE_IGNORE_PUSHED_DATA_FILTERS =
+    buildConf("spark.sql.files.scanMerge.ignorePushedDataFilters")
+      .internal()
+      .doc("When true, allow file-source `Scan`s with different pushed data filters to be merged " +
+        "via the SupportsScanMerging connector API. The merged scan widens the filter to " +
+        "OR(f1, f2) and per-scan filters are reapplied at consumers (e.g. via aggregate FILTER " +
+        "clauses). When false, scans must have identical pushed data filters to merge.")
+      .version("4.3.0")
+      .withBindingPolicy(ConfigBindingPolicy.SESSION)
+      .booleanConf
+      .createWithDefault(false)
+
   val PARQUET_WRITE_LEGACY_FORMAT = buildConf("spark.sql.parquet.writeLegacyFormat")
     .doc("If true, data will be written in a way of Spark 1.4 and earlier. For example, decimal " +
       "values will be written in Apache Parquet's fixed-length byte array format, which other " +
@@ -7896,6 +7908,9 @@ class SQLConf extends Serializable with Logging with SqlApiConf {
 
   def parquetAggregatePushDown: Boolean = getConf(PARQUET_AGGREGATE_PUSHDOWN_ENABLED)
 
+  def fileScanMergeIgnorePushedDataFilters: Boolean =
+    getConf(FILE_SCAN_MERGE_IGNORE_PUSHED_DATA_FILTERS)
+
   def orcFilterPushDown: Boolean = getConf(ORC_FILTER_PUSHDOWN_ENABLED)
 
   def orcAggregatePushDown: Boolean = getConf(ORC_AGGREGATE_PUSHDOWN_ENABLED)

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileScan.scala

@@ -16,7 +16,7 @@
  */
 package org.apache.spark.sql.execution.datasources.v2
 
-import java.util.{Locale, OptionalLong}
+import java.util.{Locale, Optional, OptionalLong}
 
 import org.apache.hadoop.fs.Path
 
@@ -25,10 +25,11 @@ import org.apache.spark.internal.LogKeys.{PATH, REASON}
 import org.apache.spark.internal.config.IO_WARNING_LARGEFILETHRESHOLD
 import org.apache.spark.sql.SparkSession
 import org.apache.spark.sql.catalyst.SQLConfHelper
-import org.apache.spark.sql.catalyst.expressions.{AttributeSet, Expression, ExpressionSet}
+import org.apache.spark.sql.catalyst.expressions.{And, AttributeSet, Expression, ExpressionSet, Or}
 import org.apache.spark.sql.catalyst.expressions.codegen.GenerateUnsafeProjection
 import org.apache.spark.sql.catalyst.plans.QueryPlan
 import org.apache.spark.sql.catalyst.types.DataTypeUtils.toAttributes
+import org.apache.spark.sql.connector.catalog.SupportsRead
 import org.apache.spark.sql.connector.read._
 import org.apache.spark.sql.errors.QueryCompilationErrors
 import org.apache.spark.sql.execution.PartitionedFileUtil
@@ -37,12 +38,14 @@ import org.apache.spark.sql.internal.{SessionStateHelper, SQLConf}
 import org.apache.spark.sql.internal.connector.SupportsMetadata
 import org.apache.spark.sql.sources.Filter
 import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
 import org.apache.spark.util.Utils
 
 trait FileScan extends Scan
   with Batch
   with SupportsReportStatistics
   with SupportsMetadata
+  with SupportsScanMerging
   with SQLConfHelper
   with Logging {
   /**
@@ -56,6 +59,9 @@ trait FileScan extends Scan
 
   def fileIndex: PartitioningAwareFileIndex
 
+  /** The scan options, used to obtain a fresh [[ScanBuilder]] when merging. */
+  def options: CaseInsensitiveStringMap
+
   def dataSchema: StructType
 
   /**
@@ -89,7 +95,7 @@ trait FileScan extends Scan
 
   protected def seqToString(seq: Seq[Any]): String = seq.mkString("[", ", ", "]")
 
-  private lazy val (normalizedPartitionFilters, normalizedDataFilters) = {
+  protected lazy val (normalizedPartitionFilters, normalizedDataFilters) = {
     val partitionFilterAttributes = AttributeSet(partitionFilters).map(a => a.name -> a).toMap
     val normalizedPartitionFilters = ExpressionSet(partitionFilters.map(
       QueryPlan.normalizeExpressions(_, toAttributes(fileIndex.partitionSchema)
@@ -211,6 +217,81 @@ trait FileScan extends Scan
     a.sortBy(_.hashCode()).sameElements(b.sortBy(_.hashCode()))
   }
 
+  // ===== SupportsScanMerging =================================================
+  //
+  // Two file scans of the same table can be fused into one when they read the same data and
+  // differ only in projected columns and/or pushed data filters. This brings V2 file sources to
+  // parity with V1, where MergeSubplans already merges such subplans (V1 keeps the Filter/Project
+  // nodes in the logical plan, so the relation leaves are identical). The logic here is
+  // format-agnostic; format-specific state is handled via the hooks below.
+
+  /**
+   * Format-specific scan state (beyond fileIndex / schema / options / partition filters / data
+   * filters / projected columns) that must match for two scans to be mergeable. The default
+   * requires no extra state. Overridden e.g. by ParquetScan to require equal variant extractions.
+   * `other` is guaranteed to be the same concrete class as `this`.
+   */
+  protected def canMergeScanStateWith(other: FileScan): Boolean = true
+
+  /**
+   * Whether this scan has pushed-down aggregation. When true, the scan emits aggregated rows with
+   * no post-scan Filter to reconcile per-side predicates, so merging is declined. Overridden by
+   * ParquetScan / OrcScan which support aggregate pushdown.
+   */
+  protected def hasAggregatePushedDown: Boolean = false
+
+  private def mergeEligible(o: FileScan): Boolean =
+    getClass == o.getClass &&
+      fileIndex == o.fileIndex &&
+      dataSchema == o.dataSchema &&
+      options == o.options &&
+      normalizedPartitionFilters == o.normalizedPartitionFilters &&
+      !hasAggregatePushedDown && !o.hasAggregatePushedDown &&
+      canMergeScanStateWith(o)
+
+  override def mergeWith(
+      other: SupportsScanMerging,
+      table: SupportsRead): Optional[SupportsScanMerging] = other match {
+    case o: FileScan if mergeEligible(o) =>
+      // Strict when the two scans already agree on pushed data filters: rebuild with the union of
+      // read schemas, rows unchanged. Otherwise (relaxed) widen the best-effort pushed filter to
+      // OR(f1, f2) so the merged scan reads a superset; the exact per-side predicate is reapplied
+      // by the post-scan Filter node (split into aggregate FILTER clauses by PlanMerger). The
+      // relaxed path is gated by `spark.sql.files.scanMerge.ignorePushedDataFilters`.
+      val strict = normalizedDataFilters == o.normalizedDataFilters
+      if (!strict && !conf.fileScanMergeIgnorePushedDataFilters) {
+        Optional.empty()
+      } else {
+        table.newScanBuilder(options) match {
+          case builder: FileScanBuilder =>
+            // Partition filters are guaranteed equal (see `mergeEligible`), so push them as-is --
+            // never widen, which would needlessly produce OR(pf, pf) and obscure partition
+            // pruning. Only data filters use the strict/relaxed distinction.
+            builder.pushFilters(mergedDataFilters(o, strict) ++ partitionFilters)
+            builder.pruneColumns(readSchema().merge(o.readSchema()))
+            builder.build() match {
+              case merged: SupportsScanMerging => Optional.of(merged)
+              case _ => Optional.empty()
+            }
+          case _ => Optional.empty()
+        }
+      }
+    case _ => Optional.empty()
+  }
+
+  // Strict: the two data-filter lists are equivalent, so push this scan's. Relaxed: widen to
+  // OR(AND(df1), AND(df2)) so the merged scan reads a superset; an empty list on either side
+  // means that side reads everything, so the union also reads everything (no data filter).
+  private def mergedDataFilters(o: FileScan, strict: Boolean): Seq[Expression] = {
+    if (strict) {
+      dataFilters
+    } else if (dataFilters.nonEmpty && o.dataFilters.nonEmpty) {
+      Seq(Or(dataFilters.reduce(And), o.dataFilters.reduce(And)))
+    } else {
+      Seq.empty
+    }
+  }
+
   private val isCaseSensitive = conf.caseSensitiveAnalysis
 
   private def normalizeName(name: String): String = {

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/orc/OrcScan.scala

@@ -104,4 +104,7 @@ case class OrcScan(
       Map("PushedAggregation" -> pushedAggregationsStr) ++
       Map("PushedGroupBy" -> pushedGroupByStr)
   }
+
+  // ORC supports aggregate pushdown, which disables scan merging (see FileScan).
+  override protected def hasAggregatePushedDown: Boolean = pushedAggregate.nonEmpty
 }

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetScan.scala

@@ -217,4 +217,16 @@ case class ParquetScan(
       Map("PushedGroupBy" -> pushedGroupByStr) ++
       Map("PushedVariantExtractions" -> variantExtractionStr)
   }
+
+  // SupportsScanMerging hooks (generic logic in FileScan): Parquet disables merging when an
+  // aggregate is pushed, and requires equal variant extractions to merge.
+  override protected def hasAggregatePushedDown: Boolean = pushedAggregate.nonEmpty
+
+  override protected def canMergeScanStateWith(other: FileScan): Boolean = other match {
+    case o: ParquetScan =>
+      java.util.Arrays.equals(
+        pushedVariantExtractions.asInstanceOf[Array[Object]],
+        o.pushedVariantExtractions.asInstanceOf[Array[Object]])
+    case _ => false
+  }
 }