Zoltán Borók-Nagy created IMPALA-15673:
------------------------------------------
Summary: Partitioned join cost ignores runtime filters, so large
builds are broadcast (TPC-DS q78 broadcasts 9.5 GB per host)
Key: IMPALA-15673
URL: https://issues.apache.org/jira/browse/IMPALA-15673
Project: IMPALA
Issue Type: Improvement
Components: Frontend
Reporter: Zoltán Borók-Nagy
DistributedPlanner.createHashJoinFragment() charges the partitioned strategy
for sending every probe row. Runtime filters from joins above often drop most
of those rows at the scan, but runtime filters are generated after the
distributed plan (Planner.createPlanFragments()), so the cost does not see them.
*TPC-DS q78 at SF1000* (8 executors, MT_DOP=8, PLANNER=CALCITE,
CTE_THRESHOLD=1): store_returns (288M rows, 4.6 GB) was broadcast at a cost of
110.6 GB against 112.9 GB for partitioning, a 2% margin. The partitioned cost
counted all 2.88B store_sales rows (103.7 GB), but the date_dim join above lets
only about 550M through as a runtime filter. Each host then built a 9.5 GB hash
table: 10.8 s partitioning rows plus 17.8 s building hash tables, on one
thread. q17, q25 and q29 hit the same problem: in q17, 2.88B store_sales rows
are costed and 81M survive.
*Prototype:* a pre-pass over the single-node plan estimates, for each hash
join, the selectivity of the filters that inner and semi joins above it apply
to its probe-side scan: cardinality / probe cardinality of each join whose
probe expr is bound by, or has a value transfer to, a slot of that scan. The
partitioned cost scales the probe cardinality by it. The pre-pass is skipped
with RUNTIME_FILTER_MODE=OFF. It changes 19 plans, all toward partitioned.
Measured on the 104 queries, against 376.8 s without these changes:
* the parallel build alone (IMPALA-15631): 353.1 s; plus this change: 345.9 s;
* this change plus the IS NOT NULL estimate (IMPALA-?): 344.4 s;
* all three: 336.0 s. q78 no longer broadcasts store_returns; with all
broadcast fixes it went from 48.0 s to 18.8 s.
IMPALA-3573 describes the same blind spot for join ordering (q29); this ticket
covers the broadcast/partition decision.
Prototype: https://gerrit.cloudera.org/#/c/25087/
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
