Zoltán Borók-Nagy created IMPALA-15685:
------------------------------------------
Summary: BROADCAST_BYTES_LIMIT (default 32 GB) lets multi-GB
broadcast builds through; lower the default and check it against the build
memory estimate
Key: IMPALA-15685
URL: https://issues.apache.org/jira/browse/IMPALA-15685
Project: IMPALA
Issue Type: Improvement
Components: Frontend
Reporter: Zoltán Borók-Nagy
DistributedPlanner.computeJoinDistributionMode()
(DistributedPlanner.java:847-852, master c580d36e61) rejects a cheaper
broadcast only if rows x serialized row size x 1.1 exceeds
BROADCAST_BYTES_LIMIT (default 32 GB, Query.thrift:493) or the query memory
limit, unbounded by default (Analyzer.java:877). 32 GB is close to an
executor's whole memory, and every host builds the full table on one thread
(IMPALA-15631).
*Example: TPC-DS SF10000, 12 executors, mem_limit=40g, MT_DOP=8, master, best
of 3.*
* q72: the 1.44B-row catalog_returns is broadcast because the probe side is
estimated at 2.10T rows (actual 108M). The limit check sees about 24 GB; the
builder estimates 64.09 GB per host. Builds average 3m22s and spill 11 of 16
partitions (peak 23.99 GB); q72 takes 286.9 s.
* q17 with PLANNER=CALCITE broadcasts a 401M-row store_returns build (1m27s,
27.71 GB per host): 114.1 s vs 10.4 s with the original planner, which
partitions this join.
Master's expected q78 plan with 3 TB statistics on 10 hosts broadcasts the
864M-row store_returns with a 33.66 GB builder estimate per host; the check
sees about 14 GB (tpcds_cpu_cost/tpcds-q78.test:415-430). At SF1000, q78 is
broadcast at a 2% cost margin (IMPALA-15673).
A prototype join order for q72 (IMPALA-15665; SF1000, 8 executors,
PLANNER=CALCITE, other pending changes present) took q72 from 16.4 s to 8.9 s
with BROADCAST_BYTES_LIMIT=1g. Without it, the new plan broadcast the 144M-row
catalog_returns (6.7 s build) and q29, q64 and q72 together gained 0.6 s.
StarRocks caps broadcasts at 15M build rows, Trino at 100 MB.
*Proposal:*
* Lower the default (e.g. 1 GB) or derive it from executor memory.
* Compare the limit with the builder's memory estimate
(HashJoinNode.java:291-293), not 1.1 x serialized bytes.
* Require a cost margin or a build/probe ratio for large broadcasts; a ratio
cannot catch q72's probe overestimate, so the cap stays.
* Fix ImpalaService.thrift:506-509, which describes the limit as total bytes
sent.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
