yashmayya commented on code in PR #19353:
URL: https://github.com/apache/pinot/pull/19353#discussion_r3906339725
##########
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:
Sounds good, I'll create a follow up.
--
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]
