924060929 commented on code in PR #68452:
URL: https://github.com/apache/doris/pull/68452#discussion_r4245489903
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/RuntimeFilterPruner.java:
##########
@@ -122,11 +136,137 @@ public PhysicalIntersect
visitPhysicalIntersect(PhysicalIntersect intersect, Cas
public PhysicalCTEAnchor<? extends Plan, ? extends Plan>
visitPhysicalCTEAnchor(
PhysicalCTEAnchor<? extends Plan, ? extends Plan> cteAnchor,
CascadesContext context) {
+ // Visit the producer subtree first: if its root is an effective RF
source
+ // (bounded/selective output, e.g. TopN/Limit/visible-column
filter/global agg),
+ // record it so that consumers of this CTE can inherit the
effectiveness.
+ // Without this, a join whose build side is a CTE consumer of such a
producer gets
+ // its runtime filters pruned as "ineffective" merely because the
consumer's
+ // statistics are unknown (always the case for external tables).
cteAnchor.child(0).accept(this, context);
+ RuntimeFilterContext rfCtx = context.getRuntimeFilterContext();
+ // Only a producer whose effectiveness is a property of the whole
relation, rather than of
+ // one join key, may pass it to consumers.
+ // - NATIVE is also set when a join's build side is selective, and
that is a property of one
+ // join key, not of the relation: for c = Project(A.k, B.v) ->
LeftJoin(A, Limit(1) -> B)
+ // the left join still preserves every A.k, yet the join visitor
marks the producer
+ // NATIVE. A consumer that only keeps B.v live would then inherit a
flag it cannot
+ // justify and keep a runtime filter that rejects no row. So NATIVE
needs the operator
+ // test below.
+ // - REF means "this relation is the target of a runtime filter".
Unlike a NATIVE bound it
+ // is a property of one column, and an output that does not shrink
can borrow it: in
+ // c = Project(A.k, B.v) -> LeftJoin(A, Join(B, D, B.v = D.v)) an
RF from D marks B REF,
+ // the inner join inherits it, and the left join inherits it again
although it preserves
+ // every A row. Tying REF to the keys it actually reduces would be
exact; we keep the
+ // inheritance anyway because the two error directions are not
symmetric: retaining an
+ // occasionally useless filter costs one more build/probe, while
dropping a producer that
+ // genuinely shrank makes the outer scan read the whole Hive table.
+ if (rfCtx.isEffectiveSrcNode(cteAnchor.child(0))) {
+ RuntimeFilterContext.EffectiveSrcType producerType =
+ rfCtx.getEffectiveSrcType(cteAnchor.child(0));
+ if (producerType == RuntimeFilterContext.EffectiveSrcType.REF
+ || (producerType ==
RuntimeFilterContext.EffectiveSrcType.NATIVE
+ &&
hasRelationGlobalEffectiveness(cteAnchor.child(0).child(0)))) {
+ effectiveCteProducers.put(cteAnchor.getCteId(), producerType);
+ }
+ }
cteAnchor.child(1).accept(this, context);
return cteAnchor;
}
+ /**
+ * Whether NATIVE effectiveness on this plan is a property of the whole
relation rather than of
+ * one join key, and may therefore be handed to another subtree through a
CTE.
+ *
+ * <p>The operator either bounds the row count on its own (Limit, TopN,
AssertNumRows, Intersect,
+ * a no-group-by aggregate, a bounded PartitionTopN) or inherits a bound
from a child it cannot
+ * add rows on top of (a grouped aggregate, or the row-preserving Project
/ Distribute / Window);
+ * or it restricts the whole relation with a predicate on a visible column.
+ * A join qualifies neither way: the join visitor also marks a join NATIVE
when its build side is
+ * selective, and that is a property of one join key — a LEFT JOIN
preserves every probe row even
+ * when its build side is limited.
+ */
+ private boolean hasRelationGlobalEffectiveness(Plan plan) {
+ if (plan instanceof PhysicalLimit || plan instanceof PhysicalTopN
+ || plan instanceof PhysicalAssertNumRows || plan instanceof
PhysicalIntersect) {
+ return true;
+ }
+ if (plan instanceof PhysicalHashAggregate) {
+ // A no-group-by aggregate returns exactly one row. A grouped one
returns at most one
+ // row per group, so it never adds rows either: a bound proven on
its input still holds
+ // on its output. Carrying that through matters because a
materialized CTE keeps its
+ // aggregate alive whenever some consumer selects the aggregate's
argument.
+ return ((PhysicalHashAggregate<?>)
plan).getGroupByExpressions().isEmpty()
+ || carryBoundFromSingleChild(plan);
+ }
+ if (plan instanceof PhysicalPartitionTopN) {
+ PhysicalPartitionTopN<?> topN = (PhysicalPartitionTopN<?>) plan;
+ return topN.hasGlobalLimit()
+ || (topN.getFunction() == WindowFuncType.ROW_NUMBER
+ && topN.getPartitionKeys().isEmpty());
+ }
+ // A predicate on a visible column is exactly what visitPhysicalFilter
already treats as an
+ // effective source. Unlike a join's build-side selectivity, that
predicate restricts the
+ // whole relation, so it is safe to hand on through a CTE.
+ if (plan instanceof PhysicalFilter) {
+ return hasVisibleColumnPredicate((PhysicalFilter<? extends Plan>)
plan);
+ }
+ // Project, Distribute and Window keep exactly one output row per
input row, so a bound
+ // proven below them still holds above them. Window matters for the
same reason as the
+ // aggregate: a materialized CTE keeps a row-preserving window alive
whenever some consumer
+ // selects one of its window columns.
+ // The walk stops at a join: a join's row count is a property of its
join keys, not of the
+ // relation, so a bound below a join says nothing about the join's
output.
+ if (plan instanceof PhysicalProject || plan instanceof
PhysicalDistribute
+ || plan instanceof PhysicalWindow) {
+ return carryBoundFromSingleChild(plan);
+ }
+ return false;
Review Comment:
[P2] Preserve the proven source bound through a chained materialized CTE
With CTE materialization enabled and `inline_cte_referenced_threshold=0`,
consider `WITH c1 AS (SELECT MAX(p) AS p FROM t), c2 AS (SELECT p FROM c1)
SELECT t.* FROM t JOIN c2 ON t.p=c2.p`. The reduced physical shape is
`Producer(c1) -> GlobalAggregate`, followed by `Producer(c2) ->
Project/Distribute -> Consumer(c1)`. Both producers represent at most one row.
`visitPhysicalCTEConsumer` marks the first consumer NATIVE, and the generic
visitor carries that mark to the second producer. However,
`visitPhysicalCTEAnchor` rechecks the second producer with
`hasRelationGlobalEffectiveness`, which reaches `PhysicalCTEConsumer` here and
returns false. The second producer is therefore absent from
`effectiveCteProducers`, and its consumer loses the bound. With unknown Hive
column statistics, the outer join can then prune the useful partition runtime
filter and scan all partitions instead.
Please carry the proven relation-level source information through CTE
consumers, using the producer/consumer output mapping, and add a
chained-materialized-CTE plan test. Avoid treating arbitrary NATIVE marks as a
relation bound, since join-key selectivity uses the same mark. This path is
inferred from the implementation; I have not executed the SQL above.
--
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]
