This is an automated email from the ASF dual-hosted git repository.
jackie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 8b24ec4 Fix the DistinctCountHLL with serialized HyperLogLog of
non-default log2m (#4655)
8b24ec4 is described below
commit 8b24ec4943456876520240934463070e9dadb21a
Author: Xiaotian (Jackie) Jiang <[email protected]>
AuthorDate: Fri Sep 27 17:26:23 2019 -0700
Fix the DistinctCountHLL with serialized HyperLogLog of non-default log2m
(#4655)
---
.../DistinctCountHLLAggregationFunction.java | 155 ++++++++++++---------
1 file changed, 87 insertions(+), 68 deletions(-)
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 d3fb94c..286c330 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
@@ -18,9 +18,9 @@
*/
package org.apache.pinot.core.query.aggregation.function;
-import com.clearspring.analytics.stream.cardinality.CardinalityMergeException;
import com.clearspring.analytics.stream.cardinality.HyperLogLog;
-import org.apache.pinot.common.data.FieldSpec;
+import com.google.common.base.Preconditions;
+import org.apache.pinot.common.data.FieldSpec.DataType;
import org.apache.pinot.common.function.AggregationFunctionType;
import org.apache.pinot.common.utils.DataSchema.ColumnDataType;
import org.apache.pinot.core.common.BlockValSet;
@@ -61,60 +61,70 @@ public class DistinctCountHLLAggregationFunction implements
AggregationFunction<
@Override
public void aggregate(int length, AggregationResultHolder
aggregationResultHolder, BlockValSet... blockValSets) {
- HyperLogLog hyperLogLog = getHyperLogLog(aggregationResultHolder);
-
- FieldSpec.DataType valueType = blockValSets[0].getValueType();
- switch (valueType) {
- case INT:
- int[] intValues = blockValSets[0].getIntValuesSV();
- for (int i = 0; i < length; i++) {
- hyperLogLog.offer(intValues[i]);
- }
- break;
- case LONG:
- long[] longValues = blockValSets[0].getLongValuesSV();
- for (int i = 0; i < length; i++) {
- hyperLogLog.offer(longValues[i]);
- }
- break;
- case FLOAT:
- float[] floatValues = blockValSets[0].getFloatValuesSV();
- for (int i = 0; i < length; i++) {
- hyperLogLog.offer(floatValues[i]);
- }
- break;
- case DOUBLE:
- double[] doubleValues = blockValSets[0].getDoubleValuesSV();
- for (int i = 0; i < length; i++) {
- hyperLogLog.offer(doubleValues[i]);
- }
- break;
- case STRING:
- String[] stringValues = blockValSets[0].getStringValuesSV();
- for (int i = 0; i < length; i++) {
- hyperLogLog.offer(stringValues[i]);
- }
- break;
- case BYTES:
- // Serialized HyperLogLog
- byte[][] bytesValues = blockValSets[0].getBytesValuesSV();
- try {
+ DataType valueType = blockValSets[0].getValueType();
+ if (valueType == DataType.BYTES) {
+ // Serialized HyperLogLog
+ byte[][] bytesValues = blockValSets[0].getBytesValuesSV();
+ try {
+ HyperLogLog hyperLogLog = aggregationResultHolder.getResult();
+ if (hyperLogLog == null) {
+ hyperLogLog =
ObjectSerDeUtils.HYPER_LOG_LOG_SER_DE.deserialize(bytesValues[0]);
+ aggregationResultHolder.setValue(hyperLogLog);
+ for (int i = 1; i < length; i++) {
+
hyperLogLog.addAll(ObjectSerDeUtils.HYPER_LOG_LOG_SER_DE.deserialize(bytesValues[i]));
+ }
+ } else {
for (int i = 0; i < length; i++) {
hyperLogLog.addAll(ObjectSerDeUtils.HYPER_LOG_LOG_SER_DE.deserialize(bytesValues[i]));
}
- } catch (Exception e) {
- throw new RuntimeException("Caught exception while aggregating
HyperLogLog", e);
}
- break;
- default:
- throw new IllegalStateException("Illegal data type for
DISTINCT_COUNT_HLL aggregation function: " + valueType);
+ } catch (Exception e) {
+ throw new RuntimeException("Caught exception while merging
HyperLogLogs", e);
+ }
+ } else {
+ HyperLogLog hyperLogLog = getHyperLogLog(aggregationResultHolder);
+ switch (valueType) {
+ case INT:
+ int[] intValues = blockValSets[0].getIntValuesSV();
+ for (int i = 0; i < length; i++) {
+ hyperLogLog.offer(intValues[i]);
+ }
+ break;
+ case LONG:
+ long[] longValues = blockValSets[0].getLongValuesSV();
+ for (int i = 0; i < length; i++) {
+ hyperLogLog.offer(longValues[i]);
+ }
+ break;
+ case FLOAT:
+ float[] floatValues = blockValSets[0].getFloatValuesSV();
+ for (int i = 0; i < length; i++) {
+ hyperLogLog.offer(floatValues[i]);
+ }
+ break;
+ case DOUBLE:
+ double[] doubleValues = blockValSets[0].getDoubleValuesSV();
+ for (int i = 0; i < length; i++) {
+ hyperLogLog.offer(doubleValues[i]);
+ }
+ break;
+ case STRING:
+ String[] stringValues = blockValSets[0].getStringValuesSV();
+ for (int i = 0; i < length; i++) {
+ hyperLogLog.offer(stringValues[i]);
+ }
+ break;
+ default:
+ throw new IllegalStateException(
+ "Illegal data type for DISTINCT_COUNT_HLL aggregation function:
" + valueType);
+ }
}
}
@Override
public void aggregateGroupBySV(int length, int[] groupKeyArray,
GroupByResultHolder groupByResultHolder,
BlockValSet... blockValSets) {
- FieldSpec.DataType valueType = blockValSets[0].getValueType();
+ DataType valueType = blockValSets[0].getValueType();
switch (valueType) {
case INT:
int[] intValues = blockValSets[0].getIntValuesSV();
@@ -149,13 +159,9 @@ public class DistinctCountHLLAggregationFunction
implements AggregationFunction<
case BYTES:
// Serialized HyperLogLog
byte[][] bytesValues = blockValSets[0].getBytesValuesSV();
- try {
- for (int i = 0; i < length; i++) {
- setValueForGroupKey(groupByResultHolder, groupKeyArray[i],
-
ObjectSerDeUtils.HYPER_LOG_LOG_SER_DE.deserialize(bytesValues[i]));
- }
- } catch (Exception e) {
- throw new RuntimeException("Caught exception while aggregating
HyperLogLog", e);
+ for (int i = 0; i < length; i++) {
+ setValueForGroupKey(groupByResultHolder, groupKeyArray[i],
+
ObjectSerDeUtils.HYPER_LOG_LOG_SER_DE.deserialize(bytesValues[i]));
}
break;
default:
@@ -166,7 +172,7 @@ public class DistinctCountHLLAggregationFunction implements
AggregationFunction<
@Override
public void aggregateGroupByMV(int length, int[][] groupKeysArray,
GroupByResultHolder groupByResultHolder,
BlockValSet... blockValSets) {
- FieldSpec.DataType valueType = blockValSets[0].getValueType();
+ DataType valueType = blockValSets[0].getValueType();
switch (valueType) {
case INT:
int[] intValues = blockValSets[0].getIntValuesSV();
@@ -201,13 +207,9 @@ public class DistinctCountHLLAggregationFunction
implements AggregationFunction<
case BYTES:
// Serialized HyperLogLog
byte[][] bytesValues = blockValSets[0].getBytesValuesSV();
- try {
- for (int i = 0; i < length; i++) {
- setValueForGroupKeys(groupByResultHolder, groupKeysArray[i],
-
ObjectSerDeUtils.HYPER_LOG_LOG_SER_DE.deserialize(bytesValues[i]));
- }
- } catch (Exception e) {
- throw new RuntimeException("Caught exception while aggregating
HyperLogLog", e);
+ for (int i = 0; i < length; i++) {
+ setValueForGroupKeys(groupByResultHolder, groupKeysArray[i],
+
ObjectSerDeUtils.HYPER_LOG_LOG_SER_DE.deserialize(bytesValues[i]));
}
break;
default:
@@ -237,10 +239,20 @@ public class DistinctCountHLLAggregationFunction
implements AggregationFunction<
@Override
public HyperLogLog merge(HyperLogLog intermediateResult1, HyperLogLog
intermediateResult2) {
+ // Can happen when aggregating serialized HyperLogLog with non-default
log2m
+ if (intermediateResult1.sizeof() != intermediateResult2.sizeof()) {
+ if (intermediateResult1.cardinality() == 0) {
+ return intermediateResult2;
+ } else {
+ Preconditions
+ .checkState(intermediateResult2.cardinality() == 0, "Cannot merge
HyperLogLogs of different sizes");
+ return intermediateResult1;
+ }
+ }
try {
intermediateResult1.addAll(intermediateResult2);
} catch (Exception e) {
- throw new RuntimeException("Caught exception while merging HyperLogLog",
e);
+ throw new RuntimeException("Caught exception while merging
HyperLogLogs", e);
}
return intermediateResult1;
}
@@ -279,10 +291,17 @@ public class DistinctCountHLLAggregationFunction
implements AggregationFunction<
* @param groupKey Group-key for which to set the value
* @param value HyperLogLog value for the group key
*/
- private static void setValueForGroupKey(GroupByResultHolder
groupByResultHolder, int groupKey, HyperLogLog value)
- throws CardinalityMergeException {
- HyperLogLog hyperLogLog = getHyperLogLog(groupByResultHolder, groupKey);
- hyperLogLog.addAll(value);
+ private static void setValueForGroupKey(GroupByResultHolder
groupByResultHolder, int groupKey, HyperLogLog value) {
+ HyperLogLog hyperLogLog = groupByResultHolder.getResult(groupKey);
+ if (hyperLogLog == null) {
+ groupByResultHolder.setValueForKey(groupKey, value);
+ } else {
+ try {
+ hyperLogLog.addAll(value);
+ } catch (Exception e) {
+ throw new RuntimeException("Caught exception while merging
HyperLogLogs", e);
+ }
+ }
}
/**
@@ -305,8 +324,8 @@ public class DistinctCountHLLAggregationFunction implements
AggregationFunction<
* @param groupKeys Group keys for which to set the value
* @param value HyperLogLog value to set
*/
- private static void setValueForGroupKeys(GroupByResultHolder
groupByResultHolder, int[] groupKeys, HyperLogLog value)
- throws CardinalityMergeException {
+ private static void setValueForGroupKeys(GroupByResultHolder
groupByResultHolder, int[] groupKeys,
+ HyperLogLog value) {
for (int groupKey : groupKeys) {
setValueForGroupKey(groupByResultHolder, groupKey, value);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
