cloud-fan commented on code in PR #58409:
URL: https://github.com/apache/spark/pull/58409#discussion_r3913900478
##########
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, 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
Review Comment:
**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.
##########
sql/core/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala:
##########
@@ -45,6 +45,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 the position that means is the
one in the full
Review Comment:
**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`.
##########
connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala:
##########
@@ -1737,6 +1737,142 @@ 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
above the scan.
Review Comment:
**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.`
--
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]
