You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ya...@apache.org on 2026/09/09 15:59:09 UTC
(spark) branch branch-4.2 updated: [SPARK-59108][4.2][SQL] Fix Avro positional matching under column pruning
This is an automated email from the ASF dual-hosted git repository.
LuciferYang pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new 84c89bfb89bf [SPARK-59108][4.2][SQL] Fix Avro positional matching under column pruning
84c89bfb89bf is described below
commit 84c89bfb89bf0ec0599f362f03e9002ed2ef032a
Author: YangJie <ya...@baidu.com>
AuthorDate: Wed Sep 9 23:58:53 2026 +0800
[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 [...]
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 [...]
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 [...]
### 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 [...]
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` twi [...]
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 <ya...@baidu.com>
Signed-off-by: yangjie01 <ya...@baidu.com>
---
.../sql/v2/avro/AvroPartitionReaderFactory.scala | 3 +-
.../sql/avro/AvroCatalystDataConversionSuite.scala | 3 +-
.../apache/spark/sql/avro/AvroRowReaderSuite.scala | 3 +-
.../spark/sql/avro/AvroSchemaHelperSuite.scala | 43 ++++++
.../org/apache/spark/sql/avro/AvroSerdeSuite.scala | 3 +-
.../org/apache/spark/sql/avro/AvroSuite.scala | 166 ++++++++++++++++++++-
.../apache/spark/sql/avro/AvroDeserializer.scala | 53 ++++++-
.../org/apache/spark/sql/avro/AvroFileFormat.scala | 3 +-
.../org/apache/spark/sql/avro/AvroUtils.scala | 24 ++-
9 files changed, 284 insertions(+), 17 deletions(-)
diff --git a/connector/avro/src/main/scala/org/apache/spark/sql/v2/avro/AvroPartitionReaderFactory.scala b/connector/avro/src/main/scala/org/apache/spark/sql/v2/avro/AvroPartitionReaderFactory.scala
index a13faf3b5156..faa519b18941 100644
--- a/connector/avro/src/main/scala/org/apache/spark/sql/v2/avro/AvroPartitionReaderFactory.scala
+++ b/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
diff --git a/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroCatalystDataConversionSuite.scala b/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroCatalystDataConversionSuite.scala
index a680634fab30..e096300e89c2 100644
--- a/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroCatalystDataConversionSuite.scala
+++ b/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)
diff --git a/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroRowReaderSuite.scala b/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroRowReaderSuite.scala
index d5b246840902..2cace154420d 100644
--- a/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroRowReaderSuite.scala
+++ b/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
diff --git a/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSchemaHelperSuite.scala b/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSchemaHelperSuite.scala
index 936458561978..72e6deb1e99b 100644
--- a/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSchemaHelperSuite.scala
+++ b/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))
diff --git a/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSerdeSuite.scala b/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSerdeSuite.scala
index 3643a95abe19..5aa6e4d703f5 100644
--- a/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSerdeSuite.scala
+++ b/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)
}
/**
diff --git a/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala b/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala
index 376d12a8a492..76a8ce62cae8 100644
--- a/connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala
+++ b/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")
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala b/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala
index ba574c091dae..003333c81ad4 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala
+++ b/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
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroFileFormat.scala b/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroFileFormat.scala
index 90781d4ad707..a93757018b63 100755
--- a/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroFileFormat.scala
+++ b/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
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala b/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala
index 02220e6c8568..1331649844c1 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala
@@ -286,17 +286,28 @@ private[sql] object AvroUtils extends Logging {
* @param positionalFieldMatch If true, perform field matching in a positional fashion
* (structural comparison between schemas, ignoring names);
* otherwise, perform field matching using field names.
+ * @param dataSchemaPositions The position of each `catalystSchema` field in the schema it was
+ * projected from, for a positional match against a projection. A
+ * positional match pairs a Catalyst field with the Avro field at the
+ * same position, and that position is the one in the full schema, so
+ * a read of only the third column still takes the third Avro field.
+ * Empty when `catalystSchema` is not a projection, in which case a
+ * field's own position is used.
*/
class AvroSchemaHelper(
avroSchema: Schema,
catalystSchema: StructType,
avroPath: Seq[String],
catalystPath: Seq[String],
- positionalFieldMatch: Boolean) {
+ positionalFieldMatch: Boolean,
+ dataSchemaPositions: Array[Int] = Array.empty) {
if (avroSchema.getType != Schema.Type.RECORD) {
throw new IncompatibleSchemaException(
s"Attempting to treat ${avroSchema.getName} as a RECORD, but it was: ${avroSchema.getType}")
}
+ require(dataSchemaPositions.isEmpty || dataSchemaPositions.length == catalystSchema.length,
+ s"Got ${dataSchemaPositions.length} data schema positions for " +
+ s"${catalystSchema.length} Catalyst fields")
private[this] val avroFieldArray = avroSchema.getFields.asScala.toArray
private[this] val fieldMap = avroSchema.getFields.asScala
@@ -320,8 +331,9 @@ private[sql] object AvroUtils extends Logging {
if (getAvroField(sqlField.name, sqlPos).isEmpty &&
(!ignoreNullable || !sqlField.nullable)) {
if (positionalFieldMatch) {
- throw new IncompatibleSchemaException("Cannot find field at position " +
- s"$sqlPos of ${toFieldStr(avroPath)} from Avro schema (using positional matching)")
+ throw new IncompatibleSchemaException(
+ s"Cannot find field at position ${avroPosition(sqlPos)} of " +
+ s"${toFieldStr(avroPath)} from Avro schema (using positional matching)")
} else {
throw new IncompatibleSchemaException(
s"Cannot find ${toFieldStr(catalystPath :+ sqlField.name)} in Avro schema")
@@ -375,11 +387,15 @@ private[sql] object AvroUtils extends Logging {
/** Get the Avro field corresponding to the provided Catalyst field name/position, if any. */
def getAvroField(fieldName: String, catalystPos: Int): Option[Schema.Field] = {
if (positionalFieldMatch) {
- avroFieldArray.lift(catalystPos)
+ avroFieldArray.lift(avroPosition(catalystPos))
} else {
getFieldByName(fieldName)
}
}
+
+ /** The Avro field position a positional match pairs the given Catalyst position with. */
+ private def avroPosition(catalystPos: Int): Int =
+ if (dataSchemaPositions.isEmpty) catalystPos else dataSchemaPositions(catalystPos)
}
/**
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org