xiangfu0 commented on code in PR #19795:
URL: https://github.com/apache/pinot/pull/19795#discussion_r4217136996
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -696,34 +685,59 @@ public double getMin() {
@Override
public double getMax() {
if (_pendingSerializedTDigest != null) {
- return
ByteBuffer.wrap(_pendingSerializedTDigest).getDouble(Integer.BYTES +
Double.BYTES);
+ return _pendingSerializedMetadata.max();
}
flush();
return _max;
}
- byte[] serialize() {
+ public byte[] serialize() {
+ if (_legacyDegraded) {
+ if (_pendingSerializedTDigest != null) {
+ return _pendingSerializedTDigest.clone();
+ }
+ if (_totalWeight == _legacySerializedTotalWeight && _min ==
_legacySerializedMin
+ && _max == _legacySerializedMax) {
+ return _legacySerializedBytes.clone();
+ }
+ // A merge with a historical invalid distribution remains unknown. Keep
that explicit in the legacy
+ // encoding instead of passing its invalid centroids into the normal
sorted K1 merge.
+ return toLegacyDegradedBytes();
+ }
if (_pendingSerializedTDigest != null) {
- getSerializedTotalWeight(_pendingSerializedTDigest);
- if (ByteBuffer.wrap(_pendingSerializedTDigest).getInt() ==
VERBOSE_ENCODING) {
- byte[] serialized =
TDigestUtils.makeLegacyCompatible(_pendingSerializedTDigest);
- return serialized == _pendingSerializedTDigest ? serialized.clone() :
serialized;
+ if (hasWeightedBoundaryCentroids(_pendingSerializedTDigest,
_pendingSerializedMetadata.centroidCount())) {
+ materializePendingSerializedTDigest();
Review Comment:
Initialized stored centroids are already finalized when their compression
matches the destination, so boundary repair no longer runs another K1 pass. A
regression retains all 200 input centroids plus only the two repaired
endpoints. Differing output compression still recompresses, preserving
configured rollup behavior; an empty merge retains all 80 compact centroids.
Addressed in
[1d9c042a3d](https://github.com/apache/pinot/commit/1d9c042a3d66a4462d7fe991b41f496f4d3dcb7f).
All 473 targeted tests and 322 legacy reader cases pass.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -1452,42 +1468,58 @@ static final class SerializedTDigestInput {
private double _totalWeight;
private boolean _decoded;
- void reset(byte[] bytes) {
- _retainedBytes = null;
- ByteBuffer input = ByteBuffer.wrap(bytes);
- int encoding = input.getInt();
- if (encoding != VERBOSE_ENCODING && encoding != SMALL_ENCODING) {
- throw new IllegalStateException("Invalid format for serialized
histogram");
- }
+ private SerializedTDigestMetadata _metadata;
+
+ public void reset(byte[] bytes) {
+ reset(bytes, TDigestUtils.inspectSerialized(ByteBuffer.wrap(bytes)));
Review Comment:
Reset now parses only the fixed-size header. Normal merged input validates
and fills reusable centroid arrays in one shared pass, including
totals/ordering/bounds flags. Standalone inspection is reserved for retaining
lazy pending input; its metadata is cached and reused. Group fanout shares the
same decoded input.
Addressed in
[1d9c042a3d](https://github.com/apache/pinot/commit/1d9c042a3d66a4462d7fe991b41f496f4d3dcb7f).
All 473 targeted tests and 322 legacy reader cases pass.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TDigestUtils.java:
##########
@@ -91,39 +77,29 @@ public static byte[] serialize(TDigest tDigest) {
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);
+ 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();
+ byte[] bytes = ((PercentileTDigestAccumulator) tDigest).serialize();
+ if (scratchBuffer != null && !scratchBuffer.isReadOnly() &&
scratchBuffer.capacity() >= bytes.length) {
+ scratchBuffer.clear();
Review Comment:
Removed the discarded scratch-buffer copy from Pinot-owned serialization.
The legacy raw preflush is retained because the independent accuracy oracle
requires it. Javadoc and the scratch regression now verify the actual
owned-buffer behavior.
Addressed in
[1d9c042a3d](https://github.com/apache/pinot/commit/1d9c042a3d66a4462d7fe991b41f496f4d3dcb7f).
All 473 targeted tests and 322 legacy reader cases pass.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/PercentileTDigestValueAggregator.java:
##########
@@ -291,45 +311,50 @@ public void add(double value, int weight) {
if (weight <= 0) {
throw new IllegalArgumentException("TDigest weight must be positive: "
+ weight);
}
+ checkTotalWeight(getTotalWeight() + weight);
invalidateCaches();
if (value == Double.NEGATIVE_INFINITY) {
- _negativeInfinityWeight = Math.addExact(_negativeInfinityWeight,
weight);
+ _negativeInfinityWeight = _negativeInfinityWeight + weight;
} else if (value == Double.POSITIVE_INFINITY) {
- _positiveInfinityWeight = Math.addExact(_positiveInfinityWeight,
weight);
+ _positiveInfinityWeight = _positiveInfinityWeight + weight;
} else {
_finiteDigest.add(value, weight);
}
}
@Override
public void add(TDigest other) {
+ 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.
+ double negativeInfinityWeight = wrapped._negativeInfinityWeight;
+ double positiveInfinityWeight = wrapped._positiveInfinityWeight;
+ // Merging shared state preserves double-precision weights without
narrowing through Centroid.count().
wrapped.invalidateCaches();
invalidateCaches();
- if (wrapped._finiteDigest.size() > 0L) {
+ if (wrapped._finiteDigest.getTotalWeight() > 0.0) {
_finiteDigest.add(List.of(wrapped._finiteDigest));
}
- _negativeInfinityWeight = Math.addExact(_negativeInfinityWeight,
negativeInfinityWeight);
- _positiveInfinityWeight = Math.addExact(_positiveInfinityWeight,
positiveInfinityWeight);
+ _negativeInfinityWeight = _negativeInfinityWeight +
negativeInfinityWeight;
+ _positiveInfinityWeight = _positiveInfinityWeight +
positiveInfinityWeight;
return;
}
- if (other.size() == 0L) {
+ if (other.getTotalWeight() == 0.0) {
return;
}
invalidateCaches();
if (Double.isFinite(other.getMin()) && Double.isFinite(other.getMax())) {
_finiteDigest.add(List.of(other));
return;
}
- for (Centroid centroid : other.centroids()) {
- add(centroid.mean(), centroid.count());
- }
+ // Replay the double-precision wire weights: Centroid.count() truncates
positive fractional mass to zero.
+ ByteBuffer encoded = ByteBuffer.allocate(other.byteSize());
Review Comment:
Normal generic merges now consume double-weight centroids directly, so
neither accumulator nor Infinity wrapper calls byteSize/asBytes to serialize a
source twice. The generic Infinity test verifies those methods are never
invoked. A degraded source uses the exceptional single-serialization path to
retain its unknown state.
Addressed in
[1d9c042a3d](https://github.com/apache/pinot/commit/1d9c042a3d66a4462d7fe991b41f496f4d3dcb7f).
All 473 targeted tests and 322 legacy reader cases pass.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/TDigest.java:
##########
@@ -0,0 +1,71 @@
+/**
+ * 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;
+
+import java.nio.ByteBuffer;
+import java.util.Collection;
+import java.util.List;
+
+/// Pinot's mutable t-digest contract using the existing K1 aggregation
behavior and legacy byte encodings.
+///
+/// Implementations are not safe for concurrent mutation; callers must
serialize access to each aggregation state.
+public abstract class TDigest {
+ public abstract void add(double value);
+
+ public abstract void add(double value, int weight);
+
+ public abstract void add(TDigest other);
+
+ public abstract void add(List<? extends TDigest> others);
+
+ public abstract void compress();
+
+ public abstract long size();
+
+ /// Returns centroid mass without narrowing fractional or large weights to
the legacy long size view.
+ public double getTotalWeight() {
+ return size();
+ }
+
+ public abstract double cdf(double value);
+
+ public abstract double quantile(double quantile);
+
+ public abstract Collection<Centroid> centroids();
+
+ public abstract double compression();
+
+ public abstract int byteSize();
+
+ public abstract int smallByteSize();
+
+ public abstract void asBytes(ByteBuffer buffer);
+
+ public abstract void asSmallBytes(ByteBuffer buffer);
+
+ public abstract int centroidCount();
+
+ public abstract double getMin();
+
+ public abstract double getMax();
+
+ /// Immutable centroid view. Counts retain the legacy int API; serialization
preserves double weights.
+ public record Centroid(double mean, int count) {
Review Comment:
The SPI centroid view now exposes double weight(), and add(value, weight)
accepts double mass. In-memory merges no longer narrow or serialize/reparse
valid centroid weights. Tests reconstruct fractional and large distributions
through the public API and preserve mass above Integer.MAX_VALUE.
Addressed in
[1d9c042a3d](https://github.com/apache/pinot/commit/1d9c042a3d66a4462d7fe991b41f496f4d3dcb7f).
All 473 targeted tests and 322 legacy reader cases pass.
--
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]
