You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by sn...@apache.org on 2022/12/05 21:14:48 UTC
[pinot] branch add-variance-function created (now a0587bfba4)
This is an automated email from the ASF dual-hosted git repository.
snlee pushed a change to branch add-variance-function
in repository https://gitbox.apache.org/repos/asf/pinot.git
at a0587bfba4 Add Variance and Standard Deviation Aggregation Functions
This branch includes the following new commits:
new a0587bfba4 Add Variance and Standard Deviation Aggregation Functions
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[pinot] 01/01: Add Variance and Standard Deviation Aggregation Functions
Posted by sn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
snlee pushed a commit to branch add-variance-function
in repository https://gitbox.apache.org/repos/asf/pinot.git
commit a0587bfba4d46d89634e3500f9406cf323060919
Author: Seunghyun Lee <se...@startree.ai>
AuthorDate: Mon Dec 5 13:12:36 2022 -0800
Add Variance and Standard Deviation Aggregation Functions
This PR adds `VAR_POP()`, `VAR_SAMP()`, `STDDEV_POP()`,
`STDDEV_SAMP()` aggregation functions.
---
.../apache/pinot/core/common/ObjectSerDeUtils.java | 30 ++-
.../function/AggregationFunctionFactory.java | 8 +
.../function/VarianceAggregationFunction.java | 204 +++++++++++++++++
.../apache/pinot/queries/VarianceQueriesTest.java | 254 +++++++++++++++++++++
.../tests/OfflineClusterIntegrationTest.java | 2 +
.../local/customobject/CovarianceTuple.java | 3 +-
.../segment/local/customobject/VarianceTuple.java | 103 +++++++++
.../pinot/segment/spi/AggregationFunctionType.java | 4 +
8 files changed, 603 insertions(+), 5 deletions(-)
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/common/ObjectSerDeUtils.java b/pinot-core/src/main/java/org/apache/pinot/core/common/ObjectSerDeUtils.java
index e22a2e7fb0..4add03657e 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/common/ObjectSerDeUtils.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/common/ObjectSerDeUtils.java
@@ -72,6 +72,7 @@ import org.apache.pinot.segment.local.customobject.LongLongPair;
import org.apache.pinot.segment.local.customobject.MinMaxRangePair;
import org.apache.pinot.segment.local.customobject.QuantileDigest;
import org.apache.pinot.segment.local.customobject.StringLongPair;
+import org.apache.pinot.segment.local.customobject.VarianceTuple;
import org.apache.pinot.segment.local.utils.GeometrySerializer;
import org.apache.pinot.spi.utils.BigDecimalUtils;
import org.apache.pinot.spi.utils.ByteArray;
@@ -123,7 +124,9 @@ public class ObjectSerDeUtils {
FloatLongPair(29),
DoubleLongPair(30),
StringLongPair(31),
- CovarianceTuple(32);
+ CovarianceTuple(32),
+
+ VarianceTuple(33);
private final int _value;
@@ -205,6 +208,8 @@ public class ObjectSerDeUtils {
return ObjectType.StringLongPair;
} else if (value instanceof CovarianceTuple) {
return ObjectType.CovarianceTuple;
+ } else if (value instanceof VarianceTuple) {
+ return ObjectType.VarianceTuple;
} else {
throw new IllegalArgumentException("Unsupported type of value: " + value.getClass().getSimpleName());
}
@@ -445,7 +450,7 @@ public class ObjectSerDeUtils {
}
};
- public static final ObjectSerDe<CovarianceTuple> COVARIANCE_TUPLE_OBJECT_SER_DE = new ObjectSerDe<CovarianceTuple>() {
+ public static final ObjectSerDe<CovarianceTuple> COVARIANCE_TUPLE_OBJECT_SER_DE = new ObjectSerDe<>() {
@Override
public byte[] serialize(CovarianceTuple covarianceTuple) {
return covarianceTuple.toBytes();
@@ -462,6 +467,24 @@ public class ObjectSerDeUtils {
}
};
+ public static final ObjectSerDe<VarianceTuple> VARIANCE_TUPLE_OBJECT_SER_DE = new ObjectSerDe<>() {
+ @Override
+ public byte[] serialize(VarianceTuple varianceTuple) {
+ return varianceTuple.toBytes();
+ }
+
+ @Override
+ public VarianceTuple deserialize(byte[] bytes) {
+ return VarianceTuple.fromBytes(bytes);
+ }
+
+ @Override
+ public VarianceTuple deserialize(ByteBuffer byteBuffer) {
+ return VarianceTuple.fromByteBuffer(byteBuffer);
+ }
+ };
+
+
public static final ObjectSerDe<HyperLogLog> HYPER_LOG_LOG_SER_DE = new ObjectSerDe<HyperLogLog>() {
@Override
@@ -1191,7 +1214,8 @@ public class ObjectSerDeUtils {
FLOAT_LONG_PAIR_SER_DE,
DOUBLE_LONG_PAIR_SER_DE,
STRING_LONG_PAIR_SER_DE,
- COVARIANCE_TUPLE_OBJECT_SER_DE
+ COVARIANCE_TUPLE_OBJECT_SER_DE,
+ VARIANCE_TUPLE_OBJECT_SER_DE
};
//@formatter:on
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 3eb3a4f5ad..ee854d3e68 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
@@ -273,6 +273,14 @@ public class AggregationFunctionFactory {
return new CovarianceAggregationFunction(arguments, false);
case COVARSAMP:
return new CovarianceAggregationFunction(arguments, true);
+ case VARPOP:
+ return new VarianceAggregationFunction(firstArgument, false, false);
+ case VARSAMP:
+ return new VarianceAggregationFunction(firstArgument, true, false);
+ case STDDEVPOP:
+ return new VarianceAggregationFunction(firstArgument, false, true);
+ case STDDEVSAMP:
+ return new VarianceAggregationFunction(firstArgument, true, true);
default:
throw new IllegalArgumentException();
}
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/VarianceAggregationFunction.java b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/VarianceAggregationFunction.java
new file mode 100644
index 0000000000..aa66f8db40
--- /dev/null
+++ b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/VarianceAggregationFunction.java
@@ -0,0 +1,204 @@
+/**
+ * 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 com.google.common.base.Preconditions;
+import java.util.Map;
+import org.apache.pinot.common.request.context.ExpressionContext;
+import org.apache.pinot.common.utils.DataSchema;
+import org.apache.pinot.core.common.BlockValSet;
+import org.apache.pinot.core.query.aggregation.AggregationResultHolder;
+import org.apache.pinot.core.query.aggregation.ObjectAggregationResultHolder;
+import org.apache.pinot.core.query.aggregation.groupby.GroupByResultHolder;
+import org.apache.pinot.core.query.aggregation.groupby.ObjectGroupByResultHolder;
+import org.apache.pinot.segment.local.customobject.VarianceTuple;
+import org.apache.pinot.segment.spi.AggregationFunctionType;
+
+
+/**
+ * Aggregation function which computes Variance and Standard Deviation
+ *
+ * References
+ * - https://en.wikipedia.org/wiki/Algorithms_for_calculating_variance
+ * - http://cpsc.yale.edu/sites/default/files/files/tr222.pdf
+ */
+public class VarianceAggregationFunction extends BaseSingleInputAggregationFunction<VarianceTuple, Double> {
+ private static final double DEFAULT_FINAL_RESULT = Double.NEGATIVE_INFINITY;
+ protected final boolean _isSample;
+
+ protected final boolean _isStdDev;
+
+ public VarianceAggregationFunction(ExpressionContext expression, boolean isSample, boolean isStdDev) {
+ super(expression);
+ _isSample = isSample;
+ _isStdDev = isStdDev;
+ }
+
+ @Override
+ public AggregationFunctionType getType() {
+ if (_isSample) {
+ return (_isStdDev) ? AggregationFunctionType.STDDEVSAMP : AggregationFunctionType.VARSAMP;
+ }
+ return (_isStdDev) ? AggregationFunctionType.STDDEVPOP : AggregationFunctionType.VARPOP;
+ }
+
+ @Override
+ public AggregationResultHolder createAggregationResultHolder() {
+ return new ObjectAggregationResultHolder();
+ }
+
+ @Override
+ public GroupByResultHolder createGroupByResultHolder(int initialCapacity, int maxCapacity) {
+ return new ObjectGroupByResultHolder(initialCapacity, maxCapacity);
+ }
+
+ @Override
+ public void aggregate(int length, AggregationResultHolder aggregationResultHolder,
+ Map<ExpressionContext, BlockValSet> blockValSetMap) {
+ double[] values = getValSet(blockValSetMap, _expression);
+
+ long count = 0;
+ double sum = 0.0;
+ double variance = 0.0;
+ for (int i = 0; i < length; i++) {
+ count++;
+ sum += values[i];
+ if (count > 1) {
+ variance = computeIntermediateVariance(count, sum, variance, values[i]);
+ }
+ }
+ setAggregationResult(aggregationResultHolder, length, sum, variance);
+ }
+
+ private double computeIntermediateVariance(long count, double sum, double m2, double value) {
+ double t = count * value - sum;
+ m2 += (t * t) / (count * (count - 1));
+ return m2;
+ }
+
+ protected void setAggregationResult(AggregationResultHolder aggregationResultHolder, long count, double sum,
+ double m2) {
+ VarianceTuple varianceTuple = aggregationResultHolder.getResult();
+ if (varianceTuple == null) {
+ aggregationResultHolder.setValue(new VarianceTuple(count, sum, m2));
+ } else {
+ varianceTuple.apply(count, sum, m2);
+ }
+ }
+
+ protected void setGroupByResult(int groupKey, GroupByResultHolder groupByResultHolder, long count, double sum,
+ double m2) {
+ VarianceTuple varianceTuple = groupByResultHolder.getResult(groupKey);
+ if (varianceTuple == null) {
+ groupByResultHolder.setValueForKey(groupKey, new VarianceTuple(count, sum, m2));
+ } else {
+ varianceTuple.apply(count, sum, m2);
+ }
+ }
+
+ private double[] getValSet(Map<ExpressionContext, BlockValSet> blockValSetMap, ExpressionContext expression) {
+ BlockValSet blockValSet = blockValSetMap.get(expression);
+ // TODO: Add MV support for variance
+ Preconditions.checkState(blockValSet.isSingleValue(),
+ "Variance function currently only supports single-valued column");
+ switch (blockValSet.getValueType().getStoredType()) {
+ case INT:
+ case LONG:
+ case FLOAT:
+ case DOUBLE:
+ return blockValSet.getDoubleValuesSV();
+ default:
+ throw new IllegalStateException("Cannot compute variance for non-numeric type: " + blockValSet.getValueType());
+ }
+ }
+
+ @Override
+ public void aggregateGroupBySV(int length, int[] groupKeyArray, GroupByResultHolder groupByResultHolder,
+ Map<ExpressionContext, BlockValSet> blockValSetMap) {
+ double[] values = getValSet(blockValSetMap, _expression);
+ for (int i = 0; i < length; i++) {
+ setGroupByResult(groupKeyArray[i], groupByResultHolder, 1L, values[i], 0.0);
+ }
+ }
+
+ @Override
+ public void aggregateGroupByMV(int length, int[][] groupKeysArray, GroupByResultHolder groupByResultHolder,
+ Map<ExpressionContext, BlockValSet> blockValSetMap) {
+ double[] values = getValSet(blockValSetMap, _expression);
+ for (int i = 0; i < length; i++) {
+ for (int groupKey : groupKeysArray[i]) {
+ setGroupByResult(groupKey, groupByResultHolder, 1L, values[i], 0.0);
+ }
+ }
+ }
+
+ @Override
+ public VarianceTuple extractAggregationResult(AggregationResultHolder aggregationResultHolder) {
+ VarianceTuple varianceTuple = aggregationResultHolder.getResult();
+ if (varianceTuple == null) {
+ return new VarianceTuple(0L, 0.0, 0.0);
+ } else {
+ return varianceTuple;
+ }
+ }
+
+ @Override
+ public VarianceTuple extractGroupByResult(GroupByResultHolder groupByResultHolder, int groupKey) {
+ return groupByResultHolder.getResult(groupKey);
+ }
+
+ @Override
+ public VarianceTuple merge(VarianceTuple intermediateResult1, VarianceTuple intermediateResult2) {
+ intermediateResult1.apply(intermediateResult2);
+ return intermediateResult1;
+ }
+
+ @Override
+ public DataSchema.ColumnDataType getIntermediateResultColumnType() {
+ return DataSchema.ColumnDataType.OBJECT;
+ }
+
+ @Override
+ public DataSchema.ColumnDataType getFinalResultColumnType() {
+ return DataSchema.ColumnDataType.DOUBLE;
+ }
+
+ @Override
+ public Double extractFinalResult(VarianceTuple varianceTuple) {
+ if (varianceTuple == null) {
+ return null;
+ }
+ long count = varianceTuple.getCount();
+ if (count == 0L) {
+ return DEFAULT_FINAL_RESULT;
+ } else {
+ double variance = varianceTuple.getM2();
+ if (_isSample) {
+ if (count - 1 == 0L) {
+ return DEFAULT_FINAL_RESULT;
+ }
+ double sampleVar = variance / (count - 1);
+ return (_isStdDev) ? Math.sqrt(sampleVar) : sampleVar;
+ } else {
+ double popVar = variance / count;
+ return (_isStdDev) ? Math.sqrt(popVar) : popVar;
+ }
+ }
+ }
+}
diff --git a/pinot-core/src/test/java/org/apache/pinot/queries/VarianceQueriesTest.java b/pinot-core/src/test/java/org/apache/pinot/queries/VarianceQueriesTest.java
new file mode 100644
index 0000000000..d2543e4e7d
--- /dev/null
+++ b/pinot-core/src/test/java/org/apache/pinot/queries/VarianceQueriesTest.java
@@ -0,0 +1,254 @@
+package org.apache.pinot.queries;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Random;
+import org.apache.commons.io.FileUtils;
+import org.apache.commons.math3.stat.descriptive.moment.StandardDeviation;
+import org.apache.commons.math3.stat.descriptive.moment.Variance;
+import org.apache.commons.math3.util.Precision;
+import org.apache.pinot.common.response.broker.BrokerResponseNative;
+import org.apache.pinot.core.operator.blocks.results.AggregationResultsBlock;
+import org.apache.pinot.core.operator.query.AggregationOperator;
+import org.apache.pinot.segment.local.customobject.VarianceTuple;
+import org.apache.pinot.segment.local.indexsegment.immutable.ImmutableSegmentLoader;
+import org.apache.pinot.segment.local.segment.creator.impl.SegmentIndexCreationDriverImpl;
+import org.apache.pinot.segment.local.segment.readers.GenericRowRecordReader;
+import org.apache.pinot.segment.spi.ImmutableSegment;
+import org.apache.pinot.segment.spi.IndexSegment;
+import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.data.readers.GenericRow;
+import org.apache.pinot.spi.utils.ReadMode;
+import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertTrue;
+
+
+public class VarianceQueriesTest extends BaseQueriesTest {
+
+ private static final File INDEX_DIR = new File(FileUtils.getTempDirectory(), "VarianceQueriesTest");
+
+ private static final String RAW_TABLE_NAME = "testTable";
+ private static final String SEGMENT_NAME = "testSegment";
+
+ private static final int NUM_RECORDS = 2000;
+ private static final int NUM_GROUPS = 10;
+ private static final int MAX_VALUE = 500;
+ private static final double RELATIVE_EPSILON = 0.0001;
+ private static final double DELTA = 0.0001;
+
+ private static final String INT_COLUMN = "intColumn";
+ private static final String LONG_COLUMN = "longColumn";
+ private static final String FLOAT_COLUMN = "floatColumn";
+ private static final String DOUBLE_COLUMN = "doubleColumn";
+ private static final String GROUP_BY_COLUMN = "groupByColumn";
+
+ private static final Schema SCHEMA =
+ new Schema.SchemaBuilder().addSingleValueDimension(INT_COLUMN, FieldSpec.DataType.INT)
+ .addSingleValueDimension(LONG_COLUMN, FieldSpec.DataType.LONG)
+ .addSingleValueDimension(FLOAT_COLUMN, FieldSpec.DataType.FLOAT)
+ .addSingleValueDimension(DOUBLE_COLUMN, FieldSpec.DataType.DOUBLE)
+ .addSingleValueDimension(GROUP_BY_COLUMN, FieldSpec.DataType.DOUBLE).build();
+
+ private static final TableConfig TABLE_CONFIG =
+ new TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME).build();
+
+ private IndexSegment _indexSegment;
+ private List<IndexSegment> _indexSegments;
+
+ int[] _intValues = new int[NUM_RECORDS];
+ long[] _longValues = new long[NUM_RECORDS];
+ float[] _floatValues = new float[NUM_RECORDS];
+ double[] _doubleValues = new double[NUM_RECORDS];
+
+ @Override
+ protected String getFilter() {
+ // filter out half of the rows based on group id
+ return " WHERE groupByColumn < " + (NUM_GROUPS / 2);
+ }
+
+ @Override
+ protected IndexSegment getIndexSegment() {
+ return _indexSegment;
+ }
+
+ @Override
+ protected List<IndexSegment> getIndexSegments() {
+ return _indexSegments;
+ }
+
+ @BeforeClass
+ public void setUp()
+ throws Exception {
+ FileUtils.deleteDirectory(INDEX_DIR);
+ Random random = new Random();
+ List<GenericRow> records = new ArrayList<>(NUM_RECORDS);
+
+ for (int i = 0; i < NUM_RECORDS; i++) {
+ GenericRow record = new GenericRow();
+ _intValues[i] = -MAX_VALUE + random.nextInt() * 2 * MAX_VALUE;
+ _longValues[i] = -MAX_VALUE + random.nextLong() * 2 * MAX_VALUE;
+ _floatValues[i] = -MAX_VALUE + random.nextFloat() * 2 * MAX_VALUE;
+ _doubleValues[i] = -MAX_VALUE + random.nextDouble() * 2 * MAX_VALUE;
+
+ record.putValue(INT_COLUMN, _intValues[i]);
+ record.putValue(LONG_COLUMN, _longValues[i]);
+ record.putValue(FLOAT_COLUMN, _floatValues[i]);
+ record.putValue(DOUBLE_COLUMN, _doubleValues[i]);
+ record.putValue(GROUP_BY_COLUMN, Math.floor(i / (NUM_RECORDS / NUM_GROUPS)));
+ records.add(record);
+ }
+
+ SegmentGeneratorConfig segmentGeneratorConfig = new SegmentGeneratorConfig(TABLE_CONFIG, SCHEMA);
+ segmentGeneratorConfig.setTableName(RAW_TABLE_NAME);
+ segmentGeneratorConfig.setSegmentName(SEGMENT_NAME);
+ segmentGeneratorConfig.setOutDir(INDEX_DIR.getPath());
+
+ SegmentIndexCreationDriverImpl driver = new SegmentIndexCreationDriverImpl();
+ driver.init(segmentGeneratorConfig, new GenericRowRecordReader(records));
+ driver.build();
+
+ ImmutableSegment immutableSegment = ImmutableSegmentLoader.load(new File(INDEX_DIR, SEGMENT_NAME), ReadMode.mmap);
+ _indexSegment = immutableSegment;
+ _indexSegments = Arrays.asList(immutableSegment, immutableSegment);
+ }
+
+ @Test
+ public void testVariancePopulationAggregationOnly() {
+ Random random = new Random();
+ for (int i = 0; i < 100; i++) {
+ System.out.println(-MAX_VALUE + random.nextDouble() * 2 * MAX_VALUE);
+ }
+ String query = "SELECT VAR_POP(intColumn), VAR_POP(longColumn), VAR_POP(floatColumn), VAR_POP(doubleColumn),"
+ + "VAR_SAMP(intColumn), VAR_SAMP(longColumn), VAR_SAMP(floatColumn), VAR_SAMP(doubleColumn) FROM testTable";
+ AggregationOperator aggregationOperator = getOperator(query);
+ AggregationResultsBlock resultsBlock = aggregationOperator.nextBlock();
+ QueriesTestUtils.testInnerSegmentExecutionStatistics(aggregationOperator.getExecutionStatistics(), NUM_RECORDS, 0,
+ NUM_RECORDS * 4, NUM_RECORDS);
+ List<Object> aggregationResult = resultsBlock.getResults();
+
+ // ADD
+ for (Object result : aggregationResult) {
+ VarianceTuple varianceTuple = (VarianceTuple) result;
+ System.out.println(varianceTuple.getCount() + " " + varianceTuple.getSum() + " " + varianceTuple.getM2());
+ }
+
+ Variance expectedVarIntPop = new Variance(false);
+ Variance expectedVarLongPop = new Variance(false);
+ Variance expectedVarFloatPop = new Variance(false);
+ Variance expectedVarDoublePop = new Variance(false);
+ Variance expectedVarIntSamp = new Variance(true);
+ Variance expectedVarLongSamp = new Variance(true);
+ Variance expectedVarFloatSamp = new Variance(true);
+ Variance expectedVarDoubleSamp = new Variance(true);
+
+ // Compute the expected variance result
+ for (int i = 0; i < NUM_RECORDS * 4; i++) {
+ int pos = i % NUM_RECORDS;
+ expectedVarIntPop.increment(_intValues[pos]);
+ expectedVarLongPop.increment(_longValues[pos]);
+ expectedVarFloatPop.increment(_floatValues[pos]);
+ expectedVarDoublePop.increment(_doubleValues[pos]);
+ expectedVarIntSamp.increment(_intValues[pos]);
+ expectedVarLongSamp.increment(_longValues[pos]);
+ expectedVarFloatSamp.increment(_floatValues[pos]);
+ expectedVarDoubleSamp.increment(_doubleValues[pos]);
+ }
+
+ BrokerResponseNative brokerResponse = getBrokerResponse(query);
+ brokerResponse.getResultTable();
+ Object[] results = brokerResponse.getResultTable().getRows().get(0);
+ assertTrue(
+ Precision.equalsWithRelativeTolerance((double) results[0], expectedVarIntPop.getResult(), RELATIVE_EPSILON));
+ assertTrue(
+ Precision.equalsWithRelativeTolerance((double) results[1], expectedVarLongPop.getResult(), RELATIVE_EPSILON));
+ assertTrue(
+ Precision.equalsWithRelativeTolerance((double) results[2], expectedVarFloatPop.getResult(), RELATIVE_EPSILON));
+ assertTrue(
+ Precision.equalsWithRelativeTolerance((double) results[3], expectedVarDoublePop.getResult(), RELATIVE_EPSILON));
+ assertTrue(
+ Precision.equalsWithRelativeTolerance((double) results[4], expectedVarIntSamp.getResult(), RELATIVE_EPSILON));
+ assertTrue(
+ Precision.equalsWithRelativeTolerance((double) results[5], expectedVarLongSamp.getResult(), RELATIVE_EPSILON));
+ assertTrue(
+ Precision.equalsWithRelativeTolerance((double) results[6], expectedVarFloatSamp.getResult(), RELATIVE_EPSILON));
+ assertTrue(Precision.equalsWithRelativeTolerance((double) results[7], expectedVarDoubleSamp.getResult(),
+ RELATIVE_EPSILON));
+ }
+
+ @Test
+ public void testStandardDeviationAggregationOnly() {
+ Random random = new Random();
+ for (int i = 0; i < 100; i++) {
+ System.out.println(-MAX_VALUE + random.nextDouble() * 2 * MAX_VALUE);
+ }
+ String query =
+ "SELECT STDDEV_POP(intColumn), STDDEV_POP(longColumn), STDDEV_POP(floatColumn), STDDEV_POP(doubleColumn),"
+ + "STDDEV_SAMP(intColumn), STDDEV_SAMP(longColumn), STDDEV_SAMP(floatColumn), STDDEV_SAMP(doubleColumn) "
+ + "FROM testTable";
+ AggregationOperator aggregationOperator = getOperator(query);
+ AggregationResultsBlock resultsBlock = aggregationOperator.nextBlock();
+ QueriesTestUtils.testInnerSegmentExecutionStatistics(aggregationOperator.getExecutionStatistics(), NUM_RECORDS, 0,
+ NUM_RECORDS * 4, NUM_RECORDS);
+ List<Object> aggregationResult = resultsBlock.getResults();
+
+ // ADD
+ for (Object result : aggregationResult) {
+ VarianceTuple varianceTuple = (VarianceTuple) result;
+ System.out.println(varianceTuple.getCount() + " " + varianceTuple.getSum() + " " + varianceTuple.getM2());
+ }
+
+ StandardDeviation expectedStdDevIntPop = new StandardDeviation(false);
+ StandardDeviation expectedStdDevLongPop = new StandardDeviation(false);
+ StandardDeviation expectedStdDevFloatPop = new StandardDeviation(false);
+ StandardDeviation expectedStdDevDoublePop = new StandardDeviation(false);
+ StandardDeviation expectedStdDevIntSamp = new StandardDeviation(true);
+ StandardDeviation expectedStdDevLongSamp = new StandardDeviation(true);
+ StandardDeviation expectedStdDevFloatSamp = new StandardDeviation(true);
+ StandardDeviation expectedStdDevDoubleSamp = new StandardDeviation(true);
+
+ // Compute the expected variance result
+ for (int i = 0; i < NUM_RECORDS * 4; i++) {
+ int pos = i % NUM_RECORDS;
+ expectedStdDevIntPop.increment(_intValues[pos]);
+ expectedStdDevLongPop.increment(_longValues[pos]);
+ expectedStdDevFloatPop.increment(_floatValues[pos]);
+ expectedStdDevDoublePop.increment(_doubleValues[pos]);
+ expectedStdDevIntSamp.increment(_intValues[pos]);
+ expectedStdDevLongSamp.increment(_longValues[pos]);
+ expectedStdDevFloatSamp.increment(_floatValues[pos]);
+ expectedStdDevDoubleSamp.increment(_doubleValues[pos]);
+ }
+
+ BrokerResponseNative brokerResponse = getBrokerResponse(query);
+ brokerResponse.getResultTable();
+ Object[] results = brokerResponse.getResultTable().getRows().get(0);
+
+ assertTrue(
+ Precision.equalsWithRelativeTolerance((double) results[0], expectedStdDevIntPop.getResult(), RELATIVE_EPSILON));
+ assertTrue(Precision.equalsWithRelativeTolerance((double) results[1], expectedStdDevLongPop.getResult(),
+ RELATIVE_EPSILON));
+ assertTrue(Precision.equalsWithRelativeTolerance((double) results[2], expectedStdDevFloatPop.getResult(),
+ RELATIVE_EPSILON));
+ assertTrue(Precision.equalsWithRelativeTolerance((double) results[3], expectedStdDevDoublePop.getResult(),
+ RELATIVE_EPSILON));
+ assertTrue(Precision.equalsWithRelativeTolerance((double) results[4], expectedStdDevIntSamp.getResult(),
+ RELATIVE_EPSILON));
+ assertTrue(Precision.equalsWithRelativeTolerance((double) results[5], expectedStdDevLongSamp.getResult(),
+ RELATIVE_EPSILON));
+ assertTrue(Precision.equalsWithRelativeTolerance((double) results[6], expectedStdDevFloatSamp.getResult(),
+ RELATIVE_EPSILON));
+ assertTrue(Precision.equalsWithRelativeTolerance((double) results[7], expectedStdDevDoubleSamp.getResult(),
+ RELATIVE_EPSILON));
+ }
+
+
+}
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/OfflineClusterIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/OfflineClusterIntegrationTest.java
index bcecc3696a..4768340bc8 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/OfflineClusterIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/OfflineClusterIntegrationTest.java
@@ -220,6 +220,8 @@ public class OfflineClusterIntegrationTest extends BaseClusterIntegrationTestSet
// Wait for all documents loaded
waitForAllDocsLoaded(600_000L);
+
+ Thread.sleep(10000000000L);
}
protected void startBrokers()
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/CovarianceTuple.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/CovarianceTuple.java
index cf705ebc45..6c1012c58d 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/CovarianceTuple.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/CovarianceTuple.java
@@ -36,8 +36,7 @@ public class CovarianceTuple implements Comparable<CovarianceTuple> {
public CovarianceTuple(double sumX, double sumY, double sumXY, long count) {
_sumX = sumX;
- _sumY = sumY;
- _sumXY = sumXY;
+ _sumY = sumY;_sumXY = sumXY;
_count = count;
}
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/VarianceTuple.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/VarianceTuple.java
new file mode 100644
index 0000000000..806e9d06d1
--- /dev/null
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/VarianceTuple.java
@@ -0,0 +1,103 @@
+/**
+ * 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.customobject;
+
+import java.nio.ByteBuffer;
+import javax.annotation.Nonnull;
+
+
+public class VarianceTuple implements Comparable<VarianceTuple> {
+ private long _count;
+ private double _sum;
+ private double _m2;
+
+ public VarianceTuple(long count, double sum, double m2) {
+ _count = count;
+ _sum = sum;
+ _m2 = m2;
+ }
+
+ public void apply(long count, double sum, double m2) {
+ double delta = (sum / count) - (_sum / _count);
+ _m2 += m2 + delta * delta * count * _count / (count + _count);
+ _count += count;
+ _sum += sum;
+ }
+
+ public void apply(@Nonnull VarianceTuple varianceTuple) {
+ double delta = (varianceTuple._sum / varianceTuple._count) - (_sum / _count);
+ _m2 += varianceTuple._m2 + delta * delta * varianceTuple._count * _count / (varianceTuple._count + _count);
+ _count += varianceTuple._count;
+ _sum += varianceTuple._sum;
+ }
+
+ public long getCount() {
+ return _count;
+ }
+
+ public double getSum() {
+ return _sum;
+ }
+
+ public double getM2() {
+ return _m2;
+ }
+
+ @Nonnull
+ public byte[] toBytes() {
+ ByteBuffer byteBuffer = ByteBuffer.allocate(Double.BYTES * 2 + Long.BYTES);
+ byteBuffer.putLong(_count);
+ byteBuffer.putDouble(_sum);
+ byteBuffer.putDouble(_m2);
+ return byteBuffer.array();
+ }
+
+ @Nonnull
+ public static VarianceTuple fromBytes(byte[] bytes) {
+ return fromByteBuffer(ByteBuffer.wrap(bytes));
+ }
+
+ @Nonnull
+ public static VarianceTuple fromByteBuffer(ByteBuffer byteBuffer) {
+ return new VarianceTuple(byteBuffer.getLong(), byteBuffer.getDouble(), byteBuffer.getDouble());
+ }
+
+ @Override
+ public int compareTo(@Nonnull VarianceTuple varianceTuple) {
+ if (_count == 0) {
+ if (varianceTuple._count == 0) {
+ return 0;
+ } else {
+ return -1;
+ }
+ } else {
+ if (varianceTuple._count == 0) {
+ return 1;
+ } else {
+ if (_m2 > varianceTuple._m2) {
+ return 1;
+ }
+ if (_m2 < varianceTuple._m2) {
+ return -1;
+ }
+ return 0;
+ }
+ }
+ }
+}
diff --git a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/AggregationFunctionType.java b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/AggregationFunctionType.java
index 48a2199918..c3ff9b24a0 100644
--- a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/AggregationFunctionType.java
+++ b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/AggregationFunctionType.java
@@ -59,6 +59,10 @@ public enum AggregationFunctionType {
HISTOGRAM("histogram"),
COVARPOP("covarPop"),
COVARSAMP("covarSamp"),
+ VARPOP("varPop"),
+ VARSAMP("varSamp"),
+ STDDEVPOP("stdDevPop"),
+ STDDEVSAMP("stdDevSamp"),
// Geo aggregation functions
STUNION("STUnion"),
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org