[SPARK-59108][SQL][4.3] Fix Avro positional matching under column pruning

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

@@ -106,7 +106,8 @@ case class AvroPartitionReaderFactory(
           avroFilters,
           options.useStableIdForUnionType,
           options.stableIdPrefixForUnionType,
-          options.recursiveFieldMaxDepth)
+          options.recursiveFieldMaxDepth,
+          dataSchema = Some(dataSchema))
         override val stopPosition = partitionedFile.start + partitionedFile.length
 
         override def next(): Boolean = hasNextRow

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

@@ -291,7 +291,8 @@ class AvroCatalystDataConversionSuite extends SharedSparkSession
       filters,
       false,
       "",
-      -1)
+      -1,
+      dataSchema = None)
     val deserialized = deserializer.deserialize(data)
     expected match {
       case None => assert(deserialized == None)

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

@@ -75,7 +75,8 @@ class AvroRowReaderSuite extends SharedSparkSession {
           new NoopFilters,
           false,
           "",
-          -1)
+          -1,
+          dataSchema = None)
         override val stopPosition = fileSize
 
         override def hasNext: Boolean = hasNextRow

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

@@ -87,6 +87,49 @@ class AvroSchemaHelperSuite extends SharedSparkSession {
     assert(nameHelper.getAvroField("nonexist", 1).isEmpty)
   }
 
