[
https://issues.apache.org/jira/browse/SPARK-59575?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Peter Toth updated SPARK-59575:
-------------------------------
Affects Version/s: 4.4.0
> 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, 4.4.0
> Reporter: Peter Toth
> Priority: Major
>
> 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]
