dongjoon-hyun commented on code in PR #58531:
URL: https://github.com/apache/spark/pull/58531#discussion_r3952220407
##########
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:
The `forall` here makes `isCompatibleWith` asymmetric with the collection
side, which answers `exists` (`ShuffleSpecCollection(Seq(h1,
h10)).isCompatibleWith(SinglePartitionShuffleSpec)` is true while the reverse
is false), and the trait doc says Spark assumes symmetry. In practice nothing
observable changes: only `KeyedShuffleSpec` can make the members disagree on
`numPartitions`, and it has no `SinglePartitionShuffleSpec` case, so that
direction is already false.
Also worth noting in the comment: `EnsureRequirements` never reaches this
branch, since a spec with `canCreatePartitioning == false` is never `best`. The
one production caller that can hit it with a collection on the `other` side is
`ValidateRequirements` (`specs.tail.forall(_.isCompatibleWith(specs.head))`),
where the stricter `forall` is the safer answer.
##########
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:
Nit: for the chosen pair `areKeysCompatible` is now evaluated three times:
once in the pairing filter, once inside `isCompatibleWith` above, and again in
`isCompatible = leftSpec.areKeysCompatible(rightSpec)` just below. Not
expensive, but the last one is known to be true whenever an agreeing pair was
found, so it could be skipped (or the pairing result reused) if the fallback
above becomes an early return.
##########
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:
Minor: since no agreeing pair means every pair fails both `isCompatibleWith`
and the `areKeysCompatible` check in the push branch, this could simply be an
early `return None` instead of falling back to the two heads. That is simpler,
and it also avoids emitting `"Pushing common partition values for
storage-partitioned join"` for a join where nothing is pushed. I understand the
intent of keeping the old path byte for byte, but with the pairing in place
there is little reason to keep it.
--
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]
