This is an automated email from the ASF dual-hosted git repository.
xiangfu0 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 310d7604fcd Materialize non-order-by columns late on the sorted
order-by path (#19879)
310d7604fcd is described below
commit 310d7604fcd89f8348eeaabb9f56de8a9b74d163
Author: Arunkumar Saravanan <[email protected]>
AuthorDate: Thu Oct 22 11:26:29 2026 +0530
Materialize non-order-by columns late on the sorted order-by path (#19879)
---
.../common/utils/config/QueryOptionsUtils.java | 8 +
.../query/LinearSelectionOrderByOperator.java | 216 ++++++++++++++++++---
.../SelectionPartiallyOrderedByDescOperation.java | 11 +-
.../SelectionPartiallyOrderedByLinearOperator.java | 11 +-
.../apache/pinot/core/plan/SelectionPlanNode.java | 27 ++-
...nnerSegmentSelectionSingleValueQueriesTest.java | 94 ++++++++-
.../queries/NullHandlingEnabledQueriesTest.java | 34 ++++
.../apache/pinot/spi/utils/CommonConstants.java | 10 +
8 files changed, 377 insertions(+), 34 deletions(-)
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/utils/config/QueryOptionsUtils.java
b/pinot-common/src/main/java/org/apache/pinot/common/utils/config/QueryOptionsUtils.java
index 936c968aaf0..491417d7315 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/utils/config/QueryOptionsUtils.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/utils/config/QueryOptionsUtils.java
@@ -964,6 +964,14 @@ public class QueryOptionsUtils {
return queryOptions.getOrDefault(QueryOptionKey.WORKLOAD_NAME,
CommonConstants.Accounting.DEFAULT_WORKLOAD_NAME);
}
+ public static boolean isLateMaterializeSortedOrderByEnabled(Map<String,
String> queryOptions) {
+ String value =
queryOptions.get(QueryOptionKey.LATE_MATERIALIZE_SORTED_ORDER_BY);
+ if (value == null) {
+ return QueryOptionKey.DEFAULT_LATE_MATERIALIZE_SORTED_ORDER_BY;
+ }
+ return Boolean.parseBoolean(value);
+ }
+
public static boolean isReverseOrderAllowed(Map<String, String>
queryOptions) {
String value = queryOptions.get(QueryOptionKey.ALLOW_REVERSE_ORDER);
if (value == null) {
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/operator/query/LinearSelectionOrderByOperator.java
b/pinot-core/src/main/java/org/apache/pinot/core/operator/query/LinearSelectionOrderByOperator.java
index 1e54d9526ed..b5a16551b25 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/operator/query/LinearSelectionOrderByOperator.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/operator/query/LinearSelectionOrderByOperator.java
@@ -22,13 +22,19 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.CaseFormat;
import com.google.common.base.Preconditions;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
+import java.util.Map;
import java.util.PriorityQueue;
+import java.util.Set;
import java.util.function.IntFunction;
import java.util.function.Supplier;
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;
@@ -36,15 +42,21 @@ import org.apache.pinot.core.common.BlockValSet;
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.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;
@@ -75,11 +87,25 @@ public abstract class LinearSelectionOrderByOperator
extends BaseOperator<Select
protected final int _numRowsToKeep;
protected final Comparator<Object[]> _comparator;
protected final Supplier<ListBuilder> _listBuilderSupplier;
+ protected final int _numOrderByExpressions;
+ /// When true the project operator only produces the order-by expressions,
and the remaining output expressions are
+ /// fetched for the rows that survive the limit. Rows carry their doc id at
index [#_numOrderByExpressions] until
+ /// that fetch overwrites it.
+ protected final boolean _lateMaterialize;
+ private long _numEntriesScannedPostFilterLate;
/// @param expressions Order-by expressions must be at the head of
the list.
/// @param numSortedExpressions Number of expressions in the order-by
expressions that are sorted.
public LinearSelectionOrderByOperator(IndexSegment indexSegment,
QueryContext queryContext,
List<ExpressionContext> expressions, BaseProjectOperator<?>
projectOperator, int numSortedExpressions) {
+ this(indexSegment, queryContext, expressions, projectOperator,
numSortedExpressions, false);
+ }
+
+ /// @param lateMaterialize whether `projectOperator` produces only the
order-by expressions, leaving the remaining
+ /// output expressions to be fetched once the limit
has been applied.
+ public LinearSelectionOrderByOperator(IndexSegment indexSegment,
QueryContext queryContext,
+ List<ExpressionContext> expressions, BaseProjectOperator<?>
projectOperator, int numSortedExpressions,
+ boolean lateMaterialize) {
_indexSegment = indexSegment;
_queryContext = queryContext;
_nullHandlingEnabled = queryContext.isNullHandlingEnabled();
@@ -90,10 +116,20 @@ public abstract class LinearSelectionOrderByOperator
extends BaseOperator<Select
assert _orderByExpressions != null;
int numOrderByExpressions = _orderByExpressions.size();
+ _numOrderByExpressions = numOrderByExpressions;
+ _lateMaterialize = lateMaterialize;
+ // Rows carry the doc id at index numOrderByExpressions, so there must be
at least one expression after the
+ // order-by ones for that slot to exist (and something worth deferring).
+ Preconditions.checkArgument(!lateMaterialize || expressions.size() >
numOrderByExpressions,
+ "Late materialization requires output expressions beyond the %s
order-by ones, got %s",
+ numOrderByExpressions, expressions.size());
+
_alreadySorted = expressions.subList(0, numSortedExpressions);
_toSort = expressions.subList(numSortedExpressions, numOrderByExpressions);
- _columnContexts = new ColumnContext[_expressions.size()];
+ // Only the projected expressions can be resolved against the project
operator. When materialising late it holds
+ // the order-by expressions alone, which is also all the comparator needs.
+ _columnContexts = new ColumnContext[lateMaterialize ?
numOrderByExpressions : _expressions.size()];
for (int i = 0; i < _columnContexts.length; i++) {
ExpressionContext expression = _expressions.get(i);
_columnContexts[i] = _projectOperator.getResultColumnContext(expression);
@@ -124,39 +160,72 @@ public abstract class LinearSelectionOrderByOperator
extends BaseOperator<Select
public ExecutionStatistics getExecutionStatistics() {
long numEntriesScannedInFilter =
_projectOperator.getExecutionStatistics().getNumEntriesScannedInFilter();
int numDocsScanned = getNumDocsScanned();
- long numEntriesScannedPostFilter = (long) numDocsScanned *
_projectOperator.getNumColumnsProjected();
+ long numEntriesScannedPostFilter =
+ (long) numDocsScanned * _projectOperator.getNumColumnsProjected() +
_numEntriesScannedPostFilterLate;
int numTotalDocs = _indexSegment.getSegmentMetadata().getTotalDocs();
return new ExecutionStatistics(numDocsScanned, numEntriesScannedInFilter,
numEntriesScannedPostFilter,
numTotalDocs);
}
+ /// Number of expressions the project operator actually produces: all of
them, or just the order-by prefix when
+ /// materialising late.
+ protected int getNumProjectedExpressions() {
+ return _columnContexts.length;
+ }
+
protected IntFunction<Object[]> fetchBlock(ValueBlock valueBlock,
BlockValSet[] blockValSets) {
- int numExpressions = _expressions.size();
+ int numProjected = getNumProjectedExpressions();
- for (int i = 0; i < numExpressions; i++) {
+ for (int i = 0; i < numProjected; i++) {
ExpressionContext expression = _expressions.get(i);
blockValSets[i] = valueBlock.getBlockValueSet(expression);
}
RowBasedBlockValueFetcher blockValueFetcher = new
RowBasedBlockValueFetcher(blockValSets);
- if (!_nullHandlingEnabled) {
- return blockValueFetcher::getRow;
+ RoaringBitmap[] nullBitmaps;
+ if (_nullHandlingEnabled) {
+ nullBitmaps = new RoaringBitmap[numProjected];
+ for (int i = 0; i < numProjected; i++) {
+ nullBitmaps[i] = blockValSets[i].getNullBitmap();
+ }
+ } else {
+ nullBitmaps = null;
}
- RoaringBitmap[] nullBitmaps = new RoaringBitmap[numExpressions];
- for (int i = 0; i < numExpressions; i++) {
- nullBitmaps[i] = blockValSets[i].getNullBitmap();
+
+ if (!_lateMaterialize) {
+ if (nullBitmaps == null) {
+ return blockValueFetcher::getRow;
+ }
+ return (rowIdInBlock) -> {
+ Object[] row = blockValueFetcher.getRow(rowIdInBlock);
+ applyNulls(row, nullBitmaps, rowIdInBlock);
+ return row;
+ };
}
- return (docId) -> {
- Object[] row = blockValueFetcher.getRow(docId);
- for (int colId = 0; colId < nullBitmaps.length; colId++) {
- if (nullBitmaps[colId] != null && nullBitmaps[colId].contains(docId)) {
- row[colId] = null;
- }
+
+ // Leave room for the expressions fetched after the limit is applied, and
carry the doc id in the first of those
+ // slots so the fetch knows which documents to read. That slot is
overwritten by the first fetched value.
+ int numExpressions = _expressions.size();
+ int[] docIds = valueBlock.getDocIds();
+ return (rowIdInBlock) -> {
+ Object[] row = new Object[numExpressions];
+ blockValueFetcher.getRow(rowIdInBlock, row, 0);
+ if (nullBitmaps != null) {
+ applyNulls(row, nullBitmaps, rowIdInBlock);
}
+ row[_numOrderByExpressions] = docIds[rowIdInBlock];
return row;
};
}
+ private static void applyNulls(Object[] row, RoaringBitmap[] nullBitmaps,
int rowIdInBlock) {
+ for (int colId = 0; colId < nullBitmaps.length; colId++) {
+ if (nullBitmaps[colId] != null &&
nullBitmaps[colId].contains(rowIdInBlock)) {
+ row[colId] = null;
+ }
+ }
+ }
+
protected abstract int getNumDocsScanned();
/// Returns a list of rows sorted that:
@@ -198,6 +267,9 @@ public abstract class LinearSelectionOrderByOperator
extends BaseOperator<Select
sb.append(", unsortedList: ");
concatList(sb, _toSort);
+ if (_lateMaterialize) {
+ sb.append(", lateMaterialized: true");
+ }
sb.append(", rest: ");
concatList(sb, _expressions.subList(_alreadySorted.size() +
_toSort.size(), _expressions.size()));
@@ -215,6 +287,9 @@ public abstract class LinearSelectionOrderByOperator
extends BaseOperator<Select
List<String> toSort =
_toSort.stream().map(ExpressionContext::toString).collect(Collectors.toList());
List<String> restStr =
rest.stream().map(ExpressionContext::toString).collect(Collectors.toList());
+ if (_lateMaterialize) {
+ attributeBuilder.putString("lateMaterialized", "true");
+ }
attributeBuilder.putStringList("sortedList", sortedList)
.putStringList("unsortedList", toSort)
.putStringList("rest", restStr);
@@ -234,25 +309,122 @@ public abstract class LinearSelectionOrderByOperator
extends BaseOperator<Select
@Override
protected SelectionResultsBlock getNextBlock() {
- return new SelectionResultsBlock(createDataSchema(),
fetch(_listBuilderSupplier), _comparator, _queryContext);
+ List<Object[]> rows = fetch(_listBuilderSupplier);
+ if (!_lateMaterialize) {
+ return new SelectionResultsBlock(createDataSchema(), rows, _comparator,
_queryContext);
+ }
+ return fetchRemainingExpressions(rows);
}
- protected DataSchema createDataSchema() {
+ /// Fills the output expressions that were not projected during the scan,
for the rows that survived the limit.
+ ///
+ /// The rows are visited in doc-id order because that is the order
[BitmapDocIdSetOperator#ascending] returns, while
+ /// `rows` keeps the result order. Both views hold the same row arrays, so
filling one fills the other.
+ private SelectionResultsBlock fetchRemainingExpressions(List<Object[]> rows)
{
int numExpressions = _expressions.size();
+ List<ExpressionContext> nonOrderByExpressions =
_expressions.subList(_numOrderByExpressions, numExpressions);
+ int numNonOrderByExpressions = nonOrderByExpressions.size();
+
+ int numRows = rows.size();
+ Object[][] rowsByDocId = rows.toArray(new Object[numRows][]);
+ Arrays.sort(rowsByDocId, Comparator.comparingInt(row -> (int)
row[_numOrderByExpressions]));
+ RoaringBitmap docIds = new RoaringBitmap();
+ for (Object[] row : rowsByDocId) {
+ docIds.add((int) row[_numOrderByExpressions]);
+ }
+ // rowsByDocId is filled by walking the bitmap in ascending order, so the
two must line up one for one.
+ // They do only because surviving rows have distinct doc ids. If that ever
stopped holding, the bitmap would
+ // dedupe and the trailing rows would keep their unfilled values instead
of failing -- wrong data returned,
+ // not an error. Checked rather than asserted because this build never
runs with -ea, and the cost is one
+ // cardinality read per query against a projection pass over every
surviving row.
+ Preconditions.checkState(docIds.getCardinality() == numRows,
+ "Expected one distinct doc id per surviving row, got %s doc ids for %s
rows", docIds.getCardinality(),
+ numRows);
+
+ Set<String> columns = new HashSet<>();
+ for (ExpressionContext expression : nonOrderByExpressions) {
+ expression.getColumns(columns);
+ }
+ Map<String, DataSource> dataSourceMap = new HashMap<>();
+ for (String column : columns) {
+ // Same resolution the eager path uses: ProjectPlanNode resolves every
selected column through this exact
+ // call, including the ones fetched here. The schema is only
dereferenced for a column the segment does not
+ // physically have, and it is set for every server query in
ServerQueryExecutorV1Impl.
+ dataSourceMap.put(column, _indexSegment.getDataSource(column,
_queryContext.getSchema()));
+ }
+ // Distinct physical columns. A column that is both an order-by column and
referenced by a non-order-by
+ // expression is read again here and counted again, which slightly
over-reports entries scanned.
+ int numColumns = columns.size();
+
+ BitmapDocIdSetOperator docIdSetOperator =
BitmapDocIdSetOperator.ascending(docIds, numRows);
+ try (ProjectionOperator projectionOperator =
+ ProjectionOperatorUtils.getProjectionOperator(dataSourceMap,
docIdSetOperator, _queryContext)) {
+ TransformOperator transformOperator =
+ new TransformOperator(_queryContext, projectionOperator,
nonOrderByExpressions);
+
+ BlockValSet[] blockValSets = new BlockValSet[numNonOrderByExpressions];
+ int rowBaseId = 0;
+ ValueBlock valueBlock;
+ while ((valueBlock = transformOperator.nextBlock()) != null) {
+ for (int i = 0; i < numNonOrderByExpressions; i++) {
+ blockValSets[i] =
valueBlock.getBlockValueSet(nonOrderByExpressions.get(i));
+ }
+ RowBasedBlockValueFetcher blockValueFetcher = new
RowBasedBlockValueFetcher(blockValSets);
+ int numDocsFetched = valueBlock.getNumDocs();
+ for (int i = 0; i < numDocsFetched; i++) {
+ blockValueFetcher.getRow(i, rowsByDocId[rowBaseId + i],
_numOrderByExpressions);
+ }
+ if (_nullHandlingEnabled) {
+ RoaringBitmap[] nullBitmaps = new
RoaringBitmap[numNonOrderByExpressions];
+ for (int i = 0; i < numNonOrderByExpressions; i++) {
+ nullBitmaps[i] = blockValSets[i].getNullBitmap();
+ }
+ for (int i = 0; i < numDocsFetched; i++) {
+ Object[] row = rowsByDocId[rowBaseId + i];
+ for (int colId = 0; colId < numNonOrderByExpressions; colId++) {
+ if (nullBitmaps[colId] != null &&
nullBitmaps[colId].contains(i)) {
+ row[_numOrderByExpressions + colId] = null;
+ }
+ }
+ }
+ }
+ long entriesScanned = (long) numDocsFetched * numColumns;
+ _numEntriesScannedPostFilterLate += entriesScanned;
+ QueryScanCostContext scanCost = getScanCostContext();
+ if (scanCost != null) {
+ scanCost.addEntriesScannedPostFilter(entriesScanned);
+ }
+ rowBaseId += numDocsFetched;
+ }
+ return new SelectionResultsBlock(createDataSchema(transformOperator,
nonOrderByExpressions), rows, _comparator,
+ _queryContext);
+ }
+ }
- // Create the data schema
+ /// Data schema covering both cases. Types for the expressions the scan
projected come from the project
+ /// operator's column contexts; types for any expressions fetched late come
from the transform operator built
+ /// for that fetch. `lateExpressions` is empty when nothing was materialised
late, which leaves the first
+ /// branch covering every expression.
+ private DataSchema createDataSchema(@Nullable TransformOperator
lateTransformOperator,
+ List<ExpressionContext> lateExpressions) {
+ int numExpressions = _expressions.size();
+ int numProjected = numExpressions - lateExpressions.size();
String[] columnNames = new String[numExpressions];
DataSchema.ColumnDataType[] columnDataTypes = new
DataSchema.ColumnDataType[numExpressions];
- for (int i = 0; i < columnNames.length; i++) {
- columnNames[i] = _expressions.get(i).toString();
- }
for (int i = 0; i < numExpressions; i++) {
+ columnNames[i] = _expressions.get(i).toString();
+ ColumnContext columnContext = i < numProjected ? _columnContexts[i]
+ : lateTransformOperator.getResultColumnContext(lateExpressions.get(i
- numProjected));
columnDataTypes[i] =
-
DataSchema.ColumnDataType.fromDataType(_columnContexts[i].getDataType(),
_columnContexts[i].isSingleValue());
+ DataSchema.ColumnDataType.fromDataType(columnContext.getDataType(),
columnContext.isSingleValue());
}
return new DataSchema(columnNames, columnDataTypes);
}
+ protected DataSchema createDataSchema() {
+ return createDataSchema(null, List.of());
+ }
+
/// A private class used to build a partially sorted list by adding
partially sorted data.
///
/// Specifically, this class has been designed to receive successive calls
to [#add(Object[])] follow by a
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/operator/query/SelectionPartiallyOrderedByDescOperation.java
b/pinot-core/src/main/java/org/apache/pinot/core/operator/query/SelectionPartiallyOrderedByDescOperation.java
index 6d348885466..396bd01ab02 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/operator/query/SelectionPartiallyOrderedByDescOperation.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/operator/query/SelectionPartiallyOrderedByDescOperation.java
@@ -44,7 +44,13 @@ public class SelectionPartiallyOrderedByDescOperation
extends LinearSelectionOrd
public SelectionPartiallyOrderedByDescOperation(IndexSegment indexSegment,
QueryContext queryContext,
List<ExpressionContext> expressions, BaseProjectOperator<?>
projectOperator, int numSortedExpressions) {
- super(indexSegment, queryContext, expressions, projectOperator,
numSortedExpressions);
+ this(indexSegment, queryContext, expressions, projectOperator,
numSortedExpressions, false);
+ }
+
+ public SelectionPartiallyOrderedByDescOperation(IndexSegment indexSegment,
QueryContext queryContext,
+ List<ExpressionContext> expressions, BaseProjectOperator<?>
projectOperator, int numSortedExpressions,
+ boolean lateMaterialize) {
+ super(indexSegment, queryContext, expressions, projectOperator,
numSortedExpressions, lateMaterialize);
assert queryContext.getOrderByExpressions() != null;
Preconditions.checkArgument(!queryContext.getOrderByExpressions().stream()
.filter(expr -> expr.getExpression().getType() ==
ExpressionContext.Type.IDENTIFIER)
@@ -62,8 +68,7 @@ public class SelectionPartiallyOrderedByDescOperation extends
LinearSelectionOrd
// the same DataBlockCache, so they may be ephemeral and being overridden
by the next block.
// The only alternative we have right now is to retrieve the last LIMIT
elements from each block
- int numExpressions = _expressions.size();
- BlockValSet[] blockValSets = new BlockValSet[numExpressions];
+ BlockValSet[] blockValSets = new BlockValSet[getNumProjectedExpressions()];
List<Object[]> localBestRows = new ArrayList<>();
ValueBlock valueBlock;
while ((valueBlock = _projectOperator.nextBlock()) != null) {
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/operator/query/SelectionPartiallyOrderedByLinearOperator.java
b/pinot-core/src/main/java/org/apache/pinot/core/operator/query/SelectionPartiallyOrderedByLinearOperator.java
index 6d71289c1c7..33a43a144c8 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/operator/query/SelectionPartiallyOrderedByLinearOperator.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/operator/query/SelectionPartiallyOrderedByLinearOperator.java
@@ -46,7 +46,13 @@ public class SelectionPartiallyOrderedByLinearOperator
extends LinearSelectionOr
public SelectionPartiallyOrderedByLinearOperator(IndexSegment indexSegment,
QueryContext queryContext,
List<ExpressionContext> expressions, BaseProjectOperator<?>
projectOperator, int numSortedExpressions) {
- super(indexSegment, queryContext, expressions, projectOperator,
numSortedExpressions);
+ this(indexSegment, queryContext, expressions, projectOperator,
numSortedExpressions, false);
+ }
+
+ public SelectionPartiallyOrderedByLinearOperator(IndexSegment indexSegment,
QueryContext queryContext,
+ List<ExpressionContext> expressions, BaseProjectOperator<?>
projectOperator, int numSortedExpressions,
+ boolean lateMaterialize) {
+ super(indexSegment, queryContext, expressions, projectOperator,
numSortedExpressions, lateMaterialize);
boolean firstColIsAsc = queryContext.getOrderByExpressions().stream()
.filter(expr -> expr.getExpression().getType() ==
ExpressionContext.Type.IDENTIFIER)
.findFirst()
@@ -61,8 +67,7 @@ public class SelectionPartiallyOrderedByLinearOperator
extends LinearSelectionOr
@Override
protected List<Object[]> fetch(Supplier<ListBuilder> listBuilderSupplier) {
- int numExpressions = _expressions.size();
- BlockValSet[] blockValSets = new BlockValSet[numExpressions];
+ BlockValSet[] blockValSets = new BlockValSet[getNumProjectedExpressions()];
ListBuilder listBuilder = listBuilderSupplier.get();
ValueBlock valueBlock;
while ((valueBlock = _projectOperator.nextBlock()) != null) {
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/plan/SelectionPlanNode.java
b/pinot-core/src/main/java/org/apache/pinot/core/plan/SelectionPlanNode.java
index 029f4811c44..04b11cec49b 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/plan/SelectionPlanNode.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/plan/SelectionPlanNode.java
@@ -88,17 +88,38 @@ public class SelectionPlanNode implements PlanNode {
maxDocsPerCall = Math.min(limit + _queryContext.getOffset(),
DocIdSetPlanNode.MAX_DOC_PER_CALL);
}
- BaseProjectOperator<?> projectOperator = getSortedByProject(expressions,
maxDocsPerCall, orderByExpressions);
+ // Whether the scan can iterate in query order, and so stop once the
list builder is full. Ascending always
+ // can; descending only when reverse iteration is permitted, because
DocIdSetPlanNode only produces an
+ // ascending DocIdSetOperator. getSortedByProject below performs the
actual switch and decides for real; this
+ // is a conservative prediction, and being wrong only costs the
optimization, never correctness.
boolean asc = orderByExpressions.get(0).isAsc();
+ boolean canIterateInQueryOrder =
+ asc ||
QueryOptionsUtils.isReverseOrderAllowed(_queryContext.getQueryOptions());
+
+ // Projecting only the order-by expressions and fetching the rest for
the rows that survive the limit pays off
+ // only when the scan cannot stop at the limit. The linear operator
stops as soon as the list builder is full,
+ // which with a fully sorted order-by prefix is after limit+offset
documents, so there is nothing to save
+ // there. The descending operator has to read the whole match set
because there is no descending cursor.
+ boolean scanBoundedByLimit = canIterateInQueryOrder &&
sortedColumnsPrefixSize == orderByExpressions.size();
+ boolean lateMaterialize = expressions.size() > numOrderByExpressions &&
!scanBoundedByLimit
+ &&
QueryOptionsUtils.isLateMaterializeSortedOrderByEnabled(_queryContext.getQueryOptions());
+
+ // Build the project operator once. Constructing one is not free -- it
resolves a DataSource and a
+ // ColumnContext per selected column, and some implementations start
prefetching those columns -- so a
+ // build-and-discard probe would read exactly the columns this is trying
to avoid reading.
+ List<ExpressionContext> expressionsToProject =
+ lateMaterialize ? expressions.subList(0, numOrderByExpressions) :
expressions;
+ BaseProjectOperator<?> projectOperator =
+ getSortedByProject(expressionsToProject, maxDocsPerCall,
orderByExpressions);
// Remember that we cannot use asc == projectOperator.isAscending()
because empty operators are considered
// both ascending and descending
DocIdOrderedOperator.DocIdOrder queryOrder =
DocIdOrderedOperator.DocIdOrder.fromAsc(asc);
if (projectOperator.isCompatibleWith(queryOrder)) {
return new SelectionPartiallyOrderedByLinearOperator(_indexSegment,
_queryContext, expressions, projectOperator,
- sortedColumnsPrefixSize);
+ sortedColumnsPrefixSize, lateMaterialize);
} else {
return new SelectionPartiallyOrderedByDescOperation(_indexSegment,
_queryContext, expressions, projectOperator,
- sortedColumnsPrefixSize);
+ sortedColumnsPrefixSize, lateMaterialize);
}
}
if (numOrderByExpressions == expressions.size()) {
diff --git
a/pinot-core/src/test/java/org/apache/pinot/queries/InnerSegmentSelectionSingleValueQueriesTest.java
b/pinot-core/src/test/java/org/apache/pinot/queries/InnerSegmentSelectionSingleValueQueriesTest.java
index 333b8db5de2..1a35f440ef9 100644
---
a/pinot-core/src/test/java/org/apache/pinot/queries/InnerSegmentSelectionSingleValueQueriesTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/queries/InnerSegmentSelectionSingleValueQueriesTest.java
@@ -290,15 +290,16 @@ public class InnerSegmentSelectionSingleValueQueriesTest
extends BaseSingleValue
public void testSelectionOrderBySingleSortedColumnDescDidOrder() {
String orderBy = " ORDER BY column5 DESC";
BaseOperator<SelectionResultsBlock> selectionOrderByOperator = getOperator(
- "SET allowReverseOrder = false; " + SELECTION_QUERY + orderBy);
+ "SET allowReverseOrder = false; SET lateMaterializeSortedOrderBy =
true; " + SELECTION_QUERY + orderBy);
assertTrue(selectionOrderByOperator instanceof
SelectionPartiallyOrderedByDescOperation, "Expected: "
+ SelectionPartiallyOrderedByLinearOperator.class + ", found: " +
selectionOrderByOperator.getClass());
SelectionResultsBlock resultsBlock = selectionOrderByOperator.nextBlock();
ExecutionStatistics executionStatistics =
selectionOrderByOperator.getExecutionStatistics();
assertEquals(executionStatistics.getNumDocsScanned(), 30000L);
assertEquals(executionStatistics.getNumEntriesScannedInFilter(), 0L);
- // 30000 * (3 columns)
- assertEquals(executionStatistics.getNumEntriesScannedPostFilter(), 90000L);
+ // The scan reads only the order-by column (30000 * 1); column1 and
column11 are fetched just for the
+ // 10 rows that survive the limit (10 * 2). Before late materialisation
this was 30000 * 3.
+ assertEquals(executionStatistics.getNumEntriesScannedPostFilter(), 30000L
+ 20L);
assertEquals(executionStatistics.getNumTotalDocs(), 30000L);
DataSchema dataSchema = resultsBlock.getDataSchema();
assertEquals(dataSchema.getColumnNames(), new String[]{"column5",
"column1", "column11"});
@@ -311,6 +312,17 @@ public class InnerSegmentSelectionSingleValueQueriesTest
extends BaseSingleValue
assertEquals(lastRow[0], "gFuH");
}
+ /// The optimization is opt-in, so the same query without the option must
still read every selected column
+ /// for the whole match set. Pins the default; delete or invert this when
the default flips.
+ @Test
+ public void testSelectionOrderBySingleSortedColumnDescIsEagerByDefault() {
+ BaseOperator<SelectionResultsBlock> operator =
+ getOperator("SET allowReverseOrder = false; " + SELECTION_QUERY + "
ORDER BY column5 DESC");
+ operator.nextBlock();
+ // 30000 docs * 3 selected columns, as before late materialisation existed.
+
assertEquals(operator.getExecutionStatistics().getNumEntriesScannedPostFilter(),
90000L);
+ }
+
@Test
public void testSelectionOrderBySingleSortedColumnDescReverseOrder() {
String orderBy = " ORDER BY column5 DESC";
@@ -725,4 +737,80 @@ public class InnerSegmentSelectionSingleValueQueriesTest
extends BaseSingleValue
assertEquals(brokerResponse.getOfflineThreadMemAllocatedBytes(), 0);
assertEquals(brokerResponse.getRealtimeThreadMemAllocatedBytes(), 0);
}
+
+ /// Runs `query` with late materialisation on and off and asserts the
answers are identical. column5 is the sorted
+ /// column in this fixture.
+ ///
+ /// @param expectFewerEntries whether late materialisation should read fewer
entries post-filter. It should not when
+ /// the scan already stops at the limit, which is
the case for the linear operator with a
+ /// fully sorted order-by prefix.
+ private void assertSameAnswerWithAndWithoutLateMaterialization(String query,
boolean expectFewerEntries) {
+ BaseOperator<SelectionResultsBlock> lateOperator = getOperator("SET
lateMaterializeSortedOrderBy = true; " + query);
+ SelectionResultsBlock lateBlock = lateOperator.nextBlock();
+ ExecutionStatistics lateStats = lateOperator.getExecutionStatistics();
+
+ BaseOperator<SelectionResultsBlock> eagerOperator =
+ getOperator("SET lateMaterializeSortedOrderBy = false; " + query);
+ SelectionResultsBlock eagerBlock = eagerOperator.nextBlock();
+ ExecutionStatistics eagerStats = eagerOperator.getExecutionStatistics();
+
+ DataSchema lateSchema = lateBlock.getDataSchema();
+ DataSchema eagerSchema = eagerBlock.getDataSchema();
+ assertEquals(lateSchema.getColumnNames(), eagerSchema.getColumnNames());
+ assertEquals(lateSchema.getColumnDataTypes(),
eagerSchema.getColumnDataTypes());
+
+ List<Object[]> lateRows = lateBlock.getRows();
+ List<Object[]> eagerRows = eagerBlock.getRows();
+ assertEquals(lateRows.size(), eagerRows.size());
+ for (int i = 0; i < lateRows.size(); i++) {
+ assertEquals(lateRows.get(i), eagerRows.get(i), "row " + i + " differs
for query: " + query);
+ }
+ assertEquals(lateStats.getNumDocsScanned(),
eagerStats.getNumDocsScanned());
+
+ long late = lateStats.getNumEntriesScannedPostFilter();
+ long eager = eagerStats.getNumEntriesScannedPostFilter();
+ if (expectFewerEntries) {
+ assertTrue(late < eager, "expected fewer entries post-filter, but read "
+ late + " vs " + eager
+ + " for query: " + query);
+ } else {
+ assertEquals(late, eager, "expected the same entries post-filter for
query: " + query);
+ }
+ }
+
+ /// The linear operator stops at the limit, so there is nothing to defer and
the plan keeps projecting everything.
+ @Test
+ public void testLateMaterializationSkippedWhenScanStopsAtLimit() {
+ assertSameAnswerWithAndWithoutLateMaterialization(SELECTION_QUERY + "
ORDER BY column5", false);
+ assertSameAnswerWithAndWithoutLateMaterialization(
+ "SET allowReverseOrder = true; " + SELECTION_QUERY + " ORDER BY
column5 DESC", false);
+ }
+
+ /// Descending without a reverse cursor reads the whole match set, so
deferring the other columns is a real saving.
+ @Test
+ public void testLateMaterializationSortedColumnDesc() {
+ assertSameAnswerWithAndWithoutLateMaterialization(
+ "SET allowReverseOrder = false; " + SELECTION_QUERY + " ORDER BY
column5 DESC", true);
+ }
+
+ @Test
+ public void testLateMaterializationWithOffsetAndLimit() {
+ assertSameAnswerWithAndWithoutLateMaterialization(
+ "SET allowReverseOrder = false; " + SELECTION_QUERY + " ORDER BY
column5 DESC LIMIT 5 OFFSET 3", true);
+ }
+
+ /// An unsorted tail in the order-by removes the limit bound on the scan, so
the linear operator benefits too. The
+ /// unsorted order-by column is still projected during the scan; only the
remaining output columns are deferred.
+ @Test
+ public void testLateMaterializationWithUnsortedOrderBySuffix() {
+ assertSameAnswerWithAndWithoutLateMaterialization(SELECTION_QUERY + "
ORDER BY column5, column6", true);
+ }
+
+ /// The late fetch builds its own TransformOperator, so a non-order-by
expression that is a function must resolve,
+ /// both for its values and for its data type in the schema.
+ @Test
+ public void testLateMaterializationWithTransformProjection() {
+ assertSameAnswerWithAndWithoutLateMaterialization(
+ "SET allowReverseOrder = false; SELECT column5, ADD(column1, 1),
column11 FROM testTable "
+ + "ORDER BY column5 DESC", true);
+ }
}
diff --git
a/pinot-core/src/test/java/org/apache/pinot/queries/NullHandlingEnabledQueriesTest.java
b/pinot-core/src/test/java/org/apache/pinot/queries/NullHandlingEnabledQueriesTest.java
index b9e1cff051e..61faccf6d37 100644
---
a/pinot-core/src/test/java/org/apache/pinot/queries/NullHandlingEnabledQueriesTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/queries/NullHandlingEnabledQueriesTest.java
@@ -2102,4 +2102,38 @@ public class NullHandlingEnabledQueriesTest extends
BaseQueriesTest {
public void destroySegments() {
_indexSegment.destroy();
}
+
+ /// Late materialisation fetches the non-order-by columns in a second pass,
so their null bitmaps come from a
+ /// different block than the order-by column's. An off-by-one there would
silently drop or invent nulls, and the
+ /// answer must not depend on whether the optimization ran.
+ @Test
+ public void testLateMaterializationPreservesNullsInDeferredColumns()
+ throws Exception {
+ initializeRows();
+ insertRowWithTwoColumns(1, null);
+ insertRowWithTwoColumns(2, 20);
+ insertRowWithTwoColumns(3, null);
+ insertRowWithTwoColumns(4, 40);
+ Schema schema = schemaBuilder()
+ .addSingleValueDimension(COLUMN1, DataType.INT)
+ .addSingleValueDimension(COLUMN2, DataType.INT)
+ .build();
+ // column1 is sorted and has no nulls, so the sorted-prefix path is used;
column2 is the deferred column
+ setUpSegments(TABLE_CONFIG_WITH_SORTED_COLUMN, schema);
+
+ String query =
+ String.format("SELECT %s, %s FROM testTable ORDER BY %s DESC LIMIT 8",
COLUMN1, COLUMN2, COLUMN1);
+ List<Object[]> lateRows = getRows("SET lateMaterializeSortedOrderBy =
true; " + query);
+ List<Object[]> eagerRows = getRows("SET lateMaterializeSortedOrderBy =
false; " + query);
+
+ assertEquals(lateRows.size(), eagerRows.size());
+ for (int i = 0; i < lateRows.size(); i++) {
+ assertEquals(lateRows.get(i), eagerRows.get(i), "row " + i + " differs");
+ }
+ // The top rows are column1=4 (column2=40) across the segment copies, then
column1=3, whose column2 is null
+ assertEquals(lateRows.get(0)[0], 4);
+ assertEquals(lateRows.get(0)[1], 40);
+ assertEquals(lateRows.get(lateRows.size() - 1)[0], 3);
+ assertNull(lateRows.get(lateRows.size() - 1)[1], "the deferred column must
keep its null");
+ }
}
diff --git
a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
index 967ab1b8dac..46b2bd84ddc 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
@@ -1058,6 +1058,16 @@ public class CommonConstants {
public static final String ALLOW_REVERSE_ORDER = "allowReverseOrder";
public static final boolean DEFAULT_ALLOW_REVERSE_ORDER = false;
+ /// Selection with an order-by prefix on a sorted column: project only
the order-by expressions during the
+ /// scan and fetch the remaining output expressions for the rows that
survive the limit. Enabling it
+ /// avoids reading every selected column for the whole match set.
+ ///
+ /// Off by default while the optimization bakes, following
`allowReverseOrder` and
+ /// `usePhysicalOptimizer`. It also changes the observable
`numEntriesScannedPostFilter` on the queries it
+ /// applies to, so turning it on is a deliberate choice. Note the win
shrinks as `limit + offset`
+ /// approaches the match-set size, where the second pass reads what
the scan would have read anyway.
+ public static final String LATE_MATERIALIZE_SORTED_ORDER_BY =
"lateMaterializeSortedOrderBy";
+ public static final boolean DEFAULT_LATE_MATERIALIZE_SORTED_ORDER_BY =
false;
public static final String MULTI_STAGE_LEAF_LIMIT =
"multiStageLeafLimit";
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
