pan3793 commented on code in PR #58514:
URL: https://github.com/apache/spark/pull/58514#discussion_r3934187734
##########
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:
Agreed, a follow-up for that sounds good.
##########
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:
Added in 224c78b252c.
##########
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:
Applied in 224c78b252c.
##########
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:
Applied in 224c78b252c.
--
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]
