yashmayya opened a new issue, #19427: URL: https://github.com/apache/pinot/issues/19427
## Background [PR #19353](https://github.com/apache/pinot/pull/19353) fixed an NPE that occurred with `useSpools = true`. `BroadcastExchange` is the only exchange that routes one block instance to more than one destination. It now copies blocks that carry aggregation intermediate results in `OBJECT` columns. Local mailboxes deliver on-heap blocks by reference, and downstream operators mutate those objects in place. That fix skips the copy for remote destinations, because a remote destination serializes the block on the calling thread and never shares the mutable object. This issue is the follow-up to [a review comment on that PR](https://github.com/apache/pinot/pull/19353#discussion_r3902651419): the skip never applies to spools, which are the case the fix was written for. ## Problem 1: the remote skip never applies to a spool A multi-send (spool) node builds one inner exchange per receiver stage. It wraps each inner exchange as a `BlockExchange.BlockExchangeSendingMailbox`. `isLocal()` on that wrapper returns `true` unconditionally, whatever the inner exchange's own mailboxes are. Every outer destination of a spool is such a wrapper, so `BroadcastExchange#route` takes the local branch for all of them. A receiver stage with no worker on the sending server therefore still gets a copy. The inner exchange serializes that copy immediately, on the same thread, and no receiver ever mutates it. Each copy costs one serialization plus one deserialization of every non-null `OBJECT` cell, for each extra receiver stage. Ordinary broadcast edges do not carry `OBJECT` columns today, so the remote branch is currently unreachable in practice. ## Problem 2: `isLocal()` carries two meanings - `BlockExchange#sendBlock` reads it as "do not split the block here". - `BroadcastExchange#route` reads it as "this destination can give the block to a receiver by reference". The two answers differ for `BlockExchangeSendingMailbox`. Only the conservative direction keeps the code correct: an unnecessary copy is safe, a missing copy corrupts data. ## Suggested fix 1. Add a predicate to `SendingMailbox`, for example `deliversByReference()`: - `InMemorySendingMailbox` returns `true`. - `GrpcSendingMailbox` returns `false`. - `BlockExchangeSendingMailbox` returns `true` if any mailbox of its inner exchange returns `true`. 2. Use the new predicate in `BroadcastExchange#route`. Then `isLocal()` keeps only the meaning that `sendBlock` needs. The delegation must be an OR over the inner mailboxes. An inner `HashExchange` builds a new block for each destination, but those blocks hold the same cell objects. One local worker in a receiver stage is therefore enough to require a copy. ## Also: state the serialization contract on the interface `BroadcastExchange#route` gives the original block to remote destinations before it gives the block to a local one. This is safe only because `send(MseBlock.Data)` serializes the block before it returns. `GrpcSendingMailbox` documents this, but `route` depends on it for every implementation that does not deliver by reference. The requirement belongs on `SendingMailbox#send`, where implementers can see it. ## Related `BlockExchangeSendingMailbox#isLocal()` also makes the outer exchange skip the splitter, and the inner exchanges get `BlockSplitter.NO_OP` (`MailboxSendOperator#getBlockExchange`). Multi-send blocks to remote receivers are therefore never split against `MAX_MAILBOX_CONTENT_SIZE_BYTES`. This is a separate defect with the same cause, and one change can correct both. -- 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]
