gortiz commented on code in PR #19120:
URL: https://github.com/apache/pinot/pull/19120#discussion_r4376711023
##########
pinot-core/src/main/java/org/apache/pinot/core/operator/InstanceResponseOperator.java:
##########
@@ -116,14 +124,14 @@ protected BaseResultsBlock getBaseBlock() {
BaseResultsBlock resultsBlock = getCombinedResults();
- long mainThreadCpuTimeNs = resourceSnapshot.getCpuTimeNs();
- long mainThreadMemAllocatedBytes = resourceSnapshot.getAllocatedBytes();
-
- long totalWallClockTimeNs = System.nanoTime() - startWallClockTimeNs;
+ _mainThreadCpuTimeNs += resourceSnapshot.getCpuTimeNs();
Review Comment:
This is more accurate than before: the old code kept only the last call,
which on a streaming response is the terminal metadata block, so main-thread
CPU and allocations were close to zero and system activities came out as 0. The
only caveat is that the summed wall clock excludes time blocked in the send, so
system activities is now a lower bound, which is still better than 0. Since
this applies to every streaming combine, please mention it in the description.
##########
pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java:
##########
@@ -1698,6 +1711,53 @@ public enum AndRestrictionPushdownMode {
// Sequential combine utilizes N + 1 threads, N for processing and 1 for
merge. Pair-wise only uses N.
public static final int
DEFAULT_SORT_AGGREGATE_SEQUENTIAL_COMBINE_NUM_SEGMENTS_THRESHOLD =
Runtime.getRuntime().availableProcessors();
+
+ /// Selects the streaming k-way-merge selection ORDER BY combine
+ /// ([Broker.Request.QueryOptionKey#SORTED_SELECTION_MERGE_MODE]).
+ ///
+ /// The merge runs single-threaded on the consumer thread, so it trades
`maxExecutionThreads` worth of parallel
+ /// segment processing for the ability to emit sorted blocks
incrementally. That trade only pays off when the
+ /// segments are physically sorted on the leading ORDER BY column, since
an unsorted child must do a full scan
+ /// plus top-K synchronously before the merge can take its first row.
+ public enum SortedSelectionMergeMode {
+ /// Never use the streaming merge. Selection ORDER BY keeps its existing
combine operators.
+ OFF,
+ /// Always use the streaming merge, including when the leading ORDER BY
expression is not an identifier and no
+ /// segment is sorted. Children then materialize their top-K eagerly and
the merge is strictly slower than
+ /// [#OFF]; intended for benchmarking and A/B, not for production
traffic.
+ ON,
+ /// Use the streaming merge only when at least
+ ///
[Broker.Request.QueryOptionKey#SORTED_SELECTION_MERGE_AUTO_MIN_SORTED_RATIO] of
the queried segments are
+ /// physically sorted on the leading ORDER BY column. Decided from
segment metadata alone, without acquiring
+ /// any segment.
+ AUTO
Review Comment:
Could we say here that AUTO is experimental for now, like ON? The default
ratio is unmeasured and the heuristic ignores segment size and overlap (tracked
in #19924).
##########
pinot-core/src/main/java/org/apache/pinot/core/operator/query/StreamingSelectionOrderByOperator.java:
##########
@@ -0,0 +1,620 @@
+/**
+ * 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.core.operator.query;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.CaseFormat;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.PriorityQueue;
+import java.util.Set;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import javax.annotation.Nullable;
+import org.apache.pinot.common.request.context.ExpressionContext;
+import org.apache.pinot.common.request.context.OrderByExpressionContext;
+import org.apache.pinot.common.utils.DataSchema;
+import org.apache.pinot.common.utils.HashUtil;
+import org.apache.pinot.core.common.BlockValSet;
+import org.apache.pinot.core.common.Operator;
+import org.apache.pinot.core.common.RowBasedBlockValueFetcher;
+import org.apache.pinot.core.operator.BaseOperator;
+import org.apache.pinot.core.operator.BaseProjectOperator;
+import org.apache.pinot.core.operator.BitmapDocIdSetOperator;
+import org.apache.pinot.core.operator.ColumnContext;
+import org.apache.pinot.core.operator.ExecutionStatistics;
+import org.apache.pinot.core.operator.ExplainAttributeBuilder;
+import org.apache.pinot.core.operator.ProjectionOperator;
+import org.apache.pinot.core.operator.ProjectionOperatorUtils;
+import org.apache.pinot.core.operator.blocks.ValueBlock;
+import org.apache.pinot.core.operator.blocks.results.SelectionResultsBlock;
+import org.apache.pinot.core.operator.transform.TransformOperator;
+import
org.apache.pinot.core.operator.transform.function.TransformFunctionFactory;
+import org.apache.pinot.core.query.request.context.QueryContext;
+import org.apache.pinot.core.query.selection.SelectionOperatorUtils;
+import org.apache.pinot.core.query.utils.OrderByComparatorFactory;
+import org.apache.pinot.segment.spi.IndexSegment;
+import org.apache.pinot.segment.spi.datasource.DataSource;
+import org.apache.pinot.spi.query.QueryScanCostContext;
+import org.roaringbitmap.RoaringBitmap;
+
+
+/// Lazy, incremental selection ORDER BY operator for segments that are
physically sorted on the first order-by column.
+///
+/// Unlike [SelectionOrderByOperator] (which materializes the segment's whole
top-K in a single block) this
+/// operator emits one globally-sorted [SelectionResultsBlock] per
[#getNextBlock()] call and returns
+/// `null` when the segment is exhausted, so that a downstream k-way-merge
combine operator can pull from many
+/// segments lazily and stop early. It relies on the underlying project
operator iterating the first order-by column in
+/// the query order (the caller must guarantee
`projectOperator.isCompatibleWith(DocIdOrder.fromAsc(asc))`).
+///
+/// Every instance emits at least one block before `null`: a segment matching
no rows emits a single empty block
+/// carrying the [DataSchema], so consumers never have to reconstruct a schema
the segment already knows.
+///
+/// It runs in one of two emission modes:
+///
+/// - **No tail to sort** (`numSortedExpressions == numOrderByExpressions`,
e.g. `ORDER BY sorted`):
+/// rows already arrive from the project operator in final order, so each
call emits the next project block
+/// (trimmed to the remaining `limit + offset` budget).
+/// - **Tail to sort** (`numSortedExpressions < numOrderByExpressions`, e.g.
+/// `ORDER BY sorted, other`):
+/// each call reads forward until a value of the sorted prefix changes (a
"run": rows that tie on every sorted
+/// order-by column), retains the run's top rows by the unsorted tail, as
many as the remaining `limit + offset`
+/// budget allows, and emits them sorted. This bounds the in-memory run
buffer to `limit + offset` rows even when
+/// the sorted prefix is near-constant (very low cardinality).
+///
+/// Like [SelectionOrderByOperator] it preserves the two-phase projection
optimization: when there are output
Review Comment:
Nit: this operator always does its own two-phase fetch and ignores
lateMaterializeSortedOrderBy. That's fine, but a sentence here would avoid
confusion now that #19879 added that option. Also, with #19879 there are three
copies of the selected-row payload fetch, so the shared helper in #19923 could
start from LinearSelectionOrderByOperator's version.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
