xiangfu0 commented on code in PR #19795:
URL: https://github.com/apache/pinot/pull/19795#discussion_r4253368232
##########
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:
Addressed in 3b8517bbdb. NaN intentionally identifies an unknown
distribution, and the notes explicitly distinguish it from SQL NULL and explain
how to locate/rebuild affected data. The global rate-limited WARN and its lack
of table/segment identity remain documented. Accurate per-table counters
require caller context absent from this digest API, so adding that plumbing is
deferred from this dependency migration.
##########
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:
Addressed in 3b8517bbdb. Kept validation at serialization because fractional
inputs can later become supported interior mass or merge into a representable
singleton; early rejection would break those cases. The error now names
add(value, weight), reports first/last endpoint weights, and explains the
unit-endpoint requirement. The existing guard test asserts the diagnostic;
genuine 3.2/3.3 probes confirm fractional singletons fail recompression with
assertions enabled while supported interior mass works.
##########
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:
Addressed in 3b8517bbdb. Added a focused same-byte regression for the
one-ULP gap and neighboring quantiles, including encoded/direct equivalence,
plus an explicit migration note. The correction preserves 90, 90, 90; genuine
3.3 returns 100, 100, 90 on that payload. Keeping the clamp avoids the legacy
non-monotonic jump.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/tdigest/TDigestAccumulator.java:
##########
@@ -104,64 +110,204 @@ public TDigestAccumulator(int compression) {
this((double) compression, true, true);
}
+ public TDigestAccumulator(double compression) {
+ this(compression, true, true);
+ }
+
private TDigestAccumulator(double compression, boolean allocateRawBuffer,
boolean allocateMergeBuffers) {
- if (!(compression > 0.0) || !Double.isFinite(compression)) {
- throw new IllegalArgumentException("TDigest compression must be
positive: " + compression);
- }
- super.setScaleFunction(ScaleFunction.K_1);
- _compression = Math.max(MIN_COMPRESSION, compression);
+ this(compression, allocateRawBuffer, allocateMergeBuffers, false);
+ }
+
+ private TDigestAccumulator(double compression, boolean allocateRawBuffer,
boolean allocateMergeBuffers,
+ boolean useTwoLevelRawCompression) {
+ _compression = TDigestCodec.normalizeCompression(compression);
+ _useTwoLevelRawCompression = useTwoLevelRawCompression;
_workingCompression = TWO_LEVEL_COMPRESSION_MULTIPLIER * _compression;
_publicMaxWeightScale = calculateK1MaxWeightScale(_compression);
_workingMaxWeightScale = calculateK1MaxWeightScale(_workingCompression);
long roundedCompression = (long) Math.ceil(_compression);
- _centroidCapacity = getCentroidCapacity(_compression);
- _pendingCentroidCapacity = getPendingCentroidCapacity(_centroidCapacity);
+ _centroidCapacity = TDigestCodec.getDefaultCentroidCapacity(_compression);
+ _pendingCentroidCapacity = getPendingCentroidCapacity(_centroidCapacity,
useTwoLevelRawCompression);
if (allocateRawBuffer) {
- _rawValues = new double[getRawBufferSize(roundedCompression)];
+ _rawValues = new double[getInitialRawBufferSize(roundedCompression)];
}
if (allocateMergeBuffers) {
- _centroidMeans = new double[_centroidCapacity];
- _centroidWeights = new double[_centroidCapacity];
- _outputMeans = new double[_centroidCapacity];
- _outputWeights = new double[_centroidCapacity];
+ int initialCapacity = Math.min(_centroidCapacity, MIN_RAW_BUFFER_SIZE);
+ _centroidMeans = new double[initialCapacity];
+ _centroidWeights = new double[initialCapacity];
+ _outputMeans = new double[initialCapacity];
+ _outputWeights = new double[initialCapacity];
}
}
- public static TDigestAccumulator forSerializedTDigest(byte[] bytes) {
- return new TDigestAccumulator(readCompression(bytes), false, false);
+ /// Creates a lazy digest with the legacy two-level compression used by
Star-tree and general digest aggregation.
+ public static TDigestAccumulator forLegacyAggregation(double compression) {
+ return new TDigestAccumulator(compression, false, false, true);
}
- public static TDigestAccumulator forSerializedTDigest(SerializedTDigestInput
input) {
- return new TDigestAccumulator(input._compression, false, false);
+ /// Reads and retains one legacy digest, preserving the general aggregation
compression policy.
+ public static TDigestAccumulator fromBytes(byte[] bytes) {
+ return fromBytes(ByteBuffer.wrap(bytes));
}
- public static TDigestAccumulator forSerializedTDigestWithMergeBuffers(byte[]
bytes) {
- return new TDigestAccumulator(readCompression(bytes), false, true);
+ /// Reads one payload and advances the buffer past it.
+ public static TDigestAccumulator fromBytes(ByteBuffer bytes) {
+ SerializedTDigestInput input = new SerializedTDigestInput();
+ input.reset(bytes);
+ return fromBytes(input);
}
- public static TDigestAccumulator forSerializedTDigest(ByteBuffer input) {
- byte[] bytes = new byte[input.remaining()];
- input.get(bytes);
- TDigestAccumulator accumulator = forSerializedTDigest(bytes);
- accumulator.addSerializedTDigest(bytes);
- return accumulator;
+ /// Retains an already validated input without rescanning its centroid bytes.
+ public static TDigestAccumulator fromBytes(SerializedTDigestInput input) {
+ TDigestAccumulator digest = forLegacyAggregation(input.getCompression());
+ digest.addSerializedTDigest(input);
+ return digest;
}
public static TDigestAccumulator forReduction(double compression) {
return new TDigestAccumulator(compression, false, false);
}
+ /// Snapshots buffered inputs and centroid state without materializing or
compressing the source. Mutable buffers
+ /// are copied; immutable retained bytes and metadata are shared. Merge and
sort scratch buffers remain lazy.
+ /// Raw capacity is retained so future flush timing is unchanged.
+ TDigestAccumulator copy() {
+ return copy(_rawValues != null ? _rawValues.length : 0);
+ }
+
+ private TDigestAccumulator copy(int rawBufferSize) {
+ TDigestAccumulator copy =
+ new TDigestAccumulator(_compression, false, false,
_useTwoLevelRawCompression);
+ copy._rawValues = rawBufferSize > 0 ? Arrays.copyOf(_rawValues,
rawBufferSize) : null;
+ copy._centroidMeans = _numCentroids > 0 ? Arrays.copyOf(_centroidMeans,
_numCentroids) : null;
+ copy._centroidWeights = _numCentroids > 0 ?
Arrays.copyOf(_centroidWeights, _numCentroids) : null;
+ copy._incomingMeans = _numIncomingCentroids > 0 ?
Arrays.copyOf(_incomingMeans, _numIncomingCentroids) : null;
+ copy._incomingWeights = _numIncomingCentroids > 0 ?
Arrays.copyOf(_incomingWeights, _numIncomingCentroids) : null;
+ copy._pendingSerializedTDigest = _pendingSerializedTDigest;
+ copy._pendingSerializedMetadata = _pendingSerializedMetadata;
+ copy._originalFractionalBytes = _originalFractionalBytes;
+ copy._originalBoundaryBytes = _originalBoundaryBytes;
+ copy._historicalFractionalMinMean = _historicalFractionalMinMean;
+ copy._historicalFractionalMaxMean = _historicalFractionalMaxMean;
+ copy._freshFractionalMinMean = _freshFractionalMinMean;
+ copy._freshFractionalMaxMean = _freshFractionalMaxMean;
+ copy._legacyDegraded = _legacyDegraded;
+ copy._hasFractionalWeights = _hasFractionalWeights;
+ copy._deferBoundaryNormalization = _deferBoundaryNormalization;
+ copy._numRawValues = _numRawValues;
+ copy._numCentroids = _numCentroids;
+ copy._numIncomingCentroids = _numIncomingCentroids;
+ copy._mergeCount = _mergeCount;
+ copy._publiclyCompressed = _publiclyCompressed;
+ copy._incomingCentroidsSorted = _incomingCentroidsSorted;
+ copy._totalWeight = _totalWeight;
+ copy._incomingWeight = _incomingWeight;
+ copy._min = _min;
+ copy._max = _max;
+ return copy;
+ }
+
+ /// Copies primitive centroids without encoding and decoding them. Zero
weights are ignored, and unordered
+ /// means are sorted when initializing a stored, already compressed
distribution; caller arrays are not mutated.
+ /// Subsequent inputs use the normal buffered merge. Positive weights and
consistent extrema are required.
+ public void addCentroids(double[] means, double[] weights, int count, double
min, double max,
+ boolean alreadyCompressed) {
+ addCentroids(means, weights, count, min, max, alreadyCompressed, false);
+ }
+
+ /// Adds validated legacy centroids without marking their existing
fractional mass as fresh input.
+ void addHistoricalCentroids(double[] means, double[] weights, int count,
double min, double max,
+ boolean alreadyCompressed) {
+ addCentroids(means, weights, count, min, max, alreadyCompressed, true);
+ }
+
+ private void addCentroids(double[] means, double[] weights, int count,
double min, double max,
+ boolean alreadyCompressed, boolean historical) {
+ requireMutable();
+ if (count < 0 || count > means.length || count > weights.length) {
+ throw new IllegalArgumentException("Invalid TDigest centroid count: " +
count);
+ }
+ double totalWeight = 0.0;
+ int positiveCount = 0;
+ boolean sorted = true;
+ boolean fractionalWeights = false;
+ double previousMean = Double.NEGATIVE_INFINITY;
+ for (int i = 0; i < count; i++) {
+ double weight = weights[i];
+ if (weight == 0.0) {
+ continue;
+ }
+ double mean = means[i];
+ if (!(weight > 0.0) || !Double.isFinite(weight) || !(mean >= min && mean
<= max)) {
+ throw new IllegalArgumentException("Invalid TDigest centroid");
+ }
+ totalWeight += weight;
+ sorted &= mean >= previousMean;
+ previousMean = mean;
+ fractionalWeights |= weight != Math.rint(weight);
+ positiveCount++;
+ }
+ checkTotalWeight(getTotalWeight() + totalWeight);
+ if (positiveCount == 0) {
+ return;
+ }
+ prepareForMutation();
+ _originalFractionalBytes = null;
+ _hasFractionalWeights |= fractionalWeights;
+ if (!historical && fractionalWeights) {
+ for (int i = 0; i < count; i++) {
+ recordFreshFractionalMean(means[i], weights[i]);
+ }
+ }
+ if (hasNoInputs() && alreadyCompressed) {
+ ensureCentroidCapacity(Math.addExact(positiveCount, 2));
+ for (int i = 0; i < count; i++) {
+ if (weights[i] > 0.0) {
+ _centroidMeans[_numCentroids] = means[i];
+ _centroidWeights[_numCentroids++] = weights[i];
+ }
+ }
+ if (!sorted) {
+ sortCentroids(_centroidMeans, _centroidWeights, _numCentroids);
+ }
+ _totalWeight = totalWeight;
+ _min = min;
+ _max = max;
+ if (!historical) {
+ normalizeBoundaryCentroids();
+ }
+ _publiclyCompressed = true;
+ } else {
+ if (sorted && positiveCount == count) {
+ bufferIncomingCentroids(means, weights, count, totalWeight, min, max);
+ } else {
+ // Caller-owned arrays are copied only when filtering or sorting is
necessary.
+ double[] sortedMeans = new double[positiveCount];
+ double[] sortedWeights = new double[positiveCount];
+ int index = 0;
+ for (int i = 0; i < count; i++) {
+ if (weights[i] > 0.0) {
+ sortedMeans[index] = means[i];
+ sortedWeights[index++] = weights[i];
+ }
+ }
+ if (!sorted) {
+ sortCentroids(sortedMeans, sortedWeights, positiveCount);
+ }
+ bufferIncomingCentroids(sortedMeans, sortedWeights, positiveCount,
totalWeight, min, max);
+ }
+ }
+ }
+
public void add(double[] values, int from, int toExclusive) {
Review Comment:
Addressed in 3b8517bbdb. Fixed: the array path now scans the requested slice
for NaN before materialization, staging or flushing. The regression starts near
a buffer boundary and verifies unchanged mass, raw contents/count, merge parity
and snapshot bytes after rejection, then retries with valid values. It fails on
the previous code and passes for both compression policies; eight valid-batch
output hashes are unchanged.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/tdigest/TDigestAccumulator.java:
##########
@@ -81,17 +82,22 @@ public final class TDigestAccumulator extends TDigest {
private double[] _incomingWeights;
private double[] _sortedIncomingMeans;
private double[] _sortedIncomingWeights;
- private int[] _incomingOrder;
private SerializedTDigestInput _serializedTDigestInput;
private byte[] _pendingSerializedTDigest;
- private double _pendingSerializedTotalWeight = Double.NaN;
- private boolean _hasSerializedInput;
+ private byte[] _originalFractionalBytes;
+ private byte[] _originalBoundaryBytes;
+ private double _historicalFractionalMinMean = Double.NaN;
+ private double _historicalFractionalMaxMean = Double.NaN;
+ private double _freshFractionalMinMean = Double.NaN;
+ private double _freshFractionalMaxMean = Double.NaN;
+ private SerializedTDigestMetadata _pendingSerializedMetadata;
+ private boolean _legacyDegraded;
Review Comment:
Addressed in 3b8517bbdb. Keeping the type refactor as follow-up work to
limit this migration. These states overlap: retained fractional bytes can
coexist with pending state, and materialization can discover an unrepresentable
weighted boundary, which requires the second serialization check. The
transition audit covered opaque corruption, copies, failed mutations and merge
order without finding a missing guard. A replacement abstraction must preserve
those lazy transitions and retained-byte semantics.
--
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]
