xiangfu0 commented on code in PR #19795:
URL: https://github.com/apache/pinot/pull/19795#discussion_r4225311296
##########
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:
Addressed in 14657b2cfb. Oversized verbose headers are inspected only on the
exceptional path: healthy over-capacity input is rejected, while historical
degraded bytes remain opaque and unchanged. Ordinary header reads stay
constant-time. Since degraded mutation now fails, no new over-capacity degraded
payload is fabricated.
##########
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:
Addressed in 14657b2cfb. Zero-weight centroids are removed before
compatibility conversion or nearest-pair reduction, including zero-weight NaN
means. The new compression-100.1 fixture repairs 220 positive plus 100 zero
centroids to 211 and passes real 3.2/3.3 read/add/compress/rewrite with
assertions enabled. The quadratic reduction loop remains confined to oversized
legacy compatibility repair; ordinary native serialization does not use it.
##########
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:
Addressed in 14657b2cfb. The wrapper now decodes through
TDigestUtils.inspectSerialized into primitive arrays, initializes/merges
through addCentroids, and writes through the shared serializeCentroids helper.
Its remaining loop only partitions finite values and infinity tails. The local
verbose writer and encode/reparse detours are removed; whole-digest
serialization also avoids rejecting valid fractional interior mass between unit
infinity endpoints.
##########
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:
Addressed in 14657b2cfb. Centralized type-10 intermediate decoding in
ObjectSerDeUtils.TDIGEST_SER_DE for both byte[] and ByteBuffer. Scalar and RAW
functions use it, and Smart no longer needs its custom type-10 branch. Explicit
function overrides remain because the aggregation interface default throws.
Stored sketch readers retain their separate legacy policy; the regression
proves appended values use reduction compression.
--
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]
