yashmayya opened a new pull request, #19428: URL: https://github.com/apache/pinot/pull/19428
Fixes #19427. Follow-up to #19353. ## Problem #19353 made `BroadcastExchange` copy blocks that carry aggregation intermediate results, because a destination can hand an on-heap block to a receiver by reference. It skips the copy for destinations that are not local, because those serialize the block within `send`. That skip never applied to spools, which are the case #19353 was written for. A multi-send (spool) node wraps each receiver stage's inner exchange as a `BlockExchange.BlockExchangeSendingMailbox`, and `isLocal()` on that wrapper returns `true` unconditionally. Every outer destination of a spool is such a wrapper, so a receiver stage with no worker on the sending server still got a copy. The inner exchange then serialized that copy immediately, on the same thread, and no receiver ever mutated it. Each wasted copy costs one serialization plus one deserialization of every non-null `OBJECT` cell. The cause is that `isLocal()` answered two different questions: - `BlockExchange#sendBlock` asks "can I pass the block whole, without splitting it?" - `BroadcastExchange#route` asks "can a receiver keep a reference to this block?" The two answers differ for `BlockExchangeSendingMailbox`. Only the conservative direction kept the code correct. ## Fix Add `SendingMailbox#deliversByReference()`, and use it in `BroadcastExchange#route`: - `InMemorySendingMailbox` returns `true`. - `GrpcSendingMailbox` returns `false`. - `BlockExchangeSendingMailbox` returns `true` if any mailbox of its inner exchange returns `true`. The answer of a mailbox never changes, so each exchange computes it once, when it is created. The delegation is an OR over the inner mailboxes. An inner `HashExchange` builds a new block for each destination, but those blocks hold the same cell objects, so one by-reference worker in a receiver stage is enough to require a copy. `isLocal()` now only means what `sendBlock` needs. This PR also moves the contract that `route` depends on — a mailbox that does not deliver by reference must finish reading the block before `send` returns — from `GrpcSendingMailbox` to `SendingMailbox#send`, where implementers can see it. ## Testing `BroadcastExchangeTest` gets two spool cases: receiver stages whose workers are all remote (no copies), and a receiver stage with one worker on this server next to one on another server (one copy, shared within that stage). The first fails without this change. `BlockExchangeTest` covers the delegation directly. `WindowFunnelTest` and `SpoolIntegrationTest` still pass. ## Out of scope The same wrapper also reports `isLocal() == true` to `BlockExchange#sendBlock`, while the inner exchanges get `BlockSplitter.NO_OP`, so multi-send blocks to remote receivers are never split against `MAX_MAILBOX_CONTENT_SIZE_BYTES`. That defect is tracked separately in #19427. -- 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]
