[Exec](performance) change colocate execution parallel num (#60677)

Previously, the parallelism for colocate execution was calculated based
on bucket count, which might not fully utilize available resources. This
PR changes the parallelism calculation logic to:

1. Use tablet count instead of bucket count for more accurate resource
utilization
2. Add a new session variable colocate_max_parallel_num to control the
maximum parallelism
3. Calculate parallelism as min(max(tablet_num,
fragment.getParallelExecNum()), colocate_max_parallel_num) to ensure
both resource utilization and system stability

Author: HappenLee

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/distribute/worker/job/UnassignedScanBucketOlapTableJob.java

@@ -505,4 +505,40 @@ private List<DistributedPlanWorker> getWorkersByReplicas(Tablet tablet, long cat
         }
         return workers;
     }
+
+    @Override
+    protected int degreeOfParallelism(int maxParallel, boolean useLocalShuffleToAddParallel) {
+        Preconditions.checkArgument(maxParallel > 0, "maxParallel must be positive");
+        if (!fragment.getDataPartition().isPartitioned()) {
+            return 1;
+        }
+        if (fragment.queryCacheParam != null) {
+            return maxParallel;
+        }
+        if (scanNodes.size() == 1 && scanNodes.get(0) instanceof OlapScanNode) {
+            OlapScanNode olapScanNode = (OlapScanNode) scanNodes.get(0);
+            ConnectContext connectContext = statementContext.getConnectContext();
+            if (connectContext != null && olapScanNode.shouldUseOneInstance(connectContext)) {
+                return 1;
+            }
+        }
+
+        long tabletNum = 0;
+        for (ScanNode scanNode : scanNodes) {
+            if (scanNode instanceof OlapScanNode) {
+                OlapScanNode olapScanNode = (OlapScanNode) scanNode;
+                tabletNum = olapScanNode.getTotalTabletsNum();
+                break;
+            }
+        }
+
+        ConnectContext connectContext = statementContext.getConnectContext();
+        int colocateMaxParallelNum = 128;
+        if (connectContext != null) {
+            colocateMaxParallelNum = connectContext.getSessionVariable().colocateMaxParallelNum;
+        }
+
+        int maxParallelism = (int) Math.max(tabletNum, fragment.getParallelExecNum());
+        return Math.min(maxParallelism, colocateMaxParallelNum);
+    }
 }

fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java

@@ -168,6 +168,7 @@ public class SessionVariable implements Serializable, Writable {
             "enable_distinct_streaming_agg_force_passthrough";
     public static final String ENABLE_BROADCAST_JOIN_FORCE_PASSTHROUGH = "enable_broadcast_join_force_passthrough";
     public static final String DISABLE_COLOCATE_PLAN = "disable_colocate_plan";
+    public static final String COLOCATE_MAX_PARALLEL_NUM = "colocate_max_parallel_num";
     public static final String ENABLE_BUCKET_SHUFFLE_JOIN = "enable_bucket_shuffle_join";
     public static final String PARALLEL_FRAGMENT_EXEC_INSTANCE_NUM = "parallel_fragment_exec_instance_num";
     public static final String PARALLEL_PIPELINE_TASK_NUM = "parallel_pipeline_task_num";
@@ -1301,6 +1302,9 @@ public void checkQuerySlotCount(String slotCnt) {
                         setter = "setFragmentInstanceNum", varType = VariableAnnotation.DEPRECATED)
     public int parallelExecInstanceNum = 8;
 
+    @VariableMgr.VarAttr(name = COLOCATE_MAX_PARALLEL_NUM, needForward = true, fuzzy = false)
+    public int colocateMaxParallelNum = 128;
+
     @VariableMgr.VarAttr(name = PARALLEL_PIPELINE_TASK_NUM, fuzzy = true, needForward = true,
                         setter = "setPipelineTaskNum")
     public int parallelPipelineTaskNum = 0;

fe/fe-core/src/test/java/org/apache/doris/qe/LocalShuffleWithBucketJoinTest.java

@@ -67,7 +67,7 @@ public void test() throws Exception {
         PipelineDistributedPlan distributedPlan
                 = (PipelineDistributedPlan) distributedPlans.get(0);
         List<AssignedJob> instances = distributedPlan.getInstanceJobs();
-        Assertions.assertEquals(beNum * parallelPipelineTaskNum, instances.size());
+        Assertions.assertEquals(beNum * bucketNum, instances.size());
         Assertions.assertTrue(instances.stream().allMatch(LocalShuffleBucketJoinAssignedJob.class::isInstance));
 
         long assignedBucketInstanceNum = instances.stream().map(LocalShuffleBucketJoinAssignedJob.class::cast)
@@ -80,7 +80,7 @@ public void test() throws Exception {
             workerToInstances.put(instance.getAssignedWorker(), instance);
         }
         for (Collection<AssignedJob> instancePerBe : workerToInstances.asMap().values()) {
-            Assertions.assertEquals(parallelPipelineTaskNum, instancePerBe.size());
+            Assertions.assertEquals(bucketNum, instancePerBe.size());
         }
     }
 }