github-actions[bot] commented on code in PR #68340:
URL: https://github.com/apache/doris/pull/68340#discussion_r4069594073
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/RuntimeFilterGenerator.java:
##########
@@ -111,136 +115,164 @@ public Plan processRoot(Plan plan, CascadesContext ctx)
{
}
Plan result = plan.accept(this, ctx);
- // try to push rf inside CTEProducer
- // collect cteProducers
+ pushRuntimeFiltersIntoCTEProducer(plan, ctx);
+ return result;
+ }
+
+ /**
+ * Push the runtime filters of the consumers of a CTE into its producer,
where one filter takes the
+ * place of the identical filters of all the consumers. See {@link
#selectPushableRuntimeFilters}.
+ */
+ private void pushRuntimeFiltersIntoCTEProducer(Plan plan, CascadesContext
ctx) {
RuntimeFilterContext rfCtx = ctx.getRuntimeFilterContext();
Map<CTEId, PhysicalCTEProducer> cteProducerMap =
plan.collect(PhysicalCTEProducer.class::isInstance)
.stream().collect(Collectors.toMap(p -> ((PhysicalCTEProducer)
p).getCteId(),
p -> (PhysicalCTEProducer) p));
- // collect cteConsumers which are RF targets
+ // collect the cte consumers that are runtime filter targets, grouped
by the cte they read
Map<CTEId, Set<PhysicalCTEConsumer>> cteIdToConsumersWithRF =
Maps.newHashMap();
Map<PhysicalCTEConsumer, Set<RuntimeFilter>> consumerToRFs =
Maps.newHashMap();
- Map<PhysicalCTEConsumer, Set<Expression>> consumerToSrcExpression =
Maps.newHashMap();
- List<RuntimeFilter> allRFs = rfCtx.getNereidsRuntimeFilter();
- for (RuntimeFilter rf : allRFs) {
- PhysicalRelation rel = rf.getTargetScan();
- if (rel instanceof PhysicalCTEConsumer) {
- PhysicalCTEConsumer consumer = (PhysicalCTEConsumer) rel;
- CTEId cteId = consumer.getCteId();
- cteIdToConsumersWithRF.computeIfAbsent(cteId, key ->
Sets.newHashSet()).add(consumer);
+ for (RuntimeFilter rf : rfCtx.getNereidsRuntimeFilter()) {
+ PhysicalRelation target = rf.getTargetScan();
+ if (target instanceof PhysicalCTEConsumer) {
+ PhysicalCTEConsumer consumer = (PhysicalCTEConsumer) target;
+ cteIdToConsumersWithRF.computeIfAbsent(consumer.getCteId(),
key -> Sets.newHashSet()).add(consumer);
consumerToRFs.computeIfAbsent(consumer, key ->
Sets.newHashSet()).add(rf);
- consumerToSrcExpression.computeIfAbsent(consumer, key ->
Sets.newHashSet())
- .add(rf.getSrcExpr());
- }
- }
- for (CTEId cteId : cteIdToConsumersWithRF.keySet()) {
- // if any consumer does not have RF, RF cannot be pushed down.
- // cteIdToConsumersWithRF.get(cteId).size() can not be 1, o.w.
this cte will be inlined.
- if (ctx.getCteIdToConsumers().get(cteId).size() ==
cteIdToConsumersWithRF.get(cteId).size()
- && cteIdToConsumersWithRF.get(cteId).size() >= 2) {
- // check if there is a common srcExpr among all the consumers
- Set<PhysicalCTEConsumer> consumers =
cteIdToConsumersWithRF.get(cteId);
- PhysicalCTEConsumer consumer0 = consumers.iterator().next();
- Set<Expression> candidateSrcExpressions =
consumerToSrcExpression.get(consumer0);
- for (PhysicalCTEConsumer currentConsumer : consumers) {
- Set<Expression> srcExpressionsOnCurrentConsumer =
consumerToSrcExpression.get(currentConsumer);
-
candidateSrcExpressions.retainAll(srcExpressionsOnCurrentConsumer);
- if (candidateSrcExpressions.isEmpty()) {
- break;
- }
- }
- if (!candidateSrcExpressions.isEmpty()) {
- // find RFs to push down
- for (Expression srcExpr : candidateSrcExpressions) {
- List<RuntimeFilter> rfsToPushDown =
Lists.newArrayList();
- for (PhysicalCTEConsumer consumer :
cteIdToConsumersWithRF.get(cteId)) {
- for (RuntimeFilter rf :
consumerToRFs.get(consumer)) {
- if (rf.getSrcExpr().equals(srcExpr)) {
- rfsToPushDown.add(rf);
- }
- }
- }
- if (rfsToPushDown.isEmpty()) {
- break;
- }
- if
(!canPushDownRuntimeFiltersIntoCTEProducer(rfsToPushDown, cteId)) {
- continue;
- }
+ }
+ }
+ for (Map.Entry<CTEId, Set<PhysicalCTEConsumer>> cteAndConsumers :
cteIdToConsumersWithRF.entrySet()) {
+ pushRuntimeFiltersIntoCTEProducer(cteAndConsumers.getKey(),
cteAndConsumers.getValue(),
+ consumerToRFs, ctx, rfCtx,
cteProducerMap.get(cteAndConsumers.getKey()));
+ }
+ }
- // the most right deep buildNode from rfsToPushDown is
used as buildNode for pushDown rf
- // since the srcExpr are the same, all buildNodes of
rfToPushDown are in the same tree path
- // the longest ancestors means its corresponding rf
build node is the most right deep one.
- List<RuntimeFilter> rightDeepRfs =
Lists.newArrayList();
- List<Plan> rightDeepAncestors =
rfsToPushDown.get(0).getBuilderNode().getAncestors();
- int rightDeepAncestorsSize = rightDeepAncestors.size();
- RuntimeFilter leftTop = rfsToPushDown.get(0);
- int leftTopAncestorsSize = rightDeepAncestorsSize;
- for (RuntimeFilter rf : rfsToPushDown) {
- List<Plan> ancestors =
rf.getBuilderNode().getAncestors();
- int currentAncestorsSize = ancestors.size();
- if (currentAncestorsSize >=
rightDeepAncestorsSize) {
- if (currentAncestorsSize ==
rightDeepAncestorsSize) {
- rightDeepRfs.add(rf);
- } else {
- rightDeepAncestorsSize =
currentAncestorsSize;
- rightDeepAncestors = ancestors;
- rightDeepRfs.clear();
- rightDeepRfs.add(rf);
- }
- }
- if (currentAncestorsSize < leftTopAncestorsSize) {
- leftTopAncestorsSize = currentAncestorsSize;
- leftTop = rf;
- }
- }
-
Preconditions.checkArgument(rightDeepAncestors.contains(leftTop.getBuilderNode()));
- // check nodes between right deep and left top are SPJ
and not denied join and not mark join
- boolean valid = true;
- for (Plan cursor : rightDeepAncestors) {
- if (cursor.equals(leftTop.getBuilderNode())) {
- break;
- }
- // valid = valid &&
SPJ_PLAN.contains(cursor.getClass());
- if (cursor instanceof AbstractPhysicalJoin) {
- AbstractPhysicalJoin cursorJoin =
(AbstractPhysicalJoin) cursor;
- valid =
(!RuntimeFilterGenerator.DENIED_JOIN_TYPES
- .contains(cursorJoin.getJoinType())
- || cursorJoin.isMarkJoin()) && valid;
- }
- if (!valid) {
- break;
- }
- }
+ /**
+ * Push the runtime filters of the consumers of one CTE into the producer
of that CTE.
+ */
+ private void pushRuntimeFiltersIntoCTEProducer(CTEId cteId,
Set<PhysicalCTEConsumer> consumers,
+ Map<PhysicalCTEConsumer, Set<RuntimeFilter>> consumerToRFs,
CascadesContext ctx,
+ RuntimeFilterContext rfCtx, PhysicalCTEProducer cteProducer) {
+ // if any consumer of this cte does not have a runtime filter, none of
them can be pushed down.
+ // there are always at least two consumers, otherwise this cte would
have been inlined.
+ if (consumers.size() < 2 ||
ctx.getCteIdToConsumers().get(cteId).size() != consumers.size()) {
+ return;
+ }
+ for (Expression srcExpr : commonSrcExpressions(consumers,
consumerToRFs)) {
+ List<RuntimeFilter> rfsOfSrcExpr =
runtimeFiltersOfSrcExpression(consumers, consumerToRFs, srcExpr);
+ for (List<RuntimeFilter> rfsOfIdentity :
selectPushableRuntimeFilters(rfsOfSrcExpr, consumers, cteId)) {
+ pushDownIdenticalFilters(rfsOfIdentity, cteId, rfCtx,
cteProducer);
+ }
+ }
+ }
- if (!valid) {
- break;
- }
+ /**
+ * The source expressions that every one of the given consumers has a
runtime filter for.
+ */
+ private static Set<Expression>
commonSrcExpressions(Set<PhysicalCTEConsumer> consumers,
+ Map<PhysicalCTEConsumer, Set<RuntimeFilter>> consumerToRFs) {
+ Iterator<PhysicalCTEConsumer> iterator = consumers.iterator();
+ Set<Expression> commonSrcExpressions =
srcExpressionsOf(consumerToRFs.get(iterator.next()));
+ while (iterator.hasNext() && !commonSrcExpressions.isEmpty()) {
+
commonSrcExpressions.retainAll(srcExpressionsOf(consumerToRFs.get(iterator.next())));
+ }
+ return commonSrcExpressions;
+ }
- for (RuntimeFilter rfToPush : rightDeepRfs) {
- Expression rightDeepTargetExpressionOnCTE = null;
- PhysicalRelation rel = rfToPush.getTargetScan();
- if (rel instanceof PhysicalCTEConsumer
- && ((PhysicalCTEConsumer)
rel).getCteId().equals(cteId)) {
- rightDeepTargetExpressionOnCTE =
rfToPush.getTargetExpression();
- }
-
- boolean pushedDown =
doPushDownIntoCTEProducerInternal(
- rfToPush,
- rightDeepTargetExpressionOnCTE,
- rfCtx,
- cteProducerMap.get(cteId)
- );
- if (pushedDown) {
- rfCtx.removeFilter(
- rfToPush,
-
rightDeepTargetExpressionOnCTE.getInputSlotExprIds().iterator().next());
- }
- }
- }
+ private static Set<Expression> srcExpressionsOf(Set<RuntimeFilter> rfs) {
+ return
rfs.stream().map(RuntimeFilter::getSrcExpr).collect(Collectors.toSet());
+ }
+
+ /**
+ * The runtime filters that all the given consumers have for one source
expression.
+ */
+ private static List<RuntimeFilter>
runtimeFiltersOfSrcExpression(Set<PhysicalCTEConsumer> consumers,
+ Map<PhysicalCTEConsumer, Set<RuntimeFilter>> consumerToRFs,
Expression srcExpr) {
+ List<RuntimeFilter> rfsOfSrcExpr = Lists.newArrayList();
+ for (PhysicalCTEConsumer consumer : consumers) {
+ for (RuntimeFilter rf : consumerToRFs.get(consumer)) {
+ if (rf.getSrcExpr().equals(srcExpr)) {
+ rfsOfSrcExpr.add(rf);
}
}
}
- return result;
+ Preconditions.checkArgument(!rfsOfSrcExpr.isEmpty());
+ return rfsOfSrcExpr;
+ }
+
+ /**
+ * Push one group of identical runtime filters into the shared CTE
producer. Only called with a group
+ * that every consumer applies, see {@link #selectPushableRuntimeFilters}.
+ */
+ private void pushDownIdenticalFilters(List<RuntimeFilter> rfsOfIdentity,
CTEId cteId,
+ RuntimeFilterContext rfCtx, PhysicalCTEProducer cteProducer) {
+ // the most right deep buildNode from rfsOfIdentity is used as
buildNode for pushDown rf
+ // since the srcExpr are the same, all buildNodes of rfsOfIdentity are
in the same tree path
+ // the longest ancestors means its corresponding rf build node is the
most right deep one.
+ List<RuntimeFilter> rightDeepRfs = Lists.newArrayList();
+ List<Plan> rightDeepAncestors =
rfsOfIdentity.get(0).getBuilderNode().getAncestors();
+ int rightDeepAncestorsSize = rightDeepAncestors.size();
+ RuntimeFilter leftTop = rfsOfIdentity.get(0);
+ int leftTopAncestorsSize = rightDeepAncestorsSize;
+ for (RuntimeFilter rf : rfsOfIdentity) {
+ List<Plan> ancestors = rf.getBuilderNode().getAncestors();
+ int currentAncestorsSize = ancestors.size();
+ if (currentAncestorsSize >= rightDeepAncestorsSize) {
+ if (currentAncestorsSize == rightDeepAncestorsSize) {
+ rightDeepRfs.add(rf);
+ } else {
+ rightDeepAncestorsSize = currentAncestorsSize;
+ rightDeepAncestors = ancestors;
+ rightDeepRfs.clear();
+ rightDeepRfs.add(rf);
+ }
+ }
+ if (currentAncestorsSize < leftTopAncestorsSize) {
+ leftTopAncestorsSize = currentAncestorsSize;
+ leftTop = rf;
+ }
+ }
+
Preconditions.checkArgument(rightDeepAncestors.contains(leftTop.getBuilderNode()));
+ // check nodes between right deep and left top are SPJ and not denied
join and not mark join
+ boolean valid = true;
+ for (Plan cursor : rightDeepAncestors) {
+ if (cursor.equals(leftTop.getBuilderNode())) {
+ break;
+ }
+ // valid = valid && SPJ_PLAN.contains(cursor.getClass());
Review Comment:
[P1] Reject value-synthesizing nodes in the builder path. A reduced
reachable tree is:
```text
Jupper [c1.k <=> g.x] build values {3, NULL}
CTEConsumer c1
HashAggregate(group by x)
Repeat(grouping sets ((x), ())) synthesizes x=NULL
Jdeep [c2.k <=> b.x] build values {3}
```
Repeat/Aggregate preserve `x`'s ExprId, so both RFs have the same source,
producer target, type, direction, and `nullAware=true`; this code chooses
`Jdeep`. Because the loop only examines joins, it accepts Repeat/Aggregate
despite the SPJ comment. BE only preserves probe NULL when the chosen build
filter actually contains NULL, so the producer RF from `{3}` removes `t.k=NULL`
before the upper `<=>` can match the synthesized NULL. Please enforce the
intended all-node SPJ check (or otherwise reject value-synthesizing ancestors)
and add this shared-CTE/grouping-sets regression.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/RuntimeFilterGenerator.java:
##########
@@ -111,136 +115,164 @@ public Plan processRoot(Plan plan, CascadesContext ctx)
{
}
Plan result = plan.accept(this, ctx);
- // try to push rf inside CTEProducer
- // collect cteProducers
+ pushRuntimeFiltersIntoCTEProducer(plan, ctx);
+ return result;
+ }
+
+ /**
+ * Push the runtime filters of the consumers of a CTE into its producer,
where one filter takes the
+ * place of the identical filters of all the consumers. See {@link
#selectPushableRuntimeFilters}.
+ */
+ private void pushRuntimeFiltersIntoCTEProducer(Plan plan, CascadesContext
ctx) {
RuntimeFilterContext rfCtx = ctx.getRuntimeFilterContext();
Map<CTEId, PhysicalCTEProducer> cteProducerMap =
plan.collect(PhysicalCTEProducer.class::isInstance)
.stream().collect(Collectors.toMap(p -> ((PhysicalCTEProducer)
p).getCteId(),
p -> (PhysicalCTEProducer) p));
- // collect cteConsumers which are RF targets
+ // collect the cte consumers that are runtime filter targets, grouped
by the cte they read
Map<CTEId, Set<PhysicalCTEConsumer>> cteIdToConsumersWithRF =
Maps.newHashMap();
Map<PhysicalCTEConsumer, Set<RuntimeFilter>> consumerToRFs =
Maps.newHashMap();
- Map<PhysicalCTEConsumer, Set<Expression>> consumerToSrcExpression =
Maps.newHashMap();
- List<RuntimeFilter> allRFs = rfCtx.getNereidsRuntimeFilter();
- for (RuntimeFilter rf : allRFs) {
- PhysicalRelation rel = rf.getTargetScan();
- if (rel instanceof PhysicalCTEConsumer) {
- PhysicalCTEConsumer consumer = (PhysicalCTEConsumer) rel;
- CTEId cteId = consumer.getCteId();
- cteIdToConsumersWithRF.computeIfAbsent(cteId, key ->
Sets.newHashSet()).add(consumer);
+ for (RuntimeFilter rf : rfCtx.getNereidsRuntimeFilter()) {
+ PhysicalRelation target = rf.getTargetScan();
+ if (target instanceof PhysicalCTEConsumer) {
+ PhysicalCTEConsumer consumer = (PhysicalCTEConsumer) target;
+ cteIdToConsumersWithRF.computeIfAbsent(consumer.getCteId(),
key -> Sets.newHashSet()).add(consumer);
consumerToRFs.computeIfAbsent(consumer, key ->
Sets.newHashSet()).add(rf);
- consumerToSrcExpression.computeIfAbsent(consumer, key ->
Sets.newHashSet())
- .add(rf.getSrcExpr());
- }
- }
- for (CTEId cteId : cteIdToConsumersWithRF.keySet()) {
- // if any consumer does not have RF, RF cannot be pushed down.
- // cteIdToConsumersWithRF.get(cteId).size() can not be 1, o.w.
this cte will be inlined.
- if (ctx.getCteIdToConsumers().get(cteId).size() ==
cteIdToConsumersWithRF.get(cteId).size()
- && cteIdToConsumersWithRF.get(cteId).size() >= 2) {
- // check if there is a common srcExpr among all the consumers
- Set<PhysicalCTEConsumer> consumers =
cteIdToConsumersWithRF.get(cteId);
- PhysicalCTEConsumer consumer0 = consumers.iterator().next();
- Set<Expression> candidateSrcExpressions =
consumerToSrcExpression.get(consumer0);
- for (PhysicalCTEConsumer currentConsumer : consumers) {
- Set<Expression> srcExpressionsOnCurrentConsumer =
consumerToSrcExpression.get(currentConsumer);
-
candidateSrcExpressions.retainAll(srcExpressionsOnCurrentConsumer);
- if (candidateSrcExpressions.isEmpty()) {
- break;
- }
- }
- if (!candidateSrcExpressions.isEmpty()) {
- // find RFs to push down
- for (Expression srcExpr : candidateSrcExpressions) {
- List<RuntimeFilter> rfsToPushDown =
Lists.newArrayList();
- for (PhysicalCTEConsumer consumer :
cteIdToConsumersWithRF.get(cteId)) {
- for (RuntimeFilter rf :
consumerToRFs.get(consumer)) {
- if (rf.getSrcExpr().equals(srcExpr)) {
- rfsToPushDown.add(rf);
- }
- }
- }
- if (rfsToPushDown.isEmpty()) {
- break;
- }
- if
(!canPushDownRuntimeFiltersIntoCTEProducer(rfsToPushDown, cteId)) {
- continue;
- }
+ }
+ }
+ for (Map.Entry<CTEId, Set<PhysicalCTEConsumer>> cteAndConsumers :
cteIdToConsumersWithRF.entrySet()) {
+ pushRuntimeFiltersIntoCTEProducer(cteAndConsumers.getKey(),
cteAndConsumers.getValue(),
+ consumerToRFs, ctx, rfCtx,
cteProducerMap.get(cteAndConsumers.getKey()));
+ }
+ }
- // the most right deep buildNode from rfsToPushDown is
used as buildNode for pushDown rf
- // since the srcExpr are the same, all buildNodes of
rfToPushDown are in the same tree path
- // the longest ancestors means its corresponding rf
build node is the most right deep one.
- List<RuntimeFilter> rightDeepRfs =
Lists.newArrayList();
- List<Plan> rightDeepAncestors =
rfsToPushDown.get(0).getBuilderNode().getAncestors();
- int rightDeepAncestorsSize = rightDeepAncestors.size();
- RuntimeFilter leftTop = rfsToPushDown.get(0);
- int leftTopAncestorsSize = rightDeepAncestorsSize;
- for (RuntimeFilter rf : rfsToPushDown) {
- List<Plan> ancestors =
rf.getBuilderNode().getAncestors();
- int currentAncestorsSize = ancestors.size();
- if (currentAncestorsSize >=
rightDeepAncestorsSize) {
- if (currentAncestorsSize ==
rightDeepAncestorsSize) {
- rightDeepRfs.add(rf);
- } else {
- rightDeepAncestorsSize =
currentAncestorsSize;
- rightDeepAncestors = ancestors;
- rightDeepRfs.clear();
- rightDeepRfs.add(rf);
- }
- }
- if (currentAncestorsSize < leftTopAncestorsSize) {
- leftTopAncestorsSize = currentAncestorsSize;
- leftTop = rf;
- }
- }
-
Preconditions.checkArgument(rightDeepAncestors.contains(leftTop.getBuilderNode()));
- // check nodes between right deep and left top are SPJ
and not denied join and not mark join
- boolean valid = true;
- for (Plan cursor : rightDeepAncestors) {
- if (cursor.equals(leftTop.getBuilderNode())) {
- break;
- }
- // valid = valid &&
SPJ_PLAN.contains(cursor.getClass());
- if (cursor instanceof AbstractPhysicalJoin) {
- AbstractPhysicalJoin cursorJoin =
(AbstractPhysicalJoin) cursor;
- valid =
(!RuntimeFilterGenerator.DENIED_JOIN_TYPES
- .contains(cursorJoin.getJoinType())
- || cursorJoin.isMarkJoin()) && valid;
- }
- if (!valid) {
- break;
- }
- }
+ /**
+ * Push the runtime filters of the consumers of one CTE into the producer
of that CTE.
+ */
+ private void pushRuntimeFiltersIntoCTEProducer(CTEId cteId,
Set<PhysicalCTEConsumer> consumers,
+ Map<PhysicalCTEConsumer, Set<RuntimeFilter>> consumerToRFs,
CascadesContext ctx,
+ RuntimeFilterContext rfCtx, PhysicalCTEProducer cteProducer) {
+ // if any consumer of this cte does not have a runtime filter, none of
them can be pushed down.
+ // there are always at least two consumers, otherwise this cte would
have been inlined.
+ if (consumers.size() < 2 ||
ctx.getCteIdToConsumers().get(cteId).size() != consumers.size()) {
+ return;
+ }
+ for (Expression srcExpr : commonSrcExpressions(consumers,
consumerToRFs)) {
+ List<RuntimeFilter> rfsOfSrcExpr =
runtimeFiltersOfSrcExpression(consumers, consumerToRFs, srcExpr);
+ for (List<RuntimeFilter> rfsOfIdentity :
selectPushableRuntimeFilters(rfsOfSrcExpr, consumers, cteId)) {
+ pushDownIdenticalFilters(rfsOfIdentity, cteId, rfCtx,
cteProducer);
+ }
+ }
+ }
- if (!valid) {
- break;
- }
+ /**
+ * The source expressions that every one of the given consumers has a
runtime filter for.
+ */
+ private static Set<Expression>
commonSrcExpressions(Set<PhysicalCTEConsumer> consumers,
+ Map<PhysicalCTEConsumer, Set<RuntimeFilter>> consumerToRFs) {
+ Iterator<PhysicalCTEConsumer> iterator = consumers.iterator();
+ Set<Expression> commonSrcExpressions =
srcExpressionsOf(consumerToRFs.get(iterator.next()));
+ while (iterator.hasNext() && !commonSrcExpressions.isEmpty()) {
+
commonSrcExpressions.retainAll(srcExpressionsOf(consumerToRFs.get(iterator.next())));
+ }
+ return commonSrcExpressions;
+ }
- for (RuntimeFilter rfToPush : rightDeepRfs) {
- Expression rightDeepTargetExpressionOnCTE = null;
- PhysicalRelation rel = rfToPush.getTargetScan();
- if (rel instanceof PhysicalCTEConsumer
- && ((PhysicalCTEConsumer)
rel).getCteId().equals(cteId)) {
- rightDeepTargetExpressionOnCTE =
rfToPush.getTargetExpression();
- }
-
- boolean pushedDown =
doPushDownIntoCTEProducerInternal(
- rfToPush,
- rightDeepTargetExpressionOnCTE,
- rfCtx,
- cteProducerMap.get(cteId)
- );
- if (pushedDown) {
- rfCtx.removeFilter(
- rfToPush,
-
rightDeepTargetExpressionOnCTE.getInputSlotExprIds().iterator().next());
- }
- }
- }
+ private static Set<Expression> srcExpressionsOf(Set<RuntimeFilter> rfs) {
+ return
rfs.stream().map(RuntimeFilter::getSrcExpr).collect(Collectors.toSet());
+ }
+
+ /**
+ * The runtime filters that all the given consumers have for one source
expression.
+ */
+ private static List<RuntimeFilter>
runtimeFiltersOfSrcExpression(Set<PhysicalCTEConsumer> consumers,
+ Map<PhysicalCTEConsumer, Set<RuntimeFilter>> consumerToRFs,
Expression srcExpr) {
+ List<RuntimeFilter> rfsOfSrcExpr = Lists.newArrayList();
+ for (PhysicalCTEConsumer consumer : consumers) {
+ for (RuntimeFilter rf : consumerToRFs.get(consumer)) {
+ if (rf.getSrcExpr().equals(srcExpr)) {
+ rfsOfSrcExpr.add(rf);
}
}
}
- return result;
+ Preconditions.checkArgument(!rfsOfSrcExpr.isEmpty());
+ return rfsOfSrcExpr;
+ }
+
+ /**
+ * Push one group of identical runtime filters into the shared CTE
producer. Only called with a group
+ * that every consumer applies, see {@link #selectPushableRuntimeFilters}.
+ */
+ private void pushDownIdenticalFilters(List<RuntimeFilter> rfsOfIdentity,
CTEId cteId,
+ RuntimeFilterContext rfCtx, PhysicalCTEProducer cteProducer) {
+ // the most right deep buildNode from rfsOfIdentity is used as
buildNode for pushDown rf
+ // since the srcExpr are the same, all buildNodes of rfsOfIdentity are
in the same tree path
+ // the longest ancestors means its corresponding rf build node is the
most right deep one.
+ List<RuntimeFilter> rightDeepRfs = Lists.newArrayList();
+ List<Plan> rightDeepAncestors =
rfsOfIdentity.get(0).getBuilderNode().getAncestors();
+ int rightDeepAncestorsSize = rightDeepAncestors.size();
+ RuntimeFilter leftTop = rfsOfIdentity.get(0);
+ int leftTopAncestorsSize = rightDeepAncestorsSize;
+ for (RuntimeFilter rf : rfsOfIdentity) {
+ List<Plan> ancestors = rf.getBuilderNode().getAncestors();
+ int currentAncestorsSize = ancestors.size();
+ if (currentAncestorsSize >= rightDeepAncestorsSize) {
+ if (currentAncestorsSize == rightDeepAncestorsSize) {
+ rightDeepRfs.add(rf);
+ } else {
+ rightDeepAncestorsSize = currentAncestorsSize;
+ rightDeepAncestors = ancestors;
+ rightDeepRfs.clear();
+ rightDeepRfs.add(rf);
+ }
+ }
+ if (currentAncestorsSize < leftTopAncestorsSize) {
+ leftTopAncestorsSize = currentAncestorsSize;
+ leftTop = rf;
+ }
+ }
+
Preconditions.checkArgument(rightDeepAncestors.contains(leftTop.getBuilderNode()));
+ // check nodes between right deep and left top are SPJ and not denied
join and not mark join
+ boolean valid = true;
+ for (Plan cursor : rightDeepAncestors) {
+ if (cursor.equals(leftTop.getBuilderNode())) {
+ break;
+ }
+ // valid = valid && SPJ_PLAN.contains(cursor.getClass());
+ if (cursor instanceof AbstractPhysicalJoin) {
+ AbstractPhysicalJoin cursorJoin = (AbstractPhysicalJoin)
cursor;
+ valid = (!RuntimeFilterGenerator.DENIED_JOIN_TYPES
Review Comment:
[P1] Also reject a representative whose source crosses a null-generating
outer-join child. For example:
```text
Jupper [c1.k <=> r.bk] upper build has {1, NULL}
CTEConsumer c1
R RIGHT OUTER [b.k = d.k]
Jdeep [c2.k <=> b.k] deeper build has {1}
CTEConsumer c2
Scan b(k)
Scan d(k) has an unmatched row
```
`R` null-extends `b.k` but preserves its ExprId, so both `<=>` RFs group
with the same `nullAware=true` identity and the deeper one is selected. RIGHT
OUTER is allowed here, and an SPJ-only check would still allow it. The selected
filter did not observe NULL, so applying it to the shared producer removes
`t.k=NULL`, even though the upper join's build does contain NULL and must match
that row. Track which outer-join child carries the source, or conservatively
reject this path, and cover it with a regression.
--
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]
