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

### What changes were proposed in this pull request?

Backport of `c809c283c2d` (#58409) to `branch-4.2`. The Avro fix comes over as it landed; the two gate removals that PR also carried are not here, because neither gate exists on this branch.

`AvroDeserializer` now takes the schema its Catalyst schema was projected from, and under `positionalFieldMatching` it resolves a Catalyst field against that field's position in the data schema rather than its position in the projection. `AvroUtils.AvroSchemaHelper` takes the resulting positions; with none it keeps using a field's own position, which is what every caller whose Catalyst schema is not a projection needs (`from_avro`, the write path, the state-store encoder).

Two read call sites pass the data schema on this branch: `AvroPartitionReaderFactory` on the V2 path and `AvroFileFormat.buildReader` on V1. Master has a third, `AvroFileFormat.readArchive`, which does not exist here. A nested record keeps resolving by its own positions, since neither read path prunes nested fields: `FileScanBuilder.supportsNestedSchemaPruning` is false and `AvroScanBuilder` does not override it, and `SchemaPruning.canPruneDataSchema` covers only Parquet and ORC.

ORC already does this for `orc.force.positional.evolution`: `OrcUtils.requestedColumnIds` maps the required schema through `dataSchema.fieldIndex(name)`, which makes its positional path projection-independent. Avro decodes the whole record whatever the projection asks for, so nothing extra is read.

Master retired two gates that kept avro out of scan merging, and neither is on this branch: SPARK-57205 (#58340) withheld the `SCAN_MERGING` capability from `AvroTable`, and SPARK-59107 (#58411) named avro in `DataSourceUtils.isProjectionSensitiveRead`. So there is no predicate to change, no `AvroTable.supportsScanMerging` to turn on, no test to delete, and `docs/sql-performance-tuning.md` has no "Merging Subplans" section stating the old behaviour. That also means subplan merging can widen an avro projection here with nothing in front of it, which makes the position mapping worth more on this branch than on 4.3, not less.

One shape stays broken, with or without this change: `recursiveFieldMaxDepth` makes `SchemaConverters` drop a field it will not recurse into, so the data schema is a gapped view of the Avro schema and positional matching misaligns from the gap onwards. The code records that where the positions are computed.

### Why are the changes needed?

With `positionalFieldMatching=true` the deserializer is built from the projected read schema while the Avro side stays the full Avro schema, and `AvroUtils.AvroSchemaHelper.getAvroField` pairs Catalyst field *i* with Avro field *i*, so a column-pruned read takes the wrong Avro field and returns wrong values with no error. Measured on a file whose fields `a`, `b`, `c` hold `id`, `100 * id`, `10000 * id` for ids 0 to 4, read with the option on:

```
sql("SELECT sum(a), sum(b), sum(c) FROM t").show()  // 10, 1000, 100000 -- all correct
sql("SELECT sum(c) FROM t").show()                  // 10       -- should be 100000
sql("SELECT sum(b) FROM t").show()                  // 10       -- should be 1000
sql("SELECT sum(a), sum(c) FROM t").show()          // 10, 1000 -- sum(c) should be 100000
```

Only a projection that is a prefix of the file's field list comes back right, so a column's value depends on which other columns the query selects. Both read paths behave the same way. Whether the failure is silent depends on the types of the mispaired fields: matching types return wrong values, as above, and incompatible ones fail the read with a schema-incompatibility error instead. A pushed filter is evaluated inside the deserializer, so the wrong pairing can also drop rows rather than only return wrong values for them.

A pruned projection is all it takes, so this does not depend on scan merging. Merging only makes it easier to reach without asking for it, and on this branch nothing keeps an avro relation out of it.

### Does this PR introduce _any_ user-facing change?

Yes, a bug fix on the Avro read path, both V1 and V2, and every 4.2.x release shipped the bug: positional matching has resolved against the projection since 3.2.0 (SPARK-34365). A read that sets `positionalFieldMatching` and prunes columns now returns the values of the columns it asked for. A query whose projection is a prefix of the Avro field list is unaffected, which is why the option's existing tests need no change. A read that used to land on a type-compatible neighbouring field now pairs with its own field and fails when the two types do not match, so a query that returned values before this change can return an error instead. That is the point of the fix rather than a side effect, but it is the shape most likely to be reported as a regression. The "Cannot find field at position N" message that positional matching raises now names the position it looked for rather than the position within the projection, which are the same number for an unprojected read. Nothing changes when the option is off, which is the default, and nothing changes on the write path or in `from_avro`.

### How was this patch tested?

Five new tests in `AvroSuite`, so each runs on both read paths (`AvroV1Suite` and `AvroV2Suite` extend it): the renamed-schema shape from the description, with each one-column and two-column projection whose values the fix changes, the ones it leaves alone being the prefixes of the field list, a pushed filter under both settings of `spark.sql.avro.filterPushdown.enabled`, `count(1)`, and mixed-case names under both case-sensitivity settings; a partition column sitting between two data columns in the schema; a nested record, which must keep resolving by its own positions, together with the `avroSchema` option supplying the Avro side; a projection that reaches past the end of the Avro schema, which reads null; and a mispaired type, which fails the read rather than returning a neighbouring field's values. One test in `AvroSchemaHelperSuite` for the helper itself. Master's `AvroArchiveReadBase` case is not here, since this branch has no archive reader.

One more test, in `AvroV1Suite`, for a merged read. The file has three columns and the two scalar aggregates read the last two, so the merged projection is a proper subset of the data schema and the read has to resolve against that schema to answer `[100, 1000]`; the scan is one widened `FileSourceScanExec` reading both columns. Unlike master's version it pins only AQE, since the strictness flags matter to a predicate that does not exist on this branch, and it has no `AvroV2Suite` twin, since avro never declares `SCAN_MERGING` here.

Mutation check, measured on this branch: with the position mapping disabled, 11 cases fail, the five `SPARK-59108` shapes on each read path and the new merge test, which answers `[10, 100]` where the file has `[100, 1000]`.

Regression, measured on this branch: the whole `avro` module, 404 tests, and `avro/scalastyle`, `avro/Test/scalastyle`, `sql/scalastyle` and `catalyst/scalastyle`. `RocksDBStateEncoderSuite` and `StateStoreSuite`, which also build an `AvroDeserializer`, were run on master rather than here; nothing in this backport differs from the master commit in that path. The existing `positionalFieldMatching` tests (SPARK-34365) needed no change, because their projections cover the whole schema.

### Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude Code

Closes #58538 from LuciferYang/SPARK-59108-4.2.

Authored-by: YangJie <yangjie01@baidu.com>
Signed-off-by: yangjie01 <yangjie01@baidu.com>

Author: LuciferYang

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

@@ -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

@@ -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, FileDataSourceV2, FileTable}
 import org.apache.spark.sql.functions._
@@ -1735,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(
@@ -3374,6 +3511,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

@@ -44,6 +44,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,
@@ -53,7 +60,8 @@ private[sql] class AvroDeserializer(
     filters: StructFilters,
     useStableIdForUnionType: Boolean,
     stableIdPrefixForUnionType: String,
-    recursiveFieldMaxDepth: Int) {
+    recursiveFieldMaxDepth: Int,
+    dataSchema: Option[StructType]) {
 
   def this(
       rootAvroType: Schema,
@@ -70,7 +78,8 @@ private[sql] class AvroDeserializer(
       new NoopFilters,
       useStableIdForUnionType,
       stableIdPrefixForUnionType,
-      recursiveFieldMaxDepth)
+      recursiveFieldMaxDepth,
+      dataSchema = None)
   }
 
   private lazy val decimalConversions = new DecimalConversion()
@@ -91,7 +100,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)
@@ -285,8 +295,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])
@@ -426,15 +436,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

@@ -148,7 +148,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