xiangfu0 commented on code in PR #19795:
URL: https://github.com/apache/pinot/pull/19795#discussion_r4253367162
##########
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:
Addressed in 3b8517bbdb. The fail-loud policy for an unknown nonempty
distribution is intentional: unchanged bytes stay readable and byte-exact,
while unsafe merges fail rather than discard unknown mass. The linked migration
notes now explicitly describe old reducers completing with NaN/garbage versus
this failure policy, including query, rollup and star-tree impact. Existing
opaque-payload and mutation-rejection regressions remain.
##########
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:
Addressed in 3b8517bbdb. Reproduced the finite-only +/-1.7e308 overflow
against genuine 3.3. Upgrade notes now call out corrupt intermediates produced
by old live servers, finishing producer upgrades, and rebuilding poisoned
stored digests. The legacy outcome depended on merge order: some old paths
already rejected NaN sources; the notes say old reducers could complete with
NaN or garbage.
##########
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:
Addressed in 3b8517bbdb. Kept the intentional point-mass correction and made
identical-byte/old-new reducer differences explicit in the notes. The existing
plateau and duplicate-heavy regressions cover this beyond the rank envelope;
restoring singleton-only exclusion fails both tests and 1/270 frozen accuracy
cases. The local genuine 3.3 same-byte probe also reproduces your weighted seed
45 result.
##########
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:
Addressed in 3b8517bbdb. Fixed this at codec inspection: zero centroids or
exclusively zero encoded weights are empty despite unused NaN/reversed extrema.
Positive/negative cancellation and non-finite weights still remain opaque.
Existing regressions now cover both encodings and reducer operand orders, byte
aggregation/group-by, and NULL-enabled/disabled extraction; the new zero-mass
cases fail before the fix and pass afterward.
--
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]
