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

[LuciferYang]

What changes were proposed in this pull request?

Backport of c809c283c2d (#58409) to branch-4.3. The Avro fix and the V1 half of the gate removal come over as they landed; the V2 half is dropped, for the reason in the fourth paragraph.

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.

One gate kept avro out of V1 scan merging because merging widens the projection, and this commit retires it: #58411 (SPARK-59107, on this branch as 04cdb66c83d) named avro in DataSourceUtils.isProjectionSensitiveRead. hasProjectionSensitiveParser loses its avro arm and, with it, its options parameter and the org.apache.spark.sql.avro import; the AvroV1Suite case that PR added goes too, since its assertion that each subquery keeps its own scan stops being true. docs/sql-performance-tuning.md no longer lists avro among the projection-sensitive V1 relations, which it has to stop doing whether or not the predicate comes off, because after this fix the position is the data schema's.

The V2 half is not here. SPARK-57205 (#58340) is not on branch-4.3, so AvroTable has no supportsScanMerging to turn on, the AvroV2Suite capability case it added does not exist, the performance guide has no V2 paragraph, and the V2 twin of the new merge test cannot hold on a branch where avro never declares SCAN_MERGING. Those four hunks are the whole difference from the master commit.

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, and 4.3.0 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, and one in AvroArchiveReadBase, which runs in the tar, zip and 7z suites, because the archive reader builds its own deserializer per entry.

One more test, in AvroV1Suite, for the shape the removed gate 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 FileSourceScanExec reading both columns. It pins the strictness flags, since a non-strict read is projection-sensitive for the other reason. Master's AvroV2Suite twin is not here, for the reason above.

Mutation check, measured on this branch: with the position mapping disabled, 14 cases fail, the five SPARK-59108 shapes on each read path, the archive one in each of the tar, zip and 7z suites, and the new merge test, which answers [10, 100] where the file has [100, 1000]. 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, measured on this branch: the whole avro module, 499 tests, the planmerging package, 108 tests, since removing the avro arm touches the shared predicate, 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

[uros-b]

Thank you @LuciferYang, LGTM!

[LuciferYang]

@peter-toth, two questions about how much further either fix should travel. I would like your read on the scope before I open anything.

This one, below 4.3. The bug is old: positional matching has resolved a column against the projection since 3.2.0 (SPARK-34365), so every 4.x line returns wrong values for a pruned positional read. Going below 4.3 is smaller than this PR, not larger: there is no gate to retire, docs/sql-performance-tuning.md has no "Merging Subplans" section before 4.3, and AvroFileFormat.readArchive does not exist either, so the deserializer change lands with two call sites instead of three. AvroDeserializer's positional path and AvroUtils.AvroSchemaHelper are the same shape on 4.0 through 4.2. If that sounds right I will open them.

SPARK-59107, below 4.3. That one reproduces there too. Measured on branch-4.2 with two csv shapes: DROPMALFORMED answers [8, 80] against [10, 80] with MergeSubplans excluded, and PERMISSIVE with _corrupt_record answers [1, 80] against [0, 80]. branch-4.0 and branch-4.1 carry the older MergeScalarSubqueries and were not run.

The fix does not port as written, though. Below 4.3 the rule lives in sql/catalyst (optimizer/MergeSubplans.scala on 4.2, optimizer/MergeScalarSubqueries.scala on 4.0 and 4.1), while the predicate it needs is about HadoopFsRelation and the FileFormat subclasses, all of which are in sql/core; catalyst references none of them today. So it would need a new seam on the catalyst side, something like a trait catalyst defines and LogicalRelation implements by asking the relation. That is a structural change on a maintenance branch, which is why I would rather agree on it first than write it: worth doing, or better recorded as a known limitation on SPARK-59107 with the fix starting at 4.3?

[peter-toth]

Below 3.5 is EOL and 3.5 only takes security fixes, so branch-4.0, branch-4.1 and branch-4.2 are the ones in scope.

This one. Please open them. The fix stands on its own there: the wrong-value bug needs only a pruned projection, no scan merging, so SELECT sum(c) FROM t is enough to hit it. Those branches also have no gate keeping avro out of V1 scan merging, since that gate is what #58411 added on 4.3, so a merged read can widen an avro projection there with nothing in front of it. That makes the position mapping worth more on those branches than on 4.3, not less.

SPARK-59107. I would not port it. A new catalyst-side seam on three maintenance branches is more than this warrants, so let us record it as a known limitation on SPARK-59107 with the fix starting at 4.3. Put the 4.2 and 3.5 numbers you measured on the ticket. They are what anyone who hits it will need.

[LuciferYang]

@dongjoon-hyun , thank you for your reminder. I have revised that comment.
@peter-toth , my apologies for causing confusion.

[LuciferYang]

Merge Summary:

• merged into branch-4.3 e03066e

Posted by merge_spark_pr.py