[opt](Nereids) optimize DistributeSpec generator of OlapScan (#15965)

use the size of selected partitions instead of olap table partition size to decide whether generate hashDistributeSpec

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/implementation/LogicalOlapScanToPhysicalOlapScan.java

@@ -48,17 +48,17 @@ public class LogicalOlapScanToPhysicalOlapScan extends OneImplementationRuleFact
     @Override
     public Rule build() {
         return logicalOlapScan().then(olapScan ->
-            new PhysicalOlapScan(
-                    olapScan.getId(),
-                    olapScan.getTable(),
-                    olapScan.getQualifier(),
-                    olapScan.getSelectedIndexId(),
-                    olapScan.getSelectedTabletIds(),
-                    olapScan.getSelectedPartitionIds(),
-                    convertDistribution(olapScan),
-                    olapScan.getPreAggStatus(),
-                    Optional.empty(),
-                    olapScan.getLogicalProperties())
+                new PhysicalOlapScan(
+                        olapScan.getId(),
+                        olapScan.getTable(),
+                        olapScan.getQualifier(),
+                        olapScan.getSelectedIndexId(),
+                        olapScan.getSelectedTabletIds(),
+                        olapScan.getSelectedPartitionIds(),
+                        convertDistribution(olapScan),
+                        olapScan.getPreAggStatus(),
+                        Optional.empty(),
+                        olapScan.getLogicalProperties())
         ).toRule(RuleType.LOGICAL_OLAP_SCAN_TO_PHYSICAL_OLAP_SCAN_RULE);
     }
 
@@ -66,10 +66,14 @@ public class LogicalOlapScanToPhysicalOlapScan extends OneImplementationRuleFact
         OlapTable olapTable = olapScan.getTable();
         DistributionInfo distributionInfo = olapTable.getDefaultDistributionInfo();
         ColocateTableIndex colocateTableIndex = Env.getCurrentColocateIndex();
-        if ((colocateTableIndex.isColocateTable(olapTable.getId())
-                && !colocateTableIndex.isGroupUnstable(colocateTableIndex.getGroup(olapTable.getId())))
-                || olapTable.getPartitionInfo().getType() == PartitionType.UNPARTITIONED
-                || olapTable.getPartitions().size() == 1) {
+        // When there are multiple partitions, olapScan tasks of different buckets are dispatched in
+        // rounded robin algorithm. Therefore, the hashDistributedSpec can be broken except they are in
+        // the same stable colocateGroup(CG)
+        boolean isBelongStableCG = colocateTableIndex.isColocateTable(olapTable.getId())
+                && !colocateTableIndex.isGroupUnstable(colocateTableIndex.getGroup(olapTable.getId()));
+        boolean isSelectUnpartition = olapTable.getPartitionInfo().getType() == PartitionType.UNPARTITIONED
+                || olapScan.getSelectedPartitionIds().size() == 1;
+        if (isBelongStableCG || isSelectUnpartition) {
             if (!(distributionInfo instanceof HashDistributionInfo)) {
                 return DistributionSpecAny.INSTANCE;
             }

regression-test/data/query_p0/join/bucket_shuffle_join.out

@@ -1,5 +1,5 @@
 -- This file is automatically generated. You should know what you did if you want to edit this
--- !bucket_shuffle_join --
+-- !test_bucket --
 1	2021-12-01T00:00
 2	2021-12-01T00:00
 

regression-test/suites/query_p0/join/bucket_shuffle_join.groovy

@@ -0,0 +1,22 @@
+// 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.
+
+suite("bucket-shuffle-join") {
+    order_qt_test_bucket """
+    select * from test_bucket_shuffle_join where rectime="2021-12-01 00:00:00" and id in (select k1 from test_join where k1 in (1,2))
+    """
+}

regression-test/suites/query_p0/join/sql/bucket_shuffle_join.sql

@@ -1 +0,0 @@
-select * from test_bucket_shuffle_join where rectime="2021-12-01 00:00:00" and id in (select k1 from test_join where k1 in (1,2))