[SPARK-57205][SQL] Declare SCAN_MERGING on the built-in file sources

connector/avro/src/main/scala/org/apache/spark/sql/v2/avro/AvroTable.scala

@@ -21,7 +21,7 @@ import scala.jdk.CollectionConverters._
 import org.apache.hadoop.fs.FileStatus
 
 import org.apache.spark.sql.SparkSession
-import org.apache.spark.sql.avro.AvroUtils
+import org.apache.spark.sql.avro.{AvroOptions, AvroUtils}
 import org.apache.spark.sql.connector.write.{LogicalWriteInfo, Write, WriteBuilder}
 import org.apache.spark.sql.execution.datasources.FileFormat
 import org.apache.spark.sql.execution.datasources.v2.FileTable
@@ -52,4 +52,17 @@ case class AvroTable(
  override def supportsDataType(dataType: DataType): Boolean = AvroUtils.supportsDataType(dataType)
 
  override def formatName: String = "Avro"
+
+ // Avro has no record-level parse verdict: a record is either decodable or the read fails, and
+ // there is no mode that drops or rewrites a record based on the columns asked for. The `mode`
+ // option in AvroOptions is read by from_avro and schema_of_avro, not by this scan.
+ //
+ // `positionalFieldMatching` is the exception. AvroPartitionReaderFactory builds the deserializer
+ // from the pruned read schema while the Avro side stays the full Avro schema, so under that
+ // option catalyst field i of the projection takes Avro field i of that schema, and widening the
+ // projection changes the values a column comes back with. Read the option off the map rather than
+ // through AvroOptions, whose constructor resolves `avroSchemaUrl` and would do I/O here, and read
+ // it leniently so a malformed value still fails where Avro reports it rather than here.
+ override protected def supportsScanMerging: Boolean =
+ !"true".equalsIgnoreCase(options.get(AvroOptions.POSITIONAL_FIELD_MATCHING))
 }

connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala

@@ -42,9 +42,10 @@ import org.apache.spark.sql.catalyst.expressions.AttributeReference
 import org.apache.spark.sql.catalyst.plans.logical.Filter
 import org.apache.spark.sql.catalyst.util.DateTimeTestUtils
 import org.apache.spark.sql.catalyst.util.DateTimeTestUtils.{withDefaultTimeZone, LA, UTC}
+import org.apache.spark.sql.connector.catalog.TableCapability
 import org.apache.spark.sql.execution.{FormattedMode, SparkPlan}
 import org.apache.spark.sql.execution.datasources.{CommonFileDataSourceSuite, DataSource, FilePartition}
-import org.apache.spark.sql.execution.datasources.v2.{BatchScanExec, FileDataSourceV2, FileTable}
+import org.apache.spark.sql.execution.datasources.v2.{BatchScanExec, DataSourceV2ScanRelation, FileDataSourceV2, FileTable}
 import org.apache.spark.sql.functions._
 import org.apache.spark.sql.internal.LegacyBehaviorPolicy
 import org.apache.spark.sql.internal.LegacyBehaviorPolicy._
@@ -3937,6 +3938,50 @@ class AvroV2Suite extends AvroSuite with ExplainSuiteHelper {
  s"V2 formatName '${v2Table.formatName}' != V1 toString '${v1Format.toString}'")
  }
 
