github-actions[bot] commented on code in PR #65129:
URL: https://github.com/apache/doris/pull/65129#discussion_r3568347208
##########
regression-test/suites/query_p0/set_operations/bucket_shuffle_set_operation.groovy:
##########
@@ -95,6 +95,208 @@ 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
+ (select id from bucket_shuffle_set_operation1 intersect select id
from bucket_shuffle_set_operation_r1) t
+ join
+ (select id from bucket_shuffle_set_operation1 intersect select id
from bucket_shuffle_set_operation_r2) s
+ on t.id = s.id"""
+
+ // A non-intersect bucket-shuffle set operation (UNION ALL) whose basic /
anchor child is a
+ // direct bucketed scan that is bucket-pruned by an IN predicate on the
distribution key, so
+ // it only scans a subset of the buckets. The other child is shuffled onto
the anchor's
+ // storage layout and has rows in the buckets the pruned anchor does not
scan. Because the
+ // union is a non-intersect bucket-shuffle set operation,
UnassignedScanBucketOlapTableJob
+ // must fill up receiver instances for those missing buckets; without the
fill-up the other
+ // child's rows in the missing buckets would have no destination instance
and be lost.
+ //
+ // The setup forces the pruned scan to be the anchor: fill_anchor has huge
injected stats so
+ // the IN-filtered branch still wins the largest-row-count basic-child
selection, while
+ // fill_spread has a mismatched bucket count (11 vs 10) so it cannot be
the natural anchor and
+ // must be shuffled. The bucket-shuffle join above the union supplies the
parent hash request
+ // that makes the union choose bucket shuffle, and the join probe
fill_probe has yet another
+ // bucket count (7) so the join is a real shuffle in its own fragment and
does not co-locate a
+ // full-bucket scan into the union fragment (which would otherwise cover
the missing buckets
+ // and hide the fill-up).
+ sql "drop table if exists bucket_shuffle_set_operation_fill_anchor"
+ sql """create table bucket_shuffle_set_operation_fill_anchor(id int)
+ distributed by hash(id) buckets 10
properties('replication_num'='1')"""
+ sql "insert into bucket_shuffle_set_operation_fill_anchor select number
from numbers('number'='20')"
+ sql "drop table if exists bucket_shuffle_set_operation_fill_spread"
+ sql """create table bucket_shuffle_set_operation_fill_spread(id int)
+ distributed by hash(id) buckets 11
properties('replication_num'='1')"""
+ sql "insert into bucket_shuffle_set_operation_fill_spread select number
from numbers('number'='20')"
+ sql "drop table if exists bucket_shuffle_set_operation_fill_probe"
+ sql """create table bucket_shuffle_set_operation_fill_probe(id int)
+ distributed by hash(id) buckets 7
properties('replication_num'='1')"""
+ sql "insert into bucket_shuffle_set_operation_fill_probe select number
from numbers('number'='20')"
+ sql """alter table bucket_shuffle_set_operation_fill_anchor modify column
id
+ set stats ('row_count'='1000000', 'ndv'='20', 'min_value'='0',
'max_value'='19')"""
+ sql """alter table bucket_shuffle_set_operation_fill_spread modify column
id
+ set stats ('row_count'='50', 'ndv'='20', 'min_value'='0',
'max_value'='19')"""
+ sql """alter table bucket_shuffle_set_operation_fill_probe modify column id
+ set stats ('row_count'='30', 'ndv'='20', 'min_value'='0',
'max_value'='19')"""
+ def bucketShuffleUnionFillUpSql = """
+ select t.id from (
+ select id from bucket_shuffle_set_operation_fill_anchor where id
in (0, 2, 4, 6)
+ union all
+ select id from bucket_shuffle_set_operation_fill_spread
+ ) t join[shuffle] bucket_shuffle_set_operation_fill_probe c on t.id =
c.id"""
+ explain {
+ sql "shape plan " + bucketShuffleUnionFillUpSql
+ check { String e ->
+ assertTrue(e.contains("PhysicalUnion[bucketShuffle]"))
+ }
+ }
+ order_qt_bucket_shuffle_union_fill_up bucketShuffleUnionFillUpSql
+
+ // Same missing-bucket fill-up contract, but the pruned basic child
exposes its storage bucket
+ // key only through an equivalent slot: the union's basic child is a
bucket-shuffle join output
+ // that projects the join key bucket_shuffle_set_operation2.id AS k, so
the storage bucket column
+ // (fill_anchor.id) is hidden and only the equivalent k is visible in the
set-operation output.
+ // The alignment proof must resolve the bucket key through its hash
equivalence set (mirroring
+ // ChildrenPropertiesRegulator.canMapBucketKeysToRequire); a direct ExprId
lookup would report
+ // the union as not bucket-aligned, drop the BUCKET_SHUFFLE marker, and
skip
+ // UnassignedScanBucketOlapTableJob.fillUpInstances(), losing the shuffled
side's rows in the
+ // buckets the pruned basic child does not scan.
+ def bucketShuffleEquivalentKeyFillUpSql = """
Review Comment:
The code now has a guard for the earlier `(k, v)` case where the candidate
basic child is bucketed by more columns than the set operation outputs, but
this suite still never exercises it: all new tables here are `distributed by
hash(id)`, and the equivalent-key cases below still use a single storage bucket
key. Without a committed case like a table `distributed by hash(k, v)` feeding
`INTERSECT(k)` (or another set op on only `k`) and asserting the plan falls
back to execution hash while preserving the result, removing
`canMapBucketKeysToRequire()` would reintroduce the planning failure without a
regression failure. Please add that shape/result coverage rather than leaving
this as only a manually verified edge case.
--
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]
