xiangfu0 commented on code in PR #19795:
URL: https://github.com/apache/pinot/pull/19795#discussion_r4224156227
##########
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:
Fixed in
[4818fb8fbc](https://github.com/apache/pinot/commit/4818fb8fbc79d81de17ce0c5bd0070eb97f300df).
The independent 3.3 reference now contains all 270
fan-in/compression/distribution/order/input cases, each with its own six
quantile errors plus only the existing 1/N allowance. This stricter gate
exposed a duplicate-centroid plateau defect in the kernel; fixing the plateau
restores all 270 cases without widening tolerances.
##########
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:
Fixed in
[4818fb8fbc](https://github.com/apache/pinot/commit/4818fb8fbc79d81de17ce0c5bd0070eb97f300df).
The real-reader manifest records native quantiles before serialization for 25
ordinary continuous/empty/singleton/huge-value fixtures; each actual 3.2/3.3
reader compares all six quantiles numerically. Weighted/repair fixtures retain
mass, header, monotonicity and add/compress checks because their interpolation
intentionally differs. A negative-control experiment scaling only interior
means by 0.9 was rejected by both real readers at P25. All 38 fixtures per
version pass.
##########
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();
+ } else {
+ checkTotalWeight(_totalWeight + _incomingWeight + _numRawValues +
other.getTotalWeight());
+ for (Centroid centroid : other.centroids()) {
+ double weight = centroid.weight();
+ if (!(weight > 0.0) || !Double.isFinite(weight) ||
Double.isNaN(centroid.mean())) {
+ throw new IllegalArgumentException("Invalid TDigest centroid");
Review Comment:
Fixed in
[4818fb8fbc](https://github.com/apache/pinot/commit/4818fb8fbc79d81de17ce0c5bd0070eb97f300df).
Object merge skips zero-weight centroids before mean validation, matching
serialized input, including an unused NaN mean. The non-finite star-tree
wrapper follows the same rule. A regression compares these object/byte
semantics.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -409,17 +409,36 @@ private void
mergeSerializedTDigest(SerializedTDigestInput input) {
_totalWeight = input._totalWeight;
_min = input._min;
_max = input._max;
+ // Stored centroids already went through the writer's public
compression. Reading or repairing their
+ // endpoints needs another K1 pass only when the output compression
differs from the stored compression.
+ _publiclyCompressed = _compression == input._compression;
+ }
+ return;
+ }
+ if (direct) {
+ flush();
+ boolean runBackwards = (_mergeCount++ & 1) != 0;
+ if (!mergeSorted(input._means, input._weights, input._numCentroids,
input._totalWeight,
+ getIncrementalCompression(), runBackwards)) {
+ return;
}
} else {
bufferIncomingCentroids(input._means, input._weights,
input._numCentroids);
- _min = Math.min(_min, input._min);
- _max = Math.max(_max, input._max);
}
+ _min = Math.min(_min, input._min);
Review Comment:
Fixed in
[4818fb8fbc](https://github.com/apache/pinot/commit/4818fb8fbc79d81de17ce0c5bd0070eb97f300df).
Buffered zero-mass inputs do not contribute header extrema. Pending zero-mass
min/max and centroid count use the same empty view before and after
materialization. Regressions cover both merge orderings and unchanged healthy
quantiles.
--
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]
