xiangfu0 commented on code in PR #10791:
URL: https://github.com/apache/pinot/pull/10791#discussion_r1205032423
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/service/dispatch/QueryDispatcher.java:
##########
@@ -139,22 +140,24 @@ int submit(long requestId, DispatchableSubPlan
dispatchableSubPlan, long timeout
for (Map.Entry<QueryServerInstance, List<Integer>> queryServerEntry
:
dispatchableSubPlan.getQueryStageList().get(stageId).getServerInstanceToWorkerIdMap().entrySet())
{
QueryServerInstance queryServerInstance = queryServerEntry.getKey();
+ Worker.QueryRequest.Builder queryRequestBuilder =
Worker.QueryRequest.newBuilder();
+ String host = queryServerInstance.getHostname();
+ int servicePort = queryServerInstance.getQueryServicePort();
+ int mailboxPort = queryServerInstance.getQueryMailboxPort();
for (int workerId : queryServerEntry.getValue()) {
- String host = queryServerInstance.getHostname();
- int servicePort = queryServerInstance.getQueryServicePort();
- int mailboxPort = queryServerInstance.getQueryMailboxPort();
VirtualServerAddress virtualServerAddress = new
VirtualServerAddress(host, mailboxPort, workerId);
- DispatchClient client = getOrCreateDispatchClient(host,
servicePort);
dispatchCalls++;
- int finalStageId = stageId;
- _executorService.submit(() ->
client.submit(Worker.QueryRequest.newBuilder().setStagePlan(
- QueryPlanSerDeUtils.serialize(
- constructDistributedStagePlan(dispatchableSubPlan,
finalStageId, virtualServerAddress)))
- .putMetadata(QueryConfig.KEY_OF_BROKER_REQUEST_ID,
String.valueOf(requestId))
- .putMetadata(QueryConfig.KEY_OF_BROKER_REQUEST_TIMEOUT_MS,
String.valueOf(timeoutMs))
- .putAllMetadata(queryOptions).build(), finalStageId,
queryServerInstance, deadline,
- dispatchCallbacks::offer));
+ queryRequestBuilder.addStagePlan(
+ QueryPlanSerDeUtils.serialize(dispatchableSubPlan, stageId,
virtualServerAddress));
Review Comment:
The basic idea is to carry all the workIds within one stagePlan then expand
it at the worker side to generate one DistributedStagePlan per worker.
--
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: commits-unsubscr...@pinot.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For additional commands, e-mail: commits-h...@pinot.apache.org
