xiangfu0 commented on code in PR #19795:
URL: https://github.com/apache/pinot/pull/19795#discussion_r4225666548
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TDigestUtils.java:
##########
@@ -202,336 +258,257 @@ public static byte[] makeLegacyCompatible(byte[]
verboseBytes) {
return small.array();
}
+ private static byte[] removeZeroWeightCentroids(byte[] verboseBytes,
SerializedTDigestMetadata metadata) {
+ ByteBuffer source = ByteBuffer.wrap(verboseBytes);
+ source.position(metadata.centroidOffset());
+ ByteBuffer filtered = ByteBuffer.allocate(metadata.encodedLength());
+ filtered.put(verboseBytes, 0, VERBOSE_HEADER_SIZE);
+ int count = 0;
+ for (int i = 0; i < metadata.centroidCount(); i++) {
+ double weight = source.getDouble();
+ double mean = source.getDouble();
+ if (weight != 0.0) {
+ filtered.putDouble(weight).putDouble(mean);
+ count++;
+ }
+ }
+ filtered.putInt(28, count);
+ return Arrays.copyOf(filtered.array(), filtered.position());
+ }
+
/// Deserializes a digest and repairs boundary centroids produced by
t-digest 3.2 when necessary.
public static TDigest deserialize(byte[] bytes) {
return deserialize(ByteBuffer.wrap(bytes));
}
- /// Validates the structure and declared capacities of a serialized merging
digest.
- public static void validateSerialized(byte[] bytes) {
- validateEncodedDigest(ByteBuffer.wrap(bytes));
+ /// Validates serialized lengths and centroid values without allocating
centroid arrays.
+ public static double validateSerialized(byte[] bytes) {
+ return validateSerialized(ByteBuffer.wrap(bytes));
}
- /// Deserializes a finite-valued digest into a merging digest with the
temporary-buffer footprint used by 3.2.
+ /// Validates one legacy digest without changing `input` and returns its
total centroid weight.
///
- /// Unlike the standard verbose decoder, this method does not retain the
substantially larger default 3.3
- /// temporary arrays. It also repairs legacy weighted boundary centroids
before the first 3.3 merge.
- public static TDigest deserializeFiniteWithLegacyBuffer(byte[] bytes) {
- ByteBuffer input = ByteBuffer.wrap(bytes);
- validateEncodedDigest(input.duplicate());
-
- int encoding = input.getInt();
- double min = input.getDouble();
- double max = input.getDouble();
- double compression;
- int centroidCount;
- boolean verbose;
- if (encoding == VERBOSE_ENCODING) {
- compression = input.getDouble();
- centroidCount = input.getInt();
- verbose = true;
- } else if (encoding == SMALL_ENCODING) {
- compression = input.getFloat();
- input.getShort();
- input.getShort();
- centroidCount = input.getShort();
- verbose = false;
- } else {
- throw new IllegalStateException("Invalid format for serialized
histogram");
- }
-
- if (centroidCount == 0) {
- return createMergingDigestWithLegacyBuffer(compression);
- }
- double[] means = new double[centroidCount + 2];
- double[] weights = new double[centroidCount + 2];
- for (int i = 0; i < centroidCount; i++) {
- double weight = verbose ? input.getDouble() : input.getFloat();
- double mean = verbose ? input.getDouble() : input.getFloat();
- if (!(weight > 0.0) || !Double.isFinite(weight)) {
- throw new IllegalArgumentException("Invalid TDigest centroid weight: "
+ weight);
- }
- if (!Double.isFinite(mean)) {
- throw new IllegalArgumentException("Expected a finite TDigest centroid
mean: " + mean);
- }
- weights[i] = weight;
- means[i] = mean;
- }
-
- int normalizedCount = weights[0] != 1.0 || weights[centroidCount - 1] !=
1.0
- ? normalizeBoundaries(means, weights, centroidCount, min, max) :
centroidCount;
- long addCount = 0L;
- for (int i = 0; i < normalizedCount; i++) {
- double weight = weights[i];
- if (weight != Math.rint(weight) || weight >= 0x1p63) {
- return copyIntoLegacyBuffer(
- deserialize(toVerboseBytes(min, max, compression, means, weights,
normalizedCount)), normalizedCount);
- }
- long integralWeight = (long) weight;
- addCount = Math.addExact(addCount, 1L + (integralWeight - 1L) /
Integer.MAX_VALUE);
- }
-
- int minimumMainCapacity = Math.addExact(Math.toIntExact(addCount), 2);
- TDigest digest = createMergingDigestWithLegacyBuffer(compression,
minimumMainCapacity);
- for (int i = 0; i < normalizedCount; i++) {
- long remainingWeight = (long) weights[i];
- while (remainingWeight > 0L) {
- int weight = (int) Math.min(remainingWeight, Integer.MAX_VALUE);
- digest.add(means[i], weight);
- remainingWeight -= weight;
- }
- }
- return digest;
+ /// Trailing bytes remain available to the caller for intermediate values
containing several objects.
+ public static double validateSerialized(ByteBuffer input) {
+ return validateSerialized(input, true);
}
- /// Deserializes a digest and advances `input` past its encoded bytes.
- public static TDigest deserialize(ByteBuffer input) {
- ByteBuffer encoded = input.slice();
- validateEncodedDigest(encoded.duplicate());
-
- int encoding = encoded.getInt();
- double min = encoded.getDouble();
- double max = encoded.getDouble();
- double compression;
- int mainCapacity;
- int bufferCapacity;
- int centroidCount;
- int centroidSize;
- if (encoding == VERBOSE_ENCODING) {
- compression = encoded.getDouble();
- centroidCount = encoded.getInt();
- mainCapacity = getDefaultCentroidCapacity(compression);
- bufferCapacity = 5 * mainCapacity;
- centroidSize = VERBOSE_CENTROID_SIZE;
- } else if (encoding == SMALL_ENCODING) {
- compression = encoded.getFloat();
- mainCapacity = encoded.getShort();
- bufferCapacity = encoded.getShort();
- centroidCount = encoded.getShort();
- centroidSize = SMALL_CENTROID_SIZE;
- } else {
- // MergingDigest.fromBytes() has already produced the canonical
exception for this case.
- throw new IllegalStateException("Invalid format for serialized
histogram");
- }
- int centroidOffset = encoded.position();
- if (centroidCount == 0) {
- return configureScaleFunction(MergingDigest.fromBytes(input));
- }
-
- double firstWeight = readWeight(encoded, centroidOffset, centroidSize);
- double lastWeight = readWeight(encoded, centroidOffset + (centroidCount -
1) * centroidSize, centroidSize);
- int defaultMainCapacity = getDefaultCentroidCapacity(compression);
- boolean needsBoundaryRepair = firstWeight != 1.0 || lastWeight != 1.0;
- boolean exceedsDefaultVerboseCapacity = encoding == VERBOSE_ENCODING &&
centroidCount > defaultMainCapacity;
- if (!needsBoundaryRepair && !exceedsDefaultVerboseCapacity) {
- return configureScaleFunction(MergingDigest.fromBytes(input));
- }
-
- double[] means = new double[centroidCount + 2];
- double[] weights = new double[centroidCount + 2];
- encoded.position(centroidOffset);
- for (int i = 0; i < centroidCount; i++) {
- if (centroidSize == VERBOSE_CENTROID_SIZE) {
- weights[i] = encoded.getDouble();
- means[i] = encoded.getDouble();
- } else {
- weights[i] = encoded.getFloat();
- means[i] = encoded.getFloat();
- }
- }
- int encodedLength = encoded.position();
- int normalizedCount = needsBoundaryRepair
- ? normalizeBoundaries(means, weights, centroidCount, min, max) :
centroidCount;
- byte[] normalizedBytes = toVerboseBytes(min, max, compression, means,
weights, normalizedCount);
- MergingDigest result;
- if (normalizedCount <= defaultMainCapacity) {
- result = MergingDigest.fromBytes(ByteBuffer.wrap(normalizedBytes));
- } else {
- int resolvedMainCapacity = mainCapacity == -1 ? defaultMainCapacity :
mainCapacity;
- int resolvedBufferCapacity = bufferCapacity == -1
- ? DEFAULT_MERGE_BUFFER_MULTIPLIER * resolvedMainCapacity :
bufferCapacity;
- int smallMainCapacity = Math.max(resolvedMainCapacity, normalizedCount);
- long smallBufferCapacity = Math.max(Math.max((long)
resolvedBufferCapacity, smallMainCapacity + 1L),
- 2L * smallMainCapacity + 1L);
- if ((double) (float) compression == compression && normalizedCount <=
Short.MAX_VALUE
- && smallMainCapacity <= Short.MAX_VALUE && smallBufferCapacity <=
Short.MAX_VALUE
- && areFloatExact(means, weights, normalizedCount)) {
- ByteBuffer normalized = ByteBuffer.allocate(SMALL_HEADER_SIZE +
SMALL_CENTROID_SIZE * normalizedCount);
- normalized.putInt(SMALL_ENCODING);
- normalized.putDouble(min);
- normalized.putDouble(max);
- normalized.putFloat((float) compression);
- normalized.putShort((short) smallMainCapacity);
- normalized.putShort((short) smallBufferCapacity);
- normalized.putShort((short) normalizedCount);
- for (int i = 0; i < normalizedCount; i++) {
- normalized.putFloat((float) weights[i]);
- normalized.putFloat((float) means[i]);
- }
- normalized.flip();
- result = MergingDigest.fromBytes(normalized);
- } else {
- result = MergingDigest.fromBytes(
- ByteBuffer.wrap(reduceVerboseCentroids(normalizedBytes,
defaultMainCapacity)));
- }
- }
- input.position(Math.addExact(input.position(), encodedLength));
- return configureScaleFunction(result);
+ private static double validateSerialized(ByteBuffer input, boolean
checkCapacity) {
+ return inspectSerialized(input, checkCapacity).totalWeight();
}
- private static TDigest configureScaleFunction(MergingDigest digest) {
- digest.setScaleFunction(ScaleFunction.K_1);
- return digest;
+ /// Header and numerical metadata for one legacy byte payload. The centroid
bytes remain owned by the caller.
+ /// Header-only inspection leaves total weight as NaN until centroid
inspection validates the numerical fields.
+ ///
+ /// Legacy writers can emit poisoned numerical state after overflow or
broken fractional boundary repair.
+ /// Such state is retained with degraded quantiles, rather than being passed
to the sorted K1 kernel.
+ public record SerializedTDigestMetadata(int encoding, double min, double
max, double compression,
+ int centroidCount, int mainCapacity, int bufferCapacity, int
centroidOffset, int centroidSize,
+ int encodedLength, double totalWeight, boolean hasNonFiniteMeans,
boolean needsLegacyFallback,
+ boolean unorderedMeans, boolean fractionalWeights, boolean
weightedBoundaries, boolean hasZeroWeightCentroids,
+ double encodedCompression) {
}
- private static TDigest copyIntoLegacyBuffer(TDigest source, int
centroidCount) {
- TDigest digest = createMergingDigestWithLegacyBuffer(source.compression(),
Math.addExact(centroidCount, 2));
- digest.add(java.util.List.of(source));
- return digest;
+ /// Inspects one payload without mutating `input`, scanning the centroid
values once.
+ public static SerializedTDigestMetadata inspectSerialized(ByteBuffer input) {
+ return inspectSerialized(input, true);
}
- private static boolean areFloatExact(double[] means, double[] weights, int
count) {
- for (int i = 0; i < count; i++) {
- if ((double) (float) weights[i] != weights[i] || (double) (float)
means[i] != means[i]) {
- return false;
- }
- }
- return 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;
}
- private static byte[] toVerboseBytes(double min, double max, double
compression, double[] means, double[] weights,
- int count) {
- ByteBuffer normalized = ByteBuffer.allocate(VERBOSE_HEADER_SIZE +
VERBOSE_CENTROID_SIZE * count);
- normalized.putInt(VERBOSE_ENCODING);
- normalized.putDouble(min);
- normalized.putDouble(max);
- normalized.putDouble(compression);
- normalized.putInt(count);
- for (int i = 0; i < count; i++) {
- normalized.putDouble(weights[i]);
- normalized.putDouble(means[i]);
- }
- return normalized.array();
+ /// 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);
}
- private static double readWeight(ByteBuffer encoded, int offset, int
centroidSize) {
- return centroidSize == VERBOSE_CENTROID_SIZE ? encoded.getDouble(offset) :
encoded.getFloat(offset);
- }
-
- private static void validateEncodedDigest(ByteBuffer input) {
- int encoding = input.getInt();
+ /// Reads a fixed-size header, optionally enforcing the declared centroid
capacity.
+ ///
+ /// NaN extrema mark legacy degraded state. Oversized payloads require
numerical inspection: historical degraded
+ /// state remains readable by Pinot even if an old decoder's default arrays
could not hold its original centroids.
+ public static SerializedTDigestMetadata readSerializedHeader(ByteBuffer
input, boolean checkCapacity) {
+ ByteBuffer encoded = input.slice().order(ByteOrder.BIG_ENDIAN);
+ int encoding = encoded.getInt();
if (encoding != VERBOSE_ENCODING && encoding != SMALL_ENCODING) {
throw new IllegalStateException("Invalid format for serialized
histogram");
}
- input.getDouble();
- input.getDouble();
+ 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) {
- compression = input.getDouble();
- mainCapacity = getDefaultCentroidCapacity(compression);
- centroidCount = input.getInt();
+ encodedCompression = encoded.getDouble();
+ compression = normalizeCompression(encodedCompression);
+ mainCapacity = Math.max(getDefaultCentroidCapacity(compression),
getLegacyDefaultCentroidCapacity(compression));
+ centroidCount = encoded.getInt();
centroidSize = VERBOSE_CENTROID_SIZE;
} else {
- compression = input.getFloat();
- mainCapacity = input.getShort();
- int bufferCapacity = input.getShort();
+ encodedCompression = encoded.getFloat();
+ compression = normalizeCompression(encodedCompression);
+ mainCapacity = encoded.getShort();
+ bufferCapacity = encoded.getShort();
checkSmallArrayCapacity(mainCapacity);
checkSmallArrayCapacity(bufferCapacity);
if (mainCapacity == -1) {
mainCapacity = getDefaultCentroidCapacity(compression);
}
- centroidCount = input.getShort();
+ centroidCount = encoded.getShort();
centroidSize = SMALL_CENTROID_SIZE;
}
if (centroidCount < 0) {
throw new IllegalArgumentException("Invalid negative TDigest centroid
count: " + centroidCount);
}
- if (centroidCount > input.remaining() / centroidSize) {
+ if (centroidCount > encoded.remaining() / centroidSize) {
throw new BufferUnderflowException();
}
- int readableMainCapacity = mainCapacity;
- if (encoding == VERBOSE_ENCODING && compression > 0.0 &&
Double.isFinite(compression)) {
- readableMainCapacity = Math.max(mainCapacity,
getLegacyDefaultCentroidCapacity(compression));
+ if (centroidCount > 0 && min > max) {
+ throw new IllegalArgumentException("Invalid TDigest extrema: " + min +
", " + max);
}
- if (centroidCount > readableMainCapacity) {
- throw new ArrayIndexOutOfBoundsException(
- "Index " + readableMainCapacity + " out of bounds for length " +
readableMainCapacity);
+ 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), false, false, false, false,
encodedCompression);
+ 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;
}
- private static void checkSmallArrayCapacity(int capacity) {
- if (capacity < -1) {
- throw new NegativeArraySizeException(Integer.toString(capacity));
+ /// 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;
+ 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);
+ }
+ // Legacy compact writers can narrow a finite double endpoint outside
the float range to infinity.
+ // Recover that identifiable rounding before classifying non-finite
state; verbose infinity remains opaque.
+ if (header.encoding() == SMALL_ENCODING) {
+ if (mean == Double.NEGATIVE_INFINITY && Double.isFinite(min) &&
roundedMin == mean) {
+ mean = min;
+ } else if (mean == Double.POSITIVE_INFINITY && Double.isFinite(max) &&
roundedMax == mean) {
+ mean = max;
+ }
+ }
+ if (means != null) {
+ means[i] = Math.max(min, Math.min(mean, max));
+ }
+ if (weights != null) {
+ weights[i] = weight;
+ }
+ // A zero-mass centroid has no distribution to merge, even when its
unused mean is NaN or stale.
+ if (weight == 0.0) {
+ hasZeroWeightCentroids = true;
+ continue;
+ }
+ if (weight > 0.0) {
+ if (positiveCentroidCount++ == 0) {
+ firstPositiveWeight = weight;
+ }
+ lastPositiveWeight = weight;
+ }
+ fractionalWeights |= weight != Math.rint(weight);
+ // A previous Pinot boundary repair emitted a negative residual for
singleton weights between one and two.
+ // Old tdunning merges could also overflow finite means to NaN or
infinity. Preserve those stored bytes,
+ // but never normalize or sort them into a plausible distribution.
+ needsLegacyFallback |= weight < 0.0 || Double.isNaN(mean)
Review Comment:
**[review] Legacy digests with infinity values degrade to NaN percentiles
(confirmed empirically).** tdunning's MergingDigest routinely emits NaN-mean
centroids when it merges two infinite centroids (`inf + (inf-inf)*k`), so NaN
means in stored bytes are the *normal* serialized form of any column whose data
contained >= 2 infinity values — not only historical corruption. This line
classifies every NaN mean as `needsLegacyFallback`, so such digests deserialize
as degraded and `quantile()`/`cdf()` return NaN for every percentile, where the
removed tdunning reader (and pre-PR Pinot) returned accurate interior
quantiles. Fuzzing 40 tdunning-3.2-written digests built from uniform data plus
~1% `add(+Inf)` calls: 31/40 now answer `quantile(0.5) == NaN` (old Pinot:
bit-identical to tdunning, e.g. 0.5013...). Since the NaN-mean centroids are
identifiable (adjacent to the infinite tail, weights intact), their mass could
be folded into the infinity tails the way `NonFiniteAwareTDigest` al
ready does, instead of degrading the whole digest.
--
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]
