peter-toth commented on code in PR #58531:
URL: https://github.com/apache/spark/pull/58531#discussion_r3956449706
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -1361,11 +1378,20 @@ trait ShuffleSpec {
*/
def createPartitioning(clustering: Seq[Expression]): Partitioning =
throw SparkUnsupportedOperationException()
+
+ override final def flatten: Seq[LeafShuffleSpec] = Seq(this)
}
-case object SinglePartitionShuffleSpec extends ShuffleSpec {
- override def isCompatibleWith(other: ShuffleSpec): Boolean = {
- other.numPartitions == 1
+case object SinglePartitionShuffleSpec extends LeafShuffleSpec {
+ override def isCompatibleWith(other: ShuffleSpec): Boolean = other match {
+ case leaf: LeafShuffleSpec => leaf.numPartitions == 1
+ // `forall`, not the `exists` the other specs use for a collection. They
ask whether *some*
+ // member matches them and then plan on that member; this one never names
a member, since
+ // `canCreatePartitioning` is false, so the answer has to hold for
whichever member the plan
+ // settles on. The counts are projected ones as everywhere here, so a
child whose every member
+ // projects to one answers yes even while holding more partitions of its
own. Members can only
+ // disagree when the subset config projects them onto different key sets.
+ case ShuffleSpecCollection(specs) => specs.forall(isCompatibleWith)
Review Comment:
Added, and the symmetry point led somewhere. The comment now names the
caller and the reachability.
I think the asymmetry is a symptom rather than a quirk. `isCompatibleWith`
serves two questions. `EnsureRequirements` asks it as a search - is some member
of this collection compatible, and which one - and then records the member it
plans on. `ValidateRequirements` asks it as a check on a built plan, where the
member is recorded nowhere, so it re-derives the choice. `exists` is right for
the first. For the second there is no right answer, only a safer one.
Enumerating the production callers says the same: this collection case is
reachable only from `ValidateRequirements`, since `canCreatePartitioning` is
false and `EnsureRequirements`' `best` is filtered on it. That is why `forall`
versus `exists` felt arbitrary here.
The old code was asymmetric too, just order-dependently:
`other.numPartitions` read `specs.head`, so `[h1, h10]` answered true and
`[h10, h1]` false, while `exists` said true for both. `forall` replaces an
order-dependent violation with a deterministic conservative one, which is what
this PR does everywhere else.
Left as `forall`, with the API split noted as a follow-up rather than done
here. A separate validation API alone would not fix it, since the validation
side would still have nothing to check against. That belongs with SPARK-59289,
which is about recording what the planner decided.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -523,28 +524,75 @@ case class EnsureRequirements(
var newLeft = left
var newRight = right
- val specs = Seq(left, right).zip(requiredChildDistribution).map { case (p,
d) =>
- if (!d.isInstanceOf[ClusteredDistribution]) return None
- val cd = d.asInstanceOf[ClusteredDistribution]
- val specOpt = createKeyedShuffleSpec(p.outputPartitioning, cd)
- if (specOpt.isEmpty) return None
- specOpt.get
- }
+ def candidatesFor(plan: SparkPlan, required: Distribution):
Seq[KeyedShuffleSpec] =
+ required match {
+ case cd: ClusteredDistribution =>
createKeyedShuffleSpecs(plan.outputPartitioning, cd)
+ case _ => Nil
+ }
+ val leftCandidates = candidatesFor(left, requiredChildDistribution.head)
+ val rightCandidates = candidatesFor(right, requiredChildDistribution(1))
+ if (leftCandidates.isEmpty || rightCandidates.isEmpty) return None
+
+ // A spec carries no `joinKeyPositions` exactly when its partitioning is
the child's own member,
+ // so this asks whether `isCompatible` below can take the pair as it
stands, without a
+ // `GroupPartitionsExec` on either side - unless partially clustered
distribution is on, which
+ // sends every pairing through the push branch anyway.
+ def bothUnprojected(l: KeyedShuffleSpec, r: KeyedShuffleSpec): Boolean =
+ l.joinKeyPositions.isEmpty && r.joinKeyPositions.isEmpty
+
+ // How many key groups the pushdown below would leave this pair.
`mergeAndDedupPartitions`
+ // keeps one side's keys and drops the other's for the filtered one-sided
join types, and there
+ // the dropped side's count says nothing, so rank on the side that
survives. The arms that
+ // really merge have no cheap answer, so they take the larger of the two
counts. Keep the join
+ // types here in step with `mergeAndDedupPartitions`.
+ def rank(l: KeyedShuffleSpec, r: KeyedShuffleSpec): Int =
+ if (!conf.getConf(SQLConf.V2_BUCKETING_PARTITION_FILTER_ENABLED)) {
+ l.numPartitions.max(r.numPartitions)
+ } else {
+ joinType match {
+ case LeftOuter | LeftAnti | LeftSingle | ExistenceJoin(_) =>
l.numPartitions
+ case RightOuter => r.numPartitions
+ case _ => l.numPartitions.max(r.numPartitions)
+ }
+ }
- val leftSpec = specs.head
- val rightSpec = specs(1)
+ // Each side may offer several members, and the right one is the one the
other side can pair
+ // with, which neither side can tell on its own. So pick the pair rather
than a member per side,
+ // and rank the pairs that agree on the keys by the parallelism they
offer, the same trade
+ // `ensureDistributionAndOrdering` makes between children when it picks
`bestSpecOpt`.
+ //
+ // Two things keep `rank` from being what the join actually gets, both on
the merging arms.
+ // `InnerLike` and `LeftSemi` intersect under
`v2BucketingPartitionFilterEnabled`, and an
+ // intersection is not monotone in member granularity: members cover
different clustering keys
+ // rather than nested ones, so a finer pair can rank above a coarser one
and still meet the
+ // other side in fewer groups. And a union does not merely exceed the rank
either, because
+ // `reduceKeys` runs between this pick and the merge and collapses
distinct keys, so a
+ // `bucket(16)` side reduced onto `bucket(8)` brings 8 keys to a merge its
spec ranked at 16.
+ // Ranking on the merged count would match what is delivered, at the cost
of merging every
+ // candidate pair, which would also raise
`storagePartitionJoinIncompatibleReducedTypesError`
+ // for pairs that are never chosen.
+ //
+ // Ties go to a pair both children report as it stands, to keep the
no-grouping-node path. A
+ // projected count never exceeds the physical one, so nothing outranks
such a pair, but a coarse
+ // member whose projected count happens to equal it ties, and enumeration
order would decide.
+ val agreeingPairs = for {
+ l <- leftCandidates
+ r <- rightCandidates
+ if l.areKeysCompatible(r)
+ } yield (l, r)
+ val (leftSpec, rightSpec) = agreeingPairs
+ .maxByOption { case (l, r) => (rank(l, r), bothUnprojected(l, r)) }
+ // No agreeing pair means every pair fails the checks below, so the
method returns `None`
+ // whichever one it reports. Reporting each side's first member keeps
that path byte for byte
+ // what the per-side pick produced, `logInfo` included.
+ .getOrElse((leftCandidates.head, rightCandidates.head))
Review Comment:
Done. Traced it first: with no agreeing pair the head-head pair fails
`isCompatibleWith` and the push branch's `areKeysCompatible` alike, so the
method returned `None` after logging a pushdown that never happened.
`SPARK-59256: no agreeing pair leaves the join alone however many members each
side has` is the guard for it.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -523,28 +524,75 @@ case class EnsureRequirements(
var newLeft = left
var newRight = right
- val specs = Seq(left, right).zip(requiredChildDistribution).map { case (p,
d) =>
- if (!d.isInstanceOf[ClusteredDistribution]) return None
- val cd = d.asInstanceOf[ClusteredDistribution]
- val specOpt = createKeyedShuffleSpec(p.outputPartitioning, cd)
- if (specOpt.isEmpty) return None
- specOpt.get
- }
+ def candidatesFor(plan: SparkPlan, required: Distribution):
Seq[KeyedShuffleSpec] =
+ required match {
+ case cd: ClusteredDistribution =>
createKeyedShuffleSpecs(plan.outputPartitioning, cd)
+ case _ => Nil
+ }
+ val leftCandidates = candidatesFor(left, requiredChildDistribution.head)
+ val rightCandidates = candidatesFor(right, requiredChildDistribution(1))
+ if (leftCandidates.isEmpty || rightCandidates.isEmpty) return None
+
+ // A spec carries no `joinKeyPositions` exactly when its partitioning is
the child's own member,
+ // so this asks whether `isCompatible` below can take the pair as it
stands, without a
+ // `GroupPartitionsExec` on either side - unless partially clustered
distribution is on, which
+ // sends every pairing through the push branch anyway.
+ def bothUnprojected(l: KeyedShuffleSpec, r: KeyedShuffleSpec): Boolean =
+ l.joinKeyPositions.isEmpty && r.joinKeyPositions.isEmpty
+
+ // How many key groups the pushdown below would leave this pair.
`mergeAndDedupPartitions`
+ // keeps one side's keys and drops the other's for the filtered one-sided
join types, and there
+ // the dropped side's count says nothing, so rank on the side that
survives. The arms that
+ // really merge have no cheap answer, so they take the larger of the two
counts. Keep the join
+ // types here in step with `mergeAndDedupPartitions`.
+ def rank(l: KeyedShuffleSpec, r: KeyedShuffleSpec): Int =
+ if (!conf.getConf(SQLConf.V2_BUCKETING_PARTITION_FILTER_ENABLED)) {
+ l.numPartitions.max(r.numPartitions)
+ } else {
+ joinType match {
+ case LeftOuter | LeftAnti | LeftSingle | ExistenceJoin(_) =>
l.numPartitions
+ case RightOuter => r.numPartitions
+ case _ => l.numPartitions.max(r.numPartitions)
+ }
+ }
- val leftSpec = specs.head
- val rightSpec = specs(1)
+ // Each side may offer several members, and the right one is the one the
other side can pair
+ // with, which neither side can tell on its own. So pick the pair rather
than a member per side,
+ // and rank the pairs that agree on the keys by the parallelism they
offer, the same trade
+ // `ensureDistributionAndOrdering` makes between children when it picks
`bestSpecOpt`.
+ //
+ // Two things keep `rank` from being what the join actually gets, both on
the merging arms.
+ // `InnerLike` and `LeftSemi` intersect under
`v2BucketingPartitionFilterEnabled`, and an
+ // intersection is not monotone in member granularity: members cover
different clustering keys
+ // rather than nested ones, so a finer pair can rank above a coarser one
and still meet the
+ // other side in fewer groups. And a union does not merely exceed the rank
either, because
+ // `reduceKeys` runs between this pick and the merge and collapses
distinct keys, so a
+ // `bucket(16)` side reduced onto `bucket(8)` brings 8 keys to a merge its
spec ranked at 16.
+ // Ranking on the merged count would match what is delivered, at the cost
of merging every
+ // candidate pair, which would also raise
`storagePartitionJoinIncompatibleReducedTypesError`
+ // for pairs that are never chosen.
+ //
+ // Ties go to a pair both children report as it stands, to keep the
no-grouping-node path. A
+ // projected count never exceeds the physical one, so nothing outranks
such a pair, but a coarse
+ // member whose projected count happens to equal it ties, and enumeration
order would decide.
+ val agreeingPairs = for {
+ l <- leftCandidates
+ r <- rightCandidates
+ if l.areKeysCompatible(r)
+ } yield (l, r)
+ val (leftSpec, rightSpec) = agreeingPairs
+ .maxByOption { case (l, r) => (rank(l, r), bothUnprojected(l, r)) }
+ // No agreeing pair means every pair fails the checks below, so the
method returns `None`
+ // whichever one it reports. Reporting each side's first member keeps
that path byte for byte
+ // what the per-side pick produced, `logInfo` included.
+ .getOrElse((leftCandidates.head, rightCandidates.head))
val leftPartitioning = leftSpec.partitioning
val rightPartitioning = rightSpec.partitioning
// We don't need to alter the existing or add new `GroupPartitionsExec`
when the child
// partitionings are not modified (projected) in specs and left and right
side partitionings are
// compatible with each other.
- // Left and right `outputPartitioning` is a `PartitioningCollection` or a
`KeyedPartitioning`
- // otherwise `createKeyedShuffleSpec()` would have returned `None`.
- var isCompatible =
- left.outputPartitioning.asInstanceOf[Expression].exists(_ ==
leftPartitioning) &&
- right.outputPartitioning.asInstanceOf[Expression].exists(_ ==
rightPartitioning) &&
- leftSpec.isCompatibleWith(rightSpec)
+ var isCompatible = bothUnprojected(leftSpec, rightSpec) &&
leftSpec.isCompatibleWith(rightSpec)
if ((!isCompatible ||
conf.v2BucketingPartiallyClusteredDistributionEnabled) &&
Review Comment:
Done, and it took the guard with it. After the early return above, the
chosen pair always comes from `agreeingPairs`, so `areKeysCompatible` is true
by construction and the `if (isCompatible)` below was dead. The branch now sets
`isCompatible = true` with the reason, and the push body is de-indented into
it. The `var` became two `val`s on the way, so nothing reassigns the answer
across those 184 lines any more.
That reformats the block, so read the hunk with `git diff -w`: four added,
three removed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
