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

[LuciferYang]

What changes were proposed in this pull request?

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).

The three read call sites pass the data schema: AvroPartitionReaderFactory on the V2 path, AvroFileFormat.buildReader and AvroFileFormat.readArchive on V1. 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.

Two gates kept avro out of scan merging because merging widens the projection, and this commit retires both, since both landed while this was open: #58340 (SPARK-57205, 484866bd80d) withheld the SCAN_MERGING capability from AvroTable on the V2 path, and #58411 (SPARK-59107, b82f9872d1c) named avro in DataSourceUtils.isProjectionSensitiveRead on the V1 path. AvroTable.supportsScanMerging becomes an unconditional true rather than a deleted override, because FileTable defaults it to false and dropping it would take the capability away from Avro V2 altogether; hasProjectionSensitiveParser loses its avro arm and, with it, its options parameter and the org.apache.spark.sql.avro import. Two tests go with them: the AvroV1Suite case #58411 added, whose assertion that each subquery keeps its own scan stops being true, and the capability assertion in the AvroV2SuiteSPARK-57205 case. The documentation has to change whether or not the predicates come off, because it states the old behaviour in two places: docs/sql-performance-tuning.md listed avro among the projection-sensitive V1 relations and said Avro withholds the capability under positionalFieldMatching, and after this fix the position is the data schema's. FileTable's class doc names the same shape as a general statement of the contract, so that one stays.

A backport has to carry the same removal wherever both prerequisites are, which today is master and branch-4.x.

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.

Does this PR introduce any user-facing change?

Yes, a bug fix on the Avro read path, both V1 and V2. 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, and one in AvroArchiveReadBase, which runs in the tar, zip and 7z suites, because the archive reader builds its own deserializer per entry.

Two more tests, one in AvroV1Suite and one in AvroV2Suite, for the shape the removed gates used to decline. 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 scans are one widened read of both columns, FileSourceScanExec on V1 and one canonically distinct DataSourceV2ScanRelation on V2, and the V2 case also asserts that the positional table now advertises SCAN_MERGING. Both pin the strictness flags, since a non-strict read is projection-sensitive for the other reason.

Mutation checks. With the position mapping disabled, all ten of the original AvroSuite cases fail (five on each path), so does the archive one, and so do the two new ones, which answer [10, 100] where the file has [100, 1000]. With the two gates put back instead, the two new ones fail the other way, the V1 one with a scan per column rather than one reading both and the V2 one at the capability assertion. Between the two mutations they pin both halves: that merging is allowed here, and that what makes it safe is the mapping. The two columns in the archive test have different types on purpose, so a wrong pairing fails the read there rather than returning plausible values.

Regression, re-measured at this head after the merge and the gate removal: the whole avro module, 501 tests, the planmerging package, 134 tests, since removing the avro arm touches the shared predicate, and avro/scalastyle, avro/Test/scalastyle, sql/scalastyle, sql/Test/scalastyle and catalyst/scalastyle. RocksDBStateEncoderSuite plus StateStoreSuite, 840 tests, were run on the pre-merge head, because the state-store encoder builds an AvroDeserializer too and nothing since has touched it. 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

[peter-toth]

Thanks for the PR, @LuciferYang!

This is the read-side root cause behind #58340's Avro gate, and the shape looks right. A column's Avro field is now fixed by its position in the data schema, so it no longer depends on what the query projects.

I re-ran it in a worktree at a3358708e6. The 11 new tests pass. With positionsInDataSchema returning empty, all 10 AvroSuite cases fail, five on each path, so your mutation check holds. I also re-ran the plan-merging repro from #58411 - SELECT (SELECT sum(a) FROM t), (SELECT sum(c) FROM t) is [10, 100000] both merged and unmerged here, against [10, 1000] and [10, 10] with the mapping off.

Nothing blocking from my side. Everything below is a description, coverage or naming item, and none of it changes the mechanism.

Non-blocking

• 1.Test coverage - one distinguishing projection is left out: The description says "every one-column and two-column projection", but the test covers z, y, (x, z) and (z, x). x and (x, y) are prefixes and come back right on base too, which I measured, so (y, z) is the only shape the fix changes that nothing asserts. [inline: connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala:1757]
• 2.A read that returned values can now fail, and the user-facing section doesn't say so: Repairing the pairing can land a projected column on an incompatible Avro field. That fails the read where it used to return a neighbouring field's values. Your "fails a mispaired type" test is exactly that shape. [inline: connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala:1865]
• 3.This also retires the open Avro request on #58411: The mapping makes the read projection-independent, which is stronger than making a pruned read correct. Worth naming #58411 beside #58340 so the three land in a consistent state. [inline: sql/core/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala:476]

