[multistage] Introduce InMemoryMailboxService for Optimizing Join Performance by ankitsultana · Pull Request #9484 · apache/pinot

ankitsultana · 2022-09-28T22:57:24Z

This does the following:

MailboxService uses TransferableBlock. The serde to MailboxContent is moved within GrpcMailbox.
Adds a new InMemoryMailboxService. If the stage is local, then the TransferableBlock can be passed from sender to receiver simply using a queue.
Adds a MultiplexingMailboxService which can choose either gRPC based mailbox-service or in-memory mailbox-service.

Optimizations:

This avoids MailboxContent serde if sender/receiver stages are local. This can save ~10% latency overhead (based on the query).
Saves memory. If we use gRPC mailbox then we'll copy the buffer to create MailboxContent.
[Not part of this PR] This will help avoid List<Object[]> _rows ==> DataTable ==> List<Object[]> _rows serde.

These optimizations are most helpful if the stages deal with large data (Amdahl's basically)

cc: @walterddr @siddharthteotia

walterddr · 2022-09-30T18:45:30Z

...t-query-runtime/src/main/java/org/apache/pinot/query/mailbox/MultiplexingMailboxService.java

this is a GOOD IDEA!

+1 very clean!

walterddr

i did a partial review. overall the approach looks good. there are some details we need to iron out and clear up

also please rebase

...ker/src/main/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandler.java

pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/GrpcReceivingMailbox.java

pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/channel/InMemoryChannel.java

pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/GrpcReceivingMailbox.java

...ry-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MailboxReceiveOperator.java

...ntime/src/main/java/org/apache/pinot/query/mailbox/channel/MailboxContentStreamObserver.java

walterddr · 2022-10-17T13:59:00Z

pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/GrpcReceivingMailbox.java

    if (waitForInitialize()) {
      mailboxContent = _contentStreamObserver.poll();
      _totalMsgReceived.incrementAndGet();
+    } else {
+      return null;
    }
-    return mailboxContent;
+    return fromMailboxContent(mailboxContent);


if (!waitForInitialize()) { return null; } mailboxContent = _contentStreamObserver.poll(); _totalMsgReceived.incrementAndGet(); return fromMailboxContent(mailboxContent);

walterddr

looks good to me overall
i have one suggestion on the wrapper around in-memory mailbox.

I will give another look once we address these. thank you for the contribution!

pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/GrpcReceivingMailbox.java

pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/channel/InMemoryChannel.java

pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/InMemoryMailboxService.java

pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/channel/InMemoryChannel.java

...ntime/src/main/java/org/apache/pinot/query/mailbox/channel/MailboxContentStreamObserver.java

ankitsultana · 2022-10-17T18:46:31Z

@walterddr : Removed InMemoryChannel. I agree using a Queue directly is better. I am using isEndOfStreamBlock to determine when the receiving mailbox is done receiving now. Can you take a look again?

codecov-commenter · 2022-10-17T19:35:52Z

Codecov Report

Merging #9484 (e345e4b) into master (209c060) will increase coverage by 0.02%.
The diff coverage is 88.07%.

@@             Coverage Diff              @@
##             master    #9484      +/-   ##
============================================
+ Coverage     60.33%   60.35%   +0.02%     
- Complexity     5148     5161      +13     
============================================
  Files          1926     1931       +5     
  Lines        103209   103450     +241     
  Branches      15670    15691      +21     
============================================
+ Hits          62266    62442     +176     
- Misses        36271    36321      +50     
- Partials       4672     4687      +15

Flag	Coverage Δ
integration2	`24.56% <0.00%> (+0.03%)`	⬆️
unittests1	`67.27% <88.88%> (-0.03%)`	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files	Coverage Δ
...requesthandler/MultiStageBrokerRequestHandler.java	`0.00% <0.00%> (ø)`
...e/pinot/query/mailbox/StringMailboxIdentifier.java	`76.66% <0.00%> (-2.65%)`	⬇️
...ot/query/runtime/executor/PhysicalPlanVisitor.java	`97.29% <ø> (ø)`
...ot/query/runtime/executor/WorkerQueryExecutor.java	`100.00% <ø> (ø)`
...rg/apache/pinot/query/service/QueryDispatcher.java	`81.00% <ø> (-3.00%)`	⬇️
...he/pinot/query/mailbox/InMemorySendingMailbox.java	`58.33% <58.33%> (ø)`
...apache/pinot/query/mailbox/GrpcSendingMailbox.java	`90.90% <75.00%> (-5.10%)`	⬇️
.../pinot/query/mailbox/InMemoryReceivingMailbox.java	`78.57% <78.57%> (ø)`
...ache/pinot/query/mailbox/GrpcReceivingMailbox.java	`93.10% <92.30%> (-2.14%)`	⬇️
...apache/pinot/query/mailbox/GrpcMailboxService.java	`94.11% <100.00%> (ø)`
... and 101 more

📣 We’re building smart automated test selection to slash your CI/CD build times. Learn more

agavra

Nice! The solution is really elegant here :) My comments are mostly around concurrency and making sure the non-happy path works properly.

