xiangfu0 commented on code in PR #19795:
URL: https://github.com/apache/pinot/pull/19795#discussion_r4229471448
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/PercentileTDigestValueAggregator.java:
##########
@@ -407,6 +465,27 @@ public int byteSize() {
return getSerializedBytes().length;
}
+ @Override
+ public int maxSerializedByteSize() {
Review Comment:
**Latent contract violation: `maxSerializedByteSize()` ignores the retained
`_originalFractionalBytes` payload while `byteSize()`/`asBytes()` return
exactly those bytes.**
A retained historical fractional payload (e.g. several `(+Inf, 0.5)`
centroids) can be longer than the reconstructed bound here (the empty finite
digest's bound caps centroid count at `ceil(totalWeight)`), so `byteSize() >
maxSerializedByteSize()`, violating the base-class contract that callers may
size `asBytes` buffers from `maxSerializedByteSize()`. It is masked today only
because `deserializeAggregatedValue` separately records `bytes.length`.
`PercentileTDigestAccumulator.maxSerializedByteSize()` handles the analogous
case with `if (_originalFractionalBytes != null) return
_originalFractionalBytes.length;` — the same one-line guard is missing at the
wrapper level.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TDigestUtils.java:
##########
@@ -86,44 +85,65 @@ public static byte[] serialize(TDigest tDigest) {
/// Serializes a digest using `scratchBuffer` for the temporary verbose
representation when it is large enough.
///
- /// The returned array is independently owned. The scratch buffer is cleared
and may be reused by the caller after
- /// this method returns.
+ /// The returned array is independently owned. Pinot-owned digests do not
use the scratch buffer; callers can
+ /// clear and reuse it after serialization.
public static byte[] serialize(TDigest tDigest, ByteBuffer scratchBuffer) {
ByteOrder scratchOrder = scratchBuffer != null ? scratchBuffer.order() :
null;
try {
- if (!(tDigest instanceof MergingDigest)) {
- // Pinot has always decoded TDigest values as MergingDigest. Preserve
the previous serialization behavior for
- // other implementations instead of interpreting their
implementation-specific payload as MergingDigest
- // bytes.
- ByteBuffer buffer = prepareScratchBuffer(scratchBuffer,
tDigest.byteSize());
- tDigest.asBytes(buffer);
- byte[] bytes = new byte[buffer.position()];
- buffer.flip();
- buffer.get(bytes);
- return bytes;
+ if (tDigest instanceof PercentileTDigestAccumulator) {
+ // Preserve the legacy two-level raw flush before final compression.
Pending serialized state answers
+ // this count from cached metadata, so stored reads do not acquire an
extra lossy compression pass.
+ tDigest.centroidCount();
Review Comment:
**Accuracy concern + contradictory rationales: this `centroidCount()`
pre-call forces a working-compression (2x) flush of buffered incoming centroids
before `serialize()` runs `compress()` at public compression — an extra lossy
pass for SQL reduction accumulators.**
For any accumulator with `_numIncomingCentroids > 0` (e.g. intermediates
built by `merge()` → `add(TDigest)` → `bufferIncomingCentroids`, then shipped
via `ObjectSerDeUtils.TDIGEST_SER_DE`), `centroidCount()` → `flush()` picks
`_workingCompression`, and `serialize()` → `compress()` then recompresses at
`_compression`. `compress()`'s own comment says flushing at working compression
first "would add an extra lossy merge pass and materially increase reducer rank
error", while `flush()`'s doc says "Buffered reducer inputs also retain the
two-level merge for its accuracy benefit" — these two rationales contradict
each other for the same scenario, and this pre-call (whose comment mentions
only the "legacy two-level raw flush") silently decides between them. Worth
resolving which is intended; if single-pass is desired for non-two-level
accumulators, the pre-call should be limited to `_useTwoLevelRawCompression`
digests.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -0,0 +1,1766 @@
+/**
+ * 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.
+/// Unchanged historical fractional boundaries retain their original bytes,
including when configured compression
+/// differs. After a merge, inherited fractional endpoints retain precise
verbose weights and finite statistics when
+/// unit endpoints cannot preserve their mass. These keep the historical
t-digest 3.3 assertion-enabled reader and
+/// recompression limitation. Newly introduced unsupported fractional
endpoints are rejected.
+///
+/// Instances are externally serialized per aggregation or group key and are
not safe for concurrent mutation.
+public final class PercentileTDigestAccumulator extends TDigest {
+ private static final int MIN_RAW_BUFFER_SIZE = 256;
+ private static final int MAX_RAW_BUFFER_SIZE = 10_000;
+ private static final int 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 double _historicalFractionalMinMean = Double.NaN;
+ private double _historicalFractionalMaxMean = Double.NaN;
+ 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());
+ inheritHistoricalFractionalBoundaries(other);
+ }
+ }
+
+ @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 && 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;
+ inheritHistoricalFractionalBoundaries(other._historicalFractionalMinMean,
other._historicalFractionalMaxMean);
+ 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) {
+ if (_legacyDegraded) {
+ SerializedTDigestInput input = getSerializedTDigestInput();
+ input.reset(bytes);
+ addSerializedTDigest(input);
+ return;
+ }
+ requireMutable();
+ materializePendingSerializedTDigest();
+ SerializedTDigestInput input = getSerializedTDigestInput();
+ input.reset(bytes);
+ addSerializedTDigest(input);
+ }
+
+ private SerializedTDigestInput getSerializedTDigestInput() {
+ if (_serializedTDigestInput == null) {
+ _serializedTDigestInput = new SerializedTDigestInput();
+ }
+ return _serializedTDigestInput;
+ }
+
+ public void addSerializedTDigest(SerializedTDigestInput input) {
+ if (ignoreEmptyInputForDegradedState(input)) {
+ return;
+ }
+ requireMutable();
+ if (prepareSerializedTDigest(input)) {
+ mergeSerializedTDigest(input, false);
+ }
+ }
+
+ public void addSerializedTDigestDirect(SerializedTDigestInput input) {
+ if (ignoreEmptyInputForDegradedState(input)) {
+ return;
+ }
+ requireMutable();
+ if (prepareSerializedTDigest(input)) {
+ mergeSerializedTDigest(input, true);
+ }
+ }
+
+ private boolean ignoreEmptyInputForDegradedState(SerializedTDigestInput
input) {
+ if (!_legacyDegraded) {
+ return false;
+ }
+ input.inspectMetadata();
+ return !input._metadata.needsLegacyFallback() &&
input._metadata.totalWeight() == 0.0;
+ }
+
+ private boolean prepareSerializedTDigest(SerializedTDigestInput input) {
+ if (_pendingSerializedTDigest == null && hasNoInputs()) {
+ input.inspectMetadata();
+ _hasFractionalWeights |= input._metadata.fractionalWeights();
+ _pendingSerializedTDigest = input.retainBytes();
+ _pendingSerializedMetadata = input._metadata;
+ if (input.recordHistoricalFractionalBoundaries(this)) {
+ // Alias the retained snapshot; historical fractional bytes remain
byte-exact after read-only materialization.
+ _originalFractionalBytes = _pendingSerializedTDigest;
+ }
+ if (input._metadata.needsLegacyFallback()) {
+ _legacyDegraded = true;
+ }
+ return false;
+ }
+ materializePendingSerializedTDigest();
+ input.decode();
+ if (input._metadata.needsLegacyFallback()) {
+ if (hasNoInputs()) {
+ _pendingSerializedTDigest = input.retainBytes();
+ _pendingSerializedMetadata = input._metadata;
+ _legacyDegraded = true;
+ return false;
+ }
+ throw corruptedMutation();
Review Comment:
**Behavior change worth an explicit decision: mixing one historically
corrupt payload with any healthy data in the same aggregation state now
hard-fails the whole operation with a generic `IllegalArgumentException`.**
`prepareSerializedTDigest` throws `corruptedMutation()` when a
`needsLegacyFallback` payload arrives after other inputs, and
`requireMutable()` throws for every later add/merge into a degraded accumulator
— including a byte-identical duplicate of the same corrupt payload. Pre-PR, the
query path deserialized these payloads through tdunning without validation and
returned (garbage) results; now a group-by where one row's stored digest is
corrupt and another row is healthy fails the entire query, merge-rollup tasks
fail permanently (retried forever until manual intervention), and star-tree
segment builds abort. The README covers the query-side semantics but not the
minion/star-tree operational impact. If the escalation is intended, consider
(a) a dedicated exception type/message that query-level error handling can
classify, and (b) documenting the operator remediation for rollup/star-tree on
tables with legacy-corrupt digests.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/PercentileTDigestValueAggregator.java:
##########
@@ -152,130 +166,102 @@ private static NonFiniteAwareTDigest
asNonFiniteAware(TDigest value) {
}
private void updateMaxByteSize(TDigest value) {
- long defaultCapacity = getDefaultCentroidCapacity(value.compression());
- long maxCentroids = Math.min(value.size(), defaultCapacity);
- _maxByteSize = Math.max(_maxByteSize, getMaxVerboseByteSize(maxCentroids));
- }
-
- private static int getMaxVerboseByteSize(long centroidCount) {
- long maxCentroids = Math.max(centroidCount, 0L);
- return Math.toIntExact(Math.addExact(VERBOSE_HEADER_SIZE,
- Math.multiplyExact(VERBOSE_CENTROID_SIZE, maxCentroids)));
- }
-
- private static long getDefaultCentroidCapacity(double compression) {
- double normalizedCompression = Math.max(MIN_COMPRESSION, compression);
- int padding = normalizedCompression < 30.0 ?
LOW_COMPRESSION_CAPACITY_PADDING
- : DEFAULT_CENTROID_CAPACITY_PADDING;
- long defaultCapacity = (long) Math.ceil(2.0 * normalizedCompression +
padding);
- long legacyCapacity = Math.addExact(Math.multiplyExact(2L, (long)
Math.ceil(compression)),
- DEFAULT_CENTROID_CAPACITY_PADDING);
- return Math.max(defaultCapacity, legacyCapacity);
+ _maxByteSize = Math.max(_maxByteSize, value.maxSerializedByteSize());
Review Comment:
**Efficiency: `maxSerializedByteSize()` counts every buffered raw value and
incoming centroid, inflating `getMaxAggregatedValueByteSize()` ~7x over master
and over any possible serialized output.**
`getCentroidCountUpperBound()` includes `_numRawValues` (up to 256) and
`_numIncomingCentroids` (up to ~1150 at compression 100), and for large total
weight the `ceil(getTotalWeight())` clamp never engages, so this reports ~26KB
where actual serialized values are legacy-capacity-bounded (<~3.7KB at
compression 100; master reported ~3.9KB).
`BaseSingleTreeBuilder.createForwardIndexes` sizes
`SingleValueVarByteRawIndexCreator` maxLength from this value, shrinking
docs-per-chunk and inflating index overhead for every TDigest star-tree segment
built after the upgrade. Clamping the bound by the legacy-compatible output
capacity (what `serialize()` can actually emit after `makeLegacyCompatible`)
would restore the tight bound.
--
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]
