Revert "[refactor](nereids) New distribute planner (apache#36531)"

This reverts commit 5b68de5.

fe/fe-core/src/main/antlr4/org/apache/doris/nereids/DorisParser.g4

@@ -262,7 +262,6 @@ planType
  | OPTIMIZED | PHYSICAL // same type
  | SHAPE
  | MEMO
- | DISTRIBUTED
  | ALL // default type
  ;
 

fe/fe-core/src/main/java/org/apache/doris/common/Id.java

@@ -25,7 +25,7 @@
 /**
  * Integer ids that cannot accidentally be compared with ints.
  */
-public class Id<IdType extends Id<IdType>> implements Comparable<Id<IdType>> {
+public class Id<IdType extends Id<IdType>> {
  protected final int id;
 
  public Id(int id) {
@@ -62,9 +62,4 @@ public ArrayList<IdType> asList() {
  public String toString() {
  return Integer.toString(id);
  }
-
- @Override
- public int compareTo(Id<IdType> idTypeId) {
- return id - idTypeId.id;
- }
 }

fe/fe-core/src/main/java/org/apache/doris/common/TreeNode.java

@@ -249,29 +249,4 @@ public void foreach(ThrowingConsumer<TreeNode<NodeType>> func) throws AnalysisEx
  child.foreach(func);
  }
  }
-
- /** anyMatch */
- public boolean anyMatch(Predicate<TreeNode<? extends NodeType>> func) {
- if (func.apply(this)) {
- return true;
- }
-
- for (NodeType child : children) {
- if (child.anyMatch(func)) {
- return true;
- }
- }
- return false;
- }
-
- /** foreachDown */
- public void foreachDown(Predicate<TreeNode<NodeType>> visitor) {
- if (!visitor.test(this)) {
- return;
- }
-
- for (TreeNode<NodeType> child : getChildren()) {
- child.foreachDown(visitor);
- }
- }
 }

fe/fe-core/src/main/java/org/apache/doris/common/profile/Profile.java

@@ -20,8 +20,6 @@
 import org.apache.doris.common.util.ProfileManager;
 import org.apache.doris.common.util.RuntimeProfile;
 import org.apache.doris.nereids.NereidsPlanner;
-import org.apache.doris.nereids.trees.plans.distribute.DistributedPlan;
-import org.apache.doris.nereids.trees.plans.distribute.FragmentIdMapping;
 import org.apache.doris.nereids.trees.plans.physical.PhysicalRelation;
 import org.apache.doris.planner.Planner;
 
@@ -110,14 +108,6 @@ public synchronized void updateSummary(long startTime, Map<String, String> summa
  }
  summaryInfo.put(SummaryProfile.PHYSICAL_PLAN,
  builder.toString().replace("\n", "\n "));
