xiangfu0 commented on code in PR #19795:
URL: https://github.com/apache/pinot/pull/19795#discussion_r4225308916
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TDigestUtils.java:
##########
@@ -35,48 +35,42 @@
/// interior centroids until the verbose representation fits the 3.2 capacity.
This keeps intermediate values readable
/// in both directions during a rolling upgrade without silently narrowing
double-precision values to floats.
///
-/// Pinot explicitly uses [ScaleFunction#K_1] for new and deserialized merging
digests. T-digest 3.3 changed its default
-/// to `K_2`, whose smaller middle-quantile centroid count caused a material
P75 accuracy regression in Pinot's
-/// deterministic 100-million-row workload. `K_1` retains the accuracy profile
of t-digest 3.2 while using the 3.3
-/// implementation and its two-level compression.
+/// Pinot owns the in-memory K1 implementation and retains these legacy bytes
for stored columns and mixed-version
+/// intermediate results. Decoding checks lengths and numerical state before
allocating centroid arrays. Historical
+/// poisoned states remain byte-readable with NaN statistics because their
original distribution cannot be recovered.
+/// Opaque poisoned payloads retain their original centroids even if an old
reader's default array cannot hold them;
+/// these historical exceptions remain readable by Pinot, but are not
guaranteed to survive a reverse upgrade.
+/// Fractional boundary masses below one, and single-centroid masses between
one and two, are retained exactly:
+/// t-digest 3.3's recompression assertion cannot be satisfied for these
inputs without changing their mass.
public final class TDigestUtils {
- private static final int VERBOSE_ENCODING = 1;
- private static final int SMALL_ENCODING = 2;
+ public static final int VERBOSE_ENCODING = 1;
+ public static final int SMALL_ENCODING = 2;
private static final int LEGACY_CAPACITY_PADDING = 10;
- private static final int VERBOSE_HEADER_SIZE = 32;
- private static final int VERBOSE_CENTROID_SIZE = 16;
- private static final int SMALL_HEADER_SIZE = 30;
- private static final int SMALL_CENTROID_SIZE = 8;
- private static final int DEFAULT_MERGE_BUFFER_MULTIPLIER = 5;
+ 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 TDigestUtils() {
}
- /// Creates a merging digest with Pinot's accuracy-preserving scale function.
+ /// Creates a digest using Pinot's accuracy-preserving K1 implementation.
public static TDigest createMergingDigest(double compression) {
- return configureScaleFunction(new MergingDigest(compression));
+ return PercentileTDigestAccumulator.forLegacyAggregation(compression);
}
- /// Creates a merging digest with the temporary-buffer footprint used by
t-digest 3.2.
- ///
- /// Segment aggregation can keep many digests live at once. T-digest 3.3
more than doubled the default temporary
- /// buffer to support two-level compression, adding about 11 KiB to every
compression-100 digest. This factory keeps
- /// the 3.3 main-buffer capacity and Pinot's K1 policy, but restores the
smaller temporary buffer for that
- /// high-cardinality use case.
- public static TDigest createMergingDigestWithLegacyBuffer(double
compression) {
- return createMergingDigestWithLegacyBuffer(compression, 0);
+ /// Rejects non-finite compression while retaining the legacy clamping of
finite settings below ten.
+ public static void validateCompression(double compression) {
Review Comment:
Addressed in 14657b2cfb. Restored the positive-compression guard in query
constructors and forReduction; historical storage headers retain their tolerant
path. The regression covers zero/negative query settings and the actual
upstream constructor behavior.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -566,22 +613,16 @@ public double quantile(double quantile) {
for (int i = 0; i < lastIndex; i++) {
double halfWeight = (_centroidWeights[i] + _centroidWeights[i + 1]) /
2.0;
if (weightSoFar + halfWeight > index) {
- double leftUnitWeight = 0.0;
- if (_centroidWeights[i] == 1.0) {
- if (index - weightSoFar < 0.5) {
- return _centroidMeans[i];
- }
- leftUnitWeight = 0.5;
+ double leftExcludedWeight = getPointMassHalfWeight(i);
Review Comment:
Addressed in 14657b2cfb. Quantile/CDF now use actual fractional tail mass,
while P0/P100 retain the encoded extrema. The 0.6-mass case and widened-extrema
interpolation boundaries are covered, including P75 equality at the last
centroid mean.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -1287,10 +1383,15 @@ private void ensureOutputCapacity(int capacity) {
}
}
- private static double getTotalWeight(double[] weights, int count) {
- double totalWeight = 0.0;
- for (int i = 0; i < count; i++) {
- totalWeight += weights[i];
+ private static void checkTotalWeight(double totalWeight) {
+ if (!(totalWeight >= 0.0) || !Double.isFinite(totalWeight)) {
+ throw new IllegalArgumentException("Invalid TDigest total weight: " +
totalWeight);
+ }
+ }
+
+ private static double checkedLegacyTotalWeight(double totalWeight) {
Review Comment:
Addressed in 14657b2cfb. Removed corrupted mass from merge arithmetic. A
corrupt historical input remains read-only with NaN statistics and its original
bytes; mixing or mutating it throws before changing a healthy receiver. Both
merge orders are covered.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -1118,12 +1207,65 @@ private void bufferIncomingCentroids(double[] means,
double[] weights, int count
private void materializePendingSerializedTDigest() {
if (_pendingSerializedTDigest != null) {
byte[] bytes = _pendingSerializedTDigest;
+ SerializedTDigestMetadata metadata = _pendingSerializedMetadata;
_pendingSerializedTDigest = null;
- _pendingSerializedTotalWeight = Double.NaN;
- mergeSerializedTDigest(bytes);
+ _pendingSerializedMetadata = null;
+ SerializedTDigestInput input = getSerializedTDigestInput();
+ input.reset(bytes, metadata);
+ input._retainedBytes = bytes;
+ if (_legacyDegraded) {
+ mergeLegacyDegradedInput(input, true);
+ } else {
+ input.decode();
+ mergeSerializedTDigest(input, false);
+ }
+ }
+ }
+
+ private void mergeLegacyDegradedInput(SerializedTDigestInput input, boolean
retainOriginal) {
+ double totalWeight = checkedLegacyTotalWeight(_totalWeight +
_incomingWeight + _numRawValues + input._totalWeight);
+ if (retainOriginal) {
+ _legacySerializedBytes = input.retainBytes();
+ } else {
+ _legacySerializedDirty = true;
+ }
+ markLegacyDegraded(input);
+ for (int i = 0; i < _numRawValues; i++) {
+ _min = Math.min(_min, _rawValues[i]);
+ _max = Math.max(_max, _rawValues[i]);
+ }
+ _totalWeight = totalWeight;
+ _hasFractionalWeights |= input._metadata.fractionalWeights();
+ _incomingWeight = 0.0;
+ _numIncomingCentroids = 0;
+ _numRawValues = 0;
+ _min = Math.min(_min, input._min);
+ _max = Math.max(_max, input._max);
+ _hasSerializedInput = true;
+ }
+
+ private void markLegacyDegraded(SerializedTDigestInput input) {
+ if (!_legacyDegraded) {
+ LOGGER.warn("Retaining historical TDigest numerical corruption: input
centroids={}, input weight={}, "
+ + "accumulated weight={}. Percentiles for this aggregation state
are NaN.",
+ input._numCentroids, input._totalWeight, _totalWeight +
_incomingWeight + _numRawValues);
+ _legacyDegraded = true;
}
}
+ private byte[] toLegacyDegradedBytes() {
Review Comment:
Addressed in 14657b2cfb. Removed the synthetic NaN-state writer. Corrupted
historical payloads can only be emitted unchanged; merging or mutating them
fails explicitly, so no newly generated poison payload reaches legacy readers.
--
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]
