Skip to content

[UUID 5/8] UUID aggregation, group-by and distinct #18873

New issue

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

Sign up for GitHub

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

Already on GitHub? Sign in to your account

Open
xiangfu0 wants to merge 8 commits into
base: master
Choose a base branch
from
+743 −54
Open

[UUID 5/8] UUID aggregation, group-by and distinct #18873

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
83dd3fe
[UUID 5/8] UUID aggregation, group-by and distinct
xiangfu0 Jul 7, 2026
0903746
Return the converted UUID form for group keys in the DataTable reduce…
xiangfu0 Aug 8, 2026
fd6df48
Cast directly to UuidKey in UuidToIdMap
xiangfu0 Aug 8, 2026
1977e2c
Hash UUID's stored bytes in distinct-count aggregations, not a canoni…
xiangfu0 Aug 9, 2026
1a5b302
Drop UUID special-casing that the stored type already covers
xiangfu0 Aug 9, 2026
0e66573
Format DISTINCT rows in one pass in BytesDistinctTable
xiangfu0 Aug 9, 2026
a7d7892
Route UUID through the normal dispatch in DistinctCountBitmap
xiangfu0 Aug 13, 2026
b50e1ee
Route UUID through the normal dispatch in DistinctCountHLL
xiangfu0 Aug 13, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
39 changes: 27 additions & 12 deletions .../main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -610,23 +610,23 @@ public static Object getAggregationResult(AggregationFunction aggregationFunctio
break;
case DISTINCTCOUNTHLL:
case DISTINCTCOUNTHLLMV:
result = getDistinctCountHLLResult(Objects.requireNonNull(dataSource.getDictionary()),
result = getDistinctCountHLLResult(dataSource,
(DistinctCountHLLAggregationFunction) aggregationFunction, explainPlanName);
break;
case DISTINCTCOUNTRAWHLL:
case DISTINCTCOUNTRAWHLLMV:
result = getDistinctCountHLLResult(Objects.requireNonNull(dataSource.getDictionary()),
result = getDistinctCountHLLResult(dataSource,
((DistinctCountRawHLLAggregationFunction) aggregationFunction).getDistinctCountHLLAggregationFunction(),
explainPlanName);
break;
case DISTINCTCOUNTHLLPLUS:
case DISTINCTCOUNTHLLPLUSMV:
result = getDistinctCountHLLPlusResult(Objects.requireNonNull(dataSource.getDictionary()),
result = getDistinctCountHLLPlusResult(dataSource,
(DistinctCountHLLPlusAggregationFunction) aggregationFunction, explainPlanName);
break;
case DISTINCTCOUNTRAWHLLPLUS:
case DISTINCTCOUNTRAWHLLPLUSMV:
result = getDistinctCountHLLPlusResult(Objects.requireNonNull(dataSource.getDictionary()),
result = getDistinctCountHLLPlusResult(dataSource,
((DistinctCountRawHLLPlusAggregationFunction) aggregationFunction)
.getDistinctCountHLLPlusAggregationFunction(), explainPlanName);
break;
Expand All @@ -642,15 +642,15 @@ public static Object getAggregationResult(AggregationFunction aggregationFunctio
(DistinctCountSmartHLLPlusAggregationFunction) aggregationFunction, explainPlanName);
break;
case DISTINCTCOUNTULL:
result = getDistinctCountULLResult(Objects.requireNonNull(dataSource.getDictionary()),
result = getDistinctCountULLResult(dataSource,
(DistinctCountULLAggregationFunction) aggregationFunction, explainPlanName);
break;
case DISTINCTCOUNTSMARTULL:
result = getDistinctCountSmartULLResult(Objects.requireNonNull(dataSource.getDictionary()),
(DistinctCountSmartULLAggregationFunction) aggregationFunction, explainPlanName);
break;
case DISTINCTCOUNTRAWULL:
result = getDistinctCountULLResult(Objects.requireNonNull(dataSource.getDictionary()),
result = getDistinctCountULLResult(dataSource,
(DistinctCountULLAggregationFunction) aggregationFunction, explainPlanName);
break;
default:
Expand Down Expand Up @@ -799,9 +799,14 @@ private static HyperLogLogPlus getDistinctValueHLLPlus(Dictionary dictionary, in
return hllPlus;
}

private static HyperLogLog getDistinctCountHLLResult(Dictionary dictionary,
private static HyperLogLog getDistinctCountHLLResult(DataSource dataSource,
DistinctCountHLLAggregationFunction function, String explainPlanName) {
if (dictionary.getValueType() == FieldSpec.DataType.BYTES) {
Dictionary dictionary = Objects.requireNonNull(dataSource.getDictionary());
// A UUID column's dictionary reports BYTES (it is a plain BytesDictionary), but its entries are logical
// scalars, not serialized sketch state. Excluding it here lets it fall through to the scalar path
// below, which offers dictionary.get(i) -- the stored byte[] -- exactly as the scan path does.
if (dataSource.getDataSourceMetadata().getDataType() != FieldSpec.DataType.UUID
&& dictionary.getValueType() == FieldSpec.DataType.BYTES) {
// Treat BYTES value as serialized HyperLogLog
try {
QueryThreadContext.checkTerminationAndSampleUsage(explainPlanName);
Expand All @@ -820,9 +825,14 @@ private static HyperLogLog getDistinctCountHLLResult(Dictionary dictionary,
}
}

private static HyperLogLogPlus getDistinctCountHLLPlusResult(Dictionary dictionary,
private static HyperLogLogPlus getDistinctCountHLLPlusResult(DataSource dataSource,
DistinctCountHLLPlusAggregationFunction function, String explainPlanName) {
if (dictionary.getValueType() == FieldSpec.DataType.BYTES) {
Dictionary dictionary = Objects.requireNonNull(dataSource.getDictionary());
// A UUID column's dictionary reports BYTES (it is a plain BytesDictionary), but its entries are logical
// scalars, not serialized sketch state. Excluding it here lets it fall through to the scalar path
// below, which offers dictionary.get(i) -- the stored byte[] -- exactly as the scan path does.
if (dataSource.getDataSourceMetadata().getDataType() != FieldSpec.DataType.UUID
&& dictionary.getValueType() == FieldSpec.DataType.BYTES) {
// Treat BYTES value as serialized HyperLogLogPlus
try {
QueryThreadContext.checkTerminationAndSampleUsage(explainPlanName);
Expand Down Expand Up @@ -861,9 +871,14 @@ private static Object getDistinctCountSmartHLLPlusResult(Dictionary dictionary,
}
}

private static UltraLogLog getDistinctCountULLResult(Dictionary dictionary,
private static UltraLogLog getDistinctCountULLResult(DataSource dataSource,
DistinctCountULLAggregationFunction function, String explainPlanName) {
if (dictionary.getValueType() == FieldSpec.DataType.BYTES) {
Dictionary dictionary = Objects.requireNonNull(dataSource.getDictionary());
// A UUID column's dictionary reports BYTES (it is a plain BytesDictionary), but its entries are logical
// scalars, not serialized sketch state. Excluding it here lets it fall through to the scalar path
// below, which offers dictionary.get(i) -- the stored byte[] -- exactly as the scan path does.
if (dataSource.getDataSourceMetadata().getDataType() != FieldSpec.DataType.UUID
&& dictionary.getValueType() == FieldSpec.DataType.BYTES) {
// Treat BYTES value as serialized UltraLogLog and merge
try {
QueryThreadContext.checkTerminationAndSampleUsage(explainPlanName);
Expand Down
78 changes: 72 additions & 6 deletions .../apache/pinot/core/query/aggregation/function/DistinctCountBitmapAggregationFunction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
*/
package org.apache.pinot.core.query.aggregation.function;

import java.util.Arrays;
import java.util.List;
import java.util.Map;
import javax.annotation.Nullable;
Expand Down Expand Up @@ -71,9 +72,11 @@ public void aggregate(int length, AggregationResultHolder aggregationResultHolde
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet blockValSet = blockValSetMap.get(_expression);

DataType dataType = blockValSet.getValueType();
DataType storedType = dataType.getStoredType();

// Treat BYTES value as serialized RoaringBitmap
DataType storedType = blockValSet.getValueType().getStoredType();
if (storedType == DataType.BYTES) {
if (storedType == DataType.BYTES && dataType != DataType.UUID) {
byte[][] bytesValues = blockValSet.getBytesValuesSV();
RoaringBitmap valueBitmap = aggregationResultHolder.getResult();
if (valueBitmap != null) {
Expand Down Expand Up @@ -138,6 +141,13 @@ protected void aggregateSV(int length, AggregationResultHolder aggregationResult
valueBitmap.add(stringValues[i].hashCode());
}
break;
// Reached only by UUID: a real BYTES column is serialized sketch state and is handled above.
case BYTES:
byte[][] uuidValues = blockValSet.getBytesValuesSV();
for (int i = 0; i < length; i++) {
valueBitmap.add(Arrays.hashCode(uuidValues[i]));
}
break;
default:
throw new IllegalStateException(
"Illegal data type for DISTINCT_COUNT_BITMAP aggregation function: " + storedType);
Expand Down Expand Up @@ -198,6 +208,15 @@ protected void aggregateMV(int length, AggregationResultHolder aggregationResult
}
}
break;
// Reached only by UUID: a real BYTES column is serialized sketch state and is handled above.
case BYTES:
byte[][][] uuidValues = blockValSet.getBytesValuesMV();
for (int i = 0; i < length; i++) {
for (byte[] value : uuidValues[i]) {
valueBitmap.add(Arrays.hashCode(value));
}
}
break;
default:
throw new IllegalStateException(
"Illegal data type for DISTINCT_COUNT_BITMAP aggregation function: " + storedType);
Expand All @@ -209,9 +228,11 @@ public void aggregateGroupBySV(int length, int[] groupKeyArray, GroupByResultHol
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet blockValSet = blockValSetMap.get(_expression);

DataType dataType = blockValSet.getValueType();
DataType storedType = dataType.getStoredType();

// Treat BYTES value as serialized RoaringBitmap
DataType storedType = blockValSet.getValueType().getStoredType();
if (storedType == DataType.BYTES) {
if (storedType == DataType.BYTES && dataType != DataType.UUID) {
byte[][] bytesValues = blockValSet.getBytesValuesSV();
for (int i = 0; i < length; i++) {
RoaringBitmap value = RoaringBitmapUtils.deserialize(bytesValues[i]);
Expand Down Expand Up @@ -277,6 +298,13 @@ protected void aggregateSVGroupBySV(int length, int[] groupKeyArray, GroupByResu
getValueBitmap(groupByResultHolder, groupKeyArray[i]).add(stringValues[i].hashCode());
}
break;
// Reached only by UUID: a real BYTES column is serialized sketch state and is handled above.
case BYTES:
byte[][] uuidValues = blockValSet.getBytesValuesSV();
for (int i = 0; i < length; i++) {
getValueBitmap(groupByResultHolder, groupKeyArray[i]).add(Arrays.hashCode(uuidValues[i]));
}
break;
default:
throw new IllegalStateException(
"Illegal data type for DISTINCT_COUNT_BITMAP aggregation function: " + storedType);
Expand Down Expand Up @@ -339,6 +367,16 @@ protected void aggregateMVGroupBySV(int length, int[] groupKeyArray, GroupByResu
}
}
break;
// Reached only by UUID: a real BYTES column is serialized sketch state and is handled above.
case BYTES:
byte[][][] uuidValues = blockValSet.getBytesValuesMV();
for (int i = 0; i < length; i++) {
RoaringBitmap bitmap = getValueBitmap(groupByResultHolder, groupKeyArray[i]);
for (byte[] value : uuidValues[i]) {
bitmap.add(Arrays.hashCode(value));
}
}
break;
default:
throw new IllegalStateException(
"Illegal data type for DISTINCT_COUNT_BITMAP aggregation function: " + storedType);
Expand All @@ -350,9 +388,11 @@ public void aggregateGroupByMV(int length, int[][] groupKeysArray, GroupByResult
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet blockValSet = blockValSetMap.get(_expression);

DataType dataType = blockValSet.getValueType();
DataType storedType = dataType.getStoredType();

// Treat BYTES value as serialized RoaringBitmap
DataType storedType = blockValSet.getValueType().getStoredType();
if (storedType == DataType.BYTES) {
if (storedType == DataType.BYTES && dataType != DataType.UUID) {
byte[][] bytesValues = blockValSet.getBytesValuesSV();
for (int i = 0; i < length; i++) {
RoaringBitmap value = RoaringBitmapUtils.deserialize(bytesValues[i]);
Expand Down Expand Up @@ -420,6 +460,13 @@ protected void aggregateSVGroupByMV(int length, int[][] groupKeysArray, GroupByR
setValueForGroupKeys(groupByResultHolder, groupKeysArray[i], stringValues[i].hashCode());
}
break;
// Reached only by UUID: a real BYTES column is serialized sketch state and is handled above.
case BYTES:
byte[][] uuidValues = blockValSet.getBytesValuesSV();
for (int i = 0; i < length; i++) {
setValueForGroupKeys(groupByResultHolder, groupKeysArray[i], Arrays.hashCode(uuidValues[i]));
}
break;
default:
throw new IllegalStateException(
"Illegal data type for DISTINCT_COUNT_BITMAP aggregation function: " + storedType);
Expand Down Expand Up @@ -494,6 +541,18 @@ protected void aggregateMVGroupByMV(int length, int[][] groupKeysArray, GroupByR
}
}
break;
// Reached only by UUID: a real BYTES column is serialized sketch state and is handled above.
case BYTES:
byte[][][] uuidValues = blockValSet.getBytesValuesMV();
for (int i = 0; i < length; i++) {
for (int groupKey : groupKeysArray[i]) {
RoaringBitmap bitmap = getValueBitmap(groupByResultHolder, groupKey);
for (byte[] value : uuidValues[i]) {
bitmap.add(Arrays.hashCode(value));
}
}
}
break;
default:
throw new IllegalStateException(
"Illegal data type for DISTINCT_COUNT_BITMAP aggregation function: " + storedType);
Expand Down Expand Up @@ -660,6 +719,13 @@ private static RoaringBitmap convertToValueBitmap(DictIdsWrapper dictIdsWrapper)
valueBitmap.add(dictionary.getStringValue(iterator.next()).hashCode());
}
break;
// A UUID column's dictionary reports BYTES (it is a plain BytesDictionary); hash the stored bytes to match
// the scan path above. A real BYTES column holds serialized bitmaps and never reaches the dictionary path.
case BYTES:
while (iterator.hasNext()) {
valueBitmap.add(Arrays.hashCode(dictionary.getBytesValue(iterator.next())));
}
break;
default:
throw new IllegalStateException(
"Illegal data type for DISTINCT_COUNT_BITMAP aggregation function: " + storedType);
Expand Down
52 changes: 46 additions & 6 deletions ...ache/pinot/core/query/aggregation/function/DistinctCountCPCSketchAggregationFunction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -135,8 +135,24 @@ public void aggregate(int length, AggregationResultHolder aggregationResultHolde
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet blockValSet = blockValSetMap.get(_expression);

FieldSpec.DataType dataType = blockValSet.getValueType();
FieldSpec.DataType storedType = dataType.getStoredType();

// UUID values are logical scalars (stored as 16-byte BYTES) — not serialized CPC Sketch state. Update
// the sketch with the canonical UUID string so DISTINCTCOUNTCPC(uuidCol) matches
// DISTINCTCOUNTCPC(CAST(uuidCol AS STRING)).
if (dataType == DataType.UUID) {
byte[][] uuidBytesValues = blockValSet.getBytesValuesSV();
// Leave the updated CpcSketch in the holder; extractAggregationResult converts it to an accumulator.
// Calling getAccumulator here would read the holder slot already occupied by the sketch and fail.
CpcSketch cpcSketch = getCpcSketch(aggregationResultHolder);
for (int i = 0; i < length; i++) {
cpcSketch.update(uuidBytesValues[i]);
}
return;
}

// Treat BYTES value as serialized CPC Sketch
FieldSpec.DataType storedType = blockValSet.getValueType().getStoredType();
if (storedType == DataType.BYTES) {
byte[][] bytesValues = blockValSet.getBytesValuesSV();
try {
Expand Down Expand Up @@ -197,17 +213,29 @@ public void aggregate(int length, AggregationResultHolder aggregationResultHolde
default:
throw new IllegalStateException("Illegal data type for DISTINCT_COUNT_CPC aggregation function: " + storedType);
}
CpcSketchAccumulator cpcSketchAccumulator = getAccumulator(aggregationResultHolder);
cpcSketchAccumulator.apply(cpcSketch);
// The updated CpcSketch already lives in the holder (getCpcSketch stored it); extractAggregationResult
// converts it to a CpcSketchAccumulator. Reading the holder as an accumulator here would
// ClassCastException — the holder slot contains the sketch, not an accumulator.
}

@Override
public void aggregateGroupBySV(int length, int[] groupKeyArray, GroupByResultHolder groupByResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet blockValSet = blockValSetMap.get(_expression);

DataType dataType = blockValSet.getValueType();
DataType storedType = dataType.getStoredType();

// UUID columns: update with canonical UUID strings converted from raw bytes (see aggregate() for rationale).
if (dataType == DataType.UUID) {
byte[][] uuidBytesValues = blockValSet.getBytesValuesSV();
for (int i = 0; i < length; i++) {
getCpcSketch(groupByResultHolder, groupKeyArray[i]).update(uuidBytesValues[i]);
}
return;
}

// Treat BYTES value as serialized CPC Sketch
DataType storedType = blockValSet.getValueType().getStoredType();
if (storedType == FieldSpec.DataType.BYTES) {
byte[][] bytesValues = blockValSet.getBytesValuesSV();
try {
Expand Down Expand Up @@ -277,10 +305,22 @@ public void aggregateGroupByMV(int length, int[][] groupKeysArray, GroupByResult
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet blockValSet = blockValSetMap.get(_expression);

// Treat BYTES value as serialized CPC Sketch
DataType storedType = blockValSet.getValueType().getStoredType();
DataType dataType = blockValSet.getValueType();
DataType storedType = dataType.getStoredType();
boolean singleValue = blockValSet.isSingleValue();

// UUID columns: update with canonical UUID strings converted from raw bytes (see aggregate() for rationale).
if (dataType == DataType.UUID && singleValue) {
byte[][] uuidBytesValues = blockValSet.getBytesValuesSV();
for (int i = 0; i < length; i++) {
byte[] canonical = uuidBytesValues[i];
for (int groupKey : groupKeysArray[i]) {
getCpcSketch(groupByResultHolder, groupKey).update(canonical);
}
}
return;
}

if (singleValue && storedType == DataType.BYTES) {
byte[][] bytesValues = blockValSet.getBytesValuesSV();
try {
Expand Down
Loading
Loading