[Enhancement] support extract common spjg as cte

Signed-off-by: Seaven <[email protected]>

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;
  }

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";
@@ -1279,6 +1281,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;
 
@@ -3515,6 +3520,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;
  }

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() +
  "}";
  }

fe/fe-core/src/main/java/com/starrocks/sql/optimizer/MaterializedViewOptimizer.java

@@ -48,6 +48,7 @@ public MvPlanContext optimize(MaterializedView mv,
  optimizerConfig.disableRuleSet(RuleSetType.INTERSECT_REWRITE);
  optimizerConfig.disableRule(RuleType.TF_REWRITE_GROUP_BY_COUNT_DISTINCT);
  optimizerConfig.disableRule(RuleType.TF_PRUNE_EMPTY_SCAN);
+ optimizerConfig.disableRule(RuleType.TF_REUSE_FUSION_RULE);
  optimizerConfig.disableRule(RuleType.TF_MV_TEXT_MATCH_REWRITE_RULE);
  optimizerConfig.disableRule(RuleType.TF_MV_TRANSPARENT_REWRITE_RULE);
  // For sync mv, no rewrite query by original sync mv rule to avoid useless rewrite.

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) {

fe/fe-core/src/main/java/com/starrocks/sql/optimizer/Optimizer.java

@@ -39,6 +39,7 @@
 import com.starrocks.sql.optimizer.rewrite.JoinPredicatePushdown;
 import com.starrocks.sql.optimizer.rule.Rule;
 import com.starrocks.sql.optimizer.rule.RuleSetType;
+import com.starrocks.sql.optimizer.rule.RuleType;
 import com.starrocks.sql.optimizer.rule.implementation.OlapScanImplementationRule;
 import com.starrocks.sql.optimizer.rule.join.ReorderJoinRule;
 import com.starrocks.sql.optimizer.rule.mv.MaterializedViewRule;
@@ -107,6 +108,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;
@@ -530,6 +532,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());
 
@@ -643,6 +646,8 @@ private OptExpression logicalRuleRewrite(
  ruleRewriteIterative(tree, rootTaskContext, new MergeProjectWithChildRule());
 
  ruleRewriteOnlyOnce(tree, rootTaskContext, new PushDownTopNBelowOuterJoinRule());
+ // intersect rewrite depend on statistics
+ Utils.calculateStatistics(tree, rootTaskContext.getOptimizerContext());
  ruleRewriteOnlyOnce(tree, rootTaskContext, RuleSetType.INTERSECT_REWRITE);
  ruleRewriteIterative(tree, rootTaskContext, new RemoveAggregationFromAggTable());
 
@@ -673,6 +678,22 @@ private OptExpression logicalRuleRewrite(
  return tree.getInputs().get(0);
  }
 
+ private OptExpression extractCommonCTE(OptExpression tree, TaskContext rootTaskContext,
+ ColumnRefSet requiredColumns) {
+ if (!context.getSessionVariable().isCboExtractCommonPlan() ||
+ optimizerConfig.isRuleDisable(RuleType.TF_REUSE_FUSION_RULE)) {
+ 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());

fe/fe-core/src/main/java/com/starrocks/sql/optimizer/operator/OperatorType.java

@@ -57,6 +57,7 @@ public enum OperatorType {
  LOGICAL_CTE_ANCHOR,
  LOGICAL_CTE_PRODUCE,
  LOGICAL_CTE_CONSUME,
+ LOGICAL_SPJG_PIECES,
 
  /**
  * Physical operator

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));
  }

fe/fe-core/src/main/java/com/starrocks/sql/optimizer/rule/RuleSet.java

@@ -256,6 +256,7 @@ public class RuleSet {
  MergeLimitDirectRule.EXCEPT,
  MergeLimitDirectRule.VALUES,
  MergeLimitDirectRule.FILTER,
+ MergeLimitDirectRule.CTE_CONSUMER,
  MergeLimitDirectRule.TABLE_FUNCTION,
  MergeLimitDirectRule.TABLE_FUNCTION_TABLE_SCAN
  ));

fe/fe-core/src/main/java/com/starrocks/sql/optimizer/rule/RuleType.java

@@ -206,6 +206,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,

fe/fe-core/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)

fe/fe-core/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;
+ }
+}