Copilot commented on code in PR #18996: URL: https://github.com/apache/pinot/pull/18996#discussion_r3593427044
########## pinot-segment-local/src/test/java/org/apache/pinot/segment/local/aggregator/PercentileTDigestValueAggregatorTest.java: ########## @@ -0,0 +1,150 @@ +/** + * 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.aggregator; + +import com.tdunning.math.stats.TDigest; +import java.nio.ByteBuffer; +import java.util.List; +import org.apache.pinot.common.request.Literal; +import org.apache.pinot.common.request.context.ExpressionContext; +import org.apache.pinot.segment.local.utils.CustomSerDeUtils; +import org.testng.annotations.DataProvider; +import org.testng.annotations.Test; + +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertTrue; + +/// Tests the serialization-size contract and deferred compression used by [PercentileTDigestValueAggregator]. +public class PercentileTDigestValueAggregatorTest { + + @Test(dataProvider = "compressionBounds") + public void testRawValuesTrackMaxVerboseSizeWithoutCompression(int compression, int numValues, + int expectedMaxByteSize) { + PercentileTDigestValueAggregator aggregator = newAggregator(compression); + assertEquals(aggregator.getMaxAggregatedValueByteSize(), 0); + + TDigest digest = aggregator.getInitialAggregatedValue(0.0); + for (int i = 1; i < numValues; i++) { + aggregator.applyRawValue(digest, i); + } + assertEquals(digest.centroidCount(), 0); + assertEquals(aggregator.getMaxAggregatedValueByteSize(), expectedMaxByteSize); + byte[] serialized = aggregator.serializeAggregatedValue(digest); + assertTrue(serialized.length <= expectedMaxByteSize); + } + + @DataProvider + public static Object[][] compressionBounds() { + return new Object[][]{ + {10, 30, 512}, + {100, 210, 3_392}, + {1_000, 2_010, 32_192} + }; + } + + @Test + public void testRawUpdatesDoNotForceCompression() { + PercentileTDigestValueAggregator aggregator = newAggregator(100); + TDigest digest = aggregator.getInitialAggregatedValue(0.0); + for (int i = 1; i < 256; i++) { + aggregator.applyRawValue(digest, i); + } + + assertEquals(digest.size(), 256L); + assertEquals(digest.centroidCount(), 0); + byte[] serialized = aggregator.serializeAggregatedValue(digest); + assertTrue(digest.centroidCount() > 0); + assertTrue(serialized.length <= aggregator.getMaxAggregatedValueByteSize()); + } + + @Test + public void testPreAggregatedCompressionExpandsRegisteredBound() { + TDigest input = TDigest.createMergingDigest(200); + input.add(42.0); + byte[] inputBytes = CustomSerDeUtils.TDIGEST_SER_DE.serialize(input); + + PercentileTDigestValueAggregator aggregator = newAggregator(10); + TDigest result = aggregator.getInitialAggregatedValue(inputBytes); + for (int i = 0; i < 409; i++) { + aggregator.applyRawValue(result, i); + } + + assertEquals(result.compression(), 200.0); + assertEquals(aggregator.getMaxAggregatedValueByteSize(), 6_592); + assertTrue(aggregator.serializeAggregatedValue(result).length <= aggregator.getMaxAggregatedValueByteSize()); Review Comment: This assertion serializes inside the check and then compares to `getMaxAggregatedValueByteSize()`, but serialization itself can increase the recorded max. To actually verify the aggregator’s registered bound is sufficient *before* serialization, capture the max first, then serialize and compare the resulting length against the captured value. ########## pinot-segment-local/src/test/java/org/apache/pinot/segment/local/aggregator/PercentileTDigestValueAggregatorTest.java: ########## @@ -0,0 +1,150 @@ +/** + * 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.aggregator; + +import com.tdunning.math.stats.TDigest; +import java.nio.ByteBuffer; +import java.util.List; +import org.apache.pinot.common.request.Literal; +import org.apache.pinot.common.request.context.ExpressionContext; +import org.apache.pinot.segment.local.utils.CustomSerDeUtils; +import org.testng.annotations.DataProvider; +import org.testng.annotations.Test; + +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertTrue; + +/// Tests the serialization-size contract and deferred compression used by [PercentileTDigestValueAggregator]. +public class PercentileTDigestValueAggregatorTest { + + @Test(dataProvider = "compressionBounds") + public void testRawValuesTrackMaxVerboseSizeWithoutCompression(int compression, int numValues, + int expectedMaxByteSize) { + PercentileTDigestValueAggregator aggregator = newAggregator(compression); + assertEquals(aggregator.getMaxAggregatedValueByteSize(), 0); + + TDigest digest = aggregator.getInitialAggregatedValue(0.0); + for (int i = 1; i < numValues; i++) { + aggregator.applyRawValue(digest, i); + } + assertEquals(digest.centroidCount(), 0); + assertEquals(aggregator.getMaxAggregatedValueByteSize(), expectedMaxByteSize); + byte[] serialized = aggregator.serializeAggregatedValue(digest); + assertTrue(serialized.length <= expectedMaxByteSize); + } + + @DataProvider + public static Object[][] compressionBounds() { + return new Object[][]{ + {10, 30, 512}, + {100, 210, 3_392}, + {1_000, 2_010, 32_192} + }; + } + + @Test + public void testRawUpdatesDoNotForceCompression() { + PercentileTDigestValueAggregator aggregator = newAggregator(100); + TDigest digest = aggregator.getInitialAggregatedValue(0.0); + for (int i = 1; i < 256; i++) { + aggregator.applyRawValue(digest, i); + } + + assertEquals(digest.size(), 256L); + assertEquals(digest.centroidCount(), 0); + byte[] serialized = aggregator.serializeAggregatedValue(digest); + assertTrue(digest.centroidCount() > 0); + assertTrue(serialized.length <= aggregator.getMaxAggregatedValueByteSize()); Review Comment: `serializeAggregatedValue()` updates `_maxByteSize` internally, so asserting `serialized.length <= aggregator.getMaxAggregatedValueByteSize()` after serialization is not validating the pre-serialization sizing contract (it will always pass unless serialization fails to update the max). Capture the max *before* serialization and compare against that instead. ########## pinot-segment-local/src/test/java/org/apache/pinot/segment/local/aggregator/PercentileTDigestValueAggregatorTest.java: ########## @@ -0,0 +1,150 @@ +/** + * 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.aggregator; + +import com.tdunning.math.stats.TDigest; +import java.nio.ByteBuffer; +import java.util.List; +import org.apache.pinot.common.request.Literal; +import org.apache.pinot.common.request.context.ExpressionContext; +import org.apache.pinot.segment.local.utils.CustomSerDeUtils; +import org.testng.annotations.DataProvider; +import org.testng.annotations.Test; + +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertTrue; + +/// Tests the serialization-size contract and deferred compression used by [PercentileTDigestValueAggregator]. +public class PercentileTDigestValueAggregatorTest { + + @Test(dataProvider = "compressionBounds") + public void testRawValuesTrackMaxVerboseSizeWithoutCompression(int compression, int numValues, + int expectedMaxByteSize) { + PercentileTDigestValueAggregator aggregator = newAggregator(compression); + assertEquals(aggregator.getMaxAggregatedValueByteSize(), 0); + + TDigest digest = aggregator.getInitialAggregatedValue(0.0); + for (int i = 1; i < numValues; i++) { + aggregator.applyRawValue(digest, i); + } + assertEquals(digest.centroidCount(), 0); + assertEquals(aggregator.getMaxAggregatedValueByteSize(), expectedMaxByteSize); + byte[] serialized = aggregator.serializeAggregatedValue(digest); + assertTrue(serialized.length <= expectedMaxByteSize); + } + + @DataProvider + public static Object[][] compressionBounds() { + return new Object[][]{ + {10, 30, 512}, + {100, 210, 3_392}, + {1_000, 2_010, 32_192} + }; + } + + @Test + public void testRawUpdatesDoNotForceCompression() { + PercentileTDigestValueAggregator aggregator = newAggregator(100); + TDigest digest = aggregator.getInitialAggregatedValue(0.0); + for (int i = 1; i < 256; i++) { + aggregator.applyRawValue(digest, i); + } + + assertEquals(digest.size(), 256L); + assertEquals(digest.centroidCount(), 0); + byte[] serialized = aggregator.serializeAggregatedValue(digest); + assertTrue(digest.centroidCount() > 0); + assertTrue(serialized.length <= aggregator.getMaxAggregatedValueByteSize()); + } + + @Test + public void testPreAggregatedCompressionExpandsRegisteredBound() { + TDigest input = TDigest.createMergingDigest(200); + input.add(42.0); + byte[] inputBytes = CustomSerDeUtils.TDIGEST_SER_DE.serialize(input); + + PercentileTDigestValueAggregator aggregator = newAggregator(10); + TDigest result = aggregator.getInitialAggregatedValue(inputBytes); + for (int i = 0; i < 409; i++) { + aggregator.applyRawValue(result, i); + } + + assertEquals(result.compression(), 200.0); + assertEquals(aggregator.getMaxAggregatedValueByteSize(), 6_592); + assertTrue(aggregator.serializeAggregatedValue(result).length <= aggregator.getMaxAggregatedValueByteSize()); + } + + @Test + public void testOversizedSmallEncodedDigestRemainsReadable() { + int centroidCount = 300; + byte[] smallEncoding = createSmallEncoding(10, 400, 500, centroidCount); + + PercentileTDigestValueAggregator aggregator = newAggregator(10); + TDigest result = aggregator.getInitialAggregatedValue(smallEncoding); + byte[] serialized = aggregator.serializeAggregatedValue(result); + TDigest clone = aggregator.cloneAggregatedValue(result); + + assertEquals(result.centroidCount(), centroidCount); + assertEquals(aggregator.getMaxAggregatedValueByteSize(), 4_832); + assertEquals(ByteBuffer.wrap(serialized).getInt(), 2); + assertEquals(serialized.length, smallEncoding.length); + assertTrue(serialized.length <= aggregator.getMaxAggregatedValueByteSize()); + assertEquals(clone.size(), result.size()); + assertEquals(clone.centroidCount(), result.centroidCount()); + assertEquals(clone.quantile(0.75), result.quantile(0.75)); + } + + @Test + public void testAggregatedUpdateDoesNotForceDestinationCompression() { + PercentileTDigestValueAggregator aggregator = newAggregator(100); + TDigest destination = aggregator.getInitialAggregatedValue(1.0); + TDigest source = TDigest.createMergingDigest(100); + source.add(2.0); + source.add(3.0); + + aggregator.applyAggregatedValue(destination, source); + + assertEquals(destination.size(), 3L); + assertEquals(destination.centroidCount(), 0); + byte[] serialized = aggregator.serializeAggregatedValue(destination); + assertTrue(serialized.length <= aggregator.getMaxAggregatedValueByteSize()); Review Comment: As written, this check compares `serialized.length` to `getMaxAggregatedValueByteSize()` *after* calling `serializeAggregatedValue()`, which updates the max internally. Capture the max before serialization so the test actually validates the sizing contract. ########## pinot-perf/src/main/java/org/apache/pinot/perf/aggregation/BenchmarkPercentileTDigestValueAggregator.java: ########## @@ -0,0 +1,104 @@ +/** + * 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.perf.aggregation; + +import com.tdunning.math.stats.TDigest; +import java.util.List; +import java.util.SplittableRandom; +import java.util.concurrent.TimeUnit; +import org.apache.pinot.segment.local.aggregator.PercentileTDigestValueAggregator; +import org.openjdk.jmh.annotations.Benchmark; +import org.openjdk.jmh.annotations.BenchmarkMode; +import org.openjdk.jmh.annotations.Fork; +import org.openjdk.jmh.annotations.Measurement; +import org.openjdk.jmh.annotations.Mode; +import org.openjdk.jmh.annotations.OutputTimeUnit; +import org.openjdk.jmh.annotations.Param; +import org.openjdk.jmh.annotations.Scope; +import org.openjdk.jmh.annotations.Setup; +import org.openjdk.jmh.annotations.State; +import org.openjdk.jmh.annotations.Threads; +import org.openjdk.jmh.annotations.Warmup; + +/// Measures the raw-value TDigest aggregation kernel used while constructing a star-tree index. +/// +/// The default models one million source rows split into groups of 1,000 rows with identical dimension keys. Each +/// group is serialized because every aggregate eventually becomes a star-tree forward-index entry. Use +/// `-p _numRows=100000000` for a 100-million-row run and vary `_rowsPerGroup` to model the table's dimension +/// cardinality. This deliberately excludes dimension sorting and forward-index I/O so it isolates +/// [PercentileTDigestValueAggregator]. +@State(Scope.Benchmark) +@BenchmarkMode(Mode.AverageTime) +@OutputTimeUnit(TimeUnit.MILLISECONDS) +@Warmup(iterations = 2, time = 1) +@Measurement(iterations = 5, time = 1) +@Fork(value = 2, jvmArgsAppend = {"-Xms4g", "-Xmx8g"}) +@Threads(1) +public class BenchmarkPercentileTDigestValueAggregator { + private static final int VALUE_BLOCK_SIZE = 1 << 16; + + @Param({"1000000"}) + int _numRows; + + @Param({"1000"}) + int _rowsPerGroup; + + private double[] _values; + + @Setup + public void setUp() { + if (_numRows <= 0) { + throw new IllegalArgumentException("numRows must be positive"); + } + if (_rowsPerGroup <= 0) { + throw new IllegalArgumentException("rowsPerGroup must be positive"); + } + _values = new double[VALUE_BLOCK_SIZE]; + SplittableRandom random = new SplittableRandom(42); + for (int i = 0; i < VALUE_BLOCK_SIZE; i++) { + _values[i] = random.nextDouble(); + } + } + + @Benchmark + public long aggregateAndSerializeRawValues() { + PercentileTDigestValueAggregator aggregator = new PercentileTDigestValueAggregator(List.of()); + long totalAggregatedRows = 0; + long totalSerializedBytes = 0; + for (int from = 0; from < _numRows; from += _rowsPerGroup) { + int to = Math.min(_numRows, from + _rowsPerGroup); + TDigest digest = aggregator.getInitialAggregatedValue(_values[from & (VALUE_BLOCK_SIZE - 1)]); + for (int i = from + 1; i < to; i++) { + aggregator.applyRawValue(digest, _values[i & (VALUE_BLOCK_SIZE - 1)]); + } + byte[] serialized = aggregator.serializeAggregatedValue(digest); + if (serialized.length > aggregator.getMaxAggregatedValueByteSize()) { + throw new IllegalStateException( + "Serialized TDigest exceeds registered maximum: " + serialized.length + " > " + + aggregator.getMaxAggregatedValueByteSize()); + } Review Comment: This benchmark’s safety check is ineffective because `serializeAggregatedValue()` updates `getMaxAggregatedValueByteSize()` before the comparison, so `serialized.length > aggregator.getMaxAggregatedValueByteSize()` should never be true. Capture the max before serialization and compare against that to validate the contract being benchmarked. -- 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]
