peter-toth commented on code in PR #58044:
URL: https://github.com/apache/spark/pull/58044#discussion_r3813801741
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala:
##########
@@ -892,22 +892,20 @@ case class UnionExec(children: Seq[SparkPlan]) extends
SparkPlan with CodegenSup
}
/**
- * Returns the output partitionings of the children, with the attributes
converted to
- * the first child's attributes at the same position.
+ * Returns the output partitionings of the children, with the attributes
converted to this
+ * union's output attributes at the same position.
*/
private def prepareOutputPartitioning(): Seq[Partitioning] = {
- // Create a map of attributes from the other children to the first child.
- val firstAttrs = children.head.output
- val attributesMap = children.tail.map(_.output).map { otherAttrs =>
- AttributeMap(otherAttrs.zip(firstAttrs))
+ // Map every child's partitioning attributes to this union's output
attributes, so all
+ // partitionings are expressed in the same attribute space before
comparison. A child's
+ // `outputPartitioning` may reference its own input attributes (e.g. a
Filter passes through
+ // its child's partitioning but adjusts the output nullability), so even
the first child is
Review Comment:
**Finding 4.** The general statement is right — a child's
`outputPartitioning` need not be expressed in its own output attributes — but
the parenthetical, and with it the PR title and the JIRA summary, names the
wrong cause. I probed the union's children on base (`5460c109699`) for exactly
the query the new test uses:
```
child0 = Project
child0.out(0): id exprId=9 nullable=true
qualifier=List(__auto_generated_subquery_name)
child0.partAttr: id exprId=9 nullable=true qualifier=List(t1)
PROBE child0.out(0) == child0.partAttr : false
union.partitioning = UnknownPartitioning(0)
```
Both sides are `nullable=true`. `AttributeReference.equals`
(`namedExpressions.scala:301`) also compares `qualifier`, and that is the bit
that differs: the shuffle's `HashPartitioning` holds `id#9` as resolved inside
the `t1` view, while the `Project` above the `Filter` holds `id#9` as resolved
against `__auto_generated_subquery_name`. `FilterExec.output`'s
`outputWithNullability` narrowing (`:267`) does happen, but it never reaches
child 0's `output` — the `Project` on top re-exposes the outer, still-nullable
attribute.
That makes this the direct continuation of SPARK-53550 (`8edc7685b97`,
`c0acf45023f`), which fixed the same qualifier/metadata sensitivity on the
*tail* children's map keys; the first child's own partitioning was the one
remaining hop. Worth saying that in the description.
Suggested rewording:
```scala
// `outputPartitioning` may reference attributes that differ
cosmetically from the ones in
// the child's own `output`: a Filter passes its child's partitioning
through verbatim while
// rewriting its `output`, and a partitioning built inside a view or
subquery carries that
// relation's qualifier. `AttributeReference.equals` compares name,
nullability, metadata and
// qualifier, so the un-remapped first child compared unequal to every
remapped sibling.
// `AttributeMap` is keyed by `ExprId`, so remapping normalizes all of
those.
```
For the title, either something cause-neutral ("UnionExec outputPartitioning
should compare children in the union's attribute space") or keep the
nullability framing and add a test whose plan isolates it — I tried three
shapes and the qualifier was the discriminator in every one, so I am not
convinced a nullability-only plan is reachable here.
##########
sql/core/src/test/scala/org/apache/spark/sql/DataFrameSetOperationsSuite.scala:
##########
@@ -1589,6 +1589,58 @@ class DataFrameSetOperationsSuite extends
SharedSparkSession with AdaptiveSparkP
}
}
+ test("SPARK-58819: union outputPartitioning ignores partition key
nullability") {
+ withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ withTempView("t1", "t2") {
+ // `id` is nullable (Option[Int]) so that `IsNotNull` in the Filter
actually adjusts it.
+ // The two branches overlap on `id = 1`, so a shared key's grouping
result depends on the
+ // union's co-location claim being honored (guarded by
`checkAnswer(grouped, ...)` below).
+ Seq((Option(1), 10), (Option(2), 20)).toDF("id",
"v").createOrReplaceTempView("t1")
+ Seq((Option(1), 30), (Option(3), 40)).toDF("id",
"v").createOrReplaceTempView("t2")
+
+ val sqlText =
+ """
+ |SELECT id, v FROM (SELECT id, v FROM t1 DISTRIBUTE BY id) WHERE
id IS NOT NULL
+ |UNION ALL
+ |SELECT id, sum(v) AS v FROM t2 GROUP BY id
+ |""".stripMargin
+ val union = spark.sql(sqlText)
+ val unionExec = union.queryExecution.executedPlan.collect { case u:
UnionExec => u }
+ assert(unionExec.size == 1)
+
+ // `IsNotNull` in the Filter adjusts the nullability of the partition
key, but nullability
+ // does not affect the hash, so the union should still propagate the
hash partitioning.
+
assert(unionExec.head.outputPartitioning.isInstanceOf[HashPartitioning],
Review Comment:
**Finding 5.** `isInstanceOf[HashPartitioning]` plus the shuffle counts
don't cover the other half of the diff — the deletion of `toUnionOutput`, whose
job was to express the chosen partitioning in the union's *own* output
attributes. Everything downstream matches by `semanticEquals`, and
`AttributeReference.canonicalized` keeps only `exprId` and `dataType`
(`namedExpressions.scala:326`), so a partitioning that leaked a child attribute
would still satisfy the group-by's `ClusteredDistribution` and
`groupedShuffles` would still be 2.
Here the two candidates are distinguishable, so the assertion is cheap and
actually discriminating — measured on this head the union reports `id#9` with
`qualifier=[__auto_generated_subquery_name]` (its own output attribute) while
child 0's partitioning attribute carries `qualifier=[t1]`:
```scala
val hashPartitioning =
unionExec.head.outputPartitioning.asInstanceOf[HashPartitioning]
assert(hashPartitioning.expressions == Seq(unionExec.head.output.head))
```
--
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]
