xiangfu0 commented on code in PR #18996:
URL: https://github.com/apache/pinot/pull/18996#discussion_r3591721451
##########
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileTDigestAggregationFunction.java:
##########
@@ -215,15 +205,9 @@ public void aggregateGroupByMV(int length, int[][]
groupKeysArray, GroupByResult
byte[][] bytesValues = blockValSet.getBytesValuesSV();
forEachNotNull(length, blockValSet, (from, to) -> {
for (int i = from; i < to; i++) {
- TDigest value =
ObjectSerDeUtils.TDIGEST_SER_DE.deserialize(bytesValues[i]);
for (int groupKey : groupKeysArray[i]) {
- TDigest tDigest = groupByResultHolder.getResult(groupKey);
- if (tDigest != null) {
- tDigest.add(value);
- } else {
- // Create a new TDigest for the group
- groupByResultHolder.setValueForKey(groupKey,
ObjectSerDeUtils.TDIGEST_SER_DE.deserialize(bytesValues[i]));
- }
+ getSerializedAccumulator(groupByResultHolder, groupKey,
bytesValues[i])
+ .addSerializedTDigest(bytesValues[i]);
}
Review Comment:
Fixed in 633aab66c6. aggregateGroupByMV() now resets one invocation-local
SerializedTDigestInput once per non-null row, lazily decodes centroids once,
and reuses the read-only primitive arrays for every group key. New groups
retain only the serialized bytes, so raw and merge buffers stay lazy. Coverage
includes compact and verbose inputs, existing and new groups, duplicate keys,
empty digests, and zero-key rows.
##########
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:
Documented and renamed both constants in 633aab66c6. Direct sorting applies
to at most 32 candidate values; Tukey's ninther starts at 128 or more. Existing
selection tests bracket 31, 32, 33 and 127, 128, 129.
##########
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:
Restored the existing holder-set-inside-foldNotNull pattern in 633aab66c6.
The range loop still uses from and toExclusive so leading and interior nulls
remain correct.
##########
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:
Not on a valid execution path. Non-group aggregation state is now typed as
PercentileTDigestAccumulator, and 633aab66c6 removes the unreachable TDigest
fallback. Null remains only for empty or all-null input.
##########
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:
Yes. Raw numeric group-by still stores a TDigest, while serialized group-by
stores PercentileTDigestAccumulator. I kept both branches and added
testRawNumericGroupByStillUsesTDigestState to document the reachable TDigest
case.
##########
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:
Yes, it works. I enabled the inherited test and added a matching real
StarTree configuration to the MV fixture. The suite now passes all 6 tests,
including testStarTreeGroupBy. Expected total digest weight counts MV values
instead of documents, and the documents-scanned assertion confirms the StarTree
path.
##########
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:
Kept this as a package-private ingestion adapter, not a TDigest replacement.
It owns only raw and serialized ingestion plus linear centroid merging.
Quantile, CDF, centroid iteration, intermediate serialization, and distributed
merging remain on the canonical MergingDigest returned by toTDigest().
Implementing TDigest directly would duplicate the full library API and wire
format. The class documentation now makes that boundary explicit.
##########
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 {
+ private static final int MIN_RAW_BUFFER_SIZE = 256;
+ private static final int MAX_RAW_BUFFER_SIZE = 10_000;
+ private static final int CENTROID_CAPACITY_PADDING = 10;
+ private static final int VERBOSE_ENCODING = 1;
+ private static final int SMALL_ENCODING = 2;
+ private static final int VERBOSE_HEADER_SIZE = 32;
+ private static final int VERBOSE_CENTROID_SIZE = 16;
+ private static final int SMALL_CENTROID_SIZE = 8;
+
+ private final double _compression;
+ private final int _centroidCapacity;
+ private double[] _rawValues;
+ private double[] _centroidMeans;
+ private double[] _centroidWeights;
+ private double[] _outputMeans;
+ private double[] _outputWeights;
+ private double[] _incomingMeans;
+ private double[] _incomingWeights;
+ private byte[] _pendingSerializedTDigest;
+ private boolean _hasSerializedInput;
+
+ private int _numRawValues;
+ private int _numCentroids;
+ private double _totalWeight;
+ private double _min = Double.POSITIVE_INFINITY;
+ private double _max = Double.NEGATIVE_INFINITY;
+
+ PercentileTDigestAccumulator(int compression) {
+ this((double) compression, true, true);
+ }
+
+ private PercentileTDigestAccumulator(double compression, boolean
allocateRawBuffer, boolean allocateMergeBuffers) {
+ if (!(compression > 0.0) || !Double.isFinite(compression)) {
+ throw new IllegalArgumentException("TDigest compression must be
positive: " + compression);
+ }
+ _compression = compression;
+ long roundedCompression = (long) Math.ceil(compression);
+ _centroidCapacity = Math.toIntExact(
+ Math.addExact(Math.multiplyExact(2L, roundedCompression),
CENTROID_CAPACITY_PADDING));
+ if (allocateRawBuffer) {
+ _rawValues = new double[getRawBufferSize(roundedCompression)];
+ }
+ if (allocateMergeBuffers) {
+ _centroidMeans = new double[_centroidCapacity];
+ _centroidWeights = new double[_centroidCapacity];
+ _outputMeans = new double[_centroidCapacity];
+ _outputWeights = new double[_centroidCapacity];
+ }
+ }
+
+ static PercentileTDigestAccumulator forSerializedTDigest(byte[] bytes) {
+ return new PercentileTDigestAccumulator(readCompression(bytes), false,
false);
+ }
+
+ static PercentileTDigestAccumulator
forSerializedTDigestWithMergeBuffers(byte[] bytes) {
+ return new PercentileTDigestAccumulator(readCompression(bytes), false,
true);
+ }
+
+ void add(double[] values, int from, int toExclusive) {
+ if (from >= toExclusive) {
+ return;
+ }
+ materializePendingSerializedTDigest();
+ ensureRawBuffer();
+ while (from < toExclusive) {
+ int numValues = Math.min(toExclusive - from, _rawValues.length -
_numRawValues);
+ int rawOffset = _numRawValues;
+ for (int i = 0; i < numValues; i++) {
+ double value = values[from + i];
+ if (Double.isNaN(value)) {
+ throw new IllegalArgumentException("Cannot add NaN to t-digest");
+ }
+ _rawValues[rawOffset + i] = value;
+ }
+ from += numValues;
+ _numRawValues += numValues;
+ if (_numRawValues == _rawValues.length) {
+ flush();
+ }
+ }
+ }
+
+ void addSerializedTDigest(byte[] bytes) {
+ flush();
+ if (!_hasSerializedInput && _numCentroids == 0 && _totalWeight == 0.0) {
+ validateSerializedTDigest(bytes);
+ _pendingSerializedTDigest = bytes;
+ _hasSerializedInput = true;
+ return;
+ }
+ materializePendingSerializedTDigest();
+ mergeSerializedTDigest(bytes);
+ _hasSerializedInput = true;
+ }
+
+ private void mergeSerializedTDigest(byte[] bytes) {
+
+ ByteBuffer input = ByteBuffer.wrap(bytes);
+ int encoding = input.getInt();
+ double min = input.getDouble();
+ double max = input.getDouble();
+ int numCentroids;
+ boolean initialize = _numCentroids == 0 && _totalWeight == 0.0;
+ double[] incomingMeans;
+ double[] incomingWeights;
+ if (encoding == VERBOSE_ENCODING) {
+ input.getDouble();
+ numCentroids = input.getInt();
+ checkCentroidCount(numCentroids);
+ checkCentroidsAvailable(input, numCentroids, VERBOSE_CENTROID_SIZE);
+ if (initialize) {
+ ensureCentroidCapacity(numCentroids);
+ incomingMeans = _centroidMeans;
+ incomingWeights = _centroidWeights;
+ } else {
+ ensureIncomingCapacity(numCentroids);
+ incomingMeans = _incomingMeans;
+ incomingWeights = _incomingWeights;
+ }
+ for (int i = 0; i < numCentroids; i++) {
+ incomingWeights[i] = input.getDouble();
+ incomingMeans[i] = input.getDouble();
+ }
+ } else if (encoding == SMALL_ENCODING) {
+ input.getFloat();
+ input.getShort();
+ input.getShort();
+ numCentroids = input.getShort();
+ checkCentroidCount(numCentroids);
+ checkCentroidsAvailable(input, numCentroids, SMALL_CENTROID_SIZE);
+ if (initialize) {
+ ensureCentroidCapacity(numCentroids);
+ incomingMeans = _centroidMeans;
+ incomingWeights = _centroidWeights;
+ } else {
+ ensureIncomingCapacity(numCentroids);
+ incomingMeans = _incomingMeans;
+ incomingWeights = _incomingWeights;
+ }
+ for (int i = 0; i < numCentroids; i++) {
+ incomingWeights[i] = input.getFloat();
+ incomingMeans[i] = input.getFloat();
+ }
+ } else {
+ throw new IllegalStateException("Invalid format for serialized
histogram");
+ }
+
+ if (initialize) {
+ double totalWeight = getTotalWeight(incomingWeights, numCentroids);
+ if (totalWeight != 0.0) {
+ _numCentroids = numCentroids;
+ _totalWeight = totalWeight;
+ _min = min;
+ _max = max;
+ }
+ } else if (mergeSorted(incomingMeans, incomingWeights, numCentroids)) {
+ _min = Math.min(_min, min);
+ _max = Math.max(_max, max);
+ }
+ }
+
+ TDigest toTDigest() {
+ if (_pendingSerializedTDigest != null) {
+ return
MergingDigest.fromBytes(ByteBuffer.wrap(_pendingSerializedTDigest));
+ }
+ flush();
+ if (_numCentroids == 0) {
+ return TDigest.createMergingDigest(_compression);
+ }
+
+ ByteBuffer buffer = ByteBuffer.allocate(VERBOSE_HEADER_SIZE +
VERBOSE_CENTROID_SIZE * _numCentroids);
Review Comment:
Fixed in 633aab66c6. Materialized oversized compact input now preserves a
valid main and merge-buffer capacity. When the selected compression is not
exactly float-representable, centroids are recompressed and emitted verbose so
compression remains exact. Fully populated oversized verbose input and
malformed compact capacities still fail with canonical exception parity.
Regression coverage includes an 80-centroid compression-20 compact input, mixed
compression 20.0000001, a subsequent add, compact sentinels, and malformed
capacities.
--
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]