Minor

• 4.The reason given for the pre-existing gap is not the one that breaks it: Differing lengths are fine on their own - AvroSuite.scala:1701 reads two Catalyst fields out of test.avro's eleven Avro fields and is correct. What breaks a gapped schema is where the dropped field sits. [inline: sql/core/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala:473]
• 5.Drop the dataSchemaPositions default: Two call sites, both in this file, and you already made AvroDeserializer.dataSchema explicit for the same reason. [inline: sql/core/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala:498]

[peter-toth]

Finding 1. The description says "with every one-column and two-column projection", but this covers four of the six: z, y, (x, z) and (z, x).

Of the three left out, two are prefixes of the field list and so come back right on base as well. I measured them with positionsInDataSchema returning empty: select("x") gives 0..4 and select("x", "y") gives (i, 100 * i), both correct. So they add nothing.

(y, z) is the one that does. On base it reads (i, 100 * i), on your head (100 * i, 10000 * i), so it is the last two-column shape the fix changes that nothing asserts.

Suggested change

	checkAnswer(df.select("z", "x"), rows.map(r =>Row(r.get(2), r.get(0))))
	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))))

With that line in, the description's claim holds for the one-column and two-column projections that distinguish the fix, and it is worth saying that rather than "every".

[peter-toth]

Finding 2. This test is the one user-facing consequence the description's "Does this PR introduce any user-facing change?" section leaves out.

On base, df.select("y") pairs y with Avro field 0, a long, and returns a's values. On your head it pairs with field 1, a string, and the read throws. So a query that produced numbers before this change now fails.

That is the right behaviour and I am not asking you to soften it. It is also the shape most likely to come back as a regression report, so it belongs in that section rather than only in a test comment. One sentence, something like: a read whose projection previously took a type-compatible neighbouring field now pairs with its own field and fails when the types do not match.

[peter-toth]

Finding 3. What this method buys is stronger than "a pruned read is now correct", and the stronger property is what two other PRs need.

A column's Avro field is derived from dataSchema alone, so it is the same whatever the projection contains. That holds even for the gapped schema you describe below: a recursiveFieldMaxDepth read stays wrong, but it is now consistently wrong rather than dependent on the projection.

Measured on a worktree at this head with fields a, b, c = id, 100 * id, 10000 * id, ids 0 to 4, positionalFieldMatching=true, V1:

	SELECT (SELECT sum(a) FROM t), (SELECT sum(c) FROM t) merged	MergeSubplans excluded
your head	[10, 100000]	[10, 100000]
mapping disabled	[10, 1000]	[10, 10]

The description names #58340's AvroTable gate. #58411 has the same item open on the V1 side - a request to add V1 Avro to DataSourceUtils.isProjectionSensitiveRead because positionalFieldMatching makes the read projection-sensitive. This removes the need for that too, and the numbers above are the evidence. Worth naming both PRs so the three land in a consistent state, and worth saying which has to go first.

[peter-toth]

Finding 4. The conclusion is right but the reason given is not the one that breaks it.

Two schemas of different lengths are fine on their own. AvroSuite.scala:1701 (SPARK-34365: support reading renamed schema using positionalFieldMatching) reads a two-field Catalyst schema out of test.avro, which has eleven Avro fields, and asserts the values are correct. Extra trailing Avro fields are simply ignored.

What breaks a gapped schema is where the dropped field sits, which is what the sentence above already says. Suggest:

*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.

[peter-toth]

Finding 5. You gave AvroDeserializer.dataSchema no default so every call site has to decide. The same argument applies here, and more so - the nested-record call at line 323 is the one place where getting it wrong returns a neighbouring field's value.

There are only two call sites and both are in this file, so the default buys nothing:

applyFilters: Int=>Boolean,
dataSchemaPositions: Array[Int])

and at line 323:

valwriteRecord= getRecordWriter(
avroType, st, avroPath, catalystPath, applyFilters = _ =>false, Array.empty)

Not a defect - your scaladoc already states the invariant and cites the two facts it rests on. And select("s.g2") in the nested-record test would fail loudly if Avro ever gained nested pruning without this being extended, since g2 is a string and Avro field 0 is a long.

