xiangfu0 commented on code in PR #19795:
URL: https://github.com/apache/pinot/pull/19795#discussion_r4215198569
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -1144,7 +1113,7 @@ private void normalizeBoundaryCentroids() {
if (_numCentroids == 1) {
double weight = _centroidWeights[0];
- if (weight <= 1.0) {
+ if (weight < 2.0) {
Review Comment:
**[code-review] The new `weight < 2.0` early-return (also line 1166) leaves
a single centroid with fractional weight in (1.0, 2.0) unrepaired, creating a
rolling-upgrade asymmetry and a repeated useless lossy re-encode.**
Reproduced with the real t-digest 3.3 jar: bytes {count=1, weight=1.5}
round-trip through deserialize/serialize emitting boundary weight 1.5; an old
3.3 server running with -ea that merges this intermediate dies with
AssertionError (unit-boundary invariant), and an old server's <=1.0 repair of
the same bytes manufactures a negative-weight (-0.5) centroid that the new
validateSerialized then hard-rejects in the reverse direction. Additionally
hasWeightedBoundaryCentroids keeps reporting such pending state as
repair-needed, so every serialize() does a lossy materialize+recompress yet
still emits a weighted boundary. The old <=1.0 behavior was itself broken
(negative residual), so this is an incomplete fix: fractional sub-2 boundary
weights should be either normalized differently or explicitly
rejected/documented.
##########
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:
**[code-review] New hard cap rejecting total weight >= 2^63
(checkTotalWeight, and validateSerialized) turns previously-working huge
aggregations and stored digests into failures.**
Pre-PR, weights were plain doubles and only size()'s long cast saturated;
merging rollup-of-rollup digests (or NonFiniteAwareTDigest states whose
infinity weights are longs up to Long.MAX_VALUE each) whose cumulative weight
crosses 2^63 completed with correct double-precision quantiles. Post-PR the
same broker reduce / segment build aborts mid-aggregation with
IllegalArgumentException('Invalid TDigest total weight') - the checks at the
add/merge entry points fire before the merge, so a single over-limit group
fails the whole query, and validateSerialized rejects such stored digests
outright. Edge-case trigger, but a hard regression with no graceful degradation
when hit.
##########
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:
**[code-review] add(TDigest) now unconditionally round-trips non-accumulator
sources through TDigestUtils.serialize -> makeLegacyCompatible, which silently
pre-merges (reduceVerboseCentroids) any source with more centroids than the
legacy 2*ceil(compression)+10 capacity before the accumulator ever sees them.**
The base add(TDigest) iterated other.centroids() and buffered every centroid
exactly (serialized fallback only for size > Integer.MAX_VALUE). Post-PR, a
TDigest implementation whose centroidCount exceeds the legacy capacity - e.g.
the benchmark's FixedTDigest FIXED_VERBOSE sources with thousands of singleton
centroids, or an out-of-tree implementation - has interior centroids merged
away during ingestion even though the accumulator itself can represent the
larger count via capacity preservation: silent resolution loss in quantiles,
plus a full verbose encode/validate/decode allocation per add(TDigest) call.
##########
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:
**[code-review] deserializeFiniteWithLegacyBuffer now ends with
digest.compress(), which unconditionally runs a lossy
recompressCentroids(_compression) merge pass over the freshly-decoded stored
centroids on every star-tree/rollup read; the base implementation buffered the
centroids without an immediate destructive pass, and no comment explains why
the eager compress is needed.**
PercentileTDigestValueAggregator.deserializeAggregatedValue calls this once
per pre-aggregated cell during star-tree builds, realtime ingestion
aggregation, and merge-rollup; each read-modify-write generation applies one
extra merge pass versus the base (alternating merge direction plus boundary
normalization can coalesce previously-separate centroids), mildly degrading
stored percentile accuracy per rollup generation. Error stays within the K1
bound so it does not compound unboundedly, but if the compress() is not
required for correctness it should be removed; if it is (e.g. memory bounding),
a comment should say so.
##########
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:
**[code-review] The new `weight <= 0` throw in add(double, int) combines
with centroids()' (int) cast of fractional weights (0 < w < 1 -> count 0) so
centroid-replay paths now throw where they previously degraded silently.**
Boundary normalization of a legacy digest whose boundary weight is in (1,2)
leaves a residual centroid with weight in (0,1); centroids() casts it to count
0, and NonFiniteAwareTDigest.add(TDigest)'s non-wrapper branch for digests with
infinite extrema (PercentileTDigestValueAggregator.java ~329,
add(centroid.mean(), centroid.count())) then hits
IllegalArgumentException('TDigest weight must be positive: 0'), failing the
star-tree build or query. Pre-PR the zero-weight add was absorbed harmlessly.
Either skip count-0 centroids at the replay sites or saturate the cast at 1.
##########
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:
**[code-review] validateSerialized is a full O(centroids) walk and is now
layered on every decode path without threading its result through, so hot paths
scan the same bytes 2-6 times and copy them twice.**
Per input row, SerializedTDigestInput.reset() full-walks the payload and
decode() re-walks it re-summing the total weight reset already computed (2x per
row in aggregate/aggregateGroupBySV over pre-aggregated BYTES columns). Per
wire digest, TDigestUtils.deserialize validates, copies the bytes, then
addSerializedTDigest validates again and clones again, and first use validates
a third time via reset (3 walks + 2 copies); readCompression full-walks the
payload to read an 8-byte field; serialize()'s pending path calls
getSerializedTotalWeight and discards the result. Cheaper: validate once where
untrusted bytes enter and pass {compression, totalWeight} through (the fields
_pendingSerializedTotalWeight and SerializedTDigestInput._totalWeight already
exist for this).
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -1041,22 +976,56 @@ private void ensureIncomingCapacity(int capacity) {
private void ensureIncomingSortCapacity(int capacity) {
ensureSortedIncomingCapacity(capacity);
if (_incomingOrder == null || capacity > _incomingOrder.length) {
- int newCapacity = _incomingOrder == null ? Math.max(capacity,
_centroidCapacity)
+ int newCapacity = _incomingOrder == null
+ ? Math.max(capacity, Math.min(_centroidCapacity,
_pendingCentroidCapacity))
: Math.max(capacity, Math.multiplyExact(_incomingOrder.length, 2));
_incomingOrder = new int[newCapacity];
+ _incomingOrderScratch = new int[newCapacity];
}
}
+ // Stable, primitive merge sort bounds the work to O(n log n), including
repeated and infinite means.
+ private void stableSortIncoming(int count) {
Review Comment:
**[code-review] stableSortIncoming hand-rolls a ~30-line bottom-up stable
merge sort with a permanently retained second scratch array
(_incomingOrderScratch), when fastutil - already used for exactly this pattern
in this module (RangeIndexCreator.java:257) - provides a stable
it.unimi.dsi.fastutil.Arrays.mergeSort(from, to, comparator, swapper).**
Group-by holds one accumulator per group (commonly 100K+ per
segment-thread); any group that receives unsorted incoming centroids allocates
both int arrays (up to 10K ints = 40KB each) and never releases them, doubling
sort-index memory for the accumulator's lifetime versus the replaced
Sort.stableSort which used no retained scratch. Either use the existing
fastutil stable sort, or allocate the merge-sort scratch transiently inside
stableSortIncoming (it is only live during the sort).
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TDigestUtils.java:
##########
@@ -48,35 +45,34 @@ public final class TDigestUtils {
private static final int SMALL_HEADER_SIZE = 30;
private static final int SMALL_CENTROID_SIZE = 8;
private static final int DEFAULT_MERGE_BUFFER_MULTIPLIER = 5;
+ private static final double MAX_COMPRESSION = 1_000_000.0;
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.
+ /// Creates Pinot's buffered digest for segment aggregation.
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.
+ /// Creates a digest whose dynamically grown arrays can hold the requested
centroid capacity.
public static TDigest createMergingDigestWithLegacyBuffer(double
compression, int minimumMainCapacity) {
Review Comment:
**[code-review] createMergingDigestWithLegacyBuffer(compression,
minimumMainCapacity) now validates and then ignores minimumMainCapacity
entirely - all three factory variants are aliases of createMergingDigest, but
the name and doc still promise capacity pre-sizing.**
The sole production caller, NonFiniteAwareTDigest.copy()
(PercentileTDigestValueAggregator.java:205), computes
Math.addExact(centroidCount(), 2) - which forces a flush/compress of the
source's buffered state - purely to produce a discarded argument, and future
readers will assume pre-allocation happens. Delete the two-arg overload and the
centroidCount() computation at the call site, and collapse the aliases.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/TDigest.java:
##########
@@ -0,0 +1,70 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.segment.local.customobject;
+
+import java.nio.ByteBuffer;
+import java.util.Collection;
+import java.util.List;
+
+/// Pinot's mutable t-digest contract using the existing K1 aggregation
behavior and legacy byte encodings.
+///
+/// Implementations are not safe for concurrent mutation; callers must
serialize access to each aggregation state.
+public abstract class TDigest {
+ public static TDigest createMergingDigest(double compression) {
Review Comment:
**[code-review] The abstract contract class carries a static factory that
hard-codes its concrete subclass
(PercentileTDigestAccumulator.forLegacyAggregation), duplicating
TDigestUtils.createMergingDigest exactly - two identical public factories
introduced in the same PR, with callers split arbitrarily.**
Base-to-derived circular reference defeats the contract abstraction, and a
future behavior change applied to one factory silently diverges query-time
digests from test/benchmark digests (production aggregation functions use the
TDigestUtils one; tests and pinot-perf use this one). Keep one factory and
delegate or migrate the other's call sites.
##########
pom.xml:
##########
@@ -253,7 +253,6 @@
<hadoop-shaded-protobuf_3_25.version>1.5.0</hadoop-shaded-protobuf_3_25.version>
<clearspring-stream-lib.version>2.9.8</clearspring-stream-lib.version>
<datasketches-java.version>9.0.0</datasketches-java.version>
- <t-digest.version>3.3</t-digest.version>
Review Comment:
**[code-review] Removing com.tdunning from the runtime classpath while
changing public API signatures (TDigestUtils factories/serde,
ObjectSerDeUtils.TDIGEST_SER_DE, ValueAggregator type parameters) from
com.tdunning.math.stats.TDigest to
org.apache.pinot.segment.local.customobject.TDigest is a binary-incompatible
break for out-of-tree plugins.**
A third-party plugin AggregationFunction compiled against a prior release
that calls TDigestUtils.createMergingDigest(...) or consumes TDIGEST_SER_DE
throws NoSuchMethodError / NoClassDefFoundError:
com/tdunning/math/stats/TDigest at runtime after a server upgrade. Inherent to
the PR's goal, but it needs the backward-incompatible label and a release note
stating that plugins must migrate to the new type or bundle t-digest themselves.
##########
pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/PercentileTDigestAggregationFunctionTest.java:
##########
@@ -1005,59 +1006,29 @@ public void testReducerMergeStructureAndAccuracy(int
fanIn, int compression, Red
sourceValues[valueIndex] = value;
}
sources[digestIndex] = createReducerSource(sourceValues, compression,
reducerInput);
- referenceSources[digestIndex] =
createReferenceReducerSource(sourceValues, compression, reducerInput);
}
PercentileTDigestAggregationFunction function =
new PercentileTDigestAggregationFunction(EXPRESSION, 75.0,
compression, false);
TDigest result = TDigest.createMergingDigest(compression);
- TDigest referenceResult = TDigest.createMergingDigest(compression);
- AssertionError referenceFailure = null;
for (int sourceIndex : mergeOrder.sourceIndexes(fanIn)) {
result = function.merge(result, sources[sourceIndex]);
- if (referenceFailure == null) {
- try {
- if (referenceResult.size() == 0L) {
- referenceResult = referenceSources[sourceIndex];
- } else {
- referenceResult.add(referenceSources[sourceIndex]);
- }
- } catch (AssertionError e) {
- // t-digest 3.3 can violate its own singleton-endpoint assertion
when many duplicate-valued digests merge.
- referenceFailure = e;
- }
- }
}
Arrays.sort(rawValues);
String caseDescription =
fanIn + "/" + compression + "/" + distribution + "/" + mergeOrder +
"/" + reducerInput;
- // Interpolation between large equal-valued centroid runs can land inside
the empirical CDF jump. Keep a fixed
- // envelope for that discontinuous distribution instead of requiring every
merge order to match the reference's
- // particular centroid partition.
- double frozenRankError = compression == 20 || distribution ==
ReducerDistribution.DUPLICATE_HEAVY ? 0.06 : 0.02;
- double[] referenceRankErrors = null;
- if (referenceFailure == null) {
- try {
- referenceRankErrors = getReducerRankErrors(referenceResult, rawValues,
caseDescription + "/reference");
- } catch (AssertionError e) {
- referenceFailure = e;
- }
- }
- double[] maxRankErrors = new double[6];
- if (referenceFailure == null) {
- for (int i = 0; i < referenceRankErrors.length; i++) {
- maxRankErrors[i] = Math.max(frozenRankError, referenceRankErrors[i] +
1e-12);
- }
- } else {
- Arrays.fill(maxRankErrors, compression == 20 ? 0.25 : compression == 100
? 0.06 : 0.02);
- }
- assertValidReducerResult(result, rawValues, caseDescription,
maxRankErrors);
+ // Validate against the empirical CDF. The former 3.3 reference allowed
0.25 after duplicate-boundary
+ // assertions at compression 20; preserve that historical envelope for
this discontinuous distribution.
+ double maxRankError = compression == 20 && distribution ==
ReducerDistribution.DUPLICATE_HEAVY
Review Comment:
**[code-review] Deleting the t-digest 3.3 reference implementation removed
the reducer test's only independent accuracy oracle, replaced by frozen
self-referential envelopes (rank error up to 0.25 for compression-20
duplicate-heavy); TDigestUtilsTest.java:271 similarly weakened the
capacity-reduction guard from an exact count (211) to a near-vacuous `<=` bound
against the input's own count.**
A future merge-path regression that degrades mid-quantile rank error from
~0.06 to 0.24 on skewed data, or one that stops reducing oversized payloads to
the legacy capacity entirely, passes CI - the accuracy contract of Pinot's now
self-owned t-digest is checked only against constants derived from the old
library's worst observed behavior. Keep an independent oracle: exact order
statistics with K1-theory per-quantile bounds, or a golden corpus of serialized
reference digests from the 3.3 build.
--
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]
