This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new b6c8c9fb39c branch-4.2: [fix](rf) Push runtime filters into a shared
CTE producer per filter identity #68340 (#70096)
b6c8c9fb39c is described below
commit b6c8c9fb39c00b9709a28c28ce65feed2b96b384
Author: morrySnow <[email protected]>
AuthorDate: Wed Nov 4 08:28:12 2026 +0800
branch-4.2: [fix](rf) Push runtime filters into a shared CTE producer per
filter identity #68340 (#70096)
picked from master #68340
Co-authored-by: minghong <[email protected]>
---
.../processor/post/RuntimeFilterContext.java | 33 +-
.../processor/post/RuntimeFilterGenerator.java | 434 +++++++++++++++------
.../post/RuntimeFilterPushDownVisitor.java | 3 +-
.../nereids/postprocess/RuntimeFilterTest.java | 160 ++++++++
.../SharedCteRuntimeFilterBackportTest.java | 178 +++++++++
..._cte_shared_producer_min_max_runtime_filter.out | 33 ++
...e_shared_producer_min_max_runtime_filter.groovy | 148 +++++++
7 files changed, 859 insertions(+), 130 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/RuntimeFilterContext.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/RuntimeFilterContext.java
index 85e76674ab7..d3be56b1379 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/RuntimeFilterContext.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/RuntimeFilterContext.java
@@ -24,6 +24,7 @@ import
org.apache.doris.nereids.trees.expressions.EqualPredicate;
import org.apache.doris.nereids.trees.expressions.ExprId;
import org.apache.doris.nereids.trees.expressions.Expression;
import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.NullSafeEqual;
import org.apache.doris.nereids.trees.expressions.Slot;
import org.apache.doris.nereids.trees.plans.Plan;
import org.apache.doris.nereids.trees.plans.physical.AbstractPhysicalJoin;
@@ -36,6 +37,7 @@ import
org.apache.doris.planner.RuntimeFilterGenerator.FilterSizeLimits;
import org.apache.doris.planner.RuntimeFilterId;
import org.apache.doris.planner.ScanNode;
import org.apache.doris.qe.SessionVariable;
+import org.apache.doris.thrift.TMinMaxRuntimeFilterType;
import org.apache.doris.thrift.TRuntimeFilterType;
import com.google.common.annotations.VisibleForTesting;
@@ -57,17 +59,22 @@ import java.util.stream.Collectors;
public class RuntimeFilterContext {
/**
- * the combination of src expr, filter type and source join node, use to
identify the filter
+ * The source, builder and predicate semantics identify a filter shared by
multiple targets.
*/
public class RuntimeFilterIdentity {
final Expression expr;
final TRuntimeFilterType type;
final AbstractPhysicalJoin join;
+ final TMinMaxRuntimeFilterType minMaxType;
+ final boolean nullAware;
- public RuntimeFilterIdentity(Expression expr, TRuntimeFilterType type,
AbstractPhysicalJoin join) {
+ public RuntimeFilterIdentity(Expression expr, TRuntimeFilterType type,
AbstractPhysicalJoin join,
+ TMinMaxRuntimeFilterType minMaxType, int exprOrder) {
this.expr = expr;
this.type = type;
this.join = join;
+ this.minMaxType = minMaxType;
+ this.nullAware = isNullAware(join, exprOrder);
}
@Override
@@ -85,7 +92,8 @@ public class RuntimeFilterContext {
}
if (other instanceof RuntimeFilterIdentity) {
RuntimeFilterIdentity otherId = (RuntimeFilterIdentity) other;
- return expr.equals(otherId.expr) && type.equals(otherId.type)
&& join.equals(otherId.join);
+ return expr.equals(otherId.expr) && type.equals(otherId.type)
&& join.equals(otherId.join)
+ && minMaxType == otherId.minMaxType && nullAware ==
otherId.nullAware;
}
return false;
}
@@ -283,14 +291,27 @@ public class RuntimeFilterContext {
this.targetOnOlapScanNodeMap.computeIfAbsent(relation, k ->
Lists.newArrayList()).add(slot);
}
+ /** Whether the builder keeps a NULL probe value for this filter. */
+ public static boolean isNullAware(AbstractPhysicalJoin<?, ?> builder, int
exprOrder) {
+ if (!(builder instanceof PhysicalHashJoin) || exprOrder < 0) {
+ return false;
+ }
+ List<Expression> conjuncts = builder.getHashJoinConjuncts();
+ Preconditions.checkArgument(exprOrder < conjuncts.size(),
+ "exprOrder %s of the runtime filter is not a hash join
conjunct", exprOrder);
+ return conjuncts.get(exprOrder) instanceof NullSafeEqual;
+ }
+
public RuntimeFilter getRuntimeFilterBySrcAndType(Expression src,
- TRuntimeFilterType type,
AbstractPhysicalJoin join) {
- return runtimeFilterIdentityToFilter.get(new
RuntimeFilterIdentity(src, type, join));
+ TRuntimeFilterType type,
AbstractPhysicalJoin join,
+ TMinMaxRuntimeFilterType minMaxType, int exprOrder) {
+ return runtimeFilterIdentityToFilter.get(new
RuntimeFilterIdentity(src, type, join, minMaxType, exprOrder));
}
public void setRuntimeFilterIdentityToFilter(Expression src,
TRuntimeFilterType type,
AbstractPhysicalJoin join,
RuntimeFilter rf) {
- runtimeFilterIdentityToFilter.put(new RuntimeFilterIdentity(src, type,
join), rf);
+ runtimeFilterIdentityToFilter.put(
+ new RuntimeFilterIdentity(src, type, join,
rf.gettMinMaxType(), rf.getExprOrder()), rf);
}
public Map<ExprId, SlotRef> getExprIdToOlapScanNodeSlotRef() {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/RuntimeFilterGenerator.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/RuntimeFilterGenerator.java
index 7f2cbe75325..8fb58e59b44 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/RuntimeFilterGenerator.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/RuntimeFilterGenerator.java
@@ -62,6 +62,7 @@ import org.apache.doris.thrift.TRuntimeFilterType;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
@@ -72,8 +73,10 @@ import org.apache.logging.log4j.Logger;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.stream.Collectors;
@@ -111,141 +114,166 @@ public class RuntimeFilterGenerator extends
PlanPostProcessor {
}
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) {
- for (PhysicalRelation rel : rf.getTargetScans()) {
- 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()) {
+ for (PhysicalRelation target : rf.getTargetScans()) {
+ 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;
- int targetCount =
rfToPush.getTargetExpressions().size();
- for (int i = 0; i < targetCount; i++) {
- PhysicalRelation rel =
rfToPush.getTargetScans().get(i);
- if (rel instanceof PhysicalCTEConsumer
- && ((PhysicalCTEConsumer)
rel).getCteId().equals(cteId)) {
- rightDeepTargetExpressionOnCTE =
rfToPush.getTargetExpressions().get(i);
- break;
- }
- }
-
- 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) {
+ Set<RuntimeFilter> rfsOfSrcExpr = Sets.newLinkedHashSet();
+ 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 Lists.newArrayList(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()));
+ // The filter of the deepest builder stands in for the filters of the
other consumers, which is sound
+ // only when it prunes at most as many rows as each of them would. The
source expression therefore has
+ // to keep the values it has on the build side of the deepest builder
while it travels up to the
+ // shallowest one: a node which can add a value to it -- the NULL an
outer join generates for the
+ // missing side of the source child, or the NULL a repeat synthesizes
for a grouping set which does
+ // not group by the source -- would make the filter below it prune the
rows the consumers above it
+ // still need.
+ if (!keepsSourceValue(rightDeepAncestors, leftTop.getBuilderNode())) {
+ return;
+ }
+ // The filter created on the producer stands in for every filter of
this group, so it has to keep the
+ // group's requirement not to be waited for. The producer feeds all
the consumers, while a filter which
+ // waits is produced by the build side of one of them: a replacement
which waits for a consumer whose
+ // own filter was non-blocking closes the cycle producer -> consumer
build side -> consumer -> producer,
+ // and the query stalls until the runtime filter or the query times
out. Waiting less than a member of
+ // the group asks for only makes the filter arrive later, it never
prunes a row the member would keep,
+ // so the replacement is non-blocking as soon as one member of the
group is.
+ boolean nonBlocking =
rfsOfIdentity.stream().anyMatch(RuntimeFilter::isNonBlocking);
+
+ for (RuntimeFilter rfToPush : rightDeepRfs) {
+ Expression rightDeepTargetExpressionOnCTE = null;
+ for (int i = 0; i < rfToPush.getTargetScans().size(); i++) {
+ PhysicalRelation rel = rfToPush.getTargetScans().get(i);
+ if (rel instanceof PhysicalCTEConsumer
+ && ((PhysicalCTEConsumer)
rel).getCteId().equals(cteId)) {
+ rightDeepTargetExpressionOnCTE =
rfToPush.getTargetExpressions().get(i);
+ break;
+ }
+ }
+
+ boolean pushedDown = doPushDownIntoCTEProducerInternal(
+ rfToPush,
+ rightDeepTargetExpressionOnCTE,
+ rfCtx,
+ cteProducer,
+ nonBlocking
+ );
+ if (pushedDown) {
+ rfCtx.removeFilter(
+ rfToPush,
+
rightDeepTargetExpressionOnCTE.getInputSlotExprIds().iterator().next());
+ }
+ }
}
/**
@@ -875,12 +903,171 @@ public class RuntimeFilterGenerator extends
PlanPostProcessor {
cteId);
return false;
}
- Set<Expression> producerTargetExpressions = rfsToPushDown.stream()
- .map(rf -> getProducerTargetExpression(rf, cteId))
- .collect(Collectors.toSet());
+ RuntimeFilter first = rfsToPushDown.get(0);
+ boolean nullAware = isNullAware(first);
+ Set<Expression> producerTargetExpressions = Sets.newHashSet();
+ for (RuntimeFilter filter : rfsToPushDown) {
+ // A MIN cannot replace a MAX, nor an equality filter a null-safe
equality filter.
+ // The multi-target pipeline keeps this group at its consumers
when semantics differ.
+ if (filter.getType() != first.getType() || filter.gettMinMaxType()
!= first.gettMinMaxType()
+ || isNullAware(filter) != nullAware) {
+ return false;
+ }
+ for (int i = 0; i < filter.getTargetScans().size(); i++) {
+ PhysicalRelation relation = filter.getTargetScans().get(i);
+ if (relation instanceof PhysicalCTEConsumer
+ && ((PhysicalCTEConsumer)
relation).getCteId().equals(cteId)) {
+ PhysicalCTEConsumer consumer = (PhysicalCTEConsumer)
relation;
+ Expression target = filter.getTargetExpressions().get(i);
+ Map<Expression, Expression> replaceMap = Maps.newHashMap();
+ for (Slot slot : target.getInputSlots()) {
+ replaceMap.put(slot, consumer.getProducerSlot(slot));
+ }
+
producerTargetExpressions.add(ExpressionUtils.replace(target, replaceMap));
+ }
+ }
+ }
return producerTargetExpressions.size() == 1;
}
+ /**
+ * Whether the nodes between the deepest builder and the given shallowest
builder only restrict the rows
+ * the source expression of the runtime filters is evaluated on, see the
shared producer pushdown.
+ * The list of ancestors starts with the deepest builder itself, whose
filter is the one which is pushed,
+ * so it is not part of the path.
+ */
+ @VisibleForTesting
+ public static boolean keepsSourceValue(List<Plan> ancestors, Plan
shallowestBuilder) {
+ if (ancestors.get(0).equals(shallowestBuilder)) {
+ // both filters are computed by the same node, their source values
are identical
+ return true;
+ }
+ for (int i = 1; i < ancestors.size(); i++) {
+ Plan node = ancestors.get(i);
+ if (node.equals(shallowestBuilder)) {
+ break;
+ }
+ if (SPJ_PLAN.stream().noneMatch(clazz -> clazz.isInstance(node))) {
+ return false;
+ }
+ if (node instanceof AbstractPhysicalJoin) {
+ AbstractPhysicalJoin<?, ?> join = (AbstractPhysicalJoin<?, ?>)
node;
+ if
(RuntimeFilterGenerator.DENIED_JOIN_TYPES.contains(join.getJoinType()) ||
join.isMarkJoin()) {
+ return false;
+ }
+ // the source is null extended when it comes from the side the
join generates NULLs for
+ if (isNullGeneratingChild(join.getJoinType(), join.child(0) ==
ancestors.get(i - 1))) {
+ return false;
+ }
+ }
+ }
+ return true;
+ }
+
+ /**
+ * Whether the given child of a join of that type is the side the join
generates NULLs for. Mirrors
+ * {@link RuntimeFilterPushDownVisitor}; the outer join types which are
denied already do not get here.
+ */
+ private static boolean isNullGeneratingChild(JoinType joinType, boolean
isLeftChild) {
+ if (joinType.isFullOuterJoin()) {
+ return true;
+ }
+ if (isLeftChild) {
+ return joinType.isRightOuterJoin() ||
joinType.isAsofRightOuterJoin();
+ }
+ return joinType.isLeftOuterJoin() || joinType.isAsofLeftOuterJoin();
+ }
+
+ /**
+ * Select the runtime filters of one source expression that may be pushed
into the shared CTE producer.
+ *
+ * <p>The producer feeds every consumer, so a filter may only be applied
on the producer when all the
+ * consumers apply the very same filter; a filter that only holds for one
consumer would prune the rows
+ * that the other consumers still need. The filters are therefore grouped
by identity -- same type, same
+ * min/max direction, same NULL semantics and same target expression on
the producer -- and only a group
+ * that every consumer applies is selected. For example, with `t c1 where
c1.k > b.x` and
+ * `t c2 where c2.k < b.x` the consumers produce a MIN and a MAX filter
on the same producer column:
+ * neither group covers both
+ * consumers, so neither is pushed. When on the other hand every consumer
applies the same pair of
+ * filters, for example a MIN_MAX and an IN_OR_BLOOM filter of the same
column, both groups are selected
+ * and each of them is still pushed once on the producer.
+ */
+ @VisibleForTesting
+ public static List<List<RuntimeFilter>> selectPushableRuntimeFilters(
+ List<RuntimeFilter> rfsOfSrcExpr, Set<PhysicalCTEConsumer>
consumers, CTEId cteId) {
+ Map<FilterIdentity, List<RuntimeFilter>> rfsByIdentity =
Maps.newLinkedHashMap();
+ for (RuntimeFilter rf : rfsOfSrcExpr) {
+ // A multi-target filter may map different consumers to different
producer columns.
+ if (canPushDownRuntimeFiltersIntoCTEProducer(ImmutableList.of(rf),
cteId)) {
+ rfsByIdentity.computeIfAbsent(new FilterIdentity(rf, cteId),
key -> Lists.newArrayList()).add(rf);
+ }
+ }
+ List<List<RuntimeFilter>> pushable = Lists.newArrayList();
+ for (List<RuntimeFilter> rfsOfIdentity : rfsByIdentity.values()) {
+ Set<PhysicalCTEConsumer> consumersApplying = rfsOfIdentity.stream()
+ .flatMap(rf -> rf.getTargetScans().stream())
+ .filter(PhysicalCTEConsumer.class::isInstance)
+ .map(PhysicalCTEConsumer.class::cast)
+ .filter(consumer -> consumer.getCteId().equals(cteId))
+ .collect(Collectors.toSet());
+ if (consumersApplying.equals(consumers)) {
+ pushable.add(rfsOfIdentity);
+ } else {
+ LOG.debug("Skip pushing runtime filters into CTE producer
because only {} of {} consumers"
+ + " of cteId: {} apply the filter {}, while
all of them have to apply it",
+ consumersApplying.size(), consumers.size(), cteId,
rfsOfIdentity.get(0));
+ }
+ }
+ return pushable;
+ }
+
+ /**
+ * Identity of a runtime filter with respect to a CTE producer. Two
filters with the same identity apply
+ * the very same predicate on the producer, therefore applying one of them
once on the producer is
+ * equivalent to applying it on every consumer. The source expression is
the same for all the filters
+ * compared here, so it does not take part in the identity.
+ */
+ private static final class FilterIdentity {
+ private final TRuntimeFilterType type;
+ private final TMinMaxRuntimeFilterType minMaxType;
+ private final boolean nullAware;
+ private final Expression producerTargetExpression;
+
+ private FilterIdentity(RuntimeFilter rf, CTEId cteId) {
+ this.type = rf.getType();
+ this.minMaxType = rf.gettMinMaxType();
+ this.nullAware = isNullAware(rf);
+ this.producerTargetExpression = getProducerTargetExpression(rf,
cteId);
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj) {
+ return true;
+ }
+ if (!(obj instanceof FilterIdentity)) {
+ return false;
+ }
+ FilterIdentity that = (FilterIdentity) obj;
+ return type == that.type && minMaxType == that.minMaxType &&
nullAware == that.nullAware
+ &&
producerTargetExpression.equals(that.producerTargetExpression);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(type, minMaxType, nullAware,
producerTargetExpression);
+ }
+
+ @Override
+ public String toString() {
+ return type + "/" + minMaxType + (nullAware ? "/nullAware" : "") +
" -> " + producerTargetExpression;
+ }
+ }
+
+ private static boolean isNullAware(RuntimeFilter rf) {
+ return RuntimeFilterContext.isNullAware(rf.getBuilderNode(),
rf.getExprOrder());
+ }
+
private static Expression getProducerTargetExpression(RuntimeFilter rf,
CTEId cteId) {
List<PhysicalRelation> targetScans = rf.getTargetScans();
List<Expression> targetExpressions = rf.getTargetExpressions();
@@ -902,7 +1089,7 @@ public class RuntimeFilterGenerator extends
PlanPostProcessor {
}
private boolean doPushDownIntoCTEProducerInternal(RuntimeFilter rf,
Expression targetExpression,
- RuntimeFilterContext ctx,
PhysicalCTEProducer cteProducer) {
+ RuntimeFilterContext ctx, PhysicalCTEProducer cteProducer, boolean
nonBlocking) {
PhysicalPlan inputPlanNode = (PhysicalPlan) cteProducer.child(0);
Slot unwrappedSlot = checkTargetChild(targetExpression);
// aliasTransMap doesn't contain the key, means that the path from the
scan to the join
@@ -955,6 +1142,7 @@ public class RuntimeFilterGenerator extends
PlanPostProcessor {
rf.getSrcExpr(), targetList, targetExpressions,
rf.getType(), rf.getExprOrder(),
rf.getBuilderNode(), buildSideNdv,
rf.isBloomFilterSizeCalculatedByNdv(),
rf.gettMinMaxType(), cteNode);
+ filter.setNonBlocking(nonBlocking);
targetNodes.forEach(node ->
node.addAppliedRuntimeFilter(filter));
for (Slot slot : targetList) {
ctx.setTargetExprIdToFilter(slot.getExprId(), filter);
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/RuntimeFilterPushDownVisitor.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/RuntimeFilterPushDownVisitor.java
index b6414d3abec..c0ebeeb85ba 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/RuntimeFilterPushDownVisitor.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/RuntimeFilterPushDownVisitor.java
@@ -216,7 +216,8 @@ public class RuntimeFilterPushDownVisitor extends
PlanVisitor<Boolean, PushDownC
}
TRuntimeFilterType type = ctx.type;
- RuntimeFilter filter =
ctx.rfContext.getRuntimeFilterBySrcAndType(ctx.srcExpr, type, ctx.builderNode);
+ RuntimeFilter filter = ctx.rfContext.getRuntimeFilterBySrcAndType(
+ ctx.srcExpr, type, ctx.builderNode, ctx.singleSideMinMax,
ctx.exprOrder);
if (filter != null) {
if (!filter.hasTargetScan(scan)) {
// A join B on A.a1=B.b and A.a1 = A.a2
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/postprocess/RuntimeFilterTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/postprocess/RuntimeFilterTest.java
index 3dcd53385db..6797afbc3db 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/postprocess/RuntimeFilterTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/postprocess/RuntimeFilterTest.java
@@ -53,10 +53,13 @@ import
org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
import org.apache.doris.nereids.trees.plans.physical.AbstractPhysicalJoin;
import org.apache.doris.nereids.trees.plans.physical.AbstractPhysicalPlan;
import org.apache.doris.nereids.trees.plans.physical.PhysicalCTEConsumer;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalCTEProducer;
import org.apache.doris.nereids.trees.plans.physical.PhysicalHashJoin;
import org.apache.doris.nereids.trees.plans.physical.PhysicalOlapScan;
import org.apache.doris.nereids.trees.plans.physical.PhysicalPlan;
import org.apache.doris.nereids.trees.plans.physical.PhysicalProject;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalRelation;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalRepeat;
import org.apache.doris.nereids.trees.plans.physical.RuntimeFilter;
import org.apache.doris.nereids.types.IntegerType;
import org.apache.doris.nereids.types.VariantType;
@@ -762,4 +765,161 @@ public class RuntimeFilterTest extends SSBTestBase {
connectContext.getSessionVariable().enableDecoupledRuntimeFilter =
false;
}
+ /**
+ * The runtime filters which every consumer of a CTE applies are pushed
into the producer, where one of
+ * them filters the rows of all the consumers; the filters of a single
consumer must stay where they are,
+ * otherwise they prune the rows the other consumers still need.
+ */
+ @Test
+ public void testPushSharedCteRuntimeFilterIntoTheProducer() {
+ int oldType =
connectContext.getSessionVariable().getRuntimeFilterType();
+ boolean oldMaterialize =
connectContext.getSessionVariable().enableCTEMaterialize;
+
connectContext.getSessionVariable().setRuntimeFilterType(TRuntimeFilterType.MIN_MAX.getValue());
+ connectContext.getSessionVariable().enableCTEMaterialize = true;
+ try {
+ // Keep a direct-slot project: 4.2 only pushes CTE filters through
direct project items.
+ // Both consumers apply the same MIN filter built from
`p_partkey`, so it reaches the producer.
+ PhysicalPlan plan = planAfterPostProcess(
+ "with t as (select lo_partkey, abs(lo_partkey) as v from
lineorder)"
+ + " select c2.lo_partkey, c1.v, c2.v from t c2
cross join t c1 cross join part"
+ + " where c1.lo_partkey > p_partkey and
c2.lo_partkey > p_partkey");
+ List<RuntimeFilter> pushedIntoProducer =
runtimeFiltersInsideCteProducers(plan);
+ Assertions.assertFalse(pushedIntoProducer.isEmpty(),
+ () -> "the filter which every consumer applies must be
pushed into the producer: "
+ + plan.treeString() + " consumers: " +
runtimeFiltersOnCteConsumers(plan));
+ // the filter which was moved into the producer is no longer
applied by the consumer it was built
+ // for, otherwise the rows it filters would be filtered twice
+ List<RuntimeFilter> onConsumers =
runtimeFiltersOnCteConsumers(plan);
+
Assertions.assertTrue(pushedIntoProducer.stream().noneMatch(onConsumers::contains),
+ () -> "a filter pushed into the producer must not stay on
its consumer: " + onConsumers);
+ } finally {
+ connectContext.getSessionVariable().setRuntimeFilterType(oldType);
+ connectContext.getSessionVariable().enableCTEMaterialize =
oldMaterialize;
+ }
+ }
+
+ @Test
+ public void testDoNotPushSingleConsumerCteRuntimeFilterIntoTheProducer() {
+ int oldType =
connectContext.getSessionVariable().getRuntimeFilterType();
+ boolean oldMaterialize =
connectContext.getSessionVariable().enableCTEMaterialize;
+
connectContext.getSessionVariable().setRuntimeFilterType(TRuntimeFilterType.MIN_MAX.getValue());
+ connectContext.getSessionVariable().enableCTEMaterialize = true;
+ try {
+ // `c1.k > p_partkey` builds a MIN filter and `c2.k < p_partkey` a
MAX one: each of them is
+ // applied by one consumer only, so neither may be applied on the
shared producer.
+ List<RuntimeFilter> pushedIntoProducer =
runtimeFiltersInsideCteProducers(planAfterPostProcess(
+ "with t as (select lo_partkey, abs(lo_partkey) as v from
lineorder)"
+ + " select c2.lo_partkey, c1.v, c2.v from t c2
cross join t c1 cross join part"
+ + " where c1.lo_partkey > p_partkey and
c2.lo_partkey < p_partkey"));
+ Assertions.assertTrue(pushedIntoProducer.isEmpty(),
+ () -> "the filters of a single consumer must not reach the
producer: " + pushedIntoProducer);
+ } finally {
+ connectContext.getSessionVariable().setRuntimeFilterType(oldType);
+ connectContext.getSessionVariable().enableCTEMaterialize =
oldMaterialize;
+ }
+ }
+
+ /**
+ * The filter created on the shared producer replaces the filters of the
consumers, so it has to keep the
+ * requirement of that group not to be waited for. The producer feeds
every consumer, while a filter which
+ * waits is built by one of them: a replacement which waits for a consumer
whose own filter was
+ * non-blocking closes the cycle producer -> build side of a consumer ->
consumer -> producer, and the
+ * query stalls until the runtime filter or the query times out.
+ */
+ @Test
+ public void
testPushSharedCteRuntimeFilterIntoTheProducerKeepsTheNonBlockingRequirement() {
+ boolean oldMaterialize =
connectContext.getSessionVariable().enableCTEMaterialize;
+ boolean oldExpandByInnerJoin =
connectContext.getSessionVariable().expandRuntimeFilterByInnerJoin;
+ boolean oldDecoupled =
connectContext.getSessionVariable().enableDecoupledRuntimeFilter;
+ long oldMinDecoupledRows =
connectContext.getSessionVariable().minDecoupledRfTargetRows;
+ connectContext.getSessionVariable().enableCTEMaterialize = true;
+ connectContext.getSessionVariable().expandRuntimeFilterByInnerJoin =
true;
+ connectContext.getSessionVariable().enableDecoupledRuntimeFilter =
true;
+ // the tables of the test catalog hold one row, while the decoupled
filter of the plan below targets a
+ // scan of it: keep the filter which describes the behavior under test
rather than the pruning of
+ // filters which cannot arrive in time on a tiny scan.
+ connectContext.getSessionVariable().minDecoupledRfTargetRows = 0;
+ try {
+ // `c1.c_custkey = s_suppkey` is the condition join: its standard
filter targets the CTE consumer
+ // `c1` and expands to `c2`, while the reverse decoupled filter is
built by the deeper join
+ // `c2.c_custkey = c1.c_custkey`, whose build side carries the
filter on `c_region`. The decoupled
+ // filter is therefore preferred and both standard filters are
marked non-blocking.
+ PhysicalPlan plan = planAfterPostProcess(
+ "with t as (select c_custkey, c_region from customer)"
+ + " select c1.c_custkey from t c2 join t c1 on
c2.c_custkey = c1.c_custkey"
+ + " join supplier on c1.c_custkey = s_suppkey"
+ + " where c1.c_region = 'ASIA'");
+ List<RuntimeFilter> pushedIntoProducer =
runtimeFiltersInsideCteProducers(plan);
+ Assertions.assertFalse(pushedIntoProducer.isEmpty(),
+ () -> "the filter which every consumer applies must be
pushed into the producer: "
+ + plan.treeString());
+
Assertions.assertTrue(pushedIntoProducer.stream().allMatch(RuntimeFilter::isNonBlocking),
+ () -> "a filter pushed into the producer must keep the
non-blocking requirement of the"
+ + " filters it replaces: " + pushedIntoProducer);
+ } finally {
+ connectContext.getSessionVariable().enableCTEMaterialize =
oldMaterialize;
+ connectContext.getSessionVariable().expandRuntimeFilterByInnerJoin
= oldExpandByInnerJoin;
+ connectContext.getSessionVariable().enableDecoupledRuntimeFilter =
oldDecoupled;
+ connectContext.getSessionVariable().minDecoupledRfTargetRows =
oldMinDecoupledRows;
+ }
+ }
+
+ /**
+ * A node which synthesizes a value of the source column -- here the NULL
the repeat adds for the grouping
+ * set which does not group by it -- makes a filter built below it prune
the rows the consumers above it
+ * still need, so the deepest filter must not stand in for them.
+ */
+ @Test
+ public void testDoNotPushRuntimeFilterWhichCrossesAValueSynthesizingNode()
{
+ boolean oldMaterialize =
connectContext.getSessionVariable().enableCTEMaterialize;
+ connectContext.getSessionVariable().enableCTEMaterialize = true;
+ try {
+ PhysicalPlan plan = planAfterPostProcess(
+ "with t as (select lo_partkey as k from lineorder)"
+ + " select c1.k from t c1 join ("
+ + " select x from ("
+ + " select c2.k as k2, p.p_partkey as x from t
c2 join part p on c2.k <=> p.p_partkey"
+ + " ) a group by grouping sets ((x), ())"
+ + " ) g on c1.k <=> g.x");
+ Assertions.assertTrue(plan.containsType(PhysicalRepeat.class),
+ "the query must plan the grouping sets which synthesize
the NULL");
+
Assertions.assertTrue(!plan.<Plan>collect(PhysicalCTEProducer.class::isInstance).isEmpty(),
+ "the query must materialize the CTE");
+
Assertions.assertFalse(runtimeFiltersOnCteConsumers(plan).isEmpty(),
+ "the consumers must keep the filters which may not be
pushed");
+ List<RuntimeFilter> pushedIntoProducer =
runtimeFiltersInsideCteProducers(plan);
+ Assertions.assertTrue(pushedIntoProducer.isEmpty(),
+ () -> "a filter which crosses a value synthesizing node
must not reach the producer: "
+ + pushedIntoProducer);
+ } finally {
+ connectContext.getSessionVariable().enableCTEMaterialize =
oldMaterialize;
+ }
+ }
+
+ private PhysicalPlan planAfterPostProcess(String sql) {
+ PlanChecker checker =
PlanChecker.from(connectContext).analyze(sql).rewrite().optimize();
+ return new
PlanPostProcessors(checker.getCascadesContext()).process(checker.getBestPlanTree());
+ }
+
+ /** The runtime filters which are still applied by the consumers of a CTE.
*/
+ private static List<RuntimeFilter>
runtimeFiltersOnCteConsumers(PhysicalPlan plan) {
+ List<RuntimeFilter> applied = new ArrayList<>();
+ for (Plan consumer :
plan.<Plan>collect(PhysicalCTEConsumer.class::isInstance)) {
+ applied.addAll(((AbstractPhysicalPlan)
consumer).getAppliedRuntimeFilters());
+ }
+ return applied;
+ }
+
+ /** The runtime filters which were installed on the relations inside the
CTE producers. */
+ private static List<RuntimeFilter>
runtimeFiltersInsideCteProducers(PhysicalPlan plan) {
+ List<RuntimeFilter> applied = new ArrayList<>();
+ for (Plan producer :
plan.<Plan>collect(PhysicalCTEProducer.class::isInstance)) {
+ for (Plan relation : ((PhysicalCTEProducer<?>) producer).child(0)
+ .<Plan>collect(PhysicalRelation.class::isInstance)) {
+ applied.addAll(((AbstractPhysicalPlan)
relation).getAppliedRuntimeFilters());
+ }
+ }
+ return applied;
+ }
+
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/postprocess/SharedCteRuntimeFilterBackportTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/postprocess/SharedCteRuntimeFilterBackportTest.java
new file mode 100644
index 00000000000..7afb8be7411
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/postprocess/SharedCteRuntimeFilterBackportTest.java
@@ -0,0 +1,178 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.postprocess;
+
+import org.apache.doris.nereids.processor.post.RuntimeFilterContext;
+import org.apache.doris.nereids.processor.post.RuntimeFilterGenerator;
+import org.apache.doris.nereids.trees.expressions.CTEId;
+import org.apache.doris.nereids.trees.expressions.EqualTo;
+import org.apache.doris.nereids.trees.expressions.NullSafeEqual;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.plans.JoinType;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalCTEConsumer;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalHashJoin;
+import org.apache.doris.nereids.trees.plans.physical.RuntimeFilter;
+import org.apache.doris.nereids.types.IntegerType;
+import org.apache.doris.planner.RuntimeFilterId;
+import org.apache.doris.qe.SessionVariable;
+import org.apache.doris.thrift.TMinMaxRuntimeFilterType;
+import org.apache.doris.thrift.TRuntimeFilterType;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.util.List;
+
+public class SharedCteRuntimeFilterBackportTest {
+ private final CTEId cteId = new CTEId(1);
+ private final SlotReference source = new SlotReference("source",
IntegerType.INSTANCE);
+ private final SlotReference producer = new SlotReference("producer",
IntegerType.INSTANCE);
+
+ @Test
+ public void testMinAndMaxCannotReplaceEachOther() {
+ RuntimeFilter min = filter(TRuntimeFilterType.MIN_MAX,
TMinMaxRuntimeFilterType.MIN, false);
+ RuntimeFilter max = filter(TRuntimeFilterType.MIN_MAX,
TMinMaxRuntimeFilterType.MAX, false);
+
Assertions.assertFalse(RuntimeFilterGenerator.canPushDownRuntimeFiltersIntoCTEProducer(
+ ImmutableList.of(min, max), cteId));
+
Assertions.assertTrue(RuntimeFilterGenerator.canPushDownRuntimeFiltersIntoCTEProducer(
+ ImmutableList.of(min, filter(TRuntimeFilterType.MIN_MAX,
TMinMaxRuntimeFilterType.MIN, false)), cteId));
+ }
+
+ @Test
+ public void testDifferentTypesAndNullSemanticsAreNotInterchangeable() {
+ RuntimeFilter ordinary = filter(TRuntimeFilterType.IN_OR_BLOOM,
TMinMaxRuntimeFilterType.MIN_MAX, false);
+
Assertions.assertFalse(RuntimeFilterGenerator.canPushDownRuntimeFiltersIntoCTEProducer(
+ ImmutableList.of(ordinary,
+ filter(TRuntimeFilterType.MIN_MAX,
TMinMaxRuntimeFilterType.MIN_MAX, false)), cteId));
+
Assertions.assertFalse(RuntimeFilterGenerator.canPushDownRuntimeFiltersIntoCTEProducer(
+ ImmutableList.of(ordinary,
+ filter(TRuntimeFilterType.IN_OR_BLOOM,
TMinMaxRuntimeFilterType.MIN_MAX, true)), cteId));
+ }
+
+ @Test
+ public void testEveryTargetOfTheSameFilterMustMatch() {
+ RuntimeFilter filter = filter(TRuntimeFilterType.IN_OR_BLOOM,
TMinMaxRuntimeFilterType.MIN_MAX, false);
+ SlotReference second = new SlotReference("second",
IntegerType.INSTANCE);
+ PhysicalCTEConsumer consumer = Mockito.mock(PhysicalCTEConsumer.class);
+ Mockito.when(consumer.getCteId()).thenReturn(cteId);
+ Mockito.when(consumer.getProducerSlot(second)).thenReturn(
+ new SlotReference("different_producer_column",
IntegerType.INSTANCE));
+ filter.addTargetSlot(second, second, consumer);
+
Assertions.assertFalse(RuntimeFilterGenerator.canPushDownRuntimeFiltersIntoCTEProducer(
+ ImmutableList.of(filter), cteId));
+ }
+
+ @Test
+ public void testSourceCannotCrossNullGeneratingSide() {
+ Plan deepest = Mockito.mock(Plan.class);
+ Plan shallowest = Mockito.mock(Plan.class);
+ PhysicalHashJoin<?, ?> join = Mockito.mock(PhysicalHashJoin.class);
+ Mockito.when(join.getJoinType()).thenReturn(JoinType.RIGHT_OUTER_JOIN);
+ Mockito.when(join.child(0)).thenReturn(deepest);
+ Plan right = Mockito.mock(Plan.class);
+ Mockito.when(join.child(1)).thenReturn(right);
+ Assertions.assertFalse(RuntimeFilterGenerator.keepsSourceValue(
+ ImmutableList.of(deepest, join), shallowest));
+ Assertions.assertTrue(RuntimeFilterGenerator.keepsSourceValue(
+ ImmutableList.of(right, join), shallowest));
+ Mockito.when(join.getJoinType()).thenReturn(JoinType.INNER_JOIN);
+ Assertions.assertTrue(RuntimeFilterGenerator.keepsSourceValue(
+ ImmutableList.of(deepest, join), shallowest));
+ }
+
+ @Test
+ public void testIdentityKeepsDirectionsAndNullSemanticsSeparate() {
+ RuntimeFilter first = filter(TRuntimeFilterType.MIN_MAX,
TMinMaxRuntimeFilterType.MIN, false);
+ RuntimeFilterContext context = new RuntimeFilterContext(new
SessionVariable(),
+ RuntimeFilterId.createGenerator());
+ RuntimeFilter max = copyFilter(first, TRuntimeFilterType.MIN_MAX,
TMinMaxRuntimeFilterType.MAX, 0);
+ context.setRuntimeFilterIdentityToFilter(source, first.getType(),
first.getBuilderNode(), first);
+ context.setRuntimeFilterIdentityToFilter(source, max.getType(),
max.getBuilderNode(), max);
+ Assertions.assertSame(first,
context.getRuntimeFilterBySrcAndType(source, first.getType(),
+ first.getBuilderNode(), TMinMaxRuntimeFilterType.MIN, 0));
+ Assertions.assertSame(max,
context.getRuntimeFilterBySrcAndType(source, first.getType(),
+ first.getBuilderNode(), TMinMaxRuntimeFilterType.MAX, 0));
+
+ RuntimeFilter ordinary = copyFilter(first,
TRuntimeFilterType.IN_OR_BLOOM,
+ TMinMaxRuntimeFilterType.MIN_MAX, 0);
+ RuntimeFilter nullSafe = copyFilter(first,
TRuntimeFilterType.IN_OR_BLOOM,
+ TMinMaxRuntimeFilterType.MIN_MAX, 1);
+
Mockito.when(first.getBuilderNode().getHashJoinConjuncts()).thenReturn(ImmutableList.of(
+ new EqualTo(first.getTargetSlots().get(0), source),
+ new NullSafeEqual(first.getTargetSlots().get(0), source)));
+ context.setRuntimeFilterIdentityToFilter(source, ordinary.getType(),
ordinary.getBuilderNode(), ordinary);
+ context.setRuntimeFilterIdentityToFilter(source, nullSafe.getType(),
nullSafe.getBuilderNode(), nullSafe);
+ Assertions.assertSame(ordinary,
context.getRuntimeFilterBySrcAndType(source, ordinary.getType(),
+ ordinary.getBuilderNode(), TMinMaxRuntimeFilterType.MIN_MAX,
0));
+ Assertions.assertSame(nullSafe,
context.getRuntimeFilterBySrcAndType(source, nullSafe.getType(),
+ nullSafe.getBuilderNode(), TMinMaxRuntimeFilterType.MIN_MAX,
1));
+ // A non-null-aware decoupled filter still shares the ordinary
equality filter.
+ Assertions.assertSame(ordinary,
context.getRuntimeFilterBySrcAndType(source, ordinary.getType(),
+ ordinary.getBuilderNode(), TMinMaxRuntimeFilterType.MIN_MAX,
-1));
+ }
+
+ @Test
+ public void testSelectCommonIdentitiesWithoutLosingOtherFilterTypes() {
+ RuntimeFilter first = filter(TRuntimeFilterType.MIN_MAX,
TMinMaxRuntimeFilterType.MIN, false);
+ RuntimeFilter second = filter(TRuntimeFilterType.MIN_MAX,
TMinMaxRuntimeFilterType.MIN, false);
+ RuntimeFilter firstBloom = copyFilter(first,
TRuntimeFilterType.IN_OR_BLOOM,
+ TMinMaxRuntimeFilterType.MIN_MAX, 0);
+ RuntimeFilter secondBloom = copyFilter(second,
TRuntimeFilterType.IN_OR_BLOOM,
+ TMinMaxRuntimeFilterType.MIN_MAX, 0);
+ RuntimeFilter firstMax = copyFilter(first, TRuntimeFilterType.MIN_MAX,
TMinMaxRuntimeFilterType.MAX, 0);
+ List<List<RuntimeFilter>> selected =
RuntimeFilterGenerator.selectPushableRuntimeFilters(
+ ImmutableList.of(first, second, firstBloom, secondBloom,
firstMax),
+ ImmutableSet.of((PhysicalCTEConsumer)
first.getTargetScans().get(0),
+ (PhysicalCTEConsumer) second.getTargetScans().get(0)),
cteId);
+ Assertions.assertEquals(2, selected.size());
+ Assertions.assertTrue(selected.contains(ImmutableList.of(first,
second)));
+ Assertions.assertTrue(selected.contains(ImmutableList.of(firstBloom,
secondBloom)));
+ }
+
+ @Test
+ public void testOppositeDirectionsDoNotCoverEveryConsumer() {
+ RuntimeFilter min = filter(TRuntimeFilterType.MIN_MAX,
TMinMaxRuntimeFilterType.MIN, false);
+ RuntimeFilter max = filter(TRuntimeFilterType.MIN_MAX,
TMinMaxRuntimeFilterType.MAX, false);
+
Assertions.assertTrue(RuntimeFilterGenerator.selectPushableRuntimeFilters(ImmutableList.of(min,
max),
+ ImmutableSet.of((PhysicalCTEConsumer)
min.getTargetScans().get(0),
+ (PhysicalCTEConsumer) max.getTargetScans().get(0)),
cteId).isEmpty());
+ }
+
+ private RuntimeFilter copyFilter(RuntimeFilter original,
TRuntimeFilterType type,
+ TMinMaxRuntimeFilterType direction, int exprOrder) {
+ return new
RuntimeFilter(RuntimeFilterId.createGenerator().getNextId(), source,
+ original.getTargetSlots(), original.getTargetExpressions(),
type, exprOrder,
+ original.getBuilderNode(), -1L, true, direction,
original.getTargetScans().get(0));
+ }
+
+ private RuntimeFilter filter(TRuntimeFilterType type,
TMinMaxRuntimeFilterType direction, boolean nullSafe) {
+ SlotReference target = new SlotReference("target",
IntegerType.INSTANCE);
+ PhysicalCTEConsumer consumer = Mockito.mock(PhysicalCTEConsumer.class);
+ Mockito.when(consumer.getCteId()).thenReturn(cteId);
+ Mockito.when(consumer.getProducerSlot(target)).thenReturn(producer);
+ PhysicalHashJoin<?, ?> builder = Mockito.mock(PhysicalHashJoin.class);
+
Mockito.when(builder.getHashJoinConjuncts()).thenReturn(ImmutableList.of(
+ nullSafe ? new NullSafeEqual(target, source) : new
EqualTo(target, source)));
+ return new
RuntimeFilter(RuntimeFilterId.createGenerator().getNextId(), source,
+ ImmutableList.of(target), ImmutableList.of(target), type, 0,
builder, -1L, true, direction, consumer);
+ }
+}
diff --git
a/regression-test/data/nereids_rules_p0/cte/test_cte_shared_producer_min_max_runtime_filter.out
b/regression-test/data/nereids_rules_p0/cte/test_cte_shared_producer_min_max_runtime_filter.out
new file mode 100644
index 00000000000..a057df8a0a7
--- /dev/null
+++
b/regression-test/data/nereids_rules_p0/cte/test_cte_shared_producer_min_max_runtime_filter.out
@@ -0,0 +1,33 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !shared_cte_min_max_rf_opposite_directions --
+0 10 0 10 3
+0 5 0 5 3
+1 10 1 10 3
+1 5 1 5 3
+
+-- !shared_cte_min_max_rf_same_direction --
+10 10 10 10 3
+10 5 10 5 3
+5 10 5 10 3
+5 5 5 5 3
+
+-- !shared_cte_min_max_rf_inlined_cte --
+0 10 0 10 3
+0 5 0 5 3
+1 10 1 10 3
+1 5 1 5 3
+
+-- !shared_cte_null_aware --
+\N \N
+3 3
+
+-- !shared_cte_grouping_sets --
+\N \N
+3 3
+
+-- !shared_cte_min_max_rf_off --
+0 10 0 10 3
+0 5 0 5 3
+1 10 1 10 3
+1 5 1 5 3
+
diff --git
a/regression-test/suites/nereids_rules_p0/cte/test_cte_shared_producer_min_max_runtime_filter.groovy
b/regression-test/suites/nereids_rules_p0/cte/test_cte_shared_producer_min_max_runtime_filter.groovy
new file mode 100644
index 00000000000..4db23ec803f
--- /dev/null
+++
b/regression-test/suites/nereids_rules_p0/cte/test_cte_shared_producer_min_max_runtime_filter.groovy
@@ -0,0 +1,148 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite("test_cte_shared_producer_min_max_runtime_filter") {
+ sql "SET enable_nereids_planner=true"
+ sql "SET enable_fallback_to_original_planner=false"
+ sql "SET enable_pipeline_engine=true"
+ // Materialize the CTE so that both references share one producer scan.
+ sql "SET enable_cte_materialize=true"
+ sql "SET inline_cte_referenced_threshold=0"
+ sql "SET runtime_filter_type='MIN_MAX'"
+ sql "SET enable_runtime_filter_prune=false"
+ sql "SET runtime_filter_wait_time_ms=10000"
+
+ sql "DROP TABLE IF EXISTS cte_rf_shared_producer_f"
+ sql """
+ CREATE TABLE cte_rf_shared_producer_f (
+ k INT
+ ) ENGINE=OLAP
+ DUPLICATE KEY(k)
+ DISTRIBUTED BY HASH(k) BUCKETS 2
+ PROPERTIES ("replication_num" = "1")
+ """
+ sql "INSERT INTO cte_rf_shared_producer_f VALUES (0), (1), (5), (10)"
+
+ sql "DROP TABLE IF EXISTS cte_rf_shared_producer_b"
+ sql """
+ CREATE TABLE cte_rf_shared_producer_b (
+ x INT
+ ) ENGINE=OLAP
+ DUPLICATE KEY(x)
+ DISTRIBUTED BY HASH(x) BUCKETS 1
+ PROPERTIES ("replication_num" = "1")
+ """
+ sql "INSERT INTO cte_rf_shared_producer_b VALUES (3)"
+
+ sql "DROP TABLE IF EXISTS cte_rf_null_f"
+ sql """
+ CREATE TABLE cte_rf_null_f (
+ k INT
+ ) ENGINE=OLAP
+ DUPLICATE KEY(k)
+ DISTRIBUTED BY HASH(k) BUCKETS 2
+ PROPERTIES ("replication_num" = "1")
+ """
+ sql "INSERT INTO cte_rf_null_f VALUES (NULL), (1), (3)"
+
+ sql "DROP TABLE IF EXISTS cte_rf_null_b"
+ sql """
+ CREATE TABLE cte_rf_null_b (
+ x INT
+ ) ENGINE=OLAP
+ DUPLICATE KEY(x)
+ DISTRIBUTED BY HASH(x) BUCKETS 1
+ PROPERTIES ("replication_num" = "1")
+ """
+ sql "INSERT INTO cte_rf_null_b VALUES (NULL), (3)"
+
+ // The two references of the CTE need disjoint row ranges: `c1.k > b.x`
asks for a MIN runtime
+ // filter and `c2.k < b.x` for a MAX one. Both target the same column of
the shared producer, so
+ // pushing both of them into the producer prunes every row (k >= 3 AND k
<= 3) and the query
+ // silently loses the rows each consumer still needs.
+ sql "SET runtime_filter_mode='GLOBAL'"
+ order_qt_shared_cte_min_max_rf_opposite_directions """
+ WITH t AS (SELECT k, ABS(k) AS v FROM cte_rf_shared_producer_f)
+ SELECT c2.k AS lo, c1.k AS hi, c2.v AS lo_v, c1.v AS hi_v, b.x
+ FROM t c2 CROSS JOIN t c1 CROSS JOIN cte_rf_shared_producer_b b
+ WHERE c1.k > b.x AND c2.k < b.x
+ ORDER BY lo, hi
+ """
+
+ // Both references filter in the same direction, so both consumers produce
the same runtime
+ // filter. Pushing it into the producer is equivalent to applying it on
every consumer.
+ order_qt_shared_cte_min_max_rf_same_direction """
+ WITH t AS (SELECT k, ABS(k) AS v FROM cte_rf_shared_producer_f)
+ SELECT c2.k AS lo, c1.k AS hi, c2.v AS lo_v, c1.v AS hi_v, b.x
+ FROM t c2 CROSS JOIN t c1 CROSS JOIN cte_rf_shared_producer_b b
+ WHERE c1.k > b.x AND c2.k > b.x
+ ORDER BY lo, hi
+ """
+
+ // Without a shared producer each reference gets its own scan and its own
runtime filter.
+ sql "SET enable_cte_materialize=false"
+ order_qt_shared_cte_min_max_rf_inlined_cte """
+ WITH t AS (SELECT k, ABS(k) AS v FROM cte_rf_shared_producer_f)
+ SELECT c2.k AS lo, c1.k AS hi, c2.v AS lo_v, c1.v AS hi_v, b.x
+ FROM t c2 CROSS JOIN t c1 CROSS JOIN cte_rf_shared_producer_b b
+ WHERE c1.k > b.x AND c2.k < b.x
+ ORDER BY lo, hi
+ """
+ sql "SET enable_cte_materialize=true"
+
+ // NULL semantics are part of the filter identity: `c1.k <=> s.x` produces
a null aware filter while the
+ // deeper `c2.k = b0.x` produces an ordinary one. They prune different
rows -- the ordinary filter removes
+ // the rows whose probe column is NULL, which are exactly the rows the
null aware predicate matches -- so
+ // neither may be applied on the shared producer, otherwise the row which
matches NULL with NULL is lost.
+ sql "SET runtime_filter_type=12"
+ order_qt_shared_cte_null_aware """
+ WITH t AS (SELECT k FROM cte_rf_null_f)
+ SELECT c1.k AS a, s.x AS bx
+ FROM t c1
+ JOIN (
+ SELECT c2.k AS k2, b0.x AS x
+ FROM t c2 RIGHT OUTER JOIN cte_rf_null_b b0 ON c2.k = b0.x
+ ) s ON c1.k <=> s.x
+ ORDER BY a, bx
+ """
+
+ // A value synthesizing node between the two builders makes the deeper
filter prune rows the upper
+ // consumer still needs: the repeat adds the NULL of the grouping set
which does not group by x, so the
+ // filter built below it, from `b.x = {3}`, must not be applied on the
shared producer. The row which
+ // matches that synthesized NULL with the NULL of the CTE is lost when it
is.
+ order_qt_shared_cte_grouping_sets """
+ WITH t AS (SELECT k FROM cte_rf_null_f)
+ SELECT c1.k AS a, g.x AS gx
+ FROM t c1
+ JOIN (
+ SELECT x FROM (
+ SELECT c2.k AS k2, b.x AS x
+ FROM t c2 JOIN cte_rf_shared_producer_b b ON c2.k <=> b.x
+ ) z GROUP BY GROUPING SETS ((x), ())
+ ) g ON c1.k <=> g.x
+ ORDER BY a, gx
+ """
+
+ sql "SET runtime_filter_mode='OFF'"
+ order_qt_shared_cte_min_max_rf_off """
+ WITH t AS (SELECT k, ABS(k) AS v FROM cte_rf_shared_producer_f)
+ SELECT c2.k AS lo, c1.k AS hi, c2.v AS lo_v, c1.v AS hi_v, b.x
+ FROM t c2 CROSS JOIN t c1 CROSS JOIN cte_rf_shared_producer_b b
+ WHERE c1.k > b.x AND c2.k < b.x
+ ORDER BY lo, hi
+ """
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
