branch-4.2 [Enhancement](optimizer) Support multi-mode CTE inline strategy

fe/fe-core/src/main/java/org/apache/doris/nereids/StatementContext.java

@@ -341,6 +341,9 @@ public enum TableFrom {
     private boolean hasNestedColumns;
 
     private final Set<CTEId> mustInlineCTE = new HashSet<>();
+
+    // CTEs that must be materialized (e.g., containing non-deterministic functions)
+    private final Set<CTEId> forceMaterializeCTEs = new HashSet<>();
     private final Set<String> usedAIResourceNames = new LinkedHashSet<>();
 
     private final Map<String, Integer> lowerCaseTableNamesCache = Maps.newHashMap();
@@ -1766,6 +1769,18 @@ public Set<CTEId> getMustInlineCTEs() {
         return mustInlineCTE;
     }
 
+    public void addForceMaterializeCTE(CTEId cteId) {
+        forceMaterializeCTEs.add(cteId);
+    }
+
+    public boolean isForceMaterializeCTE(CTEId cteId) {
+        return forceMaterializeCTEs.contains(cteId);
+    }
+
+    public Set<CTEId> getForceMaterializeCTEs() {
+        return forceMaterializeCTEs;
+    }
+
     public Optional<IcebergWriteSchemaContext> getIcebergWriteSchemaContext() {
         return icebergWriteSchemaContext;
     }

fe/fe-core/src/main/java/org/apache/doris/nereids/cost/CostModel.java

@@ -23,6 +23,7 @@
 import org.apache.doris.catalog.MTMV;
 import org.apache.doris.catalog.OlapTable;
 import org.apache.doris.nereids.PlanContext;
+import org.apache.doris.nereids.StatementContext;
 import org.apache.doris.nereids.hint.Hint;
 import org.apache.doris.nereids.hint.UseMvHint;
 import org.apache.doris.nereids.processor.post.RuntimeFilterGenerator;
@@ -40,7 +41,11 @@
 import org.apache.doris.nereids.trees.plans.Plan;
 import org.apache.doris.nereids.trees.plans.PlanNodeAndHash;
 import org.apache.doris.nereids.trees.plans.algebra.OlapScan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalCTEProducer;
+import org.apache.doris.nereids.trees.plans.logical.LogicalUnion;
 import org.apache.doris.nereids.trees.plans.physical.PhysicalAssertNumRows;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalCTEConsumer;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalCTEProducer;
 import org.apache.doris.nereids.trees.plans.physical.PhysicalDeferMaterializeOlapScan;
 import org.apache.doris.nereids.trees.plans.physical.PhysicalDeferMaterializeTopN;
 import org.apache.doris.nereids.trees.plans.physical.PhysicalDistribute;
@@ -61,6 +66,7 @@
 import org.apache.doris.nereids.trees.plans.physical.PhysicalSchemaScan;
 import org.apache.doris.nereids.trees.plans.physical.PhysicalStorageLayerAggregate;
 import org.apache.doris.nereids.trees.plans.physical.PhysicalTopN;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalUnion;
 import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
 import org.apache.doris.qe.ConnectContext;
 import org.apache.doris.qe.SessionVariable;
@@ -611,4 +617,67 @@ public Cost visitPhysicalGenerate(PhysicalGenerate<? extends Plan> generate, Pla
                 0
         );
     }
