peter-toth commented on code in PR #58409:
URL: https://github.com/apache/spark/pull/58409#discussion_r3922421191
##########
connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala:
##########
@@ -3734,37 +3870,24 @@ class AvroV1Suite extends AvroSuite {
.sparkConf
.set(SQLConf.USE_V1_SOURCE_LIST, "avro")
- test("SPARK-59107: positionalFieldMatching makes an avro read
projection-sensitive") {
- // Strictness pinned rather than inherited, so that positional matching is
the only reason the
- // read is projection-sensitive. AQE off because `AdaptiveSparkPlanExec`
is a leaf node, so with
- // it on the scans underneath it are not reachable from the executed plan.
- withSQLConf(
- SQLConf.IGNORE_CORRUPT_FILES.key -> "false",
- SQLConf.IGNORE_MISSING_FILES.key -> "false",
- SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ test("SPARK-59108: two positional reads of different columns share one
widened scan") {
+ // SPARK-59107 named avro under this option, so the two subqueries used to
keep their own scans.
+ // They share one now, and the values are the file's either way because
each column resolves
+ // against the data schema. 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").write.format("avro").save(path)
withTempView("t") {
- spark.read.option("positionalFieldMatching",
"true").format("avro").load(path)
+ spark.read.option("positionalFieldMatching",
true.toString).format("avro").load(path)
.createOrReplaceTempView("t")
- val query = "SELECT (SELECT sum(a) FROM t), (SELECT sum(b) FROM t)"
- // Compared against the same query with merging excluded rather than
against a literal
- // row: positional matching resolves a column against its position
in the read schema, so
- // what `sum(b)` answers depends on its own subquery's projection,
and SPARK-59108 changes
- // it. What merging must not change is either value.
- val unmerged = withSQLConf(
- SQLConf.OPTIMIZER_EXCLUDED_RULES.key -> MergeSubplans.ruleName) {
- sql(query).collect().toSeq
- }
- val df = sql(query)
- checkAnswer(df, unmerged)
+ val df = sql("SELECT (SELECT sum(a) FROM t), (SELECT sum(b) FROM t)")
+ checkAnswer(df, Row(10L, 100L))
Review Comment:
**Finding 8.** This test and its `AvroV2Suite` twin pass with the position
mapping removed, so between them they pin the gate removal but not the property
that makes it safe.
The file has two columns and the query reads both, so the merged projection
is the whole data schema. `positionsInDataSchema` then returns `[0, 1]`, which
is exactly what `avroPosition` computes when the array is empty, so this query
runs the same code either way. Measured in a worktree at this head with the
guard in `positionsInDataSchema` forced false: both cases still pass, 2/2.
A third column fixes it, because the union of the two subqueries is then a
proper subset of the data schema:
```scala
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")
val df = sql("SELECT (SELECT sum(b) FROM t), (SELECT sum(c) FROM
t)")
checkAnswer(df, Row(100L, 1000L))
```
with `assert(scanColumns === Seq(Seq("b", "c")))` below, and the same edit
on the V2 side. `b` and `c` sit at data schema positions 1 and 2, so the merged
read has to resolve against the data schema to answer `[100, 1000]`. I ran that
version both ways: with the mapping off both paths answer `[10, 100]` and fail,
with it on both pass. The scan assertion still fails on base for the same
reason the current one does, since the gate blocks merging whatever the column
count.
##########
connector/avro/src/main/scala/org/apache/spark/sql/v2/avro/AvroTable.scala:
##########
@@ -56,13 +56,5 @@ case class AvroTable(
// Avro has no record-level parse verdict: a record is either decodable or
the read fails, and
// there is no mode that drops or rewrites a record based on the columns
asked for. The `mode`
// option in AvroOptions is read by from_avro and schema_of_avro, not by
this scan.
- //
- // `positionalFieldMatching` is the exception. AvroPartitionReaderFactory
builds the deserializer
- // from the pruned read schema while the Avro side stays the full Avro
schema, so under that
- // option catalyst field i of the projection takes Avro field i of that
schema, and widening the
- // projection changes the values a column comes back with. Read the option
off the map rather than
- // through AvroOptions, whose constructor resolves `avroSchemaUrl` and would
do I/O here, and read
- // it leniently so a malformed value still fails where Avro reports it
rather than here.
- override protected def supportsScanMerging: Boolean =
-
!"true".equalsIgnoreCase(options.get(AvroOptions.POSITIONAL_FIELD_MATCHING))
+ override protected def supportsScanMerging: Boolean = true
Review Comment:
**Finding 10.** `FileTable`'s class doc names two things that disqualify a
format: a parser that decides what counts as a malformed record from the
columns it was asked for, and one that "resolves a column by its position in
the projection". The comment above this line answers the first and no longer
mentions the second.
The second is the half that took this PR plus #58340 and #58411 to settle,
and the paragraph you removed was the only place that recorded it. A reader who
then finds `positionalFieldMatching` has nothing here to say it was considered.
One sentence, something like:
```scala
// `positionalFieldMatching` resolves a column against its position in the
data schema rather
// than in the projection (SPARK-59108), so widening the projection does
not move a column's
// Avro field either.
```
`OrcTable` leaves `orc.force.positional.evolution` unsaid, so there is no
convention to follow here. Avro's is the case that was the exception until this
commit, which is why it is worth the line.
##########
connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala:
##########
@@ -3734,37 +3870,24 @@ class AvroV1Suite extends AvroSuite {
.sparkConf
.set(SQLConf.USE_V1_SOURCE_LIST, "avro")
- test("SPARK-59107: positionalFieldMatching makes an avro read
projection-sensitive") {
- // Strictness pinned rather than inherited, so that positional matching is
the only reason the
- // read is projection-sensitive. AQE off because `AdaptiveSparkPlanExec`
is a leaf node, so with
- // it on the scans underneath it are not reachable from the executed plan.
- withSQLConf(
- SQLConf.IGNORE_CORRUPT_FILES.key -> "false",
- SQLConf.IGNORE_MISSING_FILES.key -> "false",
- SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ test("SPARK-59108: two positional reads of different columns share one
widened scan") {
+ // SPARK-59107 named avro under this option, so the two subqueries used to
keep their own scans.
+ // They share one now, and the values are the file's either way because
each column resolves
+ // against the data schema. 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") {
Review Comment:
**Finding 9.** `isProjectionSensitiveRead` has two arms,
`!hasStrictFileReads` and `hasProjectionSensitiveParser`, so this test needs
the reads to be strict for the same reason the `AvroV2Suite` twin does. That
one pins it and says why; this one inherits it, and the `AvroV1Suite` case it
replaces pinned both.
```scala
withSQLConf(
SQLConf.IGNORE_CORRUPT_FILES.key -> "false",
SQLConf.IGNORE_MISSING_FILES.key -> "false",
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
```
It would fail loudly rather than pass quietly if a default moved, so this is
only about not depending on a default the twin already declines to depend on.
--
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]
