[SPARK-59107][SQL][4.3] Do not widen a projection-sensitive V1 file read

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.execution.{FormattedMode, SparkPlan}
+import org.apache.spark.sql.execution.{FileSourceScanExec, 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.planmerging.MergeSubplans
 import org.apache.spark.sql.functions._
 import org.apache.spark.sql.internal.LegacyBehaviorPolicy
 import org.apache.spark.sql.internal.LegacyBehaviorPolicy._
@@ -3732,6 +3733,42 @@ class AvroV1Suite extends AvroSuite {
       .sparkConf
       .set(SQLConf.USE_V1_SOURCE_LIST, "avro")
 
+  test("SPARK-59107: positionalFieldMatching makes an avro read projection-sensitive") {
+    // Strictness pinned rather than inherited, so that positional matching is the only reason the
+    // read is projection-sensitive. AQE off because `AdaptiveSparkPlanExec` is a leaf node, so with
+    // it on the scans underneath it are not reachable from the executed plan.
+    withSQLConf(
+        SQLConf.IGNORE_CORRUPT_FILES.key -> "false",
+        SQLConf.IGNORE_MISSING_FILES.key -> "false",
+        SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+      withTempPath { dir =>
+        val path = dir.getCanonicalPath
+        spark.range(0, 5).selectExpr("id AS a", "id * 10 AS b").write.format("avro").save(path)
+        withTempView("t") {
+          spark.read.option("positionalFieldMatching", "true").format("avro").load(path)
+            .createOrReplaceTempView("t")
+          val query = "SELECT (SELECT sum(a) FROM t), (SELECT sum(b) FROM t)"
+          // Compared against the same query with merging excluded rather than against a literal
+          // row: positional matching resolves a column against its position in the read schema, so
+          // what `sum(b)` answers depends on its own subquery's projection. What this test pins is
+          // that merging changes neither value.
+          val unmerged = withSQLConf(
+              SQLConf.OPTIMIZER_EXCLUDED_RULES.key -> MergeSubplans.ruleName) {
+            sql(query).collect().toSeq
+          }
+          val df = sql(query)
+          checkAnswer(df, unmerged)
+          val scanColumns = df.queryExecution.executedPlan
+            .collectWithSubqueries { case s: FileSourceScanExec => s }
+            .map(_.requiredSchema.fieldNames.sorted.toSeq)
+            .sortBy(_.mkString(","))
+          // One entry per column means the two subqueries kept their own scans.
+          assert(scanColumns === Seq(Seq("a"), Seq("b")))
+        }
+      }
+    }
+  }
+
   test("SPARK-36271: V1 insert should check schema field name too") {
     withView("v") {
       spark.range(1).createTempView("v")

docs/sql-performance-tuning.md

@@ -272,7 +272,7 @@ SELECT
 
 They are merged into one aggregate that computes `min` and `max` together, so `store_sales` is read once. In `EXPLAIN` output a merged subplan shows up as a subquery whose single output column is named `mergedValue`, and the sites that share it as `ReusedSubquery`.
 
-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.
+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. A V1 file relation whose rows depend on which columns the read asked for is merged only when both subplans read the same columns of it: `csv`, `json` and `xml`, whose parsers decide what counts as a malformed record from the required schema, `avro` read with `positionalFieldMatching`, which pairs a column with the Avro field at its position in that schema, and any file relation read with `spark.sql.files.ignoreCorruptFiles` enabled, as a read option or through the configuration, where a failure in a column only one side reads is swallowed together with the rest of that file's rows. `spark.sql.files.ignoreMissingFiles` counts too, not for that reason but because one predicate answers for both. 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.
 
 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.
 

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/FileSourceOptions.scala

@@ -52,6 +52,11 @@ class FileSourceOptions(
   val ignoreMissingFiles: Boolean = parameters.get(IGNORE_MISSING_FILES).map(_.toBoolean)
     .getOrElse(SQLConf.get.ignoreMissingFiles)
 
+  /**
+   * Whether a read under these options fails rather than skipping missing or corrupt input files.
+   */
+  def hasStrictFileReads: Boolean = !ignoreCorruptFiles && !ignoreMissingFiles
+
   /**
    * Whether the data source may read tar archives (.tar/.tar.gz/.tgz) by streaming their entries.
    * Gated by [[SQLConf.ARCHIVE_FORMAT_READER_ENABLED]] and resolved at construction (on the driver,

sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryRelation.scala

@@ -380,10 +380,7 @@ case class CachedRDDBuilder(
         val cachedPlanConf = cachedPlan.conf.clone()
 
         def hasStrictReads(conf: SQLConf): Boolean = SQLConf.withExistingConf(conf) {
-          fileSourceOptions.forall { options =>
-            val effectiveOptions = new FileSourceOptions(options)
-            !effectiveOptions.ignoreMissingFiles && !effectiveOptions.ignoreCorruptFiles
-          }
+          fileSourceOptions.forall(options => new FileSourceOptions(options).hasStrictFileReads)
         }
 
         val (inputRDD, strictPhysicalReads) = SQLConf.withExistingConf(materializationConf) {

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

@@ -29,12 +29,17 @@ import org.json4s.jackson.Serialization
 
 import org.apache.spark.{SparkException, SparkUpgradeException}
 import org.apache.spark.sql.{sources, SPARK_LEGACY_DATETIME_METADATA_KEY, SPARK_LEGACY_INT96_METADATA_KEY, SPARK_TIMEZONE_METADATA_KEY, SPARK_VERSION_METADATA_KEY}
+import org.apache.spark.sql.avro.{AvroFileFormat, AvroOptions}
+import org.apache.spark.sql.catalyst.FileSourceOptions
 import org.apache.spark.sql.catalyst.catalog.{CatalogTable, CatalogUtils}
 import org.apache.spark.sql.catalyst.expressions.{AttributeReference, AttributeSet, Expression, ExpressionSet, PredicateHelper}
 import org.apache.spark.sql.catalyst.util.{CaseInsensitiveMap, RebaseDateTime, TypeUtils}
 import org.apache.spark.sql.catalyst.util.RebaseDateTime.RebaseSpec
 import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryExecutionErrors}
+import org.apache.spark.sql.execution.datasources.csv.CSVFileFormat
+import org.apache.spark.sql.execution.datasources.json.JsonFileFormat
 import org.apache.spark.sql.execution.datasources.parquet.ParquetOptions
+import org.apache.spark.sql.execution.datasources.xml.XmlFileFormat
 import org.apache.spark.sql.internal.{LegacyBehaviorPolicy, SQLConf}
 import org.apache.spark.sql.sources.BaseRelation
 import org.apache.spark.sql.types._
@@ -161,6 +166,47 @@ object DataSourceUtils extends PredicateHelper {
       case _ => false
     }
 
+  /**
+   * Returns whether the rows this relation returns, or the values it returns for a column, depend
+   * on which columns the read was asked for. For such a relation, reading a wider set of columns is
+   * not just more work: it can return different data for the columns that were already being read.
+   *
+   * Two things put a V1 file source here. Its parser may resolve or validate a column against the
+   * set of columns it was asked for, which lets a wider read drop or rewrite rows that the narrower
+   * one returned: CSV, JSON and XML build their parser from the required schema and take `mode` and
+   * the corrupt-record column from it, and Avro under `positionalFieldMatching` pairs a column with
+   * the Avro field at its position in that schema. SPARK-59108 proposes removing that at the root;
+   * the Avro arm of `hasProjectionSensitiveParser` can go once that fix is on this branch. Or the
+   * read is not strict: under `ignoreCorruptFiles` a failure in a column only the wider read
+   * touches is swallowed together with the rest of that file's rows, whatever the format.
+   * `ignoreMissingFiles` has no such mechanism, since a missing file is skipped whatever is
+   * projected; it is here to match `FileSourceOptions.hasStrictFileReads`, the same predicate the
+   * reader and the cache-repeatability check in `InMemoryRelation` use.
+   *
+   * Callers that widen a read need this. Subplan merging is one: top-level column pruning for a V1
+   * file source happens in physical planning, from the attributes referenced above the relation, so
+   * reusing one relation for two subqueries that project different columns widens its read to the
+   * union of the two column sets.
+   */
+  private[sql] def isProjectionSensitiveRead(relation: BaseRelation): Boolean = relation match {
+    case hs: HadoopFsRelation =>
+      !new FileSourceOptions(hs.options).hasStrictFileReads ||
+        hasProjectionSensitiveParser(hs.fileFormat, hs.options)
+    case _ => false
+  }
+
+  private def hasProjectionSensitiveParser(
+      fileFormat: FileFormat, options: Map[String, String]): Boolean = fileFormat match {
+    case _: CSVFileFormat | _: JsonFileFormat | _: XmlFileFormat => true
+    // 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.
+    case _: AvroFileFormat =>
+      CaseInsensitiveMap(options).get(AvroOptions.POSITIONAL_FIELD_MATCHING)
+        .exists("true".equalsIgnoreCase)
+    case _ => false
+  }
+
   private def getRebaseSpec(
       lookupFileMeta: String => String,
       modeByConfig: String,

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

@@ -91,7 +91,7 @@ class FileScanRDD(
   private val ignoreMissingFiles = options.ignoreMissingFiles
 
   /** Whether this reader fails instead of silently skipping missing or corrupt input files. */
-  private[sql] def hasStrictFileReads: Boolean = !ignoreCorruptFiles && !ignoreMissingFiles
+  private[sql] val hasStrictFileReads: Boolean = options.hasStrictFileReads
 
   // Evaluated on the driver (sparkSession is @transient) and serialized to executors so the
   // `compute` iterator below can pass it through to ColumnVectorUtils.populate.