peter-toth commented on code in PR #58411:
URL: https://github.com/apache/spark/pull/58411#discussion_r3889751465
##########
connector/avro/src/test/scala/org/apache/spark/sql/avro/AvroSuite.scala:
##########
@@ -3732,6 +3732,23 @@ class AvroV1Suite extends AvroSuite {
.sparkConf
.set(SQLConf.USE_V1_SOURCE_LIST, "avro")
+ test("SPARK-59107: positionalFieldMatching makes an avro read
projection-sensitive") {
+ 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)
+ .createOrReplaceTempView("t")
+ // Positional matching pairs a column with the Avro field at its
position in the projected
+ // schema, so a scan of b alone reads the first Avro field and answers
10, while a scan
+ // shared with sum(a) reads both fields, pairs them by coincidence and
answers 100. The 10
+ // is wrong against the file, which is SPARK-59108; what merging must
not do is make one
+ // subquery's value depend on what the other one projects.
+ checkAnswer(sql("SELECT (SELECT sum(a) FROM t), (SELECT sum(b) FROM
t)"), Row(10L, 10L))
Review Comment:
**Finding 10.** Your description says #58409 "removes that at the root, and
this case can go with it". It takes this test with it too, and not by making it
redundant - by failing it.
I ran this exact query in a worktree at #58409's head `a3358708e60`:
```
merged = [10, 100]
unmerged = [10, 100]
```
`sum(b)` on its own is 100 there, because the fix resolves `b` against its
position in the data schema rather than in the projection. So `Row(10L, 10L)`
fails, and the avro arm of `hasProjectionSensitiveParser` is dead at the same
moment - the read stops being projection-sensitive, which is exactly what those
two numbers being equal shows.
Whichever of the two lands second pays for it, so it is worth deciding now
and saying so in the description. Two ways out:
- Land #58409 first and drop the avro arm plus this test from here. That
also drops the only reason `DataSourceUtils` imports
`org.apache.spark.sql.avro`.
- Keep them, land this first, and have #58409 delete both.
If you would rather keep the test through the transition, the assertion that
survives is the one the `sql/core` suite already uses - the columns each scan
reads, which stay `Seq(Seq("a"), Seq("b"))` under the gate whatever `sum(b)`
evaluates to. `AvroV1Suite` would need `spark.sql.adaptive.enabled` off for
that, for the reason `FileSourceV1PlanMergingSuite:369` gives.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala:
##########
@@ -52,8 +53,19 @@ case class MergeResult(
* @param merged Whether this plan is the result of merging two or more plans
(true), or
* is an original unmerged plan (false). Merged plans typically
require special
* handling such as wrapping in CTEs.
+ * @param projectionSensitiveReads The columns the plans behind this entry
read from each
+ * projection-sensitive relation, recorded
when the entry was
+ * first cached. Every plan merged into the
entry read those same
+ * columns, which is what merging one in
requires, so the record
+ * stays true of the entry. It cannot be
re-derived from `plan`,
+ * because a merge leaves projections that the
`ColumnPruning`
+ * rerun after this rule narrows again. See
+ *
`PlanMerger.collectProjectionSensitiveReads`.
*/
-case class MergedPlan(plan: LogicalPlan, merged: Boolean)
+case class MergedPlan(
+ plan: LogicalPlan,
+ merged: Boolean,
+ projectionSensitiveReads: Map[LogicalPlan, Seq[Set[String]]] = Map.empty)
Review Comment:
**Finding 11.** Nothing uses the default. `MergedPlan(plan, false,
projectionSensitiveReads)` at line 210 is the only construction, and the two
other sites are `mp.copy(...)`, which carries the record over.
The value it defaults to is also the permissive one: an entry built without
a record compares equal to any plan that reads no projection-sensitive
relation, so the merge goes through. Dropping the default makes a future
construction site say what it means.
```scala
case class MergedPlan(
plan: LogicalPlan,
merged: Boolean,
projectionSensitiveReads: Map[LogicalPlan, Seq[Set[String]]])
```
--
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]
