xiangfu0 commented on code in PR #19795:
URL: https://github.com/apache/pinot/pull/19795#discussion_r4224697562
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TDigestUtils.java:
##########
@@ -145,42 +127,48 @@ private static ByteBuffer prepareScratchBuffer(ByteBuffer
scratchBuffer, int req
///
/// The input must use the standard verbose `MergingDigest` encoding. The
returned payload stays verbose when it
/// already fits the 3.2 default capacity, uses the compact encoding only
when every narrowed field is exact, and
- /// otherwise reduces interior centroids without narrowing doubles.
+ /// 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.
public static byte[] makeLegacyCompatible(byte[] verboseBytes) {
- ByteBuffer verbose = ByteBuffer.wrap(verboseBytes);
- if (verbose.getInt() != VERBOSE_ENCODING) {
+ 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) {
+ if (metadata.encoding() != VERBOSE_ENCODING) {
throw new IllegalArgumentException("Expected verbose TDigest encoding");
}
- verbose.getDouble();
- verbose.getDouble();
- double compression = verbose.getDouble();
- if (!(compression > 0.0) || !Double.isFinite(compression)) {
- throw new IllegalArgumentException("Invalid TDigest compression: " +
compression);
+ if (metadata.needsLegacyFallback()) {
Review Comment:
**Bug (reproduced): write/read capacity asymmetry makes Pinot emit bytes it
cannot read back.** `makeLegacyCompatible` passes `needsLegacyFallback`
payloads through without capacity reduction, but every Pinot read path
(`readSerializedHeader` rejects `centroidCount > mainCapacity`) refuses them.
Repro: a 300-centroid verbose payload at compression 100 (capacity 210)
containing one NaN mean enters an accumulator through the capacity-unchecked
ingress (`add(TDigest)` / `reset(buffer, false)`); `serialize()` re-emits the
payload verbatim, and the receiving server's `deserializeIntermediateResult ->
forSerializedTDigest(ByteBuffer)` throws `IllegalArgumentException: "TDigest
centroid count exceeds capacity: 300"`, hard-failing the distributed query even
though the sender succeeded — contradicting the class doc that "these
historical exceptions remain readable by Pinot". The same IAE fires in
`addAccumulator` when a materialized degraded accumulator is merged
(`addSerializedTDigest(o
ther.serialize())` re-reads with `checkCapacity=true`). Degraded passthrough
reads likely need `checkCapacity=false` (or the writer must reduce/annotate
capacity).
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TDigestUtils.java:
##########
@@ -207,331 +195,226 @@ 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;
- }
-
- /// 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);
+ /// 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 TDigest configureScaleFunction(MergingDigest digest) {
- digest.setScaleFunction(ScaleFunction.K_1);
- return digest;
+ private static double validateSerialized(ByteBuffer input, boolean
checkCapacity) {
+ return inspectSerialized(input, checkCapacity).totalWeight();
}
- 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;
+ /// 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 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 one payload without mutating `input`, scanning the centroid
values once.
+ public static SerializedTDigestMetadata inspectSerialized(ByteBuffer input) {
+ return inspectSerialized(input, true);
}
- 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();
+ /// Inspects a payload, optionally enforcing the legacy decoder's declared
centroid capacity.
+ public static SerializedTDigestMetadata inspectSerialized(ByteBuffer input,
boolean checkCapacity) {
+ return inspectSerialized(input, readSerializedHeader(input,
checkCapacity), null, null);
}
- private static double readWeight(ByteBuffer encoded, int offset, int
centroidSize) {
- return centroidSize == VERBOSE_CENTROID_SIZE ? encoded.getDouble(offset) :
encoded.getFloat(offset);
+ /// Reads and validates the fixed-size header without scanning or allocating
centroid arrays.
+ public static SerializedTDigestMetadata readSerializedHeader(ByteBuffer
input) {
+ return readSerializedHeader(input, true);
}
- 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; centroid flags and total weight
are computed by the subsequent pass.
+ 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 (checkCapacity && centroidCount > mainCapacity) {
+ throw new IllegalArgumentException("TDigest centroid count exceeds
capacity: " + centroidCount);
}
- if (centroidCount > readableMainCapacity) {
- throw new ArrayIndexOutOfBoundsException(
- "Index " + readableMainCapacity + " out of bounds for length " +
readableMainCapacity);
- }
- }
-
- private static void checkSmallArrayCapacity(int capacity) {
- if (capacity < -1) {
- throw new NegativeArraySizeException(Integer.toString(capacity));
+ if (centroidCount > 0 && min > max) {
+ throw new IllegalArgumentException("Invalid TDigest extrema: " + min +
", " + max);
}
+ int centroidOffset = encoded.position();
+ int encodedLength = centroidOffset + centroidCount * centroidSize;
+ return 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);
}
- private static int normalizeBoundaries(double[] means, double[] weights, int
count, double min, double max) {
- if (count == 1) {
- double weight = weights[0];
- if (weight <= 1.0) {
- return count;
+ /// 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));
}
- double mean = means[0];
- means[0] = min;
- weights[0] = 1.0;
- if (weight == 2.0) {
- means[1] = max;
- weights[1] = 1.0;
- return 2;
+ 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)
+ || !Double.isFinite(mean) && (mean < min || mean > max);
+ hasNonFiniteMeans |= !Double.isFinite(mean);
+ unorderedMeans |= mean < previousMean;
+ if (!Double.isNaN(mean)) {
+ // A compact digest can be re-emitted as verbose without recovering
its double endpoint means. Old
+ // merge arithmetic can also drift by one ULP. Permit only these
identifiable endpoint roundings.
+ boolean lowerRounded = mean == roundedMin || mean == lowerAdjacent;
+ boolean upperRounded = mean == roundedMax || mean == upperAdjacent;
+ withinBounds &= (mean >= min || lowerRounded) && (mean <= max ||
upperRounded);
+ previousMean = mean;
+ }
+ totalWeight += weight;
+ if (!Double.isFinite(totalWeight)) {
+ throw new IllegalArgumentException("TDigest total weight exceeds the
supported range");
}
- means[1] = residualMean(mean, weight, min, max, weight - 2.0, min, max);
- weights[1] = weight - 2.0;
- means[2] = max;
- weights[2] = 1.0;
- return 3;
}
+ // Identifiable endpoint rounding can be restored safely. Other historical
out-of-extrema values have an
+ // unknown distribution: retain their original bytes and NaN statistics
instead of inventing clamped quantiles.
+ needsLegacyFallback |= !withinBounds;
+ boolean weightedBoundaries = positiveCentroidCount == 1 ?
firstPositiveWeight >= 2.0
+ : positiveCentroidCount > 1 && (firstPositiveWeight > 1.0 ||
lastPositiveWeight > 1.0);
+ return new SerializedTDigestMetadata(header.encoding(), min, max,
header.compression(), centroidCount,
+ header.mainCapacity(), header.bufferCapacity(),
header.centroidOffset(), centroidSize,
+ header.encodedLength(), totalWeight, hasNonFiniteMeans,
needsLegacyFallback, unorderedMeans, fractionalWeights,
+ weightedBoundaries, hasZeroWeightCentroids,
header.encodedCompression());
+ }
- if (weights[0] > 1.0) {
- System.arraycopy(means, 1, means, 2, count - 1);
- System.arraycopy(weights, 1, weights, 2, count - 1);
- double weight = weights[0];
- double mean = means[0];
- means[0] = min;
- weights[0] = 1.0;
- means[1] = residualMean(mean, weight, min, 0.0, weight - 1.0, min,
means[2]);
- weights[1] = weight - 1.0;
- count++;
- }
- int lastIndex = count - 1;
- if (weights[lastIndex] > 1.0) {
- double weight = weights[lastIndex];
- double mean = means[lastIndex];
- means[lastIndex] = residualMean(mean, weight, max, 0.0, weight - 1.0,
means[lastIndex - 1], max);
- weights[lastIndex] = weight - 1.0;
- means[count] = max;
- weights[count] = 1.0;
- count++;
+ /// Deserializes a finite-valued digest without recompressing stored
centroids on each rollup generation.
+ public static TDigest deserializeFinite(byte[] bytes) {
+ SerializedTDigestInput input = new SerializedTDigestInput();
+ input.reset(bytes);
+ return deserializeFinite(input);
+ }
+
+ /// Deserializes an already validated input without scanning the same
centroid bytes again.
+ public static TDigest deserializeFinite(SerializedTDigestInput input) {
+ SerializedTDigestMetadata metadata = input.getMetadata();
+ if (metadata.hasNonFiniteMeans() && !metadata.needsLegacyFallback()) {
+ throw new IllegalArgumentException("Expected finite TDigest centroid
means");
}
- return count;
+ PercentileTDigestAccumulator digest =
PercentileTDigestAccumulator.forLegacyAggregation(input.getCompression());
+ digest.addSerializedTDigest(input);
+ return digest;
}
- private static double clamp(double value, double min, double max) {
- return Math.max(min, Math.min(value, max));
+ /// Deserializes a digest and advances `input` past its encoded bytes.
+ public static TDigest deserialize(ByteBuffer input) {
+ SerializedTDigestInput encoded = new SerializedTDigestInput();
+ encoded.reset(input);
+ PercentileTDigestAccumulator digest =
PercentileTDigestAccumulator.forLegacyAggregation(encoded.getCompression());
+ digest.addSerializedTDigest(encoded);
+ return digest;
}
- private static double residualMean(double mean, double weight, double
firstRemovedValue,
- double secondRemovedValue, double residualWeight, double min, double
max) {
- if (!Double.isFinite(mean) || !Double.isFinite(firstRemovedValue)
- || !Double.isFinite(secondRemovedValue)) {
- return clamp(mean, min, max);
- }
- boolean removedTwoValues = weight - residualWeight == 2.0;
- int scaleShift = removedTwoValues ? 2 : 1;
- double scaledMean = Math.scalb(mean, -scaleShift);
- double scaledCorrection = scaledMean - Math.scalb(firstRemovedValue,
-scaleShift);
- if (removedTwoValues) {
- scaledCorrection += scaledMean - Math.scalb(secondRemovedValue,
-scaleShift);
+ private static void checkSmallArrayCapacity(int capacity) {
+ if (capacity < -1) {
+ throw new IllegalArgumentException("Invalid TDigest array capacity: " +
capacity);
}
- double correction = Math.scalb(scaledCorrection / residualWeight,
scaleShift);
- return clamp(mean + correction, min, max);
}
private static byte[] reduceVerboseCentroids(byte[] verboseBytes, int
maxCentroids) {
Review Comment:
**Hardening: `reduceVerboseCentroids` can write a NaN mean when merging two
adjacent zero-weight centroids.** `inspectSerialized` skips `weight == 0.0`
centroids before the fallback-flag checks, so zero-weight payloads reach this
reduction path, and `findNearestInteriorCentroids`'s tie-break actively prefers
the smallest combined weight. `weightedMean(m1, 0, m2, 0, totalWeight=0,
false)` with distinct means evaluates `0/0` and the min/max clamp does not
rescue NaN. Pinot's own readers later drop the zero-weight NaN centroid, but a
legacy 3.2/3.3 reader consuming these bytes during a rolling upgrade throws or
NaN-poisons. Only third-party payloads can trigger it today (in-tree writers
filter zero weights), but a one-line zero-weight filter when decoding here
closes it. Separately, this loop is O(n^2) in the number of reduced centroids.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/PercentileTDigestValueAggregator.java:
##########
@@ -152,128 +147,92 @@ private static NonFiniteAwareTDigest
asNonFiniteAware(TDigest value) {
}
private void updateMaxByteSize(TDigest value) {
- long defaultCapacity = getDefaultCentroidCapacity(value.compression());
- long maxCentroids = Math.min(value.size(), defaultCapacity);
- _maxByteSize = Math.max(_maxByteSize, getMaxVerboseByteSize(maxCentroids));
- }
-
- private static int getMaxVerboseByteSize(long centroidCount) {
- long maxCentroids = Math.max(centroidCount, 0L);
- return Math.toIntExact(Math.addExact(VERBOSE_HEADER_SIZE,
- Math.multiplyExact(VERBOSE_CENTROID_SIZE, maxCentroids)));
- }
-
- private static long getDefaultCentroidCapacity(double compression) {
- double normalizedCompression = Math.max(MIN_COMPRESSION, compression);
- int padding = normalizedCompression < 30.0 ?
LOW_COMPRESSION_CAPACITY_PADDING
- : DEFAULT_CENTROID_CAPACITY_PADDING;
- long defaultCapacity = (long) Math.ceil(2.0 * normalizedCompression +
padding);
- long legacyCapacity = Math.addExact(Math.multiplyExact(2L, (long)
Math.ceil(compression)),
- DEFAULT_CENTROID_CAPACITY_PADDING);
- return Math.max(defaultCapacity, legacyCapacity);
+ _maxByteSize = Math.max(_maxByteSize, value.maxSerializedByteSize());
}
/// Keeps non-finite values as exact tail masses while delegating all finite
values to the standard implementation.
///
/// The wrapper emits the standard `MergingDigest` wire format. It is
mutable, externally synchronized by the
/// segment-creation pipeline, and not safe for concurrent mutation.
private static final class NonFiniteAwareTDigest extends TDigest {
- private static final int VERBOSE_ENCODING = 1;
- private static final int SMALL_ENCODING = 2;
- private static final int VERBOSE_HEADER_SIZE = 32;
- private static final int VERBOSE_CENTROID_SIZE = 16;
-
private final TDigest _finiteDigest;
- private long _negativeInfinityWeight;
- private long _positiveInfinityWeight;
+ private double _negativeInfinityWeight;
+ private double _positiveInfinityWeight;
private byte[] _finiteSerializedBytes;
private List<Centroid> _finiteCentroids;
private byte[] _serializedBytes;
private NonFiniteAwareTDigest(double compression) {
- this(TDigestUtils.createMergingDigestWithLegacyBuffer(compression), 0L,
0L);
+ this(TDigestUtils.createMergingDigest(compression), 0L, 0L);
}
- private NonFiniteAwareTDigest(TDigest finiteDigest, long
negativeInfinityWeight, long positiveInfinityWeight) {
+ private NonFiniteAwareTDigest(TDigest finiteDigest, double
negativeInfinityWeight, double positiveInfinityWeight) {
_finiteDigest = finiteDigest;
_negativeInfinityWeight = negativeInfinityWeight;
_positiveInfinityWeight = positiveInfinityWeight;
}
private NonFiniteAwareTDigest copy() {
- invalidateCaches();
- int minimumMainCapacity = Math.addExact(_finiteDigest.centroidCount(),
2);
- TDigest finiteDigest =
TDigestUtils.createMergingDigestWithLegacyBuffer(compression(),
minimumMainCapacity);
- if (_finiteDigest.size() > 0L) {
- finiteDigest.add(List.of(_finiteDigest));
+ TDigest finiteDigest = TDigestUtils.createMergingDigest(compression());
+ if (!_finiteDigest.isEmpty()) {
+ finiteDigest.add(_finiteDigest);
}
return new NonFiniteAwareTDigest(finiteDigest, _negativeInfinityWeight,
_positiveInfinityWeight);
}
private static NonFiniteAwareTDigest fromBytes(byte[] bytes) {
Review Comment:
**Cleanup: `NonFiniteAwareTDigest` hand-rolls the centroid codec a
fourth/fifth time.** `fromBytes` manually loops the `verbose ? getDouble :
getFloat` decode with its own compact-mean clamping, and `toVerboseBytes`
re-implements the verbose writer — duplicating
`TDigestUtils.inspectSerialized(ByteBuffer, header, means, weights)` (which
already decodes into caller-supplied arrays with the identical endpoint clamp
plus validation) and the accumulator's verbose writer. The copies have already
drifted once (clamping applied in some decode loops, not others). Routing
through a single codec in `TDigestUtils` — and exposing a bulk sorted
means/weights entry point on the accumulator instead of the encode-then-reparse
roundtrip in `add`/`fromBytes` — would remove both the drift risk and an
allocation-heavy detour on the star-tree path.
##########
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileSmartTDigestAggregationFunction.java:
##########
@@ -340,8 +341,7 @@ public SerializedIntermediateResult
serializeIntermediateResult(Object o) {
@Override
public Object deserializeIntermediateResult(CustomObject customObject) {
if (customObject.getType() ==
ObjectSerDeUtils.ObjectType.TDigest.getValue()) {
- // Generic TDigest deserialization returns a plain MergingDigest. Keep
this function's TDigest intermediates as
- // accumulators so subsequent merges retain the capacity-preserving
serialization path.
+ // Retain serialized state lazily so subsequent merges preserve source
capacity and compression.
Review Comment:
**Cleanup: the accumulator-strategy choice for deserialized intermediates is
duplicated per aggregation function.** Both this function and
`PercentileTDigestAggregationFunction` override `deserializeIntermediateResult`
to bypass `ObjectSerDeUtils.TDIGEST_SER_DE` and call
`PercentileTDigestAccumulator.forSerializedTDigest`, because the shared
ser-de's `deserialize` builds a different (legacy-strategy) accumulator. A
future TDigest-producing function that forgets the override silently gets
different accuracy/perf characteristics. Centralizing the decision in the
ser-de (or a single factory like `forIntermediateResult`) would let the
per-function overrides disappear.
##########
pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/customobject/TDigest.java:
##########
@@ -0,0 +1,135 @@
+/**
+ * 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.spi.customobject;
+
+import java.nio.ByteBuffer;
+import java.util.Collection;
+import java.util.List;
+
+/// Mutable percentile digest contract for aggregation plugins and legacy
t-digest byte encodings.
+/// Implementations are not thread-safe; callers must serialize access to each
aggregation 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 {
Review Comment:
**Architecture: consider whether this belongs in `pinot-segment-spi` with
this surface.** Nothing in segment-spi uses it, implementations cannot really
be plugged (ser-de is hardwired to `PercentileTDigestAccumulator` in
segment-local), and its direct analogue `QuantileDigest` lives in
segment-local's `customobject`. Placing it in the SPI freezes the replicated
tdunning warts (`byteSize`/`asBytes`/`smallByteSize`/`asSmallBytes`, saturating
`size()`, `compress()`-as-flush) as public API whose future cleanup becomes a
breaking SPI change. All users (pinot-core, segment-local, perf) already depend
on segment-local; trimming the production contract to what call sites use
(add/merge/quantile/cdf/totalWeight/isEmpty/compression + one serialize) and
keeping the legacy encodings inside `TDigestUtils` would be a smaller frozen
surface.
--
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]
