xiangfu0 commented on code in PR #19795:
URL: https://github.com/apache/pinot/pull/19795#discussion_r4221353869
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -240,167 +271,89 @@ 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();
Review Comment:
Both paths now reuse the receiver's serialized reader. Pending receiver
state is materialized before resetting that reader, avoiding input aliasing; a
three-pending-singleton regression verifies mass, extrema and median.
Addressed in
[6246a60b4b](https://github.com/apache/pinot/commit/6246a60b4bdbaf59deb89986d2644a41178e1a89).
All 483 targeted tests and both legacy-reader checks pass.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/PercentileTDigestValueAggregator.java:
##########
@@ -152,128 +150,111 @@ private static NonFiniteAwareTDigest
asNonFiniteAware(TDigest value) {
}
private void updateMaxByteSize(TDigest value) {
- long defaultCapacity = getDefaultCentroidCapacity(value.compression());
- long maxCentroids = Math.min(value.size(), defaultCapacity);
+ long defaultCapacity =
TDigestUtils.getDefaultCentroidCapacity(value.compression());
+ long maxCentroids = (long) Math.min(Math.ceil(value.getTotalWeight()),
defaultCapacity);
+ if (value instanceof NonFiniteAwareTDigest) {
+ TDigest finite = ((NonFiniteAwareTDigest) value)._finiteDigest;
+ if (finite instanceof PercentileTDigestAccumulator) {
+ PercentileTDigestAccumulator accumulator =
(PercentileTDigestAccumulator) finite;
+ NonFiniteAwareTDigest wrapped = (NonFiniteAwareTDigest) value;
+ boolean fractionalMass = accumulator.hasFractionalWeights()
+ || wrapped._negativeInfinityWeight !=
Math.rint(wrapped._negativeInfinityWeight)
+ || wrapped._positiveInfinityWeight !=
Math.rint(wrapped._positiveInfinityWeight);
+ long bufferedBound = accumulator.getCentroidCountUpperBound() + 6L;
Review Comment:
Documented the conservative fractional/degraded buffer headroom and why it
can exceed the eventual compatibility-reduced bytes. This preserves the
allocation bound without adding a flush on each raw update.
Addressed in
[6246a60b4b](https://github.com/apache/pinot/commit/6246a60b4bdbaf59deb89986d2644a41178e1a89).
All 483 targeted tests and both legacy-reader checks pass.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -433,20 +405,39 @@ public void compress() {
@Override
public long size() {
- if (_pendingSerializedTDigest != null) {
- if (Double.isNaN(_pendingSerializedTotalWeight)) {
- _pendingSerializedTotalWeight =
getSerializedTotalWeight(_pendingSerializedTDigest);
- }
- return (long) _pendingSerializedTotalWeight;
- }
- return (long) (_totalWeight + _incomingWeight + _numRawValues);
+ return (long) getTotalWeight();
+ }
+
+ @Override
+ public boolean hasValidStatistics() {
+ return !_legacyDegraded;
+ }
+
+ @Override
+ public double getTotalWeight() {
+ return _pendingSerializedTDigest != null ? _pendingSerializedTotalWeight
+ : _totalWeight + _incomingWeight + _numRawValues;
+ }
+
+ /// Bounds output centroids from the actual buffered inputs without flushing
or using their potentially huge mass.
+ public long getCentroidCountUpperBound() {
Review Comment:
The accumulator bound now reserves four boundary splits and documents both
normalization phases. This makes the method itself an output bound, independent
of the wrapper's extra Infinity headroom.
Addressed in
[6246a60b4b](https://github.com/apache/pinot/commit/6246a60b4bdbaf59deb89986d2644a41178e1a89).
All 483 targeted tests and both legacy-reader checks pass.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -741,8 +752,9 @@ private void recompressCentroids(double compression) {
}
private byte[] toVerboseBytes() {
- ByteBuffer buffer = ByteBuffer.allocate(VERBOSE_HEADER_SIZE +
VERBOSE_CENTROID_SIZE * _numCentroids);
- buffer.putInt(VERBOSE_ENCODING);
+ ByteBuffer buffer = ByteBuffer.allocate(
+ TDigestUtils.VERBOSE_HEADER_SIZE + TDigestUtils.VERBOSE_CENTROID_SIZE
* _numCentroids);
Review Comment:
Verbose allocation now uses Math.multiplyExact and Math.addExact, matching
the checked small-encoding size calculation.
Addressed in
[6246a60b4b](https://github.com/apache/pinot/commit/6246a60b4bdbaf59deb89986d2644a41178e1a89).
All 483 targeted tests and both legacy-reader checks 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]
