[SPARK-59108][4.0][SQL] 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

@@ -292,7 +292,8 @@ class AvroCatalystDataConversionSuite extends SparkFunSuite
       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

@@ -78,7 +78,8 @@ class AvroRowReaderSuite
           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 SQLTestUtils with 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

@@ -42,7 +42,7 @@ 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
 import org.apache.spark.sql.functions.col
@@ -1587,6 +1587,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(
@@ -3030,6 +3167,33 @@ class AvroV1Suite extends AvroSuite {
       .sparkConf
       .set(SQLConf.USE_V1_SOURCE_LIST, "avro")
 
+  test("SPARK-59108: two positional reads of different columns share one widened scan") {
+    // Subplan merging widens the projection of a shared V1 file scan, and this branch has nothing
+    // that keeps an avro relation out of it, so the two subqueries share one scan whatever this
+    // fix does. What it changes is the values they get: each column resolves against the data
+    // schema rather than against the merged projection. AQE off because `AdaptiveSparkPlanExec`
+    // is a leaf node, so with it on the scan underneath is not reachable from the executed plan.
+    withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+      withTempPath { dir =>
+        val path = dir.getCanonicalPath
+        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.toString).format("avro").load(path)
+            .createOrReplaceTempView("t")
+          // 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)
+          assert(scanColumns === Seq(Seq("b", "c")))
+        }
+      }
+    }
+  }
+
   test("SPARK-36271: V1 insert should check schema field name too") {
     withView("v") {
       spark.range(1).createTempView("v")

sql/core/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala

@@ -43,6 +43,13 @@ import org.apache.spark.unsafe.types.UTF8String
 
 /**
  * A deserializer to deserialize data in avro format to data in catalyst format.
+ *
+ * @param dataSchema The schema `rootCatalystType` was projected from, for a read that prunes
+ *                   columns. A positional field match pairs a Catalyst field with the Avro field
+ *                   at the same position, and that position is the one in the full schema rather
+ *                   than in the projection: without this, reading only the third column would take
+ *                   the first Avro field. `None` when the Catalyst type is not a projection;
+ *                   unused when field matching is by name.
  */
 private[sql] class AvroDeserializer(
     rootAvroType: Schema,
@@ -52,7 +59,8 @@ private[sql] class AvroDeserializer(
     filters: StructFilters,
     useStableIdForUnionType: Boolean,
     stableIdPrefixForUnionType: String,
-    recursiveFieldMaxDepth: Int) {
+    recursiveFieldMaxDepth: Int,
+    dataSchema: Option[StructType]) {
 
   def this(
       rootAvroType: Schema,
@@ -69,7 +77,8 @@ private[sql] class AvroDeserializer(
       new NoopFilters,
       useStableIdForUnionType,
       stableIdPrefixForUnionType,
-      recursiveFieldMaxDepth)
+      recursiveFieldMaxDepth,
+      dataSchema = None)
   }
 
   private lazy val decimalConversions = new DecimalConversion()
@@ -90,7 +99,8 @@ private[sql] class AvroDeserializer(
         val resultRow = new SpecificInternalRow(st.map(_.dataType))
         val fieldUpdater = new RowUpdater(resultRow)
         val applyFilters = filters.skipRow(resultRow, _)
-        val writer = getRecordWriter(rootAvroType, st, Nil, Nil, applyFilters)
+        val writer =
+          getRecordWriter(rootAvroType, st, Nil, Nil, applyFilters, positionsInDataSchema(st))
         (data: Any) => {
           val record = data.asInstanceOf[GenericRecord]
           val skipRow = writer(fieldUpdater, record)
@@ -268,8 +278,8 @@ private[sql] class AvroDeserializer(
       case (RECORD, st: StructType) =>
         // Avro datasource doesn't accept filters with nested attributes. See SPARK-32328.
         // We can always return `false` from `applyFilters` for nested records.
-        val writeRecord =
-          getRecordWriter(avroType, st, avroPath, catalystPath, applyFilters = _ => false)
+        val writeRecord = getRecordWriter(
+          avroType, st, avroPath, catalystPath, applyFilters = _ => false, Array.empty)
         (updater, ordinal, value) =>
           val row = new SpecificInternalRow(st)
           writeRecord(new RowUpdater(row), value.asInstanceOf[GenericRecord])
@@ -409,15 +419,44 @@ private[sql] class AvroDeserializer(
     }
   }
 
+  /**
+   * The position of each `projection` field in `dataSchema`, which is what a positional field match
+   * resolves against. Empty when there is no data schema to resolve against, or when field matching
+   * is by name and the positions are unused.
+   *
+   * This takes a data schema position for an Avro field position, which `recursiveFieldMaxDepth`
+   * can break: `SchemaConverters` drops a field it will not recurse into, so the data schema is a
+   * gapped view of the Avro schema and every field after the gap resolves one position early.
+   * Positional matching is already wrong for such a schema without this method, because the fields
+   * after the gap shift by one whatever the projection is.
+   */
+  private def positionsInDataSchema(projection: StructType): Array[Int] = dataSchema match {
+    case Some(schema) if positionalFieldMatch =>
+      projection.map(field => schema.fieldIndex(field.name)).toArray
+    case _ => Array.empty
+  }
+
+  /**
+   * Creates a writer that reads a record's fields into `catalystType`'s fields.
+   *
+   * @param dataSchemaPositions The positions a positional field match resolves `catalystType`'s
+   *                            fields against, empty to use each field's own position. Only the
+   *                            root record passes them: a nested record is never a projection,
+   *                            because V1 nested pruning is limited to Parquet and ORC
+   *                            (`SchemaPruning.canPruneDataSchema`) and V2's
+   *                            `FileScanBuilder.supportsNestedSchemaPruning` is false for Avro.
+   */
   private def getRecordWriter(
       avroType: Schema,
       catalystType: StructType,
       avroPath: Seq[String],
       catalystPath: Seq[String],
-      applyFilters: Int => Boolean): (CatalystDataUpdater, GenericRecord) => Boolean = {
+      applyFilters: Int => Boolean,
+      dataSchemaPositions: Array[Int])
+      : (CatalystDataUpdater, GenericRecord) => Boolean = {
 
     val avroSchemaHelper = new AvroUtils.AvroSchemaHelper(
-      avroType, catalystType, avroPath, catalystPath, positionalFieldMatch)
+      avroType, catalystType, avroPath, catalystPath, positionalFieldMatch, dataSchemaPositions)
 
     avroSchemaHelper.validateNoExtraCatalystFields(ignoreNullable = true)
     // no need to validateNoExtraAvroFields since extra Avro fields are ignored

sql/core/src/main/scala/org/apache/spark/sql/avro/AvroFileFormat.scala

@@ -145,7 +145,8 @@ private[sql] class AvroFileFormat extends FileFormat
             avroFilters,
             parsedOptions.useStableIdForUnionType,
             parsedOptions.stableIdPrefixForUnionType,
-            parsedOptions.recursiveFieldMaxDepth)
+            parsedOptions.recursiveFieldMaxDepth,
+            dataSchema = Some(dataSchema))
           override val stopPosition = file.start + file.length
 
           override def hasNext: Boolean = hasNextRow