Peter Toth created SPARK-59575:
----------------------------------
Summary: Storage-partitioned join is lost when a marked layout's
declared key order is not sorted
Key: SPARK-59575
URL: https://issues.apache.org/jira/browse/SPARK-59575
Project: Spark
Issue Type: Bug
Components: SQL
Affects Versions: 5.0.0
Reporter: Peter Toth
A marked {{KeyedPartitioning}} (one with {{mayContainUnknownPartitionKeys}},
produced only by the one-side shuffle behind
{{spark.sql.sources.v2.bucketing.shuffle.enabled}}) pins the rows whose key it
does not declare to {{hash(key) % numPartitions}} over the keys it does
declare, *in the order it declares them*.
{{EnsureRequirements.mergeAndDedupPartitions}} sorts the merged partition key
list, for the determinism SPARK-41471 asked for. So a marked layout whose
declared order is not the sorted one is relabelled by the regrouping even where
the key set is unchanged. That is what a one-side shuffle onto a
{{UnionExec}}'s key order leaves behind. {{GroupPartitionsExec}} then gives up
the keyed claim, because only an identity regrouping keeps it, and since
SPARK-59272 the pairing gate declines and the join shuffles both sides.
This is a regression rather than a missing optimization. The "three shuffles to
four" plan change SPARK-59272 lists as its user-facing change is exactly this
shape. Measured: handing the marked side its own key list verbatim takes the
{{SPARK-59050: SPJ: regrouping a marked layout must not keep the unknown-keyed
claim}} query back to three shuffles, with a correct answer and a plan
{{ValidateRequirements}} accepts. So the fourth shuffle was not necessary in
principle, only given the rest of the machinery.
What blocks the obvious fix: {{KeyedPartitioning.createShuffleSpec}} sorts an
unmarked side through {{toGrouped}} under
{{spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled}},
while it hands a marked layout back unprojected. With the merged list left in
the marked side's order, the two children then hold one partitioning but report
two specs that {{KeyLayout.describesSameKeys}} calls different, so
{{ValidateRequirements}} rejects the join the gate just allowed. The generated
sweep added by SPARK-59272 catches it on cell {{left=id/12 right=id/312/marked
Inner}}. That asymmetry has to be settled first.
One further wrinkle for whoever takes this: the merge runs on *reduced* keys,
so "the marked side's keys verbatim" means its reduced keys, and the rule needs
a reducer exception. A reducer forfeits the claim on its own through
{{keysRewritten}}.
The other half of the same regression, where partition filtering narrows the
merged list below the marked side's declared keys, is fixed separately by
SPARK-59272's follow-up (apache/spark#58858). This ticket is only the ordering
half.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
