peter-toth commented on code in PR #58514:
URL: https://github.com/apache/spark/pull/58514#discussion_r3933601293
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -838,11 +838,18 @@ case class EnsureRequirements(
joinType: JoinType,
keyOrdering: Ordering[InternalRowComparableWrapper]):
Seq[InternalRowComparableWrapper] = {
val merged = if
(SQLConf.get.getConf(SQLConf.V2_BUCKETING_PARTITION_FILTER_ENABLED)) {
+ // Rows only match within a key group, so a group no output row can come
from is dropped.
+ // Only equi-joins reach here (SMJ/SHJ over `ExtractEquiJoinKeys`), so
Cross matches on its
+ // keys like Inner, as `canReplicateLeftSide` already assumes. Which
side's rows survive
Review Comment:
**Finding 3.** `canReplicateLeftSide` does not assume equi keys. It answers
a different question: whether replicating a side duplicates output rows.
`Cross` qualifies there for the same reason `Inner` does, because neither side
keeps unmatched rows. That answer would be unchanged for a condition-less cross
join, so it is not evidence for the clause it is attached to.
The evidence is the clause right before it, and it is worth naming the part
that actually carries the weight:
```scala
// Rows only match within a key group, so a group no output row can
come from is dropped.
// Only equi-joins reach here: every SMJ/SHJ takes its keys from
`ExtractEquiJoinKeys`, which
// needs `joinKeys.nonEmpty`. So Cross matches on its keys like Inner.
Which side's rows
// survive follows `PushExtraPredicateThroughJoin`, plus LeftSingle.
```
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -3497,6 +3539,66 @@ class KeyGroupedPartitioningSuite
}
}
+ // A cross join reaches SPJ only with equi keys, so it and a semi join drop
the key groups absent
+ // on either side. An anti join has to probe every left row, so only the
right-only groups are
+ // dropped.
+ Seq(
+ ("CROSS", Cross, Seq(Row(4, "cc", 40.0)), 1),
Review Comment:
**Finding 2.** The comment at `EnsureRequirements.scala:842` rests on two
halves. This case pins the first: a `Cross` carrying an equi condition does
reach the path, and `intersect` is right for it. The second half is that a
condition-less `Cross` never reaches it, and nothing pins that. It is the half
[pullrequestreview-5111706183](https://github.com/apache/spark/pull/58514#pullrequestreview-5111706183)
named as the fragile one, so it is worth a test rather than only a comment.
Six lines cover it. I ran this on `4c4a7be1600`:
```scala
test("SPARK-59199: a cross join without an equi condition does not reach
SPJ") {
withPartitionFilterJoinTables {
val df = sql(
s"""
|SELECT i.id, p.item_id
|FROM testcat.ns.$items i CROSS JOIN testcat.ns.$purchases p
|""".stripMargin)
val plan = df.queryExecution.executedPlan
assert(collectAllGroupPartitions(plan).isEmpty,
s"a cartesian product must not group partitions in\n$plan")
assert(df.count() == 6, "every left x right pair survives")
}
}
```
The plan is a bare `CartesianProduct` over the two `BatchScan`s, and all 3 x
2 rows come back. A future change that routed this shape through
`checkKeyGroupCompatible` would drop rows, and this is the assertion that would
catch it.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -838,11 +838,18 @@ case class EnsureRequirements(
joinType: JoinType,
keyOrdering: Ordering[InternalRowComparableWrapper]):
Seq[InternalRowComparableWrapper] = {
val merged = if
(SQLConf.get.getConf(SQLConf.V2_BUCKETING_PARTITION_FILTER_ENABLED)) {
+ // Rows only match within a key group, so a group no output row can come
from is dropped.
+ // Only equi-joins reach here (SMJ/SHJ over `ExtractEquiJoinKeys`), so
Cross matches on its
+ // keys like Inner, as `canReplicateLeftSide` already assumes. Which
side's rows survive
+ // follows `PushExtraPredicateThroughJoin`, plus LeftSingle.
joinType match {
- case Inner =>
+ // neither side keeps unmatched rows
+ case _: InnerLike | LeftSemi =>
mergeAndDedupPartitionKeys(leftPartitionKeys, rightPartitionKeys,
intersect = true)
- case LeftOuter => leftPartitionKeys.distinct
+ // every left row is kept or tested
+ case LeftOuter | LeftAnti | LeftSingle | ExistenceJoin(_) =>
leftPartitionKeys.distinct
Review Comment:
**Finding 1.** `canReplicateRightSide` at line 728 is the sibling of this
classification, and it carries the same omission you are fixing here. It lists
`Inner`, `Cross`, `LeftSemi`, `LeftAnti` and `LeftOuter`, but not `LeftSingle`
or `ExistenceJoin`. Both preserve every left row exactly like `LeftOuter`, so
replicating the right side is safe for them: each left row still lands in one
task and still sees every right row for its key.
`canReplicateLeftSide` excludes them as well, so line 610 bails out with
"Skipping partially clustered distribution as it cannot be applied for join
type" and partially-clustered distribution is off entirely for two of the join
types this PR just taught the filter about.
Measured on `items` bucketed by `id` (0, 1, 4) against `purchases` bucketed
by `item_id` (4 three times, 5), with
`spark.sql.sources.v2.bucketing.partiallyClusteredDistribution.enabled=true`,
reading `distributePartitions` off both `GroupPartitionsExec`s:
```
LEFT OUTER true, false -> applied
EXISTS (...) OR i.name = 'bb' false, false -> skipped
correlated scalar subquery false, false -> skipped
```
`OptimizeSkewedJoin.canSplitLeftSide` at `OptimizeSkewedJoin.scala:85` holds
the identical list, so a fix probably wants to move both. Happy for this to be
a follow-up.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -564,6 +565,47 @@ class KeyGroupedPartitioningSuite
|""".stripMargin)
}
+ /**
+ * Creates `items` with ids 0, 1, 4 and `purchases` with item ids 4, 5, both
bucketed by those
+ * columns, so a join on them sees two left-only key groups, one shared and
one right-only, then
+ * runs `body` with partition filtering enabled.
+ */
+ private def withPartitionFilterJoinTables(body: => Unit): Unit = {
+ createTable(items, itemsColumns, Array(bucket(8, "id")))
+ sql(s"INSERT INTO testcat.ns.$items VALUES " +
+ s"(0, 'aa', 38.0, cast('2020-01-01' as timestamp)), " +
+ s"(1, 'bb', 39.0, cast('2020-01-02' as timestamp)), " +
+ s"(4, 'cc', 40.0, cast('2020-01-02' as timestamp))")
+
+ createTable(purchases, purchasesColumns, Array(bucket(8, "item_id")))
+ sql(s"INSERT INTO testcat.ns.$purchases VALUES " +
+ s"(4, 42.0, cast('2020-01-01' as timestamp)), " +
+ s"(5, 44.0, cast('2020-01-15' as timestamp))")
+
+ withSQLConf(SQLConf.V2_BUCKETING_PUSH_PART_VALUES_ENABLED.key -> "true",
+ SQLConf.V2_BUCKETING_PARTITION_FILTER_ENABLED.key -> "true") {
+ body
+ }
+ }
+
+ /**
+ * Asserts `df` plans exactly one shuffled join accepted by `isJoinType`,
that no shuffle was
+ * added under it, and that partition filtering left it `expectedNumGroups`
key groups.
+ */
+ private def checkPartitionFilteredJoin(
+ df: DataFrame,
+ isJoinType: JoinType => Boolean,
+ expectedNumGroups: Int): Unit = {
+ val plan = df.queryExecution.executedPlan
+ val joins = collect(plan) { case j: ShuffledJoin if isJoinType(j.joinType)
=> j }
+ assert(joins.size == 1, s"expected one matching join in\n$plan")
+ assert(collectAllShuffles(joins.head).isEmpty,
+ "should not add shuffle for both sides of the join")
+ val groupPartitions = collectAllGroupPartitions(joins.head)
+ assert(groupPartitions.nonEmpty, s"expected GroupPartitionsExec in\n$plan")
+ assert(groupPartitions.forall(_.outputPartitioning.numPartitions ==
expectedNumGroups))
Review Comment:
**Finding 4.** This is the assert all five new tests turn on, and it has no
message. I reverted the `EnsureRequirements` change and re-ran them, and each
one reports only
```
groupPartitions.forall(((x$10: GroupPartitionsExec) =>
x$10.outputPartitioning.numPartitions.==(expectedNumGroups))) was false
```
with neither the expected nor the actual count. The two asserts above it
both carry a message.
```suggestion
val actualNumGroups =
groupPartitions.map(_.outputPartitioning.numPartitions)
assert(actualNumGroups.forall(_ == expectedNumGroups),
s"expected $expectedNumGroups key groups, got
${actualNumGroups.mkString(", ")}")
```
--
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]
