This is an automated email from the ASF dual-hosted git repository.
Jackie-Jiang 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 76e23b9c3bd Give the frequency sketch functions the query's null
handling option (#19227)
76e23b9c3bd is described below
commit 76e23b9c3bd1a6aa551bb025a834f8c75f62835b
Author: Xiaotian (Jackie) Jiang <[email protected]>
AuthorDate: Wed Aug 12 14:51:28 2026 -0700
Give the frequency sketch functions the query's null handling option
(#19227)
---
.../aggregation/function/AggregationFunction.java | 4 +-
.../function/AggregationFunctionFactory.java | 4 +-
.../FrequentLongsSketchAggregationFunction.java | 102 +++++++------
.../FrequentStringsSketchAggregationFunction.java | 104 ++++++++------
.../AggregationFunctionNullContractTest.java | 6 +-
.../function/FrequentSketchNullHandlingTest.java | 159 +++++++++++++++++++++
6 files changed, 283 insertions(+), 96 deletions(-)
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunction.java
index 45208aa6e02..b95b34a36f2 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunction.java
@@ -100,8 +100,8 @@ import org.apache.pinot.segment.spi.AggregationFunctionType;
/// null rows and fold the column's default value into the aggregate
whatever the query asked for: the
/// sketch-backed distinct counts (`DISTINCTCOUNTBITMAP`,
`DISTINCTCOUNTHLL`, `DISTINCTCOUNTTHETASKETCH`,
/// `DISTINCTCOUNTCPCSKETCH`, `FASTHLL`,
`SEGMENTPARTITIONEDDISTINCTCOUNT` and the raw and smart variants of
-/// each), the frequency sketches, `HISTOGRAM`, `IDSET`, `STUNION`, the
array sums, and the funnel family.
-/// Whether a function takes the option is visible at its construction
site, which is the reliable way to tell.
+/// each), `HISTOGRAM`, `IDSET`, `STUNION`, the array sums, and the
funnel family. Whether a function takes the
+/// option is visible at its construction site, which is the reliable way
to tell.
/// The family names are not a safe shorthand for this, in either
direction. The exact distinct functions
/// (`DISTINCTCOUNT`, `DISTINCTSUM`, `DISTINCTAVG`,
`DISTINCTCOUNTOFFHEAP`) do take the option and skip null
/// rows through their shared base, as do the variance,
standard-deviation and covariance functions and the
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionFactory.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionFactory.java
index cc6efe8a17d..a5adab56302 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionFactory.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionFactory.java
@@ -516,9 +516,9 @@ public class AggregationFunctionFactory {
case FUNNELEVENTSFUNCTIONEVAL:
return new FunnelEventsFunctionEvalAggregationFunction(arguments);
case FREQUENTSTRINGSSKETCH:
- return new FrequentStringsSketchAggregationFunction(arguments);
+ return new FrequentStringsSketchAggregationFunction(arguments,
nullHandlingEnabled);
case FREQUENTLONGSSKETCH:
- return new FrequentLongsSketchAggregationFunction(arguments);
+ return new FrequentLongsSketchAggregationFunction(arguments,
nullHandlingEnabled);
case DISTINCTCOUNTCPCSKETCH:
return new DistinctCountCPCSketchAggregationFunction(arguments);
case DISTINCTCOUNTRAWCPCSKETCH:
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/FrequentLongsSketchAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/FrequentLongsSketchAggregationFunction.java
index 8cf61c17507..c24e21d0be7 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/FrequentLongsSketchAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/FrequentLongsSketchAggregationFunction.java
@@ -61,13 +61,13 @@ import org.apache.pinot.spi.data.FieldSpec;
///
/// There is a variation of the function (**FREQUENT_STRINGS_SKETCH**) which
accepts STRING type input columns.
public class FrequentLongsSketchAggregationFunction
- extends BaseSingleInputAggregationFunction<FrequentLongsSketch,
Comparable<?>> {
+ extends NullableSingleInputAggregationFunction<FrequentLongsSketch,
Comparable<?>> {
protected static final int DEFAULT_MAX_MAP_SIZE = 256;
protected int _maxMapSize;
- public FrequentLongsSketchAggregationFunction(List<ExpressionContext>
arguments) {
- super(arguments.get(0));
+ public FrequentLongsSketchAggregationFunction(List<ExpressionContext>
arguments, boolean nullHandlingEnabled) {
+ super(arguments.get(0), nullHandlingEnabled);
int numArguments = arguments.size();
Preconditions.checkArgument(numArguments == 1 || numArguments == 2,
"Expecting 1 or 2 arguments for FrequentLongsSketch function:
FREQUENTITEMSSKETCH(column, maxMapSize");
@@ -95,24 +95,34 @@ public class FrequentLongsSketchAggregationFunction
BlockValSet valueSet = blockValSetMap.get(_expression);
FieldSpec.DataType valueType = valueSet.getValueType();
- FrequentLongsSketch sketch = getOrCreateSketch(aggregationResultHolder);
-
switch (valueType) {
case BYTES:
// Assuming the column contains serialized data sketch
- FrequentLongsSketch[] deserializedSketches =
-
deserializeSketches(blockValSetMap.get(_expression).getBytesValuesSV());
- sketch = getOrCreateSketch(aggregationResultHolder);
-
- for (FrequentLongsSketch colSketch : deserializedSketches) {
- sketch.merge(colSketch);
- }
+ byte[][] bytesValues = valueSet.getBytesValuesSV();
+ // The sketch is created inside the range, so a block with no non-null
row leaves the holder untouched and
+ // extractFinalResult sees the null that means nothing was aggregated
+ forEachNotNull(length, valueSet, (from, to) -> {
+ if (to == from) {
+ return;
+ }
+ FrequentLongsSketch sketch =
getOrCreateSketch(aggregationResultHolder);
+ for (int i = from; i < to; i++) {
+ sketch.merge(deserializeSketch(bytesValues[i]));
+ }
+ });
break;
case INT:
case LONG:
- for (Long val : valueSet.getLongValuesSV()) {
- sketch.update(val);
- }
+ long[] longValues = valueSet.getLongValuesSV();
+ forEachNotNull(length, valueSet, (from, to) -> {
+ if (to == from) {
+ return;
+ }
+ FrequentLongsSketch sketch =
getOrCreateSketch(aggregationResultHolder);
+ for (int i = from; i < to; i++) {
+ sketch.update(longValues[i]);
+ }
+ });
break;
default:
throw new UnsupportedOperationException("Cannot aggregate on non
int/long types");
@@ -128,20 +138,21 @@ public class FrequentLongsSketchAggregationFunction
switch (valueType) {
case BYTES:
// serialized sketch
- FrequentLongsSketch[] deserializedSketches =
-
deserializeSketches(blockValSetMap.get(_expression).getBytesValuesSV());
- for (int i = 0; i < length; i++) {
- FrequentLongsSketch sketch = getOrCreateSketch(groupByResultHolder,
groupKeyArray[i]);
- sketch.merge(deserializedSketches[i]);
- }
+ byte[][] bytesValues = valueSet.getBytesValuesSV();
+ forEachNotNull(length, valueSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ getOrCreateSketch(groupByResultHolder,
groupKeyArray[i]).merge(deserializeSketch(bytesValues[i]));
+ }
+ });
break;
case INT:
case LONG:
long[] values = valueSet.getLongValuesSV();
- for (int i = 0; i < length; i++) {
- FrequentLongsSketch sketch = getOrCreateSketch(groupByResultHolder,
groupKeyArray[i]);
- sketch.update(values[i]);
- }
+ forEachNotNull(length, valueSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ getOrCreateSketch(groupByResultHolder,
groupKeyArray[i]).update(values[i]);
+ }
+ });
break;
default:
throw new UnsupportedOperationException("Cannot aggregate on non
int/long types");
@@ -157,24 +168,27 @@ public class FrequentLongsSketchAggregationFunction
switch (valueType) {
case BYTES:
// serialized sketch
- FrequentLongsSketch[] deserializedSketches =
-
deserializeSketches(blockValSetMap.get(_expression).getBytesValuesSV());
- for (int i = 0; i < length; i++) {
- for (int groupKey : groupKeysArray[i]) {
- FrequentLongsSketch sketch =
getOrCreateSketch(groupByResultHolder, groupKey);
- sketch.merge(deserializedSketches[i]);
+ byte[][] bytesValues = valueSet.getBytesValuesSV();
+ forEachNotNull(length, valueSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ // Deserialized once per row, not once per group key the row
belongs to
+ FrequentLongsSketch rowSketch = deserializeSketch(bytesValues[i]);
+ for (int groupKey : groupKeysArray[i]) {
+ getOrCreateSketch(groupByResultHolder,
groupKey).merge(rowSketch);
+ }
}
- }
+ });
break;
case INT:
case LONG:
long[] values = valueSet.getLongValuesSV();
- for (int i = 0; i < length; i++) {
- for (int groupKey : groupKeysArray[i]) {
- FrequentLongsSketch sketch =
getOrCreateSketch(groupByResultHolder, groupKey);
- sketch.update(values[i]);
+ forEachNotNull(length, valueSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ for (int groupKey : groupKeysArray[i]) {
+ getOrCreateSketch(groupByResultHolder,
groupKey).update(values[i]);
+ }
}
- }
+ });
break;
default:
throw new UnsupportedOperationException("Cannot aggregate on non
int/long types");
@@ -202,13 +216,9 @@ public class FrequentLongsSketchAggregationFunction
return sketch;
}
- /// Deserializes the sketches from the bytes.
- protected FrequentLongsSketch[] deserializeSketches(byte[][]
serializedSketches) {
- FrequentLongsSketch[] sketches = new
FrequentLongsSketch[serializedSketches.length];
- for (int i = 0; i < serializedSketches.length; i++) {
- sketches[i] =
FrequentLongsSketch.getInstance(MemorySegment.ofArray(serializedSketches[i]));
- }
- return sketches;
+ /// Deserializes a single serialized sketch, so a row that is skipped as
null is never deserialized.
+ protected FrequentLongsSketch deserializeSketch(byte[] serializedSketch) {
+ return
FrequentLongsSketch.getInstance(MemorySegment.ofArray(serializedSketch));
}
@Nullable
@@ -261,7 +271,9 @@ public class FrequentLongsSketchAggregationFunction
@Nullable
@Override
public Comparable<?> extractFinalResult(@Nullable FrequentLongsSketch
sketch) {
- // A null intermediate result means nothing was aggregated, and there is
no sketch to serialize
+ // A null intermediate result means nothing was aggregated, and there is
no sketch to serialize. This function
+ // has never substituted an empty accumulator during extraction, so NULL
is the answer in both modes and there is
+ // no disabled-mode identity to preserve here.
return sketch != null ? new SerializedFrequentLongsSketch(sketch) : null;
}
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/FrequentStringsSketchAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/FrequentStringsSketchAggregationFunction.java
index d164b58f954..e88cf12baad 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/FrequentStringsSketchAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/FrequentStringsSketchAggregationFunction.java
@@ -62,13 +62,14 @@ import org.apache.pinot.spi.data.FieldSpec;
///
/// There is a variation of the function (**FREQUENT_LONGS_SKETCH**) which
accept INT and LONG type input columns.
public class FrequentStringsSketchAggregationFunction
- extends BaseSingleInputAggregationFunction<FrequentItemsSketch<String>,
Comparable<?>> {
+ extends
NullableSingleInputAggregationFunction<FrequentItemsSketch<String>,
Comparable<?>> {
protected static final int DEFAULT_MAX_MAP_SIZE = 256;
protected int _maxMapSize;
- public FrequentStringsSketchAggregationFunction(List<ExpressionContext>
arguments) {
- super(arguments.get(0));
+ public FrequentStringsSketchAggregationFunction(List<ExpressionContext>
arguments,
+ boolean nullHandlingEnabled) {
+ super(arguments.get(0), nullHandlingEnabled);
int numArguments = arguments.size();
Preconditions.checkArgument(numArguments == 1 || numArguments == 2,
"Expecting 1 or 2 arguments for FrequentItemsSketch function:
FREQUENTSTRINGSSKETCH(column, maxMapSize");
@@ -96,21 +97,31 @@ public class FrequentStringsSketchAggregationFunction
BlockValSet valueSet = blockValSetMap.get(_expression);
FieldSpec.DataType valueType = valueSet.getValueType();
- FrequentItemsSketch<String> sketch =
getOrCreateSketch(aggregationResultHolder);
-
if (valueType == FieldSpec.DataType.BYTES) {
// Assuming the column contains serialized data sketch
- FrequentItemsSketch<String>[] deserializedSketches =
-
deserializeSketches(blockValSetMap.get(_expression).getBytesValuesSV());
- sketch = getOrCreateSketch(aggregationResultHolder);
-
- for (FrequentItemsSketch<String> colSketch : deserializedSketches) {
- sketch.merge(colSketch);
- }
+ byte[][] bytesValues = valueSet.getBytesValuesSV();
+ // The sketch is created inside the range, so a block with no non-null
row leaves the holder untouched and
+ // extractFinalResult sees the null that means nothing was aggregated
+ forEachNotNull(length, valueSet, (from, to) -> {
+ if (to == from) {
+ return;
+ }
+ FrequentItemsSketch<String> sketch =
getOrCreateSketch(aggregationResultHolder);
+ for (int i = from; i < to; i++) {
+ sketch.merge(deserializeSketch(bytesValues[i]));
+ }
+ });
} else {
- for (String val : valueSet.getStringValuesSV()) {
- sketch.update(val);
- }
+ String[] values = valueSet.getStringValuesSV();
+ forEachNotNull(length, valueSet, (from, to) -> {
+ if (to == from) {
+ return;
+ }
+ FrequentItemsSketch<String> sketch =
getOrCreateSketch(aggregationResultHolder);
+ for (int i = from; i < to; i++) {
+ sketch.update(values[i]);
+ }
+ });
}
}
@@ -122,18 +133,19 @@ public class FrequentStringsSketchAggregationFunction
if (valueType == FieldSpec.DataType.BYTES) {
// serialized sketch
- FrequentItemsSketch<String>[] deserializedSketches =
-
deserializeSketches(blockValSetMap.get(_expression).getBytesValuesSV());
- for (int i = 0; i < length; i++) {
- FrequentItemsSketch<String> sketch =
getOrCreateSketch(groupByResultHolder, groupKeyArray[i]);
- sketch.merge(deserializedSketches[i]);
- }
+ byte[][] bytesValues = valueSet.getBytesValuesSV();
+ forEachNotNull(length, valueSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ getOrCreateSketch(groupByResultHolder,
groupKeyArray[i]).merge(deserializeSketch(bytesValues[i]));
+ }
+ });
} else {
String[] values = valueSet.getStringValuesSV();
- for (int i = 0; i < length; i++) {
- FrequentItemsSketch<String> sketch =
getOrCreateSketch(groupByResultHolder, groupKeyArray[i]);
- sketch.update(values[i]);
- }
+ forEachNotNull(length, valueSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ getOrCreateSketch(groupByResultHolder,
groupKeyArray[i]).update(values[i]);
+ }
+ });
}
}
@@ -145,22 +157,25 @@ public class FrequentStringsSketchAggregationFunction
if (valueType == FieldSpec.DataType.BYTES) {
// serialized sketch
- FrequentItemsSketch<String>[] deserializedSketches =
-
deserializeSketches(blockValSetMap.get(_expression).getBytesValuesSV());
- for (int i = 0; i < length; i++) {
- for (int groupKey : groupKeysArray[i]) {
- FrequentItemsSketch<String> sketch =
getOrCreateSketch(groupByResultHolder, groupKey);
- sketch.merge(deserializedSketches[i]);
+ byte[][] bytesValues = valueSet.getBytesValuesSV();
+ forEachNotNull(length, valueSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ // Deserialized once per row, not once per group key the row belongs
to
+ FrequentItemsSketch<String> rowSketch =
deserializeSketch(bytesValues[i]);
+ for (int groupKey : groupKeysArray[i]) {
+ getOrCreateSketch(groupByResultHolder, groupKey).merge(rowSketch);
+ }
}
- }
+ });
} else {
String[] values = valueSet.getStringValuesSV();
- for (int i = 0; i < length; i++) {
- for (int groupKey : groupKeysArray[i]) {
- FrequentItemsSketch<String> sketch =
getOrCreateSketch(groupByResultHolder, groupKey);
- sketch.update(values[i]);
+ forEachNotNull(length, valueSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ for (int groupKey : groupKeysArray[i]) {
+ getOrCreateSketch(groupByResultHolder, groupKey).update(values[i]);
+ }
}
- }
+ });
}
}
@@ -185,14 +200,9 @@ public class FrequentStringsSketchAggregationFunction
return sketch;
}
- /// Deserializes the sketches from the bytes.
- protected FrequentItemsSketch<String>[] deserializeSketches(byte[][]
serializedSketches) {
- FrequentItemsSketch<String>[] sketches = new
FrequentItemsSketch[serializedSketches.length];
- for (int i = 0; i < serializedSketches.length; i++) {
- sketches[i] =
-
FrequentItemsSketch.getInstance(MemorySegment.ofArray(serializedSketches[i]),
new ArrayOfStringsSerDe());
- }
- return sketches;
+ /// Deserializes a single serialized sketch, so a row that is skipped as
null is never deserialized.
+ protected FrequentItemsSketch<String> deserializeSketch(byte[]
serializedSketch) {
+ return
FrequentItemsSketch.getInstance(MemorySegment.ofArray(serializedSketch), new
ArrayOfStringsSerDe());
}
@Nullable
@@ -245,7 +255,9 @@ public class FrequentStringsSketchAggregationFunction
@Nullable
@Override
public Comparable<?> extractFinalResult(@Nullable
FrequentItemsSketch<String> sketch) {
- // A null intermediate result means nothing was aggregated, and there is
no sketch to serialize
+ // A null intermediate result means nothing was aggregated, and there is
no sketch to serialize. This function
+ // has never substituted an empty accumulator during extraction, so NULL
is the answer in both modes and there is
+ // no disabled-mode identity to preserve here.
return sketch != null ? new SerializedFrequentStringsSketch(sketch) : null;
}
}
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionNullContractTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionNullContractTest.java
index e511b0e7460..34ab0ea055f 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionNullContractTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionNullContractTest.java
@@ -253,7 +253,11 @@ public class AggregationFunctionNullContractTest {
AggregationFunctionType.SUMLONG, AggregationFunctionType.SUMPRECISION,
AggregationFunctionType.FIRSTWITHTIME,
AggregationFunctionType.LASTWITHTIME, AggregationFunctionType.ARRAYAGG,
AggregationFunctionType.LISTAGG,
// Given the option so they can skip null rows; a row counts only when
both input columns are non-null
- AggregationFunctionType.COVARPOP, AggregationFunctionType.COVARSAMP
+ AggregationFunctionType.COVARPOP, AggregationFunctionType.COVARSAMP,
+ // Given the option so they can skip null rows. These two were in this
set once before, on the strength of an
+ // identity comparison that reported every serializer-valued function as
honouring it; they belong here now
+ // because they genuinely do.
+ AggregationFunctionType.FREQUENTSTRINGSSKETCH,
AggregationFunctionType.FREQUENTLONGSSKETCH
);
/// Functions this test cannot drive with a one-column synthetic block,
pinned so that a silent drop-out is always a
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/FrequentSketchNullHandlingTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/FrequentSketchNullHandlingTest.java
new file mode 100644
index 00000000000..7997e18b5b1
--- /dev/null
+++
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/FrequentSketchNullHandlingTest.java
@@ -0,0 +1,159 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.core.query.aggregation.function;
+
+import java.util.List;
+import java.util.Map;
+import org.apache.datasketches.frequencies.FrequentLongsSketch;
+import org.apache.pinot.common.request.context.ExpressionContext;
+import org.apache.pinot.core.common.BlockValSet;
+import org.apache.pinot.core.common.SyntheticBlockValSets;
+import org.apache.pinot.core.query.aggregation.AggregationResultHolder;
+import org.apache.pinot.core.query.aggregation.groupby.GroupByResultHolder;
+import
org.apache.pinot.core.query.aggregation.groupby.ObjectGroupByResultHolder;
+import org.roaringbitmap.RoaringBitmap;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertNull;
+
+
+/// Null handling and row bounding for `FREQUENTLONGSSKETCH`.
+///
+/// [AggregationFunctionNullContractTest] drives these functions through one
synthetic block shape and only through
+/// `aggregate`, so the group-by paths and the row-bounding below are checked
nowhere else.
+public class FrequentSketchNullHandlingTest {
+ private static final ExpressionContext COLUMN =
ExpressionContext.forIdentifier("column");
+ private static final long[] VALUES = {10L, 20L, 30L, 40L};
+
+ private static FrequentLongsSketchAggregationFunction longs(boolean
nullHandlingEnabled) {
+ return new FrequentLongsSketchAggregationFunction(List.of(COLUMN),
nullHandlingEnabled);
+ }
+
+ private static Map<ExpressionContext, BlockValSet> block(RoaringBitmap
nullBitmap, long[] values) {
+ return Map.of(COLUMN, SyntheticBlockValSets.Long.create(nullBitmap,
values));
+ }
+
+ private static FrequentLongsSketch
aggregate(FrequentLongsSketchAggregationFunction function, int length,
+ RoaringBitmap nullBitmap, long[] values) {
+ AggregationResultHolder resultHolder =
function.createAggregationResultHolder();
+ function.aggregate(length, resultHolder, block(nullBitmap, values));
+ return function.extractAggregationResult(resultHolder);
+ }
+
+ private static long estimate(FrequentLongsSketch sketch, long item) {
+ return sketch.getEstimate(item);
+ }
+
+ /// Only the rows that carry a value are counted.
+ @Test
+ public void testNullRowsAreSkipped() {
+ FrequentLongsSketch sketch = aggregate(longs(true), VALUES.length,
RoaringBitmap.bitmapOf(1, 3), VALUES);
+
+ assertNotNull(sketch);
+ assertEquals(estimate(sketch, 10L), 1L);
+ assertEquals(estimate(sketch, 20L), 0L);
+ assertEquals(estimate(sketch, 30L), 1L);
+ assertEquals(estimate(sketch, 40L), 0L);
+ }
+
+ /// Aggregation must stop at `length`, not run to the end of the values
array.
+ ///
+ /// `ProjectionBlockValSet` hands back the `DataBlockCache` array, and
`DataBlockCache.initNewBlock` only clears the
+ /// cache when the new block is larger than the last one. A short block
after a full one therefore sees a longer
+ /// array whose tail still holds the previous block's values, and anything
that iterates the array rather than the
+ /// range counts that tail a second time.
+ @Test
+ public void testValuesBeyondLengthAreNotCounted() {
+ long[] oversized = {10L, 20L, 30L, 40L, 99L, 99L, 99L};
+
+ FrequentLongsSketch sketch = aggregate(longs(false), 4, null, oversized);
+
+ assertNotNull(sketch);
+ assertEquals(estimate(sketch, 10L), 1L);
+ assertEquals(estimate(sketch, 40L), 1L);
+ assertEquals(estimate(sketch, 99L), 0L);
+ }
+
+ @Test
+ public void testEveryRowNullYieldsNoIntermediateResult() {
+ RoaringBitmap allNull = new RoaringBitmap();
+ allNull.add(0L, VALUES.length);
+
+ assertNull(aggregate(longs(true), VALUES.length, allNull, VALUES));
+ }
+
+ /// A zero-length block still reaches the range callback, and must not mark
the holder as aggregated.
+ @Test
+ public void testZeroLengthBlockLeavesTheHolderUntouched() {
+ assertNull(aggregate(longs(true), 0, null, VALUES));
+ }
+
+ /// The group-by path skips null rows per group, and a group whose every row
is null is never created.
+ @Test
+ public void testGroupBySkipsNullRows() {
+ FrequentLongsSketchAggregationFunction function = longs(true);
+ GroupByResultHolder resultHolder = new ObjectGroupByResultHolder(4, 4);
+ // Rows 0 and 2 go to group 0, rows 1 and 3 to group 1; rows 1 and 3 are
null, so group 1 gets nothing
+ function.aggregateGroupBySV(VALUES.length, new int[]{0, 1, 0, 1},
resultHolder,
+ block(RoaringBitmap.bitmapOf(1, 3), VALUES));
+
+ FrequentLongsSketch group0 = function.extractGroupByResult(resultHolder,
0);
+ assertNotNull(group0);
+ assertEquals(estimate(group0, 10L), 1L);
+ assertEquals(estimate(group0, 30L), 1L);
+ assertNull(function.extractGroupByResult(resultHolder, 1));
+ }
+
+ /// With the option disabled the column default is counted, which is the
answer this mode has always given, and the
+ /// final result stays `NULL` for an untouched accumulator.
+ @Test
+ public void testOptionDisabledCountsNullRowsAndStillRendersNull() {
+ RoaringBitmap allNull = new RoaringBitmap();
+ allNull.add(0L, VALUES.length);
+
+ FrequentLongsSketch sketch = aggregate(longs(false), VALUES.length,
allNull, VALUES);
+ assertNotNull(sketch);
+ assertEquals(estimate(sketch, 10L), 1L);
+
+ assertNull(longs(false).extractFinalResult(null));
+ assertNull(longs(true).extractFinalResult(null));
+ }
+
+ /// A serialized sketch column is deserialized only for the rows that carry
one.
+ @Test
+ public void testSerializedRowsAreDeserializedOnlyWhenNotNull() {
+ FrequentLongsSketch first = new FrequentLongsSketch(32);
+ first.update(10L);
+ FrequentLongsSketch second = new FrequentLongsSketch(32);
+ second.update(20L);
+ byte[][] serialized = {first.toByteArray(), second.toByteArray()};
+
+ FrequentLongsSketchAggregationFunction function = longs(true);
+ AggregationResultHolder resultHolder =
function.createAggregationResultHolder();
+ function.aggregate(2, resultHolder,
+ Map.of(COLUMN,
SyntheticBlockValSets.Bytes.create(RoaringBitmap.bitmapOf(1), serialized)));
+
+ FrequentLongsSketch sketch =
function.extractAggregationResult(resultHolder);
+ assertNotNull(sketch);
+ assertEquals(estimate(sketch, 10L), 1L);
+ assertEquals(estimate(sketch, 20L), 0L);
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
