xiangfu0 commented on code in PR #19795:
URL: https://github.com/apache/pinot/pull/19795#discussion_r4217135379
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TDigestUtils.java:
##########
@@ -207,331 +180,155 @@ 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;
- }
+ /// Trailing bytes remain available to the caller for intermediate values
containing several objects.
+ public static double validateSerialized(ByteBuffer input) {
+ return validateSerialized(input, true);
+ }
- 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);
- }
+ private static double validateSerialized(ByteBuffer input, boolean
checkCapacity) {
+ return inspectSerialized(input, checkCapacity).totalWeight();
+ }
- 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;
+ /// Validated metadata for one legacy byte payload. The centroid bytes
remain owned by the caller.
+ ///
+ /// 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) {
}
- /// Deserializes a digest and advances `input` past its encoded bytes.
- public static TDigest deserialize(ByteBuffer input) {
- ByteBuffer encoded = input.slice();
- validateEncodedDigest(encoded.duplicate());
+ /// Inspects one payload without mutating `input`, scanning the centroid
values once.
+ public static SerializedTDigestMetadata inspectSerialized(ByteBuffer input) {
+ return inspectSerialized(input, true);
+ }
+ /// Inspects a payload, optionally enforcing the legacy decoder's declared
centroid capacity.
+ public static SerializedTDigestMetadata inspectSerialized(ByteBuffer input,
boolean checkCapacity) {
+ 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");
+ }
double min = encoded.getDouble();
double max = encoded.getDouble();
double compression;
int mainCapacity;
- int bufferCapacity;
+ int bufferCapacity = 0;
int centroidCount;
int centroidSize;
if (encoding == VERBOSE_ENCODING) {
compression = encoded.getDouble();
+ validateCompression(compression);
+ mainCapacity = Math.max(getDefaultCentroidCapacity(compression),
getLegacyDefaultCentroidCapacity(compression));
centroidCount = encoded.getInt();
- mainCapacity = getDefaultCentroidCapacity(compression);
- bufferCapacity = 5 * mainCapacity;
centroidSize = VERBOSE_CENTROID_SIZE;
- } else if (encoding == SMALL_ENCODING) {
+ } else {
compression = encoded.getFloat();
+ validateCompression(compression);
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 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;
- }
-
- 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;
- }
-
- 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();
- }
-
- private static double readWeight(ByteBuffer encoded, int offset, int
centroidSize) {
- return centroidSize == VERBOSE_CENTROID_SIZE ? encoded.getDouble(offset) :
encoded.getFloat(offset);
- }
-
- private static void validateEncodedDigest(ByteBuffer input) {
- int encoding = input.getInt();
- if (encoding != VERBOSE_ENCODING && encoding != SMALL_ENCODING) {
- throw new IllegalStateException("Invalid format for serialized
histogram");
- }
- input.getDouble();
- input.getDouble();
- double compression;
- int mainCapacity;
- int centroidCount;
- int centroidSize;
- if (encoding == VERBOSE_ENCODING) {
- compression = input.getDouble();
- mainCapacity = getDefaultCentroidCapacity(compression);
- centroidCount = input.getInt();
- centroidSize = VERBOSE_CENTROID_SIZE;
- } else {
- compression = input.getFloat();
- mainCapacity = input.getShort();
- int bufferCapacity = input.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);
+ if (Double.isNaN(min) || Double.isNaN(max) || (centroidCount > 0 && min >
max)) {
+ throw new IllegalArgumentException("Invalid TDigest extrema: " + min +
", " + max);
}
- }
-
- private static void checkSmallArrayCapacity(int capacity) {
- if (capacity < -1) {
- throw new NegativeArraySizeException(Integer.toString(capacity));
- }
- }
-
- 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;
+ int centroidOffset = encoded.position();
+ double totalWeight = 0.0;
+ double previousMean = Double.NEGATIVE_INFINITY;
+ boolean hasNonFiniteMeans = false;
+ boolean needsLegacyFallback = false;
+ boolean unorderedMeans = false;
+ boolean fractionalWeights = false;
+ boolean withinBounds = true;
+ 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)) {
+ // Compact means are floats but extrema are doubles. Permit only the
rounding of an actual endpoint,
+ // rather than admitting arbitrary out-of-range means that boundary
normalization would rewrite.
+ boolean lowerRounded = encoding == SMALL_ENCODING && mean == (double)
(float) min;
+ boolean upperRounded = encoding == SMALL_ENCODING && mean == (double)
(float) max;
+ 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;
- }
-
- 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:
Verbose bytes now admit the exact float-rounded endpoints produced by old
compact-to-verbose rewrites, and one-ULP endpoint drift from old arithmetic.
Decoding restores those means to the double extrema. Larger inconsistent bounds
are still rejected. Regression tests cover both historical cases and genuine
corruption.
Addressed in
[1d9c042a3d](https://github.com/apache/pinot/commit/1d9c042a3d66a4462d7fe991b41f496f4d3dcb7f).
All 473 targeted tests and 322 legacy reader cases pass.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TDigestUtils.java:
##########
@@ -52,31 +53,16 @@ public final class TDigestUtils {
private TDigestUtils() {
}
- /// Creates a merging digest with Pinot's accuracy-preserving scale function.
+ /// Creates a digest using Pinot's accuracy-preserving K1 implementation.
public static TDigest createMergingDigest(double compression) {
- return configureScaleFunction(new MergingDigest(compression));
+ return PercentileTDigestAccumulator.forLegacyAggregation(compression);
}
- /// Creates a merging digest with the temporary-buffer footprint used by
t-digest 3.2.
- ///
- /// Segment aggregation can keep many digests live at once. T-digest 3.3
more than doubled the default temporary
- /// buffer to support two-level compression, adding about 11 KiB to every
compression-100 digest. This factory keeps
- /// the 3.3 main-buffer capacity and Pinot's K1 policy, but restores the
smaller temporary buffer for that
- /// high-cardinality use case.
- public static TDigest createMergingDigestWithLegacyBuffer(double
compression) {
- return createMergingDigestWithLegacyBuffer(compression, 0);
- }
-
- /// Creates a reduced-buffer merging digest whose main array can hold at
least `minimumMainCapacity` centroids.
- public static TDigest createMergingDigestWithLegacyBuffer(double
compression, int minimumMainCapacity) {
- if (minimumMainCapacity < 0) {
- throw new IllegalArgumentException("Minimum main capacity must not be
negative: " + minimumMainCapacity);
+ /// Rejects non-positive or non-finite compression while retaining
historical positive finite settings.
+ public static void validateCompression(double compression) {
+ if (!(compression > 0.0) || !Double.isFinite(compression)) {
Review Comment:
Restored legacy clamping of finite compression settings below ten, including
zero and negative values, for factories, ingestion and serialized headers.
Non-finite compression remains rejected. Tests exercise raw updates and round
trips with those settings.
Addressed in
[1d9c042a3d](https://github.com/apache/pinot/commit/1d9c042a3d66a4462d7fe991b41f496f4d3dcb7f).
All 473 targeted tests and 322 legacy reader cases pass.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -240,61 +263,86 @@ 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);
+ SerializedTDigestInput input = new SerializedTDigestInput();
+ input.reset(other._pendingSerializedTDigest,
other._pendingSerializedMetadata);
+ input._retainedBytes = other._pendingSerializedTDigest;
+ addSerializedTDigest(input);
+ return;
+ }
+ if (other._legacyDegraded) {
+ addSerializedTDigest(other.serialize());
return;
}
other.compress();
materializePendingSerializedTDigest();
+ checkTotalWeight(_totalWeight + _incomingWeight + _numRawValues +
other._totalWeight);
+ if (_legacyDegraded) {
+ _totalWeight += other._totalWeight;
+ _min = Math.min(_min, other._min);
+ _max = Math.max(_max, other._max);
+ return;
+ }
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;
- }
+ public void addSerializedTDigest(byte[] bytes) {
materializePendingSerializedTDigest();
if (_serializedTDigestInput == null) {
_serializedTDigestInput = new SerializedTDigestInput();
}
_serializedTDigestInput.reset(bytes);
- _serializedTDigestInput.decode();
- mergeSerializedTDigest(_serializedTDigestInput);
- _hasSerializedInput = true;
+ addSerializedTDigest(_serializedTDigestInput);
}
- void addSerializedTDigest(SerializedTDigestInput input) {
+ public void addSerializedTDigest(SerializedTDigestInput input) {
+ _hasFractionalWeights |= input._metadata.fractionalWeights();
if (!_hasSerializedInput && _numCentroids == 0 && _totalWeight == 0.0 &&
_numRawValues == 0
&& _numIncomingCentroids == 0) {
_pendingSerializedTDigest = input.retainBytes();
- _pendingSerializedTotalWeight = Double.NaN;
+ _pendingSerializedTotalWeight = input._totalWeight;
Review Comment:
Legacy-degraded inputs bypass the normal nonnegative-total gate consistently
in both pending and materialized merges. Negative or zero-net-weight
distributions retain their NaN marker in either order, rather than choosing an
error based on plan shape. Tests cover buffered/direct merges, copies and raw
additions.
Addressed in
[1d9c042a3d](https://github.com/apache/pinot/commit/1d9c042a3d66a4462d7fe991b41f496f4d3dcb7f).
All 473 targeted tests and 322 legacy reader cases pass.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/PercentileTDigestValueAggregator.java:
##########
@@ -291,45 +311,50 @@ public void add(double value, int weight) {
if (weight <= 0) {
throw new IllegalArgumentException("TDigest weight must be positive: "
+ weight);
}
+ 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) {
+ 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.
+ double negativeInfinityWeight = wrapped._negativeInfinityWeight;
+ double positiveInfinityWeight = wrapped._positiveInfinityWeight;
+ // Merging shared state preserves double-precision weights without
narrowing through Centroid.count().
wrapped.invalidateCaches();
invalidateCaches();
- if (wrapped._finiteDigest.size() > 0L) {
+ if (wrapped._finiteDigest.getTotalWeight() > 0.0) {
Review Comment:
Added explicit statistical validity and empty-state semantics. The wrapper
merges and copies a degraded state even when its net mass is zero or negative,
and propagates NaN through raw and Star-tree paths. A zero-net-weight poisoned
digest is no longer replaced by an empty valid digest.
Addressed in
[1d9c042a3d](https://github.com/apache/pinot/commit/1d9c042a3d66a4462d7fe991b41f496f4d3dcb7f).
All 473 targeted tests and 322 legacy 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]