[LuciferYang]

Thanks @peter-toth, and thanks for re-running it. All five taken.

1. Added select("y", "z"). You are right that the other two are prefixes and pass on base, so the description now says "each one-column and two-column projection whose values the fix changes" and names why the rest are left out, rather than claiming every one.

2. Added to the user-facing section: a read that used to land on a type-compatible neighbouring field now pairs with its own field and fails when the types do not match, so a query that returned values can return an error instead. It reads better there than in a test comment, and it is the shape most likely to come back as a report.

3. The description now names both, #58340 on the V2 path and #58411 on the V1 path, says that this makes the read projection-independent rather than only correct under pruning, and says this is the one to land first so the two gates can come off together.

4. You are right, and the example settles it: test.avro has twelve Avro fields and the SPARK-34365 test reads two Catalyst fields out of it correctly, so trailing extras are simply ignored. The sentence now says the fields after the gap shift by one whatever the projection is.

5. Default dropped, and the nested call passes Array.empty explicitly.

Locally: the ten AvroSuite cases on both paths, AvroSchemaHelperSuite, AvroSerdeSuite and AvroCatalystDataConversionSuite, 64 tests; avro/scalastyle, avro/Test/scalastyle and sql/scalastyle.

[peter-toth]

Re-checked through d75fd68e9ae — findings 1-5 all resolved, nothing regressed. All 11 new tests pass at this head, and select("y", "z") now pins the last two-column shape the fix changes.

The merge in this head is what moves the picture: both prerequisites are in master now, #58340 as 484866bd80d and #58411 as b82f9872d1c, and both are on branch-4.x too.

Blocking

• 6.Retire both Avro gates in this commit (new): The description schedules the removal for this commit and both prerequisites have landed, so it is due. It is more than the two predicates: docs/sql-performance-tuning.md states the projection-position behaviour in two places, and that becomes false whether or not the gates come off. [inline: AvroPartitionReaderFactory.scala:110]
• 7.The description still calls the two prerequisites open pull requests (new): "Two open pull requests gate avro on this option" and "it should land after them and remove them in one commit" describe the state before the merge. Both are merged, so that paragraph should say what this commit removes rather than what it will have to.

[peter-toth]

Finding 6. Both prerequisites landed while this was open, so "it should land after them and remove them in one commit" now applies to this commit: #58340 is 484866bd80d on master and 69e1cb10f43 on branch-4.x, #58411 is b82f9872d1c and d9176bf31be. The merge at this head brings both in, and the diff still leaves both gates standing. Nothing fails while they stand, which is why it is easy to miss.

The part that makes this more than a tidy-up is the documentation. docs/sql-performance-tuning.md:343 lists "avro read with positionalFieldMatching, which pairs a column with the Avro field at its position in that schema" among the projection-sensitive V1 relations, and :345 ends with "Avro withholds it under positionalFieldMatching, which resolves a column by its position in the projection". Neither sentence survives this fix, since the position is the data schema's. Those two paragraphs have to change in this PR whether or not the predicates come off with them.

The rest, in the order I would do it:

• AvroTable.scala:67 — supportsScanMerging becomes override protected def supportsScanMerging: Boolean = true, not a deleted override. FileTable's default is false, so dropping it would take the capability away from Avro V2 altogether. The second paragraph of the comment above it goes, the first one ("Avro has no record-level parse verdict") stays, and AvroOptions leaves the import.
• DataSourceUtils.scala:203-205 — drop the case _: AvroFileFormat arm. options is then unused, so hasProjectionSensitiveParser loses that parameter and :193 becomes hasProjectionSensitiveParser(hs.fileFormat). The org.apache.spark.sql.avro import at :32 goes with it, and the doc sentence at :179-181 loses the Avro clause together with its "(SPARK-59108, which removes that at the root, so this case goes with it)".
• AvroSuite.scala:3873 — the AvroV1Suite test "SPARK-59107: positionalFieldMatching makes an avro read projection-sensitive" goes. Its scanColumns === Seq(Seq("a"), Seq("b")) fails once the arm is gone. That is the one piece of this CI catches for you; everything above it is silent.
• AvroSuite.scala:4128-4130 — the !positional.capabilities().contains(SCAN_MERGING) assertion in the SPARK-57205 test, plus the first two sentences of its comment. Keep the withSQLConf block and the hasStrictFileReads sentence.

