xiangfu0 commented on code in PR #19795:
URL: https://github.com/apache/pinot/pull/19795#discussion_r4225309998
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -1194,7 +1292,7 @@ private static int normalizeBoundaries(double[] means,
double[] weights, int cou
}
if (count == 1) {
double weight = weights[0];
- if (weight <= 1.0) {
+ if (weight < 2.0) {
Review Comment:
Addressed in 14657b2cfb. Fresh legacy serialization now rejects sub-unit
endpoint mass and fractional singleton mass below two instead of changing its
mass or writing a payload that fails the old reader's assertions. In-memory
fractional quantiles remain supported, and unchanged historical bytes survive
views, compression, copying, and returned-array mutation. The actual 3.3 reader
with -ea reproduces the 1.5 singleton restriction.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -1118,12 +1207,65 @@ private void bufferIncomingCentroids(double[] means,
double[] weights, int count
private void materializePendingSerializedTDigest() {
if (_pendingSerializedTDigest != null) {
byte[] bytes = _pendingSerializedTDigest;
+ SerializedTDigestMetadata metadata = _pendingSerializedMetadata;
_pendingSerializedTDigest = null;
- _pendingSerializedTotalWeight = Double.NaN;
- mergeSerializedTDigest(bytes);
+ _pendingSerializedMetadata = null;
+ SerializedTDigestInput input = getSerializedTDigestInput();
+ input.reset(bytes, metadata);
+ input._retainedBytes = bytes;
+ if (_legacyDegraded) {
+ mergeLegacyDegradedInput(input, true);
+ } else {
+ input.decode();
+ mergeSerializedTDigest(input, false);
+ }
+ }
+ }
+
+ private void mergeLegacyDegradedInput(SerializedTDigestInput input, boolean
retainOriginal) {
+ double totalWeight = checkedLegacyTotalWeight(_totalWeight +
_incomingWeight + _numRawValues + input._totalWeight);
+ if (retainOriginal) {
+ _legacySerializedBytes = input.retainBytes();
+ } else {
+ _legacySerializedDirty = true;
+ }
+ markLegacyDegraded(input);
+ for (int i = 0; i < _numRawValues; i++) {
+ _min = Math.min(_min, _rawValues[i]);
+ _max = Math.max(_max, _rawValues[i]);
+ }
+ _totalWeight = totalWeight;
+ _hasFractionalWeights |= input._metadata.fractionalWeights();
+ _incomingWeight = 0.0;
+ _numIncomingCentroids = 0;
+ _numRawValues = 0;
+ _min = Math.min(_min, input._min);
+ _max = Math.max(_max, input._max);
+ _hasSerializedInput = true;
+ }
+
+ private void markLegacyDegraded(SerializedTDigestInput input) {
Review Comment:
Addressed in 14657b2cfb. Added a real same-group merge-rollup regression: a
healthy sketch plus a corrupt stored sketch raises the precise corruption error
and leaves no persisted output. RAW ordering is pinned using Double.compare:
NaN sorts after finite values and positive infinity, with pairwise
antisymmetry; the ordering is documented on SerializedTDigest.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -240,167 +286,134 @@ public void add(List<? extends TDigest> others) {
}
private void addAccumulator(PercentileTDigestAccumulator other) {
+ if (other == this) {
+ addSerializedTDigest(serialize());
+ return;
+ }
if (other._pendingSerializedTDigest != null) {
- addSerializedTDigest(other._pendingSerializedTDigest);
+ // Materialize before resetting the reusable reader: materialization
uses that same reader.
+ materializePendingSerializedTDigest();
+ SerializedTDigestInput input = getSerializedTDigestInput();
+ input.reset(other._pendingSerializedTDigest,
other._pendingSerializedMetadata);
+ input._retainedBytes = other._pendingSerializedTDigest;
+ addSerializedTDigest(input);
+ return;
+ }
+ if (other._legacyDegraded) {
+ addSerializedTDigest(other.serialize());
return;
}
- other.compress();
+ // Legacy merge inputs use public compression. Apply it to a snapshot so
reading a source leaves its
+ // distribution unchanged, and copy before flushing to preserve the legacy
single-pass raw merge.
+ other = other.publiclyCompressedMergeInput();
materializePendingSerializedTDigest();
+ if (_legacyDegraded) {
+ double totalWeight = checkedLegacyTotalWeight(_totalWeight +
other._totalWeight);
+ _legacySerializedDirty = true;
+ _totalWeight = totalWeight;
+ _hasFractionalWeights |= other._hasFractionalWeights;
+ _min = Math.min(_min, other._min);
+ _max = Math.max(_max, other._max);
+ return;
+ }
+ checkTotalWeight(_totalWeight + _incomingWeight + _numRawValues +
other._totalWeight);
+ _hasFractionalWeights |= other._hasFractionalWeights;
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;
- }
+ private PercentileTDigestAccumulator publiclyCompressedMergeInput() {
Review Comment:
Addressed in 14657b2cfb. Native merges now compress the exclusively owned
source in place and ingest its primitive arrays. The SPI documents consuming
merge ownership, and wrapper caches are invalidated when that source is
compressed. Only self-merges snapshot arrays; empty and fractional self-merges
have regression coverage. All 270 accuracy cases pass with unchanged reference
envelopes.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -638,10 +702,44 @@ public int byteSize() {
return serialize().length;
Review Comment:
Addressed in 14657b2cfb. byteSize() now retains one serialized result for
the following asBytes()/asSmallBytes() write. Mutation and compression
invalidate it, and writing consumes it; the regression verifies the cache
lifecycle. maxSerializedByteSize() remains an allocation bound without
serializing.
--
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]
