xiangfu0 commented on code in PR #19795:
URL: https://github.com/apache/pinot/pull/19795#discussion_r4223704393
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -1194,7 +1243,7 @@ private static int normalizeBoundaries(double[] means,
double[] weights, int cou
}
if (count == 1) {
double weight = weights[0];
- if (weight <= 1.0) {
+ if (weight < 2.0) {
Review Comment:
**[code-review]** Serialized singleton digests with boundary weight in (1,2)
or fractional boundary mass corrupt OLD readers during a rolling upgrade
The new SPI accepts fractional double weights and normalizeBoundaries
deliberately retains a singleton weight in (1,2) unrepaired (`if (weight < 2.0)
{ return count; }`), and serialize() ships it verbatim. The previous release's
TDigestUtils.normalizeBoundaries splits a singleton weight-1.5 digest into
weights {1.0, -0.5, 1.0} (negative-weight centroid fabricated from `weight -
2.0`) and feeds that to tdunning 3.3, which produces garbage percentiles or
AssertionError under -ea. No in-tree producer passes fractional weights today,
so this bites SPI users and any future weighted caller in mixed-version
clusters.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TDigestUtils.java:
##########
@@ -86,44 +80,29 @@ public static byte[] serialize(TDigest tDigest) {
/// Serializes a digest using `scratchBuffer` for the temporary verbose
representation when it is large enough.
///
- /// The returned array is independently owned. The scratch buffer is cleared
and may be reused by the caller after
- /// this method returns.
+ /// The returned array is independently owned. Pinot-owned digests do not
use the scratch buffer; callers can
+ /// clear and reuse it after serialization.
public static byte[] serialize(TDigest tDigest, ByteBuffer scratchBuffer) {
ByteOrder scratchOrder = scratchBuffer != null ? scratchBuffer.order() :
null;
try {
- if (!(tDigest instanceof MergingDigest)) {
- // Pinot has always decoded TDigest values as MergingDigest. Preserve
the previous serialization behavior for
- // other implementations instead of interpreting their
implementation-specific payload as MergingDigest
- // bytes.
- ByteBuffer buffer = prepareScratchBuffer(scratchBuffer,
tDigest.byteSize());
- tDigest.asBytes(buffer);
- byte[] bytes = new byte[buffer.position()];
- buffer.flip();
- buffer.get(bytes);
- return bytes;
+ if (tDigest instanceof PercentileTDigestAccumulator) {
+ // Preserve the legacy two-level raw flush before final compression.
Pending serialized state answers
+ // this count from cached metadata, so stored reads do not acquire an
extra lossy compression pass.
+ tDigest.centroidCount();
+ return ((PercentileTDigestAccumulator) tDigest).serialize();
}
-
- int defaultCapacity = getDefaultCentroidCapacity(tDigest.compression());
- int legacyCapacity =
getLegacyDefaultCentroidCapacity(tDigest.compression());
- // centroidCount() drains pending values at the digest's internal
(two-level) compression, but does not perform
- // the final, destructive compression used by asBytes(). It therefore
gives us a safe bound for digests
- // deserialized from the compact encoding with an explicitly enlarged
main array while still allowing asBytes()
- // to run once.
- int currentCentroidCount = tDigest.centroidCount();
- long sizeBound = Math.min(Math.max(0L, tDigest.size()),
Math.max(defaultCapacity, legacyCapacity));
- int maxCentroids = Math.max(currentCentroidCount, (int) sizeBound);
+ // centroidCount flushes buffered input. Boundary repair can add at most
two centroids; compression and
+ // total weight are not allocation sizes, so extreme historical settings
need no enormous scratch array.
+ int maxCentroids = Math.addExact(tDigest.centroidCount(), 2);
Review Comment:
**[code-review]** Generic serialize() sizes and hard-limits the buffer from
an undocumented centroidCount()+2 bound instead of the SPI's documented
maxSerializedByteSize()
prepareScratchBuffer sets limit(VERBOSE_HEADER_SIZE +
16*(centroidCount()+2)) before tDigest.asBytes(buffer). The public SPI gives
centroidCount() no flush/bound contract and documents the asBytes bound on
maxSerializedByteSize(), which serialize() ignores. A conforming third-party
TDigest holding buffered values (or writing a larger encoding) hits
BufferOverflowException inside ObjectSerDeUtils.TDIGEST_SER_DE.serialize at
query time; an implementation using absolute puts yields a zero-length array
and BufferUnderflowException at the getInt() that follows. Fix: size from
max(requiredCapacity, tDigest.maxSerializedByteSize()) or document the bound on
the SPI.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -216,17 +243,27 @@ public void add(TDigest other) {
addAccumulator((PercentileTDigestAccumulator) 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.hasValidStatistics()) {
+ materializePendingSerializedTDigest();
+ SerializedTDigestInput input = getSerializedTDigestInput();
+ input.reset(ByteBuffer.wrap(TDigestUtils.serialize(other)), false);
+ addSerializedTDigest(input);
return;
}
-
- Collection<Centroid> centroids = other.centroids();
materializePendingSerializedTDigest();
- for (Centroid centroid : centroids) {
- bufferIncomingCentroid(centroid.mean(), centroid.count());
+ if (_legacyDegraded) {
+ _legacySerializedDirty = true;
+ _totalWeight += other.getTotalWeight();
Review Comment:
**[code-review]** Legacy-degraded accumulation paths skip checkTotalWeight,
so an overflowed/NaN total is serialized and makes the bytes unreadable one hop
later
add(double)/add(double,double)/add(TDigest)/addAccumulator/mergeLegacyDegradedInput
all do `_totalWeight += ...` with no checkTotalWeight on the degraded branch
(every non-degraded path checks). Merging two degraded payloads each carrying
~1e308 historical weight overflows to Infinity; toLegacyDegradedBytes writes it
as the single centroid weight, and TDigestUtils.inspectSerialized then throws
'Invalid TDigest centroid weight: Infinity' on every subsequent read - failing
at broker reduce or permanently poisoning a rolled-up stored column instead of
failing fast at merge time. Low likelihood, but the checked paths fail loudly
where this one defers the error into stored bytes.
##########
pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/PercentileTDigestAggregationFunctionTest.java:
##########
@@ -1005,59 +1015,32 @@ public void testReducerMergeStructureAndAccuracy(int
fanIn, int compression, Red
sourceValues[valueIndex] = value;
}
sources[digestIndex] = createReducerSource(sourceValues, compression,
reducerInput);
- referenceSources[digestIndex] =
createReferenceReducerSource(sourceValues, compression, reducerInput);
}
PercentileTDigestAggregationFunction function =
new PercentileTDigestAggregationFunction(EXPRESSION, 75.0,
compression, false);
- TDigest result = TDigest.createMergingDigest(compression);
- TDigest referenceResult = TDigest.createMergingDigest(compression);
- AssertionError referenceFailure = null;
+ TDigest result = TDigestUtils.createMergingDigest(compression);
for (int sourceIndex : mergeOrder.sourceIndexes(fanIn)) {
result = function.merge(result, sources[sourceIndex]);
- if (referenceFailure == null) {
- try {
- if (referenceResult.size() == 0L) {
- referenceResult = referenceSources[sourceIndex];
- } else {
- referenceResult.add(referenceSources[sourceIndex]);
- }
- } catch (AssertionError e) {
- // t-digest 3.3 can violate its own singleton-endpoint assertion
when many duplicate-valued digests merge.
- referenceFailure = e;
- }
- }
}
Arrays.sort(rawValues);
String caseDescription =
fanIn + "/" + compression + "/" + distribution + "/" + mergeOrder +
"/" + reducerInput;
- // Interpolation between large equal-valued centroid runs can land inside
the empirical CDF jump. Keep a fixed
- // envelope for that discontinuous distribution instead of requiring every
merge order to match the reference's
- // particular centroid partition.
- double frozenRankError = compression == 20 || distribution ==
ReducerDistribution.DUPLICATE_HEAVY ? 0.06 : 0.02;
- double[] referenceRankErrors = null;
- if (referenceFailure == null) {
- try {
- referenceRankErrors = getReducerRankErrors(referenceResult, rawValues,
caseDescription + "/reference");
- } catch (AssertionError e) {
- referenceFailure = e;
- }
- }
- double[] maxRankErrors = new double[6];
- if (referenceFailure == null) {
- for (int i = 0; i < referenceRankErrors.length; i++) {
- maxRankErrors[i] = Math.max(frozenRankError, referenceRankErrors[i] +
1e-12);
- }
- } else {
- Arrays.fill(maxRankErrors, compression == 20 ? 0.25 : compression == 100
? 0.06 : 0.02);
+ // The independent 3.3 K1 corpus uses the worst of the six merge/input
orders for each quantile, allowing
+ // one raw observation of sampling slack. It was generated with -da
because 3.3 asserts on duplicate boundaries.
+ double[] referenceRankErrors = LEGACY_REDUCER_RANK_ERRORS.get().get(fanIn
+ "/" + compression + "/" + distribution);
Review Comment:
**[code-review]** Reducer accuracy envelope weakened: every merge order is
now bounded by the worst order's CSV rank error instead of its own ~0.06 cap
The CSV oracle stores the max rank error across all 6 merge-order/input
combinations per (fanIn/compression/distribution) cell, and each combination is
tested against that worst value +1/n. At 32/20/DUPLICATE_HEAVY the p50 entry is
0.2158 (128/20: 0.1954), driven by one pathological order. A future K1-kernel
regression degrading the ORIGINAL/SERVER_LOCAL order's p50 error to ~0.20 -
which the old per-order 0.06 envelope caught - now passes silently, so
order-specific accuracy regressions up to ~3.5x the previous ceiling are
invisible to this test.
##########
compatibility-verifier/tdigest-compatibility/LegacyReader.java:
##########
@@ -0,0 +1,99 @@
+/**
+ * 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.
+ */
+
+import com.tdunning.math.stats.Centroid;
+import com.tdunning.math.stats.MergingDigest;
+import com.tdunning.math.stats.TDigest;
+import java.nio.ByteBuffer;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+
+
+/// Runs only against one legacy jar in a separate JVM; Pinot classes are
deliberately absent from its classpath.
+/// Fractional endpoint masses below one and singleton masses between one and
two are excluded: legacy 3.3 cannot
+/// recompress them without changing mass. Those preserved-byte limitations
are tested by Pinot's unit tests.
+public final class LegacyReader {
+ private LegacyReader() {
+ }
+
+ public static void main(String[] args)
+ throws Exception {
+ if (!LegacyReader.class.desiredAssertionStatus()) {
+ throw new IllegalStateException("The legacy compatibility gate requires
-ea");
+ }
+ Path directory = Path.of(args[0]);
+ List<String> manifest =
Files.readAllLines(directory.resolve("manifest.tsv"));
+ assert manifest.size() == 37 : "Missing compatibility fixtures";
+ for (String line : manifest) {
+ String[] fields = line.split("\t");
+ try {
+ TDigest digest =
MergingDigest.fromBytes(ByteBuffer.wrap(Files.readAllBytes(
+ directory.resolve(fields[0] + ".bin"))));
+ long size = Long.parseLong(fields[1]);
+ double min = Double.parseDouble(fields[2]);
+ double max = Double.parseDouble(fields[3]);
+ double compression = Double.parseDouble(fields[4]);
+ verify(digest, size, min, max, compression, true);
+ digest.add(1);
+ digest.compress();
+ verify(digest, size + 1, Math.min(min, 1), Math.max(max, 1),
compression, false);
+ if (args[1].equals("3.3")) {
+ List<Centroid> centroids = new ArrayList<>(digest.centroids());
+ if (centroids.size() > 1) {
+ assert centroids.get(0).count() == 1 : "First centroid must have
unit weight";
+ assert centroids.get(centroids.size() - 1).count() == 1 : "Last
centroid must have unit weight";
+ }
+ }
+ ByteBuffer rewritten = ByteBuffer.allocate(digest.byteSize());
+ digest.asBytes(rewritten);
+ verify(MergingDigest.fromBytes(ByteBuffer.wrap(rewritten.array())),
size + 1,
+ Math.min(min, 1), Math.max(max, 1), compression, false);
+ } catch (AssertionError | Exception e) {
+ throw new AssertionError("Legacy " + args[1] + " reader failed for " +
fields[0], e);
+ }
+ }
+ System.out.println("Legacy " + args[1] + ": " + manifest.size() + "
read/add/compress/rewrite cases passed");
+ }
+
+ private static void verify(TDigest digest, long size, double min, double
max, double compression,
Review Comment:
**[code-review]** The legacy 3.2/3.3 reader CI gate never validates quantile
values, so byte corruption that preserves monotonicity passes
verify() asserts only that quantiles are finite and monotone, min/max match
the header doubles, and size matches summed weights; centroid-mean fidelity is
unchecked. A serializer bug in makeLegacyCompatible/toCapacityPreservingBytes
that shifts, scales, or clamps interior centroid means inside [min,max]
produces bytes with correct size/header and monotone quantiles - the gate
prints all 37 cases passed for both tdunning versions while legacy servers
would compute arbitrarily wrong percentiles from Pinot-written bytes. Suggest
asserting a few quantiles against expected values recorded in the manifest.
--
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]
