gortiz commented on code in PR #19353:
URL: https://github.com/apache/pinot/pull/19353#discussion_r3902651419
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/exchange/BroadcastExchange.java:
##########
@@ -39,8 +52,34 @@ protected BroadcastExchange(List<SendingMailbox>
sendingMailboxes, BlockSplitter
@Override
protected void route(List<SendingMailbox> destinations, MseBlock.Data block)
{
+ // Serialized blocks are read-only (every receiver deserializes its own
copy of the data), so they are always
+ // safe to share
+ if (destinations.size() == 1 || !block.isRowHeap() ||
!block.asRowHeap().containsObjectColumns()) {
+ for (SendingMailbox mailbox : destinations) {
+ sendBlock(mailbox, block);
+ }
+ return;
+ }
+ // Send a copy to every active local destination except the first one,
which receives the original block without
+ // copying. Remote destinations serialize the original block on this
thread, and the copies are also made on this
+ // thread, so all reads of the original block finish before it is handed
to a local receiver that can start
+ // mutating it.
+ RowHeapDataBlock rowHeapBlock = block.asRowHeap();
+ SendingMailbox firstLocalDestination = null;
for (SendingMailbox mailbox : destinations) {
- sendBlock(mailbox, block);
+ if (mailbox.isEarlyTerminated()) {
+ continue;
+ }
+ if (!mailbox.isLocal()) {
Review Comment:
Non-blocking, but worth a follow-up: this optimisation doesn't apply to the
case the PR is actually fixing, so spools still pay serde round trips for
objects that are only going to be serialized anyway.
`BlockExchange.BlockExchangeSendingMailbox#isLocal()` returns an
unconditional `true`, regardless of whether the inner exchange's own mailboxes
are local or `GrpcSendingMailbox`. Every outer destination of a multi-send node
is one of these, so the `!mailbox.isLocal()` branch is never taken for a spool:
a spool whose receiver stages all live on other servers still gets
`copyObjectColumns()` for each extra stage, even though every one of those
inner exchanges will immediately serialize the block on this same thread and
never hand it to a mutating receiver. That's a full serialize + deserialize per
intermediate, per extra stage, purely to throw the copy away at the next hop.
Underneath it, `isLocal()` now carries two different meanings:
- the original one, used by `BlockExchange#sendBlock`: *don't split here,
pass through*;
- the new one, used here: *may hand the block to a receiver by reference*.
For `BlockExchangeSendingMailbox` those disagree, and only the conservative
direction saves us. A separate predicate on `SendingMailbox` — something like
`deliversByReference()`, which `BlockExchangeSendingMailbox` answers by
delegating to its inner mailboxes — would let the remote case skip the copy and
keep `isLocal()` meaning only what `sendBlock` needs.
Related, and also fine as a follow-up: the contract the optimisation relies
on (`send(MseBlock.Data)` serializes synchronously on the calling thread) is
documented on `GrpcSendingMailbox`, but `route` depends on it for *any* mailbox
reporting `!isLocal()`, present or future. It would be better stated on
`SendingMailbox#send` so it reads as a requirement on implementers.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
