xiangfu0 commented on code in PR #19795:
URL: https://github.com/apache/pinot/pull/19795#discussion_r4253120456
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/tdigest/TDigestAccumulator.java:
##########
@@ -186,243 +330,292 @@ public void add(double value) {
if (Double.isNaN(value)) {
throw new IllegalArgumentException("Cannot add NaN to t-digest");
}
- materializePendingSerializedTDigest();
+ requireMutable();
+ prepareForMutation();
ensureRawBuffer();
- if (_numRawValues == _rawValues.length
- || _numRawValues + _numIncomingCentroids == getPendingInputLimit()) {
- flush();
- }
+ flushIfBuffersFull(true);
+ _originalFractionalBytes = null;
_rawValues[_numRawValues++] = value;
}
@Override
- public void add(double value, int weight) {
+ public void add(double value, double weight) {
if (weight == 1) {
add(value);
return;
}
if (Double.isNaN(value)) {
throw new IllegalArgumentException("Cannot add NaN to t-digest");
}
- materializePendingSerializedTDigest();
+ if (!(weight > 0.0) || !Double.isFinite(weight)) {
+ throw new IllegalArgumentException("TDigest weight must be positive: " +
weight);
+ }
+ requireMutable();
+ prepareForMutation();
bufferIncomingCentroid(value, weight);
+ _originalFractionalBytes = null;
+ _hasFractionalWeights |= weight != Math.rint(weight);
+ recordFreshFractionalMean(value, weight);
_min = Math.min(_min, value);
_max = Math.max(_max, value);
}
@Override
public void add(TDigest other) {
+ if (other.hasValidStatistics() && other.getTotalWeight() == 0.0) {
+ return;
+ }
+ requireMutable();
if (other instanceof TDigestAccumulator) {
addAccumulator((TDigestAccumulator) 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));
+ if (other instanceof NonFiniteAwareTDigest) {
+ TDigestAccumulator finiteDigest = ((NonFiniteAwareTDigest)
other).finiteDigestIfNoTails();
+ if (finiteDigest != null) {
+ addAccumulator(finiteDigest);
+ return;
+ }
+ }
+ if (!other.hasValidStatistics()) {
+ materializePendingSerializedTDigest();
+ SerializedTDigestInput input = getSerializedTDigestInput();
+ input.reset(ByteBuffer.wrap(other.serialize()), false);
+ addSerializedTDigest(input);
return;
}
-
+ double otherWeight = other.getTotalWeight();
+ checkTotalWeight(getTotalWeight() + otherWeight);
Collection<Centroid> centroids = other.centroids();
- materializePendingSerializedTDigest();
+ boolean wrapped = other instanceof NonFiniteAwareTDigest;
+ int capacity = wrapped ? Math.addExact(centroids.size(), 2) :
centroids.size();
+ double[] means = new double[capacity];
+ double[] weights = new double[capacity];
+ int count = 0;
for (Centroid centroid : centroids) {
- bufferIncomingCentroid(centroid.mean(), centroid.count());
+ double weight = centroid.weight();
+ if (weight == 0.0) {
+ continue;
+ }
+ if (!(weight > 0.0) || !Double.isFinite(weight) ||
Double.isNaN(centroid.mean())) {
+ throw new IllegalArgumentException("Invalid TDigest centroid");
+ }
+ means[count] = centroid.mean();
+ weights[count++] = weight;
}
- _min = Math.min(_min, other.getMin());
- _max = Math.max(_max, other.getMax());
- }
-
- @Override
- public void add(List<? extends TDigest> others) {
- for (TDigest other : others) {
- add(other);
+ if (wrapped) {
+ count = normalizeSerializedBoundaries(means, weights, count,
other.getMin(), other.getMax());
}
+ // Validate all generic centroids before changing the target, then use the
same whole-source primitive path.
+ addCentroids(means, weights, count, other.getMin(), other.getMax(), false,
wrapped);
+ inheritHistoricalFractionalBoundaries(other);
}
private void addAccumulator(TDigestAccumulator other) {
+ if (other == this) {
+ compress();
+ prepareForMutation();
+ if (_numCentroids == 0) {
+ return;
+ }
+ // Snapshot only self-merges; fresh fractional mass need not be
representable by a legacy wire payload.
+ addHistoricalCentroids(Arrays.copyOf(_centroidMeans, _numCentroids),
+ Arrays.copyOf(_centroidWeights, _numCentroids), _numCentroids, _min,
_max, false);
+ return;
+ }
+ if (_deferBoundaryNormalization && !other._deferBoundaryNormalization) {
+ other.prepareForMutation();
+ }
+ if (other._originalBoundaryBytes != null) {
+ if (getTotalWeight() != 0.0) {
+ throw unrepresentableBoundaryMutation();
+ }
+ addSerializedTDigest(other._originalBoundaryBytes);
+ return;
+ }
+ if (other._originalFractionalBytes != null && getTotalWeight() == 0.0) {
+ materializePendingSerializedTDigest();
+ SerializedTDigestInput input = getSerializedTDigestInput();
+ input.reset(other._originalFractionalBytes);
+ input._retainedBytes = other._originalFractionalBytes;
+ addSerializedTDigest(input);
+ 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;
}
-
+ // MergingDigest compressed merge sources in place. Aggregation owns these
mutable inputs exclusively.
other.compress();
- materializePendingSerializedTDigest();
- preserveSerializedCapacity(other._serializedMainCapacity,
other._serializedBufferCapacity);
- bufferIncomingCentroids(other._centroidMeans, other._centroidWeights,
other._numCentroids);
- _min = Math.min(_min, other._min);
- _max = Math.max(_max, other._max);
+ other.prepareForMutation();
+ prepareForMutation();
+ checkTotalWeight(_totalWeight + _incomingWeight + _numRawValues +
other._totalWeight);
+ _hasFractionalWeights |= other._hasFractionalWeights;
+ inheritHistoricalFractionalBoundaries(other);
+ bufferIncomingCentroids(other._centroidMeans, other._centroidWeights,
other._numCentroids, other._totalWeight,
+ other._min, other._max);
+ if (other._totalWeight > 0.0) {
+ _originalFractionalBytes = null;
+ }
}
public 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;
+ if (_legacyDegraded || _originalBoundaryBytes != null) {
+ SerializedTDigestInput input = getSerializedTDigestInput();
+ input.reset(bytes);
+ addSerializedTDigest(input);
return;
}
+ requireMutable();
materializePendingSerializedTDigest();
+ SerializedTDigestInput input = getSerializedTDigestInput();
+ input.reset(bytes);
+ addSerializedTDigest(input);
+ }
+
+ private SerializedTDigestInput getSerializedTDigestInput() {
if (_serializedTDigestInput == null) {
_serializedTDigestInput = new SerializedTDigestInput();
}
- _serializedTDigestInput.reset(bytes);
- _serializedTDigestInput.decode();
- mergeSerializedTDigest(_serializedTDigestInput);
- _hasSerializedInput = true;
+ return _serializedTDigestInput;
}
public void addSerializedTDigest(SerializedTDigestInput input) {
- if (!_hasSerializedInput && _numCentroids == 0 && _totalWeight == 0.0 &&
_numRawValues == 0
- && _numIncomingCentroids == 0) {
- _pendingSerializedTDigest = input.retainBytes();
- _pendingSerializedTotalWeight = Double.NaN;
- _hasSerializedInput = true;
+ if (ignoreEmptyInputForDegradedState(input)) {
return;
}
- materializePendingSerializedTDigest();
- input.decode();
- mergeSerializedTDigest(input);
- _hasSerializedInput = true;
+ requireMutable();
+ if (prepareSerializedTDigest(input)) {
+ mergeSerializedTDigest(input, false);
+ }
}
public void addSerializedTDigestDirect(SerializedTDigestInput input) {
- if (!_hasSerializedInput && _numCentroids == 0 && _totalWeight == 0.0 &&
_numRawValues == 0
- && _numIncomingCentroids == 0) {
- _pendingSerializedTDigest = input.retainBytes();
- _pendingSerializedTotalWeight = Double.NaN;
- _hasSerializedInput = true;
+ if (ignoreEmptyInputForDegradedState(input)) {
return;
}
- materializePendingSerializedTDigest();
- input.decode();
- preserveSerializedCapacity(input._mainCapacity, input._bufferCapacity);
- boolean initialize = _numCentroids == 0 && _totalWeight == 0.0 &&
_numRawValues == 0
- && _numIncomingCentroids == 0;
- if (initialize) {
- if (input._totalWeight != 0.0) {
- ensureCentroidCapacity(input._numCentroids);
- System.arraycopy(input._means, 0, _centroidMeans, 0,
input._numCentroids);
- System.arraycopy(input._weights, 0, _centroidWeights, 0,
input._numCentroids);
- _numCentroids = input._numCentroids;
- _totalWeight = input._totalWeight;
- _min = input._min;
- _max = input._max;
- }
- } else {
- flush();
- boolean runBackwards = (_mergeCount++ & 1) != 0;
- if (mergeSorted(input._means, input._weights, input._numCentroids,
input._totalWeight,
- getIncrementalCompression(),
- runBackwards)) {
- _min = Math.min(_min, input._min);
- _max = Math.max(_max, input._max);
- }
+ requireMutable();
+ if (prepareSerializedTDigest(input)) {
+ mergeSerializedTDigest(input, true);
}
- _hasSerializedInput = true;
- }
-
- private void mergeSerializedTDigest(byte[] bytes) {
- ByteBuffer input = ByteBuffer.wrap(bytes);
- int encoding = input.getInt();
- if (encoding != VERBOSE_ENCODING && encoding != SMALL_ENCODING) {
- throw new IllegalStateException("Invalid format for serialized
histogram");
- }
- double min = input.getDouble();
- double max = input.getDouble();
- int numCentroids;
- int mainCapacity = 0;
- int bufferCapacity = 0;
- boolean initialize = _numCentroids == 0 && _totalWeight == 0.0 &&
_numRawValues == 0
- && _numIncomingCentroids == 0;
- double[] incomingMeans;
- double[] incomingWeights;
- if (encoding == VERBOSE_ENCODING) {
- double serializedCompression = input.getDouble();
- numCentroids = input.getInt();
- checkCentroidCount(numCentroids);
- checkCentroidsAvailable(input, numCentroids, VERBOSE_CENTROID_SIZE);
- checkVerboseCentroidCapacity(serializedCompression, numCentroids);
- if (initialize) {
- ensureCentroidCapacity(Math.addExact(numCentroids, 2));
- incomingMeans = _centroidMeans;
- incomingWeights = _centroidWeights;
- } else {
- incomingMeans = new double[Math.addExact(numCentroids, 2)];
- incomingWeights = new double[Math.addExact(numCentroids, 2)];
- }
- for (int i = 0; i < numCentroids; i++) {
- incomingWeights[i] = input.getDouble();
- incomingMeans[i] = input.getDouble();
+ }
+
+ private boolean ignoreEmptyInputForDegradedState(SerializedTDigestInput
input) {
+ if (!_legacyDegraded && _originalBoundaryBytes == null
+ && (_pendingSerializedMetadata == null ||
!_pendingSerializedMetadata.weightedBoundaries())) {
+ return false;
+ }
+ input.inspectMetadata();
+ return !input._metadata._needsLegacyFallback &&
input._metadata._totalWeight == 0.0;
+ }
+
+ private boolean prepareSerializedTDigest(SerializedTDigestInput input) {
+ if (_pendingSerializedTDigest == null && hasNoInputs()) {
+ input.inspectMetadata();
+ _hasFractionalWeights |= input._metadata._fractionalWeights;
+ _pendingSerializedTDigest = input.retainBytes();
+ _pendingSerializedMetadata = input.getMetadata();
+ if (input.recordHistoricalFractionalBoundaries(this)) {
+ // Alias the retained snapshot; historical fractional bytes remain
byte-exact after read-only materialization.
+ _originalFractionalBytes = _pendingSerializedTDigest;
}
- } else {
- double serializedCompression = input.getFloat();
- mainCapacity = input.getShort();
- bufferCapacity = input.getShort();
- checkSmallArrayCapacities(mainCapacity, bufferCapacity);
- mainCapacity = resolveSmallMainCapacity(serializedCompression,
mainCapacity);
- numCentroids = input.getShort();
- checkCentroidCount(numCentroids);
- checkCentroidsAvailable(input, numCentroids, SMALL_CENTROID_SIZE);
- checkSmallCentroidCapacity(mainCapacity, numCentroids);
- if (initialize) {
- ensureCentroidCapacity(Math.addExact(numCentroids, 2));
- incomingMeans = _centroidMeans;
- incomingWeights = _centroidWeights;
- } else {
- incomingMeans = new double[Math.addExact(numCentroids, 2)];
- incomingWeights = new double[Math.addExact(numCentroids, 2)];
+ if (input._metadata._needsLegacyFallback) {
+ markLegacyDegraded();
}
- for (int i = 0; i < numCentroids; i++) {
- incomingWeights[i] = input.getFloat();
- incomingMeans[i] = input.getFloat();
+ return false;
+ }
+ materializePendingSerializedTDigest();
+ input.decode();
+ if (input._metadata._needsLegacyFallback) {
+ if (hasNoInputs()) {
+ _pendingSerializedTDigest = input.retainBytes();
+ _pendingSerializedMetadata = input.getMetadata();
+ markLegacyDegraded();
+ return false;
}
+ throw corruptedMutation();
Review Comment:
**[behavior change on existing data — needs explicit sign-off]** Merging a
historically corrupted serialized digest into any non-empty state now throws
`corruptedMutation()` (both merge orders: this path for corrupt-as-source,
`requireMutable()` for corrupt-as-target). The old tdunning path silently
merged such bytes and returned a (garbage) numeric result.
Reproduced against the real t-digest 3.3 jar: verbose payload `[[email protected],
w=5@NaN, [email protected]]` → tdunning `fromBytes` + `add(42.0)` + `quantile` = NaN,
query completes; new code throws `IllegalArgumentException` on the same bytes.
Blast radius: `PERCENTILETDIGEST` queries grouping a poisoned row with healthy
rows, minion MergeRollup (`PercentileTDigestAggregator`), and star-tree builds
over existing segments all turn from degraded-result into hard failure after
upgrade.
The fail-loud policy is clearly deliberate (the rate-limited WARN says so),
but converting previously-answering production queries over existing stored
data into errors deserves a release note, and possibly an escape hatch (e.g. a
mode that drops degraded mass with a metric instead of throwing).
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/tdigest/TDigestAccumulator.java:
##########
@@ -1287,12 +1682,29 @@ 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];
+ static void checkTotalWeight(double totalWeight) {
+ if (!(totalWeight >= 0.0) || !Double.isFinite(totalWeight)) {
+ throw new IllegalArgumentException("Invalid TDigest total weight: " +
totalWeight);
}
- return totalWeight;
+ }
+
+ void requireMutable() {
+ if (_legacyDegraded) {
+ throw corruptedMutation();
Review Comment:
**[rolling upgrade]** Old tdunning servers can emit overflow-poisoned
intermediates at runtime from *healthy* data: verified empirically that
`MergingDigest(100)` fed 100k finite values of ±1.7e308 produces centroids with
NaN/Infinity means (its `mean += (x−mean)·w/W` overflows when `|x−mean| >
Double.MAX_VALUE`) and serializes them without error. A new broker deserializes
that type-10 intermediate, flags it `needsLegacyFallback`, and this
`requireMutable()` throws when it is merged with any non-empty peer
intermediate.
Result: during a mixed-version rolling upgrade, percentile queries over
wide-range data error for the whole upgrade window (pre-PR brokers returned
NaN/garbage but succeeded) — and permanently for data whose spread exceeds half
the double range, since upgraded servers keep reading tdunning-era stored
digests. Consider degrading to NaN rather than throwing for wire intermediates,
or calling this out in the upgrade notes.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/tdigest/TDigestAccumulator.java:
##########
@@ -536,59 +755,96 @@ public double quantile(double quantile) {
if (Double.isNaN(quantile) || quantile < 0.0 || quantile > 1.0) {
throw new IllegalArgumentException("q should be in [0,1], got " +
quantile);
}
+ if (_legacyDegraded) {
+ return Double.NaN;
+ }
materializePendingSerializedTDigest();
flush();
- if (_numCentroids == 0) {
+ return quantile(quantile, _numCentroids, _totalWeight, _min, _max,
_centroidMeans, _centroidWeights, null);
+ }
+
+ /// Interpolates the same distribution from primitive centroids or validated
encoded centroids, without a copy.
+ static double quantile(double quantile, int count, double totalWeight,
double min, double max, double[] means,
+ double[] weights, SerializedTDigestState encoded) {
+ if (count == 0) {
return Double.NaN;
}
- if (_numCentroids == 1) {
- return _centroidMeans[0];
+ if (count == 1) {
+ return mean(means, encoded, 0);
+ }
+ if (quantile == 0.0) {
+ return min;
+ }
+ if (quantile == 1.0) {
+ return max;
}
- double index = quantile * _totalWeight;
- if (index < 1.0) {
- return _min;
+ double index = quantile * totalWeight;
+ double firstWeight = weight(weights, encoded, 0);
+ if (firstWeight < 1.0 && index < firstWeight / 2.0) {
+ return weightedAverage(min, firstWeight / 2.0 - index, mean(means,
encoded, 0), index);
}
- if (_centroidWeights[0] > 1.0 && index < _centroidWeights[0] / 2.0) {
- return _min + (index - 1.0) / (_centroidWeights[0] / 2.0 - 1.0) *
(_centroidMeans[0] - _min);
+ if (firstWeight >= 1.0 && index < 1.0) {
+ return min;
}
- int lastIndex = _numCentroids - 1;
- if (index > _totalWeight - 1.0) {
- return _max;
+ if (weight(weights, encoded, 0) > 1.0 && index < weight(weights, encoded,
0) / 2.0) {
+ return min + (index - 1.0) / (weight(weights, encoded, 0) / 2.0 - 1.0) *
(mean(means, encoded, 0) - min);
+ }
+ int lastIndex = count - 1;
+ double lastWeight = weight(weights, encoded, lastIndex);
+ if (lastWeight < 1.0 && index >= totalWeight - lastWeight / 2.0) {
+ return weightedAverage(mean(means, encoded, lastIndex), totalWeight -
index, max,
+ index - (totalWeight - lastWeight / 2.0));
+ }
+ if (lastWeight >= 1.0 && index > totalWeight - 1.0) {
+ return max;
}
- if (_centroidWeights[lastIndex] > 1.0
- && _totalWeight - index <= _centroidWeights[lastIndex] / 2.0) {
- return _max - (_totalWeight - index - 1.0) /
(_centroidWeights[lastIndex] / 2.0 - 1.0)
- * (_max - _centroidMeans[lastIndex]);
+ if (weight(weights, encoded, lastIndex) > 1.0
+ && totalWeight - index <= weight(weights, encoded, lastIndex) / 2.0) {
+ return max - (totalWeight - index - 1.0) / (weight(weights, encoded,
lastIndex) / 2.0 - 1.0)
+ * (max - mean(means, encoded, lastIndex));
}
- double weightSoFar = _centroidWeights[0] / 2.0;
+ double weightSoFar = weight(weights, encoded, 0) / 2.0;
for (int i = 0; i < lastIndex; i++) {
- double halfWeight = (_centroidWeights[i] + _centroidWeights[i + 1]) /
2.0;
+ double halfWeight = (weight(weights, encoded, i) + weight(weights,
encoded, 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(means, weights,
encoded, count, i);
+ if (index - weightSoFar < leftExcludedWeight) {
+ return mean(means, encoded, i);
}
- double rightUnitWeight = 0.0;
- if (_centroidWeights[i + 1] == 1.0) {
- if (weightSoFar + halfWeight - index <= 0.5) {
- return _centroidMeans[i + 1];
- }
- rightUnitWeight = 0.5;
+ double rightExcludedWeight = getPointMassHalfWeight(means, weights,
encoded, count, i + 1);
+ if (weightSoFar + halfWeight - index <= rightExcludedWeight) {
+ return mean(means, encoded, i + 1);
}
- double leftWeight = index - weightSoFar - leftUnitWeight;
- double rightWeight = weightSoFar + halfWeight - index -
rightUnitWeight;
- return weightedAverage(_centroidMeans[i], rightWeight,
_centroidMeans[i + 1], leftWeight);
+ double leftWeight = index - weightSoFar - leftExcludedWeight;
+ double rightWeight = weightSoFar + halfWeight - index -
rightExcludedWeight;
+ return weightedAverage(mean(means, encoded, i), rightWeight,
mean(means, encoded, i + 1), leftWeight);
}
weightSoFar += halfWeight;
}
- double weightToMax = index - _totalWeight - _centroidWeights[lastIndex] /
2.0;
- double weightFromLastCentroid = _centroidWeights[lastIndex] / 2.0 -
weightToMax;
- return weightedAverage(_centroidMeans[lastIndex], weightToMax, _max,
weightFromLastCentroid);
+ // Rounded cumulative weights can leave a small gap before the directly
computed last centroid center.
+ double weightToMax = Math.max(0.0, index - (totalWeight - lastWeight /
2.0));
+ double weightFromLastCentroid = lastWeight / 2.0 - weightToMax;
+ return weightedAverage(mean(means, encoded, lastIndex),
weightFromLastCentroid, max, weightToMax);
+ }
+
+ private static double getPointMassHalfWeight(double[] means, double[]
weights, SerializedTDigestState encoded,
Review Comment:
**[result divergence on identical bytes]** `getPointMassHalfWeight` treats
*any* centroid adjacent to an equal-mean neighbor as point mass; tdunning 3.3
excludes interpolation mass only for `weight == 1` singletons. The new writer's
own boundary normalization (splitting endpoint weight `w` into `1 + (w−1)`
clamped at the same mean) manufactures exactly such equal-mean plateaus on
healthy weighted/merged data.
Reproduced with a cross-version harness: compression 100, 500 weighted adds
(seed 45, gaussian·50+100, weights 1–20) → serialize; on the SAME bytes q=0.001
gives −30.491 (tdunning 3.3) vs −31.349 (new) — 2.8% apart; reduce-path q=0.5:
50.4895 vs 50.5083. During a rolling upgrade the same query answers differently
depending on which server reduces, and results shift permanently vs pre-upgrade
— on stored duplicate-heavy data too, not just new ingests. If intentional,
this deviation from the 3.3 kernel deserves a comment plus coverage beyond the
frozen rank-error envelope.
##########
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileTDigestAggregationFunction.java:
##########
@@ -272,10 +273,10 @@ public TDigest extractGroupByResult(GroupByResultHolder
groupByResultHolder, int
@Override
public TDigest merge(TDigest intermediateResult1, TDigest
intermediateResult2) {
- if (intermediateResult1.size() == 0L) {
+ if (intermediateResult1.isEmpty()) {
Review Comment:
**[regression]** Switching `size() == 0L` → `isEmpty()` drops the
empty-discard for degraded zero-weight payloads. Legacy bytes with e.g. NaN min
and zero total weight (0 centroids, or all-zero-weight centroids) were accepted
by the old deserializer (structure-only validation) and silently discarded here
via `size() == 0`; now `isEmpty()` is false (`hasValidStatistics()` is false),
so the code falls through to `add()`, which throws `corruptedMutation()` in
**both** operand orders — failing at server combine or broker reduce a query
that previously succeeded (and arguably correctly, since zero mass was dropped).
For the zero-total-weight case specifically, discarding still seems right:
consider treating zero-mass degraded digests as discardable here (e.g.
`getTotalWeight() == 0.0` regardless of validity).
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/tdigest/TDigestAccumulator.java:
##########
@@ -536,59 +755,96 @@ public double quantile(double quantile) {
if (Double.isNaN(quantile) || quantile < 0.0 || quantile > 1.0) {
throw new IllegalArgumentException("q should be in [0,1], got " +
quantile);
}
+ if (_legacyDegraded) {
+ return Double.NaN;
Review Comment:
**[silent result change + observability]** A single retained degraded digest
now returns NaN from `quantile()`/`cdf()` where every prior release computed a
(garbage but finite) number from the same stored bytes — a silent
dashboard-visible shift on unchanged data (and NULL is not substituted, since
`isEmpty()` is false).
Also, the `markLegacyDegraded` WARN is rate-limited to ~1/min globally
across all accumulators, so the number of affected digests/tables is invisible.
Consider a per-table meter/counter for degraded-digest encounters so operators
can size the impact and find segments to rebuild.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/tdigest/TDigestCodec.java:
##########
@@ -0,0 +1,859 @@
+/**
+ * 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.tdigest;
+
+import java.lang.invoke.MethodHandles;
+import java.lang.invoke.VarHandle;
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.Arrays;
+
+/// Reads and writes the legacy t-digest 3.2/3.3 formats, validating lengths
before allocating centroid arrays.
+/// Fresh output respects both readers' capacities without narrowing doubles.
Historical fractional endpoints and
+/// opaque corrupted payloads retain their original limitations; byte
preservation does not ensure reverse upgrades.
+/// Endpoint repair, capacity fallback and point-mass interpolation can change
quantiles on historical data.
+/// This utility has no shared mutable state; callers must exclusively own
input buffers and parser state.
+public final class TDigestCodec {
+ public static final int VERBOSE_ENCODING = 1;
+ public static final int SMALL_ENCODING = 2;
+ private static final int LEGACY_CAPACITY_PADDING = 10;
+ public static final int VERBOSE_HEADER_SIZE = 32;
+ public static final int VERBOSE_CENTROID_SIZE = 16;
+ public static final int SMALL_HEADER_SIZE = 30;
+ public static final int SMALL_CENTROID_SIZE = 8;
+ public static final int DEFAULT_MERGE_BUFFER_MULTIPLIER = 5;
+
+ private TDigestCodec() {
+ }
+
+ /// Reads the compression from a validated header without decoding centroid
arrays.
+ public static double readCompression(byte[] bytes) {
+ return readSerializedHeader(ByteBuffer.wrap(bytes)).compression();
+ }
+
+ /// Rejects non-finite compression while retaining the legacy clamping of
finite settings below ten.
+ public static void validateCompression(double compression) {
+ if (!Double.isFinite(compression)) {
+ throw new IllegalArgumentException("Invalid TDigest compression: " +
compression);
+ }
+ }
+
+ /// Applies the legacy minimum compression to query, ingestion and
serialized-header settings.
+ public static double normalizeCompression(double compression) {
+ validateCompression(compression);
+ return Math.max(compression, 10.0);
+ }
+
+ /// Writes validated centroid arrays in the standard verbose encoding
without sorting or recompression.
+ /// The caller retains the arrays; only the first `count` entries are
written, with normalized compression.
+ public static byte[] serializeCentroids(double compression, double min,
double max, double[] means,
+ double[] weights, int count) {
+ if (count < 0 || count > means.length || count > weights.length) {
+ throw new IllegalArgumentException("Invalid TDigest centroid count: " +
count);
+ }
+ ByteBuffer encoded = ByteBuffer.allocate(Math.addExact(VERBOSE_HEADER_SIZE,
+ Math.multiplyExact(VERBOSE_CENTROID_SIZE, count)));
+
encoded.putInt(VERBOSE_ENCODING).putDouble(min).putDouble(max).putDouble(normalizeCompression(compression));
+ encoded.putInt(count);
+ for (int i = 0; i < count; i++) {
+ encoded.putDouble(weights[i]).putDouble(means[i]);
+ }
+ return encoded.array();
+ }
+
+ /// Encodes validated, sorted native centroids and selects a
legacy-compatible representation.
+ static byte[] serializeCompatibleCentroids(double compression, double min,
double max, double[] means,
+ double[] weights, int count, boolean inheritedFractionalBoundaries) {
+ if (count < 0 || count > 0 && (count > means.length || count >
weights.length || !(min <= max))) {
+ throw new IllegalArgumentException("Invalid TDigest centroid count or
extrema");
+ }
+ ByteBuffer buffer = ByteBuffer.allocate(
+ Math.addExact(VERBOSE_HEADER_SIZE,
+ Math.multiplyExact(VERBOSE_CENTROID_SIZE, count)));
+ buffer.putInt(VERBOSE_ENCODING);
+ buffer.putDouble(min);
+ buffer.putDouble(max);
+ buffer.putDouble(compression);
+ buffer.putInt(count);
+ double encodedWeight = 0.0;
+ boolean hasNonFiniteMeans = false;
+ boolean fractionalWeights = false;
+ for (int i = 0; i < count; i++) {
+ if (!(weights[i] > 0.0) || !Double.isFinite(weights[i]) || !(means[i] >=
min && means[i] <= max)) {
+ throw new IllegalArgumentException("Invalid TDigest centroid");
+ }
+ buffer.putDouble(weights[i]);
+ buffer.putDouble(means[i]);
+ encodedWeight += weights[i];
+ hasNonFiniteMeans |= !Double.isFinite(means[i]);
+ fractionalWeights |= weights[i] != Math.rint(weights[i]);
+ }
+ if (!Double.isFinite(encodedWeight)) {
+ throw new IllegalArgumentException("TDigest total weight exceeds the
supported range");
+ }
+ boolean weightedBoundaries = count == 1 ? weights[0] >= 2.0
+ : count > 1 && (weights[0] > 1.0 || weights[count - 1] > 1.0);
+ // Collect exact wire metadata during the write rather than decoding and
validating our fresh bytes again.
+ SerializedTDigestMetadata metadata = new
SerializedTDigestMetadata(VERBOSE_ENCODING, min, max,
+ compression, count,
+ VERBOSE_HEADER_SIZE, VERBOSE_CENTROID_SIZE, buffer.capacity(),
encodedWeight,
+ hasNonFiniteMeans, false, false, fractionalWeights,
weightedBoundaries, false, compression, Double.NaN,
+ Double.NaN, Double.NaN);
+ return makeLegacyCompatible(buffer.array(), metadata,
inheritedFractionalBoundaries);
+ }
+
+ /// Rejects fresh boundary masses that legacy readers cannot recompress
without inventing mass.
+ /// Call after boundary normalization and removing zero-weight centroids;
fractional interior mass is supported.
+ static void checkLegacyBoundaryWeights(int count, double firstWeight, double
lastWeight) {
+ if (count > 0 && (firstWeight < 1.0 || lastWeight < 1.0
+ || count == 1 && firstWeight != 1.0 && firstWeight < 2.0)) {
+ throw new IllegalArgumentException("Cannot serialize fractional TDigest
boundary mass for legacy readers");
Review Comment:
**[deferred failure far from cause]** `add(value, weight)` accepts
fractional boundary mass, but serialization rejects it here. Reproduced: `new
TDigestAccumulator(100); add(5.0, 0.5); quantile(0.5)` → 5.0, then
`serialize()` throws `IllegalArgumentException("Cannot serialize fractional
TDigest boundary mass for legacy readers")` — surfacing inside
`ObjectSerDeUtils.serialize` at broker-exchange time, far from the offending
`add()`. tdunning `asBytes` serialized such digests without error.
Unreachable from SQL today (all SQL paths add unit weights), but any
plugin/UDF caller of the public double-weight API hits a query failure at wire
time. Consider validating at `add()` (fail at the cause) or at least an
explicit exception message pointing at the originating input.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/tdigest/TDigestAccumulator.java:
##########
@@ -536,59 +755,96 @@ public double quantile(double quantile) {
if (Double.isNaN(quantile) || quantile < 0.0 || quantile > 1.0) {
throw new IllegalArgumentException("q should be in [0,1], got " +
quantile);
}
+ if (_legacyDegraded) {
+ return Double.NaN;
+ }
materializePendingSerializedTDigest();
flush();
- if (_numCentroids == 0) {
+ return quantile(quantile, _numCentroids, _totalWeight, _min, _max,
_centroidMeans, _centroidWeights, null);
+ }
+
+ /// Interpolates the same distribution from primitive centroids or validated
encoded centroids, without a copy.
+ static double quantile(double quantile, int count, double totalWeight,
double min, double max, double[] means,
+ double[] weights, SerializedTDigestState encoded) {
+ if (count == 0) {
return Double.NaN;
}
- if (_numCentroids == 1) {
- return _centroidMeans[0];
+ if (count == 1) {
+ return mean(means, encoded, 0);
+ }
+ if (quantile == 0.0) {
+ return min;
+ }
+ if (quantile == 1.0) {
+ return max;
}
- double index = quantile * _totalWeight;
- if (index < 1.0) {
- return _min;
+ double index = quantile * totalWeight;
+ double firstWeight = weight(weights, encoded, 0);
+ if (firstWeight < 1.0 && index < firstWeight / 2.0) {
+ return weightedAverage(min, firstWeight / 2.0 - index, mean(means,
encoded, 0), index);
}
- if (_centroidWeights[0] > 1.0 && index < _centroidWeights[0] / 2.0) {
- return _min + (index - 1.0) / (_centroidWeights[0] / 2.0 - 1.0) *
(_centroidMeans[0] - _min);
+ if (firstWeight >= 1.0 && index < 1.0) {
+ return min;
}
- int lastIndex = _numCentroids - 1;
- if (index > _totalWeight - 1.0) {
- return _max;
+ if (weight(weights, encoded, 0) > 1.0 && index < weight(weights, encoded,
0) / 2.0) {
+ return min + (index - 1.0) / (weight(weights, encoded, 0) / 2.0 - 1.0) *
(mean(means, encoded, 0) - min);
+ }
+ int lastIndex = count - 1;
+ double lastWeight = weight(weights, encoded, lastIndex);
+ if (lastWeight < 1.0 && index >= totalWeight - lastWeight / 2.0) {
+ return weightedAverage(mean(means, encoded, lastIndex), totalWeight -
index, max,
+ index - (totalWeight - lastWeight / 2.0));
+ }
+ if (lastWeight >= 1.0 && index > totalWeight - 1.0) {
+ return max;
}
- if (_centroidWeights[lastIndex] > 1.0
- && _totalWeight - index <= _centroidWeights[lastIndex] / 2.0) {
- return _max - (_totalWeight - index - 1.0) /
(_centroidWeights[lastIndex] / 2.0 - 1.0)
- * (_max - _centroidMeans[lastIndex]);
+ if (weight(weights, encoded, lastIndex) > 1.0
+ && totalWeight - index <= weight(weights, encoded, lastIndex) / 2.0) {
+ return max - (totalWeight - index - 1.0) / (weight(weights, encoded,
lastIndex) / 2.0 - 1.0)
+ * (max - mean(means, encoded, lastIndex));
}
- double weightSoFar = _centroidWeights[0] / 2.0;
+ double weightSoFar = weight(weights, encoded, 0) / 2.0;
for (int i = 0; i < lastIndex; i++) {
- double halfWeight = (_centroidWeights[i] + _centroidWeights[i + 1]) /
2.0;
+ double halfWeight = (weight(weights, encoded, i) + weight(weights,
encoded, 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(means, weights,
encoded, count, i);
+ if (index - weightSoFar < leftExcludedWeight) {
+ return mean(means, encoded, i);
}
- double rightUnitWeight = 0.0;
- if (_centroidWeights[i + 1] == 1.0) {
- if (weightSoFar + halfWeight - index <= 0.5) {
- return _centroidMeans[i + 1];
- }
- rightUnitWeight = 0.5;
+ double rightExcludedWeight = getPointMassHalfWeight(means, weights,
encoded, count, i + 1);
+ if (weightSoFar + halfWeight - index <= rightExcludedWeight) {
+ return mean(means, encoded, i + 1);
}
- double leftWeight = index - weightSoFar - leftUnitWeight;
- double rightWeight = weightSoFar + halfWeight - index -
rightUnitWeight;
- return weightedAverage(_centroidMeans[i], rightWeight,
_centroidMeans[i + 1], leftWeight);
+ double leftWeight = index - weightSoFar - leftExcludedWeight;
+ double rightWeight = weightSoFar + halfWeight - index -
rightExcludedWeight;
+ return weightedAverage(mean(means, encoded, i), rightWeight,
mean(means, encoded, i + 1), leftWeight);
}
weightSoFar += halfWeight;
}
- double weightToMax = index - _totalWeight - _centroidWeights[lastIndex] /
2.0;
- double weightFromLastCentroid = _centroidWeights[lastIndex] / 2.0 -
weightToMax;
- return weightedAverage(_centroidMeans[lastIndex], weightToMax, _max,
weightFromLastCentroid);
+ // Rounded cumulative weights can leave a small gap before the directly
computed last centroid center.
+ double weightToMax = Math.max(0.0, index - (totalWeight - lastWeight /
2.0));
Review Comment:
**[narrow divergence from legacy result]** Clamping `weightToMax` to 0 makes
the tail fallthrough return `mean[lastIndex]`, where tdunning 3.3 (and the
pre-PR code) computed `z1 = index − totalWeight − w/2` (very negative) and the
clamp effectively returned `max`. The branch is reachable in ~1-ulp windows of
`index`: verified numerically with weights `[1.0, 2.0000000000000004,
5.000000000000001, 3.0]`, q=0.8636363636363635 → index=9.5 exactly; old returns
max, new returns mean[last], strictly below max.
The new result is arguably more correct (the old max came from a tdunning
sign quirk), and the comment documents the gap — but since everything else in
this kernel is deliberately bug-compatible, this silent high-quantile
divergence is worth an explicit test and a note in the migration docs.
--
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]
