walterddr commented on code in PR #10408:
URL: https://github.com/apache/pinot/pull/10408#discussion_r1136504692
##########
pinot-query-planner/src/main/java/org/apache/pinot/query/planner/stage/MailboxReceiveNode.java:
##########
@@ -32,6 +38,14 @@ public class MailboxReceiveNode extends AbstractStageNode {
private RelDistribution.Type _exchangeType;
@ProtoProperties
private KeySelector<Object[], Object[]> _partitionKeySelector;
+ @ProtoProperties
+ private List<RexExpression> _collationKeys;
+ @ProtoProperties
+ private List<RelFieldCollation.Direction> _collationDirections;
+ @ProtoProperties
+ private boolean _isInputSorted;
+ @ProtoProperties
+ private boolean _sortOnReceiver;
Review Comment:
can we kept these 2 booleans the same name as exchange. e.g.
(`isSortOnSender` and `isSortOnReceiver`).
based on the node type itself it should know what to do. also keeps it
consistent and easier to understand.
##########
pinot-query-planner/src/main/java/org/apache/calcite/rel/logical/PinotLogicalSortExchange.java:
##########
@@ -0,0 +1,111 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.calcite.rel.logical;
+
+import org.apache.calcite.plan.Convention;
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.rel.RelCollation;
+import org.apache.calcite.rel.RelCollationTraitDef;
+import org.apache.calcite.rel.RelDistribution;
+import org.apache.calcite.rel.RelDistributionTraitDef;
+import org.apache.calcite.rel.RelInput;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.RelWriter;
+import org.apache.calcite.rel.core.SortExchange;
+
+
+/**
+ * Pinot's implementation of {@code SortExchange} which needs information
about whether to sort on the sender
+ * and/or receiver side of the exchange. Every {@code Exchange} is broken into
a send and a receive node and the
+ * decision on where to sort is made by the planner and this information has
to b passed onto the send and receive
+ * nodes for the correct execution.
+ *
+ * Note: This class does not extend {@code LogicalSortExchange} because its
constructor which takes the list of
+ * parameters is private.
+ */
+public class PinotLogicalSortExchange extends SortExchange {
+
+ protected final boolean _isSortedOnSender;
+ protected final boolean _isSortedOnReceiver;
+
+ private PinotLogicalSortExchange(RelOptCluster cluster, RelTraitSet traitSet,
+ RelNode input, RelDistribution distribution, RelCollation collation,
+ boolean isSortedOnSender, boolean isSortedOnReceiver) {
+ super(cluster, traitSet, input, distribution, collation);
+ _isSortedOnSender = isSortedOnSender;
+ _isSortedOnReceiver = isSortedOnReceiver;
+ }
+
+ /**
+ * Creates a PinotLogicalSortExchange by parsing serialized output.
+ */
+ public PinotLogicalSortExchange(RelInput input) {
+ super(input);
+ _isSortedOnSender = false;
+ _isSortedOnReceiver = true;
+ }
+
+ /**
+ * Creates a PinotLogicalSortExchange.
+ *
+ * @param input Input relational expression
+ * @param distribution Distribution specification
+ * @param collation array of sort specifications
+ * @param isSortedOnSender whether to sort on the sender
+ * @param isSortedOnReceiver whether to sort on receiver
+ */
+ public static PinotLogicalSortExchange create(
+ RelNode input,
+ RelDistribution distribution,
+ RelCollation collation,
+ boolean isSortedOnSender,
+ boolean isSortedOnReceiver) {
+ RelOptCluster cluster = input.getCluster();
+ collation = RelCollationTraitDef.INSTANCE.canonize(collation);
+ distribution = RelDistributionTraitDef.INSTANCE.canonize(distribution);
+ RelTraitSet traitSet =
+
input.getTraitSet().replace(Convention.NONE).replace(distribution).replace(collation);
+ return new PinotLogicalSortExchange(cluster, traitSet, input, distribution,
+ collation, isSortedOnSender, isSortedOnReceiver);
+ }
+
+ //~ Methods ----------------------------------------------------------------
+
+ @Override
+ public SortExchange copy(RelTraitSet traitSet, RelNode newInput,
+ RelDistribution newDistribution, RelCollation newCollation) {
+ return new PinotLogicalSortExchange(this.getCluster(), traitSet, newInput,
+ newDistribution, newCollation, _isSortedOnSender, _isSortedOnReceiver);
+ }
+
+ @Override public RelWriter explainTerms(RelWriter pw) {
Review Comment:
nit format
##########
pinot-query-planner/src/main/java/org/apache/calcite/rel/logical/PinotLogicalSortExchange.java:
##########
@@ -0,0 +1,111 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.calcite.rel.logical;
+
+import org.apache.calcite.plan.Convention;
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.rel.RelCollation;
+import org.apache.calcite.rel.RelCollationTraitDef;
+import org.apache.calcite.rel.RelDistribution;
+import org.apache.calcite.rel.RelDistributionTraitDef;
+import org.apache.calcite.rel.RelInput;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.RelWriter;
+import org.apache.calcite.rel.core.SortExchange;
+
+
+/**
+ * Pinot's implementation of {@code SortExchange} which needs information
about whether to sort on the sender
+ * and/or receiver side of the exchange. Every {@code Exchange} is broken into
a send and a receive node and the
+ * decision on where to sort is made by the planner and this information has
to b passed onto the send and receive
+ * nodes for the correct execution.
+ *
+ * Note: This class does not extend {@code LogicalSortExchange} because its
constructor which takes the list of
+ * parameters is private.
+ */
+public class PinotLogicalSortExchange extends SortExchange {
+
+ protected final boolean _isSortedOnSender;
+ protected final boolean _isSortedOnReceiver;
+
+ private PinotLogicalSortExchange(RelOptCluster cluster, RelTraitSet traitSet,
+ RelNode input, RelDistribution distribution, RelCollation collation,
+ boolean isSortedOnSender, boolean isSortedOnReceiver) {
+ super(cluster, traitSet, input, distribution, collation);
+ _isSortedOnSender = isSortedOnSender;
+ _isSortedOnReceiver = isSortedOnReceiver;
+ }
+
+ /**
+ * Creates a PinotLogicalSortExchange by parsing serialized output.
+ */
+ public PinotLogicalSortExchange(RelInput input) {
+ super(input);
+ _isSortedOnSender = false;
+ _isSortedOnReceiver = true;
+ }
+
+ /**
+ * Creates a PinotLogicalSortExchange.
+ *
+ * @param input Input relational expression
+ * @param distribution Distribution specification
+ * @param collation array of sort specifications
+ * @param isSortedOnSender whether to sort on the sender
+ * @param isSortedOnReceiver whether to sort on receiver
+ */
+ public static PinotLogicalSortExchange create(
+ RelNode input,
+ RelDistribution distribution,
+ RelCollation collation,
+ boolean isSortedOnSender,
+ boolean isSortedOnReceiver) {
+ RelOptCluster cluster = input.getCluster();
+ collation = RelCollationTraitDef.INSTANCE.canonize(collation);
+ distribution = RelDistributionTraitDef.INSTANCE.canonize(distribution);
+ RelTraitSet traitSet =
+
input.getTraitSet().replace(Convention.NONE).replace(distribution).replace(collation);
+ return new PinotLogicalSortExchange(cluster, traitSet, input, distribution,
+ collation, isSortedOnSender, isSortedOnReceiver);
+ }
+
+ //~ Methods ----------------------------------------------------------------
+
+ @Override
+ public SortExchange copy(RelTraitSet traitSet, RelNode newInput,
+ RelDistribution newDistribution, RelCollation newCollation) {
+ return new PinotLogicalSortExchange(this.getCluster(), traitSet, newInput,
+ newDistribution, newCollation, _isSortedOnSender, _isSortedOnReceiver);
+ }
+
+ @Override public RelWriter explainTerms(RelWriter pw) {
+ return super.explainTerms(pw)
+ .item("isSortedOnSender", _isSortedOnSender)
+ .item("isSortedOnReceiver", _isSortedOnReceiver);
+ }
+
+ public boolean isSortedOnSender() {
+ return _isSortedOnSender;
+ }
+
+ public boolean isSortedOnReceiver() {
+ return _isSortedOnReceiver;
+ }
Review Comment:
nit `isSortOnSender` and `isSortOnReceiver`
##########
pinot-query-planner/src/main/java/org/apache/pinot/query/planner/stage/MailboxSendNode.java:
##########
@@ -32,17 +39,38 @@ public class MailboxSendNode extends AbstractStageNode {
private RelDistribution.Type _exchangeType;
@ProtoProperties
private KeySelector<Object[], Object[]> _partitionKeySelector;
+ @ProtoProperties
+ private List<RexExpression> _collationKeys;
+ @ProtoProperties
+ private List<RelFieldCollation.Direction> _collationDirections;
+ @ProtoProperties
+ private boolean _sortOnSender;
Review Comment:
same here for naming
##########
pinot-query-planner/src/test/java/org/apache/calcite/rel/rules/PinotSortExchangeCopyRuleTest.java:
##########
@@ -81,7 +81,8 @@ public void tearDown()
@Test
public void shouldMatchLimitNoOffsetNoSort() {
// Given:
- SortExchange exchange = LogicalSortExchange.create(_input,
RelDistributions.SINGLETON, RelCollations.EMPTY);
+ SortExchange exchange = PinotLogicalSortExchange.create(_input,
RelDistributions.SINGLETON, RelCollations.EMPTY,
+ false, true);
Review Comment:
could we add a test for `(false, false)`, and a exception catcher on `(true,
false)` for now
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/QueryRunner.java:
##########
@@ -213,7 +213,7 @@ private void runLeafStage(DistributedStagePlan
distributedStagePlan, Map<String,
mailboxSendOperator = new MailboxSendOperator(_mailboxService,
new LeafStageTransferableBlockOperator(serverQueryResults,
sendNode.getDataSchema(), requestId,
sendNode.getStageId(), _rootServer),
receivingStageMetadata.getServerInstances(),
- sendNode.getExchangeType(), sendNode.getPartitionKeySelector(),
_rootServer, requestId,
+ sendNode.getExchangeType(), sendNode.getPartitionKeySelector(),
null, null, false, _rootServer, requestId,
Review Comment:
should get the collcation and others from sendNode instead of null?
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MailboxReceiveOperator.java:
##########
@@ -161,7 +194,12 @@ protected TransferableBlock getNextBlock() {
return _upstreamErrorBlock;
}
if (!block.isEndOfStreamBlock()) {
- return block;
+ if (_priorityQueue != null) {
+ List<Object[]> container = block.getContainer();
+ _priorityQueue.addAll(container);
+ } else {
+ return block;
+ }
Review Comment:
there's an issue here with this approach. I am not sure if we should return
a no-op block here or continue to grab more. is it possible to box this into a
util. say.
```suggestion
if (!block.isEndOfStreamBlock()) {
return processReceivedBlock(block); // box the logic in
processReceivedBlock(block)
```
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MailboxReceiveOperator.java:
##########
@@ -176,6 +214,18 @@ protected TransferableBlock getNextBlock() {
}
}
+ if (((openMailboxCount == 0) || (openMailboxCount <= eosMailboxCount))
+ && (!CollectionUtils.isEmpty(_priorityQueue)) &&
!_isSortedBlockConstructed) {
+ // Some data is present in the PriorityQueue, these need to be sent
upstream
+ LinkedList<Object[]> rows = new LinkedList<>();
+ while (_priorityQueue.size() > 0) {
+ Object[] row = _priorityQueue.poll();
+ rows.addFirst(row);
+ }
+ _isSortedBlockConstructed = true;
+ return new TransferableBlock(rows, _dataSchema, DataBlock.Type.ROW);
+ }
Review Comment:
based on this logic. as long as there's a _priorityQueue and size is not
zero you should return the priority queue results.
when any mailbox receives an error. it should also dropped any
_priorityQueue content and stopped the consumption.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MailboxSendOperator.java:
##########
@@ -115,6 +125,11 @@ public
MailboxSendOperator(MailboxService<TransferableBlock> mailboxService,
_exchange = blockExchangeFactory.build(mailboxService, receivingMailboxes,
exchangeType, keySelector, splitter,
deadlineMs);
+ _collationKeys = collationKeys;
+ _collationDirections = collationDirections;
+ _sortOnSender = sortOnSender;
+ Preconditions.checkState(!sortOnSender, "Sort on send is not yet
implemented!");
Review Comment:
do not check this on the sender operator. only check in the sending node
side. the support precondition is a planner constrain not a runtime constrain
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
