[Enhancement] support extract common spjg as cte

Signed-off-by: Seaven <[email protected]>
StarRocks · Oct 18, 2024 · 4fd236f · 4fd236f
1 parent acce535
commit 4fd236f
Show file tree

Hide file tree

Showing 17 changed files with 993 additions and 20 deletions.
diff --git a/fe/fe-core/src/main/java/com/starrocks/analysis/JoinOperator.java b/fe/fe-core/src/main/java/com/starrocks/analysis/JoinOperator.java
@@ -40,20 +40,19 @@
 import java.util.Set;
 
 public enum JoinOperator {
-    INNER_JOIN("INNER JOIN", TJoinOp.INNER_JOIN),
-    LEFT_OUTER_JOIN("LEFT OUTER JOIN", TJoinOp.LEFT_OUTER_JOIN),
-
-    LEFT_SEMI_JOIN("LEFT SEMI JOIN", TJoinOp.LEFT_SEMI_JOIN),
-    LEFT_ANTI_JOIN("LEFT ANTI JOIN", TJoinOp.LEFT_ANTI_JOIN),
-    RIGHT_SEMI_JOIN("RIGHT SEMI JOIN", TJoinOp.RIGHT_SEMI_JOIN),
-    RIGHT_ANTI_JOIN("RIGHT ANTI JOIN", TJoinOp.RIGHT_ANTI_JOIN),
-    RIGHT_OUTER_JOIN("RIGHT OUTER JOIN", TJoinOp.RIGHT_OUTER_JOIN),
-    FULL_OUTER_JOIN("FULL OUTER JOIN", TJoinOp.FULL_OUTER_JOIN),
-    CROSS_JOIN("CROSS JOIN", TJoinOp.CROSS_JOIN),
-    // Variant of the LEFT ANTI JOIN that is used for the equal of
+    INNER_JOIN("INNER JOIN", "⋈", TJoinOp.INNER_JOIN),
+    LEFT_OUTER_JOIN("LEFT OUTER JOIN", "⟕", TJoinOp.LEFT_OUTER_JOIN),
+
+    LEFT_SEMI_JOIN("LEFT SEMI JOIN", "⋉", TJoinOp.LEFT_SEMI_JOIN),
+    LEFT_ANTI_JOIN("LEFT ANTI JOIN", "◁", TJoinOp.LEFT_ANTI_JOIN),
+    RIGHT_SEMI_JOIN("RIGHT SEMI JOIN", "⋊", TJoinOp.RIGHT_SEMI_JOIN),
+    RIGHT_ANTI_JOIN("RIGHT ANTI JOIN", "▷", TJoinOp.RIGHT_ANTI_JOIN),
+    RIGHT_OUTER_JOIN("RIGHT OUTER JOIN", "⟖", TJoinOp.RIGHT_OUTER_JOIN),
+    FULL_OUTER_JOIN("FULL OUTER JOIN", "⟗", TJoinOp.FULL_OUTER_JOIN),
+    CROSS_JOIN("CROSS JOIN", "×", TJoinOp.CROSS_JOIN), // Variant of the LEFT ANTI JOIN that is used for the equal of
     // NOT IN subqueries. It can have a single equality join conjunct
     // that returns TRUE when the rhs is NULL.
-    NULL_AWARE_LEFT_ANTI_JOIN("NULL AWARE LEFT ANTI JOIN",
+    NULL_AWARE_LEFT_ANTI_JOIN("NULL AWARE LEFT ANTI JOIN", "▷*",
             TJoinOp.NULL_AWARE_LEFT_ANTI_JOIN);
 
     public static final String HINT_BUCKET = "BUCKET";
@@ -65,10 +64,12 @@ public enum JoinOperator {
     public static final String HINT_UNREORDER = "UNREORDER";
 
     private final String description;
+    private final String algebra;
     private final TJoinOp thriftJoinOp;
 
-    private JoinOperator(String description, TJoinOp thriftJoinOp) {
+    JoinOperator(String description, String algebra, TJoinOp thriftJoinOp) {
         this.description = description;
+        this.algebra = algebra;
         this.thriftJoinOp = thriftJoinOp;
     }
 
@@ -77,6 +78,10 @@ public String toString() {
         return description;
     }
 
+    public String toAlgebra() {
+        return algebra;
+    }
+
     public TJoinOp toThrift() {
         return thriftJoinOp;
     }

diff --git a/fe/fe-core/src/main/java/com/starrocks/qe/SessionVariable.java b/fe/fe-core/src/main/java/com/starrocks/qe/SessionVariable.java
@@ -376,6 +376,8 @@ public class SessionVariable implements Serializable, Writable, Cloneable {
 
     public static final String CBO_ENABLE_PARALLEL_PREPARE_METADATA = "enable_parallel_prepare_metadata";
 
+    public static final String CBO_EXTRACT_COMMON_PLAN = "cbo_extract_common_plan";
+
     public static final String SKEW_JOIN_RAND_RANGE = "skew_join_rand_range";
     public static final String ENABLE_STATS_TO_OPTIMIZE_SKEW_JOIN = "enable_stats_to_optimize_skew_join";
     public static final String SKEW_JOIN_OPTIMIZE_USE_MCV_COUNT = "skew_join_use_mcv_count";
@@ -1273,6 +1275,9 @@ public static MaterializedViewRewriteMode parse(String str) {
     @VariableMgr.VarAttr(name = ENABLE_GIN_FILTER)
     private boolean enableGinFilter = true;
 
+    @VariableMgr.VarAttr(name = CBO_EXTRACT_COMMON_PLAN)
+    private boolean cboExtractCommonPlan = true;
+
     @VariableMgr.VarAttr(name = CBO_MAX_REORDER_NODE_USE_EXHAUSTIVE)
     private int cboMaxReorderNodeUseExhaustive = 4;
 
@@ -3484,6 +3489,14 @@ public void setEnableMultiColumnsOnGlobbalRuntimeFilter(boolean value) {
         this.enableMultiColumnsOnGlobalRuntimeFilter = value;
     }
 
+    public boolean isCboExtractCommonPlan() {
+        return cboExtractCommonPlan;
+    }
+
+    public void setCboExtractCommonPlan(boolean cboExtractCommonPlan) {
+        this.cboExtractCommonPlan = cboExtractCommonPlan;
+    }
+
     public boolean isEnableQueryDebugTrace() {
         return enableQueryDebugTrace;
     }

diff --git a/fe/fe-core/src/main/java/com/starrocks/sql/common/DebugOperatorTracer.java b/fe/fe-core/src/main/java/com/starrocks/sql/common/DebugOperatorTracer.java
@@ -273,7 +273,7 @@ public String visitLogicalTableFunction(LogicalTableFunctionOperator node, Void
 
     @Override
     public String visitLogicalLimit(LogicalLimitOperator node, Void context) {
-        return "LogicalLimitOperator" + " {limit=" + node.getLimit() +
+        return "LogicalLimitOperator {" + node.getPhase().name() + " limit=" + node.getLimit() +
                 ", offset=" + node.getOffset() +
                 "}";
     }

diff --git a/fe/fe-core/src/main/java/com/starrocks/sql/optimizer/OptExpression.java b/fe/fe-core/src/main/java/com/starrocks/sql/optimizer/OptExpression.java
@@ -264,11 +264,9 @@ private String debugString(String headlinePrefix, String detailPrefix, int limit
         StringBuilder sb = new StringBuilder();
         sb.append(headlinePrefix).append(op.accept(new DebugOperatorTracer(), null));
         limitLine -= 1;
-        sb.append('\n');
         if (limitLine <= 0 || inputs.isEmpty()) {
             return sb.toString();
         }
-
         String childHeadlinePrefix = detailPrefix + "->  ";
         String childDetailPrefix = detailPrefix + "    ";
         for (OptExpression input : inputs) {

diff --git a/fe/fe-core/src/main/java/com/starrocks/sql/optimizer/Optimizer.java b/fe/fe-core/src/main/java/com/starrocks/sql/optimizer/Optimizer.java
@@ -106,6 +106,7 @@
 import com.starrocks.sql.optimizer.rule.tree.SimplifyCaseWhenPredicateRule;
 import com.starrocks.sql.optimizer.rule.tree.SubfieldExprNoCopyRule;
 import com.starrocks.sql.optimizer.rule.tree.lowcardinality.LowCardinalityRewriteRule;
+import com.starrocks.sql.optimizer.rule.tree.pieces.ReuseFusionPlanRule;
 import com.starrocks.sql.optimizer.rule.tree.prunesubfield.PruneSubfieldRule;
 import com.starrocks.sql.optimizer.rule.tree.prunesubfield.PushDownSubfieldRule;
 import com.starrocks.sql.optimizer.task.OptimizeGroupTask;
@@ -528,6 +529,7 @@ private OptExpression logicalRuleRewrite(
         ruleRewriteOnlyOnce(tree, rootTaskContext, RuleSetType.PRUNE_COLUMNS);
         ruleRewriteIterative(tree, rootTaskContext, RuleSetType.PRUNE_UKFK_JOIN);
         deriveLogicalProperty(tree);
+        tree = extractCommonCTE(tree, rootTaskContext, requiredColumns);
 
         ruleRewriteOnlyOnce(tree, rootTaskContext, new PushDownJoinOnExpressionToChildProject());
 
@@ -670,6 +672,21 @@ private OptExpression logicalRuleRewrite(
         return tree.getInputs().get(0);
     }
 
+    private OptExpression extractCommonCTE(OptExpression tree, TaskContext rootTaskContext,
+                                           ColumnRefSet requiredColumns) {
+        if (!context.getSessionVariable().isCboExtractCommonPlan()) {
+            return tree;
+        }
+        ReuseFusionPlanRule fusion = new ReuseFusionPlanRule();
+        tree = fusion.rewrite(tree, rootTaskContext);
+        if (fusion.hasRewrite()) {
+            deriveLogicalProperty(tree);
+            rootTaskContext.setRequiredColumns(requiredColumns.clone());
+            ruleRewriteOnlyOnce(tree, rootTaskContext, RuleSetType.PRUNE_COLUMNS);
+        }
+        return tree;
+    }
+
     private void rewriteGroupingSets(OptExpression tree, TaskContext rootTaskContext, SessionVariable sessionVariable) {
         if (sessionVariable.isEnableRewriteGroupingsetsToUnionAll()) {
             ruleRewriteIterative(tree, rootTaskContext, new RewriteGroupingSetsByCTERule());

diff --git a/fe/fe-core/src/main/java/com/starrocks/sql/optimizer/operator/OperatorType.java b/fe/fe-core/src/main/java/com/starrocks/sql/optimizer/operator/OperatorType.java
@@ -56,6 +56,7 @@ public enum OperatorType {
     LOGICAL_CTE_ANCHOR,
     LOGICAL_CTE_PRODUCE,
     LOGICAL_CTE_CONSUME,
+    LOGICAL_SPJG_PIECES,
 
     /**
      * Physical operator

diff --git a/fe/fe-core/src/main/java/com/starrocks/sql/optimizer/rewrite/ReplaceColumnRefRewriter.java b/fe/fe-core/src/main/java/com/starrocks/sql/optimizer/rewrite/ReplaceColumnRefRewriter.java
@@ -66,15 +66,18 @@ public ScalarOperator visitVariableReference(ColumnRefOperator column, Void cont
             // The rewritten predicate will be rewritten continually,
             // Rewiring predicate shouldn't change the origin project columnRefMap
 
-            ScalarOperator mapperOperator = operatorMap.get(column).clone();
-            if (isRecursively) {
-                while (mapperOperator.getChildren().isEmpty() && operatorMap.containsKey(mapperOperator)) {
+            ScalarOperator mapperOperator = operatorMap.get(column);
+            if (!isRecursively) {
+                return mapperOperator.clone();
+            } else {
+                while (mapperOperator instanceof ColumnRefOperator && operatorMap.containsKey(mapperOperator)) {
                     ScalarOperator mapped = operatorMap.get(mapperOperator);
                     if (mapped.equals(mapperOperator)) {
                         break;
                     }
-                    mapperOperator = mapped.clone();
+                    mapperOperator = mapped;
                 }
+                mapperOperator = mapperOperator.clone();
                 for (int i = 0; i < mapperOperator.getChildren().size(); ++i) {
                     mapperOperator.setChild(i, mapperOperator.getChild(i).accept(this, null));
                 }

diff --git a/fe/fe-core/src/main/java/com/starrocks/sql/optimizer/rule/RuleSet.java b/fe/fe-core/src/main/java/com/starrocks/sql/optimizer/rule/RuleSet.java
@@ -253,6 +253,7 @@ public class RuleSet {
                 MergeLimitDirectRule.EXCEPT,
                 MergeLimitDirectRule.VALUES,
                 MergeLimitDirectRule.FILTER,
+                MergeLimitDirectRule.CTE_CONSUMER,
                 MergeLimitDirectRule.TABLE_FUNCTION,
                 MergeLimitDirectRule.TABLE_FUNCTION_TABLE_SCAN
         ));

diff --git a/fe/fe-core/src/main/java/com/starrocks/sql/optimizer/rule/RuleType.java b/fe/fe-core/src/main/java/com/starrocks/sql/optimizer/rule/RuleType.java
@@ -204,6 +204,8 @@ public enum RuleType {
 
     TF_VECTOR_REWRITE_RULE,
 
+    TF_REUSE_FUSION_RULE,
+
     // The following are implementation rules:
     IMP_OLAP_LSCAN_TO_PSCAN,
     IMP_HIVE_LSCAN_TO_PSCAN,

diff --git a/...e/src/main/java/com/starrocks/sql/optimizer/rule/transformation/MergeLimitDirectRule.java b/...e/src/main/java/com/starrocks/sql/optimizer/rule/transformation/MergeLimitDirectRule.java
@@ -54,6 +54,8 @@ public class MergeLimitDirectRule extends TransformationRule {
             new MergeLimitDirectRule(OperatorType.LOGICAL_TABLE_FUNCTION);
     public static final MergeLimitDirectRule TABLE_FUNCTION_TABLE_SCAN =
             new MergeLimitDirectRule(OperatorType.LOGICAL_TABLE_FUNCTION_TABLE_SCAN);
+    public static final MergeLimitDirectRule CTE_CONSUMER =
+            new MergeLimitDirectRule(OperatorType.LOGICAL_CTE_CONSUME);
 
     private MergeLimitDirectRule(OperatorType logicalOperatorType) {
         super(RuleType.TF_MERGE_LIMIT_DIRECT, Pattern.create(OperatorType.LOGICAL_LIMIT)

diff --git a/...ore/src/main/java/com/starrocks/sql/optimizer/rule/tree/pieces/LogicalPiecesOperator.java b/...ore/src/main/java/com/starrocks/sql/optimizer/rule/tree/pieces/LogicalPiecesOperator.java
@@ -0,0 +1,40 @@
+// Copyright 2021-present StarRocks, Inc. All rights reserved.
+//
+// Licensed 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
+//
+//     https://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 com.starrocks.sql.optimizer.rule.tree.pieces;
+
+import com.starrocks.sql.optimizer.OptExpression;
+import com.starrocks.sql.optimizer.operator.Operator;
+import com.starrocks.sql.optimizer.operator.OperatorType;
+
+class LogicalPiecesOperator extends Operator {
+    public LogicalPiecesOperator(OperatorType type, OptExpression plan, QueryPiecesPlan piece) {
+        super(type);
+        this.plan = plan;
+        this.piece = piece;
+        this.predicate = plan.getOp().getPredicate();
+    }
+
+    private final OptExpression plan;
+
+    private final QueryPiecesPlan piece;
+
+    public OptExpression getPlan() {
+        return plan;
+    }
+
+    public QueryPiecesPlan getPiece() {
+        return piece;
+    }
+}