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 c07421691d3 Keep HISTOGRAM and SUM_ARRAY intermediate results as
primitive arrays (#20108)
c07421691d3 is described below
commit c07421691d3482e125331aaeece62375c79e3f06
Author: Xiaotian (Jackie) Jiang <[email protected]>
AuthorDate: Tue Nov 10 20:32:43 2026 -0800
Keep HISTOGRAM and SUM_ARRAY intermediate results as primitive arrays
(#20108)
---
.../apache/pinot/common/utils/ArrayListUtils.java | 66 -------------
.../pinot/common/utils/ArrayListUtilsTest.java | 74 ---------------
.../function/HistogramAggregationFunction.java | 78 +++++++---------
.../array/SumArrayDoubleAggregationFunction.java | 69 +++++++-------
.../array/SumArrayLongAggregationFunction.java | 69 +++++++-------
.../aggregation/utils/DoubleVectorOpUtils.java | 69 --------------
.../IntermediateResultByteBufferSerDeTest.java | 6 +-
.../function/ValueAggregationNullHandlingTest.java | 79 +++++++++++++++-
.../apache/pinot/queries/HistogramQueriesTest.java | 57 ++++++------
.../segment/local/utils/ObjectSerDeUtils.java | 102 ++++++++++++++++-----
.../segment/local/utils/ObjectSerDeUtilsTest.java | 43 +++++++++
11 files changed, 328 insertions(+), 384 deletions(-)
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/utils/ArrayListUtils.java
b/pinot-common/src/main/java/org/apache/pinot/common/utils/ArrayListUtils.java
deleted file mode 100644
index e1fa7ed640a..00000000000
---
a/pinot-common/src/main/java/org/apache/pinot/common/utils/ArrayListUtils.java
+++ /dev/null
@@ -1,66 +0,0 @@
-/**
- * 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.common.utils;
-
-import it.unimi.dsi.fastutil.doubles.DoubleArrayList;
-import it.unimi.dsi.fastutil.longs.LongArrayList;
-
-
-/// Utility class for [LongArrayList] and [DoubleArrayList].
-public class ArrayListUtils {
- private ArrayListUtils() {
- }
-
- /// Best effort extract the given [LongArrayList] to a long array without
copying the elements.
- /// The [LongArrayList#elements()] returned long array may be longer than
the actual size of the
- /// [LongArrayList], and the actual size of the [LongArrayList] can be
retrieved using
- /// [LongArrayList#size()].
- /// This method checks the length of the returned long array and returns the
same if it is equal to the size of the
- /// [LongArrayList], otherwise, it copies the elements to a new long array
and returns it.
- ///
- /// Use this method only if you are sure that the returned long array will
not be modified.
- ///
- /// Otherwise, use [LongArrayList#toLongArray()].
- ///
- /// @param longArrayList Input [LongArrayList]
- /// @return Best effort extracted long array without copying the elements
- public static long[] toLongArray(LongArrayList longArrayList) {
- long[] longArrayListElements = longArrayList.elements();
- return longArrayListElements.length == longArrayList.size() ?
longArrayListElements : longArrayList.toLongArray();
- }
-
- /// Best effort extract the given [DoubleArrayList] to a double array
without copying the elements.
- /// The [DoubleArrayList#elements()] returned double array may be longer
than the actual size of the
- /// [DoubleArrayList], and the actual size of the [DoubleArrayList] can be
retrieved using
- /// [DoubleArrayList#size()].
- /// This method checks the length of the returned double array and returns
the same if it is equal to the size of the
- /// [DoubleArrayList], otherwise, it copies the elements to a new double
array and returns it.
- ///
- /// Use this method only if you are sure that the returned double array will
not be modified.
- ///
- /// Otherwise, use [DoubleArrayList#toDoubleArray()].
- ///
- /// @param doubleArrayList Input [DoubleArrayList]
- /// @return Best effort extracted double array without copying the elements
- public static double[] toDoubleArray(DoubleArrayList doubleArrayList) {
- double[] doubleArrayListElements = doubleArrayList.elements();
- return doubleArrayListElements.length == doubleArrayList.size() ?
doubleArrayListElements
- : doubleArrayList.toDoubleArray();
- }
-}
diff --git
a/pinot-common/src/test/java/org/apache/pinot/common/utils/ArrayListUtilsTest.java
b/pinot-common/src/test/java/org/apache/pinot/common/utils/ArrayListUtilsTest.java
deleted file mode 100644
index 5adf3b084f6..00000000000
---
a/pinot-common/src/test/java/org/apache/pinot/common/utils/ArrayListUtilsTest.java
+++ /dev/null
@@ -1,74 +0,0 @@
-/**
- * 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.common.utils;
-
-import it.unimi.dsi.fastutil.doubles.DoubleArrayList;
-import it.unimi.dsi.fastutil.longs.LongArrayList;
-import org.testng.annotations.Test;
-
-import static org.testng.Assert.assertEquals;
-
-
-public class ArrayListUtilsTest {
- @Test
- public void testToLongArray() {
- // Test empty list
- LongArrayList longArrayList = new LongArrayList();
- long[] longArray = ArrayListUtils.toLongArray(longArrayList);
- assertEquals(longArray.length, 0);
-
- // Test list with one element
- longArrayList.add(1L);
- longArray = ArrayListUtils.toLongArray(longArrayList);
- assertEquals(longArray.length, 1);
- assertEquals(longArray[0], 1L);
-
- // Test list with multiple elements
- longArrayList.add(2L);
- longArrayList.add(3L);
- longArray = ArrayListUtils.toLongArray(longArrayList);
- assertEquals(longArray.length, 3);
- assertEquals(longArray[0], 1L);
- assertEquals(longArray[1], 2L);
- assertEquals(longArray[2], 3L);
- }
-
- @Test
- public void testToDoubleArray() {
- // Test empty list
- DoubleArrayList doubleArrayList = new DoubleArrayList();
- double[] doubleArray = ArrayListUtils.toDoubleArray(doubleArrayList);
- assertEquals(doubleArray.length, 0);
-
- // Test list with one element
- doubleArrayList.add(1.0);
- doubleArray = ArrayListUtils.toDoubleArray(doubleArrayList);
- assertEquals(doubleArray.length, 1);
- assertEquals(doubleArray[0], 1.0);
-
- // Test list with multiple elements
- doubleArrayList.add(2.0);
- doubleArrayList.add(3.0);
- doubleArray = ArrayListUtils.toDoubleArray(doubleArrayList);
- assertEquals(doubleArray.length, 3);
- assertEquals(doubleArray[0], 1.0);
- assertEquals(doubleArray[1], 2.0);
- assertEquals(doubleArray[2], 3.0);
- }
-}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/HistogramAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/HistogramAggregationFunction.java
index 5e581f61513..26a608a4724 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/HistogramAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/HistogramAggregationFunction.java
@@ -19,21 +19,18 @@
package org.apache.pinot.core.query.aggregation.function;
import com.google.common.base.Preconditions;
-import it.unimi.dsi.fastutil.doubles.DoubleArrayList;
import java.math.BigDecimal;
import java.util.List;
import java.util.Map;
import javax.annotation.Nullable;
import org.apache.pinot.common.CustomObject;
import org.apache.pinot.common.request.context.ExpressionContext;
-import org.apache.pinot.common.utils.ArrayListUtils;
import org.apache.pinot.common.utils.DataSchema.ColumnDataType;
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.core.query.aggregation.utils.DoubleVectorOpUtils;
import org.apache.pinot.segment.local.utils.ObjectSerDeUtils;
import org.apache.pinot.segment.spi.AggregationFunctionType;
import org.apache.pinot.spi.utils.ArrayCopyUtils;
@@ -43,7 +40,7 @@ import org.apache.pinot.spi.utils.ArrayCopyUtils;
/// usage example:
/// `Histogram(columnName, ARRAY\[0,1,10,100\])` to specify bins \[0,1),
\[1,10), \[10,1000\] or
/// `Histogram(columnName, 0, 1000, 10)` to specify 10 equal-length bins
\[0,100), \[100,200), ..., \[900,1000\]
-public class HistogramAggregationFunction extends
BaseSingleInputAggregationFunction<DoubleArrayList, double[]> {
+public class HistogramAggregationFunction extends
BaseSingleInputAggregationFunction<double[], double[]> {
private static final int DOUBLE_ARRAY_LIST_TYPE = 3;
private static final String ARRAY_CONSTRUCTOR = "arrayvalueconstructor";
@@ -204,26 +201,28 @@ public class HistogramAggregationFunction extends
BaseSingleInputAggregationFunc
@Nullable
@Override
- public DoubleArrayList extractAggregationResult(AggregationResultHolder
aggregationResultHolder) {
- DoubleArrayList histogram = aggregationResultHolder.getResult();
+ public double[] extractAggregationResult(AggregationResultHolder
aggregationResultHolder) {
+ double[] histogram = aggregationResultHolder.getResult();
if (histogram != null) {
return histogram;
}
// With the option disabled an untouched holder still renders the empty
accumulator, which is the
// intermediate this mode has always emitted; with it enabled the null is
the signal that nothing was
// aggregated.
- return _nullHandlingEnabled ? null :
DoubleVectorOpUtils.createAndInitialize(getNumBins());
+ return _nullHandlingEnabled ? null : new double[getNumBins()];
}
@Nullable
@Override
- public DoubleArrayList extractGroupByResult(GroupByResultHolder
groupByResultHolder, int groupKey) {
+ public double[] extractGroupByResult(GroupByResultHolder
groupByResultHolder, int groupKey) {
return groupByResultHolder.getResult(groupKey);
}
@Override
- public DoubleArrayList merge(DoubleArrayList intermediateResult1,
DoubleArrayList intermediateResult2) {
- DoubleVectorOpUtils.vectorAdd(intermediateResult1, intermediateResult2);
+ public double[] merge(double[] intermediateResult1, double[]
intermediateResult2) {
+ for (int i = 0; i < intermediateResult1.length; i++) {
+ intermediateResult1[i] += intermediateResult2[i];
+ }
return intermediateResult1;
}
@@ -233,13 +232,13 @@ public class HistogramAggregationFunction extends
BaseSingleInputAggregationFunc
}
@Override
- public SerializedResult serializeIntermediateResult(DoubleArrayList doubles)
{
- return new SerializedValue<>(DOUBLE_ARRAY_LIST_TYPE, doubles,
ObjectSerDeUtils.DOUBLE_ARRAY_LIST_SER_DE);
+ public SerializedResult serializeIntermediateResult(double[] histogram) {
+ return new SerializedValue<>(DOUBLE_ARRAY_LIST_TYPE, histogram,
ObjectSerDeUtils.DOUBLE_ARRAY_SER_DE);
}
@Override
- public DoubleArrayList deserializeIntermediateResult(CustomObject
customObject) {
- return
ObjectSerDeUtils.DOUBLE_ARRAY_LIST_SER_DE.deserialize(customObject.getBuffer());
+ public double[] deserializeIntermediateResult(CustomObject customObject) {
+ return
ObjectSerDeUtils.DOUBLE_ARRAY_SER_DE.deserialize(customObject.getBuffer());
}
@Override
@@ -249,22 +248,18 @@ public class HistogramAggregationFunction extends
BaseSingleInputAggregationFunc
@Nullable
@Override
- public double[] extractFinalResult(@Nullable DoubleArrayList
doubleArrayList) {
- if (doubleArrayList == null) {
+ public double[] extractFinalResult(@Nullable double[] histogram) {
+ if (histogram == null) {
// A null intermediate result means nothing was aggregated. With null
handling enabled the histogram of nothing
// is NULL; with it disabled it is the all-zero histogram, which is the
answer this mode has always given.
return _nullHandlingEnabled ? null : new double[getNumBins()];
}
- return ArrayListUtils.toDoubleArray(doubleArrayList);
+ return histogram;
}
@Override
public double[] mergeFinalResult(double[] finalResult1, double[]
finalResult2) {
- double[] mergedResult = finalResult1.clone();
- for (int i = 0; i < mergedResult.length; i++) {
- mergedResult[i] += finalResult2[i];
- }
- return mergedResult;
+ return merge(finalResult1.clone(), finalResult2);
}
@Override
@@ -279,7 +274,8 @@ public class HistogramAggregationFunction extends
BaseSingleInputAggregationFunc
}
private void aggregateSV(int length, AggregationResultHolder
aggregationResultHolder, BlockValSet blockValSet) {
- double[] histogram = new double[getNumBins()];
+ double[] aggregatedHistogram = aggregationResultHolder.getResult();
+ double[] histogram = aggregatedHistogram != null ? aggregatedHistogram :
new double[getNumBins()];
int numRows;
switch (blockValSet.getValueType().getStoredType()) {
case INT: {
@@ -336,16 +332,17 @@ public class HistogramAggregationFunction extends
BaseSingleInputAggregationFunc
throw new IllegalStateException("Cannot compute histogram for
non-numeric type: "
+ blockValSet.getValueType());
}
- // The histogram is published only when a row reached it, so a block with
no non-null row leaves the holder
- // untouched and extractFinalResult sees the null that means nothing was
aggregated. It is published once rather
- // than per range, because the buffer accumulates across ranges and adding
it again would recount earlier rows.
- if (numRows > 0) {
- setAggregationResult(aggregationResultHolder, histogram);
+ // Rows are counted into the aggregated histogram when there is one. A new
histogram is published only when a row
+ // reached it, so a block with no non-null row leaves the holder untouched
and extractFinalResult sees the null
+ // that means nothing was aggregated.
+ if (aggregatedHistogram == null && numRows > 0) {
+ aggregationResultHolder.setValue(histogram);
}
}
private void aggregateMV(int length, AggregationResultHolder
aggregationResultHolder, BlockValSet blockValSet) {
- double[] histogram = new double[getNumBins()];
+ double[] aggregatedHistogram = aggregationResultHolder.getResult();
+ double[] histogram = aggregatedHistogram != null ? aggregatedHistogram :
new double[getNumBins()];
int numRows;
switch (blockValSet.getValueType().getStoredType()) {
case INT: {
@@ -412,8 +409,8 @@ public class HistogramAggregationFunction extends
BaseSingleInputAggregationFunc
throw new IllegalStateException("Cannot compute histogram for
non-numeric type: "
+ blockValSet.getValueType());
}
- if (numRows > 0) {
- setAggregationResult(aggregationResultHolder, histogram);
+ if (aggregatedHistogram == null && numRows > 0) {
+ aggregationResultHolder.setValue(histogram);
}
}
@@ -707,22 +704,13 @@ public class HistogramAggregationFunction extends
BaseSingleInputAggregationFunc
protected void setGroupByResult(int groupKey, GroupByResultHolder
groupByResultHolder, double val) {
int binID = getBinId(val);
- DoubleArrayList byResultHolderResult =
groupByResultHolder.getResult(groupKey);
- if (byResultHolderResult == null) {
- byResultHolderResult =
DoubleVectorOpUtils.createAndInitialize(getNumBins());
- groupByResultHolder.setValueForKey(groupKey, byResultHolderResult);
+ double[] histogram = groupByResultHolder.getResult(groupKey);
+ if (histogram == null) {
+ histogram = new double[getNumBins()];
+ groupByResultHolder.setValueForKey(groupKey, histogram);
}
if (binID != INVALID_BIN) {
- DoubleVectorOpUtils.incrementElementByOne(byResultHolderResult, binID);
- }
- }
-
- protected void setAggregationResult(AggregationResultHolder
aggregationResultHolder, double[] histogram) {
- DoubleArrayList aggregatedHistogram = aggregationResultHolder.getResult();
- if (aggregatedHistogram == null) {
-
aggregationResultHolder.setValue(DoubleVectorOpUtils.createAndInitialize(histogram));
- } else {
- DoubleVectorOpUtils.vectorAdd(aggregatedHistogram, histogram);
+ histogram[binID]++;
}
}
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/array/SumArrayDoubleAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/array/SumArrayDoubleAggregationFunction.java
index 7edf6e371c1..e0f786fb9a3 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/array/SumArrayDoubleAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/array/SumArrayDoubleAggregationFunction.java
@@ -18,14 +18,12 @@
*/
package org.apache.pinot.core.query.aggregation.function.array;
-import it.unimi.dsi.fastutil.doubles.DoubleArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import javax.annotation.Nullable;
import org.apache.pinot.common.CustomObject;
import org.apache.pinot.common.request.context.ExpressionContext;
-import org.apache.pinot.common.utils.ArrayListUtils;
import org.apache.pinot.common.utils.DataSchema;
import org.apache.pinot.core.common.BlockValSet;
import org.apache.pinot.core.query.aggregation.AggregationResultHolder;
@@ -37,8 +35,7 @@ import org.apache.pinot.segment.local.utils.ObjectSerDeUtils;
import org.apache.pinot.segment.spi.AggregationFunctionType;
-public class SumArrayDoubleAggregationFunction
- extends BaseSingleInputAggregationFunction<DoubleArrayList, double[]> {
+public class SumArrayDoubleAggregationFunction extends
BaseSingleInputAggregationFunction<double[], double[]> {
private static final int DOUBLE_ARRAY_LIST_TYPE = 3;
public SumArrayDoubleAggregationFunction(List<ExpressionContext> arguments,
boolean nullHandlingEnabled) {
@@ -68,14 +65,11 @@ public class SumArrayDoubleAggregationFunction
// The accumulator is created inside the range, so a block with no
non-null row leaves the holder untouched and
// extractFinalResult sees the null that means nothing was aggregated
forEachNotNull(length, blockValSet, (from, to) -> {
- DoubleArrayList result = aggregationResultHolder.getResult();
- if (result == null) {
- result = new DoubleArrayList();
- aggregationResultHolder.setValue(result);
- }
+ double[] sums = aggregationResultHolder.getResult();
for (int i = from; i < to; i++) {
- aggregateMerge(values[i], result);
+ sums = add(sums, values[i]);
}
+ aggregationResultHolder.setValue(sums);
});
}
@@ -107,45 +101,50 @@ public class SumArrayDoubleAggregationFunction
}
private void setGroupByResult(GroupByResultHolder groupByResultHolder,
double[] values, int groupKey) {
- DoubleArrayList sumList = groupByResultHolder.getResult(groupKey);
- if (sumList == null) {
- sumList = new DoubleArrayList();
- groupByResultHolder.setValueForKey(groupKey, sumList);
+ double[] sums = groupByResultHolder.getResult(groupKey);
+ double[] newSums = add(sums, values);
+ if (newSums != sums) {
+ groupByResultHolder.setValueForKey(groupKey, newSums);
}
- aggregateMerge(values, sumList);
}
- private void aggregateMerge(double[] values, DoubleArrayList sumList) {
- for (int j = sumList.size(); j < values.length; j++) {
- sumList.add(0L);
+ /// Adds the values to the sums and returns the sums, which is a new array
when there are none yet or the values are
+ /// longer.
+ private static double[] add(@Nullable double[] sums, double[] values) {
+ if (sums == null) {
+ return values.clone();
+ }
+ if (sums.length < values.length) {
+ sums = Arrays.copyOf(sums, values.length);
}
- for (int j = 0; j < values.length; j++) {
- sumList.set(j, sumList.getDouble(j) + values[j]);
+ for (int i = 0; i < values.length; i++) {
+ sums[i] += values[i];
}
+ return sums;
}
@Nullable
@Override
- public DoubleArrayList extractAggregationResult(AggregationResultHolder
aggregationResultHolder) {
+ public double[] extractAggregationResult(AggregationResultHolder
aggregationResultHolder) {
return aggregationResultHolder.getResult();
}
@Nullable
@Override
- public DoubleArrayList extractGroupByResult(GroupByResultHolder
groupByResultHolder, int groupKey) {
+ public double[] extractGroupByResult(GroupByResultHolder
groupByResultHolder, int groupKey) {
return groupByResultHolder.getResult(groupKey);
}
@Override
- public DoubleArrayList merge(DoubleArrayList intermediateResult1,
DoubleArrayList intermediateResult2) {
- if (intermediateResult1.size() < intermediateResult2.size()) {
- for (int i = 0; i < intermediateResult1.size(); i++) {
- intermediateResult2.set(i, intermediateResult1.getDouble(i) +
intermediateResult2.getDouble(i));
+ public double[] merge(double[] intermediateResult1, double[]
intermediateResult2) {
+ if (intermediateResult1.length < intermediateResult2.length) {
+ for (int i = 0; i < intermediateResult1.length; i++) {
+ intermediateResult2[i] += intermediateResult1[i];
}
return intermediateResult2;
}
- for (int i = 0; i < intermediateResult2.size(); i++) {
- intermediateResult1.set(i, intermediateResult1.getDouble(i) +
intermediateResult2.getDouble(i));
+ for (int i = 0; i < intermediateResult2.length; i++) {
+ intermediateResult1[i] += intermediateResult2[i];
}
return intermediateResult1;
}
@@ -154,18 +153,18 @@ public class SumArrayDoubleAggregationFunction
public DataSchema.ColumnDataType getIntermediateResultColumnType() {
// AggregationResultsBlock#setIntermediateResult and
AggregationFunctionUtils#getIntermediateResult only serialize
// intermediate results whose stored type is
INT/LONG/DOUBLE/STRING/OBJECT, so we need the OBJECT/custom ser-de path
- // for DoubleArrayList payloads.
+ // for the sum array.
return DataSchema.ColumnDataType.OBJECT;
}
@Override
- public SerializedResult serializeIntermediateResult(DoubleArrayList
doubleArrayList) {
- return new SerializedValue<>(DOUBLE_ARRAY_LIST_TYPE, doubleArrayList,
ObjectSerDeUtils.DOUBLE_ARRAY_LIST_SER_DE);
+ public SerializedResult serializeIntermediateResult(double[] sums) {
+ return new SerializedValue<>(DOUBLE_ARRAY_LIST_TYPE, sums,
ObjectSerDeUtils.DOUBLE_ARRAY_SER_DE);
}
@Override
- public DoubleArrayList deserializeIntermediateResult(CustomObject
customObject) {
- return
ObjectSerDeUtils.DOUBLE_ARRAY_LIST_SER_DE.deserialize(customObject.getBuffer());
+ public double[] deserializeIntermediateResult(CustomObject customObject) {
+ return
ObjectSerDeUtils.DOUBLE_ARRAY_SER_DE.deserialize(customObject.getBuffer());
}
@Override
@@ -175,8 +174,8 @@ public class SumArrayDoubleAggregationFunction
@Nullable
@Override
- public double[] extractFinalResult(@Nullable DoubleArrayList result) {
- return result != null ? ArrayListUtils.toDoubleArray(result) : null;
+ public double[] extractFinalResult(@Nullable double[] sums) {
+ return sums;
}
@Override
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/array/SumArrayLongAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/array/SumArrayLongAggregationFunction.java
index bb6acbf4e28..a8b021b432c 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/array/SumArrayLongAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/array/SumArrayLongAggregationFunction.java
@@ -18,14 +18,12 @@
*/
package org.apache.pinot.core.query.aggregation.function.array;
-import it.unimi.dsi.fastutil.longs.LongArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import javax.annotation.Nullable;
import org.apache.pinot.common.CustomObject;
import org.apache.pinot.common.request.context.ExpressionContext;
-import org.apache.pinot.common.utils.ArrayListUtils;
import org.apache.pinot.common.utils.DataSchema;
import org.apache.pinot.core.common.BlockValSet;
import org.apache.pinot.core.query.aggregation.AggregationResultHolder;
@@ -37,8 +35,7 @@ import org.apache.pinot.segment.local.utils.ObjectSerDeUtils;
import org.apache.pinot.segment.spi.AggregationFunctionType;
-public class SumArrayLongAggregationFunction
- extends BaseSingleInputAggregationFunction<LongArrayList, long[]> {
+public class SumArrayLongAggregationFunction extends
BaseSingleInputAggregationFunction<long[], long[]> {
private static final int LONG_ARRAY_LIST_TYPE = 43;
public SumArrayLongAggregationFunction(List<ExpressionContext> arguments,
boolean nullHandlingEnabled) {
@@ -68,14 +65,11 @@ public class SumArrayLongAggregationFunction
// The accumulator is created inside the range, so a block with no
non-null row leaves the holder untouched and
// extractFinalResult sees the null that means nothing was aggregated
forEachNotNull(length, blockValSet, (from, to) -> {
- LongArrayList result = aggregationResultHolder.getResult();
- if (result == null) {
- result = new LongArrayList();
- aggregationResultHolder.setValue(result);
- }
+ long[] sums = aggregationResultHolder.getResult();
for (int i = from; i < to; i++) {
- aggregateMerge(values[i], result);
+ sums = add(sums, values[i]);
}
+ aggregationResultHolder.setValue(sums);
});
}
@@ -107,63 +101,68 @@ public class SumArrayLongAggregationFunction
}
private void setGroupByResult(GroupByResultHolder groupByResultHolder,
long[] values, int groupKey) {
- LongArrayList sumList = groupByResultHolder.getResult(groupKey);
- if (sumList == null) {
- sumList = new LongArrayList();
- groupByResultHolder.setValueForKey(groupKey, sumList);
+ long[] sums = groupByResultHolder.getResult(groupKey);
+ long[] newSums = add(sums, values);
+ if (newSums != sums) {
+ groupByResultHolder.setValueForKey(groupKey, newSums);
}
- aggregateMerge(values, sumList);
}
- private void aggregateMerge(long[] values, LongArrayList sumList) {
- for (int j = sumList.size(); j < values.length; j++) {
- sumList.add(0L);
+ /// Adds the values to the sums and returns the sums, which is a new array
when there are none yet or the values are
+ /// longer.
+ private static long[] add(@Nullable long[] sums, long[] values) {
+ if (sums == null) {
+ return values.clone();
+ }
+ if (sums.length < values.length) {
+ sums = Arrays.copyOf(sums, values.length);
}
- for (int j = 0; j < values.length; j++) {
- sumList.set(j, sumList.getLong(j) + values[j]);
+ for (int i = 0; i < values.length; i++) {
+ sums[i] += values[i];
}
+ return sums;
}
@Nullable
@Override
- public LongArrayList extractAggregationResult(AggregationResultHolder
aggregationResultHolder) {
+ public long[] extractAggregationResult(AggregationResultHolder
aggregationResultHolder) {
return aggregationResultHolder.getResult();
}
@Nullable
@Override
- public LongArrayList extractGroupByResult(GroupByResultHolder
groupByResultHolder, int groupKey) {
+ public long[] extractGroupByResult(GroupByResultHolder groupByResultHolder,
int groupKey) {
return groupByResultHolder.getResult(groupKey);
}
@Override
- public LongArrayList merge(LongArrayList intermediateResult1, LongArrayList
intermediateResult2) {
- if (intermediateResult1.size() < intermediateResult2.size()) {
- for (int i = 0; i < intermediateResult1.size(); i++) {
- intermediateResult2.set(i, intermediateResult1.getLong(i) +
intermediateResult2.getLong(i));
+ public long[] merge(long[] intermediateResult1, long[] intermediateResult2) {
+ if (intermediateResult1.length < intermediateResult2.length) {
+ for (int i = 0; i < intermediateResult1.length; i++) {
+ intermediateResult2[i] += intermediateResult1[i];
}
return intermediateResult2;
}
- for (int i = 0; i < intermediateResult2.size(); i++) {
- intermediateResult1.set(i, intermediateResult1.getLong(i) +
intermediateResult2.getLong(i));
+ for (int i = 0; i < intermediateResult2.length; i++) {
+ intermediateResult1[i] += intermediateResult2[i];
}
return intermediateResult1;
}
@Override
- public SerializedResult serializeIntermediateResult(LongArrayList
longArrayList) {
- return new SerializedValue<>(LONG_ARRAY_LIST_TYPE, longArrayList,
ObjectSerDeUtils.LONG_ARRAY_LIST_SER_DE);
+ public SerializedResult serializeIntermediateResult(long[] sums) {
+ return new SerializedValue<>(LONG_ARRAY_LIST_TYPE, sums,
ObjectSerDeUtils.LONG_ARRAY_SER_DE);
}
@Override
- public LongArrayList deserializeIntermediateResult(CustomObject
customObject) {
- return
ObjectSerDeUtils.LONG_ARRAY_LIST_SER_DE.deserialize(customObject.getBuffer());
+ public long[] deserializeIntermediateResult(CustomObject customObject) {
+ return
ObjectSerDeUtils.LONG_ARRAY_SER_DE.deserialize(customObject.getBuffer());
}
@Override
public DataSchema.ColumnDataType getIntermediateResultColumnType() {
// AggregationResultsBlock#setIntermediateResult and
AggregationFunctionUtils#getIntermediateResult only support
- // INT/LONG/DOUBLE/STRING/OBJECT stored types, so LongArrayList must use
OBJECT to piggyback on the custom ser-de
+ // INT/LONG/DOUBLE/STRING/OBJECT stored types, so the sum array must use
OBJECT to piggyback on the custom ser-de
// implemented in ObjectSerDeUtils.
return DataSchema.ColumnDataType.OBJECT;
}
@@ -175,8 +174,8 @@ public class SumArrayLongAggregationFunction
@Nullable
@Override
- public long[] extractFinalResult(@Nullable LongArrayList result) {
- return result != null ? ArrayListUtils.toLongArray(result) : null;
+ public long[] extractFinalResult(@Nullable long[] sums) {
+ return sums;
}
@Override
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/utils/DoubleVectorOpUtils.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/utils/DoubleVectorOpUtils.java
deleted file mode 100644
index 26373c1aeae..00000000000
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/utils/DoubleVectorOpUtils.java
+++ /dev/null
@@ -1,69 +0,0 @@
-/**
- * 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.utils;
-
-import com.google.common.base.Preconditions;
-import it.unimi.dsi.fastutil.doubles.DoubleArrayList;
-
-
-public class DoubleVectorOpUtils {
- private DoubleVectorOpUtils() {
- }
-
- public static DoubleArrayList vectorAdd(DoubleArrayList a, DoubleArrayList
b) {
- return vectorAdd(a, b.elements());
- }
-
- public static DoubleArrayList vectorAdd(DoubleArrayList a, double[] b) {
- double[] elements = a.elements();
- int length = elements.length;
- Preconditions.checkState(length == b.length, "The two operand arrays are
not of the same size! provided %s, %s",
- length, b.length);
- for (int i = 0; i < length; i++) {
- elements[i] += b[i];
- }
- return a;
- }
-
- /// return a new DoubleArrayList initialized with a
- /// @param a array to initialize
- /// @return a new DoubleArrayList initialized with a
- public static DoubleArrayList createAndInitialize(double[] a) {
- return new DoubleArrayList(a);
- }
-
- /// return a new DoubleArrayList with all zeros
- /// @param len length of array
- /// @return a new DoubleArrayList with all zeros
- public static DoubleArrayList createAndInitialize(int len) {
- Preconditions.checkState(len > 0, "Asking for an array of length %s", len);
- return new DoubleArrayList(new double[len]);
- }
-
- public static DoubleArrayList incrementElement(DoubleArrayList a, int
offset, double val) {
- Preconditions.checkState(a.size() > offset, "The offset %s exceeds the
array size!", offset);
- double[] elements = a.elements();
- elements[offset] += val;
- return a;
- }
-
- public static DoubleArrayList incrementElementByOne(DoubleArrayList a, int
offset) {
- return incrementElement(a, offset, 1d);
- }
-}
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/IntermediateResultByteBufferSerDeTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/IntermediateResultByteBufferSerDeTest.java
index 51e8c8812f5..a0c5463060d 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/IntermediateResultByteBufferSerDeTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/IntermediateResultByteBufferSerDeTest.java
@@ -314,9 +314,9 @@ public class IntermediateResultByteBufferSerDeTest {
{new ArrayAggDistinctStringFunction(COLUMN, false), new
ObjectOpenHashSet<>(new String[]{"a"}), 18},
{new ArrayAggDistinctBytesFunction(COLUMN, DataType.BYTES, false),
new ObjectOpenHashSet<>(new ByteArray[]{new ByteArray(new
byte[]{1})}), 19},
- {new SumArrayLongAggregationFunction(List.of(COLUMN), false), new
LongArrayList(new long[]{3, 4}), 43},
- {new SumArrayDoubleAggregationFunction(List.of(COLUMN), false), new
DoubleArrayList(new double[]{0.25, 8}), 3},
- {HISTOGRAM, new DoubleArrayList(new double[]{1, 2}), 3},
+ {new SumArrayLongAggregationFunction(List.of(COLUMN), false), new
long[]{3, 4}, 43},
+ {new SumArrayDoubleAggregationFunction(List.of(COLUMN), false), new
double[]{0.25, 8}, 3},
+ {HISTOGRAM, new double[]{1, 2}, 3},
{new PercentileAggregationFunction(COLUMN, 50, false), new
DoubleArrayList(new double[]{-3.5, 0, 7}), 3},
{PERCENTILE_TDIGEST, tDigest, 10},
{PERCENTILE_RAW_TDIGEST, tDigest, 10},
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/ValueAggregationNullHandlingTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/ValueAggregationNullHandlingTest.java
index 800d7a5a9d3..9b5f2bef870 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/ValueAggregationNullHandlingTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/ValueAggregationNullHandlingTest.java
@@ -18,7 +18,6 @@
*/
package org.apache.pinot.core.query.aggregation.function;
-import it.unimi.dsi.fastutil.doubles.DoubleArrayList;
import java.math.BigDecimal;
import java.util.List;
import java.util.Map;
@@ -238,6 +237,56 @@ public class ValueAggregationNullHandlingTest {
assertEquals(function.extractFinalResult(function.extractAggregationResult(holder)),
new double[]{11.5d, 22.5d});
}
+ /// Arrays of different lengths are added by position, so the sums are as
long as the longest array. The rows go in
+ /// twice, so the second block also adds arrays shorter than the sums it
finds.
+ @Test
+ public void testSumArrayLongAddsArraysOfDifferentLengths() {
+ Map<ExpressionContext, BlockValSet> block =
+ block(SyntheticBlockValSets.LongMV.create(null, new long[][]{{1L},
{10L, 20L, 30L}, {100L, 200L}}));
+ long[] expected = {222L, 440L, 60L};
+
+ SumArrayLongAggregationFunction function = new
SumArrayLongAggregationFunction(List.of(COLUMN), true);
+ AggregationResultHolder holder = function.createAggregationResultHolder();
+ GroupByResultHolder svHolder = function.createGroupByResultHolder(1, 1);
+ GroupByResultHolder mvHolder = function.createGroupByResultHolder(2, 2);
+ for (int i = 0; i < 2; i++) {
+ function.aggregate(3, holder, block);
+ function.aggregateGroupBySV(3, new int[]{0, 0, 0}, svHolder, block);
+ function.aggregateGroupByMV(3, new int[][]{{0, 1}, {0, 1}, {0, 1}},
mvHolder, block);
+ }
+ assertEquals(function.extractAggregationResult(holder), expected);
+ assertEquals(function.extractGroupByResult(svHolder, 0), expected);
+ assertEquals(function.extractGroupByResult(mvHolder, 0), expected);
+ assertEquals(function.extractGroupByResult(mvHolder, 1), expected);
+
+ assertEquals(function.merge(new long[]{1L}, new long[]{10L, 20L, 30L}),
new long[]{11L, 20L, 30L});
+ assertEquals(function.merge(new long[]{10L, 20L, 30L}, new long[]{1L}),
new long[]{11L, 20L, 30L});
+ }
+
+ @Test
+ public void testSumArrayDoubleAddsArraysOfDifferentLengths() {
+ Map<ExpressionContext, BlockValSet> block =
+ block(SyntheticBlockValSets.DoubleMV.create(null, new double[][]{{1d},
{10d, 20d, 30d}, {100d, 200d}}));
+ double[] expected = {222d, 440d, 60d};
+
+ SumArrayDoubleAggregationFunction function = new
SumArrayDoubleAggregationFunction(List.of(COLUMN), true);
+ AggregationResultHolder holder = function.createAggregationResultHolder();
+ GroupByResultHolder svHolder = function.createGroupByResultHolder(1, 1);
+ GroupByResultHolder mvHolder = function.createGroupByResultHolder(2, 2);
+ for (int i = 0; i < 2; i++) {
+ function.aggregate(3, holder, block);
+ function.aggregateGroupBySV(3, new int[]{0, 0, 0}, svHolder, block);
+ function.aggregateGroupByMV(3, new int[][]{{0, 1}, {0, 1}, {0, 1}},
mvHolder, block);
+ }
+ assertEquals(function.extractAggregationResult(holder), expected);
+ assertEquals(function.extractGroupByResult(svHolder, 0), expected);
+ assertEquals(function.extractGroupByResult(mvHolder, 0), expected);
+ assertEquals(function.extractGroupByResult(mvHolder, 1), expected);
+
+ assertEquals(function.merge(new double[]{1d}, new double[]{10d, 20d,
30d}), new double[]{11d, 20d, 30d});
+ assertEquals(function.merge(new double[]{10d, 20d, 30d}, new
double[]{1d}), new double[]{11d, 20d, 30d});
+ }
+
/// Unlike the others here, the sum of no arrays is NULL in both modes,
which is why neither needs a mode-aware
/// branch in extractFinalResult.
@Test
@@ -317,6 +366,28 @@ public class ValueAggregationNullHandlingTest {
new double[]{2.0d, 2.0d});
}
+ /// Every block counts into the one histogram of the holder, and a block
with no non-null row creates none.
+ @Test
+ public void testHistogramAcrossBlocks() {
+ HistogramAggregationFunction function = histogram(true);
+
+ double[] values = {1.0d, 7.0d, 6.0d};
+ AggregationResultHolder svHolder =
function.createAggregationResultHolder();
+ function.aggregate(3, svHolder,
block(SyntheticBlockValSets.Double.create(allNull(3), values)));
+ assertNull(function.extractAggregationResult(svHolder));
+ function.aggregate(3, svHolder,
block(SyntheticBlockValSets.Double.create(ROW1_NULL, values)));
+ function.aggregate(3, svHolder,
block(SyntheticBlockValSets.Double.create(ROW1_NULL, values)));
+ assertEquals(function.extractAggregationResult(svHolder), new
double[]{2.0d, 2.0d});
+
+ double[][] rows = {{1.0d, 2.0d}, {7.0d}, {6.0d, 9.0d}};
+ AggregationResultHolder mvHolder =
function.createAggregationResultHolder();
+ function.aggregate(3, mvHolder,
block(SyntheticBlockValSets.DoubleMV.create(allNull(3), rows)));
+ assertNull(function.extractAggregationResult(mvHolder));
+ function.aggregate(3, mvHolder,
block(SyntheticBlockValSets.DoubleMV.create(ROW1_NULL, rows)));
+ function.aggregate(3, mvHolder,
block(SyntheticBlockValSets.DoubleMV.create(ROW1_NULL, rows)));
+ assertEquals(function.extractAggregationResult(mvHolder), new
double[]{4.0d, 4.0d});
+ }
+
/// BIG_DECIMAL is numeric and now lands in the same bins as the other
numeric types, single- and multi-value.
@Test
public void testHistogramBigDecimalColumn() {
@@ -339,7 +410,7 @@ public class ValueAggregationNullHandlingTest {
function.aggregateGroupBySV(3, new int[]{0, 1, 0}, holder,
block(SyntheticBlockValSets.DoubleMV.create(ROW1_NULL, rows)));
- assertEquals(function.extractGroupByResult(holder, 0), new
DoubleArrayList(new double[]{2.0d, 1.0d}));
+ assertEquals(function.extractGroupByResult(holder, 0), new double[]{2.0d,
1.0d});
assertNull(function.extractGroupByResult(holder, 1));
}
@@ -353,8 +424,8 @@ public class ValueAggregationNullHandlingTest {
function.aggregateGroupByMV(2, new int[][]{{0, 1}, {0, 1}}, holder,
block(SyntheticBlockValSets.DoubleMV.create(ROW1_NULL, rows)));
- assertEquals(function.extractGroupByResult(holder, 0), new
DoubleArrayList(new double[]{2.0d, 0.0d}));
- assertEquals(function.extractGroupByResult(holder, 1), new
DoubleArrayList(new double[]{2.0d, 0.0d}));
+ assertEquals(function.extractGroupByResult(holder, 0), new double[]{2.0d,
0.0d});
+ assertEquals(function.extractGroupByResult(holder, 1), new double[]{2.0d,
0.0d});
}
// ---------- IDSET ----------
diff --git
a/pinot-core/src/test/java/org/apache/pinot/queries/HistogramQueriesTest.java
b/pinot-core/src/test/java/org/apache/pinot/queries/HistogramQueriesTest.java
index aeb93a96c79..2e8facf5e45 100644
---
a/pinot-core/src/test/java/org/apache/pinot/queries/HistogramQueriesTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/queries/HistogramQueriesTest.java
@@ -18,7 +18,6 @@
*/
package org.apache.pinot.queries;
-import it.unimi.dsi.fastutil.doubles.DoubleArrayList;
import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
@@ -136,7 +135,7 @@ public class HistogramQueriesTest extends BaseQueriesTest {
NUM_RECORDS, NUM_RECORDS);
List<Object> aggregationResult = resultsBlock.getResults();
assertNotNull(aggregationResult);
- assertEquals(((DoubleArrayList) aggregationResult.get(0)).elements(), new
double[]{1, 9, 90, 900, 1000});
+ assertEquals((double[]) aggregationResult.get(0), new double[]{1, 9, 90,
900, 1000});
// Inner Segment ARRAY-specified edges with Infinity
query = "SELECT
HISTOGRAM(intColumn,ARRAY[\"-Infinity\",1,10,100,1000,\"Infinity\"]) FROM
testTable";
@@ -148,7 +147,7 @@ public class HistogramQueriesTest extends BaseQueriesTest {
NUM_RECORDS, NUM_RECORDS);
aggregationResult = resultsBlock.getResults();
assertNotNull(aggregationResult);
- assertEquals(((DoubleArrayList) aggregationResult.get(0)).elements(), new
double[]{1, 9, 90, 900, 1000});
+ assertEquals((double[]) aggregationResult.get(0), new double[]{1, 9, 90,
900, 1000});
// Inner Segment with filter
operator = getOperatorWithFilter(query);
@@ -158,7 +157,7 @@ public class HistogramQueriesTest extends BaseQueriesTest {
NUM_RECORDS);
aggregationResult = resultsBlock.getResults();
assertNotNull(aggregationResult);
- assertEquals(((DoubleArrayList) aggregationResult.get(0)).elements(), new
double[]{0, 0, 0, 500, 1000});
+ assertEquals((double[]) aggregationResult.get(0), new double[]{0, 0, 0,
500, 1000});
// Inter segment
BrokerResponseNative brokerResponse = getBrokerResponse(query);
@@ -187,8 +186,7 @@ public class HistogramQueriesTest extends BaseQueriesTest {
NUM_RECORDS, NUM_RECORDS);
List<Object> aggregationResult = resultsBlock.getResults();
assertNotNull(aggregationResult);
- assertEquals(((DoubleArrayList) aggregationResult.get(0)).elements(),
- new double[]{100, 100, 100, 100, 100, 100, 100, 100, 100, 101});
+ assertEquals((double[]) aggregationResult.get(0), new double[]{100, 100,
100, 100, 100, 100, 100, 100, 100, 101});
operator = getOperatorWithFilter(query);
assertTrue(operator instanceof AggregationOperator);
@@ -197,8 +195,7 @@ public class HistogramQueriesTest extends BaseQueriesTest {
NUM_RECORDS);
aggregationResult = resultsBlock.getResults();
assertNotNull(aggregationResult);
- assertEquals(((DoubleArrayList) aggregationResult.get(0)).elements(),
- new double[]{0, 0, 0, 0, 0, 100, 100, 100, 100, 101});
+ assertEquals((double[]) aggregationResult.get(0), new double[]{0, 0, 0, 0,
0, 100, 100, 100, 100, 101});
// Inter segment
BrokerResponseNative brokerResponse = getBrokerResponse(query);
@@ -220,17 +217,17 @@ public class HistogramQueriesTest extends BaseQueriesTest
{
NUM_RECORDS * 2, NUM_RECORDS);
List<IntermediateRecord> resultRecords =
resultsBlock.getIntermediateRecords();
assertNotNull(resultRecords);
- assertEquals(((DoubleArrayList)
resultRecords.get(0)._record.getValues()[1]).elements(),
+ assertEquals((double[]) resultRecords.get(0)._record.getValues()[1],
new double[]{1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
0}); // [0]
- assertEquals(((DoubleArrayList)
resultRecords.get(1)._record.getValues()[1]).elements(),
+ assertEquals((double[]) resultRecords.get(1)._record.getValues()[1],
new double[]{99, 100, 100, 100, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
0, 0, 0}); // [1-400]
- assertEquals(((DoubleArrayList)
resultRecords.get(2)._record.getValues()[1]).elements(),
+ assertEquals((double[]) resultRecords.get(2)._record.getValues()[1],
new double[]{0, 0, 0, 0, 99, 100, 100, 100, 1, 0, 0, 0, 0, 0, 0, 0, 0,
0, 0, 0}); // [401-800]
- assertEquals(((DoubleArrayList)
resultRecords.get(3)._record.getValues()[1]).elements(),
+ assertEquals((double[]) resultRecords.get(3)._record.getValues()[1],
new double[]{0, 0, 0, 0, 0, 0, 0, 0, 99, 100, 100, 100, 1, 0, 0, 0, 0,
0, 0, 0}); // [801-1200]
- assertEquals(((DoubleArrayList)
resultRecords.get(4)._record.getValues()[1]).elements(),
+ assertEquals((double[]) resultRecords.get(4)._record.getValues()[1],
new double[]{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 99, 100, 100, 100, 1,
0, 0, 0}); // [1201-1600]
- assertEquals(((DoubleArrayList)
resultRecords.get(5)._record.getValues()[1]).elements(),
+ assertEquals((double[]) resultRecords.get(5)._record.getValues()[1],
new double[]{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 99, 100,
100, 100}); // [1601-2000]
// Inter segment
@@ -257,7 +254,7 @@ public class HistogramQueriesTest extends BaseQueriesTest {
2 * NUM_RECORDS, NUM_RECORDS);
List<Object> aggregationResult = resultsBlock.getResults();
assertNotNull(aggregationResult);
- assertEquals(((DoubleArrayList) aggregationResult.get(0)).elements(),
+ assertEquals((double[]) aggregationResult.get(0),
new double[]{50, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50, 50,
50, 50, 50, 50, 50, 51});
query = "SELECT HISTOGRAM(ADD(intColumn,doubleColumn),0,2000,20) FROM
testTable";
@@ -268,7 +265,7 @@ public class HistogramQueriesTest extends BaseQueriesTest {
NUM_RECORDS);
aggregationResult = resultsBlock.getResults();
assertNotNull(aggregationResult);
- assertEquals(((DoubleArrayList) aggregationResult.get(0)).elements(),
+ assertEquals((double[]) aggregationResult.get(0),
new double[]{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 50, 50, 50, 50, 50, 50, 50,
50, 50, 51});
// Inter segment
@@ -363,8 +360,7 @@ public class HistogramQueriesTest extends BaseQueriesTest {
NUM_RECORDS, NUM_RECORDS);
List<Object> aggregationResult = resultsBlock.getResults();
assertNotNull(aggregationResult);
- assertEquals(((DoubleArrayList) aggregationResult.get(0)).elements(),
- new double[]{12, 11, 11, 11, 11, 11, 11, 11, 12});
+ assertEquals((double[]) aggregationResult.get(0), new double[]{12, 11, 11,
11, 11, 11, 11, 11, 12});
// [0, 11], [12, 22], [23, 33], [34, 44], [45, 55], [56,66], [67, 77],
[78, 88], [89, 100]
query = "SELECT HISTOGRAM(intColumn,ARRAY[0,1]) FROM testTable";
@@ -375,7 +371,7 @@ public class HistogramQueriesTest extends BaseQueriesTest {
NUM_RECORDS, NUM_RECORDS);
aggregationResult = resultsBlock.getResults();
assertNotNull(aggregationResult);
- assertEquals(((DoubleArrayList) aggregationResult.get(0)).elements(), new
double[]{2});
+ assertEquals((double[]) aggregationResult.get(0), new double[]{2});
query = "SELECT HISTOGRAM(intColumn,ARRAY[1999,2000]) FROM testTable";
operator = getOperator(query);
@@ -385,7 +381,7 @@ public class HistogramQueriesTest extends BaseQueriesTest {
NUM_RECORDS, NUM_RECORDS);
aggregationResult = resultsBlock.getResults();
assertNotNull(aggregationResult);
- assertEquals(((DoubleArrayList) aggregationResult.get(0)).elements(), new
double[]{1});
+ assertEquals((double[]) aggregationResult.get(0), new double[]{1});
query = "SELECT HISTOGRAM(longColumn,ARRAY[\"-Infinity\", -999, -800,
-500, 500, 600, 800, \"Infinity\"]) FROM "
+ "testTable";
@@ -396,8 +392,7 @@ public class HistogramQueriesTest extends BaseQueriesTest {
NUM_RECORDS, NUM_RECORDS);
aggregationResult = resultsBlock.getResults();
assertNotNull(aggregationResult);
- assertEquals(((DoubleArrayList) aggregationResult.get(0)).elements(),
- new double[]{1, 199, 300, 1000, 100, 200, 200});
+ assertEquals((double[]) aggregationResult.get(0), new double[]{1, 199,
300, 1000, 100, 200, 200});
query = "SELECT HISTOGRAM(longColumn,-999.5, 500.5, 15) FROM testTable";
operator = getOperator(query);
@@ -407,7 +402,7 @@ public class HistogramQueriesTest extends BaseQueriesTest {
NUM_RECORDS, NUM_RECORDS);
aggregationResult = resultsBlock.getResults();
assertNotNull(aggregationResult);
- assertEquals(((DoubleArrayList) aggregationResult.get(0)).elements(),
+ assertEquals((double[]) aggregationResult.get(0),
new double[]{100, 100, 100, 100, 100, 100, 100, 100, 100, 100, 100,
100, 100, 100, 100});
query = "SELECT HISTOGRAM(longColumn, -1000, 500, 15) FROM testTable";
@@ -418,7 +413,7 @@ public class HistogramQueriesTest extends BaseQueriesTest {
NUM_RECORDS, NUM_RECORDS);
aggregationResult = resultsBlock.getResults();
assertNotNull(aggregationResult);
- assertEquals(((DoubleArrayList) aggregationResult.get(0)).elements(),
+ assertEquals((double[]) aggregationResult.get(0),
new double[]{100, 100, 100, 100, 100, 100, 100, 100, 100, 100, 100,
100, 100, 100, 101});
query = "SELECT HISTOGRAM(floatColumn, ARRAY[0.5,1.5,2.5,3.5,4.5,5.5])
FROM testTable";
@@ -429,7 +424,7 @@ public class HistogramQueriesTest extends BaseQueriesTest {
NUM_RECORDS, NUM_RECORDS);
aggregationResult = resultsBlock.getResults();
assertNotNull(aggregationResult);
- assertEquals(((DoubleArrayList) aggregationResult.get(0)).elements(), new
double[]{2, 2, 2, 2, 3});
+ assertEquals((double[]) aggregationResult.get(0), new double[]{2, 2, 2, 2,
3});
// {0.5, 1} {1.5, 2}, {2.5, 3}, {3.5,4}, {4.5, 5, 5.5}
query = "SELECT HISTOGRAM(doubleColumn, ARRAY[-1, 0, 1, 2, 3, 4]) FROM
testTable";
@@ -440,7 +435,7 @@ public class HistogramQueriesTest extends BaseQueriesTest {
NUM_RECORDS, NUM_RECORDS);
aggregationResult = resultsBlock.getResults();
assertNotNull(aggregationResult);
- assertEquals(((DoubleArrayList) aggregationResult.get(0)).elements(), new
double[]{0, 1, 1, 1, 2});
+ assertEquals((double[]) aggregationResult.get(0), new double[]{0, 1, 1, 1,
2});
query = "SELECT HISTOGRAM(longColumn, ARRAY[-1, 0, 1, 2, 3, 4]) FROM
testTable";
operator = getOperator(query);
@@ -450,7 +445,7 @@ public class HistogramQueriesTest extends BaseQueriesTest {
NUM_RECORDS, NUM_RECORDS);
aggregationResult = resultsBlock.getResults();
assertNotNull(aggregationResult);
- assertEquals(((DoubleArrayList) aggregationResult.get(0)).elements(), new
double[]{1, 1, 1, 1, 2});
+ assertEquals((double[]) aggregationResult.get(0), new double[]{1, 1, 1, 1,
2});
query = "SELECT HISTOGRAM(longColumn, ARRAY[-900, -850, -800, 1]) FROM
testTable WHERE "
+ "longColumn BETWEEN -1000 AND -950 OR longColumn BETWEEN -900 AND 2
OR longColumn BETWEEN 900 AND 950";
@@ -461,7 +456,7 @@ public class HistogramQueriesTest extends BaseQueriesTest {
NUM_RECORDS);
aggregationResult = resultsBlock.getResults();
assertNotNull(aggregationResult);
- assertEquals(((DoubleArrayList) aggregationResult.get(0)).elements(), new
double[]{50, 50, 802});
+ assertEquals((double[]) aggregationResult.get(0), new double[]{50, 50,
802});
query = "SELECT HISTOGRAM(longColumn, ARRAY[-900, -850, -800, 1]) FROM
testTable WHERE "
+ "longColumn BETWEEN -1000 AND -950 OR longColumn BETWEEN -900 AND 2
OR longColumn BETWEEN 900 AND 950";
@@ -472,7 +467,7 @@ public class HistogramQueriesTest extends BaseQueriesTest {
NUM_RECORDS);
aggregationResult = resultsBlock.getResults();
assertNotNull(aggregationResult);
- assertEquals(((DoubleArrayList) aggregationResult.get(0)).elements(), new
double[]{50, 50, 802});
+ assertEquals((double[]) aggregationResult.get(0), new double[]{50, 50,
802});
query = "SELECT HISTOGRAM(longColumn,
ARRAY[-1004,-1003,-1002,-1001,-1000,-999]) FROM testTable";
operator = getOperator(query);
@@ -482,7 +477,7 @@ public class HistogramQueriesTest extends BaseQueriesTest {
NUM_RECORDS, NUM_RECORDS);
aggregationResult = resultsBlock.getResults();
assertNotNull(aggregationResult);
- assertEquals(((DoubleArrayList) aggregationResult.get(0)).elements(), new
double[]{0, 0, 0, 0, 2});
+ assertEquals((double[]) aggregationResult.get(0), new double[]{0, 0, 0, 0,
2});
query = "SELECT HISTOGRAM(floatColumn,
ARRAY[995,996,997,998,999,1000,1001,1002]) FROM testTable";
operator = getOperator(query);
@@ -492,7 +487,7 @@ public class HistogramQueriesTest extends BaseQueriesTest {
NUM_RECORDS, NUM_RECORDS);
aggregationResult = resultsBlock.getResults();
assertNotNull(aggregationResult);
- assertEquals(((DoubleArrayList) aggregationResult.get(0)).elements(), new
double[]{2, 2, 2, 2, 2, 0, 0});
+ assertEquals((double[]) aggregationResult.get(0), new double[]{2, 2, 2, 2,
2, 0, 0});
}
@AfterClass
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/ObjectSerDeUtils.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/ObjectSerDeUtils.java
index 794ba81cfd8..ed5d2733506 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/ObjectSerDeUtils.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/ObjectSerDeUtils.java
@@ -176,6 +176,44 @@ public class ObjectSerDeUtils {
}
};
+ /// Writes the number of values, then the first `numValues` of `values`.
+ private static void serializeLongs(long[] values, int numValues, ByteBuffer
buffer) {
+ buffer.putInt(numValues);
+ for (int i = 0; i < numValues; i++) {
+ buffer.putLong(values[i]);
+ }
+ }
+
+ /// Serializes a `long[]` in the format of [#LONG_ARRAY_LIST_SER_DE].
+ public static final SizedObjectSerDe<long[]> LONG_ARRAY_SER_DE = new
SizedObjectSerDe<>() {
+
+ @Override
+ public int getSerializedSize(long[] values) {
+ return Math.addExact(Integer.BYTES, Math.multiplyExact(values.length,
Long.BYTES));
+ }
+
+ @Override
+ public void serialize(long[] values, ByteBuffer buffer) {
+ serializeLongs(values, values.length, buffer);
+ }
+
+ @Override
+ public void serialize(long[] values, PinotOutputStream output)
+ throws IOException {
+ output.writeInt(values.length);
+ output.writeLongs(values, 0, values.length);
+ }
+
+ @Override
+ public long[] deserialize(ByteBuffer buffer) {
+ int numValues = buffer.getInt();
+ checkRemaining(buffer, numValues, Long.BYTES);
+ long[] values = new long[numValues];
+ PrimitiveArrays.readLongs(buffer, values, 0, numValues);
+ return values;
+ }
+ };
+
public static final SizedObjectSerDe<LongArrayList> LONG_ARRAY_LIST_SER_DE =
new SizedObjectSerDe<>() {
@Override
@@ -185,12 +223,7 @@ public class ObjectSerDeUtils {
@Override
public void serialize(LongArrayList longArrayList, ByteBuffer buffer) {
- int size = longArrayList.size();
- buffer.putInt(size);
- long[] values = longArrayList.elements();
- for (int i = 0; i < size; i++) {
- buffer.putLong(values[i]);
- }
+ serializeLongs(longArrayList.elements(), longArrayList.size(), buffer);
}
@Override
@@ -203,11 +236,7 @@ public class ObjectSerDeUtils {
@Override
public LongArrayList deserialize(ByteBuffer buffer) {
- int numValues = buffer.getInt();
- checkRemaining(buffer, numValues, Long.BYTES);
- long[] values = new long[numValues];
- PrimitiveArrays.readLongs(buffer, values, 0, numValues);
- return LongArrayList.wrap(values);
+ return LongArrayList.wrap(LONG_ARRAY_SER_DE.deserialize(buffer));
}
};
@@ -246,6 +275,44 @@ public class ObjectSerDeUtils {
}
};
+ /// Writes the number of values, then the first `numValues` of `values`.
+ private static void serializeDoubles(double[] values, int numValues,
ByteBuffer buffer) {
+ buffer.putInt(numValues);
+ for (int i = 0; i < numValues; i++) {
+ buffer.putLong(Double.doubleToRawLongBits(values[i]));
+ }
+ }
+
+ /// Serializes a `double[]` in the format of [#DOUBLE_ARRAY_LIST_SER_DE].
+ public static final SizedObjectSerDe<double[]> DOUBLE_ARRAY_SER_DE = new
SizedObjectSerDe<>() {
+
+ @Override
+ public int getSerializedSize(double[] values) {
+ return Math.addExact(Integer.BYTES, Math.multiplyExact(values.length,
Double.BYTES));
+ }
+
+ @Override
+ public void serialize(double[] values, ByteBuffer buffer) {
+ serializeDoubles(values, values.length, buffer);
+ }
+
+ @Override
+ public void serialize(double[] values, PinotOutputStream output)
+ throws IOException {
+ output.writeInt(values.length);
+ output.writeDoubles(values, 0, values.length);
+ }
+
+ @Override
+ public double[] deserialize(ByteBuffer buffer) {
+ int numValues = buffer.getInt();
+ checkRemaining(buffer, numValues, Double.BYTES);
+ double[] values = new double[numValues];
+ PrimitiveArrays.readDoubles(buffer, values, 0, numValues);
+ return values;
+ }
+ };
+
public static final SizedObjectSerDe<DoubleArrayList>
DOUBLE_ARRAY_LIST_SER_DE = new SizedObjectSerDe<>() {
@Override
@@ -255,12 +322,7 @@ public class ObjectSerDeUtils {
@Override
public void serialize(DoubleArrayList doubleArrayList, ByteBuffer buffer) {
- int size = doubleArrayList.size();
- buffer.putInt(size);
- double[] values = doubleArrayList.elements();
- for (int i = 0; i < size; i++) {
- buffer.putLong(Double.doubleToRawLongBits(values[i]));
- }
+ serializeDoubles(doubleArrayList.elements(), doubleArrayList.size(),
buffer);
}
@Override
@@ -273,11 +335,7 @@ public class ObjectSerDeUtils {
@Override
public DoubleArrayList deserialize(ByteBuffer buffer) {
- int numValues = buffer.getInt();
- checkRemaining(buffer, numValues, Double.BYTES);
- double[] values = new double[numValues];
- PrimitiveArrays.readDoubles(buffer, values, 0, numValues);
- return DoubleArrayList.wrap(values);
+ return DoubleArrayList.wrap(DOUBLE_ARRAY_SER_DE.deserialize(buffer));
}
};
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/ObjectSerDeUtilsTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/ObjectSerDeUtilsTest.java
index a17435c9504..4eff2c5ac5d 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/ObjectSerDeUtilsTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/ObjectSerDeUtilsTest.java
@@ -171,6 +171,17 @@ public class ObjectSerDeUtilsTest {
{DoubleArrayList.wrap(new
double[]{Double.longBitsToDouble(0x7ff0123456789abcL), -0.0d,
Double.POSITIVE_INFINITY, Double.NEGATIVE_INFINITY, 99}, 4),
ObjectSerDeUtils.DOUBLE_ARRAY_LIST_SER_DE,
HexFormat.of().parseHex("000000047ff0123456789abc80000000000000007ff0000000000000fff0000000000000")},
+ {new long[0], ObjectSerDeUtils.LONG_ARRAY_SER_DE,
HexFormat.of().parseHex("00000000")},
+ {new long[]{Long.MIN_VALUE, Long.MAX_VALUE, 0},
ObjectSerDeUtils.LONG_ARRAY_SER_DE,
+
HexFormat.of().parseHex("0000000380000000000000007fffffffffffffff0000000000000000")},
+ {largeLongList.toLongArray(), ObjectSerDeUtils.LONG_ARRAY_SER_DE,
+ ObjectSerDeUtils.LONG_ARRAY_LIST_SER_DE.serialize(largeLongList)},
+ {new double[0], ObjectSerDeUtils.DOUBLE_ARRAY_SER_DE,
HexFormat.of().parseHex("00000000")},
+ {new double[]{Double.longBitsToDouble(0x7ff0123456789abcL), -0.0d,
Double.POSITIVE_INFINITY,
+ Double.NEGATIVE_INFINITY}, ObjectSerDeUtils.DOUBLE_ARRAY_SER_DE,
+
HexFormat.of().parseHex("000000047ff0123456789abc80000000000000007ff0000000000000fff0000000000000")},
+ {largeDoubleList.toDoubleArray(), ObjectSerDeUtils.DOUBLE_ARRAY_SER_DE,
+
ObjectSerDeUtils.DOUBLE_ARRAY_LIST_SER_DE.serialize(largeDoubleList)},
{new HyperLogLog(4), ObjectSerDeUtils.HYPER_LOG_LOG_SER_DE,
HexFormat.of().parseHex("000000040000000c" +
"000000000000000000000000")},
{hyperLogLog, ObjectSerDeUtils.HYPER_LOG_LOG_SER_DE,
hyperLogLog.getBytes()},
@@ -328,6 +339,16 @@ public class ObjectSerDeUtilsTest {
}
return bits;
}
+ if (value instanceof long[] values) {
+ return values.clone();
+ }
+ if (value instanceof double[] values) {
+ long[] bits = new long[values.length];
+ for (int i = 0; i < values.length; i++) {
+ bits[i] = Double.doubleToRawLongBits(values[i]);
+ }
+ return bits;
+ }
if (value instanceof HyperLogLog hyperLogLog) {
return hyperLogLog.getBytes();
}
@@ -362,6 +383,17 @@ public class ObjectSerDeUtilsTest {
}
}
+ @Test
+ public void testDoubleArray() {
+ for (int i = 0; i < NUM_ITERATIONS; i++) {
+ double[] expected = RANDOM.doubles(RANDOM.nextInt(100)).toArray();
+ byte[] bytes = ObjectSerDeUtils.DOUBLE_ARRAY_SER_DE.serialize(expected);
+ assertEquals(bytes,
ObjectSerDeUtils.DOUBLE_ARRAY_LIST_SER_DE.serialize(DoubleArrayList.wrap(expected)),
+ ERROR_MESSAGE);
+ assertEquals(ObjectSerDeUtils.DOUBLE_ARRAY_SER_DE.deserialize(bytes),
expected, ERROR_MESSAGE);
+ }
+ }
+
@Test
public void testHyperLogLog() {
for (int i = 0; i < NUM_ITERATIONS; i++) {
@@ -613,6 +645,17 @@ public class ObjectSerDeUtilsTest {
}
}
+ @Test
+ public void testLongArray() {
+ for (int i = 0; i < NUM_ITERATIONS; i++) {
+ long[] expected = RANDOM.longs(RANDOM.nextInt(100)).toArray();
+ byte[] bytes = ObjectSerDeUtils.LONG_ARRAY_SER_DE.serialize(expected);
+ assertEquals(bytes,
ObjectSerDeUtils.LONG_ARRAY_LIST_SER_DE.serialize(LongArrayList.wrap(expected)),
+ ERROR_MESSAGE);
+ assertEquals(ObjectSerDeUtils.LONG_ARRAY_SER_DE.deserialize(bytes),
expected, ERROR_MESSAGE);
+ }
+ }
+
@Test
public void testFloatArrayList() {
for (int i = 0; i < NUM_ITERATIONS; i++) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
