raghav-reglobe · Nov 13, 2026 · Nov 13, 2026
@@ -19,6 +19,7 @@
 
 import org.apache.doris.nereids.CascadesContext;
 import org.apache.doris.nereids.PlanProcess;
+import org.apache.doris.nereids.exceptions.AnalysisException;
 import org.apache.doris.nereids.jobs.JobContext;
 import org.apache.doris.nereids.rules.Rule;
 import org.apache.doris.nereids.rules.Rules;
@@ -33,6 +34,13 @@
 
 /** TopDownVisitorRewriteJob */
 public class TopDownVisitorRewriteJob implements RewriteJob {
+    // The most times the rules of one batch may change the plan at one position (one doRewrite call). With the
+    // planning timeout switched off, a rule that changes its node on every pass has no other way out of
+    // doRewrite. The bound counts rewrites, not memory: with the plan process recorded (EXPLAIN ... PLAN
+    // PROCESS) every rewrite also keeps two plan strings, and a rule that grows the plan can fill the heap
+    // before it reaches the bound.
+    public static final int MAX_REWRITES_OF_ONE_NODE = 10000;
+
     private final Rules rules;
     private final Predicate<Plan> isTraverseChildren;
 
@@ -73,8 +81,6 @@ private Plan rewrite(Plan parent, int childIndex, Plan plan, JobContext jobConte
             return plan;
         }
 
-        checkTimeout(jobContext);
-
         plan = doRewrite(parent, childIndex, plan, jobContext, rules, processState);
 
         ImmutableList.Builder<Plan> newChildren = ImmutableList.builderWithExpectedSize(plan.arity());
@@ -96,12 +102,14 @@ private Plan rewrite(Plan parent, int childIndex, Plan plan, JobContext jobConte
         return plan;
     }
 
-    private static Plan doRewrite(
+    private Plan doRewrite(
             Plan originParent, int childIndex, Plan plan, JobContext jobContext,
             Rules rules, ProcessState processState) {
         CascadesContext cascadesContext = jobContext.getCascadesContext();
         Plan originPlan = plan;
+        int rewrites = 0;
         while (true) {
+            checkTimeout(jobContext);
             List<Rule> currentRules = rules.getCurrentRules(originPlan);
             BitSet forbidRules = cascadesContext.getAndCacheDisableRules();
 
@@ -113,6 +121,12 @@ private static Plan doRewrite(
                 }
                 List<Plan> transform = currentRule.transform(originPlan, cascadesContext);
                 if (!transform.isEmpty() && !transform.get(0).deepEquals(originPlan)) {
+                    if (++rewrites > MAX_REWRITES_OF_ONE_NODE) {
+                        throw new AnalysisException("The rewrite does not converge: rule "
+                                + currentRule.getRuleType().name() + " still changes " + originPlan.getType()
+                                + " after " + MAX_REWRITES_OF_ONE_NODE + " rewrites at one position. A rule can be"
+                                + " switched off by name with the session variable disable_nereids_rules.");
+                    }
                     Plan newPlan = transform.get(0);
                     currentRule.acceptPlan(originPlan);
                     if (processState != null) {

@@ -46,10 +46,17 @@ public Rule build() {
                     LogicalAggregate<Plan> agg = ctx.root;
                     Set<AggregateFunction> aggregateFunctions = agg.getAggregateFunctions();
                     Set<Expression> isNotNulls = inferCommonNotNulls(aggregateFunctions, ctx.cascadesContext);
-                    Set<Expression> predicates = Collections.emptySet();
-                    if ((agg.child() instanceof Filter)) {
-                        predicates = ((Filter) agg.child()).getConjuncts();
+                    // Read every filter in the chain right under the aggregate, not only the first one. This
+                    // rule puts what it generates into a filter of its own on top, and that filter hides the
+                    // ones below it: reading only the first, the rule would generate again, on every pass, a
+                    // not-null that a lower filter holds. The filters below stay as they were written, so
+                    // InferFilterNotNull, which skips a filter that holds a generated not-null, still infers
+                    // from their conjuncts.
+                    ImmutableSet.Builder<Expression> predicatesBuilder = ImmutableSet.builder();
+                    for (Plan child = agg.child(); child instanceof Filter; child = child.child(0)) {
+                        predicatesBuilder.addAll(((Filter) child).getConjuncts());
                     }
+                    Set<Expression> predicates = predicatesBuilder.build();
                     ImmutableSet.Builder<Expression> needGenerateNotNullsBuilder = ImmutableSet.builder();
                     for (Expression isNotNull : isNotNulls) {
                         if (!predicates.contains(isNotNull)) {

@@ -0,0 +1,149 @@
+// 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.jobs.rewrite;
+
+import org.apache.doris.nereids.CascadesContext;
+import org.apache.doris.nereids.exceptions.AnalysisException;
+import org.apache.doris.nereids.rules.FilteredRules;
+import org.apache.doris.nereids.rules.Rule;
+import org.apache.doris.nereids.rules.RuleType;
+import org.apache.doris.nereids.rules.rewrite.OneRewriteRuleFactory;
+import org.apache.doris.nereids.trees.expressions.EqualTo;
+import org.apache.doris.nereids.trees.expressions.literal.IntegerLiteral;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
+import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
+import org.apache.doris.nereids.util.MemoTestUtils;
+import org.apache.doris.nereids.util.PlanConstructor;
+import org.apache.doris.qe.ConnectContext;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+class TopDownVisitorRewriteJobTest {
+    @Test
+    void testRuleThatChangesItsNodeOnEveryPassIsStopped() {
+        LogicalOlapScan scan = PlanConstructor.newLogicalOlapScan(0, "t1", 0);
+        Plan plan = new LogicalFilter<Plan>(
+                ImmutableSet.of(new EqualTo(scan.getOutput().get(0), new IntegerLiteral(0))), scan);
+        AtomicInteger applied = new AtomicInteger();
+        // One more filter on top on every pass. It stops by itself a little after the bound, so that without
+        // the bound this test fails on its assertion and does not fill the heap.
+        int stopsAt = TopDownVisitorRewriteJob.MAX_REWRITES_OF_ONE_NODE + 50;
+        Rule neverConverges = new OneRewriteRuleFactory() {
+            @Override
+            public Rule build() {
+                return logicalFilter().when(filter -> applied.get() < stopsAt).then(filter -> new LogicalFilter<Plan>(
+                        ImmutableSet.of(new EqualTo(filter.getOutput().get(0),
+                                new IntegerLiteral(applied.incrementAndGet()))), filter)
+                ).toRule(RuleType.TEST_REWRITE);
+            }
+        }.build();
+        CascadesContext cascadesContext = MemoTestUtils.createCascadesContext(
+                MemoTestUtils.createConnectContext(), plan);
+        TopDownVisitorRewriteJob job = new TopDownVisitorRewriteJob(
+                new FilteredRules(ImmutableList.of(neverConverges)), p -> true);
+
+        AnalysisException e = Assertions.assertThrows(AnalysisException.class,
+                () -> job.execute(cascadesContext.getCurrentJobContext()));
+        Assertions.assertTrue(e.getMessage().contains(RuleType.TEST_REWRITE.name()), e.getMessage());
+        Assertions.assertEquals(TopDownVisitorRewriteJob.MAX_REWRITES_OF_ONE_NODE + 1, applied.get());
+    }
+
+    @Test
+    void testTheBoundCountsOnePositionNotTheWholePlan() {
+        // Two filters, each changed 6000 times: 12000 rewrites of the plan, 6000 at each position.
+        int perPosition = 6000;
+        Assertions.assertTrue(perPosition < TopDownVisitorRewriteJob.MAX_REWRITES_OF_ONE_NODE);
+        Assertions.assertTrue(2 * perPosition > TopDownVisitorRewriteJob.MAX_REWRITES_OF_ONE_NODE);
+        LogicalOlapScan scan = PlanConstructor.newLogicalOlapScan(0, "t1", 0);
+        Plan below = new LogicalFilter<Plan>(
+                ImmutableSet.of(new EqualTo(scan.getOutput().get(0), new IntegerLiteral(0))), scan);
+        Plan plan = new LogicalFilter<Plan>(
+                ImmutableSet.of(new EqualTo(scan.getOutput().get(0), new IntegerLiteral(0))), below);
+        AtomicInteger applied = new AtomicInteger();
+        // counts a filter's literal up to perPosition, in place
+        Rule countsUp = new OneRewriteRuleFactory() {
+            @Override
+            public Rule build() {
+                return logicalFilter().when(filter -> literalOf(filter) < perPosition).then(filter -> {
+                    applied.incrementAndGet();
+                    return new LogicalFilter<Plan>(ImmutableSet.of(new EqualTo(filter.getOutput().get(0),
+                            new IntegerLiteral(literalOf(filter) + 1))), filter.child());
+                }).toRule(RuleType.TEST_REWRITE);
+            }
+        }.build();
+        CascadesContext cascadesContext = MemoTestUtils.createCascadesContext(
+                MemoTestUtils.createConnectContext(), plan);
+        TopDownVisitorRewriteJob job = new TopDownVisitorRewriteJob(
+                new FilteredRules(ImmutableList.of(countsUp)), p -> true);
+
+        job.execute(cascadesContext.getCurrentJobContext());
+        Assertions.assertEquals(2 * perPosition, applied.get());
+    }
+
+    @Test
+    void testPlanningTimeoutStopsRuleBetweenPasses() {
+        LogicalOlapScan scan = PlanConstructor.newLogicalOlapScan(0, "t1", 0);
+        Plan plan = new LogicalFilter<Plan>(
+                ImmutableSet.of(new EqualTo(scan.getOutput().get(0), new IntegerLiteral(0))), scan);
+        AtomicInteger applied = new AtomicInteger();
+        // one more filter on top on every pass; the first pass alone takes longer than the planning timeout
+        Rule neverConverges = new OneRewriteRuleFactory() {
+            @Override
+            public Rule build() {
+                return logicalFilter().then(filter -> {
+                    if (applied.incrementAndGet() == 1) {
+                        try {
+                            Thread.sleep(1100);
+                        } catch (InterruptedException interrupted) {
+                            Thread.currentThread().interrupt();
+                        }
+                    }
+                    return new LogicalFilter<Plan>(ImmutableSet.of(new EqualTo(filter.getOutput().get(0),
+                            new IntegerLiteral(applied.get()))), filter);
+                }).toRule(RuleType.TEST_REWRITE);
+            }
+        }.build();
+        ConnectContext connectContext = MemoTestUtils.createConnectContext();
+        connectContext.getSessionVariable().enableNereidsTimeout = true;
+        connectContext.getSessionVariable().nereidsTimeoutSecond = 0;
+        CascadesContext cascadesContext = MemoTestUtils.createCascadesContext(connectContext, plan);
+        // only NereidsPlanner starts this stopwatch; unstarted, the timeout check always reads zero
+        if (!cascadesContext.getStatementContext().getStopwatch().isRunning()) {
+            cascadesContext.getStatementContext().getStopwatch().start();
+        }
+        TopDownVisitorRewriteJob job = new TopDownVisitorRewriteJob(
+                new FilteredRules(ImmutableList.of(neverConverges)), p -> true);
+
+        AnalysisException e = Assertions.assertThrows(AnalysisException.class,
+                () -> job.execute(cascadesContext.getCurrentJobContext()));
+        // the timeout, read at the start of the second pass, and not the bound after ten thousand of them
+        Assertions.assertTrue(e.getMessage().contains("cost too much time"), e.getMessage());
+        Assertions.assertEquals(1, applied.get());
+    }
+
+    private static int literalOf(LogicalFilter<? extends Plan> filter) {
+        EqualTo equalTo = (EqualTo) filter.getConjuncts().iterator().next();
+        return ((IntegerLiteral) equalTo.right()).getValue();
+    }
+}
@@ -23,8 +23,12 @@
 import org.apache.doris.catalog.OlapTable;
 import org.apache.doris.catalog.PartitionInfo;
 import org.apache.doris.catalog.Type;
+import org.apache.doris.nereids.CascadesContext;
+import org.apache.doris.nereids.rules.Rule;
+import org.apache.doris.nereids.trees.expressions.Add;
 import org.apache.doris.nereids.trees.expressions.Alias;
 import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.GreaterThan;
 import org.apache.doris.nereids.trees.expressions.IsNull;
 import org.apache.doris.nereids.trees.expressions.Not;
 import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
@@ -43,8 +47,10 @@
 import org.apache.doris.nereids.trees.expressions.functions.agg.Sum;
 import org.apache.doris.nereids.trees.expressions.functions.agg.Sum0;
 import org.apache.doris.nereids.trees.expressions.literal.IntegerLiteral;
+import org.apache.doris.nereids.trees.plans.Plan;
 import org.apache.doris.nereids.trees.plans.RelationId;
 import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
+import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
 import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
 import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
 import org.apache.doris.nereids.util.LogicalPlanBuilder;
@@ -122,6 +128,69 @@ void testNotInferWhenAggregateArgumentReturnsFalseForNullInput() {
                 );
     }
 
+    @Test
+    void testInferPutsItsNotNullAboveTheFilterThatHoldsTheOther() {
+        // The filter under the aggregate already holds the not-null of c0. The not-null of c1 goes into a
+        // filter of its own on top, and the filter below stays as it was written.
+        LogicalOlapScan scan = newWideNullableScan(2);
+        Expression c0 = scan.getOutput().get(0);
+        Expression c1 = scan.getOutput().get(1);
+        Expression written = new Not(new IsNull(c0));
+        Expression generated = new Not(new IsNull(c1), true);
+        LogicalPlan plan = new LogicalPlanBuilder(scan)
+                .filter(written)
+                .aggGroupUsingIndex(ImmutableList.of(),
+                        ImmutableList.of(new Alias(new Sum(new Add(c0, c1)), "s")))
+                .build();
+        CascadesContext cascadesContext = MemoTestUtils.createCascadesContext(
+                MemoTestUtils.createConnectContext(), plan);
+        Rule rule = new InferAggNotNull().build();
+        Assertions.assertTrue(rule.getPattern().matchPlanTree(plan));
+
+        // One application at a time: run to a fixpoint, a rule that does not converge would not fail this
+        // test, it would hang it.
+        Plan once = rule.transform(plan, cascadesContext).get(0);
+        Assertions.assertTrue(once.child(0) instanceof LogicalFilter, once.treeString());
+        LogicalFilter<?> onTop = (LogicalFilter<?>) once.child(0);
+        Assertions.assertEquals(ImmutableSet.of(generated), onTop.getConjuncts());
+        Assertions.assertTrue(onTop.child() instanceof LogicalFilter, once.treeString());
+        Assertions.assertEquals(ImmutableSet.of(written), ((LogicalFilter<?>) onTop.child()).getConjuncts());
+
+        // The second application reads both filters and has nothing to add. Reading only the filter on top,
+        // it generated the not-null of c0 again, then that of c1, and so on without end.
+        Assertions.assertSame(once, rule.transform(once, cascadesContext).get(0));
+    }
+
+    @Test
+    void testFilterUnderTheAggregateStillInfersItsOwnNotNulls() {
+        // sum(c2) over filter(c0 + c1 > 0). The not-null of c2 that this rule generates must stay out of the
+        // filter below: InferFilterNotNull skips a filter that holds a generated not-null, and the not-nulls
+        // of c0 and c1 that it infers from the condition are what lets an outer join below become an inner one.
+        LogicalOlapScan scan = newWideNullableScan(3);
+        Expression c0 = scan.getOutput().get(0);
+        Expression c1 = scan.getOutput().get(1);
+        Expression c2 = scan.getOutput().get(2);
+        Expression condition = new GreaterThan(new Add(c0, c1), new IntegerLiteral(0));
+        LogicalPlan plan = new LogicalPlanBuilder(scan)
+                .filter(condition)
+                .aggGroupUsingIndex(ImmutableList.of(), ImmutableList.of(new Alias(new Sum(c2), "s")))
+                .build();
+
+        PlanChecker.from(MemoTestUtils.createConnectContext(), plan)
+                .applyTopDown(new InferAggNotNull(), new InferFilterNotNull())
+                .matches(
+                        logicalAggregate(
+                                logicalFilter(
+                                        logicalFilter(logicalOlapScan()).when(below ->
+                                                below.getConjuncts().equals(ImmutableSet.of(condition,
+                                                        new Not(new IsNull(c0), true),
+                                                        new Not(new IsNull(c1), true))))
+                                ).when(onTop ->
+                                        onTop.getConjuncts().equals(ImmutableSet.of(new Not(new IsNull(c2), true))))
+                        )
+                );
+    }
+
     @Test
     void testInferMultipleAggregateSameInput() {
         LogicalPlan plan = new LogicalPlanBuilder(scan1)