[Multi-stage] Optimize query dispatch

[Jackie-Jiang]

• 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

[codecov-commenter]

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]

looks good to me.

[walterddr]

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)

[Jackie-Jiang]

Good point! Will do it as a separate PR