xiangfu0 commented on code in PR #19795:
URL: https://github.com/apache/pinot/pull/19795#discussion_r4239363633
##########
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileSmartTDigestAggregationFunction.java:
##########
@@ -365,8 +361,8 @@ 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) {
- return null;
+ if (tDigest.isEmpty()) {
+ return _nullHandlingEnabled ? null : DEFAULT_FINAL_RESULT;
Review Comment:
The migration guide and PR description now explicitly call out the Smart
empty-digest change: NULL with null handling, otherwise -Infinity, matching its
empty value-list state. Ordinary percentileTDigest keeps its existing NaN
result without null handling. The existing on/off empty-final-extraction
regression remains in place.
Addressed in
[277a170ffe](https://github.com/apache/pinot/commit/277a170ffe224d680ee00c6f56efd422bb673990).
All 526 affected tests, actual 3.2/3.3 reader checks and the unchanged
270-case accuracy oracle pass.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/tdigest/TDigestCodec.java:
##########
@@ -0,0 +1,663 @@
+/**
+ * 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.tdigest;
+
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.Arrays;
+
+/// Compatibility helpers for the serialized t-digest format shared by
t-digest 3.2 and 3.3.
+///
+/// Version 3.3 requires the first and last centroid to have unit weight
whenever a digest is recompressed. Version
+/// 3.2 did not maintain that invariant, so an otherwise valid 3.2 digest can
fail an assertion after being read by
+/// 3.3. Deserialization repairs only those legacy boundary centroids,
preserving their total weight and first moment.
+/// Serialization uses the compact form when 3.3's larger low-compression
centroid buffer would not fit in a 3.2
+/// default buffer and every centroid is exactly representable in that format.
Otherwise, it merges the nearest
+/// interior centroids until the verbose representation fits the 3.2 capacity.
This keeps intermediate values readable
+/// in both directions during a rolling upgrade without silently narrowing
double-precision values to floats.
+///
+/// Pinot owns the in-memory K1 implementation and retains these legacy bytes
for stored columns and mixed-version
+/// intermediate results. Decoding checks lengths and numerical state before
allocating centroid arrays.
+/// NaN means at an identifiable infinite tail are restored to that infinite
bound without changing their weights.
+/// Other poisoned states remain byte-readable with NaN statistics because
their distribution cannot be recovered.
+/// They retain their original bytes; attempting to mutate them or mix them
with another distribution fails explicitly.
+/// Opaque poisoned payloads retain their original centroids even if an old
reader's default array cannot hold them;
+/// these historical exceptions remain readable by Pinot, but are not
guaranteed to survive a reverse upgrade.
+/// Historical fractional boundary masses below one, and singleton masses
between one and two, retain their original
+/// bytes. Fresh serialization rejects these boundaries: t-digest 3.3's
unit-endpoint assertion cannot be satisfied
+/// without changing their mass. Fractional interior weights remain supported.
+public final class TDigestCodec {
+ public static final int VERBOSE_ENCODING = 1;
+ public static final int SMALL_ENCODING = 2;
+ private static final int LEGACY_CAPACITY_PADDING = 10;
+ public static final int VERBOSE_HEADER_SIZE = 32;
+ public static final int VERBOSE_CENTROID_SIZE = 16;
+ public static final int SMALL_HEADER_SIZE = 30;
+ public static final int SMALL_CENTROID_SIZE = 8;
+ public static final int DEFAULT_MERGE_BUFFER_MULTIPLIER = 5;
+
+ private TDigestCodec() {
+ }
+
+ /// Reads the compression from a validated header without decoding centroid
arrays.
+ public static double readCompression(byte[] bytes) {
+ return readSerializedHeader(ByteBuffer.wrap(bytes)).compression();
+ }
+
+ /// Rejects non-finite compression while retaining the legacy clamping of
finite settings below ten.
+ public static void validateCompression(double compression) {
+ if (!Double.isFinite(compression)) {
+ throw new IllegalArgumentException("Invalid TDigest compression: " +
compression);
+ }
+ }
+
+ /// Applies the legacy minimum compression to query, ingestion and
serialized-header settings.
+ public static double normalizeCompression(double compression) {
+ validateCompression(compression);
+ return Math.max(compression, 10.0);
+ }
+
+ /// Writes validated centroid arrays in the standard verbose encoding
without sorting or recompression.
+ /// The caller retains the arrays; only the first `count` entries are
written, with normalized compression.
+ public static byte[] serializeCentroids(double compression, double min,
double max, double[] means,
+ double[] weights, int count) {
+ if (count < 0 || count > means.length || count > weights.length) {
+ throw new IllegalArgumentException("Invalid TDigest centroid count: " +
count);
+ }
+ ByteBuffer encoded = ByteBuffer.allocate(Math.addExact(VERBOSE_HEADER_SIZE,
+ Math.multiplyExact(VERBOSE_CENTROID_SIZE, count)));
+
encoded.putInt(VERBOSE_ENCODING).putDouble(min).putDouble(max).putDouble(normalizeCompression(compression));
+ encoded.putInt(count);
+ for (int i = 0; i < count; i++) {
+ encoded.putDouble(weights[i]).putDouble(means[i]);
+ }
+ return encoded.array();
+ }
+
+ /// Encodes validated, sorted native centroids and selects a
legacy-compatible representation.
+ static byte[] serializeCompatibleCentroids(double compression, double min,
double max, double[] means,
+ double[] weights, int count, boolean inheritedFractionalBoundaries) {
+ if (count < 0 || count > 0 && (count > means.length || count >
weights.length || !(min <= max))) {
+ throw new IllegalArgumentException("Invalid TDigest centroid count or
extrema");
+ }
+ ByteBuffer buffer = ByteBuffer.allocate(
+ Math.addExact(VERBOSE_HEADER_SIZE,
+ Math.multiplyExact(VERBOSE_CENTROID_SIZE, count)));
+ buffer.putInt(VERBOSE_ENCODING);
+ buffer.putDouble(min);
+ buffer.putDouble(max);
+ buffer.putDouble(compression);
+ buffer.putInt(count);
+ double encodedWeight = 0.0;
+ boolean hasNonFiniteMeans = false;
+ boolean fractionalWeights = false;
+ for (int i = 0; i < count; i++) {
+ if (!(weights[i] > 0.0) || !Double.isFinite(weights[i]) || !(means[i] >=
min && means[i] <= max)) {
+ throw new IllegalArgumentException("Invalid TDigest centroid");
+ }
+ buffer.putDouble(weights[i]);
+ buffer.putDouble(means[i]);
+ encodedWeight += weights[i];
+ hasNonFiniteMeans |= !Double.isFinite(means[i]);
+ fractionalWeights |= weights[i] != Math.rint(weights[i]);
+ }
+ if (!Double.isFinite(encodedWeight)) {
+ throw new IllegalArgumentException("TDigest total weight exceeds the
supported range");
+ }
+ boolean weightedBoundaries = count == 1 ? weights[0] >= 2.0
+ : count > 1 && (weights[0] > 1.0 || weights[count - 1] > 1.0);
+ // Collect exact wire metadata during the write rather than decoding and
validating our fresh bytes again.
+ SerializedTDigestMetadata metadata = new
SerializedTDigestMetadata(VERBOSE_ENCODING, min, max,
+ compression, count, Math.max(getDefaultCentroidCapacity(compression),
+ getLegacyDefaultCentroidCapacity(compression)), 0,
+ VERBOSE_HEADER_SIZE, VERBOSE_CENTROID_SIZE, buffer.capacity(),
encodedWeight,
+ hasNonFiniteMeans, false, false, fractionalWeights,
weightedBoundaries, false, compression, Double.NaN);
+ return makeLegacyCompatible(buffer.array(), metadata,
inheritedFractionalBoundaries);
+ }
+
+ /// Rejects fresh boundary masses that legacy readers cannot recompress
without inventing mass.
+ /// Call after boundary normalization and removing zero-weight centroids;
fractional interior mass is supported.
+ static void checkLegacyBoundaryWeights(int count, double firstWeight, double
lastWeight) {
+ if (count > 0 && (firstWeight < 1.0 || lastWeight < 1.0
+ || count == 1 && firstWeight != 1.0 && firstWeight < 2.0)) {
+ throw new IllegalArgumentException("Cannot serialize fractional TDigest
boundary mass for legacy readers");
+ }
+ }
+
+ /// Allows only unsupported endpoint means inherited from a validated legacy
source. This is shared by native,
+ /// enclosing and generic writers; new fractional global extrema do not
receive the historical exception.
+ static boolean hasInheritedFractionalBoundaryEncoding(TDigest source, int
count, double firstMean,
+ double firstWeight, double lastMean, double lastWeight) {
+ if (count <= 0 || !source.hasValidStatistics() || !(firstWeight > 0.0) ||
!(lastWeight > 0.0)
+ || !Double.isFinite(firstWeight) || !Double.isFinite(lastWeight)) {
+ return false;
+ }
+ boolean singleton = count == 1 && firstWeight != 1.0 && firstWeight < 2.0;
+ boolean firstUnsupported = firstWeight < 1.0 || singleton;
+ boolean lastUnsupported = lastWeight < 1.0 || singleton;
+ return (firstUnsupported || lastUnsupported)
+ && (!firstUnsupported || firstMean ==
source.getHistoricalFractionalBoundaryMean(true))
+ && (!lastUnsupported || lastMean ==
source.getHistoricalFractionalBoundaryMean(false));
+ }
+
+ private static void checkSerializedBoundaryWeights(byte[] bytes,
SerializedTDigestMetadata metadata) {
+ int count = metadata.centroidCount();
+ if (count > 0) {
+ ByteBuffer encoded = ByteBuffer.wrap(bytes);
+ int firstOffset = metadata.centroidOffset();
+ int lastOffset = firstOffset + (count - 1) * metadata.centroidSize();
+ double firstWeight = metadata.encoding() == VERBOSE_ENCODING
+ ? encoded.getDouble(firstOffset) : encoded.getFloat(firstOffset);
+ double lastWeight = metadata.encoding() == VERBOSE_ENCODING
+ ? encoded.getDouble(lastOffset) : encoded.getFloat(lastOffset);
+ checkLegacyBoundaryWeights(count, firstWeight, lastWeight);
+ }
+ }
+
+ /// Converts a verbose `MergingDigest` payload to a representation readable
by t-digest 3.2 and 3.3.
+ ///
+ /// The input must use the standard verbose `MergingDigest` encoding. The
returned payload stays verbose when it
+ /// already fits both legacy default capacities, uses compact encoding only
when every narrowed field is exact, and
+ /// otherwise reduces interior centroids without narrowing doubles.
Compression below ten is normalized in the
+ /// output header because the 3.2 verbose reader sizes its arrays from the
raw header before applying that minimum.
+ /// Opaque degraded state is retained without reducing its unknown
distribution, including over-capacity payloads.
+ /// Fresh fractional endpoints that legacy unit-boundary repair cannot
represent are rejected without reweighting.
+ public static byte[] makeLegacyCompatible(byte[] verboseBytes) {
+ return makeLegacyCompatible(verboseBytes,
inspectSerialized(ByteBuffer.wrap(verboseBytes), false));
+ }
+
+ /// Serializes an unchanged, independently retained payload using metadata
validated at ingestion.
+ ///
+ /// The metadata must describe these exact bytes. This avoids another
centroid validation walk for lazy digests.
+ public static byte[] makeLegacyCompatible(byte[] verboseBytes,
SerializedTDigestMetadata metadata) {
+ return makeLegacyCompatible(verboseBytes, metadata, false);
+ }
+
+ /// Retains exact verbose weights for inherited historical fractional
endpoints that cannot become unit endpoints.
+ /// All other compatibility checks apply. These payloads keep their
historical assertion-enabled 3.3 read/merge
+ /// limitation; callers must establish inherited endpoint provenance before
enabling this exception.
+ static byte[] makeLegacyCompatible(byte[] verboseBytes,
SerializedTDigestMetadata metadata,
+ boolean inheritedFractionalBoundaries) {
+ if (metadata.encoding() != VERBOSE_ENCODING) {
+ throw new IllegalArgumentException("Expected verbose TDigest encoding");
+ }
+ if (metadata.needsLegacyFallback()) {
+ return verboseBytes;
+ }
+ if (!Double.isNaN(metadata.recoveredInfinityMean())) {
+ double[] means = new double[metadata.centroidCount()];
+ double[] weights = new double[means.length];
+ decodeSerializedCentroids(ByteBuffer.wrap(verboseBytes), metadata,
means, weights);
+ if (!inheritedFractionalBoundaries) {
+ return serializeRecoveredCentroids(metadata, means, weights);
+ }
+ verboseBytes = serializeCentroids(metadata.compression(),
metadata.min(), metadata.max(), means, weights,
+ means.length);
+ metadata = inspectSerialized(ByteBuffer.wrap(verboseBytes), false);
+ }
+ if (metadata.hasZeroWeightCentroids()) {
+ verboseBytes = removeZeroWeightCentroids(verboseBytes, metadata);
+ metadata = inspectSerialized(ByteBuffer.wrap(verboseBytes), false);
+ }
+ if (!inheritedFractionalBoundaries) {
+ checkSerializedBoundaryWeights(verboseBytes, metadata);
+ }
+ double compression = metadata.compression();
+ if (metadata.encodedCompression() != compression) {
+ verboseBytes = verboseBytes.clone();
+ ByteBuffer.wrap(verboseBytes).putDouble(20, compression);
+ }
+ int centroidCount = metadata.centroidCount();
+ int legacyCapacity =
+ Math.min(getLegacyDefaultCentroidCapacity(compression),
getDefaultCentroidCapacity(compression));
+ if (centroidCount <= legacyCapacity) {
+ return verboseBytes;
+ }
+ int mainCapacity = getDefaultCentroidCapacity(compression);
+ long bufferCapacity = DEFAULT_MERGE_BUFFER_MULTIPLIER * (long)
mainCapacity;
+ if (centroidCount > mainCapacity || inheritedFractionalBoundaries ||
(double) (float) compression != compression
+ || centroidCount > Short.MAX_VALUE
+ || mainCapacity > Short.MAX_VALUE || bufferCapacity > Short.MAX_VALUE)
{
+ return reduceVerboseCentroids(verboseBytes, legacyCapacity);
+ }
+ ByteBuffer verbose = ByteBuffer.wrap(verboseBytes);
+ ByteBuffer centroids = verbose.duplicate();
+ centroids.position(metadata.centroidOffset());
+ for (int i = 0; i < centroidCount; i++) {
+ double weight = centroids.getDouble();
+ double mean = centroids.getDouble();
+ if ((double) (float) weight != weight || (double) (float) mean != mean) {
+ return reduceVerboseCentroids(verboseBytes, legacyCapacity);
+ }
+ }
+
+ ByteBuffer small = ByteBuffer.allocate(SMALL_HEADER_SIZE +
SMALL_CENTROID_SIZE * centroidCount);
+ verbose.rewind();
+ verbose.getInt();
+ small.putInt(SMALL_ENCODING);
+ small.putDouble(verbose.getDouble());
+ small.putDouble(verbose.getDouble());
+ verbose.getDouble();
+ verbose.getInt();
+ small.putFloat((float) compression);
+ small.putShort((short) mainCapacity);
+ small.putShort((short) bufferCapacity);
+ small.putShort((short) centroidCount);
+ for (int i = 0; i < centroidCount; i++) {
+ small.putFloat((float) verbose.getDouble());
+ small.putFloat((float) verbose.getDouble());
+ }
+ 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());
+ }
+
+ /// Validates serialized lengths and centroid values without allocating
centroid arrays.
+ public static double validateSerialized(byte[] bytes) {
+ return validateSerialized(ByteBuffer.wrap(bytes));
+ }
+
+ /// Validates one legacy digest without changing `input` and returns its
total centroid weight.
+ ///
+ /// Trailing bytes remain available to the caller for intermediate values
containing several objects.
+ public static double validateSerialized(ByteBuffer input) {
+ return validateSerialized(input, true);
+ }
+
+ private static double validateSerialized(ByteBuffer input, boolean
checkCapacity) {
+ return inspectSerialized(input, checkCapacity).totalWeight();
+ }
+
+ /// 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) {
+ }
+
+ /// 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 or inverted nonempty extrema mark legacy degraded state. Oversized
payloads require numerical inspection:
+ /// historical degraded state remains readable even if an old decoder's
arrays could not hold its 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 = 0;
+ int centroidCount;
+ int centroidSize;
+ if (encoding == VERBOSE_ENCODING) {
+ encodedCompression = encoded.getDouble();
+ compression = normalizeCompression(encodedCompression);
+ mainCapacity = Math.max(getDefaultCentroidCapacity(compression),
getLegacyDefaultCentroidCapacity(compression));
+ centroidCount = encoded.getInt();
+ centroidSize = VERBOSE_CENTROID_SIZE;
+ } 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;
+ }
+ if (centroidCount < 0) {
+ throw new IllegalArgumentException("Invalid negative TDigest centroid
count: " + centroidCount);
+ }
+ if (centroidCount > encoded.remaining() / centroidSize) {
+ throw new BufferUnderflowException();
+ }
+ int centroidOffset = encoded.position();
+ int encodedLength = centroidOffset + centroidCount * centroidSize;
+ SerializedTDigestMetadata header = new SerializedTDigestMetadata(encoding,
min, max, compression, centroidCount,
+ mainCapacity,
+ bufferCapacity, centroidOffset, centroidSize, encodedLength,
Double.NaN, false,
+ Double.isNaN(min) || Double.isNaN(max) || centroidCount > 0 && min >
max,
+ false, false, false, false, encodedCompression, Double.NaN);
+ if (checkCapacity && centroidCount > mainCapacity) {
+ SerializedTDigestMetadata inspected = inspectSerialized(input, header,
null, null);
+ if (!inspected.needsLegacyFallback()) {
+ throw new IllegalArgumentException("TDigest centroid count exceeds
capacity: " + centroidCount);
+ }
+ return inspected;
+ }
+ return header;
+ }
+
+ /// Validates centroids in one pass and optionally decodes their means and
weights into reusable arrays.
+ ///
+ /// The header must describe this exact input, and each supplied array must
hold all encoded centroids. Endpoint
+ /// rounding inherited from compact re-encoding or old merge arithmetic is
restored to the double header bounds.
+ public static SerializedTDigestMetadata inspectSerialized(ByteBuffer input,
SerializedTDigestMetadata header,
+ double[] means, double[] weights) {
+ ByteBuffer encoded = input.slice().order(ByteOrder.BIG_ENDIAN);
+ encoded.position(header.centroidOffset());
+ int centroidCount = header.centroidCount();
+ int centroidSize = header.centroidSize();
+ double min = header.min();
+ double max = header.max();
+ double totalWeight = 0.0;
+ double previousMean = Double.NEGATIVE_INFINITY;
+ boolean hasNonFiniteMeans = false;
+ boolean needsLegacyFallback = header.needsLegacyFallback();
+ boolean unorderedMeans = false;
+ boolean fractionalWeights = false;
+ boolean hasZeroWeightCentroids = false;
+ int positiveCentroidCount = 0;
+ double firstPositiveWeight = 0.0;
+ double lastPositiveWeight = 0.0;
+ boolean withinBounds = true;
+ // Infer only a matching one-sided tail, or a distribution consisting
entirely of one infinity. A NaN
+ // between finite centroids or in a +/-Infinity mixture has lost
information and remains opaque.
+ double infinityMean = min == max && Double.isInfinite(min) ? min
+ : Double.isFinite(min) && max == Double.POSITIVE_INFINITY ? max
+ : min == Double.NEGATIVE_INFINITY && Double.isFinite(max) ? min :
Double.NaN;
+ boolean hasNaNMeans = false;
+ boolean hasFiniteMeans = false;
+ boolean positiveInfinityTailStarted = false;
+ boolean negativeInfinityTailEnded = false;
+ boolean identifiableInfinityTail = true;
+ double roundedMin = (double) (float) min;
+ double roundedMax = (double) (float) max;
+ double lowerAdjacent = Math.nextDown(min);
+ double upperAdjacent = Math.nextUp(max);
+ for (int i = 0; i < centroidCount; i++) {
+ double weight = centroidSize == VERBOSE_CENTROID_SIZE ?
encoded.getDouble() : encoded.getFloat();
+ double mean = centroidSize == VERBOSE_CENTROID_SIZE ?
encoded.getDouble() : encoded.getFloat();
+ if (!Double.isFinite(weight)) {
+ throw new IllegalArgumentException("Invalid TDigest centroid weight: "
+ weight);
Review Comment:
Nonfinite stored weights and overflowing stored totals now retain opaque
degraded state instead of failing every read. Native and wrapper reads preserve
original bytes and return NaN statistics; mutation/mixing still fail
atomically. Fresh add/merge keeps finite-weight and total validation.
Regressions cover both encodings and fresh-versus-historical overflow.
Addressed in
[277a170ffe](https://github.com/apache/pinot/commit/277a170ffe224d680ee00c6f56efd422bb673990).
All 526 affected tests, actual 3.2/3.3 reader checks and the unchanged
270-case accuracy oracle pass.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/tdigest/NonFiniteAwareTDigest.java:
##########
@@ -0,0 +1,500 @@
+/**
+ * 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.tdigest;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import javax.annotation.Nullable;
+import
org.apache.pinot.segment.local.customobject.tdigest.TDigestCodec.SerializedTDigestMetadata;
+
+
+/// Stores non-finite values as exact tail masses while delegating finite
values to an owned native digest.
+/// Emits legacy t-digest bytes and preserves historical fractional endpoint
provenance across merges and copies.
+/// Instances are mutable and require external synchronization.
+public final class NonFiniteAwareTDigest extends TDigest {
+ private final PercentileTDigestAccumulator _finiteDigest;
+ private double _negativeInfinityWeight;
+ private double _positiveInfinityWeight;
+ private List<Centroid> _finiteCentroids;
+ private byte[] _originalFractionalBytes;
+
+ public static NonFiniteAwareTDigest forLegacyAggregation(double compression)
{
+ return new
NonFiniteAwareTDigest(PercentileTDigestAccumulator.forLegacyAggregation(compression),
0.0, 0.0);
+ }
+
+ private NonFiniteAwareTDigest(PercentileTDigestAccumulator finiteDigest,
double negativeInfinityWeight,
+ double positiveInfinityWeight) {
+ _finiteDigest = finiteDigest;
+ _negativeInfinityWeight = negativeInfinityWeight;
+ _positiveInfinityWeight = positiveInfinityWeight;
+ }
+
+ public NonFiniteAwareTDigest copy() {
+ if (_originalFractionalBytes != null) {
+ return fromBytes(_originalFractionalBytes);
+ }
+ PercentileTDigestAccumulator source = _finiteDigest;
+ PercentileTDigestAccumulator finiteDigest;
+ if (!source.hasValidStatistics() || source.hasOriginalFractionalPayload())
{
+ finiteDigest =
PercentileTDigestAccumulator.fromBytes(source.serialize());
+ } else {
+ finiteDigest =
PercentileTDigestAccumulator.forLegacyAggregation(compression());
+ if (!source.isEmpty()) {
+ finiteDigest.add(source);
+ invalidateDerivedCaches();
+ }
+ }
+ if (source.hasValidStatistics()) {
+ finiteDigest.inheritHistoricalFractionalBoundaries(source);
+ }
+ return new NonFiniteAwareTDigest(finiteDigest, _negativeInfinityWeight,
_positiveInfinityWeight);
+ }
+
+ public static NonFiniteAwareTDigest fromBytes(byte[] bytes) {
+ SerializedTDigestInput validated = new SerializedTDigestInput();
+ validated.reset(bytes);
+ SerializedTDigestMetadata metadata = validated.getMetadata();
+ if (!metadata.hasNonFiniteMeans() || metadata.needsLegacyFallback()) {
+ return new
NonFiniteAwareTDigest(PercentileTDigestAccumulator.fromBytes(validated), 0.0,
0.0);
+ }
+ double encodedMin = metadata.min();
+ double encodedMax = metadata.max();
+ double compression = metadata.compression();
+ int centroidCount = metadata.centroidCount();
+ double negativeInfinityWeight = 0.0;
+ double positiveInfinityWeight = 0.0;
+ double[] finiteMeans = new double[centroidCount];
+ double[] finiteWeights = new double[centroidCount];
+ TDigestCodec.inspectSerialized(ByteBuffer.wrap(bytes), metadata,
finiteMeans, finiteWeights);
+ int finiteCount = 0;
+ double finiteMin = Double.POSITIVE_INFINITY;
+ double finiteMax = Double.NEGATIVE_INFINITY;
+ for (int i = 0; i < centroidCount; i++) {
+ double weight = finiteWeights[i];
+ double mean = finiteMeans[i];
+ if (weight == 0.0) {
+ continue;
+ }
+ if (mean == Double.NEGATIVE_INFINITY) {
+ negativeInfinityWeight += weight;
+ } else if (mean == Double.POSITIVE_INFINITY) {
+ positiveInfinityWeight += weight;
+ } else {
+ finiteWeights[finiteCount] = weight;
+ finiteMeans[finiteCount] = mean;
+ finiteMin = Math.min(finiteMin, mean);
+ finiteMax = Math.max(finiteMax, mean);
+ finiteCount++;
+ }
+ }
+
+ PercentileTDigestAccumulator finiteDigest =
+ PercentileTDigestAccumulator.forLegacyAggregation(compression);
+ if (finiteCount > 0) {
+ finiteMin = Double.isFinite(encodedMin) ? encodedMin : finiteMin;
+ finiteMax = Double.isFinite(encodedMax) ? encodedMax : finiteMax;
+ finiteDigest.addCentroids(finiteMeans, finiteWeights, finiteCount,
finiteMin, finiteMax, true);
+ }
+ validated.recordHistoricalFractionalBoundaries(finiteDigest);
+ NonFiniteAwareTDigest wrapped =
+ new NonFiniteAwareTDigest(finiteDigest, negativeInfinityWeight,
positiveInfinityWeight);
+ if (metadata.fractionalWeights()) {
+ wrapped._originalFractionalBytes = bytes.clone();
+ }
+ return wrapped;
+ }
+
+ @Override
+ public void add(double value) {
+ add(value, 1);
+ }
+
+ @Override
+ public void add(double value, double weight) {
+ 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();
+ checkTotalWeight(getTotalWeight() + weight);
+ if (value == Double.NEGATIVE_INFINITY) {
+ _negativeInfinityWeight += weight;
+ } else if (value == Double.POSITIVE_INFINITY) {
+ _positiveInfinityWeight += weight;
+ } else {
+ _finiteDigest.add(value, weight);
+ }
+ mutated();
+ }
+
+ @Override
+ public void add(TDigest other) {
+ if (other.isEmpty()) {
+ return;
+ }
+ requireMutable();
+ if (!other.hasValidStatistics() && !isEmpty()) {
+ throw corruptedMutation();
+ }
+ if (other.hasValidStatistics()) {
+ checkTotalWeight(getTotalWeight() + other.getTotalWeight());
+ }
+ if (other instanceof NonFiniteAwareTDigest) {
+ NonFiniteAwareTDigest wrapped = (NonFiniteAwareTDigest) other;
+ double negativeInfinityWeight = wrapped._negativeInfinityWeight;
+ double positiveInfinityWeight = wrapped._positiveInfinityWeight;
+ if (!wrapped._finiteDigest.isEmpty()) {
+ _finiteDigest.add(wrapped._finiteDigest);
+ wrapped.invalidateDerivedCaches();
+ }
+ if (wrapped.hasValidStatistics()) {
+ _finiteDigest.inheritHistoricalFractionalBoundaries(wrapped);
+ }
+ _negativeInfinityWeight += negativeInfinityWeight;
+ _positiveInfinityWeight += positiveInfinityWeight;
+ mutated();
+ return;
+ }
+ if (!other.hasValidStatistics() || Double.isFinite(other.getMin()) &&
Double.isFinite(other.getMax())) {
+ _finiteDigest.add(other);
+ mutated();
+ return;
+ }
+ // Read precise centroid weights directly; generic sources need no full
digest serialization for their tails.
+ Collection<Centroid> centroids = other.centroids();
+ double[] finiteMeans = new double[centroids.size()];
+ double[] finiteWeights = new double[centroids.size()];
+ int finiteCount = 0;
+ double finiteMin = Double.POSITIVE_INFINITY;
+ double finiteMax = Double.NEGATIVE_INFINITY;
+ double negativeInfinityWeight = 0.0;
+ double positiveInfinityWeight = 0.0;
+ double totalWeight = getTotalWeight();
+ for (Centroid centroid : centroids) {
+ double mean = centroid.mean();
+ double weight = centroid.weight();
+ if (weight == 0.0) {
+ continue;
+ }
+ if (!(weight > 0.0) || !Double.isFinite(weight) || Double.isNaN(mean)) {
+ throw new IllegalArgumentException("Invalid TDigest centroid");
+ }
+ totalWeight += weight;
+ if (mean == Double.NEGATIVE_INFINITY) {
+ negativeInfinityWeight += weight;
+ } else if (mean == Double.POSITIVE_INFINITY) {
+ positiveInfinityWeight += weight;
+ } else {
+ finiteMeans[finiteCount] = mean;
+ finiteWeights[finiteCount++] = weight;
+ finiteMin = Math.min(finiteMin, mean);
+ finiteMax = Math.max(finiteMax, mean);
+ }
+ }
+ checkTotalWeight(totalWeight);
+ if (finiteCount > 0) {
+ finiteMin = Double.isFinite(other.getMin()) ? other.getMin() : finiteMin;
+ finiteMax = Double.isFinite(other.getMax()) ? other.getMax() : finiteMax;
+ _finiteDigest.addCentroids(
+ finiteMeans, finiteWeights, finiteCount, finiteMin, finiteMax,
false);
+ }
+ _finiteDigest.inheritHistoricalFractionalBoundaries(other);
+ _negativeInfinityWeight += negativeInfinityWeight;
+ _positiveInfinityWeight += positiveInfinityWeight;
+ mutated();
+ }
+
+ private void requireMutable() {
+ if (!hasValidStatistics()) {
+ throw corruptedMutation();
+ }
+ }
+
+ private static IllegalArgumentException corruptedMutation() {
+ return new IllegalArgumentException("Cannot merge or mutate a historically
corrupted TDigest; "
+ + "rebuild stored digests from source data before merging");
+ }
+
+ private void mutated() {
+ _originalFractionalBytes = null;
+ invalidateDerivedCaches();
+ }
+
+ @Override
+ public void compress() {
+ _finiteDigest.compress();
+ invalidateDerivedCaches();
+ }
+
+ @Override
+ public double getTotalWeight() {
+ return _finiteDigest.getTotalWeight() + _negativeInfinityWeight +
_positiveInfinityWeight;
+ }
+
+ @Override
+ double getHistoricalFractionalBoundaryMean(boolean lowerBoundary) {
+ return _finiteDigest.getHistoricalFractionalBoundaryMean(lowerBoundary);
+ }
+
+ @Override
+ public boolean hasValidStatistics() {
+ return _finiteDigest.hasValidStatistics();
+ }
+
+ @Override
+ public double cdf(double value) {
+ if (Double.isNaN(value) || Double.isInfinite(value)) {
+ throw new IllegalArgumentException(String.format("Invalid value: %f",
value));
+ }
+ double size = getTotalWeight();
+ if (size == 0.0 || !hasValidStatistics()) {
+ return Double.NaN;
+ }
+ double finiteSize = _finiteDigest.getTotalWeight();
+ double finiteWeight = finiteSize == 0L ? 0.0 : _finiteDigest.cdf(value) *
finiteSize;
+ return (_negativeInfinityWeight + finiteWeight) / size;
+ }
+
+ @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);
+ }
+ double size = getTotalWeight();
+ if (size == 0.0 || !hasValidStatistics()) {
+ return Double.NaN;
+ }
+ double index = quantile * size;
+ if (_negativeInfinityWeight > 0L && index < _negativeInfinityWeight) {
+ return Double.NEGATIVE_INFINITY;
+ }
+ if (_positiveInfinityWeight > 0L && index >= size -
_positiveInfinityWeight) {
+ return Double.POSITIVE_INFINITY;
+ }
+ double finiteSize = _finiteDigest.getTotalWeight();
+ if (finiteSize == 0L) {
+ return _negativeInfinityWeight > 0L ? Double.NEGATIVE_INFINITY :
Double.POSITIVE_INFINITY;
+ }
+ double finiteQuantile = (index - _negativeInfinityWeight) / finiteSize;
+ return _finiteDigest.quantile(Math.max(0.0, Math.min(finiteQuantile,
1.0)));
+ }
+
+ @Override
+ public Collection<Centroid> centroids() {
+ return getAllCentroids();
+ }
+
+ @Override
+ public double compression() {
+ return _finiteDigest.compression();
+ }
+
+ @Override
+ public int maxSerializedByteSize() {
+ if (_originalFractionalBytes != null) {
+ return _originalFractionalBytes.length;
+ }
+ if (_negativeInfinityWeight == 0.0 && _positiveInfinityWeight == 0.0) {
+ return _finiteDigest.maxSerializedByteSize();
+ }
+ int finiteByteSize = _finiteDigest.maxSerializedByteSize();
+ PercentileTDigestAccumulator finiteDigest = _finiteDigest;
+ if (finiteDigest.hasOriginalFractionalPayload()) {
+ // Infinite tails can make retained fractional boundaries interior;
their finite view may split endpoints.
+ finiteByteSize = Math.max(finiteByteSize,
+ TDigestCodec.getMaxLegacyCompatibleByteSize(compression(),
finiteDigest.getCentroidCountUpperBound()));
+ }
+ boolean hasFiniteValues = _finiteDigest.getTotalWeight() > 0.0;
+ int negativeInfinityCount =
getInfinityCentroidCount(_negativeInfinityWeight, true,
+ !hasFiniteValues && _positiveInfinityWeight == 0.0);
+ int positiveInfinityCount =
getInfinityCentroidCount(_positiveInfinityWeight,
+ !hasFiniteValues && _negativeInfinityWeight == 0.0, true);
+ int tailCount = Math.addExact(negativeInfinityCount,
positiveInfinityCount);
+ int tailByteSize = Math.multiplyExact(TDigestCodec.VERBOSE_CENTROID_SIZE,
tailCount);
+ long finiteCount = hasFiniteValues ?
finiteDigest.getCentroidCountUpperBound() : 0L;
+ int compatibleBound =
+ TDigestCodec.getMaxLegacyCompatibleByteSize(compression(),
Math.addExact(finiteCount, tailCount));
+ return Math.min(Math.addExact(finiteByteSize, tailByteSize),
compatibleBound);
+ }
+
+ @Override
+ public int centroidCount() {
+ return getAllCentroids().size();
+ }
+
+ @Override
+ public double getMin() {
+ if (_negativeInfinityWeight > 0L) {
+ return Double.NEGATIVE_INFINITY;
+ }
+ if (!_finiteDigest.isEmpty()) {
+ return _finiteDigest.getMin();
+ }
+ return Double.POSITIVE_INFINITY;
+ }
+
+ @Override
+ public double getMax() {
+ if (_positiveInfinityWeight > 0L) {
+ return Double.POSITIVE_INFINITY;
+ }
+ if (!_finiteDigest.isEmpty()) {
+ return _finiteDigest.getMax();
+ }
+ return Double.NEGATIVE_INFINITY;
+ }
+
+ private List<Centroid> getFiniteCentroids() {
+ if (_finiteCentroids == null) {
+ // Match the wire view without serializing finite-only fractional
boundaries surrounded by infinite tails.
+ PercentileTDigestAccumulator finiteDigest = _finiteDigest;
+ finiteDigest.prepareCentroidsForSerialization();
Review Comment:
Wrapper centroid reads now use the native working-compression view. Healthy
boundary repair works on the returned list without applying final public
compression to the accumulator. The seeded regression verifies both centroid
APIs and identical subsequent aggregation/serialization; its negative control
fails with the old wrapper.
Addressed in
[277a170ffe](https://github.com/apache/pinot/commit/277a170ffe224d680ee00c6f56efd422bb673990).
All 526 affected tests, actual 3.2/3.3 reader checks and the unchanged
270-case accuracy oracle pass.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/tdigest/TDigest.java:
##########
@@ -0,0 +1,139 @@
+/**
+ * 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.tdigest;
+
+import java.util.Collection;
+import java.util.List;
+
+/// Mutable percentile digest base for Pinot aggregation and legacy t-digest
byte encodings.
+/// Implementations are not thread-safe; callers must serialize access to each
aggregation state.
+/// Historical corrupted payloads may be retained unchanged, but cannot be
mutated or mixed with healthy state.
+/// Fractional mass is preserved; t-digest 3.3 can reject endpoint weights
below one or singleton masses between
+/// one and two during recompression because its endpoint invariant requires
unit weights.
+public abstract class TDigest {
+ public abstract void add(double value);
+
+ public void add(double value, int weight) {
+ add(value, (double) weight);
+ }
+
+ /// Adds positive finite mass, rejecting a valid distribution's total that
would overflow double.
+ public abstract void add(double value, double weight);
+
+ /// Merges an exclusively owned source; implementations may compress that
mutable source in place.
+ public abstract void add(TDigest other);
+
+ public void add(List<? extends TDigest> others) {
+ for (TDigest other : others) {
+ add(other);
+ }
+ }
+
+ public abstract void compress();
+
+ /// Returns the legacy long size view, saturating when the precise mass
exceeds its range.
+ public long size() {
+ return (long) getTotalWeight();
+ }
+
+ /// Returns centroid mass without narrowing fractional or large weights.
Valid totals must remain finite;
+ /// add and merge throw IllegalArgumentException when the summed mass
exceeds the finite double range.
+ /// An invalid historical payload may report its original negative total; it
cannot contribute to another digest.
+ public abstract double getTotalWeight();
+
+ /// Returns false when historical numerical corruption leaves the
distribution unknown.
+ public boolean hasValidStatistics() {
+ return true;
+ }
+
+ /// An unknown distribution remains present even when its historical weights
sum to zero.
+ public boolean isEmpty() {
+ return getTotalWeight() == 0.0 && hasValidStatistics();
+ }
+
+ public abstract double cdf(double value);
+
+ public abstract double quantile(double quantile);
+
+ /// Returns immutable centroid values with their precise double weights.
+ /// Reading this view may flush buffered inputs without applying the final
public compression.
+ public abstract Collection<Centroid> centroids();
+
+ public abstract double compression();
+
+ /// Serializes one independently owned legacy-compatible payload.
Implementations own their compression policy.
+ /// Unchanged historical encodings may be retained byte-exactly; fresh
unrepresentable boundaries are rejected.
+ public abstract byte[] serialize();
+
+ /// Bounds independently owned serialization without requiring a two-step
size/write protocol.
+ /// Mutable implementations can override this to avoid flushing buffered
inputs.
+ public int maxSerializedByteSize() {
+ return serialize().length;
+ }
+
+ public abstract int centroidCount();
+
+ public abstract double getMin();
+
+ public abstract double getMax();
+
+ /// Returns the mean of an unsupported fractional endpoint inherited from
validated legacy bytes, or NaN when
+ /// none is known. The lower or upper boundary is selected by
`lowerBoundary`. Producers of fresh values retain
+ /// this default; mergers propagate the provenance without granting
permission to new fractional extrema.
+ double getHistoricalFractionalBoundaryMean(boolean lowerBoundary) {
Review Comment:
Endpoint provenance is now cached in inspected metadata and preserved by
native/wrapper snapshots. Exact endpoint identity remains a deliberate safety
guard: approximate matching would incorrectly exempt fresh extrema. Regressions
preserve historical mass through merges, materialization and copies at
compression 10/100/1,000, and reject a newly added endpoint even one ULP away.
Custom readers can use fromBytes or the serialized-input contract to carry
validated provenance.
Addressed in
[277a170ffe](https://github.com/apache/pinot/commit/277a170ffe224d680ee00c6f56efd422bb673990).
All 526 affected tests, actual 3.2/3.3 reader checks and the unchanged
270-case accuracy oracle pass.
--
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]
