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 091eac79529 Fix HyperLogLogPlus max byte size and reject sp above 25
(#20154)
091eac79529 is described below
commit 091eac79529cfe0499bce1e498259dedc858bd9b
Author: Xiaotian (Jackie) Jiang <[email protected]>
AuthorDate: Wed Nov 4 16:47:32 2026 -0800
Fix HyperLogLogPlus max byte size and reject sp above 25 (#20154)
---
.../DistinctCountHLLPlusAggregationFunction.java | 2 +
...stinctCountSmartHLLPlusAggregationFunction.java | 2 +
.../function/AggregationFunctionFactoryTest.java | 18 +++++
.../DistinctCountHLLPlusValueAggregator.java | 12 +--
.../segment/local/utils/HyperLogLogPlusUtils.java | 62 ++++++++++++---
.../segment/local/utils/TableConfigUtils.java | 13 ++++
.../DistinctCountHLLPlusValueAggregatorTest.java | 55 +++++++++++++
.../local/aggregator/ValueAggregatorTest.java | 4 +
...MutableSegmentImplIngestionAggregationTest.java | 47 ++++++++++++
.../local/utils/HyperLogLogPlusUtilsTest.java | 89 ++++++++++++++++++++++
.../segment/local/utils/TableConfigUtilsTest.java | 59 ++++++++++++++
11 files changed, 349 insertions(+), 14 deletions(-)
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 e54e59db08f..316d3d0bbe2 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
@@ -32,6 +32,7 @@ import
org.apache.pinot.core.query.aggregation.AggregationResultHolder;
import org.apache.pinot.core.query.aggregation.ObjectAggregationResultHolder;
import org.apache.pinot.core.query.aggregation.groupby.GroupByResultHolder;
import
org.apache.pinot.core.query.aggregation.groupby.ObjectGroupByResultHolder;
+import org.apache.pinot.segment.local.utils.HyperLogLogPlusUtils;
import org.apache.pinot.segment.spi.AggregationFunctionType;
import org.apache.pinot.segment.spi.Constants;
import org.apache.pinot.segment.spi.index.reader.Dictionary;
@@ -59,6 +60,7 @@ public class DistinctCountHLLPlusAggregationFunction extends
BaseSingleInputAggr
} else if (arguments.size() == 3) {
_p = arguments.get(1).getLiteral().getIntValue();
_sp = arguments.get(2).getLiteral().getIntValue();
+ HyperLogLogPlusUtils.validateSp(_sp);
} else {
_p = CommonConstants.Helix.DEFAULT_HYPERLOGLOG_PLUS_P;
_sp = CommonConstants.Helix.DEFAULT_HYPERLOGLOG_PLUS_SP;
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountSmartHLLPlusAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountSmartHLLPlusAggregationFunction.java
index cf30ff5f30b..62dfbea39ba 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountSmartHLLPlusAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountSmartHLLPlusAggregationFunction.java
@@ -32,6 +32,7 @@ import
org.apache.pinot.common.utils.DataSchema.ColumnDataType;
import org.apache.pinot.core.common.BlockValSet;
import org.apache.pinot.core.common.ObjectSerDeUtils;
import org.apache.pinot.core.query.aggregation.AggregationResultHolder;
+import org.apache.pinot.segment.local.utils.HyperLogLogPlusUtils;
import org.apache.pinot.segment.spi.AggregationFunctionType;
import org.apache.pinot.segment.spi.index.reader.Dictionary;
import org.apache.pinot.spi.data.FieldSpec.DataType;
@@ -443,6 +444,7 @@ public class DistinctCountSmartHLLPlusAggregationFunction
extends BaseDistinctCo
break;
case SP_KEY:
_sp = Integer.parseInt(value);
+ HyperLogLogPlusUtils.validateSp(_sp);
break;
default:
throw new IllegalArgumentException("Invalid parameter key: " +
key);
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionFactoryTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionFactoryTest.java
index d10465d426c..c6652c9caca 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionFactoryTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionFactoryTest.java
@@ -18,12 +18,15 @@
*/
package org.apache.pinot.core.query.aggregation.function;
+import java.util.List;
import org.apache.pinot.common.request.context.FunctionContext;
import org.apache.pinot.common.request.context.RequestContextUtils;
import org.apache.pinot.segment.spi.AggregationFunctionType;
+import org.apache.pinot.spi.exception.BadQueryRequestException;
import org.testng.annotations.Test;
import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertThrows;
import static org.testng.Assert.assertTrue;
@@ -491,6 +494,21 @@ public class AggregationFunctionFactoryTest {
assertEquals(aggregationFunction.getResultColumnName(),
function.toString());
}
+ @Test
+ public void testDistinctCountHLLPlusSp() {
+ for (String functionName : List.of("distinctCountHLLPlus",
"distinctCountRawHLLPlus", "distinctCountHLLPlusMV",
+ "distinctCountRawHLLPlusMV")) {
+
AggregationFunctionFactory.getAggregationFunction(getFunction(functionName,
"(column, 12, 25)"), false);
+ assertThrows(BadQueryRequestException.class,
+ () ->
AggregationFunctionFactory.getAggregationFunction(getFunction(functionName,
"(column, 12, 26)"),
+ false));
+ }
+
AggregationFunctionFactory.getAggregationFunction(getFunction("distinctCountSmartHLLPlus",
"(column, 'sp=25')"),
+ false);
+ assertThrows(BadQueryRequestException.class, () ->
AggregationFunctionFactory.getAggregationFunction(
+ getFunction("distinctCountSmartHLLPlus", "(column, 'sp=26')"), false));
+ }
+
private FunctionContext getFunction(String functionName) {
return getFunction(functionName, ARGUMENT_COLUMN);
}
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/DistinctCountHLLPlusValueAggregator.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/DistinctCountHLLPlusValueAggregator.java
index 5c2855d3abb..71f90b9511e 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/DistinctCountHLLPlusValueAggregator.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/DistinctCountHLLPlusValueAggregator.java
@@ -37,7 +37,7 @@ public class DistinctCountHLLPlusValueAggregator implements
ValueAggregator<Obje
private final int _p;
private final int _sp;
- // Byte size won't change once we get the initial aggregated value
+ // Max of maxByteSize over the initial aggregated values. A merge keeps the
p and sp of the value merged into
private int _maxByteSize;
public DistinctCountHLLPlusValueAggregator(List<ExpressionContext>
arguments) {
@@ -48,6 +48,7 @@ public class DistinctCountHLLPlusValueAggregator implements
ValueAggregator<Obje
} else if (arguments.size() == 2) {
_p = arguments.get(0).getLiteral().getIntValue();
_sp = arguments.get(1).getLiteral().getIntValue();
+ HyperLogLogPlusUtils.validateSp(_sp);
} else {
_p = CommonConstants.Helix.DEFAULT_HYPERLOGLOG_PLUS_P;
_sp = CommonConstants.Helix.DEFAULT_HYPERLOGLOG_PLUS_SP;
@@ -73,11 +74,11 @@ public class DistinctCountHLLPlusValueAggregator implements
ValueAggregator<Obje
if (rawValue instanceof byte[]) {
byte[] bytes = (byte[]) rawValue;
initialValue = deserializeAggregatedValue(bytes);
- _maxByteSize = bytes.length;
+ _maxByteSize = Math.max(_maxByteSize,
HyperLogLogPlusUtils.maxByteSize(bytes));
} else {
initialValue = new HyperLogLogPlus(_p, _sp);
offerValue(initialValue, rawValue);
- _maxByteSize = HyperLogLogPlusUtils.byteSize(_p, _sp);
+ _maxByteSize = Math.max(_maxByteSize,
HyperLogLogPlusUtils.maxByteSize(_p, _sp));
}
return initialValue;
}
@@ -125,14 +126,15 @@ public class DistinctCountHLLPlusValueAggregator
implements ValueAggregator<Obje
@Override
public boolean isAggregatedValueFixedSize() {
- return true;
+ // The sparse representation grows with the number of distinct values
until it converts to the normal one
+ return _sp == 0;
}
@Override
public int getMaxAggregatedValueByteSize() {
// NOTE: For aggregated metrics, initial aggregated value might have not
been generated. Returns the byte size
// based on p and sp.
- return _maxByteSize > 0 ? _maxByteSize : HyperLogLogPlusUtils.byteSize(_p,
_sp);
+ return _maxByteSize > 0 ? _maxByteSize :
HyperLogLogPlusUtils.maxByteSize(_p, _sp);
}
@Override
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/HyperLogLogPlusUtils.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/HyperLogLogPlusUtils.java
index 6228fed3f52..f7761fa884d 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/HyperLogLogPlusUtils.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/HyperLogLogPlusUtils.java
@@ -18,22 +18,66 @@
*/
package org.apache.pinot.segment.local.utils;
-import com.clearspring.analytics.stream.cardinality.HyperLogLogPlus;
+import com.clearspring.analytics.stream.cardinality.RegisterSet;
+import com.google.common.base.Preconditions;
+import java.nio.ByteBuffer;
public class HyperLogLogPlusUtils {
private HyperLogLogPlusUtils() {
}
- /// Returns the byte size of the given HyperLogLog.
- public static int byteSize(HyperLogLogPlus value) {
- // 8 bytes header (p, sp, & register set size) & register set data
- return value.sizeof() + 3 * Integer.BYTES;
+ /// Maximum supported `sp`. The sparse representation packs the `sp`-bit
sparse index, a 6-bit run length and a flag
+ /// bit into an `int`, so a larger `sp` overflows and corrupts the registers
once the sketch converts to the normal
+ /// representation, even though stream-lib accepts `sp` up to 32.
+ public static final int MAX_SP = 25;
+
+ private static final int MAX_VAR_INT_BYTES = 5;
+
+ /// Validates that `sp` does not exceed [#MAX_SP].
+ public static void validateSp(int sp) {
+ Preconditions.checkArgument(sp <= MAX_SP, "HyperLogLogPlus sp must be at
most %s, got: %s", MAX_SP, sp);
+ }
+
+ /// Returns the maximum serialized size of a `HyperLogLogPlus` with the
given `p` and `sp`: an upper bound of
+ /// `HyperLogLogPlus.getBytes().length` over every sketch with these
parameters, not the length of any one of them.
+ ///
+ /// The serialized form is a 4-byte version, varints of `p`, `sp` and the
format flag, then the format payload:
+ /// - Normal format: a varint of the register set byte size, then the
register set. Its size depends only on `p`, so
+ /// with `sp == 0` (sparse representation disabled) every sketch
serializes to exactly the returned size.
+ /// - Sparse format (`sp > 0`, until the sketch converts to normal format):
a varint of the sparse set length, then a
+ /// delta varint of up to 5 bytes per entry. The sketch converts to normal
format once the sparse set exceeds
+ /// `0.75 * (1 << p)` entries, which bounds the entries a sparse sketch
can serialize.
+ public static int maxByteSize(int p, int sp) {
+ int headerByteSize = Integer.BYTES + varIntSize(p) + varIntSize(sp) + 1;
+ int registerSetByteSize = RegisterSet.getSizeForCount(1 << p) *
Integer.BYTES;
+ int normalByteSize = headerByteSize + varIntSize(registerSetByteSize) +
registerSetByteSize;
+ if (sp == 0) {
+ return normalByteSize;
+ }
+ int sparseSetThreshold = (int) ((1 << p) * 0.75);
+ int sparseByteSize = headerByteSize + varIntSize(sparseSetThreshold) +
sparseSetThreshold * MAX_VAR_INT_BYTES;
+ return Math.max(normalByteSize, sparseByteSize);
+ }
+
+ /// Returns the [#maxByteSize(int, int)] for the `p` and `sp` of the given
serialized `HyperLogLogPlus`, read from its
+ /// header. Both the current format (a negative version followed by varints
of `p` and `sp`) and the legacy format
+ /// without version (ints of `p` and `sp`) are supported. In the current
format, each varint is a single byte because
+ /// `p` and `sp` are below 128: stream-lib rejects `sp` above 32, and `p`
above 30 overflows the register count.
+ public static int maxByteSize(byte[] bytes) {
+ ByteBuffer buffer = ByteBuffer.wrap(bytes);
+ int version = buffer.getInt();
+ if (version < 0) {
+ int p = buffer.get();
+ int sp = buffer.get();
+ return maxByteSize(p, sp);
+ } else {
+ return maxByteSize(version, buffer.getInt());
+ }
}
- /// Returns the byte size of HyperLogLogPlus of a given p and sp.
- public static int byteSize(int p, int sp) {
- // 8 bytes header (p & sp) & register set data
- return new HyperLogLogPlus(p, sp).sizeof() + 3 * Integer.BYTES;
+ /// Returns the number of bytes in the unsigned varint encoding of the given
value.
+ private static int varIntSize(int value) {
+ return (Integer.SIZE - 1 - Integer.numberOfLeadingZeros(value | 1)) / 7 +
1;
}
}
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java
index 66f9355b3d4..f3b9fc3a809 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java
@@ -53,6 +53,7 @@ import
org.apache.pinot.segment.local.aggregator.ValueAggregator;
import org.apache.pinot.segment.local.aggregator.ValueAggregatorFactory;
import
org.apache.pinot.segment.local.recordtransformer.SchemaConformingTransformer;
import org.apache.pinot.segment.local.segment.index.loader.IndexLoadingConfig;
+import org.apache.pinot.segment.local.startree.StarTreeBuilderUtils;
import org.apache.pinot.segment.spi.AggregationFunctionType;
import org.apache.pinot.segment.spi.index.DictionaryIndexConfig;
import org.apache.pinot.segment.spi.index.FieldIndexConfigs;
@@ -2320,6 +2321,18 @@ public final class TableConfigUtils {
LOGGER.warn("StarTreeIndex config duplication: {} already matches
existing function column pair: {}. ",
columnPair, storedType);
}
+ // Create the value aggregator the same way the star-tree builder
does, so that invalid function parameters
+ // fail the config validation instead of the segment creation
+ AggregationFunctionType storedFunctionType =
storedType.getFunctionType();
+ Map<String, Object> functionParameters =
aggregationConfig.getFunctionParameters();
+ try {
+ ValueAggregatorFactory.getValueAggregator(storedFunctionType,
+
StarTreeBuilderUtils.expressionContextFromFunctionParameters(storedFunctionType,
+ functionParameters != null ? functionParameters :
Map.of()));
+ } catch (Exception e) {
+ throw new IllegalStateException("Invalid function parameters in
StarTreeIndex config: " + aggregationConfig,
+ e);
+ }
String column = columnPair.getColumn();
if (!column.equals(AggregationFunctionColumnPair.STAR)) {
aggregatedColumns.add(column);
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/aggregator/DistinctCountHLLPlusValueAggregatorTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/aggregator/DistinctCountHLLPlusValueAggregatorTest.java
new file mode 100644
index 00000000000..12efb475851
--- /dev/null
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/aggregator/DistinctCountHLLPlusValueAggregatorTest.java
@@ -0,0 +1,55 @@
+/**
+ * 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.segment.local.aggregator;
+
+import com.clearspring.analytics.stream.cardinality.HyperLogLogPlus;
+import java.util.List;
+import org.apache.pinot.segment.local.utils.HyperLogLogPlusUtils;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+
+
+public class DistinctCountHLLPlusValueAggregatorTest {
+
+ @Test
+ public void testMaxAggregatedValueByteSizeForSerializedValues() {
+ // Default p and sp, which differ from the ones of the serialized values
+ DistinctCountHLLPlusValueAggregator aggregator = new
DistinctCountHLLPlusValueAggregator(List.of());
+ int p = 12;
+ int sp = 20;
+
+ HyperLogLogPlus value =
aggregator.getInitialAggregatedValue(serialize(aggregator, p, sp, 0, 10));
+ value = aggregator.applyRawValue(value, serialize(aggregator, p, sp, 10,
2000));
+ aggregator.getInitialAggregatedValue(serialize(aggregator, p, sp, 0, 5));
+
+ int maxByteSize = aggregator.getMaxAggregatedValueByteSize();
+ assertEquals(maxByteSize, HyperLogLogPlusUtils.maxByteSize(p, sp));
+ assertTrue(aggregator.serializeAggregatedValue(value).length <=
maxByteSize);
+ }
+
+ private static byte[] serialize(DistinctCountHLLPlusValueAggregator
aggregator, int p, int sp, int start, int end) {
+ HyperLogLogPlus hllPlus = new HyperLogLogPlus(p, sp);
+ for (int i = start; i < end; i++) {
+ hllPlus.offer(i);
+ }
+ return aggregator.serializeAggregatedValue(hllPlus);
+ }
+}
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/aggregator/ValueAggregatorTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/aggregator/ValueAggregatorTest.java
index f98fac81a8c..238a80302a5 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/aggregator/ValueAggregatorTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/aggregator/ValueAggregatorTest.java
@@ -54,6 +54,10 @@ public class ValueAggregatorTest {
{AggregationFunctionType.DISTINCTCOUNTBITMAP, List.of(), false},
{AggregationFunctionType.DISTINCTCOUNTHLL, List.of(), true},
{AggregationFunctionType.DISTINCTCOUNTHLLPLUS, List.of(), true},
+ {AggregationFunctionType.DISTINCTCOUNTHLLPLUS, List.of(
+ ExpressionContext.forLiteral(Literal.intValue(12)),
+ ExpressionContext.forLiteral(Literal.intValue(20))
+ ), false},
{AggregationFunctionType.DISTINCTCOUNTULL, List.of(), true},
{AggregationFunctionType.DISTINCTCOUNTTHETASKETCH, List.of(), false},
{AggregationFunctionType.DISTINCTCOUNTTUPLESKETCH, List.of(), true},
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImplIngestionAggregationTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImplIngestionAggregationTest.java
index eab8784a9e0..9d8799a1cad 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImplIngestionAggregationTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImplIngestionAggregationTest.java
@@ -19,6 +19,7 @@
package org.apache.pinot.segment.local.indexsegment.mutable;
import com.clearspring.analytics.stream.cardinality.HyperLogLog;
+import com.clearspring.analytics.stream.cardinality.HyperLogLogPlus;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.ArrayList;
@@ -195,6 +196,33 @@ public class MutableSegmentImplIngestionAggregationTest {
mutableSegmentImpl.destroy();
}
+ @Test
+ public void testDistinctCountHLLPlus()
+ throws Exception {
+ String m1 = "hllPlus1";
+
+ Schema schema = getSchemaBuilder().addMetric(m1, DataType.BYTES).build();
+ MutableSegmentImpl mutableSegmentImpl =
+ MutableSegmentImplTestUtils.createMutableSegmentImpl(schema,
Set.of(m1), VAR_LENGTH_SET, INVERTED_INDEX_SET,
+ List.of(new AggregationConfig(m1, "distinctCountHLLPlus(metric,
12)")));
+
+ Map<String, HyperLogLogPlus> expectedValues =
addRowsDistinctCountHLLPlus(998, mutableSegmentImpl);
+
+ int numDocsIndexed = mutableSegmentImpl.getNumDocsIndexed();
+ assertTrue(numDocsIndexed < NUM_ROWS);
+ assertEquals(numDocsIndexed, expectedValues.size());
+
+ GenericRow reuse = new GenericRow();
+ for (int docId = 0; docId < numDocsIndexed; docId++) {
+ GenericRow row = mutableSegmentImpl.getRecord(docId, reuse);
+ HyperLogLogPlus actualHLLPlus = HyperLogLogPlus.Builder.build((byte[])
row.getValue(m1));
+ HyperLogLogPlus expectedHLLPlus = expectedValues.get(buildKey(row));
+ assertEquals(actualHLLPlus.cardinality(), expectedHLLPlus.cardinality());
+ }
+
+ mutableSegmentImpl.destroy();
+ }
+
@Test
public void testCount()
throws Exception {
@@ -276,6 +304,25 @@ public class MutableSegmentImplIngestionAggregationTest {
return valueMap;
}
+ private Map<String, HyperLogLogPlus> addRowsDistinctCountHLLPlus(long seed,
MutableSegmentImpl mutableSegmentImpl)
+ throws Exception {
+ Map<String, HyperLogLogPlus> valueMap = new HashMap<>();
+
+ Random random = new Random(seed);
+ for (int i = 0; i < NUM_ROWS; i++) {
+ GenericRow row = getRow(random, 1);
+ String key = buildKey(row);
+
+ int metricValue = random.nextInt(5000000);
+ row.putValue(METRIC, metricValue);
+ mutableSegmentImpl.index(row, METADATA);
+
+ valueMap.computeIfAbsent(key, k -> new
HyperLogLogPlus(12)).offer(metricValue);
+ }
+
+ return valueMap;
+ }
+
private Map<String, BigDecimal> addRowsSumPrecision(long seed,
MutableSegmentImpl mutableSegmentImpl)
throws Exception {
Map<String, BigDecimal> valueMap = new HashMap<>();
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/HyperLogLogPlusUtilsTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/HyperLogLogPlusUtilsTest.java
new file mode 100644
index 00000000000..12e79cf09f0
--- /dev/null
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/HyperLogLogPlusUtilsTest.java
@@ -0,0 +1,89 @@
+/**
+ * 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.segment.local.utils;
+
+import com.clearspring.analytics.stream.cardinality.HyperLogLogPlus;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertThrows;
+import static org.testng.Assert.assertTrue;
+
+
+public class HyperLogLogPlusUtilsTest {
+
+ @Test
+ public void testMaxByteSizeNormalFormat()
+ throws IOException {
+ for (int p = 4; p <= 16; p++) {
+ int maxByteSize = HyperLogLogPlusUtils.maxByteSize(p, 0);
+ HyperLogLogPlus hllPlus = new HyperLogLogPlus(p);
+ assertEquals(hllPlus.getBytes().length, maxByteSize);
+ for (int i = 0; i < 1000; i++) {
+ hllPlus.offer(i);
+ }
+ byte[] bytes = hllPlus.getBytes();
+ assertEquals(bytes.length, maxByteSize);
+ assertEquals(HyperLogLogPlusUtils.maxByteSize(bytes), maxByteSize);
+ }
+ }
+
+ @Test
+ public void testMaxByteSizeSparseFormat()
+ throws IOException {
+ for (int p = 4; p <= 16; p += 2) {
+ int m = 1 << p;
+ int sparseSetThreshold = (int) (m * 0.75);
+ int[] numValuesToFill = new int[]{0, 1, m / 8, m / 4, m / 2,
sparseSetThreshold, sparseSetThreshold + 1, 2 * m};
+ for (int sp : new int[]{p, 20, HyperLogLogPlusUtils.MAX_SP}) {
+ int maxByteSize = HyperLogLogPlusUtils.maxByteSize(p, sp);
+ HyperLogLogPlus hllPlus = new HyperLogLogPlus(p, sp);
+ int numValues = 0;
+ for (int numValuesToReach : numValuesToFill) {
+ for (; numValues < numValuesToReach; numValues++) {
+ hllPlus.offer(numValues);
+ }
+ byte[] bytes = hllPlus.getBytes();
+ assertTrue(bytes.length <= maxByteSize,
+ String.format("p: %d, sp: %d, numValues: %d, serialized size: %d
exceeds max byte size: %d", p, sp,
+ numValues, bytes.length, maxByteSize));
+ assertEquals(HyperLogLogPlusUtils.maxByteSize(bytes), maxByteSize);
+ }
+ }
+ }
+ }
+
+ @Test
+ public void testMaxByteSizeLegacyFormat() {
+ for (int sp : new int[]{0, 20}) {
+ byte[] legacyHeader = ByteBuffer.allocate(2 *
Integer.BYTES).putInt(12).putInt(sp).array();
+ assertEquals(HyperLogLogPlusUtils.maxByteSize(legacyHeader),
HyperLogLogPlusUtils.maxByteSize(12, sp));
+ }
+ }
+
+ @Test
+ public void testValidateSp() {
+ HyperLogLogPlusUtils.validateSp(0);
+ HyperLogLogPlusUtils.validateSp(25);
+ assertThrows(IllegalArgumentException.class, () ->
HyperLogLogPlusUtils.validateSp(26));
+ assertThrows(IllegalArgumentException.class, () ->
HyperLogLogPlusUtils.validateSp(32));
+ }
+}
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/TableConfigUtilsTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/TableConfigUtilsTest.java
index 2bc88730796..73edeac2f18 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/TableConfigUtilsTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/TableConfigUtilsTest.java
@@ -842,6 +842,40 @@ public class TableConfigUtilsTest {
fail("Should not fail due to valid aggregation function", e);
}
+ // distinctcounthllplus, expect sp above 25 to not validate
+ schema = new
Schema.SchemaBuilder().setSchemaName(TABLE_NAME).addMetric("d1",
DataType.BYTES).build();
+
+ aggregationConfigs = List.of(new AggregationConfig("d1",
"DISTINCTCOUNTHLLPLUS(s1, 12, 26)"));
+ ingestionConfig.setAggregationConfigs(aggregationConfigs);
+ tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName("myTable_REALTIME")
+ .setTimeColumnName("timeColumn")
+ .setIngestionConfig(ingestionConfig)
+ .setNoDictionaryColumns(List.of("d1"))
+ .build();
+
+ try {
+ TableConfigUtils.validateIngestionConfig(tableConfig, schema);
+ fail("Should fail due to sp above 25");
+ } catch (IllegalStateException e) {
+ // expected
+ }
+
+ // distinctcounthllplus, expect sp above 0 to not validate because the
aggregated value is not of fixed size
+ aggregationConfigs = List.of(new AggregationConfig("d1",
"DISTINCTCOUNTHLLPLUS(s1, 12, 20)"));
+ ingestionConfig.setAggregationConfigs(aggregationConfigs);
+ tableConfig = new
TableConfigBuilder(TableType.REALTIME).setTableName("myTable_REALTIME")
+ .setTimeColumnName("timeColumn")
+ .setIngestionConfig(ingestionConfig)
+ .setNoDictionaryColumns(List.of("d1"))
+ .build();
+
+ try {
+ TableConfigUtils.validateIngestionConfig(tableConfig, schema);
+ fail("Should fail due to sp above 0");
+ } catch (IllegalStateException e) {
+ assertTrue(e.getMessage().contains("must have fixed size aggregated
value"), e.getMessage());
+ }
+
// distinctcounthll, expect not specified log2m argument to default to 8
schema = new
Schema.SchemaBuilder().setSchemaName(TABLE_NAME).addMetric("d1",
DataType.BYTES).build();
@@ -2569,6 +2603,31 @@ public class TableConfigUtilsTest {
assertTrue(e.getMessage().contains("Duplicate function column pair"));
}
+ @Test
+ public void testValidateStarTreeIndexFunctionParameters() {
+ Schema schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
+ .addSingleValueDimension("myCol", DataType.STRING)
+ .build();
+
+ StarTreeIndexConfig starTreeIndexConfig = new
StarTreeIndexConfig(List.of("myCol"), null, null, List.of(
+ new StarTreeAggregationConfig("myCol", "DISTINCTCOUNTHLLPLUS",
+ Map.of(Constants.HLLPLUS_ULL_P_KEY, 12, Constants.HLLPLUS_SP_KEY,
25), null, null, null, null, null)), 1);
+ TableConfig tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
+ .setStarTreeIndexConfigs(List.of(starTreeIndexConfig))
+ .build();
+ TableConfigUtils.validate(tableConfig, schema);
+
+ starTreeIndexConfig = new StarTreeIndexConfig(List.of("myCol"), null,
null, List.of(
+ new StarTreeAggregationConfig("myCol", "DISTINCTCOUNTHLLPLUS",
+ Map.of(Constants.HLLPLUS_ULL_P_KEY, 12, Constants.HLLPLUS_SP_KEY,
26), null, null, null, null, null)), 1);
+ TableConfig invalidSpTableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME)
+ .setStarTreeIndexConfigs(List.of(starTreeIndexConfig))
+ .build();
+ IllegalStateException e =
+ expectThrows(IllegalStateException.class, () ->
TableConfigUtils.validate(invalidSpTableConfig, schema));
+ assertTrue(e.getMessage().contains("Invalid function parameters in
StarTreeIndex config"));
+ }
+
@Test
public void testValidateRetentionConfig() {
Schema schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
