xiangfu0 commented on code in PR #19795:
URL: https://github.com/apache/pinot/pull/19795#discussion_r4224691985
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -566,22 +613,16 @@ public double quantile(double quantile) {
for (int i = 0; i < lastIndex; i++) {
double halfWeight = (_centroidWeights[i] + _centroidWeights[i + 1]) /
2.0;
if (weightSoFar + halfWeight > index) {
- double leftUnitWeight = 0.0;
- if (_centroidWeights[i] == 1.0) {
- if (index - weightSoFar < 0.5) {
- return _centroidMeans[i];
- }
- leftUnitWeight = 0.5;
+ double leftExcludedWeight = getPointMassHalfWeight(i);
Review Comment:
**Bug (reproduced): sub-unit fractional total mass breaks
`quantile()`/`cdf()`.** The kernel keeps the legacy unit-mass guards (`if
(index < 1.0) { return _min; }` above, and the `1 - 0.5/_totalWeight` tail
forms in `cdf`), but the new SPI `add(double value, double weight)` explicitly
supports fractional mass. Repro: `forReduction(100); add(1.0, 0.3); add(2.0,
0.3)` (total weight 0.6, two distinct centroids) gives `quantile(0.5) =
quantile(0.99) = quantile(1.0) = 1.0` — p100 returns the minimum instead of the
maximum — and a non-monotone CDF: `cdf(1.0)=0.25, cdf(1.5)=0.5,
cdf(2.0)=0.1667`. Reachable from any stored BYTES payload whose fractional
weights sum below 1 — the exact shape the new tests pin as supported
(`testFractionalMassIsNotEmptyDuringAggregationMergeOrFinalExtraction`).
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -1287,10 +1383,15 @@ 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) {
+ if (!(totalWeight >= 0.0) || !Double.isFinite(totalWeight)) {
+ throw new IllegalArgumentException("Invalid TDigest total weight: " +
totalWeight);
+ }
+ }
+
+ private static double checkedLegacyTotalWeight(double totalWeight) {
Review Comment:
**Bug (reproduced): the degraded-path weight check permits negative
totals.** `checkedLegacyTotalWeight` only rejects non-finite values, while
every valid-statistics path goes through `checkTotalWeight`, which also rejects
negatives. A historical payload with negative centroid weights (the very
artifact this fallback exists for) merged via `mergeLegacyDegradedInput`
silently subtracts mass: a healthy 100-value accumulator merged with a verbose
payload containing one weight `-200` centroid yields `getTotalWeight() == -100`
and `size() == -100`, and the negative weight round-trips through
serialize/deserialize into downstream merges (merging that state into a
1000-value digest yields 900). Suggest rejecting (or clamping and logging)
negative legacy totals here as well.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -1118,12 +1207,65 @@ private void bufferIncomingCentroids(double[] means,
double[] weights, int count
private void materializePendingSerializedTDigest() {
if (_pendingSerializedTDigest != null) {
byte[] bytes = _pendingSerializedTDigest;
+ SerializedTDigestMetadata metadata = _pendingSerializedMetadata;
_pendingSerializedTDigest = null;
- _pendingSerializedTotalWeight = Double.NaN;
- mergeSerializedTDigest(bytes);
+ _pendingSerializedMetadata = null;
+ SerializedTDigestInput input = getSerializedTDigestInput();
+ input.reset(bytes, metadata);
+ input._retainedBytes = bytes;
+ if (_legacyDegraded) {
+ mergeLegacyDegradedInput(input, true);
+ } else {
+ input.decode();
+ mergeSerializedTDigest(input, false);
+ }
+ }
+ }
+
+ private void mergeLegacyDegradedInput(SerializedTDigestInput input, boolean
retainOriginal) {
+ double totalWeight = checkedLegacyTotalWeight(_totalWeight +
_incomingWeight + _numRawValues + input._totalWeight);
+ if (retainOriginal) {
+ _legacySerializedBytes = input.retainBytes();
+ } else {
+ _legacySerializedDirty = true;
+ }
+ markLegacyDegraded(input);
+ for (int i = 0; i < _numRawValues; i++) {
+ _min = Math.min(_min, _rawValues[i]);
+ _max = Math.max(_max, _rawValues[i]);
+ }
+ _totalWeight = totalWeight;
+ _hasFractionalWeights |= input._metadata.fractionalWeights();
+ _incomingWeight = 0.0;
+ _numIncomingCentroids = 0;
+ _numRawValues = 0;
+ _min = Math.min(_min, input._min);
+ _max = Math.max(_max, input._max);
+ _hasSerializedInput = true;
+ }
+
+ private void markLegacyDegraded(SerializedTDigestInput input) {
+ if (!_legacyDegraded) {
+ LOGGER.warn("Retaining historical TDigest numerical corruption: input
centroids={}, input weight={}, "
+ + "accumulated weight={}. Percentiles for this aggregation state
are NaN.",
+ input._numCentroids, input._totalWeight, _totalWeight +
_incomingWeight + _numRawValues);
+ _legacyDegraded = true;
}
}
+ private byte[] toLegacyDegradedBytes() {
Review Comment:
**Rolling-upgrade hazard: `toLegacyDegradedBytes` mints a wire shape
pre-upgrade readers cannot handle.** Once a degraded digest is mutated,
`serialize()` emits a synthetic verbose payload with one centroid
`{weight=totalWeight, mean=NaN}`. An old (pre-PR) node receiving it as an
intermediate result or stored value either throws — old
`NonFiniteAwareTDigest.fromBytes`: `"Cannot deserialize a TDigest with a NaN
centroid mean"`; old re-serialize validation: `"Invalid TDigest centroid"` — or
boundary-repairs the NaN into a tdunning digest returning garbage. Notably, the
finite out-of-bounds-mean corruption class (beyond the permitted endpoint
roundings) worked end-to-end on an old cluster (no bounds check existed), so
during a rolling upgrade a previously-working query can start failing on
not-yet-upgraded peers. The class docs only warn about reverse upgrades of
retained payloads, not this newly minted format.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -1194,7 +1292,7 @@ private static int normalizeBoundaries(double[] means,
double[] weights, int cou
}
if (count == 1) {
double weight = weights[0];
- if (weight <= 1.0) {
+ if (weight < 2.0) {
Review Comment:
**Rolling-upgrade hazard (reproduced): a `(1,2)`-weight singleton is
serialized unsplit and old readers mint a negative-weight centroid.** The guard
changed from the legacy `weight <= 1.0` to `weight < 2.0`, so `add(42.0, 1.5);
serialize()` emits a single centroid with weight 1.5 (verified). A pre-PR
reader's boundary repair then computes `weights[1] = 1.5 - 2.0 = -0.5`
(verified: `w0=1.0@min, w1=-0.5, w2=1.0@max`) and feeds it to
`MergingDigest.fromBytes` with no validation — persistent quantile corruption
on the un-upgraded node; raw tdunning 3.3 consumers hit `assert weight[0] == 1`
under `-ea`. Only reachable for fractional-weight SPI users (in-repo writers
stay integral), but the SPI documents fractional mass as supported. Consider
splitting/widening these singletons at serialization time, or materializing
them via the same path used when `weightedBoundaries` is true.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -1118,12 +1207,65 @@ private void bufferIncomingCentroids(double[] means,
double[] weights, int count
private void materializePendingSerializedTDigest() {
if (_pendingSerializedTDigest != null) {
byte[] bytes = _pendingSerializedTDigest;
+ SerializedTDigestMetadata metadata = _pendingSerializedMetadata;
_pendingSerializedTDigest = null;
- _pendingSerializedTotalWeight = Double.NaN;
- mergeSerializedTDigest(bytes);
+ _pendingSerializedMetadata = null;
+ SerializedTDigestInput input = getSerializedTDigestInput();
+ input.reset(bytes, metadata);
+ input._retainedBytes = bytes;
+ if (_legacyDegraded) {
+ mergeLegacyDegradedInput(input, true);
+ } else {
+ input.decode();
+ mergeSerializedTDigest(input, false);
+ }
+ }
+ }
+
+ private void mergeLegacyDegradedInput(SerializedTDigestInput input, boolean
retainOriginal) {
+ double totalWeight = checkedLegacyTotalWeight(_totalWeight +
_incomingWeight + _numRawValues + input._totalWeight);
+ if (retainOriginal) {
+ _legacySerializedBytes = input.retainBytes();
+ } else {
+ _legacySerializedDirty = true;
+ }
+ markLegacyDegraded(input);
+ for (int i = 0; i < _numRawValues; i++) {
+ _min = Math.min(_min, _rawValues[i]);
+ _max = Math.max(_max, _rawValues[i]);
+ }
+ _totalWeight = totalWeight;
+ _hasFractionalWeights |= input._metadata.fractionalWeights();
+ _incomingWeight = 0.0;
+ _numIncomingCentroids = 0;
+ _numRawValues = 0;
+ _min = Math.min(_min, input._min);
+ _max = Math.max(_max, input._max);
+ _hasSerializedInput = true;
+ }
+
+ private void markLegacyDegraded(SerializedTDigestInput input) {
Review Comment:
**Silent NaN propagation is persisted and unpinned.** Intentional per the
class docs, but two consequences are untested: (1) minion merge-rollup over one
historically corrupted digest now absorbs the healthy group's weight into
degraded state and persists the NaN payload into the rolled-up segment — where
the base commit failed the task loudly (`makeLegacyCompatible` threw) — and no
merge-rollup test covers the degraded path; (2) `SerializedTDigest.compareTo`
uses `Double.compare` on `quantile()`, so a degraded digest's NaN now sorts
above every finite value in `PERCENTILERAWTDIGEST ... ORDER BY`, and no test
pins that placement. Worth a merge-rollup degraded-path test and an explicit
decision on NaN ordering.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -240,167 +286,134 @@ public void add(List<? extends TDigest> others) {
}
private void addAccumulator(PercentileTDigestAccumulator other) {
+ if (other == this) {
+ addSerializedTDigest(serialize());
+ return;
+ }
if (other._pendingSerializedTDigest != null) {
- addSerializedTDigest(other._pendingSerializedTDigest);
+ // Materialize before resetting the reusable reader: materialization
uses that same reader.
+ materializePendingSerializedTDigest();
+ SerializedTDigestInput input = getSerializedTDigestInput();
+ input.reset(other._pendingSerializedTDigest,
other._pendingSerializedMetadata);
+ input._retainedBytes = other._pendingSerializedTDigest;
+ addSerializedTDigest(input);
+ return;
+ }
+ if (other._legacyDegraded) {
+ addSerializedTDigest(other.serialize());
return;
}
- other.compress();
+ // Legacy merge inputs use public compression. Apply it to a snapshot so
reading a source leaves its
+ // distribution unchanged, and copy before flushing to preserve the legacy
single-pass raw merge.
+ other = other.publiclyCompressedMergeInput();
materializePendingSerializedTDigest();
+ if (_legacyDegraded) {
+ double totalWeight = checkedLegacyTotalWeight(_totalWeight +
other._totalWeight);
+ _legacySerializedDirty = true;
+ _totalWeight = totalWeight;
+ _hasFractionalWeights |= other._hasFractionalWeights;
+ _min = Math.min(_min, other._min);
+ _max = Math.max(_max, other._max);
+ return;
+ }
+ checkTotalWeight(_totalWeight + _incomingWeight + _numRawValues +
other._totalWeight);
+ _hasFractionalWeights |= other._hasFractionalWeights;
preserveSerializedCapacity(other._serializedMainCapacity,
other._serializedBufferCapacity);
bufferIncomingCentroids(other._centroidMeans, other._centroidWeights,
other._numCentroids);
_min = Math.min(_min, other._min);
_max = Math.max(_max, other._max);
}
- void addSerializedTDigest(byte[] bytes) {
- if (!_hasSerializedInput && _numCentroids == 0 && _totalWeight == 0.0 &&
_numRawValues == 0
- && _numIncomingCentroids == 0) {
- byte[] retainedBytes = bytes.clone();
- _pendingSerializedTotalWeight = getSerializedTotalWeight(retainedBytes);
- _pendingSerializedTDigest = retainedBytes;
- _hasSerializedInput = true;
- return;
- }
+ private PercentileTDigestAccumulator publiclyCompressedMergeInput() {
Review Comment:
**Perf: every accumulator-to-accumulator merge deep-copies the source.**
`addAccumulator` calls `publiclyCompressedMergeInput()`, which copies up to
five arrays (centroids, raw values, incoming buffers) and runs a full K1
recompression on the copy — per segment-level merge in the combine phase —
purely to avoid mutating `other`, which the combine operator discards
immediately afterwards. tdunning's `MergingDigest.add(other)` compressed
`other` in place with zero allocation. Consider consuming-merge semantics
(flush/compress `other` in place) or merging directly from its flushed arrays.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -638,10 +702,44 @@ public int byteSize() {
return serialize().length;
Review Comment:
**Perf: `byteSize()` performs a complete serialization just to return a
length, so the documented `byteSize()`-then-`asBytes()` pattern serializes
every digest twice** (each pass allocating the full verbose array plus a
possible compact re-encode; old `MergingDigest.byteSize()` was O(1)). Consider
caching the bytes from `byteSize()` for the subsequent `asBytes()`, or
computing the size arithmetically after a single `compress()`.
--
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]
