martint
commented
2 years ago What version are you running? I'm not able to reproduce the problem with the query you provided above:
explain
select l.orderkey from (select orderkey from tpch.tiny.lineitem union all select * from (values 1) v(orderkey) ) l
join
tpch.tiny.orders o on l.orderkey = o.orderkey; Query Plan
-------------------------------------------------------------------------------------------------------------------------------------------------------
Fragment 0 [SINGLE]
Output layout: [orderkey]
Output partitioning: SINGLE []
Stage Execution Strategy: UNGROUPED_EXECUTION
Output[orderkey]
│ Layout: [orderkey:bigint]
│ Estimates: {rows: 60176 (528.89kB), cpu: 5.81M, memory: 263.67kB, network: 792.58kB}
└─ RemoteSource[1]
Layout: [orderkey:bigint]
Fragment 1 [tpch:orders:15000]
Output layout: [orderkey]
Output partitioning: SINGLE []
Stage Execution Strategy: UNGROUPED_EXECUTION
InnerJoin[("orderkey" = "orderkey_3")][$hashvalue, $hashvalue_8]
│ Layout: [orderkey:bigint]
│ Estimates: {rows: 60176 (528.89kB), cpu: 5.81M, memory: 263.67kB, network: 263.69kB}
│ Distribution: PARTITIONED
│ dynamicFilterAssignments = {orderkey_3 -> #df_395}
├─ LocalExchange[ROUND_ROBIN] ()
│ │ Layout: [orderkey:bigint, $hashvalue:bigint]
│ │ Estimates: {rows: 60176 (1.03MB), cpu: 3.10M, memory: 0B, network: 18B}
│ ├─ ScanFilterProject[table = tpch:lineitem:sf0.01, grouped = false, filterPredicate = true, dynamicFilters = {"orderkey_0" = #df_395}]
│ │ Layout: [orderkey_0:bigint, $hashvalue_5:bigint]
│ │ Estimates: {rows: 60175 (1.03MB), cpu: 528.88k, memory: 0B, network: 0B}/{rows: 60175 (1.03MB), cpu: 1.03M, memory: 0B, network: 0B}/{ro
│ │ $hashvalue_5 := combine_hash(bigint '0', COALESCE("$operator$hash_code"("orderkey_0"), 0))
│ │ orderkey_0 := tpch:orderkey
│ └─ RemoteSource[2]
│ Layout: [field_1:bigint, $hashvalue_6:bigint]
└─ LocalExchange[HASH][$hashvalue_8] ("orderkey_3")
│ Layout: [orderkey_3:bigint, $hashvalue_8:bigint]
│ Estimates: {rows: 15000 (263.67kB), cpu: 922.85k, memory: 0B, network: 263.67kB}
└─ RemoteSource[3]
Fragment 2 [SINGLE]
Output layout: [field_1, $hashvalue_7]
Output partitioning: tpch:orders:15000 [field_1]
Stage Execution Strategy: UNGROUPED_EXECUTION
Project[]
│ Layout: [field_1:bigint, $hashvalue_7:bigint]
│ Estimates: {rows: 1 (18B), cpu: 18, memory: 0B, network: 0B}
│ $hashvalue_7 := combine_hash(bigint '0', COALESCE("$operator$hash_code"("field_1"), 0))
└─ Values
Layout: [field_1:bigint]
Estimates: {rows: 1 (9B), cpu: 0, memory: 0B, network: 0B}
(BIGINT '1')
Fragment 3 [tpch:orders:15000]
Output layout: [orderkey_3, $hashvalue_10]
Output partitioning: tpch:orders:15000 [orderkey_3]
Stage Execution Strategy: UNGROUPED_EXECUTION
ScanProject[table = tpch:orders:sf0.01, grouped = false]
Layout: [orderkey_3:bigint, $hashvalue_10:bigint]
Estimates: {rows: 15000 (263.67kB), cpu: 131.84k, memory: 0B, network: 0B}/{rows: 15000 (263.67kB), cpu: 395.51k, memory: 0B, network: 0B}
$hashvalue_10 := combine_hash(bigint '0', COALESCE("$operator$hash_code"("orderkey_3"), 0))
orderkey_3 := tpch:orderkey
tpch:orderstatus
:: [[F], [O], [P]]
b-slim
Issue:
The CBO produces a sub-optimal plan (Deadly in our case) leading to schedule a Join as a Single node join thus query fail with resource exhaustion (Join of billion of rows against a small table).
How does this happen?
AddExchangesold optimizer rewrites a union node betweenSingleNode with empty values and a set of partitioned splits it produces as a single local union node. This is by design not sure why TBH.How to reproduce it We can reproduce the issue using the following statement.
Now you can push the CBO to pick a better plan in absence of stats aka avoid doing a broadcast join.
Possible fixes for the issue:
Consider adding a cost per explored plan and choose the best one based on the overall plan cost. I can not find a way how the Join costing/selection does this perhaps This fix is not possible since Trino CBO does not keep track of the cost (memo) of other explored plans when searching. Adding such a fix to Trino seems to me a major work thus this option is not explored for now. I would appreciate any pointer.
Adding a minor patch to skip empty unions when the AddExchanges fires on the union node. FYI If you wonder why the
RemoveEmptyUnionBranchesdoes not prune such an empty single node is because of the empty values are produced later on when the filter predicate is pushed into the table scan. This will fix our current issue when the union has an empty value but will not fix the general case as pointed by the query aboveAdding an arbitrary distribution when we have a mix of un-partitioned and partitioned unions node, seems like it is avoided by design, I am missing some context please feel me in if possible.
Fire again the Iterative optimizer, seems like this can be an overkill see this thread
Testing:
Plan Join with Single node