diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java
index 581c2418a3..b51bb7a700 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java
@@ -1553,6 +1553,7 @@ public class PhysicalPlanTranslator extends DefaultPlanVisitor<PlanFragment, Pla
         // according to left partition to generate right partition expr list
         DistributionSpecHash leftDistributionSpec
                 = (DistributionSpecHash) physicalHashJoin.left().getPhysicalProperties().getDistributionSpec();
+
         Pair<List<ExprId>, List<ExprId>> onClauseUsedSlots = JoinUtils.getOnClauseUsedSlots(physicalHashJoin);
         List<ExprId> rightPartitionExprIds = Lists.newArrayList(leftDistributionSpec.getOrderedShuffledColumns());
         for (int i = 0; i < leftDistributionSpec.getOrderedShuffledColumns().size(); i++) {
@@ -1572,11 +1573,14 @@ public class PhysicalPlanTranslator extends DefaultPlanVisitor<PlanFragment, Pla
         }
         // assemble fragment
         hashJoinNode.setDistributionMode(HashJoinNode.DistributionMode.BUCKET_SHUFFLE);
-        if (leftDistributionSpec.getShuffleType() != ShuffleType.NATURAL) {
+        if (leftDistributionSpec.getShuffleType() != ShuffleType.NATURAL
+                && leftDistributionSpec.getShuffleType() != ShuffleType.BUCKETED) {
             hashJoinNode.setDistributionMode(DistributionMode.PARTITIONED);
         }
         connectChildFragment(hashJoinNode, 1, leftFragment, rightFragment, context);
         leftFragment.setPlanRoot(hashJoinNode);
+        // HASH_PARTITIONED and BUCKET_SHFFULE_HASH_PARTITIONED are two type of hash algorithm
+        // And the nature left child means it use BUCKET_SHFFULE_HASH_PARTITIONED in storage layer
         TPartitionType partitionType = TPartitionType.BUCKET_SHFFULE_HASH_PARTITIONED;
         if (leftDistributionSpec.getShuffleType() != ShuffleType.NATURAL) {
             partitionType = TPartitionType.HASH_PARTITIONED;
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/DistributionSpecHash.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/DistributionSpecHash.java
index 118f67d507..72bf7e34df 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/DistributionSpecHash.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/DistributionSpecHash.java
@@ -289,10 +289,13 @@ public class DistributionSpecHash extends DistributionSpec {
      * Enums for concrete shuffle type.
      */
     public enum ShuffleType {
+        // 1. The following properties are the required properties for children
         // require, need to satisfy the distribution spec by aggregation way.
         AGGREGATE,
         // require, need to satisfy the distribution spec by join way.
         JOIN,
+
+        // 2. The following properties are the output properties from some operators
         // output, for olap scan node and colocate join
         NATURAL,
         // output, for all join except colocate join
diff --git a/regression-test/suites/nereids_p0/join/test_join5.groovy b/regression-test/suites/nereids_p0/join/test_join5.groovy
index 0fd0d2a688..292e2679cc 100644
--- a/regression-test/suites/nereids_p0/join/test_join5.groovy
+++ b/regression-test/suites/nereids_p0/join/test_join5.groovy
@@ -119,36 +119,38 @@ suite("test_join5", "nereids_p0") {
     sql " insert into c (name, a) values ('A', 'p');"
     sql " insert into c (name, a) values ('B', 'q');"
     sql " insert into c (name, a) values ('C', null);"
+    
+    sql """set parallel_fragment_exec_instance_num=8"""
+        
+    qt_join5 """
+        select c.name, ss.code, ss.b_cnt, ss.const
+        from c left join
+          (select a.code, coalesce(b_grp.cnt, 0) as b_cnt, -1 as const
+           from a left join
+             (select count(1) as cnt, b.a from b group by b.a) as b_grp
+             on a.code = b_grp.a
+          ) as ss
+          on (c.a = ss.code)
+        order by c.name;
+        """
 
-    // qt_join5 """
-    //     select c.name, ss.code, ss.b_cnt, ss.const
-    //     from c left join
-    //       (select a.code, coalesce(b_grp.cnt, 0) as b_cnt, -1 as const
-    //        from a left join
-    //          (select count(1) as cnt, b.a from b group by b.a) as b_grp
-    //          on a.code = b_grp.a
-    //       ) as ss
-    //       on (c.a = ss.code)
-    //     order by c.name;
-    //     """
-
-    // qt_join5 """
-    //     SELECT * FROM
-    //         ( SELECT 1 as key1 ) sub1
-    //         LEFT JOIN
-    //         ( SELECT sub3.key3, sub4.value2, COALESCE(sub4.value2, 66) as value3 FROM
-    //             ( SELECT 1 as key3 ) sub3
-    //             LEFT JOIN
-    //             ( SELECT sub5.key5, COALESCE(sub6.value1, 1) as value2 FROM
-    //                 ( SELECT 1 as key5 ) sub5
-    //                 LEFT JOIN
-    //                 ( SELECT 2 as key6, 42 as value1 ) sub6
-    //                 ON sub5.key5 = sub6.key6
-    //             ) sub4
-    //             ON sub4.key5 = sub3.key3
-    //         ) sub2
-    //         ON sub1.key1 = sub2.key3;
-    //         """
+    qt_join5 """
+        SELECT * FROM
+            ( SELECT 1 as key1 ) sub1
+            LEFT JOIN
+            ( SELECT sub3.key3, sub4.value2, COALESCE(sub4.value2, 66) as value3 FROM
+                ( SELECT 1 as key3 ) sub3
+                LEFT JOIN
+                ( SELECT sub5.key5, COALESCE(sub6.value1, 1) as value2 FROM
+                    ( SELECT 1 as key5 ) sub5
+                    LEFT JOIN
+                    ( SELECT 2 as key6, 42 as value1 ) sub6
+                    ON sub5.key5 = sub6.key6
+                ) sub4
+                ON sub4.key5 = sub3.key3
+            ) sub2
+            ON sub1.key1 = sub2.key3;
+            """
 
     qt_join6 """
             SELECT * FROM