xiangfu0 commented on code in PR #19795: URL: https://github.com/apache/pinot/pull/19795#discussion_r4228360211
########## pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java: ########## @@ -0,0 +1,1662 @@ +/** + * 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.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.List; +import org.apache.pinot.segment.local.utils.TDigestUtils; +import org.apache.pinot.segment.local.utils.TDigestUtils.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. +/// +/// 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 MIN_COMPRESSION = 10; + 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 SerializedTDigestMetadata _pendingSerializedMetadata; + private boolean _legacyDegraded; + private byte[] _serializedBytesForWrite; + private boolean _hasFractionalWeights; + + private int _numRawValues; + private int _numCentroids; + private int _numIncomingCentroids; + private int _serializedMainCapacity; + private int _serializedBufferCapacity; + 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(positiveCompression(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) { + TDigestUtils.validateCompression(compression); + _useTwoLevelRawCompression = useTwoLevelRawCompression; + _compression = Math.max(MIN_COMPRESSION, compression); + _workingCompression = TWO_LEVEL_COMPRESSION_MULTIPLIER * _compression; + _publicMaxWeightScale = calculateK1MaxWeightScale(_compression); + _workingMaxWeightScale = calculateK1MaxWeightScale(_workingCompression); + long roundedCompression = (long) Math.ceil(_compression); + _centroidCapacity = TDigestUtils.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); + } + + public static PercentileTDigestAccumulator forSerializedTDigest(byte[] bytes) { + return new PercentileTDigestAccumulator(readCompression(bytes), false, false); + } + + public static PercentileTDigestAccumulator forSerializedTDigest(SerializedTDigestInput input) { + return new PercentileTDigestAccumulator(input._metadata.compression(), false, false); + } + + public static PercentileTDigestAccumulator forSerializedTDigestWithMergeBuffers(byte[] bytes) { + return new PercentileTDigestAccumulator(readCompression(bytes), false, true); + } + + public static PercentileTDigestAccumulator forSerializedTDigest(ByteBuffer input) { + SerializedTDigestInput serialized = new SerializedTDigestInput(); + serialized.reset(input); + PercentileTDigestAccumulator accumulator = forSerializedTDigest(serialized); + accumulator.addSerializedTDigest(serialized); + return accumulator; + } + + public static PercentileTDigestAccumulator forReduction(double compression) { + return new PercentileTDigestAccumulator(positiveCompression(compression), false, false); + } + + /// 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) { + 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; + } + materializePendingSerializedTDigest(); + _originalFractionalBytes = null; + _hasFractionalWeights |= fractionalWeights; + 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) { + it.unimi.dsi.fastutil.Arrays.mergeSort(0, _numCentroids, + (first, second) -> Double.compare(_centroidMeans[first], _centroidMeans[second]), + (first, second) -> { + double mean = _centroidMeans[first]; + _centroidMeans[first] = _centroidMeans[second]; + _centroidMeans[second] = mean; + double weight = _centroidWeights[first]; + _centroidWeights[first] = _centroidWeights[second]; + _centroidWeights[second] = weight; + }); + } + _totalWeight = totalWeight; + _min = min; + _max = max; + normalizeBoundaryCentroids(); + _publiclyCompressed = true; + } else { + for (int i = 0; i < count; i++) { + if (weights[i] > 0.0) { + bufferIncomingCentroid(means[i], weights[i]); + } + } + _min = Math.min(_min, min); + _max = Math.max(_max, max); + } + } + + public void add(double[] values, int from, int toExclusive) { + if (from >= toExclusive) { + return; + } + requireMutable(); + materializePendingSerializedTDigest(); + while (from < toExclusive) { + ensureRawBuffer(); + if (_numRawValues == _rawValues.length + || _numRawValues + _numIncomingCentroids == getPendingInputLimit()) { + flush(); + } + 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; + if (_numRawValues == _rawValues.length + && (!_useTwoLevelRawCompression || _rawValues.length == _pendingCentroidCapacity) + || _numRawValues + _numIncomingCentroids == getPendingInputLimit()) { + flush(); + } + } + } + + @Override + public void add(double value) { + if (Double.isNaN(value)) { + throw new IllegalArgumentException("Cannot add NaN to t-digest"); + } + requireMutable(); + materializePendingSerializedTDigest(); + ensureRawBuffer(); + if (_numRawValues == _rawValues.length + || _numRawValues + _numIncomingCentroids == getPendingInputLimit()) { + flush(); + } + _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(); + materializePendingSerializedTDigest(); + bufferIncomingCentroid(value, weight); + _originalFractionalBytes = null; + _hasFractionalWeights |= weight != Math.rint(weight); + _min = Math.min(_min, value); + _max = Math.max(_max, value); + } + + @Override + public void add(TDigest other) { + requireMutable(); + if (other instanceof PercentileTDigestAccumulator) { + addAccumulator((PercentileTDigestAccumulator) other); + return; + } + if (!other.hasValidStatistics()) { + materializePendingSerializedTDigest(); + SerializedTDigestInput input = getSerializedTDigestInput(); + input.reset(ByteBuffer.wrap(TDigestUtils.serialize(other)), false); + addSerializedTDigest(input); + return; + } + materializePendingSerializedTDigest(); + double otherWeight = other.getTotalWeight(); + checkTotalWeight(_totalWeight + _incomingWeight + _numRawValues + otherWeight); + for (Centroid centroid : other.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"); + } + _hasFractionalWeights |= weight != Math.rint(weight); + bufferIncomingCentroid(centroid.mean(), weight); + _originalFractionalBytes = null; + } + if (otherWeight != 0.0) { + _min = Math.min(_min, other.getMin()); + _max = Math.max(_max, other.getMax()); + } + } + + @Override + public void add(List<? extends TDigest> others) { + for (TDigest other : others) { + add(other); + } + } + + private void addAccumulator(PercentileTDigestAccumulator other) { + if (other == this) { + compress(); + if (_numCentroids == 0) { + return; + } + // Snapshot only self-merges; fresh fractional mass need not be representable by a legacy wire payload. + addCentroids(Arrays.copyOf(_centroidMeans, _numCentroids), Arrays.copyOf(_centroidWeights, _numCentroids), + _numCentroids, _min, _max, false); + return; + } + if (other._originalFractionalBytes != null && _compression == other._compression && 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(); + materializePendingSerializedTDigest(); + checkTotalWeight(_totalWeight + _incomingWeight + _numRawValues + other._totalWeight); + _hasFractionalWeights |= other._hasFractionalWeights; + preserveSerializedCapacity(other._serializedMainCapacity, other._serializedBufferCapacity); + bufferIncomingCentroids(other._centroidMeans, other._centroidWeights, other._numCentroids); + if (other._totalWeight > 0.0) { + _originalFractionalBytes = null; + } + _min = Math.min(_min, other._min); + _max = Math.max(_max, other._max); + } + + public void addSerializedTDigest(byte[] bytes) { + 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) { + requireMutable(); + if (prepareSerializedTDigest(input)) { + mergeSerializedTDigest(input, false); + } + } + + public void addSerializedTDigestDirect(SerializedTDigestInput input) { + requireMutable(); + if (prepareSerializedTDigest(input)) { + mergeSerializedTDigest(input, true); + } + } + + private boolean prepareSerializedTDigest(SerializedTDigestInput input) { + if (_pendingSerializedTDigest == null && hasNoInputs()) { + input.inspectMetadata(); + _hasFractionalWeights |= input._metadata.fractionalWeights(); + _pendingSerializedTDigest = input.retainBytes(); + _pendingSerializedMetadata = input._metadata; + if (input.hasUnrepresentableBoundaryMass() && _compression == input._metadata.compression()) { Review Comment: **[severe] Byte-exact retention of historical fractional-boundary payloads is dropped when the accumulator's compression differs, making an unchanged payload unserializable.** This gate requires `_compression == input._metadata.compression()`, but the payload is still adopted as `_pendingSerializedTDigest` without the `_originalFractionalBytes` alias. For a sub-1.0 boundary weight, `weightedBoundaries()` is false (it only flags weight > 1.0), so `serialize()` takes the pending fast path into `makeLegacyCompatible` -> `checkSerializedBoundaryWeights` and throws `"Cannot serialize fractional TDigest boundary mass for legacy readers"` even though nothing was merged. Reachable via MergeRollup: `PercentileTDigestAggregator.aggregate` -> `mergeWithConfiguredCompression` -> `forReduction(_compressionFactor)` re-adopts stored digests at the configured compression (javadoc: "its configured output compression can differ from the compression stored in either input digest"), so any table whose stored digests use a different compression than the rollup config permanently fails the task on data that round-trips fine at matching compression. `PercentileTDigestAccumula torTest` only exercises the matching-compression copy (`forReduction(stored.compression())`). The alias (or equivalent retention) should not depend on the adopting accumulator's compression for a payload that is never mutated. ########## pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/PercentileTDigestValueAggregator.java: ########## @@ -515,71 +547,43 @@ private List<Centroid> getAllCentroids() { } private byte[] getSerializedBytes() { - return getSerializedBytes(null); - } - - private byte[] getSerializedBytes(ByteBuffer scratchBuffer) { + if (_originalFractionalBytes != null) { + return _originalFractionalBytes; + } if (_serializedBytes == null) { - byte[] finiteBytes = getFiniteSerializedBytes(scratchBuffer); - if (_negativeInfinityWeight == 0L && _positiveInfinityWeight == 0L) { - _serializedBytes = finiteBytes; + if (_negativeInfinityWeight == 0.0 && _positiveInfinityWeight == 0.0) { + _serializedBytes = getFiniteSerializedBytes(); return _serializedBytes; } - - ByteBuffer finite = ByteBuffer.wrap(finiteBytes); - int encoding = finite.getInt(); - finite.position(finite.position() + 2 * Double.BYTES); - int finiteCentroidCount; - boolean finiteVerbose; - if (encoding == VERBOSE_ENCODING) { - finite.getDouble(); - finiteCentroidCount = finite.getInt(); - finiteVerbose = true; - } else if (encoding == SMALL_ENCODING) { - finite.getFloat(); - finite.getShort(); - finite.getShort(); - finiteCentroidCount = finite.getShort(); - finiteVerbose = false; - } else { - throw new IllegalStateException("Invalid format for serialized histogram"); - } - - boolean hasFiniteValues = _finiteDigest.size() > 0L; + _finiteDigest.compress(); + invalidateCaches(); + int finiteCentroidCount = _finiteDigest.centroidCount(); + boolean hasFiniteValues = _finiteDigest.getTotalWeight() > 0.0; int negativeInfinityCentroidCount = getInfinityCentroidCount(_negativeInfinityWeight, true, - !hasFiniteValues && _positiveInfinityWeight == 0L); + !hasFiniteValues && _positiveInfinityWeight == 0.0); int positiveInfinityCentroidCount = getInfinityCentroidCount(_positiveInfinityWeight, - !hasFiniteValues && _negativeInfinityWeight == 0L, true); + !hasFiniteValues && _negativeInfinityWeight == 0.0, true); int centroidCount = Math.addExact(finiteCentroidCount, Math.addExact(negativeInfinityCentroidCount, positiveInfinityCentroidCount)); - ByteBuffer verbose = ByteBuffer.allocate(VERBOSE_HEADER_SIZE + VERBOSE_CENTROID_SIZE * centroidCount); - verbose.putInt(VERBOSE_ENCODING); - verbose.putDouble(getMin()); - verbose.putDouble(getMax()); - verbose.putDouble(compression()); - verbose.putInt(centroidCount); - appendInfinityCentroids(verbose, Double.NEGATIVE_INFINITY, _negativeInfinityWeight, true, - !hasFiniteValues && _positiveInfinityWeight == 0L); - for (int i = 0; i < finiteCentroidCount; i++) { - verbose.putDouble(finiteVerbose ? finite.getDouble() : finite.getFloat()); - verbose.putDouble(finiteVerbose ? finite.getDouble() : finite.getFloat()); - } - appendInfinityCentroids(verbose, Double.POSITIVE_INFINITY, _positiveInfinityWeight, - !hasFiniteValues && _negativeInfinityWeight == 0L, true); - _serializedBytes = TDigestUtils.makeLegacyCompatible(verbose.array()); + double[] means = new double[centroidCount]; + double[] weights = new double[centroidCount]; + ((PercentileTDigestAccumulator) _finiteDigest).copyCentroids(means, weights, negativeInfinityCentroidCount); Review Comment: **[bug] Infinity-tail serialization can emit a weighted-boundary wire payload: `copyCentroids` skips boundary normalization.** This path writes the finite digest's centroids verbatim between the infinity tails: `_finiteDigest.compress()` + `copyCentroids(...)` never call `normalizeBoundaryCentroids()` ("This does not apply public compression", per `copyCentroids`' own doc), and the `unitWeightAtStart/End` flags only split the infinity masses. A historical weighted-boundary payload (first/last weight > 1.0 — a shape `weightedBoundaries()` metadata explicitly supports) adopted at matching compression keeps `_publiclyCompressed = true`, making `compress()` a no-op, so the emitted payload's first finite centroid carries weight > 1.0. `makeLegacyCompatible`'s `checkLegacyBoundaryWeights` only rejects weights < 1.0, so the bytes go out. t-digest 3.3's `MergingDigest.merge` asserts `weight[0] == 1` / `weight[lastUsedCell - 1] == 1` — the exact invariant `compatibility-verifier/tdigest-compatibility/LegacyReader.java` enforces under `-ea` — and new-Pinot readers classify the fresh payload as `weightedBounda ries`, forcing materialization/rewrite on every hop. The pre-rewrite wrapper spliced bytes from `TDigestUtils.serialize(_finiteDigest)`, which normalizes boundaries; this is the one wire-writing path that lost that guarantee. No test feeds a weighted-boundary finite source plus infinity mass and asserts unit boundary weights in the output. ########## pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java: ########## @@ -0,0 +1,1662 @@ +/** + * 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.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.List; +import org.apache.pinot.segment.local.utils.TDigestUtils; +import org.apache.pinot.segment.local.utils.TDigestUtils.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. +/// +/// 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 MIN_COMPRESSION = 10; + 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 SerializedTDigestMetadata _pendingSerializedMetadata; + private boolean _legacyDegraded; + private byte[] _serializedBytesForWrite; + private boolean _hasFractionalWeights; + + private int _numRawValues; + private int _numCentroids; + private int _numIncomingCentroids; + private int _serializedMainCapacity; + private int _serializedBufferCapacity; + 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(positiveCompression(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) { + TDigestUtils.validateCompression(compression); + _useTwoLevelRawCompression = useTwoLevelRawCompression; + _compression = Math.max(MIN_COMPRESSION, compression); + _workingCompression = TWO_LEVEL_COMPRESSION_MULTIPLIER * _compression; + _publicMaxWeightScale = calculateK1MaxWeightScale(_compression); + _workingMaxWeightScale = calculateK1MaxWeightScale(_workingCompression); + long roundedCompression = (long) Math.ceil(_compression); + _centroidCapacity = TDigestUtils.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); + } + + public static PercentileTDigestAccumulator forSerializedTDigest(byte[] bytes) { + return new PercentileTDigestAccumulator(readCompression(bytes), false, false); + } + + public static PercentileTDigestAccumulator forSerializedTDigest(SerializedTDigestInput input) { + return new PercentileTDigestAccumulator(input._metadata.compression(), false, false); + } + + public static PercentileTDigestAccumulator forSerializedTDigestWithMergeBuffers(byte[] bytes) { + return new PercentileTDigestAccumulator(readCompression(bytes), false, true); + } + + public static PercentileTDigestAccumulator forSerializedTDigest(ByteBuffer input) { + SerializedTDigestInput serialized = new SerializedTDigestInput(); + serialized.reset(input); + PercentileTDigestAccumulator accumulator = forSerializedTDigest(serialized); + accumulator.addSerializedTDigest(serialized); + return accumulator; + } + + public static PercentileTDigestAccumulator forReduction(double compression) { + return new PercentileTDigestAccumulator(positiveCompression(compression), false, false); + } + + /// 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) { + 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; + } + materializePendingSerializedTDigest(); + _originalFractionalBytes = null; + _hasFractionalWeights |= fractionalWeights; + 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) { + it.unimi.dsi.fastutil.Arrays.mergeSort(0, _numCentroids, + (first, second) -> Double.compare(_centroidMeans[first], _centroidMeans[second]), + (first, second) -> { + double mean = _centroidMeans[first]; + _centroidMeans[first] = _centroidMeans[second]; + _centroidMeans[second] = mean; + double weight = _centroidWeights[first]; + _centroidWeights[first] = _centroidWeights[second]; + _centroidWeights[second] = weight; + }); + } + _totalWeight = totalWeight; + _min = min; + _max = max; + normalizeBoundaryCentroids(); + _publiclyCompressed = true; + } else { + for (int i = 0; i < count; i++) { + if (weights[i] > 0.0) { + bufferIncomingCentroid(means[i], weights[i]); + } + } + _min = Math.min(_min, min); + _max = Math.max(_max, max); + } + } + + public void add(double[] values, int from, int toExclusive) { + if (from >= toExclusive) { + return; + } + requireMutable(); + materializePendingSerializedTDigest(); + while (from < toExclusive) { + ensureRawBuffer(); + if (_numRawValues == _rawValues.length + || _numRawValues + _numIncomingCentroids == getPendingInputLimit()) { + flush(); + } + 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; + if (_numRawValues == _rawValues.length + && (!_useTwoLevelRawCompression || _rawValues.length == _pendingCentroidCapacity) + || _numRawValues + _numIncomingCentroids == getPendingInputLimit()) { + flush(); + } + } + } + + @Override + public void add(double value) { + if (Double.isNaN(value)) { + throw new IllegalArgumentException("Cannot add NaN to t-digest"); + } + requireMutable(); + materializePendingSerializedTDigest(); + ensureRawBuffer(); + if (_numRawValues == _rawValues.length + || _numRawValues + _numIncomingCentroids == getPendingInputLimit()) { + flush(); + } + _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(); + materializePendingSerializedTDigest(); + bufferIncomingCentroid(value, weight); + _originalFractionalBytes = null; + _hasFractionalWeights |= weight != Math.rint(weight); + _min = Math.min(_min, value); + _max = Math.max(_max, value); + } + + @Override + public void add(TDigest other) { + requireMutable(); + if (other instanceof PercentileTDigestAccumulator) { + addAccumulator((PercentileTDigestAccumulator) other); + return; + } + if (!other.hasValidStatistics()) { + materializePendingSerializedTDigest(); + SerializedTDigestInput input = getSerializedTDigestInput(); + input.reset(ByteBuffer.wrap(TDigestUtils.serialize(other)), false); + addSerializedTDigest(input); + return; + } + materializePendingSerializedTDigest(); + double otherWeight = other.getTotalWeight(); + checkTotalWeight(_totalWeight + _incomingWeight + _numRawValues + otherWeight); + for (Centroid centroid : other.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"); + } + _hasFractionalWeights |= weight != Math.rint(weight); + bufferIncomingCentroid(centroid.mean(), weight); + _originalFractionalBytes = null; + } + if (otherWeight != 0.0) { + _min = Math.min(_min, other.getMin()); + _max = Math.max(_max, other.getMax()); + } + } + + @Override + public void add(List<? extends TDigest> others) { + for (TDigest other : others) { + add(other); + } + } + + private void addAccumulator(PercentileTDigestAccumulator other) { + if (other == this) { + compress(); + if (_numCentroids == 0) { + return; + } + // Snapshot only self-merges; fresh fractional mass need not be representable by a legacy wire payload. + addCentroids(Arrays.copyOf(_centroidMeans, _numCentroids), Arrays.copyOf(_centroidWeights, _numCentroids), + _numCentroids, _min, _max, false); + return; + } + if (other._originalFractionalBytes != null && _compression == other._compression && 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(); + materializePendingSerializedTDigest(); + checkTotalWeight(_totalWeight + _incomingWeight + _numRawValues + other._totalWeight); + _hasFractionalWeights |= other._hasFractionalWeights; + preserveSerializedCapacity(other._serializedMainCapacity, other._serializedBufferCapacity); + bufferIncomingCentroids(other._centroidMeans, other._centroidWeights, other._numCentroids); + if (other._totalWeight > 0.0) { + _originalFractionalBytes = null; + } + _min = Math.min(_min, other._min); + _max = Math.max(_max, other._max); + } + + public void addSerializedTDigest(byte[] bytes) { + 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) { + requireMutable(); + if (prepareSerializedTDigest(input)) { + mergeSerializedTDigest(input, false); + } + } + + public void addSerializedTDigestDirect(SerializedTDigestInput input) { + requireMutable(); + if (prepareSerializedTDigest(input)) { + mergeSerializedTDigest(input, true); + } + } + + private boolean prepareSerializedTDigest(SerializedTDigestInput input) { + if (_pendingSerializedTDigest == null && hasNoInputs()) { + input.inspectMetadata(); + _hasFractionalWeights |= input._metadata.fractionalWeights(); + _pendingSerializedTDigest = input.retainBytes(); + _pendingSerializedMetadata = input._metadata; + if (input.hasUnrepresentableBoundaryMass() && _compression == input._metadata.compression()) { + // 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._metadata; + _legacyDegraded = true; + return false; + } + throw corruptedMutation(); + } + checkTotalWeight(_totalWeight + _incomingWeight + _numRawValues + input._metadata.totalWeight()); + 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; + } + preserveSerializedCapacity(input._metadata.mainCapacity(), input._metadata.bufferCapacity()); + if (hasNoInputs()) { + if (input._metadata.totalWeight() != 0.0) { + 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); + } + _min = Math.min(_min, input._metadata.min()); + _max = Math.max(_max, input._metadata.max()); + } + + @Override + public void compress() { + if (_legacyDegraded) { + return; + } + if (_pendingSerializedTDigest != null || _numRawValues > 0 || _numIncomingCentroids > 0 + || !_publiclyCompressed && _numCentroids > 0) { + _serializedBytesForWrite = null; + } + materializePendingSerializedTDigest(); + if (_numRawValues > 0 || _numIncomingCentroids > 0) { + // MergingDigest 3.3 combines pending values and existing centroids directly at public compression. Flushing at + // working compression first would add an extra lossy merge pass and materially increase reducer rank error. + flush(_compression); + _publiclyCompressed = true; + } else if (!_publiclyCompressed && _numCentroids > 0) { + recompressCentroids(_compression); + _publiclyCompressed = true; + } + } + + @Override + public long size() { + return (long) getTotalWeight(); + } + + @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])) { + return (weightSoFar + _centroidWeights[i]) / _totalWeight; + } + double width = _centroidMeans[i + 1] - _centroidMeans[i]; + if (width > 0.0) { + double leftExcludedWeight = getPointMassHalfWeight(i); + double rightExcludedWeight = getPointMassHalfWeight(i + 1); + double halfWeight = (_centroidWeights[i] + _centroidWeights[i + 1]) / 2.0; + double interpolatedWeight = halfWeight - leftExcludedWeight - rightExcludedWeight; + double base = weightSoFar + _centroidWeights[i] / 2.0 + leftExcludedWeight; + return (base + interpolatedWeight * (value - _centroidMeans[i]) / width) / _totalWeight; + } + return (weightSoFar + (_centroidWeights[i] + _centroidWeights[i + 1]) / 2.0) / _totalWeight; + } + weightSoFar += _centroidWeights[i]; + } + if (value == _centroidMeans[lastIndex]) { + return 1.0 - Math.min(1.0, _centroidWeights[lastIndex]) / 2.0 / _totalWeight; + } + throw new IllegalStateException("Unable to compute TDigest CDF"); + } + + @Override + public double quantile(double quantile) { + if (Double.isNaN(quantile) || quantile < 0.0 || quantile > 1.0) { + throw new IllegalArgumentException("q should be in [0,1], got " + quantile); + } + if (_legacyDegraded) { + return Double.NaN; + } + materializePendingSerializedTDigest(); + flush(); + if (_numCentroids == 0) { + return Double.NaN; + } + if (quantile == 0.0) { + return _min; + } + if (quantile == 1.0) { + return _max; + } + if (_numCentroids == 1) { + return _centroidMeans[0]; + } + + double index = quantile * _totalWeight; + double firstWeight = _centroidWeights[0]; + if (firstWeight < 1.0 && index < firstWeight / 2.0) { + return weightedAverage(_min, firstWeight / 2.0 - index, _centroidMeans[0], index); + } + if (firstWeight >= 1.0 && index < 1.0) { + return _min; + } + if (_centroidWeights[0] > 1.0 && index < _centroidWeights[0] / 2.0) { + return _min + (index - 1.0) / (_centroidWeights[0] / 2.0 - 1.0) * (_centroidMeans[0] - _min); + } + int lastIndex = _numCentroids - 1; + double lastWeight = _centroidWeights[lastIndex]; + if (lastWeight < 1.0 && index >= _totalWeight - lastWeight / 2.0) { + return weightedAverage(_centroidMeans[lastIndex], _totalWeight - index, _max, + index - (_totalWeight - lastWeight / 2.0)); + } + if (lastWeight >= 1.0 && index > _totalWeight - 1.0) { + return _max; + } + if (_centroidWeights[lastIndex] > 1.0 + && _totalWeight - index <= _centroidWeights[lastIndex] / 2.0) { + return _max - (_totalWeight - index - 1.0) / (_centroidWeights[lastIndex] / 2.0 - 1.0) + * (_max - _centroidMeans[lastIndex]); + } + + double weightSoFar = _centroidWeights[0] / 2.0; + for (int i = 0; i < lastIndex; i++) { + double halfWeight = (_centroidWeights[i] + _centroidWeights[i + 1]) / 2.0; + if (weightSoFar + halfWeight > index) { + double leftExcludedWeight = getPointMassHalfWeight(i); + if (index - weightSoFar < leftExcludedWeight) { + return _centroidMeans[i]; + } + double rightExcludedWeight = getPointMassHalfWeight(i + 1); + if (weightSoFar + halfWeight - index <= rightExcludedWeight) { + return _centroidMeans[i + 1]; + } + double leftWeight = index - weightSoFar - leftExcludedWeight; + double rightWeight = weightSoFar + halfWeight - index - rightExcludedWeight; + return weightedAverage(_centroidMeans[i], rightWeight, _centroidMeans[i + 1], leftWeight); + } + weightSoFar += halfWeight; + } + double weightToMax = index - _totalWeight - _centroidWeights[lastIndex] / 2.0; + double weightFromLastCentroid = _centroidWeights[lastIndex] / 2.0 - weightToMax; + return weightedAverage(_centroidMeans[lastIndex], weightToMax, _max, weightFromLastCentroid); + } + + private double getPointMassHalfWeight(int index) { + // Adjacent equal means establish a repeated-value plateau. Keep its complete mass out of interpolation at + // either edge; an isolated weighted centroid still represents an unknown distribution around its mean. + double mean = _centroidMeans[index]; + boolean pointMass = _centroidWeights[index] == 1.0 || index > 0 && _centroidMeans[index - 1] == mean + || index + 1 < _numCentroids && _centroidMeans[index + 1] == mean; + return pointMass ? _centroidWeights[index] / 2.0 : 0.0; + } + + private static double weightedAverage(double firstValue, double firstWeight, double secondValue, + double secondWeight) { + if (firstValue > secondValue) { + return weightedAverage(secondValue, secondWeight, firstValue, firstWeight); + } + if (firstWeight == 0.0) { + return secondValue; + } + if (secondWeight == 0.0 || firstValue == secondValue) { + return firstValue; + } + if (!Double.isFinite(firstValue)) { + return firstValue; + } + if (!Double.isFinite(secondValue)) { + return secondValue; + } + double average = (firstValue * firstWeight + secondValue * secondWeight) / (firstWeight + secondWeight); + return Math.max(firstValue, Math.min(average, secondValue)); + } + + @Override + public Collection<Centroid> centroids() { + if (_legacyDegraded) { + byte[] bytes = serialize(); + ByteBuffer encoded = ByteBuffer.wrap(bytes); + SerializedTDigestMetadata metadata = TDigestUtils.readSerializedHeader(encoded); + encoded.position(metadata.centroidOffset()); + List<Centroid> centroids = new ArrayList<>(metadata.centroidCount()); + for (int i = 0; i < metadata.centroidCount(); i++) { + double weight = metadata.centroidSize() == TDigestUtils.VERBOSE_CENTROID_SIZE + ? encoded.getDouble() : encoded.getFloat(); + double mean = metadata.centroidSize() == TDigestUtils.VERBOSE_CENTROID_SIZE + ? encoded.getDouble() : encoded.getFloat(); + centroids.add(new Centroid(mean, weight)); + } + return centroids; + } + materializePendingSerializedTDigest(); + flush(); + List<Centroid> centroids = new ArrayList<>(_numCentroids); + for (int i = 0; i < _numCentroids; i++) { + centroids.add(new Centroid(_centroidMeans[i], _centroidWeights[i])); + } + return centroids; + } + + /// Copies a valid distribution's primitive centroid arrays after flushing buffered inputs. This does not apply + /// public compression or serialize a finite sub-distribution; callers can include surrounding infinity mass. + public void copyCentroids(double[] means, double[] weights, int offset) { + if (_legacyDegraded) { + throw corruptedMutation(); + } + materializePendingSerializedTDigest(); + flush(); + if (means == weights || offset < 0 || offset > means.length - _numCentroids + || offset > weights.length - _numCentroids) { + throw new IllegalArgumentException("Insufficient distinct TDigest centroid arrays"); + } + if (_numCentroids == 0) { + return; + } + System.arraycopy(_centroidMeans, 0, means, offset, _numCentroids); + System.arraycopy(_centroidWeights, 0, weights, offset, _numCentroids); + } + + @Override + public double compression() { + return _compression; + } + + /// Returns the length of the [#serialize()] bytes rather than the raw verbose byte size, so generic `TDigest` + /// serializers ([#byteSize()] followed by [#asBytes(ByteBuffer)]) emit the mixed-version-compatible encoding. + /// Prepared bytes are cached until the next write or mutation, avoiding a second encoding pass for that pattern. + /// Emitting raw verbose bytes here could produce a digest with more centroids than an old reader of the same + /// compression allocates, which it rejects with [ArrayIndexOutOfBoundsException]. + @Override + public int byteSize() { + if (_legacyDegraded || _originalFractionalBytes != null) { + return _legacyDegraded ? _pendingSerializedTDigest.length : _originalFractionalBytes.length; + } + if (_serializedBytesForWrite == null) { + _serializedBytesForWrite = serialize(); + } + return _serializedBytesForWrite.length; + } + + /// Bounds both compatible encodings and a widened verbose view without flushing. Buffered and fractional inputs + /// can reserve more space than their eventual bytes; retained capacities need not fit default compression size. + @Override + public int maxSerializedByteSize() { + if (_legacyDegraded) { + return _pendingSerializedTDigest.length; + } + if (_originalFractionalBytes != null) { + return _originalFractionalBytes.length; + } + long centroidCount = getCentroidCountUpperBound(); + if (!_hasFractionalWeights) { + centroidCount = (long) Math.min(centroidCount, Math.ceil(getTotalWeight())); + } + int maximumBytes = getMaxVerboseByteSize(centroidCount); + return _pendingSerializedTDigest == null ? maximumBytes : Math.max(maximumBytes, _pendingSerializedTDigest.length); + } + + private static int getMaxVerboseByteSize(long centroidCount) { + return Math.toIntExact(Math.addExact(TDigestUtils.VERBOSE_HEADER_SIZE, + Math.multiplyExact(TDigestUtils.VERBOSE_CENTROID_SIZE, centroidCount))); + } + + /// Returns the small-encoding size unless float narrowing would overflow, in which case verbose bytes are used. + /// Historical degraded state retains its original encoding and rejects mutation. + @Override + public int smallByteSize() { + if (_legacyDegraded || _originalFractionalBytes != null) { + return _legacyDegraded ? _pendingSerializedTDigest.length : _originalFractionalBytes.length; + } + compress(); + normalizeBoundaryCentroids(); + checkFractionalBoundaryEncoding(); + checkCapacityPreservingCentroidCount(); + if (!canUseSmallEncoding()) { + return byteSize(); + } + return Math.addExact(TDigestUtils.SMALL_HEADER_SIZE, + Math.multiplyExact(TDigestUtils.SMALL_CENTROID_SIZE, _numCentroids)); + } + + /// Writes the [#serialize()] bytes; see [#byteSize()]. Bytes are always written in big-endian order (the t-digest + /// wire order) regardless of the destination buffer's byte order, unlike the library's `asBytes`. + @Override + public void asBytes(ByteBuffer buffer) { + byte[] bytes = _serializedBytesForWrite; + _serializedBytesForWrite = null; + buffer.put(bytes != null ? bytes : serialize()); + } + + /// Writes the representation described by [#smallByteSize()]; readers must inspect the leading encoding identifier. + @Override + public void asSmallBytes(ByteBuffer buffer) { + if (_legacyDegraded || _originalFractionalBytes != null) { + buffer.put(serialize()); + return; + } + compress(); + normalizeBoundaryCentroids(); + checkFractionalBoundaryEncoding(); + checkCapacityPreservingCentroidCount(); + if (canUseSmallEncoding()) { + buffer.put(toCapacityPreservingBytes()); + } else { + asBytes(buffer); + } + } + + @Override + public int centroidCount() { + if (_pendingSerializedTDigest != null) { + return !_legacyDegraded && _pendingSerializedMetadata.totalWeight() == 0.0 + ? 0 : _pendingSerializedMetadata.centroidCount(); + } + flush(); + return _numCentroids; + } + + @Override + public double getMin() { + if (_pendingSerializedTDigest != null) { + return !_legacyDegraded && _pendingSerializedMetadata.totalWeight() == 0.0 + ? Double.POSITIVE_INFINITY : _pendingSerializedMetadata.min(); + } + flush(); + return _min; + } + + @Override + public double getMax() { + if (_pendingSerializedTDigest != null) { + return !_legacyDegraded && _pendingSerializedMetadata.totalWeight() == 0.0 + ? Double.NEGATIVE_INFINITY : _pendingSerializedMetadata.max(); + } + flush(); + return _max; + } + + public byte[] serialize() { + if (_legacyDegraded) { + return _pendingSerializedTDigest.clone(); + } + if (_originalFractionalBytes != null) { + return _originalFractionalBytes.clone(); + } + if (_pendingSerializedTDigest != null) { + if (_pendingSerializedMetadata.weightedBoundaries() || _pendingSerializedMetadata.hasZeroWeightCentroids() Review Comment: **[bug] `serialize()`'s pending fast path misses unordered payloads whose *sorted* boundary is weighted: `weightedBoundaries` is computed in encoded order.** `inspectSerialized` derives `firstPositiveWeight`/`lastPositiveWeight` from encoded order, while boundary normalization applies in sorted-by-mean order (`decode()` mergeSorts, then `normalizeBoundaries`). A pending verbose payload encoded out of mean order (`unorderedMeans=true` is accepted as valid first-class input) with unit-weight encoded endpoints but a weight>1 smallest-mean centroid — e.g. `[[email protected], [email protected], [email protected]]` — is re-emitted verbatim here (this condition checks `weightedBoundaries()`/`hasZeroWeightCentroids()`/`recoveredInfinityMean` but not `unorderedMeans()`), so the intermediate carries a weight-4 centroid at the sorted lower boundary and trips t-digest 3.3's unit-endpoint assertion on recompression — the exact state `makeLegacyCompatible` exists to prevent, and the same payload read through `decode()` *is* repaired. Caveat: tdunning writers emit sorted centroids, so the trigger needs a non-standard writer — but since the PR deliberately accepts unorder ed payloads as input, the serializer contract should cover them: also materialize when `_pendingSerializedMetadata.unorderedMeans()` (or compute boundary weights against sorted order in `inspectSerialized`). ########## pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java: ########## @@ -0,0 +1,1662 @@ +/** + * 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.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.List; +import org.apache.pinot.segment.local.utils.TDigestUtils; +import org.apache.pinot.segment.local.utils.TDigestUtils.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. +/// +/// 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 MIN_COMPRESSION = 10; + 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 SerializedTDigestMetadata _pendingSerializedMetadata; + private boolean _legacyDegraded; + private byte[] _serializedBytesForWrite; + private boolean _hasFractionalWeights; + + private int _numRawValues; + private int _numCentroids; + private int _numIncomingCentroids; + private int _serializedMainCapacity; + private int _serializedBufferCapacity; + 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(positiveCompression(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) { + TDigestUtils.validateCompression(compression); + _useTwoLevelRawCompression = useTwoLevelRawCompression; + _compression = Math.max(MIN_COMPRESSION, compression); + _workingCompression = TWO_LEVEL_COMPRESSION_MULTIPLIER * _compression; + _publicMaxWeightScale = calculateK1MaxWeightScale(_compression); + _workingMaxWeightScale = calculateK1MaxWeightScale(_workingCompression); + long roundedCompression = (long) Math.ceil(_compression); + _centroidCapacity = TDigestUtils.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); + } + + public static PercentileTDigestAccumulator forSerializedTDigest(byte[] bytes) { + return new PercentileTDigestAccumulator(readCompression(bytes), false, false); + } + + public static PercentileTDigestAccumulator forSerializedTDigest(SerializedTDigestInput input) { + return new PercentileTDigestAccumulator(input._metadata.compression(), false, false); + } + + public static PercentileTDigestAccumulator forSerializedTDigestWithMergeBuffers(byte[] bytes) { + return new PercentileTDigestAccumulator(readCompression(bytes), false, true); + } + + public static PercentileTDigestAccumulator forSerializedTDigest(ByteBuffer input) { + SerializedTDigestInput serialized = new SerializedTDigestInput(); + serialized.reset(input); + PercentileTDigestAccumulator accumulator = forSerializedTDigest(serialized); + accumulator.addSerializedTDigest(serialized); + return accumulator; + } + + public static PercentileTDigestAccumulator forReduction(double compression) { + return new PercentileTDigestAccumulator(positiveCompression(compression), false, false); + } + + /// 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) { + 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; + } + materializePendingSerializedTDigest(); + _originalFractionalBytes = null; + _hasFractionalWeights |= fractionalWeights; + 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) { + it.unimi.dsi.fastutil.Arrays.mergeSort(0, _numCentroids, + (first, second) -> Double.compare(_centroidMeans[first], _centroidMeans[second]), + (first, second) -> { + double mean = _centroidMeans[first]; + _centroidMeans[first] = _centroidMeans[second]; + _centroidMeans[second] = mean; + double weight = _centroidWeights[first]; + _centroidWeights[first] = _centroidWeights[second]; + _centroidWeights[second] = weight; + }); + } + _totalWeight = totalWeight; + _min = min; + _max = max; + normalizeBoundaryCentroids(); + _publiclyCompressed = true; + } else { + for (int i = 0; i < count; i++) { + if (weights[i] > 0.0) { + bufferIncomingCentroid(means[i], weights[i]); + } + } + _min = Math.min(_min, min); + _max = Math.max(_max, max); + } + } + + public void add(double[] values, int from, int toExclusive) { + if (from >= toExclusive) { + return; + } + requireMutable(); + materializePendingSerializedTDigest(); + while (from < toExclusive) { + ensureRawBuffer(); + if (_numRawValues == _rawValues.length + || _numRawValues + _numIncomingCentroids == getPendingInputLimit()) { + flush(); + } + 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; + if (_numRawValues == _rawValues.length + && (!_useTwoLevelRawCompression || _rawValues.length == _pendingCentroidCapacity) + || _numRawValues + _numIncomingCentroids == getPendingInputLimit()) { + flush(); + } + } + } + + @Override + public void add(double value) { + if (Double.isNaN(value)) { + throw new IllegalArgumentException("Cannot add NaN to t-digest"); + } + requireMutable(); + materializePendingSerializedTDigest(); + ensureRawBuffer(); + if (_numRawValues == _rawValues.length + || _numRawValues + _numIncomingCentroids == getPendingInputLimit()) { + flush(); + } + _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(); + materializePendingSerializedTDigest(); + bufferIncomingCentroid(value, weight); + _originalFractionalBytes = null; + _hasFractionalWeights |= weight != Math.rint(weight); + _min = Math.min(_min, value); + _max = Math.max(_max, value); + } + + @Override + public void add(TDigest other) { + requireMutable(); + if (other instanceof PercentileTDigestAccumulator) { + addAccumulator((PercentileTDigestAccumulator) other); + return; + } + if (!other.hasValidStatistics()) { + materializePendingSerializedTDigest(); + SerializedTDigestInput input = getSerializedTDigestInput(); + input.reset(ByteBuffer.wrap(TDigestUtils.serialize(other)), false); + addSerializedTDigest(input); + return; + } + materializePendingSerializedTDigest(); + double otherWeight = other.getTotalWeight(); + checkTotalWeight(_totalWeight + _incomingWeight + _numRawValues + otherWeight); + for (Centroid centroid : other.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"); + } + _hasFractionalWeights |= weight != Math.rint(weight); + bufferIncomingCentroid(centroid.mean(), weight); + _originalFractionalBytes = null; + } + if (otherWeight != 0.0) { + _min = Math.min(_min, other.getMin()); + _max = Math.max(_max, other.getMax()); + } + } + + @Override + public void add(List<? extends TDigest> others) { + for (TDigest other : others) { + add(other); + } + } + + private void addAccumulator(PercentileTDigestAccumulator other) { + if (other == this) { + compress(); + if (_numCentroids == 0) { + return; + } + // Snapshot only self-merges; fresh fractional mass need not be representable by a legacy wire payload. + addCentroids(Arrays.copyOf(_centroidMeans, _numCentroids), Arrays.copyOf(_centroidWeights, _numCentroids), + _numCentroids, _min, _max, false); + return; + } + if (other._originalFractionalBytes != null && _compression == other._compression && 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(); + materializePendingSerializedTDigest(); + checkTotalWeight(_totalWeight + _incomingWeight + _numRawValues + other._totalWeight); + _hasFractionalWeights |= other._hasFractionalWeights; + preserveSerializedCapacity(other._serializedMainCapacity, other._serializedBufferCapacity); + bufferIncomingCentroids(other._centroidMeans, other._centroidWeights, other._numCentroids); + if (other._totalWeight > 0.0) { + _originalFractionalBytes = null; + } + _min = Math.min(_min, other._min); + _max = Math.max(_max, other._max); + } + + public void addSerializedTDigest(byte[] bytes) { + 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) { + requireMutable(); + if (prepareSerializedTDigest(input)) { + mergeSerializedTDigest(input, false); + } + } + + public void addSerializedTDigestDirect(SerializedTDigestInput input) { + requireMutable(); Review Comment: **[bug] Row-order-dependent error vs NaN: `requireMutable()` runs before the input is inspected, so an empty digest merged into a degraded accumulator throws.** `addSerializedTDigest`/`addSerializedTDigestDirect` call `requireMutable()` before looking at the input. In the BYTES group-by path (`aggregateGroupBySV` -> `addSerializedTDigestDirect`), if the degraded row is scanned first, a subsequent zero-weight serialized digest (a supported stored shape — head special-cases `totalWeight() == 0.0` pending state) throws `"Cannot merge or mutate a historically corrupted TDigest"`, even though `mergeSerializedTDigest` would have no-opped on it anyway. In the reverse docId order (empty first, degraded second) the empty pending materializes as a no-op, `hasNoInputs()` is still true, the degraded payload is adopted, and the group returns NaN. Segment/docId ordering decides between a query error and a result. The sibling paths already filter empties first (`NonFiniteAwareTDigest.add` checks `other.isEmpty()` before `requireMutable()`; `merge()` returns the other side on `isEmpty()`); the leaf `addSerializedTDigest*` path should do the same (e.g. check `input._metadata.totalWeight() == 0.0` before `requireMutable()`). ########## pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java: ########## @@ -0,0 +1,1662 @@ +/** + * 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.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.List; +import org.apache.pinot.segment.local.utils.TDigestUtils; +import org.apache.pinot.segment.local.utils.TDigestUtils.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. +/// +/// 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 MIN_COMPRESSION = 10; + 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 SerializedTDigestMetadata _pendingSerializedMetadata; + private boolean _legacyDegraded; + private byte[] _serializedBytesForWrite; + private boolean _hasFractionalWeights; + + private int _numRawValues; + private int _numCentroids; + private int _numIncomingCentroids; + private int _serializedMainCapacity; + private int _serializedBufferCapacity; + 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(positiveCompression(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) { + TDigestUtils.validateCompression(compression); + _useTwoLevelRawCompression = useTwoLevelRawCompression; + _compression = Math.max(MIN_COMPRESSION, compression); + _workingCompression = TWO_LEVEL_COMPRESSION_MULTIPLIER * _compression; + _publicMaxWeightScale = calculateK1MaxWeightScale(_compression); + _workingMaxWeightScale = calculateK1MaxWeightScale(_workingCompression); + long roundedCompression = (long) Math.ceil(_compression); + _centroidCapacity = TDigestUtils.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); + } + + public static PercentileTDigestAccumulator forSerializedTDigest(byte[] bytes) { + return new PercentileTDigestAccumulator(readCompression(bytes), false, false); + } + + public static PercentileTDigestAccumulator forSerializedTDigest(SerializedTDigestInput input) { + return new PercentileTDigestAccumulator(input._metadata.compression(), false, false); + } + + public static PercentileTDigestAccumulator forSerializedTDigestWithMergeBuffers(byte[] bytes) { + return new PercentileTDigestAccumulator(readCompression(bytes), false, true); + } + + public static PercentileTDigestAccumulator forSerializedTDigest(ByteBuffer input) { + SerializedTDigestInput serialized = new SerializedTDigestInput(); + serialized.reset(input); + PercentileTDigestAccumulator accumulator = forSerializedTDigest(serialized); + accumulator.addSerializedTDigest(serialized); + return accumulator; + } + + public static PercentileTDigestAccumulator forReduction(double compression) { + return new PercentileTDigestAccumulator(positiveCompression(compression), false, false); + } + + /// 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) { + 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; + } + materializePendingSerializedTDigest(); + _originalFractionalBytes = null; + _hasFractionalWeights |= fractionalWeights; + 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) { + it.unimi.dsi.fastutil.Arrays.mergeSort(0, _numCentroids, + (first, second) -> Double.compare(_centroidMeans[first], _centroidMeans[second]), + (first, second) -> { + double mean = _centroidMeans[first]; + _centroidMeans[first] = _centroidMeans[second]; + _centroidMeans[second] = mean; + double weight = _centroidWeights[first]; + _centroidWeights[first] = _centroidWeights[second]; + _centroidWeights[second] = weight; + }); + } + _totalWeight = totalWeight; + _min = min; + _max = max; + normalizeBoundaryCentroids(); + _publiclyCompressed = true; + } else { + for (int i = 0; i < count; i++) { + if (weights[i] > 0.0) { + bufferIncomingCentroid(means[i], weights[i]); + } + } + _min = Math.min(_min, min); + _max = Math.max(_max, max); + } + } + + public void add(double[] values, int from, int toExclusive) { + if (from >= toExclusive) { + return; + } + requireMutable(); + materializePendingSerializedTDigest(); + while (from < toExclusive) { + ensureRawBuffer(); + if (_numRawValues == _rawValues.length + || _numRawValues + _numIncomingCentroids == getPendingInputLimit()) { + flush(); + } + 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; + if (_numRawValues == _rawValues.length + && (!_useTwoLevelRawCompression || _rawValues.length == _pendingCentroidCapacity) + || _numRawValues + _numIncomingCentroids == getPendingInputLimit()) { + flush(); + } + } + } + + @Override + public void add(double value) { + if (Double.isNaN(value)) { + throw new IllegalArgumentException("Cannot add NaN to t-digest"); + } + requireMutable(); + materializePendingSerializedTDigest(); + ensureRawBuffer(); + if (_numRawValues == _rawValues.length + || _numRawValues + _numIncomingCentroids == getPendingInputLimit()) { + flush(); + } + _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(); + materializePendingSerializedTDigest(); + bufferIncomingCentroid(value, weight); + _originalFractionalBytes = null; + _hasFractionalWeights |= weight != Math.rint(weight); + _min = Math.min(_min, value); + _max = Math.max(_max, value); + } + + @Override + public void add(TDigest other) { + requireMutable(); + if (other instanceof PercentileTDigestAccumulator) { + addAccumulator((PercentileTDigestAccumulator) other); + return; + } + if (!other.hasValidStatistics()) { + materializePendingSerializedTDigest(); + SerializedTDigestInput input = getSerializedTDigestInput(); + input.reset(ByteBuffer.wrap(TDigestUtils.serialize(other)), false); + addSerializedTDigest(input); + return; + } + materializePendingSerializedTDigest(); + double otherWeight = other.getTotalWeight(); + checkTotalWeight(_totalWeight + _incomingWeight + _numRawValues + otherWeight); + for (Centroid centroid : other.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"); + } + _hasFractionalWeights |= weight != Math.rint(weight); + bufferIncomingCentroid(centroid.mean(), weight); + _originalFractionalBytes = null; + } + if (otherWeight != 0.0) { + _min = Math.min(_min, other.getMin()); + _max = Math.max(_max, other.getMax()); + } + } + + @Override + public void add(List<? extends TDigest> others) { + for (TDigest other : others) { + add(other); + } + } + + private void addAccumulator(PercentileTDigestAccumulator other) { + if (other == this) { + compress(); + if (_numCentroids == 0) { + return; + } + // Snapshot only self-merges; fresh fractional mass need not be representable by a legacy wire payload. + addCentroids(Arrays.copyOf(_centroidMeans, _numCentroids), Arrays.copyOf(_centroidWeights, _numCentroids), + _numCentroids, _min, _max, false); + return; + } + if (other._originalFractionalBytes != null && _compression == other._compression && 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(); + materializePendingSerializedTDigest(); + checkTotalWeight(_totalWeight + _incomingWeight + _numRawValues + other._totalWeight); + _hasFractionalWeights |= other._hasFractionalWeights; + preserveSerializedCapacity(other._serializedMainCapacity, other._serializedBufferCapacity); + bufferIncomingCentroids(other._centroidMeans, other._centroidWeights, other._numCentroids); + if (other._totalWeight > 0.0) { + _originalFractionalBytes = null; + } + _min = Math.min(_min, other._min); + _max = Math.max(_max, other._max); + } + + public void addSerializedTDigest(byte[] bytes) { + 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) { + requireMutable(); + if (prepareSerializedTDigest(input)) { + mergeSerializedTDigest(input, false); + } + } + + public void addSerializedTDigestDirect(SerializedTDigestInput input) { + requireMutable(); + if (prepareSerializedTDigest(input)) { + mergeSerializedTDigest(input, true); + } + } + + private boolean prepareSerializedTDigest(SerializedTDigestInput input) { + if (_pendingSerializedTDigest == null && hasNoInputs()) { + input.inspectMetadata(); + _hasFractionalWeights |= input._metadata.fractionalWeights(); + _pendingSerializedTDigest = input.retainBytes(); + _pendingSerializedMetadata = input._metadata; + if (input.hasUnrepresentableBoundaryMass() && _compression == input._metadata.compression()) { + // 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._metadata; + _legacyDegraded = true; + return false; + } + throw corruptedMutation(); + } + checkTotalWeight(_totalWeight + _incomingWeight + _numRawValues + input._metadata.totalWeight()); + 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; + } + preserveSerializedCapacity(input._metadata.mainCapacity(), input._metadata.bufferCapacity()); Review Comment: **[compat/efficiency] Synthesized verbose `mainCapacity` is fed into capacity preservation, stickily inflating compact-encoding headers.** For VERBOSE payloads, `readSerializedHeader` synthesizes `mainCapacity = max(getDefaultCentroidCapacity, getLegacyDefaultCentroidCapacity)` purely as a validation bound, but this line treats it as a source-declared buffer capacity. An accumulator at compression 100 that ingests one stored verbose digest written at compression 1000 gets `_serializedMainCapacity` ~2010 permanently (`preserveSerializedCapacity` is max-only), so every later `asSmallBytes`/`toCapacityPreservingBytes` emits headers with mainCapacity ~2010 / bufferCapacity ~10050 for a digest holding <= ~230 centroids. Legacy `MergingDigest.fromBytes` allocates its arrays from those header shorts (~10x memory per deserialization), and readers re-preserve declared capacities, so the inflation propagates across hops. Suggest passing 0 (or the declared value only for SMALL encoding) here, keeping the synthesized verbose bound for the capacity check only. ########## pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java: ########## @@ -0,0 +1,1662 @@ +/** + * 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.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.List; +import org.apache.pinot.segment.local.utils.TDigestUtils; +import org.apache.pinot.segment.local.utils.TDigestUtils.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. +/// +/// 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 MIN_COMPRESSION = 10; + 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 SerializedTDigestMetadata _pendingSerializedMetadata; + private boolean _legacyDegraded; + private byte[] _serializedBytesForWrite; + private boolean _hasFractionalWeights; + + private int _numRawValues; + private int _numCentroids; + private int _numIncomingCentroids; + private int _serializedMainCapacity; + private int _serializedBufferCapacity; + 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(positiveCompression(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) { + TDigestUtils.validateCompression(compression); + _useTwoLevelRawCompression = useTwoLevelRawCompression; + _compression = Math.max(MIN_COMPRESSION, compression); + _workingCompression = TWO_LEVEL_COMPRESSION_MULTIPLIER * _compression; + _publicMaxWeightScale = calculateK1MaxWeightScale(_compression); + _workingMaxWeightScale = calculateK1MaxWeightScale(_workingCompression); + long roundedCompression = (long) Math.ceil(_compression); + _centroidCapacity = TDigestUtils.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); + } + + public static PercentileTDigestAccumulator forSerializedTDigest(byte[] bytes) { + return new PercentileTDigestAccumulator(readCompression(bytes), false, false); + } + + public static PercentileTDigestAccumulator forSerializedTDigest(SerializedTDigestInput input) { + return new PercentileTDigestAccumulator(input._metadata.compression(), false, false); + } + + public static PercentileTDigestAccumulator forSerializedTDigestWithMergeBuffers(byte[] bytes) { + return new PercentileTDigestAccumulator(readCompression(bytes), false, true); + } + + public static PercentileTDigestAccumulator forSerializedTDigest(ByteBuffer input) { + SerializedTDigestInput serialized = new SerializedTDigestInput(); + serialized.reset(input); + PercentileTDigestAccumulator accumulator = forSerializedTDigest(serialized); + accumulator.addSerializedTDigest(serialized); + return accumulator; + } + + public static PercentileTDigestAccumulator forReduction(double compression) { + return new PercentileTDigestAccumulator(positiveCompression(compression), false, false); + } + + /// 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) { + 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; + } + materializePendingSerializedTDigest(); + _originalFractionalBytes = null; + _hasFractionalWeights |= fractionalWeights; + 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) { + it.unimi.dsi.fastutil.Arrays.mergeSort(0, _numCentroids, + (first, second) -> Double.compare(_centroidMeans[first], _centroidMeans[second]), + (first, second) -> { + double mean = _centroidMeans[first]; + _centroidMeans[first] = _centroidMeans[second]; + _centroidMeans[second] = mean; + double weight = _centroidWeights[first]; + _centroidWeights[first] = _centroidWeights[second]; + _centroidWeights[second] = weight; + }); + } + _totalWeight = totalWeight; + _min = min; + _max = max; + normalizeBoundaryCentroids(); + _publiclyCompressed = true; + } else { + for (int i = 0; i < count; i++) { + if (weights[i] > 0.0) { + bufferIncomingCentroid(means[i], weights[i]); + } + } + _min = Math.min(_min, min); + _max = Math.max(_max, max); + } + } + + public void add(double[] values, int from, int toExclusive) { + if (from >= toExclusive) { + return; + } + requireMutable(); + materializePendingSerializedTDigest(); + while (from < toExclusive) { + ensureRawBuffer(); + if (_numRawValues == _rawValues.length + || _numRawValues + _numIncomingCentroids == getPendingInputLimit()) { + flush(); + } + 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; + if (_numRawValues == _rawValues.length + && (!_useTwoLevelRawCompression || _rawValues.length == _pendingCentroidCapacity) + || _numRawValues + _numIncomingCentroids == getPendingInputLimit()) { + flush(); + } + } + } + + @Override + public void add(double value) { + if (Double.isNaN(value)) { + throw new IllegalArgumentException("Cannot add NaN to t-digest"); + } + requireMutable(); + materializePendingSerializedTDigest(); + ensureRawBuffer(); + if (_numRawValues == _rawValues.length + || _numRawValues + _numIncomingCentroids == getPendingInputLimit()) { + flush(); + } + _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(); + materializePendingSerializedTDigest(); + bufferIncomingCentroid(value, weight); + _originalFractionalBytes = null; + _hasFractionalWeights |= weight != Math.rint(weight); + _min = Math.min(_min, value); + _max = Math.max(_max, value); + } + + @Override + public void add(TDigest other) { + requireMutable(); + if (other instanceof PercentileTDigestAccumulator) { + addAccumulator((PercentileTDigestAccumulator) other); + return; + } + if (!other.hasValidStatistics()) { + materializePendingSerializedTDigest(); + SerializedTDigestInput input = getSerializedTDigestInput(); + input.reset(ByteBuffer.wrap(TDigestUtils.serialize(other)), false); + addSerializedTDigest(input); + return; + } + materializePendingSerializedTDigest(); + double otherWeight = other.getTotalWeight(); + checkTotalWeight(_totalWeight + _incomingWeight + _numRawValues + otherWeight); + for (Centroid centroid : other.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"); + } + _hasFractionalWeights |= weight != Math.rint(weight); + bufferIncomingCentroid(centroid.mean(), weight); + _originalFractionalBytes = null; + } + if (otherWeight != 0.0) { + _min = Math.min(_min, other.getMin()); + _max = Math.max(_max, other.getMax()); + } + } + + @Override + public void add(List<? extends TDigest> others) { + for (TDigest other : others) { + add(other); + } + } + + private void addAccumulator(PercentileTDigestAccumulator other) { + if (other == this) { + compress(); + if (_numCentroids == 0) { + return; + } + // Snapshot only self-merges; fresh fractional mass need not be representable by a legacy wire payload. + addCentroids(Arrays.copyOf(_centroidMeans, _numCentroids), Arrays.copyOf(_centroidWeights, _numCentroids), + _numCentroids, _min, _max, false); + return; + } + if (other._originalFractionalBytes != null && _compression == other._compression && 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(); + materializePendingSerializedTDigest(); + checkTotalWeight(_totalWeight + _incomingWeight + _numRawValues + other._totalWeight); + _hasFractionalWeights |= other._hasFractionalWeights; + preserveSerializedCapacity(other._serializedMainCapacity, other._serializedBufferCapacity); + bufferIncomingCentroids(other._centroidMeans, other._centroidWeights, other._numCentroids); + if (other._totalWeight > 0.0) { + _originalFractionalBytes = null; + } + _min = Math.min(_min, other._min); + _max = Math.max(_max, other._max); + } + + public void addSerializedTDigest(byte[] bytes) { + 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) { + requireMutable(); + if (prepareSerializedTDigest(input)) { + mergeSerializedTDigest(input, false); + } + } + + public void addSerializedTDigestDirect(SerializedTDigestInput input) { + requireMutable(); + if (prepareSerializedTDigest(input)) { + mergeSerializedTDigest(input, true); + } + } + + private boolean prepareSerializedTDigest(SerializedTDigestInput input) { + if (_pendingSerializedTDigest == null && hasNoInputs()) { + input.inspectMetadata(); + _hasFractionalWeights |= input._metadata.fractionalWeights(); + _pendingSerializedTDigest = input.retainBytes(); + _pendingSerializedMetadata = input._metadata; + if (input.hasUnrepresentableBoundaryMass() && _compression == input._metadata.compression()) { + // 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._metadata; + _legacyDegraded = true; + return false; + } + throw corruptedMutation(); + } + checkTotalWeight(_totalWeight + _incomingWeight + _numRawValues + input._metadata.totalWeight()); + 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; + } + preserveSerializedCapacity(input._metadata.mainCapacity(), input._metadata.bufferCapacity()); + if (hasNoInputs()) { + if (input._metadata.totalWeight() != 0.0) { + 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); + } + _min = Math.min(_min, input._metadata.min()); + _max = Math.max(_max, input._metadata.max()); + } + + @Override + public void compress() { + if (_legacyDegraded) { + return; + } + if (_pendingSerializedTDigest != null || _numRawValues > 0 || _numIncomingCentroids > 0 + || !_publiclyCompressed && _numCentroids > 0) { + _serializedBytesForWrite = null; + } + materializePendingSerializedTDigest(); + if (_numRawValues > 0 || _numIncomingCentroids > 0) { + // MergingDigest 3.3 combines pending values and existing centroids directly at public compression. Flushing at + // working compression first would add an extra lossy merge pass and materially increase reducer rank error. + flush(_compression); + _publiclyCompressed = true; + } else if (!_publiclyCompressed && _numCentroids > 0) { + recompressCentroids(_compression); + _publiclyCompressed = true; + } + } + + @Override + public long size() { + return (long) getTotalWeight(); + } + + @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])) { + return (weightSoFar + _centroidWeights[i]) / _totalWeight; + } + double width = _centroidMeans[i + 1] - _centroidMeans[i]; + if (width > 0.0) { + double leftExcludedWeight = getPointMassHalfWeight(i); + double rightExcludedWeight = getPointMassHalfWeight(i + 1); + double halfWeight = (_centroidWeights[i] + _centroidWeights[i + 1]) / 2.0; + double interpolatedWeight = halfWeight - leftExcludedWeight - rightExcludedWeight; + double base = weightSoFar + _centroidWeights[i] / 2.0 + leftExcludedWeight; + return (base + interpolatedWeight * (value - _centroidMeans[i]) / width) / _totalWeight; + } + return (weightSoFar + (_centroidWeights[i] + _centroidWeights[i + 1]) / 2.0) / _totalWeight; + } + weightSoFar += _centroidWeights[i]; + } + if (value == _centroidMeans[lastIndex]) { + return 1.0 - Math.min(1.0, _centroidWeights[lastIndex]) / 2.0 / _totalWeight; + } + throw new IllegalStateException("Unable to compute TDigest CDF"); + } + + @Override + public double quantile(double quantile) { + if (Double.isNaN(quantile) || quantile < 0.0 || quantile > 1.0) { + throw new IllegalArgumentException("q should be in [0,1], got " + quantile); + } + if (_legacyDegraded) { + return Double.NaN; + } + materializePendingSerializedTDigest(); + flush(); + if (_numCentroids == 0) { + return Double.NaN; + } + if (quantile == 0.0) { + return _min; + } + if (quantile == 1.0) { + return _max; + } + if (_numCentroids == 1) { + return _centroidMeans[0]; + } + + double index = quantile * _totalWeight; + double firstWeight = _centroidWeights[0]; + if (firstWeight < 1.0 && index < firstWeight / 2.0) { + return weightedAverage(_min, firstWeight / 2.0 - index, _centroidMeans[0], index); + } + if (firstWeight >= 1.0 && index < 1.0) { + return _min; + } + if (_centroidWeights[0] > 1.0 && index < _centroidWeights[0] / 2.0) { + return _min + (index - 1.0) / (_centroidWeights[0] / 2.0 - 1.0) * (_centroidMeans[0] - _min); + } + int lastIndex = _numCentroids - 1; + double lastWeight = _centroidWeights[lastIndex]; + if (lastWeight < 1.0 && index >= _totalWeight - lastWeight / 2.0) { + return weightedAverage(_centroidMeans[lastIndex], _totalWeight - index, _max, + index - (_totalWeight - lastWeight / 2.0)); + } + if (lastWeight >= 1.0 && index > _totalWeight - 1.0) { + return _max; + } + if (_centroidWeights[lastIndex] > 1.0 + && _totalWeight - index <= _centroidWeights[lastIndex] / 2.0) { + return _max - (_totalWeight - index - 1.0) / (_centroidWeights[lastIndex] / 2.0 - 1.0) + * (_max - _centroidMeans[lastIndex]); + } + + double weightSoFar = _centroidWeights[0] / 2.0; + for (int i = 0; i < lastIndex; i++) { + double halfWeight = (_centroidWeights[i] + _centroidWeights[i + 1]) / 2.0; + if (weightSoFar + halfWeight > index) { + double leftExcludedWeight = getPointMassHalfWeight(i); + if (index - weightSoFar < leftExcludedWeight) { + return _centroidMeans[i]; + } + double rightExcludedWeight = getPointMassHalfWeight(i + 1); + if (weightSoFar + halfWeight - index <= rightExcludedWeight) { + return _centroidMeans[i + 1]; + } + double leftWeight = index - weightSoFar - leftExcludedWeight; + double rightWeight = weightSoFar + halfWeight - index - rightExcludedWeight; + return weightedAverage(_centroidMeans[i], rightWeight, _centroidMeans[i + 1], leftWeight); + } + weightSoFar += halfWeight; + } + double weightToMax = index - _totalWeight - _centroidWeights[lastIndex] / 2.0; + double weightFromLastCentroid = _centroidWeights[lastIndex] / 2.0 - weightToMax; + return weightedAverage(_centroidMeans[lastIndex], weightToMax, _max, weightFromLastCentroid); + } + + private double getPointMassHalfWeight(int index) { + // Adjacent equal means establish a repeated-value plateau. Keep its complete mass out of interpolation at + // either edge; an isolated weighted centroid still represents an unknown distribution around its mean. + double mean = _centroidMeans[index]; + boolean pointMass = _centroidWeights[index] == 1.0 || index > 0 && _centroidMeans[index - 1] == mean + || index + 1 < _numCentroids && _centroidMeans[index + 1] == mean; + return pointMass ? _centroidWeights[index] / 2.0 : 0.0; + } + + private static double weightedAverage(double firstValue, double firstWeight, double secondValue, + double secondWeight) { + if (firstValue > secondValue) { + return weightedAverage(secondValue, secondWeight, firstValue, firstWeight); + } + if (firstWeight == 0.0) { + return secondValue; + } + if (secondWeight == 0.0 || firstValue == secondValue) { + return firstValue; + } + if (!Double.isFinite(firstValue)) { + return firstValue; + } + if (!Double.isFinite(secondValue)) { + return secondValue; + } + double average = (firstValue * firstWeight + secondValue * secondWeight) / (firstWeight + secondWeight); + return Math.max(firstValue, Math.min(average, secondValue)); + } + + @Override + public Collection<Centroid> centroids() { + if (_legacyDegraded) { + byte[] bytes = serialize(); + ByteBuffer encoded = ByteBuffer.wrap(bytes); + SerializedTDigestMetadata metadata = TDigestUtils.readSerializedHeader(encoded); + encoded.position(metadata.centroidOffset()); + List<Centroid> centroids = new ArrayList<>(metadata.centroidCount()); + for (int i = 0; i < metadata.centroidCount(); i++) { + double weight = metadata.centroidSize() == TDigestUtils.VERBOSE_CENTROID_SIZE + ? encoded.getDouble() : encoded.getFloat(); + double mean = metadata.centroidSize() == TDigestUtils.VERBOSE_CENTROID_SIZE + ? encoded.getDouble() : encoded.getFloat(); + centroids.add(new Centroid(mean, weight)); + } + return centroids; + } + materializePendingSerializedTDigest(); + flush(); + List<Centroid> centroids = new ArrayList<>(_numCentroids); + for (int i = 0; i < _numCentroids; i++) { + centroids.add(new Centroid(_centroidMeans[i], _centroidWeights[i])); + } + return centroids; + } + + /// Copies a valid distribution's primitive centroid arrays after flushing buffered inputs. This does not apply + /// public compression or serialize a finite sub-distribution; callers can include surrounding infinity mass. + public void copyCentroids(double[] means, double[] weights, int offset) { + if (_legacyDegraded) { + throw corruptedMutation(); + } + materializePendingSerializedTDigest(); + flush(); + if (means == weights || offset < 0 || offset > means.length - _numCentroids + || offset > weights.length - _numCentroids) { + throw new IllegalArgumentException("Insufficient distinct TDigest centroid arrays"); + } + if (_numCentroids == 0) { + return; + } + System.arraycopy(_centroidMeans, 0, means, offset, _numCentroids); + System.arraycopy(_centroidWeights, 0, weights, offset, _numCentroids); + } + + @Override + public double compression() { + return _compression; + } + + /// Returns the length of the [#serialize()] bytes rather than the raw verbose byte size, so generic `TDigest` + /// serializers ([#byteSize()] followed by [#asBytes(ByteBuffer)]) emit the mixed-version-compatible encoding. + /// Prepared bytes are cached until the next write or mutation, avoiding a second encoding pass for that pattern. + /// Emitting raw verbose bytes here could produce a digest with more centroids than an old reader of the same + /// compression allocates, which it rejects with [ArrayIndexOutOfBoundsException]. + @Override + public int byteSize() { + if (_legacyDegraded || _originalFractionalBytes != null) { + return _legacyDegraded ? _pendingSerializedTDigest.length : _originalFractionalBytes.length; + } + if (_serializedBytesForWrite == null) { + _serializedBytesForWrite = serialize(); + } + return _serializedBytesForWrite.length; + } + + /// Bounds both compatible encodings and a widened verbose view without flushing. Buffered and fractional inputs + /// can reserve more space than their eventual bytes; retained capacities need not fit default compression size. + @Override + public int maxSerializedByteSize() { + if (_legacyDegraded) { + return _pendingSerializedTDigest.length; + } + if (_originalFractionalBytes != null) { + return _originalFractionalBytes.length; + } + long centroidCount = getCentroidCountUpperBound(); + if (!_hasFractionalWeights) { + centroidCount = (long) Math.min(centroidCount, Math.ceil(getTotalWeight())); + } + int maximumBytes = getMaxVerboseByteSize(centroidCount); + return _pendingSerializedTDigest == null ? maximumBytes : Math.max(maximumBytes, _pendingSerializedTDigest.length); + } + + private static int getMaxVerboseByteSize(long centroidCount) { + return Math.toIntExact(Math.addExact(TDigestUtils.VERBOSE_HEADER_SIZE, + Math.multiplyExact(TDigestUtils.VERBOSE_CENTROID_SIZE, centroidCount))); + } + + /// Returns the small-encoding size unless float narrowing would overflow, in which case verbose bytes are used. + /// Historical degraded state retains its original encoding and rejects mutation. + @Override + public int smallByteSize() { + if (_legacyDegraded || _originalFractionalBytes != null) { + return _legacyDegraded ? _pendingSerializedTDigest.length : _originalFractionalBytes.length; + } + compress(); + normalizeBoundaryCentroids(); + checkFractionalBoundaryEncoding(); + checkCapacityPreservingCentroidCount(); + if (!canUseSmallEncoding()) { + return byteSize(); + } + return Math.addExact(TDigestUtils.SMALL_HEADER_SIZE, + Math.multiplyExact(TDigestUtils.SMALL_CENTROID_SIZE, _numCentroids)); + } + + /// Writes the [#serialize()] bytes; see [#byteSize()]. Bytes are always written in big-endian order (the t-digest + /// wire order) regardless of the destination buffer's byte order, unlike the library's `asBytes`. + @Override + public void asBytes(ByteBuffer buffer) { + byte[] bytes = _serializedBytesForWrite; + _serializedBytesForWrite = null; + buffer.put(bytes != null ? bytes : serialize()); + } + + /// Writes the representation described by [#smallByteSize()]; readers must inspect the leading encoding identifier. + @Override + public void asSmallBytes(ByteBuffer buffer) { + if (_legacyDegraded || _originalFractionalBytes != null) { + buffer.put(serialize()); + return; + } + compress(); + normalizeBoundaryCentroids(); + checkFractionalBoundaryEncoding(); + checkCapacityPreservingCentroidCount(); + if (canUseSmallEncoding()) { + buffer.put(toCapacityPreservingBytes()); + } else { + asBytes(buffer); + } + } + + @Override + public int centroidCount() { + if (_pendingSerializedTDigest != null) { + return !_legacyDegraded && _pendingSerializedMetadata.totalWeight() == 0.0 + ? 0 : _pendingSerializedMetadata.centroidCount(); + } + flush(); + return _numCentroids; + } + + @Override + public double getMin() { + if (_pendingSerializedTDigest != null) { + return !_legacyDegraded && _pendingSerializedMetadata.totalWeight() == 0.0 + ? Double.POSITIVE_INFINITY : _pendingSerializedMetadata.min(); + } + flush(); + return _min; + } + + @Override + public double getMax() { + if (_pendingSerializedTDigest != null) { + return !_legacyDegraded && _pendingSerializedMetadata.totalWeight() == 0.0 + ? Double.NEGATIVE_INFINITY : _pendingSerializedMetadata.max(); + } + flush(); + return _max; + } + + public byte[] serialize() { + if (_legacyDegraded) { + return _pendingSerializedTDigest.clone(); + } + if (_originalFractionalBytes != null) { + return _originalFractionalBytes.clone(); + } + if (_pendingSerializedTDigest != null) { + if (_pendingSerializedMetadata.weightedBoundaries() || _pendingSerializedMetadata.hasZeroWeightCentroids() + || !Double.isNaN(_pendingSerializedMetadata.recoveredInfinityMean())) { + materializePendingSerializedTDigest(); + } else { + if (ByteBuffer.wrap(_pendingSerializedTDigest).getInt() == TDigestUtils.VERBOSE_ENCODING) { + byte[] serialized = TDigestUtils.makeLegacyCompatible(_pendingSerializedTDigest, _pendingSerializedMetadata); + return serialized == _pendingSerializedTDigest ? serialized.clone() : serialized; + } + return _pendingSerializedTDigest.clone(); + } + } + compress(); + normalizeBoundaryCentroids(); + checkFractionalBoundaryEncoding(); + return toLegacyCompatibleVerboseBytes(); + } + + private void recompressCentroids(double compression) { + normalizeBoundaryCentroids(); + double[] means = _centroidMeans; + double[] weights = _centroidWeights; + int numCentroids = _numCentroids; + double totalWeight = _totalWeight; + _numCentroids = 0; + _totalWeight = 0.0; + boolean runBackwards = (_mergeCount++ & 1) != 0; + if (!mergeSorted(means, weights, numCentroids, totalWeight, compression, runBackwards)) { + throw new IllegalStateException("Cannot recompress an empty TDigest"); + } + } + + private byte[] toLegacyCompatibleVerboseBytes() { + ByteBuffer buffer = ByteBuffer.allocate( + Math.addExact(TDigestUtils.VERBOSE_HEADER_SIZE, + Math.multiplyExact(TDigestUtils.VERBOSE_CENTROID_SIZE, _numCentroids))); + buffer.putInt(TDigestUtils.VERBOSE_ENCODING); + buffer.putDouble(_min); + buffer.putDouble(_max); + buffer.putDouble(_compression); + buffer.putInt(_numCentroids); + double encodedWeight = 0.0; + boolean hasNonFiniteMeans = false; + boolean fractionalWeights = false; + for (int i = 0; i < _numCentroids; i++) { + buffer.putDouble(_centroidWeights[i]); + buffer.putDouble(_centroidMeans[i]); + encodedWeight += _centroidWeights[i]; + hasNonFiniteMeans |= !Double.isFinite(_centroidMeans[i]); + fractionalWeights |= _centroidWeights[i] != Math.rint(_centroidWeights[i]); + } + checkTotalWeight(encodedWeight); + boolean weightedBoundaries = _numCentroids == 1 ? _centroidWeights[0] >= 2.0 + : _numCentroids > 1 && (_centroidWeights[0] > 1.0 || _centroidWeights[_numCentroids - 1] > 1.0); + // Collect exact wire metadata during the write rather than decoding and validating our fresh bytes again. + SerializedTDigestMetadata metadata = new SerializedTDigestMetadata(TDigestUtils.VERBOSE_ENCODING, _min, _max, + _compression, _numCentroids, Math.max(TDigestUtils.getDefaultCentroidCapacity(_compression), + TDigestUtils.getLegacyDefaultCentroidCapacity(_compression)), 0, + TDigestUtils.VERBOSE_HEADER_SIZE, TDigestUtils.VERBOSE_CENTROID_SIZE, buffer.capacity(), encodedWeight, + hasNonFiniteMeans, false, false, fractionalWeights, weightedBoundaries, false, _compression, Double.NaN); + return TDigestUtils.makeLegacyCompatible(buffer.array(), metadata); + } + + private boolean canUseSmallEncoding() { + if (!Float.isFinite((float) _compression)) { + return false; + } + for (int i = 0; i < _numCentroids; i++) { + if (!Float.isFinite((float) _centroidWeights[i]) Review Comment: **[latent bug] `canUseSmallEncoding` handles float overflow but not underflow: a subnormal weight narrows to `0.0f` and the mass is silently dropped.** `Float.isFinite((float) 1e-46)` is true (it narrows to 0.0f), so `toCapacityPreservingBytes` emits a zero-weight centroid in the compact encoding: the mass disappears for every reader (new Pinot's decode filters zero-weight centroids; the payload also gains `hasZeroWeightCentroids`, forcing rewrites), unlike the overflow case which correctly falls back to verbose. `add(value, weight)` accepts any positive finite double weight, so the shape is constructible through the public API, though nothing in production calls `asSmallBytes` today. Treat `(float) weight == 0.0f` for a positive weight the same as non-finite narrowing. ########## pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileSmartTDigestAggregationFunction.java: ########## @@ -172,7 +173,7 @@ private void aggregateIntoValueList(int length, AggregationResultHolder aggregat } else { double[][] doubleValues = blockValSet.getDoubleValuesMV(); forEachNotNull(length, blockValSet, (from, toEx) -> { - for (int i = 0; i < length; i++) { + for (int i = from; i < toEx; i++) { Review Comment: **[process] Result-changing bug fix bundled into a dependency-removal PR: the MV not-null loop previously ignored the from/to range.** This fix is correct (the old loop `for (int i = 0; i < length; i++)` re-added all rows, including null ones, for every not-null range) and is covered by `testMultiValueAggregationRespectsEachNotNullRange` — but it changes query results for `percentileSmartTDigest` over MV columns with null handling enabled, independently of the t-digest reimplementation. Worth an explicit mention in the PR description/release notes so operators investigating result shifts after upgrade don't misattribute them to the new t-digest kernel's accuracy. ########## pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/LegacyTDigestCompatibilityTest.java: ########## @@ -0,0 +1,175 @@ +/** + * 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.utils; + +import java.nio.ByteBuffer; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.SplittableRandom; +import org.apache.pinot.segment.local.customobject.TDigest; +import org.testng.annotations.Test; + +import static org.testng.Assert.assertEquals; + + +/// Exports fresh Pinot bytes for the isolated real t-digest 3.2/3.3 reader gate in the unit-test workflow. +/// This test owns its output directory and does not share mutable state with other test classes. +public class LegacyTDigestCompatibilityTest { + private static final double[] QUANTILES = {0, 0.25, 0.5, 0.75, 0.99, 1}; + + @Test + public void testExportLegacyReaderFixtures() + throws Exception { + Path directory = Path.of(System.getProperty("basedir"), "target", "tdigest-compat-fixtures"); Review Comment: **[test robustness] `Path.of(System.getProperty("basedir"), ...)` NPEs outside Maven surefire.** `basedir` is a Maven-injected property; running this class from an IDE or a direct TestNG invocation makes `System.getProperty("basedir")` return null and `Path.of` throw NullPointerException before any t-digest logic runs. Fall back to the working directory, e.g. `System.getProperty("basedir", ".")` (module test working dir is the module root under surefire too). Also note `compatibility-verifier/tdigest-compatibility/run.sh` gates on this test's output with a bare `test -s .../manifest.tsv` under `set -euo pipefail` — if the producer test is ever excluded from testset 2, CI fails with no diagnostic; an explicit error message would save debugging time. -- 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]
