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 <[email protected]> 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: [email protected] For additional commands, e-mail: [email protected]
