xiangfu0 commented on code in PR #19795:
URL: https://github.com/apache/pinot/pull/19795#discussion_r4236116586
##########
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileSmartTDigestAggregationFunction.java:
##########
@@ -365,7 +361,7 @@ public Double extractFinalResult(@Nullable Object
intermediateResult) {
TDigest tDigest = (TDigest) intermediateResult;
// An empty digest holds the same state as the empty value list below,
so the two branches must answer alike:
// which one a query lands in depends only on whether the accumulator
crossed the conversion threshold.
- if (_nullHandlingEnabled && tDigest.size() == 0L) {
+ if (_nullHandlingEnabled && tDigest.isEmpty()) {
Review Comment:
Fixed: an empty TDigest intermediate now returns null when null handling is
enabled and DEFAULT_FINAL_RESULT when disabled, matching the empty-list path.
The existing SmartTDigest suite covers both settings.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TDigestUtils.java:
##########
@@ -86,44 +85,65 @@ 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);
- int requiredCapacity = Math.addExact(VERBOSE_HEADER_SIZE,
- Math.multiplyExact(VERBOSE_CENTROID_SIZE, maxCentroids));
+ byte[] retained =
PercentileTDigestValueAggregator.getRetainedHistoricalBytes(tDigest);
Review Comment:
Removed the codec-to-aggregator dependency and static retention special
case. The six related types now share customobject.tdigest; each digest owns
serialize() and historical retention. PercentileTDigestValueAggregator is a
thin adapter, and boundary/provenance helpers are package-private.
##########
compatibility-verifier/tdigest-compatibility/generate-rank-errors.sh:
##########
@@ -0,0 +1,67 @@
+#!/usr/bin/env bash
+#
+# 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.
+#
+
+# Regenerate with the pinned legacy utility and 3.3 jar, then compare all case
keys, inputs and rank errors.
+# Use --update to replace the checked-in reference after intentionally
changing the seeded input corpus.
+set -euo pipefail
+# Match the unit-test workflow's Maven Central retry and connection settings,
including standalone runs.
+export MAVEN_OPTS="${MAVEN_OPTS:-}
-Dmaven.wagon.httpconnectionManager.ttlSeconds=25 \
+-Dmaven.wagon.http.retryHandler.count=30 -Dhttp.keepAlive=false
-Dmaven.wagon.http.pool=false"
+TDIGEST_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)"
+TDIGEST_BASELINE=f818bae6d75a91228eb3aac95c921bca4aeb2128
+TDIGEST_WORK="$TDIGEST_ROOT/pinot-segment-local/target/tdigest-rank-oracle"
+TDIGEST_JAR_DIR="$TDIGEST_ROOT/pinot-segment-local/target/tdigest-legacy-reader"
+TDIGEST_JAR="$TDIGEST_JAR_DIR/t-digest-3.3.jar"
+TDIGEST_REFERENCE="$TDIGEST_ROOT/pinot-core/src/test/resources/data/tdigest-3.3-k1-rank-errors.csv"
+mkdir -p "$TDIGEST_WORK" "$TDIGEST_JAR_DIR"
+if ! git -C "$TDIGEST_ROOT" cat-file -e "$TDIGEST_BASELINE^{commit}"
2>/dev/null; then
+ # CI checks out shallow history. Fetch only the immutable public baseline;
leave the worktree and POMs alone.
+ for TDIGEST_ATTEMPT in 1 2 3; do
+ if git -C "$TDIGEST_ROOT" -c http.lowSpeedLimit=1 -c http.lowSpeedTime=30 \
+ fetch --no-tags --depth=1 https://github.com/apache/pinot.git
"$TDIGEST_BASELINE"; then
Review Comment:
Fixed: the oracle script checks is-shallow-repository and uses --depth=1
only for an already shallow checkout. Five disposable Git fixtures verify full
clones stay full, shallow clones stay shallow, and an available baseline does
not fetch.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/PercentileTDigestValueAggregator.java:
##########
@@ -284,52 +270,113 @@ 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);
}
- invalidateCaches();
+ requireMutable();
+ checkTotalWeight(getTotalWeight() + weight);
if (value == Double.NEGATIVE_INFINITY) {
- _negativeInfinityWeight = Math.addExact(_negativeInfinityWeight,
weight);
+ _negativeInfinityWeight += weight;
} else if (value == Double.POSITIVE_INFINITY) {
- _positiveInfinityWeight = Math.addExact(_positiveInfinityWeight,
weight);
+ _positiveInfinityWeight += weight;
} else {
_finiteDigest.add(value, weight);
}
+ mutated();
}
@Override
public void add(TDigest other) {
+ if (other.isEmpty()) {
+ return;
+ }
+ requireMutable();
+ if (!other.hasValidStatistics() && !isEmpty()) {
+ throw corruptedMutation();
+ }
+ if (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();
- invalidateCaches();
- if (wrapped._finiteDigest.size() > 0L) {
- _finiteDigest.add(List.of(wrapped._finiteDigest));
+ double negativeInfinityWeight = wrapped._negativeInfinityWeight;
+ double positiveInfinityWeight = wrapped._positiveInfinityWeight;
+ if (!wrapped._finiteDigest.isEmpty()) {
+ _finiteDigest.add(wrapped._finiteDigest);
Review Comment:
Fixed: ordinary buffered inputs update extrema for each consumed centroid
prefix, normalizing existing boundaries before widening those bounds; oversized
or unsupported-endpoint sources use a safe whole-source primitive merge. Final
source bounds are never applied to an incomplete prefix. Your exact BYTES ->
merge -> immediate serialize -> reread regression now has mass 9,999, bounds
[1, 5,000], valid statistics and q(0.198)=990.9010000000001. A seeded ordinary
straddle regression also checks mass and first moment, and its negative control
reproduces the old corruption.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TDigestUtils.java:
##########
@@ -86,44 +85,65 @@ 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();
Review Comment:
Fixed: serialize is now the native polymorphic contract, with no
centroidCount preflush or encode/reinspect bridge. SQL/reduction mode
compresses pending input directly at public compression; Type-10 decoding
preserves that mode. Added Random(9), 16 x 2,000 uniform values, compression
100 with serialization before querying P75. The independent 270-case reducer
corpus also serializes before querying its result.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/PercentileTDigestValueAggregator.java:
##########
@@ -152,130 +166,102 @@ private static NonFiniteAwareTDigest
asNonFiniteAware(TDigest value) {
}
private void updateMaxByteSize(TDigest value) {
- long defaultCapacity = getDefaultCentroidCapacity(value.compression());
- long maxCentroids = Math.min(value.size(), defaultCapacity);
- _maxByteSize = Math.max(_maxByteSize, getMaxVerboseByteSize(maxCentroids));
- }
-
- private static int getMaxVerboseByteSize(long centroidCount) {
- long maxCentroids = Math.max(centroidCount, 0L);
- return Math.toIntExact(Math.addExact(VERBOSE_HEADER_SIZE,
- Math.multiplyExact(VERBOSE_CENTROID_SIZE, maxCentroids)));
- }
-
- private static long getDefaultCentroidCapacity(double compression) {
- double normalizedCompression = Math.max(MIN_COMPRESSION, compression);
- int padding = normalizedCompression < 30.0 ?
LOW_COMPRESSION_CAPACITY_PADDING
- : DEFAULT_CENTROID_CAPACITY_PADDING;
- long defaultCapacity = (long) Math.ceil(2.0 * normalizedCompression +
padding);
- long legacyCapacity = Math.addExact(Math.multiplyExact(2L, (long)
Math.ceil(compression)),
- DEFAULT_CENTROID_CAPACITY_PADDING);
- return Math.max(defaultCapacity, legacyCapacity);
+ _maxByteSize = Math.max(_maxByteSize, value.maxSerializedByteSize());
Review Comment:
Fixed: fresh compression-100 state keeps a 3,392-byte compatible-output
bound, including the 1,049-scalar case, without flushing on each row. Temporary
raw/incoming buffer lengths no longer become a permanent Star-tree sizing
maximum. Oversized historical bytes retain a separate safe bound; tests verify
the bound covers actual serialization.
--
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]
