This is an automated email from the ASF dual-hosted git repository.
yashmayya 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 9d6545bde17 Support partial metadata based aggregation when some
projection aggregations are not metadata compatible (#18334)
9d6545bde17 is described below
commit 9d6545bde17d311129c1e52c3ff26d80f08fcb71
Author: Anil Dasari <[email protected]>
AuthorDate: Sun Aug 2 12:49:50 2026 -0700
Support partial metadata based aggregation when some projection
aggregations are not metadata compatible (#18334)
---
.../core/operator/query/AggregationOperator.java | 57 ++-
.../query/NonScanBasedAggregationOperator.java | 352 +------------------
.../pinot/core/plan/AggregationPlanNode.java | 123 +++++--
.../aggregation/DefaultAggregationExecutor.java | 27 +-
.../function/AggregationFunctionUtils.java | 385 ++++++++++++++++++++-
.../executor/StarTreeAggregationExecutor.java | 3 +
...adataAndDictionaryAggregationPlanMakerTest.java | 269 ++++++++++++++
.../DefaultAggregationExecutorTest.java | 70 +++-
.../function/AggregationFunctionUtilsTest.java | 84 +++++
.../pinot/queries/ExplainPlanQueriesTest.java | 4 +-
...nerSegmentAggregationMultiValueQueriesTest.java | 2 +-
...SegmentAggregationMultiValueRawQueriesTest.java | 2 +-
...erSegmentAggregationSingleValueQueriesTest.java | 2 +-
13 files changed, 974 insertions(+), 406 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 a03f8364929..eae9166af53 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
@@ -21,7 +21,9 @@ package org.apache.pinot.core.operator.query;
import com.google.common.base.CaseFormat;
import java.util.Arrays;
import java.util.List;
+import java.util.Objects;
import java.util.stream.Collectors;
+import javax.annotation.Nullable;
import org.apache.pinot.core.operator.BaseOperator;
import org.apache.pinot.core.operator.BaseProjectOperator;
import org.apache.pinot.core.operator.ExecutionStatistics;
@@ -31,9 +33,11 @@ import
org.apache.pinot.core.operator.blocks.results.AggregationResultsBlock;
import org.apache.pinot.core.query.aggregation.AggregationExecutor;
import org.apache.pinot.core.query.aggregation.DefaultAggregationExecutor;
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.request.context.QueryContext;
import org.apache.pinot.core.startree.executor.StarTreeAggregationExecutor;
+import org.apache.pinot.segment.spi.datasource.DataSource;
import org.apache.pinot.spi.query.QueryScanCostContext;
@@ -46,16 +50,42 @@ public class AggregationOperator extends
BaseOperator<AggregationResultsBlock> {
private final AggregationFunction[] _aggregationFunctions;
private final BaseProjectOperator<?> _projectOperator;
private final boolean _useStarTree;
- private final long _numTotalDocs;
+ private final int _numTotalDocs;
private int _numDocsScanned = 0;
- public AggregationOperator(QueryContext queryContext, AggregationInfo
aggregationInfo, long numTotalDocs) {
+ // Marks the functions that can be resolved from the column
dictionary/metadata without scanning the segment. The
+ // aligned dataSources array holds the argument data source of each such
function (a null entry is only valid for
+ // COUNT). Resolution is deferred to execution time (see getNextBlock) to
keep query planning cheap. Both are null
+ // when no function is resolvable without scanning, in which case every
function is computed by scanning.
+ @Nullable
+ private final boolean[] _nonScanResolvable;
+ @Nullable
+ private final DataSource[] _dataSources;
+
+ public AggregationOperator(QueryContext queryContext, AggregationInfo
aggregationInfo, int numTotalDocs) {
+ this(queryContext, aggregationInfo, numTotalDocs, null, null);
+ }
+
+ /// Constructs an aggregation operator that optionally resolves some
functions from the column dictionary/metadata
+ /// instead of scanning the segment. For each function flagged in {@code
nonScanResolvable}, the result is resolved
+ /// from the aligned {@code dataSources} entry (via {@link
AggregationFunctionUtils#getAggregationResult}) at
+ /// execution time and the function is skipped during the scan; all other
functions are computed by scanning the
+ /// segment. Passing {@code null} for both means every function is computed
by scanning.
+ ///
+ /// @param nonScanResolvable per-function flags (aligned by function index)
marking the functions to resolve from
+ /// dictionary/metadata, or {@code null} if none are resolvable
+ /// @param dataSources per-function argument data sources aligned by
function index (a {@code null} entry is only
+ /// valid for {@code COUNT}), or {@code null} if none are resolvable
+ public AggregationOperator(QueryContext queryContext, AggregationInfo
aggregationInfo, int numTotalDocs,
+ @Nullable boolean[] nonScanResolvable, @Nullable DataSource[]
dataSources) {
_queryContext = queryContext;
_aggregationFunctions = queryContext.getAggregationFunctions();
_projectOperator = aggregationInfo.getProjectOperator();
_useStarTree = aggregationInfo.isUseStarTree();
_numTotalDocs = numTotalDocs;
+ _nonScanResolvable = nonScanResolvable;
+ _dataSources = dataSources;
}
@Override
@@ -63,9 +93,10 @@ public class AggregationOperator extends
BaseOperator<AggregationResultsBlock> {
// Perform aggregation on all the transform blocks
AggregationExecutor aggregationExecutor;
if (_useStarTree) {
+ // StarTreeAggregationExecutor doesn't support non-scan results.
aggregationExecutor = new
StarTreeAggregationExecutor(_aggregationFunctions);
} else {
- aggregationExecutor = new
DefaultAggregationExecutor(_aggregationFunctions);
+ aggregationExecutor = new
DefaultAggregationExecutor(_aggregationFunctions, resolveNonScanResults());
}
ValueBlock valueBlock;
while ((valueBlock = _projectOperator.nextBlock()) != null) {
@@ -83,6 +114,26 @@ public class AggregationOperator extends
BaseOperator<AggregationResultsBlock> {
return new AggregationResultsBlock(_aggregationFunctions,
aggregationExecutor.getResult(), _queryContext);
}
+ /// Returns {@code null} when no function is resolvable without scanning, in
which case all functions are computed by
+ /// scanning. Each returned non-null entry is consumed by {@link
DefaultAggregationExecutor}, which skips the scan for
+ /// that function and emits the resolved value directly.
+ @Nullable
+ private Object[] resolveNonScanResults() {
+ if (_nonScanResolvable == null) {
+ return null;
+ }
+
+ Objects.requireNonNull(_dataSources);
+ Object[] nonScanResults = new Object[_aggregationFunctions.length];
+ for (int i = 0; i < _aggregationFunctions.length; i++) {
+ if (_nonScanResolvable[i]) {
+ nonScanResults[i] =
AggregationFunctionUtils.getAggregationResult(_aggregationFunctions[i],
+ _dataSources[i], _numTotalDocs, EXPLAIN_NAME);
+ }
+ }
+ return nonScanResults;
+ }
+
@Override
public List<BaseProjectOperator<?>> getChildOperators() {
return List.of(_projectOperator);
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/operator/query/NonScanBasedAggregationOperator.java
b/pinot-core/src/main/java/org/apache/pinot/core/operator/query/NonScanBasedAggregationOperator.java
index fee1fbeae90..4a00993f6b8 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/operator/query/NonScanBasedAggregationOperator.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/operator/query/NonScanBasedAggregationOperator.java
@@ -18,43 +18,16 @@
*/
package org.apache.pinot.core.operator.query;
-import com.clearspring.analytics.stream.cardinality.HyperLogLog;
-import com.clearspring.analytics.stream.cardinality.HyperLogLogPlus;
-import com.dynatrace.hash4j.distinctcount.UltraLogLog;
-import com.google.common.base.Preconditions;
-import it.unimi.dsi.fastutil.doubles.DoubleOpenHashSet;
-import it.unimi.dsi.fastutil.floats.FloatOpenHashSet;
-import it.unimi.dsi.fastutil.ints.IntOpenHashSet;
-import it.unimi.dsi.fastutil.longs.LongOpenHashSet;
-import it.unimi.dsi.fastutil.objects.ObjectOpenHashSet;
-import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.List;
-import java.util.Objects;
-import java.util.Set;
-import org.apache.pinot.core.common.ObjectSerDeUtils;
import org.apache.pinot.core.common.Operator;
import org.apache.pinot.core.operator.BaseOperator;
import org.apache.pinot.core.operator.ExecutionStatistics;
import org.apache.pinot.core.operator.blocks.results.AggregationResultsBlock;
import org.apache.pinot.core.query.aggregation.function.AggregationFunction;
-import
org.apache.pinot.core.query.aggregation.function.DistinctCountHLLAggregationFunction;
-import
org.apache.pinot.core.query.aggregation.function.DistinctCountHLLPlusAggregationFunction;
-import
org.apache.pinot.core.query.aggregation.function.DistinctCountOffHeapAggregationFunction;
-import
org.apache.pinot.core.query.aggregation.function.DistinctCountRawHLLAggregationFunction;
-import
org.apache.pinot.core.query.aggregation.function.DistinctCountRawHLLPlusAggregationFunction;
-import
org.apache.pinot.core.query.aggregation.function.DistinctCountSmartHLLAggregationFunction;
-import
org.apache.pinot.core.query.aggregation.function.DistinctCountSmartHLLPlusAggregationFunction;
-import
org.apache.pinot.core.query.aggregation.function.DistinctCountSmartULLAggregationFunction;
-import
org.apache.pinot.core.query.aggregation.function.DistinctCountULLAggregationFunction;
+import
org.apache.pinot.core.query.aggregation.function.AggregationFunctionUtils;
import org.apache.pinot.core.query.request.context.QueryContext;
-import org.apache.pinot.segment.local.customobject.MinMaxRangePair;
-import org.apache.pinot.segment.local.utils.UltraLogLogUtils;
import org.apache.pinot.segment.spi.datasource.DataSource;
-import org.apache.pinot.segment.spi.index.reader.Dictionary;
-import org.apache.pinot.spi.data.FieldSpec.DataType;
-import org.apache.pinot.spi.query.QueryThreadContext;
-import org.apache.pinot.spi.utils.ByteArray;
/// Aggregation operator that utilizes dictionary or column metadata for
serving aggregation queries to avoid scanning.
@@ -93,97 +66,8 @@ public class NonScanBasedAggregationOperator extends
BaseOperator<AggregationRes
AggregationFunction aggregationFunction = _aggregationFunctions[i];
// note that dataSource will be null for COUNT, sp do not interact with
it until it's known this isn't a COUNT
DataSource dataSource = _dataSources[i];
- Object result;
- switch (aggregationFunction.getType()) {
- case COUNT:
- result = (long) _numTotalDocs;
- break;
- case MIN:
- case MINMV:
- result = getMinValueNumeric(dataSource);
- break;
- case MINLONG:
- result = getMinValueLong(dataSource);
- break;
- case MINSTRING:
- assert dataSource.getDictionary() != null;
- result = dataSource.getDictionary().getMinVal();
- break;
- case MAX:
- case MAXMV:
- result = getMaxValueNumeric(dataSource);
- break;
- case MAXLONG:
- result = getMaxValueLong(dataSource);
- break;
- case MAXSTRING:
- assert dataSource.getDictionary() != null;
- result = dataSource.getDictionary().getMaxVal();
- break;
- case MINMAXRANGE:
- case MINMAXRANGEMV:
- result = new MinMaxRangePair(getMinValueNumeric(dataSource),
getMaxValueNumeric(dataSource));
- break;
- case DISTINCTCOUNT:
- case DISTINCTSUM:
- case DISTINCTAVG:
- case DISTINCTCOUNTMV:
- case DISTINCTSUMMV:
- case DISTINCTAVGMV:
- result =
getDistinctValueSet(Objects.requireNonNull(dataSource.getDictionary()));
- break;
- case DISTINCTCOUNTOFFHEAP:
- result = ((DistinctCountOffHeapAggregationFunction)
aggregationFunction).extractAggregationResult(
- Objects.requireNonNull(dataSource.getDictionary()));
- break;
- case DISTINCTCOUNTHLL:
- case DISTINCTCOUNTHLLMV:
- result =
getDistinctCountHLLResult(Objects.requireNonNull(dataSource.getDictionary()),
- (DistinctCountHLLAggregationFunction) aggregationFunction);
- break;
- case DISTINCTCOUNTRAWHLL:
- case DISTINCTCOUNTRAWHLLMV:
- result =
getDistinctCountHLLResult(Objects.requireNonNull(dataSource.getDictionary()),
- ((DistinctCountRawHLLAggregationFunction)
aggregationFunction).getDistinctCountHLLAggregationFunction());
- break;
- case DISTINCTCOUNTHLLPLUS:
- case DISTINCTCOUNTHLLPLUSMV:
- result =
getDistinctCountHLLPlusResult(Objects.requireNonNull(dataSource.getDictionary()),
- (DistinctCountHLLPlusAggregationFunction) aggregationFunction);
- break;
- case DISTINCTCOUNTRAWHLLPLUS:
- case DISTINCTCOUNTRAWHLLPLUSMV:
- result =
getDistinctCountHLLPlusResult(Objects.requireNonNull(dataSource.getDictionary()),
- ((DistinctCountRawHLLPlusAggregationFunction)
aggregationFunction)
- .getDistinctCountHLLPlusAggregationFunction());
- break;
- case SEGMENTPARTITIONEDDISTINCTCOUNT:
- result = (long)
Objects.requireNonNull(dataSource.getDictionary()).length();
- break;
- case DISTINCTCOUNTSMARTHLL:
- result =
getDistinctCountSmartHLLResult(Objects.requireNonNull(dataSource.getDictionary()),
- (DistinctCountSmartHLLAggregationFunction) aggregationFunction);
- break;
- case DISTINCTCOUNTSMARTHLLPLUS:
- result =
getDistinctCountSmartHLLPlusResult(Objects.requireNonNull(dataSource.getDictionary()),
- (DistinctCountSmartHLLPlusAggregationFunction)
aggregationFunction);
- break;
- case DISTINCTCOUNTULL:
- result =
getDistinctCountULLResult(Objects.requireNonNull(dataSource.getDictionary()),
- (DistinctCountULLAggregationFunction) aggregationFunction);
- break;
- case DISTINCTCOUNTSMARTULL:
- result =
getDistinctCountSmartULLResult(Objects.requireNonNull(dataSource.getDictionary()),
- (DistinctCountSmartULLAggregationFunction) aggregationFunction);
- break;
- case DISTINCTCOUNTRAWULL:
- result =
getDistinctCountULLResult(Objects.requireNonNull(dataSource.getDictionary()),
- (DistinctCountULLAggregationFunction) aggregationFunction);
- break;
- default:
- throw new IllegalStateException(
- "Non-scan based aggregation operator does not support function
type: " + aggregationFunction.getType());
- }
+ Object result =
AggregationFunctionUtils.getAggregationResult(aggregationFunction, dataSource,
+ _numTotalDocs, EXPLAIN_NAME);
aggregationResults.add(result);
}
@@ -191,236 +75,6 @@ public class NonScanBasedAggregationOperator extends
BaseOperator<AggregationRes
return new AggregationResultsBlock(_aggregationFunctions,
aggregationResults, _queryContext);
}
- private static Double getMinValueNumeric(DataSource dataSource) {
- Dictionary dictionary = dataSource.getDictionary();
- if (dictionary != null) {
- return toDouble(dictionary.getMinVal());
- }
- return toDouble(dataSource.getDataSourceMetadata().getMinValue());
- }
-
- private static Long getMinValueLong(DataSource dataSource) {
- DataType dataType =
dataSource.getDataSourceMetadata().getDataType().getStoredType();
- Preconditions.checkArgument(
- dataType == DataType.LONG || dataType == DataType.INT,
- "MINLONG aggregation function can only be applied to columns of
integer types");
- Dictionary dictionary = dataSource.getDictionary();
- if (dictionary != null) {
- return ((Number) dictionary.getMinVal()).longValue();
- }
- return ((Number)
dataSource.getDataSourceMetadata().getMinValue()).longValue();
- }
-
- private static Double getMaxValueNumeric(DataSource dataSource) {
- Dictionary dictionary = dataSource.getDictionary();
- if (dictionary != null) {
- return toDouble(dictionary.getMaxVal());
- }
- return toDouble(dataSource.getDataSourceMetadata().getMaxValue());
- }
-
- private static Long getMaxValueLong(DataSource dataSource) {
- DataType dataType =
dataSource.getDataSourceMetadata().getDataType().getStoredType();
- Preconditions.checkArgument(
- dataType == DataType.LONG || dataType == DataType.INT,
- "MAXLONG aggregation function can only be applied to columns of
integer types");
- Dictionary dictionary = dataSource.getDictionary();
- if (dictionary != null) {
- return ((Number) dictionary.getMaxVal()).longValue();
- }
- return ((Number)
dataSource.getDataSourceMetadata().getMaxValue()).longValue();
- }
-
- private static Double toDouble(Comparable<?> value) {
- if (value instanceof Double) {
- return (Double) value;
- } else if (value instanceof Number) {
- return ((Number) value).doubleValue();
- } else {
- return Double.parseDouble(value.toString());
- }
- }
-
- private static Set getDistinctValueSet(Dictionary dictionary) {
- int dictionarySize = dictionary.length();
- switch (dictionary.getValueType()) {
- case INT:
- IntOpenHashSet intSet = new IntOpenHashSet(dictionarySize);
- for (int dictId = 0; dictId < dictionarySize; dictId++) {
-
QueryThreadContext.checkTerminationAndSampleUsagePeriodically(dictId,
EXPLAIN_NAME);
- intSet.add(dictionary.getIntValue(dictId));
- }
- return intSet;
- case LONG:
- LongOpenHashSet longSet = new LongOpenHashSet(dictionarySize);
- for (int dictId = 0; dictId < dictionarySize; dictId++) {
-
QueryThreadContext.checkTerminationAndSampleUsagePeriodically(dictId,
EXPLAIN_NAME);
- longSet.add(dictionary.getLongValue(dictId));
- }
- return longSet;
- case FLOAT:
- FloatOpenHashSet floatSet = new FloatOpenHashSet(dictionarySize);
- for (int dictId = 0; dictId < dictionarySize; dictId++) {
-
QueryThreadContext.checkTerminationAndSampleUsagePeriodically(dictId,
EXPLAIN_NAME);
- floatSet.add(dictionary.getFloatValue(dictId));
- }
- return floatSet;
- case DOUBLE:
- DoubleOpenHashSet doubleSet = new DoubleOpenHashSet(dictionarySize);
- for (int dictId = 0; dictId < dictionarySize; dictId++) {
-
QueryThreadContext.checkTerminationAndSampleUsagePeriodically(dictId,
EXPLAIN_NAME);
- doubleSet.add(dictionary.getDoubleValue(dictId));
- }
- return doubleSet;
- case BIG_DECIMAL:
- ObjectOpenHashSet<BigDecimal> bigDecimalSet = new
ObjectOpenHashSet<>(dictionarySize);
- for (int dictId = 0; dictId < dictionarySize; dictId++) {
-
QueryThreadContext.checkTerminationAndSampleUsagePeriodically(dictId,
EXPLAIN_NAME);
- bigDecimalSet.add(dictionary.getBigDecimalValue(dictId));
- }
- return bigDecimalSet;
- case STRING:
- ObjectOpenHashSet<String> stringSet = new
ObjectOpenHashSet<>(dictionarySize);
- for (int dictId = 0; dictId < dictionarySize; dictId++) {
-
QueryThreadContext.checkTerminationAndSampleUsagePeriodically(dictId,
EXPLAIN_NAME);
- stringSet.add(dictionary.getStringValue(dictId));
- }
- return stringSet;
- case BYTES:
- ObjectOpenHashSet<ByteArray> bytesSet = new
ObjectOpenHashSet<>(dictionarySize);
- for (int dictId = 0; dictId < dictionarySize; dictId++) {
-
QueryThreadContext.checkTerminationAndSampleUsagePeriodically(dictId,
EXPLAIN_NAME);
- bytesSet.add(new ByteArray(dictionary.getBytesValue(dictId)));
- }
- return bytesSet;
- default:
- throw new IllegalStateException();
- }
- }
-
- private static HyperLogLog getDistinctValueHLL(Dictionary dictionary, int
log2m) {
- HyperLogLog hll = new HyperLogLog(log2m);
- int length = dictionary.length();
- for (int i = 0; i < length; i++) {
- QueryThreadContext.checkTerminationAndSampleUsagePeriodically(i,
EXPLAIN_NAME);
- hll.offer(dictionary.get(i));
- }
- return hll;
- }
-
- private static UltraLogLog getDistinctValueULL(Dictionary dictionary, int p)
{
- UltraLogLog ull = UltraLogLog.create(p);
- int length = dictionary.length();
- for (int i = 0; i < length; i++) {
- QueryThreadContext.checkTerminationAndSampleUsagePeriodically(i,
EXPLAIN_NAME);
- Object value = dictionary.get(i);
- UltraLogLogUtils.hashObject(value).ifPresent(ull::add);
- }
- return ull;
- }
-
- private static HyperLogLogPlus getDistinctValueHLLPlus(Dictionary
dictionary, int p, int sp) {
- HyperLogLogPlus hllPlus = new HyperLogLogPlus(p, sp);
- int length = dictionary.length();
- for (int i = 0; i < length; i++) {
- QueryThreadContext.checkTerminationAndSampleUsagePeriodically(i,
EXPLAIN_NAME);
- hllPlus.offer(dictionary.get(i));
- }
- return hllPlus;
- }
-
- private static HyperLogLog getDistinctCountHLLResult(Dictionary dictionary,
- DistinctCountHLLAggregationFunction function) {
- if (dictionary.getValueType() == DataType.BYTES) {
- // Treat BYTES value as serialized HyperLogLog
- try {
- QueryThreadContext.checkTerminationAndSampleUsage(EXPLAIN_NAME);
- HyperLogLog hll =
ObjectSerDeUtils.HYPER_LOG_LOG_SER_DE.deserialize(dictionary.getBytesValue(0));
- int length = dictionary.length();
- for (int i = 1; i < length; i++) {
- QueryThreadContext.checkTerminationAndSampleUsagePeriodically(i,
EXPLAIN_NAME);
-
hll.addAll(ObjectSerDeUtils.HYPER_LOG_LOG_SER_DE.deserialize(dictionary.getBytesValue(i)));
- }
- return hll;
- } catch (Exception e) {
- throw new RuntimeException("Caught exception while merging
HyperLogLogs", e);
- }
- } else {
- return getDistinctValueHLL(dictionary, function.getLog2m());
- }
- }
-
- private static HyperLogLogPlus getDistinctCountHLLPlusResult(Dictionary
dictionary,
- DistinctCountHLLPlusAggregationFunction function) {
- if (dictionary.getValueType() == DataType.BYTES) {
- // Treat BYTES value as serialized HyperLogLogPlus
- try {
- QueryThreadContext.checkTerminationAndSampleUsage(EXPLAIN_NAME);
- HyperLogLogPlus hllplus =
ObjectSerDeUtils.HYPER_LOG_LOG_PLUS_SER_DE.deserialize(dictionary.getBytesValue(0));
- int length = dictionary.length();
- for (int i = 1; i < length; i++) {
- QueryThreadContext.checkTerminationAndSampleUsagePeriodically(i,
EXPLAIN_NAME);
-
hllplus.addAll(ObjectSerDeUtils.HYPER_LOG_LOG_PLUS_SER_DE.deserialize(dictionary.getBytesValue(i)));
- }
- return hllplus;
- } catch (Exception e) {
- throw new RuntimeException("Caught exception while merging
HyperLogLogPluses", e);
- }
- } else {
- return getDistinctValueHLLPlus(dictionary, function.getP(),
function.getSp());
- }
- }
-
- private static Object getDistinctCountSmartHLLResult(Dictionary dictionary,
- DistinctCountSmartHLLAggregationFunction function) {
- if (dictionary.length() > function.getThreshold()) {
- // Store values into a HLL when the dictionary size exceeds the
conversion threshold
- return getDistinctValueHLL(dictionary, function.getLog2m());
- } else {
- return getDistinctValueSet(dictionary);
- }
- }
-
- private static Object getDistinctCountSmartHLLPlusResult(Dictionary
dictionary,
- DistinctCountSmartHLLPlusAggregationFunction function) {
- if (dictionary.length() > function.getThreshold()) {
- // Store values into a HLLPlus when the dictionary size exceeds the
conversion threshold
- return getDistinctValueHLLPlus(dictionary, function.getP(),
function.getSp());
- } else {
- return getDistinctValueSet(dictionary);
- }
- }
-
- private static UltraLogLog getDistinctCountULLResult(Dictionary dictionary,
- DistinctCountULLAggregationFunction function) {
- if (dictionary.getValueType() == DataType.BYTES) {
- // Treat BYTES value as serialized UltraLogLog and merge
- try {
- QueryThreadContext.checkTerminationAndSampleUsage(EXPLAIN_NAME);
- UltraLogLog ull =
ObjectSerDeUtils.ULTRA_LOG_LOG_OBJECT_SER_DE.deserialize(dictionary.getBytesValue(0));
- int length = dictionary.length();
- for (int i = 1; i < length; i++) {
- QueryThreadContext.checkTerminationAndSampleUsagePeriodically(i,
EXPLAIN_NAME);
-
ull.add(ObjectSerDeUtils.ULTRA_LOG_LOG_OBJECT_SER_DE.deserialize(dictionary.getBytesValue(i)));
- }
- return ull;
- } catch (Exception e) {
- throw new RuntimeException("Caught exception while merging
UltraLogLogs", e);
- }
- } else {
- return getDistinctValueULL(dictionary, function.getP());
- }
- }
-
- private static Object getDistinctCountSmartULLResult(Dictionary dictionary,
- DistinctCountSmartULLAggregationFunction function) {
- if (dictionary.length() > function.getThreshold()) {
- return getDistinctValueULL(dictionary, function.getP());
- } else {
- return getDistinctValueSet(dictionary);
- }
- }
-
@Override
public String toExplainString() {
return EXPLAIN_NAME;
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/plan/AggregationPlanNode.java
b/pinot-core/src/main/java/org/apache/pinot/core/plan/AggregationPlanNode.java
index e73e1b20c2e..054bd26cc49 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/plan/AggregationPlanNode.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/plan/AggregationPlanNode.java
@@ -20,6 +20,7 @@ package org.apache.pinot.core.plan;
import java.util.EnumSet;
import java.util.List;
+import javax.annotation.Nullable;
import org.apache.pinot.common.request.context.ExpressionContext;
import org.apache.pinot.core.common.Operator;
import org.apache.pinot.core.operator.blocks.results.AggregationResultsBlock;
@@ -60,6 +61,12 @@ public class AggregationPlanNode implements PlanNode {
private static final EnumSet<AggregationFunctionType>
METADATA_BASED_FUNCTIONS =
EnumSet.of(COUNT, MIN, MINMV, MINLONG, MAX, MAXMV, MAXLONG, MINMAXRANGE,
MINMAXRANGEMV);
+ // MIN/MAX/MINMAXRANGE derive their result numerically from the column
min/max, so they can only be resolved from
+ // metadata/dictionary for numeric columns. Non-numeric columns (e.g. BYTES)
store min/max as raw values that cannot
+ // be parsed as numbers.
+ private static final EnumSet<AggregationFunctionType>
NUMERIC_METADATA_FUNCTIONS =
+ EnumSet.of(MIN, MINMV, MINLONG, MAX, MAXMV, MAXLONG, MINMAXRANGE,
MINMAXRANGEMV);
+
private final IndexSegment _indexSegment;
private final SegmentContext _segmentContext;
private final QueryContext _queryContext;
@@ -109,17 +116,41 @@ public class AggregationPlanNode implements PlanNode {
boolean hasNullValues = _queryContext.isNullHandlingEnabled() &&
hasNullValues(aggregationFunctions);
if (!hasNullValues) {
- // Priority 2: Check if non-scan based aggregation is feasible
- if (filterOperator.isResultMatchingAll() && isFitForNonScanBasedPlan()) {
+ // when the filter matches all documents, resolve as many functions as
possible from the column
+ // dictionary/metadata without scanning the segment. Eligibility is
evaluated once per function here
+ // and reused for both the fully non-scan path (all functions
resolvable) and
+ // the partial path (some functions resolvable).
+ if (filterOperator.isResultMatchingAll()) {
+ boolean[] nonScanResolvable = new boolean[aggregationFunctions.length];
DataSource[] dataSources = new DataSource[aggregationFunctions.length];
+ int numResolved = 0;
for (int i = 0; i < aggregationFunctions.length; i++) {
- List<?> inputExpressions =
aggregationFunctions[i].getInputExpressions();
- if (!inputExpressions.isEmpty()) {
- String column = ((ExpressionContext)
inputExpressions.get(0)).getIdentifier();
- dataSources[i] = _indexSegment.getDataSource(column,
_queryContext.getSchema());
+ DataSource dataSource =
getDataSourceForAggregationFunction(aggregationFunctions[i]);
+ if (isFitForNonScanBasedPlan(aggregationFunctions[i], dataSource)) {
+ nonScanResolvable[i] = true;
+ dataSources[i] = dataSource;
+ numResolved++;
+ }
+ }
+
+ if (numResolved == aggregationFunctions.length) {
+ // Priority 2: all functions can be resolved from
dictionary/metadata -> fully non-scan based execution
+ return new NonScanBasedAggregationOperator(_queryContext,
dataSources, numTotalDocs);
+ }
+ if (numResolved > 0) {
+ // Some functions are resolved from dictionary/metadata; the rest
are scanned by the AggregationOperator.
+ // Project only the scanned functions' columns so a column that is
resolved by metadata is not counted as
+ // scanned (keeps it out of numEntriesScannedPostFilter).
+ AggregationFunction[] scannedFunctions = new
AggregationFunction[aggregationFunctions.length - numResolved];
+ for (int i = 0, j = 0; i < aggregationFunctions.length; i++) {
+ if (!nonScanResolvable[i]) {
+ scannedFunctions[j++] = aggregationFunctions[i];
+ }
}
+ aggregationInfo =
AggregationFunctionUtils.buildAggregationInfoWithoutStarTree(_segmentContext,
_queryContext,
+ aggregationFunctions, scannedFunctions, filterOperator);
+ return new AggregationOperator(_queryContext, aggregationInfo,
numTotalDocs, nonScanResolvable, dataSources);
}
- return new NonScanBasedAggregationOperator(_queryContext, dataSources,
numTotalDocs);
}
// Priority 3: Check if fast filtered count can be used
@@ -164,34 +195,40 @@ public class AggregationPlanNode implements PlanNode {
return false;
}
- /// Returns `true` if the given aggregations can be solved with dictionary
or column metadata, `false`
- /// otherwise.
- private boolean isFitForNonScanBasedPlan() {
- AggregationFunction[] aggregationFunctions =
_queryContext.getAggregationFunctions();
- assert aggregationFunctions != null;
- for (AggregationFunction<?, ?> aggregationFunction : aggregationFunctions)
{
- if (aggregationFunction.getType() == COUNT) {
- continue;
- }
- ExpressionContext argument =
aggregationFunction.getInputExpressions().get(0);
- if (argument.getType() != ExpressionContext.Type.IDENTIFIER) {
- return false;
- }
- DataSource dataSource =
_indexSegment.getDataSource(argument.getIdentifier(),
_queryContext.getSchema());
- if (DICTIONARY_BASED_FUNCTIONS.contains(aggregationFunction.getType())) {
- if (dataSource.getDictionary() != null) {
- continue;
- }
- }
- if (METADATA_BASED_FUNCTIONS.contains(aggregationFunction.getType())) {
- if (dataSource.getDataSourceMetadata().getMaxValue() != null
- && dataSource.getDataSourceMetadata().getMinValue() != null) {
- continue;
- }
- }
+ /// Returns {@code true} if the given aggregation function can be resolved
from the column dictionary or metadata
+ /// (without scanning the segment), {@code false} otherwise. {@code COUNT}
is always eligible. Functions whose result
+ /// is derived numerically from the column min/max (e.g. MIN, MAX,
MINMAXRANGE) are only eligible for numeric columns,
+ /// since non-numeric columns (e.g. BYTES) store min/max as raw values that
cannot be parsed as numbers.
+ ///
+ /// @param aggregationFunction aggregation function to test
+ /// @param dataSource the function argument's data source (see {@link
#getDataSourceForAggregationFunction})
+ private boolean isFitForNonScanBasedPlan(AggregationFunction<?, ?>
aggregationFunction,
+ @Nullable DataSource dataSource) {
+ AggregationFunctionType functionType = aggregationFunction.getType();
+ if (functionType == COUNT) {
+ return true;
+ }
+
+ if (dataSource == null) {
+ // Aggregation function does not have a single identifier argument (e.g.
SUM(1)),
+ // so it cannot be resolved from metadata
+ return false;
+ }
+
+ // MIN/MAX/MINMAXRANGE derive their result numerically from the column
min/max, which is only valid for numeric
+ // columns. Non-numeric columns (e.g. BYTES) store min/max as raw values
that cannot be parsed as numbers.
+ if (NUMERIC_METADATA_FUNCTIONS.contains(functionType)
+ &&
!dataSource.getDataSourceMetadata().getDataType().getStoredType().isNumeric()) {
return false;
}
- return true;
+
+ if (dataSource.getDictionary() != null &&
DICTIONARY_BASED_FUNCTIONS.contains(functionType)) {
+ return true;
+ }
+
+ return METADATA_BASED_FUNCTIONS.contains(functionType)
+ && dataSource.getDataSourceMetadata().getMaxValue() != null
+ && dataSource.getDataSourceMetadata().getMinValue() != null;
}
private static boolean canOptimizeFilteredCount(BaseFilterOperator
filterOperator,
@@ -199,4 +236,24 @@ public class AggregationPlanNode implements PlanNode {
return (aggregationFunctions.length == 1 &&
aggregationFunctions[0].getType() == COUNT)
&& filterOperator.canOptimizeCount();
}
+
+ /// Returns the data source for the given aggregation function's argument,
or {@code null} if the function has no
+ /// argument (e.g. {@code COUNT(*)}) or its argument is not a single column
identifier (e.g. {@code COUNT(1)} or a
+ /// transform expression), in which case it cannot be resolved from
dictionary/metadata.
+ ///
+ /// @param aggregationFunction aggregation function whose argument data
source is resolved
+ /// @return the argument's data source, or {@code null} if it has no single
identifier argument
+ @Nullable
+ private DataSource
getDataSourceForAggregationFunction(AggregationFunction<?, ?>
aggregationFunction) {
+ List<ExpressionContext> inputExpressions =
aggregationFunction.getInputExpressions();
+ if (!inputExpressions.isEmpty()) {
+ ExpressionContext argument = inputExpressions.get(0);
+ if (argument.getType() != ExpressionContext.Type.IDENTIFIER) {
+ return null;
+ }
+ return _indexSegment.getDataSource(argument.getIdentifier(),
_queryContext.getSchema());
+ }
+
+ return null;
+ }
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/DefaultAggregationExecutor.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/DefaultAggregationExecutor.java
index d6c64c1cfb6..806413253fd 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/DefaultAggregationExecutor.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/DefaultAggregationExecutor.java
@@ -20,6 +20,7 @@ package org.apache.pinot.core.query.aggregation;
import java.util.ArrayList;
import java.util.List;
+import javax.annotation.Nullable;
import org.apache.pinot.core.operator.blocks.ValueBlock;
import org.apache.pinot.core.query.aggregation.function.AggregationFunction;
import
org.apache.pinot.core.query.aggregation.function.AggregationFunctionUtils;
@@ -29,8 +30,19 @@ import
org.apache.pinot.core.query.aggregation.function.AggregationFunctionUtils
public class DefaultAggregationExecutor implements AggregationExecutor {
protected final AggregationFunction[] _aggregationFunctions;
protected final AggregationResultHolder[] _aggregationResultHolders;
+ @Nullable
+ protected final Object[] _nonScanResults;
- public DefaultAggregationExecutor(AggregationFunction[]
aggregationFunctions) {
+ /// Creates an executor that skips functions with a non-scan result
(resolved from the column dictionary or
+ /// metadata). For each index {@code i} where {@code nonScanResults[i]} is
non-null, the function is not aggregated
+ /// over the scanned blocks and the resolved value is emitted directly in
the results. A {@code null} array disables
+ /// this behavior and all functions are computed by scanning.
+ ///
+ /// @param nonScanResults per-function results resolved without scanning
(from the column dictionary or metadata),
+ /// or {@code null} if none are resolved
+ public DefaultAggregationExecutor(AggregationFunction[] aggregationFunctions,
+ @Nullable Object[] nonScanResults) {
+ _nonScanResults = nonScanResults;
_aggregationFunctions = aggregationFunctions;
int numAggregationFunctions = aggregationFunctions.length;
_aggregationResultHolders = new
AggregationResultHolder[numAggregationFunctions];
@@ -39,11 +51,18 @@ public class DefaultAggregationExecutor implements
AggregationExecutor {
}
}
+ public DefaultAggregationExecutor(AggregationFunction[]
aggregationFunctions) {
+ this(aggregationFunctions, null);
+ }
+
@Override
public void aggregate(ValueBlock valueBlock) {
int numAggregationFunctions = _aggregationFunctions.length;
int length = valueBlock.getNumDocs();
for (int i = 0; i < numAggregationFunctions; i++) {
+ if (_nonScanResults != null && _nonScanResults[i] != null) {
+ continue; // skip — already resolved without scanning
(dictionary/metadata)
+ }
AggregationFunction aggregationFunction = _aggregationFunctions[i];
aggregationFunction.aggregate(length, _aggregationResultHolders[i],
AggregationFunctionUtils.getBlockValSetMap(aggregationFunction,
valueBlock));
@@ -55,7 +74,11 @@ public class DefaultAggregationExecutor implements
AggregationExecutor {
int numFunctions = _aggregationFunctions.length;
List<Object> aggregationResults = new ArrayList<>(numFunctions);
for (int i = 0; i < numFunctions; i++) {
-
aggregationResults.add(_aggregationFunctions[i].extractAggregationResult(_aggregationResultHolders[i]));
+ if (_nonScanResults != null && _nonScanResults[i] != null) {
+ aggregationResults.add(_nonScanResults[i]);
+ } else {
+
aggregationResults.add(_aggregationFunctions[i].extractAggregationResult(_aggregationResultHolders[i]));
+ }
}
return aggregationResults;
}
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 6e9107b7444..a9c6e7d2e9f 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
@@ -18,11 +18,20 @@
*/
package org.apache.pinot.core.query.aggregation.function;
+import com.clearspring.analytics.stream.cardinality.HyperLogLog;
+import com.clearspring.analytics.stream.cardinality.HyperLogLogPlus;
+import com.dynatrace.hash4j.distinctcount.UltraLogLog;
+import com.google.common.base.Preconditions;
import it.unimi.dsi.fastutil.doubles.DoubleArrayList;
+import it.unimi.dsi.fastutil.doubles.DoubleOpenHashSet;
import it.unimi.dsi.fastutil.floats.FloatArrayList;
+import it.unimi.dsi.fastutil.floats.FloatOpenHashSet;
import it.unimi.dsi.fastutil.ints.IntArrayList;
+import it.unimi.dsi.fastutil.ints.IntOpenHashSet;
import it.unimi.dsi.fastutil.longs.LongArrayList;
+import it.unimi.dsi.fastutil.longs.LongOpenHashSet;
import it.unimi.dsi.fastutil.objects.ObjectArrayList;
+import it.unimi.dsi.fastutil.objects.ObjectOpenHashSet;
import java.io.IOException;
import java.math.BigDecimal;
import java.sql.Timestamp;
@@ -31,6 +40,7 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.Set;
import javax.annotation.Nullable;
import org.apache.commons.lang3.tuple.Pair;
@@ -42,6 +52,7 @@ import
org.apache.pinot.common.request.context.predicate.Predicate;
import org.apache.pinot.common.utils.DataSchema.ColumnDataType;
import org.apache.pinot.common.utils.config.QueryOptionsUtils;
import org.apache.pinot.core.common.BlockValSet;
+import org.apache.pinot.core.common.ObjectSerDeUtils;
import org.apache.pinot.core.common.datatable.DataTableBuilder;
import org.apache.pinot.core.operator.BaseProjectOperator;
import org.apache.pinot.core.operator.blocks.ValueBlock;
@@ -53,15 +64,22 @@ import org.apache.pinot.core.plan.FilterPlanNode;
import org.apache.pinot.core.plan.ProjectPlanNode;
import org.apache.pinot.core.query.request.context.QueryContext;
import org.apache.pinot.core.startree.StarTreeUtils;
+import org.apache.pinot.segment.local.customobject.MinMaxRangePair;
+import org.apache.pinot.segment.local.utils.UltraLogLogUtils;
import org.apache.pinot.segment.spi.AggregationFunctionType;
import org.apache.pinot.segment.spi.SegmentContext;
+import org.apache.pinot.segment.spi.datasource.DataSource;
+import org.apache.pinot.segment.spi.index.reader.Dictionary;
import
org.apache.pinot.segment.spi.index.startree.AggregationFunctionColumnPair;
+import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.query.QueryThreadContext;
import org.apache.pinot.spi.utils.ByteArray;
/// The `AggregationFunctionUtils` class provides utility methods for
aggregation function.
@SuppressWarnings({"rawtypes", "unchecked"})
public class AggregationFunctionUtils {
+
private AggregationFunctionUtils() {
}
@@ -362,7 +380,23 @@ public class AggregationFunctionUtils {
return new AggregationInfo(aggregationFunctions, projectOperator, false);
}
- /// Builds swim-lanes (list of [AggregationInfo]) for filtered aggregations.
+ /// Builds {@link AggregationInfo} for aggregations without using star-tree
index, projecting only the columns
+ /// required by {@code projectionFunctions} (a subset of {@code
allFunctions}). The partial metadata path passes just
+ /// the scanned functions here so that columns used solely by
metadata-resolved functions are excluded from the scan,
+ /// while {@link AggregationInfo} still carries the full function set. For a
full scan the two arrays are identical.
+ public static AggregationInfo
buildAggregationInfoWithoutStarTree(SegmentContext segmentContext,
+ QueryContext queryContext, AggregationFunction[] allFunctions,
AggregationFunction[] projectionFunctions,
+ BaseFilterOperator filterOperator) {
+ Set<ExpressionContext> expressionsToTransform =
+ collectExpressionsToTransform(projectionFunctions,
queryContext.getGroupByExpressions());
+ BaseProjectOperator<?> projectOperator =
+ new ProjectPlanNode(segmentContext, queryContext,
expressionsToTransform, DocIdSetPlanNode.MAX_DOC_PER_CALL,
+ filterOperator).run();
+ return new AggregationInfo(allFunctions, projectOperator, false);
+ }
+
+
+ /// Builds swim-lanes (list of {@link AggregationInfo}) for filtered
aggregations.
public static List<AggregationInfo>
buildFilteredAggregationInfos(SegmentContext segmentContext,
QueryContext queryContext) {
assert queryContext.getAggregationFunctions() != null &&
queryContext.getFilteredAggregationFunctions() != null;
@@ -477,4 +511,353 @@ public class AggregationFunctionUtils {
}
return columnName;
}
+
+ /// Resolves the result of the given aggregation function from the column
dictionary or metadata, without scanning the
+ /// segment. This is used by the non-scan based aggregation operator and by
the partial metadata-based path in
+ /// {@link org.apache.pinot.core.operator.query.AggregationOperator} to
resolve metadata-eligible functions.
+ /// <p>
+ /// {@code COUNT} is resolved directly from {@code numTotalDocs}. Every
other supported function reads its result from
+ /// the column dictionary or metadata and therefore requires a non-null
{@code dataSource}. Callers must only invoke
+ /// this method for functions that are metadata/dictionary eligible (as
determined by the fitness check in
+ /// {@link org.apache.pinot.core.plan.AggregationPlanNode}); unsupported
function types cause an
+ /// {@link IllegalStateException}.
+ ///
+ /// @param aggregationFunction aggregation function to resolve
+ /// @param dataSource data source of the function argument; may be {@code
null} only for {@code COUNT}
+ /// @param numTotalDocs total number of documents in the segment, used to
resolve {@code COUNT}
+ /// @param explainPlanName explain-plan name used for periodic query
termination checks
+ /// @return the result resolved from dictionary/metadata for the function
+ /// @throws IllegalStateException if the function type cannot be resolved
from dictionary or metadata
+ public static Object getAggregationResult(AggregationFunction
aggregationFunction, @Nullable DataSource dataSource,
+ int numTotalDocs, String explainPlanName) {
+ AggregationFunctionType functionType = aggregationFunction.getType();
+ if (functionType == AggregationFunctionType.COUNT) {
+ return (long) numTotalDocs;
+ }
+ // Every other supported function resolves its result from the column
dictionary or metadata, all of which require
+ // a non-null data source.
+ Objects.requireNonNull(dataSource, "DataSource is null for aggregation
function: " + functionType);
+
+ Object result;
+ switch (functionType) {
+ case MIN:
+ case MINMV:
+ result = getMinValueNumeric(dataSource);
+ break;
+ case MINLONG:
+ result = getMinValueLong(dataSource);
+ break;
+ case MINSTRING:
+ assert dataSource.getDictionary() != null;
+ result = dataSource.getDictionary().getMinVal();
+ break;
+ case MAX:
+ case MAXMV:
+ result = getMaxValueNumeric(dataSource);
+ break;
+ case MAXLONG:
+ result = getMaxValueLong(dataSource);
+ break;
+ case MAXSTRING:
+ assert dataSource.getDictionary() != null;
+ result = dataSource.getDictionary().getMaxVal();
+ break;
+ case MINMAXRANGE:
+ case MINMAXRANGEMV:
+ result = new MinMaxRangePair(getMinValueNumeric(dataSource),
getMaxValueNumeric(dataSource));
+ break;
+ case DISTINCTCOUNT:
+ case DISTINCTSUM:
+ case DISTINCTAVG:
+ case DISTINCTCOUNTMV:
+ case DISTINCTSUMMV:
+ case DISTINCTAVGMV:
+ result =
getDistinctValueSet(Objects.requireNonNull(dataSource.getDictionary()),
explainPlanName);
+ break;
+ case DISTINCTCOUNTOFFHEAP:
+ result = ((DistinctCountOffHeapAggregationFunction)
aggregationFunction).extractAggregationResult(
+ Objects.requireNonNull(dataSource.getDictionary()));
+ break;
+ case DISTINCTCOUNTHLL:
+ case DISTINCTCOUNTHLLMV:
+ result =
getDistinctCountHLLResult(Objects.requireNonNull(dataSource.getDictionary()),
+ (DistinctCountHLLAggregationFunction) aggregationFunction,
explainPlanName);
+ break;
+ case DISTINCTCOUNTRAWHLL:
+ case DISTINCTCOUNTRAWHLLMV:
+ result =
getDistinctCountHLLResult(Objects.requireNonNull(dataSource.getDictionary()),
+ ((DistinctCountRawHLLAggregationFunction)
aggregationFunction).getDistinctCountHLLAggregationFunction(),
+ explainPlanName);
+ break;
+ case DISTINCTCOUNTHLLPLUS:
+ case DISTINCTCOUNTHLLPLUSMV:
+ result =
getDistinctCountHLLPlusResult(Objects.requireNonNull(dataSource.getDictionary()),
+ (DistinctCountHLLPlusAggregationFunction) aggregationFunction,
explainPlanName);
+ break;
+ case DISTINCTCOUNTRAWHLLPLUS:
+ case DISTINCTCOUNTRAWHLLPLUSMV:
+ result =
getDistinctCountHLLPlusResult(Objects.requireNonNull(dataSource.getDictionary()),
+ ((DistinctCountRawHLLPlusAggregationFunction) aggregationFunction)
+ .getDistinctCountHLLPlusAggregationFunction(),
explainPlanName);
+ break;
+ case SEGMENTPARTITIONEDDISTINCTCOUNT:
+ result = (long)
Objects.requireNonNull(dataSource.getDictionary()).length();
+ break;
+ case DISTINCTCOUNTSMARTHLL:
+ result =
getDistinctCountSmartHLLResult(Objects.requireNonNull(dataSource.getDictionary()),
+ (DistinctCountSmartHLLAggregationFunction) aggregationFunction,
explainPlanName);
+ break;
+ case DISTINCTCOUNTSMARTHLLPLUS:
+ result =
getDistinctCountSmartHLLPlusResult(Objects.requireNonNull(dataSource.getDictionary()),
+ (DistinctCountSmartHLLPlusAggregationFunction)
aggregationFunction, explainPlanName);
+ break;
+ case DISTINCTCOUNTULL:
+ result =
getDistinctCountULLResult(Objects.requireNonNull(dataSource.getDictionary()),
+ (DistinctCountULLAggregationFunction) aggregationFunction,
explainPlanName);
+ break;
+ case DISTINCTCOUNTSMARTULL:
+ result =
getDistinctCountSmartULLResult(Objects.requireNonNull(dataSource.getDictionary()),
+ (DistinctCountSmartULLAggregationFunction) aggregationFunction,
explainPlanName);
+ break;
+ case DISTINCTCOUNTRAWULL:
+ result =
getDistinctCountULLResult(Objects.requireNonNull(dataSource.getDictionary()),
+ (DistinctCountULLAggregationFunction) aggregationFunction,
explainPlanName);
+ break;
+ default:
+ throw new IllegalStateException(
+ "Non-scan based aggregation operator does not support function
type: " + functionType);
+ }
+
+ return result;
+ }
+
+ private static Double getMinValueNumeric(DataSource dataSource) {
+ Dictionary dictionary = dataSource.getDictionary();
+ if (dictionary != null) {
+ return toDouble(dictionary.getMinVal());
+ }
+ return toDouble(dataSource.getDataSourceMetadata().getMinValue());
+ }
+
+ private static Long getMinValueLong(DataSource dataSource) {
+ FieldSpec.DataType dataType =
dataSource.getDataSourceMetadata().getDataType().getStoredType();
+ Preconditions.checkArgument(
+ dataType == FieldSpec.DataType.LONG || dataType ==
FieldSpec.DataType.INT,
+ "MINLONG aggregation function can only be applied to columns of
integer types");
+ Dictionary dictionary = dataSource.getDictionary();
+ if (dictionary != null) {
+ return ((Number) dictionary.getMinVal()).longValue();
+ }
+ return ((Number)
dataSource.getDataSourceMetadata().getMinValue()).longValue();
+ }
+
+ private static Double getMaxValueNumeric(DataSource dataSource) {
+ Dictionary dictionary = dataSource.getDictionary();
+ if (dictionary != null) {
+ return toDouble(dictionary.getMaxVal());
+ }
+ return toDouble(dataSource.getDataSourceMetadata().getMaxValue());
+ }
+
+ private static Long getMaxValueLong(DataSource dataSource) {
+ FieldSpec.DataType dataType =
dataSource.getDataSourceMetadata().getDataType().getStoredType();
+ Preconditions.checkArgument(
+ dataType == FieldSpec.DataType.LONG || dataType ==
FieldSpec.DataType.INT,
+ "MAXLONG aggregation function can only be applied to columns of
integer types");
+ Dictionary dictionary = dataSource.getDictionary();
+ if (dictionary != null) {
+ return ((Number) dictionary.getMaxVal()).longValue();
+ }
+ return ((Number)
dataSource.getDataSourceMetadata().getMaxValue()).longValue();
+ }
+
+ private static Double toDouble(Comparable<?> value) {
+ if (value instanceof Double) {
+ return (Double) value;
+ } else if (value instanceof Number) {
+ return ((Number) value).doubleValue();
+ } else {
+ return Double.parseDouble(value.toString());
+ }
+ }
+
+ private static Set getDistinctValueSet(Dictionary dictionary, String
explainPlanName) {
+ int dictionarySize = dictionary.length();
+ switch (dictionary.getValueType()) {
+ case INT:
+ IntOpenHashSet intSet = new IntOpenHashSet(dictionarySize);
+ for (int dictId = 0; dictId < dictionarySize; dictId++) {
+
QueryThreadContext.checkTerminationAndSampleUsagePeriodically(dictId,
explainPlanName);
+ intSet.add(dictionary.getIntValue(dictId));
+ }
+ return intSet;
+ case LONG:
+ LongOpenHashSet longSet = new LongOpenHashSet(dictionarySize);
+ for (int dictId = 0; dictId < dictionarySize; dictId++) {
+
QueryThreadContext.checkTerminationAndSampleUsagePeriodically(dictId,
explainPlanName);
+ longSet.add(dictionary.getLongValue(dictId));
+ }
+ return longSet;
+ case FLOAT:
+ FloatOpenHashSet floatSet = new FloatOpenHashSet(dictionarySize);
+ for (int dictId = 0; dictId < dictionarySize; dictId++) {
+
QueryThreadContext.checkTerminationAndSampleUsagePeriodically(dictId,
explainPlanName);
+ floatSet.add(dictionary.getFloatValue(dictId));
+ }
+ return floatSet;
+ case DOUBLE:
+ DoubleOpenHashSet doubleSet = new DoubleOpenHashSet(dictionarySize);
+ for (int dictId = 0; dictId < dictionarySize; dictId++) {
+
QueryThreadContext.checkTerminationAndSampleUsagePeriodically(dictId,
explainPlanName);
+ doubleSet.add(dictionary.getDoubleValue(dictId));
+ }
+ return doubleSet;
+ case BIG_DECIMAL:
+ ObjectOpenHashSet<BigDecimal> bigDecimalSet = new
ObjectOpenHashSet<>(dictionarySize);
+ for (int dictId = 0; dictId < dictionarySize; dictId++) {
+
QueryThreadContext.checkTerminationAndSampleUsagePeriodically(dictId,
explainPlanName);
+ bigDecimalSet.add(dictionary.getBigDecimalValue(dictId));
+ }
+ return bigDecimalSet;
+ case STRING:
+ ObjectOpenHashSet<String> stringSet = new
ObjectOpenHashSet<>(dictionarySize);
+ for (int dictId = 0; dictId < dictionarySize; dictId++) {
+
QueryThreadContext.checkTerminationAndSampleUsagePeriodically(dictId,
explainPlanName);
+ stringSet.add(dictionary.getStringValue(dictId));
+ }
+ return stringSet;
+ case BYTES:
+ ObjectOpenHashSet<ByteArray> bytesSet = new
ObjectOpenHashSet<>(dictionarySize);
+ for (int dictId = 0; dictId < dictionarySize; dictId++) {
+
QueryThreadContext.checkTerminationAndSampleUsagePeriodically(dictId,
explainPlanName);
+ bytesSet.add(new ByteArray(dictionary.getBytesValue(dictId)));
+ }
+ return bytesSet;
+ default:
+ throw new IllegalStateException();
+ }
+ }
+
+ private static HyperLogLog getDistinctValueHLL(Dictionary dictionary, int
log2m, String explainPlanName) {
+ HyperLogLog hll = new HyperLogLog(log2m);
+ int length = dictionary.length();
+ for (int i = 0; i < length; i++) {
+ QueryThreadContext.checkTerminationAndSampleUsagePeriodically(i,
explainPlanName);
+ hll.offer(dictionary.get(i));
+ }
+ return hll;
+ }
+
+ private static UltraLogLog getDistinctValueULL(Dictionary dictionary, int p,
String explainPlanName) {
+ UltraLogLog ull = UltraLogLog.create(p);
+ int length = dictionary.length();
+ for (int i = 0; i < length; i++) {
+ QueryThreadContext.checkTerminationAndSampleUsagePeriodically(i,
explainPlanName);
+ Object value = dictionary.get(i);
+ UltraLogLogUtils.hashObject(value).ifPresent(ull::add);
+ }
+ return ull;
+ }
+
+ private static HyperLogLogPlus getDistinctValueHLLPlus(Dictionary
dictionary, int p, int sp, String explainPlanName) {
+ HyperLogLogPlus hllPlus = new HyperLogLogPlus(p, sp);
+ int length = dictionary.length();
+ for (int i = 0; i < length; i++) {
+ QueryThreadContext.checkTerminationAndSampleUsagePeriodically(i,
explainPlanName);
+ hllPlus.offer(dictionary.get(i));
+ }
+ return hllPlus;
+ }
+
+ private static HyperLogLog getDistinctCountHLLResult(Dictionary dictionary,
+ DistinctCountHLLAggregationFunction function, String explainPlanName) {
+ if (dictionary.getValueType() == FieldSpec.DataType.BYTES) {
+ // Treat BYTES value as serialized HyperLogLog
+ try {
+ QueryThreadContext.checkTerminationAndSampleUsage(explainPlanName);
+ HyperLogLog hll =
ObjectSerDeUtils.HYPER_LOG_LOG_SER_DE.deserialize(dictionary.getBytesValue(0));
+ int length = dictionary.length();
+ for (int i = 1; i < length; i++) {
+ QueryThreadContext.checkTerminationAndSampleUsagePeriodically(i,
explainPlanName);
+
hll.addAll(ObjectSerDeUtils.HYPER_LOG_LOG_SER_DE.deserialize(dictionary.getBytesValue(i)));
+ }
+ return hll;
+ } catch (Exception e) {
+ throw new RuntimeException("Caught exception while merging
HyperLogLogs", e);
+ }
+ } else {
+ return getDistinctValueHLL(dictionary, function.getLog2m(),
explainPlanName);
+ }
+ }
+
+ private static HyperLogLogPlus getDistinctCountHLLPlusResult(Dictionary
dictionary,
+ DistinctCountHLLPlusAggregationFunction function, String
explainPlanName) {
+ if (dictionary.getValueType() == FieldSpec.DataType.BYTES) {
+ // Treat BYTES value as serialized HyperLogLogPlus
+ try {
+ QueryThreadContext.checkTerminationAndSampleUsage(explainPlanName);
+ HyperLogLogPlus hllplus =
ObjectSerDeUtils.HYPER_LOG_LOG_PLUS_SER_DE.deserialize(dictionary.getBytesValue(0));
+ int length = dictionary.length();
+ for (int i = 1; i < length; i++) {
+ QueryThreadContext.checkTerminationAndSampleUsagePeriodically(i,
explainPlanName);
+
hllplus.addAll(ObjectSerDeUtils.HYPER_LOG_LOG_PLUS_SER_DE.deserialize(dictionary.getBytesValue(i)));
+ }
+ return hllplus;
+ } catch (Exception e) {
+ throw new RuntimeException("Caught exception while merging
HyperLogLogPluses", e);
+ }
+ } else {
+ return getDistinctValueHLLPlus(dictionary, function.getP(),
function.getSp(), explainPlanName);
+ }
+ }
+
+ private static Object getDistinctCountSmartHLLResult(Dictionary dictionary,
+ DistinctCountSmartHLLAggregationFunction function, String
explainPlanName) {
+ if (dictionary.length() > function.getThreshold()) {
+ // Store values into a HLL when the dictionary size exceeds the
conversion threshold
+ return getDistinctValueHLL(dictionary, function.getLog2m(),
explainPlanName);
+ } else {
+ return getDistinctValueSet(dictionary, explainPlanName);
+ }
+ }
+
+ private static Object getDistinctCountSmartHLLPlusResult(Dictionary
dictionary,
+ DistinctCountSmartHLLPlusAggregationFunction function, String
explainPlanName) {
+ if (dictionary.length() > function.getThreshold()) {
+ // Store values into a HLLPlus when the dictionary size exceeds the
conversion threshold
+ return getDistinctValueHLLPlus(dictionary, function.getP(),
function.getSp(), explainPlanName);
+ } else {
+ return getDistinctValueSet(dictionary, explainPlanName);
+ }
+ }
+
+ private static UltraLogLog getDistinctCountULLResult(Dictionary dictionary,
+ DistinctCountULLAggregationFunction function, String explainPlanName) {
+ if (dictionary.getValueType() == FieldSpec.DataType.BYTES) {
+ // Treat BYTES value as serialized UltraLogLog and merge
+ try {
+ QueryThreadContext.checkTerminationAndSampleUsage(explainPlanName);
+ UltraLogLog ull =
ObjectSerDeUtils.ULTRA_LOG_LOG_OBJECT_SER_DE.deserialize(dictionary.getBytesValue(0));
+ int length = dictionary.length();
+ for (int i = 1; i < length; i++) {
+ QueryThreadContext.checkTerminationAndSampleUsagePeriodically(i,
explainPlanName);
+
ull.add(ObjectSerDeUtils.ULTRA_LOG_LOG_OBJECT_SER_DE.deserialize(dictionary.getBytesValue(i)));
+ }
+ return ull;
+ } catch (Exception e) {
+ throw new RuntimeException("Caught exception while merging
UltraLogLogs", e);
+ }
+ } else {
+ return getDistinctValueULL(dictionary, function.getP(), explainPlanName);
+ }
+ }
+
+ private static Object getDistinctCountSmartULLResult(Dictionary dictionary,
+ DistinctCountSmartULLAggregationFunction function, String
explainPlanName) {
+ if (dictionary.length() > function.getThreshold()) {
+ return getDistinctValueULL(dictionary, function.getP(), explainPlanName);
+ } else {
+ return getDistinctValueSet(dictionary, explainPlanName);
+ }
+ }
}
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 f9a032dfe77..0874619f5b5 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,7 +33,10 @@ import
org.apache.pinot.segment.spi.index.startree.AggregationFunctionColumnPair
public class StarTreeAggregationExecutor extends DefaultAggregationExecutor {
private final AggregationFunctionColumnPair[]
_aggregationFunctionColumnPairs;
+
public StarTreeAggregationExecutor(AggregationFunction[]
aggregationFunctions) {
+ // 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;
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/plan/maker/MetadataAndDictionaryAggregationPlanMakerTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/plan/maker/MetadataAndDictionaryAggregationPlanMakerTest.java
index 769bd57b9d6..8cd0278f734 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/plan/maker/MetadataAndDictionaryAggregationPlanMakerTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/plan/maker/MetadataAndDictionaryAggregationPlanMakerTest.java
@@ -22,10 +22,14 @@ import java.io.File;
import java.net.URL;
import java.util.ArrayList;
import java.util.List;
+import java.util.Set;
import java.util.concurrent.TimeUnit;
import org.apache.commons.io.FileUtils;
import org.apache.pinot.common.metrics.ServerMetrics;
import org.apache.pinot.core.common.Operator;
+import org.apache.pinot.core.operator.BaseProjectOperator;
+import org.apache.pinot.core.operator.ExecutionStatistics;
+import org.apache.pinot.core.operator.blocks.results.AggregationResultsBlock;
import org.apache.pinot.core.operator.query.AggregationOperator;
import org.apache.pinot.core.operator.query.FastFilteredCountOperator;
import org.apache.pinot.core.operator.query.GroupByOperator;
@@ -37,6 +41,7 @@ import
org.apache.pinot.segment.local.data.manager.TableDataManager;
import
org.apache.pinot.segment.local.indexsegment.immutable.ImmutableSegmentImpl;
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.upsert.ConcurrentMapPartitionUpsertMetadataManager;
import org.apache.pinot.segment.local.upsert.UpsertContext;
import org.apache.pinot.segment.local.upsert.UpsertUtils;
@@ -44,7 +49,9 @@ import org.apache.pinot.segment.spi.IndexSegment;
import org.apache.pinot.segment.spi.SegmentContext;
import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig;
import org.apache.pinot.segment.spi.creator.SegmentIndexCreationDriver;
+import org.apache.pinot.segment.spi.datasource.DataSource;
import
org.apache.pinot.segment.spi.index.mutable.ThreadSafeMutableRoaringBitmap;
+import org.apache.pinot.segment.spi.index.reader.Dictionary;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.config.table.ingestion.IngestionConfig;
@@ -52,6 +59,8 @@ 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.TimeGranularitySpec;
+import org.apache.pinot.spi.data.readers.GenericRow;
+import org.apache.pinot.spi.exception.BadQueryRequestException;
import org.apache.pinot.spi.utils.ReadMode;
import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
import org.testng.annotations.AfterClass;
@@ -61,21 +70,34 @@ import org.testng.annotations.BeforeTest;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
+import static org.mockito.AdditionalAnswers.delegatesTo;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.expectThrows;
public class MetadataAndDictionaryAggregationPlanMakerTest {
private static final String AVRO_DATA = "data" + File.separator +
"test_data-sv.avro";
private static final String SEGMENT_NAME = "testTable_201711219_20171120";
+ // A small segment with fully predictable data so aggregation results can be
asserted exactly. It holds the 10 rows
+ // metricCol = i, intCol = i + 10, bytesCol = new byte[]{(byte) i} for i =
1..10, with every row duplicated (20 rows
+ // total).
+ private static final String PREDICTABLE_TABLE_NAME = "predictableTable";
+ private static final String PREDICTABLE_SEGMENT_NAME =
"predictableTable_segment";
private static final File INDEX_DIR =
new File(FileUtils.getTempDirectory(),
"MetadataAndDictionaryAggregationPlanMakerTest");
private static final InstancePlanMakerImplV2 PLAN_MAKER = new
InstancePlanMakerImplV2();
private IndexSegment _indexSegment;
private IndexSegment _upsertIndexSegment;
+ private IndexSegment _predictableSegment;
@BeforeTest
public void buildSegment()
@@ -122,6 +144,43 @@ public class MetadataAndDictionaryAggregationPlanMakerTest
{
SegmentIndexCreationDriver driver = new SegmentIndexCreationDriverImpl();
driver.init(segmentGeneratorConfig);
driver.build();
+
+ buildPredictableSegment();
+ }
+
+ /// Builds a segment with fully predictable data so that aggregation results
can be asserted exactly. It contains the
+ /// 10 rows where row {@code i} (for {@code i = 1..10}) has {@code metricCol
= i}, {@code intCol = i + 10} and
+ /// {@code bytesCol = new byte[]{(byte) i}}, and every row is duplicated (20
rows total).
+ private void buildPredictableSegment()
+ throws Exception {
+ Schema schema = new
Schema.SchemaBuilder().setSchemaName(PREDICTABLE_TABLE_NAME)
+ .addMetric("metricCol", FieldSpec.DataType.INT)
+ .addMetric("intCol", FieldSpec.DataType.INT)
+ .addSingleValueDimension("bytesCol", FieldSpec.DataType.BYTES)
+ .build();
+ TableConfig tableConfig =
+ new
TableConfigBuilder(TableType.OFFLINE).setTableName(PREDICTABLE_TABLE_NAME).build();
+
+ List<GenericRow> records = new ArrayList<>(20);
+ // Add each row twice.
+ for (int n = 0; n < 2; n++) {
+ for (int i = 1; i <= 10; i++) {
+ GenericRow record = new GenericRow();
+ record.putValue("metricCol", i);
+ record.putValue("intCol", i + 10);
+ record.putValue("bytesCol", new byte[]{(byte) i});
+ records.add(record);
+ }
+ }
+
+ SegmentGeneratorConfig segmentGeneratorConfig = new
SegmentGeneratorConfig(tableConfig, schema);
+ segmentGeneratorConfig.setTableName(PREDICTABLE_TABLE_NAME);
+ segmentGeneratorConfig.setSegmentName(PREDICTABLE_SEGMENT_NAME);
+ segmentGeneratorConfig.setOutDir(INDEX_DIR.getAbsolutePath());
+
+ SegmentIndexCreationDriverImpl driver = new
SegmentIndexCreationDriverImpl();
+ driver.init(segmentGeneratorConfig, new GenericRowRecordReader(records));
+ driver.build();
}
@BeforeClass
@@ -130,6 +189,7 @@ public class MetadataAndDictionaryAggregationPlanMakerTest {
ServerMetrics.register(mock(ServerMetrics.class));
_indexSegment = ImmutableSegmentLoader.load(new File(INDEX_DIR,
SEGMENT_NAME), ReadMode.heap);
_upsertIndexSegment = ImmutableSegmentLoader.load(new File(INDEX_DIR,
SEGMENT_NAME), ReadMode.heap);
+ _predictableSegment = ImmutableSegmentLoader.load(new File(INDEX_DIR,
PREDICTABLE_SEGMENT_NAME), ReadMode.heap);
TableDataManager tableDataManager = mock(TableDataManager.class);
when(tableDataManager.getTableDataDir()).thenReturn(INDEX_DIR);
UpsertContext upsertContext = new UpsertContext.Builder()
@@ -150,6 +210,7 @@ public class MetadataAndDictionaryAggregationPlanMakerTest {
_indexSegment.destroy();
_upsertIndexSegment.offload();
_upsertIndexSegment.destroy();
+ _predictableSegment.destroy();
}
@AfterTest
@@ -170,6 +231,211 @@ public class
MetadataAndDictionaryAggregationPlanMakerTest {
assertTrue(upsertOperatorClass.isInstance(upsertOperator));
}
+ /// Verifies the partial metadata-based aggregation path. When a query mixes
a metadata-eligible function (MAX) with a
+ /// non-eligible one (SUM), the plan uses an {@link AggregationOperator}
that pre-aggregates the eligible function
+ /// from metadata while scanning the rest. Before this feature, a non-scan
based operator was used only when _all_
+ /// functions were metadata eligible; a mixed query would have scanned every
function.
+ ///
+ /// To prove the eligible function is actually served from metadata (and not
scanned), the column dictionary is
+ /// overridden to report a bogus max value that does not exist in the data.
MAX equals the bogus value only if the
+ /// metadata path is taken (a scan would return the true max of 10), while
SUM equals the true scanned sum of
+ /// {@code metricCol} ((1 + 2 + ... + 10) over the two row copies = 110).
+ @Test
+ public void testPartialMetadataBasedAggregationServesEligibleFromMetadata() {
+ int bogusMax = 999_999_999;
+ QueryContext queryContext = QueryContextConverterUtils.getQueryContext(
+ "select max(metricCol), sum(metricCol) from " +
PREDICTABLE_TABLE_NAME);
+
+ // Override only metricCol's dictionary max value; everything else
delegates to the real segment.
+ DataSource realDataSource = _predictableSegment.getDataSource("metricCol",
queryContext.getSchema());
+ Dictionary dictionaryWithBogusMax = mock(Dictionary.class,
delegatesTo(realDataSource.getDictionary()));
+ doReturn(bogusMax).when(dictionaryWithBogusMax).getMaxVal();
+ DataSource dataSourceWithBogusMax = mock(DataSource.class,
delegatesTo(realDataSource));
+
doReturn(dictionaryWithBogusMax).when(dataSourceWithBogusMax).getDictionary();
+ IndexSegment segment = mock(IndexSegment.class,
delegatesTo(_predictableSegment));
+
doReturn(dataSourceWithBogusMax).when(segment).getDataSource(eq("metricCol"),
any());
+
+ Operator<?> operator = PLAN_MAKER.makeSegmentPlanNode(new
SegmentContext(segment), queryContext).run();
+ // A mixed query must use the (partial) AggregationOperator, not the fully
non-scan based operator.
+ assertTrue(operator instanceof AggregationOperator);
+
+ AggregationResultsBlock resultsBlock = (AggregationResultsBlock)
operator.nextBlock();
+ List<Object> results = resultsBlock.getResults();
+ assertNotNull(results);
+ // MAX is served from the (overridden) dictionary metadata, so the bogus
value proves the metadata path was used.
+ assertEquals(((Number) results.get(0)).doubleValue(), (double) bogusMax);
+ // SUM is scanned, so the result is the true sum of metricCol: (1 + 2 +
... + 10) over the two row copies = 110.
+ assertEquals(((Number) results.get(1)).doubleValue(), 110.0);
+ }
+
+ /// {@code distinctcount} over a dictionary-encoded column is resolved
entirely from the dictionary (its cardinality
+ /// is the number of distinct values), so it must plan to the fully non-scan
{@link NonScanBasedAggregationOperator}
+ /// without a mock. {@code metricCol} holds the 10 distinct values 1..10
across 20 (duplicated) rows, so the distinct
+ /// count is 10 even though the segment has 20 docs.
+ @Test
+ public void testDistinctCountResolvedFromDictionary() {
+ // Sanity check that the fixture actually has duplicate rows, so distinct
count < total docs is a meaningful result.
+ assertEquals(_predictableSegment.getSegmentMetadata().getTotalDocs(), 20);
+
+ QueryContext queryContext =
+ QueryContextConverterUtils.getQueryContext("select
distinctcount(metricCol) from " + PREDICTABLE_TABLE_NAME);
+
+ Operator<?> operator =
+ PLAN_MAKER.makeSegmentPlanNode(new
SegmentContext(_predictableSegment), queryContext).run();
+ // The column is dictionary-encoded, so DISTINCTCOUNT is resolved from the
dictionary without scanning the segment.
+ assertTrue(operator instanceof NonScanBasedAggregationOperator);
+
+ AggregationResultsBlock resultsBlock = (AggregationResultsBlock)
operator.nextBlock();
+ List<Object> results = resultsBlock.getResults();
+ assertNotNull(results);
+ // The intermediate result is the set of distinct dictionary values:
metricCol has 10 distinct values (1..10),
+ // fewer than the 20 total docs, proving DISTINCTCOUNT counts distinct
values rather than rows.
+ assertEquals(((Set<?>) results.get(0)).size(), 10);
+ }
+
+ /// Tests for aggregations that cannot be resolved from dictionary/metadata
and must therefore fall back to
+ /// the scan-based {@link AggregationOperator}, still returning the correct
scanned result:
+ ///
+ /// - a single non-resolvable aggregation with no filter ({@code
sum(metricCol)}): the partial metadata path
+ /// evaluates eligibility per function, and when none is resolvable it
must fall back to a full scan rather than
+ /// emitting a (partial) non-scan operator with zero resolved functions;
+ /// - an aggregation over an expression argument ({@code max(add(metricCol,
intCol))}): the argument is not a plain
+ /// column reference, so it cannot be resolved from dictionary/metadata.
+ @Test(dataProvider = "nonResolvableScanQueries")
+ public void testNonResolvableAggregationFallsToScan(String description,
String query, double expectedResult) {
+ QueryContext queryContext =
QueryContextConverterUtils.getQueryContext(query);
+
+ Operator<?> operator =
+ PLAN_MAKER.makeSegmentPlanNode(new
SegmentContext(_predictableSegment), queryContext).run();
+ assertTrue(operator instanceof AggregationOperator, description);
+
+ AggregationResultsBlock resultsBlock = (AggregationResultsBlock)
operator.nextBlock();
+ List<Object> results = resultsBlock.getResults();
+ assertNotNull(results, description);
+ assertEquals(((Number) results.get(0)).doubleValue(), expectedResult,
description);
+ }
+
+ @DataProvider(name = "nonResolvableScanQueries")
+ public Object[][] nonResolvableScanQueries() {
+ return new Object[][]{
+ // SUM is not resolvable from metadata and is scanned: (1 + 2 + ... +
10) over the two row copies = 110.
+ {"single non-resolvable aggregation", "select sum(metricCol) from " +
PREDICTABLE_TABLE_NAME, 110.0},
+ // add(metricCol, intCol) = i + (i + 10) is an expression argument, so
it is scanned; max over i = 1..10 is 30.
+ {"aggregation over expression argument",
+ "select max(add(metricCol, intCol)) from " +
PREDICTABLE_TABLE_NAME, 30.0}
+ };
+ }
+
+ /// MIN/MAX derive their result numerically from the column min/max, which
is only valid for numeric columns.
+ /// For a non-numeric (BYTES) column the dictionary stores raw values that
cannot be
+ /// parsed as numbers, so {@code max(bytesCol)} and {@code min(bytesCol)}
must fall back to the scan-based
+ /// {@link AggregationOperator} rather than being (wrongly) resolved from
the dictionary by a
+ /// {@link NonScanBasedAggregationOperator}. The scan path in turn throws a
+ /// {@link BadQueryRequestException} for non-numeric aggregation.
+ @Test
+ public void testMinMaxOnNonNumericColumnFallsToScan() {
+
assertNotNull(_predictableSegment.getDataSourceNullable("bytesCol").getDictionary());
+
+ for (String function : List.of("max", "min")) {
+ String query = "select " + function + "(bytesCol) from " +
PREDICTABLE_TABLE_NAME;
+ QueryContext queryContext =
QueryContextConverterUtils.getQueryContext(query);
+ Operator<?> operator =
+ PLAN_MAKER.makeSegmentPlanNode(new
SegmentContext(_predictableSegment), queryContext).run();
+ // MIN/MAX on a non-numeric column must fall back to the scan-based
AggregationOperator, not the metadata path.
+ assertTrue(operator instanceof AggregationOperator, query);
+ assertFalse(operator instanceof NonScanBasedAggregationOperator, query);
+
+ // Running the scan path proves it still executes and correctly rejects
the non-numeric aggregation rather than
+ // producing a wrong result.
+ BadQueryRequestException exception =
expectThrows(BadQueryRequestException.class, operator::nextBlock);
+ assertTrue(exception.getMessage().contains("Cannot compute " + function
+ " for non-numeric type: BYTES"),
+ exception.getMessage());
+ }
+ }
+
+ /// Verifies that a column aggregation resolved by metadata is excluded from
the scan projection,
+ /// so the partial path only reads the columns needed by the scanned
aggregations. The reduced projection is also
+ /// reflected in numEntriesScannedPostFilter (20 docs * 1 column instead of
40).
+ @Test
+ public void testResolvedOnlyColumnExcludedFromProjection() {
+ QueryContext queryContext = QueryContextConverterUtils.getQueryContext(
+ "select max(metricCol), sum(intCol) from " + PREDICTABLE_TABLE_NAME);
+ Operator<?> operator =
+ PLAN_MAKER.makeSegmentPlanNode(new
SegmentContext(_predictableSegment), queryContext).run();
+ assertTrue(operator instanceof AggregationOperator);
+
+ BaseProjectOperator<?> projectOperator = ((AggregationOperator)
operator).getChildOperators().get(0);
+ // Only intCol (used by the scanned sum) is projected; metricCol (used
solely by the resolved max) is excluded.
+ assertEquals(projectOperator.getNumColumnsProjected(), 1);
+
assertTrue(projectOperator.getSourceColumnContextMap().containsKey("intCol"));
+
assertFalse(projectOperator.getSourceColumnContextMap().containsKey("metricCol"));
+
+ AggregationResultsBlock resultsBlock = (AggregationResultsBlock)
operator.nextBlock();
+ List<Object> results = resultsBlock.getResults();
+ assertNotNull(results);
+ // max(metricCol) is resolved from the dictionary; sum(intCol) is scanned.
+ // intCol = i + 10 for i = 1..10 over the two row copies, so sum(intCol) =
2 * (11 + 12 + ... + 20) = 310.
+ assertEquals(((Number) results.get(0)).doubleValue(), 10.0);
+ assertEquals(((Number) results.get(1)).doubleValue(), 310.0);
+
+ // All 20 docs are still scanned for the unresolved sum(intCol).
+ assertEquals(operator.getExecutionStatistics().getNumDocsScanned(), 20);
+ // With metricCol excluded, only 1 column is projected, so entries scanned
post-filter is 20 docs * 1 column = 20
+ // (it would be 40 if the resolved-only metricCol were still projected).
+
assertEquals(operator.getExecutionStatistics().getNumEntriesScannedPostFilter(),
20);
+ }
+
+ /// Verifies numDocsScanned and numEntriesScannedPostFilter across the three
routing paths: partial metadata
+ /// resolution, fully metadata-resolvable (non-scan), and full scan over
multiple aggregations.
+ @Test(dataProvider = "scanCostStatistics")
+ public void testScanCostStatistics(String description, String query,
Class<?> expectedOperatorClass,
+ long expectedNumDocsScanned, long expectedNumEntriesScannedPostFilter) {
+ QueryContext queryContext =
QueryContextConverterUtils.getQueryContext(query);
+ Operator<?> operator =
+ PLAN_MAKER.makeSegmentPlanNode(new
SegmentContext(_predictableSegment), queryContext).run();
+ assertTrue(expectedOperatorClass.isInstance(operator), description);
+ // Drain the blocks so scan operators accumulate their document counts
(harmless for the non-scan operator, whose
+ // statistics are constant).
+ operator.nextBlock();
+ ExecutionStatistics stats = operator.getExecutionStatistics();
+ assertEquals(stats.getNumDocsScanned(), expectedNumDocsScanned,
description);
+ assertEquals(stats.getNumEntriesScannedPostFilter(),
expectedNumEntriesScannedPostFilter, description);
+ }
+
+ @DataProvider(name = "scanCostStatistics")
+ public Object[][] scanCostStatistics() {
+ return new Object[][]{
+ // Partial metadata resolution: max(metricCol) is resolved from the
dictionary and only sum(intCol) is
+ // scanned, so a single column is projected -> 20 docs * 1 column = 20
entries scanned post-filter.
+ {
+ "partial metadata resolution",
+ "select max(metricCol), sum(intCol) from " +
PREDICTABLE_TABLE_NAME,
+ AggregationOperator.class,
+ 20L,
+ 20L
+ },
+ // Fully metadata-resolvable: max + min on the numeric dictionary
column are both resolved, so the non-scan
+ // operator is used and nothing is scanned post-filter. numDocsScanned
is reported as numTotalDocs (20) for
+ // backward compatibility even though no rows are actually scanned.
+ {
+ "fully metadata-resolvable",
+ "select max(metricCol), min(metricCol) from " +
PREDICTABLE_TABLE_NAME,
+ NonScanBasedAggregationOperator.class,
+ 20L,
+ 0L
+ },
+ // Full scan over multiple aggregations: neither sum is resolvable, so
both columns are scanned for all 20
+ // docs -> 20 docs * 2 columns = 40 entries scanned post-filter.
+ {
+ "full scan multiple aggregations",
+ "select sum(metricCol), sum(intCol) from " +
PREDICTABLE_TABLE_NAME,
+ AggregationOperator.class,
+ 20L,
+ 40L
+ }
+ };
+ }
+
@DataProvider(name = "testPlanMakerDataProvider")
public Object[][] testPlanMakerDataProvider() {
List<Object[]> entries = new ArrayList<>();
@@ -223,6 +489,9 @@ public class MetadataAndDictionaryAggregationPlanMakerTest {
entries.add(new Object[]{
"select sum(column1) from testTable", AggregationOperator.class,
AggregationOperator.class
});
+ entries.add(new Object[]{
+ "select count(*), sum(column1) from testTable",
AggregationOperator.class, AggregationOperator.class
+ });
// Aggregation group-by
entries.add(new Object[]{
"select sum(column1) from testTable group by daysSinceEpoch",
GroupByOperator.class, GroupByOperator.class
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/DefaultAggregationExecutorTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/DefaultAggregationExecutorTest.java
index 776a16bd6f4..38feebf12f9 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/DefaultAggregationExecutorTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/DefaultAggregationExecutorTest.java
@@ -114,6 +114,62 @@ public class DefaultAggregationExecutorTest {
/// Asserts that the aggregation results returned by the executor are as
expected.
@Test
void testAggregation() {
+ TransformBlock transformBlock = nextTransformBlock();
+ AggregationFunction[] aggregationFunctions =
_queryContext.getAggregationFunctions();
+ assert aggregationFunctions != null;
+ AggregationExecutor aggregationExecutor = new
DefaultAggregationExecutor(aggregationFunctions);
+ aggregationExecutor.aggregate(transformBlock);
+ List<Object> result = aggregationExecutor.getResult();
+ for (int i = 0; i < result.size(); i++) {
+ double actual = (double) result.get(i);
+ double expected = computeAggregation(AGGREGATION_FUNCTIONS[i],
_inputData[i]);
+ Assert.assertEquals(actual, expected,
+ "Aggregation mis-match for function " + AGGREGATION_FUNCTIONS[i] +
", Expected: " + expected + " Actual: "
+ + actual);
+ }
+ }
+
+ /// Verifies that functions with a non-scan result are not re-computed by
scanning: the injected value is
+ /// returned as-is, while functions with a {@code null} non-scan entry are
still computed from the scanned
+ /// block.
+ @Test
+ void testNonScanResultsSkipScan() {
+ TransformBlock transformBlock = nextTransformBlock();
+ AggregationFunction[] aggregationFunctions =
_queryContext.getAggregationFunctions();
+ assert aggregationFunctions != null;
+
+ // Resolve only the second function (index 1 -> MAX) without scanning; the
rest fall back to scan-based execution.
+ int nonScanIndex = 1;
+ Object[] nonScanResults = new Object[aggregationFunctions.length];
+ // inject a Double sentinel that cannot occur in the data (all values
+ // are non-negative), proving the injected value is returned untouched
rather than recomputed by scanning.
+ double injectedMax = -1.0;
+ nonScanResults[nonScanIndex] = injectedMax;
+
+ AggregationExecutor aggregationExecutor =
+ new DefaultAggregationExecutor(aggregationFunctions, nonScanResults);
+ aggregationExecutor.aggregate(transformBlock);
+ List<Object> result = aggregationExecutor.getResult();
+
+ // The non-scan function returns the injected value untouched (not a
scanned MAX).
+ Assert.assertEquals(result.get(nonScanIndex), injectedMax,
+ "Non-scan function should return the injected value, not a scanned
result");
+
+ // Remaining functions are still computed by scanning the segment.
+ for (int i = 0; i < result.size(); i++) {
+ if (i == nonScanIndex) {
+ continue;
+ }
+ double actual = (double) result.get(i);
+ double expected = computeAggregation(AGGREGATION_FUNCTIONS[i],
_inputData[i]);
+ Assert.assertEquals(actual, expected,
+ "Aggregation mis-match for function " + AGGREGATION_FUNCTIONS[i] +
", Expected: " + expected + " Actual: "
+ + actual);
+ }
+ }
+
+ /// Builds a transform block over all physical columns of the test segment
with a match-all filter.
+ private TransformBlock nextTransformBlock() {
Map<String, DataSource> dataSourceMap = new HashMap<>();
List<ExpressionContext> expressions = new ArrayList<>();
for (String column : _indexSegment.getPhysicalColumnNames()) {
@@ -127,19 +183,7 @@ public class DefaultAggregationExecutorTest {
ProjectionOperator projectionOperator =
new ProjectionOperator(dataSourceMap, docIdSetOperator, new
QueryContext.Builder().build());
TransformOperator transformOperator = new TransformOperator(_queryContext,
projectionOperator, expressions);
- TransformBlock transformBlock = transformOperator.nextBlock();
- AggregationFunction[] aggregationFunctions =
_queryContext.getAggregationFunctions();
- assert aggregationFunctions != null;
- AggregationExecutor aggregationExecutor = new
DefaultAggregationExecutor(aggregationFunctions);
- aggregationExecutor.aggregate(transformBlock);
- List<Object> result = aggregationExecutor.getResult();
- for (int i = 0; i < result.size(); i++) {
- double actual = (double) result.get(i);
- double expected = computeAggregation(AGGREGATION_FUNCTIONS[i],
_inputData[i]);
- Assert.assertEquals(actual, expected,
- "Aggregation mis-match for function " + AGGREGATION_FUNCTIONS[i] +
", Expected: " + expected + " Actual: "
- + actual);
- }
+ return transformOperator.nextBlock();
}
/// Helper method to setup the index segment on which to perform aggregation
tests.
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionUtilsTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionUtilsTest.java
new file mode 100644
index 00000000000..29041933af2
--- /dev/null
+++
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionUtilsTest.java
@@ -0,0 +1,84 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.core.query.aggregation.function;
+
+import org.apache.pinot.segment.spi.AggregationFunctionType;
+import org.apache.pinot.segment.spi.datasource.DataSource;
+import org.apache.pinot.segment.spi.index.reader.Dictionary;
+import org.testng.annotations.Test;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertThrows;
+
+
+/// Unit test for {@link AggregationFunctionUtils#getAggregationResult}, the
metadata/dictionary based aggregation
+/// result resolver used by the non-scan based and partial metadata based
aggregation paths.
+@SuppressWarnings("rawtypes")
+public class AggregationFunctionUtilsTest {
+
+ private static AggregationFunction mockFunction(AggregationFunctionType
type) {
+ AggregationFunction aggregationFunction = mock(AggregationFunction.class);
+ when(aggregationFunction.getType()).thenReturn(type);
+ return aggregationFunction;
+ }
+
+ @Test
+ public void testCountResolvedFromNumTotalDocs() {
+ AggregationFunction countFunction =
mockFunction(AggregationFunctionType.COUNT);
+ // COUNT is resolved directly from numTotalDocs and must not touch the
(possibly null) data source.
+ Object result =
AggregationFunctionUtils.getAggregationResult(countFunction, null, 42, "TEST");
+ assertEquals(result, 42L);
+ }
+
+ @Test
+ public void testMinAndMaxResolvedFromDictionary() {
+ Dictionary dictionary = mock(Dictionary.class);
+ when(dictionary.getMinVal()).thenReturn(5);
+ when(dictionary.getMaxVal()).thenReturn(10);
+ DataSource dataSource = mock(DataSource.class);
+ when(dataSource.getDictionary()).thenReturn(dictionary);
+
+ Object minResult =
AggregationFunctionUtils.getAggregationResult(mockFunction(AggregationFunctionType.MIN),
+ dataSource, 100, "TEST");
+ assertEquals(minResult, 5.0);
+
+ Object maxResult =
AggregationFunctionUtils.getAggregationResult(mockFunction(AggregationFunctionType.MAX),
+ dataSource, 100, "TEST");
+ assertEquals(maxResult, 10.0);
+ }
+
+ @Test
+ public void testUnsupportedFunctionThrows() {
+ // MODE cannot be resolved from dictionary/metadata; the resolver must
reject it rather than return a wrong result.
+ DataSource dataSource = mock(DataSource.class);
+ assertThrows(IllegalStateException.class,
+ () ->
AggregationFunctionUtils.getAggregationResult(mockFunction(AggregationFunctionType.MODE),
dataSource,
+ 100, "TEST"));
+ }
+
+ @Test
+ public void testNonCountWithNullDataSourceThrows() {
+ // Every non-COUNT function reads from the column dictionary/metadata and
therefore requires a non-null data source.
+ assertThrows(NullPointerException.class,
+ () ->
AggregationFunctionUtils.getAggregationResult(mockFunction(AggregationFunctionType.MIN),
null, 100,
+ "TEST"));
+ }
+}
diff --git
a/pinot-core/src/test/java/org/apache/pinot/queries/ExplainPlanQueriesTest.java
b/pinot-core/src/test/java/org/apache/pinot/queries/ExplainPlanQueriesTest.java
index 94492d55310..b888456b12f 100644
---
a/pinot-core/src/test/java/org/apache/pinot/queries/ExplainPlanQueriesTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/queries/ExplainPlanQueriesTest.java
@@ -1811,7 +1811,7 @@ public class ExplainPlanQueriesTest extends
BaseQueriesTest {
});
result3.add(
new Object[]{"AGGREGATE(aggregations:count(*), max(noIndexCol1),
sum(noIndexCol2), avg(noIndexCol2))", 3, 2});
- result3.add(new Object[]{"PROJECT(noIndexCol2, noIndexCol1)", 4, 3});
+ result3.add(new Object[]{"PROJECT(noIndexCol2)", 4, 3});
result3.add(new Object[]{"DOC_ID_SET", 5, 4});
result3.add(new Object[]{"FILTER_MATCH_ENTIRE_SEGMENT(docs:3)", 6, 5});
check(query3, new ResultTable(DATA_SCHEMA, result3));
@@ -1924,7 +1924,7 @@ public class ExplainPlanQueriesTest extends
BaseQueriesTest {
});
result3.add(
new Object[]{"AGGREGATE(aggregations:count(*), max(noIndexCol1),
sum(noIndexCol2), avg(noIndexCol2))", 3, 2});
- result3.add(new Object[]{"PROJECT(noIndexCol2, noIndexCol1)", 4, 3});
+ result3.add(new Object[]{"PROJECT(noIndexCol2)", 4, 3});
result3.add(new Object[]{"DOC_ID_SET", 5, 4});
result3.add(new Object[]{"FILTER_MATCH_ENTIRE_SEGMENT(docs:3)", 6, 5});
check(query3, new ResultTable(DATA_SCHEMA, result3));
diff --git
a/pinot-core/src/test/java/org/apache/pinot/queries/InnerSegmentAggregationMultiValueQueriesTest.java
b/pinot-core/src/test/java/org/apache/pinot/queries/InnerSegmentAggregationMultiValueQueriesTest.java
index 354732a376e..169511132ad 100644
---
a/pinot-core/src/test/java/org/apache/pinot/queries/InnerSegmentAggregationMultiValueQueriesTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/queries/InnerSegmentAggregationMultiValueQueriesTest.java
@@ -47,7 +47,7 @@ public class InnerSegmentAggregationMultiValueQueriesTest
extends BaseMultiValue
AggregationOperator aggregationOperator = getOperator(AGGREGATION_QUERY);
AggregationResultsBlock resultsBlock = aggregationOperator.nextBlock();
QueriesTestUtils.testInnerSegmentExecutionStatistics(aggregationOperator.getExecutionStatistics(),
100000L, 0L,
- 400000L, 100000L);
+ 200000L, 100000L);
QueriesTestUtils.testInnerSegmentAggregationResult(resultsBlock.getResults(),
100000L, 100991525475000L, 2147434110,
1182655, 83439903673981L, 100000L);
diff --git
a/pinot-core/src/test/java/org/apache/pinot/queries/InnerSegmentAggregationMultiValueRawQueriesTest.java
b/pinot-core/src/test/java/org/apache/pinot/queries/InnerSegmentAggregationMultiValueRawQueriesTest.java
index 33543cad291..7e094a64b5b 100644
---
a/pinot-core/src/test/java/org/apache/pinot/queries/InnerSegmentAggregationMultiValueRawQueriesTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/queries/InnerSegmentAggregationMultiValueRawQueriesTest.java
@@ -47,7 +47,7 @@ public class InnerSegmentAggregationMultiValueRawQueriesTest
extends BaseMultiVa
AggregationOperator aggregationOperator = getOperator(AGGREGATION_QUERY);
AggregationResultsBlock resultsBlock = aggregationOperator.nextBlock();
QueriesTestUtils.testInnerSegmentExecutionStatistics(aggregationOperator.getExecutionStatistics(),
100000L, 0L,
- 400000L, 100000L);
+ 200000L, 100000L);
QueriesTestUtils.testInnerSegmentAggregationResult(resultsBlock.getResults(),
100000L, 100991525475000L, 2147434110,
1182655, 83439903673981L, 100000L);
diff --git
a/pinot-core/src/test/java/org/apache/pinot/queries/InnerSegmentAggregationSingleValueQueriesTest.java
b/pinot-core/src/test/java/org/apache/pinot/queries/InnerSegmentAggregationSingleValueQueriesTest.java
index dbf9565f67c..ca7c283d41a 100644
---
a/pinot-core/src/test/java/org/apache/pinot/queries/InnerSegmentAggregationSingleValueQueriesTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/queries/InnerSegmentAggregationSingleValueQueriesTest.java
@@ -46,7 +46,7 @@ public class InnerSegmentAggregationSingleValueQueriesTest
extends BaseSingleVal
AggregationOperator aggregationOperator = getOperator(AGGREGATION_QUERY);
AggregationResultsBlock resultsBlock = aggregationOperator.nextBlock();
QueriesTestUtils.testInnerSegmentExecutionStatistics(aggregationOperator.getExecutionStatistics(),
30000L, 0L,
- 120000L, 30000L);
+ 60000L, 30000L);
QueriesTestUtils.testInnerSegmentAggregationResult(resultsBlock.getResults(),
30000L, 32317185437847L, 2147419555,
1689277, 28175373944314L, 30000L);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