-
- FragmentIdMapping<DistributedPlan> distributedPlans = nereidsPlanner.getDistributedPlans();
- if (distributedPlans != null) {
- summaryInfo.put(SummaryProfile.DISTRIBUTED_PLAN,
- DistributedPlan.toString(Lists.newArrayList(distributedPlans.values()))
- .replace("\n", "\n ")
- );
- }
  }
  summaryProfile.update(summaryInfo);
  for (ExecutionProfile executionProfile : executionProfiles) {

fe/fe-core/src/main/java/org/apache/doris/common/profile/SummaryProfile.java

@@ -54,7 +54,6 @@ public class SummaryProfile {
  public static final String TRACE_ID = "Trace ID";
  public static final String WORKLOAD_GROUP = "Workload Group";
  public static final String PHYSICAL_PLAN = "Physical Plan";
- public static final String DISTRIBUTED_PLAN = "Distributed Plan";
  // Execution Summary
  public static final String EXECUTION_SUMMARY_PROFILE_NAME = "Execution Summary";
  public static final String ANALYSIS_TIME = "Analysis Time";
@@ -87,7 +86,6 @@ public class SummaryProfile {
  public static final String NEREIDS_REWRITE_TIME = "Nereids Rewrite Time";
  public static final String NEREIDS_OPTIMIZE_TIME = "Nereids Optimize Time";
  public static final String NEREIDS_TRANSLATE_TIME = "Nereids Translate Time";
- public static final String NEREIDS_DISTRIBUTE_TIME = "Nereids Distribute Time";
 
  public static final String FRAGMENT_COMPRESSED_SIZE = "Fragment Compressed Size";
  public static final String FRAGMENT_RPC_COUNT = "Fragment RPC Count";
@@ -111,7 +109,6 @@ public class SummaryProfile {
  public static final ImmutableList<String> SUMMARY_KEYS = new ImmutableList.Builder<String>()
  .addAll(SUMMARY_CAPTIONS)
  .add(PHYSICAL_PLAN)
- .add(DISTRIBUTED_PLAN)
  .build();
 
  // The display order of execution summary items.
@@ -202,7 +199,6 @@ public class SummaryProfile {
  private long nereidsRewriteFinishTime = -1;
  private long nereidsOptimizeFinishTime = -1;
  private long nereidsTranslateFinishTime = -1;
- private long nereidsDistributeFinishTime = -1;
  // timestamp of query begin
  private long queryBeginTime = -1;
  // Analysis end time
@@ -319,7 +315,6 @@ private void updateExecutionSummaryProfile() {
  executionSummaryProfile.addInfoString(NEREIDS_REWRITE_TIME, getPrettyNereidsRewriteTime());
  executionSummaryProfile.addInfoString(NEREIDS_OPTIMIZE_TIME, getPrettyNereidsOptimizeTime());
  executionSummaryProfile.addInfoString(NEREIDS_TRANSLATE_TIME, getPrettyNereidsTranslateTime());
- executionSummaryProfile.addInfoString(NEREIDS_DISTRIBUTE_TIME, getPrettyNereidsDistributeTime());
  executionSummaryProfile.addInfoString(ANALYSIS_TIME,
  getPrettyTime(queryAnalysisFinishTime, queryBeginTime, TUnit.TIME_MS));
  executionSummaryProfile.addInfoString(PLAN_TIME,
@@ -424,10 +419,6 @@ public void setNereidsTranslateTime() {
  this.nereidsTranslateFinishTime = TimeUtils.getStartTimeMs();
  }
 
- public void setNereidsDistributeTime() {
- this.nereidsDistributeFinishTime = TimeUtils.getStartTimeMs();
- }
-
  public void setQueryBeginTime() {
  this.queryBeginTime = TimeUtils.getStartTimeMs();
  }
@@ -663,10 +654,6 @@ public String getPrettyNereidsTranslateTime() {
  return getPrettyTime(nereidsTranslateFinishTime, nereidsOptimizeFinishTime, TUnit.TIME_MS);
  }
 
- public String getPrettyNereidsDistributeTime() {
- return getPrettyTime(nereidsDistributeFinishTime, nereidsTranslateFinishTime, TUnit.TIME_MS);
- }
-
  private String getPrettyGetPartitionVersionTime() {
  if (getPartitionVersionTime == 0) {
  return "N/A";

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

@@ -52,9 +52,6 @@
 import org.apache.doris.nereids.trees.expressions.literal.Literal;
 import org.apache.doris.nereids.trees.plans.Plan;
 import org.apache.doris.nereids.trees.plans.commands.ExplainCommand.ExplainLevel;
-import org.apache.doris.nereids.trees.plans.distribute.DistributePlanner;
-import org.apache.doris.nereids.trees.plans.distribute.DistributedPlan;
-import org.apache.doris.nereids.trees.plans.distribute.FragmentIdMapping;
 import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
 import org.apache.doris.nereids.trees.plans.logical.LogicalSqlCache;
 import org.apache.doris.nereids.trees.plans.physical.PhysicalEmptyRelation;
@@ -73,7 +70,6 @@
 import org.apache.doris.qe.ConnectContext;
 import org.apache.doris.qe.ResultSet;
 import org.apache.doris.qe.ResultSetMetaData;
-import org.apache.doris.qe.SessionVariable;
 import org.apache.doris.qe.cache.CacheAnalyzer;
 
 import com.google.common.annotations.VisibleForTesting;
@@ -106,7 +102,6 @@ public class NereidsPlanner extends Planner {
  private Plan rewrittenPlan;
  private Plan optimizedPlan;
  private PhysicalPlan physicalPlan;
- private FragmentIdMapping<DistributedPlan> distributedPlans;
  // The cost of optimized plan
  private double cost = 0;
  private LogicalPlanAdapter logicalPlanAdapter;
@@ -135,7 +130,6 @@ public void plan(StatementBase queryStmt, org.apache.doris.thrift.TQueryOptions
  LogicalPlan parsedPlan = logicalPlanAdapter.getLogicalPlan();
  NereidsTracer.logImportantTime("EndParsePlan");
  setParsedPlan(parsedPlan);
-
  PhysicalProperties requireProperties = buildInitRequireProperties();
  statementContext.getStopwatch().start();
  Plan resultPlan = null;
@@ -146,11 +140,11 @@ public void plan(StatementBase queryStmt, org.apache.doris.thrift.TQueryOptions
  statementContext.getStopwatch().stop();
  }
  setOptimizedPlan(resultPlan);
-
- if (resultPlan instanceof PhysicalPlan) {
- physicalPlan = (PhysicalPlan) resultPlan;
- distribute(physicalPlan, explainLevel);
+ if (explainLevel.isPlanLevel) {
+ return;
  }
+ physicalPlan = (PhysicalPlan) resultPlan;
+ translate(physicalPlan);
  }
 
  @VisibleForTesting
@@ -325,7 +319,7 @@ private void optimize() {
  }
  }
 
- private void splitFragments(PhysicalPlan resultPlan) throws UserException {
+ private void translate(PhysicalPlan resultPlan) throws UserException {
  if (resultPlan instanceof PhysicalSqlCache) {
  return;
  }
@@ -370,27 +364,6 @@ private void splitFragments(PhysicalPlan resultPlan) throws UserException {
  ScanNode.setVisibleVersionForOlapScanNodes(getScanNodes());
  }
 
- private void distribute(PhysicalPlan physicalPlan, ExplainLevel explainLevel) throws UserException {
- boolean canUseNereidsDistributePlanner = SessionVariable.canUseNereidsDistributePlanner();
- if ((!canUseNereidsDistributePlanner && explainLevel.isPlanLevel)) {
- return;
- } else if ((canUseNereidsDistributePlanner && explainLevel.isPlanLevel
- && (explainLevel != ExplainLevel.ALL_PLAN && explainLevel != ExplainLevel.DISTRIBUTED_PLAN))) {
- return;
- }
-
- splitFragments(physicalPlan);
-
- if (!canUseNereidsDistributePlanner) {
- return;
- }
-
- distributedPlans = new DistributePlanner(fragments).plan();
- if (statementContext.getConnectContext().getExecutor() != null) {
- statementContext.getConnectContext().getExecutor().getSummaryProfile().setNereidsDistributeTime();
- }
- }
-
  private PhysicalPlan postProcess(PhysicalPlan physicalPlan) {
  return new PlanPostProcessors(cascadesContext).process(physicalPlan);
  }
@@ -529,17 +502,6 @@ public String getExplainString(ExplainOptions explainOptions) {
  + "\n\n========== MATERIALIZATIONS ==========\n"
  + materializationStringBuilder;
  break;
- case DISTRIBUTED_PLAN:
- StringBuilder distributedPlanStringBuilder = new StringBuilder();
-
- distributedPlanStringBuilder.append("========== DISTRIBUTED PLAN ==========\n");
- if (distributedPlans == null || distributedPlans.isEmpty()) {
- plan = "Distributed plan not generated, please set enable_nereids_distribute_planner "
- + "and enable_pipeline_x_engine to true";
- } else {
- plan += DistributedPlan.toString(Lists.newArrayList(distributedPlans.values())) + "\n\n";
- }
- break;
  case ALL_PLAN:
  plan = "========== PARSED PLAN "
  + getTimeMetricString(SummaryProfile::getPrettyParseSqlTime) + " ==========\n"
@@ -552,13 +514,7 @@ public String getExplainString(ExplainOptions explainOptions) {
  + rewrittenPlan.treeString() + "\n\n"
  + "========== OPTIMIZED PLAN "
  + getTimeMetricString(SummaryProfile::getPrettyNereidsOptimizeTime) + " ==========\n"
- + optimizedPlan.treeString() + "\n\n";
-
- if (distributedPlans != null && !distributedPlans.isEmpty()) {
- plan += "========== DISTRIBUTED PLAN "
- + getTimeMetricString(SummaryProfile::getPrettyNereidsDistributeTime) + " ==========\n";
- plan += DistributedPlan.toString(Lists.newArrayList(distributedPlans.values())) + "\n\n";
- }
+ + optimizedPlan.treeString();
  break;
  default:
  plan = super.getExplainString(explainOptions)
@@ -729,10 +685,6 @@ public PhysicalPlan getPhysicalPlan() {
  return physicalPlan;
  }
 
- public FragmentIdMapping<DistributedPlan> getDistributedPlans() {
- return distributedPlans;
- }
-
  public LogicalPlanAdapter getLogicalPlanAdapter() {
  return logicalPlanAdapter;
  }

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

@@ -3459,9 +3459,6 @@ private ExplainLevel parseExplainPlanType(PlanTypeContext planTypeContext) {
  if (planTypeContext.MEMO() != null) {
  return ExplainLevel.MEMO_PLAN;
  }
- if (planTypeContext.DISTRIBUTED() != null) {
- return ExplainLevel.DISTRIBUTED_PLAN;
- }
  return ExplainLevel.ALL_PLAN;
  }
 

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

@@ -46,7 +46,6 @@
 import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
 import org.apache.doris.nereids.util.JoinUtils;
 import org.apache.doris.qe.ConnectContext;
-import org.apache.doris.qe.SessionVariable;
 
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableList;
@@ -214,12 +213,12 @@ private boolean couldNotRightBucketShuffleJoin(JoinType joinType, DistributionSp
  || joinType == JoinType.FULL_OUTER_JOIN);
  boolean isSpecInScope = (leftHashSpec.getShuffleType() == ShuffleType.NATURAL
  || rightHashSpec.getShuffleType() == ShuffleType.NATURAL);
- return isJoinTypeInScope && isSpecInScope && !SessionVariable.canUseNereidsDistributePlanner();
+ return isJoinTypeInScope && isSpecInScope;
  }
 
  @Override
- public Boolean visitPhysicalHashJoin(
- PhysicalHashJoin<? extends Plan, ? extends Plan> hashJoin, Void context) {
+ public Boolean visitPhysicalHashJoin(PhysicalHashJoin<? extends Plan, ? extends Plan> hashJoin,
+ Void context) {
  Preconditions.checkArgument(children.size() == 2, "children.size() != 2");
  Preconditions.checkArgument(childrenProperties.size() == 2);
  Preconditions.checkArgument(requiredProperties.size() == 2);
@@ -304,24 +303,13 @@ public Boolean visitPhysicalHashJoin(
  (DistributionSpecHash) requiredProperties.get(1).getDistributionSpec()));
  } else if (leftHashSpec.getShuffleType() == ShuffleType.EXECUTION_BUCKETED
  && rightHashSpec.getShuffleType() == ShuffleType.NATURAL) {
- if (SessionVariable.canUseNereidsDistributePlanner()) {
- // nereids coordinator can exchange left side to right side to do bucket shuffle join
- // TODO: maybe we should check if left child is PhysicalDistribute.
- // If so add storage bucketed shuffle on left side. Other wise,
- // add execution bucketed shuffle on right side.
- updatedForLeft = Optional.of(calAnotherSideRequired(
- ShuffleType.STORAGE_BUCKETED, rightHashSpec, leftHashSpec,
- (DistributionSpecHash) requiredProperties.get(1).getDistributionSpec(),
- (DistributionSpecHash) requiredProperties.get(0).getDistributionSpec()));
- } else {
- // legacy coordinator could not do right be selection in this case,
- // since it always to check the left most node whether olap scan node.
- // so we can only shuffle right to left side to do normal shuffle join
- updatedForRight = Optional.of(calAnotherSideRequired(
- ShuffleType.EXECUTION_BUCKETED, leftHashSpec, rightHashSpec,
- (DistributionSpecHash) requiredProperties.get(0).getDistributionSpec(),
- (DistributionSpecHash) requiredProperties.get(1).getDistributionSpec()));
- }
+ // TODO: we must do shuffle on right because coordinator could not do right be selection in this case,
+ // since it always to check the left most node whether olap scan node.
+ // after we fix coordinator problem, we could do right to left bucket shuffle
+ updatedForRight = Optional.of(calAnotherSideRequired(
+ ShuffleType.EXECUTION_BUCKETED, leftHashSpec, rightHashSpec,
+ (DistributionSpecHash) requiredProperties.get(0).getDistributionSpec(),
+ (DistributionSpecHash) requiredProperties.get(1).getDistributionSpec()));
  } else if (leftHashSpec.getShuffleType() == ShuffleType.EXECUTION_BUCKETED
  && rightHashSpec.getShuffleType() == ShuffleType.EXECUTION_BUCKETED) {
  if (bothSideShuffleKeysAreSameOrder(rightHashSpec, leftHashSpec,
@@ -549,20 +537,20 @@ private List<ExprId> calAnotherSideRequiredShuffleIds(DistributionSpecHash notSh
  * calAnotherSideRequiredShuffleIds's comment.
  *
  * @param shuffleType real output shuffle type
- * @param notNeedShuffleSideOutput not shuffle side real output used hash spec
- * @param needShuffleSideOutput shuffle side real output used hash spec
- * @param notNeedShuffleSideRequired not shuffle side required used hash spec
- * @param needShuffleSideRequired shuffle side required hash spec
+ * @param notShuffleSideOutput not shuffle side real output used hash spec
+ * @param shuffleSideOutput shuffle side real output used hash spec
+ * @param notShuffleSideRequired not shuffle side required used hash spec
+ * @param shuffleSideRequired shuffle side required hash spec
  * @return shuffle side new required hash spec
  */
  private PhysicalProperties calAnotherSideRequired(ShuffleType shuffleType,
- DistributionSpecHash notNeedShuffleSideOutput, DistributionSpecHash needShuffleSideOutput,
- DistributionSpecHash notNeedShuffleSideRequired, DistributionSpecHash needShuffleSideRequired) {
- List<ExprId> shuffleSideIds = calAnotherSideRequiredShuffleIds(notNeedShuffleSideOutput,
- notNeedShuffleSideRequired, needShuffleSideRequired);
+ DistributionSpecHash notShuffleSideOutput, DistributionSpecHash shuffleSideOutput,
+ DistributionSpecHash notShuffleSideRequired, DistributionSpecHash shuffleSideRequired) {
+ List<ExprId> shuffleSideIds = calAnotherSideRequiredShuffleIds(notShuffleSideOutput,
+ notShuffleSideRequired, shuffleSideRequired);
  return new PhysicalProperties(new DistributionSpecHash(shuffleSideIds, shuffleType,
- needShuffleSideOutput.getTableId(), needShuffleSideOutput.getSelectedIndexId(),
- needShuffleSideOutput.getPartitionIds()));
+ shuffleSideOutput.getTableId(), shuffleSideOutput.getSelectedIndexId(),
+ shuffleSideOutput.getPartitionIds()));
  }
 
  private void updateChildEnforceAndCost(int index, PhysicalProperties targetProperties) {

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/table/TableValuedFunction.java

@@ -31,7 +31,6 @@
 import org.apache.doris.nereids.trees.expressions.shape.UnaryExpression;
 import org.apache.doris.nereids.trees.expressions.visitor.ExpressionVisitor;
 import org.apache.doris.nereids.types.DataType;
-import org.apache.doris.qe.SessionVariable;
 import org.apache.doris.statistics.ColumnStatistic;
 import org.apache.doris.statistics.Statistics;
 import org.apache.doris.tablefunction.TableValuedFunctionIf;
@@ -114,9 +113,6 @@ public boolean nullable() {
  }
 
  public PhysicalProperties getPhysicalProperties() {
- if (SessionVariable.canUseNereidsDistributePlanner()) {
- return PhysicalProperties.ANY;
- }
  return PhysicalProperties.STORAGE_ANY;
  }