xiangfu0 commented on code in PR #19795:
URL: https://github.com/apache/pinot/pull/19795#discussion_r4244257900
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/PercentileTDigestValueAggregator.java:
##########
@@ -128,562 +117,29 @@ public int getMaxAggregatedValueByteSize() {
@Override
public byte[] serializeAggregatedValue(TDigest value) {
- int requiredCapacity =
getMaxVerboseByteSize(getDefaultCentroidCapacity(value.compression()));
- if (_serializationBuffer == null || _serializationBuffer.capacity() <
requiredCapacity) {
- _serializationBuffer = ByteBuffer.allocate(requiredCapacity);
- }
- byte[] bytes = asNonFiniteAware(value).serialize(_serializationBuffer);
+ NonFiniteAwareTDigest wrapped = asNonFiniteAware(value);
+ updateMaxByteSize(wrapped);
+ byte[] bytes = wrapped.serialize();
_maxByteSize = Math.max(_maxByteSize, bytes.length);
return bytes;
}
@Override
public TDigest deserializeAggregatedValue(byte[] bytes) {
+ _maxByteSize = Math.max(_maxByteSize, bytes.length);
return NonFiniteAwareTDigest.fromBytes(bytes);
}
private static NonFiniteAwareTDigest asNonFiniteAware(TDigest value) {
Review Comment:
Fixed cloneAggregatedValue to use a snapshotting copy factory. Native
accumulator buffers are copied before conversion into the non-finite-aware
wrapper, so any consuming merge operates on the copy. The clone regression now
covers native and wrapped sources with and without an infinity tail, then
mutates/serializes the clone and continues adding to the original against an
untouched control. It reproduces the old source drift (1120-byte output versus
the control's 1024 bytes) and passes after the fix.
Validated at `3821623d46933dd233b8110e123ecb4f86ac1e96`: 554 affected tests,
genuine 3.2/3.3 bidirectional byte checks, the 270-case oracle, applicable
precommit checks and independent Pinot review pass. New-head CI and maintainer
approval remain separate gates.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/tdigest/TDigestCodec.java:
##########
@@ -0,0 +1,850 @@
+/**
+ * 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;
Review Comment:
The existing testLegacyPoisonedExtremaRemainReadableAndRetainBytes already
covers min=3 > max=-1 for both verbose and compact encodings. It verifies
invalid statistics, NaN quantile/CDF, exact original-byte retention after
repeated read/serialization, rejection of mutation, and preservation after the
failed mutation; the focused check passes. A separate boundary-validation test
also checks inverted extrema. The README already identifies inverted extrema as
preserved opaque data. No duplicate regression test was added.
Validated at `3821623d46933dd233b8110e123ecb4f86ac1e96`: 554 affected tests,
genuine 3.2/3.3 bidirectional byte checks, the 270-case oracle, applicable
precommit checks and independent Pinot review pass. New-head CI and maintainer
approval remain separate gates.
--
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]
