xiangfu0 commented on code in PR #19795:
URL: https://github.com/apache/pinot/pull/19795#discussion_r4222550443
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -1118,12 +1151,54 @@ private void bufferIncomingCentroids(double[] means,
double[] weights, int count
private void materializePendingSerializedTDigest() {
if (_pendingSerializedTDigest != null) {
byte[] bytes = _pendingSerializedTDigest;
+ SerializedTDigestMetadata metadata = _pendingSerializedMetadata;
_pendingSerializedTDigest = null;
+ _pendingSerializedMetadata = null;
_pendingSerializedTotalWeight = Double.NaN;
- mergeSerializedTDigest(bytes);
+ 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) {
+ if (retainOriginal) {
+ _legacySerializedBytes = input.retainBytes();
+ } else {
+ _legacySerializedDirty = true;
+ }
+ _legacyDegraded = true;
Review Comment:
**[correctness/behavior] One degraded legacy input silently poisons an
entire aggregation group to NaN, and rollup persists the loss.** When any
merged input has `needsLegacyFallback` (e.g. a single zero-weight centroid in a
stored 3.2-era digest), `mergeLegacyDegradedInput` flips the whole accumulator
to `_legacyDegraded`, after which `quantile()` returns `NaN` for the group even
if it also holds millions of healthy rows (verified empirically: old code
returned p50=10.0 for a zero-weight-centroid digest; new code returns NaN, and
merging it into a 1M-row group turns the group's percentile to NaN with
`size()=1000002`). `toLegacyDegradedBytes` then serializes the group as a
single NaN centroid, so a merge-rollup task **succeeds** and permanently
replaces real data with NaN digests (pre-PR the task failed loudly with IAE,
leaving source segments intact). If retaining NaN statistics for poisoned
inputs is intended, consider (a) not infecting healthy already-accumulated mass
when one
input is degraded, or at minimum (b) logging/metric-ing the degradation so it
is distinguishable from genuinely empty data.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TDigestUtils.java:
##########
@@ -624,19 +462,22 @@ private static double weightedMean(double firstMean,
double firstWeight, double
}
return Double.POSITIVE_INFINITY;
}
- if (Math.copySign(1.0, firstMean) == Math.copySign(1.0, secondMean)) {
- return firstMean + (secondMean - firstMean) * secondWeight / totalWeight;
- }
- return firstMean * (firstWeight / totalWeight) + secondMean *
(secondWeight / totalWeight);
+ double average = Math.copySign(1.0, firstMean) == Math.copySign(1.0,
secondMean)
+ ? firstMean + (secondMean - firstMean) * (secondWeight / totalWeight)
+ : firstMean * (firstWeight / totalWeight) + secondMean * (secondWeight
/ totalWeight);
+ return Math.max(Math.min(firstMean, secondMean), Math.min(average,
Math.max(firstMean, secondMean)));
}
- private static int getLegacyDefaultCentroidCapacity(double compression) {
- return Math.addExact(Math.multiplyExact(2, (int) Math.ceil(compression)),
LEGACY_CAPACITY_PADDING);
+ /// Default centroid capacity allocated by the verbose t-digest 3.2 reader.
+ public static int getLegacyDefaultCentroidCapacity(double compression) {
+ return (int) Math.min(Integer.MAX_VALUE,
Review Comment:
**[correctness/rolling-upgrade] `makeLegacyCompatible` can emit payloads the
real t-digest 3.2 reader cannot hold.** Two gaps: (1)
`getLegacyDefaultCentroidCapacity` now normalizes compression to >= 10, but the
actual 3.2 verbose reader sizes its arrays from the RAW header compression
(`size = (int)(2*Math.ceil(compression)) + 10;` with no clamp), so for a header
compression of e.g. 5 the pass-through threshold is treated as 30 while the old
reader only allocates 20 cells -> `ArrayIndexOutOfBoundsException` in a
mixed-version cluster (the previous revision of this PR reduced such payloads
instead). (2) When `metadata.needsLegacyFallback()` is true the payload passes
through verbatim even when `centroidCount > legacyCapacity`. Both only trigger
for corrupt or externally-crafted payloads (no legitimate Pinot/tdunning writer
emits them), but the function's documented contract is 'readable by both
t-digest 3.2 and 3.3', so either enforce the raw-compression capacity here or
document t
he exception.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/PercentileTDigestValueAggregator.java:
##########
@@ -284,51 +266,81 @@ public void add(double value) {
}
@Override
- public void add(double value, int weight) {
+ public void add(double value, double weight) {
if (Double.isNaN(value)) {
throw new IllegalArgumentException("Cannot add NaN to t-digest");
}
- if (weight <= 0) {
+ if (!(weight > 0.0) || !Double.isFinite(weight)) {
throw new IllegalArgumentException("TDigest weight must be positive: "
+ weight);
}
+ if (hasValidStatistics()) {
+ checkTotalWeight(getTotalWeight() + weight);
+ }
invalidateCaches();
if (value == Double.NEGATIVE_INFINITY) {
- _negativeInfinityWeight = Math.addExact(_negativeInfinityWeight,
weight);
+ _negativeInfinityWeight = _negativeInfinityWeight + weight;
} else if (value == Double.POSITIVE_INFINITY) {
- _positiveInfinityWeight = Math.addExact(_positiveInfinityWeight,
weight);
+ _positiveInfinityWeight = _positiveInfinityWeight + weight;
} else {
_finiteDigest.add(value, weight);
}
}
@Override
public void add(TDigest other) {
+ if (hasValidStatistics() && other.hasValidStatistics()) {
+ checkTotalWeight(getTotalWeight() + other.getTotalWeight());
+ }
if (other instanceof NonFiniteAwareTDigest) {
NonFiniteAwareTDigest wrapped = (NonFiniteAwareTDigest) other;
- long negativeInfinityWeight = wrapped._negativeInfinityWeight;
- long positiveInfinityWeight = wrapped._positiveInfinityWeight;
- // MergingDigest.add(List) copies the implementation's
double-precision weight arrays directly. The generic
- // add(TDigest) path goes through Centroid.count(), which is limited
to an int and truncates large weights.
- wrapped.invalidateCaches();
+ double negativeInfinityWeight = wrapped._negativeInfinityWeight;
+ double positiveInfinityWeight = wrapped._positiveInfinityWeight;
invalidateCaches();
- if (wrapped._finiteDigest.size() > 0L) {
- _finiteDigest.add(List.of(wrapped._finiteDigest));
+ if (!wrapped._finiteDigest.isEmpty()) {
+ _finiteDigest.add(wrapped._finiteDigest);
}
- _negativeInfinityWeight = Math.addExact(_negativeInfinityWeight,
negativeInfinityWeight);
- _positiveInfinityWeight = Math.addExact(_positiveInfinityWeight,
positiveInfinityWeight);
+ _negativeInfinityWeight += negativeInfinityWeight;
+ _positiveInfinityWeight += positiveInfinityWeight;
return;
}
- if (other.size() == 0L) {
+ if (other.isEmpty()) {
return;
}
invalidateCaches();
- if (Double.isFinite(other.getMin()) && Double.isFinite(other.getMax())) {
- _finiteDigest.add(List.of(other));
+ if (!other.hasValidStatistics()
+ || Double.isFinite(other.getMin()) &&
Double.isFinite(other.getMax())) {
+ _finiteDigest.add(other);
return;
}
- for (Centroid centroid : other.centroids()) {
- add(centroid.mean(), centroid.count());
+ // Read precise centroid weights directly; generic sources need no full
digest serialization for their tails.
+ Collection<Centroid> centroids = other.centroids();
+ double[] finiteMeans = new double[centroids.size()];
+ double[] finiteWeights = new double[centroids.size()];
+ int finiteCount = 0;
+ double finiteMin = Double.POSITIVE_INFINITY;
+ double finiteMax = Double.NEGATIVE_INFINITY;
+ for (Centroid centroid : centroids) {
+ double mean = centroid.mean();
+ double weight = centroid.weight();
+ if (mean == Double.NEGATIVE_INFINITY) {
+ _negativeInfinityWeight += weight;
+ } else if (mean == Double.POSITIVE_INFINITY) {
+ _positiveInfinityWeight += weight;
+ } else {
+ finiteMeans[finiteCount] = mean;
+ finiteWeights[finiteCount++] = weight;
+ finiteMin = Math.min(finiteMin, mean);
+ finiteMax = Math.max(finiteMax, mean);
+ }
+ }
+ if (finiteCount > 0) {
+ finiteMin = Double.isFinite(other.getMin()) ? other.getMin() :
finiteMin;
+ finiteMax = Double.isFinite(other.getMax()) ? other.getMax() :
finiteMax;
+ SerializedTDigestInput finiteInput = new SerializedTDigestInput();
+ finiteInput.reset(toVerboseBytes(finiteMin, finiteMax,
other.compression(),
+ finiteMeans, finiteWeights, finiteCount));
Review Comment:
**[latent bug] Inconsistent capacity check on a self-built payload.** This
`reset(byte[])` overload validates with `checkCapacity=true`, while `fromBytes`
deliberately uses `reset(ByteBuffer.wrap(bytes), false)` for the same kind of
internally generated finite payload. A third-party SPI `TDigest` with an
infinite tail (so the finite-extrema branch above is skipped) whose
`centroids()` view exposes more centroids than
`getDefaultCentroidCapacity(compression)` -- which the SPI contract explicitly
permits for working-compression flushes -- makes this throw `TDigest centroid
count exceeds capacity` and aborts the segment/star-tree build instead of
merging.
```suggestion
finiteInput.reset(ByteBuffer.wrap(toVerboseBytes(finiteMin,
finiteMax, other.compression(),
finiteMeans, finiteWeights, finiteCount)), false);
```
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -240,167 +271,128 @@ 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);
+ // 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) {
+ _legacySerializedDirty = true;
+ _totalWeight += other._totalWeight;
+ _min = Math.min(_min, other._min);
+ _max = Math.max(_max, other._max);
+ return;
+ }
+ checkTotalWeight(_totalWeight + _incomingWeight + _numRawValues +
other._totalWeight);
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() {
+ if (_publiclyCompressed && _numRawValues == 0 && _numIncomingCentroids ==
0) {
+ return this;
+ }
+ PercentileTDigestAccumulator copy =
+ new PercentileTDigestAccumulator(_compression, false, false,
_useTwoLevelRawCompression);
+ copy._numCentroids = _numCentroids;
+ copy._numRawValues = _numRawValues;
+ copy._numIncomingCentroids = _numIncomingCentroids;
+ copy._centroidMeans = _numCentroids == 0 ? null :
Arrays.copyOf(_centroidMeans, _numCentroids);
+ copy._centroidWeights = _numCentroids == 0 ? null :
Arrays.copyOf(_centroidWeights, _numCentroids);
+ copy._rawValues = _numRawValues == 0 ? null : Arrays.copyOf(_rawValues,
_numRawValues);
+ copy._incomingMeans = _numIncomingCentroids == 0 ? null :
Arrays.copyOf(_incomingMeans, _numIncomingCentroids);
+ copy._incomingWeights = _numIncomingCentroids == 0 ? null :
Arrays.copyOf(_incomingWeights, _numIncomingCentroids);
+ copy._serializedMainCapacity = _serializedMainCapacity;
+ copy._serializedBufferCapacity = _serializedBufferCapacity;
+ copy._mergeCount = _mergeCount;
+ copy._publiclyCompressed = _publiclyCompressed;
+ copy._incomingCentroidsSorted = _incomingCentroidsSorted;
+ copy._hasSerializedInput = _hasSerializedInput;
+ copy._hasFractionalWeights = _hasFractionalWeights;
+ copy._totalWeight = _totalWeight;
+ copy._incomingWeight = _incomingWeight;
+ copy._min = _min;
+ copy._max = _max;
+ copy.compress();
+ return copy;
+ }
+
+ public void addSerializedTDigest(byte[] bytes) {
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;
}
- void addSerializedTDigest(SerializedTDigestInput input) {
- if (!_hasSerializedInput && _numCentroids == 0 && _totalWeight == 0.0 &&
_numRawValues == 0
- && _numIncomingCentroids == 0) {
- _pendingSerializedTDigest = input.retainBytes();
- _pendingSerializedTotalWeight = Double.NaN;
- _hasSerializedInput = true;
- return;
+ public void addSerializedTDigest(SerializedTDigestInput input) {
+ if (prepareSerializedTDigest(input)) {
+ mergeSerializedTDigest(input, false);
}
- materializePendingSerializedTDigest();
- input.decode();
- mergeSerializedTDigest(input);
- _hasSerializedInput = true;
}
- void addSerializedTDigestDirect(SerializedTDigestInput input) {
- if (!_hasSerializedInput && _numCentroids == 0 && _totalWeight == 0.0 &&
_numRawValues == 0
- && _numIncomingCentroids == 0) {
+ public void addSerializedTDigestDirect(SerializedTDigestInput input) {
+ if (prepareSerializedTDigest(input)) {
+ mergeSerializedTDigest(input, true);
+ }
+ }
+
+ private boolean prepareSerializedTDigest(SerializedTDigestInput input) {
+ if (!_hasSerializedInput && hasNoInputs()) {
+ input.inspectMetadata();
+ _hasFractionalWeights |= input._metadata.fractionalWeights();
_pendingSerializedTDigest = input.retainBytes();
- _pendingSerializedTotalWeight = Double.NaN;
+ _pendingSerializedTotalWeight = input._totalWeight;
+ _pendingSerializedMetadata = input._metadata;
+ _legacyDegraded = input._metadata.needsLegacyFallback();
_hasSerializedInput = true;
- return;
+ return false;
}
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);
- }
+ _hasFractionalWeights |= input._metadata.fractionalWeights();
+ if (_legacyDegraded || input._metadata.needsLegacyFallback()) {
+ mergeLegacyDegradedInput(input, false);
+ return false;
}
+ checkTotalWeight(_totalWeight + _incomingWeight + _numRawValues +
input._totalWeight);
Review Comment:
**[behavior change, please confirm intent] Merging digests whose combined
mass overflows double now hard-fails the query.** Two individually valid legacy
digests with totalWeight ~1e308 each pass inspection, but merging them throws
`IllegalArgumentException("Invalid TDigest total weight: Infinity")`
(reproduced). Pre-PR the merge completed (with saturated statistics). A hard
failure may well be the better contract, but it is a new failure mode for
previously-queryable (if garbage) data -- worth an explicit note in the PR
description/migration notes.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -638,10 +686,17 @@ public int byteSize() {
return serialize().length;
}
+ /// Returns the small-encoding size for valid statistics. Historical
degraded state retains its original encoding
+ /// until modified, then uses verbose bytes so poisoned values and precise
weights survive without narrowing.
@Override
public int smallByteSize() {
Review Comment:
**[SPI contract] `smallByteSize()`/`asSmallBytes()` can disagree, and
degraded digests emit VERBOSE bytes under the small-encoding API.** (1)
`smallByteSize()` returns `30 + 8*n` for any n, but `asSmallBytes` ->
`toCapacityPreservingBytes` throws `IllegalStateException` for n >
Short.MAX_VALUE, so a caller following the size-then-write contract allocates
successfully and then crashes (tdunning silently wrapped the short instead --
failing fast is better, but the sizing method should fail too). (2) In
`_legacyDegraded` state both methods return `serialize()` output, which is
VERBOSE-encoded (flag 1) under an API whose tdunning contract always emitted
SMALL_ENCODING (flag 2); a consumer hardcoding the compact layout misparses. No
in-repo caller is affected -- but this is now the public migration target for
plugins, so at minimum document both deviations on the SPI
`TDigest.asSmallBytes`/`smallByteSize` javadoc.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -1029,34 +1032,64 @@ private void finishMerge(int numOutputCentroids, double
totalWeight, double comp
private void ensureIncomingCapacity(int capacity) {
if (_incomingMeans == null || capacity > _incomingMeans.length) {
- int newCapacity = _incomingMeans == null ? Math.max(capacity,
_centroidCapacity)
- : Math.max(capacity, Math.min(_pendingCentroidCapacity,
- Math.multiplyExact(_incomingMeans.length,
DEFAULT_MERGE_BUFFER_MULTIPLIER)));
+ int newCapacity = _incomingMeans == null
+ ? Math.max(capacity, Math.min(_centroidCapacity,
_pendingCentroidCapacity))
+ : Math.max(capacity, (int) Math.min(_pendingCentroidCapacity,
+ (long) _incomingMeans.length *
TDigestUtils.DEFAULT_MERGE_BUFFER_MULTIPLIER));
_incomingMeans = _incomingMeans == null ? new double[newCapacity] :
Arrays.copyOf(_incomingMeans, newCapacity);
_incomingWeights =
_incomingWeights == null ? new double[newCapacity] :
Arrays.copyOf(_incomingWeights, newCapacity);
}
}
- private void ensureIncomingSortCapacity(int capacity) {
- ensureSortedIncomingCapacity(capacity);
- if (_incomingOrder == null || capacity > _incomingOrder.length) {
- int newCapacity = _incomingOrder == null ? Math.max(capacity,
_centroidCapacity)
- : Math.max(capacity, Math.multiplyExact(_incomingOrder.length, 2));
- _incomingOrder = new int[newCapacity];
+ private void stableSortIncoming(int count) {
Review Comment:
**[efficiency] `stableSortIncoming` replaces tdunning's primitive stable
sort with fastutil's generic `mergeSort` driven by two capturing lambdas**
(comparator + swapper), adding per-comparison/per-swap functional-interface
indirection plus an index-array init on every flush of the incoming buffer (up
to ~1050 centroids at default compression, recurring on every server-combine /
broker-reduce / group-by trim merge). Consider a primitive stable sort over an
index array keyed by the means, or packing (meanBits, index) into a `long[]`
and using `Arrays.sort`.
--
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]
