peter-toth commented on code in PR #60197:
URL: https://github.com/apache/spark/pull/60197#discussion_r4501298586
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/ExistingRDD.scala:
##########
@@ -117,7 +116,9 @@ case class LogicalRDD(
session :: originStats :: originConstraints :: Boolean.box(fromCheckpoint)
:: Nil
override def newInstance(): LogicalRDD.this.type = {
- val rewrite = Utils.toMap(output, output.map(_.newInstance()))
+ // Keyed by exprId, because the attributes in the partitioning, the
ordering, the stats and
+ // the constraints can differ from `output` in qualifier or nullability.
Review Comment:
**Finding 3.** `newInstance` can only remap what refers to `output`. But
`fromDataset` (lines 210-211) copies `executedPlan.outputPartitioning` and
`outputOrdering` without mapping them onto `output` at all, so their exprIds
can differ from `output` too. `RemoveRedundantAliases` drops a same-name root
alias, so the executed plan outputs the source column. With AQE off when
checkpointing:
```scala
val orig = spark.range(16)
val cp = orig.repartition($"id").select($"id".as("id")).localCheckpoint()
// LogicalRDD output [id#236], outputPartitioning hashpartitioning(id#232,
5): orig's column
broadcast(orig).join(cp, orig("id") % 2 === cp("id") %
2).groupBy(orig("id")).count()
```
This returns 72 rows of partial counts instead of 16 rows with count 8, on
the base and at this head. It's issue 1 without a self-join.
`rewriteStatsAndConstraints` already maps the optimized output onto the
analyzed output for stats and constraints. Doing the same here for the executed
plan's output fixes it:
```scala
val toOutput = AttributeMap(executedPlan.output.zip(logicalPlan.output))
def remap[E <: Expression](e: E): E =
e.transform { case a: Attribute => toOutput.getOrElse(a, a)
}.asInstanceOf[E]
val partitioning =
firstLeafPartitioning(executedPlan.outputPartitioning) match {
case p: Expression => remap(p).asInstanceOf[Partitioning]
case p => p
}
// then pass `partitioning` and `executedPlan.outputOrdering.map(remap)`
```
With it, the query returns 16 rows with count 8, and `DatasetSuite` still
passes.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/ExistingRDD.scala:
##########
@@ -117,7 +116,9 @@ case class LogicalRDD(
session :: originStats :: originConstraints :: Boolean.box(fromCheckpoint)
:: Nil
override def newInstance(): LogicalRDD.this.type = {
- val rewrite = Utils.toMap(output, output.map(_.newInstance()))
+ // Keyed by exprId, because the attributes in the partitioning, the
ordering, the stats and
+ // the constraints can differ from `output` in qualifier or nullability.
+ val rewrite = AttributeMap(output.zip(output.map(_.newInstance())))
Review Comment:
**Finding 1.** With `rewrite` keyed by exprId, `output.map(rewrite)` at line
140 maps output columns that share an exprId onto one new attribute: the last
column's new instance. If they differ in name or qualifier, the renewed copy
renames a column. Case-insensitive resolution, the default, produces such
columns:
```scala
val cp = spark.range(3).select($"id", $"ID").localCheckpoint() // output
[id#0, ID#0]
cp.join(cp).columns
// base: id, ID, id, ID
// this head: id, ID, ID, ID
```
Qualifiers behave the same way. Checkpointing `df.as("s").select(df("id"),
$"s.id")` gives the output `[t.id, s.id]` on one exprId. So `$"t.id"` over
`cp.join(cp)` resolves silently at this head, where the base raises
`AMBIGUOUS_REFERENCE`.
Keeping each attribute and swapping only its exprId, as SPARK-54718 did for
`CTERelationRef.newInstance` (`cteOperators.scala:247`), fixes it:
```scala
output.map(a => a.withExprId(rewrite(a).exprId)),
```
With it, `cp.join(cp).columns` is `id, ID, id, ID` again, and the new test
still passes. A `columns` check in the test would pin it.
--
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]
