xiangfu0 commented on code in PR #19795:
URL: https://github.com/apache/pinot/pull/19795#discussion_r4223706268
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -216,17 +243,27 @@ public void add(TDigest other) {
addAccumulator((PercentileTDigestAccumulator) other);
return;
}
- if (other instanceof MergingDigest && other.size() > Integer.MAX_VALUE) {
- // Centroid.count() exposes an int even though MergingDigest stores
double weights internally. Preserve large
- // pre-aggregated weights by ingesting the serialized double-precision
representation instead.
- addSerializedTDigest(TDigestUtils.serialize(other));
+ if (!other.hasValidStatistics()) {
+ materializePendingSerializedTDigest();
+ SerializedTDigestInput input = getSerializedTDigestInput();
+ input.reset(ByteBuffer.wrap(TDigestUtils.serialize(other)), false);
+ addSerializedTDigest(input);
return;
}
-
- Collection<Centroid> centroids = other.centroids();
materializePendingSerializedTDigest();
- for (Centroid centroid : centroids) {
- bufferIncomingCentroid(centroid.mean(), centroid.count());
+ if (_legacyDegraded) {
+ _legacySerializedDirty = true;
+ _totalWeight += other.getTotalWeight();
+ } else {
+ checkTotalWeight(_totalWeight + _incomingWeight + _numRawValues +
other.getTotalWeight());
+ for (Centroid centroid : other.centroids()) {
+ double weight = centroid.weight();
+ if (!(weight > 0.0) || !Double.isFinite(weight) ||
Double.isNaN(centroid.mean())) {
+ throw new IllegalArgumentException("Invalid TDigest centroid");
Review Comment:
**[code-review]** add(TDigest) throws on zero-weight centroids that the
serialized-bytes path explicitly tolerates and filters
inspectSerialized treats zero-weight centroids as valid (`if (weight == 0.0)
{ hasZeroWeightCentroids = true; continue; }`) and decode() compacts them away,
but the object-merge path throws IllegalArgumentException('Invalid TDigest
centroid') for `!(weight > 0.0)`. The SPI centroids() javadoc imposes no
positive-weight precondition, so a conforming implementation mirroring bytes
Pinot itself deems valid kills the query or segment build when merged as an
object while the identical bytes merge fine. Either skip zero weights here to
match decode(), or document the precondition on the SPI.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -409,17 +409,36 @@ private void
mergeSerializedTDigest(SerializedTDigestInput input) {
_totalWeight = input._totalWeight;
_min = input._min;
_max = input._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._compression;
+ }
+ return;
+ }
+ if (direct) {
+ flush();
+ boolean runBackwards = (_mergeCount++ & 1) != 0;
+ if (!mergeSorted(input._means, input._weights, input._numCentroids,
input._totalWeight,
+ getIncrementalCompression(), runBackwards)) {
+ return;
}
} else {
bufferIncomingCentroids(input._means, input._weights,
input._numCentroids);
- _min = Math.min(_min, input._min);
- _max = Math.max(_max, input._max);
}
+ _min = Math.min(_min, input._min);
Review Comment:
**[code-review]** Buffered merge path applies a serialized input's header
min/max even when the input has zero total mass, unlike the guarded fast path
The hasNoInputs() fast path copies extrema only `if (input._totalWeight !=
0.0)`, but the buffered tail runs `_min = Math.min(_min, input._min)`
unconditionally. An all-zero-weight payload with finite extrema (producible by
tdunning add(x, 0), which legacy code never validated) merged into an
accumulator holding [10,20] pollutes _min to the header value, and quantile(q)
for index<1 returns that value with no sample mass there. Relatedly, a pending
zero-mass digest answers getMin()/centroidCount() from metadata before
materialization and +INF/0 after - order-dependent results from the same state.
Guard the tail with the same totalWeight check.
##########
compatibility-verifier/tdigest-compatibility/GenerateRankErrors.java:
##########
@@ -0,0 +1,154 @@
+/**
+ * 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.
+ */
+
+import com.tdunning.math.stats.Centroid;
+import com.tdunning.math.stats.TDigest;
+import java.util.Arrays;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.SplittableRandom;
+import org.apache.pinot.segment.local.utils.TDigestUtils;
Review Comment:
**[code-review]** The checked-in oracle generator cannot compile against the
tree it lives in, making the rank-error CSV practically unregenerable
GenerateRankErrors imports the pre-PR TDigestUtils API (tdunning TDigest
returns) that this PR removes; nothing in run.sh or CI compiles it, and the
regeneration recipe lives only in the CSV header, pinned to a historical commit
unavailable on shallow clones. When reducerMergeCases changes (new
distribution/fan-in/compression), loadLegacyReducerRankErrors fails demanding a
regenerated oracle, and the path of least resistance becomes hand-editing
tdigest-3.3-k1-rank-errors.csv - silently detaching the 'independent 3.3
oracle' from any real 3.3 measurement. It also duplicates the test's
distributions/merge orders/seeds formula-for-formula, so the two can drift.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -696,34 +775,38 @@ public double getMin() {
@Override
public double getMax() {
if (_pendingSerializedTDigest != null) {
- return
ByteBuffer.wrap(_pendingSerializedTDigest).getDouble(Integer.BYTES +
Double.BYTES);
+ return _pendingSerializedMetadata.max();
}
flush();
return _max;
}
- byte[] serialize() {
- if (_pendingSerializedTDigest != null) {
- getSerializedTotalWeight(_pendingSerializedTDigest);
- if (ByteBuffer.wrap(_pendingSerializedTDigest).getInt() ==
VERBOSE_ENCODING) {
- byte[] serialized =
TDigestUtils.makeLegacyCompatible(_pendingSerializedTDigest);
- return serialized == _pendingSerializedTDigest ? serialized.clone() :
serialized;
+ public byte[] serialize() {
+ if (_legacyDegraded) {
+ if (_pendingSerializedTDigest != null) {
+ return _pendingSerializedTDigest.clone();
+ }
+ if (!_legacySerializedDirty) {
+ return _legacySerializedBytes.clone();
}
- return _pendingSerializedTDigest.clone();
+ // A merge with a historical invalid distribution remains unknown. Keep
that explicit in the legacy
+ // encoding instead of passing its invalid centroids into the normal
sorted K1 merge.
+ return toLegacyDegradedBytes();
}
- compress();
- return TDigestUtils.makeLegacyCompatible(toVerboseBytes());
- }
-
- TDigest toTDigest() {
if (_pendingSerializedTDigest != null) {
- return TDigestUtils.deserialize(_pendingSerializedTDigest);
+ if (_pendingSerializedMetadata.weightedBoundaries() ||
_pendingSerializedMetadata.hasZeroWeightCentroids()) {
+ 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();
- if (_numCentroids == 0) {
- return TDigestUtils.createMergingDigest(_compression);
- }
- return
TDigestUtils.deserialize(TDigestUtils.makeLegacyCompatible(toVerboseBytes()));
+ normalizeBoundaryCentroids();
+ return TDigestUtils.makeLegacyCompatible(toVerboseBytes());
Review Comment:
**[code-review]** serialize() pays a redundant O(n) decode+validate pass
over bytes it just wrote; the metadata-carrying overload exists precisely to
skip it
The non-pending path calls single-arg
makeLegacyCompatible(toVerboseBytes()), which re-runs inspectSerialized over
the fresh bytes - recomputing totalWeight, ordering, fractional and boundary
flags the accumulator already knows from its own post-compress,
boundary-normalized state (_totalWeight, _hasFractionalWeights, sorted means).
ObjectSerDeUtils.TDIGEST_SER_DE serializes per group per block, so a 100k-group
query at compression 100 pays tens of millions of redundant getDouble() calls,
roughly doubling intermediate-result serialization CPU. Build the
SerializedTDigestMetadata from in-memory state and call the two-arg overload
(as the pending path at line 801 already does).
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/PercentileTDigestValueAggregator.java:
##########
@@ -128,17 +125,20 @@ public int getMaxAggregatedValueByteSize() {
@Override
public byte[] serializeAggregatedValue(TDigest value) {
- int requiredCapacity =
getMaxVerboseByteSize(getDefaultCentroidCapacity(value.compression()));
+ NonFiniteAwareTDigest wrapped = asNonFiniteAware(value);
+ updateMaxByteSize(wrapped);
+ int requiredCapacity = Math.max(_maxByteSize,
TDigestUtils.VERBOSE_HEADER_SIZE);
if (_serializationBuffer == null || _serializationBuffer.capacity() <
requiredCapacity) {
Review Comment:
**[code-review]** _serializationBuffer is dead weight: the scratch path is
unreachable because the wrapped finite digest is always a Pinot accumulator
TDigestUtils.serialize(tDigest, scratch) ignores the scratch buffer for
PercentileTDigestAccumulator, and NonFiniteAwareTDigest's _finiteDigest always
is one (createMergingDigest/deserializeFinite), so serializeAggregatedValue
allocates and retains a ByteBuffer that grows to _maxByteSize (tens of KB at
high compression, per metric aggregator for the whole star-tree build) and is
never written - the test even asserts it stays untouched.
NonFiniteAwareTDigest.serialize() additionally returns
getSerializedBytes(...).clone(), a full extra copy of bytes usually computed
fresh in the same call; transfer ownership (null the cache) instead. Drop the
buffer and the clone.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -1452,42 +1406,73 @@ static final class SerializedTDigestInput {
private double _totalWeight;
private boolean _decoded;
- void reset(byte[] bytes) {
- _retainedBytes = null;
- ByteBuffer input = ByteBuffer.wrap(bytes);
- int encoding = input.getInt();
- if (encoding != VERBOSE_ENCODING && encoding != SMALL_ENCODING) {
- throw new IllegalStateException("Invalid format for serialized
histogram");
- }
+ private SerializedTDigestMetadata _metadata;
+
+ public void reset(byte[] bytes) {
+ reset(bytes, TDigestUtils.readSerializedHeader(ByteBuffer.wrap(bytes)));
+ }
+
+ /// Owns one snapshot of the encoded digest and advances the source buffer
past it.
+ public void reset(ByteBuffer bytes) {
+ reset(bytes, true);
+ }
+
+ /// Reads a source digest, optionally enforcing the historical decoder's
declared centroid capacity.
+ public void reset(ByteBuffer bytes, boolean checkCapacity) {
+ SerializedTDigestMetadata metadata =
TDigestUtils.readSerializedHeader(bytes, checkCapacity);
+ byte[] snapshot = new byte[metadata.encodedLength()];
+ bytes.get(snapshot);
+ reset(snapshot, metadata);
+ _retainedBytes = snapshot;
+ }
+
+ private void reset(byte[] bytes, SerializedTDigestMetadata metadata) {
+ _metadata = metadata;
_bytes = bytes;
- _min = input.getDouble();
- _max = input.getDouble();
- if (encoding == VERBOSE_ENCODING) {
- _compression = input.getDouble();
- _numCentroids = input.getInt();
- _mainCapacity = 0;
- _bufferCapacity = 0;
- _centroidSize = VERBOSE_CENTROID_SIZE;
- } else {
- _compression = input.getFloat();
- _mainCapacity = input.getShort();
- _bufferCapacity = input.getShort();
- checkSmallArrayCapacities(_mainCapacity, _bufferCapacity);
- _mainCapacity = resolveSmallMainCapacity(_compression, _mainCapacity);
- _numCentroids = input.getShort();
- _centroidSize = SMALL_CENTROID_SIZE;
- }
- checkCentroidCount(_numCentroids);
- checkCentroidsAvailable(input, _numCentroids, _centroidSize);
- if (_centroidSize == VERBOSE_CENTROID_SIZE) {
- checkVerboseCentroidCapacity(_compression, _numCentroids);
- } else {
- checkSmallCentroidCapacity(_mainCapacity, _numCentroids);
- }
- _centroidOffset = input.position();
+ _retainedBytes = null;
+ _totalWeight = metadata.totalWeight();
+ _min = metadata.min();
+ _max = metadata.max();
+ _compression = metadata.compression();
+ _numCentroids = metadata.centroidCount();
+ _mainCapacity = metadata.mainCapacity();
+ _bufferCapacity = metadata.bufferCapacity();
+ _centroidSize = metadata.centroidSize();
+ _centroidOffset = metadata.centroidOffset();
_decoded = false;
}
+ public double getCompression() {
+ return _compression;
+ }
+
+ public int getEncodedLength() {
Review Comment:
**[code-review]** Dead API and duplicated state on the accumulator: three
uncalled SerializedTDigestInput methods, a mirror field, and a no-op NaN ternary
SerializedTDigestInput.getEncodedLength(), hasNonFiniteMeans(), and
needsLegacyFallback() have zero callers in main or test code (all users read
_metadata directly or via getMetadata()). _pendingSerializedTotalWeight is
provably always equal to _pendingSerializedMetadata.totalWeight() at its single
read site (getTotalWeight), so the pair must be kept in lockstep by hand for no
benefit. toLegacyDegradedBytes writes `Double.isNaN(_min) ? Double.NaN : _min`
- an identity ternary that reads as intentional NaN normalization but does
nothing. Delete the methods, read the metadata, and write _min directly.
--
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]
