xiangfu0 commented on code in PR #19795: URL: https://github.com/apache/pinot/pull/19795#discussion_r4241326604
########## pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/tdigest/TDigestCodec.java: ########## @@ -0,0 +1,698 @@ +/** + * 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. Pure reads retain the legacy centroids and interpolation. Mutation and serialization repair their boundaries +/// only when unit endpoints can preserve their total weight and first moment; otherwise unchanged historical bytes +/// remain readable and mutation fails explicitly. +/// 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)) { Review Comment: `validateCompression` only rejects non-finite values, so compression settings that previously failed fast now get silently clamped to 10: the old accumulator constructor threw `IllegalArgumentException("TDigest compression must be positive")`, which `PERCENTILETDIGEST(col, 50, 0)` (0 passes the factory's `>= 0` check) and `PERCENTILESMARTTDIGEST(col, 90, 'compression=-5')` (the smart function parses compression with no sign check) both reached. After this change those queries run silently at compression 10. If the clamp is intentional (it matches tdunning's and the old group-by path's behavior), consider rejecting `compression <= 0` here anyway so obviously-invalid user input still surfaces as an error instead of being masked. ########## pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/tdigest/SerializedTDigestInput.java: ########## @@ -0,0 +1,155 @@ +/** + * 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.ByteBuffer; +import org.apache.pinot.segment.local.customobject.tdigest.TDigestCodec.SerializedTDigestMetadata; + +/// Reusable, invocation-local view of one serialized TDigest for group-by-MV fanout. +/// +/// [#reset(byte\[\])] validates the header once per input row. The first decode combines numerical validation +/// with filling reusable centroid arrays, so all groups for a row merge the same primitive input without sharing +/// mutable accumulator state. A pending accumulator requests an immutable byte snapshot, created at most once per +/// reset +/// and shared by the row's group fanout. The input must remain thread-confined and must not outlive the aggregation +/// call that owns it. Pending-only inputs inspect their bytes without allocating centroid arrays; their first +/// later materialization reads the already validated values without another numerical validation pass. +public final class SerializedTDigestInput { + byte[] _bytes; + byte[] _retainedBytes; + int _numCentroids; + double[] _means; + double[] _weights; + private boolean _decoded; + private boolean _metadataInspected; + + SerializedTDigestMetadata _metadata; + + public void reset(byte[] bytes) { + reset(bytes, TDigestCodec.readSerializedHeader(ByteBuffer.wrap(bytes))); Review Comment: Per-row allocation regression on the BYTES aggregation hot path: `reset(byte[])` now allocates an escaping 18-field `SerializedTDigestMetadata` record plus `slice()`d ByteBuffers via `TDigestCodec.readSerializedHeader`, and the subsequent `decode()`/`inspectSerialized` allocates a second record — at least two records and ~4 buffer objects per input row. The pre-PR `SerializedTDigestInput` parsed the header into reused primitive fields with no escaping per-row objects. `PercentileTDigestAggregationFunction.aggregate`/`aggregateGroupBySV`/`aggregateGroupByMV` call `input.reset(bytesValues[i])` once per row over pre-aggregated BYTES columns, so this adds young-gen churn on a per-row loop that previously allocated nothing. A mutable header holder inside the reusable input (materializing the immutable record only when bytes are retained as pending) would restore the old behavior. ########## pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileTDigestAggregationFunction.java: ########## @@ -354,7 +356,7 @@ public boolean canUseStarTree(Map<String, Object> functionParameters) { protected static TDigest getDefaultTDigest(AggregationResultHolder aggregationResultHolder, int compressionFactor) { TDigest tDigest = aggregationResultHolder.getResult(); if (tDigest == null) { - tDigest = TDigestUtils.createMergingDigest(compressionFactor); + tDigest = PercentileTDigestAccumulator.forLegacyAggregation(compressionFactor); Review Comment: This `getDefaultTDigest(AggregationResultHolder, int)` overload has no callers left (the raw-value paths now go through `getAccumulator`), and it diverges from the live `GroupByResultHolder` overload: this one creates `forLegacyAggregation` (two-level working compression), the group-by one creates `new PercentileTDigestAccumulator(compressionFactor)` (incremental compression). A subclass or future caller picking this helper would get different flush/compression strategy — and different percentile accuracy — than every other raw-aggregation path. Suggest deleting it, or aligning it with the group-by overload. ########## compatibility-verifier/tdigest-compatibility/run.sh: ########## @@ -0,0 +1,53 @@ +#!/usr/bin/env bash +# +# 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. +# + +# Run after LegacyTDigestCompatibilityTest. Legacy jars are test-tool artifacts, not Pinot dependencies. +set -euo pipefail +# Match the unit-test workflow's Maven Central retry and connection settings, including standalone runs. +export MAVEN_OPTS="${MAVEN_OPTS:-} -Dmaven.wagon.httpconnectionManager.ttlSeconds=25 \ +-Dmaven.wagon.http.retryHandler.count=30 -Dhttp.keepAlive=false -Dmaven.wagon.http.pool=false" +TDIGEST_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)" +TDIGEST_FIXTURES="$TDIGEST_ROOT/pinot-segment-local/target/tdigest-compat-fixtures" +TDIGEST_WORK="$TDIGEST_ROOT/pinot-segment-local/target/tdigest-legacy-reader" +if [[ ! -s "$TDIGEST_FIXTURES/manifest.tsv" ]]; then + printf 'Missing or empty t-digest fixture manifest: %s. Run LegacyTDigestCompatibilityTest first.\n' \ + "$TDIGEST_FIXTURES/manifest.tsv" >&2 + exit 1 +fi +mkdir -p "$TDIGEST_WORK" +javac -cp "$TDIGEST_ROOT/pinot-segment-local/target/classes" -d "$TDIGEST_WORK" \ + "$TDIGEST_ROOT/compatibility-verifier/tdigest-compatibility/PinotReader.java" +for TDIGEST_VERSION in 3.2 3.3; do + TDIGEST_JAR="$TDIGEST_WORK/t-digest-$TDIGEST_VERSION.jar" + if [[ ! -s "$TDIGEST_JAR" ]]; then + # Resolution uses the workflow-cached Maven repository; retain each immutable copy for subsequent runs. + "$TDIGEST_ROOT/mvnw" -f "$TDIGEST_ROOT/pom.xml" -N -B -ntp \ + org.apache.maven.plugins:maven-dependency-plugin:3.8.1:copy \ + "-Dartifact=com.tdunning:t-digest:$TDIGEST_VERSION:jar" "-DoutputDirectory=$TDIGEST_WORK" + fi + javac -cp "$TDIGEST_JAR" -d "$TDIGEST_WORK" \ + "$TDIGEST_ROOT/compatibility-verifier/tdigest-compatibility/LegacyReader.java" + java -ea -Xmx128m -cp "$TDIGEST_WORK:$TDIGEST_JAR" LegacyReader "$TDIGEST_FIXTURES" "$TDIGEST_VERSION" Review Comment: Blanket `java -ea` also enables assertions *inside the tdunning jar*, arming t-digest 3.3's internal merge assertion (`MergingDigest.merge:379`) that spuriously fails on valid digests with non-unit boundary centroids — the same known-bad assertion the rank-error oracle generation has to run with `-da`. Today's fixtures survive only because Pinot's serializer happens to split endpoints into unit-weight singletons; any future fixture or serializer change that legitimately preserves a non-unit boundary centroid would fail CI inside tdunning's assertion and be misread as a Pinot compatibility regression. `LegacyReader` only needs its own assertions: use `java -ea:LegacyReader` (it already self-checks `desiredAssertionStatus`). -- 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]
