peter-toth commented on code in PR #59916:
URL: https://github.com/apache/spark/pull/59916#discussion_r4454921048
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,42 +439,101 @@ 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.
+ // 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.
So does every DPP
+ // filter in a subquery. The optimizer runs on a subquery on its own,
and an IN or EXISTS
+ // subquery is joined with the rest of the query only later. Then this
rule runs again, and
+ // sees all the broadcasts.
+ // 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, and into
every branch of a
+ // union. So a reused broadcast elsewhere in the join side does not
matter, such as a
+ // broadcast dimension joined to the same fact table. The key on the
application side must
+ // also be an attribute, at the join and at every leaf it comes from
(`isAttributeAtLeaves`).
+ // On a partition column, the bloom filter becomes a partition filter,
which evaluates the
+ // key on every partition before the data filters. So a key that can
fail could fail the
+ // query (SPARK-60809). And the creation side must be deterministic.
The bloom filter
+ // evaluates it again, apart from the join, so it could drop rows
(SPARK-60808).
+ // Physical planning reuses a broadcast only with exchange reuse on.
+ lazy val reused = if (conf.exchangeReuseEnabled) {
+ // After the optimizer ran on an IN subquery on its own, a DPP filter
there can be a conjunct
+ // of a larger filter.
+ val dppPlans = plan.flatMap {
+ case Filter(condition, _) =>
+ splitConjunctivePredicates(condition).collect { case dpp:
DynamicPruningSubquery => dpp }
+ case _ => Nil
+ }.flatMap(reusablePlans)
+ broadcastablePlans(plan).filter(b => dppPlans.exists(_.sameResult(b)))
+ } else {
+ Nil
+ }
+ val inSubquery = plan.isInstanceOf[Subquery]
+ def isKept(dpp: DynamicPruningSubquery): Boolean =
+ !dpp.onlyInBroadcast || inSubquery ||
+ reusablePlans(dpp).exists(p => reused.exists(_.sameResult(p)))
+ def goesIntoReused(key: Expression, side: LogicalPlan): Boolean =
+ findExpressionAndTrackAllLineagesDown(key, side).exists { case (_, leaf)
=>
+ reused.exists(_.exists(_.fastEquals(leaf)))
+ }
+ // Pushdown rewrites the bloom filter through the aliases on its way down,
so the key must be an
+ // attribute there too, not only at the join.
+ def isAttributeAtLeaves(key: Expression, side: LogicalPlan): Boolean =
+ key.isInstanceOf[Attribute] &&
findExpressionAndTrackAllLineagesDown(key, side).forall {
+ case (leafKey, _) => leafKey.isInstanceOf[Attribute]
+ }
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, l) ++
dynamicPruningSubqueries(right, 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, which
physical planning
+ // drops. An expression such as a UDF or a `LIKE` in their build
plan would also stop
+ // the search for the creation side, since `isSimpleExpression`
sees the plan of a
+ // subquery too.
+ def creationPlan(side: LogicalPlan) = withoutDppFilters(side,
keyDpps.contains)
+ def canTakeBloomFilter(key: Expression, side: LogicalPlan, other:
LogicalPlan) =
+ !hasDroppedDpp || (isAttributeAtLeaves(key, side) &&
+ creationPlan(other).deterministic && !goesIntoReused(key,
side))
Review Comment:
Fixed in
[a3b03fe](https://github.com/apache/spark/pull/59916/commits/a3b03fe2a9fa6d4073e138c844271597c2d9c6bd).
For a key whose DPP filters are all dropped, the rule now adds no bloom filter
if the creation side reads a CTE. New test "do not add bloom filter whose
creation side reads a cte", with exchange reuse off. A scratch run of your
query lost 23 to 34 rows at 2420b70, and none with bloom filters off.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,42 +439,101 @@ 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.
+ // 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.
So does every DPP
+ // filter in a subquery. The optimizer runs on a subquery on its own,
and an IN or EXISTS
+ // subquery is joined with the rest of the query only later. Then this
rule runs again, and
+ // sees all the broadcasts.
+ // 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, and into
every branch of a
+ // union. So a reused broadcast elsewhere in the join side does not
matter, such as a
+ // broadcast dimension joined to the same fact table. The key on the
application side must
+ // also be an attribute, at the join and at every leaf it comes from
(`isAttributeAtLeaves`).
+ // On a partition column, the bloom filter becomes a partition filter,
which evaluates the
+ // key on every partition before the data filters. So a key that can
fail could fail the
+ // query (SPARK-60809). And the creation side must be deterministic.
The bloom filter
+ // evaluates it again, apart from the join, so it could drop rows
(SPARK-60808).
+ // Physical planning reuses a broadcast only with exchange reuse on.
+ lazy val reused = if (conf.exchangeReuseEnabled) {
+ // After the optimizer ran on an IN subquery on its own, a DPP filter
there can be a conjunct
+ // of a larger filter.
+ val dppPlans = plan.flatMap {
+ case Filter(condition, _) =>
+ splitConjunctivePredicates(condition).collect { case dpp:
DynamicPruningSubquery => dpp }
+ case _ => Nil
+ }.flatMap(reusablePlans)
+ broadcastablePlans(plan).filter(b => dppPlans.exists(_.sameResult(b)))
+ } else {
+ Nil
+ }
+ val inSubquery = plan.isInstanceOf[Subquery]
+ def isKept(dpp: DynamicPruningSubquery): Boolean =
+ !dpp.onlyInBroadcast || inSubquery ||
+ reusablePlans(dpp).exists(p => reused.exists(_.sameResult(p)))
+ def goesIntoReused(key: Expression, side: LogicalPlan): Boolean =
+ findExpressionAndTrackAllLineagesDown(key, side).exists { case (_, leaf)
=>
+ reused.exists(_.exists(_.fastEquals(leaf)))
+ }
+ // Pushdown rewrites the bloom filter through the aliases on its way down,
so the key must be an
+ // attribute there too, not only at the join.
+ def isAttributeAtLeaves(key: Expression, side: LogicalPlan): Boolean =
+ key.isInstanceOf[Attribute] &&
findExpressionAndTrackAllLineagesDown(key, side).forall {
Review Comment:
Fixed in
[a3b03fe](https://github.com/apache/spark/pull/59916/commits/a3b03fe2a9fa6d4073e138c844271597c2d9c6bd).
`findExpressionAndTrackAllLineagesDown` has a `strict` mode now. It returns no
lineage when it cannot follow the key into a branch, and then the check fails.
Your query is the new test "do not add bloom filter whose key can fail in a
union branch the lineage does not reach". At 2420b70 it failed with
`CAST_INVALID_INPUT`.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,42 +439,101 @@ 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.
+ // 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.
So does every DPP
+ // filter in a subquery. The optimizer runs on a subquery on its own,
and an IN or EXISTS
+ // subquery is joined with the rest of the query only later. Then this
rule runs again, and
+ // sees all the broadcasts.
+ // 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, and into
every branch of a
+ // union. So a reused broadcast elsewhere in the join side does not
matter, such as a
+ // broadcast dimension joined to the same fact table. The key on the
application side must
+ // also be an attribute, at the join and at every leaf it comes from
(`isAttributeAtLeaves`).
+ // On a partition column, the bloom filter becomes a partition filter,
which evaluates the
+ // key on every partition before the data filters. So a key that can
fail could fail the
+ // query (SPARK-60809). And the creation side must be deterministic.
The bloom filter
+ // evaluates it again, apart from the join, so it could drop rows
(SPARK-60808).
+ // Physical planning reuses a broadcast only with exchange reuse on.
+ lazy val reused = if (conf.exchangeReuseEnabled) {
+ // After the optimizer ran on an IN subquery on its own, a DPP filter
there can be a conjunct
+ // of a larger filter.
+ val dppPlans = plan.flatMap {
+ case Filter(condition, _) =>
+ splitConjunctivePredicates(condition).collect { case dpp:
DynamicPruningSubquery => dpp }
+ case _ => Nil
+ }.flatMap(reusablePlans)
+ broadcastablePlans(plan).filter(b => dppPlans.exists(_.sameResult(b)))
+ } else {
+ Nil
+ }
+ val inSubquery = plan.isInstanceOf[Subquery]
+ def isKept(dpp: DynamicPruningSubquery): Boolean =
+ !dpp.onlyInBroadcast || inSubquery ||
+ reusablePlans(dpp).exists(p => reused.exists(_.sameResult(p)))
+ def goesIntoReused(key: Expression, side: LogicalPlan): Boolean =
+ findExpressionAndTrackAllLineagesDown(key, side).exists { case (_, leaf)
=>
+ reused.exists(_.exists(_.fastEquals(leaf)))
+ }
+ // Pushdown rewrites the bloom filter through the aliases on its way down,
so the key must be an
+ // attribute there too, not only at the join.
+ def isAttributeAtLeaves(key: Expression, side: LogicalPlan): Boolean =
+ key.isInstanceOf[Attribute] &&
findExpressionAndTrackAllLineagesDown(key, side).forall {
+ case (leafKey, _) => leafKey.isInstanceOf[Attribute]
+ }
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, l) ++
dynamicPruningSubqueries(right, 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, which
physical planning
+ // drops. An expression such as a UDF or a `LIKE` in their build
plan would also stop
+ // the search for the creation side, since `isSimpleExpression`
sees the plan of a
+ // subquery too.
+ def creationPlan(side: LogicalPlan) = withoutDppFilters(side,
keyDpps.contains)
+ def canTakeBloomFilter(key: Expression, side: LogicalPlan, other:
LogicalPlan) =
Review Comment:
Fixed in
[a3b03fe](https://github.com/apache/spark/pull/59916/commits/a3b03fe2a9fa6d4073e138c844271597c2d9c6bd).
The rule now adds no bloom filter if the creation side keeps an
`onlyInBroadcast` DPP filter that counts as kept. A DPP filter with a benefit
runs as its own subquery in the bloom filter's subquery too, so it does not
count. Your shape is the new test "do not add bloom filter whose creation side
keeps a dpp filter that physical planning keeps". At 2420b70 it failed with
`CAST_INVALID_INPUT`. The same happens on master for a key without a DPP
filter, which is SPARK-60842 now.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,42 +439,101 @@ 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.
+ // 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.
So does every DPP
+ // filter in a subquery. The optimizer runs on a subquery on its own,
and an IN or EXISTS
+ // subquery is joined with the rest of the query only later. Then this
rule runs again, and
+ // sees all the broadcasts.
+ // 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, and into
every branch of a
+ // union. So a reused broadcast elsewhere in the join side does not
matter, such as a
+ // broadcast dimension joined to the same fact table. The key on the
application side must
+ // also be an attribute, at the join and at every leaf it comes from
(`isAttributeAtLeaves`).
+ // On a partition column, the bloom filter becomes a partition filter,
which evaluates the
+ // key on every partition before the data filters. So a key that can
fail could fail the
+ // query (SPARK-60809). And the creation side must be deterministic.
The bloom filter
+ // evaluates it again, apart from the join, so it could drop rows
(SPARK-60808).
+ // Physical planning reuses a broadcast only with exchange reuse on.
+ lazy val reused = if (conf.exchangeReuseEnabled) {
+ // After the optimizer ran on an IN subquery on its own, a DPP filter
there can be a conjunct
+ // of a larger filter.
+ val dppPlans = plan.flatMap {
+ case Filter(condition, _) =>
+ splitConjunctivePredicates(condition).collect { case dpp:
DynamicPruningSubquery => dpp }
+ case _ => Nil
+ }.flatMap(reusablePlans)
+ broadcastablePlans(plan).filter(b => dppPlans.exists(_.sameResult(b)))
+ } else {
+ Nil
+ }
+ val inSubquery = plan.isInstanceOf[Subquery]
+ def isKept(dpp: DynamicPruningSubquery): Boolean =
+ !dpp.onlyInBroadcast || inSubquery ||
+ reusablePlans(dpp).exists(p => reused.exists(_.sameResult(p)))
+ def goesIntoReused(key: Expression, side: LogicalPlan): Boolean =
+ findExpressionAndTrackAllLineagesDown(key, side).exists { case (_, leaf)
=>
+ reused.exists(_.exists(_.fastEquals(leaf)))
+ }
+ // Pushdown rewrites the bloom filter through the aliases on its way down,
so the key must be an
+ // attribute there too, not only at the join.
+ def isAttributeAtLeaves(key: Expression, side: LogicalPlan): Boolean =
Review Comment:
Fixed in
[a3b03fe](https://github.com/apache/spark/pull/59916/commits/a3b03fe2a9fa6d4073e138c844271597c2d9c6bd).
A key is safe now if it is an attribute, a numeric up-cast of a safe key, or
the normalization of a floating point key. The cast check reuses
`Cast.canUpCast`, limited to numeric types as
`UnwrapCastInBinaryComparison.canUnwrapCast` limits it, so `DATE` to
`TIMESTAMP` stays out. New test "add bloom filter for a key with a dropped dpp
filter under a numeric up-cast".
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,42 +439,101 @@ 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.
+ // 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.
So does every DPP
+ // filter in a subquery. The optimizer runs on a subquery on its own,
and an IN or EXISTS
+ // subquery is joined with the rest of the query only later. Then this
rule runs again, and
+ // sees all the broadcasts.
+ // 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, and into
every branch of a
+ // union. So a reused broadcast elsewhere in the join side does not
matter, such as a
+ // broadcast dimension joined to the same fact table. The key on the
application side must
+ // also be an attribute, at the join and at every leaf it comes from
(`isAttributeAtLeaves`).
+ // On a partition column, the bloom filter becomes a partition filter,
which evaluates the
+ // key on every partition before the data filters. So a key that can
fail could fail the
+ // query (SPARK-60809). And the creation side must be deterministic.
The bloom filter
+ // evaluates it again, apart from the join, so it could drop rows
(SPARK-60808).
+ // Physical planning reuses a broadcast only with exchange reuse on.
+ lazy val reused = if (conf.exchangeReuseEnabled) {
+ // After the optimizer ran on an IN subquery on its own, a DPP filter
there can be a conjunct
+ // of a larger filter.
+ val dppPlans = plan.flatMap {
+ case Filter(condition, _) =>
+ splitConjunctivePredicates(condition).collect { case dpp:
DynamicPruningSubquery => dpp }
+ case _ => Nil
+ }.flatMap(reusablePlans)
+ broadcastablePlans(plan).filter(b => dppPlans.exists(_.sameResult(b)))
+ } else {
+ Nil
+ }
+ val inSubquery = plan.isInstanceOf[Subquery]
+ def isKept(dpp: DynamicPruningSubquery): Boolean =
+ !dpp.onlyInBroadcast || inSubquery ||
+ reusablePlans(dpp).exists(p => reused.exists(_.sameResult(p)))
+ def goesIntoReused(key: Expression, side: LogicalPlan): Boolean =
+ findExpressionAndTrackAllLineagesDown(key, side).exists { case (_, leaf)
=>
+ reused.exists(_.exists(_.fastEquals(leaf)))
+ }
+ // Pushdown rewrites the bloom filter through the aliases on its way down,
so the key must be an
+ // attribute there too, not only at the join.
+ def isAttributeAtLeaves(key: Expression, side: LogicalPlan): Boolean =
+ key.isInstanceOf[Attribute] &&
findExpressionAndTrackAllLineagesDown(key, side).forall {
+ case (leafKey, _) => leafKey.isInstanceOf[Attribute]
+ }
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, l) ++
dynamicPruningSubqueries(right, 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, which
physical planning
+ // drops. An expression such as a UDF or a `LIKE` in their build
plan would also stop
+ // the search for the creation side, since `isSimpleExpression`
sees the plan of a
+ // subquery too.
+ def creationPlan(side: LogicalPlan) = withoutDppFilters(side,
keyDpps.contains)
+ def canTakeBloomFilter(key: Expression, side: LogicalPlan, other:
LogicalPlan) =
+ !hasDroppedDpp || (isAttributeAtLeaves(key, side) &&
+ creationPlan(other).deterministic && !goesIntoReused(key,
side))
Review Comment:
Checking the bloom filter's `aggregate` would be exact, but it would need
the flag down in `injectBloomFilter`, and it still misses the CTE case. I kept
the check on the creation side, which errs on the safe side, and added it to
the Known limits.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -411,42 +439,101 @@ 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.
+ // 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.
So does every DPP
+ // filter in a subquery. The optimizer runs on a subquery on its own,
and an IN or EXISTS
+ // subquery is joined with the rest of the query only later. Then this
rule runs again, and
+ // sees all the broadcasts.
+ // 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, and into
every branch of a
+ // union. So a reused broadcast elsewhere in the join side does not
matter, such as a
+ // broadcast dimension joined to the same fact table. The key on the
application side must
+ // also be an attribute, at the join and at every leaf it comes from
(`isAttributeAtLeaves`).
+ // On a partition column, the bloom filter becomes a partition filter,
which evaluates the
+ // key on every partition before the data filters. So a key that can
fail could fail the
+ // query (SPARK-60809). And the creation side must be deterministic.
The bloom filter
+ // evaluates it again, apart from the join, so it could drop rows
(SPARK-60808).
+ // Physical planning reuses a broadcast only with exchange reuse on.
+ lazy val reused = if (conf.exchangeReuseEnabled) {
+ // After the optimizer ran on an IN subquery on its own, a DPP filter
there can be a conjunct
+ // of a larger filter.
+ val dppPlans = plan.flatMap {
+ case Filter(condition, _) =>
+ splitConjunctivePredicates(condition).collect { case dpp:
DynamicPruningSubquery => dpp }
+ case _ => Nil
+ }.flatMap(reusablePlans)
+ broadcastablePlans(plan).filter(b => dppPlans.exists(_.sameResult(b)))
+ } else {
+ Nil
+ }
+ val inSubquery = plan.isInstanceOf[Subquery]
+ def isKept(dpp: DynamicPruningSubquery): Boolean =
+ !dpp.onlyInBroadcast || inSubquery ||
+ reusablePlans(dpp).exists(p => reused.exists(_.sameResult(p)))
+ def goesIntoReused(key: Expression, side: LogicalPlan): Boolean =
+ findExpressionAndTrackAllLineagesDown(key, side).exists { case (_, leaf)
=>
+ reused.exists(_.exists(_.fastEquals(leaf)))
+ }
+ // Pushdown rewrites the bloom filter through the aliases on its way down,
so the key must be an
+ // attribute there too, not only at the join.
+ def isAttributeAtLeaves(key: Expression, side: LogicalPlan): Boolean =
+ key.isInstanceOf[Attribute] &&
findExpressionAndTrackAllLineagesDown(key, side).forall {
+ case (leafKey, _) => leafKey.isInstanceOf[Attribute]
+ }
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, l) ++
dynamicPruningSubqueries(right, 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, which
physical planning
+ // drops. An expression such as a UDF or a `LIKE` in their build
plan would also stop
+ // the search for the creation side, since `isSimpleExpression`
sees the plan of a
+ // subquery too.
+ def creationPlan(side: LogicalPlan) = withoutDppFilters(side,
keyDpps.contains)
Review Comment:
As you say, the outcome is never worse than master, and it needs a UDF or a
`LIKE` in the build plan, so I'd leave it.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -379,21 +377,51 @@ 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.
- @tailrec
- private def hasDynamicPruningSubquery(
- left: LogicalPlan,
- right: LogicalPlan,
- leftKey: Expression,
- rightKey: Expression): Boolean = {
- (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 _ => false
- }
+ // The plans that a broadcast hash join of `plan` could broadcast: the side
that
+ // `getBroadcastHashJoinBuildSide` picks. For an equi-join, that is the side
with a broadcast
+ // hint, or else a side small enough to broadcast that the join type can
build from, unless a
+ // `SHUFFLE_HASH` hint asks for a shuffle hash join. For a null-aware anti
join, it is the right
+ // side under the null-aware anti join thresholds. It 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 rule. It takes each side without the DPP
filters on top of it,
+ // since cleanup can still drop them.
+ private def broadcastablePlans(plan: LogicalPlan): Seq[LogicalPlan] =
plan.flatMap {
+ // AQE plans a DPP filter as `true` even if a null-aware anti join
broadcasts its build plan.
+ // Without AQE, physical planning keeps such a filter, though the filter
builds its own
+ // broadcast.
+ case ExtractSingleColumnNullAwareAntiJoin(_, _) if
conf.adaptiveExecutionEnabled => Nil
+ case join: Join =>
+ getBroadcastHashJoinBuildSide(join, conf).map {
+ case BuildLeft => withoutDppFilters(join.left)
+ case BuildRight => withoutDppFilters(join.right)
+ }
+ case _ => Nil
+ }
+
+ // `plan` without the DPP filters on top of it that `drop` picks, by default
all of them.
+ private def withoutDppFilters(
+ plan: LogicalPlan,
+ drop: DynamicPruningSubquery => Boolean = _ => true): LogicalPlan = plan
match {
+ case f @ Filter(dpp: DynamicPruningSubquery, child) =>
+ val newChild = withoutDppFilters(child, drop)
+ if (drop(dpp)) newChild else f.withNewChildren(Seq(newChild))
+ case _ => plan
+ }
+
+ // The plans whose broadcast a DPP filter may reuse: its build plan, and for
an `onlyInBroadcast`
+ // filter, the source plan of its broadcast value projection.
+ private def reusablePlans(dpp: DynamicPruningSubquery): Seq[LogicalPlan] =
+ dpp.buildQuery +: dpp.usableBroadcastValueProjection.filter(_ =>
dpp.onlyInBroadcast)
Review Comment:
Added to the Known limits.
--
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]
