xiangfu0 commented on code in PR #19795:
URL: https://github.com/apache/pinot/pull/19795#discussion_r4215691845
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -1041,22 +976,56 @@ private void ensureIncomingCapacity(int capacity) {
private void ensureIncomingSortCapacity(int capacity) {
ensureSortedIncomingCapacity(capacity);
if (_incomingOrder == null || capacity > _incomingOrder.length) {
- int newCapacity = _incomingOrder == null ? Math.max(capacity,
_centroidCapacity)
+ int newCapacity = _incomingOrder == null
+ ? Math.max(capacity, Math.min(_centroidCapacity,
_pendingCentroidCapacity))
: Math.max(capacity, Math.multiplyExact(_incomingOrder.length, 2));
_incomingOrder = new int[newCapacity];
+ _incomingOrderScratch = new int[newCapacity];
}
}
+ // Stable, primitive merge sort bounds the work to O(n log n), including
repeated and infinite means.
+ private void stableSortIncoming(int count) {
Review Comment:
Replaced the handwritten sort with fastutil stable mergeSort and removed the
retained second sort-index scratch array. Equal means keep stable order,
including signed zero.
Addressed in
[1fc37ba77c](https://github.com/apache/pinot/commit/1fc37ba77c631c574df59a6e211bb978954fb2b2).
All 463 targeted tests and 322 legacy reader cases pass.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TDigestUtils.java:
##########
@@ -48,35 +45,34 @@ public final class TDigestUtils {
private static final int SMALL_HEADER_SIZE = 30;
private static final int SMALL_CENTROID_SIZE = 8;
private static final int DEFAULT_MERGE_BUFFER_MULTIPLIER = 5;
+ private static final double MAX_COMPRESSION = 1_000_000.0;
private TDigestUtils() {
}
- /// Creates a merging digest with Pinot's accuracy-preserving scale function.
+ /// Creates a digest using Pinot's accuracy-preserving K1 implementation.
public static TDigest createMergingDigest(double compression) {
- return configureScaleFunction(new MergingDigest(compression));
+ return PercentileTDigestAccumulator.forLegacyAggregation(compression);
}
- /// Creates a merging digest with the temporary-buffer footprint used by
t-digest 3.2.
- ///
- /// Segment aggregation can keep many digests live at once. T-digest 3.3
more than doubled the default temporary
- /// buffer to support two-level compression, adding about 11 KiB to every
compression-100 digest. This factory keeps
- /// the 3.3 main-buffer capacity and Pinot's K1 policy, but restores the
smaller temporary buffer for that
- /// high-cardinality use case.
+ /// Creates Pinot's buffered digest for segment aggregation.
public static TDigest createMergingDigestWithLegacyBuffer(double
compression) {
return createMergingDigestWithLegacyBuffer(compression, 0);
}
- /// Creates a reduced-buffer merging digest whose main array can hold at
least `minimumMainCapacity` centroids.
+ /// Creates a digest whose dynamically grown arrays can hold the requested
centroid capacity.
public static TDigest createMergingDigestWithLegacyBuffer(double
compression, int minimumMainCapacity) {
Review Comment:
Removed the misleading legacy-buffer aliases and discarded
minimumMainCapacity argument. Wrapper copy uses the single factory without
forcing centroidCount() for unused pre-sizing.
Addressed in
[1fc37ba77c](https://github.com/apache/pinot/commit/1fc37ba77c631c574df59a6e211bb978954fb2b2).
All 463 targeted tests and 322 legacy reader cases pass.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/TDigest.java:
##########
@@ -0,0 +1,70 @@
+/**
+ * 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;
+
+import java.nio.ByteBuffer;
+import java.util.Collection;
+import java.util.List;
+
+/// Pinot's mutable t-digest contract using the existing K1 aggregation
behavior and legacy byte encodings.
+///
+/// Implementations are not safe for concurrent mutation; callers must
serialize access to each aggregation state.
+public abstract class TDigest {
+ public static TDigest createMergingDigest(double compression) {
Review Comment:
Removed the static factory from the TDigest contract and migrated all
callers to TDigestUtils.createMergingDigest, leaving one factory for this
behavior.
Addressed in
[1fc37ba77c](https://github.com/apache/pinot/commit/1fc37ba77c631c574df59a6e211bb978954fb2b2).
All 463 targeted tests and 322 legacy reader cases pass.
##########
pom.xml:
##########
@@ -253,7 +253,6 @@
<hadoop-shaded-protobuf_3_25.version>1.5.0</hadoop-shaded-protobuf_3_25.version>
<clearspring-stream-lib.version>2.9.8</clearspring-stream-lib.version>
<datasketches-java.version>9.0.0</datasketches-java.version>
- <t-digest.version>3.3</t-digest.version>
Review Comment:
Added backward-incompat and an explicit migration/release note in the PR
description. Java plugins must migrate their Pinot-facing signatures to the new
TDigest type and recompile. Independent tdunning usage requires a plugin-owned
dependency; bundling it alone does not restore old Pinot method signatures.
Legacy BYTES/type 10 remain supported.
Addressed in
[1fc37ba77c](https://github.com/apache/pinot/commit/1fc37ba77c631c574df59a6e211bb978954fb2b2).
All 463 targeted tests and 322 legacy reader cases pass.
##########
pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/PercentileTDigestAggregationFunctionTest.java:
##########
@@ -1005,59 +1006,29 @@ public void testReducerMergeStructureAndAccuracy(int
fanIn, int compression, Red
sourceValues[valueIndex] = value;
}
sources[digestIndex] = createReducerSource(sourceValues, compression,
reducerInput);
- referenceSources[digestIndex] =
createReferenceReducerSource(sourceValues, compression, reducerInput);
}
PercentileTDigestAggregationFunction function =
new PercentileTDigestAggregationFunction(EXPRESSION, 75.0,
compression, false);
TDigest result = TDigest.createMergingDigest(compression);
- TDigest referenceResult = TDigest.createMergingDigest(compression);
- AssertionError referenceFailure = null;
for (int sourceIndex : mergeOrder.sourceIndexes(fanIn)) {
result = function.merge(result, sources[sourceIndex]);
- if (referenceFailure == null) {
- try {
- if (referenceResult.size() == 0L) {
- referenceResult = referenceSources[sourceIndex];
- } else {
- referenceResult.add(referenceSources[sourceIndex]);
- }
- } catch (AssertionError e) {
- // t-digest 3.3 can violate its own singleton-endpoint assertion
when many duplicate-valued digests merge.
- referenceFailure = e;
- }
- }
}
Arrays.sort(rawValues);
String caseDescription =
fanIn + "/" + compression + "/" + distribution + "/" + mergeOrder +
"/" + reducerInput;
- // Interpolation between large equal-valued centroid runs can land inside
the empirical CDF jump. Keep a fixed
- // envelope for that discontinuous distribution instead of requiring every
merge order to match the reference's
- // particular centroid partition.
- double frozenRankError = compression == 20 || distribution ==
ReducerDistribution.DUPLICATE_HEAVY ? 0.06 : 0.02;
- double[] referenceRankErrors = null;
- if (referenceFailure == null) {
- try {
- referenceRankErrors = getReducerRankErrors(referenceResult, rawValues,
caseDescription + "/reference");
- } catch (AssertionError e) {
- referenceFailure = e;
- }
- }
- double[] maxRankErrors = new double[6];
- if (referenceFailure == null) {
- for (int i = 0; i < referenceRankErrors.length; i++) {
- maxRankErrors[i] = Math.max(frozenRankError, referenceRankErrors[i] +
1e-12);
- }
- } else {
- Arrays.fill(maxRankErrors, compression == 20 ? 0.25 : compression == 100
? 0.06 : 0.02);
- }
- assertValidReducerResult(result, rawValues, caseDescription,
maxRankErrors);
+ // Validate against the empirical CDF. The former 3.3 reference allowed
0.25 after duplicate-boundary
+ // assertions at compression 20; preserve that historical envelope for
this discontinuous distribution.
+ double maxRankError = compression == 20 && distribution ==
ReducerDistribution.DUPLICATE_HEAVY
Review Comment:
Added an independent real t-digest 3.3 corpus across all 270 seeded reducer
cases, with per-quantile accuracy guards, and frozen independent
custom-compression P75 values. Capacity tests now enforce the actual 3.2 limit
and check weighted quantiles, so skipping reduction is detected. The broad 0.25
allowance is removed.
Addressed in
[1fc37ba77c](https://github.com/apache/pinot/commit/1fc37ba77c631c574df59a6e211bb978954fb2b2).
All 463 targeted tests and 322 legacy reader cases pass.
--
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]