+
+    @Override
+    public Cost visitPhysicalCTEProducer(PhysicalCTEProducer<? extends Plan> cteProducer, PlanContext context) {
+        Statistics childStats = context.getChildStatistics(0);
+        double rows = childStats.getRowCount();
+        double tupleSize = childStats.computeTupleSize(cteProducer.child().getOutput());
+
+        // Determine network cost factor to guide CBO inline decision:
+        // 1. UNION ALL CTEs with N>=3 consumers: pipeline serialization means N consumers wait for producer.
+        //    Each inline copy can use consumer-specific filters (e.g. different d_year) to prune branches.
+        //    Factor = numConsumers * 3 makes materialized more expensive so CBO prefers inline.
+        // 2. Small-output CTEs (rows < 1M): computation cost far exceeds output size
+        //    (e.g. store_sales 2.87B -> 520K agg). Materialization creates a barrier preventing pipeline
+        //    parallelism. Factor = 300 reflects this overhead.
+        double networkFactor = 1.0;
+        ConnectContext connectContext = ConnectContext.get();
+        if (connectContext != null) {
+            StatementContext statCtx = connectContext.getStatementContext();
+            if (statCtx != null) {
+                Set<?> consumers = statCtx.getCteIdToConsumers().get(cteProducer.getCteId());
+                int numConsumers = (consumers != null) ? consumers.size() : 1;
+                boolean hasUnionAll = cteProducer.child().anyMatch(p -> p instanceof PhysicalUnion);
+                if (hasUnionAll && numConsumers >= 3) {
+                    // Model pipeline serialization: N consumers each forced to wait for producer
+                    networkFactor = numConsumers * 3.0;
+                } else if (rows < 1_000_000) {
+                    // Small-output CTE: high computation-to-output ratio, parallelism benefit dominates
+                    networkFactor = 300.0;
+                }
+            }
+        }
+        return Cost.of(context.getSessionVariable(), rows, 0, rows * tupleSize * networkFactor);
+    }
+
+    @Override
+    public Cost visitPhysicalCTEConsumer(PhysicalCTEConsumer cteConsumer, PlanContext context) {
+        Statistics stats = context.getStatisticsWithCheck();
+        double rows = stats.getRowCount();
+        double tupleSize = stats.computeTupleSize(cteConsumer.getOutput());
+
+        double networkFactor = 1.0;
+        ConnectContext connectContext = ConnectContext.get();
+        if (connectContext != null) {
+            StatementContext statCtx = connectContext.getStatementContext();
+            if (statCtx != null) {
+                Set<?> consumers = statCtx.getCteIdToConsumers().get(cteConsumer.getCteId());
+                int numConsumers = (consumers != null) ? consumers.size() : 1;
+                // Check UNION ALL via logical producer (physical plan may differ after optimization)
+                boolean hasUnionAll = false;
+                LogicalCTEProducer<?> logicalProducer =
+                        statCtx.getCteProducerByCteId(cteConsumer.getCteId());
+                if (logicalProducer != null) {
+                    hasUnionAll = logicalProducer.child().anyMatch(p -> p instanceof LogicalUnion);
+                }
+                if (hasUnionAll && numConsumers >= 3) {
+                    networkFactor = numConsumers * 3.0;
+                } else if (rows < 1_000_000) {
+                    networkFactor = 300.0;
+                }
+            }
+        }
+        return Cost.of(context.getSessionVariable(), rows, 0, rows * tupleSize * networkFactor);
+    }
 }

fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/executor/Optimizer.java

@@ -23,10 +23,23 @@
 import org.apache.doris.nereids.jobs.joinorder.JoinOrderJob;
 import org.apache.doris.nereids.memo.Group;
 import org.apache.doris.nereids.memo.Memo;
+import org.apache.doris.nereids.properties.PhysicalProperties;
+import org.apache.doris.nereids.rules.RuleSet;
+import org.apache.doris.nereids.rules.RuleType;
+import org.apache.doris.nereids.rules.rewrite.CTEInliner;
+import org.apache.doris.nereids.rules.rewrite.ColumnPruning;
+import org.apache.doris.nereids.rules.rewrite.EliminateEmptyRelation;
+import org.apache.doris.nereids.rules.rewrite.EliminateUnnecessaryProject;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalEmptyRelation;
 import org.apache.doris.nereids.util.MoreFieldsThread;
 import org.apache.doris.qe.ConnectContext;
 import org.apache.doris.qe.SessionVariable;
 
