yashmayya commented on code in PR #18334:
URL: https://github.com/apache/pinot/pull/18334#discussion_r3693093985
##########
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionUtils.java:
##########
@@ -501,4 +536,356 @@ public static String
getResultColumnName(AggregationFunction aggregationFunction
}
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 pre-aggregated result 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;
Review Comment:
```suggestion
return switch (functionType) {
case MIN, MINMV -> getMinValueNumeric(dataSource);
case MINLONG -> getMinValueLong(dataSource);
case MINSTRING -> {
assert dataSource.getDictionary() != null;
yield dataSource.getDictionary().getMinVal();
}
case MAX, MAXMV -> getMaxValueNumeric(dataSource);
case MAXLONG -> getMaxValueLong(dataSource);
case MAXSTRING -> {
assert dataSource.getDictionary() != null;
yield dataSource.getDictionary().getMaxVal();
}
case MINMAXRANGE, MINMAXRANGEMV ->
new MinMaxRangePair(getMinValueNumeric(dataSource),
getMaxValueNumeric(dataSource));
case DISTINCTCOUNT, DISTINCTSUM, DISTINCTAVG, DISTINCTCOUNTMV,
DISTINCTSUMMV, DISTINCTAVGMV ->
getDistinctValueSet(Objects.requireNonNull(dataSource.getDictionary()),
explainPlanName);
case DISTINCTCOUNTOFFHEAP ->
((DistinctCountOffHeapAggregationFunction)
aggregationFunction).extractAggregationResult(
Objects.requireNonNull(dataSource.getDictionary()));
case DISTINCTCOUNTHLL, DISTINCTCOUNTHLLMV ->
getDistinctCountHLLResult(Objects.requireNonNull(dataSource.getDictionary()),
(DistinctCountHLLAggregationFunction) aggregationFunction,
explainPlanName);
case DISTINCTCOUNTRAWHLL, DISTINCTCOUNTRAWHLLMV ->
getDistinctCountHLLResult(Objects.requireNonNull(dataSource.getDictionary()),
((DistinctCountRawHLLAggregationFunction)
aggregationFunction).getDistinctCountHLLAggregationFunction(),
explainPlanName);
case DISTINCTCOUNTHLLPLUS, DISTINCTCOUNTHLLPLUSMV ->
getDistinctCountHLLPlusResult(Objects.requireNonNull(dataSource.getDictionary()),
(DistinctCountHLLPlusAggregationFunction) aggregationFunction,
explainPlanName);
case DISTINCTCOUNTRAWHLLPLUS, DISTINCTCOUNTRAWHLLPLUSMV ->
getDistinctCountHLLPlusResult(Objects.requireNonNull(dataSource.getDictionary()),
((DistinctCountRawHLLPlusAggregationFunction)
aggregationFunction)
.getDistinctCountHLLPlusAggregationFunction(),
explainPlanName);
case SEGMENTPARTITIONEDDISTINCTCOUNT -> (long)
Objects.requireNonNull(dataSource.getDictionary()).length();
case DISTINCTCOUNTSMARTHLL ->
getDistinctCountSmartHLLResult(Objects.requireNonNull(dataSource.getDictionary()),
(DistinctCountSmartHLLAggregationFunction) aggregationFunction,
explainPlanName);
case DISTINCTCOUNTSMARTHLLPLUS ->
getDistinctCountSmartHLLPlusResult(Objects.requireNonNull(dataSource.getDictionary()),
(DistinctCountSmartHLLPlusAggregationFunction)
aggregationFunction, explainPlanName);
case DISTINCTCOUNTULL, DISTINCTCOUNTRAWULL ->
getDistinctCountULLResult(Objects.requireNonNull(dataSource.getDictionary()),
(DistinctCountULLAggregationFunction) aggregationFunction,
explainPlanName);
case DISTINCTCOUNTSMARTULL ->
getDistinctCountSmartULLResult(Objects.requireNonNull(dataSource.getDictionary()),
(DistinctCountSmartULLAggregationFunction) aggregationFunction,
explainPlanName);
default -> throw new IllegalStateException(
"Non-scan based aggregation operator does not support function
type: " + functionType);
};
```
nit: can use enhanced switch here
##########
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/DefaultAggregationExecutor.java:
##########
@@ -29,8 +29,18 @@
public class DefaultAggregationExecutor implements AggregationExecutor {
protected final AggregationFunction[] _aggregationFunctions;
protected final AggregationResultHolder[] _aggregationResultHolders;
+ protected final Object[] _preAggregatedResults;
- public DefaultAggregationExecutor(AggregationFunction[]
aggregationFunctions) {
+ /**
+ * Creates an executor that skips functions with a pre-aggregated result.
For each index {@code i} where
+ * {@code preAggregatedResults[i]} is non-null, the function is not
aggregated over the scanned blocks and the
+ * pre-aggregated value is emitted directly in the results. A {@code null}
array disables this behavior and all
+ * functions are computed by scanning.
+ *
+ * @param preAggregatedResults per-function pre-aggregated results, or
{@code null} if none are pre-aggregated
+ */
+ public DefaultAggregationExecutor(AggregationFunction[]
aggregationFunctions, Object[] preAggregatedResults) {
Review Comment:
nit: the `pre-aggregated` terminology used here can cause confusion because
it typically refers to values obtained from a star-tree index. Can we use some
other terminology like metadata based aggregates or non-scan aggregates instead?
##########
pinot-core/src/main/java/org/apache/pinot/core/operator/query/AggregationOperator.java:
##########
@@ -52,22 +56,51 @@ public class AggregationOperator extends
BaseOperator<AggregationResultsBlock> {
private int _numDocsScanned = 0;
+ // 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 metadata-resolvable, in which case every function is
computed by scanning.
+ @Nullable
+ private final boolean[] _metadataResolvable;
+ @Nullable
+ private final DataSource[] _dataSources;
+
public AggregationOperator(QueryContext queryContext, AggregationInfo
aggregationInfo, long 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
metadataResolvable}, 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 metadataResolvable 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
+ */
Review Comment:
We no longer use this style for Javadoc comments, see
https://github.com/apache/pinot/pull/19126/
##########
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionUtils.java:
##########
@@ -501,4 +536,356 @@ public static String
getResultColumnName(AggregationFunction aggregationFunction
}
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 pre-aggregated result 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);
+ }
+ }
+
+ @Nullable
Review Comment:
Why is this nullable?
##########
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/DefaultAggregationExecutor.java:
##########
@@ -29,8 +29,18 @@
public class DefaultAggregationExecutor implements AggregationExecutor {
protected final AggregationFunction[] _aggregationFunctions;
protected final AggregationResultHolder[] _aggregationResultHolders;
+ protected final Object[] _preAggregatedResults;
- public DefaultAggregationExecutor(AggregationFunction[]
aggregationFunctions) {
+ /**
+ * Creates an executor that skips functions with a pre-aggregated result.
For each index {@code i} where
+ * {@code preAggregatedResults[i]} is non-null, the function is not
aggregated over the scanned blocks and the
+ * pre-aggregated value is emitted directly in the results. A {@code null}
array disables this behavior and all
+ * functions are computed by scanning.
+ *
+ * @param preAggregatedResults per-function pre-aggregated results, or
{@code null} if none are pre-aggregated
+ */
+ public DefaultAggregationExecutor(AggregationFunction[]
aggregationFunctions, Object[] preAggregatedResults) {
Review Comment:
`preAggregatedResults` should be annotated as nullable
##########
pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/DefaultAggregationExecutorTest.java:
##########
@@ -120,6 +120,59 @@ public void setUp()
*/
@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 pre-aggregated result are not re-computed
by scanning: the injected value is
+ * returned as-is, while functions with a {@code null} pre-aggregated entry
are still computed from the scanned block.
+ */
+ @Test
+ void testPreAggregatedResultsSkipScan() {
+ TransformBlock transformBlock = nextTransformBlock();
+ AggregationFunction[] aggregationFunctions =
_queryContext.getAggregationFunctions();
+ assert aggregationFunctions != null;
+
+ // Pre-aggregate only the first function (index 0 -> SUM); the rest fall
back to scan-based execution.
+ Object[] preAggregatedResults = new Object[aggregationFunctions.length];
+ double injectedSum = 12345.0;
+ preAggregatedResults[0] = injectedSum;
Review Comment:
Even though it's just a unit test, why are we choosing to inject the one
aggregation function which doesn't actually support non scan based execution
out of the three aggregation functions here (sum, min, max)?
##########
pinot-core/src/main/java/org/apache/pinot/core/plan/AggregationPlanNode.java:
##########
@@ -173,40 +204,68 @@ private boolean hasNullValues(AggregationFunction[]
aggregationFunctions) {
}
/**
- * Returns {@code true} if the given aggregations can be solved with
dictionary or column metadata, {@code false}
- * otherwise.
+ * 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[] 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;
- }
- }
+ 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.
COUNT(*) or COUNT(1)),
+ // so it cannot be resolved from metadata
Review Comment:
> Aggregation function does not have a single identifier argument (e.g.
COUNT(*) or COUNT(1)), so it cannot be resolved from metadata
But we're already checking for `COUNT` before this check?
##########
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/DefaultAggregationExecutor.java:
##########
@@ -29,8 +29,18 @@
public class DefaultAggregationExecutor implements AggregationExecutor {
protected final AggregationFunction[] _aggregationFunctions;
protected final AggregationResultHolder[] _aggregationResultHolders;
+ protected final Object[] _preAggregatedResults;
Review Comment:
This should be annotated as nullable
##########
pinot-core/src/main/java/org/apache/pinot/core/operator/query/AggregationOperator.java:
##########
@@ -85,6 +118,28 @@ protected AggregationResultsBlock getNextBlock() {
return new AggregationResultsBlock(_aggregationFunctions,
aggregationExecutor.getResult(), _queryContext);
}
+ /**
+ * Returns {@code null} when no function is metadata-resolvable, 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[] resolveMetadataBasedResults() {
+ if (_metadataResolvable == null) {
+ return null;
+ }
+
+ Objects.requireNonNull(_dataSources);
+ Object[] preAggregatedResults = new Object[_aggregationFunctions.length];
+ for (int i = 0; i < _aggregationFunctions.length; i++) {
+ if (_metadataResolvable[i]) {
+ preAggregatedResults[i] =
AggregationFunctionUtils.getAggregationResult(_aggregationFunctions[i],
+ _dataSources[i], (int) _numTotalDocs, EXPLAIN_NAME);
Review Comment:
> (int) _numTotalDocs
I think `_numTotalDocs` should be changed from `long` to `int` instead of
casting here - it should always have been an `int` IMO.
##########
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionUtils.java:
##########
@@ -501,4 +536,356 @@ public static String
getResultColumnName(AggregationFunction aggregationFunction
}
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 pre-aggregated result 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;
Review Comment:
Actually on second thought, let's leave all refactors as a follow up since
this is simply a faithful copy of the implementation from
`NonScanBasedAggregationOperator`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
