xiangfu0 commented on code in PR #19795: URL: https://github.com/apache/pinot/pull/19795#discussion_r4224157934
########## 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: Fixed in [4818fb8fbc](https://github.com/apache/pinot/commit/4818fb8fbc79d81de17ce0c5bd0070eb97f300df). generate-rank-errors.sh compiles the pinned historical utility and actual 3.3 jar in isolation, fetching the immutable baseline when a shallow clone lacks it. run.sh regenerates and compares the complete reference in CI. Per-case SHA256 fingerprints include sorted raw-double bits and the exact merge permutation used by the test, catching input/seed/order drift. The shallow-history path and all 270 regenerated rows pass. ########## 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: Fixed in [4818fb8fbc](https://github.com/apache/pinot/commit/4818fb8fbc79d81de17ce0c5bd0070eb97f300df). Owned serialization gathers compatibility metadata while writing centroids and calls the metadata overload; it no longer re-inspects its fresh payload. Pending bytes retain their already-inspected metadata. ########## 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: Fixed in [4818fb8fbc](https://github.com/apache/pinot/commit/4818fb8fbc79d81de17ce0c5bd0070eb97f300df). Removed the unused retained scratch buffer and wrapper scratch plumbing. Serialization transfers byte-array ownership and invalidates both cache aliases where necessary. Finite-only and infinity-tail regressions mutate the returned array, then verify later serialization and quantiles are unaffected. ########## 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: Removed the three unused accessors and mirrored pending-total field in [4818fb8fbc](https://github.com/apache/pinot/commit/4818fb8fbc79d81de17ce0c5bd0070eb97f300df). The NaN ternary is retained with an explanation: ByteBuffer.putDouble preserves raw NaN payload bits, whereas Double.NaN canonicalizes them. It makes degraded output deterministic across merge orders. A regression uses distinct NaN bit patterns and checks byte equality. -- 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]
