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 527b7817621c..6908ef1f19d8 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 @@ -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; @@ -642,7 +642,7 @@ 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: @@ -650,7 +650,7 @@ public static Object getAggregationResult(AggregationFunction aggregationFunctio (DistinctCountSmartULLAggregationFunction) aggregationFunction, explainPlanName); break; case DISTINCTCOUNTRAWULL: - result = getDistinctCountULLResult(Objects.requireNonNull(dataSource.getDictionary()), + result = getDistinctCountULLResult(dataSource, (DistinctCountULLAggregationFunction) aggregationFunction, explainPlanName); break; default: @@ -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); @@ -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); @@ -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); diff --git a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountBitmapAggregationFunction.java b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountBitmapAggregationFunction.java index 4fe96b819dbb..20cbc7999e68 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountBitmapAggregationFunction.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountBitmapAggregationFunction.java @@ -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; @@ -71,9 +72,11 @@ public void aggregate(int length, AggregationResultHolder aggregationResultHolde Map 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) { @@ -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); @@ -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); @@ -209,9 +228,11 @@ public void aggregateGroupBySV(int length, int[] groupKeyArray, GroupByResultHol Map 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]); @@ -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); @@ -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); @@ -350,9 +388,11 @@ public void aggregateGroupByMV(int length, int[][] groupKeysArray, GroupByResult Map 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]); @@ -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); @@ -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); @@ -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); diff --git a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountCPCSketchAggregationFunction.java b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountCPCSketchAggregationFunction.java index ec273c066ab1..96e286aae1e9 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountCPCSketchAggregationFunction.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountCPCSketchAggregationFunction.java @@ -135,8 +135,24 @@ public void aggregate(int length, AggregationResultHolder aggregationResultHolde Map 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 { @@ -197,8 +213,9 @@ 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 @@ -206,8 +223,19 @@ public void aggregateGroupBySV(int length, int[] groupKeyArray, GroupByResultHol Map 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 { @@ -277,10 +305,22 @@ public void aggregateGroupByMV(int length, int[][] groupKeysArray, GroupByResult Map 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 { diff --git a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountHLLAggregationFunction.java b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountHLLAggregationFunction.java index 07bfb6bd41af..666903857e9b 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountHLLAggregationFunction.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountHLLAggregationFunction.java @@ -81,9 +81,11 @@ public void aggregate(int length, AggregationResultHolder aggregationResultHolde Map blockValSetMap) { BlockValSet blockValSet = blockValSetMap.get(_expression); + DataType dataType = blockValSet.getValueType(); + DataType storedType = dataType.getStoredType(); + // Treat BYTES value as serialized HyperLogLog - DataType storedType = blockValSet.getValueType().getStoredType(); - if (storedType == DataType.BYTES) { + if (storedType == DataType.BYTES && dataType != DataType.UUID) { byte[][] bytesValues = blockValSet.getBytesValuesSV(); try { HyperLogLog hyperLogLog = aggregationResultHolder.getResult(); @@ -159,6 +161,13 @@ protected void aggregateSV(int length, AggregationResultHolder aggregationResult hyperLogLog.offer(stringValues[i]); } break; + // Reached only by UUID: a real BYTES column is serialized HLL state and is handled above. + case BYTES: + byte[][] uuidValues = blockValSet.getBytesValuesSV(); + for (int i = 0; i < length; i++) { + hyperLogLog.offer(uuidValues[i]); + } + break; default: throw new IllegalStateException("Illegal data type for DISTINCT_COUNT_HLL aggregation function: " + storedType); } @@ -222,6 +231,15 @@ protected void aggregateMV(int length, AggregationResultHolder aggregationResult } } break; + // Reached only by UUID: a real BYTES column is serialized HLL state and is handled above. + case BYTES: + byte[][][] uuidValuesArray = blockValSet.getBytesValuesMV(); + for (int i = 0; i < length; i++) { + for (byte[] value : uuidValuesArray[i]) { + hyperLogLog.offer(value); + } + } + break; default: throw new IllegalStateException("Illegal data type for DISTINCT_COUNT_HLL aggregation function: " + storedType); } @@ -232,9 +250,11 @@ public void aggregateGroupBySV(int length, int[] groupKeyArray, GroupByResultHol Map blockValSetMap) { BlockValSet blockValSet = blockValSetMap.get(_expression); + DataType dataType = blockValSet.getValueType(); + DataType storedType = dataType.getStoredType(); + // Treat BYTES value as serialized HyperLogLog - DataType storedType = blockValSet.getValueType().getStoredType(); - if (storedType == DataType.BYTES) { + if (storedType == DataType.BYTES && dataType != DataType.UUID) { byte[][] bytesValues = blockValSet.getBytesValuesSV(); try { for (int i = 0; i < length; i++) { @@ -307,6 +327,13 @@ protected void aggregateSVGroupBySV(int length, int[] groupKeyArray, GroupByResu getHyperLogLog(groupByResultHolder, groupKeyArray[i]).offer(stringValues[i]); } break; + // Reached only by UUID: a real BYTES column is serialized HLL state and is handled above. + case BYTES: + byte[][] uuidValues = blockValSet.getBytesValuesSV(); + for (int i = 0; i < length; i++) { + getHyperLogLog(groupByResultHolder, groupKeyArray[i]).offer(uuidValues[i]); + } + break; default: throw new IllegalStateException("Illegal data type for DISTINCT_COUNT_HLL aggregation function: " + storedType); } @@ -371,6 +398,16 @@ protected void aggregateMVGroupBySV(int length, int[] groupKeyArray, GroupByResu } } break; + // Reached only by UUID: a real BYTES column is serialized HLL state and is handled above. + case BYTES: + byte[][][] uuidValuesArray = blockValSet.getBytesValuesMV(); + for (int i = 0; i < length; i++) { + HyperLogLog hyperLogLog = getHyperLogLog(groupByResultHolder, groupKeyArray[i]); + for (byte[] value : uuidValuesArray[i]) { + hyperLogLog.offer(value); + } + } + break; default: throw new IllegalStateException("Illegal data type for DISTINCT_COUNT_HLL aggregation function: " + storedType); } @@ -381,9 +418,11 @@ public void aggregateGroupByMV(int length, int[][] groupKeysArray, GroupByResult Map blockValSetMap) { BlockValSet blockValSet = blockValSetMap.get(_expression); + DataType dataType = blockValSet.getValueType(); + DataType storedType = dataType.getStoredType(); + // Treat BYTES value as serialized HyperLogLog - DataType storedType = blockValSet.getValueType().getStoredType(); - if (storedType == DataType.BYTES) { + if (storedType == DataType.BYTES && dataType != DataType.UUID) { byte[][] bytesValues = blockValSet.getBytesValuesSV(); try { for (int i = 0; i < length; i++) { @@ -459,6 +498,13 @@ protected void aggregateSVGroupByMV(int length, int[][] groupKeysArray, GroupByR setValueForGroupKeys(groupByResultHolder, groupKeysArray[i], stringValues[i]); } break; + // Reached only by UUID: a real BYTES column is serialized HLL state and is handled above. + case BYTES: + byte[][] uuidValues = blockValSet.getBytesValuesSV(); + for (int i = 0; i < length; i++) { + setValueForGroupKeys(groupByResultHolder, groupKeysArray[i], uuidValues[i]); + } + break; default: throw new IllegalStateException("Illegal data type for DISTINCT_COUNT_HLL aggregation function: " + storedType); } @@ -541,6 +587,19 @@ protected void aggregateMVGroupByMV(int length, int[][] groupKeysArray, GroupByR } } break; + // Reached only by UUID: a real BYTES column is serialized HLL state and is handled above. + case BYTES: + byte[][][] uuidValuesArray = blockValSet.getBytesValuesMV(); + for (int i = 0; i < length; i++) { + byte[][] uuidValues = uuidValuesArray[i]; + for (int groupKey : groupKeysArray[i]) { + HyperLogLog hyperLogLog = getHyperLogLog(groupByResultHolder, groupKey); + for (byte[] value : uuidValues) { + hyperLogLog.offer(value); + } + } + } + break; default: throw new IllegalStateException("Illegal data type for DISTINCT_COUNT_HLL aggregation function: " + storedType); } diff --git a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountHLLPlusAggregationFunction.java b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountHLLPlusAggregationFunction.java index fd337f433a34..452f33412099 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountHLLPlusAggregationFunction.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountHLLPlusAggregationFunction.java @@ -93,8 +93,21 @@ public void aggregate(int length, AggregationResultHolder aggregationResultHolde Map blockValSetMap) { BlockValSet blockValSet = blockValSetMap.get(_expression); + DataType dataType = blockValSet.getValueType(); + DataType storedType = dataType.getStoredType(); + + // UUID values are logical scalars (stored as 16-byte BYTES) — not serialized HyperLogLogPlus state. Offer the + // canonical UUID string so DISTINCTCOUNTHLLPLUS(uuidCol) matches DISTINCTCOUNTHLLPLUS(CAST(uuidCol AS STRING)). + if (dataType == DataType.UUID) { + byte[][] uuidBytesValues = blockValSet.getBytesValuesSV(); + HyperLogLogPlus hyperLogLogPlus = getHyperLogLogPlus(aggregationResultHolder); + for (int i = 0; i < length; i++) { + hyperLogLogPlus.offer(uuidBytesValues[i]); + } + return; + } + // Treat BYTES value as serialized HyperLogLogPlus - DataType storedType = blockValSet.getValueType().getStoredType(); if (storedType == DataType.BYTES) { byte[][] bytesValues = blockValSet.getBytesValuesSV(); try { @@ -237,8 +250,19 @@ public void aggregateGroupBySV(int length, int[] groupKeyArray, GroupByResultHol Map blockValSetMap) { BlockValSet blockValSet = blockValSetMap.get(_expression); + DataType dataType = blockValSet.getValueType(); + DataType storedType = dataType.getStoredType(); + + // UUID columns: offer 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++) { + getHyperLogLogPlus(groupByResultHolder, groupKeyArray[i]).offer(uuidBytesValues[i]); + } + return; + } + // Treat BYTES value as serialized HyperLogLogPlus - DataType storedType = blockValSet.getValueType().getStoredType(); if (storedType == DataType.BYTES) { byte[][] bytesValues = blockValSet.getBytesValuesSV(); try { @@ -385,8 +409,22 @@ public void aggregateGroupByMV(int length, int[][] groupKeysArray, GroupByResult Map blockValSetMap) { BlockValSet blockValSet = blockValSetMap.get(_expression); + DataType dataType = blockValSet.getValueType(); + DataType storedType = dataType.getStoredType(); + + // UUID columns: offer 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++) { + byte[] canonical = uuidBytesValues[i]; + for (int groupKey : groupKeysArray[i]) { + getHyperLogLogPlus(groupByResultHolder, groupKey).offer(canonical); + } + } + return; + } + // Treat BYTES value as serialized HyperLogLogPlus - DataType storedType = blockValSet.getValueType().getStoredType(); if (storedType == DataType.BYTES) { byte[][] bytesValues = blockValSet.getBytesValuesSV(); try { diff --git a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountThetaSketchAggregationFunction.java b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountThetaSketchAggregationFunction.java index a2859fe471ec..263cf8af0091 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountThetaSketchAggregationFunction.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountThetaSketchAggregationFunction.java @@ -55,6 +55,7 @@ import org.apache.pinot.segment.spi.AggregationFunctionType; import org.apache.pinot.segment.spi.Constants; import org.apache.pinot.spi.data.FieldSpec.DataType; +import org.apache.pinot.spi.utils.BytesUtils; import org.apache.pinot.spi.utils.CommonConstants; import org.apache.pinot.sql.parsers.CalciteSqlParser; @@ -189,7 +190,7 @@ public void aggregate(int length, AggregationResultHolder aggregationResultHolde boolean[] singleValues = new boolean[numExpressions]; DataType[] valueTypes = new DataType[numExpressions]; Object[] valueArrays = new Object[numExpressions]; - extractValues(blockValSetMap, singleValues, valueTypes, valueArrays); + extractValues(length, blockValSetMap, singleValues, valueTypes, valueArrays); int numFilters = _filterEvaluators.size(); // Main expression is always index 0 @@ -438,7 +439,7 @@ public void aggregateGroupBySV(int length, int[] groupKeyArray, GroupByResultHol boolean[] singleValues = new boolean[numExpressions]; DataType[] valueTypes = new DataType[numExpressions]; Object[] valueArrays = new Object[numExpressions]; - extractValues(blockValSetMap, singleValues, valueTypes, valueArrays); + extractValues(length, blockValSetMap, singleValues, valueTypes, valueArrays); int numFilters = _filterEvaluators.size(); // Main expression is always index 0 @@ -662,7 +663,7 @@ public void aggregateGroupByMV(int length, int[][] groupKeysArray, GroupByResult boolean[] singleValues = new boolean[numExpressions]; DataType[] valueTypes = new DataType[numExpressions]; Object[] valueArrays = new Object[numExpressions]; - extractValues(blockValSetMap, singleValues, valueTypes, valueArrays); + extractValues(length, blockValSetMap, singleValues, valueTypes, valueArrays); int numFilters = _filterEvaluators.size(); // Main expression is always index 0 @@ -1226,14 +1227,45 @@ private static int extractSketchId(String identifier) { } /// Extracts values from the BlockValSet map. - private void extractValues(Map blockValSetMap, boolean[] singleValues, + private void extractValues(int length, Map blockValSetMap, boolean[] singleValues, DataType[] valueTypes, Object[] valueArrays) { int numExpressions = _inputExpressions.size(); for (int i = 0; i < numExpressions; i++) { BlockValSet blockValSet = blockValSetMap.get(_inputExpressions.get(i)); boolean singleValue = blockValSet.isSingleValue(); - DataType storedType = blockValSet.getValueType().getStoredType(); + DataType dataType = blockValSet.getValueType(); + DataType storedType = dataType.getStoredType(); singleValues[i] = singleValue; + // UUID columns are stored as 16-byte BYTES but a UUID value is a logical scalar, not a pre-serialized + // theta sketch: without this branch the function takes the serialized-sketch path below and Sketch.wrap + // fails on raw 16-byte UUID content. Unlike the other distinct-count functions this one cannot consume the + // stored bytes directly -- DataType.BYTES here means "serialized sketch", with no scalar-bytes mode -- so + // the stored value is surfaced as its hex rendering, the same form used at every other String-typed UUID + // boundary (see PredicateUtils#getStoredValue and the Bloom filter key). + if (dataType == DataType.UUID) { + valueTypes[i] = DataType.STRING; + if (singleValue) { + byte[][] uuidBytesValues = blockValSet.getBytesValuesSV(); + String[] hexValues = new String[length]; + for (int j = 0; j < length; j++) { + hexValues[j] = BytesUtils.toHexString(uuidBytesValues[j]); + } + valueArrays[i] = hexValues; + } else { + byte[][][] uuidBytesValuesMV = blockValSet.getBytesValuesMV(); + String[][] hexValuesMV = new String[length][]; + for (int j = 0; j < length; j++) { + byte[][] row = uuidBytesValuesMV[j]; + String[] hexRow = new String[row.length]; + for (int k = 0; k < row.length; k++) { + hexRow[k] = BytesUtils.toHexString(row[k]); + } + hexValuesMV[j] = hexRow; + } + valueArrays[i] = hexValuesMV; + } + continue; + } valueTypes[i] = storedType; if (singleValue) { switch (storedType) { diff --git a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountULLAggregationFunction.java b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountULLAggregationFunction.java index 5309a9713af3..3a9319e00c7c 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountULLAggregationFunction.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountULLAggregationFunction.java @@ -82,8 +82,21 @@ public void aggregate(int length, AggregationResultHolder aggregationResultHolde Map blockValSetMap) { BlockValSet blockValSet = blockValSetMap.get(_expression); - // Treat BYTES value as serialized HyperLogLogPlus - DataType storedType = blockValSet.getValueType().getStoredType(); + DataType dataType = blockValSet.getValueType(); + DataType storedType = dataType.getStoredType(); + + // UUID values are logical scalars (stored as 16-byte BYTES) — not serialized UltraLogLog state. Hash the + // canonical UUID string so DISTINCTCOUNTULL(uuidCol) matches DISTINCTCOUNTULL(CAST(uuidCol AS STRING)). + if (dataType == DataType.UUID) { + byte[][] uuidBytesValues = blockValSet.getBytesValuesSV(); + UltraLogLog ull = getULL(aggregationResultHolder); + for (int i = 0; i < length; i++) { + UltraLogLogUtils.hashObject(uuidBytesValues[i]).ifPresent(ull::add); + } + return; + } + + // Treat BYTES value as serialized UltraLogLog if (storedType == DataType.BYTES) { byte[][] bytesValues = blockValSet.getBytesValuesSV(); try { @@ -155,8 +168,20 @@ public void aggregateGroupBySV(int length, int[] groupKeyArray, GroupByResultHol Map blockValSetMap) { BlockValSet blockValSet = blockValSetMap.get(_expression); + DataType dataType = blockValSet.getValueType(); + DataType storedType = dataType.getStoredType(); + + // UUID columns: hash 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++) { + UltraLogLog ull = getULL(groupByResultHolder, groupKeyArray[i]); + UltraLogLogUtils.hashObject(uuidBytesValues[i]).ifPresent(ull::add); + } + return; + } + // Treat BYTES value as serialized UltraLogLogs - DataType storedType = blockValSet.getValueType().getStoredType(); if (storedType == DataType.BYTES) { byte[][] bytesValues = blockValSet.getBytesValuesSV(); try { @@ -234,8 +259,23 @@ public void aggregateGroupByMV(int length, int[][] groupKeysArray, GroupByResult Map blockValSetMap) { BlockValSet blockValSet = blockValSetMap.get(_expression); - // Treat BYTES value as serialized HyperLogLogPlus - DataType storedType = blockValSet.getValueType().getStoredType(); + DataType dataType = blockValSet.getValueType(); + DataType storedType = dataType.getStoredType(); + + // UUID columns: hash 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++) { + byte[] canonical = uuidBytesValues[i]; + for (int groupKey : groupKeysArray[i]) { + UltraLogLog ull = getULL(groupByResultHolder, groupKey); + UltraLogLogUtils.hashObject(canonical).ifPresent(ull::add); + } + } + return; + } + + // Treat BYTES value as serialized UltraLogLogs if (storedType == DataType.BYTES) { byte[][] bytesValues = blockValSet.getBytesValuesSV(); try { diff --git a/pinot-core/src/main/java/org/apache/pinot/core/query/distinct/table/BytesDistinctTable.java b/pinot-core/src/main/java/org/apache/pinot/core/query/distinct/table/BytesDistinctTable.java index 386648b64f5b..f41794fb7420 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/query/distinct/table/BytesDistinctTable.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/query/distinct/table/BytesDistinctTable.java @@ -32,6 +32,7 @@ import org.apache.pinot.common.request.context.OrderByExpressionContext; import org.apache.pinot.common.response.broker.ResultTable; import org.apache.pinot.common.utils.DataSchema; +import org.apache.pinot.common.utils.DataSchema.ColumnDataType; import org.apache.pinot.core.common.datatable.DataTableBuilder; import org.apache.pinot.core.common.datatable.DataTableBuilderFactory; import org.apache.pinot.spi.query.QueryThreadContext; @@ -263,57 +264,59 @@ private ResultTable toResultTableWithOrderBy() { } int numValues = sortedValues.length; assert numValues <= _limit; + ColumnDataType columnDataType = _dataSchema.getColumnDataType(0); List rows; if (_hasNull) { if (numValues == _limit) { rows = new ArrayList<>(_limit); if (_orderByExpression.isNullsLast()) { - addRows(sortedValues, numValues, rows); + addRows(columnDataType, sortedValues, numValues, rows); } else { rows.add(new Object[]{null}); - addRows(sortedValues, numValues - 1, rows); + addRows(columnDataType, sortedValues, numValues - 1, rows); } } else { rows = new ArrayList<>(numValues + 1); if (_orderByExpression.isNullsLast()) { - addRows(sortedValues, numValues, rows); + addRows(columnDataType, sortedValues, numValues, rows); rows.add(new Object[]{null}); } else { rows.add(new Object[]{null}); - addRows(sortedValues, numValues, rows); + addRows(columnDataType, sortedValues, numValues, rows); } } } else { rows = new ArrayList<>(numValues); - addRows(sortedValues, numValues, rows); + addRows(columnDataType, sortedValues, numValues, rows); } return new ResultTable(_dataSchema, rows); } - private static void addRows(ByteArray[] values, int length, List rows) { + private static void addRows(ColumnDataType columnDataType, ByteArray[] values, int length, List rows) { for (int i = 0; i < length; i++) { - rows.add(new Object[]{values[i].toHexString()}); + rows.add(new Object[]{columnDataType.convertAndFormat(values[i])}); } } private ResultTable toResultTableWithoutOrderBy() { int numValues = _valueSet.size(); assert numValues <= _limit; + ColumnDataType columnDataType = _dataSchema.getColumnDataType(0); List rows; if (_hasNull && numValues < _limit) { rows = new ArrayList<>(numValues + 1); - addRows(_valueSet, rows); + addRows(columnDataType, _valueSet, rows); rows.add(new Object[]{null}); } else { rows = new ArrayList<>(numValues); - addRows(_valueSet, rows); + addRows(columnDataType, _valueSet, rows); } return new ResultTable(_dataSchema, rows); } - private static void addRows(HashSet values, List rows) { + private static void addRows(ColumnDataType columnDataType, HashSet values, List rows) { for (ByteArray value : values) { - rows.add(new Object[]{value.toHexString()}); + rows.add(new Object[]{columnDataType.convertAndFormat(value)}); } } } diff --git a/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/GroupByDataTableReducer.java b/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/GroupByDataTableReducer.java index c1add8ca7082..be757ecd9770 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/GroupByDataTableReducer.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/GroupByDataTableReducer.java @@ -531,6 +531,13 @@ private Object getConvertedKey(DataTable dataTable, ColumnDataType columnDataTyp return dataTable.getString(rowId, colId); case BYTES: return dataTable.getBytes(rowId, colId).getBytes(); + case UUID: + // Deliberately delegated to ColumnDataType#convert rather than falling through to BYTES. The other reduce + // path (reduceWithIndexedTable) converts group keys with exactly that method, and UUID is the one type + // whose converted form is not its stored bytes -- it yields a java.util.UUID. PredicateRowMatcher casts + // directly on that, so returning the raw byte[] here makes GROUP BY ... HAVING on a UUID column throw + // ClassCastException. Delegating keeps the two paths identical by construction. + return columnDataType.convert(dataTable.getBytes(rowId, colId)); default: throw new IllegalStateException("Illegal column data type in group key: " + columnDataType); } diff --git a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/DistinctCountHLLAggregationFunctionTest.java b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/DistinctCountHLLAggregationFunctionTest.java index 2520115affa0..3049fae37358 100644 --- a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/DistinctCountHLLAggregationFunctionTest.java +++ b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/DistinctCountHLLAggregationFunctionTest.java @@ -19,14 +19,22 @@ package org.apache.pinot.core.query.aggregation.function; import com.clearspring.analytics.stream.cardinality.HyperLogLog; +import java.util.Arrays; import java.util.BitSet; +import java.util.HashMap; import java.util.List; import java.util.Map; import org.apache.pinot.common.request.Literal; import org.apache.pinot.common.request.context.ExpressionContext; +import org.apache.pinot.common.request.context.RequestContextUtils; +import org.apache.pinot.core.common.BlockValSet; +import org.apache.pinot.core.query.aggregation.AggregationResultHolder; import org.apache.pinot.core.query.aggregation.ObjectAggregationResultHolder; import org.apache.pinot.segment.spi.Constants; import org.apache.pinot.segment.spi.index.reader.Dictionary; +import org.apache.pinot.spi.data.FieldSpec.DataType; +import org.apache.pinot.spi.utils.CommonConstants; +import org.apache.pinot.spi.utils.UuidUtils; import org.testng.Assert; import org.testng.annotations.Test; @@ -55,6 +63,108 @@ public void testCanUseStarTreeDefaultLog2m() { Assert.assertFalse(function.canUseStarTree(Map.of(Constants.HLL_LOG2M_KEY, "16"))); } + /// Regression: UUID columns have storedType=BYTES, but a UUID value is a logical scalar, not a serialized + /// HyperLogLog. The aggregator must route UUID columns through the same content-hash path as STRING (offering + /// canonical UUID strings) instead of trying to deserialize each 16-byte value as an HLL. + @Test + public void testAggregateOnUuidColumnOffersCanonicalStringsAndProducesExactDistinctCount() { + ExpressionContext expression = RequestContextUtils.getExpression("uuidCol"); + DistinctCountHLLAggregationFunction function = new DistinctCountHLLAggregationFunction(List.of(expression)); + + // Three distinct UUIDs across six rows; the same UUID repeats twice on rows 0/3, 1/4, 2/5. + String[] uuidStrings = new String[]{ + "550e8400-e29b-41d4-a716-446655440000", + "12345678-1234-1234-1234-1234567890ab", + "9c5e1f24-0b8e-4c9d-87f1-0aa64a3b9d12", + "550e8400-e29b-41d4-a716-446655440000", + "12345678-1234-1234-1234-1234567890ab", + "9c5e1f24-0b8e-4c9d-87f1-0aa64a3b9d12" + }; + + // Stub the BYTES fetch (raw 16-byte values) — the production path converts bytes to canonical strings + // itself because ProjectionBlockValSet.getStringValuesSV() would render stored BYTES as bare hex. + byte[][] uuidBytes = new byte[uuidStrings.length][]; + for (int i = 0; i < uuidStrings.length; i++) { + uuidBytes[i] = UuidUtils.toBytes(uuidStrings[i]); + } + BlockValSet uuidBlockValSet = mock(BlockValSet.class); + when(uuidBlockValSet.getValueType()).thenReturn(DataType.UUID); + when(uuidBlockValSet.getBytesValuesSV()).thenReturn(uuidBytes); + when(uuidBlockValSet.isSingleValue()).thenReturn(true); + when(uuidBlockValSet.getDictionary()).thenReturn(null); + + Map blockValSetMap = new HashMap<>(); + blockValSetMap.put(expression, uuidBlockValSet); + + AggregationResultHolder resultHolder = function.createAggregationResultHolder(); + function.aggregate(uuidStrings.length, resultHolder, blockValSetMap); + + Object intermediate = function.extractAggregationResult(resultHolder); + Assert.assertTrue(intermediate instanceof HyperLogLog, + "Intermediate result must be a HyperLogLog, not a dictionary bitmap"); + long cardinality = ((HyperLogLog) intermediate).cardinality(); + Assert.assertEquals(cardinality, 3L, + "HLL cardinality must equal the 3 distinct UUIDs; got " + cardinality); + } + + /// UUID columns hash their **stored bytes**, exactly as TIMESTAMP hashes its stored millis rather than a + /// formatted string. Consequence: DISTINCTCOUNTHLL(uuidCol) does NOT equal + /// DISTINCTCOUNTHLL(CAST(uuidCol AS STRING)) -- and neither does it for TIMESTAMP, so this is the consistent + /// behaviour for a logical type, not a gap. Pinned here so nobody "fixes" it back into a canonical-string + /// rendering, which would reintroduce a per-row String allocation in the aggregation loop. + @Test + public void testUuidDistinctCountHllHashesStoredBytesNotCanonicalString() + throws java.io.IOException { + String[] uuidStrings = new String[]{ + "550e8400-e29b-41d4-a716-446655440000", + "12345678-1234-1234-1234-1234567890ab", + "9c5e1f24-0b8e-4c9d-87f1-0aa64a3b9d12" + }; + + // Cardinality is still exact for a small distinct set... + Assert.assertEquals(computeHllCardinality(uuidStrings, DataType.UUID), 3L); + + // ...but the sketch is built over the 16 stored bytes, so a HyperLogLog fed the canonical strings differs. + HyperLogLog fromCanonicalStrings = new HyperLogLog(CommonConstants.Helix.DEFAULT_HYPERLOGLOG_LOG2M); + for (String uuid : uuidStrings) { + fromCanonicalStrings.offer(uuid); + } + HyperLogLog fromStoredBytes = new HyperLogLog(CommonConstants.Helix.DEFAULT_HYPERLOGLOG_LOG2M); + for (String uuid : uuidStrings) { + fromStoredBytes.offer(UuidUtils.toBytes(uuid)); + } + Assert.assertFalse(Arrays.equals(fromCanonicalStrings.getBytes(), fromStoredBytes.getBytes()), + "stored-bytes and canonical-string sketches are expected to differ"); + } + + private long computeHllCardinality(String[] values, DataType valueType) { + ExpressionContext expression = RequestContextUtils.getExpression("col"); + DistinctCountHLLAggregationFunction function = new DistinctCountHLLAggregationFunction(List.of(expression)); + + BlockValSet blockValSet = mock(BlockValSet.class); + when(blockValSet.getValueType()).thenReturn(valueType); + if (valueType == DataType.UUID) { + // UUID path fetches raw bytes and converts to canonical form itself (projection string fetch returns hex) + byte[][] uuidBytes = new byte[values.length][]; + for (int i = 0; i < values.length; i++) { + uuidBytes[i] = UuidUtils.toBytes(values[i]); + } + when(blockValSet.getBytesValuesSV()).thenReturn(uuidBytes); + } else { + when(blockValSet.getStringValuesSV()).thenReturn(values); + } + when(blockValSet.isSingleValue()).thenReturn(true); + when(blockValSet.getDictionary()).thenReturn(null); + + Map blockValSetMap = new HashMap<>(); + blockValSetMap.put(expression, blockValSet); + + AggregationResultHolder resultHolder = function.createAggregationResultHolder(); + function.aggregate(values.length, resultHolder, blockValSetMap); + Object intermediate = function.extractAggregationResult(resultHolder); + return ((HyperLogLog) intermediate).cardinality(); + } + @Test public void testCanUseStarTreeCustomLog2m() { DistinctCountHLLAggregationFunction function = new DistinctCountHLLAggregationFunction( diff --git a/pinot-core/src/test/java/org/apache/pinot/core/query/distinct/table/BytesDistinctTableTest.java b/pinot-core/src/test/java/org/apache/pinot/core/query/distinct/table/BytesDistinctTableTest.java new file mode 100644 index 000000000000..6ca1c2298cb3 --- /dev/null +++ b/pinot-core/src/test/java/org/apache/pinot/core/query/distinct/table/BytesDistinctTableTest.java @@ -0,0 +1,70 @@ +/** + * 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.distinct.table; + +import org.apache.pinot.common.request.context.ExpressionContext; +import org.apache.pinot.common.request.context.OrderByExpressionContext; +import org.apache.pinot.common.response.broker.ResultTable; +import org.apache.pinot.common.utils.DataSchema; +import org.apache.pinot.common.utils.DataSchema.ColumnDataType; +import org.apache.pinot.spi.utils.ByteArray; +import org.apache.pinot.spi.utils.BytesUtils; +import org.apache.pinot.spi.utils.UuidUtils; +import org.testng.annotations.Test; + +import static org.testng.Assert.assertEquals; + + +/// Tests for [BytesDistinctTable]. +public class BytesDistinctTableTest { + private static final String UUID_COLUMN = "uuidCol"; + private static final String UUID_VALUE_1 = "550e8400-e29b-41d4-a716-446655440000"; + private static final String UUID_VALUE_2 = "550e8400-e29b-41d4-a716-446655440001"; + + @Test + public void testToResultTableFormatsUuidAndBytesWithoutOrderBy() { + BytesDistinctTable uuidTable = new BytesDistinctTable( + new DataSchema(new String[]{UUID_COLUMN}, new ColumnDataType[]{ColumnDataType.UUID}), 10, false, null); + uuidTable.addUnbounded(new ByteArray(UuidUtils.toBytes(UUID_VALUE_1))); + + ResultTable uuidResultTable = uuidTable.toResultTable(); + assertEquals(uuidResultTable.getRows().get(0)[0], UUID_VALUE_1); + + byte[] bytesValue = new byte[]{0x01, 0x23, 0x45}; + BytesDistinctTable bytesTable = new BytesDistinctTable( + new DataSchema(new String[]{"bytesCol"}, new ColumnDataType[]{ColumnDataType.BYTES}), 10, false, null); + bytesTable.addUnbounded(new ByteArray(bytesValue)); + + ResultTable bytesResultTable = bytesTable.toResultTable(); + assertEquals(bytesResultTable.getRows().get(0)[0], BytesUtils.toHexString(bytesValue)); + } + + @Test + public void testToResultTableFormatsUuidWithOrderBy() { + BytesDistinctTable uuidTable = new BytesDistinctTable( + new DataSchema(new String[]{UUID_COLUMN}, new ColumnDataType[]{ColumnDataType.UUID}), 10, false, + new OrderByExpressionContext(ExpressionContext.forIdentifier(UUID_COLUMN), true)); + uuidTable.addUnbounded(new ByteArray(UuidUtils.toBytes(UUID_VALUE_2))); + uuidTable.addUnbounded(new ByteArray(UuidUtils.toBytes(UUID_VALUE_1))); + + ResultTable resultTable = uuidTable.toResultTable(); + assertEquals(resultTable.getRows().get(0)[0], UUID_VALUE_1); + assertEquals(resultTable.getRows().get(1)[0], UUID_VALUE_2); + } +} diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/UuidAggregationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/UuidAggregationTest.java new file mode 100644 index 000000000000..016ed27e3606 --- /dev/null +++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/UuidAggregationTest.java @@ -0,0 +1,209 @@ +/** + * 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.integration.tests.custom; + +import com.fasterxml.jackson.databind.JsonNode; +import java.io.File; +import java.util.ArrayList; +import java.util.List; +import org.apache.avro.file.DataFileWriter; +import org.apache.avro.generic.GenericData; +import org.apache.pinot.spi.config.table.TableConfig; +import org.apache.pinot.spi.config.table.TableType; +import org.apache.pinot.spi.data.FieldSpec.DataType; +import org.apache.pinot.spi.data.Schema; +import org.apache.pinot.spi.utils.builder.TableConfigBuilder; +import org.testng.annotations.Test; + +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertTrue; + + +/// End-to-end coverage for aggregating, grouping and de-duplicating a UUID column. +/// +/// These run through a real broker reduce, which is the point: the group-key conversion in +/// `GroupByDataTableReducer#getConvertedKey` is only reachable when the broker reduces a *single* data table, and the +/// unit-level `BaseQueriesTest` harness always reduces two. A `case UUID` there that returns the stored `byte[]` +/// instead of the converted `java.util.UUID` makes `GROUP BY ... HAVING` over a UUID column fail with +/// `ClassCastException: class [B cannot be cast to class java.util.UUID`, and only a query-level test catches it. +@Test(suiteName = "CustomClusterIntegrationTest") +public class UuidAggregationTest extends CustomDataQueryClusterIntegrationTest { + private static final String TABLE_NAME = "UuidAggregationTest"; + private static final String UUID_COLUMN = "uuidColumn"; + private static final String UUID_0 = "550e8400-e29b-41d4-a716-446655440000"; + private static final String UUID_0_HEX = "550e8400e29b41d4a716446655440000"; + private static final String UUID_1 = "550e8400-e29b-41d4-a716-446655440001"; + private static final String UUID_2 = "550e8400-e29b-41d4-a716-446655440002"; + + /// `UUID_0` appears twice so grouping and distinct are distinguishable from a plain row count. + private static final List ROWS = List.of(UUID_0, UUID_0, UUID_1, UUID_2); + private static final int NUM_DISTINCT = 3; + + @Override + public String getTableName() { + return TABLE_NAME; + } + + @Override + protected long getCountStarResult() { + return ROWS.size(); + } + + @Override + public int getNumAvroFiles() { + return 1; + } + + @Override + public TableConfig createOfflineTableConfig() { + return new TableConfigBuilder(TableType.OFFLINE).setTableName(getTableName()).build(); + } + + @Override + public Schema createSchema() { + return new Schema.SchemaBuilder().setSchemaName(getTableName()) + .addSingleValueDimension(UUID_COLUMN, DataType.UUID) + .build(); + } + + @Override + public List createAvroFiles() + throws Exception { + org.apache.avro.Schema avroSchema = org.apache.avro.Schema.createRecord("uuidRecord", null, null, false); + avroSchema.setFields(List.of(new org.apache.avro.Schema.Field(UUID_COLUMN, + org.apache.avro.Schema.create(org.apache.avro.Schema.Type.STRING), null, null))); + + try (AvroFilesAndWriters avroFilesAndWriters = createAvroFilesAndWriters(avroSchema)) { + DataFileWriter writer = avroFilesAndWriters.getWriters().get(0); + for (String uuid : ROWS) { + GenericData.Record record = new GenericData.Record(avroSchema); + record.put(UUID_COLUMN, uuid); + writer.append(record); + } + return avroFilesAndWriters.getAvroFiles(); + } + } + + @Test + public void testGroupByUuidColumn() + throws Exception { + setUseMultiStageQueryEngine(false); + JsonNode rows = query( + String.format("SELECT %s, COUNT(*) FROM %s GROUP BY %s ORDER BY %s", UUID_COLUMN, getTableName(), UUID_COLUMN, + UUID_COLUMN)); + assertEquals(rows.size(), NUM_DISTINCT, rows.toPrettyString()); + + // Group keys must come back as canonical UUIDs, not hex and not a byte-array rendering. + List keys = new ArrayList<>(); + for (JsonNode row : rows) { + keys.add(row.get(0).asText()); + } + assertEquals(keys, List.of(UUID_0, UUID_1, UUID_2), rows.toPrettyString()); + assertEquals(rows.get(0).get(1).asLong(), 2, rows.toPrettyString()); + assertEquals(rows.get(1).get(1).asLong(), 1, rows.toPrettyString()); + } + + /// The regression that motivated this class: `GROUP BY` a UUID column with a `HAVING` predicate on that same + /// column runs the group key through `getConvertedKey` and then straight into `PredicateRowMatcher`, which casts + /// to `java.util.UUID`. + @Test + public void testGroupByUuidColumnWithHaving() + throws Exception { + setUseMultiStageQueryEngine(false); + JsonNode rows = query( + String.format("SELECT %s, COUNT(*) FROM %s GROUP BY %s HAVING %s = '%s'", UUID_COLUMN, getTableName(), + UUID_COLUMN, UUID_COLUMN, UUID_0_HEX)); + assertEquals(rows.size(), 1, rows.toPrettyString()); + assertEquals(rows.get(0).get(0).asText(), UUID_0, rows.toPrettyString()); + assertEquals(rows.get(0).get(1).asLong(), 2, rows.toPrettyString()); + + // Same thing via an explicit CAST of the canonical form. + rows = query(String.format("SELECT %s, COUNT(*) FROM %s GROUP BY %s HAVING %s = CAST('%s' AS UUID)", UUID_COLUMN, + getTableName(), UUID_COLUMN, UUID_COLUMN, UUID_1)); + assertEquals(rows.size(), 1, rows.toPrettyString()); + assertEquals(rows.get(0).get(0).asText(), UUID_1, rows.toPrettyString()); + assertEquals(rows.get(0).get(1).asLong(), 1, rows.toPrettyString()); + } + + /// The `serverReturnFinalResult` variant. This is the option that gates the single-data-table branch in + /// `GroupByDataTableReducer#reduceAndSetResults` (`isServerReturnFinalResult() && dataTables.size() == 1`), so it + /// covers a different reduce branch from the test above. Verified by reverting the `getConvertedKey` fix: both + /// this and the plain `GROUP BY ... HAVING` test fail with + /// `ClassCastException: class [B cannot be cast to class java.util.UUID`, and both pass with it. + @Test + public void testGroupByUuidColumnWithHavingReturningFinalResult() + throws Exception { + setUseMultiStageQueryEngine(false); + JsonNode rows = query( + String.format("SELECT %s, COUNT(*) FROM %s GROUP BY %s HAVING %s = '%s' OPTION(serverReturnFinalResult=true)", + UUID_COLUMN, getTableName(), UUID_COLUMN, UUID_COLUMN, UUID_0_HEX)); + assertEquals(rows.size(), 1, rows.toPrettyString()); + assertEquals(rows.get(0).get(0).asText(), UUID_0, rows.toPrettyString()); + assertEquals(rows.get(0).get(1).asLong(), 2, rows.toPrettyString()); + } + + /// Group keys must also render canonically on the `getConvertedKey` path, not as hex. + @Test + public void testGroupByUuidColumnReturningFinalResult() + throws Exception { + setUseMultiStageQueryEngine(false); + JsonNode rows = query( + String.format("SELECT %s, COUNT(*) FROM %s GROUP BY %s ORDER BY %s OPTION(serverReturnFinalResult=true)", + UUID_COLUMN, getTableName(), UUID_COLUMN, UUID_COLUMN)); + assertEquals(rows.size(), NUM_DISTINCT, rows.toPrettyString()); + List keys = new ArrayList<>(); + for (JsonNode row : rows) { + keys.add(row.get(0).asText()); + } + assertEquals(keys, List.of(UUID_0, UUID_1, UUID_2), rows.toPrettyString()); + } + + @Test + public void testDistinctOnUuidColumn() + throws Exception { + setUseMultiStageQueryEngine(false); + JsonNode rows = + query(String.format("SELECT DISTINCT %s FROM %s ORDER BY %s", UUID_COLUMN, getTableName(), UUID_COLUMN)); + assertEquals(rows.size(), NUM_DISTINCT, rows.toPrettyString()); + + // BytesDistinctTable used to hard-code toHexString(); a UUID column must render canonically. + List values = new ArrayList<>(); + for (JsonNode row : rows) { + values.add(row.get(0).asText()); + } + assertEquals(values, List.of(UUID_0, UUID_1, UUID_2), rows.toPrettyString()); + } + + @Test + public void testDistinctCountOnUuidColumn() + throws Exception { + setUseMultiStageQueryEngine(false); + for (String function : List.of("DISTINCTCOUNT", "DISTINCTCOUNTHLL", "DISTINCTCOUNTBITMAP")) { + JsonNode rows = query(String.format("SELECT %s(%s) FROM %s", function, UUID_COLUMN, getTableName())); + assertEquals(rows.get(0).get(0).asLong(), NUM_DISTINCT, function + ": " + rows.toPrettyString()); + } + } + + private JsonNode query(String sql) + throws Exception { + JsonNode response = postQuery(sql); + assertTrue(response.path("exceptions").isEmpty(), sql + " -> " + response.toPrettyString()); + return response.path("resultTable").path("rows"); + } +}