wuxueyang96 commented on PR #61184: URL: https://github.com/apache/doris/pull/61184#issuecomment-4030967841
> > > Hi @wuxueyang96 !
> > > Thanks for submit this pr, and I already submit the same feature in
#59006. But this feature will break local shuffle and compute the wrong result,
so I revert it in #60823. So we should refactor local shuffle first, then we
can support this feature.
> >
> >
> > I'm not certainly sure that the local bucket shuffle is same to this pr.
This pr want to eliminate shuffle actually no matter local bucket shuffle or
global shuffle.
>
> My PR includes the function of eliminating exchange under set operation,
because supporting bucket shuffle itself requires the other end to distribute
according to the stored hash algorithm: the base end does not need shuffle, and
if the other end does not meet the requirements, the other end needs to use
bucket shuffle. If both ends are colocated, then neither end needs to shuffle
because they both satisfy the distribution of storing hash algorithms. So my PR
is a superset of your PR, more abstract
Actually, I rebuild the code from bf2e1c2dda944e47a5e9bf34972ae772570ec1c0,
i don't think it take effects at the same scenario:
```
MySQL [(none)]> show frontends;
+-----------------------------------------+---------------+-------------+----------+-----------+---------+--------------------+----------+----------+------------+------+-------+-------------------+---------------------+---------------------+----------+--------+------------------------+------------------+-----------+
| Name | Host | EditLogPort |
HttpPort | QueryPort | RpcPort | ArrowFlightSqlPort | Role | IsMaster |
ClusterId | Join | Alive | ReplayedJournalId | LastStartTime |
LastHeartbeat | IsHelper | ErrMsg | Version |
CurrentConnected | LiveSince |
+-----------------------------------------+---------------+-------------+----------+-----------+---------+--------------------+----------+----------+------------+------+-------+-------------------+---------------------+---------------------+----------+--------+------------------------+------------------+-----------+
| fe_781cb7e1_a9c0_49ee_845f_9ffa707ddeeb | 10.37.114.244 | 9010 |
8030 | 9030 | 9020 | 8070 | FOLLOWER | true |
1202823493 | true | true | 384 | 2026-03-10 19:50:11 |
2026-03-10 20:15:32 | true | | doris-0.0.0-bf2e1c2dda | Yes
| NULL |
+-----------------------------------------+---------------+-------------+----------+-----------+---------+--------------------+----------+----------+------------+------+-------+-------------------+---------------------+---------------------+----------+--------+------------------------+------------------+-----------+
1 row in set (0.017 sec)
MySQL [(none)]> use test;
Reading table information for completion of table and column names
You can turn off this feature to get a quicker startup with -A
Database changed
MySQL [test]> explain select d0.sum_val, d1.val, d1.id from (
select sum(sum_val) as sum_val, id from (
( SELECT sum(val) as sum_val, id
from t2 group by id )
union all ( SELECT sum(val) as
sum_val, id from t3
group by id ) ) as l group by id )
as d0 right join ( select id, val from t1 )
as d1 on d0.id = d1.id order by d0.id;
+------------------------------------------------------------------------------+
| Explain String(Nereids Planner)
|
+------------------------------------------------------------------------------+
| PLAN FRAGMENT 0
|
| OUTPUT EXPRS:
|
| sum_val[#28]
|
| val[#29]
|
| id[#30]
|
| PARTITION: UNPARTITIONED
|
|
|
| HAS_COLO_PLAN_NODE: false
|
|
|
| VRESULT SINK
|
| MYSQL_PROTOCOL
|
|
|
| 14:VMERGING-EXCHANGE
|
| offset: 0
|
| final projections: sum_val[#27], val[#25], id[#24]
|
| final project output tuple id: 11
|
| distribute expr lists: id[#24]
|
|
|
| PLAN FRAGMENT 1
|
|
|
| PARTITION: HASH_PARTITIONED: id[#14]
|
|
|
| HAS_COLO_PLAN_NODE: false
|
|
|
| STREAM DATA SINK
|
| EXCHANGE ID: 14
|
| UNPARTITIONED
|
| IS_MERGE: true
|
|
|
| 13:VSORT(587)
|
| | order by: id[#26] ASC
|
| | algorithm: full sort
|
| | local merge sort
|
| | merge by exchange
|
| | offset: 0
|
| | distribute expr lists:
|
| |
|
| 12:VHASH JOIN(583)
|
| | join op: LEFT OUTER JOIN(BROADCAST)[]
|
| | equal join conjunct: (id[#14] = id[#12])
|
| | cardinality=20
|
| | vec output tuple id: 9
|
| | output tuple id: 9
|
| | vIntermediate tuple ids: 8
|
| | hash output slot ids: 12 13 14 15
|
| | final projections: id[#16], val[#17], id[#18], sum_val[#19]
|
| | final project output tuple id: 9
|
| | distribute expr lists: id[#14]
|
| | distribute expr lists:
|
| |
|
| |----10:VEXCHANGE
|
| | offset: 0
|
| | distribute expr lists: id[#12]
|
| |
|
| 11:VOlapScanNode(541)
|
| TABLE: test.t1(t1), PREAGGREGATION: ON
|
| partitions=1/1 (t1)
|
| tablets=10/10, tabletList=1773143411383,1773143411386,1773143411389
... |
| cardinality=20, avgRowSize=1952.0, numNodes=2
|
| pushAggOp=NONE
|
|
|
| PLAN FRAGMENT 2
|
|
|
| PARTITION: HASH_PARTITIONED: id[#10]
|
|
|
| HAS_COLO_PLAN_NODE: true
|
|
|
| STREAM DATA SINK
|
| EXCHANGE ID: 10
|
| UNPARTITIONED
|
|
|
| 9:VAGGREGATE (merge finalize)(575)
|
| | output: sum(partial_sum(sum_val)[#11])[#13]
|
| | group by: id[#10]
|
| | sortByGroupKey:false
|
| | cardinality=10
|
| | distribute expr lists: id[#10]
|
| |
|
| 8:VEXCHANGE
|
| offset: 0
|
| distribute expr lists:
|
|
|
| PLAN FRAGMENT 3
|
|
|
| PARTITION: RANDOM
|
|
|
| HAS_COLO_PLAN_NODE: false
|
|
|
| STREAM DATA SINK
|
| EXCHANGE ID: 08
|
| HASH_PARTITIONED: id[#10]
|
|
|
| 7:VAGGREGATE (update serialize)(567)
|
| | STREAMING
|
| | output: partial_sum(sum_val[#9])[#11]
|
| | group by: id[#8]
|
| | sortByGroupKey:false
|
| | cardinality=10
|
| | distribute expr lists:
|
| |
|
| 6:VUNION(563)
|
| | distribute expr lists:
|
| | distribute expr lists:
|
| |
|
| |----5:VEXCHANGE
|
| | offset: 0
|
| | distribute expr lists: id[#6]
|
| |
|
| 2:VEXCHANGE
|
| offset: 0
|
| distribute expr lists: id[#2]
|
|
|
| PLAN FRAGMENT 4
|
|
|
| PARTITION: HASH_PARTITIONED: id[#4]
|
|
|
| HAS_COLO_PLAN_NODE: true
|
|
|
| STREAM DATA SINK
|
| EXCHANGE ID: 05
|
| RANDOM
|
|
|
| 4:VAGGREGATE (merge finalize)(555)
|
| | output: sum(val[#5])[#7]
|
| | group by: id[#4]
|
| | sortByGroupKey:false
|
| | cardinality=10
|
| | distribute expr lists: id[#4]
|
| |
|
| 3:VOlapScanNode(551)
|
| TABLE: test.t3(t3), PREAGGREGATION: ON
|
| partitions=1/1 (t3)
|
| tablets=10/10, tabletList=1773143411452,1773143411455,1773143411458
... |
| cardinality=20, avgRowSize=1952.0, numNodes=2
|
| pushAggOp=NONE
|
|
|
| PLAN FRAGMENT 5
|
|
|
| PARTITION: HASH_PARTITIONED: id[#0]
|
|
|
| HAS_COLO_PLAN_NODE: true
|
|
|
| STREAM DATA SINK
|
| EXCHANGE ID: 02
|
| RANDOM
|
|
|
| 1:VAGGREGATE (merge finalize)(546)
|
| | output: sum(val[#1])[#3]
|
| | group by: id[#0]
|
| | sortByGroupKey:false
|
| | cardinality=10
|
| | distribute expr lists: id[#0]
|
| |
|
| 0:VOlapScanNode(542)
|
| TABLE: test.t2(t2), PREAGGREGATION: ON
|
| partitions=1/1 (t2)
|
| tablets=10/10, tabletList=1773143411417,1773143411420,1773143411423
... |
| cardinality=20, avgRowSize=1952.0, numNodes=2
|
| pushAggOp=NONE
|
|
|
|
|
|
|
| ========== STATISTICS ==========
|
+------------------------------------------------------------------------------+
163 rows in set (0.019 sec)
```
--
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]
