This is an automated email from the ASF dual-hosted git repository.

CalvinKirs pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git


The following commit(s) were added to refs/heads/master by this push:
     new f878b28f2b4 [fix](fe) Preserve runtime filter sources through hidden 
filters (#69662)
f878b28f2b4 is described below

commit f878b28f2b46ad319f8b6ed85caf51f683d35832
Author: Calvin Kirs <[email protected]>
AuthorDate: Wed Oct 28 11:14:42 2026 +0800

    [fix](fe) Preserve runtime filter sources through hidden filters (#69662)
    
    ### What problem does this PR solve?
    
    Issue Number: None
    
    Related PR: #28010
    
    Problem Summary:
    
    A hidden delete-sign Filter can break a useful runtime-filter chain in a
    three-table join. Consider this reduced physical plan (Project and
    Distribute nodes omitted):
    
    ```text
    Join (e.bloc_id = b.id)                 RF1: b.id -> e.bloc_id
      Join (f.agent_code = e.agent_code)   RF0: e.agent_code -> f.agent_code
        Scan fact f
        Filter (e.__DORIS_DELETE_SIGN__ = 0)
          Scan agent e                    receives RF1
      Filter (b.bloc_code = 'keep')
        Scan bloc b
    ```
    
    The selective bloc predicate keeps RF1. The agent scan is then
    recognized as an effective source because it receives RF1, but
    `visitPhysicalFilter` previously discarded that information when the
    Filter contained only hidden columns. The inner join could therefore
    prune RF0 based on unknown or non-selective static join-column
    statistics, losing the filtering chain and increasing fact-table scans.
    
    Reuse the existing `propagateEffectiveSrc` helper before classifying
    visible predicates. Hidden-only Filters inherit the child's existing
    REF/NATIVE type; a visible predicate still marks its Filter NATIVE. A
    delete-sign predicate with no effective child source still does not keep
    an otherwise ineffective RF. HashAggregate's existing helper behavior is
    unchanged.
    
    Add planner unit tests covering known/non-selective and unknown
    statistics, hidden-only negative controls, and a visible-predicate
    control. Add EXPLAIN regression coverage for the retained chain and both
    controls.
---
 .../processor/post/RuntimeFilterPruner.java        |  17 +--
 .../processor/post/RuntimeFilterPrunerTest.java    | 131 +++++++++++++++++++++
 .../runtime_filter/rf_prune_hidden_filter.groovy   |  76 ++++++++++++
 3 files changed, 216 insertions(+), 8 deletions(-)

diff --git 
a/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/RuntimeFilterPruner.java
 
b/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/RuntimeFilterPruner.java
index 91ff8887385..1e35768cf19 100644
--- 
a/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/RuntimeFilterPruner.java
+++ 
b/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/RuntimeFilterPruner.java
@@ -202,7 +202,8 @@ public class RuntimeFilterPruner extends PlanPostProcessor {
 
     @Override
     public PhysicalFilter visitPhysicalFilter(PhysicalFilter<? extends Plan> 
filter, CascadesContext context) {
-        filter.child().accept(this, context);
+        // Hidden predicates do not add selectivity, but must preserve the 
child's effective source.
+        propagateEffectiveSrc(filter, context);
 
         boolean visibleFilter = false;
 
@@ -255,21 +256,21 @@ public class RuntimeFilterPruner extends 
PlanPostProcessor {
 
     /**
      * Visit child and propagate effective source type if applicable.
-     * Shared by visitPhysicalHashAggregate.
+     * Shared by visitPhysicalFilter and visitPhysicalHashAggregate.
      *
      * Note: agg is not regarded as an effective source itself. For example:
      * q1: A join (select x, sum(y) as z from B group by x) T on A.a = T.x
      * q2: A join (select x, sum(y) as z from B group by x) T on A.a = T.z
      * RF on q1 is not effective, but RF on q2 is. Let RF judge by ndv.
      */
-    private <T extends Plan> T propagateEffectiveSrc(T aggregate, 
CascadesContext context) {
+    private <T extends Plan> T propagateEffectiveSrc(T plan, CascadesContext 
context) {
         RuntimeFilterContext ctx = context.getRuntimeFilterContext();
-        aggregate.child(0).accept(this, context);
-        if (ctx.isEffectiveSrcNode(aggregate.child(0))) {
-            RuntimeFilterContext.EffectiveSrcType childType = 
ctx.getEffectiveSrcType(aggregate.child(0));
-            ctx.addEffectiveSrcNode(aggregate, childType);
+        plan.child(0).accept(this, context);
+        if (ctx.isEffectiveSrcNode(plan.child(0))) {
+            RuntimeFilterContext.EffectiveSrcType childType = 
ctx.getEffectiveSrcType(plan.child(0));
+            ctx.addEffectiveSrcNode(plan, childType);
         }
-        return aggregate;
+        return plan;
     }
 
     /**
diff --git 
a/fe/fe-core/src/test/java/org/apache/doris/nereids/processor/post/RuntimeFilterPrunerTest.java
 
b/fe/fe-core/src/test/java/org/apache/doris/nereids/processor/post/RuntimeFilterPrunerTest.java
new file mode 100644
index 00000000000..5557b0bb570
--- /dev/null
+++ 
b/fe/fe-core/src/test/java/org/apache/doris/nereids/processor/post/RuntimeFilterPrunerTest.java
@@ -0,0 +1,131 @@
+// 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.processor.post;
+
+import org.apache.doris.nereids.glue.translator.PhysicalPlanTranslator;
+import org.apache.doris.nereids.glue.translator.PlanTranslatorContext;
+import org.apache.doris.nereids.trees.expressions.EqualTo;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalHashJoin;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalPlan;
+import org.apache.doris.nereids.trees.plans.physical.RuntimeFilter;
+import org.apache.doris.nereids.util.PlanChecker;
+import org.apache.doris.statistics.model.ColumnStatistic;
+import org.apache.doris.statistics.model.ColumnStatisticBuilder;
+import org.apache.doris.thrift.TRuntimeFilterType;
+import org.apache.doris.utframe.TestWithFeService;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.List;
+import java.util.Set;
+
+class RuntimeFilterPrunerTest extends TestWithFeService {
+    @Override
+    protected void runBeforeAll() throws Exception {
+        createDatabase("rf_pruner_test");
+        useDatabase("rf_pruner_test");
+        connectContext.getSessionVariable().setDisableJoinReorder(true);
+        
connectContext.getSessionVariable().setDisableNereidsRules("PRUNE_EMPTY_PARTITION");
+        connectContext.getSessionVariable().setEnableRuntimeFilterPrune(true);
+        
connectContext.getSessionVariable().setRuntimeFilterType(TRuntimeFilterType.BLOOM.getValue());
+
+        createTable("CREATE TABLE rf_prune_fact (id INT, agent_code INT, 
username VARCHAR(32)) "
+                + "DUPLICATE KEY(id) DISTRIBUTED BY HASH(id) BUCKETS 1 "
+                + "PROPERTIES('replication_num'='1')");
+        // Merge-on-read keeps a hidden delete-sign Filter above the agent 
scan.
+        createTable("CREATE TABLE rf_prune_agent (agent_code INT, bloc_id INT) 
"
+                + "UNIQUE KEY(agent_code) DISTRIBUTED BY HASH(agent_code) 
BUCKETS 1 "
+                + "PROPERTIES('replication_num'='1', 
'enable_unique_key_merge_on_write'='false')");
+        createTable("CREATE TABLE rf_prune_bloc (id INT, bloc_code 
VARCHAR(32)) "
+                + "DUPLICATE KEY(id) DISTRIBUTED BY HASH(id) BUCKETS 1 "
+                + "PROPERTIES('replication_num'='1')");
+    }
+
+    @ParameterizedTest
+    @ValueSource(booleans = {true, false})
+    void testPreserveRuntimeFilterThroughHiddenFilter(boolean knownStats) {
+        RuntimeFilterContext context = getRuntimeFilterContext(
+                "SELECT f.id FROM rf_prune_fact f "
+                        + "JOIN [broadcast] rf_prune_agent e ON f.agent_code = 
e.agent_code "
+                        + "JOIN [broadcast] rf_prune_bloc b ON e.bloc_id = 
b.id "
+                        + "WHERE b.bloc_code = 'keep' AND f.username LIKE 
'%user%'", knownStats);
+
+        // b.id -> e.bloc_id must keep e.agent_code -> f.agent_code effective, 
even though
+        // e has no visible predicate and its static join-column statistics 
are non-selective.
+        List<RuntimeFilter> filters = context.getNereidsRuntimeFilter();
+        Assertions.assertEquals(2, filters.size());
+        Assertions.assertTrue(filters.stream().anyMatch(filter ->
+                filter.getTargetSlot().getName().equals("agent_code")));
+        Assertions.assertTrue(filters.stream().anyMatch(filter ->
+                filter.getTargetSlot().getName().equals("bloc_id")));
+        Assertions.assertTrue(context.prunedRF.isEmpty());
+    }
+
+    @ParameterizedTest
+    @ValueSource(booleans = {true, false})
+    void testHiddenFilterAloneDoesNotKeepIneffectiveRuntimeFilter(boolean 
knownStats) {
+        RuntimeFilterContext context = getRuntimeFilterContext(
+                "SELECT f.id FROM rf_prune_fact f "
+                        + "JOIN [broadcast] rf_prune_agent e ON f.agent_code = 
e.agent_code", knownStats);
+
+        Assertions.assertTrue(context.getNereidsRuntimeFilter().isEmpty());
+        Assertions.assertEquals(1, context.prunedRF.size());
+    }
+
+    @Test
+    void testVisibleFilterKeepsRuntimeFilter() {
+        RuntimeFilterContext context = getRuntimeFilterContext(
+                "SELECT f.id FROM rf_prune_fact f "
+                        + "JOIN [broadcast] rf_prune_agent e ON f.agent_code = 
e.agent_code "
+                        + "WHERE e.bloc_id = 1", true);
+
+        Assertions.assertEquals(1, context.getNereidsRuntimeFilter().size());
+        Assertions.assertTrue(context.prunedRF.isEmpty());
+    }
+
+    private RuntimeFilterContext getRuntimeFilterContext(String sql, boolean 
knownStats) {
+        PlanChecker checker = 
PlanChecker.from(connectContext).analyze(sql).rewrite().optimize();
+        PhysicalPlan plan = checker.getBestPlanTree();
+        ColumnStatistic nonSelectiveStats = new ColumnStatisticBuilder(100)
+                .setNdv(100)
+                .setNumNulls(0)
+                .setAvgSizeByte(4)
+                .setMinValue(1)
+                .setMaxValue(100)
+                .build();
+        if (knownStats) {
+            Set<PhysicalHashJoin<? extends Plan, ? extends Plan>> joins =
+                    plan.collect(PhysicalHashJoin.class::isInstance);
+            for (PhysicalHashJoin<? extends Plan, ? extends Plan> join : 
joins) {
+                EqualTo equalTo = (EqualTo) join.getEqualToConjuncts().get(0);
+                Slot probeSlot = 
equalTo.child(0).getInputSlots().iterator().next();
+                Slot buildSlot = 
equalTo.child(1).getInputSlots().iterator().next();
+                join.left().getStats().addColumnStats(probeSlot, 
nonSelectiveStats);
+                join.right().getStats().addColumnStats(buildSlot, 
nonSelectiveStats);
+            }
+        }
+        plan = new 
PlanPostProcessors(checker.getCascadesContext()).process(plan);
+        new PhysicalPlanTranslator(new 
PlanTranslatorContext(checker.getCascadesContext())).translatePlan(plan);
+        return checker.getCascadesContext().getRuntimeFilterContext();
+    }
+}
diff --git 
a/regression-test/suites/nereids_rules_p0/runtime_filter/rf_prune_hidden_filter.groovy
 
b/regression-test/suites/nereids_rules_p0/runtime_filter/rf_prune_hidden_filter.groovy
new file mode 100644
index 00000000000..195ee8dbf0d
--- /dev/null
+++ 
b/regression-test/suites/nereids_rules_p0/runtime_filter/rf_prune_hidden_filter.groovy
@@ -0,0 +1,76 @@
+// 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("rf_prune_hidden_filter") {
+    sql "set enable_runtime_filter_prune=true"
+    sql "set runtime_filter_type=2"
+    sql "set disable_join_reorder=true"
+    sql "set disable_nereids_rules=PRUNE_EMPTY_PARTITION"
+
+    sql "drop table if exists rf_prune_hidden_fact"
+    sql """
+        create table rf_prune_hidden_fact (id int, agent_code int, username 
varchar(32))
+        duplicate key(id)
+        distributed by hash(id) buckets 1
+        properties("replication_num"="1")
+    """
+    sql "drop table if exists rf_prune_hidden_agent"
+    sql """
+        create table rf_prune_hidden_agent (agent_code int, bloc_id int)
+        unique key(agent_code)
+        distributed by hash(agent_code) buckets 1
+        properties("replication_num"="1", 
"enable_unique_key_merge_on_write"="false")
+    """
+    sql "drop table if exists rf_prune_hidden_bloc"
+    sql """
+        create table rf_prune_hidden_bloc (id int, bloc_code varchar(32))
+        duplicate key(id)
+        distributed by hash(id) buckets 1
+        properties("replication_num"="1")
+    """
+
+    // The agent scan receives a bloc_id RF below its hidden delete-sign 
Filter.
+    // That Filter must pass the effective source through to the agent_code 
join.
+    explain {
+        sql """
+            shape plan select f.id from rf_prune_hidden_fact f
+            join [broadcast] rf_prune_hidden_agent e on f.agent_code = 
e.agent_code
+            join [broadcast] rf_prune_hidden_bloc b on e.bloc_id = b.id
+            where b.bloc_code = 'keep' and f.username like '%user%'
+        """
+        contains "PhysicalOlapScan[rf_prune_hidden_fact(f)] apply RFs:"
+        contains "PhysicalOlapScan[rf_prune_hidden_agent] apply RFs:"
+    }
+
+    // A hidden delete-sign predicate alone must not make an ineffective RF 
useful.
+    explain {
+        sql """
+            shape plan select f.id from rf_prune_hidden_fact f
+            join [broadcast] rf_prune_hidden_agent e on f.agent_code = 
e.agent_code
+        """
+        notContains "build RFs"
+    }
+
+    explain {
+        sql """
+            shape plan select f.id from rf_prune_hidden_fact f
+            join [broadcast] rf_prune_hidden_agent e on f.agent_code = 
e.agent_code
+            where e.bloc_id = 1
+        """
+        contains "PhysicalOlapScan[rf_prune_hidden_fact(f)] apply RFs:"
+    }
+}


---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]

Reply via email to