MSE: spool fan-out copies aggregation intermediates even when every receiver stage is remote

[yashmayya]

Background

PR #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: 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.