+  test("SPARK-59108: positional field match resolves against the data schema positions") {
+    val dataSchema = new StructType()
+      .add("a", IntegerType).add("b", IntegerType).add("c", IntegerType)
+    val avroSchema = SchemaConverters.toAvroType(dataSchema)
+    val projection = new StructType().add("c", IntegerType).add("a", IntegerType)
+
+    val helper = new AvroUtils.AvroSchemaHelper(
+      avroSchema, projection, Seq(""), Seq(""), true, Array(2, 0))
+    assert(helper.getAvroField("c", 0) === Some(avroSchema.getFields.get(2)))
+    assert(helper.getAvroField("a", 1) === Some(avroSchema.getFields.get(0)))
+    assert(helper.matchedFields.map(_.avroField.name()) === Seq("c", "a"))
+
+    // With no positions a field's own position is used, which is what an unprojected match needs.
+    val unprojected =
+      new AvroUtils.AvroSchemaHelper(avroSchema, projection, Seq(""), Seq(""), true)
+    assert(unprojected.getAvroField("c", 0) === Some(avroSchema.getFields.get(0)))
+
+    // The shape both read paths produce is an ascending subsequence of the data schema.
+    val ascending = new StructType().add("a", IntegerType).add("c", IntegerType)
+    val ascendingHelper = new AvroUtils.AvroSchemaHelper(
+      avroSchema, ascending, Seq(""), Seq(""), true, Array(0, 2))
+    assert(ascendingHelper.getAvroField("a", 0) === Some(avroSchema.getFields.get(0)))
+    assert(ascendingHelper.getAvroField("c", 1) === Some(avroSchema.getFields.get(2)))
+    assert(ascendingHelper.matchedFields.map(_.avroField.name()) === Seq("a", "c"))
+
+    val msg = intercept[IllegalArgumentException] {
+      new AvroUtils.AvroSchemaHelper(avroSchema, projection, Seq(""), Seq(""), true, Array(2))
+    }.getMessage
+    assert(msg.contains("Got 1 data schema positions for 2 Catalyst fields"))
+
+    // A missing field is reported by the position that was looked for, not by the position the
+    // field happens to have in the projection.
+    val twoFieldAvro = SchemaConverters.toAvroType(
+      new StructType().add("a", IntegerType).add("b", IntegerType))
+    val pastTheEnd = new AvroUtils.AvroSchemaHelper(
+      twoFieldAvro, new StructType().add("c", IntegerType, nullable = false),
+      Seq(""), Seq(""), true, Array(2))
+    val missing = intercept[IncompatibleSchemaException] {
+      pastTheEnd.validateNoExtraCatalystFields(ignoreNullable = false)
+    }.getMessage
+    assert(missing.contains("Cannot find field at position 2"))
+  }
+
   test("properly match fields between Avro and Catalyst schemas") {
     val catalystSchema = StructType(
       Seq("catalyst1", "catalyst2", "shared1", "shared2").map(StructField(_, IntegerType))

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

@@ -229,7 +229,8 @@ object AvroSerdeSuite {
         new NoopFilters,
         false,
         "",
-        -1)
+        -1,
+        dataSchema = None)
   }
 
   /**

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

@@ -45,7 +45,6 @@ import org.apache.spark.sql.catalyst.util.DateTimeTestUtils.{withDefaultTimeZone
 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._
@@ -1736,6 +1735,143 @@ abstract class AvroSuite
     }
   }
 
+  test("SPARK-59108: positionalFieldMatching resolves fields against the full schema") {
+    withTempPath { dir =>
+      val path = dir.getCanonicalPath
+      spark.range(0, 5).selectExpr("id AS a", "id * 100 AS b", "id * 10000 AS c")
+        .write.format("avro").save(path)
+      // The names differ from the file's, so only the positions can pair the two schemas.
+      val renamedSchema = new StructType()
+        .add("x", LongType).add("y", LongType).add("z", LongType)
+      val df = spark.read.format("avro")
+        .option("positionalFieldMatching", true.toString)
+        .schema(renamedSchema)
+        .load(path)
+
+      val rows = (0 until 5).map(i => Row(i.toLong, i * 100L, i * 10000L))
+      checkAnswer(df, rows)
+      // A column keeps its own Avro field however few of them the query projects.
+      checkAnswer(df.select("z"), rows.map(r => Row(r.get(2))))
+      checkAnswer(df.select("y"), rows.map(r => Row(r.get(1))))
+      checkAnswer(df.select("x", "z"), rows.map(r => Row(r.get(0), r.get(2))))
+      checkAnswer(df.select("z", "x"), rows.map(r => Row(r.get(2), r.get(0))))
+      checkAnswer(df.select("y", "z"), rows.map(r => Row(r.get(1), r.get(2))))
+      checkAnswer(df.selectExpr("sum(z)"), Row(100000L))
+      // With pushdown on, the filter runs inside the deserializer; with it off, it runs above the
+      // scan.
+      // Either way a wrong pairing drops rows rather than only returning wrong values for them.
+      Seq("true", "false").foreach { pushDown =>
+        withSQLConf(SQLConf.AVRO_FILTER_PUSHDOWN_ENABLED.key -> pushDown) {
+          checkAnswer(df.where("z = 20000").select("z"), Row(20000L))
+          checkAnswer(df.where("z > 20000").select("x"), Seq(Row(3L), Row(4L)))
+        }
+      }
+      // A projection of no columns at all.
+      checkAnswer(df.selectExpr("count(1)"), Row(5L))
+
+      // The projected schema carries the schema's own spelling whatever casing the query used, so
+      // the name lookup that resolves a position finds the field either way.
+      val mixedCase = spark.read.format("avro")
+        .option("positionalFieldMatching", true.toString)
+        .schema(new StructType().add("Xx", LongType).add("yY", LongType).add("ZZ", LongType))
+        .load(path)
+      Seq("true", "false").foreach { caseSensitive =>
+        withSQLConf(SQLConf.CASE_SENSITIVE.key -> caseSensitive) {
+          checkAnswer(mixedCase.select("ZZ"), rows.map(r => Row(r.get(2))))
+        }
+      }
+      withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") {
+        checkAnswer(mixedCase.select("zz"), rows.map(r => Row(r.get(2))))
+      }
+    }
+  }
+
+  test("SPARK-59108: positionalFieldMatching with a partition column in the schema") {
+    withTempPath { dir =>
+      val path = dir.getCanonicalPath
+      spark.range(0, 4).selectExpr("id AS a", "id * 100 AS b", "id % 2 AS p")
+        .write.partitionBy("p").format("avro").save(path)
+      // p is a partition column, so the files hold a and b only and the data schema is x and z.
+      val df = spark.read.format("avro")
+        .option("positionalFieldMatching", true.toString)
+        .schema("x long, p int, z long")
+        .load(path)
+
+      checkAnswer(df.select("z"), (0 until 4).map(i => Row(i * 100L)))
+      checkAnswer(df.select("x"), (0 until 4).map(i => Row(i.toLong)))
+      checkAnswer(df.select("p", "z"), (0 until 4).map(i => Row(i % 2, i * 100L)))
+      checkAnswer(df.where("p = 1").select("z"), Seq(Row(100L), Row(300L)))
+    }
+  }
+
+  test("SPARK-59108: positionalFieldMatching with a nested record and the avroSchema option") {
+    withTempPath { dir =>
+      val path = dir.getCanonicalPath
+      spark.range(0, 3).selectExpr(
+          "id AS a",
+          "named_struct('f1', id * 10, 'f2', cast(id AS string)) AS r",
+          "id * 1000 AS c")
+        .write.format("avro").save(path)
+
+      // Only the top level is a projection, so the nested record keeps resolving by its own
+      // positions. Reading the struct alone would take Avro field 0, a long, and fail.
+      val df = spark.read.format("avro")
+        .option("positionalFieldMatching", true.toString)
+        .schema("x long, s struct<g1: long, g2: string>, z long")
+        .load(path)
+      checkAnswer(df.select("s"), (0 until 3).map(i => Row(Row(i * 10L, i.toString))))
+      checkAnswer(df.select("s.g2"), (0 until 3).map(i => Row(i.toString)))
+      checkAnswer(df.select("z"), (0 until 3).map(i => Row(i * 1000L)))
+
+      // The avroSchema option supplies the Avro side, and the data schema is inferred from it, so
+      // the positions are the option's.
+      val avroSubset =
+        """{"type":"record","name":"topLevelRecord","fields":[
+          |{"name":"a","type":"long"},
+          |{"name":"c","type":"long"}]}""".stripMargin
+      val fromOption = spark.read.format("avro")
+        .option("positionalFieldMatching", true.toString)
+        .option("avroSchema", avroSubset)
+        .load(path)
+      checkAnswer(fromOption.select("c"), (0 until 3).map(i => Row(i * 1000L)))
+      checkAnswer(fromOption.select("a"), (0 until 3).map(i => Row(i.toLong)))
+    }
+  }
+
+  test("SPARK-59108: a position past the end of the Avro schema reads null") {
+    withTempPath { dir =>
+      val path = dir.getCanonicalPath
+      spark.range(0, 3).selectExpr("id AS a", "id * 100 AS b").write.format("avro").save(path)
+      val df = spark.read.format("avro")
+        .option("positionalFieldMatching", true.toString)
+        .schema("x long, y long, z long")
+        .load(path)
+
+      // z is at position 2 of the schema and the file has two fields, so it has no Avro field to
+      // read and comes back null however few columns the query projects.
+      checkAnswer(df.select("z"), Seq(Row(null), Row(null), Row(null)))
+      checkAnswer(df, (0 until 3).map(i => Row(i.toLong, i * 100L, null)))
+    }
+  }
+
+  test("SPARK-59108: positionalFieldMatching fails a mispaired type rather than reading it") {
+    withTempPath { dir =>
+      val path = dir.getCanonicalPath
+      spark.range(0, 3).selectExpr("id AS a", "cast(id AS string) AS b", "id * 10 AS c")
+        .write.format("avro").save(path)
+      val df = spark.read.format("avro")
+        .option("positionalFieldMatching", true.toString)
+        .schema("x long, y long, z long")
+        .load(path)
+
+      // y takes Avro field 1, which is a string, so the read fails instead of returning the values
+      // of a neighbouring field.
+      val ex = intercept[SparkException](df.select("y").collect())
+      assert(Utils.exceptionString(ex).contains("Cannot convert Avro"))
+      checkAnswer(df.select("z"), (0 until 3).map(i => Row(i * 10L)))
+    }
+  }
+
   test("int/long double/float conversion") {
     val catalystSchema =
       StructType(Seq(
@@ -3733,37 +3869,30 @@ 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.
+  test("SPARK-59108: two positional reads of different columns share one widened scan") {
+    // SPARK-59107 named avro under this option, so the two subqueries used to keep their own scans.
+    // They share one now, and the values are the file's either way because each column resolves
+    // against the data schema. AQE off because `AdaptiveSparkPlanExec` is a leaf node, so with it
+    // on the scan underneath is 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)
+        spark.range(0, 5).selectExpr("id AS a", "id * 10 AS b", "id * 100 AS c")
+          .write.format("avro").save(path)
         withTempView("t") {
-          spark.read.option("positionalFieldMatching", "true").format("avro").load(path)
+          spark.read.option("positionalFieldMatching", true.toString).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)
+          // b and c sit at data schema positions 1 and 2, so the merged read of the two has to
+          // resolve against the data schema rather than against its own projection.
+          val df = sql("SELECT (SELECT sum(b) FROM t), (SELECT sum(c) FROM t)")
+          checkAnswer(df, Row(100L, 1000L))
           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")))
+          assert(scanColumns === Seq(Seq("b", "c")))
         }
       }
     }

connector/avro/src/test/scala/org/apache/spark/sql/execution/datasources/AvroArchiveReadBase.scala

@@ -22,9 +22,11 @@ import java.io.{ByteArrayInputStream, IOException}
 import org.apache.avro.file.{DataFileConstants, DataFileStream}
 import org.apache.avro.generic.{GenericDatumReader, GenericRecord}
 
+import org.apache.spark.sql.Row
+
 /**
- * Binds [[ArchiveReadSuiteBase]]'s hooks to Avro, adding the streaming-reader regression tests
- * that have no format-agnostic analogue.
+ * Binds [[ArchiveReadSuiteBase]]'s hooks to Avro, adding the Avro-only tests that have no
+ * format-agnostic analogue.
  */
 trait AvroArchiveReadBase extends ArchiveReadSuiteBase {
 
@@ -47,6 +49,22 @@ trait AvroArchiveReadBase extends ArchiveReadSuiteBase {
 
   // ----- Avro-specific tests -------------------------------------------------
 
+  test("Avro: positionalFieldMatching resolves a pruned read against the full schema") {
+    // This is a second deserializer construction site, and it is handed a pruned required schema
+    // like the per-file reader is (SPARK-59108). The two columns have different types, so a wrong
+    // pairing fails the read.
+    withArchiveFile() { archive =>
+      writeArchive(archive, Seq(entryName(0) -> encodeFile(sampleDf((1, "Alice"), (2, "Bob")))))
+      val df = read(
+        archive.getCanonicalPath,
+        extraOptions = Map("positionalFieldMatching" -> "true"),
+        schema = "num INT, label STRING")
+      checkAnswer(df.select("label"), Seq(Row("Alice"), Row("Bob")))
+      checkAnswer(df.select("num"), Seq(Row(1), Row(2)))
+      checkAnswer(df, Seq(Row(1, "Alice"), Row(2, "Bob")))
+    }
+  }
+
   test("Avro: a truncated entry fails fast instead of spinning") {
     // A DataFileStream must throw on a truncated entry rather than loop forever. Cut at the header
     // and mid-file; the per-test timeout catches a regression to a spin.

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. 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.
+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, 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.