pan3793 commented on code in PR #58514:
URL: https://github.com/apache/spark/pull/58514#discussion_r3941226015
##########
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: 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.
joinType match {
- case Inner =>
+ // neither side keeps unmatched rows
+ case _: InnerLike | LeftSemi =>
mergeAndDedupPartitionKeys(leftPartitionKeys, rightPartitionKeys,
intersect = true)
Review Comment:
Pre-existing: the same shape already fails for an inner leg on this PR's
base, and SPARK-59176 (3c70da502c9, merged after this branch was cut) leaves a
reduced side with no key out of the comparison regardless of join type. Merged
master and extended the SPARK-59176 test with a semi-join leg: aba3170b537.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -851,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.
Review Comment:
Applied in aba3170b537, keeping a short pointer to `ExtractEquiJoinKeys` for
the reachability finding above.
##########
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 0d594001b89.
##########
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 0d594001b89.
##########
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 0d594001b89.
--
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]