FileTable's class doc names the same shape ("neither does one that resolves a column by its position in the projection") as a general statement of the contract, so that one can stay.

Both prerequisites are also on branch-4.x, so a backport of this fix has to carry the same removal there.

[cloud-fan]

Review summary

The projection-independent reader fix is sound for unique names, but supported duplicate schema names can collapse to one ordinal and return wrong values. The temporary V1/V2 scan-merging gates, their tests, and the performance guide must also be updated now that both prerequisites have landed; that issue is already covered by an active current-head thread. Two changed comments need minor grammar fixes.

Findings

4 total: 0 P0, 0 P1, 2 P2, 2 P3.

Non-blocking (P2)

• Preserve ordinals when the data schema has duplicate names — sql/core/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala:478 — see inline.
• Remove obsolete projection-sensitivity gates and documentation — connector/avro/src/main/scala/org/apache/spark/sql/v2/avro/AvroPartitionReaderFactory.scala:110 — remaining in an existing discussion.

Nit (P3)

• Fix the malformed Scaladoc sentence — sql/core/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala:51 — see inline.
• Delimit the two pushdown conditions — connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala:1762 — see inline.

Already raised

• Remaining: Remove obsolete projection-sensitivity gates and documentation — P2 at connector/avro/src/main/scala/org/apache/spark/sql/v2/avro/AvroPartitionReaderFactory.scala:110 — existing discussion

PR metadata suggestions

• Update the prerequisite paragraph to state that #58340 and #58411 have landed and that this PR removes their V1/V2 gates, obsolete tests, and stale performance-guide claims; the pinned body still calls them open and describes a future landing order.
• After preserving duplicate-name ordinals, add the supported duplicate-name positional-read case and its V1/V2 regression coverage to the change and testing descriptions so the claimed full-data-schema mapping is accurate.

[cloud-fan]

Non-blocking (P2): [P2] Preserve ordinals when the data schema has duplicate names

schema.fieldIndex(field.name) cannot distinguish duplicate field-name occurrences. Spark permits duplicate file-source schema names, so with positionalFieldMatching=true a two-field schema such as (x, x) maps both fields to one Avro ordinal and can silently duplicate a value, or fail when the physical types differ. Please carry each required field's source ordinal from the scan-pruning boundary into the deserializer and add V1 and V2 regression coverage with duplicate names and distinct values.

Recommended change: Preserve occurrence-specific source ordinals at the scan-pruning boundary and pass them into Avro deserialization instead of reconstructing them from StructField names.

Why this works: Capture the ordinal mapping while the pruned attributes still distinguish duplicate occurrences, thread it through the V1 and V2 reader construction paths, and let AvroSchemaHelper consume those ordinals directly.

Scope: Avro V1/V2 scan-pruning and reader/deserializer plumbing, plus focused duplicate-name regression tests in the Avro module.

Compatibility: Unique-name schemas and name-based matching retain their current behavior; supported duplicate-name positional reads are corrected to return distinct physical fields.

Risks: Ordinal plumbing could diverge between the V1 and V2 construction paths. Capturing positions after occurrence identity has already been reduced to StructField names would preserve the bug.

Constraints: Keep nested-record matching on explicit local positions. Do not change default name-based matching or broaden the separate recursiveFieldMaxDepth limitation.

Success: V1 and V2 positional reads return distinct physical values for duplicate-name user-schema fields, with regression coverage proving the mapping and no change to existing unique-name behavior.

[cloud-fan]

Nit (P3): [P3] Fix the malformed Scaladoc sentence

the position that means is the one is not grammatical and obscures which ordinal is being described. Please change it to something like that position is the one in the full schema rather than in the projection.

[cloud-fan]

Nit (P3): [P3] Delimit the two pushdown conditions

As written, pushdown on the filter initially parses as a single phrase. Please write: With pushdown on, the filter runs inside the deserializer; with it off, the filter runs above the scan.

[LuciferYang]

Thanks, both of you. 2ec57b9c2d1 retires the two gates and takes the two comment fixes; one finding I think is refuted, with the measurement below.

