peter-toth commented on code in PR #58409:
URL: https://github.com/apache/spark/pull/58409#discussion_r3889700370
##########
connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala:
##########
@@ -1735,6 +1735,141 @@ 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))))
Review Comment:
**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.
```suggestion
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".
##########
sql/core/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala:
##########
@@ -452,15 +462,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, since the two
Review Comment:
**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:
```scala
* 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.
```
##########
sql/core/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala:
##########
@@ -452,15 +462,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, since the two
+ * schemas have different lengths.
+ */
+ private def positionsInDataSchema(projection: StructType): Array[Int] =
dataSchema match {
Review Comment:
**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.
##########
sql/core/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala:
##########
@@ -452,15 +462,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, since the two
+ * schemas have different lengths.
+ */
+ 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] = Array.empty)
Review Comment:
**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:
```scala
applyFilters: Int => Boolean,
dataSchemaPositions: Array[Int])
```
and at line 323:
```scala
val writeRecord = 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.
##########
connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala:
##########
@@ -1735,6 +1735,141 @@ 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.selectExpr("sum(z)"), Row(100000L))
+ // With pushdown on the filter runs inside the deserializer, with it off
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
Review Comment:
**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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
