peter-toth commented on code in PR #59916:
URL: https://github.com/apache/spark/pull/59916#discussion_r4446907036
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,42 +452,84 @@ object InjectRuntimeFilter extends Rule[LogicalPlan] with
PredicateHelper with J
private def tryInjectRuntimeFilter(plan: LogicalPlan): LogicalPlan = {
var filterCounter = 0
val numFilterThreshold =
conf.getConf(SQLConf.RUNTIME_FILTER_NUMBER_THRESHOLD)
+ // A key with a DPP filter gets no bloom filter, unless physical planning
drops that filter. It
+ // drops an `onlyInBroadcast` DPP filter when no broadcast join broadcasts
a plan that the
+ // filter could reuse. So for a key with a DPP filter, this rule:
+ // 1. Collects `reused`: the join sides that a broadcast join could
broadcast and a DPP filter
+ // could reuse. It can hold more plans than physical planning
broadcasts. Then a dropped DPP
+ // filter still keeps the bloom filter out.
+ // 2. Counts an `onlyInBroadcast` DPP filter as kept (`isKept`) if one of
its plans has the same
+ // result as a plan in `reused`. Any other DPP filter counts as kept.
+ // 3. Adds a bloom filter for a key whose DPP filters are all dropped only
if the bloom filter
+ // cannot land in a plan in `reused` (`goesIntoReused`). There it would
change the broadcast,
+ // and physical planning would drop the DPP filter that reuses it.
Pushdown moves the bloom
+ // filter along its key, down to the leaf the key comes from. So a
reused broadcast elsewhere
+ // in the join side does not matter, such as a broadcast dimension
joined to the same fact
+ // table. If the join side has a union, pushdown can copy the bloom
filter into every
+ // branch. Then a plan in `reused` anywhere in the join side, or around
it, is enough to
+ // skip it.
+ // Physical planning reuses a broadcast only with exchange reuse on.
+ lazy val reused = if (conf.exchangeReuseEnabled) {
+ val dppPlans = plan.collect { case Filter(dpp: DynamicPruningSubquery,
_) => dpp }
+ .flatMap(reusablePlans)
+ broadcastablePlans(plan).filter(b => dppPlans.exists(_.sameResult(b)))
+ } else {
+ Nil
+ }
+ def isKept(dpp: DynamicPruningSubquery): Boolean =
+ !dpp.onlyInBroadcast || reusablePlans(dpp).exists(p =>
reused.exists(_.sameResult(p)))
Review Comment:
Fixed in
[2420b70](https://github.com/apache/spark/pull/59916/commits/2420b70b2288e0cfced46a4f2d4eb293f15f40da).
For a key whose DPP filters are all dropped, the rule now adds the bloom
filter only if its creation side is deterministic. That covers both directions:
the DPP filter's build plan, and the pruned side when the bloom filter goes
onto the other side. New tests "do not add bloom filter if the dpp filter has a
non-deterministic build plan" and "do not add bloom filter whose creation side
is a non-deterministic pruned side". The same problem on keys without a DPP
filter is SPARK-60808.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,42 +452,84 @@ object InjectRuntimeFilter extends Rule[LogicalPlan] with
PredicateHelper with J
private def tryInjectRuntimeFilter(plan: LogicalPlan): LogicalPlan = {
var filterCounter = 0
val numFilterThreshold =
conf.getConf(SQLConf.RUNTIME_FILTER_NUMBER_THRESHOLD)
+ // A key with a DPP filter gets no bloom filter, unless physical planning
drops that filter. It
+ // drops an `onlyInBroadcast` DPP filter when no broadcast join broadcasts
a plan that the
+ // filter could reuse. So for a key with a DPP filter, this rule:
+ // 1. Collects `reused`: the join sides that a broadcast join could
broadcast and a DPP filter
+ // could reuse. It can hold more plans than physical planning
broadcasts. Then a dropped DPP
+ // filter still keeps the bloom filter out.
+ // 2. Counts an `onlyInBroadcast` DPP filter as kept (`isKept`) if one of
its plans has the same
+ // result as a plan in `reused`. Any other DPP filter counts as kept.
+ // 3. Adds a bloom filter for a key whose DPP filters are all dropped only
if the bloom filter
+ // cannot land in a plan in `reused` (`goesIntoReused`). There it would
change the broadcast,
+ // and physical planning would drop the DPP filter that reuses it.
Pushdown moves the bloom
+ // filter along its key, down to the leaf the key comes from. So a
reused broadcast elsewhere
+ // in the join side does not matter, such as a broadcast dimension
joined to the same fact
+ // table. If the join side has a union, pushdown can copy the bloom
filter into every
+ // branch. Then a plan in `reused` anywhere in the join side, or around
it, is enough to
+ // skip it.
+ // Physical planning reuses a broadcast only with exchange reuse on.
+ lazy val reused = if (conf.exchangeReuseEnabled) {
+ val dppPlans = plan.collect { case Filter(dpp: DynamicPruningSubquery,
_) => dpp }
+ .flatMap(reusablePlans)
+ broadcastablePlans(plan).filter(b => dppPlans.exists(_.sameResult(b)))
+ } else {
+ Nil
+ }
+ def isKept(dpp: DynamicPruningSubquery): Boolean =
+ !dpp.onlyInBroadcast || reusablePlans(dpp).exists(p =>
reused.exists(_.sameResult(p)))
+ def goesIntoReused(key: Expression, side: LogicalPlan): Boolean =
+ if (side.containsPattern(UNION)) {
+ reused.exists(r => side.exists(_.fastEquals(r)) ||
r.exists(_.fastEquals(side)))
+ } else {
+ findExpressionAndTrackLineageDown(key, side).exists { case (_, leaf) =>
+ reused.exists(_.exists(_.fastEquals(leaf)))
+ }
+ }
plan transformUp {
case join @ ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, _, _,
left, right, hint) =>
var newLeft = left
var newRight = right
leftKeys.lazyZip(rightKeys).foreach((l, r) => {
+ lazy val keyDpps = dynamicPruningSubqueries(left, right, l, r)
// Check if:
- // 1. There is already a DPP filter on the key
- // 2. The keys are simple cheap expressions
+ // 1. The keys are simple cheap expressions
+ // 2. There is already a DPP filter on the key that physical
planning may keep
if (filterCounter < numFilterThreshold &&
- !hasDynamicPruningSubquery(left, right, l, r) &&
- isSimpleExpression(l) && isSimpleExpression(r)) {
+ isSimpleExpression(l) && isSimpleExpression(r) &&
!keyDpps.exists(isKept)) {
Review Comment:
Your SPARK-60796 fixed `prunePartitions` in the meantime, so this PR needs
nothing for it.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,42 +452,84 @@ object InjectRuntimeFilter extends Rule[LogicalPlan] with
PredicateHelper with J
private def tryInjectRuntimeFilter(plan: LogicalPlan): LogicalPlan = {
var filterCounter = 0
val numFilterThreshold =
conf.getConf(SQLConf.RUNTIME_FILTER_NUMBER_THRESHOLD)
+ // A key with a DPP filter gets no bloom filter, unless physical planning
drops that filter. It
+ // drops an `onlyInBroadcast` DPP filter when no broadcast join broadcasts
a plan that the
+ // filter could reuse. So for a key with a DPP filter, this rule:
+ // 1. Collects `reused`: the join sides that a broadcast join could
broadcast and a DPP filter
+ // could reuse. It can hold more plans than physical planning
broadcasts. Then a dropped DPP
+ // filter still keeps the bloom filter out.
+ // 2. Counts an `onlyInBroadcast` DPP filter as kept (`isKept`) if one of
its plans has the same
+ // result as a plan in `reused`. Any other DPP filter counts as kept.
+ // 3. Adds a bloom filter for a key whose DPP filters are all dropped only
if the bloom filter
+ // cannot land in a plan in `reused` (`goesIntoReused`). There it would
change the broadcast,
+ // and physical planning would drop the DPP filter that reuses it.
Pushdown moves the bloom
+ // filter along its key, down to the leaf the key comes from. So a
reused broadcast elsewhere
+ // in the join side does not matter, such as a broadcast dimension
joined to the same fact
+ // table. If the join side has a union, pushdown can copy the bloom
filter into every
+ // branch. Then a plan in `reused` anywhere in the join side, or around
it, is enough to
+ // skip it.
+ // Physical planning reuses a broadcast only with exchange reuse on.
+ lazy val reused = if (conf.exchangeReuseEnabled) {
+ val dppPlans = plan.collect { case Filter(dpp: DynamicPruningSubquery,
_) => dpp }
+ .flatMap(reusablePlans)
+ broadcastablePlans(plan).filter(b => dppPlans.exists(_.sameResult(b)))
+ } else {
+ Nil
+ }
+ def isKept(dpp: DynamicPruningSubquery): Boolean =
+ !dpp.onlyInBroadcast || reusablePlans(dpp).exists(p =>
reused.exists(_.sameResult(p)))
+ def goesIntoReused(key: Expression, side: LogicalPlan): Boolean =
+ if (side.containsPattern(UNION)) {
+ reused.exists(r => side.exists(_.fastEquals(r)) ||
r.exists(_.fastEquals(side)))
+ } else {
+ findExpressionAndTrackLineageDown(key, side).exists { case (_, leaf) =>
+ reused.exists(_.exists(_.fastEquals(leaf)))
+ }
+ }
plan transformUp {
case join @ ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, _, _,
left, right, hint) =>
var newLeft = left
var newRight = right
leftKeys.lazyZip(rightKeys).foreach((l, r) => {
+ lazy val keyDpps = dynamicPruningSubqueries(left, right, l, r)
// Check if:
- // 1. There is already a DPP filter on the key
- // 2. The keys are simple cheap expressions
+ // 1. The keys are simple cheap expressions
+ // 2. There is already a DPP filter on the key that physical
planning may keep
if (filterCounter < numFilterThreshold &&
- !hasDynamicPruningSubquery(left, right, l, r) &&
- isSimpleExpression(l) && isSimpleExpression(r)) {
+ isSimpleExpression(l) && isSimpleExpression(r) &&
!keyDpps.exists(isKept)) {
val oldLeft = newLeft
val oldRight = newRight
+ // If the key has DPP filters, they are all dropped, so step 3
above applies.
+ val hasDroppedDpp = keyDpps.nonEmpty
+ // The creation side leaves out the key's own DPP filters.
Physical planning drops them,
+ // so in the bloom filter's subquery they would only add work.
+ def creationPlan(side: LogicalPlan) = withoutDppFilters(side,
keyDpps.contains)
// Check if:
// 1. The current join type supports prune the left side with
runtime filter
// 2. The current join is a shuffle join or a broadcast join that
// has a shuffle below it
// 3. There is no bloom filter on the left key yet
+ // 4. Without a dropped DPP filter, or the bloom filter would not
go into a broadcast
+ // that a DPP filter may reuse
val hasShuffle = isProbablyShuffleJoin(left, right, hint)
if (canPruneLeft(joinType) && (hasShuffle ||
probablyHasShuffle(left)) &&
- !hasBloomFilter(newLeft, l)) {
- extractBeneficialFilterCreatePlan(left, right, l, r).foreach {
creationSide =>
- newLeft = injectFilter(l, newLeft, creationSide)
- }
+ !hasBloomFilter(newLeft, l) && !(hasDroppedDpp &&
goesIntoReused(l, left))) {
+ extractBeneficialFilterCreatePlan(left, creationPlan(right), l,
r)
+ .foreach(creationSide => newLeft = injectFilter(l, newLeft,
creationSide))
Review Comment:
Fixed in
[2420b70](https://github.com/apache/spark/pull/59916/commits/2420b70b2288e0cfced46a4f2d4eb293f15f40da).
For a key whose DPP filters are all dropped, the rule now adds the bloom
filter only if the key on the application side is an attribute, at the join and
at every leaf it comes from, since pushdown rewrites the bloom filter through
aliases. New test "do not add bloom filter if the key of the dropped dpp filter
is not an attribute". The same failure with a kept DPP filter, or with a bloom
filter on a key without one, is SPARK-60809.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,42 +452,84 @@ object InjectRuntimeFilter extends Rule[LogicalPlan] with
PredicateHelper with J
private def tryInjectRuntimeFilter(plan: LogicalPlan): LogicalPlan = {
var filterCounter = 0
val numFilterThreshold =
conf.getConf(SQLConf.RUNTIME_FILTER_NUMBER_THRESHOLD)
+ // A key with a DPP filter gets no bloom filter, unless physical planning
drops that filter. It
+ // drops an `onlyInBroadcast` DPP filter when no broadcast join broadcasts
a plan that the
+ // filter could reuse. So for a key with a DPP filter, this rule:
+ // 1. Collects `reused`: the join sides that a broadcast join could
broadcast and a DPP filter
+ // could reuse. It can hold more plans than physical planning
broadcasts. Then a dropped DPP
+ // filter still keeps the bloom filter out.
+ // 2. Counts an `onlyInBroadcast` DPP filter as kept (`isKept`) if one of
its plans has the same
+ // result as a plan in `reused`. Any other DPP filter counts as kept.
+ // 3. Adds a bloom filter for a key whose DPP filters are all dropped only
if the bloom filter
+ // cannot land in a plan in `reused` (`goesIntoReused`). There it would
change the broadcast,
+ // and physical planning would drop the DPP filter that reuses it.
Pushdown moves the bloom
+ // filter along its key, down to the leaf the key comes from. So a
reused broadcast elsewhere
+ // in the join side does not matter, such as a broadcast dimension
joined to the same fact
+ // table. If the join side has a union, pushdown can copy the bloom
filter into every
+ // branch. Then a plan in `reused` anywhere in the join side, or around
it, is enough to
+ // skip it.
+ // Physical planning reuses a broadcast only with exchange reuse on.
+ lazy val reused = if (conf.exchangeReuseEnabled) {
+ val dppPlans = plan.collect { case Filter(dpp: DynamicPruningSubquery,
_) => dpp }
+ .flatMap(reusablePlans)
+ broadcastablePlans(plan).filter(b => dppPlans.exists(_.sameResult(b)))
+ } else {
+ Nil
+ }
+ def isKept(dpp: DynamicPruningSubquery): Boolean =
+ !dpp.onlyInBroadcast || reusablePlans(dpp).exists(p =>
reused.exists(_.sameResult(p)))
+ def goesIntoReused(key: Expression, side: LogicalPlan): Boolean =
+ if (side.containsPattern(UNION)) {
+ reused.exists(r => side.exists(_.fastEquals(r)) ||
r.exists(_.fastEquals(side)))
+ } else {
+ findExpressionAndTrackLineageDown(key, side).exists { case (_, leaf) =>
+ reused.exists(_.exists(_.fastEquals(leaf)))
+ }
+ }
plan transformUp {
case join @ ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, _, _,
left, right, hint) =>
var newLeft = left
var newRight = right
leftKeys.lazyZip(rightKeys).foreach((l, r) => {
+ lazy val keyDpps = dynamicPruningSubqueries(left, right, l, r)
// Check if:
- // 1. There is already a DPP filter on the key
- // 2. The keys are simple cheap expressions
+ // 1. The keys are simple cheap expressions
+ // 2. There is already a DPP filter on the key that physical
planning may keep
if (filterCounter < numFilterThreshold &&
- !hasDynamicPruningSubquery(left, right, l, r) &&
- isSimpleExpression(l) && isSimpleExpression(r)) {
+ isSimpleExpression(l) && isSimpleExpression(r) &&
!keyDpps.exists(isKept)) {
val oldLeft = newLeft
val oldRight = newRight
+ // If the key has DPP filters, they are all dropped, so step 3
above applies.
+ val hasDroppedDpp = keyDpps.nonEmpty
+ // The creation side leaves out the key's own DPP filters.
Physical planning drops them,
+ // so in the bloom filter's subquery they would only add work.
+ def creationPlan(side: LogicalPlan) = withoutDppFilters(side,
keyDpps.contains)
// Check if:
// 1. The current join type supports prune the left side with
runtime filter
// 2. The current join is a shuffle join or a broadcast join that
// has a shuffle below it
// 3. There is no bloom filter on the left key yet
+ // 4. Without a dropped DPP filter, or the bloom filter would not
go into a broadcast
+ // that a DPP filter may reuse
val hasShuffle = isProbablyShuffleJoin(left, right, hint)
if (canPruneLeft(joinType) && (hasShuffle ||
probablyHasShuffle(left)) &&
- !hasBloomFilter(newLeft, l)) {
- extractBeneficialFilterCreatePlan(left, right, l, r).foreach {
creationSide =>
- newLeft = injectFilter(l, newLeft, creationSide)
- }
+ !hasBloomFilter(newLeft, l) && !(hasDroppedDpp &&
goesIntoReused(l, left))) {
+ extractBeneficialFilterCreatePlan(left, creationPlan(right), l,
r)
+ .foreach(creationSide => newLeft = injectFilter(l, newLeft,
creationSide))
}
// Did we actually inject on the left? If not, try on the right
// Check if:
// 1. The current join type supports prune the right side with
runtime filter
// 2. The current join is a shuffle join or a broadcast join that
// has a shuffle below it
// 3. There is no bloom filter on the right key yet
+ // 4. Without a dropped DPP filter, or the bloom filter would not
go into a broadcast
+ // that a DPP filter may reuse
if (newLeft.fastEquals(oldLeft) && canPruneRight(joinType) &&
- (hasShuffle || probablyHasShuffle(right)) &&
!hasBloomFilter(newRight, r)) {
- extractBeneficialFilterCreatePlan(right, left, r, l).foreach {
creationSide =>
- newRight = injectFilter(r, newRight, creationSide)
- }
+ (hasShuffle || probablyHasShuffle(right)) &&
!hasBloomFilter(newRight, r) &&
+ !(hasDroppedDpp && goesIntoReused(r, right))) {
+ extractBeneficialFilterCreatePlan(right, creationPlan(left), r,
l)
+ .foreach(creationSide => newRight = injectFilter(r, newRight,
creationSide))
Review Comment:
Added to the Known limits.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,42 +452,84 @@ object InjectRuntimeFilter extends Rule[LogicalPlan] with
PredicateHelper with J
private def tryInjectRuntimeFilter(plan: LogicalPlan): LogicalPlan = {
var filterCounter = 0
val numFilterThreshold =
conf.getConf(SQLConf.RUNTIME_FILTER_NUMBER_THRESHOLD)
+ // A key with a DPP filter gets no bloom filter, unless physical planning
drops that filter. It
+ // drops an `onlyInBroadcast` DPP filter when no broadcast join broadcasts
a plan that the
+ // filter could reuse. So for a key with a DPP filter, this rule:
+ // 1. Collects `reused`: the join sides that a broadcast join could
broadcast and a DPP filter
+ // could reuse. It can hold more plans than physical planning
broadcasts. Then a dropped DPP
+ // filter still keeps the bloom filter out.
+ // 2. Counts an `onlyInBroadcast` DPP filter as kept (`isKept`) if one of
its plans has the same
+ // result as a plan in `reused`. Any other DPP filter counts as kept.
+ // 3. Adds a bloom filter for a key whose DPP filters are all dropped only
if the bloom filter
+ // cannot land in a plan in `reused` (`goesIntoReused`). There it would
change the broadcast,
+ // and physical planning would drop the DPP filter that reuses it.
Pushdown moves the bloom
+ // filter along its key, down to the leaf the key comes from. So a
reused broadcast elsewhere
+ // in the join side does not matter, such as a broadcast dimension
joined to the same fact
+ // table. If the join side has a union, pushdown can copy the bloom
filter into every
+ // branch. Then a plan in `reused` anywhere in the join side, or around
it, is enough to
+ // skip it.
+ // Physical planning reuses a broadcast only with exchange reuse on.
+ lazy val reused = if (conf.exchangeReuseEnabled) {
+ val dppPlans = plan.collect { case Filter(dpp: DynamicPruningSubquery,
_) => dpp }
+ .flatMap(reusablePlans)
+ broadcastablePlans(plan).filter(b => dppPlans.exists(_.sameResult(b)))
+ } else {
+ Nil
+ }
+ def isKept(dpp: DynamicPruningSubquery): Boolean =
+ !dpp.onlyInBroadcast || reusablePlans(dpp).exists(p =>
reused.exists(_.sameResult(p)))
+ def goesIntoReused(key: Expression, side: LogicalPlan): Boolean =
Review Comment:
Added to the Known limits.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -379,20 +379,61 @@ 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.
+ // The plans that a broadcast hash join of `plan` could broadcast. For an
equi-join, it is the
+ // side that `getBroadcastBuildSide` picks by a broadcast hint, or else by
size, as
+ // `JoinSelection` asks it. That ignores the other join hints, so the list
can hold more plans
+ // than physical planning broadcasts. It holds no fewer, unless the
statistics change after this
Review Comment:
Fixed in
[2420b70](https://github.com/apache/spark/pull/59916/commits/2420b70b2288e0cfced46a4f2d4eb293f15f40da),
thanks! In a subquery, every DPP filter now counts as kept. The rule runs
again on the whole plan once `RewritePredicateSubquery` has joined an IN or
EXISTS subquery with the rest of the query, and that run sees every broadcast.
Your query is a new test. A scalar subquery keeps master's behavior.
--
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]
