You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by "morrySnow (via GitHub)" <gi...@apache.org> on 2023/05/25 09:45:51 UTC
[GitHub] [doris] morrySnow commented on a diff in pull request #19934: [feat](optimizer) Support materailze CTE
morrySnow commented on code in PR #19934:
URL: https://github.com/apache/doris/pull/19934#discussion_r1202109118
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/analyzer/CTEContext.java:
##########
@@ -38,16 +38,21 @@ public class CTEContext {
private String name;
private LogicalSubQueryAlias<Plan> parsedPlan;
// this cache only use once
- private LogicalPlan analyzedPlanCacheOnce;
- private Function<Plan, LogicalPlan> analyzePlanBuilder;
+ private LogicalPlan analyzedPlan;
+ private Callable<LogicalPlan> analyzePlanBuilder;
+
+ private int uniqueId;
Review Comment:
```suggestion
private int cteId;
```
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/analyzer/CTEContext.java:
##########
@@ -38,16 +38,21 @@ public class CTEContext {
private String name;
private LogicalSubQueryAlias<Plan> parsedPlan;
// this cache only use once
- private LogicalPlan analyzedPlanCacheOnce;
- private Function<Plan, LogicalPlan> analyzePlanBuilder;
+ private LogicalPlan analyzedPlan;
+ private Callable<LogicalPlan> analyzePlanBuilder;
+
+ private int uniqueId;
/* build head CTEContext */
public CTEContext() {
- this(null, null);
+ this(null, null, Integer.MIN_VALUE);
}
- /** CTEContext */
- public CTEContext(@Nullable LogicalSubQueryAlias<Plan> parsedPlan, @Nullable CTEContext previousCteContext) {
+ /**
+ * CTEContext
+ */
+ public CTEContext(@Nullable LogicalSubQueryAlias<Plan> parsedPlan, @Nullable CTEContext previousCteContext,
+ int uniqueId) {
Review Comment:
```suggestion
public CTEContext(@Nullable LogicalSubQueryAlias<Plan> parsedPlan,
@Nullable CTEContext previousCteContext, int uniqueId) {
```
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/analyzer/CTEContext.java:
##########
@@ -38,16 +38,21 @@ public class CTEContext {
private String name;
private LogicalSubQueryAlias<Plan> parsedPlan;
// this cache only use once
- private LogicalPlan analyzedPlanCacheOnce;
- private Function<Plan, LogicalPlan> analyzePlanBuilder;
+ private LogicalPlan analyzedPlan;
+ private Callable<LogicalPlan> analyzePlanBuilder;
+
+ private int uniqueId;
Review Comment:
use IdGenerator to generate cteId
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/logical/CTEProducerRewrite.java:
##########
@@ -0,0 +1,113 @@
+// 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.rules.rewrite.logical;
+
+import org.apache.doris.nereids.CascadesContext;
+import org.apache.doris.nereids.jobs.batch.NereidsRewriter;
+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.Expression;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalCTEConsumer;
+import org.apache.doris.nereids.trees.plans.logical.LogicalCTEProducer;
+import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
+import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.util.ExpressionUtils;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import org.apache.commons.collections.CollectionUtils;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * Rewrite CTE Producer recursively.
+ */
+public class CTEProducerRewrite extends OneRewriteRuleFactory {
+
+ @Override
+ public Rule build() {
+ return logicalCTEProducer().when(p -> !p.isRewritten()).thenApply(ctx -> {
+ LogicalCTEProducer<? extends Plan> cteProducer = ctx.root;
+ Set<Expression> projects = ctx.cascadesContext.findProjectForProducer(cteProducer.getCteId());
+ LogicalPlan child = tryToConstructFilter(ctx.cascadesContext, cteProducer.getCteId(),
+ (LogicalPlan) ctx.root.child());
+ // LogicalPlan child = (LogicalPlan) ctx.root.child();
+ if (CollectionUtils.isNotEmpty(projects)
+ && ctx.cascadesContext.couldProjectCTEProducerBasedOnCTEConsumer(cteProducer.getCteId())) {
+ child = new LogicalProject(ImmutableList.copyOf(projects), child);
+ }
+ CascadesContext rewrittenCtx = ctx.cascadesContext.forkForCTEProducer(child);
+ NereidsRewriter rewriter = new NereidsRewriter(rewrittenCtx);
+ rewriter.execute();
+ return cteProducer.withChildren(ImmutableList.of(rewrittenCtx.getRewritePlan()),
+ new ArrayList<>(child.getOutput()), true);
+ }).toRule(RuleType.CTE_PRODUCER_REWRITE);
+ }
+
+ private LogicalPlan tryToConstructFilter(CascadesContext cascadesContext, int cteId, LogicalPlan child) {
Review Comment:
it is very hard to understand, please add comment to explain what it do
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/batch/NereidsRewriter.java:
##########
@@ -85,51 +92,58 @@
*/
public class NereidsRewriter extends BatchRewriteJob {
private static final List<RewriteJob> REWRITE_JOBS = jobs(
+
+ bottomUp(new InlineCTE()),
+
topic("Plan Normalization",
- topDown(
- new EliminateOrderByConstant(),
- new EliminateGroupByConstant(),
- // MergeProjects depends on this rule
- new LogicalSubQueryAliasToLogicalProject(),
- // TODO: we should do expression normalization after plan normalization
- // because some rewritten depends on sub expression tree matching
- // such as group by key matching and replaced
- // but we need to do some normalization before subquery unnesting,
- // such as extract common expression.
- new ExpressionNormalization(),
- new ExpressionOptimization(),
- new AvgDistinctToSumDivCount(),
- new CountDistinctRewrite(),
- new ExtractFilterFromCrossJoin()
- ),
- topDown(
- // ExtractSingleTableExpressionFromDisjunction conflict to InPredicateToEqualToRule
- // in the ExpressionNormalization, so must invoke in another job, or else run into
- // dead loop
- new ExtractSingleTableExpressionFromDisjunction()
- )
+ topDown(
+ new EliminateOrderByConstant(),
+ new EliminateGroupByConstant(),
+ // MergeProjects depends on this rule
+ new LogicalSubQueryAliasToLogicalProject(),
+ // TODO: we should do expression normalization after plan normalization
+ // because some rewritten depends on sub expression tree matching
+ // such as group by key matching and replaced
+ // but we need to do some normalization before subquery unnesting,
+ // such as extract common expression.
+ new ExpressionNormalization(),
+ new ExpressionOptimization(),
+ new AvgDistinctToSumDivCount(),
+ new CountDistinctRewrite(),
+ new ExtractFilterFromCrossJoin()
+ ),
+ topDown(
+ // ExtractSingleTableExpressionFromDisjunction conflict to InPredicateToEqualToRule
+ // in the ExpressionNormalization, so must invoke in another job, or else run into
+ // dead loop
+ new ExtractSingleTableExpressionFromDisjunction()
+ )
),
+ topic("Rewrite CTE", topDown(
+ new PushdownFilterThroughCTEAnchor(),
+ new PushdownProjectThroughCTEAnchor())),
+
Review Comment:
why we need these rules? why we could see anchor here?
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindExpression.java:
##########
@@ -413,6 +415,18 @@ protected boolean condition(Rule rule, Plan plan) {
LogicalProject<Plan> project = sort.child();
return bindSort(sort, project, ctx.cascadesContext);
})
+ ), RuleType.BINDING_SORT_SLOT.build(
+ logicalSort(logicalCTEConsumer()).thenApply(ctx -> {
+ LogicalSort<LogicalCTEConsumer> sort = ctx.root;
+ LogicalCTEConsumer cteConsumer = sort.child();
+ return bindSort(sort, cteConsumer, ctx.cascadesContext);
+ })
Review Comment:
which sql could generate sort(consumer())?
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalCTE.java:
##########
@@ -40,14 +42,35 @@
private final List<LogicalSubQueryAlias<Plan>> aliasQueries;
+ private final Map<String, Integer> subQueryAliasToUniqueId;
+
+ private final boolean registered;
+
public LogicalCTE(List<LogicalSubQueryAlias<Plan>> aliasQueries, CHILD_TYPE child) {
- this(aliasQueries, Optional.empty(), Optional.empty(), child);
+ this(aliasQueries, Optional.empty(), Optional.empty(), child, false, null);
+ }
+
+ public LogicalCTE(List<LogicalSubQueryAlias<Plan>> aliasQueries, CHILD_TYPE child, boolean registered,
+ Map<String, Integer> subQueryAliasToUniqueId) {
+ this(aliasQueries, Optional.empty(), Optional.empty(), child, registered,
+ subQueryAliasToUniqueId);
}
public LogicalCTE(List<LogicalSubQueryAlias<Plan>> aliasQueries, Optional<GroupExpression> groupExpression,
- Optional<LogicalProperties> logicalProperties, CHILD_TYPE child) {
+ Optional<LogicalProperties> logicalProperties, CHILD_TYPE child, boolean registered,
+ Map<String, Integer> subQueryAliasToUniqueId) {
super(PlanType.LOGICAL_CTE, groupExpression, logicalProperties, child);
this.aliasQueries = ImmutableList.copyOf(Objects.requireNonNull(aliasQueries, "aliasQueries can not be null"));
+ this.registered = registered;
+ this.subQueryAliasToUniqueId = subQueryAliasToUniqueId == null ? initUniqueId() : subQueryAliasToUniqueId;
+ }
+
+ private Map<String, Integer> initUniqueId() {
+ Map<String, Integer> subQueryAliasToUniqueId = new HashMap<>();
+ for (LogicalSubQueryAlias<Plan> subQueryAlias : aliasQueries) {
+ subQueryAliasToUniqueId.put(subQueryAlias.getAlias(), subQueryAlias.getUniqueId());
+ }
+ return subQueryAliasToUniqueId;
Review Comment:
use ImmutableMap
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalCTE.java:
##########
@@ -40,14 +42,35 @@
private final List<LogicalSubQueryAlias<Plan>> aliasQueries;
+ private final Map<String, Integer> subQueryAliasToUniqueId;
+
+ private final boolean registered;
+
public LogicalCTE(List<LogicalSubQueryAlias<Plan>> aliasQueries, CHILD_TYPE child) {
- this(aliasQueries, Optional.empty(), Optional.empty(), child);
+ this(aliasQueries, Optional.empty(), Optional.empty(), child, false, null);
Review Comment:
do not use `null` as Map object. use `ImmutableMap.of()`
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalCTEAnchor.java:
##########
@@ -0,0 +1,92 @@
+// 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.trees.plans.logical;
+
+import org.apache.doris.nereids.memo.GroupExpression;
+import org.apache.doris.nereids.properties.LogicalProperties;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.PlanType;
+import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
+
+import com.google.common.collect.ImmutableList;
+
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * LogicalCTEAnchor
+ */
+public class LogicalCTEAnchor<
+ LEFT_CHILD_TYPE extends Plan,
+ RIGHT_CHILD_TYPE extends Plan> extends LogicalBinary<LEFT_CHILD_TYPE, RIGHT_CHILD_TYPE> {
Review Comment:
```suggestion
public class LogicalCTEAnchor<LEFT_CHILD_TYPE extends Plan, RIGHT_CHILD_TYPE extends Plan>
extends LogicalBinary<LEFT_CHILD_TYPE, RIGHT_CHILD_TYPE> {
```
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalCTEAnchor.java:
##########
@@ -0,0 +1,92 @@
+// 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.trees.plans.logical;
+
+import org.apache.doris.nereids.memo.GroupExpression;
+import org.apache.doris.nereids.properties.LogicalProperties;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.PlanType;
+import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
+
+import com.google.common.collect.ImmutableList;
+
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * LogicalCTEAnchor
+ */
+public class LogicalCTEAnchor<
+ LEFT_CHILD_TYPE extends Plan,
+ RIGHT_CHILD_TYPE extends Plan> extends LogicalBinary<LEFT_CHILD_TYPE, RIGHT_CHILD_TYPE> {
+
+ private final int cteId;
Review Comment:
useless, remove it
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/analyzer/CTEContext.java:
##########
@@ -80,26 +86,40 @@ public Optional<LogicalSubQueryAlias<Plan>> getParsedCtePlan(String cteName) {
return findCTEContext(cteName).map(cte -> cte.parsedPlan);
}
- /** getAnalyzedCTE */
+ /**
+ * getAnalyzedCTE
+ */
public Optional<LogicalPlan> getAnalyzedCTE(String cteName) {
Review Comment:
```suggestion
public Optional<LogicalPlan> getForReuse(String cteName) {
```
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/CollectFilterAboveConsumer.java:
##########
@@ -0,0 +1,52 @@
+// 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.trees.plans.logical;
+
+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.Expression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+
+import java.util.Set;
+
+/**
+ * CollectFilterOnConsumer
+ */
+public class CollectFilterAboveConsumer extends OneRewriteRuleFactory {
+
+ @Override
+ public Rule build() {
+ return logicalFilter(logicalCTEConsumer()).thenApply(ctx -> {
+ LogicalFilter<LogicalCTEConsumer> filter = ctx.root;
+ LogicalCTEConsumer cteConsumer = filter.child();
+ Set<Expression> exprs = filter.getConjuncts();
+ for (Expression expr : exprs) {
+ Expression rewrittenExpr = expr.rewriteUp(e -> {
+ if (e instanceof Slot) {
+ return cteConsumer.findProducerSlot((Slot) e);
+ }
+ return e;
+ });
+ ctx.cascadesContext.putCTEIdToFilter(cteConsumer.getCteId(), rewrittenExpr);
Review Comment:
`CTEIdToFilter` seem not use anymore
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/analyzer/CTEContext.java:
##########
@@ -80,26 +86,40 @@ public Optional<LogicalSubQueryAlias<Plan>> getParsedCtePlan(String cteName) {
return findCTEContext(cteName).map(cte -> cte.parsedPlan);
}
- /** getAnalyzedCTE */
+ /**
+ * getAnalyzedCTE
+ */
public Optional<LogicalPlan> getAnalyzedCTE(String cteName) {
+ if (!findCTEContext(cteName).isPresent()) {
+ return Optional.empty();
+ }
+ return Optional.of(findCTEContext(cteName).get().analyzedPlan);
+ }
Review Comment:
```suggestion
return findCTEContext(cteName).map(CTEContext::analyzedPlan);
}
```
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/logical/ColumnPruning.java:
##########
@@ -262,6 +263,9 @@ private <P extends Plan> P pruneChildren(P plan, Set<Slot> parentRequiredSlots)
}
private Plan doPruneChild(Plan plan, Plan child, Set<Slot> childRequiredSlots) {
+ if (child instanceof LogicalCTEProducer) {
+ return child;
+ }
Review Comment:
why need this, i think when u do column prune, u could not see any LogicalCTEProducer
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/logical/CTEProducerRewrite.java:
##########
@@ -0,0 +1,113 @@
+// 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.rules.rewrite.logical;
+
+import org.apache.doris.nereids.CascadesContext;
+import org.apache.doris.nereids.jobs.batch.NereidsRewriter;
+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.Expression;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalCTEConsumer;
+import org.apache.doris.nereids.trees.plans.logical.LogicalCTEProducer;
+import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
+import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.util.ExpressionUtils;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import org.apache.commons.collections.CollectionUtils;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * Rewrite CTE Producer recursively.
+ */
+public class CTEProducerRewrite extends OneRewriteRuleFactory {
+
+ @Override
+ public Rule build() {
+ return logicalCTEProducer().when(p -> !p.isRewritten()).thenApply(ctx -> {
+ LogicalCTEProducer<? extends Plan> cteProducer = ctx.root;
+ Set<Expression> projects = ctx.cascadesContext.findProjectForProducer(cteProducer.getCteId());
+ LogicalPlan child = tryToConstructFilter(ctx.cascadesContext, cteProducer.getCteId(),
+ (LogicalPlan) ctx.root.child());
+ // LogicalPlan child = (LogicalPlan) ctx.root.child();
Review Comment:
remove useless code
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/logical/CTEProducerRewrite.java:
##########
@@ -0,0 +1,113 @@
+// 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.rules.rewrite.logical;
+
+import org.apache.doris.nereids.CascadesContext;
+import org.apache.doris.nereids.jobs.batch.NereidsRewriter;
+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.Expression;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalCTEConsumer;
+import org.apache.doris.nereids.trees.plans.logical.LogicalCTEProducer;
+import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
+import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.util.ExpressionUtils;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import org.apache.commons.collections.CollectionUtils;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * Rewrite CTE Producer recursively.
+ */
+public class CTEProducerRewrite extends OneRewriteRuleFactory {
+
+ @Override
+ public Rule build() {
+ return logicalCTEProducer().when(p -> !p.isRewritten()).thenApply(ctx -> {
+ LogicalCTEProducer<? extends Plan> cteProducer = ctx.root;
+ Set<Expression> projects = ctx.cascadesContext.findProjectForProducer(cteProducer.getCteId());
+ LogicalPlan child = tryToConstructFilter(ctx.cascadesContext, cteProducer.getCteId(),
+ (LogicalPlan) ctx.root.child());
+ // LogicalPlan child = (LogicalPlan) ctx.root.child();
+ if (CollectionUtils.isNotEmpty(projects)
+ && ctx.cascadesContext.couldProjectCTEProducerBasedOnCTEConsumer(cteProducer.getCteId())) {
+ child = new LogicalProject(ImmutableList.copyOf(projects), child);
+ }
+ CascadesContext rewrittenCtx = ctx.cascadesContext.forkForCTEProducer(child);
+ NereidsRewriter rewriter = new NereidsRewriter(rewrittenCtx);
+ rewriter.execute();
+ return cteProducer.withChildren(ImmutableList.of(rewrittenCtx.getRewritePlan()),
+ new ArrayList<>(child.getOutput()), true);
+ }).toRule(RuleType.CTE_PRODUCER_REWRITE);
+ }
+
+ private LogicalPlan tryToConstructFilter(CascadesContext cascadesContext, int cteId, LogicalPlan child) {
+ Set<Integer> consumerIds = cascadesContext.getCteIdToConsumer().get(cteId).stream()
+ .map(LogicalCTEConsumer::getConsumerId).collect(
Review Comment:
please wrap at each chain function after `stream()` when u use stream api
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/analyzer/CTEContext.java:
##########
@@ -80,26 +86,40 @@ public Optional<LogicalSubQueryAlias<Plan>> getParsedCtePlan(String cteName) {
return findCTEContext(cteName).map(cte -> cte.parsedPlan);
}
- /** getAnalyzedCTE */
+ /**
+ * getAnalyzedCTE
+ */
public Optional<LogicalPlan> getAnalyzedCTE(String cteName) {
+ if (!findCTEContext(cteName).isPresent()) {
+ return Optional.empty();
+ }
+ return Optional.of(findCTEContext(cteName).get().analyzedPlan);
+ }
+
+ public Optional<LogicalPlan> getForInline(String cteName) {
return findCTEContext(cteName).map(CTEContext::doAnalyzeCTE);
}
- /** findCTEContext */
+ /**
+ * findCTEContext
+ */
public Optional<CTEContext> findCTEContext(String cteName) {
+ if (cteName.equals(name)) {
+ return Optional.of(this);
+ }
CTEContext cteContext = cteContextMap.get(cteName);
return Optional.ofNullable(cteContext);
}
private LogicalPlan doAnalyzeCTE() {
- // we always analyze a cte as least once, if the cte only use once, we can return analyzedPlanCacheOnce.
- // but if the cte use more then once, we should return difference analyzed plan to generate difference
- // relation id, so the relation will not conflict in the memo.
- if (analyzedPlanCacheOnce != null) {
- LogicalPlan analyzedPlan = analyzedPlanCacheOnce;
- analyzedPlanCacheOnce = null;
- return analyzedPlan;
+ try {
+ return analyzePlanBuilder.call();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
Review Comment:
need to return clear error msg
```suggestion
throw new AnalysisException(..., e);
```
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalCTEConsumer.java:
##########
@@ -0,0 +1,175 @@
+// 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.trees.plans.logical;
+
+import org.apache.doris.nereids.memo.GroupExpression;
+import org.apache.doris.nereids.properties.LogicalProperties;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.StatementScopeIdGenerator;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.PlanType;
+import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+/**
+ * LogicalCTEConsumer
+ */
+public class LogicalCTEConsumer extends LogicalLeaf {
+
+ private final LogicalPlan childPlan;
+
+ private final int cteId;
+
+ private final Map<Slot, Slot> consumerToProducerOutputMap = new LinkedHashMap<>();
+
+ private final Map<Slot, Slot> producerToConsumerOutputMap = new LinkedHashMap<>();
+
+ private final int consumerId;
Review Comment:
use IdGenerator to generate this id
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalCTEConsumer.java:
##########
@@ -0,0 +1,175 @@
+// 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.trees.plans.logical;
+
+import org.apache.doris.nereids.memo.GroupExpression;
+import org.apache.doris.nereids.properties.LogicalProperties;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.StatementScopeIdGenerator;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.PlanType;
+import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+/**
+ * LogicalCTEConsumer
+ */
+public class LogicalCTEConsumer extends LogicalLeaf {
+
+ private final LogicalPlan childPlan;
+
+ private final int cteId;
+
+ private final Map<Slot, Slot> consumerToProducerOutputMap = new LinkedHashMap<>();
+
+ private final Map<Slot, Slot> producerToConsumerOutputMap = new LinkedHashMap<>();
+
+ private final int consumerId;
+
+ private final String name;
+
+ /**
+ * Logical CTE consumer.
+ */
+ public LogicalCTEConsumer(Optional<GroupExpression> groupExpression,
+ Optional<LogicalProperties> logicalProperties, LogicalPlan childPlan, int cteId, String name) {
+ super(PlanType.LOGICAL_CTE_RELATION, groupExpression, logicalProperties);
+ this.childPlan = childPlan;
+ this.cteId = cteId;
+ this.name = name;
+ initProducerToConsumerOutputMap(childPlan);
+ for (Map.Entry<Slot, Slot> entry : producerToConsumerOutputMap.entrySet()) {
+ this.consumerToProducerOutputMap.put(entry.getValue(), entry.getKey());
+ }
+ this.consumerId = StatementScopeIdGenerator.newExprId().asInt();
+ }
+
+ /**
+ * Logical CTE consumer.
+ */
+ public LogicalCTEConsumer(Optional<GroupExpression> groupExpression,
+ Optional<LogicalProperties> logicalProperties, LogicalPlan childPlan, int cteId,
+ Map<Slot, Slot> consumerToProducerOutputMap,
+ Map<Slot, Slot> producerToConsumerOutputMap, int consumerId, String name) {
+ super(PlanType.LOGICAL_CTE_RELATION, groupExpression, logicalProperties);
+ this.childPlan = childPlan;
+ this.cteId = cteId;
+ this.consumerToProducerOutputMap.putAll(consumerToProducerOutputMap);
+ this.producerToConsumerOutputMap.putAll(producerToConsumerOutputMap);
+ this.consumerId = consumerId;
+ this.name = name;
+ }
+
+ private void initProducerToConsumerOutputMap(LogicalPlan childPlan) {
+ List<Slot> producerOutput = childPlan.getOutput();
+ for (Slot producerOutputSlot : producerOutput) {
+ Slot consumerSlot = new SlotReference(producerOutputSlot.getName(),
+ producerOutputSlot.getDataType(), producerOutputSlot.nullable(), ImmutableList.of(name));
+ producerToConsumerOutputMap.put(producerOutputSlot, consumerSlot);
+ }
+ }
+
+ public Map<Slot, Slot> getConsumerToProducerOutputMap() {
+ return consumerToProducerOutputMap;
+ }
+
+ public Map<Slot, Slot> getProducerToConsumerOutputMap() {
+ return producerToConsumerOutputMap;
+ }
+
+ @Override
+ public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
+ return visitor.visitLogicalCTEConsumer(this, context);
+ }
+
+ @Override
+ public List<? extends Expression> getExpressions() {
+ return ImmutableList.of();
+ }
+
+ @Override
+ public Plan withGroupExpression(Optional<GroupExpression> groupExpression) {
+ return new LogicalCTEConsumer(groupExpression, Optional.of(getLogicalProperties()), childPlan, cteId,
+ consumerToProducerOutputMap,
+ producerToConsumerOutputMap,
+ consumerId, name);
+ }
+
+ @Override
+ public Plan withLogicalProperties(Optional<LogicalProperties> logicalProperties) {
+ return new LogicalCTEConsumer(groupExpression, logicalProperties, childPlan, cteId,
+ consumerToProducerOutputMap,
+ producerToConsumerOutputMap,
+ consumerId, name);
+ }
+
+ @Override
+ public List<Slot> computeOutput() {
+ return new ArrayList<>(producerToConsumerOutputMap.values());
+ }
+
+ public int getCteId() {
+ return cteId;
+ }
+
+ @Override
+ public int hashCode() {
+ return consumerId;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ return this.consumerId == ((LogicalCTEConsumer) o).consumerId;
+ }
+
+ public int getConsumerId() {
+ return consumerId;
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ public Slot findProducerSlot(Slot consumerSlot) {
+ Slot slot = consumerToProducerOutputMap.get(consumerSlot);
+ Preconditions.checkArgument(slot != null, String.format("Required producer"
+ + "slot for :%s doesn't exist", consumerSlot));
+ return slot;
+ }
+
+ @Override
+ public String toString() {
+ return String.format("LOGICAL_CTE_CONSUMER#%d#%d", cteId, consumerId);
Review Comment:
```suggestion
return Utils.toSqlString("LogicalCteConsumer[" + id.asInt() + "]",
"cteId", cteId,
"consumerId", consumerId);
```
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/logical/CTEProducerRewrite.java:
##########
@@ -0,0 +1,113 @@
+// 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.rules.rewrite.logical;
+
+import org.apache.doris.nereids.CascadesContext;
+import org.apache.doris.nereids.jobs.batch.NereidsRewriter;
+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.Expression;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalCTEConsumer;
+import org.apache.doris.nereids.trees.plans.logical.LogicalCTEProducer;
+import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
+import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.util.ExpressionUtils;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import org.apache.commons.collections.CollectionUtils;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * Rewrite CTE Producer recursively.
+ */
+public class CTEProducerRewrite extends OneRewriteRuleFactory {
+
+ @Override
+ public Rule build() {
+ return logicalCTEProducer().when(p -> !p.isRewritten()).thenApply(ctx -> {
+ LogicalCTEProducer<? extends Plan> cteProducer = ctx.root;
+ Set<Expression> projects = ctx.cascadesContext.findProjectForProducer(cteProducer.getCteId());
+ LogicalPlan child = tryToConstructFilter(ctx.cascadesContext, cteProducer.getCteId(),
+ (LogicalPlan) ctx.root.child());
+ // LogicalPlan child = (LogicalPlan) ctx.root.child();
+ if (CollectionUtils.isNotEmpty(projects)
+ && ctx.cascadesContext.couldProjectCTEProducerBasedOnCTEConsumer(cteProducer.getCteId())) {
+ child = new LogicalProject(ImmutableList.copyOf(projects), child);
+ }
+ CascadesContext rewrittenCtx = ctx.cascadesContext.forkForCTEProducer(child);
+ NereidsRewriter rewriter = new NereidsRewriter(rewrittenCtx);
+ rewriter.execute();
+ return cteProducer.withChildren(ImmutableList.of(rewrittenCtx.getRewritePlan()),
+ new ArrayList<>(child.getOutput()), true);
+ }).toRule(RuleType.CTE_PRODUCER_REWRITE);
+ }
+
+ private LogicalPlan tryToConstructFilter(CascadesContext cascadesContext, int cteId, LogicalPlan child) {
+ Set<Integer> consumerIds = cascadesContext.getCteIdToConsumer().get(cteId).stream()
+ .map(LogicalCTEConsumer::getConsumerId).collect(
+ Collectors.toSet());
+ Set<Set<Expression>> filtersAboveEachConsumer = cascadesContext.getConsumerIdToFilter().entrySet().stream()
+ .filter(kv -> consumerIds.contains(
+ kv.getKey())).map(Entry::getValue).collect(Collectors.toSet());
Review Comment:
why not generate set<set> directly in `CollectFilterAboveConsumer`
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/cascades/ApplyRuleJob.java:
##########
@@ -55,6 +56,7 @@ public ApplyRuleJob(GroupExpression groupExpression, Rule rule, JobContext conte
super(JobType.APPLY_RULE, context);
this.groupExpression = groupExpression;
this.rule = rule;
+ super.cteIdToStats = new HashMap<>();
Review Comment:
maybe u need u change super's ctor
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalCTEProducer.java:
##########
@@ -0,0 +1,139 @@
+// 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.trees.plans.logical;
+
+import org.apache.doris.nereids.memo.GroupExpression;
+import org.apache.doris.nereids.properties.LogicalProperties;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.PlanType;
+import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
+
+import com.google.common.base.Objects;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * LogicalCTEProducer
+ */
+public class LogicalCTEProducer<CHILD_TYPE extends Plan>
+ extends LogicalUnary<CHILD_TYPE> {
+
+ private final int cteId;
+
+ private final List<Slot> projects;
+
+ private final boolean rewritten;
+
+ public LogicalCTEProducer(CHILD_TYPE child, int cteId) {
+ super(PlanType.LOGICAL_CTE_PRODUCER, child);
+ this.cteId = cteId;
+ this.projects = Collections.emptyList();
+ this.rewritten = false;
+ }
+
+ public LogicalCTEProducer(Optional<GroupExpression> groupExpression,
+ Optional<LogicalProperties> logicalProperties, CHILD_TYPE child, int cteId,
+ List<Slot> projects, boolean rewritten) {
+ super(PlanType.LOGICAL_CTE_PRODUCER, groupExpression, logicalProperties, child);
+ this.cteId = cteId;
+ this.projects = projects;
+ this.rewritten = rewritten;
+ }
+
+ public int getCteId() {
+ return cteId;
+ }
+
+ public List<Slot> getProjects() {
+ return projects;
+ }
+
+ @Override
+ public Plan withChildren(List<Plan> children) {
+ return new LogicalCTEProducer<>(groupExpression, Optional.of(getLogicalProperties()), children.get(0),
+ cteId, projects, rewritten);
+ }
+
+ public Plan withChildren(List<Plan> children, List<Slot> projects, boolean rewritten) {
+ return new LogicalCTEProducer<>(groupExpression, Optional.of(getLogicalProperties()), children.get(0),
+ cteId, projects, rewritten);
+ }
+
+ @Override
+ public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
+ return visitor.visitLogicalCTEProducer(this, context);
+ }
+
+ @Override
+ public List<? extends Expression> getExpressions() {
+ return child().getExpressions();
+ }
+
+ @Override
+ public Plan withGroupExpression(Optional<GroupExpression> groupExpression) {
+ return new LogicalCTEProducer<>(groupExpression, Optional.of(getLogicalProperties()), child(), cteId,
+ projects, rewritten);
+ }
+
+ @Override
+ public Plan withLogicalProperties(Optional<LogicalProperties> logicalProperties) {
+ return new LogicalCTEProducer<>(groupExpression, logicalProperties, child(), cteId,
+ projects, rewritten);
+ }
+
+ @Override
+ public List<Slot> computeOutput() {
+ return child().computeOutput();
+ }
+
+ @Override
+ public String toString() {
+ return String.format("LOGICAL_CTE_PRODUCER#%d", cteId);
Review Comment:
```suggestion
return Utils.toSqlString("LogicalCteProducer[" + id.asInt() + "]",
"cteId", cteId);
```
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalCTEAnchor.java:
##########
@@ -0,0 +1,92 @@
+// 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.trees.plans.logical;
+
+import org.apache.doris.nereids.memo.GroupExpression;
+import org.apache.doris.nereids.properties.LogicalProperties;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.PlanType;
+import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
+
+import com.google.common.collect.ImmutableList;
+
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * LogicalCTEAnchor
+ */
+public class LogicalCTEAnchor<
+ LEFT_CHILD_TYPE extends Plan,
+ RIGHT_CHILD_TYPE extends Plan> extends LogicalBinary<LEFT_CHILD_TYPE, RIGHT_CHILD_TYPE> {
+
+ private final int cteId;
+
+ public LogicalCTEAnchor(LEFT_CHILD_TYPE leftChild, RIGHT_CHILD_TYPE rightChild, int cteId) {
+ this(Optional.empty(), Optional.empty(), leftChild, rightChild, cteId);
+ }
+
+ public LogicalCTEAnchor(Optional<GroupExpression> groupExpression,
+ Optional<LogicalProperties> logicalProperties,
+ LEFT_CHILD_TYPE leftChild, RIGHT_CHILD_TYPE rightChild, int cteId) {
+ super(PlanType.LOGICAL_CTE_ANCHOR, groupExpression, logicalProperties, leftChild, rightChild);
+ this.cteId = cteId;
+ }
+
+ @Override
+ public Plan withChildren(List<Plan> children) {
+ return new LogicalCTEAnchor<>(groupExpression, Optional.of(getLogicalProperties()),
+ children.get(0), children.get(1), cteId);
+ }
+
+ @Override
+ public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
+ return visitor.visitLogicalCTEAnchor(this, context);
+ }
+
+ @Override
+ public List<? extends Expression> getExpressions() {
+ return ImmutableList.of();
+ }
+
+ @Override
+ public Plan withGroupExpression(Optional<GroupExpression> groupExpression) {
+ return new LogicalCTEAnchor<>(groupExpression, Optional.of(getLogicalProperties()), left(), right(), cteId);
+ }
+
+ @Override
+ public Plan withLogicalProperties(Optional<LogicalProperties> logicalProperties) {
+ return new LogicalCTEAnchor<>(groupExpression, logicalProperties, left(), right(), cteId);
+ }
+
+ @Override
+ public List<Slot> computeOutput() {
+ return right().getOutput();
+ }
+
+ public int getCteId() {
+ return cteId;
+ }
+
+ @Override
+ public String toString() {
+ return String.format("LOGICAL_CTE_ANCHOR#%d", cteId);
Review Comment:
```suggestion
return Utils.toSqlString("LogicalCteAnchor[" + id.asInt() + "]",
"cteId", cteId);
```
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/logical/CTEProducerRewrite.java:
##########
@@ -0,0 +1,113 @@
+// 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.rules.rewrite.logical;
+
+import org.apache.doris.nereids.CascadesContext;
+import org.apache.doris.nereids.jobs.batch.NereidsRewriter;
+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.Expression;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalCTEConsumer;
+import org.apache.doris.nereids.trees.plans.logical.LogicalCTEProducer;
+import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
+import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.util.ExpressionUtils;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import org.apache.commons.collections.CollectionUtils;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * Rewrite CTE Producer recursively.
+ */
+public class CTEProducerRewrite extends OneRewriteRuleFactory {
+
+ @Override
+ public Rule build() {
+ return logicalCTEProducer().when(p -> !p.isRewritten()).thenApply(ctx -> {
+ LogicalCTEProducer<? extends Plan> cteProducer = ctx.root;
+ Set<Expression> projects = ctx.cascadesContext.findProjectForProducer(cteProducer.getCteId());
+ LogicalPlan child = tryToConstructFilter(ctx.cascadesContext, cteProducer.getCteId(),
+ (LogicalPlan) ctx.root.child());
+ // LogicalPlan child = (LogicalPlan) ctx.root.child();
+ if (CollectionUtils.isNotEmpty(projects)
+ && ctx.cascadesContext.couldProjectCTEProducerBasedOnCTEConsumer(cteProducer.getCteId())) {
+ child = new LogicalProject(ImmutableList.copyOf(projects), child);
+ }
+ CascadesContext rewrittenCtx = ctx.cascadesContext.forkForCTEProducer(child);
+ NereidsRewriter rewriter = new NereidsRewriter(rewrittenCtx);
+ rewriter.execute();
+ return cteProducer.withChildren(ImmutableList.of(rewrittenCtx.getRewritePlan()),
+ new ArrayList<>(child.getOutput()), true);
+ }).toRule(RuleType.CTE_PRODUCER_REWRITE);
+ }
+
+ private LogicalPlan tryToConstructFilter(CascadesContext cascadesContext, int cteId, LogicalPlan child) {
+ Set<Integer> consumerIds = cascadesContext.getCteIdToConsumer().get(cteId).stream()
+ .map(LogicalCTEConsumer::getConsumerId).collect(
+ Collectors.toSet());
+ Set<Set<Expression>> filtersAboveEachConsumer = cascadesContext.getConsumerIdToFilter().entrySet().stream()
+ .filter(kv -> consumerIds.contains(
+ kv.getKey())).map(Entry::getValue).collect(Collectors.toSet());
+ Set<Expression> someone = filtersAboveEachConsumer.stream().findFirst().orElse(null);
+ if (someone == null) {
+ return child;
+ }
+ int filterSize = cascadesContext.getCteIdToConsumer().get(cteId).size();
+ Set<Expression> filter = new HashSet<>();
+ for (Expression f : someone) {
+ int matchCount = 1;
+ Set<SlotReference> slots = f.collect(e -> e instanceof SlotReference);
+ Set<Expression> mightBeJoined = new HashSet<>();
+ for (Set<Expression> another : filtersAboveEachConsumer) {
+ if (another.equals(someone)) {
+ continue;
+ }
+ Set<Expression> matched = new HashSet<>();
+ for (Expression e : another) {
+ Set<SlotReference> otherSlots = e.collect(ae -> ae instanceof SlotReference);
+ if (otherSlots.equals(slots)) {
+ matched.add(e);
+ }
+ }
+ if (!matched.isEmpty()) {
+ matchCount++;
+ }
+ mightBeJoined.addAll(matched);
+ }
+ if (matchCount >= filterSize) {
+ mightBeJoined.add(f);
+ filter.add(ExpressionUtils.or(mightBeJoined));
Review Comment:
why only push join? i don't think it is enough.
we should push all filter into produce to reduce exchange data size
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindExpression.java:
##########
@@ -413,6 +415,18 @@ protected boolean condition(Rule rule, Plan plan) {
LogicalProject<Plan> project = sort.child();
return bindSort(sort, project, ctx.cascadesContext);
})
+ ), RuleType.BINDING_SORT_SLOT.build(
+ logicalSort(logicalCTEConsumer()).thenApply(ctx -> {
+ LogicalSort<LogicalCTEConsumer> sort = ctx.root;
+ LogicalCTEConsumer cteConsumer = sort.child();
+ return bindSort(sort, cteConsumer, ctx.cascadesContext);
+ })
+ ), RuleType.BINDING_SORT_SLOT.build(
+ logicalSort(logicalCTE()).thenApply(ctx -> {
+ LogicalSort<LogicalCTE<Plan>> sort = ctx.root;
+ LogicalCTE<Plan> cteConsumer = sort.child();
+ return bindSort(sort, cteConsumer, ctx.cascadesContext);
Review Comment:
which sql could generate sort(cte)?
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalCTEProducer.java:
##########
@@ -0,0 +1,139 @@
+// 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.trees.plans.logical;
+
+import org.apache.doris.nereids.memo.GroupExpression;
+import org.apache.doris.nereids.properties.LogicalProperties;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.PlanType;
+import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
+
+import com.google.common.base.Objects;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * LogicalCTEProducer
+ */
+public class LogicalCTEProducer<CHILD_TYPE extends Plan>
+ extends LogicalUnary<CHILD_TYPE> {
+
+ private final int cteId;
+
+ private final List<Slot> projects;
+
+ private final boolean rewritten;
+
+ public LogicalCTEProducer(CHILD_TYPE child, int cteId) {
+ super(PlanType.LOGICAL_CTE_PRODUCER, child);
+ this.cteId = cteId;
+ this.projects = Collections.emptyList();
+ this.rewritten = false;
+ }
+
+ public LogicalCTEProducer(Optional<GroupExpression> groupExpression,
+ Optional<LogicalProperties> logicalProperties, CHILD_TYPE child, int cteId,
+ List<Slot> projects, boolean rewritten) {
+ super(PlanType.LOGICAL_CTE_PRODUCER, groupExpression, logicalProperties, child);
+ this.cteId = cteId;
+ this.projects = projects;
+ this.rewritten = rewritten;
+ }
+
+ public int getCteId() {
+ return cteId;
+ }
+
+ public List<Slot> getProjects() {
+ return projects;
+ }
+
+ @Override
+ public Plan withChildren(List<Plan> children) {
Review Comment:
check children size
##########
fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java:
##########
@@ -330,6 +330,8 @@ public class SessionVariable implements Serializable, Writable {
public static final String ENABLE_ORC_LAZY_MAT = "enable_orc_lazy_materialization";
+ public static final String INLINE_CTE_REFERENCED_THRESHOLD = "inline_cte_referenced_threshold";
Review Comment:
cte_reuse_threshold?
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalSubQueryAlias.java:
##########
@@ -156,4 +169,16 @@ public LogicalSubQueryAlias<CHILD_TYPE> withLogicalProperties(Optional<LogicalPr
return new LogicalSubQueryAlias<>(qualifier, columnAliases, Optional.empty(),
logicalProperties, child());
}
+
+ public int uniqueId() {
+ return StatementScopeIdGenerator.newExprId().asInt();
Review Comment:
do not use this generator, use a new one for cte only
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalSubQueryAlias.java:
##########
@@ -44,6 +45,8 @@
private final List<String> qualifier;
private final Optional<List<String>> columnAliases;
+ private final int uniqueId;
Review Comment:
```suggestion
private final int cteId;
```
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalCTEProducer.java:
##########
@@ -0,0 +1,139 @@
+// 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.trees.plans.logical;
+
+import org.apache.doris.nereids.memo.GroupExpression;
+import org.apache.doris.nereids.properties.LogicalProperties;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.PlanType;
+import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
+
+import com.google.common.base.Objects;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * LogicalCTEProducer
+ */
+public class LogicalCTEProducer<CHILD_TYPE extends Plan>
+ extends LogicalUnary<CHILD_TYPE> {
+
+ private final int cteId;
+
+ private final List<Slot> projects;
+
+ private final boolean rewritten;
+
+ public LogicalCTEProducer(CHILD_TYPE child, int cteId) {
+ super(PlanType.LOGICAL_CTE_PRODUCER, child);
+ this.cteId = cteId;
+ this.projects = Collections.emptyList();
+ this.rewritten = false;
+ }
+
+ public LogicalCTEProducer(Optional<GroupExpression> groupExpression,
+ Optional<LogicalProperties> logicalProperties, CHILD_TYPE child, int cteId,
+ List<Slot> projects, boolean rewritten) {
+ super(PlanType.LOGICAL_CTE_PRODUCER, groupExpression, logicalProperties, child);
+ this.cteId = cteId;
+ this.projects = projects;
Review Comment:
ImmutableList.copyOf(Objects.requireNotNull(projects, "xxxx should not null"))
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalCTEProducer.java:
##########
@@ -0,0 +1,139 @@
+// 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.trees.plans.logical;
+
+import org.apache.doris.nereids.memo.GroupExpression;
+import org.apache.doris.nereids.properties.LogicalProperties;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.PlanType;
+import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
+
+import com.google.common.base.Objects;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * LogicalCTEProducer
+ */
+public class LogicalCTEProducer<CHILD_TYPE extends Plan>
+ extends LogicalUnary<CHILD_TYPE> {
+
+ private final int cteId;
+
+ private final List<Slot> projects;
+
+ private final boolean rewritten;
+
+ public LogicalCTEProducer(CHILD_TYPE child, int cteId) {
+ super(PlanType.LOGICAL_CTE_PRODUCER, child);
+ this.cteId = cteId;
+ this.projects = Collections.emptyList();
+ this.rewritten = false;
+ }
+
+ public LogicalCTEProducer(Optional<GroupExpression> groupExpression,
+ Optional<LogicalProperties> logicalProperties, CHILD_TYPE child, int cteId,
+ List<Slot> projects, boolean rewritten) {
+ super(PlanType.LOGICAL_CTE_PRODUCER, groupExpression, logicalProperties, child);
+ this.cteId = cteId;
+ this.projects = projects;
+ this.rewritten = rewritten;
+ }
+
+ public int getCteId() {
+ return cteId;
+ }
+
+ public List<Slot> getProjects() {
+ return projects;
+ }
+
+ @Override
+ public Plan withChildren(List<Plan> children) {
+ return new LogicalCTEProducer<>(groupExpression, Optional.of(getLogicalProperties()), children.get(0),
+ cteId, projects, rewritten);
+ }
+
+ public Plan withChildren(List<Plan> children, List<Slot> projects, boolean rewritten) {
Review Comment:
withChildrenAndProjects
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalSubQueryAlias.java:
##########
@@ -44,6 +45,8 @@
private final List<String> qualifier;
private final Optional<List<String>> columnAliases;
+ private final int uniqueId;
Review Comment:
do not use int, use a class CteId like ExprId
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindRelation.java:
##########
@@ -127,15 +128,17 @@ private TableIf getTable(String catalogName, String dbName, String tableName, En
private LogicalPlan bindWithCurrentDb(CascadesContext cascadesContext, UnboundRelation unboundRelation) {
String tableName = unboundRelation.getNameParts().get(0);
// check if it is a CTE's name
- CTEContext cteContext = cascadesContext.getCteContext();
- Optional<LogicalPlan> analyzedCte = cteContext.getAnalyzedCTE(tableName);
- if (analyzedCte.isPresent()) {
- LogicalPlan ctePlan = analyzedCte.get();
- if (ctePlan instanceof LogicalSubQueryAlias
- && ((LogicalSubQueryAlias<?>) ctePlan).getAlias().equals(tableName)) {
- return ctePlan;
+ CTEContext cteContext = cascadesContext.getCteContext().findCTEContext(tableName).orElse(null);
+ if (cteContext != null) {
+ Optional<LogicalPlan> analyzedCte = cteContext.getAnalyzedCTE(tableName);
+ if (analyzedCte.isPresent()) {
+ LogicalPlan realChild = ((LogicalSubQueryAlias) (analyzedCte.get())).toLogicalProject();
Review Comment:
why do toProject here? we have a rewrite rule to trans LogicalSubQueryAlias to project.
if u just want get output from realChild, should call getOutput from LogicalSubQueryAlias directly, no need to do toLogicalProject
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/RuleType.java:
##########
@@ -56,7 +56,6 @@ public enum RuleType {
BINDING_SET_OPERATION_SLOT(RuleTypeClass.REWRITE),
BINDING_GENERATE_FUNCTION(RuleTypeClass.REWRITE),
BINDING_INSERT_TARGET_TABLE(RuleTypeClass.REWRITE),
-
Review Comment:
add back this blank line
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/CollectFilterAboveConsumer.java:
##########
Review Comment:
this file should under `rules/rewrite/logical` dir
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalCTE.java:
##########
@@ -40,14 +42,35 @@
private final List<LogicalSubQueryAlias<Plan>> aliasQueries;
+ private final Map<String, Integer> subQueryAliasToUniqueId;
+
+ private final boolean registered;
+
public LogicalCTE(List<LogicalSubQueryAlias<Plan>> aliasQueries, CHILD_TYPE child) {
- this(aliasQueries, Optional.empty(), Optional.empty(), child);
+ this(aliasQueries, Optional.empty(), Optional.empty(), child, false, null);
+ }
+
+ public LogicalCTE(List<LogicalSubQueryAlias<Plan>> aliasQueries, CHILD_TYPE child, boolean registered,
+ Map<String, Integer> subQueryAliasToUniqueId) {
+ this(aliasQueries, Optional.empty(), Optional.empty(), child, registered,
+ subQueryAliasToUniqueId);
}
public LogicalCTE(List<LogicalSubQueryAlias<Plan>> aliasQueries, Optional<GroupExpression> groupExpression,
- Optional<LogicalProperties> logicalProperties, CHILD_TYPE child) {
+ Optional<LogicalProperties> logicalProperties, CHILD_TYPE child, boolean registered,
+ Map<String, Integer> subQueryAliasToUniqueId) {
super(PlanType.LOGICAL_CTE, groupExpression, logicalProperties, child);
this.aliasQueries = ImmutableList.copyOf(Objects.requireNonNull(aliasQueries, "aliasQueries can not be null"));
+ this.registered = registered;
+ this.subQueryAliasToUniqueId = subQueryAliasToUniqueId == null ? initUniqueId() : subQueryAliasToUniqueId;
Review Comment:
use `ImmutableMap.copyOf()`
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/visitor/DefaultPlanVisitor.java:
##########
@@ -25,9 +25,10 @@
public class DefaultPlanVisitor<R, C> extends PlanVisitor<R, C> {
@Override
public R visit(Plan plan, C context) {
+ R ret = null;
Review Comment:
change this return object is a bad idea. if return null is not what u want, u should override visit function
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/CollectProjectAboveConsumer.java:
##########
Review Comment:
this file should under `rules/rewrite/logical` dir
##########
fe/fe-core/src/main/java/org/apache/doris/planner/SelectNode.java:
##########
@@ -50,7 +50,7 @@ public SelectNode(PlanNodeId id, PlanNode child) {
this.nullableTupleIds = child.nullableTupleIds;
}
- protected SelectNode(PlanNodeId id, PlanNode child, List<Expr> conjuncts) {
+ public SelectNode(PlanNodeId id, PlanNode child, List<Expr> conjuncts) {
Review Comment:
if Nereids use this ctor, add comment on it to avoid change the behavior when we change legacy planner
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalCTE.java:
##########
@@ -40,14 +42,35 @@
private final List<LogicalSubQueryAlias<Plan>> aliasQueries;
+ private final Map<String, Integer> subQueryAliasToUniqueId;
+
+ private final boolean registered;
+
public LogicalCTE(List<LogicalSubQueryAlias<Plan>> aliasQueries, CHILD_TYPE child) {
- this(aliasQueries, Optional.empty(), Optional.empty(), child);
+ this(aliasQueries, Optional.empty(), Optional.empty(), child, false, null);
+ }
+
+ public LogicalCTE(List<LogicalSubQueryAlias<Plan>> aliasQueries, CHILD_TYPE child, boolean registered,
+ Map<String, Integer> subQueryAliasToUniqueId) {
+ this(aliasQueries, Optional.empty(), Optional.empty(), child, registered,
+ subQueryAliasToUniqueId);
}
public LogicalCTE(List<LogicalSubQueryAlias<Plan>> aliasQueries, Optional<GroupExpression> groupExpression,
- Optional<LogicalProperties> logicalProperties, CHILD_TYPE child) {
+ Optional<LogicalProperties> logicalProperties, CHILD_TYPE child, boolean registered,
+ Map<String, Integer> subQueryAliasToUniqueId) {
super(PlanType.LOGICAL_CTE, groupExpression, logicalProperties, child);
this.aliasQueries = ImmutableList.copyOf(Objects.requireNonNull(aliasQueries, "aliasQueries can not be null"));
+ this.registered = registered;
+ this.subQueryAliasToUniqueId = subQueryAliasToUniqueId == null ? initUniqueId() : subQueryAliasToUniqueId;
+ }
+
+ private Map<String, Integer> initUniqueId() {
Review Comment:
```suggestion
private Map<String, Integer> initCteId() {
```
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalCTEConsumer.java:
##########
@@ -0,0 +1,175 @@
+// 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.trees.plans.logical;
+
+import org.apache.doris.nereids.memo.GroupExpression;
+import org.apache.doris.nereids.properties.LogicalProperties;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.StatementScopeIdGenerator;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.PlanType;
+import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+/**
+ * LogicalCTEConsumer
+ */
+public class LogicalCTEConsumer extends LogicalLeaf {
+
+ private final LogicalPlan childPlan;
Review Comment:
why consumer has childPlan?
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalCTE.java:
##########
@@ -40,14 +42,35 @@
private final List<LogicalSubQueryAlias<Plan>> aliasQueries;
+ private final Map<String, Integer> subQueryAliasToUniqueId;
Review Comment:
```suggestion
private final Map<String, Integer> cteNameToId;
```
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/RegisterCTE.java:
##########
@@ -45,18 +46,22 @@ public class RegisterCTE extends OneAnalysisRuleFactory {
@Override
public Rule build() {
- return logicalCTE().thenApply(ctx -> {
+ return logicalCTE().whenNot(LogicalCTE::isRegistered).thenApply(ctx -> {
LogicalCTE<Plan> logicalCTE = ctx.root;
- register(logicalCTE.getAliasQueries(), ctx.cascadesContext);
- return logicalCTE.child();
+ List<LogicalSubQueryAlias<Plan>> analyzedCTE = register(logicalCTE, ctx.cascadesContext);
+ return new LogicalCTE<>(analyzedCTE, logicalCTE.child(), true,
+ logicalCTE.getSubQueryAliasToUniqueId());
}).toRule(RuleType.REGISTER_CTE);
}
/**
* register and store CTEs in CTEContext
*/
- private void register(List<LogicalSubQueryAlias<Plan>> aliasQueryList, CascadesContext cascadesContext) {
+ private List<LogicalSubQueryAlias<Plan>> register(LogicalCTE<Plan> logicalCTE,
+ CascadesContext cascadesContext) {
CTEContext cteCtx = cascadesContext.getCteContext();
+ List<LogicalSubQueryAlias<Plan>> aliasQueryList = logicalCTE.getAliasQueries();
Review Comment:
```suggestion
List<LogicalSubQueryAlias<Plan>> aliasQuerys = logicalCTE.getAliasQueries();
```
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/visitor/PlanVisitor.java:
##########
@@ -396,4 +415,21 @@ public R visitPhysicalDistribute(PhysicalDistribute<? extends Plan> distribute,
public R visitPhysicalAssertNumRows(PhysicalAssertNumRows<? extends Plan> assertNumRows, C context) {
return visit(assertNumRows, context);
}
+
+ public R visitLogicalCTEProducer(LogicalCTEProducer<? extends Plan> cteProducer, C context) {
+ return visit(cteProducer, context);
+ }
+
+ public R visitLogicalCTEConsumer(LogicalCTEConsumer cteConsumer, C context) {
+ return visit(cteConsumer, context);
+ }
+
+ public R visitLogicalCTEAnchor(LogicalCTEAnchor<? extends Plan, ? extends Plan> cteAnchor, C context) {
+ return visit(cteAnchor, context);
+ }
+
+ public R visitLogicalCTERelation(LogicalCTEConsumer cteRelation, C context) {
Review Comment:
remove this function
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalCTEProducer.java:
##########
@@ -0,0 +1,143 @@
+// 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.trees.plans.physical;
+
+import org.apache.doris.nereids.memo.GroupExpression;
+import org.apache.doris.nereids.properties.LogicalProperties;
+import org.apache.doris.nereids.properties.PhysicalProperties;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.PlanType;
+import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
+import org.apache.doris.statistics.Statistics;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+
+/**
+ * Physical CTE producer.
+ */
+public class PhysicalCTEProducer<CHILD_TYPE extends Plan> extends PhysicalUnary<CHILD_TYPE> {
+ private final int cteId;
+
+ private final List<Slot> projects;
+
+ public PhysicalCTEProducer(int cteId, List<Slot> projects,
+ LogicalProperties logicalProperties, CHILD_TYPE child) {
+ super(PlanType.PHYSICAL_CTE_PRODUCE, Optional.empty(), logicalProperties, child);
+ this.cteId = cteId;
+ this.projects = projects;
+ }
+
+ public PhysicalCTEProducer(int cteId, List<Slot> projects,
+ Optional<GroupExpression> groupExpression,
+ LogicalProperties logicalProperties, CHILD_TYPE child) {
+ super(PlanType.PHYSICAL_CTE_PRODUCE, groupExpression, logicalProperties, child);
+ this.cteId = cteId;
+ this.projects = projects;
+ }
+
+ public PhysicalCTEProducer(int cteId, List<Slot> projects, Optional<GroupExpression> groupExpression,
+ LogicalProperties logicalProperties, PhysicalProperties physicalProperties,
+ Statistics statistics, CHILD_TYPE child) {
+ super(PlanType.PHYSICAL_CTE_PRODUCE, groupExpression, logicalProperties, physicalProperties, statistics, child);
+ this.cteId = cteId;
+ this.projects = projects;
Review Comment:
ImmutableList.copyOf and null check
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalCTEProducer.java:
##########
@@ -0,0 +1,143 @@
+// 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.trees.plans.physical;
+
+import org.apache.doris.nereids.memo.GroupExpression;
+import org.apache.doris.nereids.properties.LogicalProperties;
+import org.apache.doris.nereids.properties.PhysicalProperties;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.PlanType;
+import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
+import org.apache.doris.statistics.Statistics;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+
+/**
+ * Physical CTE producer.
+ */
+public class PhysicalCTEProducer<CHILD_TYPE extends Plan> extends PhysicalUnary<CHILD_TYPE> {
+ private final int cteId;
+
+ private final List<Slot> projects;
+
+ public PhysicalCTEProducer(int cteId, List<Slot> projects,
+ LogicalProperties logicalProperties, CHILD_TYPE child) {
+ super(PlanType.PHYSICAL_CTE_PRODUCE, Optional.empty(), logicalProperties, child);
+ this.cteId = cteId;
+ this.projects = projects;
Review Comment:
this(cteId, projects, Optional.empty(), logicalProperties, child)
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalCTEConsumer.java:
##########
@@ -0,0 +1,175 @@
+// 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.trees.plans.logical;
+
+import org.apache.doris.nereids.memo.GroupExpression;
+import org.apache.doris.nereids.properties.LogicalProperties;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.StatementScopeIdGenerator;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.PlanType;
+import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+/**
+ * LogicalCTEConsumer
+ */
+public class LogicalCTEConsumer extends LogicalLeaf {
+
+ private final LogicalPlan childPlan;
+
+ private final int cteId;
+
+ private final Map<Slot, Slot> consumerToProducerOutputMap = new LinkedHashMap<>();
+
+ private final Map<Slot, Slot> producerToConsumerOutputMap = new LinkedHashMap<>();
+
+ private final int consumerId;
+
+ private final String name;
+
+ /**
+ * Logical CTE consumer.
+ */
+ public LogicalCTEConsumer(Optional<GroupExpression> groupExpression,
+ Optional<LogicalProperties> logicalProperties, LogicalPlan childPlan, int cteId, String name) {
+ super(PlanType.LOGICAL_CTE_RELATION, groupExpression, logicalProperties);
+ this.childPlan = childPlan;
+ this.cteId = cteId;
+ this.name = name;
+ initProducerToConsumerOutputMap(childPlan);
+ for (Map.Entry<Slot, Slot> entry : producerToConsumerOutputMap.entrySet()) {
+ this.consumerToProducerOutputMap.put(entry.getValue(), entry.getKey());
+ }
+ this.consumerId = StatementScopeIdGenerator.newExprId().asInt();
+ }
+
+ /**
+ * Logical CTE consumer.
+ */
+ public LogicalCTEConsumer(Optional<GroupExpression> groupExpression,
+ Optional<LogicalProperties> logicalProperties, LogicalPlan childPlan, int cteId,
+ Map<Slot, Slot> consumerToProducerOutputMap,
+ Map<Slot, Slot> producerToConsumerOutputMap, int consumerId, String name) {
+ super(PlanType.LOGICAL_CTE_RELATION, groupExpression, logicalProperties);
+ this.childPlan = childPlan;
+ this.cteId = cteId;
+ this.consumerToProducerOutputMap.putAll(consumerToProducerOutputMap);
+ this.producerToConsumerOutputMap.putAll(producerToConsumerOutputMap);
+ this.consumerId = consumerId;
+ this.name = name;
+ }
+
+ private void initProducerToConsumerOutputMap(LogicalPlan childPlan) {
+ List<Slot> producerOutput = childPlan.getOutput();
+ for (Slot producerOutputSlot : producerOutput) {
+ Slot consumerSlot = new SlotReference(producerOutputSlot.getName(),
+ producerOutputSlot.getDataType(), producerOutputSlot.nullable(), ImmutableList.of(name));
+ producerToConsumerOutputMap.put(producerOutputSlot, consumerSlot);
Review Comment:
i think there have a bug if producer's output has two slot with same exprid
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalCTEConsumer.java:
##########
@@ -0,0 +1,175 @@
+// 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.trees.plans.logical;
+
+import org.apache.doris.nereids.memo.GroupExpression;
+import org.apache.doris.nereids.properties.LogicalProperties;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.StatementScopeIdGenerator;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.PlanType;
+import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+/**
+ * LogicalCTEConsumer
+ */
+public class LogicalCTEConsumer extends LogicalLeaf {
+
+ private final LogicalPlan childPlan;
+
+ private final int cteId;
+
+ private final Map<Slot, Slot> consumerToProducerOutputMap = new LinkedHashMap<>();
+
+ private final Map<Slot, Slot> producerToConsumerOutputMap = new LinkedHashMap<>();
Review Comment:
do not use mutable container implementation
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalCTEConsumer.java:
##########
@@ -0,0 +1,175 @@
+// 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.trees.plans.logical;
+
+import org.apache.doris.nereids.memo.GroupExpression;
+import org.apache.doris.nereids.properties.LogicalProperties;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.StatementScopeIdGenerator;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.PlanType;
+import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+/**
+ * LogicalCTEConsumer
+ */
+public class LogicalCTEConsumer extends LogicalLeaf {
+
+ private final LogicalPlan childPlan;
+
+ private final int cteId;
+
+ private final Map<Slot, Slot> consumerToProducerOutputMap = new LinkedHashMap<>();
+
+ private final Map<Slot, Slot> producerToConsumerOutputMap = new LinkedHashMap<>();
+
+ private final int consumerId;
+
+ private final String name;
+
+ /**
+ * Logical CTE consumer.
+ */
+ public LogicalCTEConsumer(Optional<GroupExpression> groupExpression,
+ Optional<LogicalProperties> logicalProperties, LogicalPlan childPlan, int cteId, String name) {
+ super(PlanType.LOGICAL_CTE_RELATION, groupExpression, logicalProperties);
+ this.childPlan = childPlan;
+ this.cteId = cteId;
+ this.name = name;
+ initProducerToConsumerOutputMap(childPlan);
+ for (Map.Entry<Slot, Slot> entry : producerToConsumerOutputMap.entrySet()) {
+ this.consumerToProducerOutputMap.put(entry.getValue(), entry.getKey());
+ }
+ this.consumerId = StatementScopeIdGenerator.newExprId().asInt();
+ }
+
+ /**
+ * Logical CTE consumer.
+ */
+ public LogicalCTEConsumer(Optional<GroupExpression> groupExpression,
+ Optional<LogicalProperties> logicalProperties, LogicalPlan childPlan, int cteId,
+ Map<Slot, Slot> consumerToProducerOutputMap,
+ Map<Slot, Slot> producerToConsumerOutputMap, int consumerId, String name) {
+ super(PlanType.LOGICAL_CTE_RELATION, groupExpression, logicalProperties);
+ this.childPlan = childPlan;
+ this.cteId = cteId;
+ this.consumerToProducerOutputMap.putAll(consumerToProducerOutputMap);
+ this.producerToConsumerOutputMap.putAll(producerToConsumerOutputMap);
+ this.consumerId = consumerId;
+ this.name = name;
+ }
+
+ private void initProducerToConsumerOutputMap(LogicalPlan childPlan) {
+ List<Slot> producerOutput = childPlan.getOutput();
+ for (Slot producerOutputSlot : producerOutput) {
+ Slot consumerSlot = new SlotReference(producerOutputSlot.getName(),
+ producerOutputSlot.getDataType(), producerOutputSlot.nullable(), ImmutableList.of(name));
+ producerToConsumerOutputMap.put(producerOutputSlot, consumerSlot);
+ }
+ }
+
+ public Map<Slot, Slot> getConsumerToProducerOutputMap() {
+ return consumerToProducerOutputMap;
+ }
+
+ public Map<Slot, Slot> getProducerToConsumerOutputMap() {
+ return producerToConsumerOutputMap;
+ }
+
+ @Override
+ public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
+ return visitor.visitLogicalCTEConsumer(this, context);
+ }
+
+ @Override
+ public List<? extends Expression> getExpressions() {
+ return ImmutableList.of();
+ }
+
+ @Override
+ public Plan withGroupExpression(Optional<GroupExpression> groupExpression) {
+ return new LogicalCTEConsumer(groupExpression, Optional.of(getLogicalProperties()), childPlan, cteId,
+ consumerToProducerOutputMap,
+ producerToConsumerOutputMap,
+ consumerId, name);
+ }
+
+ @Override
+ public Plan withLogicalProperties(Optional<LogicalProperties> logicalProperties) {
+ return new LogicalCTEConsumer(groupExpression, logicalProperties, childPlan, cteId,
+ consumerToProducerOutputMap,
+ producerToConsumerOutputMap,
+ consumerId, name);
+ }
+
+ @Override
+ public List<Slot> computeOutput() {
+ return new ArrayList<>(producerToConsumerOutputMap.values());
Review Comment:
return ImmutableList
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/RewriteCTEProducer.java:
##########
@@ -0,0 +1,31 @@
+// 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.trees.plans.logical;
+
+import org.apache.doris.nereids.rules.Rule;
+import org.apache.doris.nereids.rules.rewrite.OneRewriteRuleFactory;
+
+/**
+ * CTE producer rewriter.
+ */
+public class RewriteCTEProducer extends OneRewriteRuleFactory {
Review Comment:
useless file
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalCTEProducer.java:
##########
@@ -0,0 +1,143 @@
+// 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.trees.plans.physical;
+
+import org.apache.doris.nereids.memo.GroupExpression;
+import org.apache.doris.nereids.properties.LogicalProperties;
+import org.apache.doris.nereids.properties.PhysicalProperties;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.PlanType;
+import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
+import org.apache.doris.statistics.Statistics;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+
+/**
+ * Physical CTE producer.
+ */
+public class PhysicalCTEProducer<CHILD_TYPE extends Plan> extends PhysicalUnary<CHILD_TYPE> {
+ private final int cteId;
+
+ private final List<Slot> projects;
+
+ public PhysicalCTEProducer(int cteId, List<Slot> projects,
+ LogicalProperties logicalProperties, CHILD_TYPE child) {
+ super(PlanType.PHYSICAL_CTE_PRODUCE, Optional.empty(), logicalProperties, child);
+ this.cteId = cteId;
+ this.projects = projects;
+ }
+
+ public PhysicalCTEProducer(int cteId, List<Slot> projects,
+ Optional<GroupExpression> groupExpression,
+ LogicalProperties logicalProperties, CHILD_TYPE child) {
+ super(PlanType.PHYSICAL_CTE_PRODUCE, groupExpression, logicalProperties, child);
+ this.cteId = cteId;
+ this.projects = projects;
+ }
+
+ public PhysicalCTEProducer(int cteId, List<Slot> projects, Optional<GroupExpression> groupExpression,
+ LogicalProperties logicalProperties, PhysicalProperties physicalProperties,
+ Statistics statistics, CHILD_TYPE child) {
+ super(PlanType.PHYSICAL_CTE_PRODUCE, groupExpression, logicalProperties, physicalProperties, statistics, child);
+ this.cteId = cteId;
+ this.projects = projects;
+ }
+
+ public int getCteId() {
+ return cteId;
+ }
+
+ public List<Slot> getProjects() {
+ return projects;
+ }
+
+ @Override
+ public List<Expression> getExpressions() {
+ return ImmutableList.of();
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+
+ if (!super.equals(o)) {
+ return false;
+ }
+
+ PhysicalCTEProducer that = (PhysicalCTEProducer) o;
+ return Objects.equals(cteId, that.cteId);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(super.hashCode(), cteId);
+ }
+
+ @Override
+ public String toString() {
+ return "PhysicalCTEProduce{"
Review Comment:
use Utils.toSqlString
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/CascadesContext.java:
##########
@@ -454,4 +484,98 @@ public void close() {
}
}
}
+
+ public void putCTEIdToCTEClosure(int cteId, Callable<LogicalPlan> cteClosure) {
+ this.cteIdToCTEClosure.put(cteId, cteClosure);
+ }
+
+ public void putCTEIdToCTEClosure(Map<Integer, Callable<LogicalPlan>> cteConsumers) {
+ this.cteIdToCTEClosure.putAll(cteConsumers);
+ }
+
+ public void putCTEIdToConsumer(LogicalCTEConsumer cteConsumer) {
+ Set<LogicalCTEConsumer> consumers =
+ this.cteIdToConsumer.computeIfAbsent(cteConsumer.getCteId(), k -> new HashSet<>());
+ consumers.add(cteConsumer);
+ }
+
+ public void putCTEIdToConsumer(Map<Integer, Set<LogicalCTEConsumer>> cteConsumers) {
+ this.cteIdToConsumer.putAll(cteConsumers);
+ }
+
+ public void putCTEIdToFilter(int cteId, Expression f) {
+ Set<Expression> filters = this.cteIdToFilter.computeIfAbsent(cteId, k -> new HashSet<>());
+ filters.add(f);
+ }
+
+ public void putCTEIdToProject(int cteId, Expression p) {
+ Set<Expression> projects = this.cteIdToProject.computeIfAbsent(cteId, k -> new HashSet<>());
+ projects.add(p);
+ }
+
+ public Set<Expression> findProjectForProducer(int cteId) {
+ return this.cteIdToProject.get(cteId);
+ }
+
+ /**
+ * Fork for rewritten child tree of CTEProducer.
+ */
+ public CascadesContext forkForCTEProducer(Plan plan) {
+ CascadesContext cascadesContext = new CascadesContext(plan, memo, statementContext, PhysicalProperties.ANY);
+ cascadesContext.cteIdToConsumer = cteIdToConsumer;
+ cascadesContext.cteIdToFilter = cteIdToFilter;
+ cascadesContext.cteIdToProject = cteIdToProject;
+ cascadesContext.cteContext = cteContext;
+ cascadesContext.cteIdToCTEClosure = cteIdToCTEClosure;
+ cascadesContext.consumerIdToFilter = consumerIdToFilter;
Review Comment:
do we need do deep copy?
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalCTEProducer.java:
##########
@@ -0,0 +1,143 @@
+// 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.trees.plans.physical;
+
+import org.apache.doris.nereids.memo.GroupExpression;
+import org.apache.doris.nereids.properties.LogicalProperties;
+import org.apache.doris.nereids.properties.PhysicalProperties;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.PlanType;
+import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
+import org.apache.doris.statistics.Statistics;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+
+/**
+ * Physical CTE producer.
+ */
+public class PhysicalCTEProducer<CHILD_TYPE extends Plan> extends PhysicalUnary<CHILD_TYPE> {
+ private final int cteId;
+
+ private final List<Slot> projects;
+
+ public PhysicalCTEProducer(int cteId, List<Slot> projects,
+ LogicalProperties logicalProperties, CHILD_TYPE child) {
+ super(PlanType.PHYSICAL_CTE_PRODUCE, Optional.empty(), logicalProperties, child);
+ this.cteId = cteId;
+ this.projects = projects;
+ }
+
+ public PhysicalCTEProducer(int cteId, List<Slot> projects,
+ Optional<GroupExpression> groupExpression,
+ LogicalProperties logicalProperties, CHILD_TYPE child) {
+ super(PlanType.PHYSICAL_CTE_PRODUCE, groupExpression, logicalProperties, child);
+ this.cteId = cteId;
+ this.projects = projects;
+ }
+
+ public PhysicalCTEProducer(int cteId, List<Slot> projects, Optional<GroupExpression> groupExpression,
+ LogicalProperties logicalProperties, PhysicalProperties physicalProperties,
+ Statistics statistics, CHILD_TYPE child) {
+ super(PlanType.PHYSICAL_CTE_PRODUCE, groupExpression, logicalProperties, physicalProperties, statistics, child);
+ this.cteId = cteId;
+ this.projects = projects;
+ }
+
+ public int getCteId() {
+ return cteId;
+ }
+
+ public List<Slot> getProjects() {
+ return projects;
+ }
+
+ @Override
+ public List<Expression> getExpressions() {
+ return ImmutableList.of();
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+
+ if (!super.equals(o)) {
+ return false;
+ }
+
+ PhysicalCTEProducer that = (PhysicalCTEProducer) o;
+ return Objects.equals(cteId, that.cteId);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(super.hashCode(), cteId);
+ }
+
+ @Override
+ public String toString() {
+ return "PhysicalCTEProduce{"
+ + "cteId='"
+ + cteId
+ + '\''
+ + '}';
+ }
+
+ @Override
+ public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
+ return visitor.visitPhysicalCTEProduce(this, context);
+ }
+
+ @Override
+ public PhysicalCTEProducer<Plan> withChildren(List<Plan> children) {
+ Preconditions.checkArgument(children.size() == 1);
+ return new PhysicalCTEProducer<>(cteId, projects, getLogicalProperties(), children.get(0));
+ }
+
+ @Override
+ public PhysicalCTEProducer<CHILD_TYPE> withGroupExpression(Optional<GroupExpression> groupExpression) {
+ return new PhysicalCTEProducer<>(cteId, projects, groupExpression, getLogicalProperties(), child());
+ }
+
+ @Override
+ public PhysicalCTEProducer<CHILD_TYPE> withLogicalProperties(Optional<LogicalProperties> logicalProperties) {
+ return new PhysicalCTEProducer<>(cteId, projects, Optional.empty(), logicalProperties.get(), child());
+ }
+
+ @Override
+ public PhysicalCTEProducer<CHILD_TYPE> withPhysicalPropertiesAndStats(PhysicalProperties physicalProperties,
+ Statistics statistics) {
Review Comment:
```suggestion
public PhysicalCTEProducer<CHILD_TYPE> withPhysicalPropertiesAndStats(
PhysicalProperties physicalProperties, Statistics statistics) {
```
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalCTEConsumer.java:
##########
@@ -0,0 +1,174 @@
+// 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.trees.plans.physical;
+
+import org.apache.doris.nereids.memo.GroupExpression;
+import org.apache.doris.nereids.properties.LogicalProperties;
+import org.apache.doris.nereids.properties.PhysicalProperties;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.PlanType;
+import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
+import org.apache.doris.statistics.Statistics;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+
+/**
+ * Physical CTE consumer.
+ */
+public class PhysicalCTEConsumer extends PhysicalLeaf {
+ private final int cteId;
+
+ private final Map<Slot, Slot> producerToConsumerSlotMap;
+
+ private final Map<Slot, Slot> consumerToProducerSlotMap;
+
+ /**
+ * Constructor
+ */
+ public PhysicalCTEConsumer(int cteId, Map<Slot, Slot> consumerToProducerSlotMap,
+ Map<Slot, Slot> producerToConsumerSlotMap,
+ LogicalProperties logicalProperties) {
+ super(PlanType.PHYSICAL_CTE_CONSUME, Optional.empty(), logicalProperties);
+ this.cteId = cteId;
+ this.consumerToProducerSlotMap = consumerToProducerSlotMap;
+ this.producerToConsumerSlotMap = producerToConsumerSlotMap;
Review Comment:
reuse the second ctor, `this(.....)`
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalCTEConsumer.java:
##########
@@ -0,0 +1,174 @@
+// 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.trees.plans.physical;
+
+import org.apache.doris.nereids.memo.GroupExpression;
+import org.apache.doris.nereids.properties.LogicalProperties;
+import org.apache.doris.nereids.properties.PhysicalProperties;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.PlanType;
+import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
+import org.apache.doris.statistics.Statistics;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+
+/**
+ * Physical CTE consumer.
+ */
+public class PhysicalCTEConsumer extends PhysicalLeaf {
+ private final int cteId;
+
+ private final Map<Slot, Slot> producerToConsumerSlotMap;
+
+ private final Map<Slot, Slot> consumerToProducerSlotMap;
+
+ /**
+ * Constructor
+ */
+ public PhysicalCTEConsumer(int cteId, Map<Slot, Slot> consumerToProducerSlotMap,
+ Map<Slot, Slot> producerToConsumerSlotMap,
+ LogicalProperties logicalProperties) {
+ super(PlanType.PHYSICAL_CTE_CONSUME, Optional.empty(), logicalProperties);
+ this.cteId = cteId;
+ this.consumerToProducerSlotMap = consumerToProducerSlotMap;
+ this.producerToConsumerSlotMap = producerToConsumerSlotMap;
+ }
+
+ /**
+ * Constructor
+ */
+ public PhysicalCTEConsumer(int cteId, Map<Slot, Slot> consumerToProducerSlotMap,
+ Map<Slot, Slot> producerToConsumerSlotMap,
+ Optional<GroupExpression> groupExpression,
+ LogicalProperties logicalProperties) {
+ super(PlanType.PHYSICAL_CTE_CONSUME, groupExpression, logicalProperties);
+ this.cteId = cteId;
+ this.consumerToProducerSlotMap = consumerToProducerSlotMap;
+ this.producerToConsumerSlotMap = producerToConsumerSlotMap;
+ }
+
+ /**
+ * Constructor
+ */
+ public PhysicalCTEConsumer(int cteId, Map<Slot, Slot> consumerToProducerSlotMap,
+ Map<Slot, Slot> producerToConsumerSlotMap,
+ Optional<GroupExpression> groupExpression,
+ LogicalProperties logicalProperties, PhysicalProperties physicalProperties,
+ Statistics statistics) {
+ super(PlanType.PHYSICAL_CTE_CONSUME, groupExpression, logicalProperties, physicalProperties, statistics);
+ this.cteId = cteId;
+ this.consumerToProducerSlotMap = consumerToProducerSlotMap;
+ this.producerToConsumerSlotMap = producerToConsumerSlotMap;
+ }
+
+ public int getCteId() {
+ return cteId;
+ }
+
+ public Map<Slot, Slot> getProducerToConsumerSlotMap() {
+ return producerToConsumerSlotMap;
+ }
+
+ @Override
+ public List<Expression> getExpressions() {
+ return Lists.newArrayList();
Review Comment:
return ImmutableList
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/logical/InlineCTE.java:
##########
@@ -0,0 +1,67 @@
+// 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.rules.rewrite.logical;
+
+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.Alias;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.plans.logical.LogicalCTEConsumer;
+import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.qe.ConnectContext;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * A CTEConsumer would be converted to a inlined plan if corresponding CTE referenced less than or
+ * equal inline_cte_referenced_threshold (it's a session variable, by default is 1).
+ */
+public class InlineCTE extends OneRewriteRuleFactory {
+
+ @Override
+ public Rule build() {
+ return logicalCTEConsumer().thenApply(ctx -> {
+ LogicalCTEConsumer cteConsumer = ctx.root;
+ int refCount = ctx.cascadesContext.cteReferencedCount(cteConsumer.getCteId());
+ if (ConnectContext.get().getSessionVariable().enablePipelineEngine
+ && refCount > ConnectContext.get().getSessionVariable().inlineCTEReferencedThreshold) {
+ return cteConsumer;
+ }
Review Comment:
add comment to explain this if statement
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java:
##########
@@ -1901,6 +1907,115 @@ public PlanFragment visitPhysicalGenerate(PhysicalGenerate<? extends Plan> gener
return currentFragment;
}
+ @Override
+ public PlanFragment visitPhysicalCTEConsume(PhysicalCTEConsumer consume,
+ PlanTranslatorContext context) {
+ int cteId = consume.getCteId();
+
+ MultiCastPlanFragment multCastFragment = (MultiCastPlanFragment) context.getCteProduceFragments().get(cteId);
+ Preconditions.checkState(multCastFragment.getSink() instanceof MultiCastDataSink,
+ "invalid multCastFragment");
+
+ MultiCastDataSink multiCastDataSink = (MultiCastDataSink) multCastFragment.getSink();
+ Preconditions.checkState(multiCastDataSink != null, "invalid multiCastDataSink");
+
+ PhysicalCTEProducer cteProducer = context.getCteProduceMap().get(cteId);
+ Preconditions.checkState(cteProducer != null, "invalid cteProducer");
+
+ ExchangeNode exchangeNode = new ExchangeNode(context.nextPlanNodeId(),
+ multCastFragment.getPlanRoot(), false);
+
+ DataStreamSink streamSink = new DataStreamSink(exchangeNode.getId());
+ streamSink.setPartition(DataPartition.RANDOM);
+ streamSink.setFragment(multCastFragment);
+
+ multiCastDataSink.getDataStreamSinks().add(streamSink);
+ multiCastDataSink.getDestinations().add(Lists.newArrayList());
+
+ exchangeNode.setNumInstances(multCastFragment.getPlanRoot().getNumInstances());
+
+ PlanFragment consumeFragment = new PlanFragment(context.nextFragmentId(), exchangeNode,
+ multCastFragment.getDataPartition());
+
+ Map<Slot, Slot> projectMap = Maps.newHashMap();
+ projectMap.putAll(consume.getProducerToConsumerSlotMap());
+
+ List<NamedExpression> execList = new ArrayList<>();
+ PlanNode inputPlanNode = consumeFragment.getPlanRoot();
+ List<Slot> cteProjects = cteProducer.getProjects();
+ for (Slot slot : cteProjects) {
+ if (projectMap.containsKey(slot)) {
+ execList.add(projectMap.get(slot));
+ } else {
+ throw new RuntimeException("could not find slot in cte producer consumer projectMap");
+ }
+ }
+
+ List<Slot> slotList = execList
+ .stream()
+ .map(e -> e.toSlot())
+ .collect(Collectors.toList());
+
+ TupleDescriptor tupleDescriptor = generateTupleDesc(slotList, null, context);
+
+ // update tuple list and tblTupleList
+ inputPlanNode.getTupleIds().clear();
+ inputPlanNode.getTupleIds().add(tupleDescriptor.getId());
+ inputPlanNode.getTblRefIds().clear();
+ inputPlanNode.getTblRefIds().add(tupleDescriptor.getId());
+ inputPlanNode.getNullableTupleIds().clear();
+ inputPlanNode.getNullableTupleIds().add(tupleDescriptor.getId());
+
+ List<Expr> execExprList = execList
+ .stream()
+ .map(e -> ExpressionTranslator.translate(e, context))
+ .collect(Collectors.toList());
+
+ inputPlanNode.setProjectList(execExprList);
+ inputPlanNode.setOutputTupleDesc(tupleDescriptor);
+
+ // update data partition
+ DataPartition dataPartition = new DataPartition(TPartitionType.HASH_PARTITIONED, execExprList);
+ consumeFragment.setDataPartition(dataPartition);
+
+ SelectNode projectNode = new SelectNode(context.nextPlanNodeId(), inputPlanNode);
+ consumeFragment.setPlanRoot(projectNode);
+
+ multCastFragment.getDestNodeList().add(exchangeNode);
+ consumeFragment.addChild(multCastFragment);
+ context.getPlanFragments().add(consumeFragment);
+
+ return consumeFragment;
+ }
+
+ @Override
+ public PlanFragment visitPhysicalCTEProduce(PhysicalCTEProducer<? extends Plan> produceOperator,
+ PlanTranslatorContext context) {
+ PlanFragment child = produceOperator.child().accept(this, context);
+ int cteId = produceOperator.getCteId();
+ context.getPlanFragments().remove(child);
+ MultiCastPlanFragment cteProduce = new MultiCastPlanFragment(child);
+ MultiCastDataSink multiCastDataSink = new MultiCastDataSink();
+ cteProduce.setSink(multiCastDataSink);
+
+ List<Expr> outputs = produceOperator.getProjects().stream()
+ .map(e -> ExpressionTranslator.translate(e, context))
+ .collect(Collectors.toList());
+
+ cteProduce.setOutputExprs(outputs);
+ context.getCteProduceFragments().put(cteId, cteProduce);
+ context.getCteProduceMap().put(cteId, produceOperator);
+ context.getPlanFragments().add(cteProduce);
+ return child;
+ }
+
+ @Override
+ public PlanFragment visitPhysicalCTEAnchor(PhysicalCTEAnchor<? extends Plan, ? extends Plan> anchorOperator,
+ PlanTranslatorContext context) {
+ anchorOperator.child(0).accept(this, context);
+ return anchorOperator.child(1).accept(this, context);
Review Comment:
add a NOTICE comment to explain we must translate producer first
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalCTEProducer.java:
##########
@@ -0,0 +1,143 @@
+// 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.trees.plans.physical;
+
+import org.apache.doris.nereids.memo.GroupExpression;
+import org.apache.doris.nereids.properties.LogicalProperties;
+import org.apache.doris.nereids.properties.PhysicalProperties;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.PlanType;
+import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
+import org.apache.doris.statistics.Statistics;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+
+/**
+ * Physical CTE producer.
+ */
+public class PhysicalCTEProducer<CHILD_TYPE extends Plan> extends PhysicalUnary<CHILD_TYPE> {
+ private final int cteId;
+
+ private final List<Slot> projects;
+
+ public PhysicalCTEProducer(int cteId, List<Slot> projects,
+ LogicalProperties logicalProperties, CHILD_TYPE child) {
+ super(PlanType.PHYSICAL_CTE_PRODUCE, Optional.empty(), logicalProperties, child);
+ this.cteId = cteId;
+ this.projects = projects;
+ }
+
+ public PhysicalCTEProducer(int cteId, List<Slot> projects,
+ Optional<GroupExpression> groupExpression,
+ LogicalProperties logicalProperties, CHILD_TYPE child) {
+ super(PlanType.PHYSICAL_CTE_PRODUCE, groupExpression, logicalProperties, child);
+ this.cteId = cteId;
+ this.projects = projects;
Review Comment:
ImmutableList.copyOf and null check
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalCTEConsumer.java:
##########
@@ -0,0 +1,174 @@
+// 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.trees.plans.physical;
+
+import org.apache.doris.nereids.memo.GroupExpression;
+import org.apache.doris.nereids.properties.LogicalProperties;
+import org.apache.doris.nereids.properties.PhysicalProperties;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.PlanType;
+import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
+import org.apache.doris.statistics.Statistics;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+
+/**
+ * Physical CTE consumer.
+ */
+public class PhysicalCTEConsumer extends PhysicalLeaf {
+ private final int cteId;
+
+ private final Map<Slot, Slot> producerToConsumerSlotMap;
+
+ private final Map<Slot, Slot> consumerToProducerSlotMap;
+
+ /**
+ * Constructor
+ */
+ public PhysicalCTEConsumer(int cteId, Map<Slot, Slot> consumerToProducerSlotMap,
+ Map<Slot, Slot> producerToConsumerSlotMap,
+ LogicalProperties logicalProperties) {
+ super(PlanType.PHYSICAL_CTE_CONSUME, Optional.empty(), logicalProperties);
+ this.cteId = cteId;
+ this.consumerToProducerSlotMap = consumerToProducerSlotMap;
+ this.producerToConsumerSlotMap = producerToConsumerSlotMap;
+ }
+
+ /**
+ * Constructor
+ */
+ public PhysicalCTEConsumer(int cteId, Map<Slot, Slot> consumerToProducerSlotMap,
+ Map<Slot, Slot> producerToConsumerSlotMap,
+ Optional<GroupExpression> groupExpression,
+ LogicalProperties logicalProperties) {
+ super(PlanType.PHYSICAL_CTE_CONSUME, groupExpression, logicalProperties);
+ this.cteId = cteId;
+ this.consumerToProducerSlotMap = consumerToProducerSlotMap;
+ this.producerToConsumerSlotMap = producerToConsumerSlotMap;
+ }
+
+ /**
+ * Constructor
+ */
+ public PhysicalCTEConsumer(int cteId, Map<Slot, Slot> consumerToProducerSlotMap,
+ Map<Slot, Slot> producerToConsumerSlotMap,
+ Optional<GroupExpression> groupExpression,
+ LogicalProperties logicalProperties, PhysicalProperties physicalProperties,
+ Statistics statistics) {
+ super(PlanType.PHYSICAL_CTE_CONSUME, groupExpression, logicalProperties, physicalProperties, statistics);
+ this.cteId = cteId;
+ this.consumerToProducerSlotMap = consumerToProducerSlotMap;
+ this.producerToConsumerSlotMap = producerToConsumerSlotMap;
+ }
+
+ public int getCteId() {
+ return cteId;
+ }
+
+ public Map<Slot, Slot> getProducerToConsumerSlotMap() {
+ return producerToConsumerSlotMap;
+ }
+
+ @Override
+ public List<Expression> getExpressions() {
+ return Lists.newArrayList();
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+
+ if (!super.equals(o)) {
+ return false;
+ }
+
+ PhysicalCTEConsumer that = (PhysicalCTEConsumer) o;
+ return Objects.equals(cteId, that.cteId)
+ && Objects.equals(producerToConsumerSlotMap, that.producerToConsumerSlotMap)
+ && Objects.equals(consumerToProducerSlotMap, that.consumerToProducerSlotMap);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(super.hashCode(), cteId, producerToConsumerSlotMap, consumerToProducerSlotMap);
+ }
+
+ @Override
+ public String toString() {
+ return "PhysicalCTEConsume{"
+ + "cteId='"
+ + cteId
+ + '\''
+ + '}';
+ }
+
+ @Override
+ public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
+ return visitor.visitPhysicalCTEConsume(this, context);
+ }
+
+ @Override
+ public PhysicalCTEConsumer withChildren(List<Plan> children) {
+ Preconditions.checkArgument(children.size() == 0);
Review Comment:
```suggestion
Preconditions.checkArgument(children.isEmpty());
```
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/CascadesContext.java:
##########
@@ -454,4 +484,98 @@ public void close() {
}
}
}
+
+ public void putCTEIdToCTEClosure(int cteId, Callable<LogicalPlan> cteClosure) {
+ this.cteIdToCTEClosure.put(cteId, cteClosure);
+ }
+
+ public void putCTEIdToCTEClosure(Map<Integer, Callable<LogicalPlan>> cteConsumers) {
+ this.cteIdToCTEClosure.putAll(cteConsumers);
+ }
+
+ public void putCTEIdToConsumer(LogicalCTEConsumer cteConsumer) {
+ Set<LogicalCTEConsumer> consumers =
+ this.cteIdToConsumer.computeIfAbsent(cteConsumer.getCteId(), k -> new HashSet<>());
+ consumers.add(cteConsumer);
+ }
+
+ public void putCTEIdToConsumer(Map<Integer, Set<LogicalCTEConsumer>> cteConsumers) {
+ this.cteIdToConsumer.putAll(cteConsumers);
+ }
+
+ public void putCTEIdToFilter(int cteId, Expression f) {
+ Set<Expression> filters = this.cteIdToFilter.computeIfAbsent(cteId, k -> new HashSet<>());
+ filters.add(f);
+ }
+
+ public void putCTEIdToProject(int cteId, Expression p) {
+ Set<Expression> projects = this.cteIdToProject.computeIfAbsent(cteId, k -> new HashSet<>());
+ projects.add(p);
+ }
+
+ public Set<Expression> findProjectForProducer(int cteId) {
+ return this.cteIdToProject.get(cteId);
+ }
+
+ /**
+ * Fork for rewritten child tree of CTEProducer.
+ */
+ public CascadesContext forkForCTEProducer(Plan plan) {
+ CascadesContext cascadesContext = new CascadesContext(plan, memo, statementContext, PhysicalProperties.ANY);
+ cascadesContext.cteIdToConsumer = cteIdToConsumer;
+ cascadesContext.cteIdToFilter = cteIdToFilter;
+ cascadesContext.cteIdToProject = cteIdToProject;
+ cascadesContext.cteContext = cteContext;
+ cascadesContext.cteIdToCTEClosure = cteIdToCTEClosure;
+ cascadesContext.consumerIdToFilter = consumerIdToFilter;
+ return cascadesContext;
+ }
+
+ public int cteReferencedCount(int cteId) {
+ Set<LogicalCTEConsumer> cteConsumer = cteIdToConsumer.get(cteId);
+ if (cteConsumer == null) {
+ return 0;
+ }
+ return cteIdToConsumer.get(cteId).size();
+ }
+
+ public Map<Integer, Set<LogicalCTEConsumer>> getCteIdToConsumer() {
+ return cteIdToConsumer;
+ }
+
+ public Map<Integer, Callable<LogicalPlan>> getCteIdToCTEClosure() {
+ return cteIdToCTEClosure;
+ }
+
+ public LogicalPlan findCTEPlanForInline(int cteId) {
+ try {
+ return cteIdToCTEClosure.get(cteId).call();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ public void putConsumerIdToFilter(int id, Expression filter) {
+ Set<Expression> filters = this.consumerIdToFilter.computeIfAbsent(id, k -> new HashSet<>());
+ filters.add(filter);
+ }
+
+ public Set<Expression> findFiltersForConsumer(int id) {
+ return this.consumerIdToFilter.get(id);
+ }
Review Comment:
useless function
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalCTEProducer.java:
##########
@@ -0,0 +1,143 @@
+// 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.trees.plans.physical;
+
+import org.apache.doris.nereids.memo.GroupExpression;
+import org.apache.doris.nereids.properties.LogicalProperties;
+import org.apache.doris.nereids.properties.PhysicalProperties;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.PlanType;
+import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
+import org.apache.doris.statistics.Statistics;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+
+/**
+ * Physical CTE producer.
+ */
+public class PhysicalCTEProducer<CHILD_TYPE extends Plan> extends PhysicalUnary<CHILD_TYPE> {
+ private final int cteId;
+
+ private final List<Slot> projects;
+
+ public PhysicalCTEProducer(int cteId, List<Slot> projects,
+ LogicalProperties logicalProperties, CHILD_TYPE child) {
+ super(PlanType.PHYSICAL_CTE_PRODUCE, Optional.empty(), logicalProperties, child);
+ this.cteId = cteId;
+ this.projects = projects;
+ }
+
+ public PhysicalCTEProducer(int cteId, List<Slot> projects,
+ Optional<GroupExpression> groupExpression,
+ LogicalProperties logicalProperties, CHILD_TYPE child) {
+ super(PlanType.PHYSICAL_CTE_PRODUCE, groupExpression, logicalProperties, child);
+ this.cteId = cteId;
+ this.projects = projects;
+ }
+
+ public PhysicalCTEProducer(int cteId, List<Slot> projects, Optional<GroupExpression> groupExpression,
+ LogicalProperties logicalProperties, PhysicalProperties physicalProperties,
+ Statistics statistics, CHILD_TYPE child) {
+ super(PlanType.PHYSICAL_CTE_PRODUCE, groupExpression, logicalProperties, physicalProperties, statistics, child);
+ this.cteId = cteId;
+ this.projects = projects;
+ }
+
+ public int getCteId() {
+ return cteId;
+ }
+
+ public List<Slot> getProjects() {
+ return projects;
+ }
+
+ @Override
+ public List<Expression> getExpressions() {
+ return ImmutableList.of();
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+
+ if (!super.equals(o)) {
+ return false;
+ }
+
+ PhysicalCTEProducer that = (PhysicalCTEProducer) o;
+ return Objects.equals(cteId, that.cteId);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(super.hashCode(), cteId);
+ }
+
+ @Override
+ public String toString() {
+ return "PhysicalCTEProduce{"
+ + "cteId='"
+ + cteId
+ + '\''
+ + '}';
+ }
+
+ @Override
+ public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
+ return visitor.visitPhysicalCTEProduce(this, context);
+ }
+
+ @Override
+ public PhysicalCTEProducer<Plan> withChildren(List<Plan> children) {
+ Preconditions.checkArgument(children.size() == 1);
+ return new PhysicalCTEProducer<>(cteId, projects, getLogicalProperties(), children.get(0));
+ }
+
+ @Override
+ public PhysicalCTEProducer<CHILD_TYPE> withGroupExpression(Optional<GroupExpression> groupExpression) {
+ return new PhysicalCTEProducer<>(cteId, projects, groupExpression, getLogicalProperties(), child());
+ }
+
+ @Override
+ public PhysicalCTEProducer<CHILD_TYPE> withLogicalProperties(Optional<LogicalProperties> logicalProperties) {
+ return new PhysicalCTEProducer<>(cteId, projects, Optional.empty(), logicalProperties.get(), child());
+ }
+
+ @Override
+ public PhysicalCTEProducer<CHILD_TYPE> withPhysicalPropertiesAndStats(PhysicalProperties physicalProperties,
+ Statistics statistics) {
+ return new PhysicalCTEProducer<>(cteId, projects, groupExpression, getLogicalProperties(), physicalProperties,
+ statistics, child());
+ }
+
+ @Override
+ public String shapeInfo() {
+ StringBuilder builder = new StringBuilder();
+ builder.append("cte producer(" + cteId + ")");
+ return builder.toString();
Review Comment:
```suggestion
return "CteProducer[cteId=" + cteId + "]";
```
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalCTEConsumer.java:
##########
@@ -0,0 +1,174 @@
+// 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.trees.plans.physical;
+
+import org.apache.doris.nereids.memo.GroupExpression;
+import org.apache.doris.nereids.properties.LogicalProperties;
+import org.apache.doris.nereids.properties.PhysicalProperties;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.PlanType;
+import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
+import org.apache.doris.statistics.Statistics;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+
+/**
+ * Physical CTE consumer.
+ */
+public class PhysicalCTEConsumer extends PhysicalLeaf {
+ private final int cteId;
+
+ private final Map<Slot, Slot> producerToConsumerSlotMap;
+
+ private final Map<Slot, Slot> consumerToProducerSlotMap;
Review Comment:
```suggestion
public class PhysicalCTEConsumer extends PhysicalLeaf {
private final int cteId;
private final Map<Slot, Slot> producerToConsumerSlotMap;
private final Map<Slot, Slot> consumerToProducerSlotMap;
```
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalSubQueryAlias.java:
##########
@@ -66,6 +69,16 @@ public LogicalSubQueryAlias(List<String> qualifier, Optional<List<String>> colum
super(PlanType.LOGICAL_SUBQUERY_ALIAS, groupExpression, logicalProperties, child);
this.qualifier = ImmutableList.copyOf(Objects.requireNonNull(qualifier));
this.columnAliases = columnAliases;
+ this.uniqueId = uniqueId();
+ }
+
+ public LogicalSubQueryAlias(List<String> qualifier, Optional<List<String>> columnAliases,
+ Optional<GroupExpression> groupExpression,
+ Optional<LogicalProperties> logicalProperties, CHILD_TYPE child, int uniqueId) {
+ super(PlanType.LOGICAL_SUBQUERY_ALIAS, groupExpression, logicalProperties, child);
+ this.qualifier = ImmutableList.copyOf(Objects.requireNonNull(qualifier));
Review Comment:
add check msg
```suggestion
this.qualifier = ImmutableList.copyOf(Objects.requireNonNull(qualifier, "xxxx"));
```
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/cascades/CostAndEnforcerJob.java:
##########
@@ -244,13 +245,15 @@ private boolean calculateEnforce(List<PhysicalProperties> requestChildrenPropert
// update current group statistics and re-compute costs.
if (groupExpression.children().stream().anyMatch(group -> group.getStatistics() == null)) {
+ // TODO: If it's error, add some warning log at least.
// if we come here, mean that we have some error in stats calculator and should fix it.
return false;
}
StatsCalculator statsCalculator = StatsCalculator.estimate(groupExpression,
context.getCascadesContext().getConnectContext().getSessionVariable().getForbidUnknownColStats(),
context.getCascadesContext().getConnectContext().getTotalColumnStatisticMap(),
- context.getCascadesContext().getConnectContext().getSessionVariable().isPlayNereidsDump());
+ context.getCascadesContext().getConnectContext().getSessionVariable().isPlayNereidsDump(),
+ new HashMap<>());
Review Comment:
add new estimate function with default empty map
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/logical/BuildCTEAnchorAndCTEProducer.java:
##########
@@ -0,0 +1,64 @@
+// 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.rules.rewrite.logical;
+
+import org.apache.doris.nereids.CascadesContext;
+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.plans.logical.LogicalCTE;
+import org.apache.doris.nereids.trees.plans.logical.LogicalCTEAnchor;
+import org.apache.doris.nereids.trees.plans.logical.LogicalCTEProducer;
+import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalSubQueryAlias;
+import org.apache.doris.qe.ConnectContext;
+
+/**
+ * BuildCTEAnchorAndCTEProducer.
+ */
+public class BuildCTEAnchorAndCTEProducer extends OneRewriteRuleFactory {
+
+ @Override
+ public Rule build() {
+ return logicalCTE().thenApply(ctx -> {
+ return rewrite(ctx.root, ctx.cascadesContext);
+ }).toRule(RuleType.BUILD_CTE_ANCHOR_AND_CTE_PRODUCER);
+ }
+
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ private LogicalPlan rewrite(LogicalPlan p, CascadesContext cascadesContext) {
+ if (!(p instanceof LogicalCTE)) {
+ return p;
+ }
+ LogicalCTE logicalCTE = (LogicalCTE) p;
+ LogicalPlan child = (LogicalPlan) logicalCTE.child();
+ for (int i = logicalCTE.getAliasQueries().size() - 1; i >= 0; i--) {
Review Comment:
add comment to explain why need traverse from rear to front
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/logical/BuildCTEAnchorAndCTEProducer.java:
##########
@@ -0,0 +1,64 @@
+// 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.rules.rewrite.logical;
+
+import org.apache.doris.nereids.CascadesContext;
+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.plans.logical.LogicalCTE;
+import org.apache.doris.nereids.trees.plans.logical.LogicalCTEAnchor;
+import org.apache.doris.nereids.trees.plans.logical.LogicalCTEProducer;
+import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalSubQueryAlias;
+import org.apache.doris.qe.ConnectContext;
+
+/**
+ * BuildCTEAnchorAndCTEProducer.
+ */
+public class BuildCTEAnchorAndCTEProducer extends OneRewriteRuleFactory {
+
+ @Override
+ public Rule build() {
+ return logicalCTE().thenApply(ctx -> {
+ return rewrite(ctx.root, ctx.cascadesContext);
+ }).toRule(RuleType.BUILD_CTE_ANCHOR_AND_CTE_PRODUCER);
+ }
+
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ private LogicalPlan rewrite(LogicalPlan p, CascadesContext cascadesContext) {
+ if (!(p instanceof LogicalCTE)) {
+ return p;
+ }
Review Comment:
remove this if
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/CollectProjectAboveConsumer.java:
##########
@@ -0,0 +1,79 @@
+// 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.trees.plans.logical;
+
+import org.apache.doris.nereids.pattern.MatchingContext;
+import org.apache.doris.nereids.rules.Rule;
+import org.apache.doris.nereids.rules.RuleType;
+import org.apache.doris.nereids.rules.rewrite.RewriteRuleFactory;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+
+import com.google.common.collect.ImmutableList;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * Collect Projects Above CTE Consumer.
+ */
+public class CollectProjectAboveConsumer implements RewriteRuleFactory {
+
+ @Override
+ public List<Rule> buildRules() {
+ return ImmutableList.of(RuleType.COLLECT_PROJECT_ABOVE_CONSUMER
+ .build(logicalProject(logicalCTEConsumer()).thenApply(ctx -> {
+ LogicalProject<LogicalCTEConsumer> project = ctx.root;
+ List<NamedExpression> namedExpressions = project.getProjects();
+ LogicalCTEConsumer cteConsumer = project.child();
+ collectProject(ctx, namedExpressions, cteConsumer);
+ return ctx.root;
+ })),
+ RuleType.COLLECT_PROJECT_ABOVE_FILTER_CONSUMER.build(logicalProject(logicalFilter(logicalCTEConsumer()))
+ .thenApply(ctx -> {
+ LogicalProject<LogicalFilter<LogicalCTEConsumer>> project = ctx.root;
+ LogicalFilter<LogicalCTEConsumer> filter = project.child();
+ Set<Slot> filterSlots = filter.getInputSlots();
+ List<NamedExpression> namedExpressions = new ArrayList<>(project.getProjects());
+ for (Slot slot : filterSlots) {
+ if (!project.getOutput().contains(slot)) {
+ namedExpressions.add(slot);
+ }
+ }
+ collectProject(ctx, namedExpressions, filter.child());
+ return ctx.root;
+ }))
+ );
+ }
+
+ private static void collectProject(MatchingContext ctx,
Review Comment:
```suggestion
private static void collectProject(CascadesContext ctx,
```
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalCTEConsumer.java:
##########
@@ -0,0 +1,174 @@
+// 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.trees.plans.physical;
+
+import org.apache.doris.nereids.memo.GroupExpression;
+import org.apache.doris.nereids.properties.LogicalProperties;
+import org.apache.doris.nereids.properties.PhysicalProperties;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.PlanType;
+import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
+import org.apache.doris.statistics.Statistics;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+
+/**
+ * Physical CTE consumer.
+ */
+public class PhysicalCTEConsumer extends PhysicalLeaf {
+ private final int cteId;
+
+ private final Map<Slot, Slot> producerToConsumerSlotMap;
+
+ private final Map<Slot, Slot> consumerToProducerSlotMap;
+
+ /**
+ * Constructor
+ */
+ public PhysicalCTEConsumer(int cteId, Map<Slot, Slot> consumerToProducerSlotMap,
+ Map<Slot, Slot> producerToConsumerSlotMap,
+ LogicalProperties logicalProperties) {
+ super(PlanType.PHYSICAL_CTE_CONSUME, Optional.empty(), logicalProperties);
+ this.cteId = cteId;
+ this.consumerToProducerSlotMap = consumerToProducerSlotMap;
Review Comment:
ImmutableMap.CopyOf() and null check
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/logical/CTEProducerRewrite.java:
##########
@@ -0,0 +1,113 @@
+// 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.rules.rewrite.logical;
+
+import org.apache.doris.nereids.CascadesContext;
+import org.apache.doris.nereids.jobs.batch.NereidsRewriter;
+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.Expression;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalCTEConsumer;
+import org.apache.doris.nereids.trees.plans.logical.LogicalCTEProducer;
+import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
+import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.util.ExpressionUtils;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import org.apache.commons.collections.CollectionUtils;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * Rewrite CTE Producer recursively.
+ */
+public class CTEProducerRewrite extends OneRewriteRuleFactory {
+
+ @Override
+ public Rule build() {
+ return logicalCTEProducer().when(p -> !p.isRewritten()).thenApply(ctx -> {
+ LogicalCTEProducer<? extends Plan> cteProducer = ctx.root;
+ Set<Expression> projects = ctx.cascadesContext.findProjectForProducer(cteProducer.getCteId());
+ LogicalPlan child = tryToConstructFilter(ctx.cascadesContext, cteProducer.getCteId(),
+ (LogicalPlan) ctx.root.child());
+ // LogicalPlan child = (LogicalPlan) ctx.root.child();
+ if (CollectionUtils.isNotEmpty(projects)
+ && ctx.cascadesContext.couldProjectCTEProducerBasedOnCTEConsumer(cteProducer.getCteId())) {
Review Comment:
```suggestion
&& ctx.cascadesContext.couldPruneColumnOnProducer(cteProducer.getCteId())) {
```
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalCTEConsumer.java:
##########
@@ -0,0 +1,174 @@
+// 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.trees.plans.physical;
+
+import org.apache.doris.nereids.memo.GroupExpression;
+import org.apache.doris.nereids.properties.LogicalProperties;
+import org.apache.doris.nereids.properties.PhysicalProperties;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.PlanType;
+import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
+import org.apache.doris.statistics.Statistics;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+
+/**
+ * Physical CTE consumer.
+ */
+public class PhysicalCTEConsumer extends PhysicalLeaf {
+ private final int cteId;
+
+ private final Map<Slot, Slot> producerToConsumerSlotMap;
+
+ private final Map<Slot, Slot> consumerToProducerSlotMap;
+
+ /**
+ * Constructor
+ */
+ public PhysicalCTEConsumer(int cteId, Map<Slot, Slot> consumerToProducerSlotMap,
+ Map<Slot, Slot> producerToConsumerSlotMap,
+ LogicalProperties logicalProperties) {
+ super(PlanType.PHYSICAL_CTE_CONSUME, Optional.empty(), logicalProperties);
+ this.cteId = cteId;
+ this.consumerToProducerSlotMap = consumerToProducerSlotMap;
+ this.producerToConsumerSlotMap = producerToConsumerSlotMap;
+ }
+
+ /**
+ * Constructor
+ */
+ public PhysicalCTEConsumer(int cteId, Map<Slot, Slot> consumerToProducerSlotMap,
+ Map<Slot, Slot> producerToConsumerSlotMap,
+ Optional<GroupExpression> groupExpression,
+ LogicalProperties logicalProperties) {
+ super(PlanType.PHYSICAL_CTE_CONSUME, groupExpression, logicalProperties);
+ this.cteId = cteId;
+ this.consumerToProducerSlotMap = consumerToProducerSlotMap;
+ this.producerToConsumerSlotMap = producerToConsumerSlotMap;
+ }
+
+ /**
+ * Constructor
+ */
+ public PhysicalCTEConsumer(int cteId, Map<Slot, Slot> consumerToProducerSlotMap,
+ Map<Slot, Slot> producerToConsumerSlotMap,
+ Optional<GroupExpression> groupExpression,
+ LogicalProperties logicalProperties, PhysicalProperties physicalProperties,
+ Statistics statistics) {
+ super(PlanType.PHYSICAL_CTE_CONSUME, groupExpression, logicalProperties, physicalProperties, statistics);
+ this.cteId = cteId;
+ this.consumerToProducerSlotMap = consumerToProducerSlotMap;
+ this.producerToConsumerSlotMap = producerToConsumerSlotMap;
+ }
+
+ public int getCteId() {
+ return cteId;
+ }
+
+ public Map<Slot, Slot> getProducerToConsumerSlotMap() {
+ return producerToConsumerSlotMap;
+ }
+
+ @Override
+ public List<Expression> getExpressions() {
+ return Lists.newArrayList();
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+
+ if (!super.equals(o)) {
+ return false;
+ }
+
+ PhysicalCTEConsumer that = (PhysicalCTEConsumer) o;
+ return Objects.equals(cteId, that.cteId)
+ && Objects.equals(producerToConsumerSlotMap, that.producerToConsumerSlotMap)
+ && Objects.equals(consumerToProducerSlotMap, that.consumerToProducerSlotMap);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(super.hashCode(), cteId, producerToConsumerSlotMap, consumerToProducerSlotMap);
+ }
+
+ @Override
+ public String toString() {
+ return "PhysicalCTEConsume{"
+ + "cteId='"
+ + cteId
+ + '\''
+ + '}';
+ }
+
+ @Override
+ public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
+ return visitor.visitPhysicalCTEConsume(this, context);
+ }
+
+ @Override
+ public PhysicalCTEConsumer withChildren(List<Plan> children) {
+ Preconditions.checkArgument(children.size() == 0);
+ return new PhysicalCTEConsumer(cteId, consumerToProducerSlotMap, producerToConsumerSlotMap,
+ getLogicalProperties());
+ }
+
+ @Override
+ public PhysicalCTEConsumer withGroupExpression(Optional<GroupExpression> groupExpression) {
+ return new PhysicalCTEConsumer(cteId, consumerToProducerSlotMap, producerToConsumerSlotMap,
+ groupExpression, getLogicalProperties());
+ }
+
+ @Override
+ public PhysicalCTEConsumer withLogicalProperties(Optional<LogicalProperties> logicalProperties) {
+ return new PhysicalCTEConsumer(cteId, consumerToProducerSlotMap, producerToConsumerSlotMap,
+ Optional.empty(), logicalProperties.get());
+ }
+
+ @Override
+ public PhysicalCTEConsumer withPhysicalPropertiesAndStats(PhysicalProperties physicalProperties,
+ Statistics statistics) {
Review Comment:
```suggestion
public PhysicalCTEConsumer withPhysicalPropertiesAndStats(
PhysicalProperties physicalProperties, Statistics statistics) {
```
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalCTEProducer.java:
##########
@@ -0,0 +1,143 @@
+// 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.trees.plans.physical;
+
+import org.apache.doris.nereids.memo.GroupExpression;
+import org.apache.doris.nereids.properties.LogicalProperties;
+import org.apache.doris.nereids.properties.PhysicalProperties;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.PlanType;
+import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
+import org.apache.doris.statistics.Statistics;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+
+/**
+ * Physical CTE producer.
+ */
+public class PhysicalCTEProducer<CHILD_TYPE extends Plan> extends PhysicalUnary<CHILD_TYPE> {
+ private final int cteId;
+
+ private final List<Slot> projects;
Review Comment:
```suggestion
public class PhysicalCTEProducer<CHILD_TYPE extends Plan> extends PhysicalUnary<CHILD_TYPE> {
private final int cteId;
private final List<Slot> projects;
```
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalCTEAnchor.java:
##########
@@ -0,0 +1,142 @@
+// 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.trees.plans.physical;
+
+import org.apache.doris.nereids.memo.GroupExpression;
+import org.apache.doris.nereids.properties.LogicalProperties;
+import org.apache.doris.nereids.properties.PhysicalProperties;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.PlanType;
+import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
+import org.apache.doris.statistics.Statistics;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+
+/**
+ * Physical CTE anchor.
+ */
+public class PhysicalCTEAnchor<
+ LEFT_CHILD_TYPE extends Plan,
+ RIGHT_CHILD_TYPE extends Plan>
+ extends PhysicalBinary<LEFT_CHILD_TYPE, RIGHT_CHILD_TYPE> {
+ private final int cteId;
+
+ public PhysicalCTEAnchor(int cteId, LogicalProperties logicalProperties,
+ LEFT_CHILD_TYPE leftChild,
+ RIGHT_CHILD_TYPE rightChild) {
+ super(PlanType.PHYSICAL_CTE_ANCHOR, Optional.empty(),
+ logicalProperties, leftChild, rightChild);
+ this.cteId = cteId;
+ }
+
+ public PhysicalCTEAnchor(int cteId, Optional<GroupExpression> groupExpression,
+ LogicalProperties logicalProperties,
+ LEFT_CHILD_TYPE leftChild,
+ RIGHT_CHILD_TYPE rightChild) {
+ super(PlanType.PHYSICAL_CTE_ANCHOR, groupExpression, logicalProperties, leftChild, rightChild);
+ this.cteId = cteId;
+ }
+
+ public PhysicalCTEAnchor(int cteId, Optional<GroupExpression> groupExpression,
+ LogicalProperties logicalProperties, PhysicalProperties physicalProperties,
+ Statistics statistics, LEFT_CHILD_TYPE leftChild,
+ RIGHT_CHILD_TYPE rightChild) {
+ super(PlanType.PHYSICAL_CTE_ANCHOR, groupExpression, logicalProperties, physicalProperties, statistics,
+ leftChild, rightChild);
+ this.cteId = cteId;
+ }
+
+ public int getCteId() {
+ return cteId;
+ }
+
+ @Override
+ public List<Expression> getExpressions() {
+ return ImmutableList.of();
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+
+ if (!super.equals(o)) {
+ return false;
+ }
+
+ PhysicalCTEAnchor that = (PhysicalCTEAnchor) o;
+ return Objects.equals(cteId, that.cteId);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(super.hashCode(), cteId);
+ }
+
+ @Override
+ public String toString() {
+ return "PhysicalCTEAnchor{"
+ + "cteId='"
+ + cteId
+ + '\''
+ + '}';
+ }
+
+ @Override
+ public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
+ return visitor.visitPhysicalCTEAnchor(this, context);
+ }
+
+ @Override
+ public PhysicalCTEAnchor<Plan, Plan> withChildren(List<Plan> children) {
+ Preconditions.checkArgument(children.size() == 2);
+ return new PhysicalCTEAnchor<>(cteId, getLogicalProperties(), children.get(0), children.get(1));
+ }
+
+ @Override
+ public PhysicalCTEAnchor<Plan, Plan> withGroupExpression(Optional<GroupExpression> groupExpression) {
+ return new PhysicalCTEAnchor<>(cteId, groupExpression, getLogicalProperties(), child(0), child(1));
+ }
+
+ @Override
+ public PhysicalCTEAnchor<Plan, Plan> withLogicalProperties(Optional<LogicalProperties> logicalProperties) {
+ return new PhysicalCTEAnchor<>(cteId,
+ Optional.empty(), logicalProperties.get(), child(0), child(1));
+ }
+
+ @Override
+ public PhysicalCTEAnchor<Plan, Plan> withPhysicalPropertiesAndStats(PhysicalProperties physicalProperties,
+ Statistics statistics) {
+ return new PhysicalCTEAnchor<>(cteId, groupExpression, getLogicalProperties(), physicalProperties,
+ statistics, child(0), child(1));
+ }
+
+ @Override
+ public String shapeInfo() {
+ StringBuilder builder = new StringBuilder();
+ builder.append("cte Anchor(" + cteId + ")");
+ return builder.toString();
Review Comment:
```suggestion
return "CteAnchor[cteId=" + cteId + "]";
```
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/CascadesContext.java:
##########
@@ -98,6 +101,18 @@ public class CascadesContext implements ScheduleContext, PlanSource {
private Optional<Scope> outerScope = Optional.empty();
+ private Map<Integer, Set<LogicalCTEConsumer>> cteIdToConsumer = new HashMap<>();
+
+ private Map<Integer, Callable<LogicalPlan>> cteIdToCTEClosure = new HashMap<>();
+
+ private Map<Integer, Set<Expression>> cteIdToFilter = new HashMap<>();
+
+ private Map<Integer, Set<Expression>> cteIdToProject = new HashMap<>();
+
+ private Map<Integer, Set<Expression>> consumerIdToFilter = new HashMap<>();
Review Comment:
```suggestion
private Map<Integer, Set<Expression>> consumerIdToFilters = new HashMap<>();
```
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/stats/StatsCalculator.java:
##########
@@ -127,12 +134,16 @@ public class StatsCalculator extends DefaultPlanVisitor<Statistics, Void> {
private Map<String, Histogram> totalHistogramMap = new HashMap<>();
+ private Map<Integer, Statistics> cteIdToStats;
+
private StatsCalculator(GroupExpression groupExpression, boolean forbidUnknownColStats,
- Map<String, ColumnStatistic> columnStatisticMap, boolean isPlayNereidsDump) {
+ Map<String, ColumnStatistic> columnStatisticMap, boolean isPlayNereidsDump,
+ Map<Integer, Statistics> cteIdToStats) {
this.groupExpression = groupExpression;
this.forbidUnknownColStats = forbidUnknownColStats;
this.totalColumnStatisticMap = columnStatisticMap;
this.isPlayNereidsDump = isPlayNereidsDump;
+ this.cteIdToStats = cteIdToStats;
Review Comment:
check null
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildOutputPropertyDeriver.java:
##########
@@ -86,6 +89,27 @@ public PhysicalProperties visit(Plan plan, PlanContext context) {
return PhysicalProperties.ANY;
}
+ @Override
+ public PhysicalProperties visitPhysicalCTEProduce(
+ PhysicalCTEProducer<? extends Plan> cteProducer, PlanContext context) {
+ Preconditions.checkState(childrenOutputProperties.size() == 1);
+ return childrenOutputProperties.get(0);
+ }
+
+ @Override
+ public PhysicalProperties visitPhysicalCTEConsume(
+ PhysicalCTEConsumer cteConsumeOperator, PlanContext context) {
+ Preconditions.checkState(childrenOutputProperties.size() == 0);
+ return PhysicalProperties.ANY;
+ }
+
+ @Override
+ public PhysicalProperties visitPhysicalCTEAnchor(
+ PhysicalCTEAnchor<? extends Plan, ? extends Plan> cteAnchorOperator, PlanContext context) {
+ Preconditions.checkState(childrenOutputProperties.size() == 2);
+ return childrenOutputProperties.get(1);
Review Comment:
add comment to explain why return right child's properties
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/CollectProjectAboveConsumer.java:
##########
@@ -0,0 +1,79 @@
+// 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.trees.plans.logical;
+
+import org.apache.doris.nereids.pattern.MatchingContext;
+import org.apache.doris.nereids.rules.Rule;
+import org.apache.doris.nereids.rules.RuleType;
+import org.apache.doris.nereids.rules.rewrite.RewriteRuleFactory;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+
+import com.google.common.collect.ImmutableList;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * Collect Projects Above CTE Consumer.
+ */
+public class CollectProjectAboveConsumer implements RewriteRuleFactory {
+
+ @Override
+ public List<Rule> buildRules() {
+ return ImmutableList.of(RuleType.COLLECT_PROJECT_ABOVE_CONSUMER
+ .build(logicalProject(logicalCTEConsumer()).thenApply(ctx -> {
+ LogicalProject<LogicalCTEConsumer> project = ctx.root;
+ List<NamedExpression> namedExpressions = project.getProjects();
+ LogicalCTEConsumer cteConsumer = project.child();
+ collectProject(ctx, namedExpressions, cteConsumer);
+ return ctx.root;
+ })),
+ RuleType.COLLECT_PROJECT_ABOVE_FILTER_CONSUMER.build(logicalProject(logicalFilter(logicalCTEConsumer()))
+ .thenApply(ctx -> {
+ LogicalProject<LogicalFilter<LogicalCTEConsumer>> project = ctx.root;
+ LogicalFilter<LogicalCTEConsumer> filter = project.child();
+ Set<Slot> filterSlots = filter.getInputSlots();
+ List<NamedExpression> namedExpressions = new ArrayList<>(project.getProjects());
+ for (Slot slot : filterSlots) {
+ if (!project.getOutput().contains(slot)) {
+ namedExpressions.add(slot);
+ }
+ }
+ collectProject(ctx, namedExpressions, filter.child());
+ return ctx.root;
+ }))
+ );
+ }
+
+ private static void collectProject(MatchingContext ctx,
+ List<NamedExpression> namedExpressions, LogicalCTEConsumer cteConsumer) {
+ for (Expression expr : namedExpressions) {
+ expr.foreach(node -> {
+ if (!(node instanceof Slot)) {
+ return;
+ }
+ Slot slot = cteConsumer.findProducerSlot((Slot) node);
+ ctx.cascadesContext.putCTEIdToProject(cteConsumer.getCteId(), slot);
+ ctx.cascadesContext.markConsumerUnderProject(cteConsumer);
Review Comment:
put some in ctx but not use, just remove this line
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/CascadesContext.java:
##########
@@ -98,6 +101,18 @@ public class CascadesContext implements ScheduleContext, PlanSource {
private Optional<Scope> outerScope = Optional.empty();
+ private Map<Integer, Set<LogicalCTEConsumer>> cteIdToConsumer = new HashMap<>();
+
+ private Map<Integer, Callable<LogicalPlan>> cteIdToCTEClosure = new HashMap<>();
+
+ private Map<Integer, Set<Expression>> cteIdToFilter = new HashMap<>();
+
+ private Map<Integer, Set<Expression>> cteIdToProject = new HashMap<>();
Review Comment:
cteIdToRetainedSlots
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/CascadesContext.java:
##########
@@ -98,6 +101,18 @@ public class CascadesContext implements ScheduleContext, PlanSource {
private Optional<Scope> outerScope = Optional.empty();
+ private Map<Integer, Set<LogicalCTEConsumer>> cteIdToConsumer = new HashMap<>();
+
+ private Map<Integer, Callable<LogicalPlan>> cteIdToCTEClosure = new HashMap<>();
+
+ private Map<Integer, Set<Expression>> cteIdToFilter = new HashMap<>();
+
+ private Map<Integer, Set<Expression>> cteIdToProject = new HashMap<>();
+
+ private Map<Integer, Set<Expression>> consumerIdToFilter = new HashMap<>();
+
+ private Map<Integer, Set<Integer>> cteIdToConsumerUnderProject = new HashMap<>();
Review Comment:
remove it
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/logical/InlineCTE.java:
##########
@@ -0,0 +1,67 @@
+// 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.rules.rewrite.logical;
+
+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.Alias;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.plans.logical.LogicalCTEConsumer;
+import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.qe.ConnectContext;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * A CTEConsumer would be converted to a inlined plan if corresponding CTE referenced less than or
+ * equal inline_cte_referenced_threshold (it's a session variable, by default is 1).
+ */
+public class InlineCTE extends OneRewriteRuleFactory {
+
+ @Override
+ public Rule build() {
+ return logicalCTEConsumer().thenApply(ctx -> {
+ LogicalCTEConsumer cteConsumer = ctx.root;
+ int refCount = ctx.cascadesContext.cteReferencedCount(cteConsumer.getCteId());
+ if (ConnectContext.get().getSessionVariable().enablePipelineEngine
+ && refCount > ConnectContext.get().getSessionVariable().inlineCTEReferencedThreshold) {
+ return cteConsumer;
+ }
+ LogicalPlan inlinedPlan = ctx.cascadesContext.findCTEPlanForInline(cteConsumer.getCteId());
+ List<Slot> inlinedPlanOutput = inlinedPlan.getOutput();
+ List<Slot> cteConsumerOutput = cteConsumer.getOutput();
+ List<NamedExpression> projects = new ArrayList<>();
+ for (Slot inlineSlot : inlinedPlanOutput) {
+ String name = inlineSlot.getName();
+ for (Slot consumerSlot : cteConsumerOutput) {
+ if (consumerSlot.getName().equals(name)) {
Review Comment:
use name is not good idea, if we have same name output in producer
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildOutputPropertyDeriver.java:
##########
@@ -86,6 +89,27 @@ public PhysicalProperties visit(Plan plan, PlanContext context) {
return PhysicalProperties.ANY;
}
+ @Override
+ public PhysicalProperties visitPhysicalCTEProduce(
+ PhysicalCTEProducer<? extends Plan> cteProducer, PlanContext context) {
+ Preconditions.checkState(childrenOutputProperties.size() == 1);
+ return childrenOutputProperties.get(0);
+ }
+
+ @Override
+ public PhysicalProperties visitPhysicalCTEConsume(
Review Comment:
```suggestion
public PhysicalProperties visitPhysicalCTEConsumer(
```
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildOutputPropertyDeriver.java:
##########
@@ -86,6 +89,27 @@ public PhysicalProperties visit(Plan plan, PlanContext context) {
return PhysicalProperties.ANY;
}
+ @Override
+ public PhysicalProperties visitPhysicalCTEProduce(
Review Comment:
```suggestion
public PhysicalProperties visitPhysicalCTEProducer(
```
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalCTEConsumer.java:
##########
@@ -0,0 +1,174 @@
+// 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.trees.plans.physical;
+
+import org.apache.doris.nereids.memo.GroupExpression;
+import org.apache.doris.nereids.properties.LogicalProperties;
+import org.apache.doris.nereids.properties.PhysicalProperties;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.PlanType;
+import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
+import org.apache.doris.statistics.Statistics;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+
+/**
+ * Physical CTE consumer.
+ */
+public class PhysicalCTEConsumer extends PhysicalLeaf {
+ private final int cteId;
+
+ private final Map<Slot, Slot> producerToConsumerSlotMap;
+
+ private final Map<Slot, Slot> consumerToProducerSlotMap;
+
+ /**
+ * Constructor
+ */
+ public PhysicalCTEConsumer(int cteId, Map<Slot, Slot> consumerToProducerSlotMap,
+ Map<Slot, Slot> producerToConsumerSlotMap,
+ LogicalProperties logicalProperties) {
+ super(PlanType.PHYSICAL_CTE_CONSUME, Optional.empty(), logicalProperties);
+ this.cteId = cteId;
+ this.consumerToProducerSlotMap = consumerToProducerSlotMap;
+ this.producerToConsumerSlotMap = producerToConsumerSlotMap;
+ }
+
+ /**
+ * Constructor
+ */
+ public PhysicalCTEConsumer(int cteId, Map<Slot, Slot> consumerToProducerSlotMap,
+ Map<Slot, Slot> producerToConsumerSlotMap,
+ Optional<GroupExpression> groupExpression,
+ LogicalProperties logicalProperties) {
+ super(PlanType.PHYSICAL_CTE_CONSUME, groupExpression, logicalProperties);
+ this.cteId = cteId;
+ this.consumerToProducerSlotMap = consumerToProducerSlotMap;
+ this.producerToConsumerSlotMap = producerToConsumerSlotMap;
+ }
+
+ /**
+ * Constructor
+ */
+ public PhysicalCTEConsumer(int cteId, Map<Slot, Slot> consumerToProducerSlotMap,
+ Map<Slot, Slot> producerToConsumerSlotMap,
+ Optional<GroupExpression> groupExpression,
+ LogicalProperties logicalProperties, PhysicalProperties physicalProperties,
+ Statistics statistics) {
+ super(PlanType.PHYSICAL_CTE_CONSUME, groupExpression, logicalProperties, physicalProperties, statistics);
+ this.cteId = cteId;
+ this.consumerToProducerSlotMap = consumerToProducerSlotMap;
+ this.producerToConsumerSlotMap = producerToConsumerSlotMap;
+ }
+
+ public int getCteId() {
+ return cteId;
+ }
+
+ public Map<Slot, Slot> getProducerToConsumerSlotMap() {
+ return producerToConsumerSlotMap;
+ }
+
+ @Override
+ public List<Expression> getExpressions() {
+ return Lists.newArrayList();
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+
+ if (!super.equals(o)) {
+ return false;
+ }
+
+ PhysicalCTEConsumer that = (PhysicalCTEConsumer) o;
+ return Objects.equals(cteId, that.cteId)
+ && Objects.equals(producerToConsumerSlotMap, that.producerToConsumerSlotMap)
+ && Objects.equals(consumerToProducerSlotMap, that.consumerToProducerSlotMap);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(super.hashCode(), cteId, producerToConsumerSlotMap, consumerToProducerSlotMap);
+ }
+
+ @Override
+ public String toString() {
+ return "PhysicalCTEConsume{"
+ + "cteId='"
+ + cteId
+ + '\''
+ + '}';
+ }
+
+ @Override
+ public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
+ return visitor.visitPhysicalCTEConsume(this, context);
+ }
+
+ @Override
+ public PhysicalCTEConsumer withChildren(List<Plan> children) {
+ Preconditions.checkArgument(children.size() == 0);
+ return new PhysicalCTEConsumer(cteId, consumerToProducerSlotMap, producerToConsumerSlotMap,
+ getLogicalProperties());
+ }
+
+ @Override
+ public PhysicalCTEConsumer withGroupExpression(Optional<GroupExpression> groupExpression) {
+ return new PhysicalCTEConsumer(cteId, consumerToProducerSlotMap, producerToConsumerSlotMap,
+ groupExpression, getLogicalProperties());
+ }
+
+ @Override
+ public PhysicalCTEConsumer withLogicalProperties(Optional<LogicalProperties> logicalProperties) {
+ return new PhysicalCTEConsumer(cteId, consumerToProducerSlotMap, producerToConsumerSlotMap,
+ Optional.empty(), logicalProperties.get());
+ }
+
+ @Override
+ public PhysicalCTEConsumer withPhysicalPropertiesAndStats(PhysicalProperties physicalProperties,
+ Statistics statistics) {
+ return new PhysicalCTEConsumer(cteId, consumerToProducerSlotMap, producerToConsumerSlotMap,
+ groupExpression, getLogicalProperties(), physicalProperties, statistics);
+ }
+
+ @Override
+ public String shapeInfo() {
+ StringBuilder builder = new StringBuilder();
+ builder.append("cte consumer(" + cteId + ")");
+ return builder.toString();
Review Comment:
```suggestion
return "CteConsumer[cteId=" + cteId + "]";
```
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/CascadesContext.java:
##########
@@ -138,6 +153,21 @@ public static CascadesContext newRewriteContext(StatementContext statementContex
return new CascadesContext(initPlan, null, statementContext, cteContext, PhysicalProperties.ANY);
}
+ /**
+ * New rewrite context.
+ */
+ public static CascadesContext newRewriteContext(CascadesContext context, Plan plan) {
+ CascadesContext cascadesContext = CascadesContext.newRewriteContext(
+ context.getStatementContext(), plan, context.getCteContext());
+ cascadesContext.cteIdToConsumer = context.cteIdToConsumer;
+ cascadesContext.cteIdToFilter = context.cteIdToFilter;
+ cascadesContext.cteIdToProject = context.cteIdToProject;
+ cascadesContext.cteContext = context.cteContext;
+ cascadesContext.cteIdToCTEClosure = context.cteIdToCTEClosure;
+ cascadesContext.consumerIdToFilter = context.consumerIdToFilter;
Review Comment:
do we need to do deepCopy?
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/logical/BuildCTEAnchorAndCTEProducer.java:
##########
@@ -0,0 +1,64 @@
+// 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.rules.rewrite.logical;
+
+import org.apache.doris.nereids.CascadesContext;
+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.plans.logical.LogicalCTE;
+import org.apache.doris.nereids.trees.plans.logical.LogicalCTEAnchor;
+import org.apache.doris.nereids.trees.plans.logical.LogicalCTEProducer;
+import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalSubQueryAlias;
+import org.apache.doris.qe.ConnectContext;
+
+/**
+ * BuildCTEAnchorAndCTEProducer.
+ */
+public class BuildCTEAnchorAndCTEProducer extends OneRewriteRuleFactory {
+
+ @Override
+ public Rule build() {
+ return logicalCTE().thenApply(ctx -> {
+ return rewrite(ctx.root, ctx.cascadesContext);
+ }).toRule(RuleType.BUILD_CTE_ANCHOR_AND_CTE_PRODUCER);
+ }
+
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ private LogicalPlan rewrite(LogicalPlan p, CascadesContext cascadesContext) {
+ if (!(p instanceof LogicalCTE)) {
+ return p;
+ }
+ LogicalCTE logicalCTE = (LogicalCTE) p;
+ LogicalPlan child = (LogicalPlan) logicalCTE.child();
Review Comment:
```suggestion
Plan child = logicalCTE.child();
```
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/CascadesContext.java:
##########
@@ -454,4 +484,98 @@ public void close() {
}
}
}
+
+ public void putCTEIdToCTEClosure(int cteId, Callable<LogicalPlan> cteClosure) {
+ this.cteIdToCTEClosure.put(cteId, cteClosure);
+ }
+
+ public void putCTEIdToCTEClosure(Map<Integer, Callable<LogicalPlan>> cteConsumers) {
+ this.cteIdToCTEClosure.putAll(cteConsumers);
+ }
+
+ public void putCTEIdToConsumer(LogicalCTEConsumer cteConsumer) {
+ Set<LogicalCTEConsumer> consumers =
+ this.cteIdToConsumer.computeIfAbsent(cteConsumer.getCteId(), k -> new HashSet<>());
+ consumers.add(cteConsumer);
+ }
+
+ public void putCTEIdToConsumer(Map<Integer, Set<LogicalCTEConsumer>> cteConsumers) {
+ this.cteIdToConsumer.putAll(cteConsumers);
+ }
+
+ public void putCTEIdToFilter(int cteId, Expression f) {
+ Set<Expression> filters = this.cteIdToFilter.computeIfAbsent(cteId, k -> new HashSet<>());
+ filters.add(f);
+ }
+
+ public void putCTEIdToProject(int cteId, Expression p) {
+ Set<Expression> projects = this.cteIdToProject.computeIfAbsent(cteId, k -> new HashSet<>());
+ projects.add(p);
+ }
+
+ public Set<Expression> findProjectForProducer(int cteId) {
+ return this.cteIdToProject.get(cteId);
+ }
+
+ /**
+ * Fork for rewritten child tree of CTEProducer.
+ */
+ public CascadesContext forkForCTEProducer(Plan plan) {
+ CascadesContext cascadesContext = new CascadesContext(plan, memo, statementContext, PhysicalProperties.ANY);
+ cascadesContext.cteIdToConsumer = cteIdToConsumer;
+ cascadesContext.cteIdToFilter = cteIdToFilter;
+ cascadesContext.cteIdToProject = cteIdToProject;
+ cascadesContext.cteContext = cteContext;
+ cascadesContext.cteIdToCTEClosure = cteIdToCTEClosure;
+ cascadesContext.consumerIdToFilter = consumerIdToFilter;
+ return cascadesContext;
+ }
+
+ public int cteReferencedCount(int cteId) {
+ Set<LogicalCTEConsumer> cteConsumer = cteIdToConsumer.get(cteId);
+ if (cteConsumer == null) {
+ return 0;
+ }
+ return cteIdToConsumer.get(cteId).size();
+ }
+
+ public Map<Integer, Set<LogicalCTEConsumer>> getCteIdToConsumer() {
+ return cteIdToConsumer;
+ }
+
+ public Map<Integer, Callable<LogicalPlan>> getCteIdToCTEClosure() {
+ return cteIdToCTEClosure;
+ }
+
+ public LogicalPlan findCTEPlanForInline(int cteId) {
+ try {
+ return cteIdToCTEClosure.get(cteId).call();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ public void putConsumerIdToFilter(int id, Expression filter) {
+ Set<Expression> filters = this.consumerIdToFilter.computeIfAbsent(id, k -> new HashSet<>());
+ filters.add(filter);
+ }
+
+ public Set<Expression> findFiltersForConsumer(int id) {
+ return this.consumerIdToFilter.get(id);
+ }
+
+ public Map<Integer, Set<Expression>> getConsumerIdToFilter() {
+ return consumerIdToFilter;
+ }
+
+ public void markConsumerUnderProject(LogicalCTEConsumer cteConsumer) {
+ Set<Integer> consumerIds =
+ this.cteIdToConsumerUnderProject.computeIfAbsent(cteConsumer.getCteId(), k -> new HashSet<>());
+ consumerIds.add(cteConsumer.getConsumerId());
+ }
Review Comment:
useless function
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalCTEConsumer.java:
##########
@@ -0,0 +1,174 @@
+// 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.trees.plans.physical;
+
+import org.apache.doris.nereids.memo.GroupExpression;
+import org.apache.doris.nereids.properties.LogicalProperties;
+import org.apache.doris.nereids.properties.PhysicalProperties;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.PlanType;
+import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
+import org.apache.doris.statistics.Statistics;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+
+/**
+ * Physical CTE consumer.
+ */
+public class PhysicalCTEConsumer extends PhysicalLeaf {
+ private final int cteId;
+
+ private final Map<Slot, Slot> producerToConsumerSlotMap;
+
+ private final Map<Slot, Slot> consumerToProducerSlotMap;
+
+ /**
+ * Constructor
+ */
+ public PhysicalCTEConsumer(int cteId, Map<Slot, Slot> consumerToProducerSlotMap,
+ Map<Slot, Slot> producerToConsumerSlotMap,
+ LogicalProperties logicalProperties) {
+ super(PlanType.PHYSICAL_CTE_CONSUME, Optional.empty(), logicalProperties);
+ this.cteId = cteId;
+ this.consumerToProducerSlotMap = consumerToProducerSlotMap;
+ this.producerToConsumerSlotMap = producerToConsumerSlotMap;
+ }
+
+ /**
+ * Constructor
+ */
+ public PhysicalCTEConsumer(int cteId, Map<Slot, Slot> consumerToProducerSlotMap,
+ Map<Slot, Slot> producerToConsumerSlotMap,
+ Optional<GroupExpression> groupExpression,
+ LogicalProperties logicalProperties) {
+ super(PlanType.PHYSICAL_CTE_CONSUME, groupExpression, logicalProperties);
+ this.cteId = cteId;
+ this.consumerToProducerSlotMap = consumerToProducerSlotMap;
+ this.producerToConsumerSlotMap = producerToConsumerSlotMap;
+ }
+
+ /**
+ * Constructor
+ */
+ public PhysicalCTEConsumer(int cteId, Map<Slot, Slot> consumerToProducerSlotMap,
+ Map<Slot, Slot> producerToConsumerSlotMap,
+ Optional<GroupExpression> groupExpression,
+ LogicalProperties logicalProperties, PhysicalProperties physicalProperties,
+ Statistics statistics) {
+ super(PlanType.PHYSICAL_CTE_CONSUME, groupExpression, logicalProperties, physicalProperties, statistics);
+ this.cteId = cteId;
+ this.consumerToProducerSlotMap = consumerToProducerSlotMap;
+ this.producerToConsumerSlotMap = producerToConsumerSlotMap;
+ }
+
+ public int getCteId() {
+ return cteId;
+ }
+
+ public Map<Slot, Slot> getProducerToConsumerSlotMap() {
+ return producerToConsumerSlotMap;
+ }
+
+ @Override
+ public List<Expression> getExpressions() {
+ return Lists.newArrayList();
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+
+ if (!super.equals(o)) {
+ return false;
+ }
+
+ PhysicalCTEConsumer that = (PhysicalCTEConsumer) o;
+ return Objects.equals(cteId, that.cteId)
+ && Objects.equals(producerToConsumerSlotMap, that.producerToConsumerSlotMap)
+ && Objects.equals(consumerToProducerSlotMap, that.consumerToProducerSlotMap);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(super.hashCode(), cteId, producerToConsumerSlotMap, consumerToProducerSlotMap);
+ }
+
+ @Override
+ public String toString() {
+ return "PhysicalCTEConsume{"
Review Comment:
use `Utils.toSqlString`
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalCTEAnchor.java:
##########
@@ -0,0 +1,142 @@
+// 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.trees.plans.physical;
+
+import org.apache.doris.nereids.memo.GroupExpression;
+import org.apache.doris.nereids.properties.LogicalProperties;
+import org.apache.doris.nereids.properties.PhysicalProperties;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.PlanType;
+import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
+import org.apache.doris.statistics.Statistics;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+
+/**
+ * Physical CTE anchor.
+ */
+public class PhysicalCTEAnchor<
+ LEFT_CHILD_TYPE extends Plan,
+ RIGHT_CHILD_TYPE extends Plan>
+ extends PhysicalBinary<LEFT_CHILD_TYPE, RIGHT_CHILD_TYPE> {
+ private final int cteId;
+
+ public PhysicalCTEAnchor(int cteId, LogicalProperties logicalProperties,
+ LEFT_CHILD_TYPE leftChild,
+ RIGHT_CHILD_TYPE rightChild) {
+ super(PlanType.PHYSICAL_CTE_ANCHOR, Optional.empty(),
+ logicalProperties, leftChild, rightChild);
+ this.cteId = cteId;
+ }
+
+ public PhysicalCTEAnchor(int cteId, Optional<GroupExpression> groupExpression,
+ LogicalProperties logicalProperties,
+ LEFT_CHILD_TYPE leftChild,
+ RIGHT_CHILD_TYPE rightChild) {
+ super(PlanType.PHYSICAL_CTE_ANCHOR, groupExpression, logicalProperties, leftChild, rightChild);
+ this.cteId = cteId;
+ }
+
+ public PhysicalCTEAnchor(int cteId, Optional<GroupExpression> groupExpression,
+ LogicalProperties logicalProperties, PhysicalProperties physicalProperties,
+ Statistics statistics, LEFT_CHILD_TYPE leftChild,
+ RIGHT_CHILD_TYPE rightChild) {
+ super(PlanType.PHYSICAL_CTE_ANCHOR, groupExpression, logicalProperties, physicalProperties, statistics,
+ leftChild, rightChild);
+ this.cteId = cteId;
+ }
+
+ public int getCteId() {
+ return cteId;
+ }
+
+ @Override
+ public List<Expression> getExpressions() {
+ return ImmutableList.of();
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+
+ if (!super.equals(o)) {
+ return false;
+ }
+
+ PhysicalCTEAnchor that = (PhysicalCTEAnchor) o;
+ return Objects.equals(cteId, that.cteId);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(super.hashCode(), cteId);
+ }
+
+ @Override
+ public String toString() {
+ return "PhysicalCTEAnchor{"
+ + "cteId='"
+ + cteId
+ + '\''
+ + '}';
+ }
+
+ @Override
+ public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
+ return visitor.visitPhysicalCTEAnchor(this, context);
+ }
+
+ @Override
+ public PhysicalCTEAnchor<Plan, Plan> withChildren(List<Plan> children) {
+ Preconditions.checkArgument(children.size() == 2);
+ return new PhysicalCTEAnchor<>(cteId, getLogicalProperties(), children.get(0), children.get(1));
+ }
+
+ @Override
+ public PhysicalCTEAnchor<Plan, Plan> withGroupExpression(Optional<GroupExpression> groupExpression) {
+ return new PhysicalCTEAnchor<>(cteId, groupExpression, getLogicalProperties(), child(0), child(1));
+ }
+
+ @Override
+ public PhysicalCTEAnchor<Plan, Plan> withLogicalProperties(Optional<LogicalProperties> logicalProperties) {
+ return new PhysicalCTEAnchor<>(cteId,
+ Optional.empty(), logicalProperties.get(), child(0), child(1));
+ }
+
+ @Override
+ public PhysicalCTEAnchor<Plan, Plan> withPhysicalPropertiesAndStats(PhysicalProperties physicalProperties,
+ Statistics statistics) {
Review Comment:
```suggestion
public PhysicalCTEAnchor<Plan, Plan> withPhysicalPropertiesAndStats(
PhysicalProperties physicalProperties, Statistics statistics) {
```
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/logical/BuildCTEAnchorAndCTEProducer.java:
##########
@@ -0,0 +1,64 @@
+// 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.rules.rewrite.logical;
+
+import org.apache.doris.nereids.CascadesContext;
+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.plans.logical.LogicalCTE;
+import org.apache.doris.nereids.trees.plans.logical.LogicalCTEAnchor;
+import org.apache.doris.nereids.trees.plans.logical.LogicalCTEProducer;
+import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalSubQueryAlias;
+import org.apache.doris.qe.ConnectContext;
+
+/**
+ * BuildCTEAnchorAndCTEProducer.
+ */
+public class BuildCTEAnchorAndCTEProducer extends OneRewriteRuleFactory {
+
+ @Override
+ public Rule build() {
+ return logicalCTE().thenApply(ctx -> {
+ return rewrite(ctx.root, ctx.cascadesContext);
+ }).toRule(RuleType.BUILD_CTE_ANCHOR_AND_CTE_PRODUCER);
+ }
+
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ private LogicalPlan rewrite(LogicalPlan p, CascadesContext cascadesContext) {
+ if (!(p instanceof LogicalCTE)) {
+ return p;
+ }
+ LogicalCTE logicalCTE = (LogicalCTE) p;
+ LogicalPlan child = (LogicalPlan) logicalCTE.child();
+ for (int i = logicalCTE.getAliasQueries().size() - 1; i >= 0; i--) {
+ LogicalSubQueryAlias s = (LogicalSubQueryAlias) logicalCTE.getAliasQueries().get(i);
+ int id = logicalCTE.findUniqueId(s.getAlias());
+ if (cascadesContext.cteReferencedCount(id)
+ <= ConnectContext.get().getSessionVariable().inlineCTEReferencedThreshold
+ || !ConnectContext.get().getSessionVariable().enablePipelineEngine) {
+ continue;
Review Comment:
add comment to explain this if
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalCTEAnchor.java:
##########
@@ -0,0 +1,142 @@
+// 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.trees.plans.physical;
+
+import org.apache.doris.nereids.memo.GroupExpression;
+import org.apache.doris.nereids.properties.LogicalProperties;
+import org.apache.doris.nereids.properties.PhysicalProperties;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.PlanType;
+import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
+import org.apache.doris.statistics.Statistics;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+
+/**
+ * Physical CTE anchor.
+ */
+public class PhysicalCTEAnchor<
+ LEFT_CHILD_TYPE extends Plan,
+ RIGHT_CHILD_TYPE extends Plan>
+ extends PhysicalBinary<LEFT_CHILD_TYPE, RIGHT_CHILD_TYPE> {
+ private final int cteId;
+
+ public PhysicalCTEAnchor(int cteId, LogicalProperties logicalProperties,
+ LEFT_CHILD_TYPE leftChild,
+ RIGHT_CHILD_TYPE rightChild) {
+ super(PlanType.PHYSICAL_CTE_ANCHOR, Optional.empty(),
+ logicalProperties, leftChild, rightChild);
+ this.cteId = cteId;
+ }
+
+ public PhysicalCTEAnchor(int cteId, Optional<GroupExpression> groupExpression,
+ LogicalProperties logicalProperties,
+ LEFT_CHILD_TYPE leftChild,
+ RIGHT_CHILD_TYPE rightChild) {
+ super(PlanType.PHYSICAL_CTE_ANCHOR, groupExpression, logicalProperties, leftChild, rightChild);
+ this.cteId = cteId;
+ }
+
+ public PhysicalCTEAnchor(int cteId, Optional<GroupExpression> groupExpression,
+ LogicalProperties logicalProperties, PhysicalProperties physicalProperties,
+ Statistics statistics, LEFT_CHILD_TYPE leftChild,
+ RIGHT_CHILD_TYPE rightChild) {
+ super(PlanType.PHYSICAL_CTE_ANCHOR, groupExpression, logicalProperties, physicalProperties, statistics,
+ leftChild, rightChild);
+ this.cteId = cteId;
+ }
+
+ public int getCteId() {
+ return cteId;
+ }
+
+ @Override
+ public List<Expression> getExpressions() {
+ return ImmutableList.of();
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+
+ if (!super.equals(o)) {
+ return false;
+ }
+
+ PhysicalCTEAnchor that = (PhysicalCTEAnchor) o;
+ return Objects.equals(cteId, that.cteId);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(super.hashCode(), cteId);
+ }
+
+ @Override
+ public String toString() {
+ return "PhysicalCTEAnchor{"
Review Comment:
use `Utils.toSqlString`
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/CollectProjectAboveConsumer.java:
##########
@@ -0,0 +1,79 @@
+// 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.trees.plans.logical;
+
+import org.apache.doris.nereids.pattern.MatchingContext;
+import org.apache.doris.nereids.rules.Rule;
+import org.apache.doris.nereids.rules.RuleType;
+import org.apache.doris.nereids.rules.rewrite.RewriteRuleFactory;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+
+import com.google.common.collect.ImmutableList;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * Collect Projects Above CTE Consumer.
+ */
+public class CollectProjectAboveConsumer implements RewriteRuleFactory {
+
+ @Override
+ public List<Rule> buildRules() {
+ return ImmutableList.of(RuleType.COLLECT_PROJECT_ABOVE_CONSUMER
+ .build(logicalProject(logicalCTEConsumer()).thenApply(ctx -> {
+ LogicalProject<LogicalCTEConsumer> project = ctx.root;
+ List<NamedExpression> namedExpressions = project.getProjects();
+ LogicalCTEConsumer cteConsumer = project.child();
+ collectProject(ctx, namedExpressions, cteConsumer);
+ return ctx.root;
+ })),
+ RuleType.COLLECT_PROJECT_ABOVE_FILTER_CONSUMER.build(logicalProject(logicalFilter(logicalCTEConsumer()))
+ .thenApply(ctx -> {
+ LogicalProject<LogicalFilter<LogicalCTEConsumer>> project = ctx.root;
+ LogicalFilter<LogicalCTEConsumer> filter = project.child();
+ Set<Slot> filterSlots = filter.getInputSlots();
+ List<NamedExpression> namedExpressions = new ArrayList<>(project.getProjects());
+ for (Slot slot : filterSlots) {
+ if (!project.getOutput().contains(slot)) {
+ namedExpressions.add(slot);
+ }
+ }
+ collectProject(ctx, namedExpressions, filter.child());
+ return ctx.root;
+ }))
+ );
+ }
+
+ private static void collectProject(MatchingContext ctx,
+ List<NamedExpression> namedExpressions, LogicalCTEConsumer cteConsumer) {
+ for (Expression expr : namedExpressions) {
+ expr.foreach(node -> {
Review Comment:
use `expr.collect(Slot.class::isInstance)`
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalCTEAnchor.java:
##########
@@ -0,0 +1,142 @@
+// 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.trees.plans.physical;
+
+import org.apache.doris.nereids.memo.GroupExpression;
+import org.apache.doris.nereids.properties.LogicalProperties;
+import org.apache.doris.nereids.properties.PhysicalProperties;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.PlanType;
+import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
+import org.apache.doris.statistics.Statistics;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+
+/**
+ * Physical CTE anchor.
+ */
+public class PhysicalCTEAnchor<
+ LEFT_CHILD_TYPE extends Plan,
+ RIGHT_CHILD_TYPE extends Plan>
+ extends PhysicalBinary<LEFT_CHILD_TYPE, RIGHT_CHILD_TYPE> {
+ private final int cteId;
Review Comment:
useless attribute
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/CascadesContext.java:
##########
@@ -454,4 +484,98 @@ public void close() {
}
}
}
+
+ public void putCTEIdToCTEClosure(int cteId, Callable<LogicalPlan> cteClosure) {
+ this.cteIdToCTEClosure.put(cteId, cteClosure);
+ }
+
+ public void putCTEIdToCTEClosure(Map<Integer, Callable<LogicalPlan>> cteConsumers) {
+ this.cteIdToCTEClosure.putAll(cteConsumers);
+ }
+
+ public void putCTEIdToConsumer(LogicalCTEConsumer cteConsumer) {
+ Set<LogicalCTEConsumer> consumers =
+ this.cteIdToConsumer.computeIfAbsent(cteConsumer.getCteId(), k -> new HashSet<>());
+ consumers.add(cteConsumer);
+ }
+
+ public void putCTEIdToConsumer(Map<Integer, Set<LogicalCTEConsumer>> cteConsumers) {
+ this.cteIdToConsumer.putAll(cteConsumers);
+ }
+
+ public void putCTEIdToFilter(int cteId, Expression f) {
+ Set<Expression> filters = this.cteIdToFilter.computeIfAbsent(cteId, k -> new HashSet<>());
+ filters.add(f);
+ }
+
+ public void putCTEIdToProject(int cteId, Expression p) {
+ Set<Expression> projects = this.cteIdToProject.computeIfAbsent(cteId, k -> new HashSet<>());
+ projects.add(p);
+ }
+
+ public Set<Expression> findProjectForProducer(int cteId) {
+ return this.cteIdToProject.get(cteId);
+ }
+
+ /**
+ * Fork for rewritten child tree of CTEProducer.
+ */
+ public CascadesContext forkForCTEProducer(Plan plan) {
+ CascadesContext cascadesContext = new CascadesContext(plan, memo, statementContext, PhysicalProperties.ANY);
+ cascadesContext.cteIdToConsumer = cteIdToConsumer;
+ cascadesContext.cteIdToFilter = cteIdToFilter;
+ cascadesContext.cteIdToProject = cteIdToProject;
+ cascadesContext.cteContext = cteContext;
+ cascadesContext.cteIdToCTEClosure = cteIdToCTEClosure;
+ cascadesContext.consumerIdToFilter = consumerIdToFilter;
+ return cascadesContext;
+ }
+
+ public int cteReferencedCount(int cteId) {
+ Set<LogicalCTEConsumer> cteConsumer = cteIdToConsumer.get(cteId);
+ if (cteConsumer == null) {
+ return 0;
+ }
+ return cteIdToConsumer.get(cteId).size();
+ }
+
+ public Map<Integer, Set<LogicalCTEConsumer>> getCteIdToConsumer() {
+ return cteIdToConsumer;
+ }
+
+ public Map<Integer, Callable<LogicalPlan>> getCteIdToCTEClosure() {
+ return cteIdToCTEClosure;
+ }
+
+ public LogicalPlan findCTEPlanForInline(int cteId) {
+ try {
+ return cteIdToCTEClosure.get(cteId).call();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ public void putConsumerIdToFilter(int id, Expression filter) {
+ Set<Expression> filters = this.consumerIdToFilter.computeIfAbsent(id, k -> new HashSet<>());
+ filters.add(filter);
+ }
+
+ public Set<Expression> findFiltersForConsumer(int id) {
+ return this.consumerIdToFilter.get(id);
+ }
+
+ public Map<Integer, Set<Expression>> getConsumerIdToFilter() {
+ return consumerIdToFilter;
+ }
+
+ public void markConsumerUnderProject(LogicalCTEConsumer cteConsumer) {
+ Set<Integer> consumerIds =
+ this.cteIdToConsumerUnderProject.computeIfAbsent(cteConsumer.getCteId(), k -> new HashSet<>());
+ consumerIds.add(cteConsumer.getConsumerId());
+ }
+
+ public boolean couldProjectCTEProducerBasedOnCTEConsumer(int cteId) {
+ Set<Integer> consumerIds = this.cteIdToConsumerUnderProject.get(cteId);
+ return consumerIds.size() == this.cteIdToConsumer.get(cteId).size();
+ }
Review Comment:
useless function
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalCTEConsumer.java:
##########
@@ -0,0 +1,175 @@
+// 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.trees.plans.logical;
+
+import org.apache.doris.nereids.memo.GroupExpression;
+import org.apache.doris.nereids.properties.LogicalProperties;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.StatementScopeIdGenerator;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.PlanType;
+import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+/**
+ * LogicalCTEConsumer
+ */
+public class LogicalCTEConsumer extends LogicalLeaf {
+
+ private final LogicalPlan childPlan;
+
+ private final int cteId;
+
+ private final Map<Slot, Slot> consumerToProducerOutputMap = new LinkedHashMap<>();
+
+ private final Map<Slot, Slot> producerToConsumerOutputMap = new LinkedHashMap<>();
+
+ private final int consumerId;
+
+ private final String name;
+
+ /**
+ * Logical CTE consumer.
+ */
+ public LogicalCTEConsumer(Optional<GroupExpression> groupExpression,
+ Optional<LogicalProperties> logicalProperties, LogicalPlan childPlan, int cteId, String name) {
+ super(PlanType.LOGICAL_CTE_RELATION, groupExpression, logicalProperties);
+ this.childPlan = childPlan;
+ this.cteId = cteId;
+ this.name = name;
+ initProducerToConsumerOutputMap(childPlan);
+ for (Map.Entry<Slot, Slot> entry : producerToConsumerOutputMap.entrySet()) {
+ this.consumerToProducerOutputMap.put(entry.getValue(), entry.getKey());
+ }
+ this.consumerId = StatementScopeIdGenerator.newExprId().asInt();
Review Comment:
do not use StatementScopeIdGenerator, use a new Generator for cunsumer id
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/CascadesContext.java:
##########
@@ -98,6 +101,18 @@ public class CascadesContext implements ScheduleContext, PlanSource {
private Optional<Scope> outerScope = Optional.empty();
+ private Map<Integer, Set<LogicalCTEConsumer>> cteIdToConsumer = new HashMap<>();
Review Comment:
```suggestion
private Map<Integer, Set<LogicalCTEConsumer>> cteIdToConsumers = new HashMap<>();
```
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/PlanType.java:
##########
@@ -59,10 +59,18 @@ public enum PlanType {
LOGICAL_EXCEPT,
LOGICAL_INTERSECT,
LOGICAL_USING_JOIN,
+ LOGICAL_CTE_RELATION,
+ LOGICAL_CTE_ANCHOR,
+ LOGICAL_CTE_PRODUCER,
+ LOGICAL_CTE_CONSUMER,
+
GROUP_PLAN,
// physical plan
PHYSICAL_OLAP_TABLE_SINK,
+ PHYSICAL_CTE_PRODUCE,
+ PHYSICAL_CTE_CONSUME,
Review Comment:
```suggestion
PHYSICAL_CTE_PRODUCER,
PHYSICAL_CTE_CONSUMER,
```
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/logical/BuildCTEAnchorAndCTEProducer.java:
##########
@@ -0,0 +1,64 @@
+// 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.rules.rewrite.logical;
+
+import org.apache.doris.nereids.CascadesContext;
+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.plans.logical.LogicalCTE;
+import org.apache.doris.nereids.trees.plans.logical.LogicalCTEAnchor;
+import org.apache.doris.nereids.trees.plans.logical.LogicalCTEProducer;
+import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalSubQueryAlias;
+import org.apache.doris.qe.ConnectContext;
+
+/**
+ * BuildCTEAnchorAndCTEProducer.
+ */
+public class BuildCTEAnchorAndCTEProducer extends OneRewriteRuleFactory {
+
+ @Override
+ public Rule build() {
+ return logicalCTE().thenApply(ctx -> {
+ return rewrite(ctx.root, ctx.cascadesContext);
+ }).toRule(RuleType.BUILD_CTE_ANCHOR_AND_CTE_PRODUCER);
+ }
+
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ private LogicalPlan rewrite(LogicalPlan p, CascadesContext cascadesContext) {
Review Comment:
```suggestion
private LogicalPlan rewrite(LogicalCTE<Plan> p, CascadesContext cascadesContext) {
```
--
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: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org