xiangfu0 commented on code in PR #19795:
URL: https://github.com/apache/pinot/pull/19795#discussion_r4223007349
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TDigestUtils.java:
##########
@@ -207,331 +181,195 @@ 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;
- }
-
- 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) {
}
- 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;
int mainCapacity;
+ int bufferCapacity = 0;
int centroidCount;
int centroidSize;
if (encoding == VERBOSE_ENCODING) {
- compression = input.getDouble();
- mainCapacity = getDefaultCentroidCapacity(compression);
- centroidCount = input.getInt();
+ compression = normalizeCompression(encoded.getDouble());
+ 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();
+ compression = normalizeCompression(encoded.getFloat());
+ 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);
}
- 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 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);
}
- double mean = means[0];
- means[0] = min;
- weights[0] = 1.0;
- if (weight == 2.0) {
- means[1] = max;
- weights[1] = 1.0;
- return 2;
+ 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;
+ }
+ if (means != null) {
+ means[i] = Math.max(min, Math.min(mean, max));
+ }
+ if (weights != null) {
+ weights[i] = weight;
+ }
+ 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;
- }
-
- 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++;
+ if (!needsLegacyFallback && !withinBounds) {
+ throw new IllegalArgumentException("TDigest centroid mean is outside its
extrema: " + min + ", " + max);
Review Comment:
Finite nonpositive compression was already clamped to ten; regressions
confirm it. Historical means beyond identifiable endpoint rounding now enter
the unknown-state fallback instead of throwing, retaining original bytes and
mass with NaN statistics. Inverted bounds remain rejected.
Addressed in
[33c0b75b60](https://github.com/apache/pinot/commit/33c0b75b60f5ed7561cbdc186e241167992c0a2e).
All 491 selected tests, the 322-case compatibility matrix and 74 real-reader
cases pass.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -1118,12 +1151,54 @@ private void bufferIncomingCentroids(double[] means,
double[] weights, int count
private void materializePendingSerializedTDigest() {
if (_pendingSerializedTDigest != null) {
byte[] bytes = _pendingSerializedTDigest;
+ SerializedTDigestMetadata metadata = _pendingSerializedMetadata;
_pendingSerializedTDigest = null;
+ _pendingSerializedMetadata = null;
_pendingSerializedTotalWeight = Double.NaN;
- mergeSerializedTDigest(bytes);
+ SerializedTDigestInput input = getSerializedTDigestInput();
+ input.reset(bytes, metadata);
+ input._retainedBytes = bytes;
+ if (_legacyDegraded) {
+ mergeLegacyDegradedInput(input, true);
+ } else {
+ input.decode();
+ mergeSerializedTDigest(input, false);
+ }
}
}
+ private void mergeLegacyDegradedInput(SerializedTDigestInput input, boolean
retainOriginal) {
+ if (retainOriginal) {
+ _legacySerializedBytes = input.retainBytes();
+ } else {
+ _legacySerializedDirty = true;
+ }
+ _legacyDegraded = true;
Review Comment:
Zero-weight centroids now contribute no distribution and are removed before
merging; the million-row regression retains its median and precise mass in
either merge order. Truly unrecoverable negative/NaN states retain their mass
with unknown statistics, and the centralized transition emits a warning once
per digest so degradation is distinguishable from empty data. Dropping unknown
input mass would manufacture a plausible percentile.
Addressed in
[33c0b75b60](https://github.com/apache/pinot/commit/33c0b75b60f5ed7561cbdc186e241167992c0a2e).
All 491 selected tests, the 322-case compatibility matrix and 74 real-reader
cases pass.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TDigestUtils.java:
##########
@@ -624,19 +462,22 @@ private static double weightedMean(double firstMean,
double firstWeight, double
}
return Double.POSITIVE_INFINITY;
}
- if (Math.copySign(1.0, firstMean) == Math.copySign(1.0, secondMean)) {
- return firstMean + (secondMean - firstMean) * secondWeight / totalWeight;
- }
- return firstMean * (firstWeight / totalWeight) + secondMean *
(secondWeight / totalWeight);
+ double average = Math.copySign(1.0, firstMean) == Math.copySign(1.0,
secondMean)
+ ? firstMean + (secondMean - firstMean) * (secondWeight / totalWeight)
+ : firstMean * (firstWeight / totalWeight) + secondMean * (secondWeight
/ totalWeight);
+ return Math.max(Math.min(firstMean, secondMean), Math.min(average,
Math.max(firstMean, secondMean)));
}
- private static int getLegacyDefaultCentroidCapacity(double compression) {
- return Math.addExact(Math.multiplyExact(2, (int) Math.ceil(compression)),
LEGACY_CAPACITY_PADDING);
+ /// Default centroid capacity allocated by the verbose t-digest 3.2 reader.
+ public static int getLegacyDefaultCentroidCapacity(double compression) {
+ return (int) Math.min(Integer.MAX_VALUE,
Review Comment:
The raw 3.2 capacity helper no longer clamps compression before computing
capacity. Healthy verbose output normalizes a below-ten header to ten before
compatibility decisions; actual 3.2/3.3 readers now gate the compression=5,
count=25 case. Opaque unknown over-capacity state is explicitly documented as a
retained historical exception: reducing it would invent a distribution, so
reverse-upgrade readability is not guaranteed.
Addressed in
[33c0b75b60](https://github.com/apache/pinot/commit/33c0b75b60f5ed7561cbdc186e241167992c0a2e).
All 491 selected tests, the 322-case compatibility matrix and 74 real-reader
cases pass.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/PercentileTDigestValueAggregator.java:
##########
@@ -284,51 +266,81 @@ public void add(double value) {
}
@Override
- public void add(double value, int weight) {
+ public void add(double value, double weight) {
if (Double.isNaN(value)) {
throw new IllegalArgumentException("Cannot add NaN to t-digest");
}
- if (weight <= 0) {
+ if (!(weight > 0.0) || !Double.isFinite(weight)) {
throw new IllegalArgumentException("TDigest weight must be positive: "
+ weight);
}
+ if (hasValidStatistics()) {
+ checkTotalWeight(getTotalWeight() + weight);
+ }
invalidateCaches();
if (value == Double.NEGATIVE_INFINITY) {
- _negativeInfinityWeight = Math.addExact(_negativeInfinityWeight,
weight);
+ _negativeInfinityWeight = _negativeInfinityWeight + weight;
} else if (value == Double.POSITIVE_INFINITY) {
- _positiveInfinityWeight = Math.addExact(_positiveInfinityWeight,
weight);
+ _positiveInfinityWeight = _positiveInfinityWeight + weight;
} else {
_finiteDigest.add(value, weight);
}
}
@Override
public void add(TDigest other) {
+ if (hasValidStatistics() && other.hasValidStatistics()) {
+ checkTotalWeight(getTotalWeight() + other.getTotalWeight());
+ }
if (other instanceof NonFiniteAwareTDigest) {
NonFiniteAwareTDigest wrapped = (NonFiniteAwareTDigest) other;
- long negativeInfinityWeight = wrapped._negativeInfinityWeight;
- long positiveInfinityWeight = wrapped._positiveInfinityWeight;
- // MergingDigest.add(List) copies the implementation's
double-precision weight arrays directly. The generic
- // add(TDigest) path goes through Centroid.count(), which is limited
to an int and truncates large weights.
- wrapped.invalidateCaches();
+ double negativeInfinityWeight = wrapped._negativeInfinityWeight;
+ double positiveInfinityWeight = wrapped._positiveInfinityWeight;
invalidateCaches();
- if (wrapped._finiteDigest.size() > 0L) {
- _finiteDigest.add(List.of(wrapped._finiteDigest));
+ if (!wrapped._finiteDigest.isEmpty()) {
+ _finiteDigest.add(wrapped._finiteDigest);
}
- _negativeInfinityWeight = Math.addExact(_negativeInfinityWeight,
negativeInfinityWeight);
- _positiveInfinityWeight = Math.addExact(_positiveInfinityWeight,
positiveInfinityWeight);
+ _negativeInfinityWeight += negativeInfinityWeight;
+ _positiveInfinityWeight += positiveInfinityWeight;
return;
}
- if (other.size() == 0L) {
+ if (other.isEmpty()) {
return;
}
invalidateCaches();
- if (Double.isFinite(other.getMin()) && Double.isFinite(other.getMax())) {
- _finiteDigest.add(List.of(other));
+ if (!other.hasValidStatistics()
+ || Double.isFinite(other.getMin()) &&
Double.isFinite(other.getMax())) {
+ _finiteDigest.add(other);
return;
}
- for (Centroid centroid : other.centroids()) {
- add(centroid.mean(), centroid.count());
+ // Read precise centroid weights directly; generic sources need no full
digest serialization for their tails.
+ Collection<Centroid> centroids = other.centroids();
+ double[] finiteMeans = new double[centroids.size()];
+ double[] finiteWeights = new double[centroids.size()];
+ int finiteCount = 0;
+ double finiteMin = Double.POSITIVE_INFINITY;
+ double finiteMax = Double.NEGATIVE_INFINITY;
+ for (Centroid centroid : centroids) {
+ double mean = centroid.mean();
+ double weight = centroid.weight();
+ if (mean == Double.NEGATIVE_INFINITY) {
+ _negativeInfinityWeight += weight;
+ } else if (mean == Double.POSITIVE_INFINITY) {
+ _positiveInfinityWeight += weight;
+ } else {
+ finiteMeans[finiteCount] = mean;
+ finiteWeights[finiteCount++] = weight;
+ finiteMin = Math.min(finiteMin, mean);
+ finiteMax = Math.max(finiteMax, mean);
+ }
+ }
+ if (finiteCount > 0) {
+ finiteMin = Double.isFinite(other.getMin()) ? other.getMin() :
finiteMin;
+ finiteMax = Double.isFinite(other.getMax()) ? other.getMax() :
finiteMax;
+ SerializedTDigestInput finiteInput = new SerializedTDigestInput();
+ finiteInput.reset(toVerboseBytes(finiteMin, finiteMax,
other.compression(),
+ finiteMeans, finiteWeights, finiteCount));
Review Comment:
Internally generated generic finite payloads now reset through ByteBuffer
with checkCapacity=false, matching the existing internal decode path. A
regression merges a generic digest with an Infinity tail and more working
centroids than the default capacity, retaining precise mass and extrema.
Addressed in
[33c0b75b60](https://github.com/apache/pinot/commit/33c0b75b60f5ed7561cbdc186e241167992c0a2e).
All 491 selected tests, the 322-case compatibility matrix and 74 real-reader
cases pass.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -240,167 +271,128 @@ public void add(List<? extends TDigest> others) {
}
private void addAccumulator(PercentileTDigestAccumulator other) {
+ _hasFractionalWeights |= other._hasFractionalWeights;
+ if (other == this) {
+ addSerializedTDigest(serialize());
+ return;
+ }
if (other._pendingSerializedTDigest != null) {
- addSerializedTDigest(other._pendingSerializedTDigest);
+ // Materialize before resetting the reusable reader: materialization
uses that same reader.
+ materializePendingSerializedTDigest();
+ SerializedTDigestInput input = getSerializedTDigestInput();
+ input.reset(other._pendingSerializedTDigest,
other._pendingSerializedMetadata);
+ input._retainedBytes = other._pendingSerializedTDigest;
+ addSerializedTDigest(input);
+ return;
+ }
+ if (other._legacyDegraded) {
+ addSerializedTDigest(other.serialize());
return;
}
- other.compress();
+ // Legacy merge inputs use public compression. Apply it to a snapshot so
reading a source leaves its
+ // distribution unchanged, and copy before flushing to preserve the legacy
single-pass raw merge.
+ other = other.publiclyCompressedMergeInput();
materializePendingSerializedTDigest();
+ if (_legacyDegraded) {
+ _legacySerializedDirty = true;
+ _totalWeight += other._totalWeight;
+ _min = Math.min(_min, other._min);
+ _max = Math.max(_max, other._max);
+ return;
+ }
+ checkTotalWeight(_totalWeight + _incomingWeight + _numRawValues +
other._totalWeight);
preserveSerializedCapacity(other._serializedMainCapacity,
other._serializedBufferCapacity);
bufferIncomingCentroids(other._centroidMeans, other._centroidWeights,
other._numCentroids);
_min = Math.min(_min, other._min);
_max = Math.max(_max, other._max);
}
- void addSerializedTDigest(byte[] bytes) {
- if (!_hasSerializedInput && _numCentroids == 0 && _totalWeight == 0.0 &&
_numRawValues == 0
- && _numIncomingCentroids == 0) {
- byte[] retainedBytes = bytes.clone();
- _pendingSerializedTotalWeight = getSerializedTotalWeight(retainedBytes);
- _pendingSerializedTDigest = retainedBytes;
- _hasSerializedInput = true;
- return;
- }
+ private PercentileTDigestAccumulator publiclyCompressedMergeInput() {
+ if (_publiclyCompressed && _numRawValues == 0 && _numIncomingCentroids ==
0) {
+ return this;
+ }
+ PercentileTDigestAccumulator copy =
+ new PercentileTDigestAccumulator(_compression, false, false,
_useTwoLevelRawCompression);
+ copy._numCentroids = _numCentroids;
+ copy._numRawValues = _numRawValues;
+ copy._numIncomingCentroids = _numIncomingCentroids;
+ copy._centroidMeans = _numCentroids == 0 ? null :
Arrays.copyOf(_centroidMeans, _numCentroids);
+ copy._centroidWeights = _numCentroids == 0 ? null :
Arrays.copyOf(_centroidWeights, _numCentroids);
+ copy._rawValues = _numRawValues == 0 ? null : Arrays.copyOf(_rawValues,
_numRawValues);
+ copy._incomingMeans = _numIncomingCentroids == 0 ? null :
Arrays.copyOf(_incomingMeans, _numIncomingCentroids);
+ copy._incomingWeights = _numIncomingCentroids == 0 ? null :
Arrays.copyOf(_incomingWeights, _numIncomingCentroids);
+ copy._serializedMainCapacity = _serializedMainCapacity;
+ copy._serializedBufferCapacity = _serializedBufferCapacity;
+ copy._mergeCount = _mergeCount;
+ copy._publiclyCompressed = _publiclyCompressed;
+ copy._incomingCentroidsSorted = _incomingCentroidsSorted;
+ copy._hasSerializedInput = _hasSerializedInput;
+ copy._hasFractionalWeights = _hasFractionalWeights;
+ copy._totalWeight = _totalWeight;
+ copy._incomingWeight = _incomingWeight;
+ copy._min = _min;
+ copy._max = _max;
+ copy.compress();
+ return copy;
+ }
+
+ public void addSerializedTDigest(byte[] bytes) {
materializePendingSerializedTDigest();
+ SerializedTDigestInput input = getSerializedTDigestInput();
+ input.reset(bytes);
+ addSerializedTDigest(input);
+ }
+
+ private SerializedTDigestInput getSerializedTDigestInput() {
if (_serializedTDigestInput == null) {
_serializedTDigestInput = new SerializedTDigestInput();
}
- _serializedTDigestInput.reset(bytes);
- _serializedTDigestInput.decode();
- mergeSerializedTDigest(_serializedTDigestInput);
- _hasSerializedInput = true;
+ return _serializedTDigestInput;
}
- void addSerializedTDigest(SerializedTDigestInput input) {
- if (!_hasSerializedInput && _numCentroids == 0 && _totalWeight == 0.0 &&
_numRawValues == 0
- && _numIncomingCentroids == 0) {
- _pendingSerializedTDigest = input.retainBytes();
- _pendingSerializedTotalWeight = Double.NaN;
- _hasSerializedInput = true;
- return;
+ public void addSerializedTDigest(SerializedTDigestInput input) {
+ if (prepareSerializedTDigest(input)) {
+ mergeSerializedTDigest(input, false);
}
- materializePendingSerializedTDigest();
- input.decode();
- mergeSerializedTDigest(input);
- _hasSerializedInput = true;
}
- void addSerializedTDigestDirect(SerializedTDigestInput input) {
- if (!_hasSerializedInput && _numCentroids == 0 && _totalWeight == 0.0 &&
_numRawValues == 0
- && _numIncomingCentroids == 0) {
+ public void addSerializedTDigestDirect(SerializedTDigestInput input) {
+ if (prepareSerializedTDigest(input)) {
+ mergeSerializedTDigest(input, true);
+ }
+ }
+
+ private boolean prepareSerializedTDigest(SerializedTDigestInput input) {
+ if (!_hasSerializedInput && hasNoInputs()) {
+ input.inspectMetadata();
+ _hasFractionalWeights |= input._metadata.fractionalWeights();
_pendingSerializedTDigest = input.retainBytes();
- _pendingSerializedTotalWeight = Double.NaN;
+ _pendingSerializedTotalWeight = input._totalWeight;
+ _pendingSerializedMetadata = input._metadata;
+ _legacyDegraded = input._metadata.needsLegacyFallback();
_hasSerializedInput = true;
- return;
+ return false;
}
materializePendingSerializedTDigest();
input.decode();
- preserveSerializedCapacity(input._mainCapacity, input._bufferCapacity);
- boolean initialize = _numCentroids == 0 && _totalWeight == 0.0 &&
_numRawValues == 0
- && _numIncomingCentroids == 0;
- if (initialize) {
- if (input._totalWeight != 0.0) {
- ensureCentroidCapacity(input._numCentroids);
- System.arraycopy(input._means, 0, _centroidMeans, 0,
input._numCentroids);
- System.arraycopy(input._weights, 0, _centroidWeights, 0,
input._numCentroids);
- _numCentroids = input._numCentroids;
- _totalWeight = input._totalWeight;
- _min = input._min;
- _max = input._max;
- }
- } else {
- flush();
- boolean runBackwards = (_mergeCount++ & 1) != 0;
- if (mergeSorted(input._means, input._weights, input._numCentroids,
input._totalWeight,
- getIncrementalCompression(),
- runBackwards)) {
- _min = Math.min(_min, input._min);
- _max = Math.max(_max, input._max);
- }
+ _hasFractionalWeights |= input._metadata.fractionalWeights();
+ if (_legacyDegraded || input._metadata.needsLegacyFallback()) {
+ mergeLegacyDegradedInput(input, false);
+ return false;
}
+ checkTotalWeight(_totalWeight + _incomingWeight + _numRawValues +
input._totalWeight);
Review Comment:
Confirmed this is intentional for valid distributions: combined mass beyond
the finite double range fails explicitly. The SPI contract and PR migration
note now describe that failure, while finite totals above Long.MAX_VALUE
continue to work with the saturating size() view.
Addressed in
[33c0b75b60](https://github.com/apache/pinot/commit/33c0b75b60f5ed7561cbdc186e241167992c0a2e).
All 491 selected tests, the 322-case compatibility matrix and 74 real-reader
cases pass.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
