You are viewing a plain text version of this content. The canonical link for it is here.

Posted to commits@pinot.apache.org by "xiangfu0 (via GitHub)" <gi...@apache.org> on 2026/08/29 20:23:25 UTC

[PR] [MSE] Support sender-side sorting for sort exchanges [pinot]

xiangfu0 opened a new pull request, #19396:
URL: https://github.com/apache/pinot/pull/19396

   Fixes #19395.
   
   `PinotLogicalSortExchange` has carried an `isSortOnSender` flag that nothing honored: `MailboxSendOperator` always streamed its input unchanged, and `SortedMailboxReceiveOperator` always buffered every row of every sender and sorted them all at once. The planner set the flag to `false` everywhere, with TODOs saying sender-side sorting was not implemented.
   
   This PR implements it and turns it on for the window exchanges, so a WINDOW with an `ORDER BY` inside `OVER()` no longer makes one node hold the whole exchanged input before emitting its first row.
   
   ### Sender side
   
   `MailboxSendOperator` now sorts when `MailboxSendNode.isSort()` is set and the node carries a collation, using the same `SortUtils.SortComparator` the receiver used, so direction and null ordering are unchanged. It reads its input in full, sorts it, and sends it in blocks of at most 10k rows rather than as one large block, so the receiving mailbox queue keeps providing backpressure and the receiver can start merging before the whole run has arrived.
   
   Every destination receives a subsequence of that order: each exchange (`HashExchange`, `RandomExchange`, `BroadcastExchange`, `SingletonExchange`) routes the rows of a block to its destinations without reordering them, and `BlockSplitter` splits a block sequentially.
   
   The schema and aggregation functions of the blocks it builds come from the input blocks rather than from the plan node, so what the receiving end reads is what this operator would have forwarded had it not sorted. An error from the input supersedes the rows read so far, as before, and the operator stops sending as soon as the exchange reports every receiver early terminated.
   
   ### Receiver side
   
   `SortedMailboxReceiveOperator` now has two paths, chosen by `MailboxReceiveNode.isSortedOnSender()`:
   
   - **Senders sorted**: a k-way merge with one row cursor per sender. It emits the smallest head row as soon as every sender that has not finished has one, in blocks of at most 10k rows, so rows flow downstream while the exchange is still running and only the rows the merge is ahead of are held.
   - **Senders did not sort**: the existing buffer-everything-and-sort, unchanged. This is what still runs for plain `ORDER BY` exchanges.
   
   The merge reads whichever mailbox has a block ready rather than blocking on the sender it needs next. Blocking on one sender would hold the other senders' rows in their own mailboxes — less memory here — but it deadlocks: a sender blocked on a full mailbox of some other receiver can be the very sender this operator is waiting for, and with two receivers merging two senders each can end up waiting on the sender the other one is blocking, with neither making progress until the query times out. Reading whatever is ready keeps every mailbox drained, cannot deadlock, and buys incremental output at the cost of buffering how far ahead the fastest senders run — never more than the buffer-everything path it replaces.
   
   `BlockingMultiStreamConsumer` gains two accessors for this: `getLastReadStream()`, to tell which sender the block just read belongs to, and `isStreamLive(stream)`, because the round-robin read consumes each stream's EOS without returning it and the merge must know that one sender is over while the others still produce. Cancellation, deadlines, backpressure and error propagation are untouched — the merge reads through the same `readMseBlockBlocking()` the unsorted receive uses.
   
   ### Planner
   
   `PinotWindowExchangeNodeInsertRule` now creates its two sort exchanges with `isSortOnSender=true`, which is what the stale TODOs in that rule described. Nothing else changes: `PinotSortExchangeNodeInsertRule` (plain `ORDER BY`) and `PinotAggregateExchangeNodeInsertRule` keep sorting on the receiver only, so the plan of every non-window query is byte-identical. Turning it on for `ORDER BY` is a larger decision — `PinotSortExchangeCopyRule` may already have pushed a `Sort` below the exchange, so the sender would be re-sorting sorted rows — and is left to a separate change; the TODO there now says so.
   
   For a window group with `PARTITION BY` and `ORDER BY` on different keys, hashing on the partition keys sends every row of a partition to the same receiver, so merging by the order keys leaves each partition ordered — the invariant `WindowAggregateOperator` documents it depends on.
   
   ### Rolling-upgrade boundary
   
   A server that has not been upgraded ignores `MailboxSendNode.isSort()` and sends its rows unsorted, while an upgraded receiver told `sortedOnSender=true` merges them as if they were sorted. **Upgrade servers before brokers.** Until every server runs this code, a broker running it can produce a window plan whose sender does not sort and whose receiver assumes it did, which returns wrongly ordered window results for the duration of a broker-first rollout. There is no persisted-data or plan-format change: `sort` and `sortedOnSender` are existing protobuf fields, only their value changes.
   
   ### Testing
   
   - `MailboxSendOperatorTest`: the sender sorts by the collation over duplicates and nulls, honours descending with nulls first, splits the sorted run across blocks, drops the buffered rows when the input ends in error, and stops sending when the exchange reports every receiver early terminated.
   - `SortedMailboxReceiveOperatorTest`: the merge over two senders, with duplicate keys, with nulls, with one sender empty, with every sender empty, plus the error and timeout paths. The existing cases cover the unchanged buffer-and-sort path.
   - `WindowFunctions.json` gains a `sorted_exchange_window_functions` group, validated against H2 on the two-server runtime harness: `RANK`/`DENSE_RANK` over duplicate order keys, descending order keys, a peer-based running aggregate, `ORDER BY`-only windows where every sender feeds one receiver, and filters that leave most or all senders with nothing to send.
   - Those new cases were checked to be non-vacuous: with the sender-side sort removed (an old server) but the receiver still merging, 5 of the 8 new queries fail, along with 173 of the pre-existing window cases.
   - `WindowFunctionPlans.json` expectations updated for the `isSortOnSender=[true]` flag on window exchanges; no other plan expectation moved.
   
   Full `pinot-query-planner` (1539 tests) and `pinot-query-runtime` (4599 tests) suites pass.
   


-- 
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-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] [MSE] Add opt-in sender-sorted merging for global ordered windows [pinot]

Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.

xiangfu0 commented on PR #19396:
URL: https://github.com/apache/pinot/pull/19396#issuecomment-6050622277

@gortiz rebased on master `3a36f13959` and pushed `6f63db504b9858b45707ee13b08a30049ad55910` with the review fixes. All 14 inline requests have individual evidence-backed replies and are resolved.

The gate now reads only cached state on an enabled query, and shared incremental live-version tracking replaces the duplicated scanner. Merge exchange/receive are sibling nodes with dedicated visitors; legacy flags are deprecated. The single-run requirement is an explicit serialized hint, unused merge limits are removed, and the merge uses passive receive deadlines.

The new hybrid integration test passed through actual planner/proto/opchain/local+gRPC execution, comparing receiver-sort and merge results for ROWS/FOLLOWING and RANGE/peer frames with nulls, duplicate keys and all 32 offline/realtime rows. Other local coverage passed: 968 planner/serde/plan-fixture cases, 177 focused runtime cases, plus tracker/capability/query-option tests. Scoped formatting, license, checkstyle and added-line warning checks passed.

For the non-inline comments: the receiver MAX-fetch hardening, 124 JSON plan edits and unrelated annotations dependency are removed; the option/config now use `windowKWayMerge`; the no-op local/wrapper/blank-line leftovers are removed. The description now documents usage, tag 19, the visitors, the SNAPSHOT exclusion and the actual offset semantics. Heap remove/add and single-remaining-stream improvements are left out as the informational performance follow-up, without adding speculative optimization or benchmark code.

The feature remains opt-in and the memory limitation tracked by #19395 remains open. This update does not claim universal performance superiority or a validated AUTO selector. Exact-head hosted CI is running; maintainer re-review is still required.

--
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-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] [MSE] Merge sender-sorted streams for global ordered windows [pinot]

Posted by "gortiz (via GitHub)" <gi...@apache.org>.

gortiz commented on code in PR #19396:
URL: https://github.com/apache/pinot/pull/19396#discussion_r4106395375


##########
pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotWindowExchangeNodeInsertRule.java:
##########
@@ -120,13 +120,13 @@ public void onMatch(RelOptRuleCall call) {
         exchange = PinotLogicalExchange.create(input, RelDistributions.hash(List.of()));
       } else {
         // Only ORDER BY
-        // Add a LogicalSortExchange with collation on the order by key(s) and an empty hash partition key.
-        // 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.
+        // Sort each sender explicitly and merge the sorted mailbox streams at the receiver. The Sort retained above
+        // the exchange is the semantic ordering boundary and becomes a streaming limit when the merge receiver
+        // advertises this exact collation.
         // TODO: Revisit whether we should use hash distribution
         exchange =
-            PinotLogicalSortExchange.create(input, RelDistributions.hash(List.of()), windowGroup.orderKeys, false,
-                false);
+            PinotLogicalSortExchange.create(input, RelDistributions.hash(List.of()), windowGroup.orderKeys, true,

Review Comment:
   RISK (rolling upgrade): with a new broker and old servers, which is the usual controller→broker→server upgrade window, one query can sort the same rows up to three times.
   
   1) An old leaf ServerPlanRequestVisitor.visitSort pushes the new SortNode into V1 as ORDER BY ... LIMIT 2147483647. 2) An old receiver sees isSort=true and builds the deprecated SortedMailboxReceiveOperator, which does a full sort. 3) The old SortOperator above it does not know its input is sorted and sorts again. The base plan used (false,false), so it sorted once. The results stay correct, but queries are slower until all servers are upgraded. A kill switch would cover this case too.



##########
pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotWindowExchangeNodeInsertRule.java:
##########
@@ -141,18 +141,17 @@ 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.
+    // where the exchange carries a collation the ordering has to be established above it. Keep an explicit Sort as
+    // the semantic boundary. A confirmed merge receiver advertises the exact collation, allowing SortOperator to
+    // stream through it; a legacy receiver still performs the full sort.

Review Comment:
   Drop the Sort created just below (line 158) for the sortOnReceiver=true branch. It does no work at runtime, and a no-op operator should not stay in the plan. Removing a Sort whose input already provides the collation is a normal logical rewrite.
   
   With (sortOnSender=true, sortOnReceiver=true), every receiver already returns rows in order. A new receiver merges, or falls back to a full sort. An old receiver uses SortedMailboxReceiveOperator, which does a full sort. The Sort above has no fetch or offset. On new servers it becomes a LimitSortOperator with limit MAX_VALUE, which passes every row through. On old servers it sorts a second time, and that second sort is part of the triple sort during the upgrade window. Proposal: only wrap the exchange in LogicalSort when it does not sort on the receiver. Today that is the partitioned (false,false) branch. Or let a SortRemove-style rule drop the Sort because the exchange's collation already covers it. Consequences: (1) SortOperator.isInputSorted, the SortedMultiStageOperator interface, and the LimitSortOperator/SortOperator Javadoc edits exist only to turn this Sort into a no-op, so they can be removed. (2) WindowFunctionPlans.json and QueryCompilationTest expectations lose the Sor
 t above the exchange. (3) The comment that cites #19412's reason for the explicit Sort (fetch/offset handling, deprecated receive op) no longer applies here, because this Sort has no fetch.



