Jackie-Jiang commented on code in PR #19795:
URL: https://github.com/apache/pinot/pull/19795#discussion_r4235761745
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TDigestUtils.java:
##########
@@ -86,44 +85,65 @@ public static byte[] serialize(TDigest tDigest) {
/// Serializes a digest using `scratchBuffer` for the temporary verbose
representation when it is large enough.
///
- /// The returned array is independently owned. The scratch buffer is cleared
and may be reused by the caller after
- /// this method returns.
+ /// The returned array is independently owned. Pinot-owned digests do not
use the scratch buffer; callers can
+ /// clear and reuse it after serialization.
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);
- return 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();
Review Comment:
**[P2] Avoid the additional lossy compression pass for SQL/reducer
accumulators**
For an accumulator with buffered incoming centroids, `centroidCount()` calls
`flush()` at working compression, after which `serialize()` recompresses at
public compression. The previous specialized accumulator serialization merged
pending input directly at public compression. This therefore adds an
unnecessary merge pass per serialized group/metric and can reduce accuracy. The
warning inside `compress()` already describes why this extra pass is
undesirable.
An isolated exact-head probe holding the inputs and kernel fixed confirms
two serialization merge passes through `TDigestUtils.serialize()`, versus one
through `accumulator.serialize()`, across fan-ins 2, 4, 8, 16, 32 and 64. For
16 sources of 2,000 uniform values generated sequentially with `new Random(9)`,
compression 100, the P75 rank error was `0.0023125` with this preflush versus
`0.000125` without it. This is a deterministic example, not a claim that every
distribution regresses or a production-throughput measurement.
Please restrict the historical preflush to the legacy aggregation mode and
preserve direct final compression for SQL/reduction accumulators. Add a
serialize-before-query accuracy case: the main independent reducer-rank test
currently calls `result.compress()` before serialization, so it bypasses this
path.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/PercentileTDigestValueAggregator.java:
##########
@@ -284,52 +270,113 @@ public void add(double value) {
}
@Override
- public void add(double value, int weight) {
+ public void add(double value, double weight) {
if (Double.isNaN(value)) {
throw new IllegalArgumentException("Cannot add NaN to t-digest");
}
- if (weight <= 0) {
+ if (!(weight > 0.0) || !Double.isFinite(weight)) {
throw new IllegalArgumentException("TDigest weight must be positive: "
+ weight);
}
- invalidateCaches();
+ requireMutable();
+ checkTotalWeight(getTotalWeight() + weight);
if (value == Double.NEGATIVE_INFINITY) {
- _negativeInfinityWeight = Math.addExact(_negativeInfinityWeight,
weight);
+ _negativeInfinityWeight += weight;
} else if (value == Double.POSITIVE_INFINITY) {
- _positiveInfinityWeight = Math.addExact(_positiveInfinityWeight,
weight);
+ _positiveInfinityWeight += weight;
} else {
_finiteDigest.add(value, weight);
}
+ mutated();
}
@Override
public void add(TDigest other) {
+ if (other.isEmpty()) {
+ return;
+ }
+ requireMutable();
+ if (!other.hasValidStatistics() && !isEmpty()) {
+ throw corruptedMutation();
+ }
+ if (other.hasValidStatistics()) {
+ 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.
- wrapped.invalidateCaches();
- invalidateCaches();
- if (wrapped._finiteDigest.size() > 0L) {
- _finiteDigest.add(List.of(wrapped._finiteDigest));
+ double negativeInfinityWeight = wrapped._negativeInfinityWeight;
+ double positiveInfinityWeight = wrapped._positiveInfinityWeight;
+ if (!wrapped._finiteDigest.isEmpty()) {
+ _finiteDigest.add(wrapped._finiteDigest);
Review Comment:
**[P1] The native merge can corrupt valid mixed-compression sketches**
Replacing the library batch merge with this native merge exposes the
accumulator's stale-extrema ordering to Star-tree/segment aggregation. In
`addAccumulator()`, `bufferIncomingCentroids()` can flush multiple times before
the destination extrema are updated. Boundary normalization then operates on
the empty destination's `+Infinity/-Infinity` extrema and introduces infinite
centroid means from entirely finite input. The underlying ordering existed in
the old SQL accumulator, but this value-aggregator path previously used the
library and was safe.
Reproduced against exact base `6f87b715` and head `2f808539`, without
reading any intermediate quantiles or centroid views:
```java
ByteBuffer sourceBytes = ByteBuffer.allocate(32 + 16 * 5000);
sourceBytes.putInt(1).putDouble(1).putDouble(5000)
.putDouble(10000).putInt(5000);
for (int i = 1; i <= 5000; i++) {
sourceBytes.putDouble(i == 1 || i == 5000 ? 1 : 2).putDouble(i);
}
byte[] empty = ByteBuffer.allocate(32).putInt(1)
.putDouble(Double.POSITIVE_INFINITY)
.putDouble(Double.NEGATIVE_INFINITY)
.putDouble(100).putInt(0).array();
byte[] singleton = ByteBuffer.allocate(48).putInt(1)
.putDouble(2500).putDouble(2500).putDouble(10000)
.putInt(1).putDouble(1).putDouble(2500).array();
var aggregator = new PercentileTDigestValueAggregator(List.of());
TDigest target = aggregator.getInitialAggregatedValue(empty);
TDigest source = aggregator.deserializeAggregatedValue(sourceBytes.array());
source = aggregator.applyRawValue(source, singleton);
target = aggregator.applyAggregatedValue(target, source);
TDigest restored = aggregator.deserializeAggregatedValue(
aggregator.serializeAggregatedValue(target));
System.out.println(restored.quantile(0.198));
```
Base prints `990.901`; head prints `NaN` and `restored.hasValidStatistics()`
is false. Both inputs are valid legacy payloads with positive integral weights.
Please keep extrema consistent with each consumed batch before boundary
normalization, or use a safe whole-source merge. Simply assigning the final
source bounds before processing a partial prefix also needs care. Add a
regression covering mixed-compression BYTES aggregation followed immediately by
serialization and rereading.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/PercentileTDigestValueAggregator.java:
##########
@@ -152,130 +166,102 @@ private static NonFiniteAwareTDigest
asNonFiniteAware(TDigest value) {
}
private void updateMaxByteSize(TDigest value) {
- long defaultCapacity = getDefaultCentroidCapacity(value.compression());
- long maxCentroids = Math.min(value.size(), defaultCapacity);
- _maxByteSize = Math.max(_maxByteSize, getMaxVerboseByteSize(maxCentroids));
- }
-
- private static int getMaxVerboseByteSize(long centroidCount) {
- long maxCentroids = Math.max(centroidCount, 0L);
- return Math.toIntExact(Math.addExact(VERBOSE_HEADER_SIZE,
- Math.multiplyExact(VERBOSE_CENTROID_SIZE, maxCentroids)));
- }
-
- private static long getDefaultCentroidCapacity(double compression) {
- double normalizedCompression = Math.max(MIN_COMPRESSION, compression);
- int padding = normalizedCompression < 30.0 ?
LOW_COMPRESSION_CAPACITY_PADDING
- : DEFAULT_CENTROID_CAPACITY_PADDING;
- long defaultCapacity = (long) Math.ceil(2.0 * normalizedCompression +
padding);
- long legacyCapacity = Math.addExact(Math.multiplyExact(2L, (long)
Math.ceil(compression)),
- DEFAULT_CENTROID_CAPACITY_PADDING);
- return Math.max(defaultCapacity, legacyCapacity);
+ _maxByteSize = Math.max(_maxByteSize, value.maxSerializedByteSize());
Review Comment:
**[P2] Do not persist the temporary raw-buffer size as the Star-tree output
bound**
`maxSerializedByteSize()` counts buffered raw rows through
`getCentroidCountUpperBound()`, and this maximum never shrinks after
compression. With compression 100 and 1,049 scalar inputs, an exact-head probe
reports a bound of **16,816 bytes**, while the actual serialized payload is
**912 bytes**. The previous registration capped this ordinary raw-input case at
**3,392 bytes**.
This estimate flows through `BaseSingleTreeBuilder.createForwardIndexes()`
into `SingleValueVarByteRawIndexCreator`. For supported index versions 2/3 with
`deriveNumDocsPerChunk=false` and 1,000 documents per chunk, the two writer
buffers grow from roughly **6.8 MB to 33.6 MB per metric**. With derived chunk
sizing and a 1 MiB target, the chunk drops from **308 to 62 documents**. The
default v4 writer's 1 MiB cap avoids this particular buffer-growth impact, so
the concern is specifically the older supported writer configurations.
Please retain safe sizing for oversized historical payloads while deriving
the fresh-state bound from possible compressed output, or collecting actual
serialized sizes before constructing the forward index. Avoid forcing a
compression on every input just to tighten the bound.
--
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]
