Jackie-Jiang commented on code in PR #19879:
URL: https://github.com/apache/pinot/pull/19879#discussion_r4298071865
##########
pinot-core/src/main/java/org/apache/pinot/core/operator/query/LinearSelectionOrderByOperator.java:
##########
@@ -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;
Review Comment:
Please remove the lateMaterialize boolean plumbing as well as the
lateMaterializeSortedOrderBy query option: the planner switch, constructor
arguments, and _lateMaterialize field/branches. Use the same two cases as
SelectionOrderByOperator: when all output expressions are order-by expressions
there is nothing to defer; otherwise, fetch the remaining expressions after
selection. This keeps the sorted and unsorted paths consistent and avoids
maintaining a separate eager/late mode.
##########
pinot-core/src/main/java/org/apache/pinot/core/operator/query/LinearSelectionOrderByOperator.java:
##########
@@ -234,25 +309,122 @@ private void concatList(StringBuilder sb, List<?> list) {
@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,
Review Comment:
This helper can drop the lateExpressions parameter.
getNumProjectedExpressions() already gives the split point, and
_expressions.get(i) is the expression to resolve from lateTransformOperator for
the remaining slots. That removes a redundant list argument and the i -
numProjected mapping. The no-argument wrapper can delegate to
createDataSchema(null).
##########
pinot-core/src/main/java/org/apache/pinot/core/operator/query/LinearSelectionOrderByOperator.java:
##########
@@ -234,25 +309,122 @@ private void concatList(StringBuilder sb, List<?> list) {
@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
Review Comment:
Please shorten this to the invariant that surviving doc IDs must be unique
for bitmap/row alignment. The statement that the build never runs with -ea is
incorrect: Surefire enables assertions by default, independently of an explicit
-ea in argLine. If keeping checkState, cache docIds.getCardinality() once; the
current call computes it twice, including on success when evaluating the
message arguments.
##########
pinot-core/src/main/java/org/apache/pinot/core/operator/query/LinearSelectionOrderByOperator.java:
##########
@@ -234,25 +309,122 @@ private void concatList(StringBuilder sb, List<?> list) {
@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.
Review Comment:
Counting an actual reread is correct scan-work accounting, so this does not
over-report entries scanned. The comment can simply say that the count includes
columns reread from the order-by pass.
##########
pinot-core/src/main/java/org/apache/pinot/core/operator/query/LinearSelectionOrderByOperator.java:
##########
@@ -234,25 +309,122 @@ private void concatList(StringBuilder sb, List<?> list) {
@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++) {
Review Comment:
We can reuse applyNulls here by adding a destination-column offset: 0 during
selection and _numOrderByExpressions during the late fetch. That puts null
restoration in one helper; the null-bitmap position remains the block-local row
index i in this 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: [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]
