xiangfu0 commented on code in PR #19795:
URL: https://github.com/apache/pinot/pull/19795#discussion_r4224159699
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -756,12 +841,14 @@ private byte[] toVerboseBytes() {
private byte[] toCapacityPreservingBytes() {
checkCapacityPreservingCentroidCount();
- int mainCapacity = Math.min(Short.MAX_VALUE, Math.max(_numCentroids,
_serializedMainCapacity));
- long defaultBufferCapacity =
Math.multiplyExact(DEFAULT_MERGE_BUFFER_MULTIPLIER, (long) mainCapacity);
+ int mainCapacity = Math.min(Short.MAX_VALUE,
+ Math.max(_centroidCapacity, Math.max(_numCentroids,
_serializedMainCapacity)));
+ long defaultBufferCapacity =
Math.multiplyExact(TDigestUtils.DEFAULT_MERGE_BUFFER_MULTIPLIER, (long)
mainCapacity);
int bufferCapacity = Math.toIntExact(Math.min(Short.MAX_VALUE,
Math.max(Math.max((long) _serializedBufferCapacity, mainCapacity +
1L), defaultBufferCapacity)));
- ByteBuffer buffer = ByteBuffer.allocate(SMALL_HEADER_SIZE +
SMALL_CENTROID_SIZE * _numCentroids);
- buffer.putInt(SMALL_ENCODING);
+ ByteBuffer buffer = ByteBuffer.allocate(
+ TDigestUtils.SMALL_HEADER_SIZE + TDigestUtils.SMALL_CENTROID_SIZE *
_numCentroids);
+ buffer.putInt(TDigestUtils.SMALL_ENCODING);
buffer.putDouble(_min);
buffer.putDouble(_max);
buffer.putFloat((float) _compression);
Review Comment:
Fixed in
[4818fb8fbc](https://github.com/apache/pinot/commit/4818fb8fbc79d81de17ce0c5bd0070eb97f300df).
asSmallBytes and smallByteSize choose VERBOSE when a finite centroid mean or
weight would overflow float; ordinary compact rounding remains supported.
Regressions cover huge means and weights, consistent reported/written sizes,
and numeric round trips.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -1243,19 +1292,12 @@ private void ensureRawBuffer() {
}
private static int getRawBufferSize(long roundedCompression) {
- return (int) Math.min(MAX_RAW_BUFFER_SIZE, Math.max(MIN_RAW_BUFFER_SIZE,
2L * roundedCompression));
- }
-
- private static int getCentroidCapacity(double compression) {
- double normalizedCompression = Math.max(MIN_COMPRESSION, compression);
- int padding = normalizedCompression < 30.0 ?
LOW_COMPRESSION_CAPACITY_PADDING
- : DEFAULT_CENTROID_CAPACITY_PADDING;
- return Math.toIntExact((long) Math.ceil(2.0 * normalizedCompression +
padding));
+ return (int) Math.min(MAX_RAW_BUFFER_SIZE, Math.max(MIN_RAW_BUFFER_SIZE,
2.0 * roundedCompression));
}
private static int getPendingCentroidCapacity(int centroidCapacity) {
- return Math.min(MAX_RAW_BUFFER_SIZE,
- Math.max(MIN_RAW_BUFFER_SIZE,
Math.multiplyExact(DEFAULT_MERGE_BUFFER_MULTIPLIER, centroidCapacity)));
+ return (int) Math.min(MAX_RAW_BUFFER_SIZE,
+ Math.max(MIN_RAW_BUFFER_SIZE, (long)
TDigestUtils.DEFAULT_MERGE_BUFFER_MULTIPLIER * centroidCapacity));
}
private int getPendingInputLimit() {
Review Comment:
Fixed in
[4818fb8fbc](https://github.com/apache/pinot/commit/4818fb8fbc79d81de17ce0c5bd0070eb97f300df).
At the capped high-compression capacity, pending input uses an independent
bounded 10,000-element batch instead of subtracting retained centroids until
the limit becomes one. Regression coverage verifies multiple weighted and raw
batches do not trigger one full merge per input; the ordinary-compression
compatibility corpus remains unchanged.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -912,13 +928,8 @@ private void flushIncoming(double compression) {
double[] incomingMeans = _incomingMeans;
double[] incomingWeights = _incomingWeights;
if (!_incomingCentroidsSorted) {
- ensureIncomingSortCapacity(incomingCount);
- Sort.stableSort(_incomingOrder, _incomingMeans, incomingCount);
- for (int i = 0; i < incomingCount; i++) {
- int sourceIndex = _incomingOrder[i];
- _sortedIncomingMeans[i] = _incomingMeans[sourceIndex];
- _sortedIncomingWeights[i] = _incomingWeights[sourceIndex];
- }
+ ensureSortedIncomingCapacity(incomingCount);
+ stableSortIncoming(incomingCount);
incomingMeans = _sortedIncomingMeans;
incomingWeights = _sortedIncomingWeights;
}
Review Comment:
Fixed in
[4818fb8fbc](https://github.com/apache/pinot/commit/4818fb8fbc79d81de17ce0c5bd0070eb97f300df).
flushIncoming clears pending count/weight only after mergeSorted succeeds. A
near-Double.MAX_VALUE regression forces merge-time overflow from floating-point
reassociation and verifies buffered mass remains available after the exception.
--
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]
