xiangfu0 commented on code in PR #19795: URL: https://github.com/apache/pinot/pull/19795#discussion_r4244255668
########## pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/tdigest/PercentileTDigestAccumulator.java: ########## @@ -0,0 +1,1725 @@ +/** + * 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. + */ +/* + * Portions of the CDF, quantile, weighted-average and K1 weight-limit implementation are adapted from + * MergingDigest, AbstractTDigest and ScaleFunction in t-digest 3.3 (https://github.com/tdunning/t-digest). + * Modified for Pinot's primitive centroid storage, lazy serialized inputs and legacy boundary handling. + * Original license notice for these portions: + * + * Licensed to Ted Dunning 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.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.List; +import org.apache.pinot.segment.local.customobject.tdigest.TDigestCodec.SerializedTDigestMetadata; + + + +/// Accumulates raw values and serialized TDigests into primitive centroids using Pinot's K1 weight-limit rule. +/// +/// Raw values are sorted in small batches; serialized centroids retain their existing sorted order. Both are +/// linearly merged with accumulated centroids. K1 preserves Pinot's existing middle-quantile behavior, and the +/// legacy verbose and small encodings preserve stored data and mixed-version wire compatibility. +/// +/// Serialized state keeps its first digest pending and allocates centroid buffers only when another input must be +/// merged or the digest is queried. Raw-value buffers remain unallocated until a raw value is added. +/// +/// Historical corrupted payloads remain opaque and byte-exact, with NaN statistics. Their reported historical mass +/// may be invalid; they cannot be mutated or mixed with another distribution. Such an operation fails before +/// changing aggregate mass, because neither invented statistics nor synthetic NaN bytes are safe for old readers. +/// Unchanged historical fractional boundaries retain their original bytes, including when configured compression +/// differs. After a merge, inherited fractional endpoints retain precise verbose weights and finite statistics when +/// unit endpoints cannot preserve their mass. These keep the historical t-digest 3.3 assertion-enabled reader and +/// recompression limitation. Newly introduced unsupported fractional endpoints are rejected. +/// +/// Instances are externally serialized per aggregation or group key and are not safe for concurrent mutation. +public final class PercentileTDigestAccumulator extends TDigest { + private static final int MIN_RAW_BUFFER_SIZE = 256; + private static final int MAX_RAW_BUFFER_SIZE = 10_000; + private static final int TWO_LEVEL_COMPRESSION_MULTIPLIER = 2; + private static final int MIN_INCREMENTAL_COMPRESSION = 50; + + private final double _compression; + private final boolean _useTwoLevelRawCompression; + private final double _workingCompression; + private final double _publicMaxWeightScale; + private final double _workingMaxWeightScale; + private final int _centroidCapacity; + private final int _pendingCentroidCapacity; + private double[] _rawValues; + private double[] _centroidMeans; + private double[] _centroidWeights; + private double[] _outputMeans; + private double[] _outputWeights; + private double[] _incomingMeans; + private double[] _incomingWeights; + private double[] _sortedIncomingMeans; + private double[] _sortedIncomingWeights; + private SerializedTDigestInput _serializedTDigestInput; + private byte[] _pendingSerializedTDigest; + private byte[] _originalFractionalBytes; + private byte[] _originalBoundaryBytes; + private double _historicalFractionalMinMean = Double.NaN; + private double _historicalFractionalMaxMean = Double.NaN; + private double _freshFractionalMinMean = Double.NaN; + private double _freshFractionalMaxMean = Double.NaN; + private SerializedTDigestMetadata _pendingSerializedMetadata; + private boolean _legacyDegraded; + private boolean _hasFractionalWeights; + private boolean _deferBoundaryNormalization; + + private int _numRawValues; + private int _numCentroids; + private int _numIncomingCentroids; + private int _mergeCount; + private boolean _publiclyCompressed; + private boolean _incomingCentroidsSorted = true; + private double _totalWeight; + private double _incomingWeight; + private double _min = Double.POSITIVE_INFINITY; + private double _max = Double.NEGATIVE_INFINITY; + + public PercentileTDigestAccumulator(int compression) { + this((double) compression, true, true); + } + + public PercentileTDigestAccumulator(double compression) { + this(compression, true, true); + } + + private PercentileTDigestAccumulator(double compression, boolean allocateRawBuffer, boolean allocateMergeBuffers) { + this(compression, allocateRawBuffer, allocateMergeBuffers, false); + } + + private PercentileTDigestAccumulator(double compression, boolean allocateRawBuffer, boolean allocateMergeBuffers, + boolean useTwoLevelRawCompression) { + _compression = TDigestCodec.normalizeCompression(compression); + _useTwoLevelRawCompression = useTwoLevelRawCompression; + _workingCompression = TWO_LEVEL_COMPRESSION_MULTIPLIER * _compression; + _publicMaxWeightScale = calculateK1MaxWeightScale(_compression); + _workingMaxWeightScale = calculateK1MaxWeightScale(_workingCompression); + long roundedCompression = (long) Math.ceil(_compression); + _centroidCapacity = TDigestCodec.getDefaultCentroidCapacity(_compression); + _pendingCentroidCapacity = getPendingCentroidCapacity(_centroidCapacity, useTwoLevelRawCompression); + if (allocateRawBuffer) { + _rawValues = new double[getInitialRawBufferSize(roundedCompression)]; + } + if (allocateMergeBuffers) { + int initialCapacity = Math.min(_centroidCapacity, MIN_RAW_BUFFER_SIZE); + _centroidMeans = new double[initialCapacity]; + _centroidWeights = new double[initialCapacity]; + _outputMeans = new double[initialCapacity]; + _outputWeights = new double[initialCapacity]; + } + } + + /// Creates a lazy digest with the legacy two-level compression used by Star-tree and general digest aggregation. + public static PercentileTDigestAccumulator forLegacyAggregation(double compression) { + return new PercentileTDigestAccumulator(compression, false, false, true); + } + + /// Reads and retains one legacy digest, preserving the general aggregation compression policy. + public static PercentileTDigestAccumulator fromBytes(byte[] bytes) { + return fromBytes(ByteBuffer.wrap(bytes)); + } + + /// Reads one payload and advances the buffer past it. + public static PercentileTDigestAccumulator fromBytes(ByteBuffer bytes) { + SerializedTDigestInput input = new SerializedTDigestInput(); + input.reset(bytes); + return fromBytes(input); + } + + /// Retains an already validated input without rescanning its centroid bytes. + public static PercentileTDigestAccumulator fromBytes(SerializedTDigestInput input) { + PercentileTDigestAccumulator digest = forLegacyAggregation(input.getCompression()); + digest.addSerializedTDigest(input); + return digest; + } + + public static PercentileTDigestAccumulator forReduction(double compression) { + return new PercentileTDigestAccumulator(compression, false, false); + } + + /// Snapshots buffered inputs and centroid state without materializing or compressing the source. Mutable buffers + /// are copied; immutable retained bytes and metadata are shared. Merge and sort scratch buffers remain lazy. + PercentileTDigestAccumulator copy() { + PercentileTDigestAccumulator copy = + new PercentileTDigestAccumulator(_compression, false, false, _useTwoLevelRawCompression); + copy._rawValues = _rawValues != null ? _rawValues.clone() : null; + copy._centroidMeans = _numCentroids > 0 ? Arrays.copyOf(_centroidMeans, _numCentroids) : null; + copy._centroidWeights = _numCentroids > 0 ? Arrays.copyOf(_centroidWeights, _numCentroids) : null; + copy._incomingMeans = _numIncomingCentroids > 0 ? Arrays.copyOf(_incomingMeans, _numIncomingCentroids) : null; + copy._incomingWeights = _numIncomingCentroids > 0 ? Arrays.copyOf(_incomingWeights, _numIncomingCentroids) : null; + copy._pendingSerializedTDigest = _pendingSerializedTDigest; + copy._pendingSerializedMetadata = _pendingSerializedMetadata; + copy._originalFractionalBytes = _originalFractionalBytes; + copy._originalBoundaryBytes = _originalBoundaryBytes; + copy._historicalFractionalMinMean = _historicalFractionalMinMean; + copy._historicalFractionalMaxMean = _historicalFractionalMaxMean; + copy._freshFractionalMinMean = _freshFractionalMinMean; + copy._freshFractionalMaxMean = _freshFractionalMaxMean; + copy._legacyDegraded = _legacyDegraded; + copy._hasFractionalWeights = _hasFractionalWeights; + copy._deferBoundaryNormalization = _deferBoundaryNormalization; + copy._numRawValues = _numRawValues; + copy._numCentroids = _numCentroids; + copy._numIncomingCentroids = _numIncomingCentroids; + copy._mergeCount = _mergeCount; + copy._publiclyCompressed = _publiclyCompressed; + copy._incomingCentroidsSorted = _incomingCentroidsSorted; + copy._totalWeight = _totalWeight; + copy._incomingWeight = _incomingWeight; + copy._min = _min; + copy._max = _max; + return copy; + } + + /// Copies primitive centroids without encoding and decoding them. Zero weights are ignored, and unordered + /// means are sorted when initializing a stored, already compressed distribution; caller arrays are not mutated. + /// Subsequent inputs use the normal buffered merge. Positive weights and consistent extrema are required. + public void addCentroids(double[] means, double[] weights, int count, double min, double max, + boolean alreadyCompressed) { + addCentroids(means, weights, count, min, max, alreadyCompressed, false); + } + + /// Adds validated legacy centroids without marking their existing fractional mass as fresh input. + void addHistoricalCentroids(double[] means, double[] weights, int count, double min, double max, + boolean alreadyCompressed) { + addCentroids(means, weights, count, min, max, alreadyCompressed, true); + } + + private void addCentroids(double[] means, double[] weights, int count, double min, double max, + boolean alreadyCompressed, boolean historical) { + requireMutable(); + if (count < 0 || count > means.length || count > weights.length) { + throw new IllegalArgumentException("Invalid TDigest centroid count: " + count); + } + double totalWeight = 0.0; + int positiveCount = 0; + boolean sorted = true; + boolean fractionalWeights = false; + double previousMean = Double.NEGATIVE_INFINITY; + for (int i = 0; i < count; i++) { + double weight = weights[i]; + if (weight == 0.0) { + continue; + } + double mean = means[i]; + if (!(weight > 0.0) || !Double.isFinite(weight) || !(mean >= min && mean <= max)) { + throw new IllegalArgumentException("Invalid TDigest centroid"); + } + totalWeight += weight; + sorted &= mean >= previousMean; + previousMean = mean; + fractionalWeights |= weight != Math.rint(weight); + positiveCount++; + } + checkTotalWeight(getTotalWeight() + totalWeight); + if (positiveCount == 0) { + return; + } + prepareForMutation(); + _originalFractionalBytes = null; + _hasFractionalWeights |= fractionalWeights; + if (!historical && fractionalWeights) { + for (int i = 0; i < count; i++) { + recordFreshFractionalMean(means[i], weights[i]); + } + } + if (hasNoInputs() && alreadyCompressed) { + ensureCentroidCapacity(Math.addExact(positiveCount, 2)); + for (int i = 0; i < count; i++) { + if (weights[i] > 0.0) { + _centroidMeans[_numCentroids] = means[i]; + _centroidWeights[_numCentroids++] = weights[i]; + } + } + if (!sorted) { + sortCentroids(_centroidMeans, _centroidWeights, _numCentroids); + } + _totalWeight = totalWeight; + _min = min; + _max = max; + if (!historical) { + normalizeBoundaryCentroids(); + } + _publiclyCompressed = true; + } else { + if (sorted && positiveCount == count) { + bufferIncomingCentroids(means, weights, count, totalWeight, min, max); + } else { + // Caller-owned arrays are copied only when filtering or sorting is necessary. + double[] sortedMeans = new double[positiveCount]; + double[] sortedWeights = new double[positiveCount]; + int index = 0; + for (int i = 0; i < count; i++) { + if (weights[i] > 0.0) { + sortedMeans[index] = means[i]; + sortedWeights[index++] = weights[i]; + } + } + if (!sorted) { + sortCentroids(sortedMeans, sortedWeights, positiveCount); + } + bufferIncomingCentroids(sortedMeans, sortedWeights, positiveCount, totalWeight, min, max); + } + _min = Math.min(_min, min); + _max = Math.max(_max, max); + } + } + + public void add(double[] values, int from, int toExclusive) { + if (from >= toExclusive) { + return; + } + requireMutable(); + prepareForMutation(); + while (from < toExclusive) { + ensureRawBuffer(); + flushIfBuffersFull(true); + int numValues = Math.min(toExclusive - from, Math.min(_rawValues.length - _numRawValues, + getPendingInputLimit() - _numRawValues - _numIncomingCentroids)); + int rawOffset = _numRawValues; + for (int i = 0; i < numValues; i++) { + double value = values[from + i]; + if (Double.isNaN(value)) { + throw new IllegalArgumentException("Cannot add NaN to t-digest"); + } + _rawValues[rawOffset + i] = value; + } + _originalFractionalBytes = null; + from += numValues; + _numRawValues += numValues; + flushIfBuffersFull(!_useTwoLevelRawCompression || _rawValues.length == _pendingCentroidCapacity); + } + } + + @Override + public void add(double value) { + if (Double.isNaN(value)) { + throw new IllegalArgumentException("Cannot add NaN to t-digest"); + } + requireMutable(); + prepareForMutation(); + ensureRawBuffer(); + flushIfBuffersFull(true); + _originalFractionalBytes = null; + _rawValues[_numRawValues++] = value; + } + + @Override + public void add(double value, double weight) { + if (weight == 1) { + add(value); + return; + } + if (Double.isNaN(value)) { + throw new IllegalArgumentException("Cannot add NaN to t-digest"); + } + if (!(weight > 0.0) || !Double.isFinite(weight)) { + throw new IllegalArgumentException("TDigest weight must be positive: " + weight); + } + requireMutable(); + prepareForMutation(); + bufferIncomingCentroid(value, weight); + _originalFractionalBytes = null; + _hasFractionalWeights |= weight != Math.rint(weight); + recordFreshFractionalMean(value, weight); + _min = Math.min(_min, value); + _max = Math.max(_max, value); + } + + @Override + public void add(TDigest other) { + if (other.hasValidStatistics() && other.getTotalWeight() == 0.0) { + return; + } + requireMutable(); + if (other instanceof PercentileTDigestAccumulator) { + addAccumulator((PercentileTDigestAccumulator) other); + return; + } + if (other instanceof NonFiniteAwareTDigest) { + PercentileTDigestAccumulator finiteDigest = ((NonFiniteAwareTDigest) other).finiteDigestIfNoTails(); + if (finiteDigest != null) { + addAccumulator(finiteDigest); + return; + } + } + if (!other.hasValidStatistics()) { + materializePendingSerializedTDigest(); + SerializedTDigestInput input = getSerializedTDigestInput(); + input.reset(ByteBuffer.wrap(other.serialize()), false); + addSerializedTDigest(input); + return; + } + double otherWeight = other.getTotalWeight(); + checkTotalWeight(getTotalWeight() + otherWeight); + Collection<Centroid> centroids = other.centroids(); + boolean wrapped = other instanceof NonFiniteAwareTDigest; + int capacity = wrapped ? Math.addExact(centroids.size(), 2) : centroids.size(); + double[] means = new double[capacity]; + double[] weights = new double[capacity]; + int count = 0; + for (Centroid centroid : centroids) { + double weight = centroid.weight(); + if (weight == 0.0) { + continue; + } + if (!(weight > 0.0) || !Double.isFinite(weight) || Double.isNaN(centroid.mean())) { + throw new IllegalArgumentException("Invalid TDigest centroid"); + } + means[count] = centroid.mean(); + weights[count++] = weight; + } + if (wrapped) { + count = normalizeSerializedBoundaries(means, weights, count, other.getMin(), other.getMax()); + } + // Validate all generic centroids before changing the target, then use the same whole-source primitive path. + addCentroids(means, weights, count, other.getMin(), other.getMax(), false, wrapped); + inheritHistoricalFractionalBoundaries(other); + } + + private void addAccumulator(PercentileTDigestAccumulator other) { + if (other == this) { + compress(); + prepareForMutation(); + if (_numCentroids == 0) { + return; + } + // Snapshot only self-merges; fresh fractional mass need not be representable by a legacy wire payload. + addHistoricalCentroids(Arrays.copyOf(_centroidMeans, _numCentroids), + Arrays.copyOf(_centroidWeights, _numCentroids), _numCentroids, _min, _max, false); + return; + } + if (_deferBoundaryNormalization && !other._deferBoundaryNormalization) { + other.prepareForMutation(); + } + if (other._originalBoundaryBytes != null) { + if (getTotalWeight() != 0.0) { + throw unrepresentableBoundaryMutation(); + } + addSerializedTDigest(other._originalBoundaryBytes); + return; + } + if (other._originalFractionalBytes != null && getTotalWeight() == 0.0) { + materializePendingSerializedTDigest(); + SerializedTDigestInput input = getSerializedTDigestInput(); + input.reset(other._originalFractionalBytes); + input._retainedBytes = other._originalFractionalBytes; + addSerializedTDigest(input); + return; + } + if (other._pendingSerializedTDigest != null) { + // Materialize before resetting the reusable reader: materialization uses that same reader. + materializePendingSerializedTDigest(); + SerializedTDigestInput input = getSerializedTDigestInput(); + input.reset(other._pendingSerializedTDigest, other._pendingSerializedMetadata); + input._retainedBytes = other._pendingSerializedTDigest; + addSerializedTDigest(input); + return; + } + if (other._legacyDegraded) { + addSerializedTDigest(other.serialize()); + return; + } + + // MergingDigest compressed merge sources in place. Aggregation owns these mutable inputs exclusively. + other.compress(); + other.prepareForMutation(); + prepareForMutation(); + checkTotalWeight(_totalWeight + _incomingWeight + _numRawValues + other._totalWeight); + _hasFractionalWeights |= other._hasFractionalWeights; + inheritHistoricalFractionalBoundaries(other); + bufferIncomingCentroids(other._centroidMeans, other._centroidWeights, other._numCentroids, other._totalWeight, + other._min, other._max); + if (other._totalWeight > 0.0) { + _originalFractionalBytes = null; + } + _min = Math.min(_min, other._min); + _max = Math.max(_max, other._max); + } + + public void addSerializedTDigest(byte[] bytes) { + if (_legacyDegraded || _originalBoundaryBytes != null) { + SerializedTDigestInput input = getSerializedTDigestInput(); + input.reset(bytes); + addSerializedTDigest(input); + return; + } + requireMutable(); + materializePendingSerializedTDigest(); + SerializedTDigestInput input = getSerializedTDigestInput(); + input.reset(bytes); + addSerializedTDigest(input); + } + + private SerializedTDigestInput getSerializedTDigestInput() { + if (_serializedTDigestInput == null) { + _serializedTDigestInput = new SerializedTDigestInput(); + } + return _serializedTDigestInput; + } + + public void addSerializedTDigest(SerializedTDigestInput input) { + if (ignoreEmptyInputForDegradedState(input)) { + return; + } + requireMutable(); + if (prepareSerializedTDigest(input)) { + mergeSerializedTDigest(input, false); + } + } + + public void addSerializedTDigestDirect(SerializedTDigestInput input) { + if (ignoreEmptyInputForDegradedState(input)) { + return; + } + requireMutable(); + if (prepareSerializedTDigest(input)) { + mergeSerializedTDigest(input, true); + } + } + + private boolean ignoreEmptyInputForDegradedState(SerializedTDigestInput input) { + if (!_legacyDegraded && _originalBoundaryBytes == null + && (_pendingSerializedMetadata == null || !_pendingSerializedMetadata.weightedBoundaries())) { + return false; + } + input.inspectMetadata(); + return !input._metadata._needsLegacyFallback && input._metadata._totalWeight == 0.0; + } + + private boolean prepareSerializedTDigest(SerializedTDigestInput input) { + if (_pendingSerializedTDigest == null && hasNoInputs()) { + input.inspectMetadata(); + _hasFractionalWeights |= input._metadata._fractionalWeights; + _pendingSerializedTDigest = input.retainBytes(); + _pendingSerializedMetadata = input.getMetadata(); + if (input.recordHistoricalFractionalBoundaries(this)) { + // Alias the retained snapshot; historical fractional bytes remain byte-exact after read-only materialization. + _originalFractionalBytes = _pendingSerializedTDigest; + } + if (input._metadata._needsLegacyFallback) { + _legacyDegraded = true; + } + return false; + } + materializePendingSerializedTDigest(); + input.decode(); + if (input._metadata._needsLegacyFallback) { + if (hasNoInputs()) { + _pendingSerializedTDigest = input.retainBytes(); + _pendingSerializedMetadata = input.getMetadata(); + _legacyDegraded = true; + return false; + } + throw corruptedMutation(); + } + if (!_deferBoundaryNormalization) { + input.normalizeBoundaries(); + } + prepareForMutation(); + checkTotalWeight(_totalWeight + _incomingWeight + _numRawValues + input._metadata._totalWeight); + input.recordHistoricalFractionalBoundaries(this); + if (input._metadata._totalWeight > 0.0) { + _originalFractionalBytes = null; + } + _hasFractionalWeights |= input._metadata._fractionalWeights; + return true; + } + + private boolean hasNoInputs() { + return _numCentroids == 0 && _totalWeight == 0.0 && _numRawValues == 0 && _numIncomingCentroids == 0; + } + + private void mergeSerializedTDigest(SerializedTDigestInput input, boolean direct) { + if (input._metadata._totalWeight == 0.0) { + return; + } + if (hasNoInputs()) { + ensureCentroidCapacity(input._numCentroids); + System.arraycopy(input._means, 0, _centroidMeans, 0, input._numCentroids); + System.arraycopy(input._weights, 0, _centroidWeights, 0, input._numCentroids); + _numCentroids = input._numCentroids; + _totalWeight = input._metadata._totalWeight; + _min = input._metadata._min; + _max = input._metadata._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._metadata._compression; + return; + } + if (direct) { + flush(); + boolean runBackwards = (_mergeCount++ & 1) != 0; + if (!mergeSorted(input._means, input._weights, input._numCentroids, input._metadata._totalWeight, + getIncrementalCompression(), runBackwards)) { + return; + } + } else { + bufferIncomingCentroids(input._means, input._weights, input._numCentroids, input._metadata._totalWeight, + input._metadata._min, input._metadata._max); + } + _min = Math.min(_min, input._metadata._min); + _max = Math.max(_max, input._metadata._max); + } + + @Override + public void compress() { + if (_legacyDegraded) { + return; + } + materializePendingSerializedTDigest(); + if (_useTwoLevelRawCompression) { + // Public compression and serialization both retain the legacy working pass before final compression. + flush(); + } + if (_numRawValues > 0 || _numIncomingCentroids > 0) { + // SQL reduction combines pending values and existing centroids directly at public compression, matching + // MergingDigest 3.3. Its writer avoids an extra working-compression pass that can increase reducer rank error. + flush(_compression); + _publiclyCompressed = true; + } else if (!_publiclyCompressed && _numCentroids > 0) { + recompressCentroids(_compression); + _publiclyCompressed = true; + } + } + + @Override + public boolean hasValidStatistics() { + return !_legacyDegraded; + } + + @Override + public double getTotalWeight() { + return _pendingSerializedTDigest != null ? _pendingSerializedMetadata.totalWeight() + : _totalWeight + _incomingWeight + _numRawValues; + } + + /// Bounds serialized output centroids from buffered inputs, including up to two boundary splits before a merge + /// and two more before writing, without flushing or using the potentially huge input mass. + public long getCentroidCountUpperBound() { + return _pendingSerializedTDigest != null ? (long) _pendingSerializedMetadata.centroidCount() + 4L + : (long) _numCentroids + _numRawValues + _numIncomingCentroids + 4L; + } + + /// Whether unchanged historical fractional boundaries still require their original byte representation. + public boolean hasOriginalFractionalPayload() { + return _originalFractionalBytes != null; + } + + @Override + public double cdf(double value) { + if (Double.isNaN(value) || Double.isInfinite(value)) { + throw new IllegalArgumentException(String.format("Invalid value: %f", value)); + } + if (_legacyDegraded) { + return Double.NaN; + } + materializePendingSerializedTDigest(); + flush(); + if (_numCentroids == 0) { + return Double.NaN; + } + if (_numCentroids == 1) { + double width = _max - _min; + if (value < _min) { + return 0.0; + } else if (value > _max) { + return 1.0; + } else if (value - _min <= width) { + return 0.5; + } else { + return (value - _min) / width; + } + } + if (value < _min) { + return 0.0; + } + if (value > _max) { + return 1.0; + } + if (value < _centroidMeans[0]) { + double width = _centroidMeans[0] - _min; + if (width > 0.0) { + if (_centroidWeights[0] < 1.0) { + return (_centroidWeights[0] / 2.0 * (value - _min) / width) / _totalWeight; + } + return value == _min ? 0.5 / _totalWeight + : (1.0 + (value - _min) / width * (_centroidWeights[0] / 2.0 - 1.0)) / _totalWeight; + } + return 0.0; + } + int lastIndex = _numCentroids - 1; + if (value > _centroidMeans[lastIndex]) { + double width = _max - _centroidMeans[lastIndex]; + if (width > 0.0) { + if (_centroidWeights[lastIndex] < 1.0) { + return 1.0 - (_centroidWeights[lastIndex] / 2.0 * (_max - value) / width) / _totalWeight; + } + return value == _max ? 1.0 - 0.5 / _totalWeight + : 1.0 - (1.0 + (_max - value) / width * (_centroidWeights[lastIndex] / 2.0 - 1.0)) + / _totalWeight; + } + return 1.0; + } + + double weightSoFar = 0.0; + for (int i = 0; i < lastIndex; i++) { + if (_centroidMeans[i] == value) { + double equalWeight = 0.0; + int equalIndex = i; + while (equalIndex < _numCentroids && _centroidMeans[equalIndex] == value) { + equalWeight += _centroidWeights[equalIndex]; + equalIndex++; + } + return (weightSoFar + equalWeight / 2.0) / _totalWeight; + } + if (_centroidMeans[i] <= value && _centroidMeans[i + 1] > value) { + if (!Double.isFinite(_centroidMeans[i]) || !Double.isFinite(_centroidMeans[i + 1])) { + if (_centroidMeans[i] == Double.NEGATIVE_INFINITY) { + return (weightSoFar + _centroidWeights[i]) / _totalWeight; + } + if (_centroidWeights[i] == 1.0 && _centroidWeights[i + 1] == 1.0) { + return (weightSoFar + 1.0) / _totalWeight; + } + double leftExcludedWeight = _centroidWeights[i] == 1.0 ? 0.5 : 0.0; + double rightExcludedWeight = _centroidWeights[i + 1] == 1.0 ? 0.5 : 0.0; + double interpolatedWeight = (_centroidWeights[i] + _centroidWeights[i + 1]) / 2.0 + - leftExcludedWeight - rightExcludedWeight; + double base = weightSoFar + _centroidWeights[i] / 2.0 + leftExcludedWeight; + return (base + interpolatedWeight * (value - _centroidMeans[i]) + / (_centroidMeans[i + 1] - _centroidMeans[i])) / _totalWeight; Review Comment: Verified the same bytes against both implementations: [(-Infinity,3),(10,4),(20,1)] at cdf(0) returns 3/8 in both the old Pinot accumulator and this implementation; the genuine 3.3 reader returns NaN. The README now documents this distinction, and testCdfInfiniteNeighborsRetainsLegacyUnitExclusions already pins that exact case. This is a difference from tdunning 3.3, not a new change from the old Pinot accumulator. Validated at `3821623d46933dd233b8110e123ecb4f86ac1e96`: 554 affected tests, genuine 3.2/3.3 bidirectional byte checks, the 270-case oracle, applicable precommit checks and independent Pinot review pass. New-head CI and maintainer approval remain separate gates. ########## pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/tdigest/PercentileTDigestAccumulator.java: ########## @@ -0,0 +1,1725 @@ +/** + * 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. + */ +/* + * Portions of the CDF, quantile, weighted-average and K1 weight-limit implementation are adapted from + * MergingDigest, AbstractTDigest and ScaleFunction in t-digest 3.3 (https://github.com/tdunning/t-digest). + * Modified for Pinot's primitive centroid storage, lazy serialized inputs and legacy boundary handling. + * Original license notice for these portions: + * + * Licensed to Ted Dunning 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.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.List; +import org.apache.pinot.segment.local.customobject.tdigest.TDigestCodec.SerializedTDigestMetadata; + + + +/// Accumulates raw values and serialized TDigests into primitive centroids using Pinot's K1 weight-limit rule. +/// +/// Raw values are sorted in small batches; serialized centroids retain their existing sorted order. Both are +/// linearly merged with accumulated centroids. K1 preserves Pinot's existing middle-quantile behavior, and the +/// legacy verbose and small encodings preserve stored data and mixed-version wire compatibility. +/// +/// Serialized state keeps its first digest pending and allocates centroid buffers only when another input must be +/// merged or the digest is queried. Raw-value buffers remain unallocated until a raw value is added. +/// +/// Historical corrupted payloads remain opaque and byte-exact, with NaN statistics. Their reported historical mass +/// may be invalid; they cannot be mutated or mixed with another distribution. Such an operation fails before +/// changing aggregate mass, because neither invented statistics nor synthetic NaN bytes are safe for old readers. +/// Unchanged historical fractional boundaries retain their original bytes, including when configured compression +/// differs. After a merge, inherited fractional endpoints retain precise verbose weights and finite statistics when +/// unit endpoints cannot preserve their mass. These keep the historical t-digest 3.3 assertion-enabled reader and +/// recompression limitation. Newly introduced unsupported fractional endpoints are rejected. +/// +/// Instances are externally serialized per aggregation or group key and are not safe for concurrent mutation. +public final class PercentileTDigestAccumulator extends TDigest { + private static final int MIN_RAW_BUFFER_SIZE = 256; + private static final int MAX_RAW_BUFFER_SIZE = 10_000; + private static final int TWO_LEVEL_COMPRESSION_MULTIPLIER = 2; + private static final int MIN_INCREMENTAL_COMPRESSION = 50; + + private final double _compression; + private final boolean _useTwoLevelRawCompression; + private final double _workingCompression; + private final double _publicMaxWeightScale; + private final double _workingMaxWeightScale; + private final int _centroidCapacity; + private final int _pendingCentroidCapacity; + private double[] _rawValues; + private double[] _centroidMeans; + private double[] _centroidWeights; + private double[] _outputMeans; + private double[] _outputWeights; + private double[] _incomingMeans; + private double[] _incomingWeights; + private double[] _sortedIncomingMeans; + private double[] _sortedIncomingWeights; + private SerializedTDigestInput _serializedTDigestInput; + private byte[] _pendingSerializedTDigest; + private byte[] _originalFractionalBytes; + private byte[] _originalBoundaryBytes; + private double _historicalFractionalMinMean = Double.NaN; + private double _historicalFractionalMaxMean = Double.NaN; + private double _freshFractionalMinMean = Double.NaN; + private double _freshFractionalMaxMean = Double.NaN; + private SerializedTDigestMetadata _pendingSerializedMetadata; + private boolean _legacyDegraded; + private boolean _hasFractionalWeights; + private boolean _deferBoundaryNormalization; + + private int _numRawValues; + private int _numCentroids; + private int _numIncomingCentroids; + private int _mergeCount; + private boolean _publiclyCompressed; + private boolean _incomingCentroidsSorted = true; + private double _totalWeight; + private double _incomingWeight; + private double _min = Double.POSITIVE_INFINITY; + private double _max = Double.NEGATIVE_INFINITY; + + public PercentileTDigestAccumulator(int compression) { + this((double) compression, true, true); + } + + public PercentileTDigestAccumulator(double compression) { + this(compression, true, true); + } + + private PercentileTDigestAccumulator(double compression, boolean allocateRawBuffer, boolean allocateMergeBuffers) { + this(compression, allocateRawBuffer, allocateMergeBuffers, false); + } + + private PercentileTDigestAccumulator(double compression, boolean allocateRawBuffer, boolean allocateMergeBuffers, + boolean useTwoLevelRawCompression) { + _compression = TDigestCodec.normalizeCompression(compression); + _useTwoLevelRawCompression = useTwoLevelRawCompression; + _workingCompression = TWO_LEVEL_COMPRESSION_MULTIPLIER * _compression; + _publicMaxWeightScale = calculateK1MaxWeightScale(_compression); + _workingMaxWeightScale = calculateK1MaxWeightScale(_workingCompression); + long roundedCompression = (long) Math.ceil(_compression); + _centroidCapacity = TDigestCodec.getDefaultCentroidCapacity(_compression); + _pendingCentroidCapacity = getPendingCentroidCapacity(_centroidCapacity, useTwoLevelRawCompression); + if (allocateRawBuffer) { + _rawValues = new double[getInitialRawBufferSize(roundedCompression)]; + } + if (allocateMergeBuffers) { + int initialCapacity = Math.min(_centroidCapacity, MIN_RAW_BUFFER_SIZE); + _centroidMeans = new double[initialCapacity]; + _centroidWeights = new double[initialCapacity]; + _outputMeans = new double[initialCapacity]; + _outputWeights = new double[initialCapacity]; + } + } + + /// Creates a lazy digest with the legacy two-level compression used by Star-tree and general digest aggregation. + public static PercentileTDigestAccumulator forLegacyAggregation(double compression) { + return new PercentileTDigestAccumulator(compression, false, false, true); + } + + /// Reads and retains one legacy digest, preserving the general aggregation compression policy. + public static PercentileTDigestAccumulator fromBytes(byte[] bytes) { + return fromBytes(ByteBuffer.wrap(bytes)); + } + + /// Reads one payload and advances the buffer past it. + public static PercentileTDigestAccumulator fromBytes(ByteBuffer bytes) { + SerializedTDigestInput input = new SerializedTDigestInput(); + input.reset(bytes); + return fromBytes(input); + } + + /// Retains an already validated input without rescanning its centroid bytes. + public static PercentileTDigestAccumulator fromBytes(SerializedTDigestInput input) { + PercentileTDigestAccumulator digest = forLegacyAggregation(input.getCompression()); + digest.addSerializedTDigest(input); + return digest; + } + + public static PercentileTDigestAccumulator forReduction(double compression) { + return new PercentileTDigestAccumulator(compression, false, false); + } + + /// Snapshots buffered inputs and centroid state without materializing or compressing the source. Mutable buffers + /// are copied; immutable retained bytes and metadata are shared. Merge and sort scratch buffers remain lazy. + PercentileTDigestAccumulator copy() { + PercentileTDigestAccumulator copy = + new PercentileTDigestAccumulator(_compression, false, false, _useTwoLevelRawCompression); + copy._rawValues = _rawValues != null ? _rawValues.clone() : null; + copy._centroidMeans = _numCentroids > 0 ? Arrays.copyOf(_centroidMeans, _numCentroids) : null; + copy._centroidWeights = _numCentroids > 0 ? Arrays.copyOf(_centroidWeights, _numCentroids) : null; + copy._incomingMeans = _numIncomingCentroids > 0 ? Arrays.copyOf(_incomingMeans, _numIncomingCentroids) : null; + copy._incomingWeights = _numIncomingCentroids > 0 ? Arrays.copyOf(_incomingWeights, _numIncomingCentroids) : null; + copy._pendingSerializedTDigest = _pendingSerializedTDigest; + copy._pendingSerializedMetadata = _pendingSerializedMetadata; + copy._originalFractionalBytes = _originalFractionalBytes; + copy._originalBoundaryBytes = _originalBoundaryBytes; + copy._historicalFractionalMinMean = _historicalFractionalMinMean; + copy._historicalFractionalMaxMean = _historicalFractionalMaxMean; + copy._freshFractionalMinMean = _freshFractionalMinMean; + copy._freshFractionalMaxMean = _freshFractionalMaxMean; + copy._legacyDegraded = _legacyDegraded; + copy._hasFractionalWeights = _hasFractionalWeights; + copy._deferBoundaryNormalization = _deferBoundaryNormalization; + copy._numRawValues = _numRawValues; + copy._numCentroids = _numCentroids; + copy._numIncomingCentroids = _numIncomingCentroids; + copy._mergeCount = _mergeCount; + copy._publiclyCompressed = _publiclyCompressed; + copy._incomingCentroidsSorted = _incomingCentroidsSorted; + copy._totalWeight = _totalWeight; + copy._incomingWeight = _incomingWeight; + copy._min = _min; + copy._max = _max; + return copy; + } + + /// Copies primitive centroids without encoding and decoding them. Zero weights are ignored, and unordered + /// means are sorted when initializing a stored, already compressed distribution; caller arrays are not mutated. + /// Subsequent inputs use the normal buffered merge. Positive weights and consistent extrema are required. + public void addCentroids(double[] means, double[] weights, int count, double min, double max, + boolean alreadyCompressed) { + addCentroids(means, weights, count, min, max, alreadyCompressed, false); + } + + /// Adds validated legacy centroids without marking their existing fractional mass as fresh input. + void addHistoricalCentroids(double[] means, double[] weights, int count, double min, double max, + boolean alreadyCompressed) { + addCentroids(means, weights, count, min, max, alreadyCompressed, true); + } + + private void addCentroids(double[] means, double[] weights, int count, double min, double max, + boolean alreadyCompressed, boolean historical) { + requireMutable(); + if (count < 0 || count > means.length || count > weights.length) { + throw new IllegalArgumentException("Invalid TDigest centroid count: " + count); + } + double totalWeight = 0.0; + int positiveCount = 0; + boolean sorted = true; + boolean fractionalWeights = false; + double previousMean = Double.NEGATIVE_INFINITY; + for (int i = 0; i < count; i++) { + double weight = weights[i]; + if (weight == 0.0) { + continue; + } + double mean = means[i]; + if (!(weight > 0.0) || !Double.isFinite(weight) || !(mean >= min && mean <= max)) { + throw new IllegalArgumentException("Invalid TDigest centroid"); + } + totalWeight += weight; + sorted &= mean >= previousMean; + previousMean = mean; + fractionalWeights |= weight != Math.rint(weight); + positiveCount++; + } + checkTotalWeight(getTotalWeight() + totalWeight); + if (positiveCount == 0) { + return; + } + prepareForMutation(); + _originalFractionalBytes = null; + _hasFractionalWeights |= fractionalWeights; + if (!historical && fractionalWeights) { + for (int i = 0; i < count; i++) { + recordFreshFractionalMean(means[i], weights[i]); + } + } + if (hasNoInputs() && alreadyCompressed) { + ensureCentroidCapacity(Math.addExact(positiveCount, 2)); + for (int i = 0; i < count; i++) { + if (weights[i] > 0.0) { + _centroidMeans[_numCentroids] = means[i]; + _centroidWeights[_numCentroids++] = weights[i]; + } + } + if (!sorted) { + sortCentroids(_centroidMeans, _centroidWeights, _numCentroids); + } + _totalWeight = totalWeight; + _min = min; + _max = max; + if (!historical) { + normalizeBoundaryCentroids(); + } + _publiclyCompressed = true; + } else { + if (sorted && positiveCount == count) { + bufferIncomingCentroids(means, weights, count, totalWeight, min, max); + } else { + // Caller-owned arrays are copied only when filtering or sorting is necessary. + double[] sortedMeans = new double[positiveCount]; + double[] sortedWeights = new double[positiveCount]; + int index = 0; + for (int i = 0; i < count; i++) { + if (weights[i] > 0.0) { + sortedMeans[index] = means[i]; + sortedWeights[index++] = weights[i]; + } + } + if (!sorted) { + sortCentroids(sortedMeans, sortedWeights, positiveCount); + } + bufferIncomingCentroids(sortedMeans, sortedWeights, positiveCount, totalWeight, min, max); + } + _min = Math.min(_min, min); + _max = Math.max(_max, max); + } + } + + public void add(double[] values, int from, int toExclusive) { + if (from >= toExclusive) { + return; + } + requireMutable(); + prepareForMutation(); + while (from < toExclusive) { + ensureRawBuffer(); + flushIfBuffersFull(true); + int numValues = Math.min(toExclusive - from, Math.min(_rawValues.length - _numRawValues, + getPendingInputLimit() - _numRawValues - _numIncomingCentroids)); + int rawOffset = _numRawValues; + for (int i = 0; i < numValues; i++) { + double value = values[from + i]; + if (Double.isNaN(value)) { + throw new IllegalArgumentException("Cannot add NaN to t-digest"); + } + _rawValues[rawOffset + i] = value; + } + _originalFractionalBytes = null; + from += numValues; + _numRawValues += numValues; + flushIfBuffersFull(!_useTwoLevelRawCompression || _rawValues.length == _pendingCentroidCapacity); + } + } + + @Override + public void add(double value) { + if (Double.isNaN(value)) { + throw new IllegalArgumentException("Cannot add NaN to t-digest"); + } + requireMutable(); + prepareForMutation(); + ensureRawBuffer(); + flushIfBuffersFull(true); + _originalFractionalBytes = null; + _rawValues[_numRawValues++] = value; + } + + @Override + public void add(double value, double weight) { + if (weight == 1) { + add(value); + return; + } + if (Double.isNaN(value)) { + throw new IllegalArgumentException("Cannot add NaN to t-digest"); + } + if (!(weight > 0.0) || !Double.isFinite(weight)) { + throw new IllegalArgumentException("TDigest weight must be positive: " + weight); + } + requireMutable(); + prepareForMutation(); + bufferIncomingCentroid(value, weight); + _originalFractionalBytes = null; + _hasFractionalWeights |= weight != Math.rint(weight); + recordFreshFractionalMean(value, weight); + _min = Math.min(_min, value); + _max = Math.max(_max, value); + } + + @Override + public void add(TDigest other) { + if (other.hasValidStatistics() && other.getTotalWeight() == 0.0) { + return; + } + requireMutable(); + if (other instanceof PercentileTDigestAccumulator) { + addAccumulator((PercentileTDigestAccumulator) other); + return; + } + if (other instanceof NonFiniteAwareTDigest) { + PercentileTDigestAccumulator finiteDigest = ((NonFiniteAwareTDigest) other).finiteDigestIfNoTails(); + if (finiteDigest != null) { + addAccumulator(finiteDigest); + return; + } + } + if (!other.hasValidStatistics()) { + materializePendingSerializedTDigest(); + SerializedTDigestInput input = getSerializedTDigestInput(); + input.reset(ByteBuffer.wrap(other.serialize()), false); + addSerializedTDigest(input); + return; + } + double otherWeight = other.getTotalWeight(); + checkTotalWeight(getTotalWeight() + otherWeight); + Collection<Centroid> centroids = other.centroids(); + boolean wrapped = other instanceof NonFiniteAwareTDigest; + int capacity = wrapped ? Math.addExact(centroids.size(), 2) : centroids.size(); + double[] means = new double[capacity]; + double[] weights = new double[capacity]; + int count = 0; + for (Centroid centroid : centroids) { + double weight = centroid.weight(); + if (weight == 0.0) { + continue; + } + if (!(weight > 0.0) || !Double.isFinite(weight) || Double.isNaN(centroid.mean())) { + throw new IllegalArgumentException("Invalid TDigest centroid"); + } + means[count] = centroid.mean(); + weights[count++] = weight; + } + if (wrapped) { + count = normalizeSerializedBoundaries(means, weights, count, other.getMin(), other.getMax()); + } + // Validate all generic centroids before changing the target, then use the same whole-source primitive path. + addCentroids(means, weights, count, other.getMin(), other.getMax(), false, wrapped); + inheritHistoricalFractionalBoundaries(other); + } + + private void addAccumulator(PercentileTDigestAccumulator other) { + if (other == this) { + compress(); + prepareForMutation(); + if (_numCentroids == 0) { + return; + } + // Snapshot only self-merges; fresh fractional mass need not be representable by a legacy wire payload. + addHistoricalCentroids(Arrays.copyOf(_centroidMeans, _numCentroids), + Arrays.copyOf(_centroidWeights, _numCentroids), _numCentroids, _min, _max, false); + return; + } + if (_deferBoundaryNormalization && !other._deferBoundaryNormalization) { + other.prepareForMutation(); + } + if (other._originalBoundaryBytes != null) { + if (getTotalWeight() != 0.0) { + throw unrepresentableBoundaryMutation(); + } + addSerializedTDigest(other._originalBoundaryBytes); + return; + } + if (other._originalFractionalBytes != null && getTotalWeight() == 0.0) { + materializePendingSerializedTDigest(); + SerializedTDigestInput input = getSerializedTDigestInput(); + input.reset(other._originalFractionalBytes); + input._retainedBytes = other._originalFractionalBytes; + addSerializedTDigest(input); + return; + } + if (other._pendingSerializedTDigest != null) { + // Materialize before resetting the reusable reader: materialization uses that same reader. + materializePendingSerializedTDigest(); + SerializedTDigestInput input = getSerializedTDigestInput(); + input.reset(other._pendingSerializedTDigest, other._pendingSerializedMetadata); + input._retainedBytes = other._pendingSerializedTDigest; + addSerializedTDigest(input); + return; + } + if (other._legacyDegraded) { + addSerializedTDigest(other.serialize()); + return; + } + + // MergingDigest compressed merge sources in place. Aggregation owns these mutable inputs exclusively. + other.compress(); + other.prepareForMutation(); + prepareForMutation(); + checkTotalWeight(_totalWeight + _incomingWeight + _numRawValues + other._totalWeight); + _hasFractionalWeights |= other._hasFractionalWeights; + inheritHistoricalFractionalBoundaries(other); + bufferIncomingCentroids(other._centroidMeans, other._centroidWeights, other._numCentroids, other._totalWeight, + other._min, other._max); + if (other._totalWeight > 0.0) { + _originalFractionalBytes = null; + } + _min = Math.min(_min, other._min); + _max = Math.max(_max, other._max); + } + + public void addSerializedTDigest(byte[] bytes) { + if (_legacyDegraded || _originalBoundaryBytes != null) { + SerializedTDigestInput input = getSerializedTDigestInput(); + input.reset(bytes); + addSerializedTDigest(input); + return; + } + requireMutable(); + materializePendingSerializedTDigest(); + SerializedTDigestInput input = getSerializedTDigestInput(); + input.reset(bytes); + addSerializedTDigest(input); + } + + private SerializedTDigestInput getSerializedTDigestInput() { + if (_serializedTDigestInput == null) { + _serializedTDigestInput = new SerializedTDigestInput(); + } + return _serializedTDigestInput; + } + + public void addSerializedTDigest(SerializedTDigestInput input) { + if (ignoreEmptyInputForDegradedState(input)) { + return; + } + requireMutable(); + if (prepareSerializedTDigest(input)) { + mergeSerializedTDigest(input, false); + } + } + + public void addSerializedTDigestDirect(SerializedTDigestInput input) { + if (ignoreEmptyInputForDegradedState(input)) { + return; + } + requireMutable(); + if (prepareSerializedTDigest(input)) { + mergeSerializedTDigest(input, true); + } + } + + private boolean ignoreEmptyInputForDegradedState(SerializedTDigestInput input) { + if (!_legacyDegraded && _originalBoundaryBytes == null + && (_pendingSerializedMetadata == null || !_pendingSerializedMetadata.weightedBoundaries())) { + return false; + } + input.inspectMetadata(); + return !input._metadata._needsLegacyFallback && input._metadata._totalWeight == 0.0; + } + + private boolean prepareSerializedTDigest(SerializedTDigestInput input) { + if (_pendingSerializedTDigest == null && hasNoInputs()) { + input.inspectMetadata(); + _hasFractionalWeights |= input._metadata._fractionalWeights; + _pendingSerializedTDigest = input.retainBytes(); + _pendingSerializedMetadata = input.getMetadata(); + if (input.recordHistoricalFractionalBoundaries(this)) { + // Alias the retained snapshot; historical fractional bytes remain byte-exact after read-only materialization. + _originalFractionalBytes = _pendingSerializedTDigest; + } + if (input._metadata._needsLegacyFallback) { + _legacyDegraded = true; + } + return false; + } + materializePendingSerializedTDigest(); + input.decode(); + if (input._metadata._needsLegacyFallback) { + if (hasNoInputs()) { + _pendingSerializedTDigest = input.retainBytes(); + _pendingSerializedMetadata = input.getMetadata(); + _legacyDegraded = true; + return false; + } + throw corruptedMutation(); + } + if (!_deferBoundaryNormalization) { + input.normalizeBoundaries(); + } + prepareForMutation(); + checkTotalWeight(_totalWeight + _incomingWeight + _numRawValues + input._metadata._totalWeight); + input.recordHistoricalFractionalBoundaries(this); + if (input._metadata._totalWeight > 0.0) { + _originalFractionalBytes = null; + } + _hasFractionalWeights |= input._metadata._fractionalWeights; + return true; + } + + private boolean hasNoInputs() { + return _numCentroids == 0 && _totalWeight == 0.0 && _numRawValues == 0 && _numIncomingCentroids == 0; + } + + private void mergeSerializedTDigest(SerializedTDigestInput input, boolean direct) { + if (input._metadata._totalWeight == 0.0) { + return; + } + if (hasNoInputs()) { + ensureCentroidCapacity(input._numCentroids); + System.arraycopy(input._means, 0, _centroidMeans, 0, input._numCentroids); + System.arraycopy(input._weights, 0, _centroidWeights, 0, input._numCentroids); + _numCentroids = input._numCentroids; + _totalWeight = input._metadata._totalWeight; + _min = input._metadata._min; + _max = input._metadata._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._metadata._compression; + return; + } + if (direct) { + flush(); + boolean runBackwards = (_mergeCount++ & 1) != 0; + if (!mergeSorted(input._means, input._weights, input._numCentroids, input._metadata._totalWeight, + getIncrementalCompression(), runBackwards)) { + return; + } + } else { + bufferIncomingCentroids(input._means, input._weights, input._numCentroids, input._metadata._totalWeight, + input._metadata._min, input._metadata._max); + } + _min = Math.min(_min, input._metadata._min); + _max = Math.max(_max, input._metadata._max); + } + + @Override + public void compress() { + if (_legacyDegraded) { + return; + } + materializePendingSerializedTDigest(); + if (_useTwoLevelRawCompression) { + // Public compression and serialization both retain the legacy working pass before final compression. + flush(); + } + if (_numRawValues > 0 || _numIncomingCentroids > 0) { + // SQL reduction combines pending values and existing centroids directly at public compression, matching + // MergingDigest 3.3. Its writer avoids an extra working-compression pass that can increase reducer rank error. + flush(_compression); + _publiclyCompressed = true; + } else if (!_publiclyCompressed && _numCentroids > 0) { + recompressCentroids(_compression); + _publiclyCompressed = true; + } Review Comment: Agreed that direct MergingDigest 3.3 compress() uses one public pass. The retained policy is Pinot's previous TDigestUtils.serialize(): it first called centroidCount() to drain at working compression, then asBytes() for public compression. I clarified that distinction in the code comment. A probe against the real 3.3 JAR and old Pinot helper confirms 1 merge for direct compress() versus 2 for Pinot serialization; the existing test pins explicit compress()+serialize to the same policy as serialize alone. This does not promise identical centroids to direct library compress(). Validated at `3821623d46933dd233b8110e123ecb4f86ac1e96`: 554 affected tests, genuine 3.2/3.3 bidirectional byte checks, the 270-case oracle, applicable precommit checks and independent Pinot review pass. New-head CI and maintainer approval remain separate gates. -- 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]
