You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by ja...@apache.org on 2026/09/22 00:36:29 UTC
(pinot) branch master updated: Add null-aware star-tree index (#19367)
This is an automated email from the ASF dual-hosted git repository.
Jackie-Jiang 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 554f9cda26b Add null-aware star-tree index (#19367)
554f9cda26b is described below
commit 554f9cda26b5be57a261099b050e07fb09d96520
Author: Xiaotian (Jackie) Jiang <17...@users.noreply.github.com>
AuthorDate: Mon Sep 21 17:36:22 2026 -0700
Add null-aware star-tree index (#19367)
---
.../core/operator/query/AggregationOperator.java | 11 +-
.../query/FilteredAggregationOperator.java | 6 +-
.../operator/query/FilteredGroupByOperator.java | 6 +-
.../pinot/core/operator/query/GroupByOperator.java | 13 +-
.../function/AggregationFunctionUtils.java | 72 ++--
.../function/AvgAggregationFunction.java | 32 +-
.../apache/pinot/core/startree/StarTreeUtils.java | 204 +++++++----
.../executor/StarTreeAggregationExecutor.java | 17 +-
.../startree/executor/StarTreeGroupByExecutor.java | 26 +-
.../pinot/core/startree/StarTreeUtilsTest.java | 15 +-
.../queries/NullAwareStarTreeQueriesTest.java | 387 +++++++++++++++++++++
.../local/aggregator/CountValueAggregator.java | 11 +
.../DistinctCountBitmapValueAggregator.java | 4 +-
.../DistinctCountThetaSketchValueAggregator.java | 4 +-
.../segment/local/aggregator/ValueAggregator.java | 22 ++
.../local/segment/store/StarTreeIndexReader.java | 21 ++
.../local/startree/StarTreeBuilderUtils.java | 4 +
.../startree/v2/builder/BaseSingleTreeBuilder.java | 236 ++++++++++---
.../v2/builder/OffHeapSingleTreeBuilder.java | 73 +++-
.../startree/v2/builder/StarTreeIndexCombiner.java | 16 +
.../v2/builder/StarTreeIndexSeparator.java | 5 +
.../v2/builder/StarTreeV2BuilderConfig.java | 39 ++-
.../startree/v2/store/StarTreeDataSource.java | 9 +-
.../startree/v2/store/StarTreeIndexMapUtils.java | 14 +-
.../startree/v2/store/StarTreeLoaderUtils.java | 24 +-
.../segment/local/utils/TableConfigUtils.java | 7 +-
.../v2/builder/NullAwareStarTreeBuilderTest.java | 230 ++++++++++++
.../v2/builder/StarTreeBuilderUtilsTest.java | 2 +-
.../startree/AggregationFunctionColumnPair.java | 49 ++-
.../spi/index/startree/StarTreeV2Constants.java | 2 +
.../spi/index/startree/StarTreeV2Metadata.java | 21 +-
.../spi/index/startree/StarTreeV2MetadataTest.java | 3 +-
.../api/resources/SegmentMetadataFetcher.java | 4 +
.../spi/config/table/StarTreeIndexConfig.java | 25 +-
34 files changed, 1367 insertions(+), 247 deletions(-)
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/operator/query/AggregationOperator.java b/pinot-core/src/main/java/org/apache/pinot/core/operator/query/AggregationOperator.java
index eae9166af53..b98e2abf4c2 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/operator/query/AggregationOperator.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/operator/query/AggregationOperator.java
@@ -38,6 +38,7 @@ import org.apache.pinot.core.query.aggregation.function.AggregationFunctionUtils
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;
@@ -49,7 +50,9 @@ public class AggregationOperator extends BaseOperator<AggregationResultsBlock> {
private final QueryContext _queryContext;
private final AggregationFunction[] _aggregationFunctions;
private final BaseProjectOperator<?> _projectOperator;
- private final boolean _useStarTree;
+ /// Non-null when the aggregation reads a star-tree, holding the pairs resolved against the one it was routed to.
+ @Nullable
+ private final AggregationFunctionColumnPair[] _starTreeFunctionColumnPairs;
private final int _numTotalDocs;
private int _numDocsScanned = 0;
@@ -82,7 +85,7 @@ public class AggregationOperator extends BaseOperator<AggregationResultsBlock> {
_queryContext = queryContext;
_aggregationFunctions = queryContext.getAggregationFunctions();
_projectOperator = aggregationInfo.getProjectOperator();
- _useStarTree = aggregationInfo.isUseStarTree();
+ _starTreeFunctionColumnPairs = aggregationInfo.getStarTreeFunctionColumnPairs();
_numTotalDocs = numTotalDocs;
_nonScanResolvable = nonScanResolvable;
_dataSources = dataSources;
@@ -92,9 +95,9 @@ public class AggregationOperator extends BaseOperator<AggregationResultsBlock> {
protected AggregationResultsBlock getNextBlock() {
// Perform aggregation on all the transform blocks
AggregationExecutor aggregationExecutor;
- if (_useStarTree) {
+ if (_starTreeFunctionColumnPairs != null) {
// StarTreeAggregationExecutor doesn't support non-scan results.
- aggregationExecutor = new StarTreeAggregationExecutor(_aggregationFunctions);
+ aggregationExecutor = new StarTreeAggregationExecutor(_aggregationFunctions, _starTreeFunctionColumnPairs);
} else {
aggregationExecutor = new DefaultAggregationExecutor(_aggregationFunctions, resolveNonScanResults());
}
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/operator/query/FilteredAggregationOperator.java b/pinot-core/src/main/java/org/apache/pinot/core/operator/query/FilteredAggregationOperator.java
index ba433e14474..5b9daef759d 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/operator/query/FilteredAggregationOperator.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/operator/query/FilteredAggregationOperator.java
@@ -35,6 +35,7 @@ import org.apache.pinot.core.query.aggregation.function.AggregationFunction;
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;
@@ -75,9 +76,10 @@ public class FilteredAggregationOperator extends BaseOperator<AggregationResults
for (AggregationInfo aggregationInfo : _aggregationInfos) {
AggregationFunction[] aggregationFunctions = aggregationInfo.getFunctions();
BaseProjectOperator<?> projectOperator = aggregationInfo.getProjectOperator();
+ AggregationFunctionColumnPair[] starTreeFunctionColumnPairs = aggregationInfo.getStarTreeFunctionColumnPairs();
AggregationExecutor aggregationExecutor;
- if (aggregationInfo.isUseStarTree()) {
- aggregationExecutor = new StarTreeAggregationExecutor(aggregationFunctions);
+ if (starTreeFunctionColumnPairs != null) {
+ aggregationExecutor = new StarTreeAggregationExecutor(aggregationFunctions, starTreeFunctionColumnPairs);
} else {
aggregationExecutor = new DefaultAggregationExecutor(aggregationFunctions);
}
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/operator/query/FilteredGroupByOperator.java b/pinot-core/src/main/java/org/apache/pinot/core/operator/query/FilteredGroupByOperator.java
index f910cd0bda6..166948c106c 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/operator/query/FilteredGroupByOperator.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/operator/query/FilteredGroupByOperator.java
@@ -49,6 +49,7 @@ import org.apache.pinot.core.query.aggregation.groupby.GroupKeyGenerator;
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;
@@ -141,12 +142,13 @@ public class FilteredGroupByOperator extends BaseOperator<GroupByResultsBlock> {
BaseProjectOperator<?> projectOperator = aggregationInfo.getProjectOperator();
// Perform aggregation group-by on all the blocks
+ AggregationFunctionColumnPair[] starTreeFunctionColumnPairs = aggregationInfo.getStarTreeFunctionColumnPairs();
DefaultGroupByExecutor groupByExecutor;
- if (aggregationInfo.isUseStarTree()) {
+ if (starTreeFunctionColumnPairs != null) {
groupByExecutor =
new StarTreeGroupByExecutor(_queryContext, aggregationFunctions, _groupByExpressions, projectOperator,
- groupKeyGenerator);
+ starTreeFunctionColumnPairs, groupKeyGenerator);
} else {
groupByExecutor =
new DefaultGroupByExecutor(_queryContext, aggregationFunctions, _groupByExpressions, projectOperator,
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/operator/query/GroupByOperator.java b/pinot-core/src/main/java/org/apache/pinot/core/operator/query/GroupByOperator.java
index 50afa03d62a..e138a8c7973 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/operator/query/GroupByOperator.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/operator/query/GroupByOperator.java
@@ -22,6 +22,7 @@ import com.google.common.base.CaseFormat;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;
+import javax.annotation.Nullable;
import org.apache.pinot.common.metrics.ServerMeter;
import org.apache.pinot.common.metrics.ServerMetrics;
import org.apache.pinot.common.request.context.ExpressionContext;
@@ -43,6 +44,7 @@ import org.apache.pinot.core.query.aggregation.groupby.GroupByExecutor;
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;
@@ -59,7 +61,9 @@ public class GroupByOperator extends BaseOperator<GroupByResultsBlock> {
private final AggregationFunction[] _aggregationFunctions;
private final ExpressionContext[] _groupByExpressions;
private final BaseProjectOperator<?> _projectOperator;
- private final boolean _useStarTree;
+ /// Non-null when the aggregation reads a star-tree, holding the pairs resolved against the one it was routed to.
+ @Nullable
+ private final AggregationFunctionColumnPair[] _starTreeFunctionColumnPairs;
private final long _numTotalDocs;
private final DataSchema _dataSchema;
@@ -71,7 +75,7 @@ public class GroupByOperator extends BaseOperator<GroupByResultsBlock> {
_aggregationFunctions = queryContext.getAggregationFunctions();
_groupByExpressions = queryContext.getGroupByExpressions().toArray(new ExpressionContext[0]);
_projectOperator = aggregationInfo.getProjectOperator();
- _useStarTree = aggregationInfo.isUseStarTree();
+ _starTreeFunctionColumnPairs = aggregationInfo.getStarTreeFunctionColumnPairs();
_numTotalDocs = numTotalDocs;
// NOTE: The indexedTable expects that the data schema will have group by columns before aggregation columns
@@ -120,8 +124,9 @@ public class GroupByOperator extends BaseOperator<GroupByResultsBlock> {
// Perform aggregation group-by on all the blocks
GroupByExecutor groupByExecutor;
// TODO: pass trimGroupSize to executor, who creates the result holder
- if (_useStarTree) {
- groupByExecutor = new StarTreeGroupByExecutor(_queryContext, _groupByExpressions, _projectOperator);
+ if (_starTreeFunctionColumnPairs != null) {
+ groupByExecutor = new StarTreeGroupByExecutor(_queryContext, _groupByExpressions, _projectOperator,
+ _starTreeFunctionColumnPairs);
} else {
groupByExecutor = new DefaultGroupByExecutor(_queryContext, _groupByExpressions, _projectOperator);
}
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionUtils.java b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionUtils.java
index 33b7abf005e..4f96018d661 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionUtils.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionUtils.java
@@ -85,16 +85,27 @@ public class AggregationFunctionUtils {
private AggregationFunctionUtils() {
}
- /// (For Star-Tree) Creates an [AggregationFunctionColumnPair] in stored type from the
- /// [AggregationFunction]. Returns `null` if the [AggregationFunction] cannot be represented as an
- /// [AggregationFunctionColumnPair] (e.g. has multiple arguments, argument is not column etc.).
+ /// (For Star-Tree) Resolves the [AggregationFunctionColumnPair] stored by a regular star-tree, or by a null-aware
+ /// one when `nullHandlingEnabled` is `true`. Returns `null` if the [AggregationFunction] cannot be represented as a
+ /// pair (e.g. has multiple arguments, argument is not a column etc.).
+ ///
+ /// The two only differ for `COUNT`: a regular star-tree stores a single `count__*` of every row, while a null-aware
+ /// star-tree stores a `count__column` holding the count of that column's non-null values.
@Nullable
- public static AggregationFunctionColumnPair getStoredFunctionColumnPair(AggregationFunction aggregationFunction) {
+ public static AggregationFunctionColumnPair getStoredFunctionColumnPair(AggregationFunction aggregationFunction,
+ boolean nullHandlingEnabled) {
AggregationFunctionType functionType = aggregationFunction.getType();
+ List<ExpressionContext> inputExpressions = aggregationFunction.getInputExpressions();
if (functionType == AggregationFunctionType.COUNT) {
- return AggregationFunctionColumnPair.COUNT_STAR;
+ // CountAggregationFunction only reports an input expression when null handling is enabled and the argument is a
+ // non-star identifier or function, which is exactly when nulls have to be excluded from the count. Everything
+ // else counts every row and maps to COUNT(*).
+ if (!nullHandlingEnabled || inputExpressions.size() != 1
+ || inputExpressions.get(0).getType() != ExpressionContext.Type.IDENTIFIER) {
+ return AggregationFunctionColumnPair.COUNT_STAR;
+ }
+ return AggregationFunctionColumnPair.countColumn(inputExpressions.get(0).getIdentifier());
}
- List<ExpressionContext> inputExpressions = aggregationFunction.getInputExpressions();
if (inputExpressions.size() == 1) {
ExpressionContext inputExpression = inputExpressions.get(0);
if (inputExpression.getType() == ExpressionContext.Type.IDENTIFIER) {
@@ -184,7 +195,13 @@ public class AggregationFunctionUtils {
/// marker, so they get the unwrapped BlockValSet.
public static Map<ExpressionContext, BlockValSet> getBlockValSetMap(
AggregationFunctionColumnPair aggregationFunctionColumnPair, ValueBlock valueBlock) {
- ExpressionContext expression = ExpressionContext.forIdentifier(aggregationFunctionColumnPair.getColumn());
+ // A COUNT column holds pre-aggregated counts that have to be summed rather than counted, which
+ // CountAggregationFunction recognizes from the STAR identifier. This also applies to the per-column COUNT of a
+ // null-aware star-tree, so the key is STAR there as well.
+ String column = aggregationFunctionColumnPair.getFunctionType() == AggregationFunctionType.COUNT
+ ? AggregationFunctionColumnPair.STAR
+ : aggregationFunctionColumnPair.getColumn();
+ ExpressionContext expression = ExpressionContext.forIdentifier(column);
BlockValSet blockValSet = valueBlock.getBlockValueSet(aggregationFunctionColumnPair.toColumnName());
if (aggregationFunctionColumnPair.getFunctionType() == AggregationFunctionType.ARRAYAGG) {
blockValSet = new StarTreePreAggregatedBlockValSet(blockValSet);
@@ -352,13 +369,21 @@ public class AggregationFunctionUtils {
public static class AggregationInfo {
private final AggregationFunction[] _functions;
private final BaseProjectOperator<?> _projectOperator;
- private final boolean _useStarTree;
+ @Nullable
+ private final AggregationFunctionColumnPair[] _starTreeFunctionColumnPairs;
+
+ /// Creates the info for aggregations that do not read a star-tree.
+ public AggregationInfo(AggregationFunction[] functions, BaseProjectOperator<?> projectOperator) {
+ this(functions, projectOperator, null);
+ }
+ /// Creates the info for aggregations that read the star-tree the given pairs were resolved against, or for
+ /// aggregations that do not read one at all when `starTreeFunctionColumnPairs` is `null`.
public AggregationInfo(AggregationFunction[] functions, BaseProjectOperator<?> projectOperator,
- boolean useStarTree) {
+ @Nullable AggregationFunctionColumnPair[] starTreeFunctionColumnPairs) {
_functions = functions;
_projectOperator = projectOperator;
- _useStarTree = useStarTree;
+ _starTreeFunctionColumnPairs = starTreeFunctionColumnPairs;
}
public AggregationFunction[] getFunctions() {
@@ -369,8 +394,13 @@ public class AggregationFunctionUtils {
return _projectOperator;
}
- public boolean isUseStarTree() {
- return _useStarTree;
+ /// Returns the function-column pairs projected from the star-tree, or `null` when star-tree is not used.
+ ///
+ /// These are resolved against the specific star-tree the query was routed to, so the aggregation executors must
+ /// use them rather than re-deriving the pairs, which cannot tell a regular star-tree from a null-aware one.
+ @Nullable
+ public AggregationFunctionColumnPair[] getStarTreeFunctionColumnPairs() {
+ return _starTreeFunctionColumnPairs;
}
}
@@ -397,15 +427,11 @@ public class AggregationFunctionUtils {
if (queryContext.isGroupingSets()) {
return null;
}
- if (!filterOperator.isResultEmpty()) {
- BaseProjectOperator<?> projectOperator =
- StarTreeUtils.createStarTreeBasedProjectOperator(segmentContext.getIndexSegment(), queryContext,
- aggregationFunctions, filter, predicateEvaluators);
- if (projectOperator != null) {
- return new AggregationInfo(aggregationFunctions, projectOperator, true);
- }
+ if (filterOperator.isResultEmpty()) {
+ return null;
}
- return null;
+ return StarTreeUtils.createStarTreeBasedAggregationInfo(segmentContext.getIndexSegment(), queryContext,
+ aggregationFunctions, filter, predicateEvaluators);
}
/// Builds [AggregationInfo] for aggregations without using star-tree index.
@@ -417,7 +443,7 @@ public class AggregationFunctionUtils {
BaseProjectOperator<?> projectOperator =
new ProjectPlanNode(segmentContext, queryContext, expressionsToTransform, DocIdSetPlanNode.MAX_DOC_PER_CALL,
filterOperator).run();
- return new AggregationInfo(aggregationFunctions, projectOperator, false);
+ return new AggregationInfo(aggregationFunctions, projectOperator);
}
/// Builds {@link AggregationInfo} for aggregations without using star-tree index, projecting only the columns
@@ -432,7 +458,7 @@ public class AggregationFunctionUtils {
BaseProjectOperator<?> projectOperator =
new ProjectPlanNode(segmentContext, queryContext, expressionsToTransform, DocIdSetPlanNode.MAX_DOC_PER_CALL,
filterOperator).run();
- return new AggregationInfo(allFunctions, projectOperator, false);
+ return new AggregationInfo(allFunctions, projectOperator);
}
@@ -453,7 +479,7 @@ public class AggregationFunctionUtils {
BaseProjectOperator<?> projectOperator =
new ProjectPlanNode(segmentContext, queryContext, expressions, DocIdSetPlanNode.MAX_DOC_PER_CALL,
mainFilterOperator).run();
- return List.of(new AggregationInfo(aggregationFunctions, projectOperator, false));
+ return List.of(new AggregationInfo(aggregationFunctions, projectOperator));
}
// For each aggregation function, check if the aggregation function is a filtered aggregate. If so, populate the
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AvgAggregationFunction.java b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AvgAggregationFunction.java
index d7393fdf173..481e03908ff 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AvgAggregationFunction.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AvgAggregationFunction.java
@@ -82,10 +82,12 @@ public class AvgAggregationFunction extends BaseSingleInputAggregationFunction<A
// Serialized AvgPair
byte[][] bytesValues = blockValSet.getBytesValuesSV();
AvgPair avgPair = new AvgPair();
- for (int i = 0; i < length; i++) {
- AvgPair value = ObjectSerDeUtils.AVG_PAIR_SER_DE.deserialize(bytesValues[i]);
- avgPair.apply(value);
- }
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ AvgPair value = ObjectSerDeUtils.AVG_PAIR_SER_DE.deserialize(bytesValues[i]);
+ avgPair.apply(value);
+ }
+ });
// Only set the aggregation result when there is at least one non-null input value
if (avgPair.getCount() != 0) {
updateAggregationResult(aggregationResultHolder, avgPair.getSum(), avgPair.getCount());
@@ -150,10 +152,12 @@ public class AvgAggregationFunction extends BaseSingleInputAggregationFunction<A
GroupByResultHolder groupByResultHolder) {
// Serialized AvgPair
byte[][] bytesValues = blockValSet.getBytesValuesSV();
- for (int i = 0; i < length; i++) {
- AvgPair avgPair = ObjectSerDeUtils.AVG_PAIR_SER_DE.deserialize(bytesValues[i]);
- updateGroupByResult(groupKeyArray[i], groupByResultHolder, avgPair.getSum(), avgPair.getCount());
- }
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ AvgPair avgPair = ObjectSerDeUtils.AVG_PAIR_SER_DE.deserialize(bytesValues[i]);
+ updateGroupByResult(groupKeyArray[i], groupByResultHolder, avgPair.getSum(), avgPair.getCount());
+ }
+ });
}
protected void aggregateSVGroupBySV(BlockValSet blockValSet, int length, int[] groupKeyArray,
@@ -207,12 +211,14 @@ public class AvgAggregationFunction extends BaseSingleInputAggregationFunction<A
GroupByResultHolder groupByResultHolder) {
// Serialized AvgPair
byte[][] bytesValues = blockValSet.getBytesValuesSV();
- for (int i = 0; i < length; i++) {
- AvgPair avgPair = ObjectSerDeUtils.AVG_PAIR_SER_DE.deserialize(bytesValues[i]);
- for (int groupKey : groupKeysArray[i]) {
- updateGroupByResult(groupKey, groupByResultHolder, avgPair.getSum(), avgPair.getCount());
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ AvgPair avgPair = ObjectSerDeUtils.AVG_PAIR_SER_DE.deserialize(bytesValues[i]);
+ for (int groupKey : groupKeysArray[i]) {
+ updateGroupByResult(groupKey, groupByResultHolder, avgPair.getSum(), avgPair.getCount());
+ }
}
- }
+ });
}
protected void aggregateSVGroupByMV(BlockValSet blockValSet, int length, int[][] groupKeysArray,
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/startree/StarTreeUtils.java b/pinot-core/src/main/java/org/apache/pinot/core/startree/StarTreeUtils.java
index 66e641e98b3..05cc42ddad1 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/startree/StarTreeUtils.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/startree/StarTreeUtils.java
@@ -39,6 +39,7 @@ import org.apache.pinot.core.operator.filter.predicate.PredicateEvaluator;
import org.apache.pinot.core.operator.filter.predicate.PredicateEvaluatorProvider;
import org.apache.pinot.core.query.aggregation.function.AggregationFunction;
import org.apache.pinot.core.query.aggregation.function.AggregationFunctionUtils;
+import org.apache.pinot.core.query.aggregation.function.AggregationFunctionUtils.AggregationInfo;
import org.apache.pinot.core.query.aggregation.function.array.BaseArrayAggFunction;
import org.apache.pinot.core.query.request.context.QueryContext;
import org.apache.pinot.core.startree.plan.StarTreeProjectPlanNode;
@@ -66,12 +67,24 @@ public class StarTreeUtils {
@Nullable
public static AggregationFunctionColumnPair[] extractAggregationFunctionPairs(
AggregationFunction[] aggregationFunctions) {
+ return extractAggregationFunctionPairs(aggregationFunctions, false);
+ }
+
+ /// Extracts the [AggregationFunctionColumnPair]s from the given [AggregationFunction]s, resolving them against
+ /// either a regular or a null-aware star-tree. Returns `null` if any [AggregationFunction] cannot be represented as
+ /// an [AggregationFunctionColumnPair].
+ ///
+ /// The only pair that differs between the two is `COUNT`: a regular star-tree stores a single `count__*` of every
+ /// row, while a null-aware star-tree stores a `count__column` holding the count of that column's non-null values.
+ @Nullable
+ public static AggregationFunctionColumnPair[] extractAggregationFunctionPairs(
+ AggregationFunction[] aggregationFunctions, boolean nullHandlingEnabled) {
int numAggregationFunctions = aggregationFunctions.length;
AggregationFunctionColumnPair[] aggregationFunctionColumnPairs =
new AggregationFunctionColumnPair[numAggregationFunctions];
for (int i = 0; i < numAggregationFunctions; i++) {
AggregationFunctionColumnPair aggregationFunctionColumnPair =
- AggregationFunctionUtils.getStoredFunctionColumnPair(aggregationFunctions[i]);
+ AggregationFunctionUtils.getStoredFunctionColumnPair(aggregationFunctions[i], nullHandlingEnabled);
if (aggregationFunctionColumnPair != null) {
aggregationFunctionColumnPairs[i] = aggregationFunctionColumnPair;
} else {
@@ -350,6 +363,16 @@ public class StarTreeUtils {
// Do not use star-tree for the following predicates because:
// - REGEXP_LIKE: Need to scan the whole dictionary to gather the matching dictionary ids
// - TEXT_MATCH/IS_NULL/IS_NOT_NULL: No way to gather the matching dictionary ids
+ // TODO: Support IS_NULL / IS_NOT_NULL on a null-aware star-tree.
+ // Nothing in the tree prevents it: a null-aware star-tree groups nulls under a reserved dictionary id one
+ // past the column's last real id, so IS_NULL matches that id alone and IS_NOT_NULL matches every real id.
+ // Null rows form their own child node, and StarTreeFilterOperator already skips the star node for a
+ // predicated dimension, so nulls cannot leak in through it (at the cost of enumerating real children for
+ // IS_NOT_NULL). Two gaps: FilterPlanNode answers both straight from the segment's null vector with a
+ // BitmapBasedFilterOperator and never builds a predicate evaluator, so there is no getMatchingDictIds() to
+ // call here; and the reserved id only exists in the tree, since the forward index stores the column's
+ // default null value, so a leftover predicate could not re-apply it. Supporting them needs a star-tree
+ // specific evaluator that knows the reserved id.
case REGEXP_LIKE:
case TEXT_MATCH:
case IS_NULL:
@@ -379,9 +402,19 @@ public class StarTreeUtils {
dataSource.getDataSourceMetadata().getDataType(), null);
}
- /// Returns a [BaseProjectOperator] when the filter can be solved with star-tree, or `null` otherwise.
+ /// Returns an [AggregationInfo] reading a star-tree when the filter can be solved with one, or `null` otherwise.
+ ///
+ /// The resolved [AggregationFunctionColumnPair]s travel with the project operator because they depend on which
+ /// star-tree was picked: a null-aware star-tree resolves `COUNT(column)` to `count__column`, while a regular one
+ /// resolves it to `count__*`. The aggregation executors must read back the same columns that were projected.
+ ///
+ /// A star-tree is only consistent with one null-handling mode. A regular star-tree folds nulls into the column's
+ /// default null value and includes them in the pre-aggregation, matching null-handling-off semantics; a null-aware
+ /// star-tree keeps nulls apart and excludes them, matching null-handling-on semantics. Queries are therefore routed
+ /// to a star-tree built in the matching mode, except that a null-handling-on query may still fall back to a regular
+ /// star-tree when none of the columns it touches actually contains a null value.
@Nullable
- public static BaseProjectOperator<?> createStarTreeBasedProjectOperator(IndexSegment indexSegment,
+ public static AggregationInfo createStarTreeBasedAggregationInfo(IndexSegment indexSegment,
QueryContext queryContext, AggregationFunction[] aggregationFunctions, @Nullable FilterContext filter,
List<Pair<Predicate, PredicateEvaluator>> predicateEvaluators) {
List<StarTreeV2> starTrees = indexSegment.getStarTrees();
@@ -389,6 +422,8 @@ public class StarTreeUtils {
return null;
}
+ // Resolved here only for the checks below, which do not depend on which star-tree is picked. The pairs the
+ // projection reads are resolved per star-tree in createAggregationInfo, because COUNT differs between the two.
AggregationFunctionColumnPair[] aggregationFunctionColumnPairs =
extractAggregationFunctionPairs(aggregationFunctions);
if (aggregationFunctionColumnPairs == null) {
@@ -423,82 +458,117 @@ public class StarTreeUtils {
.toArray(new ExpressionContext[0]) : null;
if (queryContext.isNullHandlingEnabled()) {
- // We can still use the star-tree index if there aren't actually any null values in this segment for all the
- // metrics being aggregated, all the dimensions being filtered on / grouped by.
- for (int i = 0; i < aggregationFunctionColumnPairs.length; i++) {
- AggregationFunctionColumnPair aggregationFunctionColumnPair = aggregationFunctionColumnPairs[i];
- if (aggregationFunctionColumnPair == AggregationFunctionColumnPair.COUNT_STAR) {
- // COUNT aggregation function returns a non-empty input expressions list only when null handling is enabled
- // and the input operand is a non-star identifier or function.
- List<ExpressionContext> inputExpressions = aggregationFunctions[i].getInputExpressions();
- if (!inputExpressions.isEmpty()) {
- if (inputExpressions.get(0).getType() == ExpressionContext.Type.IDENTIFIER) {
- DataSource dataSource = indexSegment.getDataSource(inputExpressions.get(0).getIdentifier());
- if (FilterOperatorUtils.hasNulls(dataSource)) {
- return null;
- }
- }
- }
- // Null handling is irrelevant for COUNT(*), COUNT(literal), COUNT(nonNullColumn)
- continue;
- }
-
- String column = aggregationFunctionColumnPair.getColumn();
- DataSource dataSource = indexSegment.getDataSourceNullable(column);
- if (dataSource == null) {
- LOGGER.debug("Cannot use star-tree index because aggregation column: '{}' does not exist", column);
- return null;
- }
- if (FilterOperatorUtils.hasNulls(dataSource)) {
- LOGGER.debug("Cannot use star-tree index because aggregation column: '{}' has null values", column);
- return null;
- }
- }
-
- for (String column : predicateEvaluatorsMap.keySet()) {
- DataSource dataSource = indexSegment.getDataSourceNullable(column);
- if (dataSource == null) {
- LOGGER.debug("Cannot use star-tree index because filter column: '{}' does not exist", column);
- return null;
- }
- if (FilterOperatorUtils.hasNulls(dataSource)) {
- LOGGER.debug("Cannot use star-tree index because filter column: '{}' has null values", column);
- return null;
- }
+ // A null-aware star-tree pre-aggregates with exactly the semantics the query asks for
+ AggregationInfo aggregationInfo = createAggregationInfo(indexSegment, queryContext, starTrees, true,
+ aggregationFunctions, groupByExpressions, predicateEvaluatorsMap);
+ if (aggregationInfo != null) {
+ return aggregationInfo;
}
+ }
+ return createAggregationInfo(indexSegment, queryContext, starTrees, false, aggregationFunctions,
+ groupByExpressions, predicateEvaluatorsMap);
+ }
- Set<String> groupByColumns = new HashSet<>();
- if (groupByExpressions != null) {
- for (ExpressionContext groupByExpression : groupByExpressions) {
- groupByExpression.getColumns(groupByColumns);
- }
- }
- for (String column : groupByColumns) {
- DataSource dataSource = indexSegment.getDataSourceNullable(column);
- if (dataSource == null) {
- LOGGER.debug("Cannot use star-tree index because group-by column: '{}' does not exist", column);
- return null;
- }
- if (FilterOperatorUtils.hasNulls(dataSource)) {
- LOGGER.debug("Cannot use star-tree index because group-by column: '{}' has null values", column);
- return null;
- }
- }
+ /// Returns an [AggregationInfo] built on the first star-tree that both matches `nullAware` and fits the query, or
+ /// `null` if there is none.
+ ///
+ /// Resolves the function-column pairs against the same mode, because a null-aware star-tree stores `COUNT` per
+ /// column while a regular one stores a single count of every row, and the executors have to read back whichever
+ /// was projected.
+ @Nullable
+ private static AggregationInfo createAggregationInfo(IndexSegment indexSegment, QueryContext queryContext,
+ List<StarTreeV2> starTrees, boolean nullAware, AggregationFunction[] aggregationFunctions,
+ @Nullable ExpressionContext[] groupByExpressions,
+ Map<String, List<CompositePredicateEvaluator>> predicateEvaluatorsMap) {
+ // Only `COUNT` resolves differently between the two, and never to `null`, so a query that cannot be represented
+ // as pairs at all fails here for either kind of star-tree
+ AggregationFunctionColumnPair[] functionColumnPairs =
+ extractAggregationFunctionPairs(aggregationFunctions, nullAware);
+ if (functionColumnPairs == null) {
+ return null;
+ }
+ // A regular star-tree folded nulls into the column's default value and counted them, so it can only answer a
+ // null-handling-on query when nothing the query touches is actually null
+ if (!nullAware && queryContext.isNullHandlingEnabled() && !hasNoNullValues(indexSegment, aggregationFunctions,
+ functionColumnPairs, predicateEvaluatorsMap.keySet(), groupByExpressions)) {
+ return null;
}
List<Pair<AggregationFunction, AggregationFunctionColumnPair>> aggregations =
new ArrayList<>(aggregationFunctions.length);
for (int i = 0; i < aggregationFunctions.length; i++) {
- aggregations.add(Pair.of(aggregationFunctions[i], aggregationFunctionColumnPairs[i]));
+ aggregations.add(Pair.of(aggregationFunctions[i], functionColumnPairs[i]));
}
for (StarTreeV2 starTreeV2 : starTrees) {
- if (isFitForStarTree(starTreeV2.getMetadata(), aggregations, groupByExpressions,
- predicateEvaluatorsMap.keySet())) {
- return new StarTreeProjectPlanNode(queryContext, starTreeV2, aggregationFunctionColumnPairs, groupByExpressions,
- predicateEvaluatorsMap).run();
+ StarTreeV2Metadata metadata = starTreeV2.getMetadata();
+ if (metadata.isNullHandlingEnabled() != nullAware) {
+ continue;
+ }
+ if (isFitForStarTree(metadata, aggregations, groupByExpressions, predicateEvaluatorsMap.keySet())) {
+ BaseProjectOperator<?> projectOperator =
+ new StarTreeProjectPlanNode(queryContext, starTreeV2, functionColumnPairs, groupByExpressions,
+ predicateEvaluatorsMap).run();
+ return new AggregationInfo(aggregationFunctions, projectOperator, functionColumnPairs);
}
}
return null;
}
+
+ /// Returns whether none of the columns the query touches contains a null value in this segment, in which case a
+ /// regular star-tree produces the same result as a null-aware one and can serve a null-handling-on query.
+ private static boolean hasNoNullValues(IndexSegment indexSegment, AggregationFunction[] aggregationFunctions,
+ AggregationFunctionColumnPair[] functionColumnPairs, Set<String> predicateColumns,
+ @Nullable ExpressionContext[] groupByExpressions) {
+ for (int i = 0; i < functionColumnPairs.length; i++) {
+ AggregationFunctionColumnPair functionColumnPair = functionColumnPairs[i];
+ if (functionColumnPair == AggregationFunctionColumnPair.COUNT_STAR) {
+ // COUNT aggregation function returns a non-empty input expressions list only when null handling is enabled
+ // and the input operand is a non-star identifier or function. Null handling is irrelevant for COUNT(*),
+ // COUNT(literal) and COUNT(nonNullColumn).
+ List<ExpressionContext> inputExpressions = aggregationFunctions[i].getInputExpressions();
+ if (!inputExpressions.isEmpty() && inputExpressions.get(0).getType() == ExpressionContext.Type.IDENTIFIER
+ && !hasNoNullValues(indexSegment, inputExpressions.get(0).getIdentifier(), "aggregation")) {
+ return false;
+ }
+ continue;
+ }
+
+ if (!hasNoNullValues(indexSegment, functionColumnPair.getColumn(), "aggregation")) {
+ return false;
+ }
+ }
+
+ for (String column : predicateColumns) {
+ if (!hasNoNullValues(indexSegment, column, "filter")) {
+ return false;
+ }
+ }
+
+ Set<String> groupByColumns = new HashSet<>();
+ if (groupByExpressions != null) {
+ for (ExpressionContext groupByExpression : groupByExpressions) {
+ groupByExpression.getColumns(groupByColumns);
+ }
+ }
+ for (String column : groupByColumns) {
+ if (!hasNoNullValues(indexSegment, column, "group-by")) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static boolean hasNoNullValues(IndexSegment indexSegment, String column, String columnRole) {
+ DataSource dataSource = indexSegment.getDataSourceNullable(column);
+ if (dataSource == null) {
+ LOGGER.debug("Cannot use star-tree index because {} column: '{}' does not exist", columnRole, column);
+ return false;
+ }
+ if (FilterOperatorUtils.hasNulls(dataSource)) {
+ LOGGER.debug("Cannot use star-tree index because {} column: '{}' has null values", columnRole, column);
+ return false;
+ }
+ return true;
+ }
}
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/startree/executor/StarTreeAggregationExecutor.java b/pinot-core/src/main/java/org/apache/pinot/core/startree/executor/StarTreeAggregationExecutor.java
index 0874619f5b5..21f8fa9aecc 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/startree/executor/StarTreeAggregationExecutor.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/startree/executor/StarTreeAggregationExecutor.java
@@ -33,18 +33,17 @@ import org.apache.pinot.segment.spi.index.startree.AggregationFunctionColumnPair
public class StarTreeAggregationExecutor extends DefaultAggregationExecutor {
private final AggregationFunctionColumnPair[] _aggregationFunctionColumnPairs;
-
- public StarTreeAggregationExecutor(AggregationFunction[] aggregationFunctions) {
+ /// Creates an executor over the pre-aggregated columns the query was routed to.
+ ///
+ /// `aggregationFunctionColumnPairs` must be the pairs the star-tree project operator was built with, because they
+ /// depend on which star-tree was picked: a null-aware star-tree resolves `COUNT(column)` to `count__column` while a
+ /// regular one resolves it to `count__*`.
+ public StarTreeAggregationExecutor(AggregationFunction[] aggregationFunctions,
+ AggregationFunctionColumnPair[] aggregationFunctionColumnPairs) {
// StarTreeAggregationExecutor doesn't support pre-aggregated results.
// So, we don't need to pass pre-aggregated results to the super class.
super(aggregationFunctions);
-
- int numAggregationFunctions = aggregationFunctions.length;
- _aggregationFunctionColumnPairs = new AggregationFunctionColumnPair[numAggregationFunctions];
- for (int i = 0; i < numAggregationFunctions; i++) {
- _aggregationFunctionColumnPairs[i] =
- AggregationFunctionUtils.getStoredFunctionColumnPair(aggregationFunctions[i]);
- }
+ _aggregationFunctionColumnPairs = aggregationFunctionColumnPairs;
}
@Override
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/startree/executor/StarTreeGroupByExecutor.java b/pinot-core/src/main/java/org/apache/pinot/core/startree/executor/StarTreeGroupByExecutor.java
index f72a76a547a..67ccb5fd5a8 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/startree/executor/StarTreeGroupByExecutor.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/startree/executor/StarTreeGroupByExecutor.java
@@ -43,27 +43,21 @@ public class StarTreeGroupByExecutor extends DefaultGroupByExecutor {
private final AggregationFunctionColumnPair[] _aggregationFunctionColumnPairs;
public StarTreeGroupByExecutor(QueryContext queryContext, ExpressionContext[] groupByExpressions,
- BaseProjectOperator<?> projectOperator) {
- this(queryContext, queryContext.getAggregationFunctions(), groupByExpressions, projectOperator, null);
- }
-
- public StarTreeGroupByExecutor(QueryContext queryContext, AggregationFunction[] aggregationFunctions,
- ExpressionContext[] groupByExpressions, BaseProjectOperator<?> projectOperator) {
- this(queryContext, aggregationFunctions, groupByExpressions, projectOperator, null);
+ BaseProjectOperator<?> projectOperator, AggregationFunctionColumnPair[] aggregationFunctionColumnPairs) {
+ this(queryContext, queryContext.getAggregationFunctions(), groupByExpressions, projectOperator,
+ aggregationFunctionColumnPairs, null);
}
+ /// Creates an executor over the pre-aggregated columns the query was routed to.
+ ///
+ /// `aggregationFunctionColumnPairs` must be the pairs the star-tree project operator was built with, because they
+ /// depend on which star-tree was picked: a null-aware star-tree resolves `COUNT(column)` to `count__column` while a
+ /// regular one resolves it to `count__*`.
public StarTreeGroupByExecutor(QueryContext queryContext, AggregationFunction[] aggregationFunctions,
ExpressionContext[] groupByExpressions, BaseProjectOperator<?> projectOperator,
- @Nullable GroupKeyGenerator groupKeyGenerator) {
+ AggregationFunctionColumnPair[] aggregationFunctionColumnPairs, @Nullable GroupKeyGenerator groupKeyGenerator) {
super(queryContext, aggregationFunctions, groupByExpressions, projectOperator, groupKeyGenerator);
-
- assert aggregationFunctions != null;
- int numAggregationFunctions = aggregationFunctions.length;
- _aggregationFunctionColumnPairs = new AggregationFunctionColumnPair[numAggregationFunctions];
- for (int i = 0; i < numAggregationFunctions; i++) {
- _aggregationFunctionColumnPairs[i] =
- AggregationFunctionUtils.getStoredFunctionColumnPair(aggregationFunctions[i]);
- }
+ _aggregationFunctionColumnPairs = aggregationFunctionColumnPairs;
}
@Override
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/startree/StarTreeUtilsTest.java b/pinot-core/src/test/java/org/apache/pinot/core/startree/StarTreeUtilsTest.java
index cd18702ad88..284937c86b0 100644
--- a/pinot-core/src/test/java/org/apache/pinot/core/startree/StarTreeUtilsTest.java
+++ b/pinot-core/src/test/java/org/apache/pinot/core/startree/StarTreeUtilsTest.java
@@ -25,12 +25,12 @@ import java.util.List;
import org.apache.commons.io.FileUtils;
import org.apache.pinot.common.request.context.ExpressionContext;
import org.apache.pinot.common.request.context.predicate.EqPredicate;
-import org.apache.pinot.core.operator.BaseProjectOperator;
import org.apache.pinot.core.operator.blocks.ValueBlock;
import org.apache.pinot.core.operator.filter.predicate.BaseRawValueBasedPredicateEvaluator;
import org.apache.pinot.core.operator.filter.predicate.EqualsPredicateEvaluatorFactory;
import org.apache.pinot.core.operator.filter.predicate.PredicateEvaluator;
import org.apache.pinot.core.plan.FilterPlanNode;
+import org.apache.pinot.core.query.aggregation.function.AggregationFunctionUtils.AggregationInfo;
import org.apache.pinot.core.query.request.context.QueryContext;
import org.apache.pinot.core.query.request.context.utils.QueryContextConverterUtils;
import org.apache.pinot.segment.local.indexsegment.immutable.ImmutableSegmentLoader;
@@ -216,7 +216,7 @@ public class StarTreeUtilsTest {
/// End-to-end assertion equivalent to `EXPLAIN PLAN` showing a `STAR_TREE` node:
/// a filter query returning `COUNT(*)` from a RAW-forward + separated-dictionary star-tree dimension is served
- /// via the star-tree operator (non-null result from `createStarTreeBasedProjectOperator`) and reads a single
+ /// via the star-tree operator (non-null result from `createStarTreeBasedAggregationInfo`) and reads a single
/// aggregated record per matching path — same as a `DICTIONARY`-encoded dim would.
@Test
public void testStarTreeAcceleratesEqualityFilterOnRawWithDictionary() {
@@ -226,13 +226,14 @@ public class StarTreeUtilsTest {
FilterPlanNode filterPlanNode = new FilterPlanNode(new SegmentContext(_segment), queryContext);
filterPlanNode.run();
- BaseProjectOperator<?> operator = StarTreeUtils.createStarTreeBasedProjectOperator(_segment, queryContext,
- queryContext.getAggregationFunctions(), queryContext.getFilter(),
- filterPlanNode.getPredicateEvaluators());
- assertNotNull(operator, "Star-tree plan expected for EQ on RAW+dict dimension");
+ AggregationInfo aggregationInfo =
+ StarTreeUtils.createStarTreeBasedAggregationInfo(_segment, queryContext,
+ queryContext.getAggregationFunctions(), queryContext.getFilter(),
+ filterPlanNode.getPredicateEvaluators());
+ assertNotNull(aggregationInfo, "Star-tree plan expected for EQ on RAW+dict dimension");
// Traversal must not throw; before the fix, StarTreeFilterOperator.getMatchingDictIds threw UOE here.
- ValueBlock block = operator.nextBlock();
+ ValueBlock block = aggregationInfo.getProjectOperator().nextBlock();
assertNotNull(block, "Star-tree traversal returned no block");
// Star-tree yields one aggregated document per matching path — same behavior as DICTIONARY-encoded columns.
diff --git a/pinot-core/src/test/java/org/apache/pinot/queries/NullAwareStarTreeQueriesTest.java b/pinot-core/src/test/java/org/apache/pinot/queries/NullAwareStarTreeQueriesTest.java
new file mode 100644
index 00000000000..4d08a76a504
--- /dev/null
+++ b/pinot-core/src/test/java/org/apache/pinot/queries/NullAwareStarTreeQueriesTest.java
@@ -0,0 +1,387 @@
+/**
+ * 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.queries;
+
+import java.io.File;
+import java.io.IOException;
+import java.lang.reflect.Array;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import javax.annotation.Nullable;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.common.response.broker.BrokerResponseNative;
+import org.apache.pinot.common.response.broker.ResultTable;
+import org.apache.pinot.segment.local.indexsegment.immutable.ImmutableSegmentLoader;
+import org.apache.pinot.segment.local.segment.creator.impl.SegmentIndexCreationDriverImpl;
+import org.apache.pinot.segment.local.segment.readers.GenericRowRecordReader;
+import org.apache.pinot.segment.local.startree.v2.builder.MultipleTreesBuilder;
+import org.apache.pinot.segment.local.startree.v2.builder.MultipleTreesBuilder.BuildMode;
+import org.apache.pinot.segment.spi.ImmutableSegment;
+import org.apache.pinot.segment.spi.IndexSegment;
+import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig;
+import org.apache.pinot.spi.config.table.StarTreeAggregationConfig;
+import org.apache.pinot.spi.config.table.StarTreeIndexConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.data.readers.GenericRow;
+import org.apache.pinot.spi.utils.ReadMode;
+import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertTrue;
+
+
+/// Queries answered from a null-aware star-tree, where the dimension itself contains nulls.
+///
+/// `NullAwareStarTreeBuilderTest` only inspects what the builder stored. These go through the query path instead,
+/// which is where a null row has to read back as a null rather than as the value standing in for it.
+///
+/// The segment is built twice, once per build mode, and both copies are queried together: the harness fans the
+/// segment list out to two servers, so every aggregate is scaled by [#SEGMENT_COPIES], and a builder that disagreed
+/// with the other would show up as a wrong total. Each segment holds a null-aware and a regular star-tree over the
+/// same columns, so routing between the two is exercised as well.
+public class NullAwareStarTreeQueriesTest extends BaseQueriesTest {
+ private static final File INDEX_DIR = new File(FileUtils.getTempDirectory(), "NullAwareStarTreeQueriesTest");
+ private static final String RAW_TABLE_NAME = "testTable";
+ private static final String SEGMENT_NAME = "testSegment";
+ private static final String DIMENSION = "d";
+ private static final String METRIC = "m";
+ /// A metric that is null for every row of one dimension group, so that group has no aggregated value.
+ private static final String SPARSE_METRIC = "m2";
+ /// A metric that is null in every row, so no group has an aggregated value and the column has no size to derive.
+ private static final String PRECISE_METRIC = "m3";
+
+ private static final Map<String, String> NULL_HANDLING_ENABLED = Map.of("enableNullHandling", "true");
+
+ /// Two segments per server and two servers.
+ private static final int SEGMENT_COPIES = 4;
+
+ /// One record per leaf, so each distinct dimension value gets its own pre-aggregated document.
+ private static final int MAX_LEAF_RECORDS = 1;
+
+ /// `d` is null for two rows, so the null-aware star-tree stores those under the reserved dictionary id. The metric
+ /// sums are distinct per group so a group picking up the wrong rows is visible in the answer.
+ private static final Integer[] DIMENSION_VALUES = {1, 1, 1, 1, null, null, 2, 2, 2, 2};
+ private static final int[] METRIC_VALUES = {1, 2, 3, 4, 5, 6, 7, 8, 9, 10};
+ private static final Integer[] SPARSE_METRIC_VALUES = {10, null, 30, 40, 50, null, null, null, null, null};
+ private static final int SUM_WHERE_D_IS_1 = 1 + 2 + 3 + 4;
+ private static final int SUM_WHERE_D_IS_NULL = 5 + 6;
+ private static final int SUM_WHERE_D_IS_2 = 7 + 8 + 9 + 10;
+ private static final int SPARSE_SUM_WHERE_D_IS_1 = 10 + 30 + 40;
+ private static final int SPARSE_COUNT_WHERE_D_IS_1 = 3;
+ private static final int SPARSE_SUM_WHERE_D_IS_NULL = 50;
+ private static final int SPARSE_COUNT_WHERE_D_IS_NULL = 1;
+ /// A value no row holds, so a `<>` against it is true over every real value and UNKNOWN over a null.
+ private static final int ABSENT_VALUE = 99999;
+
+ private IndexSegment _indexSegment;
+ private List<IndexSegment> _indexSegments;
+
+ @Override
+ protected String getFilter() {
+ return "";
+ }
+
+ @Override
+ protected IndexSegment getIndexSegment() {
+ return _indexSegment;
+ }
+
+ @Override
+ protected List<IndexSegment> getIndexSegments() {
+ return _indexSegments;
+ }
+
+ @BeforeClass
+ public void setUp()
+ throws Exception {
+ FileUtils.deleteDirectory(INDEX_DIR);
+ _indexSegments = new ArrayList<>(2);
+ for (BuildMode buildMode : BuildMode.values()) {
+ _indexSegments.add(buildSegment(buildMode));
+ }
+ _indexSegment = _indexSegments.get(0);
+ }
+
+ private static ImmutableSegment buildSegment(BuildMode buildMode)
+ throws Exception {
+ Schema schema = new Schema.SchemaBuilder().setSchemaName(RAW_TABLE_NAME)
+ .addSingleValueDimension(DIMENSION, DataType.INT)
+ .addMetric(METRIC, DataType.INT)
+ .addMetric(SPARSE_METRIC, DataType.INT)
+ .addMetric(PRECISE_METRIC, DataType.BIG_DECIMAL)
+ .build();
+ TableConfig tableConfig =
+ new TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME).setNullHandlingEnabled(true).build();
+
+ List<GenericRow> rows = new ArrayList<>(DIMENSION_VALUES.length);
+ for (int i = 0; i < DIMENSION_VALUES.length; i++) {
+ GenericRow row = new GenericRow();
+ row.putValue(DIMENSION, DIMENSION_VALUES[i]);
+ row.putValue(METRIC, METRIC_VALUES[i]);
+ row.putValue(SPARSE_METRIC, SPARSE_METRIC_VALUES[i]);
+ row.putValue(PRECISE_METRIC, null);
+ rows.add(row);
+ }
+
+ File outDir = new File(INDEX_DIR, buildMode.name());
+ SegmentGeneratorConfig segmentGeneratorConfig = new SegmentGeneratorConfig(tableConfig, schema);
+ segmentGeneratorConfig.setTableName(RAW_TABLE_NAME);
+ segmentGeneratorConfig.setSegmentName(SEGMENT_NAME);
+ segmentGeneratorConfig.setDefaultNullHandlingEnabled(true);
+ segmentGeneratorConfig.setOutDir(outDir.getPath());
+ SegmentIndexCreationDriverImpl driver = new SegmentIndexCreationDriverImpl();
+ driver.init(segmentGeneratorConfig, new GenericRowRecordReader(rows));
+ driver.build();
+
+ File indexDir = new File(outDir, SEGMENT_NAME);
+ StarTreeIndexConfig nullAwareConfig = new StarTreeIndexConfig(List.of(DIMENSION), null,
+ List.of("SUM__" + METRIC, "SUM__" + SPARSE_METRIC, "COUNT__" + SPARSE_METRIC, "AVG__" + SPARSE_METRIC,
+ "SUMPRECISION__" + PRECISE_METRIC, "COUNT__*"),
+ List.of(new StarTreeAggregationConfig(SPARSE_METRIC, "arrayAgg")), MAX_LEAF_RECORDS, true);
+ StarTreeIndexConfig regularConfig = new StarTreeIndexConfig(List.of(DIMENSION), null,
+ List.of("SUM__" + METRIC, "SUM__" + SPARSE_METRIC, "COUNT__*"), null, MAX_LEAF_RECORDS);
+ try (MultipleTreesBuilder builder =
+ new MultipleTreesBuilder(List.of(nullAwareConfig, regularConfig), false, indexDir, buildMode)) {
+ builder.build();
+ }
+ return ImmutableSegmentLoader.load(indexDir, ReadMode.mmap);
+ }
+
+ @AfterClass
+ public void tearDown()
+ throws IOException {
+ for (IndexSegment indexSegment : _indexSegments) {
+ indexSegment.destroy();
+ }
+ FileUtils.deleteDirectory(INDEX_DIR);
+ }
+
+ /// A null row is stored under the column's default null value and marked in the dimension's null vector, so the
+ /// group key generator has to read that vector and give the row a group of its own.
+ @Test
+ public void groupingByANullDimensionReturnsTheNullGroup() {
+ BrokerResponseNative response = getBrokerResponse(
+ String.format("SELECT %s, SUM(%s) FROM testTable GROUP BY %s", DIMENSION, METRIC, DIMENSION),
+ NULL_HANDLING_ENABLED);
+
+ assertStarTreeWasUsed(response);
+ ResultTable resultTable = response.getResultTable();
+ assertEquals(resultTable.getRows().size(), 3);
+ assertEquals(aggregateForGroup(resultTable, 1), (double) SUM_WHERE_D_IS_1 * SEGMENT_COPIES);
+ assertEquals(aggregateForGroup(resultTable, 2), (double) SUM_WHERE_D_IS_2 * SEGMENT_COPIES);
+ assertEquals(aggregateForGroup(resultTable, null), (double) SUM_WHERE_D_IS_NULL * SEGMENT_COPIES,
+ "Rows whose dimension is null must form their own group rather than joining a real value's group");
+ }
+
+ /// A group-by on a transform over the dimension reads its values rather than its dictionary ids, so every id the
+ /// star-tree's forward index holds has to resolve against the dictionary it shares with the segment.
+ @Test
+ public void groupingByATransformOverANullDimensionReturnsTheNullGroup() {
+ BrokerResponseNative response = getBrokerResponse(
+ String.format("SELECT %s + 1, SUM(%s) FROM testTable GROUP BY %s + 1", DIMENSION, METRIC, DIMENSION),
+ NULL_HANDLING_ENABLED);
+
+ assertStarTreeWasUsed(response);
+ ResultTable resultTable = response.getResultTable();
+ assertEquals(resultTable.getRows().size(), 3);
+ assertEquals(aggregateForGroup(resultTable, 2.0), (double) SUM_WHERE_D_IS_1 * SEGMENT_COPIES);
+ assertEquals(aggregateForGroup(resultTable, 3.0), (double) SUM_WHERE_D_IS_2 * SEGMENT_COPIES);
+ assertEquals(aggregateForGroup(resultTable, null), (double) SUM_WHERE_D_IS_NULL * SEGMENT_COPIES,
+ "A null dimension makes the transform null, so those rows must form their own group");
+ }
+
+ /// A predicate that is always true over real values is UNKNOWN over a null, so the null rows are not selected.
+ @Test
+ public void anAlwaysTruePredicateExcludesNullRows() {
+ BrokerResponseNative response = getBrokerResponse(
+ String.format("SELECT SUM(%s) FROM testTable WHERE %s <> %d", METRIC, DIMENSION, ABSENT_VALUE),
+ NULL_HANDLING_ENABLED);
+
+ assertStarTreeWasUsed(response);
+ assertEquals(response.getResultTable().getRows().get(0)[0],
+ (double) (SUM_WHERE_D_IS_1 + SUM_WHERE_D_IS_2) * SEGMENT_COPIES,
+ "A null dimension makes the predicate UNKNOWN, so the row must not be aggregated");
+ }
+
+ /// A predicate excluding one real value matches every other one but not a null, whose reserved id lies one past
+ /// the last real value and which the predicate is UNKNOWN over anyway. The negated form reaches the star-tree as a
+ /// negated composite predicate rather than as an exclusive one, so both shapes are covered.
+ @Test
+ public void aPredicateExcludingOneValueAlsoExcludesNullRows() {
+ for (String predicate : List.of(DIMENSION + " <> 1", "NOT (" + DIMENSION + " = 1)")) {
+ BrokerResponseNative response =
+ getBrokerResponse(String.format("SELECT SUM(%s) FROM testTable WHERE %s", METRIC, predicate),
+ NULL_HANDLING_ENABLED);
+
+ assertStarTreeWasUsed(response);
+ assertEquals(response.getResultTable().getRows().get(0)[0], (double) SUM_WHERE_D_IS_2 * SEGMENT_COPIES,
+ "Wrong answer for predicate: " + predicate);
+ }
+ }
+
+ /// With a predicate on the dimension, the null rows are UNKNOWN and no null group may appear.
+ @Test
+ public void aFilteredGroupByHasNoNullGroup() {
+ BrokerResponseNative response = getBrokerResponse(
+ String.format("SELECT %s, SUM(%s) FROM testTable WHERE %s <> 1 GROUP BY %s", DIMENSION, METRIC, DIMENSION,
+ DIMENSION), NULL_HANDLING_ENABLED);
+
+ assertStarTreeWasUsed(response);
+ ResultTable resultTable = response.getResultTable();
+ assertEquals(resultTable.getRows().size(), 1);
+ assertEquals(aggregateForGroup(resultTable, 2), (double) SUM_WHERE_D_IS_2 * SEGMENT_COPIES);
+ }
+
+ /// A group whose metric is null in every row has no aggregated value: `SUM` is `NULL` and `COUNT(column)` is 0,
+ /// which the null-aware star-tree stores per column rather than as the row count.
+ @Test
+ public void anAllNullMetricGroupAggregatesToNull() {
+ BrokerResponseNative response = getBrokerResponse(
+ String.format("SELECT %s, SUM(%s), COUNT(%s) FROM testTable GROUP BY %s", DIMENSION, SPARSE_METRIC,
+ SPARSE_METRIC, DIMENSION), NULL_HANDLING_ENABLED);
+
+ assertStarTreeWasUsed(response);
+ ResultTable resultTable = response.getResultTable();
+ assertEquals(resultTable.getRows().size(), 3);
+ assertEquals(aggregateForGroup(resultTable, 1), (double) SPARSE_SUM_WHERE_D_IS_1 * SEGMENT_COPIES);
+ assertEquals(countForGroup(resultTable, 1), (long) SPARSE_COUNT_WHERE_D_IS_1 * SEGMENT_COPIES);
+ assertEquals(aggregateForGroup(resultTable, null), (double) SPARSE_SUM_WHERE_D_IS_NULL * SEGMENT_COPIES);
+ assertEquals(countForGroup(resultTable, null), (long) SPARSE_COUNT_WHERE_D_IS_NULL * SEGMENT_COPIES);
+ assertNull(rowForGroup(resultTable, 2)[1], "A group with no non-null value has no sum");
+ assertEquals(countForGroup(resultTable, 2), 0L);
+ }
+
+ /// `AVG` reads its pre-aggregated pairs straight from the forward index, where a group that aggregated over no
+ /// non-null input holds a placeholder no aggregator can decode, so it has to skip the rows the null vector marks.
+ @Test
+ public void anAllNullMetricGroupAveragesToNull() {
+ BrokerResponseNative response = getBrokerResponse(
+ String.format("SELECT %s, AVG(%s) FROM testTable GROUP BY %s", DIMENSION, SPARSE_METRIC, DIMENSION),
+ NULL_HANDLING_ENABLED);
+
+ assertStarTreeWasUsed(response);
+ ResultTable resultTable = response.getResultTable();
+ assertEquals(resultTable.getRows().size(), 3);
+ // An average needs no scaling by the segment copies: both its sum and its count scale with them
+ assertEquals(aggregateForGroup(resultTable, 1), (double) SPARSE_SUM_WHERE_D_IS_1 / SPARSE_COUNT_WHERE_D_IS_1);
+ assertEquals(aggregateForGroup(resultTable, null),
+ (double) SPARSE_SUM_WHERE_D_IS_NULL / SPARSE_COUNT_WHERE_D_IS_NULL);
+ assertNull(rowForGroup(resultTable, 2)[1], "A group with no non-null value has no average");
+ }
+
+ /// `arrayAgg` reads its pre-aggregated cells as serialized sets, and a group that aggregated over no non-null input
+ /// holds a placeholder that is not one, so it has to skip the rows the null vector marks. Only the distinct variant
+ /// is stored in a star-tree, and a distinct set absorbs the repeated segment copies rather than scaling with them.
+ @Test
+ public void anAllNullMetricGroupArrayAggregatesToAnEmptyArray() {
+ BrokerResponseNative response = getBrokerResponse(
+ String.format("SELECT %s, arrayAgg(%s, 'INT', true) FROM testTable GROUP BY %s", DIMENSION, SPARSE_METRIC,
+ DIMENSION), NULL_HANDLING_ENABLED);
+
+ assertStarTreeWasUsed(response);
+ ResultTable resultTable = response.getResultTable();
+ assertEquals(resultTable.getRows().size(), 3);
+ assertEquals(arrayForGroup(resultTable, 1), List.of(10, 30, 40));
+ assertEquals(arrayForGroup(resultTable, null), List.of(50));
+ assertEquals(arrayForGroup(resultTable, 2), List.of(),
+ "A group with no non-null value aggregates to an empty array");
+ }
+
+ /// A metric that is null in every row leaves the star-tree with no aggregated value anywhere, so its column holds
+ /// nothing but placeholders and has no value to size itself from.
+ @Test
+ public void anAllNullMetricColumnAggregatesToNull() {
+ BrokerResponseNative response =
+ getBrokerResponse(String.format("SELECT SUMPRECISION(%s) FROM testTable", PRECISE_METRIC),
+ NULL_HANDLING_ENABLED);
+
+ assertStarTreeWasUsed(response);
+ assertNull(response.getResultTable().getRows().get(0)[0]);
+ }
+
+ /// A query without null handling is served by the regular star-tree in the same segment, which folds a null
+ /// dimension into the column's default value and a null metric into `0`: the answers null handling disabled asks
+ /// for, and different from the null-aware ones above.
+ @Test
+ public void aQueryWithoutNullHandlingUsesTheRegularStarTree() {
+ BrokerResponseNative response = getBrokerResponse(
+ String.format("SELECT %s, SUM(%s) FROM testTable GROUP BY %s", DIMENSION, SPARSE_METRIC, DIMENSION));
+
+ assertStarTreeWasUsed(response);
+ ResultTable resultTable = response.getResultTable();
+ assertEquals(resultTable.getRows().size(), 3);
+ assertEquals(aggregateForGroup(resultTable, 1), (double) SPARSE_SUM_WHERE_D_IS_1 * SEGMENT_COPIES);
+ assertEquals(aggregateForGroup(resultTable, FieldSpec.DEFAULT_DIMENSION_NULL_VALUE_OF_INT),
+ (double) SPARSE_SUM_WHERE_D_IS_NULL * SEGMENT_COPIES,
+ "Without null handling the null rows fold into the dimension's default value");
+ assertEquals(aggregateForGroup(resultTable, 2), 0d, "Without null handling a null metric folds into 0");
+ }
+
+ /// Guards against the checks above passing because the query silently fell back to a raw scan, which would answer
+ /// correctly and prove nothing. A star-tree reads one pre-aggregated document per group rather than every row.
+ private static void assertStarTreeWasUsed(BrokerResponseNative response) {
+ long numRowsScanned = (long) DIMENSION_VALUES.length * SEGMENT_COPIES;
+ assertTrue(response.getNumDocsScanned() < numRowsScanned,
+ "Expected the star-tree to be used, but " + response.getNumDocsScanned() + " documents were scanned out of "
+ + numRowsScanned);
+ }
+
+ private static double aggregateForGroup(ResultTable resultTable, @Nullable Object groupKey) {
+ return ((Number) rowForGroup(resultTable, groupKey)[1]).doubleValue();
+ }
+
+ private static long countForGroup(ResultTable resultTable, @Nullable Object groupKey) {
+ return ((Number) rowForGroup(resultTable, groupKey)[2]).longValue();
+ }
+
+ /// Returns the `arrayAgg` cell of a group, sorted, because the aggregate is a set whose order is arbitrary. The
+ /// cell is read reflectively since a multi-value result column arrives as a primitive array.
+ private static List<Integer> arrayForGroup(ResultTable resultTable, @Nullable Object groupKey) {
+ Object cell = rowForGroup(resultTable, groupKey)[1];
+ int length = Array.getLength(cell);
+ List<Integer> values = new ArrayList<>(length);
+ for (int i = 0; i < length; i++) {
+ values.add(((Number) Array.get(cell, i)).intValue());
+ }
+ values.sort(Comparator.naturalOrder());
+ return values;
+ }
+
+ /// Returns the row whose group key matches, or fails when no such group exists.
+ private static Object[] rowForGroup(ResultTable resultTable, @Nullable Object groupKey) {
+ for (Object[] row : resultTable.getRows()) {
+ if (Objects.equals(groupKey, row[0])) {
+ return row;
+ }
+ }
+ throw new AssertionError("No group for dimension value: " + groupKey + " in " + resultTable.getRows().size()
+ + " rows");
+ }
+}
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/CountValueAggregator.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/CountValueAggregator.java
index 2b7e01a6575..498eaf669f0 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/CountValueAggregator.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/CountValueAggregator.java
@@ -41,6 +41,17 @@ public class CountValueAggregator implements ValueAggregator<Object, Long> {
return rawValue != null ? 1L : 0L;
}
+ /// The only aggregator that still answers a group with no non-null input itself.
+ ///
+ /// `COUNT` is read back from the pre-aggregated column by summing it rather than through the null vector, which
+ /// [org.apache.pinot.core.query.aggregation.function.CountAggregationFunction] recognizes from the `__STAR__`
+ /// identifier. Returning `0` is also exactly the placeholder a null would leave behind, so recording the group in
+ /// the null vector would cost a bit per group and buy nothing.
+ @Override
+ public Long getAllNullAggregatedValue() {
+ return 0L;
+ }
+
@Override
public Long applyRawValue(Long value, Object rawValue) {
return value + 1;
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/DistinctCountBitmapValueAggregator.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/DistinctCountBitmapValueAggregator.java
index f1d1d4c161c..74fbb95f2d8 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/DistinctCountBitmapValueAggregator.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/DistinctCountBitmapValueAggregator.java
@@ -41,7 +41,9 @@ public class DistinctCountBitmapValueAggregator implements ValueAggregator<Objec
@Override
public RoaringBitmap getInitialAggregatedValue(Object rawValue) {
- // NOTE: rawValue cannot be null because this aggregator can only be used for star-tree index.
+ // NOTE: rawValue cannot be null because this aggregator can only be used for star-tree index, and the builder
+ // never passes a null raw value: a null-aware star-tree leaves the aggregated value null until the group sees
+ // its first non-null input.
assert rawValue != null;
RoaringBitmap initialValue;
if (rawValue instanceof byte[]) {
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/DistinctCountThetaSketchValueAggregator.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/DistinctCountThetaSketchValueAggregator.java
index c735ea4f906..b2e72d99d6f 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/DistinctCountThetaSketchValueAggregator.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/DistinctCountThetaSketchValueAggregator.java
@@ -116,7 +116,9 @@ public class DistinctCountThetaSketchValueAggregator implements ValueAggregator<
@Override
public Object getInitialAggregatedValue(Object rawValue) {
- // NOTE: rawValue cannot be null because this aggregator can only be used for star-tree index.
+ // NOTE: rawValue cannot be null because this aggregator can only be used for star-tree index, and the builder
+ // never passes a null raw value: a null-aware star-tree leaves the aggregated value null until the group
+ // sees its first non-null input.
assert rawValue != null;
ThetaUnion thetaUnion = _setOperationBuilder.buildUnion();
if (rawValue instanceof byte[]) { // Serialized ThetaSketch
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/ValueAggregator.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/ValueAggregator.java
index 506f02fc0f7..350535d76d9 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/ValueAggregator.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/ValueAggregator.java
@@ -41,6 +41,28 @@ public interface ValueAggregator<R, A> {
/// specified in the schema.
A getInitialAggregatedValue(@Nullable R rawValue);
+ /// Returns the aggregated value of a group whose input values are all null, or `null` to have the star-tree record
+ /// the group in its null vector instead.
+ ///
+ /// Only consulted by null-aware star-trees, which exclude null input values from the pre-aggregation and can
+ /// therefore produce a group with no values at all.
+ ///
+ /// Returning `null` is safe whenever the aggregation function skips null rows while reading the pre-aggregated
+ /// column, which every aggregation function does apart from `COUNT`. A group recorded in the null vector is never
+ /// read back, so the placeholder left in the forward index is never deserialized.
+ ///
+ /// `COUNT` is the exception and overrides this: it is read back by summing the pre-aggregated column rather than
+ /// through the null vector, so it answers `0` itself. Every other aggregator takes the default, which keeps an
+ /// all-null group down to a placeholder plus one null-vector bit instead of a serialized empty sketch.
+ ///
+ /// An aggregator whose [#getAggregatedValueType] is `BYTES` must also make [#getMaxAggregatedValueByteSize] account
+ /// for the value returned here, because the star-tree sizes the variable-length forward index from that and would
+ /// otherwise under-allocate for a group that has no value of its own.
+ @Nullable
+ default A getAllNullAggregatedValue() {
+ return null;
+ }
+
/// Applies a raw value to the current aggregated value.
///
/// NOTE: if value is mutable, will directly modify the value.
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/store/StarTreeIndexReader.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/store/StarTreeIndexReader.java
index 4e00fcde59f..e53f33435ed 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/store/StarTreeIndexReader.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/store/StarTreeIndexReader.java
@@ -102,11 +102,15 @@ public class StarTreeIndexReader implements Closeable {
new StarTreeIndexEntry(indexMap.get(StarTreeIndexMapUtils.STAR_TREE_INDEX_KEY), _dataBuffer,
ByteOrder.LITTLE_ENDIAN));
StarTreeV2Metadata starTreeMetadata = _starTreeMetadataList.get(starTreeId);
+ boolean nullHandlingEnabled = starTreeMetadata.isNullHandlingEnabled();
// Load dimension forward indexes
for (String dimension : starTreeMetadata.getDimensionsSplitOrder()) {
columnEntries.put(new IndexKey(dimension, StandardIndexes.forward()), new StarTreeIndexEntry(
indexMap.get(new StarTreeIndexMapUtils.IndexKey(StarTreeIndexMapUtils.IndexType.FORWARD_INDEX, dimension)),
_dataBuffer, ByteOrder.BIG_ENDIAN));
+ if (nullHandlingEnabled) {
+ mapNullValueVectorEntry(columnEntries, indexMap, dimension);
+ }
}
// Load metric (function-column pair) forward indexes
for (AggregationFunctionColumnPair functionColumnPair : starTreeMetadata.getFunctionColumnPairs()) {
@@ -114,6 +118,23 @@ public class StarTreeIndexReader implements Closeable {
columnEntries.put(new IndexKey(metric, StandardIndexes.forward()), new StarTreeIndexEntry(
indexMap.get(new StarTreeIndexMapUtils.IndexKey(StarTreeIndexMapUtils.IndexType.FORWARD_INDEX, metric)),
_dataBuffer, ByteOrder.BIG_ENDIAN));
+ if (nullHandlingEnabled) {
+ mapNullValueVectorEntry(columnEntries, indexMap, metric);
+ }
+ }
+ }
+
+ /// Maps the null value vector of a null-aware star-tree column, if it has one.
+ ///
+ /// The builder only writes a null value vector for columns that actually contain null values, so a missing entry
+ /// simply means that the column has none.
+ private void mapNullValueVectorEntry(Map<IndexKey, StarTreeIndexEntry> columnEntries,
+ Map<StarTreeIndexMapUtils.IndexKey, StarTreeIndexMapUtils.IndexValue> indexMap, String column) {
+ StarTreeIndexMapUtils.IndexValue indexValue =
+ indexMap.get(new StarTreeIndexMapUtils.IndexKey(StarTreeIndexMapUtils.IndexType.NULL_VALUE_VECTOR, column));
+ if (indexValue != null) {
+ columnEntries.put(new IndexKey(column, StandardIndexes.nullValueVector()),
+ new StarTreeIndexEntry(indexValue, _dataBuffer, ByteOrder.BIG_ENDIAN));
}
}
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/startree/StarTreeBuilderUtils.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/startree/StarTreeBuilderUtils.java
index 465a5737f40..ad2a5ad5d84 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/startree/StarTreeBuilderUtils.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/startree/StarTreeBuilderUtils.java
@@ -266,6 +266,10 @@ public class StarTreeBuilderUtils {
if (builderConfig.getMaxLeafRecords() != metadata.getMaxLeafRecords()) {
return true;
}
+ // Switching null handling on or off changes how the whole star-tree is pre-aggregated, so it has to be rebuilt
+ if (builderConfig.isNullHandlingEnabled() != metadata.isNullHandlingEnabled()) {
+ return true;
+ }
TreeMap<AggregationFunctionColumnPair, AggregationSpec> newSpecs = builderConfig.getAggregationSpecs();
TreeMap<AggregationFunctionColumnPair, AggregationSpec> existingSpecs = metadata.getAggregationSpecs();
if (newSpecs.size() != existingSpecs.size()) {
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/startree/v2/builder/BaseSingleTreeBuilder.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/startree/v2/builder/BaseSingleTreeBuilder.java
index e40b4860125..5ef46fc14ed 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/startree/v2/builder/BaseSingleTreeBuilder.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/startree/v2/builder/BaseSingleTreeBuilder.java
@@ -31,12 +31,14 @@ import java.util.Set;
import java.util.TreeMap;
import javax.annotation.Nullable;
import org.apache.commons.configuration2.Configuration;
+import org.apache.commons.lang3.ArrayUtils;
import org.apache.pinot.common.request.context.ExpressionContext;
import org.apache.pinot.segment.local.aggregator.ValueAggregator;
import org.apache.pinot.segment.local.aggregator.ValueAggregatorFactory;
import org.apache.pinot.segment.local.segment.creator.impl.fwd.SingleValueFixedByteRawIndexCreator;
import org.apache.pinot.segment.local.segment.creator.impl.fwd.SingleValueUnsortedForwardIndexCreator;
import org.apache.pinot.segment.local.segment.creator.impl.fwd.SingleValueVarByteRawIndexCreator;
+import org.apache.pinot.segment.local.segment.creator.impl.nullvalue.NullValueVectorCreator;
import org.apache.pinot.segment.local.segment.readers.PinotSegmentColumnReader;
import org.apache.pinot.segment.local.startree.StarTreeBuilderUtils;
import org.apache.pinot.segment.local.startree.StarTreeBuilderUtils.TreeNode;
@@ -44,11 +46,13 @@ import org.apache.pinot.segment.spi.AggregationFunctionType;
import org.apache.pinot.segment.spi.ImmutableSegment;
import org.apache.pinot.segment.spi.compression.ChunkCompressionType;
import org.apache.pinot.segment.spi.index.creator.ForwardIndexCreator;
+import org.apache.pinot.segment.spi.index.reader.Dictionary;
import org.apache.pinot.segment.spi.index.startree.AggregationFunctionColumnPair;
import org.apache.pinot.segment.spi.index.startree.AggregationSpec;
import org.apache.pinot.segment.spi.index.startree.StarTreeNode;
import org.apache.pinot.segment.spi.index.startree.StarTreeV2Constants;
import org.apache.pinot.segment.spi.memory.PinotDataBuffer;
+import org.apache.pinot.spi.data.FieldSpec;
import org.apache.pinot.spi.data.FieldSpec.DataType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -71,6 +75,18 @@ abstract class BaseSingleTreeBuilder implements SingleTreeBuilder {
final String[] _dimensionsSplitOrder;
final Set<Integer> _skipStarNodeCreationForDimensions;
final PinotSegmentColumnReader[] _dimensionReaders;
+ // Cardinality of each dimension's segment dictionary. In a null-aware star-tree this doubles as the dictionary id
+ // reserved for null values while records are built and the tree is split: one past the last real id, so nulls sort
+ // after every real value and form their own tree node instead of folding into the column's default null value. The
+ // reserved id is mapped back to the default null value's id when the forward index is written.
+ final int[] _dimensionCardinalities;
+ /// Dictionary id of each dimension's default null value, or `null` when null handling is disabled.
+ ///
+ /// A null row is stored the way a regular column stores one: the default null value in the forward index, and the
+ /// row marked in the null vector. Only read for a dimension that has a null row, which is exactly when the segment
+ /// dictionary holds the default null value.
+ @Nullable
+ private final int[] _defaultNullValueDictIds;
final int _numMetrics;
// Name of the function-column pairs
@@ -79,8 +95,14 @@ abstract class BaseSingleTreeBuilder implements SingleTreeBuilder {
// Readers and data types for column in function-column pair
final PinotSegmentColumnReader[] _metricReaders;
final AggregationSpec[] _aggregationSpecs;
+ /// Whether each metric has produced an aggregated value. A null-aware star-tree leaves this `false` for a metric
+ /// whose every group aggregated over null input only, which is a metric with no value to size its column from.
+ ///
+ /// Set where raw values are first aggregated; a star record only re-aggregates values that already passed there.
+ private final boolean[] _metricHasAggregatedValue;
final int _maxLeafRecords;
+ final boolean _nullHandlingEnabled;
final TreeNode _rootNode = getNewNode();
@@ -110,12 +132,15 @@ abstract class BaseSingleTreeBuilder implements SingleTreeBuilder {
_outputDir = outputDir;
_segment = segment;
_metadataProperties = metadataProperties;
+ _nullHandlingEnabled = builderConfig.isNullHandlingEnabled();
List<String> dimensionsSplitOrder = builderConfig.getDimensionsSplitOrder();
_numDimensions = dimensionsSplitOrder.size();
_dimensionsSplitOrder = new String[_numDimensions];
_skipStarNodeCreationForDimensions = new HashSet<>();
_dimensionReaders = new PinotSegmentColumnReader[_numDimensions];
+ _dimensionCardinalities = new int[_numDimensions];
+ _defaultNullValueDictIds = _nullHandlingEnabled ? new int[_numDimensions] : null;
Set<String> skipStarNodeCreationForDimensions = builderConfig.getSkipStarNodeCreationForDimensions();
for (int i = 0; i < _numDimensions; i++) {
String dimension = dimensionsSplitOrder.get(i);
@@ -126,12 +151,19 @@ abstract class BaseSingleTreeBuilder implements SingleTreeBuilder {
_dimensionReaders[i] = new PinotSegmentColumnReader(segment, dimension);
Preconditions.checkState(_dimensionReaders[i].hasDictionary(),
"Dimension: " + dimension + " does not have dictionary");
+ Dictionary dictionary = segment.getDictionary(dimension);
+ _dimensionCardinalities[i] = dictionary.length();
+ if (_nullHandlingEnabled) {
+ FieldSpec fieldSpec = segment.getSegmentMetadata().getSchema().getFieldSpecFor(dimension);
+ _defaultNullValueDictIds[i] = dictionary.indexOf(FieldSpec.getStringValue(fieldSpec.getDefaultNullValue()));
+ }
}
TreeMap<AggregationFunctionColumnPair, AggregationSpec> aggregationSpecs = builderConfig.getAggregationSpecs();
_numMetrics = aggregationSpecs.size();
_metrics = new String[_numMetrics];
_valueAggregators = new ValueAggregator[_numMetrics];
+ _metricHasAggregatedValue = new boolean[_numMetrics];
_metricReaders = new PinotSegmentColumnReader[_numMetrics];
_aggregationSpecs = new AggregationSpec[_numMetrics];
@@ -145,9 +177,12 @@ abstract class BaseSingleTreeBuilder implements SingleTreeBuilder {
_valueAggregators[index] =
ValueAggregatorFactory.getValueAggregator(functionColumnPair.getFunctionType(), arguments);
_aggregationSpecs[index] = aggregationSpec;
- // Ignore the column for COUNT aggregation function
- if (_valueAggregators[index].getAggregationType() != AggregationFunctionType.COUNT) {
- String column = functionColumnPair.getColumn();
+ // COUNT(*) counts rows rather than values and needs no reader. Every other pair reads a real column, including
+ // the COUNT(column) a null-aware star-tree stores, which counts that column's non-null values. A non-COUNT
+ // pair on STAR is an invalid config, and is left to fail here on the missing column as it always has.
+ String column = functionColumnPair.getColumn();
+ if (_valueAggregators[index].getAggregationType() != AggregationFunctionType.COUNT || !column.equals(
+ AggregationFunctionColumnPair.STAR)) {
PinotSegmentColumnReader metricReader = new PinotSegmentColumnReader(segment, column);
// The distinct arrayAgg star-tree aggregator only supports single-value source columns (dictionary-encoded or
// raw). See ArrayAggDistinctValueAggregator.
@@ -217,8 +252,17 @@ abstract class BaseSingleTreeBuilder implements SingleTreeBuilder {
/// @return Dimensions (dictionary Ids) for a segment record
int[] getSegmentRecordDimensions(int docId) {
int[] dimensions = new int[_numDimensions];
- for (int i = 0; i < _numDimensions; i++) {
- dimensions[i] = _dimensionReaders[i].getDictId(docId);
+ if (_nullHandlingEnabled) {
+ for (int i = 0; i < _numDimensions; i++) {
+ PinotSegmentColumnReader dimensionReader = _dimensionReaders[i];
+ // A null value is stored under the reserved dictionary id instead of the dictionary id of the column's
+ // default null value, so that null rows are not grouped together with rows holding that default value
+ dimensions[i] = dimensionReader.isNull(docId) ? _dimensionCardinalities[i] : dimensionReader.getDictId(docId);
+ }
+ } else {
+ for (int i = 0; i < _numDimensions; i++) {
+ dimensions[i] = _dimensionReaders[i].getDictId(docId);
+ }
}
return dimensions;
}
@@ -228,18 +272,10 @@ abstract class BaseSingleTreeBuilder implements SingleTreeBuilder {
/// @param docId Document Id
/// @return Segment record
Record getSegmentRecord(int docId) {
- int[] dimensions = getSegmentRecordDimensions(docId);
- Object[] metrics = new Object[_numMetrics];
- for (int i = 0; i < _numMetrics; i++) {
- // Ignore the column for COUNT aggregation function
- if (_metricReaders[i] != null) {
- metrics[i] = _metricReaders[i].getValue(docId);
- }
- }
- return new Record(dimensions, metrics);
+ return new Record(getSegmentRecordDimensions(docId), getSegmentRecordMetrics(docId));
}
- /// Same as {@link #getSegmentRecord(int)} but reads dims from `dimBuffer` (row-major, indexed by
+ /// Same as [#getSegmentRecord] but reads dims from `dimBuffer` (row-major, indexed by
/// `docId * _numDimensions * Integer.BYTES`) instead of re-fetching from the segment.
Record getSegmentRecordWithBufferDims(int docId, PinotDataBuffer dimBuffer) {
int[] dimensions = new int[_numDimensions];
@@ -248,14 +284,21 @@ abstract class BaseSingleTreeBuilder implements SingleTreeBuilder {
dimensions[i] = dimBuffer.getInt(off);
off += Integer.BYTES;
}
+ return new Record(dimensions, getSegmentRecordMetrics(docId));
+ }
+
+ /// Returns the raw metric values of a segment record. `COUNT(*)` has no reader and leaves its slot `null`. In a
+ /// null-aware star-tree a null value is passed down as `null` rather than as the column's default null value, so
+ /// that the aggregation excludes it.
+ private Object[] getSegmentRecordMetrics(int docId) {
Object[] metrics = new Object[_numMetrics];
for (int i = 0; i < _numMetrics; i++) {
- // Ignore the column for COUNT aggregation function
- if (_metricReaders[i] != null) {
- metrics[i] = _metricReaders[i].getValue(docId);
+ PinotSegmentColumnReader metricReader = _metricReaders[i];
+ if (metricReader != null) {
+ metrics[i] = _nullHandlingEnabled && metricReader.isNull(docId) ? null : metricReader.getValue(docId);
}
}
- return new Record(dimensions, metrics);
+ return metrics;
}
/// Merges a segment record (raw) into the aggregated record.
@@ -270,24 +313,35 @@ abstract class BaseSingleTreeBuilder implements SingleTreeBuilder {
int[] dimensions = Arrays.copyOf(segmentRecord._dimensions, _numDimensions);
Object[] metrics = new Object[_numMetrics];
for (int i = 0; i < _numMetrics; i++) {
- Object rawValue = segmentRecord._metrics[i];
- if (rawValue != null) {
- metrics[i] = _valueAggregators[i].getInitialAggregatedValue(rawValue);
- } else {
+ if (_metricReaders[i] == null) {
+ // COUNT(*) has no reader and counts every row
assert _valueAggregators[i].getAggregationType() == AggregationFunctionType.COUNT;
metrics[i] = 1L;
+ } else {
+ // A null raw value only occurs in a null-aware star-tree, where it is excluded from the aggregation. The
+ // aggregated value stays null until the group sees its first non-null value.
+ Object rawValue = segmentRecord._metrics[i];
+ metrics[i] = rawValue != null ? _valueAggregators[i].getInitialAggregatedValue(rawValue) : null;
}
+ _metricHasAggregatedValue[i] |= metrics[i] != null;
}
return new Record(dimensions, metrics);
} else {
for (int i = 0; i < _numMetrics; i++) {
- Object rawValue = segmentRecord._metrics[i];
- if (rawValue != null) {
- aggregatedRecord._metrics[i] = _valueAggregators[i].applyRawValue(aggregatedRecord._metrics[i], rawValue);
- } else {
+ if (_metricReaders[i] == null) {
assert _valueAggregators[i].getAggregationType() == AggregationFunctionType.COUNT;
aggregatedRecord._metrics[i] = ((long) aggregatedRecord._metrics[i]) + 1;
+ continue;
+ }
+ Object rawValue = segmentRecord._metrics[i];
+ if (rawValue == null) {
+ continue;
}
+ Object aggregatedValue = aggregatedRecord._metrics[i];
+ aggregatedRecord._metrics[i] =
+ aggregatedValue != null ? _valueAggregators[i].applyRawValue(aggregatedValue, rawValue)
+ : _valueAggregators[i].getInitialAggregatedValue(rawValue);
+ _metricHasAggregatedValue[i] = true;
}
return aggregatedRecord;
}
@@ -305,13 +359,21 @@ abstract class BaseSingleTreeBuilder implements SingleTreeBuilder {
int[] dimensions = Arrays.copyOf(starTreeRecord._dimensions, _numDimensions);
Object[] metrics = new Object[_numMetrics];
for (int i = 0; i < _numMetrics; i++) {
- metrics[i] = _valueAggregators[i].cloneAggregatedValue(starTreeRecord._metrics[i]);
+ // A null value means the group aggregated over no non-null input, which only occurs in a null-aware star-tree
+ Object value = starTreeRecord._metrics[i];
+ metrics[i] = value != null ? _valueAggregators[i].cloneAggregatedValue(value) : null;
}
return new Record(dimensions, metrics);
} else {
for (int i = 0; i < _numMetrics; i++) {
+ Object value = starTreeRecord._metrics[i];
+ if (value == null) {
+ continue;
+ }
+ Object aggregatedValue = aggregatedRecord._metrics[i];
aggregatedRecord._metrics[i] =
- _valueAggregators[i].applyAggregatedValue(aggregatedRecord._metrics[i], starTreeRecord._metrics[i]);
+ aggregatedValue != null ? _valueAggregators[i].applyAggregatedValue(aggregatedValue, value)
+ : _valueAggregators[i].cloneAggregatedValue(value);
}
return aggregatedRecord;
}
@@ -482,10 +544,24 @@ abstract class BaseSingleTreeBuilder implements SingleTreeBuilder {
SingleValueUnsortedForwardIndexCreator[] dimensionIndexCreators =
new SingleValueUnsortedForwardIndexCreator[_numDimensions];
for (int i = 0; i < _numDimensions; i++) {
- String dimension = _dimensionsSplitOrder[i];
- int cardinality = _segment.getDictionary(dimension).length();
dimensionIndexCreators[i] =
- new SingleValueUnsortedForwardIndexCreator(_outputDir, _dimensionsSplitOrder[i], cardinality, _numDocs);
+ new SingleValueUnsortedForwardIndexCreator(_outputDir, _dimensionsSplitOrder[i], _dimensionCardinalities[i],
+ _numDocs);
+ }
+
+ // Null vectors are only created for a null-aware star-tree. Dimensions need one as well as metrics: a null row
+ // is stored as the column's default null value, which is indistinguishable from a row that holds that value.
+ NullValueVectorCreator[] dimensionNullValueVectorCreators = null;
+ NullValueVectorCreator[] metricNullValueVectorCreators = null;
+ if (_nullHandlingEnabled) {
+ dimensionNullValueVectorCreators = new NullValueVectorCreator[_numDimensions];
+ for (int i = 0; i < _numDimensions; i++) {
+ dimensionNullValueVectorCreators[i] = new NullValueVectorCreator(_outputDir, _dimensionsSplitOrder[i]);
+ }
+ metricNullValueVectorCreators = new NullValueVectorCreator[_numMetrics];
+ for (int i = 0; i < _numMetrics; i++) {
+ metricNullValueVectorCreators[i] = new NullValueVectorCreator(_outputDir, _metrics[i]);
+ }
}
ForwardIndexCreator[] metricIndexCreators = new ForwardIndexCreator[_numMetrics];
@@ -498,7 +574,7 @@ abstract class BaseSingleTreeBuilder implements SingleTreeBuilder {
if (valueType == BYTES) {
metricIndexCreators[i] =
new SingleValueVarByteRawIndexCreator(_outputDir, compressionType, metric, _numDocs, BYTES,
- valueAggregator.getMaxAggregatedValueByteSize(), aggregationSpec.isDeriveNumDocsPerChunk(),
+ getMaxAggregatedValueByteSize(i), aggregationSpec.isDeriveNumDocsPerChunk(),
aggregationSpec.getIndexVersion(), aggregationSpec.getTargetMaxChunkSizeBytes(),
aggregationSpec.getTargetDocsPerChunk());
} else {
@@ -513,30 +589,40 @@ abstract class BaseSingleTreeBuilder implements SingleTreeBuilder {
for (int docId = 0; docId < _numDocs; docId++) {
Record record = getStarTreeRecord(docId);
for (int i = 0; i < _numDimensions; i++) {
- dimensionIndexCreators[i].putDictId(record._dimensions[i]);
+ int dictId = record._dimensions[i];
+ // The reserved null dictionary id is out of range for the segment dictionary the star-tree shares, so the
+ // column's default null value is stored in its place and the row is marked in the dimension's null vector,
+ // which is how a regular column stores a null. A value-based read of the forward index then resolves like
+ // any other. Star records hold STAR_IN_FORWARD_INDEX (0) for the starred dimension, which never collides
+ // with the reserved id, so they are correctly left out of the null vector.
+ if (dimensionNullValueVectorCreators != null && dictId == _dimensionCardinalities[i]) {
+ dimensionNullValueVectorCreators[i].setNull(docId);
+ dictId = _defaultNullValueDictIds[i];
+ }
+ dimensionIndexCreators[i].putDictId(dictId);
}
for (int i = 0; i < _numMetrics; i++) {
ValueAggregator valueAggregator = _valueAggregators[i];
- ForwardIndexCreator metricIndexCreator = metricIndexCreators[i];
- switch (valueAggregator.getAggregatedValueType()) {
- case INT:
- metricIndexCreator.putInt((int) record._metrics[i]);
- break;
- case LONG:
- metricIndexCreator.putLong((long) record._metrics[i]);
- break;
- case FLOAT:
- metricIndexCreator.putFloat((float) record._metrics[i]);
- break;
- case DOUBLE:
- metricIndexCreator.putDouble((double) record._metrics[i]);
- break;
- case BYTES:
- metricIndexCreator.putBytes(valueAggregator.serializeAggregatedValue(record._metrics[i]));
- break;
- default:
- throw new IllegalStateException();
+ Object value = record._metrics[i];
+ if (value == null) {
+ // The group aggregated over no non-null input. Only COUNT still has a well-defined result of its own
+ // (0); every other aggregator answers SQL NULL and gets a placeholder in the forward index masked by
+ // the null vector.
+ assert _nullHandlingEnabled;
+ value = valueAggregator.getAllNullAggregatedValue();
+ if (value == null) {
+ metricNullValueVectorCreators[i].setNull(docId);
+ }
}
+ putMetricValue(metricIndexCreators[i], valueAggregator, value);
+ }
+ }
+ if (_nullHandlingEnabled) {
+ for (NullValueVectorCreator nullValueVectorCreator : dimensionNullValueVectorCreators) {
+ nullValueVectorCreator.seal();
+ }
+ for (NullValueVectorCreator nullValueVectorCreator : metricNullValueVectorCreators) {
+ nullValueVectorCreator.seal();
}
}
} catch (Exception e) {
@@ -570,6 +656,50 @@ abstract class BaseSingleTreeBuilder implements SingleTreeBuilder {
}
}
+ /// Returns the maximum serialized size of a metric's aggregated values.
+ ///
+ /// A null-aware star-tree can leave a metric with no aggregated value at all, when every one of its groups
+ /// aggregated over null input only. The aggregator then has no value to derive a size from, and some require one,
+ /// so the size comes from whatever stands in for a null instead: the aggregator's own answer for such a group, or
+ /// the empty placeholder when it has none.
+ private int getMaxAggregatedValueByteSize(int metricId) {
+ ValueAggregator valueAggregator = _valueAggregators[metricId];
+ if (_metricHasAggregatedValue[metricId]) {
+ return valueAggregator.getMaxAggregatedValueByteSize();
+ }
+ Object allNullValue = valueAggregator.getAllNullAggregatedValue();
+ return allNullValue != null ? valueAggregator.serializeAggregatedValue(allNullValue).length : 0;
+ }
+
+ /// Writes an aggregated metric value into the forward index.
+ ///
+ /// A `null` value is a group that aggregates to SQL `NULL`; it is recorded in the metric's null vector and stored
+ /// here as the aggregated type's zero value, which the query side never reads.
+ private static void putMetricValue(ForwardIndexCreator metricIndexCreator, ValueAggregator valueAggregator,
+ @Nullable Object value)
+ throws IOException {
+ switch (valueAggregator.getAggregatedValueType()) {
+ case INT:
+ metricIndexCreator.putInt(value != null ? (int) value : 0);
+ break;
+ case LONG:
+ metricIndexCreator.putLong(value != null ? (long) value : 0L);
+ break;
+ case FLOAT:
+ metricIndexCreator.putFloat(value != null ? (float) value : 0f);
+ break;
+ case DOUBLE:
+ metricIndexCreator.putDouble(value != null ? (double) value : 0d);
+ break;
+ case BYTES:
+ metricIndexCreator.putBytes(
+ value != null ? valueAggregator.serializeAggregatedValue(value) : ArrayUtils.EMPTY_BYTE_ARRAY);
+ break;
+ default:
+ throw new IllegalStateException();
+ }
+ }
+
@Override
public void close()
throws IOException {
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/startree/v2/builder/OffHeapSingleTreeBuilder.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/startree/v2/builder/OffHeapSingleTreeBuilder.java
index f00b7c03823..f9d16f92c02 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/startree/v2/builder/OffHeapSingleTreeBuilder.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/startree/v2/builder/OffHeapSingleTreeBuilder.java
@@ -32,9 +32,11 @@ import java.util.Arrays;
import java.util.Iterator;
import org.apache.commons.configuration2.Configuration;
import org.apache.commons.io.FileUtils;
+import org.apache.commons.lang3.ArrayUtils;
import org.apache.pinot.segment.spi.ImmutableSegment;
import org.apache.pinot.segment.spi.index.startree.StarTreeV2Constants;
import org.apache.pinot.segment.spi.memory.PinotDataBuffer;
+import org.roaringbitmap.RoaringBitmap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -52,6 +54,15 @@ public class OffHeapSingleTreeBuilder extends BaseSingleTreeBuilder {
private final BufferedOutputStream _starTreeRecordOutputStream;
private final RecordOffsets _starTreeRecordOffsets;
+ /// One bitmap per metric holding the doc ids whose group aggregated over no non-null input, or `null` when null
+ /// handling is disabled. A record is stored the way a regular column is: the metric keeps a placeholder in the
+ /// serialized record and its nullness lives beside it, so the record layout is unchanged and the arithmetic in
+ /// [FixedSizeRecordOffsets] still holds.
+ ///
+ /// Keying on the doc id is safe because records are never moved: sorting permutes an array of doc ids and compares
+ /// through it, leaving each record where it was written.
+ private final RoaringBitmap[] _metricNullBitmaps;
+
private PinotDataBuffer _segmentRecordBuffer;
private PinotDataBuffer _starTreeRecordBuffer;
private int _numReadableStarTreeRecords;
@@ -68,6 +79,7 @@ public class OffHeapSingleTreeBuilder extends BaseSingleTreeBuilder {
_starTreeRecordFile);
_starTreeRecordOutputStream = new BufferedOutputStream(new FileOutputStream(_starTreeRecordFile));
_starTreeRecordOffsets = createRecordOffsets();
+ _metricNullBitmaps = _nullHandlingEnabled ? new RoaringBitmap[_numMetrics] : null;
}
/// Returns [FixedSizeRecordOffsets] when all metrics are serialized with a fixed size (see
@@ -90,6 +102,11 @@ public class OffHeapSingleTreeBuilder extends BaseSingleTreeBuilder {
return new FixedSizeRecordOffsets(recordSize);
}
+ /// Serializes a record into the temporary star-tree record store.
+ ///
+ /// A metric that aggregated over no non-null input has no value to write, so the aggregated type's zero is written
+ /// in its place and [#appendRecord] records the nullness in [#_metricNullBitmaps]. The placeholder is never read
+ /// back: [#deserializeStarTreeRecord] consults the bitmap first.
@SuppressWarnings("unchecked")
private byte[] serializeStarTreeRecord(Record starTreeRecord) {
int numBytes = _numDimensions * Integer.BYTES;
@@ -103,7 +120,9 @@ public class OffHeapSingleTreeBuilder extends BaseSingleTreeBuilder {
numBytes += Double.BYTES;
break;
case BYTES:
- metricBytes[i] = _valueAggregators[i].serializeAggregatedValue(starTreeRecord._metrics[i]);
+ Object bytesValue = starTreeRecord._metrics[i];
+ metricBytes[i] = bytesValue != null ? _valueAggregators[i].serializeAggregatedValue(bytesValue)
+ : ArrayUtils.EMPTY_BYTE_ARRAY;
numBytes += Integer.BYTES + metricBytes[i].length;
break;
default:
@@ -116,12 +135,13 @@ public class OffHeapSingleTreeBuilder extends BaseSingleTreeBuilder {
byteBuffer.putInt(dimension);
}
for (int i = 0; i < _numMetrics; i++) {
+ Object value = starTreeRecord._metrics[i];
switch (_valueAggregators[i].getAggregatedValueType()) {
case LONG:
- byteBuffer.putLong((Long) starTreeRecord._metrics[i]);
+ byteBuffer.putLong(value != null ? (long) value : 0L);
break;
case DOUBLE:
- byteBuffer.putDouble((Double) starTreeRecord._metrics[i]);
+ byteBuffer.putDouble(value != null ? (double) value : 0d);
break;
case BYTES:
byteBuffer.putInt(metricBytes[i].length);
@@ -134,7 +154,11 @@ public class OffHeapSingleTreeBuilder extends BaseSingleTreeBuilder {
return bytes;
}
- private Record deserializeStarTreeRecord(PinotDataBuffer buffer, long offset) {
+ /// Deserializes the record at `docId`, whose metrics start at `offset`.
+ ///
+ /// A metric marked null in [#_metricNullBitmaps] comes back as `null` without its placeholder being deserialized,
+ /// which matters for `BYTES`: the placeholder is an empty array that no aggregator can decode.
+ private Record deserializeStarTreeRecord(PinotDataBuffer buffer, long offset, int docId) {
int[] dimensions = new int[_numDimensions];
for (int i = 0; i < _numDimensions; i++) {
dimensions[i] = buffer.getInt(offset);
@@ -144,20 +168,22 @@ public class OffHeapSingleTreeBuilder extends BaseSingleTreeBuilder {
for (int i = 0; i < _numMetrics; i++) {
switch (_valueAggregators[i].getAggregatedValueType()) {
case LONG:
- metrics[i] = buffer.getLong(offset);
+ metrics[i] = hasMetricValue(docId, i) ? buffer.getLong(offset) : null;
offset += Long.BYTES;
break;
case DOUBLE:
- metrics[i] = buffer.getDouble(offset);
+ metrics[i] = hasMetricValue(docId, i) ? buffer.getDouble(offset) : null;
offset += Double.BYTES;
break;
case BYTES:
int numBytes = buffer.getInt(offset);
offset += Integer.BYTES;
- byte[] bytes = new byte[numBytes];
- buffer.copyTo(offset, bytes);
+ if (hasMetricValue(docId, i)) {
+ byte[] bytes = new byte[numBytes];
+ buffer.copyTo(offset, bytes);
+ metrics[i] = _valueAggregators[i].deserializeAggregatedValue(bytes);
+ }
offset += numBytes;
- metrics[i] = _valueAggregators[i].deserializeAggregatedValue(bytes);
break;
default:
throw new IllegalStateException();
@@ -172,13 +198,40 @@ public class OffHeapSingleTreeBuilder extends BaseSingleTreeBuilder {
byte[] bytes = serializeStarTreeRecord(record);
_starTreeRecordOutputStream.write(bytes);
_starTreeRecordOffsets.addRecord(bytes.length);
+ if (_metricNullBitmaps != null) {
+ // The caller assigns this record _numDocs and increments it afterwards, so it is this record's doc id
+ markNullMetrics(record, _numDocs);
+ }
+ }
+
+ /// Records which of the record's metrics aggregated over no non-null input.
+ private void markNullMetrics(Record record, int docId) {
+ for (int i = 0; i < _numMetrics; i++) {
+ if (record._metrics[i] == null) {
+ RoaringBitmap nullBitmap = _metricNullBitmaps[i];
+ if (nullBitmap == null) {
+ nullBitmap = new RoaringBitmap();
+ _metricNullBitmaps[i] = nullBitmap;
+ }
+ nullBitmap.add(docId);
+ }
+ }
+ }
+
+ /// Returns whether the record holds a value for the metric, as opposed to having aggregated over no non-null input.
+ private boolean hasMetricValue(int docId, int metricId) {
+ if (_metricNullBitmaps == null) {
+ return true;
+ }
+ RoaringBitmap nullBitmap = _metricNullBitmaps[metricId];
+ return nullBitmap == null || !nullBitmap.contains(docId);
}
@Override
Record getStarTreeRecord(int docId)
throws IOException {
ensureBufferReadable(docId);
- return deserializeStarTreeRecord(_starTreeRecordBuffer, _starTreeRecordOffsets.getStartOffset(docId));
+ return deserializeStarTreeRecord(_starTreeRecordBuffer, _starTreeRecordOffsets.getStartOffset(docId), docId);
}
@Override
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/startree/v2/builder/StarTreeIndexCombiner.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/startree/v2/builder/StarTreeIndexCombiner.java
index c3f4d685826..64d0804ea5b 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/startree/v2/builder/StarTreeIndexCombiner.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/startree/v2/builder/StarTreeIndexCombiner.java
@@ -62,6 +62,7 @@ public class StarTreeIndexCombiner implements Closeable {
File dimensionIndexFile =
new File(starTreeIndexDir, dimension + V1Constants.Indexes.UNSORTED_SV_FORWARD_INDEX_FILE_EXTENSION);
indexMap.add(Pair.of(new IndexKey(IndexType.FORWARD_INDEX, dimension), writeFile(dimensionIndexFile)));
+ writeNullValueVectorIfExists(starTreeIndexDir, dimension, indexMap);
}
// Write metric (function-column pair) indexes
@@ -70,12 +71,27 @@ public class StarTreeIndexCombiner implements Closeable {
File metricIndexFile =
new File(starTreeIndexDir, metric + V1Constants.Indexes.RAW_SV_FORWARD_INDEX_FILE_EXTENSION);
indexMap.add(Pair.of(new IndexKey(IndexType.FORWARD_INDEX, metric), writeFile(metricIndexFile)));
+ writeNullValueVectorIfExists(starTreeIndexDir, metric, indexMap);
}
FileUtils.cleanDirectory(starTreeIndexDir);
return indexMap;
}
+ /// Writes the null value vector for the given column if the builder created one.
+ ///
+ /// Only a null-aware star-tree creates them, and only for columns that actually contain null values, so a missing
+ /// file simply means there is nothing to record.
+ private void writeNullValueVectorIfExists(File starTreeIndexDir, String column,
+ List<Pair<IndexKey, IndexValue>> indexMap)
+ throws IOException {
+ File nullValueVectorFile =
+ new File(starTreeIndexDir, column + V1Constants.Indexes.NULLVALUE_VECTOR_FILE_EXTENSION);
+ if (nullValueVectorFile.exists()) {
+ indexMap.add(Pair.of(new IndexKey(IndexType.NULL_VALUE_VECTOR, column), writeFile(nullValueVectorFile)));
+ }
+ }
+
private IndexValue writeFile(File srcFile)
throws IOException {
try (FileChannel src = new RandomAccessFile(srcFile, "r").getChannel()) {
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/startree/v2/builder/StarTreeIndexSeparator.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/startree/v2/builder/StarTreeIndexSeparator.java
index 6e28093c9c9..f69505d1174 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/startree/v2/builder/StarTreeIndexSeparator.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/startree/v2/builder/StarTreeIndexSeparator.java
@@ -112,6 +112,11 @@ public class StarTreeIndexSeparator implements Closeable {
destIndexFile = new File(starTreeOutputDir, key._column + suffix);
writeIndexToFile(destIndexFile, indexMap.get(key));
break;
+ case NULL_VALUE_VECTOR:
+ destIndexFile =
+ new File(starTreeOutputDir, key._column + V1Constants.Indexes.NULLVALUE_VECTOR_FILE_EXTENSION);
+ writeIndexToFile(destIndexFile, indexMap.get(key));
+ break;
default:
}
}
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/startree/v2/builder/StarTreeV2BuilderConfig.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/startree/v2/builder/StarTreeV2BuilderConfig.java
index 765177c9acd..d2455145e86 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/startree/v2/builder/StarTreeV2BuilderConfig.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/startree/v2/builder/StarTreeV2BuilderConfig.java
@@ -55,8 +55,10 @@ public class StarTreeV2BuilderConfig {
private final Set<String> _skipStarNodeCreationForDimensions;
private final TreeMap<AggregationFunctionColumnPair, AggregationSpec> _aggregationSpecs;
private final int _maxLeafRecords;
+ private final boolean _nullHandlingEnabled;
public static StarTreeV2BuilderConfig fromIndexConfig(StarTreeIndexConfig indexConfig) {
+ boolean nullHandlingEnabled = indexConfig.isNullHandlingEnabled();
List<String> dimensionsSplitOrder = indexConfig.getDimensionsSplitOrder();
Set<String> skipStarNodeCreationForDimensions;
@@ -72,7 +74,7 @@ public class StarTreeV2BuilderConfig {
if (indexConfig.getFunctionColumnPairs() != null) {
for (String functionColumnPair : indexConfig.getFunctionColumnPairs()) {
AggregationFunctionColumnPair aggregationFunctionColumnPair =
- AggregationFunctionColumnPair.fromColumnName(functionColumnPair);
+ AggregationFunctionColumnPair.fromColumnName(functionColumnPair, nullHandlingEnabled);
AggregationFunctionColumnPair storedType =
AggregationFunctionColumnPair.resolveToStoredType(aggregationFunctionColumnPair);
// If there is already an equivalent functionColumnPair in the map, do not load another.
@@ -83,7 +85,7 @@ public class StarTreeV2BuilderConfig {
if (indexConfig.getAggregationConfigs() != null) {
for (StarTreeAggregationConfig aggregationConfig : indexConfig.getAggregationConfigs()) {
AggregationFunctionColumnPair aggregationFunctionColumnPair =
- AggregationFunctionColumnPair.fromAggregationConfig(aggregationConfig);
+ AggregationFunctionColumnPair.fromAggregationConfig(aggregationConfig, nullHandlingEnabled);
AggregationFunctionColumnPair storedType =
AggregationFunctionColumnPair.resolveToStoredType(aggregationFunctionColumnPair);
// If there is already an equivalent functionColumnPair in the map, do not load another.
@@ -98,13 +100,13 @@ public class StarTreeV2BuilderConfig {
}
return new StarTreeV2BuilderConfig(dimensionsSplitOrder, skipStarNodeCreationForDimensions, aggregationSpecs,
- maxLeafRecords);
+ maxLeafRecords, nullHandlingEnabled);
}
public static StarTreeV2BuilderConfig fromMetadata(StarTreeV2Metadata starTreeV2Metadata) {
return new StarTreeV2BuilderConfig(starTreeV2Metadata.getDimensionsSplitOrder(),
starTreeV2Metadata.getSkipStarNodeCreationForDimensions(), starTreeV2Metadata.getAggregationSpecs(),
- starTreeV2Metadata.getMaxLeafRecords());
+ starTreeV2Metadata.getMaxLeafRecords(), starTreeV2Metadata.isNullHandlingEnabled());
}
/// Generates default config based on the segment metadata.
@@ -173,8 +175,8 @@ public class StarTreeV2BuilderConfig {
AggregationSpec.DEFAULT);
}
- return new StarTreeV2BuilderConfig(dimensionsSplitOrder, Set.of(), aggregationSpecs,
- DEFAULT_MAX_LEAF_RECORDS);
+ return new StarTreeV2BuilderConfig(dimensionsSplitOrder, Set.of(), aggregationSpecs, DEFAULT_MAX_LEAF_RECORDS,
+ false);
}
public static StarTreeV2BuilderConfig generateDefaultConfig(Schema schema, JsonNode columnsMetadata) {
@@ -239,8 +241,8 @@ public class StarTreeV2BuilderConfig {
AggregationSpec.DEFAULT);
}
- return new StarTreeV2BuilderConfig(dimensionsSplitOrder, Set.of(), aggregationSpecs,
- DEFAULT_MAX_LEAF_RECORDS);
+ return new StarTreeV2BuilderConfig(dimensionsSplitOrder, Set.of(), aggregationSpecs, DEFAULT_MAX_LEAF_RECORDS,
+ false);
}
public static Map<String, JsonNode> convertJsonNodeToMap(JsonNode columnsMetadata) {
@@ -253,11 +255,13 @@ public class StarTreeV2BuilderConfig {
}
private StarTreeV2BuilderConfig(List<String> dimensionsSplitOrder, Set<String> skipStarNodeCreationForDimensions,
- TreeMap<AggregationFunctionColumnPair, AggregationSpec> aggregationSpecs, int maxLeafRecords) {
+ TreeMap<AggregationFunctionColumnPair, AggregationSpec> aggregationSpecs, int maxLeafRecords,
+ boolean nullHandlingEnabled) {
_dimensionsSplitOrder = dimensionsSplitOrder;
_skipStarNodeCreationForDimensions = skipStarNodeCreationForDimensions;
_aggregationSpecs = aggregationSpecs;
_maxLeafRecords = maxLeafRecords;
+ _nullHandlingEnabled = nullHandlingEnabled;
}
public List<String> getDimensionsSplitOrder() {
@@ -280,10 +284,16 @@ public class StarTreeV2BuilderConfig {
return _maxLeafRecords;
}
+ /// Returns whether the star-tree should be built with null-aware semantics. See
+ /// [StarTreeIndexConfig#isNullHandlingEnabled].
+ public boolean isNullHandlingEnabled() {
+ return _nullHandlingEnabled;
+ }
+
/// Writes the metadata which is used to initialize the [StarTreeV2Metadata] when loading the segment.
public void writeMetadata(Configuration metadataProperties, int totalDocs) {
StarTreeV2Metadata.writeMetadata(metadataProperties, totalDocs, _dimensionsSplitOrder, _aggregationSpecs,
- _maxLeafRecords, _skipStarNodeCreationForDimensions);
+ _maxLeafRecords, _skipStarNodeCreationForDimensions, _nullHandlingEnabled);
}
@Override
@@ -295,20 +305,23 @@ public class StarTreeV2BuilderConfig {
return false;
}
StarTreeV2BuilderConfig that = (StarTreeV2BuilderConfig) o;
- return _maxLeafRecords == that._maxLeafRecords && Objects.equals(_dimensionsSplitOrder, that._dimensionsSplitOrder)
+ return _maxLeafRecords == that._maxLeafRecords && _nullHandlingEnabled == that._nullHandlingEnabled
+ && Objects.equals(_dimensionsSplitOrder, that._dimensionsSplitOrder)
&& Objects.equals(_skipStarNodeCreationForDimensions, that._skipStarNodeCreationForDimensions)
&& Objects.equals(_aggregationSpecs, that._aggregationSpecs);
}
@Override
public int hashCode() {
- return Objects.hash(_dimensionsSplitOrder, _skipStarNodeCreationForDimensions, _aggregationSpecs, _maxLeafRecords);
+ return Objects.hash(_dimensionsSplitOrder, _skipStarNodeCreationForDimensions, _aggregationSpecs, _maxLeafRecords,
+ _nullHandlingEnabled);
}
@Override
public String toString() {
return new ToStringBuilder(this, ToStringStyle.SHORT_PREFIX_STYLE).append("splitOrder", _dimensionsSplitOrder)
.append("skipStarNodeCreation", _skipStarNodeCreationForDimensions)
- .append("aggregationSpecs", _aggregationSpecs).append("maxLeafRecords", _maxLeafRecords).toString();
+ .append("aggregationSpecs", _aggregationSpecs).append("maxLeafRecords", _maxLeafRecords)
+ .append("nullHandlingEnabled", _nullHandlingEnabled).toString();
}
}
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/startree/v2/store/StarTreeDataSource.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/startree/v2/store/StarTreeDataSource.java
index 0b9df128648..47c38cf68da 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/startree/v2/store/StarTreeDataSource.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/startree/v2/store/StarTreeDataSource.java
@@ -26,19 +26,26 @@ import org.apache.pinot.segment.spi.index.StandardIndexes;
import org.apache.pinot.segment.spi.index.column.ColumnIndexContainer;
import org.apache.pinot.segment.spi.index.reader.Dictionary;
import org.apache.pinot.segment.spi.index.reader.ForwardIndexReader;
+import org.apache.pinot.segment.spi.index.reader.NullValueVectorReader;
import org.apache.pinot.segment.spi.partition.PartitionFunction;
import org.apache.pinot.spi.data.FieldSpec;
public class StarTreeDataSource extends BaseDataSource {
+ /// Creates a data source over a star-tree column.
+ ///
+ /// `nullValueVector` is only non-null for a null-aware star-tree column that contains null values. Dimensions need
+ /// one just as metrics do: a null is stored in the forward index as the column's default null value, which is
+ /// indistinguishable from a genuine occurrence of that value, so the query side recognizes nulls from the vector.
public StarTreeDataSource(FieldSpec fieldSpec, int numDocs, ForwardIndexReader<?> forwardIndex,
- @Nullable Dictionary dictionary) {
+ @Nullable Dictionary dictionary, @Nullable NullValueVectorReader nullValueVector) {
super(
new StarTreeDataSourceMetadata(fieldSpec, numDocs),
new ColumnIndexContainer.FromMap.Builder()
.with(StandardIndexes.forward(), forwardIndex)
.with(StandardIndexes.dictionary(), dictionary)
+ .with(StandardIndexes.nullValueVector(), nullValueVector)
.build());
}
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/startree/v2/store/StarTreeIndexMapUtils.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/startree/v2/store/StarTreeIndexMapUtils.java
index 25217b881e1..032793f58df 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/startree/v2/store/StarTreeIndexMapUtils.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/startree/v2/store/StarTreeIndexMapUtils.java
@@ -62,6 +62,9 @@ import org.apache.pinot.spi.env.CommonsConfigurationUtils;
/// - 1.null.STAR_TREE.OFFSET = 5500
/// - 1.null.STAR_TREE.SIZE = 2500
/// - ...
+///
+/// A null-aware star-tree additionally stores a NULL_VALUE_VECTOR entry per dimension and metric that has at least one
+/// null value, e.g. `0.sum__metric.NULL_VALUE_VECTOR.OFFSET`.
public class StarTreeIndexMapUtils {
private StarTreeIndexMapUtils() {
}
@@ -75,7 +78,9 @@ public class StarTreeIndexMapUtils {
/// Type of the index.
public enum IndexType {
- STAR_TREE, FORWARD_INDEX
+ // NOTE: NULL_VALUE_VECTOR is only written by a null-aware star-tree. A server running a version that predates it
+ // fails to parse the index map of such a segment, so it must not be enabled before the whole cluster is upgraded.
+ STAR_TREE, FORWARD_INDEX, NULL_VALUE_VECTOR
}
/// Key of the index map.
@@ -189,10 +194,13 @@ public class StarTreeIndexMapUtils {
}
// Convert metric (function-column pair) to stored name for backward-compatibility
if (!dimensionSet.contains(column)) {
- AggregationFunctionColumnPair functionColumnPair = AggregationFunctionColumnPair.fromColumnName(column);
+ // A count__column pair is only ever stored by a null-aware star-tree, where it holds the count of the
+ // non-null values of that column rather than the row count
+ AggregationFunctionColumnPair functionColumnPair = AggregationFunctionColumnPair.fromColumnName(column,
+ starTreeMetadataList.get(starTreeId).isNullHandlingEnabled());
column = AggregationFunctionColumnPair.resolveToStoredType(functionColumnPair).toColumnName();
}
- indexKey = new IndexKey(IndexType.FORWARD_INDEX, column);
+ indexKey = new IndexKey(indexType, column);
}
long value = configuration.getLong(key);
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/startree/v2/store/StarTreeLoaderUtils.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/startree/v2/store/StarTreeLoaderUtils.java
index 07e1a8c7d1f..86184fe38a7 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/startree/v2/store/StarTreeLoaderUtils.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/startree/v2/store/StarTreeLoaderUtils.java
@@ -23,8 +23,10 @@ import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import javax.annotation.Nullable;
import org.apache.pinot.segment.local.aggregator.ValueAggregatorFactory;
import org.apache.pinot.segment.local.segment.index.forward.ForwardIndexReaderFactory;
+import org.apache.pinot.segment.local.segment.index.readers.NullValueVectorReaderImpl;
import org.apache.pinot.segment.local.segment.index.readers.forward.FixedBitSVForwardIndexReaderV2;
import org.apache.pinot.segment.local.startree.OffHeapStarTree;
import org.apache.pinot.segment.local.startree.StarTreeBuilderUtils;
@@ -34,6 +36,7 @@ import org.apache.pinot.segment.spi.index.StandardIndexes;
import org.apache.pinot.segment.spi.index.column.ColumnIndexContainer;
import org.apache.pinot.segment.spi.index.metadata.SegmentMetadataImpl;
import org.apache.pinot.segment.spi.index.reader.ForwardIndexReader;
+import org.apache.pinot.segment.spi.index.reader.NullValueVectorReader;
import org.apache.pinot.segment.spi.index.startree.AggregationFunctionColumnPair;
import org.apache.pinot.segment.spi.index.startree.StarTree;
import org.apache.pinot.segment.spi.index.startree.StarTreeV2;
@@ -80,6 +83,7 @@ public class StarTreeLoaderUtils {
StarTree starTree = new OffHeapStarTree(indexReader.getIndexFor(String.valueOf(i), StandardIndexes.inverted()));
int numDocs = starTreeMetadata.getNumDocs();
+ boolean nullHandlingEnabled = starTreeMetadata.isNullHandlingEnabled();
Map<String, DataSource> dataSourceMap = new HashMap<>();
// Load dimension forward indexes
@@ -89,7 +93,8 @@ public class StarTreeLoaderUtils {
FixedBitSVForwardIndexReaderV2 forwardIndex =
new FixedBitSVForwardIndexReaderV2(forwardIndexDataBuffer, numDocs, columnMetadata.getBitsPerElement());
dataSourceMap.put(dimension, new StarTreeDataSource(columnMetadata.getFieldSpec(), numDocs, forwardIndex,
- indexContainerMap.get(dimension).getIndex(StandardIndexes.dictionary())));
+ indexContainerMap.get(dimension).getIndex(StandardIndexes.dictionary()),
+ loadNullValueVector(indexReader, dimension, nullHandlingEnabled)));
}
// Load metric (function-column pair) forward indexes
@@ -100,7 +105,8 @@ public class StarTreeLoaderUtils {
FieldSpec fieldSpec = new MetricFieldSpec(metric, dataType);
ForwardIndexReader<?> forwardIndex = ForwardIndexReaderFactory.getInstance()
.createRawIndexReader(forwardIndexDataBuffer, dataType.getStoredType(), true);
- dataSourceMap.put(metric, new StarTreeDataSource(fieldSpec, numDocs, forwardIndex, null));
+ dataSourceMap.put(metric, new StarTreeDataSource(fieldSpec, numDocs, forwardIndex, null,
+ loadNullValueVector(indexReader, metric, nullHandlingEnabled)));
}
starTrees.add(new StarTreeV2() {
@@ -132,4 +138,18 @@ public class StarTreeLoaderUtils {
}
return starTrees;
}
+
+ /// Returns the null value vector of a null-aware star-tree column, or `null` when the column has none.
+ ///
+ /// A regular star-tree never stores one, and a null-aware star-tree only stores one for columns that actually
+ /// contain null values.
+ @Nullable
+ private static NullValueVectorReader loadNullValueVector(SegmentDirectory.Reader indexReader, String column,
+ boolean nullHandlingEnabled)
+ throws IOException {
+ if (!nullHandlingEnabled || !indexReader.hasIndexFor(column, StandardIndexes.nullValueVector())) {
+ return null;
+ }
+ return new NullValueVectorReaderImpl(indexReader.getIndexFor(column, StandardIndexes.nullValueVector()));
+ }
}
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java
index 8ef3c48727e..2ece7905724 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java
@@ -2000,11 +2000,14 @@ public final class TableConfigUtils {
Set<AggregationFunctionColumnPair> functionColumnPairsSet = new HashSet<>();
Set<AggregationFunctionColumnPair> storedTypes = new HashSet<>();
Set<String> aggregatedColumns = new HashSet<>();
+ // A null-aware star-tree stores COUNT per column rather than collapsing every COUNT to COUNT(*), so the column
+ // of a COUNT pair has to be kept and validated against the schema
+ boolean nullHandlingEnabled = starTreeIndexConfig.isNullHandlingEnabled();
if (functionColumnPairs != null) {
for (String functionColumnPair : functionColumnPairs) {
AggregationFunctionColumnPair columnPair;
try {
- columnPair = AggregationFunctionColumnPair.fromColumnName(functionColumnPair);
+ columnPair = AggregationFunctionColumnPair.fromColumnName(functionColumnPair, nullHandlingEnabled);
} catch (Exception e) {
throw new IllegalStateException("Invalid StarTreeIndex config: " + functionColumnPair + ". Must be"
+ "in the form <Aggregation function>__<Column name>");
@@ -2032,7 +2035,7 @@ public final class TableConfigUtils {
for (StarTreeAggregationConfig aggregationConfig : aggregationConfigs) {
AggregationFunctionColumnPair columnPair;
try {
- columnPair = AggregationFunctionColumnPair.fromAggregationConfig(aggregationConfig);
+ columnPair = AggregationFunctionColumnPair.fromAggregationConfig(aggregationConfig, nullHandlingEnabled);
} catch (Exception e) {
throw new IllegalStateException("Invalid StarTreeIndex config: " + aggregationConfig);
}
diff --git a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/startree/v2/builder/NullAwareStarTreeBuilderTest.java b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/startree/v2/builder/NullAwareStarTreeBuilderTest.java
new file mode 100644
index 00000000000..11dd454ad0f
--- /dev/null
+++ b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/startree/v2/builder/NullAwareStarTreeBuilderTest.java
@@ -0,0 +1,230 @@
+/**
+ * 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.segment.local.startree.v2.builder;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.segment.local.indexsegment.immutable.ImmutableSegmentLoader;
+import org.apache.pinot.segment.local.segment.creator.impl.SegmentIndexCreationDriverImpl;
+import org.apache.pinot.segment.local.segment.readers.GenericRowRecordReader;
+import org.apache.pinot.segment.local.startree.v2.builder.MultipleTreesBuilder.BuildMode;
+import org.apache.pinot.segment.spi.AggregationFunctionType;
+import org.apache.pinot.segment.spi.ImmutableSegment;
+import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig;
+import org.apache.pinot.segment.spi.index.reader.ForwardIndexReader;
+import org.apache.pinot.segment.spi.index.reader.ForwardIndexReaderContext;
+import org.apache.pinot.segment.spi.index.reader.NullValueVectorReader;
+import org.apache.pinot.segment.spi.index.startree.AggregationFunctionColumnPair;
+import org.apache.pinot.segment.spi.index.startree.StarTreeV2;
+import org.apache.pinot.spi.config.table.StarTreeIndexConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.data.readers.GenericRow;
+import org.apache.pinot.spi.utils.ReadMode;
+import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
+import org.roaringbitmap.buffer.ImmutableRoaringBitmap;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.*;
+
+
+/// Builds star-trees over a segment whose metric is null for a whole group.
+///
+/// The group with no non-null input is the case a null-aware star-tree exists to represent, and it is the case the
+/// regular star-tree cannot distinguish from one that genuinely aggregated the column's default null value.
+///
+/// The build mode is a parameter rather than a random choice, because the two builders represent an all-null group
+/// differently: the on-heap one keeps the aggregated value in memory where `null` needs no encoding, while the
+/// off-heap one serializes every record to a temporary store and has to carry the nullness beside it.
+public class NullAwareStarTreeBuilderTest {
+ private static final File TEMP_DIR = new File(FileUtils.getTempDirectory(), "NullAwareStarTreeBuilderTest");
+ private static final String TABLE_NAME = "testTable";
+ private static final String SEGMENT_NAME = "testSegment";
+ private static final String DIMENSION = "d";
+ private static final String METRIC = "m";
+
+ /// One record per leaf, so every parent node has to aggregate rather than pass a single record through.
+ private static final int MAX_LEAF_RECORDS = 1;
+
+ /// `d = 0` holds only null metrics, `d = 1` holds only non-null ones. The default null value of an `INT` metric is
+ /// `0`, which is below both real values, so a regular star-tree reports `0` as the minimum where a null-aware one
+ /// reports `10`.
+ private static final int[] DIMENSION_VALUES = {0, 0, 1, 1};
+ private static final Integer[] METRIC_VALUES = {null, null, 10, 20};
+
+ private static final String MIN_COLUMN =
+ new AggregationFunctionColumnPair(AggregationFunctionType.MIN, METRIC).toColumnName();
+
+ @AfterMethod
+ public void cleanUp()
+ throws IOException {
+ FileUtils.deleteDirectory(TEMP_DIR);
+ }
+
+ @DataProvider(name = "buildModes")
+ public Object[][] buildModes() {
+ return new Object[][]{{BuildMode.ON_HEAP}, {BuildMode.OFF_HEAP}};
+ }
+
+ /// A group with no non-null input has no aggregated value to store, and the off-heap builder used to have nowhere
+ /// to put that: it serialized every metric unconditionally and dereferenced the missing value.
+ @Test(dataProvider = "buildModes")
+ public void anAllNullGroupSurvivesTheBuild(BuildMode buildMode)
+ throws Exception {
+ File indexDir = createSegment();
+ buildStarTrees(indexDir, buildMode, starTreeConfig(true));
+
+ ImmutableSegment segment = ImmutableSegmentLoader.load(indexDir, ReadMode.mmap);
+ try {
+ StarTreeV2 starTree = segment.getStarTrees().get(0);
+ assertTrue(starTree.getMetadata().isNullHandlingEnabled());
+
+ // The all-null group is recorded in the metric's null vector rather than as an aggregated value
+ ImmutableRoaringBitmap nullBitmap = nullBitmap(starTree, MIN_COLUMN);
+ assertNotNull(nullBitmap, "A null-aware star-tree must write a null vector for a metric with an all-null group");
+ assertFalse(nullBitmap.isEmpty());
+
+ // Every group that did aggregate something reports a real minimum, never the column's default null value
+ List<Double> minimums = nonNullValues(starTree, MIN_COLUMN, nullBitmap);
+ assertFalse(minimums.isEmpty(), "Some group must have aggregated a value, or the check below proves nothing");
+ for (double minimum : minimums) {
+ assertEquals(minimum, 10.0, "A null row must not be aggregated as the column default");
+ }
+ } finally {
+ segment.destroy();
+ }
+ }
+
+ /// The two variants answer differently, so a segment has to be able to hold both: the builder config's identity
+ /// includes the flag, which is what stops one being reused for the other.
+ @Test(dataProvider = "buildModes")
+ public void bothVariantsCoexistAndDisagreeOnNulls(BuildMode buildMode)
+ throws Exception {
+ File indexDir = createSegment();
+ buildStarTrees(indexDir, buildMode, starTreeConfig(false), starTreeConfig(true));
+
+ ImmutableSegment segment = ImmutableSegmentLoader.load(indexDir, ReadMode.mmap);
+ try {
+ List<StarTreeV2> starTrees = segment.getStarTrees();
+ assertEquals(starTrees.size(), 2, "The two variants differ only by the flag, and must not be deduplicated");
+
+ StarTreeV2 regular = variant(starTrees, false);
+ StarTreeV2 nullAware = variant(starTrees, true);
+
+ // Same shape: the flag is the only difference between the two configs
+ assertEquals(regular.getMetadata().getDimensionsSplitOrder(), nullAware.getMetadata().getDimensionsSplitOrder());
+ assertEquals(regular.getMetadata().getFunctionColumnPairs(), nullAware.getMetadata().getFunctionColumnPairs());
+
+ // The regular tree folds nulls into the column default, so it has no null vector and reports that default
+ assertNull(nullBitmap(regular, MIN_COLUMN), "A regular star-tree does not write a null vector");
+ assertTrue(nonNullValues(regular, MIN_COLUMN, null).contains(0.0),
+ "A regular star-tree aggregates a null row as the column's default null value");
+
+ // The null-aware tree excludes them, so the default never appears as a minimum
+ ImmutableRoaringBitmap nullBitmap = nullBitmap(nullAware, MIN_COLUMN);
+ assertNotNull(nullBitmap);
+ assertFalse(nonNullValues(nullAware, MIN_COLUMN, nullBitmap).contains(0.0),
+ "A null-aware star-tree excludes null rows from the pre-aggregation");
+ } finally {
+ segment.destroy();
+ }
+ }
+
+ private static StarTreeV2 variant(List<StarTreeV2> starTrees, boolean nullHandlingEnabled) {
+ for (StarTreeV2 starTree : starTrees) {
+ if (starTree.getMetadata().isNullHandlingEnabled() == nullHandlingEnabled) {
+ return starTree;
+ }
+ }
+ throw new AssertionError("No star-tree with nullHandlingEnabled=" + nullHandlingEnabled);
+ }
+
+ private static ImmutableRoaringBitmap nullBitmap(StarTreeV2 starTree, String column) {
+ NullValueVectorReader nullValueVector = starTree.getDataSource(column).getNullValueVector();
+ return nullValueVector != null ? nullValueVector.getNullBitmap() : null;
+ }
+
+ /// Returns the pre-aggregated values of every doc the null vector does not mark, so that a placeholder left behind
+ /// for a null group is never read as data.
+ private static List<Double> nonNullValues(StarTreeV2 starTree, String column, ImmutableRoaringBitmap nullBitmap)
+ throws IOException {
+ List<Double> values = new ArrayList<>();
+ ForwardIndexReader reader = starTree.getDataSource(column).getForwardIndex();
+ assertNotNull(reader);
+ try (ForwardIndexReaderContext context = reader.createContext()) {
+ for (int docId = 0; docId < starTree.getMetadata().getNumDocs(); docId++) {
+ if (nullBitmap == null || !nullBitmap.contains(docId)) {
+ values.add(reader.getDouble(docId, context));
+ }
+ }
+ }
+ return values;
+ }
+
+ private static StarTreeIndexConfig starTreeConfig(boolean nullHandlingEnabled) {
+ return new StarTreeIndexConfig(List.of(DIMENSION), null,
+ List.of(new AggregationFunctionColumnPair(AggregationFunctionType.MIN, METRIC).toColumnName()), null,
+ MAX_LEAF_RECORDS, nullHandlingEnabled);
+ }
+
+ private static void buildStarTrees(File indexDir, BuildMode buildMode, StarTreeIndexConfig... configs)
+ throws Exception {
+ try (MultipleTreesBuilder builder = new MultipleTreesBuilder(List.of(configs), false, indexDir, buildMode)) {
+ builder.build();
+ }
+ }
+
+ private static File createSegment()
+ throws Exception {
+ Schema schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
+ .addSingleValueDimension(DIMENSION, DataType.INT)
+ .addMetric(METRIC, DataType.INT)
+ .build();
+ TableConfig tableConfig =
+ new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).setNullHandlingEnabled(true).build();
+
+ List<GenericRow> rows = new ArrayList<>(DIMENSION_VALUES.length);
+ for (int i = 0; i < DIMENSION_VALUES.length; i++) {
+ GenericRow row = new GenericRow();
+ row.putValue(DIMENSION, DIMENSION_VALUES[i]);
+ if (METRIC_VALUES[i] != null) {
+ row.putValue(METRIC, METRIC_VALUES[i]);
+ } else {
+ // Ingestion stores the column default and records the row in the null vector
+ row.putDefaultNullValue(METRIC, schema.getFieldSpecFor(METRIC).getDefaultNullValue());
+ }
+ rows.add(row);
+ }
+
+ SegmentGeneratorConfig segmentGeneratorConfig = new SegmentGeneratorConfig(tableConfig, schema);
+ segmentGeneratorConfig.setOutDir(TEMP_DIR.getPath());
+ segmentGeneratorConfig.setSegmentName(SEGMENT_NAME);
+ SegmentIndexCreationDriverImpl driver = new SegmentIndexCreationDriverImpl();
+ driver.init(segmentGeneratorConfig, new GenericRowRecordReader(rows));
+ driver.build();
+ return new File(TEMP_DIR, SEGMENT_NAME);
+ }
+}
diff --git a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/startree/v2/builder/StarTreeBuilderUtilsTest.java b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/startree/v2/builder/StarTreeBuilderUtilsTest.java
index c9480238963..8a819f2e2cb 100644
--- a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/startree/v2/builder/StarTreeBuilderUtilsTest.java
+++ b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/startree/v2/builder/StarTreeBuilderUtilsTest.java
@@ -214,7 +214,7 @@ public class StarTreeBuilderUtilsTest {
TreeMap<AggregationFunctionColumnPair, AggregationSpec> aggregationSpecs = new TreeMap<>();
aggregationSpecs.put(new AggregationFunctionColumnPair(AggregationFunctionType.DISTINCTCOUNTHLL, "col2"),
new AggregationSpec(null, null, null, null, null, Map.of(Constants.HLL_LOG2M_KEY, 16)));
- StarTreeV2Metadata.writeMetadata(metadataProperties, 1, List.of("col1"), aggregationSpecs, 100, Set.of());
+ StarTreeV2Metadata.writeMetadata(metadataProperties, 1, List.of("col1"), aggregationSpecs, 100, Set.of(), false);
StarTreeV2Metadata existingStarTreeMetadata = new StarTreeV2Metadata(metadataProperties);
StarTreeIndexConfig starTreeIndexConfig = new StarTreeIndexConfig(List.of("col1"), null, null, List.of(
diff --git a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/startree/AggregationFunctionColumnPair.java b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/startree/AggregationFunctionColumnPair.java
index 8c5a3347f48..60801ab5bab 100644
--- a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/startree/AggregationFunctionColumnPair.java
+++ b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/startree/AggregationFunctionColumnPair.java
@@ -34,14 +34,29 @@ public class AggregationFunctionColumnPair implements Comparable<AggregationFunc
private final String _column;
public AggregationFunctionColumnPair(AggregationFunctionType functionType, String column) {
+ this(functionType, column, false);
+ }
+
+ private AggregationFunctionColumnPair(AggregationFunctionType functionType, String column,
+ boolean preserveCountColumn) {
_functionType = functionType;
- if (functionType == AggregationFunctionType.COUNT) {
+ // A regular star-tree counts rows rather than values, so every COUNT collapses to COUNT(*). A null-aware star-tree
+ // instead stores the count of the non-null values of a specific column, which requires keeping the column name.
+ if (functionType == AggregationFunctionType.COUNT && !preserveCountColumn) {
_column = STAR;
} else {
_column = column;
}
}
+ /// Returns the pair representing `COUNT(column)`, counting only the non-null values of the column.
+ ///
+ /// Unlike the regular constructor, which normalizes every `COUNT` to [#COUNT_STAR], this keeps the column so that a
+ /// null-aware star-tree can pre-aggregate per-column non-null counts.
+ public static AggregationFunctionColumnPair countColumn(String column) {
+ return new AggregationFunctionColumnPair(AggregationFunctionType.COUNT, column, true);
+ }
+
public AggregationFunctionType getFunctionType() {
return _functionType;
}
@@ -59,12 +74,23 @@ public class AggregationFunctionColumnPair implements Comparable<AggregationFunc
}
public static AggregationFunctionColumnPair fromColumnName(String columnName) {
+ return fromColumnName(columnName, false);
+ }
+
+ /// Parses a function-column pair name such as `sum__col`.
+ ///
+ /// When `preserveCountColumn` is `false`, `count__col` resolves to [#COUNT_STAR], matching how a regular star-tree
+ /// stores counts. Pass `true` for a null-aware star-tree, where `count__col` denotes the non-null count of `col`.
+ public static AggregationFunctionColumnPair fromColumnName(String columnName, boolean preserveCountColumn) {
String[] parts = columnName.split(DELIMITER, 2);
- return fromFunctionAndColumnName(parts[0], parts[1]);
+ return fromFunctionAndColumnName(parts[0], parts[1], preserveCountColumn);
}
- public static AggregationFunctionColumnPair fromAggregationConfig(StarTreeAggregationConfig aggregationConfig) {
- return fromFunctionAndColumnName(aggregationConfig.getAggregationFunction(), aggregationConfig.getColumnName());
+ /// Builds a pair from an aggregation config. See [#fromColumnName] for the meaning of `preserveCountColumn`.
+ public static AggregationFunctionColumnPair fromAggregationConfig(StarTreeAggregationConfig aggregationConfig,
+ boolean preserveCountColumn) {
+ return fromFunctionAndColumnName(aggregationConfig.getAggregationFunction(), aggregationConfig.getColumnName(),
+ preserveCountColumn);
}
/// Return a new `AggregationFunctionColumnPair` from an existing functionColumnPair where the new pair
@@ -72,7 +98,13 @@ public class AggregationFunctionColumnPair implements Comparable<AggregationFunc
/// @param functionColumnPair the existing functionColumnPair
/// @return the new functionColumnPair
public static AggregationFunctionColumnPair resolveToStoredType(AggregationFunctionColumnPair functionColumnPair) {
- AggregationFunctionType storedType = getStoredType(functionColumnPair.getFunctionType());
+ AggregationFunctionType functionType = functionColumnPair.getFunctionType();
+ AggregationFunctionType storedType = getStoredType(functionType);
+ // Already in stored form. Returning it as-is also preserves the column of a per-column COUNT, which the
+ // constructor would otherwise normalize back to STAR.
+ if (storedType == functionType) {
+ return functionColumnPair;
+ }
return new AggregationFunctionColumnPair(storedType, functionColumnPair.getColumn());
}
@@ -114,12 +146,13 @@ public class AggregationFunctionColumnPair implements Comparable<AggregationFunc
}
}
- private static AggregationFunctionColumnPair fromFunctionAndColumnName(String functionName, String columnName) {
+ private static AggregationFunctionColumnPair fromFunctionAndColumnName(String functionName, String columnName,
+ boolean preserveCountColumn) {
AggregationFunctionType functionType = AggregationFunctionType.getAggregationFunctionType(functionName);
- if (functionType == AggregationFunctionType.COUNT) {
+ if (functionType == AggregationFunctionType.COUNT && !preserveCountColumn) {
return COUNT_STAR;
} else {
- return new AggregationFunctionColumnPair(functionType, columnName);
+ return new AggregationFunctionColumnPair(functionType, columnName, preserveCountColumn);
}
}
diff --git a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/startree/StarTreeV2Constants.java b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/startree/StarTreeV2Constants.java
index 2bd582a05e0..01f26a068ed 100644
--- a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/startree/StarTreeV2Constants.java
+++ b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/startree/StarTreeV2Constants.java
@@ -47,6 +47,8 @@ public class StarTreeV2Constants {
public static final String FUNCTION_COLUMN_PAIRS = "function.column.pairs";
public static final String MAX_LEAF_RECORDS = "max.leaf.records";
public static final String SKIP_STAR_NODE_CREATION_FOR_DIMENSIONS = "skip.star.node.creation";
+ // Absent for star-trees built before null-aware star-trees were introduced, which read back as not null-aware
+ public static final String NULL_HANDLING_ENABLED = "null.handling.enabled";
public static final String AGGREGATION_COUNT = "aggregation.count";
public static final String AGGREGATION_PREFIX = "aggregation.";
public static final String FUNCTION_TYPE = "function.type";
diff --git a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/startree/StarTreeV2Metadata.java b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/startree/StarTreeV2Metadata.java
index a67b1f55fc2..6b86196a060 100644
--- a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/startree/StarTreeV2Metadata.java
+++ b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/startree/StarTreeV2Metadata.java
@@ -42,10 +42,13 @@ public class StarTreeV2Metadata {
// The following properties are useful for generating the builder config
private final int _maxLeafRecords;
private final Set<String> _skipStarNodeCreationForDimensions;
+ private final boolean _nullHandlingEnabled;
public StarTreeV2Metadata(Configuration metadataProperties) {
_numDocs = metadataProperties.getInt(MetadataKey.TOTAL_DOCS);
_dimensionsSplitOrder = Arrays.asList(metadataProperties.getStringArray(MetadataKey.DIMENSIONS_SPLIT_ORDER));
+ // NOTE: Must be read before the aggregation specs, which are parsed differently for a null-aware star-tree
+ _nullHandlingEnabled = metadataProperties.getBoolean(MetadataKey.NULL_HANDLING_ENABLED, false);
_aggregationSpecs = new TreeMap<>();
int numAggregations = metadataProperties.getInt(MetadataKey.AGGREGATION_COUNT, 0);
if (numAggregations > 0) {
@@ -54,7 +57,10 @@ public class StarTreeV2Metadata {
AggregationFunctionType functionType =
AggregationFunctionType.getAggregationFunctionType(aggregationConfig.getString(MetadataKey.FUNCTION_TYPE));
String columnName = aggregationConfig.getString(MetadataKey.COLUMN_NAME);
- AggregationFunctionColumnPair functionColumnPair = new AggregationFunctionColumnPair(functionType, columnName);
+ AggregationFunctionColumnPair functionColumnPair =
+ functionType == AggregationFunctionType.COUNT && _nullHandlingEnabled
+ ? AggregationFunctionColumnPair.countColumn(columnName)
+ : new AggregationFunctionColumnPair(functionType, columnName);
// Lookup the stored aggregation type
AggregationFunctionColumnPair storedType =
AggregationFunctionColumnPair.resolveToStoredType(functionColumnPair);
@@ -79,7 +85,7 @@ public class StarTreeV2Metadata {
// Backward compatibility with columnName format
for (String functionColumnPairName : metadataProperties.getStringArray(MetadataKey.FUNCTION_COLUMN_PAIRS)) {
AggregationFunctionColumnPair functionColumnPair =
- AggregationFunctionColumnPair.fromColumnName(functionColumnPairName);
+ AggregationFunctionColumnPair.fromColumnName(functionColumnPairName, _nullHandlingEnabled);
// Lookup the stored aggregation type
AggregationFunctionColumnPair storedType =
AggregationFunctionColumnPair.resolveToStoredType(functionColumnPair);
@@ -120,9 +126,17 @@ public class StarTreeV2Metadata {
return _skipStarNodeCreationForDimensions;
}
+ /// Returns whether this star-tree was pre-aggregated with null-aware semantics.
+ ///
+ /// A null-aware star-tree stores null dimension values under a dedicated dictionary id and excludes null metric
+ /// values from the pre-aggregation, so it is only consistent with queries that have null handling enabled.
+ public boolean isNullHandlingEnabled() {
+ return _nullHandlingEnabled;
+ }
+
public static void writeMetadata(Configuration metadataProperties, int totalDocs, List<String> dimensionsSplitOrder,
TreeMap<AggregationFunctionColumnPair, AggregationSpec> aggregationSpecs, int maxLeafRecords,
- Set<String> skipStarNodeCreationForDimensions) {
+ Set<String> skipStarNodeCreationForDimensions, boolean nullHandlingEnabled) {
metadataProperties.setProperty(MetadataKey.TOTAL_DOCS, totalDocs);
metadataProperties.setProperty(MetadataKey.DIMENSIONS_SPLIT_ORDER, dimensionsSplitOrder);
metadataProperties.setProperty(MetadataKey.FUNCTION_COLUMN_PAIRS, aggregationSpecs.keySet());
@@ -154,5 +168,6 @@ public class StarTreeV2Metadata {
metadataProperties.setProperty(MetadataKey.MAX_LEAF_RECORDS, maxLeafRecords);
metadataProperties.setProperty(MetadataKey.SKIP_STAR_NODE_CREATION_FOR_DIMENSIONS,
skipStarNodeCreationForDimensions);
+ metadataProperties.setProperty(MetadataKey.NULL_HANDLING_ENABLED, nullHandlingEnabled);
}
}
diff --git a/pinot-segment-spi/src/test/java/org/apache/pinot/segment/spi/index/startree/StarTreeV2MetadataTest.java b/pinot-segment-spi/src/test/java/org/apache/pinot/segment/spi/index/startree/StarTreeV2MetadataTest.java
index e29d73abc2f..1a68afaee62 100644
--- a/pinot-segment-spi/src/test/java/org/apache/pinot/segment/spi/index/startree/StarTreeV2MetadataTest.java
+++ b/pinot-segment-spi/src/test/java/org/apache/pinot/segment/spi/index/startree/StarTreeV2MetadataTest.java
@@ -98,7 +98,8 @@ public class StarTreeV2MetadataTest {
private static Configuration createMetadata(List<String> dimensionsSplitOrder,
TreeMap<AggregationFunctionColumnPair, AggregationSpec> aggregationSpecs) {
Configuration metadataProperties = new PropertiesConfiguration();
- StarTreeV2Metadata.writeMetadata(metadataProperties, 1, dimensionsSplitOrder, aggregationSpecs, 10000, Set.of());
+ StarTreeV2Metadata.writeMetadata(metadataProperties, 1, dimensionsSplitOrder, aggregationSpecs, 10000, Set.of(),
+ false);
return metadataProperties;
}
diff --git a/pinot-server/src/main/java/org/apache/pinot/server/api/resources/SegmentMetadataFetcher.java b/pinot-server/src/main/java/org/apache/pinot/server/api/resources/SegmentMetadataFetcher.java
index d684a4f60c3..648de844925 100644
--- a/pinot-server/src/main/java/org/apache/pinot/server/api/resources/SegmentMetadataFetcher.java
+++ b/pinot-server/src/main/java/org/apache/pinot/server/api/resources/SegmentMetadataFetcher.java
@@ -64,6 +64,7 @@ public class SegmentMetadataFetcher {
private static final String STAR_TREE_METRIC_AGGREGATIONS = "metric-aggregations";
private static final String STAR_TREE_MAX_LEAF_RECORDS = "max-leaf-records";
private static final String STAR_TREE_DIMENSION_COLUMNS_SKIPPED = "dimension-columns-skipped";
+ private static final String STAR_TREE_NULL_HANDLING_ENABLED = "null-handling-enabled";
/// This is a helper method that fetches the segment metadata for a given segment.
/// @param columns Columns to include for metadata
@@ -188,6 +189,9 @@ public class SegmentMetadataFetcher {
starTreeIndexMap.put(STAR_TREE_MAX_LEAF_RECORDS, starTreeMetadata.getMaxLeafRecords());
starTreeIndexMap.put(STAR_TREE_DIMENSION_COLUMNS_SKIPPED,
starTreeMetadata.getSkipStarNodeCreationForDimensions());
+ // Part of the star-tree's identity: a null-aware and a regular star-tree over the same columns are different
+ // indexes, so a consumer rebuilding a config from this metadata has to carry it
+ starTreeIndexMap.put(STAR_TREE_NULL_HANDLING_ENABLED, starTreeMetadata.isNullHandlingEnabled());
startreeDetails.add(starTreeIndexMap);
}
return startreeDetails;
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/StarTreeIndexConfig.java b/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/StarTreeIndexConfig.java
index 542ee055c6e..1e11483f5cd 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/StarTreeIndexConfig.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/StarTreeIndexConfig.java
@@ -38,6 +38,8 @@ public class StarTreeIndexConfig extends BaseJsonConfig {
private final List<StarTreeAggregationConfig> _aggregationConfigs;
// The upper bound of records to be scanned at the leaf node
private final int _maxLeafRecords;
+ // Whether the star-tree pre-aggregates with null-aware semantics
+ private final boolean _nullHandlingEnabled;
@JsonCreator
public StarTreeIndexConfig(
@@ -46,7 +48,8 @@ public class StarTreeIndexConfig extends BaseJsonConfig {
List<String> skipStarNodeCreationForDimensions,
@JsonProperty(value = "functionColumnPairs") @Nullable List<String> functionColumnPairs,
@JsonProperty(value = "aggregationConfigs") @Nullable List<StarTreeAggregationConfig> aggregationConfigs,
- @JsonProperty(value = "maxLeafRecords") int maxLeafRecords) {
+ @JsonProperty(value = "maxLeafRecords") int maxLeafRecords,
+ @JsonProperty(value = "nullHandlingEnabled") boolean nullHandlingEnabled) {
Preconditions.checkArgument(CollectionUtils.isNotEmpty(dimensionsSplitOrder),
"'dimensionsSplitOrder' must be configured");
_dimensionsSplitOrder = dimensionsSplitOrder;
@@ -55,10 +58,20 @@ public class StarTreeIndexConfig extends BaseJsonConfig {
_functionColumnPairs = CollectionUtils.isNotEmpty(functionColumnPairs) ? functionColumnPairs : null;
_aggregationConfigs = CollectionUtils.isNotEmpty(aggregationConfigs) ? aggregationConfigs : null;
_maxLeafRecords = maxLeafRecords;
+ _nullHandlingEnabled = nullHandlingEnabled;
Preconditions.checkArgument(_functionColumnPairs != null || _aggregationConfigs != null,
"Either 'functionColumnPairs' or 'aggregationConfigs' must be configured");
}
+ /// Convenience constructor for a star-tree that is not null-aware, matching the behavior before
+ /// `nullHandlingEnabled` was introduced.
+ public StarTreeIndexConfig(List<String> dimensionsSplitOrder,
+ @Nullable List<String> skipStarNodeCreationForDimensions, @Nullable List<String> functionColumnPairs,
+ @Nullable List<StarTreeAggregationConfig> aggregationConfigs, int maxLeafRecords) {
+ this(dimensionsSplitOrder, skipStarNodeCreationForDimensions, functionColumnPairs, aggregationConfigs,
+ maxLeafRecords, false);
+ }
+
public List<String> getDimensionsSplitOrder() {
return _dimensionsSplitOrder;
}
@@ -81,4 +94,14 @@ public class StarTreeIndexConfig extends BaseJsonConfig {
public int getMaxLeafRecords() {
return _maxLeafRecords;
}
+
+ /// Returns whether this star-tree pre-aggregates with null-aware semantics.
+ ///
+ /// A null-aware star-tree keeps null dimension values in their own group instead of folding them into the column's
+ /// default null value, and excludes null metric values from the pre-aggregation. It can therefore only serve queries
+ /// with null handling enabled, while a regular star-tree can only serve queries with null handling disabled (or
+ /// enabled queries over columns that happen to contain no nulls).
+ public boolean isNullHandlingEnabled() {
+ return _nullHandlingEnabled;
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org