Retiring the gates (@peter-toth's 6, and cloud-fan's P2 on the same thread). Done in the order you gave. AvroTable.supportsScanMerging is an unconditional true rather than a deleted override, since FileTable defaults it to false; hasProjectionSensitiveParser loses its avro arm and with it the options parameter and the org.apache.spark.sql.avro import; the AvroV1Suite case from #58411 and the capability assertion in the AvroV2SuiteSPARK-57205 case are gone. Two things the compiler caught that the list did not: MergeSubplans and FileSourceScanExec became unused imports in AvroSuite with that test, and AvroOptions became one in AvroTable, where the only mention left is prose. The documentation went with them: docs/sql-performance-tuning.md no longer lists avro among the projection-sensitive V1 relations, and the V2 paragraph no longer ends with Avro withholding the capability. FileTable's class doc keeps the shape as a general statement of the contract, as you said.

Duplicate names (cloud-fan's P2 at AvroDeserializer.scala:478) I believe is refuted. The premise is that Spark permits a duplicate-name file-source schema, so I tried to build one on both read paths before writing the ordinal plumbing:

V1, caseSensitive=false: AnalysisException [COLUMN_ALREADY_EXISTS] The column `x` already exists
V2, caseSensitive=false: same
V1, caseSensitive=true: same
V2, caseSensitive=true: same

spark.read.schema("x long, x long") is rejected before the read on all four combinations, so a duplicate-name data schema never reaches the deserializer and fieldIndex has no two occurrences to collapse. The other routes to one are closed too: the Avro spec forbids duplicate field names in a record, so an inferred schema cannot have them; the avroSchema option supplies the Avro side rather than the Catalyst data schema; and x beside X is rejected by the same check when analysis is case-insensitive, while under case-sensitive analysis they are two distinct names that fieldIndex separates, which the mixed-case test already covers. If you would rather have the state ruled out in code as well, I can add a uniqueness assertion where the positions are computed, but I would rather not add a check for a state that cannot arise.

Nits. Both taken: that position is the one in the full schema rather than in the projection, and With pushdown on, the filter runs inside the deserializer; with it off, it runs above the scan.

The description now says what this commit removes rather than what it will have to, and names the two commits the prerequisites landed as. It does not claim duplicate-name coverage, for the reason above. A backport carries the same removal wherever both prerequisites are, which today is master and branch-4.x.

[cloud-fan]

Review summary

One existing P2 remains after narrowing it to the supported V1 schema(StructType) path, and one new P2 asks for option-specific positive plan-and-result coverage after the V1 and V2 scan-merging gates were removed.

Findings

2 total: 0 P0, 0 P1, 2 P2, 0 P3.

Non-blocking (P2)

• Preserve duplicate field occurrences on the V1 StructType path — sql/core/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala:478
  The existing duplicate-name discussion remains valid on the narrower V1 schema(StructType) path: both occurrences are resolved with fieldIndex, so (x, x) maps to [1, 1] and returns the second physical value twice. Preserve occurrence-specific ordinals and add a focused V1 regression with distinct values; V2 rejects duplicate names before Avro and needs no duplicate-schema test.
• Keep positive positional scan-merging coverage for both read paths — General
  Please add positive positionalFieldMatching scan-merging regressions in AvroV1Suite and AvroV2Suite. Each should run two scalar aggregates over different columns, assert the literal result, and assert one widened scan (FileSourceScanExec for V1 and DataSourceV2ScanRelation for V2); the V2 case should also assert that the positional table advertises SCAN_MERGING. This keeps both the option-specific eligibility change and widened-read correctness observable.

Re-review status

4 addressed, 0 remaining, 2 new to this AI review.

New attribution: 0 newly introduced, 1 late catch, 0 previously raised, 1 unattributed.

Remaining findings

No prior AI findings remain.

[peter-toth]

Re-checked through 9a7cde1ea1e — findings 6 and 7 resolved, nothing regressed. Both gates are gone, AvroTable.supportsScanMerging is an unconditional true rather than a deleted override, both docs/sql-performance-tuning.md sentences went with them, and the description says what this commit removes. The 13 SPARK-59108 cases pass at this head.

On the duplicate-name thread at r3913900478: your refutation holds and I would not add the assertion. DataFrameReader.schema(String) is schema(StructType.fromDDL(schemaString)) (sql/api/src/main/scala/org/apache/spark/sql/DataFrameReader.scala:77), so the four measurements you ran already are the schema(StructType) path. The rejection is on the data schema itself in both readers, DataSource.scala:462 for V1 and FileTable.scala:101 for V2, so nothing with two occurrences of a name can reach positionsInDataSchema.

Non-blocking

• 8.The two new merge tests pass with the position mapping removed (new): Both write a two-column file and read both columns, so the merged projection is the whole data schema and the mapping is the identity. Measured: with positionsInDataSchema disabled both still pass. A third column, with the two subqueries over the last two, makes the union a proper subset and pins both halves. inline: connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala:3886

Minor

• 9.The V1 merge test dropped the strictness pin its V2 twin keeps (new):isProjectionSensitiveRead is also true when the read is not strict, so this test rests on ignoreCorruptFiles and ignoreMissingFiles being off exactly as the V2 one does, and the case it replaces pinned both. inline: connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala:3878
• 10.AvroTable's comment now answers only half the SCAN_MERGING contract (new):FileTable names two disqualifiers and the surviving paragraph answers one of them. The positional half is the one three PRs went into making true, and the removed paragraph was the only place that said so. inline: connector/avro/src/main/scala/org/apache/spark/sql/v2/avro/AvroTable.scala:59

[peter-toth]

Finding 8. This test and its AvroV2Suite twin pass with the position mapping removed, so between them they pin the gate removal but not the property that makes it safe.

The file has two columns and the query reads both, so the merged projection is the whole data schema. positionsInDataSchema then returns [0, 1], which is exactly what avroPosition computes when the array is empty, so this query runs the same code either way. Measured in a worktree at this head with the guard in positionsInDataSchema forced false: both cases still pass, 2/2.

A third column fixes it, because the union of the two subqueries is then a proper subset of the data schema:

 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")
valdf= sql("SELECT (SELECT sum(b) FROM t), (SELECT sum(c) FROM t)")
checkAnswer(df, Row(100L, 1000L))

with assert(scanColumns === Seq(Seq("b", "c"))) below, and the same edit on the V2 side. b and c sit at data schema positions 1 and 2, so the merged read has to resolve against the data schema to answer [100, 1000]. I ran that version both ways: with the mapping off both paths answer [10, 100] and fail, with it on both pass. The scan assertion still fails on base for the same reason the current one does, since the gate blocks merging whatever the column count.

[peter-toth]

Finding 9.isProjectionSensitiveRead has two arms, !hasStrictFileReads and hasProjectionSensitiveParser, so this test needs the reads to be strict for the same reason the AvroV2Suite twin does. That one pins it and says why; this one inherits it, and the AvroV1Suite case it replaces pinned both.

 withSQLConf(
SQLConf.IGNORE_CORRUPT_FILES.key ->"false",
SQLConf.IGNORE_MISSING_FILES.key ->"false",
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key ->"false") {

It would fail loudly rather than pass quietly if a default moved, so this is only about not depending on a default the twin already declines to depend on.

[peter-toth]

Finding 10.FileTable's class doc names two things that disqualify a format: a parser that decides what counts as a malformed record from the columns it was asked for, and one that "resolves a column by its position in the projection". The comment above this line answers the first and no longer mentions the second.

The second is the half that took this PR plus #58340 and #58411 to settle, and the paragraph you removed was the only place that recorded it. A reader who then finds positionalFieldMatching has nothing here to say it was considered. One sentence, something like:

// `positionalFieldMatching` resolves a column against its position in the data schema rather// than in the projection (SPARK-59108), so widening the projection does not move a column's// Avro field either.

OrcTable leaves orc.force.positional.evolution unsaid, so there is no convention to follow here. Avro's is the case that was the exception until this commit, which is why it is worth the line.

[LuciferYang]

Thanks @peter-toth, and thanks for the DataFrameReader.schema(String) pointer, that settles the duplicate-name thread better than my measurements did: schema(String) is schema(StructType.fromDDL(...)), so the four runs I posted were already the schema(StructType) path, and the narrowing had nothing left to narrow. No assertion added.

All three are in at 01a676f2575.

8. You are right, and this was the gap I talked myself out of. Both cases read every column of a two-column file, so the merged projection was the whole data schema and positionsInDataSchema returned the identity, which is what the empty array already means. Three columns now, with the two subqueries over b and c, so the union is a proper subset and the answer [100, 1000] is only reachable through the mapping. Re-measured both ways at this head: with the mapping disabled both fail with [10, 100], with the gates put back instead the V1 one fails with a scan per column and the V2 one at the capability assertion. So the pair pins both halves now rather than one.

9. Pinned, same three configurations as the twin. isProjectionSensitiveRead is true for a non-strict read whatever the parser does, so the test rested on two defaults its twin already declines to rest on.

10. Added, close to your wording: the comment now says positionalFieldMatching resolves a column against its position in the data schema rather than in the projection, so widening the projection does not move a column's Avro field either. That is the half FileTable names second, and with the paragraph I removed it was unrecorded.

The description carries the new shape and both mutation results.

[peter-toth]

Re-checked through 01a676f2575 — findings 8, 9, 10 resolved, nothing new.

The two merge tests are the three-column shape I measured at the previous head, and it discriminates in both directions there: [100, 1000] with the mapping on, [10, 100] with it off. The 13 SPARK-59108 cases pass at this head.

Thanks for working through all of these, @LuciferYang — nothing left open from my side.

[uros-b]

Thank you @LuciferYang and @peter-toth!

[cloud-fan]

Review summary

The implementation, tests, comments, and performance documentation are consistent with the projection-invariant positional-read contract. Earlier review requests are resolved, and the remaining scanner candidates are either unreachable or immaterial.

Findings

0 total: 0 P0, 0 P1, 0 P2, 0 P3.

No findings.

Re-review status

Prior AI findings: 2 addressed, 0 still present; other current findings: 0.

New attribution: 0 newly introduced, 0 late catch, 0 previously raised, 0 unattributed.

Remaining findings

No prior AI findings remain.

Verification

• V1 DataSource and V2 FileTable reject duplicate names within the data schema before constructing the Avro reader, so occurrence-collapsing duplicate schemas do not reach the changed mapping.
• The V1 and V2 scan-merging tests use a proper-subset union over non-leading columns, so their expected values require mapping through the unpruned data schema as well as removal of the eligibility gates.

[LuciferYang]

Merge Summary:

• merged into master c809c28
• merged into branch-4.x c2f8c58

Posted by merge_spark_pr.py

[LuciferYang]

Thanks @cloud-fan@peter-toth@uros-b

As #58340 is missing from branch‑4.3, I’ll create a manual PR to merge it into branch‑4.3/branch‑4.2/branch‑4.1

[cloud-fan]

Review summary

The projection-invariant positional mapping, gate removal, documentation, and unique-name coverage are coherent, but the name-based mapping regresses V1's accepted duplicate-schema path. Preserve occurrence-specific ordinals at the V1 scan-to-deserializer boundary and add focused distinct-value coverage; V2 already rejects duplicate names. No tests were run as part of this source review.

Findings

1 total: 0 P0, 0 P1, 1 P2, 0 P3.

Non-blocking (P2)

• Preserve duplicate field occurrences on the V1 explicit-schema path — sql/core/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala:478 — see inline.

Re-review status

Prior AI findings: 0 addressed, 0 still present; additional unresolved findings in this review: 1.

New attribution: 0 newly introduced, 1 late catch, 0 previously raised, 0 unattributed.

Remaining prior AI findings

No prior AI findings remain.

Existing discussions

• existing discussion — The gate-removal work is present, but the response's duplicate-schema refutation is contradicted by the pinned V1 DataSource path, which catches duplicate-column validation and proceeds before the changed fieldIndex mapping.
• existing discussion — The merge-test changes are present, but treating schema(String) as proof that V1 rejects duplicate StructTypes overlooks DataSource's warning-and-continue compatibility branch.
• existing discussion — The broad V1/V2 request narrows on the pinned tree: V2 rejects duplicate schema names, but V1 explicit schemas still reach schema.fieldIndex and collapse duplicate occurrences.

PR metadata suggestions

• After fixing the retained V1 duplicate-occurrence case, update the change, user-facing behavior, and testing sections to state that accepted duplicate-name V1 schemas preserve occurrence-specific ordinals and are covered by a distinct-value V1 regression; V2 rejects duplicate data-column names.

[cloud-fan]

Non-blocking (P2): V1 file reads intentionally catch duplicate-column validation and continue with a warning, so an explicit schema such as (x, x) reaches this lookup. StructType.fieldIndex retains one index per name, which makes both occurrences resolve to the same Avro ordinal: same-typed fields silently return the second value twice, and differing types fail converter construction. Please preserve each required field's source ordinal while the V1 scan boundary can still distinguish occurrences, carry that mapping into top-level deserialization, keep nested records on local ordinals, and add a V1 regression with duplicate names and distinct physical values. V2 rejects duplicate data-column names and needs no duplicate-schema change.

Recommended change: Preserve occurrence-specific source ordinals before V1 required fields are reduced to StructField names, and pass that immutable mapping through the normal and archive V1 reader paths into top-level Avro deserialization.

Why this works: Capture each required attribute's ordinal in the full V1 data schema while duplicate occurrences are still distinguishable; have AvroSchemaHelper consume those ordinals directly at the root, while nested records continue to use their explicit local positions.

Scope: Avro V1 scan-pruning and normal/archive reader-to-deserializer plumbing, plus one focused V1 duplicate-schema regression; no V2 duplicate-schema work.

Compatibility: Unique-name positional reads, default name-based matching, V2 reads, and nested-record local matching retain their current behavior; supported V1 duplicate-name positional reads regain occurrence-correct values.

Risks: Capturing the mapping after attributes have already been reduced to names would preserve the bug. The normal and archive V1 paths could diverge, or root ordinal plumbing could accidentally affect nested-record local matching.

Constraints: Keep nested-record matching on explicit local positions. Do not change default name-based matching or broaden the recursiveFieldMaxDepth limitation. Do not add V2 duplicate-schema plumbing or coverage because V2 rejects duplicate data-column names.

Success: A V1 positional read with two same-named explicit-schema fields and distinct physical values returns the two distinct values in order on both applicable V1 reader paths, while existing unique-name and nested-record tests remain unchanged.

[LuciferYang]

Thanks for staying on this, @cloud-fan, and you are right about the branch: getOrInferFileFormatSchema does catch the duplicate-column check and continue with a warning (DataSource.scala:239-245, SPARK-18108 and SPARK-21144). I had not found it, and my earlier replies answered the symptom rather than the mechanism, which is probably why the question kept coming back. Let me lay out what I read this time so you can tell me where it is wrong.

As I read it there are two checks on the V1 path, looking at different things. The warn-and-continue one checks (dataSchema ++ partitionSchema).map(_.name), so what it tolerates is a name shared between the data schema and the partition schema. The throwing one is checkSchemaColumnNameDuplication(hs.dataSchema, equality) in resolveRelation, and that is what a duplicate inside the data schema hits. It sits after the relation-building match and applies to its result, so every HadoopFsRelation branch goes through it, the streaming-sink metadata one and the CatalogFileIndex one included, and equality is conf.resolver.

The tolerated case seems unable to leave two occurrences behind, because of DataSource.scala:222-223: a user-specified schema becomes the data schema minus every field whose name matches a partition column, so the field that made the concatenation ambiguous is the one that gets removed, and if the duplicated name is the partition column's own then both copies go. Printing the resolved relation's two schemas:

user schema (a, p), partition column p: loaded, data=a partition=p, rows=[2,2],[0,0],[1,1]
user schema (p, p), partition column p: loaded, data=<empty> partition=p, rows=[2],[0],[1]
user schema (x, x), no partition column: AnalysisException [COLUMN_ALREADY_EXISTS] The column `x` already exists

Two places did look like they could bite, and I think your instinct about the weak check is right about one of them. Streaming is where a duplicate really does travel on it: readStream.schema("x int, x int") survives stream setup, since sourceSchema goes through getOrInferFileFormatSchema. FileStreamSource.getBatch then builds each micro-batch with resolveRelation(checkFilesExist = false, readOnly = true), so the query fails at first-batch planning rather than at analysis, which is worth its own JIRA even though no relation reaches buildReader. The other is HiveMetastoreCatalog.convertToLogicalRelation, the one construction site that skips the check outright, but it is only ever called with ParquetFileFormat or OrcFileFormat, and a Hive avro table stays a HiveTableRelation and reads through the Hive SerDe.

So as far as I can tell dataSchema.fieldIndex is never asked to choose between two occurrences. You know this code much better than I do, though, so if there is a path I have walked past, please say which one and I will open a follow-up JIRA for the ordinal plumbing, since this PR is already merged and cannot carry the change anyway. A query I can run would help me see it fastest.

[cloud-fan]

Thanks for the detailed trace. Your reading is correct: the warning path only tolerates data/partition overlap, while duplicates within dataSchema are rejected before reader construction. I don't see a supported path where dataSchema.fieldIndex must distinguish duplicate occurrences, so my earlier finding was incorrect. The streaming late-failure case is separate from this PR.