agavra · 2022-10-18T20:51:23Z

pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/InMemoryMailboxService.java

+  private final ConcurrentHashMap<String, ReceivingMailbox<TransferableBlock>> _receivingMailboxMap =
+      new ConcurrentHashMap<>();
+  private final ConcurrentHashMap<String, SendingMailbox<TransferableBlock>> _sendingMailboxMap =
+      new ConcurrentHashMap<>();
+  private final ConcurrentHashMap<String, ArrayBlockingQueue<TransferableBlock>> _channelMap =
+      new ConcurrentHashMap<>();


it's usually safer to wrap ReceivingMailbox, SendingMailbox and ArrayBlockingQueue all in a single MailboxState class, and then maintain just a single Map<String, MailboxState> that sets all three of these at once in a single computeIfAbsent call.

I know it's unlikely that there will be a race, but it's easier to reason about when it's guaranteed there won't be!

i think this was my original split in grpc mailbox ( this was necessary as the channels and mailbox are separately managed)
but in this case all 3 are guaranteed 1-1 mapping b/c they are all in the same JVM

I am ok with both approaches since

grouping them make more clear code smell, we can precisely managed them together because it is a special property of the in-mem mailbox

separating them makes logical sense as separation of concern (e.g. the sender normally won't have access to the receiving side info, this just happen to be the case for in-mem mailbox)

agavra · 2022-10-18T20:57:35Z

pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/InMemoryMailboxService.java

+  }
+
+  @Override
+  public void shutdown() {


when do the maps get cleaned up? (esp. regarding a cancelled query or one that times out)

This is a known leak. @walterddr is planning to fix it with query cancellation afaik

yeah we had a discussion and i created a tracker but never get to it.

but upon thinking although the affect is not much (just a simple object reference), it would be a big problem when stuff error out, see my comment in the new issue: #9626

pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/InMemoryReceivingMailbox.java

agavra · 2022-10-18T21:27:05Z

pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/InMemoryReceivingMailbox.java

+
+  @Override
+  public boolean isClosed() {
+    return _closed && _queue.size() == 0;


why do we check that the _queue.size() == 0? if I cancel the query or it times out and I close the mailbox I shouldn't be forced to process the queue

in GRPC isClosed() <=> isCompleted(); i think the wording might be misleading in this case

Ideally there shouldn't be any case where _closed is true but _queue.size() is not 0. Only case where it may happen is when the sender sends something immediately after sending the EOS block. We can also throw an exception here in that case.

To better understand this, this is how I believe some of the scenarios will be handled:

If the sender has died, it would have sent an end of stream block and then terminated on its end. The receiver here would then get the end of stream block and send it upstream to MailboxReceiveOperator. The queue should be empty after _closed is marked true.

If for some reason the sender was not able to send the EOS block, MailboxReceiveOperator will time out after the query timeout is reached.

upon checking the usage of the interface method isClosed(). it is actually in the non-threadsafe mailboxreceiveoperator.

i think for now we can keep this b/c grpc mailbox does basically the same thing - it should really be called "shouldAnotherBlockBeExpected"

later with Support non-blocking MailboxReceivedOperator #9615 we should redefine this API (or create a new one)

agavra · 2022-10-18T21:29:28Z

pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/InMemorySendingMailbox.java

+  }
+
+  @Override
+  public void complete() {


do we want to send a sentinel block here to make sure that the receiving side will terminate (instead of waiting for DEFAULT_CHANNEL_TIMEOUT_SECONDS)? or is there any other mechanism to make sure that in the non-happy path the receiver won't need to wait the whole timeout?

it is guarantee that the last message is the EOS message which serves the purpose of marking it "closed/completed".

this is probably also the reason why the close flag didn't set to volatile b/c the thread that process the EOS content is also the one that sets the flag (not the otherway around)

it is guarantee that the last message is the EOS

is that true even in the error scenarios? I was worried about cases like timeout, where we'd want to timeout quickly instead of blocking for 120s

The 120s timeout was definitely a bug. Reduced it to 1s. I have mentioned some timeout scenario handling in the other comment regarding _queue.size() == 0.

is that true even in the error scenarios? I was worried about cases like timeout, where we'd want to timeout quickly instead of blocking for 120s

if the error is sent from the sender then yes that will be the last message;

if the error occurs after a normal message was delivered then the error will originate from where the error was thrown, from that point the originated operator will stop returning any new data blocks

so yes it is guaranteed in the operator level; but it is not guarantee in the mailbox level. meaning the mailbox could still be receiving data afterwards, just no thread will dequeue from that receiving buffer. exactly the scenario i described in #9626

The 120s timeout was definitely a bug. Reduced it to 1s. I have mentioned some timeout scenario handling in the other comment regarding _queue.size() == 0.

bare in mind that the 100ms timeout for GRPCReceivingMailbox is only the timeout upon initialization; from that point there's no pull-with-time-out. it is a pushed model managed by GRPC

pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/InMemorySendingMailbox.java

agavra · 2022-10-18T21:31:30Z

...t-query-runtime/src/main/java/org/apache/pinot/query/mailbox/MultiplexingMailboxService.java

+1 very clean!

walterddr

lgtm overall thank you!

there are sevearl remaining items but I think we can follow them up later

walterddr reviewed Sep 30, 2022

View reviewed changes

ankitsultana changed the title ~~[Draft] [multistage] Introduce InMemoryMailboxService for Optimizing Join Performance~~ [multistage] Introduce InMemoryMailboxService for Optimizing Join Performance Oct 11, 2022

ankitsultana marked this pull request as ready for review October 11, 2022 06:16

ankitsultana requested a review from walterddr October 11, 2022 06:16

walterddr reviewed Oct 13, 2022

View reviewed changes

Ankit Sultana added 5 commits October 13, 2022 22:09

Use In-Memory Mailbox for Local Stages

9b6c6bd

Fix UTs and some bugs

8186653

Embellishments.. tests

4a4f1f4

Fix checkstyle

116bd8a

Rebase on master

ab7acfb

ankitsultana force-pushed the join-opt-2 branch from 7b88262 to ab7acfb Compare October 14, 2022 06:02

Address some feedback + add some comments

0ebf757

ankitsultana commented Oct 15, 2022

View reviewed changes

...ntime/src/main/java/org/apache/pinot/query/mailbox/channel/MailboxContentStreamObserver.java Show resolved Hide resolved

walterddr reviewed Oct 17, 2022

View reviewed changes

Remove InMemoryChannel + Address Other Feedback

ed23523

Ankit Sultana added 2 commits October 17, 2022 15:05

Fix test

fce4d7a

Merge branch 'master' into join-opt-2

afbbea2

walterddr mentioned this pull request Oct 18, 2022

Support non-blocking MailboxReceivedOperator #9615

Closed

Fix merge conflict with master

7746b11

agavra reviewed Oct 18, 2022

View reviewed changes

Ankit Sultana added 2 commits October 18, 2022 17:42

Address feedback

69eac60

Fix null handling

e345e4b

walterddr approved these changes Oct 19, 2022

View reviewed changes

walterddr merged commit 510051a into apache:master Oct 19, 2022

This was referenced Oct 21, 2022

Add integration test for distinct #9637

Merged

[hotfix] mailbox NPE #9639

Merged

walterddr mentioned this pull request Nov 8, 2022

[hotfix][multistage] fix error not returned issue #9760

Merged

Conversation

ankitsultana commented Sep 28, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

walterddr left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

walterddr left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

ankitsultana commented Oct 17, 2022

Uh oh!

codecov-commenter commented Oct 17, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

agavra left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

walterddr Oct 18, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

walterddr Oct 19, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

walterddr left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

ankitsultana commented Sep 28, 2022 •

edited

Loading

codecov-commenter commented Oct 17, 2022 •

edited

Loading

walterddr Oct 18, 2022 •

edited

Loading

walterddr Oct 19, 2022 •

edited

Loading