siddharthteotia commented on code in PR #8558:
URL: https://github.com/apache/pinot/pull/8558#discussion_r855405640
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/executor/WorkerQueryExecutor.java:
##########
@@ -98,22 +100,27 @@ public void runJob() {
private BaseOperator<DataTableBlock> getOperator(long requestId, StageNode
stageNode,
Map<Integer, StageMetadata> metadataMap) {
// TODO: optimize this into a framework. (physical planner)
- if (stageNode instanceof MailboxSendNode) {
- MailboxSendNode sendNode = (MailboxSendNode) stageNode;
- BaseOperator<DataTableBlock> nextOperator = getOperator(requestId,
sendNode.getInputs().get(0), metadataMap);
- StageMetadata receivingStageMetadata =
metadataMap.get(sendNode.getReceiverStageId());
- return new MailboxSendOperator(_mailboxService, nextOperator,
receivingStageMetadata.getServerInstances(),
- sendNode.getExchangeType(), _hostName, _port, requestId,
sendNode.getStageId());
- } else if (stageNode instanceof MailboxReceiveNode) {
+ if (stageNode instanceof MailboxReceiveNode) {
MailboxReceiveNode receiveNode = (MailboxReceiveNode) stageNode;
List<ServerInstance> sendingInstances =
metadataMap.get(receiveNode.getSenderStageId()).getServerInstances();
return new MailboxReceiveOperator(_mailboxService,
RelDistribution.Type.ANY, sendingInstances, _hostName, _port,
requestId, receiveNode.getSenderStageId());
+ } else if (stageNode instanceof MailboxSendNode) {
+ MailboxSendNode sendNode = (MailboxSendNode) stageNode;
+ BaseOperator<DataTableBlock> nextOperator = getOperator(requestId,
sendNode.getInputs().get(0), metadataMap);
+ StageMetadata receivingStageMetadata =
metadataMap.get(sendNode.getReceiverStageId());
+ return new MailboxSendOperator(_mailboxService, nextOperator,
receivingStageMetadata.getServerInstances(),
+ sendNode.getExchangeType(), sendNode.getPartitionKeySelector(),
_hostName, _port, requestId,
+ sendNode.getStageId());
} else if (stageNode instanceof JoinNode) {
JoinNode joinNode = (JoinNode) stageNode;
BaseOperator<DataTableBlock> leftOperator = getOperator(requestId,
joinNode.getInputs().get(0), metadataMap);
BaseOperator<DataTableBlock> rightOperator = getOperator(requestId,
joinNode.getInputs().get(1), metadataMap);
return new BroadcastJoinOperator(leftOperator, rightOperator,
joinNode.getCriteria());
Review Comment:
+1 - it is a physical relational operator for executing join using hash. A
nested loop join can also be implemented in the future and the distribution
strategy could still possibly be broadcast
--
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
