xiangfu0 commented on code in PR #19795:
URL: https://github.com/apache/pinot/pull/19795#discussion_r4222551901
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -696,34 +744,60 @@ 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() {
+ public byte[] serialize() {
+ if (_legacyDegraded) {
+ if (_pendingSerializedTDigest != null) {
+ return _pendingSerializedTDigest.clone();
+ }
+ if (!_legacySerializedDirty) {
+ return _legacySerializedBytes.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();
+ }
if (_pendingSerializedTDigest != null) {
- getSerializedTotalWeight(_pendingSerializedTDigest);
- if (ByteBuffer.wrap(_pendingSerializedTDigest).getInt() ==
VERBOSE_ENCODING) {
- byte[] serialized =
TDigestUtils.makeLegacyCompatible(_pendingSerializedTDigest);
- return serialized == _pendingSerializedTDigest ? serialized.clone() :
serialized;
+ if (hasWeightedBoundaryCentroids(_pendingSerializedTDigest,
_pendingSerializedMetadata.centroidCount())) {
+ 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();
}
- return _pendingSerializedTDigest.clone();
}
compress();
+ normalizeBoundaryCentroids();
return TDigestUtils.makeLegacyCompatible(toVerboseBytes());
}
- TDigest toTDigest() {
- if (_pendingSerializedTDigest != null) {
- return TDigestUtils.deserialize(_pendingSerializedTDigest);
+ private static boolean hasWeightedBoundaryCentroids(byte[] bytes, int count)
{
Review Comment:
**[simplification] `hasWeightedBoundaryCentroids` is a hand-rolled
byte-level re-parse** (duplicating the VERBOSE/SMALL header offsets and the
boundary-weight rule) over bytes whose centroids `inspectSerialized` already
scanned once, with the result cached in `_pendingSerializedMetadata`. Adding a
`weightedBoundaries` flag to `SerializedTDigestMetadata`, computed in the
existing single scan, deletes this second parser and removes one more place the
encoding layout is hardcoded.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/PercentileTDigestValueAggregator.java:
##########
@@ -152,128 +150,112 @@ private static NonFiniteAwareTDigest
asNonFiniteAware(TDigest value) {
}
private void updateMaxByteSize(TDigest value) {
Review Comment:
**[fragility] `updateMaxByteSize` reaches through two classes' internals to
size serialization buffers**: an instanceof chain unwrapping
`NonFiniteAwareTDigest._finiteDigest`, testing it for
`PercentileTDigestAccumulator`, and combining
`hasFractionalWeights()`/`getCentroidCountUpperBound()` with a magic `+ 6L` for
infinity-centroid splits. If either class changes its centroid accounting (e.g.
boundary repair adding a centroid, or the infinity split rules changing), this
bound silently under-estimates and `serializeAggregatedValue` fails at
segment-build time, far from the change. A `maxSerializedByteSize()` on the
`TDigest` contract would let each implementation own its bound.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -81,11 +61,15 @@ final class PercentileTDigestAccumulator extends TDigest {
private double[] _incomingWeights;
private double[] _sortedIncomingMeans;
private double[] _sortedIncomingWeights;
- private int[] _incomingOrder;
private SerializedTDigestInput _serializedTDigestInput;
private byte[] _pendingSerializedTDigest;
private double _pendingSerializedTotalWeight = Double.NaN;
+ private SerializedTDigestMetadata _pendingSerializedMetadata;
+ private boolean _legacyDegraded;
Review Comment:
**[architecture] `_legacyDegraded` is threaded through the whole hot
accumulator as many special-case branches** (add paths, compress, cdf/quantile,
both serialize encodings, all serialized-merge entry points, plus the
`_legacySerializedBytes`/`_legacySerializedDirty` bookkeeping). Every future
accumulator change must remember to re-check the flag or it silently merges
poisoned state into the K1 kernel. Since the decision is made once at decode
time (`needsLegacyFallback`), a small separate degraded `TDigest`
implementation (bytes + min/max/totalWeight, NaN quantiles, pass-through
serialize) returned by the decode layer would keep this class single-invariant;
merging degraded-into-live becomes one case in `add(TDigest)`.
##########
pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/PercentileTDigestAggregationFunctionTest.java:
##########
@@ -37,17 +39,26 @@
import org.apache.pinot.core.query.aggregation.AggregationResultHolder;
import
org.apache.pinot.core.query.aggregation.function.AggregationFunction.SerializedIntermediateResult;
import org.apache.pinot.core.query.aggregation.groupby.GroupByResultHolder;
+import
org.apache.pinot.segment.local.customobject.PercentileTDigestAccumulator;
import org.apache.pinot.segment.local.utils.TDigestUtils;
import org.apache.pinot.segment.spi.Constants;
+import org.apache.pinot.segment.spi.customobject.TDigest;
+import org.apache.pinot.segment.spi.customobject.TDigest.Centroid;
import org.apache.pinot.spi.data.FieldSpec.DataType;
import org.roaringbitmap.RoaringBitmap;
import org.testng.Assert;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertTrue;
public class PercentileTDigestAggregationFunctionTest {
private static final ExpressionContext EXPRESSION =
ExpressionContext.forIdentifier("col");
+ private static final Map<String, double[]> LEGACY_REDUCER_RANK_ERRORS =
loadLegacyReducerRankErrors();
Review Comment:
**[test robustness] The CSV oracle is loaded in a static initializer**, so a
missing/filtered resource or a malformed row fails every test in this class
with `ExceptionInInitializerError` rather than one reducer test with a clear
message; a row with fewer than the expected fields passes loading and later
surfaces as `ArrayIndexOutOfBoundsException` in the assertion helper. Also note
the oracle is effectively non-regenerable (requires checking out a pre-PR
commit, re-adding the removed tdunning dependency, and running with `-da`), and
its key omits merge order/reducer input, so new test dimensions can silently
validate against stale maxima. Consider lazy loading with field-count
validation, and documenting the regeneration recipe next to the CSV.
##########
pinot-perf/src/main/java/org/apache/pinot/perf/aggregation/TDigestBenchmarkUtils.java:
##########
@@ -75,13 +58,18 @@ static byte[] createFixedVerboseBytes(double[]
sortedValues, double compression)
start = 1 + (centroidId - 1) * numInteriorValues /
numInteriorCentroids;
end = 1 + centroidId * numInteriorValues / numInteriorCentroids;
}
- double sum = 0.0;
- for (int valueId = start; valueId < end; valueId++) {
- sum += sortedValues[valueId];
+ double mean = sortedValues[start];
+ for (int valueId = start + 1; valueId < end; valueId++) {
+ double fraction = 1.0 / (valueId - start + 1);
+ double value = sortedValues[valueId];
+ mean = Math.copySign(1.0, mean) == Math.copySign(1.0, value)
Review Comment:
**[benchmark validity] The FIXED_VERBOSE generator changed its bucket-mean
formula in this same PR** (exact sum/weight -> incremental sign-aware lerp +
range clamp), so the one benchmark mode whose purpose is feeding byte-identical
inputs across the implementation swap now produces different bytes pre- vs
post-PR. Any before/after comparison using FIXED_VERBOSE attributes
quality/perf drift partly to the input change rather than the merge
implementation. If the lerp form is needed for the new validation, consider
noting in the class doc that FIXED_VERBOSE outputs are not comparable across
this PR boundary.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -1501,26 +1479,35 @@ private void decode() {
}
int encodedCentroidCount = _numCentroids;
ensureCapacity(Math.addExact(encodedCentroidCount, 2));
- ByteBuffer input = ByteBuffer.wrap(_bytes);
- input.position(_centroidOffset);
- double totalWeight = 0.0;
- if (_centroidSize == VERBOSE_CENTROID_SIZE) {
- for (int i = 0; i < encodedCentroidCount; i++) {
- double weight = input.getDouble();
- _weights[i] = weight;
- _means[i] = input.getDouble();
- totalWeight += weight;
- }
+ if (Double.isNaN(_metadata.totalWeight())) {
+ _metadata = TDigestUtils.inspectSerialized(ByteBuffer.wrap(_bytes),
_metadata, _means, _weights);
+ _totalWeight = _metadata.totalWeight();
} else {
+ // Pending state was already inspected before retaining bytes. Its
later first read only materializes
+ // those validated centroids; regular merged inputs compute flags and
weights in the decode pass above.
+ ByteBuffer input = ByteBuffer.wrap(_bytes);
+ input.position(_centroidOffset);
for (int i = 0; i < encodedCentroidCount; i++) {
- double weight = input.getFloat();
- _weights[i] = weight;
- _means[i] = input.getFloat();
- totalWeight += weight;
+ _weights[i] = _centroidSize == TDigestUtils.VERBOSE_CENTROID_SIZE ?
input.getDouble() : input.getFloat();
+ double mean = _centroidSize == TDigestUtils.VERBOSE_CENTROID_SIZE ?
input.getDouble() : input.getFloat();
+ _means[i] = clamp(mean, _min, _max);
+ }
+ }
+ if (!_metadata.needsLegacyFallback()) {
+ if (_metadata.unorderedMeans()) {
+ it.unimi.dsi.fastutil.Arrays.mergeSort(0, encodedCentroidCount,
Review Comment:
**[convention] CLAUDE.md: "Prefer imports over fully qualified class names
(e.g., use `import com.foo.Bar` and refer to `Bar`, not `com.foo.Bar`
inline)."** `it.unimi.dsi.fastutil.Arrays.mergeSort(...)` is written inline
because `java.util.Arrays` occupies the simple name, but `import static
it.unimi.dsi.fastutil.Arrays.mergeSort;` satisfies the rule with no clash
(java.util.Arrays has no mergeSort member).
--
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]
