github-actions[bot] commented on code in PR #68340:
URL: https://github.com/apache/doris/pull/68340#discussion_r4070597456
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/RuntimeFilterGenerator.java:
##########
@@ -111,136 +115,153 @@ 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()));
+ // 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;
+ }
+
+ 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(
Review Comment:
[P1] Preserve the nonblocking state when these identity groups are recreated
in the CTE producer. A PR-specific reachable tree is:
```text
PhysicalCTEAnchor t(k, v)
|-- CTEProducer t -> Scan f(k, v)
`-- Jupper [c1.k = b.x, c1.v = b.x]
|-- Jdeep [c2.k = c1.k, c2.v = c1.v]
| |-- CTEConsumer c2
| `-- CTEConsumer c1
`-- Scan b(x)
```
Each upper standard `b.x` RF expands to both consumers. When the reverse
decoupled RFs from `Jdeep` are preferred, `markStandardRfAsNonBlocking` marks
both standard groups nonblocking to break the wait cycle. Previously the
combined producer targets `{k,v}` failed the single-target check, but this PR
splits them into pushable `k` and `v` identities. This call recreates each
producer RF without copying `isNonBlocking()`, removes the originals, and
translation gives the replacements positive waits. The producer then waits for
`Jupper`/`b`, `b` waits for `Jdeep`, and `Jdeep` needs the producer, so
execution stalls until RF/query timeout. Please preserve the group's
nonblocking requirement (including the dedup path; OR it, or decline the push
on disagreement) and add a translated shared-CTE/decoupled test asserting zero
wait.
--
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]
