github-actions[bot] commented on code in PR #68340: URL: https://github.com/apache/doris/pull/68340#discussion_r4062787220
########## regression-test/suites/nereids_rules_p0/cte/test_cte_shared_producer_min_max_runtime_filter.groovy: ########## @@ -0,0 +1,93 @@ +// 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)" + + // 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 Review Comment: [P2] Add a plan-level oracle for the safe-push path. This same-direction query returns the same rows whether the MIN filter is pushed into the shared producer, remains on the consumers, or is not generated, while the FE tests only call `selectPushableRuntimeFilters` on mocks. Thus the refactored producer insertion, original-filter cleanup, and translation can regress without any added assertion failing. Please add an `EXPLAIN SHAPE PLAN` or real postprocessor assertion that the same-direction filter reaches the producer and that the opposite-direction MIN/MAX filters do not. ########## fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/RuntimeFilterGenerator.java: ########## @@ -881,20 +911,80 @@ public static Slot checkTargetChild(Expression leftChild) { } /** - * Check whether runtime filters on CTE consumers can be pushed into their shared CTE producer. + * 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 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 boolean canPushDownRuntimeFiltersIntoCTEProducer( - List<RuntimeFilter> rfsToPushDown, CTEId cteId) { - if (rfsToPushDown.isEmpty()) { - LOG.warn("Skip pushing runtime filters into CTE producer because no runtime filters exist for cteId: {}", - cteId); - return false; + 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) { + 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() + .map(rf -> (PhysicalCTEConsumer) rf.getTargetScan()) + .collect(Collectors.toSet()); + if (consumersApplying.size() == consumers.size()) { + pushable.add(rfsOfIdentity); + } else { + LOG.warn("Skip pushing runtime filters into CTE producer because only {} of {} consumers of" Review Comment: [P2] Avoid WARN for this expected optimization skip. The valid opposite-direction query added by this PR creates separate MIN and MAX identities, so it emits two warnings during ordinary planning; other consumer-specific identities add more. Nothing is broken and no operator action is possible, so repeated materialized-CTE queries can fill the warning log with normal decisions. Please remove this log or lower it to DEBUG. ########## fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/RuntimeFilterGenerator.java: ########## @@ -881,20 +911,80 @@ public static Slot checkTargetChild(Expression leftChild) { } /** - * Check whether runtime filters on CTE consumers can be pushed into their shared CTE producer. + * 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 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 boolean canPushDownRuntimeFiltersIntoCTEProducer( - List<RuntimeFilter> rfsToPushDown, CTEId cteId) { - if (rfsToPushDown.isEmpty()) { - LOG.warn("Skip pushing runtime filters into CTE producer because no runtime filters exist for cteId: {}", - cteId); - return false; + 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) { + 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() + .map(rf -> (PhysicalCTEConsumer) rf.getTargetScan()) + .collect(Collectors.toSet()); + if (consumersApplying.size() == consumers.size()) { + pushable.add(rfsOfIdentity); + } else { + LOG.warn("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 Expression producerTargetExpression; + + private FilterIdentity(RuntimeFilter rf, CTEId cteId) { Review Comment: [P1] Preserve null-aware equality in this identity. `nullAware` is part of the runtime-filter predicate, but it is not represented by `type`, `minMaxType`, or the producer target. A concrete triggering tree is: ```text HashJoin INNER [c1.k <=> b.x] (upper RF is null-aware) CTEConsumer c1 HashJoin RIGHT OUTER [c2.k = b.x] (deeper RF is ordinary) CTEConsumer c2 Scan b(x) (contains NULL) ``` The right outer join preserves `b.x = NULL`, and both RFs have the same source ExprId, type, placeholder min/max mode, and mapped producer target. They therefore group here, after which `pushDownIdenticalFilters` selects the deeper ordinary RF. Legacy translation derives `nullAware=false` from that builder's `=`, so the producer filter removes `t.k IS NULL` before the upper `<=>` join can match it; the remaining upper consumer filter cannot restore the row. Please include the effective null-aware mode in the identity (or otherwise reject this substitution) and add a shared-CTE regression for this NULL/right-outer shape. -- 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]
