xiangfu0 commented on code in PR #19795:
URL: https://github.com/apache/pinot/pull/19795#discussion_r4215690038
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -1287,12 +1256,10 @@ private void ensureOutputCapacity(int capacity) {
}
}
- private static double getTotalWeight(double[] weights, int count) {
- double totalWeight = 0.0;
- for (int i = 0; i < count; i++) {
- totalWeight += weights[i];
+ private static void checkTotalWeight(double totalWeight) {
Review Comment:
Removed the 2^63 rejection while retaining non-finite-total checks.
Quantile/CDF paths use precise double mass, and size() retains the legacy
saturating long view. Infinity tails also preserve double weights in bounded
wire output. Tests cover totals above Long.MAX_VALUE and mixed finite/infinite
masses.
Addressed in
[1fc37ba77c](https://github.com/apache/pinot/commit/1fc37ba77c631c574df59a6e211bb978954fb2b2).
All 463 targeted tests and 322 legacy reader cases pass.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -216,20 +219,9 @@ public void add(TDigest other) {
addAccumulator((PercentileTDigestAccumulator) other);
return;
}
- if (other instanceof MergingDigest && other.size() > Integer.MAX_VALUE) {
- // Centroid.count() exposes an int even though MergingDigest stores
double weights internally. Preserve large
- // pre-aggregated weights by ingesting the serialized double-precision
representation instead.
- addSerializedTDigest(TDigestUtils.serialize(other));
- return;
- }
-
- Collection<Centroid> centroids = other.centroids();
- materializePendingSerializedTDigest();
- for (Centroid centroid : centroids) {
- bufferIncomingCentroid(centroid.mean(), centroid.count());
- }
- _min = Math.min(_min, other.getMin());
- _max = Math.max(_max, other.getMax());
+ // The centroid API exposes int counts for compatibility. The legacy
verbose representation preserves
+ // fractional and large double weights from other implementations.
+ addSerializedTDigest(TDigestUtils.serialize(other));
Review Comment:
Generic ingestion now reads byteSize/asBytes directly, without calling the
compatibility reducer first. Its internal validated path permits the actual
source centroid count, retaining resolution and fractional/large weights until
the normal K1 merge.
Addressed in
[1fc37ba77c](https://github.com/apache/pinot/commit/1fc37ba77c631c574df59a6e211bb978954fb2b2).
All 463 targeted tests and 322 legacy reader cases pass.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TDigestUtils.java:
##########
@@ -207,331 +183,144 @@ 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());
-
+ private static double validateSerialized(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 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();
+ int 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);
+ 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;
+ double totalWeight = 0.0;
+ double previousMean = Double.NEGATIVE_INFINITY;
+ 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 (!(weight > 0.0) || !Double.isFinite(weight) || Double.isNaN(mean) ||
mean < previousMean) {
+ throw new IllegalArgumentException("Invalid TDigest centroid: mean=" +
mean + ", 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;
+ totalWeight += weight;
+ if (!Double.isFinite(totalWeight) || totalWeight >= 0x1p63) {
+ 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;
+ previousMean = mean;
}
+ return totalWeight;
+ }
- 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++;
+ /// Deserializes a finite-valued digest into Pinot's buffered implementation.
+ public static TDigest deserializeFiniteWithLegacyBuffer(byte[] bytes) {
+ validateSerialized(bytes);
+ ByteBuffer input = ByteBuffer.wrap(bytes);
+ int encoding = input.getInt();
+ input.position(Integer.BYTES + 2 * Double.BYTES);
+ int count;
+ boolean verbose = encoding == VERBOSE_ENCODING;
+ if (verbose) {
+ input.getDouble();
+ count = input.getInt();
+ } else {
+ input.getFloat();
+ input.getShort();
+ input.getShort();
+ count = input.getShort();
}
- 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++;
+ for (int i = 0; i < count; i++) {
+ if (verbose) {
+ input.getDouble();
+ } else {
+ input.getFloat();
+ }
+ double mean = verbose ? input.getDouble() : input.getFloat();
+ if (!Double.isFinite(mean)) {
+ throw new IllegalArgumentException("Expected a finite TDigest centroid
mean: " + mean);
+ }
}
- return count;
+ TDigest digest = deserialize(bytes);
+ digest.compress();
Review Comment:
Removed eager compression from finite deserialization. Stored centroids
remain pending until a merge or serialization needs compression, avoiding an
extra pass on each rollup generation. Oversized compact inputs retain their
bytes and expand the registered byte bound.
Addressed in
[1fc37ba77c](https://github.com/apache/pinot/commit/1fc37ba77c631c574df59a6e211bb978954fb2b2).
All 463 targeted tests and 322 legacy reader cases pass.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -204,6 +204,9 @@ public void add(double value, int weight) {
if (Double.isNaN(value)) {
throw new IllegalArgumentException("Cannot add NaN to t-digest");
}
+ if (weight <= 0) {
Review Comment:
Replay uses precision-bearing wire weights instead of Centroid.count().
Fractional finite and Infinity masses survive merges, copies and round trips
without zero-count throws. Byte reservation also accounts for fractional
centroids before serialization, tested with 0.5+0.5 and 0.5+1.5.
Addressed in
[1fc37ba77c](https://github.com/apache/pinot/commit/1fc37ba77c631c574df59a6e211bb978954fb2b2).
All 463 targeted tests and 322 legacy reader cases pass.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -1452,7 +1387,8 @@ static final class SerializedTDigestInput {
private double _totalWeight;
private boolean _decoded;
- void reset(byte[] bytes) {
+ public void reset(byte[] bytes) {
+ _totalWeight = TDigestUtils.validateSerialized(bytes);
Review Comment:
Ingestion now validates once and retains its metadata with one immutable
snapshot. Lazy decode reuses the validated total, pending serialization reuses
metadata, and header/compression reads no longer scan centroids. The finite
wrapper reuses the validated input too.
Addressed in
[1fc37ba77c](https://github.com/apache/pinot/commit/1fc37ba77c631c574df59a6e211bb978954fb2b2).
All 463 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]
