xiangfu0 commented on code in PR #19795:
URL: https://github.com/apache/pinot/pull/19795#discussion_r4218023744
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/PercentileTDigestValueAggregator.java:
##########
@@ -527,33 +508,21 @@ private byte[] getSerializedBytes(ByteBuffer
scratchBuffer) {
}
ByteBuffer finite = ByteBuffer.wrap(finiteBytes);
- int encoding = finite.getInt();
- finite.position(finite.position() + 2 * Double.BYTES);
- int finiteCentroidCount;
- boolean finiteVerbose;
- if (encoding == VERBOSE_ENCODING) {
- finite.getDouble();
- finiteCentroidCount = finite.getInt();
- finiteVerbose = true;
- } else if (encoding == SMALL_ENCODING) {
- finite.getFloat();
- finite.getShort();
- finite.getShort();
- finiteCentroidCount = finite.getShort();
- finiteVerbose = false;
- } else {
- throw new IllegalStateException("Invalid format for serialized
histogram");
- }
+ SerializedTDigestMetadata metadata =
TDigestUtils.readSerializedHeader(finite, false);
Review Comment:
[code-review] The re-encode loop below (`verbose.putDouble(...
finite.getFloat())`) transcribes SMALL-encoded finite float means into the
verbose payload without clamping them to the extrema, while the header writes
`getMin()`/`getMax()` (−∞/+∞ once infinity mass exists). `fromBytes` and
`SerializedTDigestInput.decode()` both clamp small means to `[min, max]`
because float rounding can land just outside the double extrema — but here a
mean equal to `(double)(float) max > max` is emitted into a VERBOSE payload,
and `inspectSerialized`'s float-rounding allowance applies only to SMALL
encoding, so the final `makeLegacyCompatible` throws `TDigest centroid mean is
outside its extrema`.
Reproduced trigger: a stored pre-aggregated digest in small encoding whose
max (e.g. 0.3) is not float-representable enters via
`deserializeAggregatedValue` (kept as pending bytes, re-emitted verbatim by the
finite digest), then the same star-tree/realtime group receives a −Infinity raw
value; `serializeAggregatedValue` → `getSerializedBytes` throws and the
star-tree build / merge-rollup fails. Symmetric case: +Infinity mass with a
non-float-exact min. Clamp the re-widened means like `decode()` does.
##########
pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/PercentileRawTDigestAggregationFunctionTest.java:
##########
@@ -155,17 +153,32 @@ static byte[] createSmallUnitCentroidDigest(int
numCentroids, double compression
return buffer.array();
}
- private static void assertLegacyCompatibleShape(byte[] bytes, double
compression) {
+ static void assertLegacyCompatibleShape(byte[] bytes, double compression) {
Review Comment:
[code-review] With the tdunning dependency gone, the wire-compat tests lost
their real legacy-reader oracle: cross-version readability is now asserted only
via Pinot's own lenient decoder (`TDigestUtils.deserialize`, which repairs
boundary centroids and accepts counts up to `max(default, legacy)` capacity)
plus this structural shape check (header/count/capacity arithmetic only). A
regression in `makeLegacyCompatible` that emits in-capacity verbose bytes whose
first/last centroids have weight > 1 would pass CI but trip t-digest 3.3's
`assert weight[0] == 1` (or corrupt its interpolation) on an old server during
a rolling upgrade — the exact contract this PR is preserving. Consider keeping
`com.tdunning:t-digest` as a test-scoped dependency so
`MergingDigest.fromBytes` + a subsequent merge still gate the emitted bytes.
##########
pinot-segment-local/src/test/java/org/apache/pinot/segment/local/aggregator/PercentileTDigestValueAggregatorTest.java:
##########
@@ -46,9 +52,10 @@ public void testRawValuesTrackMaxVerboseSize(int
compression, int numValues,
for (int i = 1; i < numValues; i++) {
aggregator.applyRawValue(digest, i);
}
- assertEquals(aggregator.getMaxAggregatedValueByteSize(),
expectedMaxByteSize);
+ int registeredBound = aggregator.getMaxAggregatedValueByteSize();
+ assertTrue(registeredBound <= expectedMaxByteSize);
Review Comment:
[code-review] This assertion (and the similar one at the `6_592` check
below) was weakened from equality to `<=`, so the test no longer verifies that
the registered max-byte-size actually *expands* when a higher-compression
pre-aggregated digest enters the aggregator. `assertTrue(registeredBound <=
expectedMaxByteSize)` passes even if `updateMaxByteSize` ignored the input's
larger compression entirely, and the companion `serialized.length <=
registeredBound` check can't catch that because the test digest holds a single
value (~tens of bytes). An under-registration regression here can overflow
star-tree var-byte writers sized from `getMaxAggregatedValueByteSize()`.
Suggest also asserting a lower bound (e.g. the bound computed for the larger
compression) or serializing a digest large enough to be discriminating.
##########
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:
[code-review] (perf nit) This path allocates a fresh
`SerializedTDigestInput` on every pending-state merge (and `add(TDigest)` does
the same for the invalid-statistics path), while the class already maintains
the reusable `_serializedTDigestInput` field for exactly this purpose (used by
`addSerializedTDigest(byte[])` and `materializePendingSerializedTDigest`).
`merge(TDigest, TDigest)` runs per group per segment during combine and per
group during broker reduce, so a high-cardinality group-by allocates millions
of short-lived inputs the reused field would amortize. Reuse is safe here
because `addSerializedTDigest` retains only the bytes, never the input object.
##########
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:
[code-review] (sizing nit) For fractional-mass (or invalid-statistics)
inputs, `bufferedBound` is not capped by the default centroid capacity, and
`getCentroidCountUpperBound()` counts un-flushed buffered inputs — up to the
pending-buffer limit (~1050 at compression 100, 10000 max). `_maxByteSize` can
therefore be registered as ~17KB–160KB while actual serialized payloads stay
~3.4KB (`makeLegacyCompatible` reduces to legacy capacity). This only wastes
forward-index chunk headroom (no corruption — `serializeAggregatedValue`
records actual lengths), and the mass-based bound is genuinely invalid for
fractional weights, but it may be worth flushing before computing the bound (or
documenting the inflation) since the estimate depends on call-time buffer fill.
##########
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:
[code-review] (doc nit) "Bounds output centroids" can be exceeded by up to
2: `serialize()` normalizes boundaries twice (once inside the merge via
`normalizeBoundaryCentroids()` on existing centroids before merging, +≤2
inputs; once again before `toVerboseBytes`, +≤2), so output can reach
`_numCentroids + _numRawValues + _numIncomingCentroids + 4`. Example at
compression 100: centroids `[[email protected], [email protected]]` plus buffered `[email protected], [email protected]` →
bound 6, serialized 7 (both-end variants reach 8). The sole current caller's `+
6L` provably absorbs it (infinity centroids are ≤4 when finite values exist),
but any future caller sizing a buffer from this "upper bound" under-allocates.
Suggest `+ 4L` here or correcting the doc.
##########
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:
[code-review] (hardening nit) This size expression uses unchecked int
arithmetic, unlike `smallByteSize()`'s `Math.addExact`/`multiplyExact`: with
query-controlled compression (`PERCENTILETDIGEST(col, p, compression)` has no
upper bound — `validateCompression` only requires positive finite) and ≥ ~134M
centroids, `VERBOSE_CENTROID_SIZE * _numCentroids` wraps and
`ByteBuffer.allocate` fails with a confusing `capacity < 0` instead of a clear
error. Needs a multi-GB heap to reach, so purely latent — but a one-line
`multiplyExact` makes the failure mode match the small-encoding path.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -1041,22 +986,40 @@ 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];
}
}
+ private void stableSortIncoming(int count) {
Review Comment:
[code-review] (perf nit) This replaces tdunning's primitive stable sort with
`it.unimi.dsi.fastutil.Arrays.mergeSort` driven by two capturing lambdas: every
comparison takes two levels of indirection
(`_incomingMeans[_incomingOrder[i]]`) through an interface call, plus the
identity-init loop and a gather pass. `flushIncoming` runs once per
pending-buffer fill (up to ~1050 elements at compression 100, 10000 max) on
every weighted/serialized merge path. A hand-rolled primitive stable merge of
the mean/weight pairs directly into the existing
`_sortedIncomingMeans`/`_sortedIncomingWeights` scratch arrays would drop the
`_incomingOrder` indirection and lambda dispatch entirely.
--
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]
