[
https://issues.apache.org/jira/browse/SPARK-59234?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
XiDuo You updated SPARK-59234:
------------------------------
Description:
h2. Problem
{{GroupPartitionsExec.alignToExpectedKeys}}
({{sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala}})
derives {{isGrouped}} from the expected counts, while {{padTo}} never
truncates:
{code:java}
var isGrouped = true
val alignedPartitions = expectedPartitionKeys.get.flatMap { case (key,
numSplits) =>
if (numSplits > 1) isGrouped = false
val splits = keyMap.getOrElse(key, Seq.empty)
if (distributePartitions) {
val paddedSplits = splits.map(Seq(_)).padTo(numSplits, Seq.empty)
paddedSplits.map((key, _))
} else {
Seq.fill(numSplits)((key, splits))
}
}
{code}
When the expected count of a key is lower than the number of splits the child
actually holds for it, the distribute branch emits more partitions than
declared for that key, all carrying the same key, while {{isGrouped}} can still
be reported as {{true}} (whenever {{numSplits == 1}}). The emitted layout then
contradicts its own report: duplicate keys under a grouped report.
h2. Why it matters
{{isGrouped = true}} is consumed as fact downstream: by {{satisfies}} for
{{ClusteredDistribution}}, by join pairing and by shuffle-spec construction. A
wrong expected count therefore surfaces several steps away from its cause.
During the SPARK-58996 review this was observed as {{requirement failed: All
KeyedPartitionings in a PartitioningCollection must have equal partitionKeys}}
from {{PartitioningCollection.fromPartitionings}} through
{{SortMergeJoinExec.outputPartitioning}}, while the miscount itself was
produced in {{alignToExpectedKeys}}.
h2. Possible fixes
* derive {{isGrouped}} from the partitions actually emitted, so the report is
honest and downstream gates react to the real layout; or
* assert {{splits.size <= numSplits}} in the distribute branch, failing fast at
the node that owns the alignment, with an error pointing at the cause.
h2. Context
Follow-up from the review of
[https://github.com/apache/spark/pull/58279|#58279] (SPARK-58996, finding 5,
raised by Peter Toth). That PR removed the known producers of a wrong expected
count: the statistics, the original partition keys and the positions projecting
them are now read from the pre-alignment plan through one descent. This ticket
is the node-level guard that would localise any future producer of a wrong
count at {{GroupPartitionsExec}} instead of letting the inconsistency propagate.
> GroupPartitionsExec should catch a wrong expected partition key count at the
> node
> ---------------------------------------------------------------------------------
>
> Key: SPARK-59234
> URL: https://issues.apache.org/jira/browse/SPARK-59234
> Project: Spark
> Issue Type: Improvement
> Components: SQL
> Affects Versions: 5.0.0
> Reporter: XiDuo You
> Priority: Major
>
> h2. Problem
> {{GroupPartitionsExec.alignToExpectedKeys}}
> ({{sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala}})
> derives {{isGrouped}} from the expected counts, while {{padTo}} never
> truncates:
> {code:java}
> var isGrouped = true
> val alignedPartitions = expectedPartitionKeys.get.flatMap { case (key,
> numSplits) =>
> if (numSplits > 1) isGrouped = false
> val splits = keyMap.getOrElse(key, Seq.empty)
> if (distributePartitions) {
> val paddedSplits = splits.map(Seq(_)).padTo(numSplits, Seq.empty)
> paddedSplits.map((key, _))
> } else {
> Seq.fill(numSplits)((key, splits))
> }
> }
> {code}
> When the expected count of a key is lower than the number of splits the child
> actually holds for it, the distribute branch emits more partitions than
> declared for that key, all carrying the same key, while {{isGrouped}} can
> still be reported as {{true}} (whenever {{numSplits == 1}}). The emitted
> layout then contradicts its own report: duplicate keys under a grouped report.
> h2. Why it matters
> {{isGrouped = true}} is consumed as fact downstream: by {{satisfies}} for
> {{ClusteredDistribution}}, by join pairing and by shuffle-spec construction.
> A wrong expected count therefore surfaces several steps away from its cause.
> During the SPARK-58996 review this was observed as {{requirement failed: All
> KeyedPartitionings in a PartitioningCollection must have equal
> partitionKeys}} from {{PartitioningCollection.fromPartitionings}} through
> {{SortMergeJoinExec.outputPartitioning}}, while the miscount itself was
> produced in {{alignToExpectedKeys}}.
> h2. Possible fixes
> * derive {{isGrouped}} from the partitions actually emitted, so the report is
> honest and downstream gates react to the real layout; or
> * assert {{splits.size <= numSplits}} in the distribute branch, failing fast
> at the node that owns the alignment, with an error pointing at the cause.
> h2. Context
> Follow-up from the review of
> [https://github.com/apache/spark/pull/58279|#58279] (SPARK-58996, finding 5,
> raised by Peter Toth). That PR removed the known producers of a wrong
> expected count: the statistics, the original partition keys and the positions
> projecting them are now read from the pre-alignment plan through one descent.
> This ticket is the node-level guard that would localise any future producer
> of a wrong count at {{GroupPartitionsExec}} instead of letting the
> inconsistency propagate.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
