xiangfu0 commented on code in PR #19795:
URL: https://github.com/apache/pinot/pull/19795#discussion_r4243462406
##########
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileTDigestAggregationFunction.java:
##########
@@ -271,10 +273,10 @@ public TDigest extractGroupByResult(GroupByResultHolder
groupByResultHolder, int
@Override
public TDigest merge(TDigest intermediateResult1, TDigest
intermediateResult2) {
- if (intermediateResult1.size() == 0L) {
+ if (intermediateResult1.isEmpty()) {
Review Comment:
This remains the explicit fail-fast policy: present corrupted data is not
treated as an empty distribution or silently dropped. A NaN-weight legacy
payload does have size() == 0 in the old reader; the isolated reproduction
confirms the new digest retains its bytes/NaN statistics and rejects both
healthy+corrupt merge orders while preserving the healthy operand. The
migration guide documents whole-query/rollup/star-tree impact and the
rebuild/reingestion remedy. I retained that contract rather than inventing
combined RAW bytes or silently discarding unknown mass.
Verified at
[b5fa1ba341](https://github.com/apache/pinot/commit/b5fa1ba3419356294fd431a4ee2539809f1bf159):
551 affected tests and the real 3.2/3.3 compatibility checks pass.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/tdigest/TDigestCodec.java:
##########
@@ -0,0 +1,854 @@
+/**
+ * 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.lang.invoke.MethodHandles;
+import java.lang.invoke.VarHandle;
+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. Pure reads retain the legacy weighted boundary centroids. Adjacent
equal means use Pinot's point-mass
+/// interpolation, so quantile/CDF values can change on unchanged bytes,
including intervals next to a plateau.
+/// Mutation and serialization repair their boundaries only when unit
endpoints can preserve their total weight and
+/// first moment; otherwise unchanged historical bytes remain readable and
mutation fails explicitly.
+/// 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,
+ VERBOSE_HEADER_SIZE, VERBOSE_CENTROID_SIZE, buffer.capacity(),
encodedWeight,
+ hasNonFiniteMeans, false, false, fractionalWeights,
weightedBoundaries, false, compression, Double.NaN,
+ Double.NaN, 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.
+ /// Exact equality is intentional: endpoint repair preserves these means,
and admitting nearby means would grant
+ /// the exception to fresh extrema. Fresh fractional input at or beyond a
recorded endpoint permanently excludes
+ /// that endpoint, even if its historical mass was merged away or the new
weight equals the historical weight.
+ 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)
+ && !(source.getFreshFractionalBoundaryMean(true) <= firstMean))
+ && (!lastUnsupported || lastMean ==
source.getHistoricalFractionalBoundaryMean(false)
+ && !(source.getFreshFractionalBoundaryMean(false) >= lastMean));
+ }
+
+ 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.addHistoricalCentroids(means, weights, means.length,
metadata.min(), metadata.max(), true);
+
digest.inheritHistoricalFractionalBoundaries(metadata.historicalFractionalMinMean(),
+ metadata.historicalFractionalMaxMean());
+ 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 centroidOffset, int centroidSize,
+ int encodedLength, double totalWeight, boolean hasNonFiniteMeans,
boolean needsLegacyFallback,
+ boolean unorderedMeans, boolean fractionalWeights, boolean
weightedBoundaries, boolean hasZeroWeightCentroids,
+ double encodedCompression, double recoveredInfinityMean, double
historicalFractionalMinMean,
+ double historicalFractionalMaxMean) {
+ /// Whether the encoded centroid view can be used unchanged by healthy
pending-state readers and writers.
+ public boolean isCanonicalPayload() {
+ return !needsLegacyFallback && !weightedBoundaries && !unorderedMeans &&
!hasZeroWeightCentroids
+ && Double.isNaN(recoveredInfinityMean);
+ }
+ }
+
+ /// Invocation-local parser state. Primitive byte-array reads avoid per-row
buffer and metadata allocations.
+ /// Public metadata and retained pending payloads receive immutable
snapshots instead of sharing this state.
+ static final class SerializedTDigestState {
+ private static final VarHandle SHORT_VIEW =
+ MethodHandles.byteArrayViewVarHandle(short[].class,
ByteOrder.BIG_ENDIAN);
+ private static final VarHandle INT_VIEW =
MethodHandles.byteArrayViewVarHandle(int[].class, ByteOrder.BIG_ENDIAN);
+ private static final VarHandle LONG_VIEW =
MethodHandles.byteArrayViewVarHandle(long[].class, ByteOrder.BIG_ENDIAN);
+
+ int _encoding;
+ double _min;
+ double _max;
+ double _compression;
+ int _centroidCount;
+ int _centroidOffset;
+ int _centroidSize;
+ int _encodedLength;
+ double _totalWeight;
+ boolean _hasNonFiniteMeans;
+ boolean _needsLegacyFallback;
+ boolean _unorderedMeans;
+ boolean _fractionalWeights;
+ boolean _weightedBoundaries;
+ boolean _hasZeroWeightCentroids;
+ double _encodedCompression;
+ double _recoveredInfinityMean;
+ double _historicalFractionalMinMean;
+ double _historicalFractionalMaxMean;
+ private byte[] _bytes;
+ private ByteBuffer _buffer;
+ private int _position;
+ private int _limit;
+
+ void reset(byte[] bytes) {
+ _bytes = bytes;
+ _buffer = null;
+ _position = 0;
+ _limit = bytes.length;
+ }
+
+ void reset(ByteBuffer input) {
+ _bytes = null;
+ _buffer = input.slice().order(ByteOrder.BIG_ENDIAN);
+ _position = 0;
+ _limit = _buffer.limit();
+ }
+
+ void copyFrom(SerializedTDigestMetadata metadata) {
+ _encoding = metadata.encoding();
+ _min = metadata.min();
+ _max = metadata.max();
+ _compression = metadata.compression();
+ _centroidCount = metadata.centroidCount();
+ _centroidOffset = metadata.centroidOffset();
+ _centroidSize = metadata.centroidSize();
+ _encodedLength = metadata.encodedLength();
+ _totalWeight = metadata.totalWeight();
+ _hasNonFiniteMeans = metadata.hasNonFiniteMeans();
+ _needsLegacyFallback = metadata.needsLegacyFallback();
+ _unorderedMeans = metadata.unorderedMeans();
+ _fractionalWeights = metadata.fractionalWeights();
+ _weightedBoundaries = metadata.weightedBoundaries();
+ _hasZeroWeightCentroids = metadata.hasZeroWeightCentroids();
+ _encodedCompression = metadata.encodedCompression();
+ _recoveredInfinityMean = metadata.recoveredInfinityMean();
+ _historicalFractionalMinMean = metadata.historicalFractionalMinMean();
+ _historicalFractionalMaxMean = metadata.historicalFractionalMaxMean();
+ }
+
+ SerializedTDigestMetadata snapshot() {
+ return new SerializedTDigestMetadata(_encoding, _min, _max,
_compression, _centroidCount, _centroidOffset,
+ _centroidSize, _encodedLength, _totalWeight, _hasNonFiniteMeans,
_needsLegacyFallback, _unorderedMeans,
+ _fractionalWeights, _weightedBoundaries, _hasZeroWeightCentroids,
_encodedCompression, _recoveredInfinityMean,
+ _historicalFractionalMinMean, _historicalFractionalMaxMean);
+ }
+
+ private int readPosition(int length) {
+ if (length > _limit - _position) {
+ throw new BufferUnderflowException();
+ }
+ int position = _position;
+ _position += length;
+ return position;
+ }
+
+ private int readInt() {
+ int position = readPosition(Integer.BYTES);
+ return _bytes != null ? (int) INT_VIEW.get(_bytes, position) :
_buffer.getInt(position);
+ }
+
+ private short readShort() {
+ int position = readPosition(Short.BYTES);
+ return _bytes != null ? (short) SHORT_VIEW.get(_bytes, position) :
_buffer.getShort(position);
+ }
+
+ private double readDouble() {
+ int position = readPosition(Double.BYTES);
+ return _bytes != null
+ ? Double.longBitsToDouble((long) LONG_VIEW.get(_bytes, position))
+ : _buffer.getDouble(position);
+ }
+
+ private float readFloat() {
+ int position = readPosition(Float.BYTES);
+ return _bytes != null ? Float.intBitsToFloat((int) INT_VIEW.get(_bytes,
position)) : _buffer.getFloat(position);
+ }
+ }
+
+ /// 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) {
+ SerializedTDigestState state = new SerializedTDigestState();
+ state.reset(input);
+ readSerializedHeader(state, checkCapacity);
+ if (Double.isNaN(state._totalWeight)) {
+ inspectSerialized(state, null, null);
+ }
+ return state.snapshot();
+ }
+
+ /// 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) {
+ SerializedTDigestState state = new SerializedTDigestState();
+ state.reset(input);
+ readSerializedHeader(state, checkCapacity);
+ return state.snapshot();
+ }
+
+ static void readSerializedHeader(SerializedTDigestState state, boolean
checkCapacity) {
+ int encoding = state.readInt();
+ if (encoding != VERBOSE_ENCODING && encoding != SMALL_ENCODING) {
+ throw new IllegalStateException("Invalid format for serialized
histogram");
+ }
+ double min = state.readDouble();
+ double max = state.readDouble();
+ double compression;
+ double encodedCompression;
+ int mainCapacity;
+ int bufferCapacity = 0;
+ int centroidCount;
+ int centroidSize;
+ if (encoding == VERBOSE_ENCODING) {
+ encodedCompression = state.readDouble();
+ compression = normalizeCompression(encodedCompression);
+ mainCapacity = Math.max(getDefaultCentroidCapacity(compression),
getLegacyDefaultCentroidCapacity(compression));
+ centroidCount = state.readInt();
+ centroidSize = VERBOSE_CENTROID_SIZE;
+ } else {
+ encodedCompression = state.readFloat();
+ compression = normalizeCompression(encodedCompression);
+ mainCapacity = state.readShort();
+ bufferCapacity = state.readShort();
+ checkSmallArrayCapacity(mainCapacity);
+ checkSmallArrayCapacity(bufferCapacity);
+ if (mainCapacity == -1) {
+ mainCapacity = getDefaultCentroidCapacity(compression);
+ }
+ centroidCount = state.readShort();
+ centroidSize = SMALL_CENTROID_SIZE;
+ }
+ if (centroidCount < 0) {
+ throw new IllegalArgumentException("Invalid negative TDigest centroid
count: " + centroidCount);
+ }
+ if (centroidCount > (state._limit - state._position) / centroidSize) {
+ throw new BufferUnderflowException();
+ }
+ int centroidOffset = state._position;
+ int encodedLength = centroidOffset + centroidCount * centroidSize;
+ state._encoding = encoding;
+ state._min = min;
+ state._max = max;
+ state._compression = compression;
+ state._centroidCount = centroidCount;
+ state._centroidOffset = centroidOffset;
+ state._centroidSize = centroidSize;
+ state._encodedLength = encodedLength;
+ state._totalWeight = Double.NaN;
+ state._hasNonFiniteMeans = false;
+ state._needsLegacyFallback = Double.isNaN(min) || Double.isNaN(max) ||
centroidCount > 0 && min > max;
+ state._unorderedMeans = false;
+ state._fractionalWeights = false;
+ state._weightedBoundaries = false;
+ state._hasZeroWeightCentroids = false;
+ state._encodedCompression = encodedCompression;
+ state._recoveredInfinityMean = Double.NaN;
+ state._historicalFractionalMinMean = Double.NaN;
+ state._historicalFractionalMaxMean = Double.NaN;
+ if (checkCapacity && centroidCount > mainCapacity) {
+ inspectSerialized(state, null, null);
+ if (!state._needsLegacyFallback) {
+ throw new IllegalArgumentException("TDigest centroid count exceeds
capacity: " + centroidCount);
+ }
+ }
+ }
+
+ /// 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) {
+ SerializedTDigestState state = new SerializedTDigestState();
+ state.reset(input);
+ state.copyFrom(header);
+ inspectSerialized(state, means, weights);
+ return state.snapshot();
+ }
+
+ static void inspectSerialized(SerializedTDigestState state, double[] means,
double[] weights) {
+ state._position = state._centroidOffset;
+ int centroidCount = state._centroidCount;
+ int centroidSize = state._centroidSize;
+ double min = state._min;
+ double max = state._max;
+ double totalWeight = 0.0;
+ double previousMean = Double.NEGATIVE_INFINITY;
+ boolean hasNonFiniteMeans = false;
+ boolean needsLegacyFallback = state._needsLegacyFallback;
+ boolean unorderedMeans = false;
+ boolean fractionalWeights = false;
+ boolean hasZeroWeightCentroids = false;
+ int positiveCentroidCount = 0;
+ double firstPositiveWeight = 0.0;
+ double lastPositiveWeight = 0.0;
+ double lowerPositiveMean = Double.POSITIVE_INFINITY;
+ double upperPositiveMean = Double.NEGATIVE_INFINITY;
+ double lowerPositiveWeight = 0.0;
+ double upperPositiveWeight = 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;
+ // Legacy mean += (incoming - mean) * weight / totalWeight can cancel near
a small endpoint. Bound
+ // its value displacement using the finite input range's scale, rather
than ULPs of the small result.
+ double roundingTolerance = 4.0 * Math.max(Double.isFinite(min) ?
Math.ulp(min) : 0.0,
Review Comment:
Retained the bounded legacy-roundoff policy. Unmodified pre-PR Pinot
produced 6, 26 and 410 endpoint-ULP undershoots from ordinary finite inputs;
the 410-ULP example at min=0.1 is only 0.05005 ULP at max=1000. Thus scaling by
the arithmetic operands is deliberate: these bytes cannot distinguish
historical cancellation from a similar small corrupt displacement. Arbitrarily
clamping all finite outliers would instead invent statistics for contradictory
extrema. The old-source probe covers 180 cases and an additional 960
merge-chain states; neither establishes a universal bound. The existing tests
retain bytes/NaN beyond the allowance. A concrete legacy-writer output outside
that allowance would justify extending it; the current recommendation supplies
no such reproduction.
Verified at
[b5fa1ba341](https://github.com/apache/pinot/commit/b5fa1ba3419356294fd431a4ee2539809f1bf159):
551 affected tests and the real 3.2/3.3 compatibility checks 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]
