enable rf prune

fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/MergeProjectPostProcessor.java

@@ -37,8 +37,12 @@ public PhysicalProject visitPhysicalProject(PhysicalProject<? extends Plan> proj
         Plan newChild = child.accept(this, ctx);
         if (newChild instanceof PhysicalProject) {
             List<NamedExpression> projections = project.mergeProjections((PhysicalProject) newChild);
-            return project.withProjectionsAndChild(projections, newChild.child(0));
+            return (PhysicalProject) project
+                    .withProjectionsAndChild(projections, newChild.child(0))
+                    .copyStatsAndGroupIdFrom(project);
         }
-        return child != newChild ? project.withChildren(Lists.newArrayList(newChild)) : project;
+        return child != newChild
+                ? (PhysicalProject) project.withChildren(Lists.newArrayList(newChild)).copyStatsAndGroupIdFrom(project)
+                : project;
     }
 }

fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/PlanPostProcessors.java

@@ -61,6 +61,7 @@ public List<PlanPostProcessor> getProcessors() {
         builder.add(new PushdownFilterThroughProject());
         builder.add(new MergeProjectPostProcessor());
         builder.add(new RecomputeLogicalPropertiesProcessor());
+        builder.add(new TopNScanOpt());
         // after generate rf, DO NOT replace PLAN NODE
         builder.add(new FragmentProcessor());
         if (!cascadesContext.getConnectContext().getSessionVariable().getRuntimeFilterMode()
@@ -71,7 +72,6 @@ public List<PlanPostProcessor> getProcessors() {
             }
         }
         builder.add(new Validator());
-        builder.add(new TopNScanOpt());
         return builder.build();
     }
 }

fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/PushdownFilterThroughProject.java

@@ -19,6 +19,7 @@
 
 import org.apache.doris.nereids.CascadesContext;
 import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.physical.AbstractPhysicalPlan;
 import org.apache.doris.nereids.trees.plans.physical.PhysicalFilter;
 import org.apache.doris.nereids.trees.plans.physical.PhysicalProject;
 import org.apache.doris.nereids.util.ExpressionUtils;
@@ -31,14 +32,20 @@ public class PushdownFilterThroughProject extends PlanPostProcessor {
     public Plan visitPhysicalFilter(PhysicalFilter<? extends Plan> filter, CascadesContext context) {
         Plan child = filter.child();
         if (!(child instanceof PhysicalProject)) {
-            return filter.withChildren(child.accept(this, context));
+            Plan newChild = child.accept(this, context);
+            if (newChild == child) {
+                return filter;
+            } else {
+                return ((AbstractPhysicalPlan) filter.withChildren(child.accept(this, context)))
+                        .copyStatsAndGroupIdFrom(filter);
+            }
         }
 
         PhysicalProject<? extends Plan> project = (PhysicalProject<? extends Plan>) child;
         PhysicalFilter<? extends Plan> newFilter = filter.withConjunctsAndChild(
                 ExpressionUtils.replace(filter.getConjuncts(), project.getAliasToProducer()),
                 project.child());
-
-        return project.withChildren(newFilter.accept(this, context));
+        return ((PhysicalProject) project.withChildren(newFilter.accept(this, context)))
+                .copyStatsAndGroupIdFrom(project);
     }
 }

fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/RecomputeLogicalPropertiesProcessor.java

@@ -18,22 +18,17 @@
 package org.apache.doris.nereids.processor.post;
 
 import org.apache.doris.nereids.CascadesContext;
-import org.apache.doris.nereids.trees.plans.AbstractPlan;
 import org.apache.doris.nereids.trees.plans.Plan;
-import org.apache.doris.nereids.trees.plans.physical.PhysicalPlan;
-import org.apache.doris.nereids.util.MutableState;
+import org.apache.doris.nereids.trees.plans.physical.AbstractPhysicalPlan;
 
 /**
  * merge consecutive projects
  */
 public class RecomputeLogicalPropertiesProcessor extends PlanPostProcessor {
     @Override
     public Plan visit(Plan plan, CascadesContext ctx) {
-        PhysicalPlan physicalPlan = (PhysicalPlan) visitChildren(this, plan, ctx);
-        physicalPlan = physicalPlan.resetLogicalProperties();
-        physicalPlan = physicalPlan.withPhysicalPropertiesAndStats(physicalPlan.getPhysicalProperties(),
-                ((AbstractPlan) plan).getStats());
-        physicalPlan.setMutableState(MutableState.KEY_GROUP, plan.getGroupIdAsString());
-        return physicalPlan;
+        AbstractPhysicalPlan newPlan = (AbstractPhysicalPlan) visitChildren(this, plan, ctx);
+        return ((AbstractPhysicalPlan) newPlan.resetLogicalProperties())
+                .copyStatsAndGroupIdFrom((AbstractPhysicalPlan) plan);
     }
 }

fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/TopNScanOpt.java

@@ -44,7 +44,7 @@ public PhysicalTopN<? extends Plan> visitPhysicalTopN(PhysicalTopN<? extends Pla
         Plan child = topN.child().accept(this, ctx);
         topN = rewriteTopN(topN);
         if (child != topN.child()) {
-            topN.withChildren(child);
+            topN = ((PhysicalTopN) topN.withChildren(child)).copyStatsAndGroupIdFrom(topN);
         }
         return topN;
     }
@@ -54,11 +54,11 @@ public Plan visitPhysicalDeferMaterializeTopN(PhysicalDeferMaterializeTopN<? ext
             CascadesContext context) {
         Plan child = topN.child().accept(this, context);
         if (child != topN.child()) {
-            topN = topN.withChildren(ImmutableList.of(child));
+            topN = topN.withChildren(ImmutableList.of(child)).copyStatsAndGroupIdFrom(topN);
         }
         PhysicalTopN<? extends Plan> rewrittenTopN = rewriteTopN(topN.getPhysicalTopN());
         if (topN.getPhysicalTopN() != rewrittenTopN) {
-            topN = topN.withPhysicalTopN(rewrittenTopN);
+            topN = topN.withPhysicalTopN(rewrittenTopN).copyStatsAndGroupIdFrom(topN);
         }
         return topN;
     }
