Skip to content

Add null-aware star-tree index #19367

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
Jackie-Jiang wants to merge 1 commit into
base: master
Choose a base branch
from
Draft

Add null-aware star-tree index #19367

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
57 changes: 29 additions & 28 deletions pinot-core/src/main/java/org/apache/pinot/core/operator/query/AggregationOperator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,7 @@
import org.apache.pinot.core.query.request.context.QueryContext;
import org.apache.pinot.core.startree.executor.StarTreeAggregationExecutor;
import org.apache.pinot.segment.spi.datasource.DataSource;
import org.apache.pinot.segment.spi.index.startree.AggregationFunctionColumnPair;
import org.apache.pinot.spi.query.QueryScanCostContext;


Expand All @@ -47,9 +48,7 @@ public class AggregationOperator extends BaseOperator<AggregationResultsBlock> {
private static final String EXPLAIN_NAME = "AGGREGATE";

private final QueryContext _queryContext;
private final AggregationFunction[] _aggregationFunctions;
private final BaseProjectOperator<?> _projectOperator;
private final boolean _useStarTree;
private final AggregationInfo _aggregationInfo;
private final int _numTotalDocs;

private int _numDocsScanned = 0;
Expand Down Expand Up @@ -80,9 +79,7 @@ public AggregationOperator(QueryContext queryContext, AggregationInfo aggregatio
public AggregationOperator(QueryContext queryContext, AggregationInfo aggregationInfo, int numTotalDocs,
@Nullable boolean[] nonScanResolvable, @Nullable DataSource[] dataSources) {
_queryContext = queryContext;
_aggregationFunctions = queryContext.getAggregationFunctions();
_projectOperator = aggregationInfo.getProjectOperator();
_useStarTree = aggregationInfo.isUseStarTree();
_aggregationInfo = aggregationInfo;
_numTotalDocs = numTotalDocs;
_nonScanResolvable = nonScanResolvable;
_dataSources = dataSources;
Expand All @@ -91,27 +88,27 @@ public AggregationOperator(QueryContext queryContext, AggregationInfo aggregatio
@Override
protected AggregationResultsBlock getNextBlock() {
// Perform aggregation on all the transform blocks
AggregationExecutor aggregationExecutor;
if (_useStarTree) {
// StarTreeAggregationExecutor doesn't support non-scan results.
aggregationExecutor = new StarTreeAggregationExecutor(_aggregationFunctions);
} else {
aggregationExecutor = new DefaultAggregationExecutor(_aggregationFunctions, resolveNonScanResults());
}
AggregationFunction[] aggregationFunctions = _aggregationInfo.getFunctions();
AggregationFunctionColumnPair[] starTreeFunctionColumnPairs = _aggregationInfo.getStarTreeFunctionColumnPairs();
AggregationExecutor aggregationExecutor = starTreeFunctionColumnPairs != null
// StarTreeAggregationExecutor doesn't support non-scan results.
? new StarTreeAggregationExecutor(aggregationFunctions, starTreeFunctionColumnPairs)
: new DefaultAggregationExecutor(aggregationFunctions, resolveNonScanResults());
BaseProjectOperator<?> projectOperator = _aggregationInfo.getProjectOperator();
ValueBlock valueBlock;
while ((valueBlock = _projectOperator.nextBlock()) != null) {
while ((valueBlock = projectOperator.nextBlock()) != null) {
_numDocsScanned += valueBlock.getNumDocs();
QueryScanCostContext scanCost = getScanCostContext();
if (scanCost != null) {
scanCost.addDocsScanned(valueBlock.getNumDocs());
scanCost.addEntriesScannedPostFilter(
(long) valueBlock.getNumDocs() * _projectOperator.getNumColumnsProjected());
(long) valueBlock.getNumDocs() * projectOperator.getNumColumnsProjected());
}
aggregationExecutor.aggregate(valueBlock);
}

// Build intermediate result block based on aggregation result from the executor
return new AggregationResultsBlock(_aggregationFunctions, aggregationExecutor.getResult(), _queryContext);
return new AggregationResultsBlock(aggregationFunctions, aggregationExecutor.getResult(), _queryContext);
}

/// Returns {@code null} when no function is resolvable without scanning, in which case all functions are computed by
Expand All @@ -124,10 +121,11 @@ private Object[] resolveNonScanResults() {
}

Objects.requireNonNull(_dataSources);
Object[] nonScanResults = new Object[_aggregationFunctions.length];
for (int i = 0; i < _aggregationFunctions.length; i++) {
AggregationFunction[] aggregationFunctions = _aggregationInfo.getFunctions();
Object[] nonScanResults = new Object[aggregationFunctions.length];
for (int i = 0; i < aggregationFunctions.length; i++) {
if (_nonScanResolvable[i]) {
nonScanResults[i] = AggregationFunctionUtils.getAggregationResult(_aggregationFunctions[i],
nonScanResults[i] = AggregationFunctionUtils.getAggregationResult(aggregationFunctions[i],
_dataSources[i], _numTotalDocs, EXPLAIN_NAME);
}
}
Expand All @@ -136,24 +134,26 @@ private Object[] resolveNonScanResults() {

@Override
public List<BaseProjectOperator<?>> getChildOperators() {
return List.of(_projectOperator);
return List.of(_aggregationInfo.getProjectOperator());
}

@Override
public ExecutionStatistics getExecutionStatistics() {
long numEntriesScannedInFilter = _projectOperator.getExecutionStatistics().getNumEntriesScannedInFilter();
long numEntriesScannedPostFilter = (long) _numDocsScanned * _projectOperator.getNumColumnsProjected();
BaseProjectOperator<?> projectOperator = _aggregationInfo.getProjectOperator();
long numEntriesScannedInFilter = projectOperator.getExecutionStatistics().getNumEntriesScannedInFilter();
long numEntriesScannedPostFilter = (long) _numDocsScanned * projectOperator.getNumColumnsProjected();
return new ExecutionStatistics(_numDocsScanned, numEntriesScannedInFilter, numEntriesScannedPostFilter,
_numTotalDocs);
}

@Override
public String toExplainString() {
StringBuilder stringBuilder = new StringBuilder(EXPLAIN_NAME).append("(aggregations:");
if (_aggregationFunctions.length > 0) {
stringBuilder.append(_aggregationFunctions[0].toExplainString());
for (int i = 1; i < _aggregationFunctions.length; i++) {
stringBuilder.append(", ").append(_aggregationFunctions[i].toExplainString());
AggregationFunction[] aggregationFunctions = _aggregationInfo.getFunctions();
if (aggregationFunctions.length > 0) {
stringBuilder.append(aggregationFunctions[0].toExplainString());
for (int i = 1; i < aggregationFunctions.length; i++) {
stringBuilder.append(", ").append(aggregationFunctions[i].toExplainString());
}
}

Expand All @@ -168,10 +168,11 @@ protected String getExplainName() {
@Override
protected void explainAttributes(ExplainAttributeBuilder attributeBuilder) {
super.explainAttributes(attributeBuilder);
if (_aggregationFunctions.length == 0) {
AggregationFunction[] aggregationFunctions = _aggregationInfo.getFunctions();
if (aggregationFunctions.length == 0) {
return;
}
List<String> aggregations = Arrays.stream(_aggregationFunctions)
List<String> aggregations = Arrays.stream(aggregationFunctions)
.map(AggregationFunction::toExplainString)
.collect(Collectors.toList());
attributeBuilder.putStringList("aggregations", aggregations);
Expand Down
16 changes: 7 additions & 9 deletions ...-core/src/main/java/org/apache/pinot/core/operator/query/FilteredAggregationOperator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@
import org.apache.pinot.core.query.aggregation.function.AggregationFunctionUtils.AggregationInfo;
import org.apache.pinot.core.query.request.context.QueryContext;
import org.apache.pinot.core.startree.executor.StarTreeAggregationExecutor;
import org.apache.pinot.segment.spi.index.startree.AggregationFunctionColumnPair;
import org.apache.pinot.spi.query.QueryScanCostContext;


Expand Down Expand Up @@ -74,14 +75,12 @@ protected AggregationResultsBlock getNextBlock() {

for (AggregationInfo aggregationInfo : _aggregationInfos) {
AggregationFunction[] aggregationFunctions = aggregationInfo.getFunctions();
BaseProjectOperator<?> projectOperator = aggregationInfo.getProjectOperator();
AggregationExecutor aggregationExecutor;
if (aggregationInfo.isUseStarTree()) {
aggregationExecutor = new StarTreeAggregationExecutor(aggregationFunctions);
} else {
aggregationExecutor = new DefaultAggregationExecutor(aggregationFunctions);
}
AggregationFunctionColumnPair[] starTreeFunctionColumnPairs = aggregationInfo.getStarTreeFunctionColumnPairs();
AggregationExecutor aggregationExecutor = starTreeFunctionColumnPairs != null
? new StarTreeAggregationExecutor(aggregationFunctions, starTreeFunctionColumnPairs)
: new DefaultAggregationExecutor(aggregationFunctions);

BaseProjectOperator<?> projectOperator = aggregationInfo.getProjectOperator();
ValueBlock valueBlock;
int numDocsScanned = 0;
while ((valueBlock = projectOperator.nextBlock()) != null) {
Expand All @@ -97,8 +96,7 @@ protected AggregationResultsBlock getNextBlock() {
QueryScanCostContext scanCost = getScanCostContext();
if (scanCost != null) {
scanCost.addDocsScanned(numDocsScanned);
scanCost.addEntriesScannedPostFilter(
(long) numDocsScanned * projectOperator.getNumColumnsProjected());
scanCost.addEntriesScannedPostFilter((long) numDocsScanned * projectOperator.getNumColumnsProjected());
}
_numEntriesScannedInFilter += projectOperator.getExecutionStatistics().getNumEntriesScannedInFilter();
_numEntriesScannedPostFilter += (long) numDocsScanned * projectOperator.getNumColumnsProjected();
Expand Down
32 changes: 14 additions & 18 deletions pinot-core/src/main/java/org/apache/pinot/core/operator/query/FilteredGroupByOperator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,7 @@
import org.apache.pinot.core.query.request.context.QueryContext;
import org.apache.pinot.core.startree.executor.StarTreeGroupByExecutor;
import org.apache.pinot.core.util.GroupByUtils;
import org.apache.pinot.segment.spi.index.startree.AggregationFunctionColumnPair;
import org.apache.pinot.spi.query.QueryScanCostContext;
import org.apache.pinot.spi.trace.Tracing;
import org.slf4j.Logger;
Expand Down Expand Up @@ -85,7 +86,7 @@ public FilteredGroupByOperator(QueryContext queryContext, List<AggregationInfo>
// NOTE: The indexedTable expects that the data schema will have group by columns before aggregation columns
int numGroupByExpressions = _groupByExpressions.length;
int numAggregationFunctions = _aggregationFunctions.length;
/// Grouping-set queries append a synthetic $groupingId key column after the union group-by columns.
// Grouping-set queries append a synthetic $groupingId key column after the union group-by columns.
int numExtraKeyColumns = queryContext.getNumExtraGroupByKeyColumns();
int numKeyColumns = numGroupByExpressions + numExtraKeyColumns;
int numColumns = numKeyColumns + numAggregationFunctions;
Expand All @@ -101,7 +102,7 @@ public FilteredGroupByOperator(QueryContext queryContext, List<AggregationInfo>
projectOperator.getResultColumnContext(groupByExpression).getDataType());
}

/// Synthetic grouping-id discriminator column for GROUP BY GROUPING SETS / ROLLUP / CUBE
// Synthetic grouping-id discriminator column for GROUP BY GROUPING SETS / ROLLUP / CUBE
if (numExtraKeyColumns > 0) {
columnNames[numGroupByExpressions] = GroupingSets.GROUPING_ID_COLUMN;
columnDataTypes[numGroupByExpressions] = DataSchema.ColumnDataType.INT;
Expand Down Expand Up @@ -141,17 +142,13 @@ protected GroupByResultsBlock getNextBlock() {
BaseProjectOperator<?> projectOperator = aggregationInfo.getProjectOperator();

// Perform aggregation group-by on all the blocks
DefaultGroupByExecutor groupByExecutor;
AggregationFunctionColumnPair[] starTreeFunctionColumnPairs = aggregationInfo.getStarTreeFunctionColumnPairs();
DefaultGroupByExecutor groupByExecutor = starTreeFunctionColumnPairs != null
? new StarTreeGroupByExecutor(_queryContext, aggregationFunctions, _groupByExpressions, projectOperator,
starTreeFunctionColumnPairs, groupKeyGenerator)
: new DefaultGroupByExecutor(_queryContext, aggregationFunctions, _groupByExpressions, projectOperator,
groupKeyGenerator);

if (aggregationInfo.isUseStarTree()) {
groupByExecutor =
new StarTreeGroupByExecutor(_queryContext, aggregationFunctions, _groupByExpressions, projectOperator,
groupKeyGenerator);
} else {
groupByExecutor =
new DefaultGroupByExecutor(_queryContext, aggregationFunctions, _groupByExpressions, projectOperator,
groupKeyGenerator);
}
// The group key generator should be shared across all AggregationFunctions so that agg results can be
// aligned. Given that filtered aggregations are stored as an iterable of iterables so that all filtered aggs
// with the same filter can share transform blocks, rather than a singular flat iterable in the case where
Expand All @@ -172,8 +169,7 @@ protected GroupByResultsBlock getNextBlock() {
QueryScanCostContext scanCost = getScanCostContext();
if (scanCost != null) {
scanCost.addDocsScanned(numDocsScanned);
scanCost.addEntriesScannedPostFilter(
(long) numDocsScanned * projectOperator.getNumColumnsProjected());
scanCost.addEntriesScannedPostFilter((long) numDocsScanned * projectOperator.getNumColumnsProjected());
}
_numEntriesScannedInFilter += projectOperator.getExecutionStatistics().getNumEntriesScannedInFilter();
_numEntriesScannedPostFilter += (long) numDocsScanned * projectOperator.getNumColumnsProjected();
Expand Down Expand Up @@ -212,11 +208,11 @@ protected GroupByResultsBlock getNextBlock() {
boolean unsafeTrim = _queryContext.isUnsafeTrim();

GroupByResultsBlock resultsBlock;
/// Grouping-set queries use a per-set bucketed segment trim (keyed on the $groupingId discriminator) so
/// that a global top-K cannot starve low-magnitude sets such as the grand total. The broker still applies
/// the final ORDER BY + LIMIT across all sets.
// Grouping-set queries use a per-set bucketed segment trim (keyed on the $groupingId discriminator) so
// that a global top-K cannot starve low-magnitude sets such as the grand total. The broker still applies
// the final ORDER BY + LIMIT across all sets.
if (_queryContext.isGroupingSets()) {
/// The $groupingId discriminator is the key column immediately after the union group-by columns.
// The $groupingId discriminator is the key column immediately after the union group-by columns.
return GroupByUtils.buildGroupingSetsResultsBlock(_queryContext, _dataSchema, groupKeyGenerator,
groupByResultHolders, groupKeyGenerator.getNumKeys(), _groupByExpressions.length, numGroupsLimitReached,
numGroupsWarningLimitReached);
Expand Down
Loading
Loading