[multistage] mailbox receiving stage not correctly created #11880
Description
running the following with colocated join
explain implementation plan for
-- explain plan for
WITH tmp2 AS (
SELECT * FROM userGroups
WHERE groupUUID NOT IN ('group-3', 'group-4')
),
tmp3 AS (
SELECT *
FROM userAttributes
WHERE userUUID IN (SELECT userUUID FROM tmp2)
AND totalTrips < 100
)
SELECT * FROM tmp3 WHERE userUUID IN (SELECT userUUID from tmp2)
produces this physical plan:
[0]@100.114.228.138:50723 MAIL_RECEIVE(BROADCAST_DISTRIBUTED)
├── [1]@100.114.228.138:50752 MAIL_SEND(BROADCAST_DISTRIBUTED)->{[0]@100.114.228.138@{50723,50723}|[0]} (Subtree Omitted)
├── [1]@100.114.228.138:50746 MAIL_SEND(BROADCAST_DISTRIBUTED)->{[0]@100.114.228.138@{50723,50723}|[0]} (Subtree Omitted)
├── [1]@100.114.228.138:50737 MAIL_SEND(BROADCAST_DISTRIBUTED)->{[0]@100.114.228.138@{50723,50723}|[0]} (Subtree Omitted)
└── [1]@100.114.228.138:50731 MAIL_SEND(BROADCAST_DISTRIBUTED)->{[0]@100.114.228.138@{50723,50723}|[0]}
└── [1]@100.114.228.138:50731 JOIN
├── [1]@100.114.228.138:50731 MAIL_RECEIVE(HASH_DISTRIBUTED)
│ ├── [2]@100.114.228.138:50752 MAIL_SEND(HASH_DISTRIBUTED)->{[1]@100.114.228.138@{50752,50753}|[3],[1]@100.114.228.138@{50746,50747}|[2],[1]@100.114.228.138@{50737,50738}|[1],[1]@100.114.228.138@{50731,50732}|[0]} (Subtree Omitted)
│ ├── [2]@100.114.228.138:50746 MAIL_SEND(HASH_DISTRIBUTED)->{[1]@100.114.228.138@{50752,50753}|[3],[1]@100.114.228.138@{50746,50747}|[2],[1]@100.114.228.138@{50737,50738}|[1],[1]@100.114.228.138@{50731,50732}|[0]} (Subtree Omitted)
│ ├── [2]@100.114.228.138:50737 MAIL_SEND(HASH_DISTRIBUTED)->{[1]@100.114.228.138@{50752,50753}|[3],[1]@100.114.228.138@{50746,50747}|[2],[1]@100.114.228.138@{50737,50738}|[1],[1]@100.114.228.138@{50731,50732}|[0]} (Subtree Omitted)
│ └── [2]@100.114.228.138:50731 MAIL_SEND(HASH_DISTRIBUTED)->{[1]@100.114.228.138@{50752,50753}|[3],[1]@100.114.228.138@{50746,50747}|[2],[1]@100.114.228.138@{50737,50738}|[1],[1]@100.114.228.138@{50731,50732}|[0]}
------ This receiving stage should indicate a receiving stage of [2] not [1] ------
│ └── [1]@100.114.228.138:50731 MAIL_RECEIVE(HASH_DISTRIBUTED).
│ ├── [3]@100.114.228.138:50746 MAIL_SEND(HASH_DISTRIBUTED)->{[1]@100.114.228.138@{50752,50753}|[3],[1]@100.114.228.138@{50746,50747}|[2],[1]@100.114.228.138@{50737,50738}|[1],[1]@100.114.228.138@{50731,50732}|[0]}
│ │ └── [3]@100.114.228.138:50746 JOIN
│ │ ├── [3]@100.114.228.138:50746 PROJECT
│ │ │ └── [3]@100.114.228.138:50746 FILTER
│ │ │ └── [3]@100.114.228.138:50746 TABLE SCAN (userAttributes) null
│ │ └── [3]@100.114.228.138:50746 MAIL_RECEIVE(BROADCAST_DISTRIBUTED)
│ │ ├── [4]@100.114.228.138:50746 MAIL_SEND(BROADCAST_DISTRIBUTED)->{[3]@100.114.228.138@{50746,50747}|[1],[3]@100.114.228.138@{50731,50732}|[0]}
│ │ │ └── [4]@100.114.228.138:50746 PROJECT
│ │ │ └── [4]@100.114.228.138:50746 FILTER
│ │ │ └── [4]@100.114.228.138:50746 TABLE SCAN (userGroups) null
│ │ └── [4]@100.114.228.138:50731 MAIL_SEND(BROADCAST_DISTRIBUTED)->{[3]@100.114.228.138@{50746,50747}|[1],[3]@100.114.228.138@{50731,50732}|[0]}
│ │ └── [4]@100.114.228.138:50731 PROJECT
│ │ └── [4]@100.114.228.138:50731 FILTER
│ │ └── [4]@100.114.228.138:50731 TABLE SCAN (userGroups) null
│ └── [3]@100.114.228.138:50731 MAIL_SEND(HASH_DISTRIBUTED)->{[1]@100.114.228.138@{50752,50753}|[3],[1]@100.114.228.138@{50746,50747}|[2],[1]@100.114.228.138@{50737,50738}|[1],[1]@100.114.228.138@{50731,50732}|[0]}
│ └── [3]@100.114.228.138:50731 JOIN
│ ├── [3]@100.114.228.138:50731 PROJECT
│ │ └── [3]@100.114.228.138:50731 FILTER
│ │ └── [3]@100.114.228.138:50731 TABLE SCAN (userAttributes) null
│ └── [3]@100.114.228.138:50731 MAIL_RECEIVE(BROADCAST_DISTRIBUTED)
│ ├── [4]@100.114.228.138:50746 MAIL_SEND(BROADCAST_DISTRIBUTED)->{[3]@100.114.228.138@{50746,50747}|[1],[3]@100.114.228.138@{50731,50732}|[0]}
│ │ └── [4]@100.114.228.138:50746 PROJECT
│ │ └── [4]@100.114.228.138:50746 FILTER
│ │ └── [4]@100.114.228.138:50746 TABLE SCAN (userGroups) null
│ └── [4]@100.114.228.138:50731 MAIL_SEND(BROADCAST_DISTRIBUTED)->{[3]@100.114.228.138@{50746,50747}|[1],[3]@100.114.228.138@{50731,50732}|[0]}
│ └── [4]@100.114.228.138:50731 PROJECT
│ └── [4]@100.114.228.138:50731 FILTER
│ └── [4]@100.114.228.138:50731 TABLE SCAN (userGroups) null
└── [1]@100.114.228.138:50731 MAIL_RECEIVE(HASH_DISTRIBUTED)
├── [5]@100.114.228.138:50746 MAIL_SEND(HASH_DISTRIBUTED)->{[1]@100.114.228.138@{50752,50753}|[3],[1]@100.114.228.138@{50746,50747}|[2],[1]@100.114.228.138@{50737,50738}|[1],[1]@100.114.228.138@{50731,50732}|[0]}
│ └── [5]@100.114.228.138:50746 PROJECT
│ └── [5]@100.114.228.138:50746 FILTER
│ └── [5]@100.114.228.138:50746 TABLE SCAN (userGroups) null
└── [5]@100.114.228.138:50731 MAIL_SEND(HASH_DISTRIBUTED)->{[1]@100.114.228.138@{50752,50753}|[3],[1]@100.114.228.138@{50746,50747}|[2],[1]@100.114.228.138@{50737,50738}|[1],[1]@100.114.228.138@{50731,50732}|[0]}
└── [5]@100.114.228.138:50731 PROJECT
└── [5]@100.114.228.138:50731 FILTER
└── [5]@100.114.228.138:50731 TABLE SCAN (userGroups) null
logical plan
Execution Plan
LogicalJoin(condition=[=($3, $5)], joinType=[semi])
PinotLogicalExchange(distribution=[hash[3]]) <-- THIS should be the right exchange
PinotLogicalExchange(distribution=[hash[3]]) <-- THIS exchange is redundant
LogicalJoin(condition=[=($3, $5)], joinType=[semi])
LogicalProject(daysSinceFirstTrip=[$3], deviceOS=[$4], totalTrips=[$5], userUUID=[$6])
LogicalFilter(condition=[<($5, 100)])
LogicalTableScan(table=[[userAttributes]])
PinotLogicalExchange(distribution=[broadcast], relExchangeType=[PIPELINE_BREAKER])
LogicalProject(groupUUID=[$3], userUUID=[$4])
LogicalFilter(condition=[AND(<>($3, 'group-3'), <>($3, 'group-4'))])
LogicalTableScan(table=[[userGroups]])
PinotLogicalExchange(distribution=[hash[1]])
LogicalProject(groupUUID=[$3], userUUID=[$4])
LogicalFilter(condition=[AND(<>($3, 'group-3'), <>($3, 'group-4'))])
LogicalTableScan(table=[[userGroups]])