Skip to content

[Multi-stage] Optimize query dispatch #12358

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom
Merged
Jackie-Jiang merged 1 commit into from
Feb 3, 2024
Merged

[Multi-stage] Optimize query dispatch #12358

Jackie-Jiang merged 1 commit into from
Feb 3, 2024

Conversation

@Jackie-Jiang
Copy link
Contributor

@Jackie-Jiang Jackie-Jiang commented Feb 3, 2024

  • Do the server independent serialization only once per stage
  • Send one plan per server instead of one plan per stage per server
  • Parallel execute the server dependent serialization

@Jackie-Jiang Jackie-Jiang added enhancement multi-stage Related to the multi-stage query engine labels Feb 3, 2024
@codecov-commenter
Copy link

codecov-commenter commented Feb 3, 2024 •
edited
Loading

Codecov Report

Attention: 8 lines in your changes are missing coverage. Please review.

Comparison is base (33074e1) 61.71% compared to head (9d24f79) 61.68%.
Report is 4 commits behind head on master.

Files Patch % Lines
.../pinot/query/service/dispatch/QueryDispatcher.java 84.31% 7 Missing and 1 partial ⚠️
Additional details and impacted files 
@@             Coverage Diff              @@
##             master   #12358      +/-   ##
============================================
- Coverage     61.71%   61.68%   -0.04%     
+ Complexity     1153      207     -946     
============================================
  Files          2424     2426       +2     
  Lines        132512   132674     +162     
  Branches      20481    20502      +21     
============================================
+ Hits          81782    81839      +57     
- Misses        44732    44831      +99     
- Partials       5998     6004       +6
Flag Coverage Δ
custom-integration1 <0.01% <0.00%> (-0.01%) ⬇️
integration <0.01% <0.00%> (-0.01%) ⬇️
integration1 <0.01% <0.00%> (-0.01%) ⬇️
integration2 0.00% <0.00%> (ø)
java-11 61.63% <89.04%> (-0.05%) ⬇️
java-21 61.57% <89.04%> (-0.03%) ⬇️
skip-bytebuffers-false 61.66% <89.04%> (-0.04%) ⬇️
skip-bytebuffers-true 61.54% <89.04%> (-0.03%) ⬇️
temurin 61.68% <89.04%> (-0.04%) ⬇️
unittests 61.67% <89.04%> (-0.04%) ⬇️
unittests1 46.90% <89.04%> (-0.02%) ⬇️
unittests2 27.68% <0.00%> (+<0.01%) ⬆️

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

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

walterddr
walterddr approved these changes Feb 3, 2024
View reviewed changes
Copy link
Contributor

@walterddr walterddr left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

looks good to me.

walterddr
walterddr reviewed Feb 3, 2024
View reviewed changes
pinot-query-runtime/src/main/java/org/apache/pinot/query/service/dispatch/QueryDispatcher.java
Comment on lines +135 to +136
for (QueryServerInstance serverInstance : serverInstances) {
_executorService.submit(() -> {
Copy link
Contributor

@walterddr walterddr Feb 3, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

note that QueryServer could also apply similar technique

pinot/pinot-query-runtime/src/main/java/org/apache/pinot/query/service/server/QueryServer.java

Lines 108 to 123 in 0a43986

try {
distributedStagePlans = QueryPlanSerDeUtils.deserializeStagePlan(request);
} catch (Exception e) {
LOGGER.error("Caught exception while deserializing the request: {}", requestId, e);
responseObserver.onError(Status.INVALID_ARGUMENT.withDescription("Bad request").withCause(e).asException());
return;
}
// 2. Submit distributed stage plans, await response successful or any failure which cancels all other tasks.
int numSubmission = distributedStagePlans.size();
CompletableFuture<?>[] submissionStubs = new CompletableFuture[numSubmission];
for (int i = 0; i < numSubmission; i++) {
DistributedStagePlan distributedStagePlan = distributedStagePlans.get(i);
submissionStubs[i] =
CompletableFuture.runAsync(() -> _queryRunner.processQuery(distributedStagePlan, requestMetadata),
_querySubmissionExecutorService);
}

deserialization (line 109) can be move into the runAsync (line 121)

Copy link
Contributor Author

@Jackie-Jiang Jackie-Jiang Feb 3, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good point! Will do it as a separate PR

@Jackie-Jiang Jackie-Jiang merged commit c9a82c4 into apache:master Feb 3, 2024
@Jackie-Jiang Jackie-Jiang deleted the optimiza_dispatch branch February 3, 2024 19:59
@walterddr walterddr mentioned this pull request Feb 5, 2024
Merged
suyashpatel98 pushed a commit to suyashpatel98/pinot that referenced this pull request Feb 28, 2024
@Jackie-Jiang @suyashpatel98
[Multi-stage] Optimize query dispatch (apache#12358)
1338ec2
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

@walterddr walterddr walterddr approved these changes

@xiangfu0 xiangfu0 xiangfu0 approved these changes

Assignees

No one assigned

Labels

enhancement multi-stage Related to the multi-stage query engine

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

4 participants

@Jackie-Jiang @codecov-commenter @xiangfu0 @walterddr