xiangfu0 commented on code in PR #19795:
URL: https://github.com/apache/pinot/pull/19795#discussion_r4216258962
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TDigestUtils.java:
##########
@@ -52,31 +53,16 @@ public final class TDigestUtils {
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.
- 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.
- public static TDigest createMergingDigestWithLegacyBuffer(double
compression, int minimumMainCapacity) {
- if (minimumMainCapacity < 0) {
- throw new IllegalArgumentException("Minimum main capacity must not be
negative: " + minimumMainCapacity);
+ /// Rejects non-positive or non-finite compression while retaining
historical positive finite settings.
+ public static void validateCompression(double compression) {
+ if (!(compression > 0.0) || !Double.isFinite(compression)) {
Review Comment:
**[compat regression] compression <= 0 now throws where it previously
clamped to 10.**
The deleted factory path tolerated non-positive compression (old code:
`double normalizedCompression = Math.max(compression, 10.0)`; tdunning's
`MergingDigest` constructor itself clamps `compression < 10` to 10).
`validateCompression` now throws `IllegalArgumentException` for `compression <=
0`, and the star-tree/ingestion-aggregation path reaches it via
`TDigestUtils.createMergingDigest` (NonFiniteAwareTDigest constructor).
Reproduced: `createMergingDigest(0)` worked on the base (clamped to 10) and
throws on this PR. A table with star-tree index or ingestion aggregation
configured as `percentileTDigest(col, 0)` that built fine before will fail
segment build/consumption after upgrade. Consider clamping (with a warning)
instead of throwing for the config-driven path.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -240,61 +263,86 @@ public void add(List<? extends TDigest> others) {
}
private void addAccumulator(PercentileTDigestAccumulator other) {
+ _hasFractionalWeights |= other._hasFractionalWeights;
+ if (other == this) {
+ addSerializedTDigest(serialize());
+ return;
+ }
if (other._pendingSerializedTDigest != null) {
- addSerializedTDigest(other._pendingSerializedTDigest);
+ SerializedTDigestInput input = new SerializedTDigestInput();
+ input.reset(other._pendingSerializedTDigest,
other._pendingSerializedMetadata);
+ input._retainedBytes = other._pendingSerializedTDigest;
+ addSerializedTDigest(input);
+ return;
+ }
+ if (other._legacyDegraded) {
+ addSerializedTDigest(other.serialize());
return;
}
other.compress();
materializePendingSerializedTDigest();
+ checkTotalWeight(_totalWeight + _incomingWeight + _numRawValues +
other._totalWeight);
+ if (_legacyDegraded) {
+ _totalWeight += other._totalWeight;
+ _min = Math.min(_min, other._min);
+ _max = Math.max(_max, other._max);
+ return;
+ }
preserveSerializedCapacity(other._serializedMainCapacity,
other._serializedBufferCapacity);
bufferIncomingCentroids(other._centroidMeans, other._centroidWeights,
other._numCentroids);
_min = Math.min(_min, other._min);
_max = Math.max(_max, other._max);
}
- void addSerializedTDigest(byte[] bytes) {
- if (!_hasSerializedInput && _numCentroids == 0 && _totalWeight == 0.0 &&
_numRawValues == 0
- && _numIncomingCentroids == 0) {
- byte[] retainedBytes = bytes.clone();
- _pendingSerializedTotalWeight = getSerializedTotalWeight(retainedBytes);
- _pendingSerializedTDigest = retainedBytes;
- _hasSerializedInput = true;
- return;
- }
+ public void addSerializedTDigest(byte[] bytes) {
materializePendingSerializedTDigest();
if (_serializedTDigestInput == null) {
_serializedTDigestInput = new SerializedTDigestInput();
}
_serializedTDigestInput.reset(bytes);
- _serializedTDigestInput.decode();
- mergeSerializedTDigest(_serializedTDigestInput);
- _hasSerializedInput = true;
+ addSerializedTDigest(_serializedTDigestInput);
}
- void addSerializedTDigest(SerializedTDigestInput input) {
+ public void addSerializedTDigest(SerializedTDigestInput input) {
+ _hasFractionalWeights |= input._metadata.fractionalWeights();
if (!_hasSerializedInput && _numCentroids == 0 && _totalWeight == 0.0 &&
_numRawValues == 0
&& _numIncomingCentroids == 0) {
_pendingSerializedTDigest = input.retainBytes();
- _pendingSerializedTotalWeight = Double.NaN;
+ _pendingSerializedTotalWeight = input._totalWeight;
Review Comment:
**[consistency] Pending-adoption branch skips `checkTotalWeight`, making
corrupt negative-weight legacy digests merge-order dependent.**
The fast path stores `_pendingSerializedTotalWeight = input._totalWeight`
without validation, while the non-pending branch below enforces
`checkTotalWeight(_totalWeight + ... + input._totalWeight)`. Verified
dynamically: a verbose digest with a single weight of -5 deserializes alone
with `size() == -5` and NaN quantiles, but merging the same bytes into a
smaller non-empty accumulator throws `IllegalArgumentException("Invalid TDigest
total weight")` — so for the same stored bytes, one plan shape errors and
another silently returns NaN, decided by merge order. Either validate in the
pending branch too, or exempt legacy-degraded totals from `checkTotalWeight` in
the merge path.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/PercentileTDigestValueAggregator.java:
##########
@@ -291,45 +311,50 @@ public void add(double value, int weight) {
if (weight <= 0) {
throw new IllegalArgumentException("TDigest weight must be positive: "
+ weight);
}
+ checkTotalWeight(getTotalWeight() + weight);
invalidateCaches();
if (value == Double.NEGATIVE_INFINITY) {
- _negativeInfinityWeight = Math.addExact(_negativeInfinityWeight,
weight);
+ _negativeInfinityWeight = _negativeInfinityWeight + weight;
} else if (value == Double.POSITIVE_INFINITY) {
- _positiveInfinityWeight = Math.addExact(_positiveInfinityWeight,
weight);
+ _positiveInfinityWeight = _positiveInfinityWeight + weight;
} else {
_finiteDigest.add(value, weight);
}
}
@Override
public void add(TDigest other) {
+ checkTotalWeight(getTotalWeight() + other.getTotalWeight());
if (other instanceof NonFiniteAwareTDigest) {
NonFiniteAwareTDigest wrapped = (NonFiniteAwareTDigest) other;
- long negativeInfinityWeight = wrapped._negativeInfinityWeight;
- long positiveInfinityWeight = wrapped._positiveInfinityWeight;
- // MergingDigest.add(List) copies the implementation's
double-precision weight arrays directly. The generic
- // add(TDigest) path goes through Centroid.count(), which is limited
to an int and truncates large weights.
+ double negativeInfinityWeight = wrapped._negativeInfinityWeight;
+ double positiveInfinityWeight = wrapped._positiveInfinityWeight;
+ // Merging shared state preserves double-precision weights without
narrowing through Centroid.count().
wrapped.invalidateCaches();
invalidateCaches();
- if (wrapped._finiteDigest.size() > 0L) {
+ if (wrapped._finiteDigest.getTotalWeight() > 0.0) {
Review Comment:
**[consistency] Zero-net-weight legacy-degraded digests are silently dropped
on the star-tree path.**
`if (wrapped._finiteDigest.getTotalWeight() > 0.0)` (and the same gate in
`copy()` at L219, which backs `cloneAggregatedValue`) skips merging when the
poisoned digest's net total weight is <= 0. A stored digest with historical
poisoned weights summing to zero (e.g. {1.0, -1.0}) — which this PR now accepts
as legacy-degraded instead of rejecting — then contributes nothing to the
star-tree aggregate, so the star-tree answer is a clean quantile while the raw
query path over the same rows degrades to NaN via
`PercentileTDigestAccumulator.addSerializedTDigest`. `copy()` likewise returns
a fresh empty digest (min=+Inf/max=-Inf, no NaN marker) for such a source. The
degraded flag should propagate through `add`/`copy` regardless of net weight.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -696,34 +685,59 @@ 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 (_totalWeight == _legacySerializedTotalWeight && _min ==
_legacySerializedMin
+ && _max == _legacySerializedMax) {
+ 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();
Review Comment:
**[perf/accuracy] Re-serializing an untouched legacy digest with weighted
boundary centroids now runs a full lossy K1 recompression instead of the
previous byte pass-through.**
Pre-PR `serialize()` returned pending verbose bytes via
`makeLegacyCompatible` without materializing. Now
`hasWeightedBoundaryCentroids(...)` forces
`materializePendingSerializedTDigest()`, and since `mergeSerializedTDigest`'s
initialize path never sets `_publiclyCompressed`, the subsequent `compress()`
runs `recompressCentroids` (full K1 merge) plus `normalizeBoundaryCentroids` +
`toVerboseBytes` + `makeLegacyCompatible`. Weighted boundaries are the norm for
bytes written by tdunning 3.2 / pre-upgrade Pinot, so every pass-through
intermediate (one digest per group, high-cardinality group-by) pays decode +
recompression + two encode passes per hop, and the recompression is lossy, so
bytes are no longer preserved. If boundary repair is all that is needed,
initializing from an already-finalized digest could set `_publiclyCompressed =
true` and skip the K1 pass (boundary repair alone adds at most 2 centroids).
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -1452,42 +1468,58 @@ static final class SerializedTDigestInput {
private double _totalWeight;
private boolean _decoded;
- void reset(byte[] bytes) {
- _retainedBytes = null;
- ByteBuffer input = ByteBuffer.wrap(bytes);
- int encoding = input.getInt();
- if (encoding != VERBOSE_ENCODING && encoding != SMALL_ENCODING) {
- throw new IllegalStateException("Invalid format for serialized
histogram");
- }
+ private SerializedTDigestMetadata _metadata;
+
+ public void reset(byte[] bytes) {
+ reset(bytes, TDigestUtils.inspectSerialized(ByteBuffer.wrap(bytes)));
Review Comment:
**[perf] Every serialized input is now scanned twice: `reset()` runs
`inspectSerialized`'s full per-centroid walk, then `decode()` re-reads the same
centroids.**
The pre-PR `reset()`/`validateEncodedDigest` parsed only the fixed-size
header in O(1); total weight was accumulated during the single `decode()` pass
(or computed lazily for pending state). Now every call to
`aggregate`/`aggregateGroupBySV`/`aggregateGroupByMV` over a BYTES t-digest
column pays a validation walk (weights, means, rint, ordering, bounds) plus the
decode walk per input row — roughly doubling deserialization CPU for
pre-aggregated percentile queries. Consider computing the metadata flags inside
`decode()`'s existing loop and only running the standalone inspection when the
input goes pending without being decoded.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TDigestUtils.java:
##########
@@ -91,39 +77,29 @@ public static byte[] serialize(TDigest tDigest) {
public static byte[] serialize(TDigest tDigest, ByteBuffer scratchBuffer) {
ByteOrder scratchOrder = scratchBuffer != null ? scratchBuffer.order() :
null;
try {
- if (!(tDigest instanceof MergingDigest)) {
- // Pinot has always decoded TDigest values as MergingDigest. Preserve
the previous serialization behavior for
- // other implementations instead of interpreting their
implementation-specific payload as MergingDigest
- // bytes.
- ByteBuffer buffer = prepareScratchBuffer(scratchBuffer,
tDigest.byteSize());
- tDigest.asBytes(buffer);
- byte[] bytes = new byte[buffer.position()];
- buffer.flip();
- buffer.get(bytes);
+ if (tDigest instanceof PercentileTDigestAccumulator) {
+ // Preserve the legacy two-level raw flush before final compression.
Pending serialized state answers
+ // this count from cached metadata, so stored reads do not acquire an
extra lossy compression pass.
+ tDigest.centroidCount();
+ byte[] bytes = ((PercentileTDigestAccumulator) tDigest).serialize();
+ if (scratchBuffer != null && !scratchBuffer.isReadOnly() &&
scratchBuffer.capacity() >= bytes.length) {
+ scratchBuffer.clear();
Review Comment:
**[dead work] The scratch-buffer copy in the accumulator branch is never
read.**
`scratchBuffer.clear(); scratchBuffer.put(bytes);` copies the
already-materialized array into the caller's scratch buffer, but the method
returns `bytes` and no caller reads the scratch contents (the only
scratch-passing chain,
`PercentileTDigestValueAggregator.serializeAggregatedValue ->
getSerializedBytes -> getFiniteSerializedBytes`, uses only the returned array)
— so every star-tree record serialization performs an extra full memcpy whose
result is discarded. The buffer is also left unflipped (position ==
bytes.length), which contradicts the javadoc's 'cleared and may be reused' and
would hand stale/underflowing state to any caller that did rely on it. `return
((PercentileTDigestAccumulator) tDigest).serialize();` does the same job.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/PercentileTDigestValueAggregator.java:
##########
@@ -291,45 +311,50 @@ public void add(double value, int weight) {
if (weight <= 0) {
throw new IllegalArgumentException("TDigest weight must be positive: "
+ weight);
}
+ checkTotalWeight(getTotalWeight() + weight);
invalidateCaches();
if (value == Double.NEGATIVE_INFINITY) {
- _negativeInfinityWeight = Math.addExact(_negativeInfinityWeight,
weight);
+ _negativeInfinityWeight = _negativeInfinityWeight + weight;
} else if (value == Double.POSITIVE_INFINITY) {
- _positiveInfinityWeight = Math.addExact(_positiveInfinityWeight,
weight);
+ _positiveInfinityWeight = _positiveInfinityWeight + weight;
} else {
_finiteDigest.add(value, weight);
}
}
@Override
public void add(TDigest other) {
+ checkTotalWeight(getTotalWeight() + other.getTotalWeight());
if (other instanceof NonFiniteAwareTDigest) {
NonFiniteAwareTDigest wrapped = (NonFiniteAwareTDigest) other;
- long negativeInfinityWeight = wrapped._negativeInfinityWeight;
- long positiveInfinityWeight = wrapped._positiveInfinityWeight;
- // MergingDigest.add(List) copies the implementation's
double-precision weight arrays directly. The generic
- // add(TDigest) path goes through Centroid.count(), which is limited
to an int and truncates large weights.
+ double negativeInfinityWeight = wrapped._negativeInfinityWeight;
+ double positiveInfinityWeight = wrapped._positiveInfinityWeight;
+ // Merging shared state preserves double-precision weights without
narrowing through Centroid.count().
wrapped.invalidateCaches();
invalidateCaches();
- if (wrapped._finiteDigest.size() > 0L) {
+ if (wrapped._finiteDigest.getTotalWeight() > 0.0) {
_finiteDigest.add(List.of(wrapped._finiteDigest));
}
- _negativeInfinityWeight = Math.addExact(_negativeInfinityWeight,
negativeInfinityWeight);
- _positiveInfinityWeight = Math.addExact(_positiveInfinityWeight,
positiveInfinityWeight);
+ _negativeInfinityWeight = _negativeInfinityWeight +
negativeInfinityWeight;
+ _positiveInfinityWeight = _positiveInfinityWeight +
positiveInfinityWeight;
return;
}
- if (other.size() == 0L) {
+ if (other.getTotalWeight() == 0.0) {
return;
}
invalidateCaches();
if (Double.isFinite(other.getMin()) && Double.isFinite(other.getMax())) {
_finiteDigest.add(List.of(other));
return;
}
- for (Centroid centroid : other.centroids()) {
- add(centroid.mean(), centroid.count());
- }
+ // Replay the double-precision wire weights: Centroid.count() truncates
positive fractional mass to zero.
+ ByteBuffer encoded = ByteBuffer.allocate(other.byteSize());
Review Comment:
**[perf] The other digest is fully serialized twice per infinity merge.**
`ByteBuffer.allocate(other.byteSize())` then `other.asBytes(encoded)`: for a
`PercentileTDigestAccumulator`, `byteSize()` is `serialize().length` and
`asBytes()` is `put(serialize())`, so compress + normalize + encode +
`makeLegacyCompatible` run end-to-end twice and the first result is discarded —
for every `applyAggregatedValue` merge whose extrema include +/-Infinity during
segment build. The same `byteSize()/asBytes()` double-serialize trap exists in
`PercentileTDigestAccumulator.add(TDigest)`. Serialize once
(`TDigestUtils.serialize(other)`) and wrap the resulting array.
--
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]
