[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>
9 files changed
tree: 723deba99d6c2d34277370e94a2ed93e67f57d20
  1. .github/
  2. .mvn/
  3. assembly/
  4. bin/
  5. binder/
  6. build/
  7. common/
  8. conf/
  9. connector/
  10. core/
  11. data/
  12. dev/
  13. docs/
  14. examples/
  15. graphx/
  16. hadoop-cloud/
  17. launcher/
  18. licenses/
  19. licenses-binary/
  20. mllib/
  21. mllib-local/
  22. project/
  23. python/
  24. R/
  25. repl/
  26. resource-managers/
  27. sbin/
  28. sql/
  29. streaming/
  30. tools/
  31. udf/
  32. ui-test/
  33. .asf.yaml
  34. .gitattributes
  35. .gitignore
  36. .nojekyll
  37. .pre-commit-config.yaml
  38. .sbtopts
  39. AGENTS.md
  40. CONTRIBUTING.md
  41. LICENSE
  42. LICENSE-binary
  43. NOTICE
  44. NOTICE-binary
  45. pom.xml
  46. pyproject.toml
  47. README.md
  48. scalastyle-config.xml
  49. SECURITY.md
README.md

Apache Spark

Spark is a unified analytics engine for large-scale data processing. It provides high-level APIs in Scala, Java, Python, and R (Deprecated), and an optimized engine that supports general computation graphs for data analysis. It also supports a rich set of higher-level tools including Spark SQL for SQL and DataFrames, pandas API on Spark for pandas workloads, MLlib for machine learning, GraphX for graph processing, and Structured Streaming for stream processing.

License Maven Central Java GitHub Actions Build PySpark Coverage PyPI Downloads

Online Documentation

You can find the latest Spark documentation, including a programming guide, on the project web page. This README file only contains basic setup instructions.

Build Pipeline Status

BranchStatus
masterGitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
branch-4.1GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
branch-4.0GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
GitHub Actions Build
branch-3.5GitHub Actions Build
GitHub Actions Build

Building Spark

Spark is built using Apache Maven. To build Spark and its example programs, run:

./build/mvn -DskipTests clean package

(You do not need to do this if you downloaded a pre-built package.)

More detailed documentation is available from the project site, at “Building Spark”.

For general development tips, including info on developing Spark using an IDE, see “Useful Developer Tools”.

Interactive Scala Shell

The easiest way to start using Spark is through the Scala shell:

./bin/spark-shell

Try the following command, which should return 1,000,000,000:

scala> spark.range(1000 * 1000 * 1000).count()

Interactive Python Shell

Alternatively, if you prefer Python, you can use the Python shell:

./bin/pyspark

And run the following command, which should also return 1,000,000,000:

>>> spark.range(1000 * 1000 * 1000).count()

Example Programs

Spark also comes with several sample programs in the examples directory. To run one of them, use ./bin/run-example <class> [params]. For example:

./bin/run-example SparkPi

will run the Pi example locally.

You can set the MASTER environment variable when running examples to submit examples to a cluster. This can be spark:// URL, “yarn” to run on YARN, and “local” to run locally with one thread, or “local[N]” to run locally with N threads. You can also use an abbreviated class name if the class is in the examples package. For instance:

MASTER=spark://host:7077 ./bin/run-example SparkPi

Many of the example programs print usage help if no params are given.

Running Tests

Testing first requires building Spark. Once Spark is built, tests can be run using:

./dev/run-tests

Please see the guidance on how to run tests for a module, or individual tests.

There is also a Kubernetes integration test, see resource-managers/kubernetes/integration-tests/README.md

A Note About Hadoop Versions

Spark uses the Hadoop core library to talk to HDFS and other Hadoop-supported storage systems. Because the protocols have changed in different versions of Hadoop, you must build Spark against the same version that your cluster runs.

Please refer to the build documentation at “Specifying the Hadoop Version and Enabling YARN” for detailed guidance on building for a particular distribution of Hadoop, including building for particular Hive and Hive Thriftserver distributions.

Configuration

Please refer to the Configuration Guide in the online documentation for an overview on how to configure Spark.

Contributing

Please review the Contribution to Spark guide for information on how to get started contributing to the project.