+ test("SPARK-57205: Avro V2 declares SCAN_MERGING and merges scans differing only in columns") {
+ // AvroTable withholds the capability under positionalFieldMatching, because the deserializer is
+ // built from the pruned read schema while the Avro side stays unpruned, so catalyst field i of
+ // the projection takes Avro field i of that schema and widening the projection shifts values.
+ // FileTable also withholds it when the reads are not strict, so pin that rather than inherit.
+ withSQLConf(
+ SQLConf.IGNORE_CORRUPT_FILES.key -> "false",
+ SQLConf.IGNORE_MISSING_FILES.key -> "false") {
+ val v2Provider = DataSource.lookupDataSourceV2("avro", spark.sessionState.conf)
+ assert(v2Provider.isDefined)
+ val dsV2 = v2Provider.get.asInstanceOf[FileDataSourceV2]
+ val v2Table = dsV2.getTable(
+ new StructType(), Array.empty, JCollections.emptyMap[String, String]())
+ assert(v2Table.capabilities().contains(TableCapability.SCAN_MERGING))
+ val positional = dsV2.getTable(new StructType(), Array.empty,
+ JCollections.singletonMap("positionalFieldMatching", "true"))
+ assert(!positional.capabilities().contains(TableCapability.SCAN_MERGING))
+
+ withTempPath { dir =>
+ val path = dir.getCanonicalPath
+ spark.range(0, 20).selectExpr("id AS a", "id * 2 AS b", "id % 3 AS c")
+ .write.format("avro").save(path)
+ withTempView("avro_scan_merging") {
+ spark.read.format("avro").load(path).createOrReplaceTempView("avro_scan_merging")
+ val df = sql(
+ """
+ |SELECT
+ | (SELECT sum(a) FROM avro_scan_merging WHERE c = 1),
+ | (SELECT sum(b) FROM avro_scan_merging WHERE c = 1)
+ |""".stripMargin)
+ checkAnswer(df, Row(70, 140))
+ val scans = df.queryExecution.optimizedPlan.collectWithSubqueries {
+ case s: DataSourceV2ScanRelation => s
+ }
+ assert(scans.map(_.canonicalized).distinct.length == 1,
+ s"the two Avro scans should be fused into one:\n${df.queryExecution.optimizedPlan}")
+ // c is read because the filter stays above the merged scan.
+ assert(scans.head.output.map(_.name).toSet == Set("a", "b", "c"),
+ s"the merged scan should read the union of both columns; got ${scans.head.output}")
+ }
+ }
+ }
+ }
+
  test("Geospatial types are not supported in Avro") {
  withTempDir { dir =>
  // Temporary directory for writing the test data.

docs/sql-performance-tuning.md

@@ -342,6 +342,8 @@ They are merged into one aggregate that computes `min` and `max` together, so `s
 
 Two subplans are merged when their plans match node by node: `Project` lists are unioned, `Aggregate`s must have the same grouping and use the same aggregation implementation (so a `min` is not merged with a `collect_list`), `Filter`s must have the same condition, `Join`s must have the same type, condition and hints, and the leaves must read the same input. Subplans that differ only in their `WHERE` conditions can be merged as well, by turning each side's condition into a boolean column and giving each side's aggregate expressions a `FILTER (WHERE ...)` clause. That is controlled by the configurations below. Queries that still contain a `WITH` clause when this rule runs (one that was not inlined) are skipped.
 
+On the DataSource V2 read path the requirement that the leaves read the same input is relaxed for a source that declares the `SCAN_MERGING` table capability: two leaves that differ only in their projected columns merge into a single scan reading the union of those columns. Among the built-in file formats Parquet, ORC, text and Avro declare it; a format reaches its V2 read path only when it is removed from `spark.sql.sources.useV1SourceList`. A file table withholds the capability when `spark.sql.files.ignoreCorruptFiles` is true, because a read failure in a column that only the other subplan projects would then be swallowed along with the rest of that file's rows, and when `spark.sql.files.ignoreMissingFiles` is true, to match the strictness predicate the file reader uses. Avro withholds it under `positionalFieldMatching`, which resolves a column by its position in the projection.
+
 When only one of the two subplans has a filter, merging is always beneficial, because the unfiltered side reads all the data anyway. This case is on by default, unless the filter has to cross a `Join` to reach the aggregate, which needs the through-join configuration below. When both sides have a filter (the symmetric case), the merged scan filter becomes `OR(f1, f2)`, which is less selective than either original filter and can therefore read more data - for example when the filters prune partitions or Parquet row groups. That is why the symmetric case is disabled by default.
 
 Still, it is worth considering on queries that compute several differently filtered aggregates over the same table, which is a common analytical shape:
@@ -384,7 +386,7 @@ In TPC-DS benchmark runs, enabling symmetric filter propagation made `q9` and `q
  <td><code>spark.sql.optimizer.mergeSubplans.filterPropagation.dsv2SymmetricFilterPropagation.enabled</code></td>
  <td>false</td>
  <td>
- When true, two DataSource V2 scans that pushed the same strictly enforced filters but carry different best-effort (post-scan) filters can be merged even when <code>spark.sql.optimizer.mergeSubplans.filterPropagation.symmetricFilterPropagation.enabled</code> is false. In this case widening cannot change the set of rows the scan is required to return, as the strict filters are re-pushed unchanged and the enclosing <code>Filter</code> re-checks the rest above the scan. This applies only to V2 sources that opt in to scan merging with the <code>SCAN_MERGING</code> table capability; no built-in source does.
+ When true, two DataSource V2 scans that pushed the same strictly enforced filters but carry different best-effort (post-scan) filters can be merged even when <code>spark.sql.optimizer.mergeSubplans.filterPropagation.symmetricFilterPropagation.enabled</code> is false. In this case widening cannot change the set of rows the scan is required to return, as the strict filters are re-pushed unchanged and the enclosing <code>Filter</code> re-checks the rest above the scan. This applies only to V2 sources that opt in to scan merging with the <code>SCAN_MERGING</code> table capability. For a file source the strictly enforced filters are the partition filters, so this configuration is what lets two scans over the same partitions but with different data filters merge.
  </td>
  <td>4.3.0</td>
  </tr>

sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableCapability.java

@@ -139,6 +139,12 @@ public enum TableCapability {
  * 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>
+ * Determinism alone is not enough: widening the set of pruned columns, with the options and
+ * pushed filters held constant, must not change which rows the scan returns nor the values it
+ * returns for the columns already asked for. It may at most surface a read error. A source whose
+ * parser decides what counts as a malformed record from the set of columns it was asked for does
+ * not meet this, and neither does one that resolves a column by its position in the projection.
+ * <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

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

@@ -37,6 +37,18 @@ import org.apache.spark.sql.util.CaseInsensitiveStringMap
 import org.apache.spark.sql.util.SchemaUtils
 import org.apache.spark.util.ArrayImplicits._
 
+/**
+ * A [[Table]] backed by files.
+ *
+ * A subclass opts in to the `SCAN_MERGING` capability by overriding [[supportsScanMerging]], which
+ * holds it to this: with the scan options and the pushed filters held constant, widening the set of
+ * columns pruned on its builder must not change which rows the scan returns, nor the values it
+ * returns for the columns it was already asked for. It may at most surface a read error. A format
+ * whose parser decides what counts as a malformed record from the set of columns it was asked for
+ * does not meet that, and neither does one that resolves a column by its position in the
+ * projection. The capability is also withheld from a table whose reads are not strict, see
+ * `hasStrictFileReads`.
+ */
 abstract class FileTable(
  sparkSession: SparkSession,
  options: CaseInsensitiveStringMap,
@@ -111,7 +123,32 @@ abstract class FileTable(
 
  override def properties: util.Map[String, String] = options.asCaseSensitiveMap
 
- override def capabilities: java.util.Set[TableCapability] = FileTable.CAPABILITIES
+ override def capabilities: java.util.Set[TableCapability] =
+ if (supportsScanMerging && hasStrictFileReads) {
+ FileTable.CAPABILITIES_WITH_SCAN_MERGING
+ } else {
+ FileTable.CAPABILITIES
+ }
+
+ /**
+ * Whether this table meets the `SCAN_MERGING` contract described on this class. Defaults to
+ * false: a format that does not merge only misses an optimization, while a format that merges
+ * when its parser is projection-sensitive returns wrong rows.
+ */
+ protected def supportsScanMerging: Boolean = false
+
+ /**
+ * Whether a read of this table is strict. Under `ignoreCorruptFiles`, a read failure in a column
+ * that only the other scan projects is swallowed and the remaining rows of that file are dropped,
+ * so the merged scan would not read a superset of either input's rows. `ignoreMissingFiles` drops
+ * the same rows whatever is projected, and is included to match `FileScanRDD.hasStrictFileReads`,
+ * the same predicate on the physical side. Evaluated per call rather than cached, so a table
+ * built before either configuration was set still answers for the read that is running.
+ */
+ private def hasStrictFileReads: Boolean = {
+ val fileSourceOptions = new FileSourceOptions(options.asCaseSensitiveMap.asScala.toMap)
+ !fileSourceOptions.ignoreCorruptFiles && !fileSourceOptions.ignoreMissingFiles
+ }
 
  /**
  * When possible, this method should return the schema of the given `files`. When the format
@@ -182,4 +219,10 @@ abstract class FileTable(
 
 object FileTable {
  private val CAPABILITIES = util.EnumSet.of(BATCH_READ, BATCH_WRITE)
+
+ // For the formats that override supportsScanMerging. `fileIndex` is a lazy val, so every scan
+ // built from one table lists the same files, and `newScanBuilder` returns a fresh builder over
+ // `mergedOptions(options)`.
+ private val CAPABILITIES_WITH_SCAN_MERGING =
+ util.EnumSet.of(BATCH_READ, BATCH_WRITE, SCAN_MERGING)
 }

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

@@ -68,4 +68,10 @@ case class OrcTable(
  }
 
  override def formatName: String = "ORC"
+
+ // A row is decoded from the columns the scan asked for, so under strict file reads reading more
+ // columns can only surface an error, never silently change which rows come back. When the read is
+ // not strict that error is swallowed and the rest of the file's rows go with it, which is why
+ // FileTable withholds the capability there.
+ override protected def supportsScanMerging: Boolean = true
 }

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

@@ -70,4 +70,10 @@ case class ParquetTable(
  }
 
  override def formatName: String = "Parquet"
+
+ // A row is decoded from the column chunks the scan asked for, so under strict file reads reading
+ // more columns can only surface an error, never silently change which rows come back. When the
+ // read is not strict that error is swallowed and the rest of the file's rows go with it, which is
+ // why FileTable withholds the capability there.
+ override protected def supportsScanMerging: Boolean = true
 }

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

@@ -49,4 +49,8 @@ case class TextTable(
  override def supportsDataType(dataType: DataType): Boolean = dataType == StringType
 
  override def formatName: String = "Text"
+
+ // The schema is a single `value` column, and every line -- or every file under `wholetext` --
+ // becomes a row, so there is no parse step whose outcome a wider column set could change.
+ override protected def supportsScanMerging: Boolean = true
 }

sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/DSv2PlanMergingSuite.scala

@@ -23,7 +23,7 @@ import org.apache.spark.sql.{DataFrame, QueryTest, Row}
 import org.apache.spark.sql.catalyst.plans.physical.KeyedPartitioning
 import org.apache.spark.sql.connector.FakeV2ProviderWithCustomSchema
 import org.apache.spark.sql.connector.catalog.{InMemoryScanMergingPartitionFilterCatalog, InMemoryScanMergingReportingCatalog}
-import org.apache.spark.sql.execution.datasources.v2.{BatchScanExec, DataSourceV2Relation, DataSourceV2ScanRelation}
+import org.apache.spark.sql.execution.datasources.v2.{BatchScanExec, DataSourceV2Relation}
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.test.SharedSparkSession
 
@@ -37,7 +37,7 @@ import org.apache.spark.sql.test.SharedSparkSession
  * mis-classify it as non-strict and decline the merge (leaving two scans).
  */
 class DSv2PlanMergingSuite extends QueryTest with SharedSparkSession
- with BeforeAndAfter {
+ with BeforeAndAfter with V2ScanMergingTestHelper {
 
  private val v2Source = classOf[FakeV2ProviderWithCustomSchema].getName
  private val tbl = "scanmerge.t"
@@ -56,11 +56,6 @@ class DSv2PlanMergingSuite extends QueryTest with SharedSparkSession
  spark.conf.unset("spark.sql.catalog.scanmergereport")
  }
 
- 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 --