[
https://issues.apache.org/jira/browse/SPARK-61211?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
XiDuo You updated SPARK-61211:
------------------------------
Description:
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 union output partitioning mapping dates back to
SPARK-52921 and shipped in 4.1.0; the two analyzer confs this shape needs were
also added in 4.1.0, so 4.1.0 is the earliest affected release.
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
was:
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
> 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: 4.1.0, 4.2.0, 4.3.0, 5.0.0, 4.4.0
> Reporter: XiDuo You
> Priority: Major
>
> 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 union output partitioning mapping dates back to
> SPARK-52921 and shipped in 4.1.0; the two analyzer confs this shape needs
> were also added in 4.1.0, so 4.1.0 is the earliest affected release.
> 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]