+import com.google.common.collect.ImmutableList;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
 import java.util.Objects;
 
 /**
@@ -35,6 +48,7 @@
  * try to find best plan under the guidance of statistic information and cost model.
  */
 public class Optimizer {
+    private static final Logger LOG = LogManager.getLogger(Optimizer.class);
 
     private final CascadesContext cascadesContext;
 
@@ -47,11 +61,16 @@ public Optimizer(CascadesContext cascadesContext) {
      */
     public void execute() {
         MoreFieldsThread.keepFunctionSignature(() -> {
+            // generate inlined CTE alternative for CBO comparison
+            Plan cboInlinedPlan = generateCTEInlineAlternative();
             // init memo
             cascadesContext.toMemo();
+            if (cboInlinedPlan != null) {
+                cascadesContext.getMemo().copyIn(cboInlinedPlan, cascadesContext.getMemo().getRoot(), false);
+            }
             // stats derive
-            cascadesContext.getMemo().getRoot().getLogicalExpressions().forEach(groupExpression ->
-                    cascadesContext.pushJob(
+            cascadesContext.getMemo().getRoot().getLogicalExpressions()
+                    .forEach(groupExpression -> cascadesContext.pushJob(
                             new DeriveStatsJob(groupExpression, cascadesContext.getCurrentJobContext())));
             cascadesContext.getJobScheduler().executeJobPool(cascadesContext);
             if (cascadesContext.getStatementContext().isDpHyp() || isDpHyp(cascadesContext)) {
@@ -101,6 +120,97 @@ private void dpHypOptimize() {
         cascadesContext.getJobScheduler().executeJobPool(cascadesContext);
     }
 
+    /**
+     * Generate a fully inlined CTE alternative plan and add it to the Memo root group.
+     * This gives the CBO the ability to compare costs of materialized vs inlined CTE approaches.
+     *
+     * After inlining, runs filter pushdown and column pruning on the inlined plan so that
+     * each inlined CTE body gets consumer-specific filters pushed down into it, producing
+     * different optimized sub-trees per consumer position (e.g., different date/type filters
+     * can eliminate branches in UNION queries inside the CTE body).
+     */
+    private Plan generateCTEInlineAlternative() {
+        int mode = getSessionVariable().cteInlineMode;
+        if (mode < 0) {
+            return null;
+        }
+        try {
+            if (mode == 0) {
+                return generateSelectiveCTEInline();
+            } else {
+                return generateFullCTEInline();
+            }
+        } catch (Exception e) {
+            LOG.warn("Failed to generate CTE inline alternative for CBO, fall back to default behavior", e);
+            return null;
+        }
+    }
+
+    private Plan generateFullCTEInline() {
+        Plan rewritePlan = cascadesContext.getRewritePlan();
+        CTEInliner cteInliner = new CTEInliner(cascadesContext.getStatementContext());
+        Plan inlinedPlan = cteInliner.generateInlinedPlan(rewritePlan);
+        if (inlinedPlan != null) {
+            return rewriteInlinedPlan(inlinedPlan);
+        }
+        return null;
+    }
+
+    // Returns null because mode=0 directly replaces rewritePlan via
+    // setRewritePlan(),
+    // so toMemo() will use the inlined plan. No need to copyIn as an alternative.
+    private Plan generateSelectiveCTEInline() {
+        Plan rewritePlan = cascadesContext.getRewritePlan();
+        CTEInliner cteInliner = new CTEInliner(cascadesContext.getStatementContext(), true);
+        Plan inlinedPlan = cteInliner.generateInlinedPlan(rewritePlan);
+        if (inlinedPlan != null) {
+            inlinedPlan = rewriteInlinedPlan(inlinedPlan);
+            if (inlinedPlan.anyMatch(p -> p instanceof LogicalEmptyRelation)) {
+                inlinedPlan = eliminateEmptyRelation(inlinedPlan);
+                cascadesContext.setRewritePlan(inlinedPlan);
+                return null;
+            }
+        }
+        return null;
+    }
+
+    private Plan eliminateEmptyRelation(Plan plan) {
+        CascadesContext ctx = CascadesContext.initContext(
+                cascadesContext.getStatementContext(), plan, PhysicalProperties.ANY);
+        // Use getCteChildrenRewriter for the same reason as rewriteInlinedPlan:
+        // getWholeTreeRewriterWithCustomJobs would invoke RewriteCteChildren which
+        // reads stale rewrittenCteConsumer cache from the main Rewriter phase,
+        // reverting the inlined CTE subtrees back to the original structure.
+        Rewriter.getCteChildrenRewriter(ctx, ImmutableList.of(
+                Rewriter.bottomUp(new EliminateEmptyRelation()),
+                Rewriter.custom(RuleType.COLUMN_PRUNING, ColumnPruning::new),
+                Rewriter.custom(RuleType.ELIMINATE_UNNECESSARY_PROJECT, EliminateUnnecessaryProject::new))).execute();
+        return ctx.getRewritePlan();
+    }
+
+    /**
+     * Run filter pushdown and column pruning on the inlined plan using a temporary
+     * CascadesContext.
+     *
+     * We deliberately use getCteChildrenRewriter (no notTraverseChildrenOf wrapper) so that
+     * PUSH_DOWN_FILTERS traverses the ENTIRE inlined plan tree, including inside any remaining
+     * LogicalCTEAnchor subtrees (e.g. for CTEs that were NOT inlined). Using
+     * getWholeTreeRewriterWithCustomJobs would invoke RewriteCteChildren, which reads from the
+     * shared StatementContext cache (rewrittenCteConsumer) populated during the main Rewriter
+     * phase. That cached outer query still contains LogicalCTEConsumer nodes for the inlined CTE,
+     * preventing the filter from ever reaching the inlined union body.
+     */
+    private Plan rewriteInlinedPlan(Plan inlinedPlan) {
+        CascadesContext inlinedContext = CascadesContext.initContext(
+                cascadesContext.getStatementContext(), inlinedPlan, PhysicalProperties.ANY);
+        Rewriter.getCteChildrenRewriter(inlinedContext, ImmutableList.of(
+                Rewriter.bottomUp(RuleSet.PUSH_DOWN_FILTERS),
+                Rewriter.custom(RuleType.COLUMN_PRUNING, ColumnPruning::new),
+                Rewriter.bottomUp(RuleSet.PUSH_DOWN_FILTERS),
+                Rewriter.custom(RuleType.ELIMINATE_UNNECESSARY_PROJECT, EliminateUnnecessaryProject::new))).execute();
+        return inlinedContext.getRewritePlan();
+    }
+
     private SessionVariable getSessionVariable() {
         return cascadesContext.getConnectContext().getSessionVariable();
     }

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/CTEInline.java

@@ -34,6 +34,7 @@
 import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
 import org.apache.doris.nereids.trees.plans.visitor.CustomRewriter;
 import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.NondeterministicFunctionCollector;
 import org.apache.doris.qe.ConnectContext;
 
 import com.google.common.collect.ImmutableList;
@@ -91,14 +92,16 @@ public Plan visitLogicalCTEAnchor(LogicalCTEAnchor<? extends Plan, ? extends Pla
                 return root.accept(this, null);
             } else {
                 ConnectContext connectContext = ConnectContext.get();
+                LogicalCTEProducer<?> cteProducer = (LogicalCTEProducer<?>) cteAnchor.left();
                 if (connectContext.getSessionVariable().enableCTEMaterialize
-                        && consumers.size() > connectContext.getSessionVariable().inlineCTEReferencedThreshold) {
+                        && (consumers.size() > connectContext.getSessionVariable().inlineCTEReferencedThreshold
+                                || containsNondeterministicFunction(cteProducer))) {
                     // not inline
                     Plan right = cteAnchor.right().accept(this, null);
                     return cteAnchor.withChildren(cteAnchor.left(), right);
                 } else {
                     // should inline
-                    Plan root = cteAnchor.right().accept(this, (LogicalCTEProducer<?>) cteAnchor.left());
+                    Plan root = cteAnchor.right().accept(this, cteProducer);
                     // process child
                     return root.accept(this, null);
                 }
@@ -126,4 +129,10 @@ public Plan visitLogicalCTEConsumer(LogicalCTEConsumer cteConsumer, LogicalCTEPr
         }
         return cteConsumer;
     }
+
+    private boolean containsNondeterministicFunction(LogicalCTEProducer<?> producer) {
+        List<Expression> nondeterministicFunctions = new ArrayList<>();
+        producer.accept(NondeterministicFunctionCollector.INSTANCE, nondeterministicFunctions);
+        return !nondeterministicFunctions.isEmpty();
+    }
 }