Parallel fragment exec instance

fe/src/main/java/org/apache/doris/common/Config.java

@@ -522,6 +522,20 @@ public class Config extends ConfigBase {
     @ConfField(mutable = true)
     public static int query_colocate_join_memory_limit_penalty_factor = 8;
 
+    /*
+     * the parallel exec instance num for one Fragment in one BE
+     * 1 means disable this feature
+     */
+    @ConfField(mutable = true)
+    public static int parallel_fragment_exec_instance_num = 1;
+
+    /*
+     * if table scan data size larger than parallel_scan_data_size_threshold
+     * we will parallel the fragment exec instance
+     */
+    @ConfField(mutable = true)
+    public static long parallel_scan_data_size_threshold = 500 * 1024 * 1024L; //500M
+
     /*
      * co-location join is an experimental feature now.
      * Set to false if you know what it is and really want to use it.

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

@@ -149,11 +149,6 @@ public void createPlanFragments(StatementBase statment, Analyzer analyzer, TQuer
 
         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

@@ -158,15 +158,8 @@ public PlanNode createSingleNodePlan() throws UserException, AnalysisException {
      * 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)) {
+      if (ctx_.isSingleNodeExec()) {
           return;
       }