spark-test commented on code in PR #59916:
URL: https://github.com/apache/spark/pull/59916#discussion_r4416577000
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -415,12 +424,15 @@ object InjectRuntimeFilter extends Rule[LogicalPlan] with
PredicateHelper with J
case join @ ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, _, _,
left, right, hint) =>
var newLeft = left
var newRight = right
+ // The side the join is likely to broadcast, picked as `JoinSelection`
picks it
+ val broadcastSide = getBroadcastBuildSide(join, hintOnly = true, conf)
+ .orElse(getBroadcastBuildSide(join, hintOnly = false, conf))
Review Comment:
**Nit (P3):** This fallback is what keeps counting an `onlyInBroadcast` DPP
filter that reuses an automatic (unhinted) broadcast, which is the default way
such a filter finds its broadcast. No test in this change reaches it. Every new
`InjectRuntimeFilterSuite` case runs with `AUTO_BROADCASTJOIN_THRESHOLD=2000`,
where neither the filtered `bf2` (3409 bytes) nor the `bf5part` side fits, and
forces each broadcast with a hint. `StorageFilterPushDownSuite` runs with
broadcasts disabled. So removing this `.orElse(...)`, or moving it ahead of the
hint-only call, changes no assertion.
Without it, a key whose DPP filter reuses an automatic broadcast would also
get a redundant bloom subquery. With the calls swapped, sizes would override a
broadcast hint when the other side also fits the threshold. Results would not
change either way, but the default path is left unpinned.
See **Shared repair plan 1** in the review body.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -415,12 +424,15 @@ object InjectRuntimeFilter extends Rule[LogicalPlan] with
PredicateHelper with J
case join @ ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, _, _,
left, right, hint) =>
var newLeft = left
var newRight = right
+ // The side the join is likely to broadcast, picked as `JoinSelection`
picks it
Review Comment:
**Nit (P3):** This says the side is picked as `JoinSelection` picks it, but
the code only chains `getBroadcastBuildSide(join, hintOnly = true, conf)` and
the size-based `getBroadcastBuildSide(join, hintOnly = false, conf)`. After a
broadcast hint, `JoinSelection` tries a `SHUFFLE_MERGE` hint, a `SHUFFLE_HASH`
hint and `SHUFFLE_REPLICATE_NL` before it falls back to a size-based broadcast
(`SparkStrategies.scala:339-347`). It also never plans a broadcast hash join on
keys that `hashJoinSupported` rejects. The size-based call honors only the
`NO_BROADCAST_HASH` / `NO_BROADCAST_AND_REPLICATION` hints.
For example, take `select /*+ merge(bf2) */ * from (select f5, sum(a5) from
bf5part group by f5) t join bf2 on t.f5 = bf2.c2 where bf2.a2 = 62`, with the
filtered `bf2` under the broadcast threshold and `t` above it. Here
`broadcastSide` is `Some(BuildRight)`, so the `onlyInBroadcast` DPP filter
counts and `t.f5` gets no bloom filter. Physical planning picks a sort merge
join, finds no broadcast of `bf2` to reuse, and replaces the DPP filter with
`true`. The key ends up with no runtime filter, which is the outcome this PR
fixes for unhinted joins. Results are unchanged and this matches the merge
base, but the comment hides the gap.
The repo already models this precedence. `getBroadcastHashJoinBuildSide`
vetoes the size-based broadcast under a `SHUFFLE_HASH` hint that
`JoinSelection` would honor, and
`PushDownJoinThroughUnion.broadcastsRightForEveryBranch` adds the
`SHUFFLE_MERGE` / `SHUFFLE_REPLICATE_NL` veto. Reusing that logic here would
make the comment true. Otherwise the comment should say which hints are not
modeled.
See **Shared repair plan 1** in the review body.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -380,18 +380,27 @@ object InjectRuntimeFilter extends Rule[LogicalPlan] with
PredicateHelper with J
}
// This checks if there is already a DPP filter, as this rule is called just
after DPP.
+ // `broadcastSide` is the side the join is likely to broadcast. A DPP filter
that runs only by
+ // reusing a broadcast counts only if physical planning is likely to find
one, which needs both:
+ // - exchange reuse is on;
+ // - `broadcastSide` is the filter's build side, or the filter has a
broadcast value projection.
@tailrec
private def hasDynamicPruningSubquery(
left: LogicalPlan,
right: LogicalPlan,
leftKey: Expression,
- rightKey: Expression): Boolean = {
+ rightKey: Expression,
+ broadcastSide: Option[BuildSide]): Boolean = {
+ def isKept(dpp: DynamicPruningSubquery, buildSide: BuildSide): Boolean =
+ !dpp.onlyInBroadcast || (conf.exchangeReuseEnabled &&
+ (broadcastSide.contains(buildSide) ||
dpp.usableBroadcastValueProjection.isDefined))
(left, right) match {
- case (Filter(DynamicPruningSubquery(pruningKey, _, _, _, _, _, _),
plan), _) =>
- pruningKey.fastEquals(leftKey) || hasDynamicPruningSubquery(plan,
right, leftKey, rightKey)
- case (_, Filter(DynamicPruningSubquery(pruningKey, _, _, _, _, _, _),
plan)) =>
- pruningKey.fastEquals(rightKey) ||
- hasDynamicPruningSubquery(left, plan, leftKey, rightKey)
+ case (Filter(dpp: DynamicPruningSubquery, plan), _) =>
+ (dpp.pruningKey.fastEquals(leftKey) && isKept(dpp, BuildRight)) ||
+ hasDynamicPruningSubquery(plan, right, leftKey, rightKey,
broadcastSide)
+ case (_, Filter(dpp: DynamicPruningSubquery, plan)) =>
+ (dpp.pruningKey.fastEquals(rightKey) && isKept(dpp, BuildLeft)) ||
Review Comment:
**Nit (P3):** This arm maps a DPP filter on the right child to a broadcast
of the left side, the only side of this join whose broadcast the filter can
reuse. The shape is common: TPC-DS q42 lists `date_dim` before `store_sales`,
so its DPP filter on `ss_sold_date_sk` sits on the right child and reuses a
`BuildLeft` broadcast of `date_dim`. But no test in this change reaches the
arm. Every new and modified DPP case in `InjectRuntimeFilterSuite` puts
`bf5part`, or the aggregate over it, on the join's left side, and
`StorageFilterPushDownSuite` keeps the probe on the left. Changing this to
`isKept(dpp, BuildRight)`, or dropping the `isKept` check, changes no assertion
in these suites. A broken mapping would give right-child keys a redundant bloom
subquery, or no runtime filter at all. Results would not change.
**Verification:**
- **Behavior:** With the partitioned table on the right side of a shuffle
join and a selective filter on the left side, the right-side key gets the
runtime bloom filter, and its DPP filter is planned as true. The coverage fails
if the right-child arm counts the filter unconditionally.
- **Regression:** With the left side broadcast and reused by the right-child
DPP filter, and a shuffle below the pruned right side so that the same query
without DPP gets the bloom filter, no bloom filter is added. The coverage fails
if the right-child arm checks the wrong build side.
--
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]
