924060929 commented on code in PR #65129:
URL: https://github.com/apache/doris/pull/65129#discussion_r3568667134


##########
regression-test/suites/query_p0/set_operations/bucket_shuffle_set_operation.groovy:
##########
@@ -95,6 +95,235 @@ suite("bucket_shuffle_set_operation") {
         select id from bucket_shuffle_set_operation2 where id=1
         """)
 
+    // The basic child of a bucket-shuffle set operation can be a join output 
instead of a
+    // direct scan. In that shape the local exchange planned for the basic 
side must still
+    // partition by the storage bucket function: an execution-hash local 
exchange would not
+    // align with the bucket-distributed side and the set operation would 
compute wrong results.
+    checkShapeAndResult("bucket_shuffle_join_as_basic_child", """
+        select a.id from bucket_shuffle_set_operation1 a
+        join bucket_shuffle_set_operation2 b on a.id = b.id
+        intersect
+        select id from bucket_shuffle_set_operation3""")
+
+    // a set operation child can itself be a set operation whose output claims 
a bucket
+    // distribution; the outer set operation must only treat its children as 
bucket-aligned
+    // when they share the same storage layout
+    checkShapeAndResult("bucket_shuffle_nested_set_operation", """
+        select id from bucket_shuffle_set_operation3
+        union all
+        (select a.id from bucket_shuffle_set_operation1 a
+        join bucket_shuffle_set_operation2 b on a.id = b.id
+        intersect
+        select id from bucket_shuffle_set_operation2)""")
+
+    // when local shuffle is disabled entirely, every pipeline runs a single 
task per
+    // instance so the bucket alignment holds naturally and bucket shuffle is 
still allowed
+    sql "set enable_local_shuffle=false"
+    checkShapeAndResult("bucket_shuffle_when_local_shuffle_off", """
+        select id from bucket_shuffle_set_operation1
+        intersect
+        select id from bucket_shuffle_set_operation2""")
+    sql "set enable_local_shuffle=true"
+
+    // A shuffle join above the union pushes a hash request into the union
+    // (createHashRequestAccordingToParent, the parent-hash request path). 
When the FE does not
+    // plan the local shuffle, that request must be downgraded so the union 
does not choose
+    // bucket shuffle, while the result stays correct.
+    def unionParentHashSql = """
+        select b.id from (
+            select id from bucket_shuffle_set_operation1
+            union all
+            select id from bucket_shuffle_set_operation2
+        ) u join[shuffle] bucket_shuffle_set_operation3 b on u.id = b.id
+        """
+    sql "set enable_local_shuffle_planner=false"
+    // Golden shape: the union must stay a plain PhysicalUnion whose two 
children are each a
+    // PhysicalDistribute[DistributionSpecHash]. That is the actual proof that 
the parent hash
+    // request was pushed down into the union 
(createHashRequestAccordingToParent) and downgraded
+    // to execution hash, not merely that the PhysicalUnion line lacks the 
[bucketShuffle] tag.
+    // If that path regressed, the optimizer could instead keep an unbucketed 
union and add a
+    // single PhysicalDistribute[DistributionSpecHash] above it for the 
shuffle join; the golden
+    // shape below (union children are distributes, not direct scans) would 
catch that.
+    qt_union_parent_hash_shape_when_local_shuffle_planner_off("explain shape 
plan " + unionParentHashSql)
+    explain {
+        sql "shape plan " + unionParentHashSql
+        check { String e ->
+            def unionIndex = e.indexOf("PhysicalUnion")
+            assertTrue(unionIndex >= 0)
+            // the union must not be a bucket shuffle union when the FE local 
shuffle planner is off
+            assertFalse(e.substring(unionIndex,
+                    Math.min(unionIndex + "PhysicalUnion".length() + 20, 
e.length())).contains("bucketShuffle"))
+            // and the parent hash request must have been pushed into the 
union: each union child
+            // arrives through a PhysicalDistribute[DistributionSpecHash], so 
no union child is a
+            // direct scan.
+            def afterUnion = e.substring(unionIndex)
+            def joinProbeIndex = 
afterUnion.indexOf("bucket_shuffle_set_operation3")
+            def unionSubtree = joinProbeIndex >= 0 ? afterUnion.substring(0, 
joinProbeIndex) : afterUnion
+            assertEquals(2, 
unionSubtree.split("PhysicalDistribute\\[DistributionSpecHash\\]", -1).length - 
1)
+        }
+    }
+    order_qt_union_parent_hash_when_local_shuffle_planner_off 
unionParentHashSql
+    sql "set enable_local_shuffle_planner=true"
+
+    // A plain intersect/except without a parent hash request goes through
+    // visitPhysicalSetOperation directly (not the parent-hash request path); 
with the FE local
+    // shuffle planner disabled it must not choose bucket shuffle either, and 
the result stays
+    // correct.
+    sql "set enable_local_shuffle_planner=false"
+    explain {
+        sql "shape plan select id from bucket_shuffle_set_operation1 intersect 
select id from bucket_shuffle_set_operation2"
+        check { String e ->
+            assertFalse(e.contains("bucketShuffle"))
+        }
+    }
+    order_qt_plain_intersect_when_local_shuffle_planner_off "select id from 
bucket_shuffle_set_operation1 intersect select id from 
bucket_shuffle_set_operation2"
+    sql "set enable_local_shuffle_planner=true"
+
+    // The right child can be selected as the bucket-shuffle basic child 
(larger row count). The
+    // set operation output must then advertise a non-specific distribution 
rather than a plain
+    // execution hash: otherwise two such set operations with different 
storage layouts are
+    // co-located under a join and fail bucket assignment ("Can not find 
tablet ... in the
+    // bucket"). r1 / r2 are larger than the small tables so they become the 
basic child on the
+    // right, and they are different tables so their bucket layouts differ.
+    sql "drop table if exists bucket_shuffle_set_operation_r1"
+    sql "create table bucket_shuffle_set_operation_r1(id int) distributed by 
hash(id) buckets 10 properties('replication_num'='1')"
+    sql "insert into bucket_shuffle_set_operation_r1 select number from 
numbers('number'='20')"
+    sql "drop table if exists bucket_shuffle_set_operation_r2"
+    sql "create table bucket_shuffle_set_operation_r2(id int) distributed by 
hash(id) buckets 10 properties('replication_num'='1')"
+    sql "insert into bucket_shuffle_set_operation_r2 select number from 
numbers('number'='20')"
+    sql "alter table bucket_shuffle_set_operation_r1 modify column id set 
stats ('row_count'='1000', 'ndv'='1000', 'min_value'='0', 'max_value'='19')"
+    sql "alter table bucket_shuffle_set_operation_r2 modify column id set 
stats ('row_count'='1000', 'ndv'='1000', 'min_value'='0', 'max_value'='19')"
+    order_qt_intersect_right_basic_parent_hash """
+        select t.id from

Review Comment:
   Good point — added a shape assertion to `intersect_right_basic_parent_hash` 
(kept the ordered result). It now asserts each `INTERSECT` is 
`[bucketShuffle]`, that `r1` / `r2` (the larger, right side) are the direct 
bucketed basic children of the intersect (their parent in the shape tree is the 
`PhysicalIntersect[bucketShuffle]`, not a distribute), and that each intersect 
shuffles its left `op1` child through an enforced 
`PhysicalDistribute[DistributionSpecHash]` directly under it. So if the 
optimizer stopped selecting the right side as the basic child, the 
`distributeToChildIndex > 0` branch would no longer be covered and the 
assertion fails, even though the result stays `1, 2, 3`. Verified passing on a 
4-BE cluster.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: [email protected]

For queries about this service, please contact Infrastructure at:
[email protected]


---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]

Reply via email to