xiangfu0 commented on code in PR #19795:
URL: https://github.com/apache/pinot/pull/19795#discussion_r4229474611
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TDigestUtils.java:
##########
@@ -202,338 +334,321 @@ public static byte[] makeLegacyCompatible(byte[]
verboseBytes) {
return small.array();
}
+ private static byte[] serializeRecoveredCentroids(SerializedTDigestMetadata
metadata, double[] means,
+ double[] weights) {
+ PercentileTDigestAccumulator digest =
PercentileTDigestAccumulator.forLegacyAggregation(metadata.compression());
+ digest.addCentroids(means, weights, means.length, metadata.min(),
metadata.max(), true);
+ return digest.serialize();
+ }
+
+ private static byte[] removeZeroWeightCentroids(byte[] verboseBytes,
SerializedTDigestMetadata metadata) {
+ ByteBuffer source = ByteBuffer.wrap(verboseBytes);
+ source.position(metadata.centroidOffset());
+ ByteBuffer filtered = ByteBuffer.allocate(metadata.encodedLength());
+ filtered.put(verboseBytes, 0, VERBOSE_HEADER_SIZE);
+ int count = 0;
+ for (int i = 0; i < metadata.centroidCount(); i++) {
+ double weight = source.getDouble();
+ double mean = source.getDouble();
+ if (weight != 0.0) {
+ filtered.putDouble(weight).putDouble(mean);
+ count++;
+ }
+ }
+ filtered.putInt(28, count);
+ return Arrays.copyOf(filtered.array(), filtered.position());
+ }
+
/// Deserializes a digest and repairs boundary centroids produced by
t-digest 3.2 when necessary.
public static TDigest deserialize(byte[] bytes) {
return deserialize(ByteBuffer.wrap(bytes));
}
- /// Validates the structure and declared capacities of a serialized merging
digest.
- public static void validateSerialized(byte[] bytes) {
- validateEncodedDigest(ByteBuffer.wrap(bytes));
+ /// Validates serialized lengths and centroid values without allocating
centroid arrays.
+ public static double validateSerialized(byte[] bytes) {
+ return validateSerialized(ByteBuffer.wrap(bytes));
}
- /// Deserializes a finite-valued digest into a merging digest with the
temporary-buffer footprint used by 3.2.
+ /// Validates one legacy digest without changing `input` and returns its
total centroid weight.
///
- /// Unlike the standard verbose decoder, this method does not retain the
substantially larger default 3.3
- /// temporary arrays. It also repairs legacy weighted boundary centroids
before the first 3.3 merge.
- public static TDigest deserializeFiniteWithLegacyBuffer(byte[] bytes) {
- ByteBuffer input = ByteBuffer.wrap(bytes);
- validateEncodedDigest(input.duplicate());
+ /// Trailing bytes remain available to the caller for intermediate values
containing several objects.
+ public static double validateSerialized(ByteBuffer input) {
+ return validateSerialized(input, true);
+ }
- int encoding = input.getInt();
- double min = input.getDouble();
- double max = input.getDouble();
- double compression;
- int centroidCount;
- boolean verbose;
- if (encoding == VERBOSE_ENCODING) {
- compression = input.getDouble();
- centroidCount = input.getInt();
- verbose = true;
- } else if (encoding == SMALL_ENCODING) {
- compression = input.getFloat();
- input.getShort();
- input.getShort();
- centroidCount = input.getShort();
- verbose = false;
- } else {
- throw new IllegalStateException("Invalid format for serialized
histogram");
- }
+ private static double validateSerialized(ByteBuffer input, boolean
checkCapacity) {
+ return inspectSerialized(input, checkCapacity).totalWeight();
+ }
- if (centroidCount == 0) {
- return createMergingDigestWithLegacyBuffer(compression);
- }
- double[] means = new double[centroidCount + 2];
- double[] weights = new double[centroidCount + 2];
- for (int i = 0; i < centroidCount; i++) {
- double weight = verbose ? input.getDouble() : input.getFloat();
- double mean = verbose ? input.getDouble() : input.getFloat();
- if (!(weight > 0.0) || !Double.isFinite(weight)) {
- throw new IllegalArgumentException("Invalid TDigest centroid weight: "
+ weight);
- }
- if (!Double.isFinite(mean)) {
- throw new IllegalArgumentException("Expected a finite TDigest centroid
mean: " + mean);
- }
- weights[i] = weight;
- means[i] = mean;
- }
-
- int normalizedCount = weights[0] != 1.0 || weights[centroidCount - 1] !=
1.0
- ? normalizeBoundaries(means, weights, centroidCount, min, max) :
centroidCount;
- long addCount = 0L;
- for (int i = 0; i < normalizedCount; i++) {
- double weight = weights[i];
- if (weight != Math.rint(weight) || weight >= 0x1p63) {
- return copyIntoLegacyBuffer(
- deserialize(toVerboseBytes(min, max, compression, means, weights,
normalizedCount)), normalizedCount);
- }
- long integralWeight = (long) weight;
- addCount = Math.addExact(addCount, 1L + (integralWeight - 1L) /
Integer.MAX_VALUE);
- }
-
- int minimumMainCapacity = Math.addExact(Math.toIntExact(addCount), 2);
- TDigest digest = createMergingDigestWithLegacyBuffer(compression,
minimumMainCapacity);
- for (int i = 0; i < normalizedCount; i++) {
- long remainingWeight = (long) weights[i];
- while (remainingWeight > 0L) {
- int weight = (int) Math.min(remainingWeight, Integer.MAX_VALUE);
- digest.add(means[i], weight);
- remainingWeight -= weight;
- }
- }
- return digest;
+ /// Header and numerical metadata for one legacy byte payload. The centroid
bytes remain owned by the caller.
+ /// Header-only inspection leaves total weight as NaN until centroid
inspection validates the numerical fields.
+ ///
+ /// Legacy writers can emit poisoned numerical state after overflow or
broken fractional boundary repair.
+ /// Such state is retained with degraded quantiles, rather than being passed
to the sorted K1 kernel.
+ public record SerializedTDigestMetadata(int encoding, double min, double
max, double compression,
+ int centroidCount, int mainCapacity, int bufferCapacity, int
centroidOffset, int centroidSize,
+ int encodedLength, double totalWeight, boolean hasNonFiniteMeans,
boolean needsLegacyFallback,
+ boolean unorderedMeans, boolean fractionalWeights, boolean
weightedBoundaries, boolean hasZeroWeightCentroids,
+ double encodedCompression, double recoveredInfinityMean) {
}
- /// Deserializes a digest and advances `input` past its encoded bytes.
- public static TDigest deserialize(ByteBuffer input) {
- ByteBuffer encoded = input.slice();
- validateEncodedDigest(encoded.duplicate());
+ /// Inspects one payload without mutating `input`, scanning the centroid
values once.
+ public static SerializedTDigestMetadata inspectSerialized(ByteBuffer input) {
+ return inspectSerialized(input, true);
+ }
+ /// Inspects a payload, optionally enforcing the legacy decoder's declared
centroid capacity.
+ public static SerializedTDigestMetadata inspectSerialized(ByteBuffer input,
boolean checkCapacity) {
+ SerializedTDigestMetadata header = readSerializedHeader(input,
checkCapacity);
+ return Double.isNaN(header.totalWeight()) ? inspectSerialized(input,
header, null, null) : header;
+ }
+
+ /// Reads the fixed-size header without allocating centroid arrays;
oversized historical state is also inspected.
+ public static SerializedTDigestMetadata readSerializedHeader(ByteBuffer
input) {
+ return readSerializedHeader(input, true);
+ }
+
+ /// Reads a fixed-size header, optionally enforcing the declared centroid
capacity.
+ ///
+ /// NaN extrema mark legacy degraded state. Oversized payloads require
numerical inspection: historical degraded
+ /// state remains readable by Pinot even if an old decoder's default arrays
could not hold its original centroids.
+ public static SerializedTDigestMetadata readSerializedHeader(ByteBuffer
input, boolean checkCapacity) {
+ ByteBuffer encoded = input.slice().order(ByteOrder.BIG_ENDIAN);
int encoding = encoded.getInt();
+ if (encoding != VERBOSE_ENCODING && encoding != SMALL_ENCODING) {
+ throw new IllegalStateException("Invalid format for serialized
histogram");
+ }
double min = encoded.getDouble();
double max = encoded.getDouble();
double compression;
+ double encodedCompression;
int mainCapacity;
- int bufferCapacity;
+ int bufferCapacity = 0;
int centroidCount;
int centroidSize;
if (encoding == VERBOSE_ENCODING) {
- compression = encoded.getDouble();
+ encodedCompression = encoded.getDouble();
+ compression = normalizeCompression(encodedCompression);
+ mainCapacity = Math.max(getDefaultCentroidCapacity(compression),
getLegacyDefaultCentroidCapacity(compression));
centroidCount = encoded.getInt();
- mainCapacity = getDefaultCentroidCapacity(compression);
- bufferCapacity = 5 * mainCapacity;
centroidSize = VERBOSE_CENTROID_SIZE;
- } else if (encoding == SMALL_ENCODING) {
- compression = encoded.getFloat();
+ } else {
+ encodedCompression = encoded.getFloat();
+ compression = normalizeCompression(encodedCompression);
mainCapacity = encoded.getShort();
bufferCapacity = encoded.getShort();
+ checkSmallArrayCapacity(mainCapacity);
+ checkSmallArrayCapacity(bufferCapacity);
+ if (mainCapacity == -1) {
+ mainCapacity = getDefaultCentroidCapacity(compression);
+ }
centroidCount = encoded.getShort();
centroidSize = SMALL_CENTROID_SIZE;
- } else {
- // MergingDigest.fromBytes() has already produced the canonical
exception for this case.
- throw new IllegalStateException("Invalid format for serialized
histogram");
}
- int centroidOffset = encoded.position();
- if (centroidCount == 0) {
- return configureScaleFunction(MergingDigest.fromBytes(input));
+ if (centroidCount < 0) {
+ throw new IllegalArgumentException("Invalid negative TDigest centroid
count: " + centroidCount);
}
-
- double firstWeight = readWeight(encoded, centroidOffset, centroidSize);
- double lastWeight = readWeight(encoded, centroidOffset + (centroidCount -
1) * centroidSize, centroidSize);
- int defaultMainCapacity = getDefaultCentroidCapacity(compression);
- boolean needsBoundaryRepair = firstWeight != 1.0 || lastWeight != 1.0;
- boolean exceedsDefaultVerboseCapacity = encoding == VERBOSE_ENCODING &&
centroidCount > defaultMainCapacity;
- if (!needsBoundaryRepair && !exceedsDefaultVerboseCapacity) {
- return configureScaleFunction(MergingDigest.fromBytes(input));
+ if (centroidCount > encoded.remaining() / centroidSize) {
+ throw new BufferUnderflowException();
+ }
+ if (centroidCount > 0 && min > max) {
Review Comment:
**Inconsistency: inverted extrema hard-throw on every read while NaN extrema
get legacy-fallback retention.**
A historical corrupt payload with `min > max` (and `centroidCount > 0`)
throws `IllegalArgumentException` from `readSerializedHeader` on every path
(queries over stored BYTES, deserialize, validateSerialized), making the stored
column unqueryable, while NaN extrema two lines below are retained via
`needsLegacyFallback`. Master's reader (`validateEncodedDigest` + tdunning
`fromBytes`) accepted such bytes, and this file's doc promises poisoned states
"remain byte-readable with NaN statistics". No known legacy writer produces
`count>0 && min>max` (empty tdunning digests write +Inf/-Inf with count 0), so
this is corruption-only — but for consistency with the stated design, inverted
extrema should degrade to fallback rather than throw.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -0,0 +1,1766 @@
+/**
+ * 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 java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import org.apache.pinot.segment.local.utils.TDigestUtils;
+import
org.apache.pinot.segment.local.utils.TDigestUtils.SerializedTDigestMetadata;
+
+
+
+/// Accumulates raw values and serialized TDigests into primitive centroids
using Pinot's K1 weight-limit rule.
+///
+/// Raw values are sorted in small batches; serialized centroids retain their
existing sorted order. Both are
+/// linearly merged with accumulated centroids. K1 preserves Pinot's existing
middle-quantile behavior, and the
+/// legacy verbose and small encodings preserve stored data and mixed-version
wire compatibility.
+///
+/// Serialized state keeps its first digest pending and allocates centroid
buffers only when another input must be
+/// merged or the digest is queried. Raw-value buffers remain unallocated
until a raw value is added.
+///
+/// Historical corrupted payloads remain opaque and byte-exact, with NaN
statistics. Their reported historical mass
+/// may be invalid; they cannot be mutated or mixed with another distribution.
Such an operation fails before
+/// changing aggregate mass, because neither invented statistics nor synthetic
NaN bytes are safe for old readers.
+/// Unchanged historical fractional boundaries retain their original bytes,
including when configured compression
+/// differs. After a merge, inherited fractional endpoints retain precise
verbose weights and finite statistics when
+/// unit endpoints cannot preserve their mass. These keep the historical
t-digest 3.3 assertion-enabled reader and
+/// recompression limitation. Newly introduced unsupported fractional
endpoints are rejected.
+///
+/// Instances are externally serialized per aggregation or group key and are
not safe for concurrent mutation.
+public final class PercentileTDigestAccumulator extends TDigest {
+ private static final int MIN_RAW_BUFFER_SIZE = 256;
+ private static final int MAX_RAW_BUFFER_SIZE = 10_000;
+ private static final int MIN_COMPRESSION = 10;
+ private static final int TWO_LEVEL_COMPRESSION_MULTIPLIER = 2;
+ private static final int MIN_INCREMENTAL_COMPRESSION = 50;
+
+ private final double _compression;
+ private final boolean _useTwoLevelRawCompression;
+ private final double _workingCompression;
+ private final double _publicMaxWeightScale;
+ private final double _workingMaxWeightScale;
+ private final int _centroidCapacity;
+ private final int _pendingCentroidCapacity;
+ private double[] _rawValues;
+ private double[] _centroidMeans;
+ private double[] _centroidWeights;
+ private double[] _outputMeans;
+ private double[] _outputWeights;
+ private double[] _incomingMeans;
+ private double[] _incomingWeights;
+ private double[] _sortedIncomingMeans;
+ private double[] _sortedIncomingWeights;
+ private SerializedTDigestInput _serializedTDigestInput;
+ private byte[] _pendingSerializedTDigest;
+ private byte[] _originalFractionalBytes;
+ private double _historicalFractionalMinMean = Double.NaN;
+ private double _historicalFractionalMaxMean = Double.NaN;
+ private SerializedTDigestMetadata _pendingSerializedMetadata;
+ private boolean _legacyDegraded;
+ private byte[] _serializedBytesForWrite;
+ private boolean _hasFractionalWeights;
+
+ private int _numRawValues;
+ private int _numCentroids;
+ private int _numIncomingCentroids;
+ private int _serializedMainCapacity;
+ private int _serializedBufferCapacity;
+ private int _mergeCount;
+ private boolean _publiclyCompressed;
+ private boolean _incomingCentroidsSorted = true;
+ private double _totalWeight;
+ private double _incomingWeight;
+ private double _min = Double.POSITIVE_INFINITY;
+ private double _max = Double.NEGATIVE_INFINITY;
+
+ public PercentileTDigestAccumulator(int compression) {
+ this(positiveCompression(compression), true, true);
+ }
+
+ public PercentileTDigestAccumulator(double compression) {
+ this(compression, true, true);
+ }
+
+ private PercentileTDigestAccumulator(double compression, boolean
allocateRawBuffer, boolean allocateMergeBuffers) {
+ this(compression, allocateRawBuffer, allocateMergeBuffers, false);
+ }
+
+ private PercentileTDigestAccumulator(double compression, boolean
allocateRawBuffer, boolean allocateMergeBuffers,
+ boolean useTwoLevelRawCompression) {
+ TDigestUtils.validateCompression(compression);
+ _useTwoLevelRawCompression = useTwoLevelRawCompression;
+ _compression = Math.max(MIN_COMPRESSION, compression);
+ _workingCompression = TWO_LEVEL_COMPRESSION_MULTIPLIER * _compression;
+ _publicMaxWeightScale = calculateK1MaxWeightScale(_compression);
+ _workingMaxWeightScale = calculateK1MaxWeightScale(_workingCompression);
+ long roundedCompression = (long) Math.ceil(_compression);
+ _centroidCapacity = TDigestUtils.getDefaultCentroidCapacity(_compression);
+ _pendingCentroidCapacity = getPendingCentroidCapacity(_centroidCapacity,
useTwoLevelRawCompression);
+ if (allocateRawBuffer) {
+ _rawValues = new double[getInitialRawBufferSize(roundedCompression)];
+ }
+ if (allocateMergeBuffers) {
+ int initialCapacity = Math.min(_centroidCapacity, MIN_RAW_BUFFER_SIZE);
+ _centroidMeans = new double[initialCapacity];
+ _centroidWeights = new double[initialCapacity];
+ _outputMeans = new double[initialCapacity];
+ _outputWeights = new double[initialCapacity];
+ }
+ }
+
+ /// Creates a lazy digest with the legacy two-level compression used by
Star-tree and general digest aggregation.
+ public static PercentileTDigestAccumulator forLegacyAggregation(double
compression) {
+ return new PercentileTDigestAccumulator(compression, false, false, true);
+ }
+
+ public static PercentileTDigestAccumulator forSerializedTDigest(byte[]
bytes) {
+ return new PercentileTDigestAccumulator(readCompression(bytes), false,
false);
+ }
+
+ public static PercentileTDigestAccumulator
forSerializedTDigest(SerializedTDigestInput input) {
+ return new PercentileTDigestAccumulator(input._metadata.compression(),
false, false);
+ }
+
+ public static PercentileTDigestAccumulator
forSerializedTDigestWithMergeBuffers(byte[] bytes) {
+ return new PercentileTDigestAccumulator(readCompression(bytes), false,
true);
+ }
+
+ public static PercentileTDigestAccumulator forSerializedTDigest(ByteBuffer
input) {
+ SerializedTDigestInput serialized = new SerializedTDigestInput();
+ serialized.reset(input);
+ PercentileTDigestAccumulator accumulator =
forSerializedTDigest(serialized);
+ accumulator.addSerializedTDigest(serialized);
+ return accumulator;
+ }
+
+ public static PercentileTDigestAccumulator forReduction(double compression) {
+ return new PercentileTDigestAccumulator(positiveCompression(compression),
false, false);
+ }
+
+ /// Copies primitive centroids without encoding and decoding them. Zero
weights are ignored, and unordered
+ /// means are sorted when initializing a stored, already compressed
distribution; caller arrays are not mutated.
+ /// Subsequent inputs use the normal buffered merge. Positive weights and
consistent extrema are required.
+ public void addCentroids(double[] means, double[] weights, int count, double
min, double max,
+ boolean alreadyCompressed) {
+ requireMutable();
+ if (count < 0 || count > means.length || count > weights.length) {
+ throw new IllegalArgumentException("Invalid TDigest centroid count: " +
count);
+ }
+ double totalWeight = 0.0;
+ int positiveCount = 0;
+ boolean sorted = true;
+ boolean fractionalWeights = false;
+ double previousMean = Double.NEGATIVE_INFINITY;
+ for (int i = 0; i < count; i++) {
+ double weight = weights[i];
+ if (weight == 0.0) {
+ continue;
+ }
+ double mean = means[i];
+ if (!(weight > 0.0) || !Double.isFinite(weight) || !(mean >= min && mean
<= max)) {
+ throw new IllegalArgumentException("Invalid TDigest centroid");
+ }
+ totalWeight += weight;
+ sorted &= mean >= previousMean;
+ previousMean = mean;
+ fractionalWeights |= weight != Math.rint(weight);
+ positiveCount++;
+ }
+ checkTotalWeight(getTotalWeight() + totalWeight);
+ if (positiveCount == 0) {
+ return;
+ }
+ materializePendingSerializedTDigest();
+ _originalFractionalBytes = null;
+ _hasFractionalWeights |= fractionalWeights;
+ if (hasNoInputs() && alreadyCompressed) {
+ ensureCentroidCapacity(Math.addExact(positiveCount, 2));
+ for (int i = 0; i < count; i++) {
+ if (weights[i] > 0.0) {
+ _centroidMeans[_numCentroids] = means[i];
+ _centroidWeights[_numCentroids++] = weights[i];
+ }
+ }
+ if (!sorted) {
+ it.unimi.dsi.fastutil.Arrays.mergeSort(0, _numCentroids,
+ (first, second) -> Double.compare(_centroidMeans[first],
_centroidMeans[second]),
+ (first, second) -> {
+ double mean = _centroidMeans[first];
+ _centroidMeans[first] = _centroidMeans[second];
+ _centroidMeans[second] = mean;
+ double weight = _centroidWeights[first];
+ _centroidWeights[first] = _centroidWeights[second];
+ _centroidWeights[second] = weight;
+ });
+ }
+ _totalWeight = totalWeight;
+ _min = min;
+ _max = max;
+ normalizeBoundaryCentroids();
+ _publiclyCompressed = true;
+ } else {
+ for (int i = 0; i < count; i++) {
+ if (weights[i] > 0.0) {
+ bufferIncomingCentroid(means[i], weights[i]);
+ }
+ }
+ _min = Math.min(_min, min);
+ _max = Math.max(_max, max);
+ }
+ }
+
+ public void add(double[] values, int from, int toExclusive) {
+ if (from >= toExclusive) {
+ return;
+ }
+ requireMutable();
+ materializePendingSerializedTDigest();
+ while (from < toExclusive) {
+ ensureRawBuffer();
+ if (_numRawValues == _rawValues.length
+ || _numRawValues + _numIncomingCentroids == getPendingInputLimit()) {
+ flush();
+ }
+ int numValues = Math.min(toExclusive - from, Math.min(_rawValues.length
- _numRawValues,
+ getPendingInputLimit() - _numRawValues - _numIncomingCentroids));
+ 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;
+ }
+ _originalFractionalBytes = null;
+ from += numValues;
+ _numRawValues += numValues;
+ if (_numRawValues == _rawValues.length
+ && (!_useTwoLevelRawCompression || _rawValues.length ==
_pendingCentroidCapacity)
+ || _numRawValues + _numIncomingCentroids == getPendingInputLimit()) {
+ flush();
+ }
+ }
+ }
+
+ @Override
+ public void add(double value) {
+ if (Double.isNaN(value)) {
+ throw new IllegalArgumentException("Cannot add NaN to t-digest");
+ }
+ requireMutable();
+ materializePendingSerializedTDigest();
+ ensureRawBuffer();
+ if (_numRawValues == _rawValues.length
+ || _numRawValues + _numIncomingCentroids == getPendingInputLimit()) {
+ flush();
+ }
+ _originalFractionalBytes = null;
+ _rawValues[_numRawValues++] = value;
+ }
+
+ @Override
+ public void add(double value, double weight) {
+ if (weight == 1) {
+ add(value);
+ return;
+ }
+ if (Double.isNaN(value)) {
+ throw new IllegalArgumentException("Cannot add NaN to t-digest");
+ }
+ if (!(weight > 0.0) || !Double.isFinite(weight)) {
+ throw new IllegalArgumentException("TDigest weight must be positive: " +
weight);
+ }
+ requireMutable();
+ materializePendingSerializedTDigest();
+ bufferIncomingCentroid(value, weight);
+ _originalFractionalBytes = null;
+ _hasFractionalWeights |= weight != Math.rint(weight);
+ _min = Math.min(_min, value);
+ _max = Math.max(_max, value);
+ }
+
+ @Override
+ public void add(TDigest other) {
+ requireMutable();
+ if (other instanceof PercentileTDigestAccumulator) {
+ addAccumulator((PercentileTDigestAccumulator) other);
+ return;
+ }
+ if (!other.hasValidStatistics()) {
+ materializePendingSerializedTDigest();
+ SerializedTDigestInput input = getSerializedTDigestInput();
+ input.reset(ByteBuffer.wrap(TDigestUtils.serialize(other)), false);
+ addSerializedTDigest(input);
+ return;
+ }
+ materializePendingSerializedTDigest();
+ double otherWeight = other.getTotalWeight();
+ checkTotalWeight(_totalWeight + _incomingWeight + _numRawValues +
otherWeight);
+ for (Centroid centroid : other.centroids()) {
+ double weight = centroid.weight();
+ if (weight == 0.0) {
+ continue;
+ }
+ if (!(weight > 0.0) || !Double.isFinite(weight) ||
Double.isNaN(centroid.mean())) {
+ throw new IllegalArgumentException("Invalid TDigest centroid");
+ }
+ _hasFractionalWeights |= weight != Math.rint(weight);
+ bufferIncomingCentroid(centroid.mean(), weight);
+ _originalFractionalBytes = null;
+ }
+ if (otherWeight != 0.0) {
+ _min = Math.min(_min, other.getMin());
+ _max = Math.max(_max, other.getMax());
+ inheritHistoricalFractionalBoundaries(other);
+ }
+ }
+
+ @Override
+ public void add(List<? extends TDigest> others) {
+ for (TDigest other : others) {
+ add(other);
+ }
+ }
+
+ private void addAccumulator(PercentileTDigestAccumulator other) {
+ if (other == this) {
+ compress();
+ if (_numCentroids == 0) {
+ return;
+ }
+ // Snapshot only self-merges; fresh fractional mass need not be
representable by a legacy wire payload.
+ addCentroids(Arrays.copyOf(_centroidMeans, _numCentroids),
Arrays.copyOf(_centroidWeights, _numCentroids),
+ _numCentroids, _min, _max, false);
+ return;
+ }
+ if (other._originalFractionalBytes != null && getTotalWeight() == 0.0) {
+ materializePendingSerializedTDigest();
+ SerializedTDigestInput input = getSerializedTDigestInput();
+ input.reset(other._originalFractionalBytes);
+ input._retainedBytes = other._originalFractionalBytes;
+ addSerializedTDigest(input);
+ return;
+ }
+ if (other._pendingSerializedTDigest != null) {
+ // Materialize before resetting the reusable reader: materialization
uses that same reader.
+ materializePendingSerializedTDigest();
+ SerializedTDigestInput input = getSerializedTDigestInput();
+ input.reset(other._pendingSerializedTDigest,
other._pendingSerializedMetadata);
+ input._retainedBytes = other._pendingSerializedTDigest;
+ addSerializedTDigest(input);
+ return;
+ }
+ if (other._legacyDegraded) {
+ addSerializedTDigest(other.serialize());
+ return;
+ }
+
+ // MergingDigest compressed merge sources in place. Aggregation owns these
mutable inputs exclusively.
+ other.compress();
+ materializePendingSerializedTDigest();
+ checkTotalWeight(_totalWeight + _incomingWeight + _numRawValues +
other._totalWeight);
+ _hasFractionalWeights |= other._hasFractionalWeights;
+ inheritHistoricalFractionalBoundaries(other._historicalFractionalMinMean,
other._historicalFractionalMaxMean);
+ preserveSerializedCapacity(other._serializedMainCapacity,
other._serializedBufferCapacity);
+ bufferIncomingCentroids(other._centroidMeans, other._centroidWeights,
other._numCentroids);
+ if (other._totalWeight > 0.0) {
+ _originalFractionalBytes = null;
+ }
+ _min = Math.min(_min, other._min);
+ _max = Math.max(_max, other._max);
+ }
+
+ public void addSerializedTDigest(byte[] bytes) {
+ if (_legacyDegraded) {
+ SerializedTDigestInput input = getSerializedTDigestInput();
+ input.reset(bytes);
+ addSerializedTDigest(input);
+ return;
+ }
+ requireMutable();
+ materializePendingSerializedTDigest();
+ SerializedTDigestInput input = getSerializedTDigestInput();
+ input.reset(bytes);
+ addSerializedTDigest(input);
+ }
+
+ private SerializedTDigestInput getSerializedTDigestInput() {
+ if (_serializedTDigestInput == null) {
+ _serializedTDigestInput = new SerializedTDigestInput();
+ }
+ return _serializedTDigestInput;
+ }
+
+ public void addSerializedTDigest(SerializedTDigestInput input) {
+ if (ignoreEmptyInputForDegradedState(input)) {
+ return;
+ }
+ requireMutable();
+ if (prepareSerializedTDigest(input)) {
+ mergeSerializedTDigest(input, false);
+ }
+ }
+
+ public void addSerializedTDigestDirect(SerializedTDigestInput input) {
+ if (ignoreEmptyInputForDegradedState(input)) {
+ return;
+ }
+ requireMutable();
+ if (prepareSerializedTDigest(input)) {
+ mergeSerializedTDigest(input, true);
+ }
+ }
+
+ private boolean ignoreEmptyInputForDegradedState(SerializedTDigestInput
input) {
+ if (!_legacyDegraded) {
+ return false;
+ }
+ input.inspectMetadata();
+ return !input._metadata.needsLegacyFallback() &&
input._metadata.totalWeight() == 0.0;
+ }
+
+ private boolean prepareSerializedTDigest(SerializedTDigestInput input) {
+ if (_pendingSerializedTDigest == null && hasNoInputs()) {
+ input.inspectMetadata();
+ _hasFractionalWeights |= input._metadata.fractionalWeights();
+ _pendingSerializedTDigest = input.retainBytes();
+ _pendingSerializedMetadata = input._metadata;
+ if (input.recordHistoricalFractionalBoundaries(this)) {
+ // Alias the retained snapshot; historical fractional bytes remain
byte-exact after read-only materialization.
+ _originalFractionalBytes = _pendingSerializedTDigest;
+ }
+ if (input._metadata.needsLegacyFallback()) {
+ _legacyDegraded = true;
+ }
+ return false;
+ }
+ materializePendingSerializedTDigest();
+ input.decode();
+ if (input._metadata.needsLegacyFallback()) {
+ if (hasNoInputs()) {
+ _pendingSerializedTDigest = input.retainBytes();
+ _pendingSerializedMetadata = input._metadata;
+ _legacyDegraded = true;
+ return false;
+ }
+ throw corruptedMutation();
+ }
+ checkTotalWeight(_totalWeight + _incomingWeight + _numRawValues +
input._metadata.totalWeight());
+ input.recordHistoricalFractionalBoundaries(this);
+ if (input._metadata.totalWeight() > 0.0) {
+ _originalFractionalBytes = null;
+ }
+ _hasFractionalWeights |= input._metadata.fractionalWeights();
+ return true;
+ }
+
+ private boolean hasNoInputs() {
+ return _numCentroids == 0 && _totalWeight == 0.0 && _numRawValues == 0 &&
_numIncomingCentroids == 0;
+ }
+
+ private void mergeSerializedTDigest(SerializedTDigestInput input, boolean
direct) {
+ if (input._metadata.totalWeight() == 0.0) {
+ return;
+ }
+ if (input._metadata.encoding() == TDigestUtils.SMALL_ENCODING) {
+ preserveSerializedCapacity(input._metadata.mainCapacity(),
input._metadata.bufferCapacity());
+ }
+ if (hasNoInputs()) {
+ if (input._metadata.totalWeight() != 0.0) {
+ ensureCentroidCapacity(input._numCentroids);
+ System.arraycopy(input._means, 0, _centroidMeans, 0,
input._numCentroids);
+ System.arraycopy(input._weights, 0, _centroidWeights, 0,
input._numCentroids);
+ _numCentroids = input._numCentroids;
+ _totalWeight = input._metadata.totalWeight();
+ _min = input._metadata.min();
+ _max = input._metadata.max();
+ // Stored centroids already went through the writer's public
compression. Reading or repairing their
+ // endpoints needs another K1 pass only when the output compression
differs from the stored compression.
+ _publiclyCompressed = _compression == input._metadata.compression();
+ }
+ return;
+ }
+ if (direct) {
+ flush();
+ boolean runBackwards = (_mergeCount++ & 1) != 0;
+ if (!mergeSorted(input._means, input._weights, input._numCentroids,
input._metadata.totalWeight(),
+ getIncrementalCompression(), runBackwards)) {
+ return;
+ }
+ } else {
+ bufferIncomingCentroids(input._means, input._weights,
input._numCentroids);
+ }
+ _min = Math.min(_min, input._metadata.min());
+ _max = Math.max(_max, input._metadata.max());
+ }
+
+ @Override
+ public void compress() {
+ if (_legacyDegraded) {
+ return;
+ }
+ if (_pendingSerializedTDigest != null || _numRawValues > 0 ||
_numIncomingCentroids > 0
+ || !_publiclyCompressed && _numCentroids > 0) {
+ _serializedBytesForWrite = null;
+ }
+ materializePendingSerializedTDigest();
+ if (_numRawValues > 0 || _numIncomingCentroids > 0) {
+ // MergingDigest 3.3 combines pending values and existing centroids
directly at public compression. Flushing at
+ // working compression first would add an extra lossy merge pass and
materially increase reducer rank error.
+ flush(_compression);
+ _publiclyCompressed = true;
+ } else if (!_publiclyCompressed && _numCentroids > 0) {
+ recompressCentroids(_compression);
+ _publiclyCompressed = true;
+ }
+ }
+
+ @Override
+ public long size() {
+ return (long) getTotalWeight();
+ }
+
+ @Override
+ public boolean hasValidStatistics() {
+ return !_legacyDegraded;
+ }
+
+ @Override
+ public double getTotalWeight() {
+ return _pendingSerializedTDigest != null ?
_pendingSerializedMetadata.totalWeight()
+ : _totalWeight + _incomingWeight + _numRawValues;
+ }
+
+ /// Bounds serialized output centroids from buffered inputs, including up to
two boundary splits before a merge
+ /// and two more before writing, without flushing or using the potentially
huge input mass.
+ public long getCentroidCountUpperBound() {
+ return _pendingSerializedTDigest != null ? (long)
_pendingSerializedMetadata.centroidCount() + 4L
+ : (long) _numCentroids + _numRawValues + _numIncomingCentroids + 4L;
+ }
+
+ /// Whether unchanged historical fractional boundaries still require their
original byte representation.
+ public boolean hasOriginalFractionalPayload() {
+ return _originalFractionalBytes != null;
+ }
+
+ @Override
+ public double cdf(double value) {
+ if (Double.isNaN(value) || Double.isInfinite(value)) {
+ throw new IllegalArgumentException(String.format("Invalid value: %f",
value));
+ }
+ if (_legacyDegraded) {
+ return Double.NaN;
+ }
+ materializePendingSerializedTDigest();
+ flush();
+ if (_numCentroids == 0) {
+ return Double.NaN;
+ }
+ if (_numCentroids == 1) {
+ double width = _max - _min;
+ if (value < _min) {
+ return 0.0;
+ } else if (value > _max) {
+ return 1.0;
+ } else if (value - _min <= width) {
+ return 0.5;
+ } else {
+ return (value - _min) / width;
+ }
+ }
+ if (value < _min) {
+ return 0.0;
+ }
+ if (value > _max) {
+ return 1.0;
+ }
+ if (value < _centroidMeans[0]) {
+ double width = _centroidMeans[0] - _min;
+ if (width > 0.0) {
+ if (_centroidWeights[0] < 1.0) {
+ return (_centroidWeights[0] / 2.0 * (value - _min) / width) /
_totalWeight;
+ }
+ return value == _min ? 0.5 / _totalWeight
+ : (1.0 + (value - _min) / width * (_centroidWeights[0] / 2.0 -
1.0)) / _totalWeight;
+ }
+ return 0.0;
+ }
+ int lastIndex = _numCentroids - 1;
+ if (value > _centroidMeans[lastIndex]) {
+ double width = _max - _centroidMeans[lastIndex];
+ if (width > 0.0) {
+ if (_centroidWeights[lastIndex] < 1.0) {
+ return 1.0 - (_centroidWeights[lastIndex] / 2.0 * (_max - value) /
width) / _totalWeight;
+ }
+ return value == _max ? 1.0 - 0.5 / _totalWeight
+ : 1.0 - (1.0 + (_max - value) / width *
(_centroidWeights[lastIndex] / 2.0 - 1.0))
+ / _totalWeight;
+ }
+ return 1.0;
+ }
+
+ double weightSoFar = 0.0;
+ for (int i = 0; i < lastIndex; i++) {
+ if (_centroidMeans[i] == value) {
+ double equalWeight = 0.0;
+ int equalIndex = i;
+ while (equalIndex < _numCentroids && _centroidMeans[equalIndex] ==
value) {
+ equalWeight += _centroidWeights[equalIndex];
+ equalIndex++;
+ }
+ return (weightSoFar + equalWeight / 2.0) / _totalWeight;
+ }
+ if (_centroidMeans[i] <= value && _centroidMeans[i + 1] > value) {
+ if (!Double.isFinite(_centroidMeans[i]) ||
!Double.isFinite(_centroidMeans[i + 1])) {
+ return (weightSoFar + _centroidWeights[i]) / _totalWeight;
+ }
+ double width = _centroidMeans[i + 1] - _centroidMeans[i];
+ if (width > 0.0) {
+ double leftExcludedWeight = getPointMassHalfWeight(i);
+ double rightExcludedWeight = getPointMassHalfWeight(i + 1);
+ double halfWeight = (_centroidWeights[i] + _centroidWeights[i + 1])
/ 2.0;
+ double interpolatedWeight = halfWeight - leftExcludedWeight -
rightExcludedWeight;
+ double base = weightSoFar + _centroidWeights[i] / 2.0 +
leftExcludedWeight;
+ return (base + interpolatedWeight * (value - _centroidMeans[i]) /
width) / _totalWeight;
+ }
+ return (weightSoFar + (_centroidWeights[i] + _centroidWeights[i + 1])
/ 2.0) / _totalWeight;
+ }
+ weightSoFar += _centroidWeights[i];
+ }
+ if (value == _centroidMeans[lastIndex]) {
+ return 1.0 - Math.min(1.0, _centroidWeights[lastIndex]) / 2.0 /
_totalWeight;
+ }
+ throw new IllegalStateException("Unable to compute TDigest CDF");
+ }
+
+ @Override
+ public double quantile(double quantile) {
+ if (Double.isNaN(quantile) || quantile < 0.0 || quantile > 1.0) {
+ throw new IllegalArgumentException("q should be in [0,1], got " +
quantile);
+ }
+ if (_legacyDegraded) {
+ return Double.NaN;
+ }
+ materializePendingSerializedTDigest();
+ flush();
+ if (_numCentroids == 0) {
+ return Double.NaN;
+ }
+ if (quantile == 0.0) {
+ return _min;
+ }
+ if (quantile == 1.0) {
+ return _max;
+ }
+ if (_numCentroids == 1) {
+ return _centroidMeans[0];
+ }
+
+ double index = quantile * _totalWeight;
+ double firstWeight = _centroidWeights[0];
+ if (firstWeight < 1.0 && index < firstWeight / 2.0) {
+ return weightedAverage(_min, firstWeight / 2.0 - index,
_centroidMeans[0], index);
+ }
+ if (firstWeight >= 1.0 && index < 1.0) {
+ return _min;
+ }
+ if (_centroidWeights[0] > 1.0 && index < _centroidWeights[0] / 2.0) {
+ return _min + (index - 1.0) / (_centroidWeights[0] / 2.0 - 1.0) *
(_centroidMeans[0] - _min);
+ }
+ int lastIndex = _numCentroids - 1;
+ double lastWeight = _centroidWeights[lastIndex];
+ if (lastWeight < 1.0 && index >= _totalWeight - lastWeight / 2.0) {
+ return weightedAverage(_centroidMeans[lastIndex], _totalWeight - index,
_max,
+ index - (_totalWeight - lastWeight / 2.0));
+ }
+ if (lastWeight >= 1.0 && index > _totalWeight - 1.0) {
+ return _max;
+ }
+ if (_centroidWeights[lastIndex] > 1.0
+ && _totalWeight - index <= _centroidWeights[lastIndex] / 2.0) {
+ return _max - (_totalWeight - index - 1.0) /
(_centroidWeights[lastIndex] / 2.0 - 1.0)
+ * (_max - _centroidMeans[lastIndex]);
+ }
+
+ double weightSoFar = _centroidWeights[0] / 2.0;
+ for (int i = 0; i < lastIndex; i++) {
+ double halfWeight = (_centroidWeights[i] + _centroidWeights[i + 1]) /
2.0;
+ if (weightSoFar + halfWeight > index) {
+ double leftExcludedWeight = getPointMassHalfWeight(i);
+ if (index - weightSoFar < leftExcludedWeight) {
+ return _centroidMeans[i];
+ }
+ double rightExcludedWeight = getPointMassHalfWeight(i + 1);
+ if (weightSoFar + halfWeight - index <= rightExcludedWeight) {
+ return _centroidMeans[i + 1];
+ }
+ double leftWeight = index - weightSoFar - leftExcludedWeight;
+ double rightWeight = weightSoFar + halfWeight - index -
rightExcludedWeight;
+ return weightedAverage(_centroidMeans[i], rightWeight,
_centroidMeans[i + 1], leftWeight);
+ }
+ weightSoFar += halfWeight;
+ }
+ double weightToMax = index - _totalWeight - _centroidWeights[lastIndex] /
2.0;
+ double weightFromLastCentroid = _centroidWeights[lastIndex] / 2.0 -
weightToMax;
+ return weightedAverage(_centroidMeans[lastIndex], weightToMax, _max,
weightFromLastCentroid);
+ }
+
+ private double getPointMassHalfWeight(int index) {
+ // Adjacent equal means establish a repeated-value plateau. Keep its
complete mass out of interpolation at
+ // either edge; an isolated weighted centroid still represents an unknown
distribution around its mean.
+ double mean = _centroidMeans[index];
+ boolean pointMass = _centroidWeights[index] == 1.0 || index > 0 &&
_centroidMeans[index - 1] == mean
+ || index + 1 < _numCentroids && _centroidMeans[index + 1] == mean;
+ return pointMass ? _centroidWeights[index] / 2.0 : 0.0;
+ }
+
+ private static double weightedAverage(double firstValue, double firstWeight,
double secondValue,
+ double secondWeight) {
+ if (firstValue > secondValue) {
+ return weightedAverage(secondValue, secondWeight, firstValue,
firstWeight);
+ }
+ if (firstWeight == 0.0) {
+ return secondValue;
+ }
+ if (secondWeight == 0.0 || firstValue == secondValue) {
+ return firstValue;
+ }
+ if (!Double.isFinite(firstValue)) {
+ return firstValue;
+ }
+ if (!Double.isFinite(secondValue)) {
+ return secondValue;
+ }
+ double average = (firstValue * firstWeight + secondValue * secondWeight) /
(firstWeight + secondWeight);
+ return Math.max(firstValue, Math.min(average, secondValue));
+ }
+
+ @Override
+ public Collection<Centroid> centroids() {
+ if (_legacyDegraded) {
+ byte[] bytes = serialize();
+ ByteBuffer encoded = ByteBuffer.wrap(bytes);
+ SerializedTDigestMetadata metadata =
TDigestUtils.readSerializedHeader(encoded);
+ encoded.position(metadata.centroidOffset());
+ List<Centroid> centroids = new ArrayList<>(metadata.centroidCount());
+ for (int i = 0; i < metadata.centroidCount(); i++) {
+ double weight = metadata.centroidSize() ==
TDigestUtils.VERBOSE_CENTROID_SIZE
+ ? encoded.getDouble() : encoded.getFloat();
+ double mean = metadata.centroidSize() ==
TDigestUtils.VERBOSE_CENTROID_SIZE
+ ? encoded.getDouble() : encoded.getFloat();
+ centroids.add(new Centroid(mean, weight));
+ }
+ return centroids;
+ }
+ materializePendingSerializedTDigest();
+ flush();
+ List<Centroid> centroids = new ArrayList<>(_numCentroids);
+ for (int i = 0; i < _numCentroids; i++) {
+ centroids.add(new Centroid(_centroidMeans[i], _centroidWeights[i]));
+ }
+ return centroids;
+ }
+
+ /// Copies a valid distribution's primitive centroid arrays after flushing
buffered inputs. This does not apply
+ /// public compression or serialize a finite sub-distribution; callers can
include surrounding infinity mass.
+ public void copyCentroids(double[] means, double[] weights, int offset) {
+ if (_legacyDegraded) {
+ throw corruptedMutation();
+ }
+ materializePendingSerializedTDigest();
+ flush();
+ if (means == weights || offset < 0 || offset > means.length - _numCentroids
+ || offset > weights.length - _numCentroids) {
+ throw new IllegalArgumentException("Insufficient distinct TDigest
centroid arrays");
+ }
+ if (_numCentroids == 0) {
+ return;
+ }
+ System.arraycopy(_centroidMeans, 0, means, offset, _numCentroids);
+ System.arraycopy(_centroidWeights, 0, weights, offset, _numCentroids);
+ }
+
+ @Override
+ public double compression() {
+ return _compression;
+ }
+
+ /// Returns the length of the [#serialize()] bytes rather than the raw
verbose byte size, so generic `TDigest`
+ /// serializers ([#byteSize()] followed by [#asBytes(ByteBuffer)]) emit the
mixed-version-compatible encoding.
+ /// Prepared bytes are cached until the next write or mutation, avoiding a
second encoding pass for that pattern.
+ /// Emitting raw verbose bytes here could produce a digest with more
centroids than an old reader of the same
+ /// compression allocates, which it rejects with
[ArrayIndexOutOfBoundsException].
+ @Override
+ public int byteSize() {
+ if (_legacyDegraded || _originalFractionalBytes != null) {
+ return _legacyDegraded ? _pendingSerializedTDigest.length :
_originalFractionalBytes.length;
+ }
+ if (_serializedBytesForWrite == null) {
+ _serializedBytesForWrite = serialize();
+ }
+ return _serializedBytesForWrite.length;
+ }
+
+ /// Bounds both compatible encodings and a widened verbose view without
flushing. Buffered and fractional inputs
+ /// can reserve more space than their eventual bytes; retained capacities
need not fit default compression size.
+ @Override
+ public int maxSerializedByteSize() {
+ if (_legacyDegraded) {
+ return _pendingSerializedTDigest.length;
+ }
+ if (_originalFractionalBytes != null) {
+ return _originalFractionalBytes.length;
+ }
+ long centroidCount = getCentroidCountUpperBound();
+ if (!_hasFractionalWeights) {
+ centroidCount = (long) Math.min(centroidCount,
Math.ceil(getTotalWeight()));
+ }
+ int maximumBytes = getMaxVerboseByteSize(centroidCount);
+ return _pendingSerializedTDigest == null ? maximumBytes :
Math.max(maximumBytes, _pendingSerializedTDigest.length);
+ }
+
+ private static int getMaxVerboseByteSize(long centroidCount) {
+ return Math.toIntExact(Math.addExact(TDigestUtils.VERBOSE_HEADER_SIZE,
+ Math.multiplyExact(TDigestUtils.VERBOSE_CENTROID_SIZE,
centroidCount)));
+ }
+
+ /// Returns the small-encoding size unless float narrowing loses a field or
an inherited fractional endpoint needs
+ /// exact verbose bytes.
+ /// Historical degraded state retains its original encoding and rejects
mutation.
+ @Override
+ public int smallByteSize() {
+ if (_legacyDegraded || _originalFractionalBytes != null) {
+ return _legacyDegraded ? _pendingSerializedTDigest.length :
_originalFractionalBytes.length;
+ }
+ compress();
+ normalizeBoundaryCentroids();
+ checkFractionalBoundaryEncoding();
+ if (!canUseSmallEncoding()) {
+ return byteSize();
+ }
+ checkCapacityPreservingCentroidCount();
+ return Math.addExact(TDigestUtils.SMALL_HEADER_SIZE,
+ Math.multiplyExact(TDigestUtils.SMALL_CENTROID_SIZE, _numCentroids));
+ }
+
+ /// Writes the [#serialize()] bytes; see [#byteSize()]. Bytes are always
written in big-endian order (the t-digest
+ /// wire order) regardless of the destination buffer's byte order, unlike
the library's `asBytes`.
+ @Override
+ public void asBytes(ByteBuffer buffer) {
+ byte[] bytes = _serializedBytesForWrite;
+ _serializedBytesForWrite = null;
+ buffer.put(bytes != null ? bytes : serialize());
+ }
+
+ /// Writes the representation described by [#smallByteSize()]; readers must
inspect the leading encoding identifier.
+ @Override
+ public void asSmallBytes(ByteBuffer buffer) {
+ if (_legacyDegraded || _originalFractionalBytes != null) {
+ buffer.put(serialize());
+ return;
+ }
+ compress();
+ normalizeBoundaryCentroids();
+ checkFractionalBoundaryEncoding();
+ if (canUseSmallEncoding()) {
+ buffer.put(toCapacityPreservingBytes());
+ } else {
+ asBytes(buffer);
+ }
+ }
+
+ @Override
+ public int centroidCount() {
+ if (_pendingSerializedTDigest != null) {
+ return !_legacyDegraded && _pendingSerializedMetadata.totalWeight() ==
0.0
Review Comment:
**Latent inconsistency: `centroidCount()` on a pending digest returns the
encoded count, which disagrees with the materialized count every other read
path reports.**
The encoded count (a) includes zero-weight centroids that `decode()` drops
and (b) excludes the up-to-2 boundary splits `normalizeBoundaries` applies for
payloads with `weightedBoundaries`. A caller sizing arrays from
`centroidCount()` and then calling `copyCentroids(means, weights, 0)` triggers
materialization; for a payload whose first/last weight > 1, `_numCentroids`
becomes N+1/N+2 and the guard throws `Insufficient distinct TDigest centroid
arrays` on correctly-sized arrays. In-repo callers dodge it by calling
`prepareCentroidsForSerialization()` first, but the natural public pairing
`centroidCount()` + `copyCentroids()` is broken for pending state.
##########
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileSmartTDigestAggregationFunction.java:
##########
@@ -365,7 +361,7 @@ public Double extractFinalResult(@Nullable Object
intermediateResult) {
TDigest tDigest = (TDigest) intermediateResult;
// An empty digest holds the same state as the empty value list below,
so the two branches must answer alike:
// which one a query lands in depends only on whether the accumulator
crossed the conversion threshold.
- if (_nullHandlingEnabled && tDigest.size() == 0L) {
+ if (_nullHandlingEnabled && tDigest.isEmpty()) {
Review Comment:
**Latent inconsistency with the comment above: with null handling disabled,
an empty digest returns `quantile()` = NaN while an empty value list returns
`DEFAULT_FINAL_RESULT`.**
The comment says the two branches "must answer alike", and the `isEmpty()`
guard enforces that only when `_nullHandlingEnabled` is true. With it disabled,
an empty TDigest intermediate renders NaN but an empty list renders the
sentinel. Unreachable today (a digest intermediate only exists after crossing
the threshold, so it is never empty), but the guard should also apply with
`DEFAULT_FINAL_RESULT` when null handling is off, exactly as the comment
promises: `if (tDigest.isEmpty()) { return _nullHandlingEnabled ? null :
DEFAULT_FINAL_RESULT; }`.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TDigestUtils.java:
##########
@@ -86,44 +85,65 @@ public static byte[] serialize(TDigest tDigest) {
/// Serializes a digest using `scratchBuffer` for the temporary verbose
representation when it is large enough.
///
- /// The returned array is independently owned. The scratch buffer is cleared
and may be reused by the caller after
- /// this method returns.
+ /// The returned array is independently owned. Pinot-owned digests do not
use the scratch buffer; callers can
+ /// clear and reuse it after serialization.
public static byte[] serialize(TDigest tDigest, ByteBuffer scratchBuffer) {
ByteOrder scratchOrder = scratchBuffer != null ? scratchBuffer.order() :
null;
try {
- if (!(tDigest instanceof MergingDigest)) {
- // Pinot has always decoded TDigest values as MergingDigest. Preserve
the previous serialization behavior for
- // other implementations instead of interpreting their
implementation-specific payload as MergingDigest
- // bytes.
- ByteBuffer buffer = prepareScratchBuffer(scratchBuffer,
tDigest.byteSize());
- tDigest.asBytes(buffer);
- byte[] bytes = new byte[buffer.position()];
- buffer.flip();
- buffer.get(bytes);
- return bytes;
+ if (tDigest instanceof PercentileTDigestAccumulator) {
+ // Preserve the legacy two-level raw flush before final compression.
Pending serialized state answers
+ // this count from cached metadata, so stored reads do not acquire an
extra lossy compression pass.
+ tDigest.centroidCount();
+ return ((PercentileTDigestAccumulator) tDigest).serialize();
}
-
- int defaultCapacity = getDefaultCentroidCapacity(tDigest.compression());
- int legacyCapacity =
getLegacyDefaultCentroidCapacity(tDigest.compression());
- // centroidCount() drains pending values at the digest's internal
(two-level) compression, but does not perform
- // the final, destructive compression used by asBytes(). It therefore
gives us a safe bound for digests
- // deserialized from the compact encoding with an explicitly enlarged
main array while still allowing asBytes()
- // to run once.
- int currentCentroidCount = tDigest.centroidCount();
- long sizeBound = Math.min(Math.max(0L, tDigest.size()),
Math.max(defaultCapacity, legacyCapacity));
- int maxCentroids = Math.max(currentCentroidCount, (int) sizeBound);
- int requiredCapacity = Math.addExact(VERBOSE_HEADER_SIZE,
- Math.multiplyExact(VERBOSE_CENTROID_SIZE, maxCentroids));
+ byte[] retained =
PercentileTDigestValueAggregator.getRetainedHistoricalBytes(tDigest);
Review Comment:
**Layering: the shared codec now compiles against an aggregator
implementation detail, creating a utils → aggregator → utils package cycle.**
`TDigestUtils.serialize` special-cases the `NonFiniteAwareTDigest` wrapper
via the public static hook
`PercentileTDigestValueAggregator.getRetainedHistoricalBytes`, which probes the
wrapper's private fields; any future TDigest wrapper needing byte-exact
retention must be special-cased here too. A polymorphic hook on the base class
removes the import, the instanceof chain, and the static accessor: e.g.
`@Nullable byte[] retainedHistoricalBytes() { return null; }` on `TDigest`,
overridden by the wrapper, with `serialize()` calling it unconditionally.
--
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]
