[
https://issues.apache.org/jira/browse/SPARK-61211?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
XiDuo You updated SPARK-61211:
------------------------------
Affects Version/s: 4.1.0
4.2.0
4.3.0
4.4.0
> 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 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]
