xiangfu0 commented on code in PR #19396:
URL: https://github.com/apache/pinot/pull/19396#discussion_r4213853325
##########
pinot-common/src/main/proto/plan.proto:
##########
@@ -201,6 +202,18 @@ message MailboxReceiveNode {
bool sortedOnSender = 7;
}
+// Each input stream is ordered by the logical plan. Older servers reject the
unknown PlanNode oneof tag.
+// Brokers only emit this node after verifying homogeneous supported versions.
+message MailboxMergeReceiveNode {
+ int32 senderStageId = 1;
+ ExchangeType exchangeType = 2;
+ DistributionType distributionType = 3;
+ repeated int32 keys = 4;
+ repeated Collation collations = 5;
+ int32 fetch = 6;
Review Comment:
Updated at `85be3b0e80e4265ceeb6258e7db4582c4415511c` to reflect the current
implementation.
Fetch/offset are removed from MailboxMergeReceiveNode in the proto. The
current message contains only fields 1–5; tags 6 and 7 are not reserved in this
source. There is no unset proto3 integer limit with empty-window semantics.
Earlier merge-node serde and hybrid integration validation is historical
evidence; fresh rebased-head CI remains separate.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/plan/server/ServerPlanRequestVisitor.java:
##########
@@ -282,6 +283,12 @@ public Void visitProject(ProjectNode node,
ServerPlanRequestContext context) {
@Override
public Void visitSort(SortNode node, ServerPlanRequestContext context) {
if (visit(node.getInputs().get(0), context)) {
+ if (node.getFetch() == Integer.MAX_VALUE) {
Review Comment:
Updated at `85be3b0e80e4265ceeb6258e7db4582c4415511c` to reflect the current
implementation.
The fetch==MAX_VALUE sentinel is replaced by the serialized single_run_sort
node hint. RelToPlanNodeConverter attaches it to the merge exchange's sender
Sort, and ServerPlanRequestVisitor uses isSingleRunRequired() to retain that
Sort above the combined hybrid/logical leaf input. The physical window fallback
also marks its complete-input Sort and encodes MAX_VALUE for older servers that
ignore the hint; ordinary user MAX_VALUE limits remain unmarked and keep V1
pushdown. The existing visitor tests cover the marked/unmarked distinction;
prior local passes are historical evidence.
##########
pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotWindowExchangeNodeInsertRule.java:
##########
@@ -141,27 +158,44 @@ public void onMatch(RelOptRuleCall call) {
exchange = PinotLogicalExchange.create(input,
RelDistributions.hash(windowGroup.keys.toList()), prePartitioned);
} else {
// PARTITION BY and ORDER BY on different key(s)
- // Add a LogicalSortExchange hashed on the partition by keys and
collation based on order by keys.
- // The ordering itself is established by the Sort placed over the
exchange below, not by the receive
- // operator - see the comment at the transformTo call.
+ // Keep the receiver full-sort path for a partitioned exchange.
Sorting before the hash exchange compares rows
+ // routed to different receivers and blocks streaming; the explicit
Sort retained above the exchange establishes
+ // the required ordering after partitioning.
exchange = PinotLogicalSortExchange.create(input,
RelDistributions.hash(windowGroup.keys.toList()),
windowGroup.orderKeys, false, false, prePartitioned);
}
}
- // WindowAggregateOperator requires its input ordered on the ORDER BY keys
and does no ordering of its own, so
- // where the exchange carries a collation the ordering has to be
established above it. Place an explicit Sort
- // rather than asking the receive operator to sort: SortOperator is the
operator that knows fetch/offset, and
- // SortedMailboxReceiveOperator is deprecated. The Sort carries no fetch,
so it keeps every row - the same
- // semantics as the unbounded list the receive operator used.
+ // WindowAggregateOperator requires its input ordered on the ORDER BY keys
and does no ordering of its own. A sort
+ // exchange that sorts on the receiver establishes that contract itself;
adding another Sort above it would make
+ // old servers sort the same rows twice during a rolling upgrade. Keep an
explicit Sort only for post-exchange
+ // full-sort paths.
// PinotSortExchangeNodeInsertRule does not re-fire on it: its matches()
rejects a Sort whose input is an
- // exchange. PinotSortExchangeCopyRule does not either: it declines when
there is no fetch.
+ // exchange. Window sorts must retain every input row, independently of
the broker response limit.
RelNode windowInput = exchange instanceof PinotLogicalSortExchange
- ? LogicalSort.create(exchange, ((PinotLogicalSortExchange)
exchange).getCollation(), null, null) : exchange;
+ && !((PinotLogicalSortExchange) exchange).isSortOnReceiver()
Review Comment:
Updated at `85be3b0e80e4265ceeb6258e7db4582c4415511c` to reflect the current
implementation.
Removed the dead isSortOnReceiver guard and obsolete upgrade comment.
Ordinary sort exchanges retain an explicit receiver Sort; a k-way exchange
establishes ordered output. The earlier Calcite receiver MAX-fetch hardening
and 124 window JSON edits are removed. Separately, the physical converter
encodes MAX_VALUE for a complete-input window fallback so older servers that
ignore the single-run hint do not truncate it. The current core has no diff in
planner resource JSON; previous resource-suite passes are historical validation.
--
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]
