Parallel fragment exec instance

fe/src/main/java/org/apache/doris/common/util/ListUtil.java

@@ -0,0 +1,58 @@
+// 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.common.util;
+
+import com.google.common.base.Preconditions;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class ListUtil {
+    /**
+     * split a list to multi expected number sublist
+     * for example:
+     *
+     * list is : [1, 2, 3, 4, 5, 6, 7]
+     * expectedSize is : 3
+     *
+     * return :
+     * [1, 4, 7]
+     * [2, 5]
+     * [3, 6]
+     */
+    public static  <T> List<List<T>> splitBySize(List<T> list, int expectedSize)
+            throws NullPointerException, IllegalArgumentException {
+        Preconditions.checkNotNull(list, "list must not be null");
+        Preconditions.checkArgument(expectedSize > 0, "expectedSize must larger than 0");
+
+        int splitSize = Math.min(expectedSize, list.size());
+
+        List<List<T>> result = new ArrayList<List<T>>(splitSize);
+        for (int i = 0; i < splitSize; i++) {
+            result.add(new ArrayList<>());
+        }
+
+        int index = 0;
+        for (T t : list) {
+            result.get(index).add(t);
+            index = (index + 1) % splitSize;
+        }
+
+        return result;
+    }
+}

fe/src/main/java/org/apache/doris/planner/HashJoinNode.java

@@ -294,6 +294,10 @@ public int getNumInstances() {
         return Math.max(children.get(0).getNumInstances(), children.get(1).getNumInstances());
     }
 
+    public boolean isShuffleJoin() {
+        return distrMode == DistributionMode.PARTITIONED;
+    }
+
     enum DistributionMode {
         NONE("NONE"),
         BROADCAST("BROADCAST"),

fe/src/main/java/org/apache/doris/planner/Planner.java

@@ -145,15 +145,8 @@ public void createPlanFragments(StatementBase statment, Analyzer analyzer, TQuer
         singleNodePlanner = new SingleNodePlanner(plannerContext);
         PlanNode singleNodePlan = singleNodePlanner.createSingleNodePlan();
 
-        singleNodePlanner.validatePlan(singleNodePlan);
-
         List<Expr> resultExprs = queryStmt.getResultExprs();
         if (statment instanceof InsertStmt) {
-            if (queryOptions.isSetMt_dop() && queryOptions.mt_dop > 0) {
-                throw new NotImplementedException(
-                        "MT_DOP not supported for plans with insert.");
-            }
-
             InsertStmt insertStmt = (InsertStmt) statment;
             if (insertStmt.getOlapTuple() != null && !insertStmt.isStreaming()) {
                 singleNodePlan = new OlapRewriteNode(plannerContext.getNextNodeId(), singleNodePlan, insertStmt);

fe/src/main/java/org/apache/doris/planner/SingleNodePlanner.java

@@ -150,31 +150,6 @@ public PlanNode createSingleNodePlan() throws UserException, AnalysisException {
         return singleNodePlan;
     }
 
-    /**
-     * Checks that the given single-node plan is executable:
-     * - It may not contain right or full outer joins with no equi-join conjuncts that
-     *   are not inside the right child of a SubplanNode.
-     * - MT_DOP > 0 is not supported for plans with base table joins or table sinks.
-     * Throws a NotImplementedException if plan validation fails.
-     */
-    public void validatePlan(PlanNode planNode) throws NotImplementedException {
-      if (ctx_.getQueryOptions().isSetMt_dop() && ctx_.getQueryOptions().mt_dop > 0
-          && (planNode instanceof HashJoinNode || planNode instanceof CrossJoinNode)) {
-          throw new NotImplementedException(
-              "MT_DOP not supported for plans with base table joins or table sinks.");
-      }
-
-      // As long as MT_DOP is unset or 0 any join can run in a single-node plan.
-      if (ctx_.isSingleNodeExec() &&
-          (!ctx_.getQueryOptions().isSetMt_dop() || ctx_.getQueryOptions().mt_dop == 0)) {
-          return;
-      }
-
-      for (PlanNode child : planNode.getChildren()) {
-          validatePlan(child);
-      }
-    }
-
     /**
      * Creates an EmptyNode that 'materializes' the tuples of the given stmt.
      * Marks all collection-typed slots referenced in stmt as non-materialized because