Zoltán Borók-Nagy created IMPALA-15519:
------------------------------------------
Summary: Allow more than one in-flight TransmitData RPC per data
stream channel
Key: IMPALA-15519
URL: https://issues.apache.org/jira/browse/IMPALA-15519
Project: IMPALA
Issue Type: Improvement
Components: Backend
Reporter: Zoltán Borók-Nagy
KrpcDataStreamSender::Channel allows one TransmitData RPC in flight
(rpc_in_flight_, krpc-data-stream-sender.cc:284; see the TODO at :245, "replace
this with a queue"). The sender serializes the next batch while an RPC is in
flight, then waits in WaitForRpcLocked() (:437) before it can send it. A
channel therefore moves at most one batch per network round trip, whatever the
bandwidth. With the default 16K buffer, hash senders whose receivers are not
pushing back spend most of their time blocked on round trips.
*Data:* TPC-DS SF100, 5 executors (AWS m6id.4xlarge), 104 queries, master
1f34ea915e, MT_DOP=0.
* Hash exchanges send 10.2M TransmitData RPCs of about 17 KB each (compressed).
* In the 630 hash exchanges whose receivers are not pushing back, the slowest
sender instances are blocked 150 s in total. Examples: q72 34.5 s, q51 5.1 s,
q67a 4.6 s per exchange, q64 3.6 s.
* Network time per RPC is about 390 us. It stays flat when the payload grows 8x
(IMPALA-15512).
A prototype with 2 RPCs in flight per channel for HASH_PARTITIONED and RANDOM
senders:
* MT_DOP=0: 374.6 -> 307.1 s (-18%; -10% without q72 and q78), geomean 0.91.
* With a 128K buffer and no exchange compression: 251.6 s.
* MT_DOP=8: the same combination gives -5% (187.0 -> 176.9 s).
*Design notes from the prototype:*
* Move the per-RPC state (RpcController, response, in-flight batch, start time,
outbound batch) into a slot struct. A channel owns K slots and counts the RPCs
in flight.
* TransmitData waits until fewer than K RPCs are in flight. FlushFinal and
EndDataStream wait for all of them.
* Keep the first RPC error. Teardown cancels and waits for every slot.
* Keep K=1 for merging exchanges, broadcast/unpartitioned senders, Kudu and
directed (Iceberg) senders.
* Each channel now holds K+1 batches instead of 2. The planner's memory
estimate (DataStreamSink) and any adaptive buffer sizing must use K+1.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
