XiDuo You created SPARK-61211:
---------------------------------
Summary: Avoid incorrect union partitioning with duplicate child
expression IDs
Key: SPARK-61211
URL: https://issues.apache.org/jira/browse/SPARK-61211
Project: Spark
Issue Type: Bug
Components: SQL
Affects Versions: 5.0.0
Reporter: XiDuo You
UnionExec.prepareOutputPartitioning can report an incorrect hash layout when a
child output repeats an exprId. Runtime BindReferences selects the first
occurrence, while AttributeMap(child.output.zip(union.output)) retains the last
occurrence. A union can consequently advertise clustering on a column whose
rows were actually shuffled by another column, causing EnsureRequirements to
omit a necessary exchange.
This is independent of SPARK-60379's shuffle push-down. Reviewers reproduced
the existing bug on merge base 352ac9bf228 with push-down off, and with AQE
both on and off. Disabling spark.sql.unionOutputPartitioning restores the
expected result. The exact earliest affected release has not yet been
determined.
Reported reproduction (Scala):
{code:scala}
import org.apache.spark.sql.functions._
import org.apache.spark.sql.expressions.Window
import spark.implicits._
spark.conf.set("spark.sql.analyzer.unionIsResolvedWhenDuplicatesPerChildResolved",
"false")
spark.conf.set("spark.sql.enforceTypeCoercionBeforeUnionDeduplication.enabled",
"false")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "false")
spark.conf.set("spark.sql.shuffle.partitions", "5")
spark.conf.set("spark.sql.unionOutputPartitioning", "true")
// Test harness controls; avoid resolver dual-run validation and codegen
compilation of duplicates.
spark.conf.set("spark.sql.analyzer.singlePassResolver.dualRunWithLegacy",
"false")
spark.conf.set("spark.sql.codegen.wholeStage", "false")
val p1 = spark.range(1000).selectExpr("id p", "id w").select($"p", $"p", $"w")
val p2 = spark.range(1000, 11000).selectExpr("id q", "id % 1000 r", "id w2")
val x = spark.range(1000).selectExpr("id x", "id c", "id u").repartition($"c")
val r = p1.union(p2).repartition(p1.col("p"))
val result = x.union(r).withColumn("rn",
row_number().over(Window.partitionBy(x.col("c")).orderBy(x.col("u"))))
result.collect().count(_.getInt(3) == 1)
{code}
Expected: 1000. Reported actual: 4582. Collect all columns before counting to
retain the triggering plan shape.
The duplicate second child reports hash(p), physically bound to its first
column. Its mapping into the outer union reports hash(c), corresponding to the
second column. Both outer children then appear to report hash(c), although
their rows are not co-located by c.
Investigate a conservative refusal to propagate partitioning from ambiguous
child outputs, or a mapping aligned with first-occurrence runtime binding, with
coverage for both mapping directions and all supported partitioning types. Keep
this general UnionExec fix separate from the narrower push-down eligibility
guards in SPARK-60379.
Reviewer evidence:
https://github.com/apache/spark/pull/59630#discussion_r4495483202
https://github.com/apache/spark/pull/59630#discussion_r4493424741
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
