Jackie-Jiang commented on code in PR #18996:
URL: https://github.com/apache/pinot/pull/18996#discussion_r3591466702
##########
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileAggregationFunction.java:
##########
@@ -35,6 +35,8 @@
public class PercentileAggregationFunction extends
NullableSingleInputAggregationFunction<DoubleArrayList, Double> {
private static final double DEFAULT_FINAL_RESULT = Double.NEGATIVE_INFINITY;
+ private static final int SORT_THRESHOLD = 32;
+ private static final int NINTHER_THRESHOLD = 128;
Review Comment:
Can you document the meaning of these 2 thresholds? How are they used?
##########
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileTDigestAggregationFunction.java:
##########
@@ -108,20 +108,20 @@ public void aggregate(int length, AggregationResultHolder
aggregationResultHolde
if (blockValSet.getValueType() == DataType.BYTES) {
// Serialized TDigest
byte[][] bytesValues = blockValSet.getBytesValuesSV();
- foldNotNull(length, blockValSet, (TDigest)
aggregationResultHolder.getResult(), (tDigest, from, toEx) -> {
- if (tDigest != null) {
- for (int i = from; i < toEx; i++) {
-
tDigest.add(ObjectSerDeUtils.TDIGEST_SER_DE.deserialize(bytesValues[i]));
- }
- } else {
- tDigest =
ObjectSerDeUtils.TDIGEST_SER_DE.deserialize(bytesValues[0]);
- aggregationResultHolder.setValue(tDigest);
- for (int i = 1; i < length; i++) {
-
tDigest.add(ObjectSerDeUtils.TDIGEST_SER_DE.deserialize(bytesValues[i]));
+ PercentileTDigestAccumulator accumulator = foldNotNull(length,
blockValSet,
Review Comment:
Can we keep the same pattern?
##########
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileTDigestAccumulator.java:
##########
@@ -0,0 +1,448 @@
+/**
+ * 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.tdunning.math.stats.MergingDigest;
+import com.tdunning.math.stats.TDigest;
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+
+/// Accumulates raw values and serialized TDigests into centroids without
repeatedly sorting existing centroids.
+///
+/// The accumulator is used while a percentile TDigest result, or one of its
group-by results, is being built. Raw
+/// values are sorted in small batches, while serialized TDigest centroids are
decoded in their existing sorted order.
+/// Both are linearly merged with the accumulated centroids and compressed
using the same weight-limit rule as
+/// [MergingDigest]. [#toTDigest()] normalizes the result to a standard
[MergingDigest], so intermediate-result
+/// serialization and distributed merging remain unchanged.
+///
+/// Serialized group state keeps its first digest pending, and allocates
primitive centroid buffers only when another
+/// input must be merged. The raw-value buffer remains unallocated for
serialized state until a raw value is added.
+///
+/// Instances are thread-confined to one aggregation result or group-by result.
+final class PercentileTDigestAccumulator {
Review Comment:
Is this almost re-implementing the TDigest? Should we consider directly
implementing TDigest if the original library has poor support?
##########
pinot-core/src/test/java/org/apache/pinot/queries/PercentileTDigestMVQueriesTest.java:
##########
@@ -55,6 +56,12 @@
public class PercentileTDigestMVQueriesTest extends
PercentileTDigestQueriesTest {
private static final int MAX_NUM_MULTI_VALUES = 10;
+ @Override
+ @Test(enabled = false)
+ public void testStarTreeGroupBy() {
+ // The StarTree fixture stores the single-value TDigest aggregation only.
Review Comment:
Does this test fail? It should work right?
##########
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileTDigestAggregationFunction.java:
##########
@@ -268,21 +252,25 @@ protected void aggregateMVGroupByMV(int length, int[][]
groupKeysArray, GroupByR
@Override
public TDigest extractAggregationResult(AggregationResultHolder
aggregationResultHolder) {
- TDigest tDigest = aggregationResultHolder.getResult();
- if (tDigest == null) {
+ Object result = aggregationResultHolder.getResult();
+ if (result == null) {
return TDigest.createMergingDigest(_compressionFactor);
+ } else if (result instanceof PercentileTDigestAccumulator) {
+ return ((PercentileTDigestAccumulator) result).toTDigest();
} else {
- return tDigest;
+ return (TDigest) result;
Review Comment:
Is this possible to reach?
##########
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileTDigestAggregationFunction.java:
##########
@@ -268,21 +252,25 @@ protected void aggregateMVGroupByMV(int length, int[][]
groupKeysArray, GroupByR
@Override
public TDigest extractAggregationResult(AggregationResultHolder
aggregationResultHolder) {
- TDigest tDigest = aggregationResultHolder.getResult();
- if (tDigest == null) {
+ Object result = aggregationResultHolder.getResult();
+ if (result == null) {
return TDigest.createMergingDigest(_compressionFactor);
+ } else if (result instanceof PercentileTDigestAccumulator) {
+ return ((PercentileTDigestAccumulator) result).toTDigest();
} else {
- return tDigest;
+ return (TDigest) result;
}
}
@Override
public TDigest extractGroupByResult(GroupByResultHolder groupByResultHolder,
int groupKey) {
- TDigest tDigest = groupByResultHolder.getResult(groupKey);
- if (tDigest == null) {
+ Object result = groupByResultHolder.getResult(groupKey);
+ if (result == null) {
return TDigest.createMergingDigest(_compressionFactor);
+ } else if (result instanceof PercentileTDigestAccumulator) {
+ return ((PercentileTDigestAccumulator) result).toTDigest();
} else {
- return tDigest;
+ return (TDigest) result;
Review Comment:
Is this possible to reach?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
