xiangfu0 commented on code in PR #19795: URL: https://github.com/apache/pinot/pull/19795#discussion_r4240091215
########## pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/tdigest/TDigestCodec.java: ########## @@ -0,0 +1,691 @@ +/** + * 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.tdigest; + +import java.nio.BufferUnderflowException; +import java.nio.ByteBuffer; +import java.nio.ByteOrder; +import java.util.Arrays; + +/// Compatibility helpers for the serialized t-digest format shared by t-digest 3.2 and 3.3. +/// +/// Version 3.3 requires the first and last centroid to have unit weight whenever a digest is recompressed. Version +/// 3.2 did not maintain that invariant, so an otherwise valid 3.2 digest can fail an assertion after being read by +/// 3.3. Deserialization repairs only those legacy boundary centroids, preserving their total weight and first moment. +/// Serialization uses the compact form when 3.3's larger low-compression centroid buffer would not fit in a 3.2 +/// default buffer and every centroid is exactly representable in that format. Otherwise, it merges the nearest +/// interior centroids until the verbose representation fits the 3.2 capacity. This keeps intermediate values readable +/// in both directions during a rolling upgrade without silently narrowing double-precision values to floats. +/// +/// Pinot owns the in-memory K1 implementation and retains these legacy bytes for stored columns and mixed-version +/// intermediate results. Decoding checks lengths and numerical state before allocating centroid arrays. +/// NaN means at an identifiable infinite tail are restored to that infinite bound without changing their weights. +/// Other poisoned states remain byte-readable with NaN statistics because their distribution cannot be recovered. +/// They retain their original bytes; attempting to mutate them or mix them with another distribution fails explicitly. +/// Opaque poisoned payloads retain their original centroids even if an old reader's default array cannot hold them; +/// these historical exceptions remain readable by Pinot, but are not guaranteed to survive a reverse upgrade. +/// Historical fractional boundary masses below one, and singleton masses between one and two, retain their original +/// bytes. Fresh serialization rejects these boundaries: t-digest 3.3's unit-endpoint assertion cannot be satisfied +/// without changing their mass. Fractional interior weights remain supported. +public final class TDigestCodec { + public static final int VERBOSE_ENCODING = 1; + public static final int SMALL_ENCODING = 2; + private static final int LEGACY_CAPACITY_PADDING = 10; + public static final int VERBOSE_HEADER_SIZE = 32; + public static final int VERBOSE_CENTROID_SIZE = 16; + public static final int SMALL_HEADER_SIZE = 30; + public static final int SMALL_CENTROID_SIZE = 8; + public static final int DEFAULT_MERGE_BUFFER_MULTIPLIER = 5; + + private TDigestCodec() { + } + + /// Reads the compression from a validated header without decoding centroid arrays. + public static double readCompression(byte[] bytes) { + return readSerializedHeader(ByteBuffer.wrap(bytes)).compression(); + } + + /// Rejects non-finite compression while retaining the legacy clamping of finite settings below ten. + public static void validateCompression(double compression) { + if (!Double.isFinite(compression)) { + throw new IllegalArgumentException("Invalid TDigest compression: " + compression); + } + } + + /// Applies the legacy minimum compression to query, ingestion and serialized-header settings. + public static double normalizeCompression(double compression) { + validateCompression(compression); + return Math.max(compression, 10.0); + } + + /// Writes validated centroid arrays in the standard verbose encoding without sorting or recompression. + /// The caller retains the arrays; only the first `count` entries are written, with normalized compression. + public static byte[] serializeCentroids(double compression, double min, double max, double[] means, + double[] weights, int count) { + if (count < 0 || count > means.length || count > weights.length) { + throw new IllegalArgumentException("Invalid TDigest centroid count: " + count); + } + ByteBuffer encoded = ByteBuffer.allocate(Math.addExact(VERBOSE_HEADER_SIZE, + Math.multiplyExact(VERBOSE_CENTROID_SIZE, count))); + encoded.putInt(VERBOSE_ENCODING).putDouble(min).putDouble(max).putDouble(normalizeCompression(compression)); + encoded.putInt(count); + for (int i = 0; i < count; i++) { + encoded.putDouble(weights[i]).putDouble(means[i]); + } + return encoded.array(); + } + + /// Encodes validated, sorted native centroids and selects a legacy-compatible representation. + static byte[] serializeCompatibleCentroids(double compression, double min, double max, double[] means, + double[] weights, int count, boolean inheritedFractionalBoundaries) { + if (count < 0 || count > 0 && (count > means.length || count > weights.length || !(min <= max))) { + throw new IllegalArgumentException("Invalid TDigest centroid count or extrema"); + } + ByteBuffer buffer = ByteBuffer.allocate( + Math.addExact(VERBOSE_HEADER_SIZE, + Math.multiplyExact(VERBOSE_CENTROID_SIZE, count))); + buffer.putInt(VERBOSE_ENCODING); + buffer.putDouble(min); + buffer.putDouble(max); + buffer.putDouble(compression); + buffer.putInt(count); + double encodedWeight = 0.0; + boolean hasNonFiniteMeans = false; + boolean fractionalWeights = false; + for (int i = 0; i < count; i++) { + if (!(weights[i] > 0.0) || !Double.isFinite(weights[i]) || !(means[i] >= min && means[i] <= max)) { + throw new IllegalArgumentException("Invalid TDigest centroid"); + } + buffer.putDouble(weights[i]); + buffer.putDouble(means[i]); + encodedWeight += weights[i]; + hasNonFiniteMeans |= !Double.isFinite(means[i]); + fractionalWeights |= weights[i] != Math.rint(weights[i]); + } + if (!Double.isFinite(encodedWeight)) { + throw new IllegalArgumentException("TDigest total weight exceeds the supported range"); + } + boolean weightedBoundaries = count == 1 ? weights[0] >= 2.0 + : count > 1 && (weights[0] > 1.0 || weights[count - 1] > 1.0); + // Collect exact wire metadata during the write rather than decoding and validating our fresh bytes again. + SerializedTDigestMetadata metadata = new SerializedTDigestMetadata(VERBOSE_ENCODING, min, max, + compression, count, + VERBOSE_HEADER_SIZE, VERBOSE_CENTROID_SIZE, buffer.capacity(), encodedWeight, + hasNonFiniteMeans, false, false, fractionalWeights, weightedBoundaries, false, compression, Double.NaN, + Double.NaN, Double.NaN); + return makeLegacyCompatible(buffer.array(), metadata, inheritedFractionalBoundaries); + } + + /// Rejects fresh boundary masses that legacy readers cannot recompress without inventing mass. + /// Call after boundary normalization and removing zero-weight centroids; fractional interior mass is supported. + static void checkLegacyBoundaryWeights(int count, double firstWeight, double lastWeight) { + if (count > 0 && (firstWeight < 1.0 || lastWeight < 1.0 + || count == 1 && firstWeight != 1.0 && firstWeight < 2.0)) { + throw new IllegalArgumentException("Cannot serialize fractional TDigest boundary mass for legacy readers"); + } + } + + /// Allows only unsupported endpoint means inherited from a validated legacy source. This is shared by native, + /// enclosing and generic writers; new fractional global extrema do not receive the historical exception. + /// Exact equality is intentional: endpoint repair preserves these means, and admitting nearby means would grant + /// the exception to fresh extrema. This check must never permit an approximate match or a digest-wide exemption. + static boolean hasInheritedFractionalBoundaryEncoding(TDigest source, int count, double firstMean, + double firstWeight, double lastMean, double lastWeight) { + if (count <= 0 || !source.hasValidStatistics() || !(firstWeight > 0.0) || !(lastWeight > 0.0) + || !Double.isFinite(firstWeight) || !Double.isFinite(lastWeight)) { + return false; + } + boolean singleton = count == 1 && firstWeight != 1.0 && firstWeight < 2.0; + boolean firstUnsupported = firstWeight < 1.0 || singleton; + boolean lastUnsupported = lastWeight < 1.0 || singleton; + return (firstUnsupported || lastUnsupported) Review Comment: Fixed in [df09e87b8f](https://github.com/apache/pinot/commit/df09e87b8f99948e03442cb3afbb887b3100dc5d). Fixed by tracking the minimum and maximum means of fresh fractional inputs through copies and merges. A historical endpoint loses its exception when any fresh fractional input is at or beyond that mean, so neither a different weight nor an equal weight can reuse the old endpoint permission; this also covers stale recorded means and fresh-before-history order. Historical-only mass and added unit weights preserve prior behavior. Regressions cover both weights/input orders, native/wrapper copies and merges, and a valid fractional interior control. ########## pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/tdigest/TDigestCodec.java: ########## @@ -0,0 +1,691 @@ +/** + * 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.tdigest; + +import java.nio.BufferUnderflowException; +import java.nio.ByteBuffer; +import java.nio.ByteOrder; +import java.util.Arrays; + +/// Compatibility helpers for the serialized t-digest format shared by t-digest 3.2 and 3.3. +/// +/// Version 3.3 requires the first and last centroid to have unit weight whenever a digest is recompressed. Version +/// 3.2 did not maintain that invariant, so an otherwise valid 3.2 digest can fail an assertion after being read by +/// 3.3. Deserialization repairs only those legacy boundary centroids, preserving their total weight and first moment. +/// Serialization uses the compact form when 3.3's larger low-compression centroid buffer would not fit in a 3.2 +/// default buffer and every centroid is exactly representable in that format. Otherwise, it merges the nearest +/// interior centroids until the verbose representation fits the 3.2 capacity. This keeps intermediate values readable +/// in both directions during a rolling upgrade without silently narrowing double-precision values to floats. +/// +/// Pinot owns the in-memory K1 implementation and retains these legacy bytes for stored columns and mixed-version +/// intermediate results. Decoding checks lengths and numerical state before allocating centroid arrays. +/// NaN means at an identifiable infinite tail are restored to that infinite bound without changing their weights. +/// Other poisoned states remain byte-readable with NaN statistics because their distribution cannot be recovered. +/// They retain their original bytes; attempting to mutate them or mix them with another distribution fails explicitly. +/// Opaque poisoned payloads retain their original centroids even if an old reader's default array cannot hold them; +/// these historical exceptions remain readable by Pinot, but are not guaranteed to survive a reverse upgrade. +/// Historical fractional boundary masses below one, and singleton masses between one and two, retain their original +/// bytes. Fresh serialization rejects these boundaries: t-digest 3.3's unit-endpoint assertion cannot be satisfied +/// without changing their mass. Fractional interior weights remain supported. +public final class TDigestCodec { + public static final int VERBOSE_ENCODING = 1; + public static final int SMALL_ENCODING = 2; + private static final int LEGACY_CAPACITY_PADDING = 10; + public static final int VERBOSE_HEADER_SIZE = 32; + public static final int VERBOSE_CENTROID_SIZE = 16; + public static final int SMALL_HEADER_SIZE = 30; + public static final int SMALL_CENTROID_SIZE = 8; + public static final int DEFAULT_MERGE_BUFFER_MULTIPLIER = 5; + + private TDigestCodec() { + } + + /// Reads the compression from a validated header without decoding centroid arrays. + public static double readCompression(byte[] bytes) { + return readSerializedHeader(ByteBuffer.wrap(bytes)).compression(); + } + + /// Rejects non-finite compression while retaining the legacy clamping of finite settings below ten. + public static void validateCompression(double compression) { + if (!Double.isFinite(compression)) { + throw new IllegalArgumentException("Invalid TDigest compression: " + compression); + } + } + + /// Applies the legacy minimum compression to query, ingestion and serialized-header settings. + public static double normalizeCompression(double compression) { + validateCompression(compression); + return Math.max(compression, 10.0); + } + + /// Writes validated centroid arrays in the standard verbose encoding without sorting or recompression. + /// The caller retains the arrays; only the first `count` entries are written, with normalized compression. + public static byte[] serializeCentroids(double compression, double min, double max, double[] means, + double[] weights, int count) { + if (count < 0 || count > means.length || count > weights.length) { + throw new IllegalArgumentException("Invalid TDigest centroid count: " + count); + } + ByteBuffer encoded = ByteBuffer.allocate(Math.addExact(VERBOSE_HEADER_SIZE, + Math.multiplyExact(VERBOSE_CENTROID_SIZE, count))); + encoded.putInt(VERBOSE_ENCODING).putDouble(min).putDouble(max).putDouble(normalizeCompression(compression)); + encoded.putInt(count); + for (int i = 0; i < count; i++) { + encoded.putDouble(weights[i]).putDouble(means[i]); + } + return encoded.array(); + } + + /// Encodes validated, sorted native centroids and selects a legacy-compatible representation. + static byte[] serializeCompatibleCentroids(double compression, double min, double max, double[] means, + double[] weights, int count, boolean inheritedFractionalBoundaries) { + if (count < 0 || count > 0 && (count > means.length || count > weights.length || !(min <= max))) { + throw new IllegalArgumentException("Invalid TDigest centroid count or extrema"); + } + ByteBuffer buffer = ByteBuffer.allocate( + Math.addExact(VERBOSE_HEADER_SIZE, + Math.multiplyExact(VERBOSE_CENTROID_SIZE, count))); + buffer.putInt(VERBOSE_ENCODING); + buffer.putDouble(min); + buffer.putDouble(max); + buffer.putDouble(compression); + buffer.putInt(count); + double encodedWeight = 0.0; + boolean hasNonFiniteMeans = false; + boolean fractionalWeights = false; + for (int i = 0; i < count; i++) { + if (!(weights[i] > 0.0) || !Double.isFinite(weights[i]) || !(means[i] >= min && means[i] <= max)) { + throw new IllegalArgumentException("Invalid TDigest centroid"); + } + buffer.putDouble(weights[i]); + buffer.putDouble(means[i]); + encodedWeight += weights[i]; + hasNonFiniteMeans |= !Double.isFinite(means[i]); + fractionalWeights |= weights[i] != Math.rint(weights[i]); + } + if (!Double.isFinite(encodedWeight)) { + throw new IllegalArgumentException("TDigest total weight exceeds the supported range"); + } + boolean weightedBoundaries = count == 1 ? weights[0] >= 2.0 + : count > 1 && (weights[0] > 1.0 || weights[count - 1] > 1.0); + // Collect exact wire metadata during the write rather than decoding and validating our fresh bytes again. + SerializedTDigestMetadata metadata = new SerializedTDigestMetadata(VERBOSE_ENCODING, min, max, + compression, count, + VERBOSE_HEADER_SIZE, VERBOSE_CENTROID_SIZE, buffer.capacity(), encodedWeight, + hasNonFiniteMeans, false, false, fractionalWeights, weightedBoundaries, false, compression, Double.NaN, + Double.NaN, Double.NaN); + return makeLegacyCompatible(buffer.array(), metadata, inheritedFractionalBoundaries); + } + + /// Rejects fresh boundary masses that legacy readers cannot recompress without inventing mass. + /// Call after boundary normalization and removing zero-weight centroids; fractional interior mass is supported. + static void checkLegacyBoundaryWeights(int count, double firstWeight, double lastWeight) { + if (count > 0 && (firstWeight < 1.0 || lastWeight < 1.0 + || count == 1 && firstWeight != 1.0 && firstWeight < 2.0)) { + throw new IllegalArgumentException("Cannot serialize fractional TDigest boundary mass for legacy readers"); + } + } + + /// Allows only unsupported endpoint means inherited from a validated legacy source. This is shared by native, + /// enclosing and generic writers; new fractional global extrema do not receive the historical exception. + /// Exact equality is intentional: endpoint repair preserves these means, and admitting nearby means would grant + /// the exception to fresh extrema. This check must never permit an approximate match or a digest-wide exemption. + static boolean hasInheritedFractionalBoundaryEncoding(TDigest source, int count, double firstMean, + double firstWeight, double lastMean, double lastWeight) { + if (count <= 0 || !source.hasValidStatistics() || !(firstWeight > 0.0) || !(lastWeight > 0.0) + || !Double.isFinite(firstWeight) || !Double.isFinite(lastWeight)) { + return false; + } + boolean singleton = count == 1 && firstWeight != 1.0 && firstWeight < 2.0; + boolean firstUnsupported = firstWeight < 1.0 || singleton; + boolean lastUnsupported = lastWeight < 1.0 || singleton; + return (firstUnsupported || lastUnsupported) + && (!firstUnsupported || firstMean == source.getHistoricalFractionalBoundaryMean(true)) + && (!lastUnsupported || lastMean == source.getHistoricalFractionalBoundaryMean(false)); + } + + private static void checkSerializedBoundaryWeights(byte[] bytes, SerializedTDigestMetadata metadata) { + int count = metadata.centroidCount(); + if (count > 0) { + ByteBuffer encoded = ByteBuffer.wrap(bytes); + int firstOffset = metadata.centroidOffset(); + int lastOffset = firstOffset + (count - 1) * metadata.centroidSize(); + double firstWeight = metadata.encoding() == VERBOSE_ENCODING + ? encoded.getDouble(firstOffset) : encoded.getFloat(firstOffset); + double lastWeight = metadata.encoding() == VERBOSE_ENCODING + ? encoded.getDouble(lastOffset) : encoded.getFloat(lastOffset); + checkLegacyBoundaryWeights(count, firstWeight, lastWeight); + } + } + + /// Converts a verbose `MergingDigest` payload to a representation readable by t-digest 3.2 and 3.3. + /// + /// The input must use the standard verbose `MergingDigest` encoding. The returned payload stays verbose when it + /// already fits both legacy default capacities, uses compact encoding only when every narrowed field is exact, and + /// otherwise reduces interior centroids without narrowing doubles. Compression below ten is normalized in the + /// output header because the 3.2 verbose reader sizes its arrays from the raw header before applying that minimum. + /// Opaque degraded state is retained without reducing its unknown distribution, including over-capacity payloads. + /// Fresh fractional endpoints that legacy unit-boundary repair cannot represent are rejected without reweighting. + public static byte[] makeLegacyCompatible(byte[] verboseBytes) { + return makeLegacyCompatible(verboseBytes, inspectSerialized(ByteBuffer.wrap(verboseBytes), false)); + } + + /// Serializes an unchanged, independently retained payload using metadata validated at ingestion. + /// + /// The metadata must describe these exact bytes. This avoids another centroid validation walk for lazy digests. + public static byte[] makeLegacyCompatible(byte[] verboseBytes, SerializedTDigestMetadata metadata) { + return makeLegacyCompatible(verboseBytes, metadata, false); + } + + /// Retains exact verbose weights for inherited historical fractional endpoints that cannot become unit endpoints. + /// All other compatibility checks apply. These payloads keep their historical assertion-enabled 3.3 read/merge + /// limitation; callers must establish inherited endpoint provenance before enabling this exception. + static byte[] makeLegacyCompatible(byte[] verboseBytes, SerializedTDigestMetadata metadata, + boolean inheritedFractionalBoundaries) { + if (metadata.encoding() != VERBOSE_ENCODING) { + throw new IllegalArgumentException("Expected verbose TDigest encoding"); + } + if (metadata.needsLegacyFallback()) { + return verboseBytes; + } + if (!Double.isNaN(metadata.recoveredInfinityMean())) { + double[] means = new double[metadata.centroidCount()]; + double[] weights = new double[means.length]; + decodeSerializedCentroids(ByteBuffer.wrap(verboseBytes), metadata, means, weights); + if (!inheritedFractionalBoundaries) { + return serializeRecoveredCentroids(metadata, means, weights); Review Comment: Fixed in [df09e87b8f](https://github.com/apache/pinot/commit/df09e87b8f99948e03442cb3afbb887b3100dc5d). Fixed the recovered-infinity branch to import validated historical endpoint provenance into its rebuilt accumulator. The reported {0.5@1, 1@3, 1@NaN; max=+Infinity} payload now converts with total weight 2.5, the 0.5 lower endpoint preserved and its NaN tail restored to +Infinity. The original fromBytes path remains byte-exact; a regression covers both paths. ########## compatibility-verifier/tdigest-compatibility/LegacyReader.java: ########## @@ -0,0 +1,141 @@ +/** + * 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.MergingDigest; +import com.tdunning.math.stats.TDigest; +import java.nio.ByteBuffer; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; + + +/// Runs only against one legacy jar in a separate JVM; Pinot classes are deliberately absent from its classpath. +/// Fractional endpoint masses below one and singleton masses between one and two are excluded: legacy 3.3 cannot +/// recompress them without changing mass. Those preserved-byte limitations are tested by Pinot's unit tests. +/// Initial value parity covers ordinary continuous/empty/singleton inputs and huge-mean fallback. Weighted endpoint +/// or capacity repair can change interpolation, and 3.2 has older singleton rules; those fixtures retain structural, +/// mass, extrema, and monotonicity checks rather than widening the numerical tolerance. +public final class LegacyReader { + private static final double[] QUANTILES = {0, 0.25, 0.5, 0.75, 0.99, 1}; + + private LegacyReader() { + } + + public static void main(String[] args) + throws Exception { + if (!LegacyReader.class.desiredAssertionStatus()) { + throw new IllegalStateException("The legacy compatibility gate requires -ea"); + } + Path directory = Path.of(args[0]); + List<String> manifest = Files.readAllLines(directory.resolve("manifest.tsv")); + assert manifest.size() == 39 : "Missing compatibility fixtures"; Review Comment: Fixed in [df09e87b8f](https://github.com/apache/pinot/commit/df09e87b8f99948e03442cb3afbb887b3100dc5d). Added the reverse direction to the same isolated CI gate. Each real 3.2/3.3 JVM now writes verbose and compact payloads for a separate Pinot-only JVM to decode, including directly constructed legacy weighted endpoints and custom-capacity merged digests in addition to the existing rewrites. The Pinot reader checks centroid mass/means against the real legacy decoder, exact total mass/extrema/compression, selected initial quantiles, and add/serialize/read behavior. This passes 81 genuine legacy-written inputs and 34 initial quantile comparisons per version; a negative control changing a legacy-written centroid mean fails as expected. The old jars remain standalone test-tool artifacts, outside Pinot dependencies. -- 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]
