rohityadav1993 commented on code in PR #19120:
URL: https://github.com/apache/pinot/pull/19120#discussion_r4378283946
##########
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:
Added to the description. "No behavior change when the option is off" now
names this as the one exception: the consumer-thread CPU and allocation stats
are summed over all streamed blocks for every streaming combine, not only this
one. "Thread CPU and allocation stats" now says `systemActivitiesCpuTimeNs` is
a lower bound, because the time blocked in the send is excluded.
##########
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:
Done in 91386dcdfd. The `AUTO` Javadoc now says it is experimental like
`ON`: the default ratio is not measured, and the check ignores segment size and
overlap (#19924). The `sortedSelectionMergeMode` query option Javadoc and the
options table in the description say the same.
##########
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:
Done in 91386dcdfd, with one change: #20003 removed
`lateMaterializeSortedOrderBy` (late materialization is now always on for the
materialized path), so this PR is now rebased on it. The class Javadoc now says
the two-phase fetch is the operator's own and needs no query option, and that
the materialized sorted-prefix operators late-materialize the same way since
#20003. Agreed on the three copies of the payload fetch
(`SelectionOrderByOperator`, `LinearSelectionOrderByOperator` and this
operator). The shared helper in #19923 (A2) will 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]