@@ -103,7 +103,7 @@ private PhysicalTopN<? extends Plan> rewriteTopN(PhysicalTopN<? extends Plan> to
         olapScan = (OlapScan) child;
 
         if (olapScan.getTable().isDupKeysOrMergeOnWrite()) {
-            return topN.withEnableRuntimeFilter(true);
+            return topN.withEnableRuntimeFilter(true).copyStatsAndGroupIdFrom(topN);
         }
 
         return topN;

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/AbstractPhysicalPlan.java

@@ -32,6 +32,7 @@
 import org.apache.doris.nereids.trees.plans.Explainable;
 import org.apache.doris.nereids.trees.plans.Plan;
 import org.apache.doris.nereids.trees.plans.PlanType;
+import org.apache.doris.nereids.util.MutableState;
 import org.apache.doris.planner.RuntimeFilterId;
 import org.apache.doris.qe.ConnectContext;
 import org.apache.doris.statistics.Statistics;
@@ -136,4 +137,11 @@ public boolean pushDownRuntimeFilter(CascadesContext context, IdGenerator<Runtim
     public Plan getExplainPlan(ConnectContext ctx) {
         return this;
     }
+
+    public <T extends AbstractPhysicalPlan> T copyStatsAndGroupIdFrom(T from) {
+        T newPlan = (T) withPhysicalPropertiesAndStats(
+                from.getPhysicalProperties(), from.getStats());
+        newPlan.setMutableState(MutableState.KEY_GROUP, from.getGroupIdAsString());
+        return newPlan;
+    }
 }

fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java

@@ -904,7 +904,7 @@ public void setMaxJoinNumberOfReorder(int maxJoinNumberOfReorder) {
     private double broadcastHashtableMemLimitPercentage = 0.2;
 
     @VariableMgr.VarAttr(name = ENABLE_RUNTIME_FILTER_PRUNE, needForward = true)
-    public boolean enableRuntimeFilterPrune = false;
+    public boolean enableRuntimeFilterPrune = true;
 
     /**
      * The client can pass some special information by setting this session variable in the format: "k1:v1;k2:v2".

fe/fe-core/src/test/java/org/apache/doris/nereids/postprocess/RuntimeFilterTest.java

@@ -45,6 +45,8 @@ public void runBeforeAll() throws Exception {
         super.runBeforeAll();
         connectContext.getSessionVariable().setRuntimeFilterMode("Global");
         connectContext.getSessionVariable().setRuntimeFilterType(8);
+        connectContext.getSessionVariable().setEnableRuntimeFilterPrune(false);
+        connectContext.getSessionVariable().expandRuntimeFilterByInnerJoin = false;
     }
 
     @Test

regression-test/suites/nereids_tpcds_shape_sf100_p0/ddl/rf.tmpl

@@ -30,6 +30,7 @@ suite("ds_rf{--}") {
     sql 'set broadcast_row_count_limit = 30000000'
     sql 'set enable_nereids_timeout = false'
     sql 'set enable_pipeline_engine=true'
+    sql 'set enable_runtime_filter_prune=false'
     String stmt = '''
     explain physical plan
     {query}

regression-test/suites/nereids_tpcds_shape_sf100_p0/rf/ds_rf1.groovy

@@ -30,6 +30,8 @@ suite("ds_rf1") {
     sql 'set broadcast_row_count_limit = 30000000'
     sql 'set enable_nereids_timeout = false'
     sql 'set enable_pipeline_engine=true'
+    sql 'set enable_runtime_filter_prune=false'
+    sql 'set expand_runtime_filter_by_inner_join=false'
     String stmt = '''
     explain physical plan
     with customer_total_return as

regression-test/suites/nereids_tpcds_shape_sf100_p0/rf/ds_rf10.groovy

@@ -30,6 +30,8 @@ suite("ds_rf10") {
     sql 'set broadcast_row_count_limit = 30000000'
     sql 'set enable_nereids_timeout = false'
     sql 'set enable_pipeline_engine=true'
+    sql 'set enable_runtime_filter_prune=false'
+    sql 'set expand_runtime_filter_by_inner_join=false'
     String stmt = '''
     explain physical plan
     select  

regression-test/suites/nereids_tpcds_shape_sf100_p0/rf/ds_rf11.groovy

@@ -30,6 +30,8 @@ suite("ds_rf11") {
     sql 'set broadcast_row_count_limit = 30000000'
     sql 'set enable_nereids_timeout = false'
     sql 'set enable_pipeline_engine=true'
+    sql 'set enable_runtime_filter_prune=false'
+    sql 'set expand_runtime_filter_by_inner_join=false'
     String stmt = '''
     explain physical plan
     with year_total as (

regression-test/suites/nereids_tpcds_shape_sf100_p0/rf/ds_rf12.groovy

@@ -30,6 +30,8 @@ suite("ds_rf12") {
     sql 'set broadcast_row_count_limit = 30000000'
     sql 'set enable_nereids_timeout = false'
     sql 'set enable_pipeline_engine=true'
+    sql 'set enable_runtime_filter_prune=false'
+    sql 'set expand_runtime_filter_by_inner_join=false'
     String stmt = '''
     explain physical plan
     select  i_item_id

regression-test/suites/nereids_tpcds_shape_sf100_p0/rf/ds_rf13.groovy

@@ -30,6 +30,8 @@ suite("ds_rf13") {
     sql 'set broadcast_row_count_limit = 30000000'
     sql 'set enable_nereids_timeout = false'
     sql 'set enable_pipeline_engine=true'
+    sql 'set enable_runtime_filter_prune=false'
+    sql 'set expand_runtime_filter_by_inner_join=false'
     String stmt = '''
     explain physical plan
     select avg(ss_quantity)

regression-test/suites/nereids_tpcds_shape_sf100_p0/rf/ds_rf14.groovy

@@ -30,6 +30,8 @@ suite("ds_rf14") {
     sql 'set broadcast_row_count_limit = 30000000'
     sql 'set enable_nereids_timeout = false'
     sql 'set enable_pipeline_engine=true'
+    sql 'set enable_runtime_filter_prune=false'
+    sql 'set expand_runtime_filter_by_inner_join=false'
     String stmt = '''
     explain physical plan
     with  cross_items as

regression-test/suites/nereids_tpcds_shape_sf100_p0/rf/ds_rf15.groovy

@@ -30,6 +30,8 @@ suite("ds_rf15") {
     sql 'set broadcast_row_count_limit = 30000000'
     sql 'set enable_nereids_timeout = false'
     sql 'set enable_pipeline_engine=true'
+    sql 'set enable_runtime_filter_prune=false'
+    sql 'set expand_runtime_filter_by_inner_join=false'
     String stmt = '''
     explain physical plan
     select  ca_zip

regression-test/suites/nereids_tpcds_shape_sf100_p0/rf/ds_rf16.groovy

@@ -30,6 +30,8 @@ suite("ds_rf16") {
     sql 'set broadcast_row_count_limit = 30000000'
     sql 'set enable_nereids_timeout = false'
     sql 'set enable_pipeline_engine=true'
+    sql 'set enable_runtime_filter_prune=false'
+    sql 'set expand_runtime_filter_by_inner_join=false'
     String stmt = '''
     explain physical plan
     select  

regression-test/suites/nereids_tpcds_shape_sf100_p0/rf/ds_rf17.groovy

@@ -30,6 +30,8 @@ suite("ds_rf17") {
     sql 'set broadcast_row_count_limit = 30000000'
     sql 'set enable_nereids_timeout = false'
     sql 'set enable_pipeline_engine=true'
+    sql 'set enable_runtime_filter_prune=false'
+    sql 'set expand_runtime_filter_by_inner_join=false'
     String stmt = '''
     explain physical plan
     select  i_item_id

regression-test/suites/nereids_tpcds_shape_sf100_p0/rf/ds_rf18.groovy

@@ -30,6 +30,8 @@ suite("ds_rf18") {
     sql 'set broadcast_row_count_limit = 30000000'
     sql 'set enable_nereids_timeout = false'
     sql 'set enable_pipeline_engine=true'
+    sql 'set enable_runtime_filter_prune=false'
+    sql 'set expand_runtime_filter_by_inner_join=false'
     String stmt = '''
     explain physical plan
     select  i_item_id,

regression-test/suites/nereids_tpcds_shape_sf100_p0/rf/ds_rf19.groovy

@@ -30,6 +30,8 @@ suite("ds_rf19") {
     sql 'set broadcast_row_count_limit = 30000000'
     sql 'set enable_nereids_timeout = false'
     sql 'set enable_pipeline_engine=true'
+    sql 'set enable_runtime_filter_prune=false'
+    sql 'set expand_runtime_filter_by_inner_join=false'
     String stmt = '''
     explain physical plan
     select  i_brand_id brand_id, i_brand brand, i_manufact_id, i_manufact,

regression-test/suites/nereids_tpcds_shape_sf100_p0/rf/ds_rf2.groovy

@@ -30,6 +30,8 @@ suite("ds_rf2") {
     sql 'set broadcast_row_count_limit = 30000000'
     sql 'set enable_nereids_timeout = false'
     sql 'set enable_pipeline_engine=true'
+    sql 'set enable_runtime_filter_prune=false'
+    sql 'set expand_runtime_filter_by_inner_join=false'
     String stmt = '''
     explain physical plan
     with wscs as

regression-test/suites/nereids_tpcds_shape_sf100_p0/rf/ds_rf20.groovy

@@ -30,6 +30,8 @@ suite("ds_rf20") {
     sql 'set broadcast_row_count_limit = 30000000'
     sql 'set enable_nereids_timeout = false'
     sql 'set enable_pipeline_engine=true'
+    sql 'set enable_runtime_filter_prune=false'
+    sql 'set expand_runtime_filter_by_inner_join=false'
     String stmt = '''
     explain physical plan
     select  i_item_id

regression-test/suites/nereids_tpcds_shape_sf100_p0/rf/ds_rf21.groovy

@@ -30,6 +30,8 @@ suite("ds_rf21") {
     sql 'set broadcast_row_count_limit = 30000000'
     sql 'set enable_nereids_timeout = false'
     sql 'set enable_pipeline_engine=true'
+    sql 'set enable_runtime_filter_prune=false'
+    sql 'set expand_runtime_filter_by_inner_join=false'
     String stmt = '''
     explain physical plan
     select  *

regression-test/suites/nereids_tpcds_shape_sf100_p0/rf/ds_rf22.groovy

@@ -30,6 +30,8 @@ suite("ds_rf22") {
     sql 'set broadcast_row_count_limit = 30000000'
     sql 'set enable_nereids_timeout = false'
     sql 'set enable_pipeline_engine=true'
+    sql 'set enable_runtime_filter_prune=false'
+    sql 'set expand_runtime_filter_by_inner_join=false'
     String stmt = '''
     explain physical plan
     select  i_product_name

regression-test/suites/nereids_tpcds_shape_sf100_p0/rf/ds_rf23.groovy

@@ -30,6 +30,8 @@ suite("ds_rf23") {
     sql 'set broadcast_row_count_limit = 30000000'
     sql 'set enable_nereids_timeout = false'
     sql 'set enable_pipeline_engine=true'
+    sql 'set enable_runtime_filter_prune=false'
+    sql 'set expand_runtime_filter_by_inner_join=false'
     String stmt = '''
     explain physical plan
     with frequent_ss_items as 

regression-test/suites/nereids_tpcds_shape_sf100_p0/rf/ds_rf24.groovy

@@ -30,6 +30,8 @@ suite("ds_rf24") {
     sql 'set broadcast_row_count_limit = 30000000'
     sql 'set enable_nereids_timeout = false'
     sql 'set enable_pipeline_engine=true'
+    sql 'set enable_runtime_filter_prune=false'
+    sql 'set expand_runtime_filter_by_inner_join=false'
     String stmt = '''
     explain physical plan
     with ssales as

regression-test/suites/nereids_tpcds_shape_sf100_p0/rf/ds_rf25.groovy

@@ -30,6 +30,8 @@ suite("ds_rf25") {
     sql 'set broadcast_row_count_limit = 30000000'
     sql 'set enable_nereids_timeout = false'
     sql 'set enable_pipeline_engine=true'
+    sql 'set enable_runtime_filter_prune=false'
+    sql 'set expand_runtime_filter_by_inner_join=false'
     String stmt = '''
     explain physical plan
     select  

regression-test/suites/nereids_tpcds_shape_sf100_p0/rf/ds_rf26.groovy

@@ -30,6 +30,8 @@ suite("ds_rf26") {
     sql 'set broadcast_row_count_limit = 30000000'
     sql 'set enable_nereids_timeout = false'
     sql 'set enable_pipeline_engine=true'
+    sql 'set enable_runtime_filter_prune=false'
+    sql 'set expand_runtime_filter_by_inner_join=false'
     String stmt = '''
     explain physical plan
     select  i_item_id, 

regression-test/suites/nereids_tpcds_shape_sf100_p0/rf/ds_rf27.groovy

@@ -30,6 +30,8 @@ suite("ds_rf27") {
     sql 'set broadcast_row_count_limit = 30000000'
     sql 'set enable_nereids_timeout = false'
     sql 'set enable_pipeline_engine=true'
+    sql 'set enable_runtime_filter_prune=false'
+    sql 'set expand_runtime_filter_by_inner_join=false'
     String stmt = '''
     explain physical plan
     select  i_item_id,

regression-test/suites/nereids_tpcds_shape_sf100_p0/rf/ds_rf28.groovy

@@ -30,6 +30,8 @@ suite("ds_rf28") {
     sql 'set broadcast_row_count_limit = 30000000'
     sql 'set enable_nereids_timeout = false'
     sql 'set enable_pipeline_engine=true'
+    sql 'set enable_runtime_filter_prune=false'
+    sql 'set expand_runtime_filter_by_inner_join=false'
     String stmt = '''
     explain physical plan
     select  *

regression-test/suites/nereids_tpcds_shape_sf100_p0/rf/ds_rf29.groovy

@@ -30,6 +30,8 @@ suite("ds_rf29") {
     sql 'set broadcast_row_count_limit = 30000000'
     sql 'set enable_nereids_timeout = false'
     sql 'set enable_pipeline_engine=true'
+    sql 'set enable_runtime_filter_prune=false'
+    sql 'set expand_runtime_filter_by_inner_join=false'
     String stmt = '''
     explain physical plan
     select   

regression-test/suites/nereids_tpcds_shape_sf100_p0/rf/ds_rf3.groovy

@@ -30,6 +30,8 @@ suite("ds_rf3") {
     sql 'set broadcast_row_count_limit = 30000000'
     sql 'set enable_nereids_timeout = false'
     sql 'set enable_pipeline_engine=true'
+    sql 'set enable_runtime_filter_prune=false'
+    sql 'set expand_runtime_filter_by_inner_join=false'
     String stmt = '''
     explain physical plan
     select  dt.d_year 

regression-test/suites/nereids_tpcds_shape_sf100_p0/rf/ds_rf30.groovy

@@ -30,6 +30,8 @@ suite("ds_rf30") {
     sql 'set broadcast_row_count_limit = 30000000'
     sql 'set enable_nereids_timeout = false'
     sql 'set enable_pipeline_engine=true'
+    sql 'set enable_runtime_filter_prune=false'
+    sql 'set expand_runtime_filter_by_inner_join=false'
     String stmt = '''
     explain physical plan
     with customer_total_return as

regression-test/suites/nereids_tpcds_shape_sf100_p0/rf/ds_rf31.groovy

@@ -30,6 +30,8 @@ suite("ds_rf31") {
     sql 'set broadcast_row_count_limit = 30000000'
     sql 'set enable_nereids_timeout = false'
     sql 'set enable_pipeline_engine=true'
+    sql 'set enable_runtime_filter_prune=false'
+    sql 'set expand_runtime_filter_by_inner_join=false'
     String stmt = '''
     explain physical plan
     with ss as