xiangfu0 commented on code in PR #19795:
URL: https://github.com/apache/pinot/pull/19795#discussion_r4223009563
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -638,10 +686,17 @@ public int byteSize() {
return serialize().length;
}
+ /// Returns the small-encoding size for valid statistics. Historical
degraded state retains its original encoding
+ /// until modified, then uses verbose bytes so poisoned values and precise
weights survive without narrowing.
@Override
public int smallByteSize() {
Review Comment:
smallByteSize() now calls the writer's same signed-short centroid-count
guard, so size and write reject oversized compact output consistently. The
32768-centroid regression checks both failures. The degraded original/verbose
encoding exception is documented on both SPI methods.
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:
##########
@@ -1029,34 +1032,64 @@ private void finishMerge(int numOutputCentroids, double
totalWeight, double comp
private void ensureIncomingCapacity(int capacity) {
if (_incomingMeans == null || capacity > _incomingMeans.length) {
- int newCapacity = _incomingMeans == null ? Math.max(capacity,
_centroidCapacity)
- : Math.max(capacity, Math.min(_pendingCentroidCapacity,
- Math.multiplyExact(_incomingMeans.length,
DEFAULT_MERGE_BUFFER_MULTIPLIER)));
+ int newCapacity = _incomingMeans == null
+ ? Math.max(capacity, Math.min(_centroidCapacity,
_pendingCentroidCapacity))
+ : Math.max(capacity, (int) Math.min(_pendingCentroidCapacity,
+ (long) _incomingMeans.length *
TDigestUtils.DEFAULT_MERGE_BUFFER_MULTIPLIER));
_incomingMeans = _incomingMeans == null ? new double[newCapacity] :
Arrays.copyOf(_incomingMeans, newCapacity);
_incomingWeights =
_incomingWeights == null ? new double[newCapacity] :
Arrays.copyOf(_incomingWeights, newCapacity);
}
}
- private void ensureIncomingSortCapacity(int capacity) {
- ensureSortedIncomingCapacity(capacity);
- if (_incomingOrder == null || capacity > _incomingOrder.length) {
- int newCapacity = _incomingOrder == null ? Math.max(capacity,
_centroidCapacity)
- : Math.max(capacity, Math.multiplyExact(_incomingOrder.length, 2));
- _incomingOrder = new int[newCapacity];
+ private void stableSortIncoming(int count) {
Review Comment:
This was already fixed in the previous published revision:
stableSortIncoming directly merges primitive mean/weight pairs into the
existing scratch arrays. There is no order array, capturing comparator/swapper,
identity initialization or gather pass on that hot path.
Verified at
[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:
##########
@@ -696,34 +744,60 @@ public double getMin() {
@Override
public double getMax() {
if (_pendingSerializedTDigest != null) {
- return
ByteBuffer.wrap(_pendingSerializedTDigest).getDouble(Integer.BYTES +
Double.BYTES);
+ return _pendingSerializedMetadata.max();
}
flush();
return _max;
}
- byte[] serialize() {
+ public byte[] serialize() {
+ if (_legacyDegraded) {
+ if (_pendingSerializedTDigest != null) {
+ return _pendingSerializedTDigest.clone();
+ }
+ if (!_legacySerializedDirty) {
+ return _legacySerializedBytes.clone();
+ }
+ // A merge with a historical invalid distribution remains unknown. Keep
that explicit in the legacy
+ // encoding instead of passing its invalid centroids into the normal
sorted K1 merge.
+ return toLegacyDegradedBytes();
+ }
if (_pendingSerializedTDigest != null) {
- getSerializedTotalWeight(_pendingSerializedTDigest);
- if (ByteBuffer.wrap(_pendingSerializedTDigest).getInt() ==
VERBOSE_ENCODING) {
- byte[] serialized =
TDigestUtils.makeLegacyCompatible(_pendingSerializedTDigest);
- return serialized == _pendingSerializedTDigest ? serialized.clone() :
serialized;
+ if (hasWeightedBoundaryCentroids(_pendingSerializedTDigest,
_pendingSerializedMetadata.centroidCount())) {
+ materializePendingSerializedTDigest();
+ } else {
+ if (ByteBuffer.wrap(_pendingSerializedTDigest).getInt() ==
TDigestUtils.VERBOSE_ENCODING) {
+ byte[] serialized =
TDigestUtils.makeLegacyCompatible(_pendingSerializedTDigest,
_pendingSerializedMetadata);
+ return serialized == _pendingSerializedTDigest ? serialized.clone()
: serialized;
+ }
+ return _pendingSerializedTDigest.clone();
}
- return _pendingSerializedTDigest.clone();
}
compress();
+ normalizeBoundaryCentroids();
return TDigestUtils.makeLegacyCompatible(toVerboseBytes());
}
- TDigest toTDigest() {
- if (_pendingSerializedTDigest != null) {
- return TDigestUtils.deserialize(_pendingSerializedTDigest);
+ private static boolean hasWeightedBoundaryCentroids(byte[] bytes, int count)
{
Review Comment:
Boundary weights and zero-weight presence are now computed during the
existing codec scan and cached in SerializedTDigestMetadata. Pending
serialization consumes those flags; the second byte-level boundary parser is
removed.
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:
##########
@@ -152,128 +150,112 @@ private static NonFiniteAwareTDigest
asNonFiniteAware(TDigest value) {
}
private void updateMaxByteSize(TDigest value) {
Review Comment:
Added a nonabstract maxSerializedByteSize() contract with
implementation-owned native/wrapper bounds. The aggregator delegates to it and
registers the bound before emission; native state does not flush. Infinity
split counts come from the same helper used by the writer. Regressions cover
verbose expansion, unobserved state, and degraded compact mutations without
under-allocation.
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]
