xiangfu0 commented on code in PR #19795: URL: https://github.com/apache/pinot/pull/19795#discussion_r4253010256
########## pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/tdigest/NonFiniteAwareTDigest.java: ########## @@ -0,0 +1,749 @@ +/** + * 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.tdigest; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; +import javax.annotation.Nullable; +import org.apache.pinot.segment.local.customobject.tdigest.TDigestCodec.SerializedTDigestMetadata; + + +/// Stores non-finite values as tail masses while delegating finite values to an owned native digest. +/// Emits legacy t-digest bytes and preserves historical fractional endpoint provenance across merges and copies. +/// Instances are mutable and require external synchronization. +public final class NonFiniteAwareTDigest extends TDigest { + private TDigestAccumulator _finiteDigest; + private double _negativeInfinityWeight; + private double _positiveInfinityWeight; + private List<Centroid> _finiteCentroids; + private byte[] _originalBytes; + private boolean _unrepresentableInfinityTail; + private boolean _pendingHistoricalBoundaryRepair; + private double _historicalMin; + private double _historicalMax; + + public static NonFiniteAwareTDigest forLegacyAggregation(double compression) { + return new NonFiniteAwareTDigest(TDigestAccumulator.forLegacyAggregation(compression), 0.0, 0.0); + } + + private NonFiniteAwareTDigest(TDigestAccumulator finiteDigest, double negativeInfinityWeight, + double positiveInfinityWeight) { + _finiteDigest = finiteDigest; + _negativeInfinityWeight = negativeInfinityWeight; + _positiveInfinityWeight = positiveInfinityWeight; + } + + /// Snapshots the aggregation without flushing or compressing the source's buffered inputs. + public NonFiniteAwareTDigest copy() { + NonFiniteAwareTDigest copy = + new NonFiniteAwareTDigest(_finiteDigest.copy(), _negativeInfinityWeight, _positiveInfinityWeight); + copy._originalBytes = _originalBytes == null ? null : _originalBytes.clone(); Review Comment: Addressed in 564fec0064. Shared the immutable retained bytes in copy/adoption; input and output cloning remains. The local ownership probe verifies caller isolation and independent mutation of both copies. ########## pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/tdigest/TDigest.java: ########## @@ -0,0 +1,146 @@ +/** + * 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.tdigest; + +import java.util.Collection; +import java.util.List; + +/// Mutable percentile digest base for Pinot aggregation and legacy t-digest byte encodings. +/// Implementations require external synchronization. Historical corrupted payloads can be retained unchanged, +/// but cannot be mutated or mixed with healthy state. +public abstract class TDigest { + public abstract void add(double value); + + public void add(double value, int weight) { + add(value, (double) weight); + } + + /// Adds positive finite mass; rejects total-weight overflow. Fractional interior mass is supported, but + /// serialization rejects fresh endpoint mass below one or singleton mass between one and two: legacy readers + /// require unit endpoints. Thus `add(5, 0.5)` is accepted but cannot be serialized alone. + public abstract void add(double value, double weight); + + /// Merges an exclusively owned source; implementations may compress that mutable source in place. + public abstract void add(TDigest other); + + public void add(List<? extends TDigest> others) { + for (TDigest other : others) { + add(other); + } + } + + public abstract void compress(); + + /// Returns the legacy long size view, saturating when the precise mass exceeds its range. + public long size() { + return (long) getTotalWeight(); + } + + /// Returns mass without narrowing fractional or large weights. Add/merge reject overflow of a valid total. + /// Invalid historical payloads may report their original negative or non-finite total, but cannot be merged. + public abstract double getTotalWeight(); + + /// Returns false when historical numerical corruption leaves the distribution unknown. + public boolean hasValidStatistics() { + return true; + } + + /// An unknown distribution remains present even when its historical weights sum to zero. + public boolean isEmpty() { + return getTotalWeight() == 0.0 && hasValidStatistics(); + } + + public abstract double cdf(double value); + + public abstract double quantile(double quantile); + + /// Returns immutable centroid values with their precise double weights. + /// Reading this view may flush buffered inputs without applying the final public compression. + public abstract Collection<Centroid> centroids(); + + public abstract double compression(); + + /// Serializes one independently owned legacy-compatible payload. Implementations own their compression policy. + /// Unchanged historical encodings may be retained byte-exactly; fresh unrepresentable boundaries are rejected. + public abstract byte[] serialize(); + + /// Finalizes an independently owned result without changing a live aggregation that may receive more inputs. + protected abstract byte[] serializeSnapshot(); + + /// Bounds independently owned serialization without requiring a two-step size/write protocol. + /// Mutable implementations can override this to avoid flushing buffered inputs. + public int maxSerializedByteSize() { + return serialize().length; + } + + public abstract int centroidCount(); + + public abstract double getMin(); + + public abstract double getMax(); + + /// Returns the inherited unsupported fractional endpoint mean, or NaN. Mergers propagate this provenance + /// from validated legacy bytes without granting permission to new fractional extrema. + double getHistoricalFractionalBoundaryMean(boolean lowerBoundary) { + return Double.NaN; + } + + /// Extremal fresh fractional input means, retained across merges to prevent reusing historical permission for + /// new mass at the same endpoint. NaN means this digest has received no fresh fractional inputs. + double getFreshFractionalBoundaryMean(boolean lowerBoundary) { + return Double.NaN; + } + + /// Immutable centroid value preserving fractional and large mass. + public static final class Centroid { + private final double _mean; + private final double _weight; Review Comment: Addressed in 564fec0064. Replaced Centroid with a record, removing 33 net lines. A local probe checked 14,641 equality pairs plus matching hash codes, including NaNs and signed zero. -- 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]
