This is an automated email from the ASF dual-hosted git repository.
xiangfu0 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 1de06413cad [UUID 5/8] UUID aggregation, group-by and distinct (#18873)
1de06413cad is described below
commit 1de06413cadb84edd0d48bbbb6d458ca200fbd69
Author: Xiang Fu <[email protected]>
AuthorDate: Mon Aug 17 14:25:03 2026 -0700
[UUID 5/8] UUID aggregation, group-by and distinct (#18873)
* [UUID 5/8] UUID aggregation, group-by and distinct
Part 5/8 of splitting apache/pinot#18140 (logical UUID type). Rebased onto
latest master; stacked on uuid-split/04-sse-predicates-cast.
Downstream references use the UuidKey class merged in #18869.
* Return the converted UUID form for group keys in the DataTable reduce path
getConvertedKey had `case UUID` falling through to BYTES, returning the raw
byte[]. The other reduce path converts group keys via
ColumnDataType#convert,
and UUID is the one type whose converted form is not its stored bytes -- it
yields a java.util.UUID.
PredicateRowMatcher casts that directly (see #18872), so the byte[] made
GROUP BY ... HAVING over a UUID column fail with
"ClassCastException: class [B cannot be cast to class java.util.UUID".
Delegating to columnDataType.convert(...) keeps the two paths identical by
construction rather than by duplicated knowledge.
Covered by a new UuidAggregationTest integration test rather than a unit
test.
The unit-level BaseQueriesTest harness cannot reach this code, which is why
it
was uncovered; a query-level test goes through the real broker reduce.
Verified
by reverting the fix: testGroupByUuidColumnWithHaving and
testGroupByUuidColumnWithHavingReturningFinalResult both fail with the
ClassCastException and both pass with it.
The test also covers GROUP BY key rendering, DISTINCT (BytesDistinctTable no
longer hard-codes hex) and DISTINCTCOUNT / DISTINCTCOUNTHLL /
DISTINCTCOUNTBITMAP
over a UUID column.
* Cast directly to UuidKey in UuidToIdMap
Both callers key on UuidKey already
(NoDictionary{Single,Multi}ColumnGroupKey
Generator, via UuidKey.fromBytes), so UuidKey.fromObject accepted five input
types where exactly one is ever passed, and ran an instanceof chain per row
in
the group-by loop.
Casting directly matches the sibling maps -- DoubleToIdMap casts to double
--
and keeps the input type deterministic, which is what was asked for on the
equivalent PredicateRowMatcher branch in #18872.
* Hash UUID's stored bytes in distinct-count aggregations, not a canonical
string
The previous version rendered each UUID as its 36-char canonical string so
that
DISTINCTCOUNTHLL(uuidCol) would equal DISTINCTCOUNTHLL(CAST(uuidCol AS
STRING)).
No other logical type provides that guarantee: the scan path switches on the
stored type, so TIMESTAMP offers its raw millis and BOOLEAN its int, and
neither
matches a CAST to STRING. The UUID rendering was inventing a cross-type
equivalence at the cost of a String allocation per row in the aggregation
loop.
UUID now hashes its stored 16 bytes. Verified byte[] is content-hashed
rather
than identity-hashed by both HyperLogLog (clearspring MurmurHash) and
UltraLogLogUtils.OBJECT_FUNNEL (putBytes).
A minimal guard is still needed at each site, because unlike LONG or INT the
stored BYTES type is not a scalar case in this family: the scan path has no
`case BYTES`, and the dictionary path reads BYTES as serialized sketch
state.
- AggregationFunctionUtils: the three UUID blocks are gone; the BYTES guard
now
excludes UUID so it falls through to the scalar path, which offers
dictionary.get(i) -- the stored byte[] -- exactly as the scan path does.
- DistinctCountBitmap hashes Arrays.hashCode(bytes).
- DistinctCountThetaSketch cannot take scalar bytes (BYTES there means
"serialized sketch"), so it surfaces the stored hex rendering instead.
Replaces testUuidDistinctCountHllMatchesStringDistinctCountHll, which
asserted
the invariant being dropped, with one pinning the new behaviour.
* Drop UUID special-casing that the stored type already covers
Applying the rule that if TIMESTAMP and BIG_DECIMAL need no special
handling at
a site, UUID does not either -- their stored types carry them, and UUID's
should too.
Removed:
- NoDictionary{Single,Multi}ColumnGroupKeyGenerator, UuidToIdMap and its
ValueToIdMapFactory entry. `case BYTES` is already supported there and is
structurally identical to the UUID branch: same getBytesValuesSV(), same
loop, same map, differing only in wrapping UuidKey vs ByteArray. Both
yield
ByteArray downstream, which is what ColumnDataType.UUID#convert consumes,
so
this was a pure key-representation optimization -- two primitive longs
instead of a byte[] wrapper -- with no benchmark to justify 5 files and
+237/-36. UuidAggregationTest#testGroupByUuidColumn still passes,
confirming
group keys render identically without it.
- AnyValueAggregationFunction. Its switch is on getStoredType(), so
TIMESTAMP
already collapses to LONG (raw millis) and BOOLEAN to INT. UUID
collapsing to
BYTES is the consistent behaviour; the branch made it the odd one out.
- IntegerTupleSketchAggregationFunction. The UUID branch only produced a
friendlier error message; TIMESTAMP gets no such treatment when it is
equally
unusable there.
Kept, because the same rule shows they are needed:
- BytesDistinctTable: LongDistinctTable and BigDecimalDistinctTable already
render by ColumnDataType, and MultiColumnDistinctTable already calls
convertAndFormat. BytesDistinctTable hard-coding toHexString() was the
outlier; this brings it in line rather than special-casing UUID.
- The distinct-count guards and GroupByDataTableReducer, both verified
necessary by probe -- BYTES means "serialized sketch" in one and the group
key must convert in the other.
19 files / +958 -80 -> 12 files / +694 -39.
* Format DISTINCT rows in one pass in BytesDistinctTable
The previous version added every row holding a ByteArray, then walked the
whole
list again in formatRows() to rewrite row[0]. Two traversals and two writes
per
row where master did one.
The formatting now happens inline in addRows, with the ColumnDataType
passed in
-- the same shape LongDistinctTable already uses for its TIMESTAMP handling.
Not an extra String allocation either way: ColumnDataType#convertAndFormat
for
BYTES is ((ByteArray) value).toHexString(), the exact call master made, and
ByteArray#getBytes() returns the internal array without copying.
* Route UUID through the normal dispatch in DistinctCountBitmap
Replaces the three `if (dataType == DataType.UUID)` early-returns with a
narrowed guard on the serialized-bitmap branch:
if (storedType == DataType.BYTES && dataType != DataType.UUID)
UUID now falls through to aggregateSV/aggregateMV and the group-by variants
like any other type, handled by a `case BYTES` in each switch. That fixes MV
UUID, which the early-returns silently broke -- they called
getBytesValuesSV()
above the isSingleValue() dispatch, so an MV column took the SV accessor.
Seven switches needed the case, not six: the dictionary path in
convertToValueBitmap also throws on BYTES, which the integration test
caught --
DISTINCTCOUNTBITMAP(uuidColumn) failed with "Illegal data type for
DISTINCT_COUNT_BITMAP aggregation function: BYTES" until it was added.
DistinctCountHLL, DistinctCountHLLPlus, DistinctCountCPCSketch and
DistinctCountULL still use the early-return shape and need the same
treatment.
* Route UUID through the normal dispatch in DistinctCountHLL
Same shape as DistinctCountBitmap: the three `if (dataType ==
DataType.UUID)`
early-returns are gone, the serialized-HLL guard is narrowed to
if (storedType == DataType.BYTES && dataType != DataType.UUID)
and UUID falls through to aggregateSV/aggregateMV and the group-by variants,
handled by a `case BYTES` in each of the six switches. This fixes MV UUID,
which the early-returns broke by calling getBytesValuesSV() above the
isSingleValue() dispatch.
No seventh switch here: unlike DistinctCountBitmap, HLL's dictionary path
lives
in AggregationFunctionUtils#getDistinctCountHLLResult, which already
excludes
UUID from the serialized branch so it falls through to the scalar path.
* Route UUID through the normal dispatch in DistinctCountHLLPlus and ULL
Same shape as Bitmap and HLL: the three `if (dataType == DataType.UUID)`
early-returns are removed, each serialized-sketch guard is narrowed to
if (storedType == DataType.BYTES && dataType != DataType.UUID)
and UUID reaches the ordinary dispatch via a `case BYTES` in each switch.
HLLPlus has 6 switches (SV/MV x aggregate, groupBySV, groupByMV). ULL has 3:
it is single-value only, so all of its switches read getStringValuesSV and
the
BYTES cases mirror that -- no MV variants to add.
Both dictionary paths live in AggregationFunctionUtils, which already
excludes
UUID from the serialized branch, so neither needed a seventh switch.
* Read UUID as bytes in DistinctCountThetaSketch
extractValues no longer special-cases UUID: the block that materialized a
String[] / String[][] of hex renderings is gone, so a UUID column now
reports
its stored type (BYTES) and carries the raw byte[][] like any bytes column.
All the handling moved into the aggregateXXX methods. Each of the three
dispatch sites now reads the logical type from the block val set it already
has
and narrows its guard, so UUID takes the scalar path rather than the
serialized-sketch path, and a "case BYTES" in each of the six scalar
switches
feeds the stored bytes straight to UpdatableThetaSketch#update(byte[]) -- no
String allocated anywhere.
Same shape as Bitmap, HLL, HLLPlus and ULL. This also fixes MV UUID here for
the same reason: the work now happens inside the SV and MV branches instead
of
above them.
* Leave extractValues untouched in DistinctCountThetaSketch
The length parameter was only needed to size the hex String arrays that the
UUID block used to build. With that block gone, length was unused in the
body,
so the signature change and its three call sites were dead weight.
extractValues is now byte-identical to master: all UUID handling lives in
the
aggregateXXX methods.
* Route UUID through the normal dispatch in DistinctCountCPCSketch
Last of the six. Same shape as the others: the UUID early-returns are gone,
the
serialized-sketch guard is narrowed with `&& dataType != DataType.UUID`,
and a
`case BYTES` in each switch feeds the stored bytes to CpcSketch#update.
This one had 2 UUID blocks and 1 BYTES guard rather than 3 and 3 -- the
counts
differ per function, so they were verified rather than assumed.
No `if (dataType == DataType.UUID)` early-return remains anywhere in the
aggregation function package.
* Fix UUID aggregation dispatch for CPC and Theta sketches
Route UUID values through the single-value and multi-value switch branches
for aggregate and group-by paths while preserving serialized BYTES sketch
handling.
Add unit and ingestion/query coverage for UUID and multi-value serialized
sketches.
* Address UUID aggregation review feedback
* Clarify serialized aggregation BYTES handling
* Restore legacy serialized BYTES dispatch
* Simplify UUID aggregation coverage
* Address UUID aggregation review comments
---
.../function/AggregationFunctionUtils.java | 40 ++--
.../DistinctCountBitmapAggregationFunction.java | 76 +++++++-
.../DistinctCountCPCSketchAggregationFunction.java | 164 ++++++++++++++--
.../DistinctCountHLLAggregationFunction.java | 71 ++++++-
.../DistinctCountHLLPlusAggregationFunction.java | 71 ++++++-
...istinctCountThetaSketchAggregationFunction.java | 158 +++++++++++++--
.../DistinctCountULLAggregationFunction.java | 44 ++++-
.../query/distinct/table/BytesDistinctTable.java | 28 ++-
.../core/query/reduce/GroupByDataTableReducer.java | 3 +
.../tests/custom/UuidAggregationTest.java | 212 +++++++++++++++++++++
10 files changed, 783 insertions(+), 84 deletions(-)
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 527b7817621..e4071beaa47 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
@@ -69,6 +69,7 @@ import org.apache.pinot.segment.local.utils.UltraLogLogUtils;
import org.apache.pinot.segment.spi.AggregationFunctionType;
import org.apache.pinot.segment.spi.SegmentContext;
import org.apache.pinot.segment.spi.datasource.DataSource;
+import org.apache.pinot.segment.spi.datasource.DataSourceMetadata;
import org.apache.pinot.segment.spi.index.reader.Dictionary;
import
org.apache.pinot.segment.spi.index.startree.AggregationFunctionColumnPair;
import org.apache.pinot.spi.data.FieldSpec;
@@ -610,23 +611,23 @@ public class AggregationFunctionUtils {
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 +643,7 @@ public class AggregationFunctionUtils {
(DistinctCountSmartHLLPlusAggregationFunction)
aggregationFunction, explainPlanName);
break;
case DISTINCTCOUNTULL:
- result =
getDistinctCountULLResult(Objects.requireNonNull(dataSource.getDictionary()),
+ result = getDistinctCountULLResult(dataSource,
(DistinctCountULLAggregationFunction) aggregationFunction,
explainPlanName);
break;
case DISTINCTCOUNTSMARTULL:
@@ -650,7 +651,7 @@ public class AggregationFunctionUtils {
(DistinctCountSmartULLAggregationFunction) aggregationFunction,
explainPlanName);
break;
case DISTINCTCOUNTRAWULL:
- result =
getDistinctCountULLResult(Objects.requireNonNull(dataSource.getDictionary()),
+ result = getDistinctCountULLResult(dataSource,
(DistinctCountULLAggregationFunction) aggregationFunction,
explainPlanName);
break;
default:
@@ -799,10 +800,13 @@ public class AggregationFunctionUtils {
return hllPlus;
}
- private static HyperLogLog getDistinctCountHLLResult(Dictionary dictionary,
+ private static HyperLogLog getDistinctCountHLLResult(DataSource dataSource,
DistinctCountHLLAggregationFunction function, String explainPlanName) {
- if (dictionary.getValueType() == FieldSpec.DataType.BYTES) {
- // Treat BYTES value as serialized HyperLogLog
+ Dictionary dictionary = dataSource.getDictionary();
+ assert dictionary != null;
+ DataSourceMetadata metadata = dataSource.getDataSourceMetadata();
+ if (metadata.getDataType() == FieldSpec.DataType.BYTES) {
+ // Logical BYTES dictionary entries are serialized HyperLogLog objects.
try {
QueryThreadContext.checkTerminationAndSampleUsage(explainPlanName);
HyperLogLog hll =
ObjectSerDeUtils.HYPER_LOG_LOG_SER_DE.deserialize(dictionary.getBytesValue(0));
@@ -820,10 +824,13 @@ public class AggregationFunctionUtils {
}
}
- private static HyperLogLogPlus getDistinctCountHLLPlusResult(Dictionary
dictionary,
+ private static HyperLogLogPlus getDistinctCountHLLPlusResult(DataSource
dataSource,
DistinctCountHLLPlusAggregationFunction function, String
explainPlanName) {
- if (dictionary.getValueType() == FieldSpec.DataType.BYTES) {
- // Treat BYTES value as serialized HyperLogLogPlus
+ Dictionary dictionary = dataSource.getDictionary();
+ assert dictionary != null;
+ DataSourceMetadata metadata = dataSource.getDataSourceMetadata();
+ if (metadata.getDataType() == FieldSpec.DataType.BYTES) {
+ // Logical BYTES dictionary entries are serialized HyperLogLogPlus
objects.
try {
QueryThreadContext.checkTerminationAndSampleUsage(explainPlanName);
HyperLogLogPlus hllplus =
ObjectSerDeUtils.HYPER_LOG_LOG_PLUS_SER_DE.deserialize(dictionary.getBytesValue(0));
@@ -861,10 +868,13 @@ public class AggregationFunctionUtils {
}
}
- private static UltraLogLog getDistinctCountULLResult(Dictionary dictionary,
+ private static UltraLogLog getDistinctCountULLResult(DataSource dataSource,
DistinctCountULLAggregationFunction function, String explainPlanName) {
- if (dictionary.getValueType() == FieldSpec.DataType.BYTES) {
- // Treat BYTES value as serialized UltraLogLog and merge
+ Dictionary dictionary = dataSource.getDictionary();
+ assert dictionary != null;
+ DataSourceMetadata metadata = dataSource.getDataSourceMetadata();
+ if (metadata.getDataType() == FieldSpec.DataType.BYTES) {
+ // Logical BYTES dictionary entries are serialized UltraLogLog objects.
try {
QueryThreadContext.checkTerminationAndSampleUsage(explainPlanName);
UltraLogLog ull =
ObjectSerDeUtils.ULTRA_LOG_LOG_OBJECT_SER_DE.deserialize(dictionary.getBytesValue(0));
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 4fe96b819db..e71632673b2 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,9 @@ public class DistinctCountBitmapAggregationFunction extends
BaseSingleInputAggre
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet blockValSet = blockValSetMap.get(_expression);
- // Treat BYTES value as serialized RoaringBitmap
- DataType storedType = blockValSet.getValueType().getStoredType();
- if (storedType == DataType.BYTES) {
+ DataType dataType = blockValSet.getValueType();
+ if (dataType == DataType.BYTES) {
+ // Logical BYTES is a serialized RoaringBitmap and always uses the
single-value representation.
byte[][] bytesValues = blockValSet.getBytesValuesSV();
RoaringBitmap valueBitmap = aggregationResultHolder.getResult();
if (valueBitmap != null) {
@@ -90,6 +91,8 @@ public class DistinctCountBitmapAggregationFunction extends
BaseSingleInputAggre
return;
}
+ DataType storedType = dataType.getStoredType();
+
if (blockValSet.isSingleValue()) {
aggregateSV(length, aggregationResultHolder, blockValSet, storedType);
} else {
@@ -138,6 +141,12 @@ public class DistinctCountBitmapAggregationFunction
extends BaseSingleInputAggre
valueBitmap.add(stringValues[i].hashCode());
}
break;
+ case BYTES:
+ byte[][] bytesValues = blockValSet.getBytesValuesSV();
+ for (int i = 0; i < length; i++) {
+ valueBitmap.add(Arrays.hashCode(bytesValues[i]));
+ }
+ break;
default:
throw new IllegalStateException(
"Illegal data type for DISTINCT_COUNT_BITMAP aggregation function:
" + storedType);
@@ -198,6 +207,14 @@ public class DistinctCountBitmapAggregationFunction
extends BaseSingleInputAggre
}
}
break;
+ case BYTES:
+ byte[][][] bytesValues = blockValSet.getBytesValuesMV();
+ for (int i = 0; i < length; i++) {
+ for (byte[] value : bytesValues[i]) {
+ valueBitmap.add(Arrays.hashCode(value));
+ }
+ }
+ break;
default:
throw new IllegalStateException(
"Illegal data type for DISTINCT_COUNT_BITMAP aggregation function:
" + storedType);
@@ -209,9 +226,9 @@ public class DistinctCountBitmapAggregationFunction extends
BaseSingleInputAggre
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet blockValSet = blockValSetMap.get(_expression);
- // Treat BYTES value as serialized RoaringBitmap
- DataType storedType = blockValSet.getValueType().getStoredType();
- if (storedType == DataType.BYTES) {
+ DataType dataType = blockValSet.getValueType();
+ if (dataType == DataType.BYTES) {
+ // Logical BYTES is a serialized RoaringBitmap and always uses the
single-value representation.
byte[][] bytesValues = blockValSet.getBytesValuesSV();
for (int i = 0; i < length; i++) {
RoaringBitmap value = RoaringBitmapUtils.deserialize(bytesValues[i]);
@@ -226,6 +243,8 @@ public class DistinctCountBitmapAggregationFunction extends
BaseSingleInputAggre
return;
}
+ DataType storedType = dataType.getStoredType();
+
if (blockValSet.isSingleValue()) {
aggregateSVGroupBySV(length, groupKeyArray, groupByResultHolder,
blockValSet, storedType);
} else {
@@ -277,6 +296,12 @@ public class DistinctCountBitmapAggregationFunction
extends BaseSingleInputAggre
getValueBitmap(groupByResultHolder,
groupKeyArray[i]).add(stringValues[i].hashCode());
}
break;
+ case BYTES:
+ byte[][] bytesValues = blockValSet.getBytesValuesSV();
+ for (int i = 0; i < length; i++) {
+ getValueBitmap(groupByResultHolder,
groupKeyArray[i]).add(Arrays.hashCode(bytesValues[i]));
+ }
+ break;
default:
throw new IllegalStateException(
"Illegal data type for DISTINCT_COUNT_BITMAP aggregation function:
" + storedType);
@@ -339,6 +364,15 @@ public class DistinctCountBitmapAggregationFunction
extends BaseSingleInputAggre
}
}
break;
+ case BYTES:
+ byte[][][] bytesValues = blockValSet.getBytesValuesMV();
+ for (int i = 0; i < length; i++) {
+ RoaringBitmap bitmap = getValueBitmap(groupByResultHolder,
groupKeyArray[i]);
+ for (byte[] value : bytesValues[i]) {
+ bitmap.add(Arrays.hashCode(value));
+ }
+ }
+ break;
default:
throw new IllegalStateException(
"Illegal data type for DISTINCT_COUNT_BITMAP aggregation function:
" + storedType);
@@ -350,9 +384,9 @@ public class DistinctCountBitmapAggregationFunction extends
BaseSingleInputAggre
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet blockValSet = blockValSetMap.get(_expression);
- // Treat BYTES value as serialized RoaringBitmap
- DataType storedType = blockValSet.getValueType().getStoredType();
- if (storedType == DataType.BYTES) {
+ DataType dataType = blockValSet.getValueType();
+ if (dataType == DataType.BYTES) {
+ // Logical BYTES is a serialized RoaringBitmap and always uses the
single-value representation.
byte[][] bytesValues = blockValSet.getBytesValuesSV();
for (int i = 0; i < length; i++) {
RoaringBitmap value = RoaringBitmapUtils.deserialize(bytesValues[i]);
@@ -369,6 +403,8 @@ public class DistinctCountBitmapAggregationFunction extends
BaseSingleInputAggre
return;
}
+ DataType storedType = dataType.getStoredType();
+
if (blockValSet.isSingleValue()) {
aggregateSVGroupByMV(length, groupKeysArray, groupByResultHolder,
blockValSet, storedType);
} else {
@@ -420,6 +456,12 @@ public class DistinctCountBitmapAggregationFunction
extends BaseSingleInputAggre
setValueForGroupKeys(groupByResultHolder, groupKeysArray[i],
stringValues[i].hashCode());
}
break;
+ case BYTES:
+ byte[][] bytesValues = blockValSet.getBytesValuesSV();
+ for (int i = 0; i < length; i++) {
+ setValueForGroupKeys(groupByResultHolder, groupKeysArray[i],
Arrays.hashCode(bytesValues[i]));
+ }
+ break;
default:
throw new IllegalStateException(
"Illegal data type for DISTINCT_COUNT_BITMAP aggregation function:
" + storedType);
@@ -494,6 +536,17 @@ public class DistinctCountBitmapAggregationFunction
extends BaseSingleInputAggre
}
}
break;
+ case BYTES:
+ byte[][][] bytesValues = blockValSet.getBytesValuesMV();
+ for (int i = 0; i < length; i++) {
+ for (int groupKey : groupKeysArray[i]) {
+ RoaringBitmap bitmap = getValueBitmap(groupByResultHolder,
groupKey);
+ for (byte[] value : bytesValues[i]) {
+ bitmap.add(Arrays.hashCode(value));
+ }
+ }
+ }
+ break;
default:
throw new IllegalStateException(
"Illegal data type for DISTINCT_COUNT_BITMAP aggregation function:
" + storedType);
@@ -660,6 +713,11 @@ public class DistinctCountBitmapAggregationFunction
extends BaseSingleInputAggre
valueBitmap.add(dictionary.getStringValue(iterator.next()).hashCode());
}
break;
+ 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 ec273c066ab..8194222fb69 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,9 +135,9 @@ public class DistinctCountCPCSketchAggregationFunction
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet blockValSet = blockValSetMap.get(_expression);
- // Treat BYTES value as serialized CPC Sketch
- FieldSpec.DataType storedType = blockValSet.getValueType().getStoredType();
- if (storedType == DataType.BYTES) {
+ DataType dataType = blockValSet.getValueType();
+ if (dataType == DataType.BYTES) {
+ // Logical BYTES stores serialized CpcSketch objects in the single-value
representation.
byte[][] bytesValues = blockValSet.getBytesValuesSV();
try {
CpcSketchAccumulator cpcSketchAccumulator =
getAccumulator(aggregationResultHolder);
@@ -153,6 +153,16 @@ public class DistinctCountCPCSketchAggregationFunction
return;
}
+ DataType storedType = dataType.getStoredType();
+ if (blockValSet.isSingleValue()) {
+ aggregateSV(length, aggregationResultHolder, blockValSet, storedType);
+ } else {
+ aggregateMV(length, aggregationResultHolder, blockValSet, storedType);
+ }
+ }
+
+ protected void aggregateSV(int length, AggregationResultHolder
aggregationResultHolder, BlockValSet blockValSet,
+ DataType storedType) {
// For dictionary-encoded expression, store dictionary ids into the bitmap
Dictionary dictionary = blockValSet.isDictionaryEncoded() ?
blockValSet.getDictionary() : null;
if (dictionary != null) {
@@ -194,11 +204,44 @@ public class DistinctCountCPCSketchAggregationFunction
cpcSketch.update(stringValues[i]);
}
break;
+ case BYTES:
+ byte[][] bytesValues = blockValSet.getBytesValuesSV();
+ for (int i = 0; i < length; i++) {
+ cpcSketch.update(bytesValues[i]);
+ }
+ break;
+ default:
+ throw new IllegalStateException("Illegal data type for
DISTINCT_COUNT_CPC aggregation function: " + storedType);
+ }
+ }
+
+ protected void aggregateMV(int length, AggregationResultHolder
aggregationResultHolder, BlockValSet blockValSet,
+ DataType storedType) {
+ // For dictionary-encoded expression, store dictionary ids into the bitmap
+ Dictionary dictionary = blockValSet.isDictionaryEncoded() ?
blockValSet.getDictionary() : null;
+ if (dictionary != null) {
+ int[][] dictIds = blockValSet.getDictionaryIdsMV();
+ RoaringBitmap dictIdBitmap = getDictIdBitmap(aggregationResultHolder,
dictionary);
+ for (int i = 0; i < length; i++) {
+ dictIdBitmap.add(dictIds[i]);
+ }
+ return;
+ }
+
+ // For non-dictionary-encoded expression, store values into the CpcSketch
+ switch (storedType) {
+ case BYTES:
+ byte[][][] bytesValues = blockValSet.getBytesValuesMV();
+ CpcSketch cpcSketch = getCpcSketch(aggregationResultHolder);
+ for (int i = 0; i < length; i++) {
+ for (byte[] value : bytesValues[i]) {
+ cpcSketch.update(value);
+ }
+ }
+ break;
default:
throw new IllegalStateException("Illegal data type for
DISTINCT_COUNT_CPC aggregation function: " + storedType);
}
- CpcSketchAccumulator cpcSketchAccumulator =
getAccumulator(aggregationResultHolder);
- cpcSketchAccumulator.apply(cpcSketch);
}
@Override
@@ -206,9 +249,9 @@ public class DistinctCountCPCSketchAggregationFunction
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet blockValSet = blockValSetMap.get(_expression);
- // Treat BYTES value as serialized CPC Sketch
- DataType storedType = blockValSet.getValueType().getStoredType();
- if (storedType == FieldSpec.DataType.BYTES) {
+ DataType dataType = blockValSet.getValueType();
+ if (dataType == DataType.BYTES) {
+ // Logical BYTES stores serialized CpcSketch objects in the single-value
representation.
byte[][] bytesValues = blockValSet.getBytesValuesSV();
try {
CpcSketch[] sketches = deserializeSketches(bytesValues, length);
@@ -225,6 +268,16 @@ public class DistinctCountCPCSketchAggregationFunction
return;
}
+ DataType storedType = dataType.getStoredType();
+ if (blockValSet.isSingleValue()) {
+ aggregateSVGroupBySV(length, groupKeyArray, groupByResultHolder,
blockValSet, storedType);
+ } else {
+ aggregateMVGroupBySV(length, groupKeyArray, groupByResultHolder,
blockValSet, storedType);
+ }
+ }
+
+ protected void aggregateSVGroupBySV(int length, int[] groupKeyArray,
GroupByResultHolder groupByResultHolder,
+ BlockValSet blockValSet, DataType storedType) {
// For dictionary-encoded expression, store dictionary ids into the bitmap
Dictionary dictionary = blockValSet.isDictionaryEncoded() ?
blockValSet.getDictionary() : null;
if (dictionary != null) {
@@ -267,6 +320,40 @@ public class DistinctCountCPCSketchAggregationFunction
getCpcSketch(groupByResultHolder,
groupKeyArray[i]).update(stringValues[i]);
}
break;
+ case BYTES:
+ byte[][] bytesValues = blockValSet.getBytesValuesSV();
+ for (int i = 0; i < length; i++) {
+ getCpcSketch(groupByResultHolder,
groupKeyArray[i]).update(bytesValues[i]);
+ }
+ break;
+ default:
+ throw new IllegalStateException("Illegal data type for
DISTINCT_COUNT_CPC aggregation function: " + storedType);
+ }
+ }
+
+ protected void aggregateMVGroupBySV(int length, int[] groupKeyArray,
GroupByResultHolder groupByResultHolder,
+ BlockValSet blockValSet, DataType storedType) {
+ // For dictionary-encoded expression, store dictionary ids into the bitmap
+ Dictionary dictionary = blockValSet.isDictionaryEncoded() ?
blockValSet.getDictionary() : null;
+ if (dictionary != null) {
+ int[][] dictIds = blockValSet.getDictionaryIdsMV();
+ for (int i = 0; i < length; i++) {
+ getDictIdBitmap(groupByResultHolder, groupKeyArray[i],
dictionary).add(dictIds[i]);
+ }
+ return;
+ }
+
+ // For non-dictionary-encoded expression, store values into the CpcSketch
+ switch (storedType) {
+ case BYTES:
+ byte[][][] bytesValues = blockValSet.getBytesValuesMV();
+ for (int i = 0; i < length; i++) {
+ CpcSketch cpcSketch = getCpcSketch(groupByResultHolder,
groupKeyArray[i]);
+ for (byte[] value : bytesValues[i]) {
+ cpcSketch.update(value);
+ }
+ }
+ break;
default:
throw new IllegalStateException("Illegal data type for
DISTINCT_COUNT_CPC aggregation function: " + storedType);
}
@@ -277,11 +364,9 @@ public class DistinctCountCPCSketchAggregationFunction
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet blockValSet = blockValSetMap.get(_expression);
- // Treat BYTES value as serialized CPC Sketch
- DataType storedType = blockValSet.getValueType().getStoredType();
- boolean singleValue = blockValSet.isSingleValue();
-
- if (singleValue && storedType == DataType.BYTES) {
+ DataType dataType = blockValSet.getValueType();
+ if (dataType == DataType.BYTES) {
+ // Logical BYTES stores serialized CpcSketch objects in the single-value
representation.
byte[][] bytesValues = blockValSet.getBytesValuesSV();
try {
CpcSketch[] sketches = deserializeSketches(bytesValues, length);
@@ -298,6 +383,16 @@ public class DistinctCountCPCSketchAggregationFunction
return;
}
+ DataType storedType = dataType.getStoredType();
+ if (blockValSet.isSingleValue()) {
+ aggregateSVGroupByMV(length, groupKeysArray, groupByResultHolder,
blockValSet, storedType);
+ } else {
+ aggregateMVGroupByMV(length, groupKeysArray, groupByResultHolder,
blockValSet, storedType);
+ }
+ }
+
+ protected void aggregateSVGroupByMV(int length, int[][] groupKeysArray,
GroupByResultHolder groupByResultHolder,
+ BlockValSet blockValSet, DataType storedType) {
// For dictionary-encoded expression, store dictionary ids into the bitmap
Dictionary dictionary = blockValSet.isDictionaryEncoded() ?
blockValSet.getDictionary() : null;
if (dictionary != null) {
@@ -350,6 +445,47 @@ public class DistinctCountCPCSketchAggregationFunction
}
}
break;
+ case BYTES:
+ byte[][] bytesValues = blockValSet.getBytesValuesSV();
+ for (int i = 0; i < length; i++) {
+ for (int groupKey : groupKeysArray[i]) {
+ getCpcSketch(groupByResultHolder, groupKey).update(bytesValues[i]);
+ }
+ }
+ break;
+ default:
+ throw new IllegalStateException("Illegal data type for
DISTINCT_COUNT_CPC aggregation function: " + storedType);
+ }
+ }
+
+ protected void aggregateMVGroupByMV(int length, int[][] groupKeysArray,
GroupByResultHolder groupByResultHolder,
+ BlockValSet blockValSet, DataType storedType) {
+ // For dictionary-encoded expression, store dictionary ids into the bitmap
+ Dictionary dictionary = blockValSet.isDictionaryEncoded() ?
blockValSet.getDictionary() : null;
+ if (dictionary != null) {
+ int[][] dictIds = blockValSet.getDictionaryIdsMV();
+ for (int i = 0; i < length; i++) {
+ int[] rowDictIds = dictIds[i];
+ for (int groupKey : groupKeysArray[i]) {
+ getDictIdBitmap(groupByResultHolder, groupKey,
dictionary).add(rowDictIds);
+ }
+ }
+ return;
+ }
+
+ // For non-dictionary-encoded expression, store values into the CpcSketch
+ switch (storedType) {
+ case BYTES:
+ byte[][][] bytesValues = blockValSet.getBytesValuesMV();
+ for (int i = 0; i < length; i++) {
+ for (int groupKey : groupKeysArray[i]) {
+ CpcSketch cpcSketch = getCpcSketch(groupByResultHolder, groupKey);
+ for (byte[] value : bytesValues[i]) {
+ cpcSketch.update(value);
+ }
+ }
+ }
+ break;
default:
throw new IllegalStateException("Illegal data type for
DISTINCT_COUNT_CPC aggregation function: " + storedType);
}
@@ -524,6 +660,8 @@ public class DistinctCountCPCSketchAggregationFunction
private void addObjectToSketch(Object rawValue, CpcSketch sketch) {
if (rawValue instanceof String) {
sketch.update((String) rawValue);
+ } else if (rawValue instanceof byte[]) {
+ sketch.update((byte[]) rawValue);
} else if (rawValue instanceof Integer) {
sketch.update((Integer) rawValue);
} else if (rawValue instanceof Long) {
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 07bfb6bd41a..1d68cde5958 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,9 @@ public class DistinctCountHLLAggregationFunction extends
BaseSingleInputAggregat
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet blockValSet = blockValSetMap.get(_expression);
- // Treat BYTES value as serialized HyperLogLog
- DataType storedType = blockValSet.getValueType().getStoredType();
- if (storedType == DataType.BYTES) {
+ DataType dataType = blockValSet.getValueType();
+ if (dataType == DataType.BYTES) {
+ // Logical BYTES is a serialized HyperLogLog and always uses the
single-value representation.
byte[][] bytesValues = blockValSet.getBytesValuesSV();
try {
HyperLogLog hyperLogLog = aggregationResultHolder.getResult();
@@ -104,6 +104,8 @@ public class DistinctCountHLLAggregationFunction extends
BaseSingleInputAggregat
return;
}
+ DataType storedType = dataType.getStoredType();
+
if (blockValSet.isSingleValue()) {
aggregateSV(length, aggregationResultHolder, blockValSet, storedType);
} else {
@@ -159,6 +161,12 @@ public class DistinctCountHLLAggregationFunction extends
BaseSingleInputAggregat
hyperLogLog.offer(stringValues[i]);
}
break;
+ case BYTES:
+ byte[][] bytesValues = blockValSet.getBytesValuesSV();
+ for (int i = 0; i < length; i++) {
+ hyperLogLog.offer(bytesValues[i]);
+ }
+ break;
default:
throw new IllegalStateException("Illegal data type for
DISTINCT_COUNT_HLL aggregation function: " + storedType);
}
@@ -222,6 +230,14 @@ public class DistinctCountHLLAggregationFunction extends
BaseSingleInputAggregat
}
}
break;
+ case BYTES:
+ byte[][][] bytesValuesArray = blockValSet.getBytesValuesMV();
+ for (int i = 0; i < length; i++) {
+ for (byte[] value : bytesValuesArray[i]) {
+ hyperLogLog.offer(value);
+ }
+ }
+ break;
default:
throw new IllegalStateException("Illegal data type for
DISTINCT_COUNT_HLL aggregation function: " + storedType);
}
@@ -232,9 +248,9 @@ public class DistinctCountHLLAggregationFunction extends
BaseSingleInputAggregat
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet blockValSet = blockValSetMap.get(_expression);
- // Treat BYTES value as serialized HyperLogLog
- DataType storedType = blockValSet.getValueType().getStoredType();
- if (storedType == DataType.BYTES) {
+ DataType dataType = blockValSet.getValueType();
+ if (dataType == DataType.BYTES) {
+ // Logical BYTES is a serialized HyperLogLog and always uses the
single-value representation.
byte[][] bytesValues = blockValSet.getBytesValuesSV();
try {
for (int i = 0; i < length; i++) {
@@ -253,6 +269,8 @@ public class DistinctCountHLLAggregationFunction extends
BaseSingleInputAggregat
return;
}
+ DataType storedType = dataType.getStoredType();
+
if (blockValSet.isSingleValue()) {
aggregateSVGroupBySV(length, groupKeyArray, groupByResultHolder,
blockValSet, storedType);
} else {
@@ -307,6 +325,12 @@ public class DistinctCountHLLAggregationFunction extends
BaseSingleInputAggregat
getHyperLogLog(groupByResultHolder,
groupKeyArray[i]).offer(stringValues[i]);
}
break;
+ case BYTES:
+ byte[][] bytesValues = blockValSet.getBytesValuesSV();
+ for (int i = 0; i < length; i++) {
+ getHyperLogLog(groupByResultHolder,
groupKeyArray[i]).offer(bytesValues[i]);
+ }
+ break;
default:
throw new IllegalStateException("Illegal data type for
DISTINCT_COUNT_HLL aggregation function: " + storedType);
}
@@ -371,6 +395,15 @@ public class DistinctCountHLLAggregationFunction extends
BaseSingleInputAggregat
}
}
break;
+ case BYTES:
+ byte[][][] bytesValuesArray = blockValSet.getBytesValuesMV();
+ for (int i = 0; i < length; i++) {
+ HyperLogLog hyperLogLog = getHyperLogLog(groupByResultHolder,
groupKeyArray[i]);
+ for (byte[] value : bytesValuesArray[i]) {
+ hyperLogLog.offer(value);
+ }
+ }
+ break;
default:
throw new IllegalStateException("Illegal data type for
DISTINCT_COUNT_HLL aggregation function: " + storedType);
}
@@ -381,9 +414,9 @@ public class DistinctCountHLLAggregationFunction extends
BaseSingleInputAggregat
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet blockValSet = blockValSetMap.get(_expression);
- // Treat BYTES value as serialized HyperLogLog
- DataType storedType = blockValSet.getValueType().getStoredType();
- if (storedType == DataType.BYTES) {
+ DataType dataType = blockValSet.getValueType();
+ if (dataType == DataType.BYTES) {
+ // Logical BYTES is a serialized HyperLogLog and always uses the
single-value representation.
byte[][] bytesValues = blockValSet.getBytesValuesSV();
try {
for (int i = 0; i < length; i++) {
@@ -405,6 +438,8 @@ public class DistinctCountHLLAggregationFunction extends
BaseSingleInputAggregat
return;
}
+ DataType storedType = dataType.getStoredType();
+
if (blockValSet.isSingleValue()) {
aggregateSVGroupByMV(length, groupKeysArray, groupByResultHolder,
blockValSet, storedType);
} else {
@@ -459,6 +494,12 @@ public class DistinctCountHLLAggregationFunction extends
BaseSingleInputAggregat
setValueForGroupKeys(groupByResultHolder, groupKeysArray[i],
stringValues[i]);
}
break;
+ case BYTES:
+ byte[][] bytesValues = blockValSet.getBytesValuesSV();
+ for (int i = 0; i < length; i++) {
+ setValueForGroupKeys(groupByResultHolder, groupKeysArray[i],
bytesValues[i]);
+ }
+ break;
default:
throw new IllegalStateException("Illegal data type for
DISTINCT_COUNT_HLL aggregation function: " + storedType);
}
@@ -541,6 +582,18 @@ public class DistinctCountHLLAggregationFunction extends
BaseSingleInputAggregat
}
}
break;
+ case BYTES:
+ byte[][][] bytesValuesArray = blockValSet.getBytesValuesMV();
+ for (int i = 0; i < length; i++) {
+ byte[][] bytesValues = bytesValuesArray[i];
+ for (int groupKey : groupKeysArray[i]) {
+ HyperLogLog hyperLogLog = getHyperLogLog(groupByResultHolder,
groupKey);
+ for (byte[] value : bytesValues) {
+ 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 fd337f433a3..dfda4ed0bc7 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,9 +93,9 @@ public class DistinctCountHLLPlusAggregationFunction extends
BaseSingleInputAggr
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet blockValSet = blockValSetMap.get(_expression);
- // Treat BYTES value as serialized HyperLogLogPlus
- DataType storedType = blockValSet.getValueType().getStoredType();
- if (storedType == DataType.BYTES) {
+ DataType dataType = blockValSet.getValueType();
+ if (dataType == DataType.BYTES) {
+ // Logical BYTES is a serialized HyperLogLogPlus and always uses the
single-value representation.
byte[][] bytesValues = blockValSet.getBytesValuesSV();
try {
HyperLogLogPlus hyperLogLogPlus = aggregationResultHolder.getResult();
@@ -114,6 +114,8 @@ public class DistinctCountHLLPlusAggregationFunction
extends BaseSingleInputAggr
return;
}
+ DataType storedType = dataType.getStoredType();
+
if (blockValSet.isSingleValue()) {
aggregateSV(length, aggregationResultHolder, blockValSet, storedType);
} else {
@@ -164,6 +166,12 @@ public class DistinctCountHLLPlusAggregationFunction
extends BaseSingleInputAggr
hyperLogLogPlus.offer(stringValues[i]);
}
break;
+ case BYTES:
+ byte[][] bytesValues = blockValSet.getBytesValuesSV();
+ for (int i = 0; i < length; i++) {
+ hyperLogLogPlus.offer(bytesValues[i]);
+ }
+ break;
default:
throw new IllegalStateException(
"Illegal data type for DISTINCT_COUNT_HLL_PLUS aggregation
function: " + storedType);
@@ -226,6 +234,14 @@ public class DistinctCountHLLPlusAggregationFunction
extends BaseSingleInputAggr
}
}
break;
+ case BYTES:
+ byte[][][] bytesValuesArray = blockValSet.getBytesValuesMV();
+ for (int i = 0; i < length; i++) {
+ for (byte[] value : bytesValuesArray[i]) {
+ hyperLogLogPlus.offer(value);
+ }
+ }
+ break;
default:
throw new IllegalStateException(
"Illegal data type for DISTINCT_COUNT_HLL_PLUS aggregation
function: " + storedType);
@@ -237,9 +253,9 @@ public class DistinctCountHLLPlusAggregationFunction
extends BaseSingleInputAggr
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet blockValSet = blockValSetMap.get(_expression);
- // Treat BYTES value as serialized HyperLogLogPlus
- DataType storedType = blockValSet.getValueType().getStoredType();
- if (storedType == DataType.BYTES) {
+ DataType dataType = blockValSet.getValueType();
+ if (dataType == DataType.BYTES) {
+ // Logical BYTES is a serialized HyperLogLogPlus and always uses the
single-value representation.
byte[][] bytesValues = blockValSet.getBytesValuesSV();
try {
for (int i = 0; i < length; i++) {
@@ -258,6 +274,8 @@ public class DistinctCountHLLPlusAggregationFunction
extends BaseSingleInputAggr
return;
}
+ DataType storedType = dataType.getStoredType();
+
if (blockValSet.isSingleValue()) {
aggregateSVGroupBySV(length, groupKeyArray, groupByResultHolder,
blockValSet, storedType);
} else {
@@ -309,6 +327,12 @@ public class DistinctCountHLLPlusAggregationFunction
extends BaseSingleInputAggr
getHyperLogLogPlus(groupByResultHolder,
groupKeyArray[i]).offer(stringValues[i]);
}
break;
+ case BYTES:
+ byte[][] bytesValues = blockValSet.getBytesValuesSV();
+ for (int i = 0; i < length; i++) {
+ getHyperLogLogPlus(groupByResultHolder,
groupKeyArray[i]).offer(bytesValues[i]);
+ }
+ break;
default:
throw new IllegalStateException(
"Illegal data type for DISTINCT_COUNT_HLL_PLUS aggregation
function: " + storedType);
@@ -374,6 +398,15 @@ public class DistinctCountHLLPlusAggregationFunction
extends BaseSingleInputAggr
}
}
break;
+ case BYTES:
+ byte[][][] bytesValuesArray = blockValSet.getBytesValuesMV();
+ for (int i = 0; i < length; i++) {
+ HyperLogLogPlus hyperLogLogPlus =
getHyperLogLogPlus(groupByResultHolder, groupKeyArray[i]);
+ for (byte[] value : bytesValuesArray[i]) {
+ hyperLogLogPlus.offer(value);
+ }
+ }
+ break;
default:
throw new IllegalStateException(
"Illegal data type for DISTINCT_COUNT_HLL_PLUS aggregation
function: " + storedType);
@@ -385,9 +418,9 @@ public class DistinctCountHLLPlusAggregationFunction
extends BaseSingleInputAggr
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet blockValSet = blockValSetMap.get(_expression);
- // Treat BYTES value as serialized HyperLogLogPlus
- DataType storedType = blockValSet.getValueType().getStoredType();
- if (storedType == DataType.BYTES) {
+ DataType dataType = blockValSet.getValueType();
+ if (dataType == DataType.BYTES) {
+ // Logical BYTES is a serialized HyperLogLogPlus and always uses the
single-value representation.
byte[][] bytesValues = blockValSet.getBytesValuesSV();
try {
for (int i = 0; i < length; i++) {
@@ -409,6 +442,8 @@ public class DistinctCountHLLPlusAggregationFunction
extends BaseSingleInputAggr
return;
}
+ DataType storedType = dataType.getStoredType();
+
if (blockValSet.isSingleValue()) {
aggregateSVGroupByMV(length, groupKeysArray, groupByResultHolder,
blockValSet, storedType);
} else {
@@ -460,6 +495,12 @@ public class DistinctCountHLLPlusAggregationFunction
extends BaseSingleInputAggr
setValueForGroupKeys(groupByResultHolder, groupKeysArray[i],
stringValues[i]);
}
break;
+ case BYTES:
+ byte[][] bytesValues = blockValSet.getBytesValuesSV();
+ for (int i = 0; i < length; i++) {
+ setValueForGroupKeys(groupByResultHolder, groupKeysArray[i],
bytesValues[i]);
+ }
+ break;
default:
throw new IllegalStateException(
"Illegal data type for DISTINCT_COUNT_HLL_PLUS aggregation
function: " + storedType);
@@ -542,6 +583,18 @@ public class DistinctCountHLLPlusAggregationFunction
extends BaseSingleInputAggr
}
}
break;
+ case BYTES:
+ byte[][][] bytesValuesArray = blockValSet.getBytesValuesMV();
+ for (int i = 0; i < length; i++) {
+ byte[][] bytesValues = bytesValuesArray[i];
+ for (int groupKey : groupKeysArray[i]) {
+ HyperLogLogPlus hyperLogLogPlus =
getHyperLogLogPlus(groupByResultHolder, groupKey);
+ for (byte[] value : bytesValues) {
+ hyperLogLogPlus.offer(value);
+ }
+ }
+ }
+ break;
default:
throw new IllegalStateException(
"Illegal data type for DISTINCT_COUNT_HLL_PLUS aggregation
function: " + storedType);
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 a2859fe471e..9118d766124 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
@@ -196,7 +196,7 @@ public class DistinctCountThetaSketchAggregationFunction
if (valueTypes[0] != DataType.BYTES) {
List<UpdatableThetaSketch> updateSketches =
getUpdateSketches(aggregationResultHolder);
if (singleValues[0]) {
- switch (valueTypes[0]) {
+ switch (valueTypes[0].getStoredType()) {
case INT:
int[] intValues = (int[]) valueArrays[0];
if (_includeDefaultSketch) {
@@ -287,13 +287,31 @@ public class DistinctCountThetaSketchAggregationFunction
}
}
break;
+ case BYTES:
+ byte[][] bytesValues = (byte[][]) valueArrays[0];
+ if (_includeDefaultSketch) {
+ UpdatableThetaSketch defaultSketch = updateSketches.get(0);
+ for (int i = 0; i < length; i++) {
+ defaultSketch.update(bytesValues[i]);
+ }
+ }
+ for (int i = 0; i < numFilters; i++) {
+ FilterEvaluator filterEvaluator = _filterEvaluators.get(i);
+ UpdatableThetaSketch updateSketch = updateSketches.get(i + 1);
+ for (int j = 0; j < length; j++) {
+ if (filterEvaluator.evaluate(singleValues, valueTypes,
valueArrays, j)) {
+ updateSketch.update(bytesValues[j]);
+ }
+ }
+ }
+ break;
default:
throw new IllegalStateException(
"Illegal single-value data type for
DISTINCT_COUNT_THETA_SKETCH aggregation function: "
+ valueTypes[0]);
}
} else {
- switch (valueTypes[0]) {
+ switch (valueTypes[0].getStoredType()) {
case INT:
int[][] intValues = (int[][]) valueArrays[0];
if (_includeDefaultSketch) {
@@ -404,13 +422,35 @@ public class DistinctCountThetaSketchAggregationFunction
}
}
break;
+ case BYTES:
+ byte[][][] bytesValues = (byte[][][]) valueArrays[0];
+ if (_includeDefaultSketch) {
+ UpdatableThetaSketch defaultSketch = updateSketches.get(0);
+ for (int i = 0; i < length; i++) {
+ for (byte[] value : bytesValues[i]) {
+ defaultSketch.update(value);
+ }
+ }
+ }
+ for (int i = 0; i < numFilters; i++) {
+ FilterEvaluator filterEvaluator = _filterEvaluators.get(i);
+ UpdatableThetaSketch updateSketch = updateSketches.get(i + 1);
+ for (int j = 0; j < length; j++) {
+ if (filterEvaluator.evaluate(singleValues, valueTypes,
valueArrays, j)) {
+ for (byte[] value : bytesValues[j]) {
+ updateSketch.update(value);
+ }
+ }
+ }
+ }
+ break;
default:
throw new IllegalStateException(
"Illegal multi-value data type for DISTINCT_COUNT_THETA_SKETCH
aggregation function: " + valueTypes[0]);
}
}
} else {
- // Serialized sketch
+ // Logical BYTES stores serialized ThetaSketch objects in the
single-value representation.
List<ThetaSketchAccumulator> thetaSketchAccumulators =
getUnions(aggregationResultHolder);
ThetaSketch[] sketches = deserializeSketches((byte[][]) valueArrays[0],
length);
if (_includeDefaultSketch) {
@@ -444,7 +484,7 @@ public class DistinctCountThetaSketchAggregationFunction
// Main expression is always index 0
if (valueTypes[0] != DataType.BYTES) {
if (singleValues[0]) {
- switch (valueTypes[0]) {
+ switch (valueTypes[0].getStoredType()) {
case INT:
int[] intValues = (int[]) valueArrays[0];
for (int i = 0; i < length; i++) {
@@ -520,13 +560,29 @@ public class DistinctCountThetaSketchAggregationFunction
}
}
break;
+ case BYTES:
+ byte[][] bytesValues = (byte[][]) valueArrays[0];
+ for (int i = 0; i < length; i++) {
+ List<UpdatableThetaSketch> updateSketches =
+ getUpdateSketches(groupByResultHolder, groupKeyArray[i]);
+ byte[] value = bytesValues[i];
+ if (_includeDefaultSketch) {
+ updateSketches.get(0).update(value);
+ }
+ for (int j = 0; j < numFilters; j++) {
+ if (_filterEvaluators.get(j).evaluate(singleValues,
valueTypes, valueArrays, i)) {
+ updateSketches.get(j + 1).update(value);
+ }
+ }
+ }
+ break;
default:
throw new IllegalStateException(
"Illegal single-value data type for
DISTINCT_COUNT_THETA_SKETCH aggregation function: "
+ valueTypes[0]);
}
} else {
- switch (valueTypes[0]) {
+ switch (valueTypes[0].getStoredType()) {
case INT:
int[][] intValues = (int[][]) valueArrays[0];
for (int i = 0; i < length; i++) {
@@ -632,13 +688,35 @@ public class DistinctCountThetaSketchAggregationFunction
}
}
break;
+ case BYTES:
+ byte[][][] bytesValues = (byte[][][]) valueArrays[0];
+ for (int i = 0; i < length; i++) {
+ List<UpdatableThetaSketch> updateSketches =
+ getUpdateSketches(groupByResultHolder, groupKeyArray[i]);
+ byte[][] values = bytesValues[i];
+ if (_includeDefaultSketch) {
+ UpdatableThetaSketch defaultSketch = updateSketches.get(0);
+ for (byte[] value : values) {
+ defaultSketch.update(value);
+ }
+ }
+ for (int j = 0; j < numFilters; j++) {
+ if (_filterEvaluators.get(j).evaluate(singleValues,
valueTypes, valueArrays, i)) {
+ UpdatableThetaSketch updateSketch = updateSketches.get(j +
1);
+ for (byte[] value : values) {
+ updateSketch.update(value);
+ }
+ }
+ }
+ }
+ break;
default:
throw new IllegalStateException(
"Illegal multi-value data type for DISTINCT_COUNT_THETA_SKETCH
aggregation function: " + valueTypes[0]);
}
}
} else {
- // Serialized sketch
+ // Logical BYTES stores serialized ThetaSketch objects in the
single-value representation.
ThetaSketch[] sketches = deserializeSketches((byte[][]) valueArrays[0],
length);
for (int i = 0; i < length; i++) {
List<ThetaSketchAccumulator> thetaSketchAccumulators =
getUnions(groupByResultHolder, groupKeyArray[i]);
@@ -668,7 +746,7 @@ public class DistinctCountThetaSketchAggregationFunction
// Main expression is always index 0
if (valueTypes[0] != DataType.BYTES) {
if (singleValues[0]) {
- switch (valueTypes[0]) {
+ switch (valueTypes[0].getStoredType()) {
case INT:
int[] intValues = (int[]) valueArrays[0];
if (_includeDefaultSketch) {
@@ -769,13 +847,33 @@ public class DistinctCountThetaSketchAggregationFunction
}
}
break;
+ case BYTES:
+ byte[][] bytesValues = (byte[][]) valueArrays[0];
+ if (_includeDefaultSketch) {
+ for (int i = 0; i < length; i++) {
+ for (int groupKey : groupKeysArray[i]) {
+ getUpdateSketches(groupByResultHolder,
groupKey).get(0).update(bytesValues[i]);
+ }
+ }
+ }
+ for (int i = 0; i < numFilters; i++) {
+ FilterEvaluator filterEvaluator = _filterEvaluators.get(i);
+ for (int j = 0; j < length; j++) {
+ if (filterEvaluator.evaluate(singleValues, valueTypes,
valueArrays, j)) {
+ for (int groupKey : groupKeysArray[j]) {
+ getUpdateSketches(groupByResultHolder, groupKey).get(i +
1).update(bytesValues[j]);
+ }
+ }
+ }
+ }
+ break;
default:
throw new IllegalStateException(
"Illegal single-value data type for
DISTINCT_COUNT_THETA_SKETCH aggregation function: "
+ valueTypes[0]);
}
} else {
- switch (valueTypes[0]) {
+ switch (valueTypes[0].getStoredType()) {
case INT:
int[][] intValues = (int[][]) valueArrays[0];
if (_includeDefaultSketch) {
@@ -906,13 +1004,40 @@ public class DistinctCountThetaSketchAggregationFunction
}
}
break;
+ case BYTES:
+ byte[][][] bytesValues = (byte[][][]) valueArrays[0];
+ if (_includeDefaultSketch) {
+ for (int i = 0; i < length; i++) {
+ for (int groupKey : groupKeysArray[i]) {
+ UpdatableThetaSketch defaultSketch =
getUpdateSketches(groupByResultHolder, groupKey).get(0);
+ for (byte[] value : bytesValues[i]) {
+ defaultSketch.update(value);
+ }
+ }
+ }
+ }
+ for (int i = 0; i < numFilters; i++) {
+ FilterEvaluator filterEvaluator = _filterEvaluators.get(i);
+ for (int j = 0; j < length; j++) {
+ if (filterEvaluator.evaluate(singleValues, valueTypes,
valueArrays, j)) {
+ for (int groupKey : groupKeysArray[j]) {
+ UpdatableThetaSketch updateSketch =
+ getUpdateSketches(groupByResultHolder, groupKey).get(i
+ 1);
+ for (byte[] value : bytesValues[j]) {
+ updateSketch.update(value);
+ }
+ }
+ }
+ }
+ }
+ break;
default:
throw new IllegalStateException(
"Illegal multi-value data type for DISTINCT_COUNT_THETA_SKETCH
aggregation function: " + valueTypes[0]);
}
}
} else {
- // Serialized sketch
+ // Logical BYTES stores serialized ThetaSketch objects in the
single-value representation.
ThetaSketch[] sketches = deserializeSketches((byte[][]) valueArrays[0],
length);
if (_includeDefaultSketch) {
for (int i = 0; i < length; i++) {
@@ -1232,9 +1357,10 @@ public class DistinctCountThetaSketchAggregationFunction
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;
- valueTypes[i] = storedType;
+ valueTypes[i] = dataType;
if (singleValue) {
switch (storedType) {
case INT:
@@ -1275,6 +1401,9 @@ public class DistinctCountThetaSketchAggregationFunction
case STRING:
valueArrays[i] = blockValSet.getStringValuesMV();
break;
+ case BYTES:
+ valueArrays[i] = blockValSet.getBytesValuesMV();
+ break;
default:
throw new IllegalStateException();
}
@@ -1513,7 +1642,7 @@ public class DistinctCountThetaSketchAggregationFunction
_predicateEvaluator =
PredicateEvaluatorProvider.getPredicateEvaluator(_predicate, null, valueType,
null);
}
if (singleValue) {
- switch (valueType) {
+ switch (valueType.getStoredType()) {
case INT:
return _predicateEvaluator.applySV(((int[]) valueArray)[index]);
case LONG:
@@ -1530,7 +1659,7 @@ public class DistinctCountThetaSketchAggregationFunction
throw new IllegalStateException();
}
} else {
- switch (valueType) {
+ switch (valueType.getStoredType()) {
case INT:
int[] intValues = ((int[][]) valueArray)[index];
return _predicateEvaluator.applyMV(intValues, intValues.length);
@@ -1546,6 +1675,9 @@ public class DistinctCountThetaSketchAggregationFunction
case STRING:
String[] stringValues = ((String[][]) valueArray)[index];
return _predicateEvaluator.applyMV(stringValues,
stringValues.length);
+ case BYTES:
+ byte[][] bytesValues = ((byte[][][]) valueArray)[index];
+ return _predicateEvaluator.applyMV(bytesValues,
bytesValues.length);
default:
throw new IllegalStateException();
}
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 5309a9713af..7ef3142ddb5 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,9 +82,9 @@ public class DistinctCountULLAggregationFunction extends
BaseSingleInputAggregat
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet blockValSet = blockValSetMap.get(_expression);
- // Treat BYTES value as serialized HyperLogLogPlus
- DataType storedType = blockValSet.getValueType().getStoredType();
- if (storedType == DataType.BYTES) {
+ DataType dataType = blockValSet.getValueType();
+ if (dataType == DataType.BYTES) {
+ // Logical BYTES is a serialized UltraLogLog and always uses the
single-value representation.
byte[][] bytesValues = blockValSet.getBytesValuesSV();
try {
UltraLogLog ull = aggregationResultHolder.getResult();
@@ -103,6 +103,8 @@ public class DistinctCountULLAggregationFunction extends
BaseSingleInputAggregat
return;
}
+ DataType storedType = dataType.getStoredType();
+
// For dictionary-encoded expression, store dictionary ids into the bitmap
Dictionary dictionary = blockValSet.isDictionaryEncoded() ?
blockValSet.getDictionary() : null;
if (dictionary != null) {
@@ -144,6 +146,12 @@ public class DistinctCountULLAggregationFunction extends
BaseSingleInputAggregat
UltraLogLogUtils.hashObject(stringValues[i]).ifPresent(ull::add);
}
break;
+ case BYTES:
+ byte[][] bytesValues = blockValSet.getBytesValuesSV();
+ for (int i = 0; i < length; i++) {
+ UltraLogLogUtils.hashObject(bytesValues[i]).ifPresent(ull::add);
+ }
+ break;
default:
throw new IllegalStateException(
"Illegal data type for DISTINCT_COUNT_ULL aggregation function: "
+ storedType);
@@ -155,9 +163,9 @@ public class DistinctCountULLAggregationFunction extends
BaseSingleInputAggregat
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet blockValSet = blockValSetMap.get(_expression);
- // Treat BYTES value as serialized UltraLogLogs
- DataType storedType = blockValSet.getValueType().getStoredType();
- if (storedType == DataType.BYTES) {
+ DataType dataType = blockValSet.getValueType();
+ if (dataType == DataType.BYTES) {
+ // Logical BYTES is a serialized UltraLogLog and always uses the
single-value representation.
byte[][] bytesValues = blockValSet.getBytesValuesSV();
try {
for (int i = 0; i < length; i++) {
@@ -176,6 +184,8 @@ public class DistinctCountULLAggregationFunction extends
BaseSingleInputAggregat
return;
}
+ DataType storedType = dataType.getStoredType();
+
// For dictionary-encoded expression, store dictionary ids into the bitmap
Dictionary dictionary = blockValSet.isDictionaryEncoded() ?
blockValSet.getDictionary() : null;
if (dictionary != null) {
@@ -223,6 +233,13 @@ public class DistinctCountULLAggregationFunction extends
BaseSingleInputAggregat
.ifPresent(getULL(groupByResultHolder, groupKeyArray[i])::add);
}
break;
+ case BYTES:
+ byte[][] bytesValues = blockValSet.getBytesValuesSV();
+ for (int i = 0; i < length; i++) {
+ UltraLogLogUtils.hashObject(bytesValues[i])
+ .ifPresent(getULL(groupByResultHolder, groupKeyArray[i])::add);
+ }
+ break;
default:
throw new IllegalStateException(
"Illegal data type for DISTINCT_COUNT_ULL aggregation function: "
+ storedType);
@@ -234,9 +251,9 @@ public class DistinctCountULLAggregationFunction extends
BaseSingleInputAggregat
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet blockValSet = blockValSetMap.get(_expression);
- // Treat BYTES value as serialized HyperLogLogPlus
- DataType storedType = blockValSet.getValueType().getStoredType();
- if (storedType == DataType.BYTES) {
+ DataType dataType = blockValSet.getValueType();
+ if (dataType == DataType.BYTES) {
+ // Logical BYTES is a serialized UltraLogLog and always uses the
single-value representation.
byte[][] bytesValues = blockValSet.getBytesValuesSV();
try {
for (int i = 0; i < length; i++) {
@@ -246,7 +263,6 @@ public class DistinctCountULLAggregationFunction extends
BaseSingleInputAggregat
if (ull != null) {
ull.add(value);
} else {
- // Create a new HyperLogLogPlus for the group
groupByResultHolder.setValueForKey(groupKey,
ObjectSerDeUtils.ULTRA_LOG_LOG_OBJECT_SER_DE.deserialize(bytesValues[i]));
}
@@ -258,6 +274,8 @@ public class DistinctCountULLAggregationFunction extends
BaseSingleInputAggregat
return;
}
+ DataType storedType = dataType.getStoredType();
+
// For dictionary-encoded expression, store dictionary ids into the bitmap
Dictionary dictionary = blockValSet.isDictionaryEncoded() ?
blockValSet.getDictionary() : null;
if (dictionary != null) {
@@ -300,6 +318,12 @@ public class DistinctCountULLAggregationFunction extends
BaseSingleInputAggregat
setValueForGroupKeys(groupByResultHolder, groupKeysArray[i],
stringValues[i]);
}
break;
+ case BYTES:
+ byte[][] bytesValues = blockValSet.getBytesValuesSV();
+ for (int i = 0; i < length; i++) {
+ setValueForGroupKeys(groupByResultHolder, groupKeysArray[i],
bytesValues[i]);
+ }
+ break;
default:
throw new IllegalStateException(
"Illegal data type for DISTINCT_COUNT_HLL_PLUS aggregation
function: " + storedType);
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 386648b64f5..5a2565e49f6 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,11 +32,13 @@ import org.apache.pinot.common.datatable.DataTable;
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;
import org.apache.pinot.spi.utils.ByteArray;
import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.UuidUtils;
import org.roaringbitmap.RoaringBitmap;
@@ -290,9 +292,16 @@ public class BytesDistinctTable extends DistinctTable {
return new ResultTable(_dataSchema, rows);
}
- private static void addRows(ByteArray[] values, int length, List<Object[]>
rows) {
- for (int i = 0; i < length; i++) {
- rows.add(new Object[]{values[i].toHexString()});
+ private void addRows(ByteArray[] values, int length, List<Object[]> rows) {
+ ColumnDataType columnDataType = _dataSchema.getColumnDataType(0);
+ if (columnDataType == ColumnDataType.UUID) {
+ for (int i = 0; i < length; i++) {
+ rows.add(new Object[]{UuidUtils.toString(values[i])});
+ }
+ } else {
+ for (int i = 0; i < length; i++) {
+ rows.add(new Object[]{values[i].toHexString()});
+ }
}
}
@@ -311,9 +320,16 @@ public class BytesDistinctTable extends DistinctTable {
return new ResultTable(_dataSchema, rows);
}
- private static void addRows(HashSet<ByteArray> values, List<Object[]> rows) {
- for (ByteArray value : values) {
- rows.add(new Object[]{value.toHexString()});
+ private void addRows(HashSet<ByteArray> values, List<Object[]> rows) {
+ ColumnDataType columnDataType = _dataSchema.getColumnDataType(0);
+ if (columnDataType == ColumnDataType.UUID) {
+ for (ByteArray value : values) {
+ rows.add(new Object[]{UuidUtils.toString(value)});
+ }
+ } else {
+ for (ByteArray value : values) {
+ rows.add(new Object[]{value.toHexString()});
+ }
}
}
}
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 c1add8ca708..3c36ebdf920 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
@@ -63,6 +63,7 @@ import org.apache.pinot.core.util.trace.TraceRunnable;
import org.apache.pinot.spi.exception.EarlyTerminationException;
import org.apache.pinot.spi.exception.QueryErrorCode;
import org.apache.pinot.spi.query.QueryThreadContext;
+import org.apache.pinot.spi.utils.UuidUtils;
import org.roaringbitmap.RoaringBitmap;
@@ -531,6 +532,8 @@ public class GroupByDataTableReducer implements
DataTableReducer {
return dataTable.getString(rowId, colId);
case BYTES:
return dataTable.getBytes(rowId, colId).getBytes();
+ case UUID:
+ return UuidUtils.toUUID(dataTable.getBytes(rowId, colId));
default:
throw new IllegalStateException("Illegal column data type in group
key: " + columnDataType);
}
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 00000000000..1b828a49998
--- /dev/null
+++
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/UuidAggregationTest.java
@@ -0,0 +1,212 @@
+/**
+ * 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.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 UUID aggregation coverage over dictionary-encoded and raw SV/MV
columns.
+@Test(suiteName = "CustomClusterIntegrationTest")
+public class UuidAggregationTest extends CustomDataQueryClusterIntegrationTest
{
+ private static final String TABLE_NAME = "UuidAggregationTest";
+ private static final String UUID_DICT_SV_COLUMN = "uuidDictSv";
+ private static final String UUID_DICT_MV_COLUMN = "uuidDictMv";
+ private static final String UUID_RAW_SV_COLUMN = "uuidRawSv";
+ private static final String UUID_RAW_MV_COLUMN = "uuidRawMv";
+
+ 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";
+ private static final String UUID_3 = "550e8400-e29b-41d4-a716-446655440003";
+
+ private static final List<String> UUID_SV_VALUES = List.of(UUID_0, UUID_0,
UUID_1, UUID_2);
+ private static final List<List<String>> UUID_MV_VALUES =
+ List.of(List.of(UUID_0, UUID_1), List.of(UUID_1, UUID_2),
List.of(UUID_0), List.of(UUID_3));
+
+ @Override
+ public String getTableName() {
+ return TABLE_NAME;
+ }
+
+ @Override
+ protected long getCountStarResult() {
+ return UUID_SV_VALUES.size();
+ }
+
+ @Override
+ public int getNumAvroFiles() {
+ return 1;
+ }
+
+ @Override
+ public TableConfig createOfflineTableConfig() {
+ return new
TableConfigBuilder(TableType.OFFLINE).setTableName(getTableName())
+ .setNoDictionaryColumns(List.of(UUID_RAW_SV_COLUMN,
UUID_RAW_MV_COLUMN)).build();
+ }
+
+ @Override
+ public Schema createSchema() {
+ return new Schema.SchemaBuilder().setSchemaName(getTableName())
+ .addSingleValueDimension(UUID_DICT_SV_COLUMN, DataType.UUID)
+ .addMultiValueDimension(UUID_DICT_MV_COLUMN, DataType.UUID)
+ .addSingleValueDimension(UUID_RAW_SV_COLUMN, DataType.UUID)
+ .addMultiValueDimension(UUID_RAW_MV_COLUMN, DataType.UUID)
+ .build();
+ }
+
+ @Override
+ public List<File> createAvroFiles()
+ throws Exception {
+ org.apache.avro.Schema uuidSchema =
org.apache.avro.Schema.create(org.apache.avro.Schema.Type.STRING);
+ 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_DICT_SV_COLUMN, uuidSchema,
null, null),
+ new org.apache.avro.Schema.Field(UUID_DICT_MV_COLUMN,
org.apache.avro.Schema.createArray(uuidSchema), null,
+ null),
+ new org.apache.avro.Schema.Field(UUID_RAW_SV_COLUMN, uuidSchema, null,
null),
+ new org.apache.avro.Schema.Field(UUID_RAW_MV_COLUMN,
org.apache.avro.Schema.createArray(uuidSchema), null,
+ null)));
+
+ try (AvroFilesAndWriters avroFilesAndWriters =
createAvroFilesAndWriters(avroSchema)) {
+ DataFileWriter<GenericData.Record> writer =
avroFilesAndWriters.getWriters().get(0);
+ for (int i = 0; i < UUID_SV_VALUES.size(); i++) {
+ GenericData.Record record = new GenericData.Record(avroSchema);
+ record.put(UUID_DICT_SV_COLUMN, UUID_SV_VALUES.get(i));
+ record.put(UUID_DICT_MV_COLUMN, UUID_MV_VALUES.get(i));
+ record.put(UUID_RAW_SV_COLUMN, UUID_SV_VALUES.get(i));
+ record.put(UUID_RAW_MV_COLUMN, UUID_MV_VALUES.get(i));
+ writer.append(record);
+ }
+ return avroFilesAndWriters.getAvroFiles();
+ }
+ }
+
+ @Test
+ public void testGroupByHavingReturningFinalResult()
+ throws Exception {
+ setUseMultiStageQueryEngine(false);
+ JsonNode rows = query(String.format(
+ "SELECT %1$s, COUNT(*) FROM %2$s GROUP BY %1$s HAVING %1$s = '%3$s' "
+ + "OPTION(serverReturnFinalResult=true)",
+ UUID_DICT_SV_COLUMN, getTableName(), 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(), 2L, rows.toPrettyString());
+ }
+
+ @Test(dataProvider = "useBothQueryEngines")
+ public void testDistinctOnUuidColumn(boolean useMultiStageQueryEngine)
+ throws Exception {
+ setUseMultiStageQueryEngine(useMultiStageQueryEngine);
+ JsonNode rows = query(String.format("SELECT DISTINCT %1$s FROM %2$s ORDER
BY %1$s", UUID_DICT_SV_COLUMN,
+ getTableName()));
+
+ assertEquals(rows.size(), 3, rows.toPrettyString());
+ for (int i = 0; i < rows.size(); i++) {
+ assertEquals(rows.get(i).get(0).asText(), List.of(UUID_0, UUID_1,
UUID_2).get(i), rows.toPrettyString());
+ }
+ }
+
+ @Test
+ public void testDistinctCountOnUuidColumns()
+ throws Exception {
+ setUseMultiStageQueryEngine(false);
+ for (String function : List.of("DISTINCTCOUNT", "DISTINCTCOUNTHLL",
"DISTINCTCOUNTHLLPLUS",
+ "DISTINCTCOUNTBITMAP", "DISTINCTCOUNTTHETASKETCH",
"DISTINCTCOUNTCPCSKETCH")) {
+ JsonNode rows = query(String.format("SELECT %1$s(%2$s), %1$s(%3$s),
%1$s(%4$s) FROM %5$s", function,
+ UUID_DICT_SV_COLUMN, UUID_RAW_SV_COLUMN, UUID_RAW_MV_COLUMN,
getTableName()));
+ assertCounts(rows.get(0), 3L, 3L, 4L);
+ }
+
+ // DISTINCTCOUNTULL currently supports only single-value inputs.
+ JsonNode rows = query(String.format("SELECT DISTINCTCOUNTULL(%s),
DISTINCTCOUNTULL(%s) FROM %s",
+ UUID_DICT_SV_COLUMN, UUID_RAW_SV_COLUMN, getTableName()));
+ assertCounts(rows.get(0), 3L, 3L);
+
+ rows = query(String.format(
+ "SELECT DISTINCTCOUNTTHETASKETCH(%1$s, '', '%1$s = ''%3$s''', '$1'), "
+ + "DISTINCTCOUNTTHETASKETCH(%2$s, '', '%2$s = ''%3$s''', '$1')
FROM %4$s",
+ UUID_RAW_SV_COLUMN, UUID_RAW_MV_COLUMN, UUID_0, getTableName()));
+ assertCounts(rows.get(0), 1L, 2L);
+ }
+
+ @Test
+ public void testCpcAndThetaGroupByUuidColumns()
+ throws Exception {
+ setUseMultiStageQueryEngine(false);
+ JsonNode rows = queryGroupBy(UUID_RAW_SV_COLUMN);
+ assertEquals(rows.size(), 3, rows.toPrettyString());
+ assertGroupRow(rows.get(0), UUID_0, 1L, 3L, 1L, 3L);
+ assertGroupRow(rows.get(1), UUID_1, 1L, 1L, 1L, 1L);
+ assertGroupRow(rows.get(2), UUID_2, 1L, 1L, 1L, 1L);
+
+ // UUID multi-value group keys require a dictionary, while the aggregate
input remains raw.
+ rows = queryGroupBy(UUID_DICT_MV_COLUMN);
+ assertEquals(rows.size(), 4, rows.toPrettyString());
+ assertGroupRow(rows.get(0), UUID_0, 2L, 2L, 2L, 2L);
+ assertGroupRow(rows.get(1), UUID_1, 1L, 3L, 1L, 3L);
+ assertGroupRow(rows.get(2), UUID_2, 1L, 2L, 1L, 2L);
+ assertGroupRow(rows.get(3), UUID_3, 1L, 1L, 1L, 1L);
+ }
+
+ private JsonNode queryGroupBy(String groupByColumn)
+ throws Exception {
+ return query(String.format(
+ "SELECT %1$s, DISTINCTCOUNTCPCSKETCH(%2$s),
DISTINCTCOUNTCPCSKETCH(%3$s), "
+ + "DISTINCTCOUNTTHETASKETCH(%2$s), DISTINCTCOUNTTHETASKETCH(%3$s) "
+ + "FROM %4$s GROUP BY %1$s ORDER BY %1$s",
+ groupByColumn, UUID_RAW_SV_COLUMN, UUID_RAW_MV_COLUMN,
getTableName()));
+ }
+
+ 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");
+ }
+
+ private static void assertGroupRow(JsonNode row, String groupKey, long...
expectedCounts) {
+ assertEquals(row.get(0).asText(), groupKey, row.toPrettyString());
+ for (int i = 0; i < expectedCounts.length; i++) {
+ assertEquals(row.get(i + 1).asLong(), expectedCounts[i],
row.toPrettyString());
+ }
+ }
+
+ private static void assertCounts(JsonNode row, long... expectedCounts) {
+ assertEquals(row.size(), expectedCounts.length, row.toPrettyString());
+ for (int i = 0; i < expectedCounts.length; i++) {
+ assertEquals(row.get(i).asLong(), expectedCounts[i],
row.toPrettyString());
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