##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/SortedMailboxMergeReceiveOperator.java:
##########
@@ -0,0 +1,668 @@
+/**
+ * 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.pinot.query.runtime.operator;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.Deque;
+import java.util.IdentityHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import javax.annotation.Nullable;
+import org.apache.calcite.rel.RelFieldCollation;
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.pinot.common.utils.DataSchema;
+import org.apache.pinot.query.mailbox.ReceivingMailbox;
+import org.apache.pinot.query.planner.plannode.MailboxReceiveNode;
+import org.apache.pinot.query.runtime.blocks.MseBlock;
+import org.apache.pinot.query.runtime.blocks.RowHeapDataBlock;
+import org.apache.pinot.query.runtime.blocks.SuccessMseBlock;
+import org.apache.pinot.query.runtime.operator.utils.AsyncStream;
+import org.apache.pinot.query.runtime.operator.utils.SortUtils;
+import org.apache.pinot.query.runtime.plan.OpChainExecutionContext;
+import org.apache.pinot.spi.query.QueryThreadContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/// Receives streams that the plan declares sorted on the sender and merges them by the exchange collation.
+///
+/// An explicit sender [SortOperator] establishes the row ordering; [MailboxSendOperator] only transports that
+/// ordering. The transport marker confirms rollout compatibility and is not itself a sorting mechanism.
+///
+/// The plan declaration alone is not trusted during a rolling upgrade. Every data block must carry the transport's
+/// sender-sort confirmation. Before this operator emits its first row it obtains a head row, or EOS, from every live
+/// sender. If any sender's first data is unconfirmed, all tentatively buffered rows are folded into a full receiver
+/// sort. Once output starts, losing the confirmation is a protocol violation because already emitted rows cannot be
+/// recovered into that fallback.
+///
+/// The merge reads whichever mailbox is ready instead of blocking on one sender. This prevents a sender that is
+/// backpressured by another receiver from creating a cross-receiver wait cycle. Rows are emitted in blocks of at most
+/// 10,000 while cursor state carries the ordering frontier across calls. A fast sender can be read ahead while another
+/// sender is starved, so retained input is workload-dependent and can approach the legacy full receiver sort in the
+/// worst case.
+///
+/// This operator is driven by a single consumer thread and is not thread-safe.
+public class SortedMailboxMergeReceiveOperator extends BaseMailboxReceiveOperator implements SortedMultiStageOperator {
+  private static final Logger LOGGER = LoggerFactory.getLogger(SortedMailboxMergeReceiveOperator.class);
+
+  private static final String EXPLAIN_NAME = "SORTED_MAILBOX_MERGE_RECEIVE";
+  private static final String MERGE_SCOPE = "SortedMailboxMergeReceiveOperator";
+  private final DataSchema _dataSchema;
+  private final List<RelFieldCollation> _collations;
+  private final Comparator<Object[]> _comparator;
+  private final SenderCursorHeap _readyCursors;
+  private final boolean _singleSortedSender;
+  /// Senders that have not finished but do not currently have a row ready. Nothing can be emitted while this is
+  /// non-empty because any one of these senders may hold the next row.
+  private final Set<SenderCursor> _starvedCursors = Collections.newSetFromMap(new IdentityHashMap<>());
+  private final Map<AsyncStream<ReceivingMailbox.MseBlockWithStats>, SenderCursor> _cursorsByStream =
+      new IdentityHashMap<>();
+  private boolean _mergeOutputStarted;
+  private boolean _fallbackToSort;
+  private boolean _tryEqualHeadMerge;
+  private int _fallbackOutputIndex = -1;
+
+  /// Rows buffered only for the mixed-version fallback. The sorted list is handed downstream as-is, so cleanup must
+  /// drop this reference rather than clear it.
+  @Nullable
+  private List<Object[]> _rows;
+
+  @Nullable
+  private MseBlock _eosBlock;
+
+  public SortedMailboxMergeReceiveOperator(OpChainExecutionContext context, MailboxReceiveNode node) {
+    super(context, node);
+    Preconditions.checkState(node.isSort(), "Receiver-side sorting must be enabled");
+    Preconditions.checkState(node.isSortedOnSender(), "Sender-side sorting must be enabled");
+    Preconditions.checkState(!CollectionUtils.isEmpty(node.getCollations()), "Field collations must be set");
+    _dataSchema = node.getDataSchema();
+    _collations = List.copyOf(node.getCollations());
+    _comparator = new SortUtils.SortComparator(_collations, false);

Review Comment:
   The comparator matches the sender side (FullSortOperator also uses SortComparator(collations, false)). Null ordering is consistent. OK.
   
   I checked this because a mismatch in null direction between sender and receiver would silently corrupt the merge.



##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/SortedMailboxMergeReceiveOperator.java:
##########
@@ -0,0 +1,668 @@
+/**
+ * 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.pinot.query.runtime.operator;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.Deque;
+import java.util.IdentityHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import javax.annotation.Nullable;
+import org.apache.calcite.rel.RelFieldCollation;
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.pinot.common.utils.DataSchema;
+import org.apache.pinot.query.mailbox.ReceivingMailbox;
+import org.apache.pinot.query.planner.plannode.MailboxReceiveNode;
+import org.apache.pinot.query.runtime.blocks.MseBlock;
+import org.apache.pinot.query.runtime.blocks.RowHeapDataBlock;
+import org.apache.pinot.query.runtime.blocks.SuccessMseBlock;
+import org.apache.pinot.query.runtime.operator.utils.AsyncStream;
+import org.apache.pinot.query.runtime.operator.utils.SortUtils;
+import org.apache.pinot.query.runtime.plan.OpChainExecutionContext;
+import org.apache.pinot.spi.query.QueryThreadContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/// Receives streams that the plan declares sorted on the sender and merges them by the exchange collation.
+///
+/// An explicit sender [SortOperator] establishes the row ordering; [MailboxSendOperator] only transports that
+/// ordering. The transport marker confirms rollout compatibility and is not itself a sorting mechanism.
+///
+/// The plan declaration alone is not trusted during a rolling upgrade. Every data block must carry the transport's
+/// sender-sort confirmation. Before this operator emits its first row it obtains a head row, or EOS, from every live
+/// sender. If any sender's first data is unconfirmed, all tentatively buffered rows are folded into a full receiver
+/// sort. Once output starts, losing the confirmation is a protocol violation because already emitted rows cannot be
+/// recovered into that fallback.
+///
+/// The merge reads whichever mailbox is ready instead of blocking on one sender. This prevents a sender that is
+/// backpressured by another receiver from creating a cross-receiver wait cycle. Rows are emitted in blocks of at most
+/// 10,000 while cursor state carries the ordering frontier across calls. A fast sender can be read ahead while another
+/// sender is starved, so retained input is workload-dependent and can approach the legacy full receiver sort in the
+/// worst case.
+///
+/// This operator is driven by a single consumer thread and is not thread-safe.
+public class SortedMailboxMergeReceiveOperator extends BaseMailboxReceiveOperator implements SortedMultiStageOperator {
+  private static final Logger LOGGER = LoggerFactory.getLogger(SortedMailboxMergeReceiveOperator.class);
+
+  private static final String EXPLAIN_NAME = "SORTED_MAILBOX_MERGE_RECEIVE";
+  private static final String MERGE_SCOPE = "SortedMailboxMergeReceiveOperator";
+  private final DataSchema _dataSchema;
+  private final List<RelFieldCollation> _collations;
+  private final Comparator<Object[]> _comparator;
+  private final SenderCursorHeap _readyCursors;
+  private final boolean _singleSortedSender;
+  /// Senders that have not finished but do not currently have a row ready. Nothing can be emitted while this is
+  /// non-empty because any one of these senders may hold the next row.
+  private final Set<SenderCursor> _starvedCursors = Collections.newSetFromMap(new IdentityHashMap<>());
+  private final Map<AsyncStream<ReceivingMailbox.MseBlockWithStats>, SenderCursor> _cursorsByStream =
+      new IdentityHashMap<>();
+  private boolean _mergeOutputStarted;
+  private boolean _fallbackToSort;
+  private boolean _tryEqualHeadMerge;
+  private int _fallbackOutputIndex = -1;
+
+  /// Rows buffered only for the mixed-version fallback. The sorted list is handed downstream as-is, so cleanup must
+  /// drop this reference rather than clear it.
+  @Nullable
+  private List<Object[]> _rows;
+
+  @Nullable
+  private MseBlock _eosBlock;
+
+  public SortedMailboxMergeReceiveOperator(OpChainExecutionContext context, MailboxReceiveNode node) {
+    super(context, node);
+    Preconditions.checkState(node.isSort(), "Receiver-side sorting must be enabled");
+    Preconditions.checkState(node.isSortedOnSender(), "Sender-side sorting must be enabled");
+    Preconditions.checkState(!CollectionUtils.isEmpty(node.getCollations()), "Field collations must be set");
+    _dataSchema = node.getDataSchema();
+    _collations = List.copyOf(node.getCollations());
+    _comparator = new SortUtils.SortComparator(_collations, false);
+    List<AsyncStream<ReceivingMailbox.MseBlockWithStats>> streams = _multiConsumer.getLiveStreamsSnapshot();
+    _readyCursors = new SenderCursorHeap(streams.size(), _comparator);
+    _singleSortedSender = streams.size() == 1;
+    if (!_singleSortedSender) {
+      for (AsyncStream<ReceivingMailbox.MseBlockWithStats> stream : streams) {
+        SenderCursor cursor = new SenderCursor(stream);
+        _cursorsByStream.put(stream, cursor);
+        _starvedCursors.add(cursor);
+      }
+    }
+  }
+
+  @Override
+  protected Logger logger() {
+    return LOGGER;
+  }
+
+  @Override
+  public String toExplainString() {
+    return EXPLAIN_NAME;
+  }
+
+  @Override
+  public List<RelFieldCollation> getCollations() {
+    return _collations;
+  }
+
+  @Override
+  protected MseBlock getNextBlock() {
+    if (_fallbackOutputIndex >= 0 && _rows != null) {
+      return emitFallbackBlock();
+    }
+    if (_eosBlock != null) {
+      return _eosBlock;
+    }
+    if (_isEarlyTerminated) {
+      return readUntilEos();
+    }
+    return _singleSortedSender ? readSingleSortedSender() : mergeNextBlock();
+  }
+
+  /// Passes through one confirmed sorted sender without copying its rows through the merge heap.
+  private MseBlock readSingleSortedSender() {
+    while (true) {
+      MseBlock block = _multiConsumer.readMseBlockBlocking();
+      if (block.isEos()) {
+        return terminate(block);
+      }
+      MseBlock.Data dataBlock = (MseBlock.Data) block;
+      checkActiveTerminationAndSampleUsage();
+      if (!_multiConsumer.isLastBlockSortedOnSender()) {
+        fallbackToFullSort(dataBlock.asRowHeap().getRows());
+        return sortAllRows();
+      }
+      if (dataBlock.getNumRows() > 0) {
+        _mergeOutputStarted = true;
+        return dataBlock;
+      }
+    }
+  }
+
+  /// Merges the sorted senders, emitting at most [SortOperator#DEFAULT_MAX_ROWS_PER_BLOCK] rows per call.
+  private MseBlock mergeNextBlock() {
+    ArrayList<Object[]> rows = new ArrayList<>(0);
+    while (rows.size() < SortOperator.DEFAULT_MAX_ROWS_PER_BLOCK) {
+      if (!_starvedCursors.isEmpty()) {
+        MseBlock.Eos error;
+        boolean receivedMoreRows = false;
+        if (rows.isEmpty()) {
+          error = readOneBlock();
+        } else {
+          // Rows already removed from the heap are a globally ordered prefix. Consume any immediately available
+          // cursor progress so blocks can still be coalesced, but return that safe prefix instead of waiting merely
+          // to fill the output block.
+          MseBlock block = _multiConsumer.pollMseBlockOrStreamCompletion();
+          if (block == null && _multiConsumer.getFinishedStreamsLastRead().isEmpty()) {
+            break;
+          }
+          error = processReadBlock(block);
+          receivedMoreRows = block != null && block.isData() && ((MseBlock.Data) block).getNumRows() > 0;
+        }
+        if (error != null) {
+          return terminate(error);
+        }
+        if (_fallbackToSort) {
+          // These rows were already removed from cursors while building this not-yet-emitted block.
+          _rows.addAll(rows);
+          return sortAllRows();
+        }
+        if (receivedMoreRows) {
+          // A refill proves this is not a one-block result. Restore the established full-block capacity once so
+          // fragmented input cannot trigger repeated growth while this output block is assembled.
+          rows.ensureCapacity(SortOperator.DEFAULT_MAX_ROWS_PER_BLOCK);
+        }
+        continue;
+      }
+      if (_readyCursors.isEmpty()) {
+        break;
+      }
+      if (rows.isEmpty()) {
+        // Keep empty and tiny results cheap without sacrificing the one-allocation path for full output blocks.
+        // Sum all currently buffered rows once; later output blocks retain the established full-block capacity.
+        int initialCapacity = _mergeOutputStarted ? SortOperator.DEFAULT_MAX_ROWS_PER_BLOCK
+            : _readyCursors.getCappedAvailableRowCount(SortOperator.DEFAULT_MAX_ROWS_PER_BLOCK);
+        // ArrayList grows by 1.5x. Round near-full first blocks up now so a small refill cannot allocate an oversized
+        // replacement in addition to the nearly full initial array.
+        if (initialCapacity + (initialCapacity >> 1) >= SortOperator.DEFAULT_MAX_ROWS_PER_BLOCK) {
+          initialCapacity = SortOperator.DEFAULT_MAX_ROWS_PER_BLOCK;
+        }
+        rows.ensureCapacity(initialCapacity);
+      }
+      if (_tryEqualHeadMerge && _readyCursors.size() > 1) {

Review Comment:
   The equal-head fast path together with the lazy 'dirty' heap adds the most complexity in this PR. I traced it and it looks correct, but the invariants are implicit.
   
   Correctness depends on: dirty=true only when active==oldSize>1; every path that sets _tryEqualHeadMerge=false either heapified already or hits the else branch at 236; add() is only reached when a cursor was starved, which implies the heap was heapified. Only one targeted test covers this (shouldMergeEqualHeadsAcrossRefillIntoBoundedBlocks). A randomized test would be much cheaper insurance than the reasoning above: N senders, random block splits, random tie density, random starvation order, and output compared against a full sort.



##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/SortedMailboxMergeReceiveOperator.java:
##########
@@ -0,0 +1,668 @@
+/**
+ * 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.pinot.query.runtime.operator;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.Deque;
+import java.util.IdentityHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import javax.annotation.Nullable;
+import org.apache.calcite.rel.RelFieldCollation;
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.pinot.common.utils.DataSchema;
+import org.apache.pinot.query.mailbox.ReceivingMailbox;
+import org.apache.pinot.query.planner.plannode.MailboxReceiveNode;
+import org.apache.pinot.query.runtime.blocks.MseBlock;
+import org.apache.pinot.query.runtime.blocks.RowHeapDataBlock;
+import org.apache.pinot.query.runtime.blocks.SuccessMseBlock;
+import org.apache.pinot.query.runtime.operator.utils.AsyncStream;
+import org.apache.pinot.query.runtime.operator.utils.SortUtils;
+import org.apache.pinot.query.runtime.plan.OpChainExecutionContext;
+import org.apache.pinot.spi.query.QueryThreadContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/// Receives streams that the plan declares sorted on the sender and merges them by the exchange collation.
+///
+/// An explicit sender [SortOperator] establishes the row ordering; [MailboxSendOperator] only transports that
+/// ordering. The transport marker confirms rollout compatibility and is not itself a sorting mechanism.
+///
+/// The plan declaration alone is not trusted during a rolling upgrade. Every data block must carry the transport's
+/// sender-sort confirmation. Before this operator emits its first row it obtains a head row, or EOS, from every live
+/// sender. If any sender's first data is unconfirmed, all tentatively buffered rows are folded into a full receiver
+/// sort. Once output starts, losing the confirmation is a protocol violation because already emitted rows cannot be
+/// recovered into that fallback.
+///
+/// The merge reads whichever mailbox is ready instead of blocking on one sender. This prevents a sender that is
+/// backpressured by another receiver from creating a cross-receiver wait cycle. Rows are emitted in blocks of at most
+/// 10,000 while cursor state carries the ordering frontier across calls. A fast sender can be read ahead while another
+/// sender is starved, so retained input is workload-dependent and can approach the legacy full receiver sort in the
+/// worst case.
+///
+/// This operator is driven by a single consumer thread and is not thread-safe.
+public class SortedMailboxMergeReceiveOperator extends BaseMailboxReceiveOperator implements SortedMultiStageOperator {
+  private static final Logger LOGGER = LoggerFactory.getLogger(SortedMailboxMergeReceiveOperator.class);
+
+  private static final String EXPLAIN_NAME = "SORTED_MAILBOX_MERGE_RECEIVE";
+  private static final String MERGE_SCOPE = "SortedMailboxMergeReceiveOperator";
+  private final DataSchema _dataSchema;
+  private final List<RelFieldCollation> _collations;
+  private final Comparator<Object[]> _comparator;
+  private final SenderCursorHeap _readyCursors;
+  private final boolean _singleSortedSender;
+  /// Senders that have not finished but do not currently have a row ready. Nothing can be emitted while this is
+  /// non-empty because any one of these senders may hold the next row.
+  private final Set<SenderCursor> _starvedCursors = Collections.newSetFromMap(new IdentityHashMap<>());
+  private final Map<AsyncStream<ReceivingMailbox.MseBlockWithStats>, SenderCursor> _cursorsByStream =
+      new IdentityHashMap<>();
+  private boolean _mergeOutputStarted;
+  private boolean _fallbackToSort;
+  private boolean _tryEqualHeadMerge;
+  private int _fallbackOutputIndex = -1;
+
+  /// Rows buffered only for the mixed-version fallback. The sorted list is handed downstream as-is, so cleanup must
+  /// drop this reference rather than clear it.
+  @Nullable
+  private List<Object[]> _rows;
+
+  @Nullable
+  private MseBlock _eosBlock;
+
+  public SortedMailboxMergeReceiveOperator(OpChainExecutionContext context, MailboxReceiveNode node) {
+    super(context, node);
+    Preconditions.checkState(node.isSort(), "Receiver-side sorting must be enabled");
+    Preconditions.checkState(node.isSortedOnSender(), "Sender-side sorting must be enabled");
+    Preconditions.checkState(!CollectionUtils.isEmpty(node.getCollations()), "Field collations must be set");
+    _dataSchema = node.getDataSchema();
+    _collations = List.copyOf(node.getCollations());
+    _comparator = new SortUtils.SortComparator(_collations, false);
+    List<AsyncStream<ReceivingMailbox.MseBlockWithStats>> streams = _multiConsumer.getLiveStreamsSnapshot();
+    _readyCursors = new SenderCursorHeap(streams.size(), _comparator);
+    _singleSortedSender = streams.size() == 1;
+    if (!_singleSortedSender) {
+      for (AsyncStream<ReceivingMailbox.MseBlockWithStats> stream : streams) {
+        SenderCursor cursor = new SenderCursor(stream);
+        _cursorsByStream.put(stream, cursor);
+        _starvedCursors.add(cursor);
+      }
+    }
+  }
+
+  @Override
+  protected Logger logger() {
+    return LOGGER;
+  }
+
+  @Override
+  public String toExplainString() {
+    return EXPLAIN_NAME;
+  }
+
+  @Override
+  public List<RelFieldCollation> getCollations() {
+    return _collations;
+  }
+
+  @Override
+  protected MseBlock getNextBlock() {
+    if (_fallbackOutputIndex >= 0 && _rows != null) {
+      return emitFallbackBlock();
+    }
+    if (_eosBlock != null) {
+      return _eosBlock;
+    }
+    if (_isEarlyTerminated) {
+      return readUntilEos();
+    }
+    return _singleSortedSender ? readSingleSortedSender() : mergeNextBlock();
+  }
+
+  /// Passes through one confirmed sorted sender without copying its rows through the merge heap.
+  private MseBlock readSingleSortedSender() {
+    while (true) {
+      MseBlock block = _multiConsumer.readMseBlockBlocking();
+      if (block.isEos()) {
+        return terminate(block);
+      }
+      MseBlock.Data dataBlock = (MseBlock.Data) block;
+      checkActiveTerminationAndSampleUsage();
+      if (!_multiConsumer.isLastBlockSortedOnSender()) {
+        fallbackToFullSort(dataBlock.asRowHeap().getRows());
+        return sortAllRows();
+      }
+      if (dataBlock.getNumRows() > 0) {
+        _mergeOutputStarted = true;
+        return dataBlock;
+      }
+    }
+  }
+
+  /// Merges the sorted senders, emitting at most [SortOperator#DEFAULT_MAX_ROWS_PER_BLOCK] rows per call.
+  private MseBlock mergeNextBlock() {
+    ArrayList<Object[]> rows = new ArrayList<>(0);
+    while (rows.size() < SortOperator.DEFAULT_MAX_ROWS_PER_BLOCK) {
+      if (!_starvedCursors.isEmpty()) {

Review Comment:
   MEMORY: when one cursor is starved, readOneBlock pulls from any ready mailbox into unbounded per-cursor _pending deques. This effectively removes mailbox backpressure for the fast senders.
   
   The PR describes this, and the worst case equals the old full receiver sort. The difference is that sender-side FullSortOperators now also hold their full buffers during the send, so peak cluster memory can be higher than on master when one sender is slow. Consider a follow-up that bounds pending rows per cursor, or at least a stat (for example max retained rows) so this can be observed in production.



##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/SortedMailboxMergeReceiveOperator.java:
##########
@@ -0,0 +1,668 @@
+/**
+ * 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.pinot.query.runtime.operator;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.Deque;
+import java.util.IdentityHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import javax.annotation.Nullable;
+import org.apache.calcite.rel.RelFieldCollation;
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.pinot.common.utils.DataSchema;
+import org.apache.pinot.query.mailbox.ReceivingMailbox;
+import org.apache.pinot.query.planner.plannode.MailboxReceiveNode;
+import org.apache.pinot.query.runtime.blocks.MseBlock;
+import org.apache.pinot.query.runtime.blocks.RowHeapDataBlock;
+import org.apache.pinot.query.runtime.blocks.SuccessMseBlock;
+import org.apache.pinot.query.runtime.operator.utils.AsyncStream;
+import org.apache.pinot.query.runtime.operator.utils.SortUtils;
+import org.apache.pinot.query.runtime.plan.OpChainExecutionContext;
+import org.apache.pinot.spi.query.QueryThreadContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/// Receives streams that the plan declares sorted on the sender and merges them by the exchange collation.
+///
+/// An explicit sender [SortOperator] establishes the row ordering; [MailboxSendOperator] only transports that
+/// ordering. The transport marker confirms rollout compatibility and is not itself a sorting mechanism.
+///
+/// The plan declaration alone is not trusted during a rolling upgrade. Every data block must carry the transport's
+/// sender-sort confirmation. Before this operator emits its first row it obtains a head row, or EOS, from every live
+/// sender. If any sender's first data is unconfirmed, all tentatively buffered rows are folded into a full receiver
+/// sort. Once output starts, losing the confirmation is a protocol violation because already emitted rows cannot be
+/// recovered into that fallback.
+///
+/// The merge reads whichever mailbox is ready instead of blocking on one sender. This prevents a sender that is
+/// backpressured by another receiver from creating a cross-receiver wait cycle. Rows are emitted in blocks of at most
+/// 10,000 while cursor state carries the ordering frontier across calls. A fast sender can be read ahead while another
+/// sender is starved, so retained input is workload-dependent and can approach the legacy full receiver sort in the
+/// worst case.
+///
+/// This operator is driven by a single consumer thread and is not thread-safe.
+public class SortedMailboxMergeReceiveOperator extends BaseMailboxReceiveOperator implements SortedMultiStageOperator {
+  private static final Logger LOGGER = LoggerFactory.getLogger(SortedMailboxMergeReceiveOperator.class);
+
+  private static final String EXPLAIN_NAME = "SORTED_MAILBOX_MERGE_RECEIVE";
+  private static final String MERGE_SCOPE = "SortedMailboxMergeReceiveOperator";
+  private final DataSchema _dataSchema;
+  private final List<RelFieldCollation> _collations;
+  private final Comparator<Object[]> _comparator;
+  private final SenderCursorHeap _readyCursors;
+  private final boolean _singleSortedSender;
+  /// Senders that have not finished but do not currently have a row ready. Nothing can be emitted while this is
+  /// non-empty because any one of these senders may hold the next row.
+  private final Set<SenderCursor> _starvedCursors = Collections.newSetFromMap(new IdentityHashMap<>());
+  private final Map<AsyncStream<ReceivingMailbox.MseBlockWithStats>, SenderCursor> _cursorsByStream =
+      new IdentityHashMap<>();
+  private boolean _mergeOutputStarted;
+  private boolean _fallbackToSort;
+  private boolean _tryEqualHeadMerge;
+  private int _fallbackOutputIndex = -1;
+
+  /// Rows buffered only for the mixed-version fallback. The sorted list is handed downstream as-is, so cleanup must
+  /// drop this reference rather than clear it.
+  @Nullable
+  private List<Object[]> _rows;
+
+  @Nullable
+  private MseBlock _eosBlock;
+
+  public SortedMailboxMergeReceiveOperator(OpChainExecutionContext context, MailboxReceiveNode node) {
+    super(context, node);
+    Preconditions.checkState(node.isSort(), "Receiver-side sorting must be enabled");
+    Preconditions.checkState(node.isSortedOnSender(), "Sender-side sorting must be enabled");
+    Preconditions.checkState(!CollectionUtils.isEmpty(node.getCollations()), "Field collations must be set");
+    _dataSchema = node.getDataSchema();
+    _collations = List.copyOf(node.getCollations());
+    _comparator = new SortUtils.SortComparator(_collations, false);
+    List<AsyncStream<ReceivingMailbox.MseBlockWithStats>> streams = _multiConsumer.getLiveStreamsSnapshot();
+    _readyCursors = new SenderCursorHeap(streams.size(), _comparator);
+    _singleSortedSender = streams.size() == 1;
+    if (!_singleSortedSender) {
+      for (AsyncStream<ReceivingMailbox.MseBlockWithStats> stream : streams) {
+        SenderCursor cursor = new SenderCursor(stream);
+        _cursorsByStream.put(stream, cursor);
+        _starvedCursors.add(cursor);
+      }
+    }
+  }
+
+  @Override
+  protected Logger logger() {
+    return LOGGER;
+  }
+
+  @Override
+  public String toExplainString() {
+    return EXPLAIN_NAME;
+  }
+
+  @Override
+  public List<RelFieldCollation> getCollations() {
+    return _collations;
+  }
+
+  @Override
+  protected MseBlock getNextBlock() {
+    if (_fallbackOutputIndex >= 0 && _rows != null) {
+      return emitFallbackBlock();
+    }
+    if (_eosBlock != null) {
+      return _eosBlock;
+    }
+    if (_isEarlyTerminated) {
+      return readUntilEos();
+    }
+    return _singleSortedSender ? readSingleSortedSender() : mergeNextBlock();
+  }
+
+  /// Passes through one confirmed sorted sender without copying its rows through the merge heap.
+  private MseBlock readSingleSortedSender() {
+    while (true) {
+      MseBlock block = _multiConsumer.readMseBlockBlocking();
+      if (block.isEos()) {
+        return terminate(block);
+      }
+      MseBlock.Data dataBlock = (MseBlock.Data) block;
+      checkActiveTerminationAndSampleUsage();
+      if (!_multiConsumer.isLastBlockSortedOnSender()) {

Review Comment:
   On the single-sender path, blocks pass through unsplit, so output blocks can be larger than 10k rows. That is the same as today's receive, but the class Javadoc says output blocks are capped at 10,000 rows.
   
   Align the Javadoc or the PR text.



##########
pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotWindowExchangeNodeInsertRule.java:
##########
@@ -120,13 +120,13 @@ public void onMatch(RelOptRuleCall call) {
         exchange = PinotLogicalExchange.create(input, RelDistributions.hash(List.of()));
       } else {
         // Only ORDER BY
-        // Add a LogicalSortExchange with collation on the order by key(s) and an empty hash partition key.
-        // 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.
+        // Sort each sender explicitly and merge the sorted mailbox streams at the receiver. The Sort retained above
+        // the exchange is the semantic ordering boundary and becomes a streaming limit when the merge receiver
+        // advertises this exact collation.
         // TODO: Revisit whether we should use hash distribution
         exchange =
-            PinotLogicalSortExchange.create(input, RelDistributions.hash(List.of()), windowGroup.orderKeys, false,
-                false);
+            PinotLogicalSortExchange.create(input, RelDistributions.hash(List.of()), windowGroup.orderKeys, true,

Review Comment:
   RISK: there is no kill switch. Every global ORDER BY window now takes the sender-sort + merge path, and the PR's own numbers show +53% on presorted input and +2-4% on small or single-sender cases. Please add a broker config plus a query option to turn it off.
   
   Use the same style of mechanism as the normal sort-exchange copy (broker config CONFIG_OF_SORT_EXCHANGE_COPY_THRESHOLD + query option sortExchangeCopyThreshold). The trigger must not be that threshold itself, though. The threshold is keyed on the Sort's fetch, and the window Sort has no fetch: a query LIMIT sits above the window and is not pushed through it. The sender sort is therefore always a full sort, and a fetch threshold would never enable this path. The L x W amplification that motivates the threshold also does not happen here: a global window already sends all N rows to one receiver, so sorting on the senders does not change how many rows are sent. The costs here are different: merge CPU, memory held by both sender buffers and receiver read-ahead, and the upgrade-window extra sorts. So a boolean (for example a windowSortOnSender option, default on or off per the author's call) fits better. A threshold on something meaningful, such as the number of senders, would also work
 . If a LIMIT is ever pushed through ROW_NUMBER/RANK or other prefix-frame windows, the senders would do a top-L; that is exactly the normal-sort case and should reuse sortExchangeCopyThreshold. The switch also gives operators a way back from a production regression without a rollback.



##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/GrpcSendingMailbox.java:
##########
@@ -432,12 +435,14 @@ protected void sendContent(ByteString byteString, boolean waitForMore, boolean b
       if (!bypassReady && isTerminated()) {
         return;
       }
-      MailboxContent content = MailboxContent.newBuilder()
+      MailboxContent.Builder contentBuilder = MailboxContent.newBuilder()
           .setMailboxId(_id)
           .setPayload(byteString)
-          .setWaitForMore(waitForMore)
-          .build();
-      _contentObserver.onNext(content);
+          .setWaitForMore(waitForMore);
+      if (_sortedOnSender) {
+        contentBuilder.putMetadata(ChannelUtils.MAILBOX_METADATA_SORTED_ON_SENDER, Boolean.TRUE.toString());

Review Comment:
   nit: the marker is added to every chunk, including split parts (waitForMore=true) and EOS.
   
   MailboxContentObserver reads the flag only from the message that completes the block, so the flag on the other chunks is unused. It costs 26 bytes each, which is negligible. The flag is set per stream but repeated per message; that is fine for simplicity.



##########
pinot-query-planner/src/main/java/org/apache/pinot/query/planner/logical/PlanFragmenter.java:
##########
@@ -185,6 +186,16 @@ public PlanNode visitExchange(ExchangeNode node, Context context) {
 
     // Create a new context for the next PlanFragment with MailboxSendNode as the root node.
     PlanNode nextPlanFragmentRoot = node.getInputs().get(0).visit(this, new Context(senderPlanFragmentId));
+    if (node.isSortOnSender()) {
+      Preconditions.checkState(!node.getCollations().isEmpty(),
+          "Sender sorting requires a non-empty exchange collation");
+      // Ordering belongs to an explicit operator in the sender fragment. MailboxSendOperator only preserves and
+      // transports this output; ServerPlanRequestVisitor keeps this SortNode above the V1 leaf boundary.
+      // SortOperator applies the broker response limit when fetch is absent. This internal sort must retain every
+      // sender row, so use the largest representable fetch with a zero effective offset.
+      nextPlanFragmentRoot = new SortNode(senderPlanFragmentId, nextPlanFragmentRoot.getDataSchema(), null,
+          List.of(nextPlanFragmentRoot), node.getCollations(), Integer.MAX_VALUE, -1);

Review Comment:
   Inserting a physical SortNode inside the fragmenter works, but it makes the stage plan differ from the RelNode plan.
   
   EXPLAIN (logical) shows only isSortOnSender=[true]. The SortNode with fetch=2147483647 appears only in the fragmented or implementation plan. That is acceptable, but please confirm that EXPLAIN IMPLEMENTATION PLAN renders it clearly. Also note that the MAX_VALUE fetch reaches old servers' V1 leaf as LIMIT 2147483647, and that LIMIT replaces any multiStageLeafLimit. On the new server path it does not, because the sort stays above the leaf boundary.



##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MailboxSendOperator.java:
##########
@@ -64,13 +67,17 @@ public class MailboxSendOperator extends MultiStageOperator {
   private final BlockExchange _exchange;
   private final StatMap<StatKey> _statMap = new StatMap<>(StatKey.class);
 
-  // TODO: Support sort on sender
   public MailboxSendOperator(OpChainExecutionContext context, MultiStageOperator input, MailboxSendNode node) {
-    this(context, input, statMap -> getBlockExchange(context, node, statMap));
+    this(context, input, statMap -> getBlockExchange(context, node, statMap, isSortedOnSender(input, node)));
     _statMap.merge(StatKey.STAGE, context.getStageId());
     _statMap.merge(StatKey.PARALLELISM, 1);
   }
 
+  @VisibleForTesting
+  static boolean isSortedOnSender(MultiStageOperator input, MailboxSendNode node) {
+    return input instanceof SortOperator && node.hasExplicitSortInput();

Review Comment:
   The marker is keyed on 'input instanceof SortOperator'. That holds for all three implementations (Limit/TopN/Full), and each one produces a sorted stream here. OK, but the check is fragile.
   
   If someone later wraps the sender op-chain (for example a stats or tracing decorator), the marker silently disappears. Every query then takes fallbackToFullSort, and nothing fails. Consider asking the op for its collation instead, since SortedMultiStageOperator already exists. Or add a test in MailboxSendOperatorTest that builds the real op-chain from PlanNodeToOpChain.



##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/utils/BlockingMultiStreamConsumer.java:
##########
@@ -269,6 +315,37 @@ private E readDroppingSuccessEos() {
     return block;
   }
 
+  /// Returns the stream that produced the element the last blocking read returned.
+  ///
+  /// This is only meaningful right after that call returned a data or an error element. The element that ends all
+  /// the streams is not produced by any of them, and a stream that emitted its EOS is dropped from the ones this
+  /// consumer tracks, so `null` is returned in both cases.
+  ///
+  /// Consumers that need to keep the elements of each stream apart, like a merge of already sorted streams, use
+  /// this to tell which stream the element they just read belongs to.
+  ///
+  /// This method is called by the consumer thread.
+  @Nullable
+  public AsyncStream<E> getLastReadStream() {

Review Comment:
   API surface: this adds 3 accessors with 'only valid right after the last read' semantics (getLastReadStream, getFinishedStreamsLastRead view, live snapshot). They are easy to misuse.
   
   Only the merge receiver uses them. A single read method that returns a small result (block, source stream, finished streams) would make the protocol explicit instead of relying on temporal coupling. This is not blocking.



##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/SortedMailboxMergeReceiveOperator.java:
##########
@@ -0,0 +1,668 @@
+/**
+ * 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.pinot.query.runtime.operator;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.Deque;
+import java.util.IdentityHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import javax.annotation.Nullable;
+import org.apache.calcite.rel.RelFieldCollation;
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.pinot.common.utils.DataSchema;
+import org.apache.pinot.query.mailbox.ReceivingMailbox;
+import org.apache.pinot.query.planner.plannode.MailboxReceiveNode;
+import org.apache.pinot.query.runtime.blocks.MseBlock;
+import org.apache.pinot.query.runtime.blocks.RowHeapDataBlock;
+import org.apache.pinot.query.runtime.blocks.SuccessMseBlock;
+import org.apache.pinot.query.runtime.operator.utils.AsyncStream;
+import org.apache.pinot.query.runtime.operator.utils.SortUtils;
+import org.apache.pinot.query.runtime.plan.OpChainExecutionContext;
+import org.apache.pinot.spi.query.QueryThreadContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/// Receives streams that the plan declares sorted on the sender and merges them by the exchange collation.
+///
+/// An explicit sender [SortOperator] establishes the row ordering; [MailboxSendOperator] only transports that
+/// ordering. The transport marker confirms rollout compatibility and is not itself a sorting mechanism.
+///
+/// The plan declaration alone is not trusted during a rolling upgrade. Every data block must carry the transport's
+/// sender-sort confirmation. Before this operator emits its first row it obtains a head row, or EOS, from every live
+/// sender. If any sender's first data is unconfirmed, all tentatively buffered rows are folded into a full receiver
+/// sort. Once output starts, losing the confirmation is a protocol violation because already emitted rows cannot be
+/// recovered into that fallback.
+///
+/// The merge reads whichever mailbox is ready instead of blocking on one sender. This prevents a sender that is
+/// backpressured by another receiver from creating a cross-receiver wait cycle. Rows are emitted in blocks of at most
+/// 10,000 while cursor state carries the ordering frontier across calls. A fast sender can be read ahead while another
+/// sender is starved, so retained input is workload-dependent and can approach the legacy full receiver sort in the
+/// worst case.
+///
+/// This operator is driven by a single consumer thread and is not thread-safe.
+public class SortedMailboxMergeReceiveOperator extends BaseMailboxReceiveOperator implements SortedMultiStageOperator {
+  private static final Logger LOGGER = LoggerFactory.getLogger(SortedMailboxMergeReceiveOperator.class);
+
+  private static final String EXPLAIN_NAME = "SORTED_MAILBOX_MERGE_RECEIVE";
+  private static final String MERGE_SCOPE = "SortedMailboxMergeReceiveOperator";
+  private final DataSchema _dataSchema;
+  private final List<RelFieldCollation> _collations;
+  private final Comparator<Object[]> _comparator;
+  private final SenderCursorHeap _readyCursors;
+  private final boolean _singleSortedSender;
+  /// Senders that have not finished but do not currently have a row ready. Nothing can be emitted while this is
+  /// non-empty because any one of these senders may hold the next row.
+  private final Set<SenderCursor> _starvedCursors = Collections.newSetFromMap(new IdentityHashMap<>());
+  private final Map<AsyncStream<ReceivingMailbox.MseBlockWithStats>, SenderCursor> _cursorsByStream =
+      new IdentityHashMap<>();
+  private boolean _mergeOutputStarted;
+  private boolean _fallbackToSort;
+  private boolean _tryEqualHeadMerge;
+  private int _fallbackOutputIndex = -1;
+
+  /// Rows buffered only for the mixed-version fallback. The sorted list is handed downstream as-is, so cleanup must
+  /// drop this reference rather than clear it.
+  @Nullable
+  private List<Object[]> _rows;
+
+  @Nullable
+  private MseBlock _eosBlock;
+
+  public SortedMailboxMergeReceiveOperator(OpChainExecutionContext context, MailboxReceiveNode node) {
+    super(context, node);
+    Preconditions.checkState(node.isSort(), "Receiver-side sorting must be enabled");
+    Preconditions.checkState(node.isSortedOnSender(), "Sender-side sorting must be enabled");
+    Preconditions.checkState(!CollectionUtils.isEmpty(node.getCollations()), "Field collations must be set");
+    _dataSchema = node.getDataSchema();
+    _collations = List.copyOf(node.getCollations());
+    _comparator = new SortUtils.SortComparator(_collations, false);
+    List<AsyncStream<ReceivingMailbox.MseBlockWithStats>> streams = _multiConsumer.getLiveStreamsSnapshot();
+    _readyCursors = new SenderCursorHeap(streams.size(), _comparator);
+    _singleSortedSender = streams.size() == 1;
+    if (!_singleSortedSender) {
+      for (AsyncStream<ReceivingMailbox.MseBlockWithStats> stream : streams) {
+        SenderCursor cursor = new SenderCursor(stream);
+        _cursorsByStream.put(stream, cursor);
+        _starvedCursors.add(cursor);
+      }
+    }
+  }
+
+  @Override
+  protected Logger logger() {
+    return LOGGER;
+  }
+
+  @Override
+  public String toExplainString() {
+    return EXPLAIN_NAME;
+  }
+
+  @Override
+  public List<RelFieldCollation> getCollations() {
+    return _collations;
+  }
+
+  @Override
+  protected MseBlock getNextBlock() {
+    if (_fallbackOutputIndex >= 0 && _rows != null) {
+      return emitFallbackBlock();
+    }
+    if (_eosBlock != null) {
+      return _eosBlock;
+    }
+    if (_isEarlyTerminated) {
+      return readUntilEos();
+    }
+    return _singleSortedSender ? readSingleSortedSender() : mergeNextBlock();
+  }
+
+  /// Passes through one confirmed sorted sender without copying its rows through the merge heap.
+  private MseBlock readSingleSortedSender() {
+    while (true) {
+      MseBlock block = _multiConsumer.readMseBlockBlocking();
+      if (block.isEos()) {
+        return terminate(block);
+      }
+      MseBlock.Data dataBlock = (MseBlock.Data) block;
+      checkActiveTerminationAndSampleUsage();
+      if (!_multiConsumer.isLastBlockSortedOnSender()) {
+        fallbackToFullSort(dataBlock.asRowHeap().getRows());
+        return sortAllRows();
+      }
+      if (dataBlock.getNumRows() > 0) {
+        _mergeOutputStarted = true;
+        return dataBlock;
+      }
+    }
+  }
+
+  /// Merges the sorted senders, emitting at most [SortOperator#DEFAULT_MAX_ROWS_PER_BLOCK] rows per call.
+  private MseBlock mergeNextBlock() {
+    ArrayList<Object[]> rows = new ArrayList<>(0);
+    while (rows.size() < SortOperator.DEFAULT_MAX_ROWS_PER_BLOCK) {
+      if (!_starvedCursors.isEmpty()) {
+        MseBlock.Eos error;
+        boolean receivedMoreRows = false;
+        if (rows.isEmpty()) {
+          error = readOneBlock();
+        } else {
+          // Rows already removed from the heap are a globally ordered prefix. Consume any immediately available
+          // cursor progress so blocks can still be coalesced, but return that safe prefix instead of waiting merely
+          // to fill the output block.
+          MseBlock block = _multiConsumer.pollMseBlockOrStreamCompletion();
+          if (block == null && _multiConsumer.getFinishedStreamsLastRead().isEmpty()) {
+            break;
+          }
+          error = processReadBlock(block);
+          receivedMoreRows = block != null && block.isData() && ((MseBlock.Data) block).getNumRows() > 0;
+        }
+        if (error != null) {
+          return terminate(error);
+        }
+        if (_fallbackToSort) {
+          // These rows were already removed from cursors while building this not-yet-emitted block.
+          _rows.addAll(rows);
+          return sortAllRows();
+        }
+        if (receivedMoreRows) {
+          // A refill proves this is not a one-block result. Restore the established full-block capacity once so
+          // fragmented input cannot trigger repeated growth while this output block is assembled.
+          rows.ensureCapacity(SortOperator.DEFAULT_MAX_ROWS_PER_BLOCK);
+        }
+        continue;
+      }
+      if (_readyCursors.isEmpty()) {
+        break;
+      }
+      if (rows.isEmpty()) {
+        // Keep empty and tiny results cheap without sacrificing the one-allocation path for full output blocks.
+        // Sum all currently buffered rows once; later output blocks retain the established full-block capacity.
+        int initialCapacity = _mergeOutputStarted ? SortOperator.DEFAULT_MAX_ROWS_PER_BLOCK
+            : _readyCursors.getCappedAvailableRowCount(SortOperator.DEFAULT_MAX_ROWS_PER_BLOCK);
+        // ArrayList grows by 1.5x. Round near-full first blocks up now so a small refill cannot allocate an oversized
+        // replacement in addition to the nearly full initial array.
+        if (initialCapacity + (initialCapacity >> 1) >= SortOperator.DEFAULT_MAX_ROWS_PER_BLOCK) {
+          initialCapacity = SortOperator.DEFAULT_MAX_ROWS_PER_BLOCK;
+        }
+        rows.ensureCapacity(initialCapacity);
+      }
+      if (_tryEqualHeadMerge && _readyCursors.size() > 1) {
+        int remaining = SortOperator.DEFAULT_MAX_ROWS_PER_BLOCK - rows.size();
+        if (remaining >= _readyCursors.size()) {
+          if (_readyCursors.allHeadsEqual()) {
+            int previousRowCount = rows.size();
+            List<SenderCursor> exhausted = _readyCursors.advanceEqualHeads(rows);
+            if (exhausted != null) {
+              for (SenderCursor cursor : exhausted) {
+                if (!cursor._finished) {
+                  _starvedCursors.add(cursor);
+                }
+              }
+            }
+            if (!_starvedCursors.isEmpty()) {
+              _tryEqualHeadMerge = false;
+            }
+            checkActiveTerminationAndSampleUsageAfterBatch(previousRowCount, rows.size());
+            continue;
+          } else {
+            _tryEqualHeadMerge = false;
+          }
+        } else {
+          _readyCursors.ensureOrdered();
+        }
+      }
+      SenderCursor cursor = _readyCursors.peek();
+      rows.add(cursor.next());
+      if (cursor.hasRow()) {
+        // The cursor's key can only move forward, so restoring the heap from the root takes one sift-down. A generic
+        // PriorityQueue poll followed by add performs two independent heap repairs for every emitted row.
+        _readyCursors.updateTop();
+      } else if (!cursor._finished) {
+        _readyCursors.removeTop();
+        _starvedCursors.add(cursor);
+        _tryEqualHeadMerge = false;
+      } else {
+        _readyCursors.removeTop();
+        _tryEqualHeadMerge = true;
+      }
+      QueryThreadContext.checkTerminationAndSampleUsagePeriodically(rows.size(), MERGE_SCOPE,
+          _context.getActiveDeadlineMs());
+    }
+    if (rows.isEmpty()) {
+      return terminate(SuccessMseBlock.INSTANCE);
+    }
+    _mergeOutputStarted = true;
+    return new RowHeapDataBlock(rows, _dataSchema);
+  }
+
+  /// Reads one block from whichever sender is ready and updates only the cursor that produced it.
+  ///
+  /// @return the error that ended the read, or `null` when the read succeeded
+  @Nullable
+  private MseBlock.Eos readOneBlock() {
+    return processReadBlock(_multiConsumer.readMseBlockOrStreamCompletionBlocking());
+  }
+
+  @Nullable
+  private MseBlock.Eos processReadBlock(@Nullable MseBlock block) {
+    if (block == null) {
+      updateFinishedCursors();
+      if (_starvedCursors.isEmpty()) {
+        _tryEqualHeadMerge = true;
+      }
+      return null;
+    }
+    if (block.isEos()) {
+      updateFinishedCursors();
+      MseBlock.Eos eos = (MseBlock.Eos) block;
+      if (eos.isError()) {
+        return eos;
+      }
+      // Aggregate success is returned only after every sender has emitted EOS.
+      _starvedCursors.clear();
+      _tryEqualHeadMerge = true;
+      return null;
+    }
+    AsyncStream<ReceivingMailbox.MseBlockWithStats> stream = _multiConsumer.getLastReadStream();
+    Preconditions.checkState(stream != null, "Read a data block from no mailbox on stage: %s", _context.getStageId());
+    SenderCursor cursor = _cursorsByStream.get(stream);
+    Preconditions.checkState(cursor != null, "Read a data block from unknown mailbox: %s", stream.getId());
+    List<Object[]> rows = ((MseBlock.Data) block).asRowHeap().getRows();
+    checkActiveTerminationAndSampleUsage();
+    if (!_multiConsumer.isLastBlockSortedOnSender()) {
+      fallbackToFullSort(rows);
+      return null;
+    }
+    cursor.offer(rows);
+    updateFinishedCursors();
+    if (cursor.hasRow() && _starvedCursors.remove(cursor)) {
+      _readyCursors.add(cursor);
+    }
+    if (_starvedCursors.isEmpty()) {
+      _tryEqualHeadMerge = true;
+    }
+    return null;
+  }
+
+  /// Removes only the starved cursors whose EOS was consumed by the last read.
+  private void updateFinishedCursors() {
+    for (AsyncStream<ReceivingMailbox.MseBlockWithStats> stream : _multiConsumer.getFinishedStreamsLastRead()) {
+      SenderCursor cursor = _cursorsByStream.get(stream);
+      if (cursor != null) {
+        cursor._finished = true;
+        if (!cursor.hasRow()) {
+          _starvedCursors.remove(cursor);
+          _cursorsByStream.remove(stream);
+        }
+      }
+    }
+  }
+
+  /// Switches to a full receiver sort when a legacy sender omits the transport confirmation.
+  private void fallbackToFullSort(List<Object[]> unconfirmedRows) {
+    Preconditions.checkState(!_mergeOutputStarted,

Review Comment:
   This Preconditions check is correct as a safety net, but the message is internal-only. If it fires, users see an opaque error.
   
   New senders always mark every message and old senders never do. The check can therefore only fire on a bug or a truly mixed stream. Fine, but consider a QueryErrorCode, such as INTERNAL, and a message that tells the user to retry once the rolling upgrade is complete.



##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/plan/server/ServerPlanRequestVisitor.java:
##########
@@ -259,8 +259,17 @@ public Void visitMailboxReceive(MailboxReceiveNode node, ServerPlanRequestContex
 
   @Override
   public Void visitMailboxSend(MailboxSendNode node, ServerPlanRequestContext context) {
-    if (visit(node.getInputs().get(0), context)) {
-      context.setLeafStageBoundaryNode(node.getInputs().get(0));
+    PlanNode input = node.getInputs().get(0);
+    if (node.hasExplicitSortInput()) {

Review Comment:
   Good catch on hybrid and logical tables: if the sort were pushed into V1 per physical request, the output would be interleaved sorted runs on one mailbox.
   
   The ServerPlanRequestVisitorTest covers this. The cost is that the leaf no longer gets V1 ORDER BY pushdown. That pushdown could have used a sorted index or early segment pruning, but the sort has no limit, so little is lost.



##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/plan/PlanNodeToOpChain.java:
##########
@@ -193,7 +194,9 @@ void record(PlanNode node, MultiStageOperator operator) {
     @Override
     public MultiStageOperator visitMailboxReceive(MailboxReceiveNode node, OpChainExecutionContext context) {
       try {
-        if (node.isSort()) {
+        if (node.isSort() && node.isSortedOnSender()) {

Review Comment:
   If the Sort above the exchange is dropped (see the comment at PinotWindowExchangeNodeInsertRule:154), isSort()=true becomes the ONLY thing that makes a receiver produce ordered rows for the window. Please document that here, as a contract.
   
   New receivers: isSort && isSortedOnSender selects the merge, and the merge falls back to a full sort on unmarked data. Old receivers: isSort selects SortedMailboxReceiveOperator, which does a full sort. Both are needed for correct results, and nothing above the receiver sorts again. So (a) SortedMailboxReceiveOperator cannot be removed while old servers may still receive plans from new brokers, even though it is deprecated; and (b) the planner must never emit sortedOnSender=true with sort=false. Consider a Preconditions check in the planner or in PlanFragmenter, and a test that builds the op-chain for a global ordered window and asserts that the receive operator orders its output.



-- 
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-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] [MSE] Add opt-in sender-sorted merging for global ordered windows [pinot]

Posted by "gortiz (via GitHub)" <gi...@apache.org>.

gortiz commented on PR #19396:
URL: https://github.com/apache/pinot/pull/19396#issuecomment-5951299327

I'd like to use this PR to fix how MSE assigns the responsibility for ordering. Today, ordering should be a requirement that the logical plan guarantees. Instead, the physical plan and the runtime keep getting extra machinery to guarantee it: `sort`/`sortedOnSender` on the exchange and mailbox nodes, `SortedMailboxReceiveOperator`, and now a per-block `sorted.on.sender` marker. I consider the marker a blocker.

The runtime has to trust the plan. Re-checking ordering on every block is like a JIT adding checks for invariants that the front-end compiler already proved. It is also fragile: `MailboxSendOperator.isSortedOnSender()` requires a `SortOperator` right below the send, so any order-preserving operator we add there later (a buffer, for example) silently disables the optimization. And it goes through the whole mailbox layer (`GrpcSendingMailbox`, `InMemorySendingMailbox`, `MailboxService`, `ReceivingMailbox`, `MailboxContentObserver`, `ChannelUtils`, `BlockingMultiStreamConsumer`) for an opt-in feature.

What I'd suggest instead is to express ordering with simple plan nodes, not flags:

- The sender has no ordering flags. If the receiver needs ordered streams, the plan guarantees the order upstream: usually with a Sort, but also by construction, for example after a sorted join.
- The receiver uses a new plan node: a receive that k-way merges streams that are each already sorted, with an optional limit/offset. That's `SortedMailboxMergeReceiveOperator`, and the optional limit also covers the streaming ORDER BY LIMIT in #19711.
- When the plan doesn't guarantee order upstream, it uses a plain receive and a Sort downstream, as #19412 already does.

On the Calcite side, this could be a new `PinotKWayMergeSortExchange`: an exchange that requires its input to be sorted on the collation and preserves that order in its output. `PlanFragmenter` would turn it into a plain `MailboxSendNode` plus the new merge receive node. A rule similar to `PinotSortExchangeCopyRule` could push the limit/offset of a Sort above it into the receive.

In practice, for this PR that means removing the mailbox metadata and the `sortedOnSender` parameter of `offer()`/`offerRaw()`, `MseBlockWithStats.isSortedOnSender()`, `isLastBlockSortedOnSender()`, `MailboxSendOperator.isSortedOnSender()`, `hasExplicitSortInput()` and the full-sort fallback in the merge receiver. A per-row order check is fine in tests, but not as a production path.

Compatibility gets simpler too. On a new server:

| Plan | Comes from | New server |
|---|---|---|
| plain receive | current broker | `MailboxReceiveOperator` |
| receive with `sort=true` | broker older than #19412 | `MailboxReceiveOperator` + `SortOperator` |
| new merge receive node | new broker, option enabled | k-way merge |

For a new broker with old servers, it's the broker's responsibility not to send the new node and to keep generating a plan that old servers understand, i.e. a plain receive with a Sort downstream. We already have the mechanism for this: `SendStatsPredicate` in `SAFE` mode watches the instance configs and detects when any broker or server reports a different Pinot version. Today only servers use it to decide whether to send MSE stats, but the broker can reuse the same logic and only pick `PinotKWayMergeSortExchange` when the cluster is homogeneous. As a last line of defense, an old server can't deserialize the new node (`PlanNodeDeserializer` throws `Unsupported PlanNode type`), so the query fails loudly instead of returning wrong results.

With this, `sortedOnSender` and `sort` in `plan.proto` and the flags in `PinotLogicalSortExchange` can be deprecated, and `SortedMailboxReceiveOperator` can go away. Old ORDER BY plans already have a Sort with the fetch above the receive, so they get the bounded implementation with a LIMIT. For old window plans, the server injects a Sort without a fetch. That part can be a follow-up.

--
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-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] [MSE] Merge sender-sorted streams for global ordered windows [pinot]

Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.

xiangfu0 commented on code in PR #19396:
URL: https://github.com/apache/pinot/pull/19396#discussion_r4110236587


##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/SortedMailboxMergeReceiveOperator.java:
##########
@@ -0,0 +1,668 @@
+/**
+ * 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.pinot.query.runtime.operator;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.Deque;
+import java.util.IdentityHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import javax.annotation.Nullable;
+import org.apache.calcite.rel.RelFieldCollation;
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.pinot.common.utils.DataSchema;
+import org.apache.pinot.query.mailbox.ReceivingMailbox;
+import org.apache.pinot.query.planner.plannode.MailboxReceiveNode;
+import org.apache.pinot.query.runtime.blocks.MseBlock;
+import org.apache.pinot.query.runtime.blocks.RowHeapDataBlock;
+import org.apache.pinot.query.runtime.blocks.SuccessMseBlock;
+import org.apache.pinot.query.runtime.operator.utils.AsyncStream;
+import org.apache.pinot.query.runtime.operator.utils.SortUtils;
+import org.apache.pinot.query.runtime.plan.OpChainExecutionContext;
+import org.apache.pinot.spi.query.QueryThreadContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/// Receives streams that the plan declares sorted on the sender and merges them by the exchange collation.
+///
+/// An explicit sender [SortOperator] establishes the row ordering; [MailboxSendOperator] only transports that
+/// ordering. The transport marker confirms rollout compatibility and is not itself a sorting mechanism.
+///
+/// The plan declaration alone is not trusted during a rolling upgrade. Every data block must carry the transport's
+/// sender-sort confirmation. Before this operator emits its first row it obtains a head row, or EOS, from every live
+/// sender. If any sender's first data is unconfirmed, all tentatively buffered rows are folded into a full receiver
+/// sort. Once output starts, losing the confirmation is a protocol violation because already emitted rows cannot be
+/// recovered into that fallback.
+///
+/// The merge reads whichever mailbox is ready instead of blocking on one sender. This prevents a sender that is
+/// backpressured by another receiver from creating a cross-receiver wait cycle. Rows are emitted in blocks of at most
+/// 10,000 while cursor state carries the ordering frontier across calls. A fast sender can be read ahead while another
+/// sender is starved, so retained input is workload-dependent and can approach the legacy full receiver sort in the
+/// worst case.
+///
+/// This operator is driven by a single consumer thread and is not thread-safe.
+public class SortedMailboxMergeReceiveOperator extends BaseMailboxReceiveOperator implements SortedMultiStageOperator {
+  private static final Logger LOGGER = LoggerFactory.getLogger(SortedMailboxMergeReceiveOperator.class);
+
+  private static final String EXPLAIN_NAME = "SORTED_MAILBOX_MERGE_RECEIVE";
+  private static final String MERGE_SCOPE = "SortedMailboxMergeReceiveOperator";
+  private final DataSchema _dataSchema;
+  private final List<RelFieldCollation> _collations;
+  private final Comparator<Object[]> _comparator;
+  private final SenderCursorHeap _readyCursors;
+  private final boolean _singleSortedSender;
+  /// Senders that have not finished but do not currently have a row ready. Nothing can be emitted while this is
+  /// non-empty because any one of these senders may hold the next row.
+  private final Set<SenderCursor> _starvedCursors = Collections.newSetFromMap(new IdentityHashMap<>());
+  private final Map<AsyncStream<ReceivingMailbox.MseBlockWithStats>, SenderCursor> _cursorsByStream =
+      new IdentityHashMap<>();
+  private boolean _mergeOutputStarted;
+  private boolean _fallbackToSort;
+  private boolean _tryEqualHeadMerge;
+  private int _fallbackOutputIndex = -1;
+
+  /// Rows buffered only for the mixed-version fallback. The sorted list is handed downstream as-is, so cleanup must
+  /// drop this reference rather than clear it.
+  @Nullable
+  private List<Object[]> _rows;
+
+  @Nullable
+  private MseBlock _eosBlock;
+
+  public SortedMailboxMergeReceiveOperator(OpChainExecutionContext context, MailboxReceiveNode node) {
+    super(context, node);
+    Preconditions.checkState(node.isSort(), "Receiver-side sorting must be enabled");
+    Preconditions.checkState(node.isSortedOnSender(), "Sender-side sorting must be enabled");
+    Preconditions.checkState(!CollectionUtils.isEmpty(node.getCollations()), "Field collations must be set");
+    _dataSchema = node.getDataSchema();
+    _collations = List.copyOf(node.getCollations());
+    _comparator = new SortUtils.SortComparator(_collations, false);

Review Comment:
   Confirmed. The merge uses the same comparator construction and null-direction handling as the regular sort path, so ascending, descending, and null ordering remain consistent. No change was needed.



##########
pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotWindowExchangeNodeInsertRule.java:
##########
@@ -141,18 +141,17 @@ 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.
+    // where the exchange carries a collation the ordering has to be established above it. Keep an explicit Sort as
+    // the semantic boundary. A confirmed merge receiver advertises the exact collation, allowing SortOperator to
+    // stream through it; a legacy receiver still performs the full sort.

Review Comment:
   Done. I removed the redundant Sort above a receiver-sorted exchange and deleted SortedMultiStageOperator, SortOperator.isInputSorted, and their associated tests and documentation. The old plan-fixture changes are reverted as well.



##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/plan/PlanNodeToOpChain.java:
##########
@@ -193,7 +194,9 @@ void record(PlanNode node, MultiStageOperator operator) {
     @Override
     public MultiStageOperator visitMailboxReceive(MailboxReceiveNode node, OpChainExecutionContext context) {
       try {
-        if (node.isSort()) {
+        if (node.isSort() && node.isSortedOnSender()) {

Review Comment:
   Addressed by documenting the isSortOnSender/isSortOnReceiver ordering contract and enforcing sortOnSender implies sortOnReceiver in PlanFragmenter. Plan-shape, real op-chain selection, and randomized merge tests now cover the contract end to end.



##########
pinot-query-planner/src/main/java/org/apache/pinot/query/planner/logical/PlanFragmenter.java:
##########
@@ -185,6 +186,16 @@ public PlanNode visitExchange(ExchangeNode node, Context context) {
 
     // Create a new context for the next PlanFragment with MailboxSendNode as the root node.
     PlanNode nextPlanFragmentRoot = node.getInputs().get(0).visit(this, new Context(senderPlanFragmentId));
+    if (node.isSortOnSender()) {
+      Preconditions.checkState(!node.getCollations().isEmpty(),
+          "Sender sorting requires a non-empty exchange collation");
+      // Ordering belongs to an explicit operator in the sender fragment. MailboxSendOperator only preserves and
+      // transports this output; ServerPlanRequestVisitor keeps this SortNode above the V1 leaf boundary.
+      // SortOperator applies the broker response limit when fetch is absent. This internal sort must retain every
+      // sender row, so use the largest representable fetch with a zero effective offset.
+      nextPlanFragmentRoot = new SortNode(senderPlanFragmentId, nextPlanFragmentRoot.getDataSchema(), null,
+          List.of(nextPlanFragmentRoot), node.getCollations(), Integer.MAX_VALUE, -1);

Review Comment:
   Added implementation-EXPLAIN coverage. The enabled plan asserts both MAIL_SEND [SORTED] and the sender-side SORT LIMIT 2147483647, in addition to checking that no redundant Sort remains above the merge receiver.



##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MailboxSendOperator.java:
##########
@@ -64,13 +67,17 @@ public class MailboxSendOperator extends MultiStageOperator {
   private final BlockExchange _exchange;
   private final StatMap<StatKey> _statMap = new StatMap<>(StatKey.class);
 
-  // TODO: Support sort on sender
   public MailboxSendOperator(OpChainExecutionContext context, MultiStageOperator input, MailboxSendNode node) {
-    this(context, input, statMap -> getBlockExchange(context, node, statMap));
+    this(context, input, statMap -> getBlockExchange(context, node, statMap, isSortedOnSender(input, node)));
     _statMap.merge(StatKey.STAGE, context.getStageId());
     _statMap.merge(StatKey.PARALLELISM, 1);
   }
 
+  @VisibleForTesting
+  static boolean isSortedOnSender(MultiStageOperator input, MailboxSendNode node) {
+    return input instanceof SortOperator && node.hasExplicitSortInput();

Review Comment:
   I kept the structural check as a fail-safe and added a real op-chain conversion test that verifies an explicit SortNode selects the sorted MailboxService overload. If a future wrapper hides the sort, the sender omits confirmation and the receiver safely falls back instead of risking incorrect ordering.



##########
pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotWindowExchangeNodeInsertRule.java:
##########
@@ -120,13 +120,13 @@ public void onMatch(RelOptRuleCall call) {
         exchange = PinotLogicalExchange.create(input, RelDistributions.hash(List.of()));
       } else {
         // Only ORDER BY
-        // Add a LogicalSortExchange with collation on the order by key(s) and an empty hash partition key.
-        // 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.
+        // Sort each sender explicitly and merge the sorted mailbox streams at the receiver. The Sort retained above
+        // the exchange is the semantic ordering boundary and becomes a streaming limit when the merge receiver
+        // advertises this exact collation.
         // TODO: Revisit whether we should use hash distribution
         exchange =
-            PinotLogicalSortExchange.create(input, RelDistributions.hash(List.of()), windowGroup.orderKeys, false,
-                false);
+            PinotLogicalSortExchange.create(input, RelDistributions.hash(List.of()), windowGroup.orderKeys, true,

Review Comment:
   Addressed. The merge receiver now establishes final ordering directly, so the redundant post-exchange Sort is gone. The optimization is also default-off with query and broker controls: disabled mode preserves the master plan, while enabled mixed-version execution is limited to sender plus legacy receiver sorting rather than three sorts.



##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/SortedMailboxMergeReceiveOperator.java:
##########
@@ -0,0 +1,668 @@
+/**
+ * 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.pinot.query.runtime.operator;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.Deque;
+import java.util.IdentityHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import javax.annotation.Nullable;
+import org.apache.calcite.rel.RelFieldCollation;
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.pinot.common.utils.DataSchema;
+import org.apache.pinot.query.mailbox.ReceivingMailbox;
+import org.apache.pinot.query.planner.plannode.MailboxReceiveNode;
+import org.apache.pinot.query.runtime.blocks.MseBlock;
+import org.apache.pinot.query.runtime.blocks.RowHeapDataBlock;
+import org.apache.pinot.query.runtime.blocks.SuccessMseBlock;
+import org.apache.pinot.query.runtime.operator.utils.AsyncStream;
+import org.apache.pinot.query.runtime.operator.utils.SortUtils;
+import org.apache.pinot.query.runtime.plan.OpChainExecutionContext;
+import org.apache.pinot.spi.query.QueryThreadContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/// Receives streams that the plan declares sorted on the sender and merges them by the exchange collation.
+///
+/// An explicit sender [SortOperator] establishes the row ordering; [MailboxSendOperator] only transports that
+/// ordering. The transport marker confirms rollout compatibility and is not itself a sorting mechanism.
+///
+/// The plan declaration alone is not trusted during a rolling upgrade. Every data block must carry the transport's
+/// sender-sort confirmation. Before this operator emits its first row it obtains a head row, or EOS, from every live
+/// sender. If any sender's first data is unconfirmed, all tentatively buffered rows are folded into a full receiver
+/// sort. Once output starts, losing the confirmation is a protocol violation because already emitted rows cannot be
+/// recovered into that fallback.
+///
+/// The merge reads whichever mailbox is ready instead of blocking on one sender. This prevents a sender that is
+/// backpressured by another receiver from creating a cross-receiver wait cycle. Rows are emitted in blocks of at most
+/// 10,000 while cursor state carries the ordering frontier across calls. A fast sender can be read ahead while another
+/// sender is starved, so retained input is workload-dependent and can approach the legacy full receiver sort in the
+/// worst case.
+///
+/// This operator is driven by a single consumer thread and is not thread-safe.
+public class SortedMailboxMergeReceiveOperator extends BaseMailboxReceiveOperator implements SortedMultiStageOperator {
+  private static final Logger LOGGER = LoggerFactory.getLogger(SortedMailboxMergeReceiveOperator.class);
+
+  private static final String EXPLAIN_NAME = "SORTED_MAILBOX_MERGE_RECEIVE";
+  private static final String MERGE_SCOPE = "SortedMailboxMergeReceiveOperator";
+  private final DataSchema _dataSchema;
+  private final List<RelFieldCollation> _collations;
+  private final Comparator<Object[]> _comparator;
+  private final SenderCursorHeap _readyCursors;
+  private final boolean _singleSortedSender;
+  /// Senders that have not finished but do not currently have a row ready. Nothing can be emitted while this is
+  /// non-empty because any one of these senders may hold the next row.
+  private final Set<SenderCursor> _starvedCursors = Collections.newSetFromMap(new IdentityHashMap<>());
+  private final Map<AsyncStream<ReceivingMailbox.MseBlockWithStats>, SenderCursor> _cursorsByStream =
+      new IdentityHashMap<>();
+  private boolean _mergeOutputStarted;
+  private boolean _fallbackToSort;
+  private boolean _tryEqualHeadMerge;
+  private int _fallbackOutputIndex = -1;
+
+  /// Rows buffered only for the mixed-version fallback. The sorted list is handed downstream as-is, so cleanup must
+  /// drop this reference rather than clear it.
+  @Nullable
+  private List<Object[]> _rows;
+
+  @Nullable
+  private MseBlock _eosBlock;
+
+  public SortedMailboxMergeReceiveOperator(OpChainExecutionContext context, MailboxReceiveNode node) {
+    super(context, node);
+    Preconditions.checkState(node.isSort(), "Receiver-side sorting must be enabled");
+    Preconditions.checkState(node.isSortedOnSender(), "Sender-side sorting must be enabled");
+    Preconditions.checkState(!CollectionUtils.isEmpty(node.getCollations()), "Field collations must be set");
+    _dataSchema = node.getDataSchema();
+    _collations = List.copyOf(node.getCollations());
+    _comparator = new SortUtils.SortComparator(_collations, false);
+    List<AsyncStream<ReceivingMailbox.MseBlockWithStats>> streams = _multiConsumer.getLiveStreamsSnapshot();
+    _readyCursors = new SenderCursorHeap(streams.size(), _comparator);
+    _singleSortedSender = streams.size() == 1;
+    if (!_singleSortedSender) {
+      for (AsyncStream<ReceivingMailbox.MseBlockWithStats> stream : streams) {
+        SenderCursor cursor = new SenderCursor(stream);
+        _cursorsByStream.put(stream, cursor);
+        _starvedCursors.add(cursor);
+      }
+    }
+  }
+
+  @Override
+  protected Logger logger() {
+    return LOGGER;
+  }
+
+  @Override
+  public String toExplainString() {
+    return EXPLAIN_NAME;
+  }
+
+  @Override
+  public List<RelFieldCollation> getCollations() {
+    return _collations;
+  }
+
+  @Override
+  protected MseBlock getNextBlock() {
+    if (_fallbackOutputIndex >= 0 && _rows != null) {
+      return emitFallbackBlock();
+    }
+    if (_eosBlock != null) {
+      return _eosBlock;
+    }
+    if (_isEarlyTerminated) {
+      return readUntilEos();
+    }
+    return _singleSortedSender ? readSingleSortedSender() : mergeNextBlock();
+  }
+
+  /// Passes through one confirmed sorted sender without copying its rows through the merge heap.
+  private MseBlock readSingleSortedSender() {
+    while (true) {
+      MseBlock block = _multiConsumer.readMseBlockBlocking();
+      if (block.isEos()) {
+        return terminate(block);
+      }
+      MseBlock.Data dataBlock = (MseBlock.Data) block;
+      checkActiveTerminationAndSampleUsage();
+      if (!_multiConsumer.isLastBlockSortedOnSender()) {

Review Comment:
   Corrected the class and PR documentation. The 10,000-row cap applies to multi-sender merge and fallback output; a confirmed single sender is passed through with its original block boundaries.



##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/plan/server/ServerPlanRequestVisitor.java:
##########
@@ -259,8 +259,17 @@ public Void visitMailboxReceive(MailboxReceiveNode node, ServerPlanRequestContex
 
   @Override
   public Void visitMailboxSend(MailboxSendNode node, ServerPlanRequestContext context) {
-    if (visit(node.getInputs().get(0), context)) {
-      context.setLeafStageBoundaryNode(node.getInputs().get(0));
+    PlanNode input = node.getInputs().get(0);
+    if (node.hasExplicitSortInput()) {

Review Comment:
   Confirmed. ServerPlanRequestVisitorTest covers the leaf-boundary invariant needed for hybrid and logical-table physical requests. The conservative loss of unbounded-sort pushdown is accepted here; no extra hybrid-specific code change was needed.



##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/SortedMailboxMergeReceiveOperator.java:
##########
@@ -0,0 +1,668 @@
+/**
+ * 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.pinot.query.runtime.operator;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.Deque;
+import java.util.IdentityHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import javax.annotation.Nullable;
+import org.apache.calcite.rel.RelFieldCollation;
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.pinot.common.utils.DataSchema;
+import org.apache.pinot.query.mailbox.ReceivingMailbox;
+import org.apache.pinot.query.planner.plannode.MailboxReceiveNode;
+import org.apache.pinot.query.runtime.blocks.MseBlock;
+import org.apache.pinot.query.runtime.blocks.RowHeapDataBlock;
+import org.apache.pinot.query.runtime.blocks.SuccessMseBlock;
+import org.apache.pinot.query.runtime.operator.utils.AsyncStream;
+import org.apache.pinot.query.runtime.operator.utils.SortUtils;
+import org.apache.pinot.query.runtime.plan.OpChainExecutionContext;
+import org.apache.pinot.spi.query.QueryThreadContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/// Receives streams that the plan declares sorted on the sender and merges them by the exchange collation.
+///
+/// An explicit sender [SortOperator] establishes the row ordering; [MailboxSendOperator] only transports that
+/// ordering. The transport marker confirms rollout compatibility and is not itself a sorting mechanism.
+///
+/// The plan declaration alone is not trusted during a rolling upgrade. Every data block must carry the transport's
+/// sender-sort confirmation. Before this operator emits its first row it obtains a head row, or EOS, from every live
+/// sender. If any sender's first data is unconfirmed, all tentatively buffered rows are folded into a full receiver
+/// sort. Once output starts, losing the confirmation is a protocol violation because already emitted rows cannot be
+/// recovered into that fallback.
+///
+/// The merge reads whichever mailbox is ready instead of blocking on one sender. This prevents a sender that is
+/// backpressured by another receiver from creating a cross-receiver wait cycle. Rows are emitted in blocks of at most
+/// 10,000 while cursor state carries the ordering frontier across calls. A fast sender can be read ahead while another
+/// sender is starved, so retained input is workload-dependent and can approach the legacy full receiver sort in the
+/// worst case.
+///
+/// This operator is driven by a single consumer thread and is not thread-safe.
+public class SortedMailboxMergeReceiveOperator extends BaseMailboxReceiveOperator implements SortedMultiStageOperator {
+  private static final Logger LOGGER = LoggerFactory.getLogger(SortedMailboxMergeReceiveOperator.class);
+
+  private static final String EXPLAIN_NAME = "SORTED_MAILBOX_MERGE_RECEIVE";
+  private static final String MERGE_SCOPE = "SortedMailboxMergeReceiveOperator";
+  private final DataSchema _dataSchema;
+  private final List<RelFieldCollation> _collations;
+  private final Comparator<Object[]> _comparator;
+  private final SenderCursorHeap _readyCursors;
+  private final boolean _singleSortedSender;
+  /// Senders that have not finished but do not currently have a row ready. Nothing can be emitted while this is
+  /// non-empty because any one of these senders may hold the next row.
+  private final Set<SenderCursor> _starvedCursors = Collections.newSetFromMap(new IdentityHashMap<>());
+  private final Map<AsyncStream<ReceivingMailbox.MseBlockWithStats>, SenderCursor> _cursorsByStream =
+      new IdentityHashMap<>();
+  private boolean _mergeOutputStarted;
+  private boolean _fallbackToSort;
+  private boolean _tryEqualHeadMerge;
+  private int _fallbackOutputIndex = -1;
+
+  /// Rows buffered only for the mixed-version fallback. The sorted list is handed downstream as-is, so cleanup must
+  /// drop this reference rather than clear it.
+  @Nullable
+  private List<Object[]> _rows;
+
+  @Nullable
+  private MseBlock _eosBlock;
+
+  public SortedMailboxMergeReceiveOperator(OpChainExecutionContext context, MailboxReceiveNode node) {
+    super(context, node);
+    Preconditions.checkState(node.isSort(), "Receiver-side sorting must be enabled");
+    Preconditions.checkState(node.isSortedOnSender(), "Sender-side sorting must be enabled");
+    Preconditions.checkState(!CollectionUtils.isEmpty(node.getCollations()), "Field collations must be set");
+    _dataSchema = node.getDataSchema();
+    _collations = List.copyOf(node.getCollations());
+    _comparator = new SortUtils.SortComparator(_collations, false);
+    List<AsyncStream<ReceivingMailbox.MseBlockWithStats>> streams = _multiConsumer.getLiveStreamsSnapshot();
+    _readyCursors = new SenderCursorHeap(streams.size(), _comparator);
+    _singleSortedSender = streams.size() == 1;
+    if (!_singleSortedSender) {
+      for (AsyncStream<ReceivingMailbox.MseBlockWithStats> stream : streams) {
+        SenderCursor cursor = new SenderCursor(stream);
+        _cursorsByStream.put(stream, cursor);
+        _starvedCursors.add(cursor);
+      }
+    }
+  }
+
+  @Override
+  protected Logger logger() {
+    return LOGGER;
+  }
+
+  @Override
+  public String toExplainString() {
+    return EXPLAIN_NAME;
+  }
+
+  @Override
+  public List<RelFieldCollation> getCollations() {
+    return _collations;
+  }
+
+  @Override
+  protected MseBlock getNextBlock() {
+    if (_fallbackOutputIndex >= 0 && _rows != null) {
+      return emitFallbackBlock();
+    }
+    if (_eosBlock != null) {
+      return _eosBlock;
+    }
+    if (_isEarlyTerminated) {
+      return readUntilEos();
+    }
+    return _singleSortedSender ? readSingleSortedSender() : mergeNextBlock();
+  }
+
+  /// Passes through one confirmed sorted sender without copying its rows through the merge heap.
+  private MseBlock readSingleSortedSender() {
+    while (true) {
+      MseBlock block = _multiConsumer.readMseBlockBlocking();
+      if (block.isEos()) {
+        return terminate(block);
+      }
+      MseBlock.Data dataBlock = (MseBlock.Data) block;
+      checkActiveTerminationAndSampleUsage();
+      if (!_multiConsumer.isLastBlockSortedOnSender()) {
+        fallbackToFullSort(dataBlock.asRowHeap().getRows());
+        return sortAllRows();
+      }
+      if (dataBlock.getNumRows() > 0) {
+        _mergeOutputStarted = true;
+        return dataBlock;
+      }
+    }
+  }
+
+  /// Merges the sorted senders, emitting at most [SortOperator#DEFAULT_MAX_ROWS_PER_BLOCK] rows per call.
+  private MseBlock mergeNextBlock() {
+    ArrayList<Object[]> rows = new ArrayList<>(0);
+    while (rows.size() < SortOperator.DEFAULT_MAX_ROWS_PER_BLOCK) {
+      if (!_starvedCursors.isEmpty()) {
+        MseBlock.Eos error;
+        boolean receivedMoreRows = false;
+        if (rows.isEmpty()) {
+          error = readOneBlock();
+        } else {
+          // Rows already removed from the heap are a globally ordered prefix. Consume any immediately available
+          // cursor progress so blocks can still be coalesced, but return that safe prefix instead of waiting merely
+          // to fill the output block.
+          MseBlock block = _multiConsumer.pollMseBlockOrStreamCompletion();
+          if (block == null && _multiConsumer.getFinishedStreamsLastRead().isEmpty()) {
+            break;
+          }
+          error = processReadBlock(block);
+          receivedMoreRows = block != null && block.isData() && ((MseBlock.Data) block).getNumRows() > 0;
+        }
+        if (error != null) {
+          return terminate(error);
+        }
+        if (_fallbackToSort) {
+          // These rows were already removed from cursors while building this not-yet-emitted block.
+          _rows.addAll(rows);
+          return sortAllRows();
+        }
+        if (receivedMoreRows) {
+          // A refill proves this is not a one-block result. Restore the established full-block capacity once so
+          // fragmented input cannot trigger repeated growth while this output block is assembled.
+          rows.ensureCapacity(SortOperator.DEFAULT_MAX_ROWS_PER_BLOCK);
+        }
+        continue;
+      }
+      if (_readyCursors.isEmpty()) {
+        break;
+      }
+      if (rows.isEmpty()) {
+        // Keep empty and tiny results cheap without sacrificing the one-allocation path for full output blocks.
+        // Sum all currently buffered rows once; later output blocks retain the established full-block capacity.
+        int initialCapacity = _mergeOutputStarted ? SortOperator.DEFAULT_MAX_ROWS_PER_BLOCK
+            : _readyCursors.getCappedAvailableRowCount(SortOperator.DEFAULT_MAX_ROWS_PER_BLOCK);
+        // ArrayList grows by 1.5x. Round near-full first blocks up now so a small refill cannot allocate an oversized
+        // replacement in addition to the nearly full initial array.
+        if (initialCapacity + (initialCapacity >> 1) >= SortOperator.DEFAULT_MAX_ROWS_PER_BLOCK) {
+          initialCapacity = SortOperator.DEFAULT_MAX_ROWS_PER_BLOCK;
+        }
+        rows.ensureCapacity(initialCapacity);
+      }
+      if (_tryEqualHeadMerge && _readyCursors.size() > 1) {
+        int remaining = SortOperator.DEFAULT_MAX_ROWS_PER_BLOCK - rows.size();
+        if (remaining >= _readyCursors.size()) {
+          if (_readyCursors.allHeadsEqual()) {
+            int previousRowCount = rows.size();
+            List<SenderCursor> exhausted = _readyCursors.advanceEqualHeads(rows);
+            if (exhausted != null) {
+              for (SenderCursor cursor : exhausted) {
+                if (!cursor._finished) {
+                  _starvedCursors.add(cursor);
+                }
+              }
+            }
+            if (!_starvedCursors.isEmpty()) {
+              _tryEqualHeadMerge = false;
+            }
+            checkActiveTerminationAndSampleUsageAfterBatch(previousRowCount, rows.size());
+            continue;
+          } else {
+            _tryEqualHeadMerge = false;
+          }
+        } else {
+          _readyCursors.ensureOrdered();
+        }
+      }
+      SenderCursor cursor = _readyCursors.peek();
+      rows.add(cursor.next());
+      if (cursor.hasRow()) {
+        // The cursor's key can only move forward, so restoring the heap from the root takes one sift-down. A generic
+        // PriorityQueue poll followed by add performs two independent heap repairs for every emitted row.
+        _readyCursors.updateTop();
+      } else if (!cursor._finished) {
+        _readyCursors.removeTop();
+        _starvedCursors.add(cursor);
+        _tryEqualHeadMerge = false;
+      } else {
+        _readyCursors.removeTop();
+        _tryEqualHeadMerge = true;
+      }
+      QueryThreadContext.checkTerminationAndSampleUsagePeriodically(rows.size(), MERGE_SCOPE,
+          _context.getActiveDeadlineMs());
+    }
+    if (rows.isEmpty()) {
+      return terminate(SuccessMseBlock.INSTANCE);
+    }
+    _mergeOutputStarted = true;
+    return new RowHeapDataBlock(rows, _dataSchema);
+  }
+
+  /// Reads one block from whichever sender is ready and updates only the cursor that produced it.
+  ///
+  /// @return the error that ended the read, or `null` when the read succeeded
+  @Nullable
+  private MseBlock.Eos readOneBlock() {
+    return processReadBlock(_multiConsumer.readMseBlockOrStreamCompletionBlocking());
+  }
+
+  @Nullable
+  private MseBlock.Eos processReadBlock(@Nullable MseBlock block) {
+    if (block == null) {
+      updateFinishedCursors();
+      if (_starvedCursors.isEmpty()) {
+        _tryEqualHeadMerge = true;
+      }
+      return null;
+    }
+    if (block.isEos()) {
+      updateFinishedCursors();
+      MseBlock.Eos eos = (MseBlock.Eos) block;
+      if (eos.isError()) {
+        return eos;
+      }
+      // Aggregate success is returned only after every sender has emitted EOS.
+      _starvedCursors.clear();
+      _tryEqualHeadMerge = true;
+      return null;
+    }
+    AsyncStream<ReceivingMailbox.MseBlockWithStats> stream = _multiConsumer.getLastReadStream();
+    Preconditions.checkState(stream != null, "Read a data block from no mailbox on stage: %s", _context.getStageId());
+    SenderCursor cursor = _cursorsByStream.get(stream);
+    Preconditions.checkState(cursor != null, "Read a data block from unknown mailbox: %s", stream.getId());
+    List<Object[]> rows = ((MseBlock.Data) block).asRowHeap().getRows();
+    checkActiveTerminationAndSampleUsage();
+    if (!_multiConsumer.isLastBlockSortedOnSender()) {
+      fallbackToFullSort(rows);
+      return null;
+    }
+    cursor.offer(rows);
+    updateFinishedCursors();
+    if (cursor.hasRow() && _starvedCursors.remove(cursor)) {
+      _readyCursors.add(cursor);
+    }
+    if (_starvedCursors.isEmpty()) {
+      _tryEqualHeadMerge = true;
+    }
+    return null;
+  }
+
+  /// Removes only the starved cursors whose EOS was consumed by the last read.
+  private void updateFinishedCursors() {
+    for (AsyncStream<ReceivingMailbox.MseBlockWithStats> stream : _multiConsumer.getFinishedStreamsLastRead()) {
+      SenderCursor cursor = _cursorsByStream.get(stream);
+      if (cursor != null) {
+        cursor._finished = true;
+        if (!cursor.hasRow()) {
+          _starvedCursors.remove(cursor);
+          _cursorsByStream.remove(stream);
+        }
+      }
+    }
+  }
+
+  /// Switches to a full receiver sort when a legacy sender omits the transport confirmation.
+  private void fallbackToFullSort(List<Object[]> unconfirmedRows) {
+    Preconditions.checkState(!_mergeOutputStarted,

Review Comment:
   Fixed. A confirmation change after output begins now returns an INTERNAL query error with an actionable message to retry after the rolling upgrade or disable windowSortOnSender. The regression test asserts both the error code and the message.



##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/GrpcSendingMailbox.java:
##########
@@ -432,12 +435,14 @@ protected void sendContent(ByteString byteString, boolean waitForMore, boolean b
       if (!bypassReady && isTerminated()) {
         return;
       }
-      MailboxContent content = MailboxContent.newBuilder()
+      MailboxContent.Builder contentBuilder = MailboxContent.newBuilder()
           .setMailboxId(_id)
           .setPayload(byteString)
-          .setWaitForMore(waitForMore)
-          .build();
-      _contentObserver.onNext(content);
+          .setWaitForMore(waitForMore);
+      if (_sortedOnSender) {
+        contentBuilder.putMetadata(ChannelUtils.MAILBOX_METADATA_SORTED_ON_SENDER, Boolean.TRUE.toString());

Review Comment:
   Confirmed. The ordering marker intentionally remains on every mailbox message so each message is self-describing; this is a fixed 26-byte per-message cost rather than a per-row cost. No code change was needed here.



##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/utils/BlockingMultiStreamConsumer.java:
##########
@@ -269,6 +315,37 @@ private E readDroppingSuccessEos() {
     return block;
   }
 
+  /// Returns the stream that produced the element the last blocking read returned.
+  ///
+  /// This is only meaningful right after that call returned a data or an error element. The element that ends all
+  /// the streams is not produced by any of them, and a stream that emitted its EOS is dropped from the ones this
+  /// consumer tracks, so `null` is returned in both cases.
+  ///
+  /// Consumers that need to keep the elements of each stream apart, like a merge of already sorted streams, use
+  /// this to tell which stream the element they just read belongs to.
+  ///
+  /// This method is called by the consumer thread.
+  @Nullable
+  public AsyncStream<E> getLastReadStream() {

Review Comment:
   Agreed that a combined ReadResult API would make the temporal coupling clearer. I am deferring that shared mailbox API refactor because this operator is the single consumer here and the comment is nonblocking; it can be handled separately without expanding this PR.



##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/SortedMailboxMergeReceiveOperator.java:
##########
@@ -0,0 +1,668 @@
+/**
+ * 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.pinot.query.runtime.operator;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.Deque;
+import java.util.IdentityHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import javax.annotation.Nullable;
+import org.apache.calcite.rel.RelFieldCollation;
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.pinot.common.utils.DataSchema;
+import org.apache.pinot.query.mailbox.ReceivingMailbox;
+import org.apache.pinot.query.planner.plannode.MailboxReceiveNode;
+import org.apache.pinot.query.runtime.blocks.MseBlock;
+import org.apache.pinot.query.runtime.blocks.RowHeapDataBlock;
+import org.apache.pinot.query.runtime.blocks.SuccessMseBlock;
+import org.apache.pinot.query.runtime.operator.utils.AsyncStream;
+import org.apache.pinot.query.runtime.operator.utils.SortUtils;
+import org.apache.pinot.query.runtime.plan.OpChainExecutionContext;
+import org.apache.pinot.spi.query.QueryThreadContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/// Receives streams that the plan declares sorted on the sender and merges them by the exchange collation.
+///
+/// An explicit sender [SortOperator] establishes the row ordering; [MailboxSendOperator] only transports that
+/// ordering. The transport marker confirms rollout compatibility and is not itself a sorting mechanism.
+///
+/// The plan declaration alone is not trusted during a rolling upgrade. Every data block must carry the transport's
+/// sender-sort confirmation. Before this operator emits its first row it obtains a head row, or EOS, from every live
+/// sender. If any sender's first data is unconfirmed, all tentatively buffered rows are folded into a full receiver
+/// sort. Once output starts, losing the confirmation is a protocol violation because already emitted rows cannot be
+/// recovered into that fallback.
+///
+/// The merge reads whichever mailbox is ready instead of blocking on one sender. This prevents a sender that is
+/// backpressured by another receiver from creating a cross-receiver wait cycle. Rows are emitted in blocks of at most
+/// 10,000 while cursor state carries the ordering frontier across calls. A fast sender can be read ahead while another
+/// sender is starved, so retained input is workload-dependent and can approach the legacy full receiver sort in the
+/// worst case.
+///
+/// This operator is driven by a single consumer thread and is not thread-safe.
+public class SortedMailboxMergeReceiveOperator extends BaseMailboxReceiveOperator implements SortedMultiStageOperator {
+  private static final Logger LOGGER = LoggerFactory.getLogger(SortedMailboxMergeReceiveOperator.class);
+
+  private static final String EXPLAIN_NAME = "SORTED_MAILBOX_MERGE_RECEIVE";
+  private static final String MERGE_SCOPE = "SortedMailboxMergeReceiveOperator";
+  private final DataSchema _dataSchema;
+  private final List<RelFieldCollation> _collations;
+  private final Comparator<Object[]> _comparator;
+  private final SenderCursorHeap _readyCursors;
+  private final boolean _singleSortedSender;
+  /// Senders that have not finished but do not currently have a row ready. Nothing can be emitted while this is
+  /// non-empty because any one of these senders may hold the next row.
+  private final Set<SenderCursor> _starvedCursors = Collections.newSetFromMap(new IdentityHashMap<>());
+  private final Map<AsyncStream<ReceivingMailbox.MseBlockWithStats>, SenderCursor> _cursorsByStream =
+      new IdentityHashMap<>();
+  private boolean _mergeOutputStarted;
+  private boolean _fallbackToSort;
+  private boolean _tryEqualHeadMerge;
+  private int _fallbackOutputIndex = -1;
+
+  /// Rows buffered only for the mixed-version fallback. The sorted list is handed downstream as-is, so cleanup must
+  /// drop this reference rather than clear it.
+  @Nullable
+  private List<Object[]> _rows;
+
+  @Nullable
+  private MseBlock _eosBlock;
+
+  public SortedMailboxMergeReceiveOperator(OpChainExecutionContext context, MailboxReceiveNode node) {
+    super(context, node);
+    Preconditions.checkState(node.isSort(), "Receiver-side sorting must be enabled");
+    Preconditions.checkState(node.isSortedOnSender(), "Sender-side sorting must be enabled");
+    Preconditions.checkState(!CollectionUtils.isEmpty(node.getCollations()), "Field collations must be set");
+    _dataSchema = node.getDataSchema();
+    _collations = List.copyOf(node.getCollations());
+    _comparator = new SortUtils.SortComparator(_collations, false);
+    List<AsyncStream<ReceivingMailbox.MseBlockWithStats>> streams = _multiConsumer.getLiveStreamsSnapshot();
+    _readyCursors = new SenderCursorHeap(streams.size(), _comparator);
+    _singleSortedSender = streams.size() == 1;
+    if (!_singleSortedSender) {
+      for (AsyncStream<ReceivingMailbox.MseBlockWithStats> stream : streams) {
+        SenderCursor cursor = new SenderCursor(stream);
+        _cursorsByStream.put(stream, cursor);
+        _starvedCursors.add(cursor);
+      }
+    }
+  }
+
+  @Override
+  protected Logger logger() {
+    return LOGGER;
+  }
+
+  @Override
+  public String toExplainString() {
+    return EXPLAIN_NAME;
+  }
+
+  @Override
+  public List<RelFieldCollation> getCollations() {
+    return _collations;
+  }
+
+  @Override
+  protected MseBlock getNextBlock() {
+    if (_fallbackOutputIndex >= 0 && _rows != null) {
+      return emitFallbackBlock();
+    }
+    if (_eosBlock != null) {
+      return _eosBlock;
+    }
+    if (_isEarlyTerminated) {
+      return readUntilEos();
+    }
+    return _singleSortedSender ? readSingleSortedSender() : mergeNextBlock();
+  }
+
+  /// Passes through one confirmed sorted sender without copying its rows through the merge heap.
+  private MseBlock readSingleSortedSender() {
+    while (true) {
+      MseBlock block = _multiConsumer.readMseBlockBlocking();
+      if (block.isEos()) {
+        return terminate(block);
+      }
+      MseBlock.Data dataBlock = (MseBlock.Data) block;
+      checkActiveTerminationAndSampleUsage();
+      if (!_multiConsumer.isLastBlockSortedOnSender()) {
+        fallbackToFullSort(dataBlock.asRowHeap().getRows());
+        return sortAllRows();
+      }
+      if (dataBlock.getNumRows() > 0) {
+        _mergeOutputStarted = true;
+        return dataBlock;
+      }
+    }
+  }
+
+  /// Merges the sorted senders, emitting at most [SortOperator#DEFAULT_MAX_ROWS_PER_BLOCK] rows per call.
+  private MseBlock mergeNextBlock() {
+    ArrayList<Object[]> rows = new ArrayList<>(0);
+    while (rows.size() < SortOperator.DEFAULT_MAX_ROWS_PER_BLOCK) {
+      if (!_starvedCursors.isEmpty()) {

Review Comment:
   Agreed: pending per-sender blocks remain unbounded in the worst case. I kept bounded queues and retained-row telemetry out of this minimal patch; issue #19395 remains open to track the broader strict-memory-bound work. The PR description now states this limitation explicitly.



##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/SortedMailboxMergeReceiveOperator.java:
##########
@@ -0,0 +1,668 @@
+/**
+ * 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.pinot.query.runtime.operator;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.Deque;
+import java.util.IdentityHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import javax.annotation.Nullable;
+import org.apache.calcite.rel.RelFieldCollation;
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.pinot.common.utils.DataSchema;
+import org.apache.pinot.query.mailbox.ReceivingMailbox;
+import org.apache.pinot.query.planner.plannode.MailboxReceiveNode;
+import org.apache.pinot.query.runtime.blocks.MseBlock;
+import org.apache.pinot.query.runtime.blocks.RowHeapDataBlock;
+import org.apache.pinot.query.runtime.blocks.SuccessMseBlock;
+import org.apache.pinot.query.runtime.operator.utils.AsyncStream;
+import org.apache.pinot.query.runtime.operator.utils.SortUtils;
+import org.apache.pinot.query.runtime.plan.OpChainExecutionContext;
+import org.apache.pinot.spi.query.QueryThreadContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/// Receives streams that the plan declares sorted on the sender and merges them by the exchange collation.
+///
+/// An explicit sender [SortOperator] establishes the row ordering; [MailboxSendOperator] only transports that
+/// ordering. The transport marker confirms rollout compatibility and is not itself a sorting mechanism.
+///
+/// The plan declaration alone is not trusted during a rolling upgrade. Every data block must carry the transport's
+/// sender-sort confirmation. Before this operator emits its first row it obtains a head row, or EOS, from every live
+/// sender. If any sender's first data is unconfirmed, all tentatively buffered rows are folded into a full receiver
+/// sort. Once output starts, losing the confirmation is a protocol violation because already emitted rows cannot be
+/// recovered into that fallback.
+///
+/// The merge reads whichever mailbox is ready instead of blocking on one sender. This prevents a sender that is
+/// backpressured by another receiver from creating a cross-receiver wait cycle. Rows are emitted in blocks of at most
+/// 10,000 while cursor state carries the ordering frontier across calls. A fast sender can be read ahead while another
+/// sender is starved, so retained input is workload-dependent and can approach the legacy full receiver sort in the
+/// worst case.
+///
+/// This operator is driven by a single consumer thread and is not thread-safe.
+public class SortedMailboxMergeReceiveOperator extends BaseMailboxReceiveOperator implements SortedMultiStageOperator {
+  private static final Logger LOGGER = LoggerFactory.getLogger(SortedMailboxMergeReceiveOperator.class);
+
+  private static final String EXPLAIN_NAME = "SORTED_MAILBOX_MERGE_RECEIVE";
+  private static final String MERGE_SCOPE = "SortedMailboxMergeReceiveOperator";
+  private final DataSchema _dataSchema;
+  private final List<RelFieldCollation> _collations;
+  private final Comparator<Object[]> _comparator;
+  private final SenderCursorHeap _readyCursors;
+  private final boolean _singleSortedSender;
+  /// Senders that have not finished but do not currently have a row ready. Nothing can be emitted while this is
+  /// non-empty because any one of these senders may hold the next row.
+  private final Set<SenderCursor> _starvedCursors = Collections.newSetFromMap(new IdentityHashMap<>());
+  private final Map<AsyncStream<ReceivingMailbox.MseBlockWithStats>, SenderCursor> _cursorsByStream =
+      new IdentityHashMap<>();
+  private boolean _mergeOutputStarted;
+  private boolean _fallbackToSort;
+  private boolean _tryEqualHeadMerge;
+  private int _fallbackOutputIndex = -1;
+
+  /// Rows buffered only for the mixed-version fallback. The sorted list is handed downstream as-is, so cleanup must
+  /// drop this reference rather than clear it.
+  @Nullable
+  private List<Object[]> _rows;
+
+  @Nullable
+  private MseBlock _eosBlock;
+
+  public SortedMailboxMergeReceiveOperator(OpChainExecutionContext context, MailboxReceiveNode node) {
+    super(context, node);
+    Preconditions.checkState(node.isSort(), "Receiver-side sorting must be enabled");
+    Preconditions.checkState(node.isSortedOnSender(), "Sender-side sorting must be enabled");
+    Preconditions.checkState(!CollectionUtils.isEmpty(node.getCollations()), "Field collations must be set");
+    _dataSchema = node.getDataSchema();
+    _collations = List.copyOf(node.getCollations());
+    _comparator = new SortUtils.SortComparator(_collations, false);
+    List<AsyncStream<ReceivingMailbox.MseBlockWithStats>> streams = _multiConsumer.getLiveStreamsSnapshot();
+    _readyCursors = new SenderCursorHeap(streams.size(), _comparator);
+    _singleSortedSender = streams.size() == 1;
+    if (!_singleSortedSender) {
+      for (AsyncStream<ReceivingMailbox.MseBlockWithStats> stream : streams) {
+        SenderCursor cursor = new SenderCursor(stream);
+        _cursorsByStream.put(stream, cursor);
+        _starvedCursors.add(cursor);
+      }
+    }
+  }
+
+  @Override
+  protected Logger logger() {
+    return LOGGER;
+  }
+
+  @Override
+  public String toExplainString() {
+    return EXPLAIN_NAME;
+  }
+
+  @Override
+  public List<RelFieldCollation> getCollations() {
+    return _collations;
+  }
+
+  @Override
+  protected MseBlock getNextBlock() {
+    if (_fallbackOutputIndex >= 0 && _rows != null) {
+      return emitFallbackBlock();
+    }
+    if (_eosBlock != null) {
+      return _eosBlock;
+    }
+    if (_isEarlyTerminated) {
+      return readUntilEos();
+    }
+    return _singleSortedSender ? readSingleSortedSender() : mergeNextBlock();
+  }
+
+  /// Passes through one confirmed sorted sender without copying its rows through the merge heap.
+  private MseBlock readSingleSortedSender() {
+    while (true) {
+      MseBlock block = _multiConsumer.readMseBlockBlocking();
+      if (block.isEos()) {
+        return terminate(block);
+      }
+      MseBlock.Data dataBlock = (MseBlock.Data) block;
+      checkActiveTerminationAndSampleUsage();
+      if (!_multiConsumer.isLastBlockSortedOnSender()) {
+        fallbackToFullSort(dataBlock.asRowHeap().getRows());
+        return sortAllRows();
+      }
+      if (dataBlock.getNumRows() > 0) {
+        _mergeOutputStarted = true;
+        return dataBlock;
+      }
+    }
+  }
+
+  /// Merges the sorted senders, emitting at most [SortOperator#DEFAULT_MAX_ROWS_PER_BLOCK] rows per call.
+  private MseBlock mergeNextBlock() {
+    ArrayList<Object[]> rows = new ArrayList<>(0);
+    while (rows.size() < SortOperator.DEFAULT_MAX_ROWS_PER_BLOCK) {
+      if (!_starvedCursors.isEmpty()) {
+        MseBlock.Eos error;
+        boolean receivedMoreRows = false;
+        if (rows.isEmpty()) {
+          error = readOneBlock();
+        } else {
+          // Rows already removed from the heap are a globally ordered prefix. Consume any immediately available
+          // cursor progress so blocks can still be coalesced, but return that safe prefix instead of waiting merely
+          // to fill the output block.
+          MseBlock block = _multiConsumer.pollMseBlockOrStreamCompletion();
+          if (block == null && _multiConsumer.getFinishedStreamsLastRead().isEmpty()) {
+            break;
+          }
+          error = processReadBlock(block);
+          receivedMoreRows = block != null && block.isData() && ((MseBlock.Data) block).getNumRows() > 0;
+        }
+        if (error != null) {
+          return terminate(error);
+        }
+        if (_fallbackToSort) {
+          // These rows were already removed from cursors while building this not-yet-emitted block.
+          _rows.addAll(rows);
+          return sortAllRows();
+        }
+        if (receivedMoreRows) {
+          // A refill proves this is not a one-block result. Restore the established full-block capacity once so
+          // fragmented input cannot trigger repeated growth while this output block is assembled.
+          rows.ensureCapacity(SortOperator.DEFAULT_MAX_ROWS_PER_BLOCK);
+        }
+        continue;
+      }
+      if (_readyCursors.isEmpty()) {
+        break;
+      }
+      if (rows.isEmpty()) {
+        // Keep empty and tiny results cheap without sacrificing the one-allocation path for full output blocks.
+        // Sum all currently buffered rows once; later output blocks retain the established full-block capacity.
+        int initialCapacity = _mergeOutputStarted ? SortOperator.DEFAULT_MAX_ROWS_PER_BLOCK
+            : _readyCursors.getCappedAvailableRowCount(SortOperator.DEFAULT_MAX_ROWS_PER_BLOCK);
+        // ArrayList grows by 1.5x. Round near-full first blocks up now so a small refill cannot allocate an oversized
+        // replacement in addition to the nearly full initial array.
+        if (initialCapacity + (initialCapacity >> 1) >= SortOperator.DEFAULT_MAX_ROWS_PER_BLOCK) {
+          initialCapacity = SortOperator.DEFAULT_MAX_ROWS_PER_BLOCK;
+        }
+        rows.ensureCapacity(initialCapacity);
+      }
+      if (_tryEqualHeadMerge && _readyCursors.size() > 1) {

Review Comment:
   Added a deterministic randomized comparison against an independent full sort. It covers 50 fixed-seed scenarios, 2-8 senders, key cardinalities 1-32, empty and duplicate-heavy inputs, randomized block boundaries, and temporary mailbox starvation, while also checking for lost or duplicated rows.



##########
pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotWindowExchangeNodeInsertRule.java:
##########
@@ -120,13 +120,13 @@ public void onMatch(RelOptRuleCall call) {
         exchange = PinotLogicalExchange.create(input, RelDistributions.hash(List.of()));
       } else {
         // Only ORDER BY
-        // Add a LogicalSortExchange with collation on the order by key(s) and an empty hash partition key.
-        // 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.
+        // Sort each sender explicitly and merge the sorted mailbox streams at the receiver. The Sort retained above
+        // the exchange is the semantic ordering boundary and becomes a streaming limit when the merge receiver
+        // advertises this exact collation.
         // TODO: Revisit whether we should use hash distribution
         exchange =
-            PinotLogicalSortExchange.create(input, RelDistributions.hash(List.of()), windowGroup.orderKeys, false,
-                false);
+            PinotLogicalSortExchange.create(input, RelDistributions.hash(List.of()), windowGroup.orderKeys, true,

Review Comment:
   Added both controls: SET windowSortOnSender=true|false and broker config pinot.broker.multistage.window.sort.on.sender. The query option overrides the broker default, and the default is false because the expanded benchmarks found a stable presorted-input regression.



-- 
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-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] [MSE] Support sender-side sorting for sort exchanges [pinot]

Posted by "gortiz (via GitHub)" <gi...@apache.org>.

gortiz commented on PR #19396:
URL: https://github.com/apache/pinot/pull/19396#issuecomment-5478898366

   I've been working on my own fix for `SortedMailboxReceiverOperator`, which goes back to what used to be:
   
   ```
   ┌───────────────────────────────────┬─────────────────────────┬──────────────────────┐
   │                                   │ #19396                  │      My branch       │
   ├───────────────────────────────────┼─────────────────────────┼──────────────────────┤
   │                                   │ sortOnSender =          │ sortOnReceiver =     │
   │ PinotWindowExchangeNodeInsertRule │ **true**, keeps         │ **false** + explicit │
   │                                   │ sortOnReceiver = true   │  Sort                │
   ├───────────────────────────────────┼─────────────────────────┼──────────────────────┤
   │                                   │ +336 lines — k-way      │                      │
   │ SortedMailboxReceiveOperator      │ merge over sorted       │ @Deprecated          │
   │                                   │ senders                 │                      │
   ├───────────────────────────────────┼─────────────────────────┼──────────────────────┤
   │ MailboxSendOperator               │ +143 — sorts sender     │ untouched            │
   │                                   │ output, 10k blocks      │                      │
   ├───────────────────────────────────┼─────────────────────────┼──────────────────────┤
   │                                   │ new sortedOnSender      │                      │
   │ Mailbox protocol                  │ confirmation over gRPC  │ untouched            │
   │                                   │ + in-memory             │                      │
   ├───────────────────────────────────┼─────────────────────────┼──────────────────────┤
   │ PinotSortExchangeNodeInsertRule   │ comment only —          │ sortOnReceiver =     │
   │                                   │ deliberately deferred   │ false                │
   └───────────────────────────────────┴─────────────────────────┴──────────────────────┘
   ```
   
   The idea I have is to stop generating SortedMailboxReceiveOperator and substitute that with a sort on the receiver side (with optional sort on the sender side when the limit is small). Once we have that, we can start thinking about recovering SortedMailboxReceiveOperator as a k-way merge when the senders guarantee data is send in order, which is what you and #19121 are doing (although #19121 keeps both the current and the k-way merge).
   
   Also, I don't think `MailboxSendOperator` should sort on send. Instead, we should add a sort operator on the sender opchain and keep MailboxSendOperator agnostic about ordering.


-- 
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-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] [MSE] Add opt-in sender-sorted merging for global ordered windows [pinot]

Posted by "xiangfu0 (via GitHub) via commits" <co...@pinot.apache.org>.

xiangfu0 commented on PR #19396:
URL: https://github.com/apache/pinot/pull/19396#issuecomment-6644706871

   @gortiz Could you re-review #19396? It is rebased onto `fd1d3f7b59` after #19120 landed. This PR provides the cross-server merge for global ordered windows and the receiver that #19121 builds on.
   
   All 13 hosted checks passed on `0f9aad4d410`, including both unit and integration suites and the compatibility checks. [Checks](https://github.com/apache/pinot/pull/19396/checks)
   


-- 
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-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] [MSE] Merge sender-sorted streams for ordered windows [pinot]

Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.

xiangfu0 commented on PR #19396:
URL: https://github.com/apache/pinot/pull/19396#issuecomment-5492145419

@gortiz I rebased this onto current `master` and refactored it around the coordination feedback.

- Sender sorting is now owned by an explicit `SortNode` / `SortOperator`; `MailboxSendOperator` no longer performs sorting and only advertises a guarantee it can structurally verify.
- Ordered-window fan-in now uses a dedicated `SortedMailboxMergeReceiveOperator`, leaving the legacy `SortedMailboxReceiveOperator` execution path intact. That isolates the overlap with #19121 while #19120 and #19122 remain complementary producer/consumer work.
- This follows the ownership direction in #19412. If #19412 lands first, the remaining overlap should be mechanical rather than two competing sender-sort implementations.
- Mixed-version behavior is safe in either upgrade direction: missing per-block confirmation falls back to the legacy full receiver sort before output, while old receivers retain `sortOnReceiver = true`.

Validation on head `f151331206ca6a13157b1d93e55bb889d4c39c25`: the 20-module runtime reactor passed (planner 1,540 tests; runtime 4,640 tests, 18 skips), 348 focused planner/runtime tests passed, all four hygiene gates passed for planner/runtime/perf, and the eight-domain Pinot review completed with no remaining findings. The refreshed 400k-row matched JMH comparison was noisy in both directions, so I reported the raw values in the PR body without making a performance claim.

--
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-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] [MSE] Support sender-side sorting for sort exchanges [pinot]

Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.

codecov-commenter commented on PR #19396:
URL: https://github.com/apache/pinot/pull/19396#issuecomment-5464888337

   ## [Codecov](https://app.codecov.io/gh/apache/pinot/pull/19396?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
   :x: Patch coverage is `90.84967%` with `14 lines` in your changes missing coverage. Please review.
   :white_check_mark: Project coverage is 57.72%. Comparing base ([`bbbed25`](https://app.codecov.io/gh/apache/pinot/commit/bbbed251444e00d58bdcbe59a6f233902eb7e79c?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`c9f873c`](https://app.codecov.io/gh/apache/pinot/commit/c9f873c12a0d17e4a0c061ca8950cf10bc1e1e71?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
   
   | [Files with missing lines](https://app.codecov.io/gh/apache/pinot/pull/19396?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
   |---|---|---|
   | [...runtime/operator/SortedMailboxReceiveOperator.java](https://app.codecov.io/gh/apache/pinot/pull/19396?src=pr&el=tree&filepath=pinot-query-runtime%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fquery%2Fruntime%2Foperator%2FSortedMailboxReceiveOperator.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtcXVlcnktcnVudGltZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvcXVlcnkvcnVudGltZS9vcGVyYXRvci9Tb3J0ZWRNYWlsYm94UmVjZWl2ZU9wZXJhdG9yLmphdmE=) | 91.75% | [2 Missing and 6 partials :warning: ](https://app.codecov.io/gh/apache/pinot/pull/19396?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
   | [...ot/query/runtime/operator/MailboxSendOperator.java](https://app.codecov.io/gh/apache/pinot/pull/19396?src=pr&el=tree&filepath=pinot-query-runtime%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fquery%2Fruntime%2Foperator%2FMailboxSendOperator.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtcXVlcnktcnVudGltZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvcXVlcnkvcnVudGltZS9vcGVyYXRvci9NYWlsYm94U2VuZE9wZXJhdG9yLmphdmE=) | 89.13% | [2 Missing and 3 partials :warning: ](https://app.codecov.io/gh/apache/pinot/pull/19396?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
   | [...me/operator/utils/BlockingMultiStreamConsumer.java](https://app.codecov.io/gh/apache/pinot/pull/19396?src=pr&el=tree&filepath=pinot-query-runtime%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fquery%2Fruntime%2Foperator%2Futils%2FBlockingMultiStreamConsumer.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtcXVlcnktcnVudGltZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvcXVlcnkvcnVudGltZS9vcGVyYXRvci91dGlscy9CbG9ja2luZ011bHRpU3RyZWFtQ29uc3VtZXIuamF2YQ==) | 83.33% | [0 Missing and 1 partial :warning: ](https://app.codecov.io/gh/apache/pinot/pull/19396?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
   
   <details><summary>Additional details and impacted files</summary>
   
   
   
   ```diff
   @@             Coverage Diff              @@
   ##             master   #19396      +/-   ##
   ============================================
   + Coverage     57.71%   57.72%   +0.01%     
   + Complexity        7        1       -6     
   ============================================
     Files          2686     2683       -3     
     Lines        163987   164114     +127     
     Branches      26627    26653      +26     
   ============================================
   + Hits          94640    94742     +102     
   - Misses        61352    61360       +8     
   - Partials       7995     8012      +17     
   ```
   
   | [Flag](https://app.codecov.io/gh/apache/pinot/pull/19396/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
   |---|---|---|
   | [integration](https://app.codecov.io/gh/apache/pinot/pull/19396/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
   | [integration1](https://app.codecov.io/gh/apache/pinot/pull/19396/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
   | [integration2](https://app.codecov.io/gh/apache/pinot/pull/19396/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
   | [java-25](https://app.codecov.io/gh/apache/pinot/pull/19396/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `57.72% <90.84%> (+0.01%)` | :arrow_up: |
   | [lane-a](https://app.codecov.io/gh/apache/pinot/pull/19396/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
   | [lane-b](https://app.codecov.io/gh/apache/pinot/pull/19396/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
   | [temurin](https://app.codecov.io/gh/apache/pinot/pull/19396/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `57.72% <90.84%> (+0.01%)` | :arrow_up: |
   | [unittests](https://app.codecov.io/gh/apache/pinot/pull/19396/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `57.72% <90.84%> (+0.01%)` | :arrow_up: |
   | [unittests1](https://app.codecov.io/gh/apache/pinot/pull/19396/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `57.72% <90.84%> (+0.01%)` | :arrow_up: |
   
   Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
   </details>
   
   [:umbrella: View full report in Codecov by Harness](https://app.codecov.io/gh/apache/pinot/pull/19396?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).   
   :loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
   <details><summary> :rocket: New features to boost your workflow: </summary>
   
   - :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
   </details>


-- 
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-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] [MSE] Add opt-in sender-sorted merging for global ordered windows [pinot]

Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.

xiangfu0 commented on PR #19396:
URL: https://github.com/apache/pinot/pull/19396#issuecomment-5975636970

   @gortiz updated at `01af90617f`. Ordering is now a logical-plan guarantee: an [explicit Sort feeds PinotKWayMergeSortExchange](https://github.com/apache/pinot/blob/01af90617fc8789384187dd9007b40ad642314e7/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotWindowExchangeNodeInsertRule.java#L132-L177), and [fragmentation creates a plain sender plus a distinct merge receive](https://github.com/apache/pinot/blob/01af90617fc8789384187dd9007b40ad642314e7/pinot-query-planner/src/main/java/org/apache/pinot/query/planner/logical/PlanFragmenter.java#L193-L205) with serialized optional fetch/offset. Mailbox block markers, sender structural checks, production row-order checks, and the new merge receiver's full-sort fallback are removed.
   
   The [broker gate](https://github.com/apache/pinot/blob/01af90617fc8789384187dd9007b40ad642314e7/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/KWayMergeSupportPredicate.java#L38-L97) requires matching, non-SNAPSHOT broker/server release versions and fails closed on missing/unreadable configs, registration failure or disconnect; multi-cluster queries disable it. Unsupported clusters use plain receive plus an explicit MAX-fetch Sort, preserving complete window input. [Legacy `sort=true` plans still use the deprecated full-sort receiver](https://github.com/apache/pinot/blob/01af90617fc8789384187dd9007b40ad642314e7/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/plan/PlanNodeToOpChain.java#L195-L205); its replacement and old-flag cleanup remain follow-up work. The gate is a point-in-time snapshot, so a concurrent version/topology change can still cause a loud unsupported-node query failure.
   
   Recent unchanged planner/window coverage passed 962 cases, including full-input following/RANGE regressions. This latest commit only corrects the existing SNAPSHOT assertion (+7/-3 lines); both broker gate tests, JDK25 warning-enabled compilation, formatting, checkstyle and license checks passed, with no warnings on added lines. All 17 reported checks were green at `b047dd4`; CI is running on the new head. Please re-review the architecture and retained legacy compatibility path.


-- 
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-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] [MSE] Support sender-side sorting for sort exchanges [pinot]

Posted by "gortiz (via GitHub)" <gi...@apache.org>.

gortiz commented on PR #19396:
URL: https://github.com/apache/pinot/pull/19396#issuecomment-5476959551

   We are already working on something like that in #19120, #19121 and #19122. I would recommend coordinating our efforts.


-- 
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-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] [MSE] Add opt-in sender-sorted merging for global ordered windows [pinot]

Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.

xiangfu0 commented on code in PR #19396:
URL: https://github.com/apache/pinot/pull/19396#discussion_r4213852934


##########
pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandler.java:
##########
@@ -623,6 +632,9 @@ private ImmutableQueryEnvironment.Config getQueryEnvConf(HttpHeaders httpHeaders
         .defaultDisabledPlannerRules(_defaultDisabledPlannerRules)
         .defaultSortExchangeCopyLimit(sortExchangeCopyThreshold)
         .defaultSealedInListThreshold(sealedInListThreshold)
+        .defaultWindowSortOnSender(windowSortOnSender)
+        .isKWayMergeSupported(!QueryOptionsUtils.isMultiClusterRoutingEnabled(queryOptions, false)
+            && _kWayMergeSupported.getAsBoolean())

Review Comment:
   Addressed in `6f63db504b9858b45707ee13b08a30049ad55910`.
   
   The option-off path now short-circuits before consulting the capability supplier. The supplier itself only reads the shared tracker's volatile snapshot: it does not call Helix, ZooKeeper, or isConnected(). The focused capability tests verify repeated reads make no manager calls.



##########
pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/KWayMergeSupportPredicate.java:
##########
@@ -0,0 +1,99 @@
+/**
+ * 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.pinot.broker.requesthandler;
+
+import java.util.List;
+import java.util.function.BooleanSupplier;
+import org.apache.helix.HelixAdmin;
+import org.apache.helix.HelixManager;
+import org.apache.helix.NotificationContext;
+import org.apache.helix.api.listeners.BatchMode;
+import org.apache.helix.api.listeners.InstanceConfigChangeListener;
+import org.apache.helix.api.listeners.PreFetch;
+import org.apache.helix.model.InstanceConfig;
+import org.apache.pinot.common.version.PinotVersion;
+import org.apache.pinot.spi.config.instance.InstanceType;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.InstanceTypeUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/// Enables new merge plan nodes only after a complete homogeneous broker/server release-version snapshot.
+/// Uses the SAFE stats policy's version comparison, but fails closed on initialization and read failures.
+/// Snapshot builds are excluded because equal snapshot versions can contain different protocol implementations.
+/// Helix drives updates serially; query threads read the volatile result without locking.
+@BatchMode(enabled = false)
+@PreFetch(enabled = false)
+public class KWayMergeSupportPredicate implements InstanceConfigChangeListener, BooleanSupplier {
+  private static final Logger LOGGER = LoggerFactory.getLogger(KWayMergeSupportPredicate.class);
+  private final HelixManager _manager;
+  private final String _version;
+  private volatile boolean _supported;
+
+  public KWayMergeSupportPredicate(HelixManager manager) {
+    this(manager, PinotVersion.VERSION);
+  }
+
+  KWayMergeSupportPredicate(HelixManager manager, String version) {
+    _manager = manager;
+    _version = version;
+  }
+
+  @Override
+  public boolean getAsBoolean() {
+    return _supported && _manager.isConnected();
+  }
+
+  @Override
+  public synchronized void onInstanceConfigChange(List<InstanceConfig> configs, NotificationContext context) {
+    // Clear first: no partial snapshot or failed refresh may leave the new wire node enabled.
+    _supported = false;

Review Comment:
   Addressed in `6f63db504b9858b45707ee13b08a30049ad55910`.
   
   The shared tracker now computes and publishes one snapshot after a successful update, without clearing the previous result during reads. It logs incompatible live instance IDs and reported versions, plus read failures. Failures intentionally fail closed; config/live/INIT notifications refresh the state and allow recovery. Tracker tests cover publication during an update, failure/recovery and finalization.



##########
pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/KWayMergeSupportPredicate.java:
##########
@@ -0,0 +1,99 @@
+/**
+ * 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.pinot.broker.requesthandler;
+
+import java.util.List;
+import java.util.function.BooleanSupplier;
+import org.apache.helix.HelixAdmin;
+import org.apache.helix.HelixManager;
+import org.apache.helix.NotificationContext;
+import org.apache.helix.api.listeners.BatchMode;
+import org.apache.helix.api.listeners.InstanceConfigChangeListener;
+import org.apache.helix.api.listeners.PreFetch;
+import org.apache.helix.model.InstanceConfig;
+import org.apache.pinot.common.version.PinotVersion;
+import org.apache.pinot.spi.config.instance.InstanceType;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.InstanceTypeUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/// Enables new merge plan nodes only after a complete homogeneous broker/server release-version snapshot.
+/// Uses the SAFE stats policy's version comparison, but fails closed on initialization and read failures.
+/// Snapshot builds are excluded because equal snapshot versions can contain different protocol implementations.
+/// Helix drives updates serially; query threads read the volatile result without locking.
+@BatchMode(enabled = false)
+@PreFetch(enabled = false)
+public class KWayMergeSupportPredicate implements InstanceConfigChangeListener, BooleanSupplier {
+  private static final Logger LOGGER = LoggerFactory.getLogger(KWayMergeSupportPredicate.class);
+  private final HelixManager _manager;
+  private final String _version;
+  private volatile boolean _supported;
+
+  public KWayMergeSupportPredicate(HelixManager manager) {
+    this(manager, PinotVersion.VERSION);
+  }
+
+  KWayMergeSupportPredicate(HelixManager manager, String version) {
+    _manager = manager;
+    _version = version;
+  }
+
+  @Override
+  public boolean getAsBoolean() {
+    return _supported && _manager.isConnected();

Review Comment:
   Addressed in `6f63db504b9858b45707ee13b08a30049ad55910`.
   
   Removed isConnected() from the query path. The shared version tracker publishes a cached result and resets it on FINALIZE/non-update notifications; INIT refreshes it. The predicate tests cover initialization, cached reads and finalization.



##########
pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/KWayMergeSupportPredicate.java:
##########
@@ -0,0 +1,99 @@
+/**
+ * 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.pinot.broker.requesthandler;
+
+import java.util.List;
+import java.util.function.BooleanSupplier;
+import org.apache.helix.HelixAdmin;
+import org.apache.helix.HelixManager;
+import org.apache.helix.NotificationContext;
+import org.apache.helix.api.listeners.BatchMode;
+import org.apache.helix.api.listeners.InstanceConfigChangeListener;
+import org.apache.helix.api.listeners.PreFetch;
+import org.apache.helix.model.InstanceConfig;
+import org.apache.pinot.common.version.PinotVersion;
+import org.apache.pinot.spi.config.instance.InstanceType;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.InstanceTypeUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/// Enables new merge plan nodes only after a complete homogeneous broker/server release-version snapshot.
+/// Uses the SAFE stats policy's version comparison, but fails closed on initialization and read failures.
+/// Snapshot builds are excluded because equal snapshot versions can contain different protocol implementations.
+/// Helix drives updates serially; query threads read the volatile result without locking.
+@BatchMode(enabled = false)
+@PreFetch(enabled = false)
+public class KWayMergeSupportPredicate implements InstanceConfigChangeListener, BooleanSupplier {
+  private static final Logger LOGGER = LoggerFactory.getLogger(KWayMergeSupportPredicate.class);
+  private final HelixManager _manager;
+  private final String _version;
+  private volatile boolean _supported;
+
+  public KWayMergeSupportPredicate(HelixManager manager) {
+    this(manager, PinotVersion.VERSION);
+  }
+
+  KWayMergeSupportPredicate(HelixManager manager, String version) {
+    _manager = manager;
+    _version = version;
+  }
+
+  @Override
+  public boolean getAsBoolean() {
+    return _supported && _manager.isConnected();
+  }
+
+  @Override
+  public synchronized void onInstanceConfigChange(List<InstanceConfig> configs, NotificationContext context) {
+    // Clear first: no partial snapshot or failed refresh may leave the new wire node enabled.
+    _supported = false;
+    if (context.getType() != NotificationContext.Type.INIT
+        && context.getType() != NotificationContext.Type.CALLBACK) {
+      return;
+    }
+    if (_version == null || _version.equals(PinotVersion.UNKNOWN) || _version.contains("SNAPSHOT")) {

Review Comment:
   Addressed in `6f63db504b9858b45707ee13b08a30049ad55910`.
   
   Added WindowKWayMergeIntegrationTest with a scoped test-only supplier override, restored in finally; no production option bypasses the release-version gate. It passed locally and compares windowKWayMerge=true/false for ROWS/FOLLOWING and RANGE/peer frames with DESC NULLS FIRST, null/duplicate keys and 16 offline + 16 realtime rows. It asserts the physical merge plan, exact complete results and executed local/gRPC mailbox statistics.



##########
pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/KWayMergeSupportPredicate.java:
##########
@@ -0,0 +1,99 @@
+/**
+ * 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.pinot.broker.requesthandler;
+
+import java.util.List;
+import java.util.function.BooleanSupplier;
+import org.apache.helix.HelixAdmin;
+import org.apache.helix.HelixManager;
+import org.apache.helix.NotificationContext;
+import org.apache.helix.api.listeners.BatchMode;
+import org.apache.helix.api.listeners.InstanceConfigChangeListener;
+import org.apache.helix.api.listeners.PreFetch;
+import org.apache.helix.model.InstanceConfig;
+import org.apache.pinot.common.version.PinotVersion;
+import org.apache.pinot.spi.config.instance.InstanceType;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.InstanceTypeUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/// Enables new merge plan nodes only after a complete homogeneous broker/server release-version snapshot.
+/// Uses the SAFE stats policy's version comparison, but fails closed on initialization and read failures.
+/// Snapshot builds are excluded because equal snapshot versions can contain different protocol implementations.
+/// Helix drives updates serially; query threads read the volatile result without locking.
+@BatchMode(enabled = false)
+@PreFetch(enabled = false)
+public class KWayMergeSupportPredicate implements InstanceConfigChangeListener, BooleanSupplier {
+  private static final Logger LOGGER = LoggerFactory.getLogger(KWayMergeSupportPredicate.class);
+  private final HelixManager _manager;
+  private final String _version;
+  private volatile boolean _supported;
+
+  public KWayMergeSupportPredicate(HelixManager manager) {
+    this(manager, PinotVersion.VERSION);
+  }
+
+  KWayMergeSupportPredicate(HelixManager manager, String version) {
+    _manager = manager;
+    _version = version;
+  }
+
+  @Override
+  public boolean getAsBoolean() {
+    return _supported && _manager.isConnected();
+  }
+
+  @Override
+  public synchronized void onInstanceConfigChange(List<InstanceConfig> configs, NotificationContext context) {
+    // Clear first: no partial snapshot or failed refresh may leave the new wire node enabled.
+    _supported = false;
+    if (context.getType() != NotificationContext.Type.INIT
+        && context.getType() != NotificationContext.Type.CALLBACK) {
+      return;
+    }
+    if (_version == null || _version.equals(PinotVersion.UNKNOWN) || _version.contains("SNAPSHOT")) {
+      return;
+    }
+    try {
+      HelixAdmin admin = _manager.getClusterManagmentTool();
+      String cluster = _manager.getClusterName();
+      List<String> instances = admin.getInstancesInCluster(cluster);

Review Comment:
   Addressed in `6f63db504b9858b45707ee13b08a30049ad55910`.
   
   Extracted ClusterVersionTracker and reused it from the merge gate and SendStatsPredicate.Safe. Config data changes read only the changed live broker/server config; unrelated/non-live configs do not trigger reads or block the gate. Membership changes read live records to detect same-name restarts but reuse configs for unchanged sessions, refreshing only new/restarted/unreadable participants. INIT/config child changes perform a full live refresh. The tracker tests cover read counts, stale configs, restarts and failure recovery.



##########
pinot-query-planner/src/main/java/org/apache/pinot/query/planner/plannode/MailboxMergeReceiveNode.java:
##########
@@ -0,0 +1,87 @@
+/**
+ * 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.pinot.query.planner.plannode;
+
+import java.util.List;
+import java.util.Objects;
+import javax.annotation.Nullable;
+import org.apache.calcite.rel.RelDistribution;
+import org.apache.calcite.rel.RelFieldCollation;
+import org.apache.pinot.calcite.rel.logical.PinotRelExchangeType;
+import org.apache.pinot.common.utils.DataSchema;
+
+
+/// Receives streams ordered by the logical plan and preserves that order with a k-way merge.
+/// This distinct wire node is emitted only when the broker verifies homogeneous cluster versions.
+/// Instances belong to a single query plan and are not thread-safe.
+public class MailboxMergeReceiveNode extends MailboxReceiveNode {
+  private final int _fetch;
+  private final int _offset;
+
+  public MailboxMergeReceiveNode(int stageId, DataSchema schema, int senderStageId,
+      PinotRelExchangeType exchangeType, RelDistribution.Type distribution, @Nullable List<Integer> keys,
+      List<RelFieldCollation> collations, @Nullable MailboxSendNode sender) {
+    this(stageId, schema, senderStageId, exchangeType, distribution, keys, collations, -1, -1, sender);
+  }
+
+  public MailboxMergeReceiveNode(int stageId, DataSchema schema, int senderStageId,
+      PinotRelExchangeType exchangeType, RelDistribution.Type distribution, @Nullable List<Integer> keys,
+      List<RelFieldCollation> collations, int fetch, int offset, @Nullable MailboxSendNode sender) {
+    super(stageId, schema, senderStageId, exchangeType, distribution, keys, collations, false, false, sender);

Review Comment:
   Addressed in `6f63db504b9858b45707ee13b08a30049ad55910`.
   
   MailboxMergeReceiveNode is now a sibling BasePlanNode with visitMailboxMergeReceive; the ordinary receive visitors no longer distinguish it with instanceof or expose misleading sort flags. The legacy sort/sortedOnSender flags and proto fields (including sender field 7) are deprecated but retained for decoding old plans. Serde, planner, opchain, stats, pipeline and cancellation tests cover the new node.



##########
pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/BaseBrokerStarter.java:
##########
@@ -583,6 +584,13 @@ public void start()
               _accessControlFactory, _queryQuotaManager, _tableCache, _multiStageQueryThrottler, _failureDetector,
               _threadAccountant, multiClusterRoutingContext, workerManager, multiClusterWorkerManager,
               _serverRoutingStatsManager);
+      KWayMergeSupportPredicate mergeSupport = new KWayMergeSupportPredicate(_spectatorHelixManager);
+      try {
+        _spectatorHelixManager.addInstanceConfigChangeListener(mergeSupport);

Review Comment:
   Addressed in `6f63db504b9858b45707ee13b08a30049ad55910`.
   
   SNAPSHOT/unknown brokers now skip merge listener registration entirely. Release brokers retain the incremental listeners because an explicit windowKWayMerge=true query can override a false broker default. Config data events read at most the changed live participant, and the option-off query path never consults the supplier. Partial registration failure permanently stops the tracker and leaves the feature disabled.



##########
pinot-query-planner/src/main/java/org/apache/pinot/query/planner/plannode/KWayMergeExchangeNode.java:
##########
@@ -0,0 +1,44 @@
+/**
+ * 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.pinot.query.planner.plannode;
+
+import java.util.List;
+import javax.annotation.Nullable;
+import org.apache.calcite.rel.RelDistribution;
+import org.apache.calcite.rel.RelFieldCollation;
+import org.apache.pinot.calcite.rel.logical.PinotRelExchangeType;
+import org.apache.pinot.common.utils.DataSchema;
+
+
+/// Planning-only exchange requiring ordered inputs and producing a k-way merge receive and a plain send.
+/// Instances are local to a query's planning thread and are never serialized.
+public class KWayMergeExchangeNode extends ExchangeNode {
+  public KWayMergeExchangeNode(int stageId, DataSchema schema, List<PlanNode> inputs,
+      RelDistribution.Type distribution, @Nullable List<Integer> keys, boolean prePartitioned,
+      List<RelFieldCollation> collations, String hashFunction) {
+    super(stageId, schema, inputs, PinotRelExchangeType.STREAMING, distribution, keys, prePartitioned, collations,

Review Comment:
   Addressed in `6f63db504b9858b45707ee13b08a30049ad55910`.
   
   KWayMergeExchangeNode is now a separate BasePlanNode with visitKWayMergeExchange, not an ExchangeNode subtype. Its own fields express the ordered exchange contract, and fragmentation creates a plain mailbox sender plus the sibling merge receive. Routing metadata, explain, stage equivalence and traversal visitors handle the distinct node explicitly.



##########
pinot-query-planner/src/main/java/org/apache/pinot/query/planner/logical/PlanFragmenter.java:
##########
@@ -197,8 +199,10 @@ public PlanNode visitExchange(ExchangeNode node, Context context) {
     _mailboxSendToExchangeNodeMap.put(mailboxSendNode, node);
 
     // Return the MailboxReceiveNode as the leave node of the current PlanFragment.
-    MailboxReceiveNode mailboxReceiveNode =
-        new MailboxReceiveNode(receiverPlanFragmentId, nextPlanFragmentRoot.getDataSchema(),
+    MailboxReceiveNode mailboxReceiveNode = node instanceof KWayMergeExchangeNode
+        ? new MailboxMergeReceiveNode(receiverPlanFragmentId, nextPlanFragmentRoot.getDataSchema(),

Review Comment:
   Addressed in `6f63db504b9858b45707ee13b08a30049ad55910`.
   
   Removed the unused merge fetch/offset support from the constructor, node, operator, explain, serde, equivalence and limit-only tests. The merge receives complete ordered streams; limit pushdown can be added separately with a producing rule and its tests.



##########
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:
   Addressed in `6f63db504b9858b45707ee13b08a30049ad55910`.
   
   Dropped fetch/offset from the merge proto and reserved tags 6 and 7. There is no unset proto3 integer limit with empty-window semantics anymore. Merge-node serde and the hybrid integration test pass.



##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/SortedMailboxMergeReceiveOperator.java:
##########
@@ -0,0 +1,351 @@
+/**
+ * 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.pinot.query.runtime.operator;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.Deque;
+import java.util.IdentityHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.PriorityQueue;
+import java.util.Set;
+import javax.annotation.Nullable;
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.pinot.common.utils.DataSchema;
+import org.apache.pinot.query.mailbox.ReceivingMailbox;
+import org.apache.pinot.query.planner.plannode.MailboxMergeReceiveNode;
+import org.apache.pinot.query.runtime.blocks.MseBlock;
+import org.apache.pinot.query.runtime.blocks.RowHeapDataBlock;
+import org.apache.pinot.query.runtime.blocks.SuccessMseBlock;
+import org.apache.pinot.query.runtime.operator.utils.AsyncStream;
+import org.apache.pinot.query.runtime.operator.utils.SortUtils;
+import org.apache.pinot.query.runtime.plan.OpChainExecutionContext;
+import org.apache.pinot.spi.query.QueryThreadContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/// K-way merges streams whose ordering is guaranteed by the logical plan.
+///
+/// The broker only emits this operator's plan node in a homogeneous cluster. Ordering is established upstream;
+/// mailbox senders preserve it without transport metadata or runtime structural checks.
+///
+/// Reads whichever mailbox is ready to avoid cross-receiver backpressure cycles. Multi-sender output blocks contain
+/// at most 10,000 rows; a single sender preserves its input blocks. Read-ahead retained while another sender is starved
+/// can approach the legacy full receiver sort's memory footprint.
+///
+/// Driven by a single consumer thread; not thread-safe.
+public class SortedMailboxMergeReceiveOperator extends BaseMailboxReceiveOperator {
+  private static final Logger LOGGER = LoggerFactory.getLogger(SortedMailboxMergeReceiveOperator.class);
+
+  private static final String EXPLAIN_NAME = "SORTED_MAILBOX_MERGE_RECEIVE";
+  private static final String MERGE_SCOPE = "SortedMailboxMergeReceiveOperator";
+  private final DataSchema _dataSchema;
+  private final Comparator<Object[]> _comparator;
+  private final PriorityQueue<SenderCursor> _readyCursors;
+  private final boolean _singleSortedSender;
+  private int _rowsToSkip;
+  private long _rowsToEmit;
+  /// Senders that have not finished but do not currently have a row ready. Nothing can be emitted while this is
+  /// non-empty because any one of these senders may hold the next row.
+  private final Set<SenderCursor> _starvedCursors = Collections.newSetFromMap(new IdentityHashMap<>());
+  private final Map<AsyncStream<ReceivingMailbox.MseBlockWithStats>, SenderCursor> _cursorsByStream =
+      new IdentityHashMap<>();
+
+  @Nullable
+  private MseBlock _eosBlock;
+
+  public SortedMailboxMergeReceiveOperator(OpChainExecutionContext context, MailboxMergeReceiveNode node) {
+    super(context, node);
+    Preconditions.checkState(!CollectionUtils.isEmpty(node.getCollations()), "Field collations must be set");
+    _dataSchema = node.getDataSchema();
+    _rowsToSkip = Math.max(node.getOffset(), 0);
+    _rowsToEmit = node.getFetch() < 0 ? Long.MAX_VALUE : node.getFetch();
+    _comparator = new SortUtils.SortComparator(List.copyOf(node.getCollations()), false);
+    List<AsyncStream<ReceivingMailbox.MseBlockWithStats>> streams = _multiConsumer.getLiveStreamsSnapshot();
+    _readyCursors = new PriorityQueue<>(Math.max(streams.size(), 1),
+        (left, right) -> _comparator.compare(left.peek(), right.peek()));
+    _singleSortedSender = streams.size() == 1;
+    if (!_singleSortedSender) {
+      for (AsyncStream<ReceivingMailbox.MseBlockWithStats> stream : streams) {
+        SenderCursor cursor = new SenderCursor();
+        _cursorsByStream.put(stream, cursor);
+        _starvedCursors.add(cursor);
+      }
+    }
+  }
+
+  @Override
+  protected Logger logger() {
+    return LOGGER;
+  }
+
+  @Override
+  public String toExplainString() {
+    return EXPLAIN_NAME;
+  }
+
+  @Override
+  protected MseBlock getNextBlock() {
+    if (_eosBlock != null) {
+      return _eosBlock;
+    }
+    if (_isEarlyTerminated) {
+      return readUntilEos();
+    }
+    if (_rowsToEmit == 0) {
+      earlyTerminate();
+      return readUntilEos();
+    }
+    while (true) {
+      MseBlock block = _singleSortedSender ? readSingleSortedSender() : mergeNextBlock();
+      if (block.isEos()) {
+        return block;
+      }
+      if (_rowsToSkip == 0 && _rowsToEmit == Long.MAX_VALUE) {
+        return block;
+      }
+      List<Object[]> rows = ((MseBlock.Data) block).asRowHeap().getRows();
+      int from = Math.min(_rowsToSkip, rows.size());
+      _rowsToSkip -= from;
+      int count = (int) Math.min(rows.size() - from, _rowsToEmit);
+      _rowsToEmit -= count;
+      if (_rowsToEmit == 0) {
+        earlyTerminate();
+      }
+      if (count > 0) {
+        return from == 0 && count == rows.size() ? block
+            : new RowHeapDataBlock(rows.subList(from, from + count), _dataSchema);
+      }
+    }
+  }
+
+  /// Passes through one sorted sender without copying its rows through the merge heap.
+  private MseBlock readSingleSortedSender() {
+    while (true) {
+      MseBlock block = _multiConsumer.readMseBlockBlocking();
+      if (block.isEos()) {
+        return terminate(block);
+      }
+      MseBlock.Data dataBlock = (MseBlock.Data) block;
+      checkActiveTerminationAndSampleUsage();
+
+      if (dataBlock.getNumRows() > 0) {
+        return dataBlock;
+      }
+    }
+  }
+
+  /// Merges the sorted senders, emitting at most [SortOperator#DEFAULT_MAX_ROWS_PER_BLOCK] rows per call.
+  private MseBlock mergeNextBlock() {
+    List<Object[]> rows = new ArrayList<>();
+    while (rows.size() < SortOperator.DEFAULT_MAX_ROWS_PER_BLOCK) {
+      if (!_starvedCursors.isEmpty()) {
+        MseBlock.Eos error;
+        if (rows.isEmpty()) {
+          error = readOneBlock();
+        } else {
+          // Rows already removed from the heap are a globally ordered prefix. Consume any immediately available
+          // cursor progress so blocks can still be coalesced, but return that safe prefix instead of waiting merely
+          // to fill the output block.
+          MseBlock block = _multiConsumer.pollMseBlockOrStreamCompletion();
+          if (block == null && _multiConsumer.getFinishedStreamsLastRead().isEmpty()) {
+            break;
+          }
+          error = processReadBlock(block);
+        }
+        if (error != null) {
+          return terminate(error);
+        }
+
+        continue;
+      }
+      if (_readyCursors.isEmpty()) {
+        break;
+      }
+      SenderCursor cursor = _readyCursors.remove();
+      rows.add(cursor.next());
+      if (cursor.hasRow()) {
+        _readyCursors.add(cursor);
+      } else if (!cursor._finished) {
+        _starvedCursors.add(cursor);
+      }
+      QueryThreadContext.checkTerminationAndSampleUsagePeriodically(rows.size(), MERGE_SCOPE,
+          _context.getActiveDeadlineMs());

Review Comment:
   Addressed in `6f63db504b9858b45707ee13b08a30049ad55910`.
   
   Removed the active-deadline checks. The merge loop now uses the inherited termination/resource-sampling methods and therefore BaseMailboxReceiveOperator's passive deadline. The merge/operator and focused runtime tests pass.



##########
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:
   Addressed in `6f63db504b9858b45707ee13b08a30049ad55910`.
   
   Replaced the fetch==MAX_VALUE sentinel with the serialized requiresSingleRun SortNode hint. PlanFragmenter attaches it only to the merge exchange's explicit sender sort, and ServerPlanRequestVisitor retains that sort above the combined hybrid/logical leaf input. The visitor tests verify that marked sorts remain above the boundary while an ordinary user MAX_VALUE limit still receives V1 pushdown; the hybrid integration test exercises the combined input.



##########
pinot-segment-local/pom.xml:
##########
@@ -88,6 +88,13 @@
       <artifactId>h3</artifactId>
     </dependency>
 
+    <!-- Zstd's API annotations are required when javac resolves annotated return types with warnings enabled. -->

Review Comment:
   Addressed in `6f63db504b9858b45707ee13b08a30049ad55910`.
   
   Removed the annotations dependency. pinot-segment-local/pom.xml now has no diff against rebased master. Warning-enabled validation used freshly installed current dependencies for the affected modules without reintroducing the unrelated POM change.



##########
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:
   Addressed in `6f63db504b9858b45707ee13b08a30049ad55910`.
   
   Removed the dead isSortOnReceiver guard and obsolete upgrade comment. Ordinary sort exchanges retain the explicit receiver sort; the distinct k-way exchange establishes its own ordered output. The unrelated receiver MAX-fetch hardening and window JSON changes were also removed, and the full resource-based plan suite passes.



-- 
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-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] [MSE] Add opt-in sender-sorted merging for global ordered windows [pinot]

Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.

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: commits-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] [MSE] Merge sender-sorted streams for ordered windows [pinot]

Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.

xiangfu0 commented on PR #19396:
URL: https://github.com/apache/pinot/pull/19396#issuecomment-5805916311

   Rebased and simplified on top of `master` `257a0e7df2` (which includes #19412); the updated PR head is `7ecc01c0fd`. The patch shrank from 38 to 27 files and from 2,379 to 1,957 changed lines, with the benchmark harness kept temporary and out of the PR.
   
   Fresh matched JMH results on JDK 25, 400k rows, two servers, local + gRPC fan-in:
   
   | Ordered window | Rebased master | PR | Mean change |
   |---|---:|---:|---:|
   | Global | 193.400 ± 9.810 ms/op (3 forks, 15 measurements) | 137.801 ± 14.177 ms/op | -28.7% |
   | Partitioned | 247.859 ± 27.999 ms/op (2 forks, 10 measurements) | 184.532 ± 31.146 ms/op | -25.5% |
   
   Both JMH-reported intervals are non-overlapping. Local validation also passed: 89 focused runtime tests, all 243 `QueryCompilationTest` cases, and planner/runtime Spotless, Checkstyle, and license gates. Fresh exact-head CI is now running after the force-with-lease update.
   


-- 
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-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] [MSE] Merge sender-sorted streams for ordered windows [pinot]

Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.

xiangfu0 commented on PR #19396:
URL: https://github.com/apache/pinot/pull/19396#issuecomment-5803358904

   Rebased now that #19412 is merged. This uses its explicit sender Sort, keeps MailboxSendOperator order-agnostic, isolates k-way merge in the ordered-window receiver, and preserves safe fallback for mixed-version/custom transports.
   
   Focused planner/runtime and real local/gRPC transport tests pass on f87d986d5c; fresh CI is running.


-- 
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-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] [MSE] Add opt-in sender-sorted merging for global ordered windows [pinot]

Posted by "gortiz (via GitHub)" <gi...@apache.org>.

gortiz commented on code in PR #19396:
URL: https://github.com/apache/pinot/pull/19396#discussion_r4207838063


##########
pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/KWayMergeSupportPredicate.java:
##########
@@ -0,0 +1,99 @@
+/**
+ * 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.pinot.broker.requesthandler;
+
+import java.util.List;
+import java.util.function.BooleanSupplier;
+import org.apache.helix.HelixAdmin;
+import org.apache.helix.HelixManager;
+import org.apache.helix.NotificationContext;
+import org.apache.helix.api.listeners.BatchMode;
+import org.apache.helix.api.listeners.InstanceConfigChangeListener;
+import org.apache.helix.api.listeners.PreFetch;
+import org.apache.helix.model.InstanceConfig;
+import org.apache.pinot.common.version.PinotVersion;
+import org.apache.pinot.spi.config.instance.InstanceType;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.InstanceTypeUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/// Enables new merge plan nodes only after a complete homogeneous broker/server release-version snapshot.
+/// Uses the SAFE stats policy's version comparison, but fails closed on initialization and read failures.
+/// Snapshot builds are excluded because equal snapshot versions can contain different protocol implementations.
+/// Helix drives updates serially; query threads read the volatile result without locking.
+@BatchMode(enabled = false)
+@PreFetch(enabled = false)
+public class KWayMergeSupportPredicate implements InstanceConfigChangeListener, BooleanSupplier {
+  private static final Logger LOGGER = LoggerFactory.getLogger(KWayMergeSupportPredicate.class);
+  private final HelixManager _manager;
+  private final String _version;
+  private volatile boolean _supported;
+
+  public KWayMergeSupportPredicate(HelixManager manager) {
+    this(manager, PinotVersion.VERSION);
+  }
+
+  KWayMergeSupportPredicate(HelixManager manager, String version) {
+    _manager = manager;
+    _version = version;
+  }
+
+  @Override
+  public boolean getAsBoolean() {
+    return _supported && _manager.isConnected();
+  }
+
+  @Override
+  public synchronized void onInstanceConfigChange(List<InstanceConfig> configs, NotificationContext context) {
+    // Clear first: no partial snapshot or failed refresh may leave the new wire node enabled.
+    _supported = false;

Review Comment:
   Clearing the flag first makes the feature flap off during every rescan, and if the scan throws it stays off until the next instance config event, which may never come in a stable cluster. I'd compute into a local and publish once. Also, unlike `SendStatsPredicate`, nothing logs which instances keep the gate closed, so an operator who enables the option can't tell why it does nothing.



##########
pinot-query-planner/src/main/java/org/apache/pinot/query/planner/plannode/MailboxMergeReceiveNode.java:
##########
@@ -0,0 +1,87 @@
+/**
+ * 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.pinot.query.planner.plannode;
+
+import java.util.List;
+import java.util.Objects;
+import javax.annotation.Nullable;
+import org.apache.calcite.rel.RelDistribution;
+import org.apache.calcite.rel.RelFieldCollation;
+import org.apache.pinot.calcite.rel.logical.PinotRelExchangeType;
+import org.apache.pinot.common.utils.DataSchema;
+
+
+/// Receives streams ordered by the logical plan and preserves that order with a k-way merge.
+/// This distinct wire node is emitted only when the broker verifies homogeneous cluster versions.
+/// Instances belong to a single query plan and are not thread-safe.
+public class MailboxMergeReceiveNode extends MailboxReceiveNode {
+  private final int _fetch;
+  private final int _offset;
+
+  public MailboxMergeReceiveNode(int stageId, DataSchema schema, int senderStageId,
+      PinotRelExchangeType exchangeType, RelDistribution.Type distribution, @Nullable List<Integer> keys,
+      List<RelFieldCollation> collations, @Nullable MailboxSendNode sender) {
+    this(stageId, schema, senderStageId, exchangeType, distribution, keys, collations, -1, -1, sender);
+  }
+
+  public MailboxMergeReceiveNode(int stageId, DataSchema schema, int senderStageId,
+      PinotRelExchangeType exchangeType, RelDistribution.Type distribution, @Nullable List<Integer> keys,
+      List<RelFieldCollation> collations, int fetch, int offset, @Nullable MailboxSendNode sender) {
+    super(stageId, schema, senderStageId, exchangeType, distribution, keys, collations, false, false, sender);

Review Comment:
   The subclass passes `sort=false, sortedOnSender=false` to a parent whose flags mean the opposite of what this node does. Any code that reads `isSort()` gets the wrong answer for it, and every visitor needs an `instanceof` (serializer, `PlanNodeToRelConverter`, `PlanNodeMerger`, `EquivalentStagesFinder`, `PlanFragmenter`, `PlanNodeToOpChain`). I'd make it a sibling `PlanNode` with its own `visitMailboxMergeReceive`, and deprecate the old `sort`/`sortedOnSender` flags in this PR so we end up with a single model.



##########
pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/KWayMergeSupportPredicate.java:
##########
@@ -0,0 +1,99 @@
+/**
+ * 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.pinot.broker.requesthandler;
+
+import java.util.List;
+import java.util.function.BooleanSupplier;
+import org.apache.helix.HelixAdmin;
+import org.apache.helix.HelixManager;
+import org.apache.helix.NotificationContext;
+import org.apache.helix.api.listeners.BatchMode;
+import org.apache.helix.api.listeners.InstanceConfigChangeListener;
+import org.apache.helix.api.listeners.PreFetch;
+import org.apache.helix.model.InstanceConfig;
+import org.apache.pinot.common.version.PinotVersion;
+import org.apache.pinot.spi.config.instance.InstanceType;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.InstanceTypeUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/// Enables new merge plan nodes only after a complete homogeneous broker/server release-version snapshot.
+/// Uses the SAFE stats policy's version comparison, but fails closed on initialization and read failures.
+/// Snapshot builds are excluded because equal snapshot versions can contain different protocol implementations.
+/// Helix drives updates serially; query threads read the volatile result without locking.
+@BatchMode(enabled = false)
+@PreFetch(enabled = false)
+public class KWayMergeSupportPredicate implements InstanceConfigChangeListener, BooleanSupplier {
+  private static final Logger LOGGER = LoggerFactory.getLogger(KWayMergeSupportPredicate.class);
+  private final HelixManager _manager;
+  private final String _version;
+  private volatile boolean _supported;
+
+  public KWayMergeSupportPredicate(HelixManager manager) {
+    this(manager, PinotVersion.VERSION);
+  }
+
+  KWayMergeSupportPredicate(HelixManager manager, String version) {
+    _manager = manager;
+    _version = version;
+  }
+
+  @Override
+  public boolean getAsBoolean() {
+    return _supported && _manager.isConnected();

Review Comment:
   The `isConnected()` check doesn't add anything here: a session expiry already triggers FINALIZE and a new INIT. Returning `_supported` alone avoids taking the ZK event lock on the query path.



##########
pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/KWayMergeSupportPredicate.java:
##########
@@ -0,0 +1,99 @@
+/**
+ * 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.pinot.broker.requesthandler;
+
+import java.util.List;
+import java.util.function.BooleanSupplier;
+import org.apache.helix.HelixAdmin;
+import org.apache.helix.HelixManager;
+import org.apache.helix.NotificationContext;
+import org.apache.helix.api.listeners.BatchMode;
+import org.apache.helix.api.listeners.InstanceConfigChangeListener;
+import org.apache.helix.api.listeners.PreFetch;
+import org.apache.helix.model.InstanceConfig;
+import org.apache.pinot.common.version.PinotVersion;
+import org.apache.pinot.spi.config.instance.InstanceType;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.InstanceTypeUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/// Enables new merge plan nodes only after a complete homogeneous broker/server release-version snapshot.
+/// Uses the SAFE stats policy's version comparison, but fails closed on initialization and read failures.
+/// Snapshot builds are excluded because equal snapshot versions can contain different protocol implementations.
+/// Helix drives updates serially; query threads read the volatile result without locking.
+@BatchMode(enabled = false)
+@PreFetch(enabled = false)
+public class KWayMergeSupportPredicate implements InstanceConfigChangeListener, BooleanSupplier {
+  private static final Logger LOGGER = LoggerFactory.getLogger(KWayMergeSupportPredicate.class);
+  private final HelixManager _manager;
+  private final String _version;
+  private volatile boolean _supported;
+
+  public KWayMergeSupportPredicate(HelixManager manager) {
+    this(manager, PinotVersion.VERSION);
+  }
+
+  KWayMergeSupportPredicate(HelixManager manager, String version) {
+    _manager = manager;
+    _version = version;
+  }
+
+  @Override
+  public boolean getAsBoolean() {
+    return _supported && _manager.isConnected();
+  }
+
+  @Override
+  public synchronized void onInstanceConfigChange(List<InstanceConfig> configs, NotificationContext context) {
+    // Clear first: no partial snapshot or failed refresh may leave the new wire node enabled.
+    _supported = false;
+    if (context.getType() != NotificationContext.Type.INIT
+        && context.getType() != NotificationContext.Type.CALLBACK) {
+      return;
+    }
+    if (_version == null || _version.equals(PinotVersion.UNKNOWN) || _version.contains("SNAPSHOT")) {

Review Comment:
   Excluding SNAPSHOT means no CI run, quickstart or locally built cluster can take the merge path, so it will run for the first time in a release cluster. I'd keep a test-only way to force the gate open and add an integration test that compares the results with the receiver sort.



##########
pinot-segment-local/pom.xml:
##########
@@ -88,6 +88,13 @@
       <artifactId>h3</artifactId>
     </dependency>
 
+    <!-- Zstd's API annotations are required when javac resolves annotated return types with warnings enabled. -->

Review Comment:
   This looks unrelated to the PR. Nothing in `pinot-segment-local` uses these annotations. If it's needed for warning-enabled builds, I'd move it to its own PR.



##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/SortedMailboxMergeReceiveOperator.java:
##########
@@ -0,0 +1,351 @@
+/**
+ * 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.pinot.query.runtime.operator;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.Deque;
+import java.util.IdentityHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.PriorityQueue;
+import java.util.Set;
+import javax.annotation.Nullable;
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.pinot.common.utils.DataSchema;
+import org.apache.pinot.query.mailbox.ReceivingMailbox;
+import org.apache.pinot.query.planner.plannode.MailboxMergeReceiveNode;
+import org.apache.pinot.query.runtime.blocks.MseBlock;
+import org.apache.pinot.query.runtime.blocks.RowHeapDataBlock;
+import org.apache.pinot.query.runtime.blocks.SuccessMseBlock;
+import org.apache.pinot.query.runtime.operator.utils.AsyncStream;
+import org.apache.pinot.query.runtime.operator.utils.SortUtils;
+import org.apache.pinot.query.runtime.plan.OpChainExecutionContext;
+import org.apache.pinot.spi.query.QueryThreadContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/// K-way merges streams whose ordering is guaranteed by the logical plan.
+///
+/// The broker only emits this operator's plan node in a homogeneous cluster. Ordering is established upstream;
+/// mailbox senders preserve it without transport metadata or runtime structural checks.
+///
+/// Reads whichever mailbox is ready to avoid cross-receiver backpressure cycles. Multi-sender output blocks contain
+/// at most 10,000 rows; a single sender preserves its input blocks. Read-ahead retained while another sender is starved
+/// can approach the legacy full receiver sort's memory footprint.
+///
+/// Driven by a single consumer thread; not thread-safe.
+public class SortedMailboxMergeReceiveOperator extends BaseMailboxReceiveOperator {
+  private static final Logger LOGGER = LoggerFactory.getLogger(SortedMailboxMergeReceiveOperator.class);
+
+  private static final String EXPLAIN_NAME = "SORTED_MAILBOX_MERGE_RECEIVE";
+  private static final String MERGE_SCOPE = "SortedMailboxMergeReceiveOperator";
+  private final DataSchema _dataSchema;
+  private final Comparator<Object[]> _comparator;
+  private final PriorityQueue<SenderCursor> _readyCursors;
+  private final boolean _singleSortedSender;
+  private int _rowsToSkip;
+  private long _rowsToEmit;
+  /// Senders that have not finished but do not currently have a row ready. Nothing can be emitted while this is
+  /// non-empty because any one of these senders may hold the next row.
+  private final Set<SenderCursor> _starvedCursors = Collections.newSetFromMap(new IdentityHashMap<>());
+  private final Map<AsyncStream<ReceivingMailbox.MseBlockWithStats>, SenderCursor> _cursorsByStream =
+      new IdentityHashMap<>();
+
+  @Nullable
+  private MseBlock _eosBlock;
+
+  public SortedMailboxMergeReceiveOperator(OpChainExecutionContext context, MailboxMergeReceiveNode node) {
+    super(context, node);
+    Preconditions.checkState(!CollectionUtils.isEmpty(node.getCollations()), "Field collations must be set");
+    _dataSchema = node.getDataSchema();
+    _rowsToSkip = Math.max(node.getOffset(), 0);
+    _rowsToEmit = node.getFetch() < 0 ? Long.MAX_VALUE : node.getFetch();
+    _comparator = new SortUtils.SortComparator(List.copyOf(node.getCollations()), false);
+    List<AsyncStream<ReceivingMailbox.MseBlockWithStats>> streams = _multiConsumer.getLiveStreamsSnapshot();
+    _readyCursors = new PriorityQueue<>(Math.max(streams.size(), 1),
+        (left, right) -> _comparator.compare(left.peek(), right.peek()));
+    _singleSortedSender = streams.size() == 1;
+    if (!_singleSortedSender) {
+      for (AsyncStream<ReceivingMailbox.MseBlockWithStats> stream : streams) {
+        SenderCursor cursor = new SenderCursor();
+        _cursorsByStream.put(stream, cursor);
+        _starvedCursors.add(cursor);
+      }
+    }
+  }
+
+  @Override
+  protected Logger logger() {
+    return LOGGER;
+  }
+
+  @Override
+  public String toExplainString() {
+    return EXPLAIN_NAME;
+  }
+
+  @Override
+  protected MseBlock getNextBlock() {
+    if (_eosBlock != null) {
+      return _eosBlock;
+    }
+    if (_isEarlyTerminated) {
+      return readUntilEos();
+    }
+    if (_rowsToEmit == 0) {
+      earlyTerminate();
+      return readUntilEos();
+    }
+    while (true) {
+      MseBlock block = _singleSortedSender ? readSingleSortedSender() : mergeNextBlock();
+      if (block.isEos()) {
+        return block;
+      }
+      if (_rowsToSkip == 0 && _rowsToEmit == Long.MAX_VALUE) {
+        return block;
+      }
+      List<Object[]> rows = ((MseBlock.Data) block).asRowHeap().getRows();
+      int from = Math.min(_rowsToSkip, rows.size());
+      _rowsToSkip -= from;
+      int count = (int) Math.min(rows.size() - from, _rowsToEmit);
+      _rowsToEmit -= count;
+      if (_rowsToEmit == 0) {
+        earlyTerminate();
+      }
+      if (count > 0) {
+        return from == 0 && count == rows.size() ? block
+            : new RowHeapDataBlock(rows.subList(from, from + count), _dataSchema);
+      }
+    }
+  }
+
+  /// Passes through one sorted sender without copying its rows through the merge heap.
+  private MseBlock readSingleSortedSender() {
+    while (true) {
+      MseBlock block = _multiConsumer.readMseBlockBlocking();
+      if (block.isEos()) {
+        return terminate(block);
+      }
+      MseBlock.Data dataBlock = (MseBlock.Data) block;
+      checkActiveTerminationAndSampleUsage();
+
+      if (dataBlock.getNumRows() > 0) {
+        return dataBlock;
+      }
+    }
+  }
+
+  /// Merges the sorted senders, emitting at most [SortOperator#DEFAULT_MAX_ROWS_PER_BLOCK] rows per call.
+  private MseBlock mergeNextBlock() {
+    List<Object[]> rows = new ArrayList<>();
+    while (rows.size() < SortOperator.DEFAULT_MAX_ROWS_PER_BLOCK) {
+      if (!_starvedCursors.isEmpty()) {
+        MseBlock.Eos error;
+        if (rows.isEmpty()) {
+          error = readOneBlock();
+        } else {
+          // Rows already removed from the heap are a globally ordered prefix. Consume any immediately available
+          // cursor progress so blocks can still be coalesced, but return that safe prefix instead of waiting merely
+          // to fill the output block.
+          MseBlock block = _multiConsumer.pollMseBlockOrStreamCompletion();
+          if (block == null && _multiConsumer.getFinishedStreamsLastRead().isEmpty()) {
+            break;
+          }
+          error = processReadBlock(block);
+        }
+        if (error != null) {
+          return terminate(error);
+        }
+
+        continue;
+      }
+      if (_readyCursors.isEmpty()) {
+        break;
+      }
+      SenderCursor cursor = _readyCursors.remove();
+      rows.add(cursor.next());
+      if (cursor.hasRow()) {
+        _readyCursors.add(cursor);
+      } else if (!cursor._finished) {
+        _starvedCursors.add(cursor);
+      }
+      QueryThreadContext.checkTerminationAndSampleUsagePeriodically(rows.size(), MERGE_SCOPE,
+          _context.getActiveDeadlineMs());

Review Comment:
   Receive operators use the passive deadline on purpose (`BaseMailboxReceiveOperator.getDeadlineMs()`), so the upstream operator that is actually working times out first and reports a useful error. Using the active deadline here makes this stage time out first with a generic error. I'd use the inherited `checkTerminationAndSampleUsage*` methods. Same at line 257.



##########
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:
   These fields are never set by the planner, and once released they are permanent. Their semantics also differ from `SortNode`: here `fetch=0` means "emit nothing", while `SortOperator` treats `fetch<=0` as unbounded. Since proto3 encodes an unset int32 as 0, any producer that forgets `setFetch(-1)` silently returns empty windows. I'd drop them for now.



##########
pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandler.java:
##########
@@ -623,6 +632,9 @@ private ImmutableQueryEnvironment.Config getQueryEnvConf(HttpHeaders httpHeaders
         .defaultDisabledPlannerRules(_defaultDisabledPlannerRules)
         .defaultSortExchangeCopyLimit(sortExchangeCopyThreshold)
         .defaultSealedInListThreshold(sealedInListThreshold)
+        .defaultWindowSortOnSender(windowSortOnSender)
+        .isKWayMergeSupported(!QueryOptionsUtils.isMultiClusterRoutingEnabled(queryOptions, false)
+            && _kWayMergeSupported.getAsBoolean())

Review Comment:
   This runs for every MSE query, with or without windows and with the option on or off. `getAsBoolean()` calls `isConnected()`, which goes through `ZkClient.waitForKeeperState` and acquires the ZK event lock (I checked the Helix 2.0.1 bytecode). During a session expiry and reconnect, every MSE query blocks here. Could the predicate return only the volatile flag, and could we evaluate it only when the window option is enabled?



##########
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:
   `fetch == Integer.MAX_VALUE` becomes an implicit contract between the planner and the server. A user query with `ORDER BY x LIMIT 2147483647` also hits this branch and loses the V1 sort pushdown. The requirement belongs to the exchange (the merge needs a single run per sender), so I'd use an explicit marker for it.



##########
pinot-query-planner/src/main/java/org/apache/pinot/query/planner/plannode/KWayMergeExchangeNode.java:
##########
@@ -0,0 +1,44 @@
+/**
+ * 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.pinot.query.planner.plannode;
+
+import java.util.List;
+import javax.annotation.Nullable;
+import org.apache.calcite.rel.RelDistribution;
+import org.apache.calcite.rel.RelFieldCollation;
+import org.apache.pinot.calcite.rel.logical.PinotRelExchangeType;
+import org.apache.pinot.common.utils.DataSchema;
+
+
+/// Planning-only exchange requiring ordered inputs and producing a k-way merge receive and a plain send.
+/// Instances are local to a query's planning thread and are never serialized.
+public class KWayMergeExchangeNode extends ExchangeNode {
+  public KWayMergeExchangeNode(int stageId, DataSchema schema, List<PlanNode> inputs,
+      RelDistribution.Type distribution, @Nullable List<Integer> keys, boolean prePartitioned,
+      List<RelFieldCollation> collations, String hashFunction) {
+    super(stageId, schema, inputs, PinotRelExchangeType.STREAMING, distribution, keys, prePartitioned, collations,

Review Comment:
   Same here: the parent's `sortOnSender`/`sortOnReceiver` are hard-coded to false. Since `ExchangeNode` is planner-internal, I'd either keep a single exchange node, or make the merge exchange a separate type, but not a subtype that contradicts the parent.



##########
pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/BaseBrokerStarter.java:
##########
@@ -583,6 +584,13 @@ public void start()
               _accessControlFactory, _queryQuotaManager, _tableCache, _multiStageQueryThrottler, _failureDetector,
               _threadAccountant, multiClusterRoutingContext, workerManager, multiClusterWorkerManager,
               _serverRoutingStatsManager);
+      KWayMergeSupportPredicate mergeSupport = new KWayMergeSupportPredicate(_spectatorHelixManager);
+      try {
+        _spectatorHelixManager.addInstanceConfigChangeListener(mergeSupport);

Review Comment:
   This is registered on every broker, even when the broker config is off and no query will ever enable the option. I'd register it only when the feature can be enabled, or make it cheap enough that it doesn't matter.



##########
pinot-query-planner/src/main/java/org/apache/pinot/query/planner/logical/PlanFragmenter.java:
##########
@@ -197,8 +199,10 @@ public PlanNode visitExchange(ExchangeNode node, Context context) {
     _mailboxSendToExchangeNodeMap.put(mailboxSendNode, node);
 
     // Return the MailboxReceiveNode as the leave node of the current PlanFragment.
-    MailboxReceiveNode mailboxReceiveNode =
-        new MailboxReceiveNode(receiverPlanFragmentId, nextPlanFragmentRoot.getDataSchema(),
+    MailboxReceiveNode mailboxReceiveNode = node instanceof KWayMergeExchangeNode
+        ? new MailboxMergeReceiveNode(receiverPlanFragmentId, nextPlanFragmentRoot.getDataSchema(),

Review Comment:
   This is the only production constructor call and it always passes fetch/offset -1, so the fetch/offset support in the node, the proto, the operator, the explain and the equivalence check is only reachable from tests. I'd remove it until a rule pushes a limit into the merge receive, or add that rule and its plan tests in this PR.



##########
pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/KWayMergeSupportPredicate.java:
##########
@@ -0,0 +1,99 @@
+/**
+ * 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.pinot.broker.requesthandler;
+
+import java.util.List;
+import java.util.function.BooleanSupplier;
+import org.apache.helix.HelixAdmin;
+import org.apache.helix.HelixManager;
+import org.apache.helix.NotificationContext;
+import org.apache.helix.api.listeners.BatchMode;
+import org.apache.helix.api.listeners.InstanceConfigChangeListener;
+import org.apache.helix.api.listeners.PreFetch;
+import org.apache.helix.model.InstanceConfig;
+import org.apache.pinot.common.version.PinotVersion;
+import org.apache.pinot.spi.config.instance.InstanceType;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.InstanceTypeUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/// Enables new merge plan nodes only after a complete homogeneous broker/server release-version snapshot.
+/// Uses the SAFE stats policy's version comparison, but fails closed on initialization and read failures.
+/// Snapshot builds are excluded because equal snapshot versions can contain different protocol implementations.
+/// Helix drives updates serially; query threads read the volatile result without locking.
+@BatchMode(enabled = false)
+@PreFetch(enabled = false)
+public class KWayMergeSupportPredicate implements InstanceConfigChangeListener, BooleanSupplier {
+  private static final Logger LOGGER = LoggerFactory.getLogger(KWayMergeSupportPredicate.class);
+  private final HelixManager _manager;
+  private final String _version;
+  private volatile boolean _supported;
+
+  public KWayMergeSupportPredicate(HelixManager manager) {
+    this(manager, PinotVersion.VERSION);
+  }
+
+  KWayMergeSupportPredicate(HelixManager manager, String version) {
+    _manager = manager;
+    _version = version;
+  }
+
+  @Override
+  public boolean getAsBoolean() {
+    return _supported && _manager.isConnected();
+  }
+
+  @Override
+  public synchronized void onInstanceConfigChange(List<InstanceConfig> configs, NotificationContext context) {
+    // Clear first: no partial snapshot or failed refresh may leave the new wire node enabled.
+    _supported = false;
+    if (context.getType() != NotificationContext.Type.INIT
+        && context.getType() != NotificationContext.Type.CALLBACK) {
+      return;
+    }
+    if (_version == null || _version.equals(PinotVersion.UNKNOWN) || _version.contains("SNAPSHOT")) {
+      return;
+    }
+    try {
+      HelixAdmin admin = _manager.getClusterManagmentTool();
+      String cluster = _manager.getClusterName();
+      List<String> instances = admin.getInstancesInCluster(cluster);

Review Comment:
   This rescans every broker and server on every callback: 1 + N ZK reads on the spectator event thread, for data changes too. Servers write their instance config on start and shutdown, so a rolling restart costs O(N²) reads per broker. `SendStatsPredicate.Safe` only rescans on INIT or child changes and reads the single changed path otherwise. `getInstancesInCluster` also returns non-live instances, so a stale znode from a decommissioned host keeps the feature off. Could we extract the version tracking from `SendStatsPredicate` into something shared instead of duplicating it?



##########
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:
   Every `PinotLogicalSortExchange.create` in this rule passes `sortOnReceiver=false`, so this condition is always true, and the comment describes a rolling-upgrade case this rule can't produce. I'd remove both.



-- 
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-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] [MSE] Add opt-in sender-sorted merging for global ordered windows [pinot]

Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.

xiangfu0 commented on code in PR #19396:
URL: https://github.com/apache/pinot/pull/19396#discussion_r4213852932


##########
pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/KWayMergeSupportPredicate.java:
##########
@@ -0,0 +1,99 @@
+/**
+ * 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.pinot.broker.requesthandler;
+
+import java.util.List;
+import java.util.function.BooleanSupplier;
+import org.apache.helix.HelixAdmin;
+import org.apache.helix.HelixManager;
+import org.apache.helix.NotificationContext;
+import org.apache.helix.api.listeners.BatchMode;
+import org.apache.helix.api.listeners.InstanceConfigChangeListener;
+import org.apache.helix.api.listeners.PreFetch;
+import org.apache.helix.model.InstanceConfig;
+import org.apache.pinot.common.version.PinotVersion;
+import org.apache.pinot.spi.config.instance.InstanceType;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.InstanceTypeUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/// Enables new merge plan nodes only after a complete homogeneous broker/server release-version snapshot.
+/// Uses the SAFE stats policy's version comparison, but fails closed on initialization and read failures.
+/// Snapshot builds are excluded because equal snapshot versions can contain different protocol implementations.
+/// Helix drives updates serially; query threads read the volatile result without locking.
+@BatchMode(enabled = false)
+@PreFetch(enabled = false)
+public class KWayMergeSupportPredicate implements InstanceConfigChangeListener, BooleanSupplier {
+  private static final Logger LOGGER = LoggerFactory.getLogger(KWayMergeSupportPredicate.class);
+  private final HelixManager _manager;
+  private final String _version;
+  private volatile boolean _supported;
+
+  public KWayMergeSupportPredicate(HelixManager manager) {
+    this(manager, PinotVersion.VERSION);
+  }
+
+  KWayMergeSupportPredicate(HelixManager manager, String version) {
+    _manager = manager;
+    _version = version;
+  }
+
+  @Override
+  public boolean getAsBoolean() {
+    return _supported && _manager.isConnected();
+  }
+
+  @Override
+  public synchronized void onInstanceConfigChange(List<InstanceConfig> configs, NotificationContext context) {
+    // Clear first: no partial snapshot or failed refresh may leave the new wire node enabled.
+    _supported = false;
+    if (context.getType() != NotificationContext.Type.INIT
+        && context.getType() != NotificationContext.Type.CALLBACK) {
+      return;
+    }
+    if (_version == null || _version.equals(PinotVersion.UNKNOWN) || _version.contains("SNAPSHOT")) {

Review Comment:
   Updated at `85be3b0e80e4265ceeb6258e7db4582c4415511c` to reflect the current implementation.
   
   The current WindowKWayMergeIntegrationTest forces the test-only supplier true inside try/finally and resets it to false in finally; it does not restore a saved production supplier. No production query option bypasses the release-version gate. The fixture contains 8 offline + 8 realtime rows and compares windowKWayMerge=true/false for ROWS/FOLLOWING and RANGE/peer frames with DESC NULLS FIRST, null and duplicate keys. It asserts complete 16-row results and a physical merge plan through the real cluster planner/proto/server/opchain path; it does not assert local/gRPC mailbox statistics. The earlier local pass is historical validation, not a claim about fresh rebased-head CI.



##########
pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/KWayMergeSupportPredicate.java:
##########
@@ -0,0 +1,99 @@
+/**
+ * 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.pinot.broker.requesthandler;
+
+import java.util.List;
+import java.util.function.BooleanSupplier;
+import org.apache.helix.HelixAdmin;
+import org.apache.helix.HelixManager;
+import org.apache.helix.NotificationContext;
+import org.apache.helix.api.listeners.BatchMode;
+import org.apache.helix.api.listeners.InstanceConfigChangeListener;
+import org.apache.helix.api.listeners.PreFetch;
+import org.apache.helix.model.InstanceConfig;
+import org.apache.pinot.common.version.PinotVersion;
+import org.apache.pinot.spi.config.instance.InstanceType;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.InstanceTypeUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/// Enables new merge plan nodes only after a complete homogeneous broker/server release-version snapshot.
+/// Uses the SAFE stats policy's version comparison, but fails closed on initialization and read failures.
+/// Snapshot builds are excluded because equal snapshot versions can contain different protocol implementations.
+/// Helix drives updates serially; query threads read the volatile result without locking.
+@BatchMode(enabled = false)
+@PreFetch(enabled = false)
+public class KWayMergeSupportPredicate implements InstanceConfigChangeListener, BooleanSupplier {
+  private static final Logger LOGGER = LoggerFactory.getLogger(KWayMergeSupportPredicate.class);
+  private final HelixManager _manager;
+  private final String _version;
+  private volatile boolean _supported;
+
+  public KWayMergeSupportPredicate(HelixManager manager) {
+    this(manager, PinotVersion.VERSION);
+  }
+
+  KWayMergeSupportPredicate(HelixManager manager, String version) {
+    _manager = manager;
+    _version = version;
+  }
+
+  @Override
+  public boolean getAsBoolean() {
+    return _supported && _manager.isConnected();
+  }
+
+  @Override
+  public synchronized void onInstanceConfigChange(List<InstanceConfig> configs, NotificationContext context) {
+    // Clear first: no partial snapshot or failed refresh may leave the new wire node enabled.
+    _supported = false;
+    if (context.getType() != NotificationContext.Type.INIT
+        && context.getType() != NotificationContext.Type.CALLBACK) {
+      return;
+    }
+    if (_version == null || _version.equals(PinotVersion.UNKNOWN) || _version.contains("SNAPSHOT")) {
+      return;
+    }
+    try {
+      HelixAdmin admin = _manager.getClusterManagmentTool();
+      String cluster = _manager.getClusterName();
+      List<String> instances = admin.getInstancesInCluster(cluster);

Review Comment:
   Updated at `85be3b0e80e4265ceeb6258e7db4582c4415511c` to reflect the current implementation.
   
   The current shared ClusterVersionTracker is reused by the merge gate and SendStatsPredicate.Safe. Config data changes read only a changed tracked live broker/server config. Topology callbacks read live child names, reuse cached configs for still-present names and read newly live participants; INIT or recovery after a failed read refreshes every live participant. Non-live configs do not block the gate. This implementation does not track live-session IDs or separately detect same-name restarts; config notifications provide version updates. Tests cover incremental read counts, stale non-live configs and failure recovery, rather than session-based restart detection.



##########
pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/BaseBrokerStarter.java:
##########
@@ -583,6 +584,13 @@ public void start()
               _accessControlFactory, _queryQuotaManager, _tableCache, _multiStageQueryThrottler, _failureDetector,
               _threadAccountant, multiClusterRoutingContext, workerManager, multiClusterWorkerManager,
               _serverRoutingStatsManager);
+      KWayMergeSupportPredicate mergeSupport = new KWayMergeSupportPredicate(_spectatorHelixManager);
+      try {
+        _spectatorHelixManager.addInstanceConfigChangeListener(mergeSupport);

Review Comment:
   Updated at `85be3b0e80e4265ceeb6258e7db4582c4415511c` to reflect the current implementation.
   
   The current implementation registers both incremental listeners even for SNAPSHOT/unknown brokers, but the tracker disables unsupported local versions and returns before callback reads. Release brokers keep those listeners because an explicit windowKWayMerge=true query can override a false broker default. Option-off queries do not consult the supplier. Config data events read only the changed tracked live participant; topology updates reuse still-present configs. If registration fails, both listeners are removed and the handler retains its default disabled supplier; there is no permanent-stop method on the tracker.



##########
pinot-query-planner/src/main/java/org/apache/pinot/query/planner/plannode/MailboxMergeReceiveNode.java:
##########
@@ -0,0 +1,87 @@
+/**
+ * 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.pinot.query.planner.plannode;
+
+import java.util.List;
+import java.util.Objects;
+import javax.annotation.Nullable;
+import org.apache.calcite.rel.RelDistribution;
+import org.apache.calcite.rel.RelFieldCollation;
+import org.apache.pinot.calcite.rel.logical.PinotRelExchangeType;
+import org.apache.pinot.common.utils.DataSchema;
+
+
+/// Receives streams ordered by the logical plan and preserves that order with a k-way merge.
+/// This distinct wire node is emitted only when the broker verifies homogeneous cluster versions.
+/// Instances belong to a single query plan and are not thread-safe.
+public class MailboxMergeReceiveNode extends MailboxReceiveNode {
+  private final int _fetch;
+  private final int _offset;
+
+  public MailboxMergeReceiveNode(int stageId, DataSchema schema, int senderStageId,
+      PinotRelExchangeType exchangeType, RelDistribution.Type distribution, @Nullable List<Integer> keys,
+      List<RelFieldCollation> collations, @Nullable MailboxSendNode sender) {
+    this(stageId, schema, senderStageId, exchangeType, distribution, keys, collations, -1, -1, sender);
+  }
+
+  public MailboxMergeReceiveNode(int stageId, DataSchema schema, int senderStageId,
+      PinotRelExchangeType exchangeType, RelDistribution.Type distribution, @Nullable List<Integer> keys,
+      List<RelFieldCollation> collations, int fetch, int offset, @Nullable MailboxSendNode sender) {
+    super(stageId, schema, senderStageId, exchangeType, distribution, keys, collations, false, false, sender);

Review Comment:
   Updated at `85be3b0e80e4265ceeb6258e7db4582c4415511c` to reflect the current implementation.
   
   MailboxMergeReceiveNode and ordinary MailboxReceiveNode are now sibling classes under BaseMailboxReceiveNode, with a dedicated visitMailboxMergeReceive method. The merge receive exposes collations without the ordinary receiver sorting flags. The legacy sort/sortedOnSender Java accessors and proto fields, including sender field 7, are deprecated but retained for decoding old plans. Earlier serde/planner/opchain/runtime validation is historical evidence; fresh rebased-head CI remains separate.



##########
pinot-query-planner/src/main/java/org/apache/pinot/query/planner/plannode/KWayMergeExchangeNode.java:
##########
@@ -0,0 +1,44 @@
+/**
+ * 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.pinot.query.planner.plannode;
+
+import java.util.List;
+import javax.annotation.Nullable;
+import org.apache.calcite.rel.RelDistribution;
+import org.apache.calcite.rel.RelFieldCollation;
+import org.apache.pinot.calcite.rel.logical.PinotRelExchangeType;
+import org.apache.pinot.common.utils.DataSchema;
+
+
+/// Planning-only exchange requiring ordered inputs and producing a k-way merge receive and a plain send.
+/// Instances are local to a query's planning thread and are never serialized.
+public class KWayMergeExchangeNode extends ExchangeNode {
+  public KWayMergeExchangeNode(int stageId, DataSchema schema, List<PlanNode> inputs,
+      RelDistribution.Type distribution, @Nullable List<Integer> keys, boolean prePartitioned,
+      List<RelFieldCollation> collations, String hashFunction) {
+    super(stageId, schema, inputs, PinotRelExchangeType.STREAMING, distribution, keys, prePartitioned, collations,

Review Comment:
   Updated at `85be3b0e80e4265ceeb6258e7db4582c4415511c` to reflect the current implementation.
   
   The current implementation uses the single-exchange alternative you suggested: ExchangeNode.kWayMerge(...) records an explicit isKWayMerge mode, rather than a separate KWayMergeExchangeNode subtype. It retains the existing visitExchange method; there is no visitKWayMergeExchange method. Fragmentation creates a plain mailbox sender plus the sibling MailboxMergeReceiveNode, whose dedicated receive visitor carries the ordered-stream contract.



-- 
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-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] [MSE] Merge sender-sorted streams for global ordered windows [pinot]

Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.

xiangfu0 commented on PR #19396:
URL: https://github.com/apache/pinot/pull/19396#issuecomment-5843323460

   @gortiz Thanks for the detailed review. Final head `2673b6149f` is rebased on current master and addresses the requests: the optimization is default-off with broker/query controls, the redundant receiver-side Sort and no-op machinery are removed, and randomized merge plus implementation-plan/op-chain coverage are added. I also documented the remaining unbounded read-ahead under open #19395 and deferred the nonblocking shared consumer API cleanup. All 14 inline threads now have evidence-backed replies and are resolved; fresh exact-head CI is running.


-- 
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-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] [MSE] Merge sender-sorted streams for global ordered windows [pinot]

Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.

xiangfu0 commented on PR #19396:
URL: https://github.com/apache/pinot/pull/19396#issuecomment-5880261791

   Update to supersede my September 26 default-off summary: current head `1feb42217b` is rebased on master `a3af29e92b` and makes `windowSortOnSender=auto` the default. AUTO chooses separately for each global ordered-window exchange; unknown or nonqualifying shapes keep the receiver full sort, and explicit `true`/`false` remain available.
   
   The PR description now includes the new A–B–B–A and same-cluster paired measurements, including the remaining small/uncertain fallback overhead. Focused tests, package build, formatting, checkstyle, license, and added-line warning checks passed locally. Exact-head CI is running; no unresolved review thread remains. I am not claiming a universal no-regression guarantee or requesting merge before CI/re-review.
   


-- 
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-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] [MSE] Add opt-in sender-sorted merging for global ordered windows [pinot]

Posted by "xiangfu0 (via GitHub) via commits" <co...@pinot.apache.org>.

xiangfu0 commented on PR #19396:
URL: https://github.com/apache/pinot/pull/19396#issuecomment-6535640637

   @gortiz could you re-review this and #19711? Ordering is now guaranteed by the logical plan through a dedicated merge-receive node; mailbox markers and runtime ordering checks are removed. The version gate reads cached incremental cluster state, and explicit single-run sorts plus passive receive deadlines address the compatibility and timeout feedback.
   
   Randomized merge tests and the real two-server hybrid ROWS/RANGE comparison pass. Rebased onto master; all [13 current-head checks](https://github.com/apache/pinot/pull/19396/checks?sha=be31a33d114c5df89475c4da102b2f23595bc54f) pass on `be31a33d11`.
   


-- 
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-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org