xiangfu0 commented on code in PR #19795:
URL: https://github.com/apache/pinot/pull/19795#discussion_r4224154641
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TDigestUtils.java:
##########
@@ -207,331 +192,217 @@ public static TDigest deserialize(byte[] bytes) {
return deserialize(ByteBuffer.wrap(bytes));
}
- /// Validates the structure and declared capacities of a serialized merging
digest.
- public static void validateSerialized(byte[] bytes) {
- validateEncodedDigest(ByteBuffer.wrap(bytes));
+ /// Validates serialized lengths and centroid values without allocating
centroid arrays.
+ public static double validateSerialized(byte[] bytes) {
+ return validateSerialized(ByteBuffer.wrap(bytes));
}
- /// Deserializes a finite-valued digest into a merging digest with the
temporary-buffer footprint used by 3.2.
+ /// Validates one legacy digest without changing `input` and returns its
total centroid weight.
///
- /// Unlike the standard verbose decoder, this method does not retain the
substantially larger default 3.3
- /// temporary arrays. It also repairs legacy weighted boundary centroids
before the first 3.3 merge.
- public static TDigest deserializeFiniteWithLegacyBuffer(byte[] bytes) {
- ByteBuffer input = ByteBuffer.wrap(bytes);
- validateEncodedDigest(input.duplicate());
-
- int encoding = input.getInt();
- double min = input.getDouble();
- double max = input.getDouble();
- double compression;
- int centroidCount;
- boolean verbose;
- if (encoding == VERBOSE_ENCODING) {
- compression = input.getDouble();
- centroidCount = input.getInt();
- verbose = true;
- } else if (encoding == SMALL_ENCODING) {
- compression = input.getFloat();
- input.getShort();
- input.getShort();
- centroidCount = input.getShort();
- verbose = false;
- } else {
- throw new IllegalStateException("Invalid format for serialized
histogram");
- }
-
- if (centroidCount == 0) {
- return createMergingDigestWithLegacyBuffer(compression);
- }
- double[] means = new double[centroidCount + 2];
- double[] weights = new double[centroidCount + 2];
- for (int i = 0; i < centroidCount; i++) {
- double weight = verbose ? input.getDouble() : input.getFloat();
- double mean = verbose ? input.getDouble() : input.getFloat();
- if (!(weight > 0.0) || !Double.isFinite(weight)) {
- throw new IllegalArgumentException("Invalid TDigest centroid weight: "
+ weight);
- }
- if (!Double.isFinite(mean)) {
- throw new IllegalArgumentException("Expected a finite TDigest centroid
mean: " + mean);
- }
- weights[i] = weight;
- means[i] = mean;
- }
-
- int normalizedCount = weights[0] != 1.0 || weights[centroidCount - 1] !=
1.0
- ? normalizeBoundaries(means, weights, centroidCount, min, max) :
centroidCount;
- long addCount = 0L;
- for (int i = 0; i < normalizedCount; i++) {
- double weight = weights[i];
- if (weight != Math.rint(weight) || weight >= 0x1p63) {
- return copyIntoLegacyBuffer(
- deserialize(toVerboseBytes(min, max, compression, means, weights,
normalizedCount)), normalizedCount);
- }
- long integralWeight = (long) weight;
- addCount = Math.addExact(addCount, 1L + (integralWeight - 1L) /
Integer.MAX_VALUE);
- }
-
- int minimumMainCapacity = Math.addExact(Math.toIntExact(addCount), 2);
- TDigest digest = createMergingDigestWithLegacyBuffer(compression,
minimumMainCapacity);
- for (int i = 0; i < normalizedCount; i++) {
- long remainingWeight = (long) weights[i];
- while (remainingWeight > 0L) {
- int weight = (int) Math.min(remainingWeight, Integer.MAX_VALUE);
- digest.add(means[i], weight);
- remainingWeight -= weight;
- }
- }
- return digest;
+ /// Trailing bytes remain available to the caller for intermediate values
containing several objects.
+ public static double validateSerialized(ByteBuffer input) {
+ return validateSerialized(input, true);
}
- /// Deserializes a digest and advances `input` past its encoded bytes.
- public static TDigest deserialize(ByteBuffer input) {
- ByteBuffer encoded = input.slice();
- validateEncodedDigest(encoded.duplicate());
-
- int encoding = encoded.getInt();
- double min = encoded.getDouble();
- double max = encoded.getDouble();
- double compression;
- int mainCapacity;
- int bufferCapacity;
- int centroidCount;
- int centroidSize;
- if (encoding == VERBOSE_ENCODING) {
- compression = encoded.getDouble();
- centroidCount = encoded.getInt();
- mainCapacity = getDefaultCentroidCapacity(compression);
- bufferCapacity = 5 * mainCapacity;
- centroidSize = VERBOSE_CENTROID_SIZE;
- } else if (encoding == SMALL_ENCODING) {
- compression = encoded.getFloat();
- mainCapacity = encoded.getShort();
- bufferCapacity = encoded.getShort();
- centroidCount = encoded.getShort();
- centroidSize = SMALL_CENTROID_SIZE;
- } else {
- // MergingDigest.fromBytes() has already produced the canonical
exception for this case.
- throw new IllegalStateException("Invalid format for serialized
histogram");
- }
- int centroidOffset = encoded.position();
- if (centroidCount == 0) {
- return configureScaleFunction(MergingDigest.fromBytes(input));
- }
-
- double firstWeight = readWeight(encoded, centroidOffset, centroidSize);
- double lastWeight = readWeight(encoded, centroidOffset + (centroidCount -
1) * centroidSize, centroidSize);
- int defaultMainCapacity = getDefaultCentroidCapacity(compression);
- boolean needsBoundaryRepair = firstWeight != 1.0 || lastWeight != 1.0;
- boolean exceedsDefaultVerboseCapacity = encoding == VERBOSE_ENCODING &&
centroidCount > defaultMainCapacity;
- if (!needsBoundaryRepair && !exceedsDefaultVerboseCapacity) {
- return configureScaleFunction(MergingDigest.fromBytes(input));
- }
-
- double[] means = new double[centroidCount + 2];
- double[] weights = new double[centroidCount + 2];
- encoded.position(centroidOffset);
- for (int i = 0; i < centroidCount; i++) {
- if (centroidSize == VERBOSE_CENTROID_SIZE) {
- weights[i] = encoded.getDouble();
- means[i] = encoded.getDouble();
- } else {
- weights[i] = encoded.getFloat();
- means[i] = encoded.getFloat();
- }
- }
- int encodedLength = encoded.position();
- int normalizedCount = needsBoundaryRepair
- ? normalizeBoundaries(means, weights, centroidCount, min, max) :
centroidCount;
- byte[] normalizedBytes = toVerboseBytes(min, max, compression, means,
weights, normalizedCount);
- MergingDigest result;
- if (normalizedCount <= defaultMainCapacity) {
- result = MergingDigest.fromBytes(ByteBuffer.wrap(normalizedBytes));
- } else {
- int resolvedMainCapacity = mainCapacity == -1 ? defaultMainCapacity :
mainCapacity;
- int resolvedBufferCapacity = bufferCapacity == -1
- ? DEFAULT_MERGE_BUFFER_MULTIPLIER * resolvedMainCapacity :
bufferCapacity;
- int smallMainCapacity = Math.max(resolvedMainCapacity, normalizedCount);
- long smallBufferCapacity = Math.max(Math.max((long)
resolvedBufferCapacity, smallMainCapacity + 1L),
- 2L * smallMainCapacity + 1L);
- if ((double) (float) compression == compression && normalizedCount <=
Short.MAX_VALUE
- && smallMainCapacity <= Short.MAX_VALUE && smallBufferCapacity <=
Short.MAX_VALUE
- && areFloatExact(means, weights, normalizedCount)) {
- ByteBuffer normalized = ByteBuffer.allocate(SMALL_HEADER_SIZE +
SMALL_CENTROID_SIZE * normalizedCount);
- normalized.putInt(SMALL_ENCODING);
- normalized.putDouble(min);
- normalized.putDouble(max);
- normalized.putFloat((float) compression);
- normalized.putShort((short) smallMainCapacity);
- normalized.putShort((short) smallBufferCapacity);
- normalized.putShort((short) normalizedCount);
- for (int i = 0; i < normalizedCount; i++) {
- normalized.putFloat((float) weights[i]);
- normalized.putFloat((float) means[i]);
- }
- normalized.flip();
- result = MergingDigest.fromBytes(normalized);
- } else {
- result = MergingDigest.fromBytes(
- ByteBuffer.wrap(reduceVerboseCentroids(normalizedBytes,
defaultMainCapacity)));
- }
- }
- input.position(Math.addExact(input.position(), encodedLength));
- return configureScaleFunction(result);
+ private static double validateSerialized(ByteBuffer input, boolean
checkCapacity) {
+ return inspectSerialized(input, checkCapacity).totalWeight();
}
- private static TDigest configureScaleFunction(MergingDigest digest) {
- digest.setScaleFunction(ScaleFunction.K_1);
- return digest;
- }
-
- private static TDigest copyIntoLegacyBuffer(TDigest source, int
centroidCount) {
- TDigest digest = createMergingDigestWithLegacyBuffer(source.compression(),
Math.addExact(centroidCount, 2));
- digest.add(java.util.List.of(source));
- return digest;
+ /// Header and numerical metadata for one legacy byte payload. The centroid
bytes remain owned by the caller.
+ /// Header-only inspection leaves total weight as NaN until centroid
inspection validates the numerical fields.
+ ///
+ /// Legacy writers can emit poisoned numerical state after overflow or
broken fractional boundary repair.
+ /// Such state is retained with degraded quantiles, rather than being passed
to the sorted K1 kernel.
+ public record SerializedTDigestMetadata(int encoding, double min, double
max, double compression,
+ int centroidCount, int mainCapacity, int bufferCapacity, int
centroidOffset, int centroidSize,
+ int encodedLength, double totalWeight, boolean hasNonFiniteMeans,
boolean needsLegacyFallback,
+ boolean unorderedMeans, boolean fractionalWeights, boolean
weightedBoundaries, boolean hasZeroWeightCentroids,
+ double encodedCompression) {
}
- private static boolean areFloatExact(double[] means, double[] weights, int
count) {
- for (int i = 0; i < count; i++) {
- if ((double) (float) weights[i] != weights[i] || (double) (float)
means[i] != means[i]) {
- return false;
- }
- }
- return true;
+ /// Inspects one payload without mutating `input`, scanning the centroid
values once.
+ public static SerializedTDigestMetadata inspectSerialized(ByteBuffer input) {
+ return inspectSerialized(input, true);
}
- private static byte[] toVerboseBytes(double min, double max, double
compression, double[] means, double[] weights,
- int count) {
- ByteBuffer normalized = ByteBuffer.allocate(VERBOSE_HEADER_SIZE +
VERBOSE_CENTROID_SIZE * count);
- normalized.putInt(VERBOSE_ENCODING);
- normalized.putDouble(min);
- normalized.putDouble(max);
- normalized.putDouble(compression);
- normalized.putInt(count);
- for (int i = 0; i < count; i++) {
- normalized.putDouble(weights[i]);
- normalized.putDouble(means[i]);
- }
- return normalized.array();
+ /// Inspects a payload, optionally enforcing the legacy decoder's declared
centroid capacity.
+ public static SerializedTDigestMetadata inspectSerialized(ByteBuffer input,
boolean checkCapacity) {
+ return inspectSerialized(input, readSerializedHeader(input,
checkCapacity), null, null);
}
- private static double readWeight(ByteBuffer encoded, int offset, int
centroidSize) {
- return centroidSize == VERBOSE_CENTROID_SIZE ? encoded.getDouble(offset) :
encoded.getFloat(offset);
+ /// Reads and validates the fixed-size header without scanning or allocating
centroid arrays.
+ public static SerializedTDigestMetadata readSerializedHeader(ByteBuffer
input) {
+ return readSerializedHeader(input, true);
}
- private static void validateEncodedDigest(ByteBuffer input) {
- int encoding = input.getInt();
+ /// Reads a fixed-size header, optionally enforcing the declared centroid
capacity.
+ ///
+ /// NaN extrema mark legacy degraded state; centroid flags and total weight
are computed by the subsequent pass.
+ public static SerializedTDigestMetadata readSerializedHeader(ByteBuffer
input, boolean checkCapacity) {
+ ByteBuffer encoded = input.slice().order(ByteOrder.BIG_ENDIAN);
+ int encoding = encoded.getInt();
if (encoding != VERBOSE_ENCODING && encoding != SMALL_ENCODING) {
throw new IllegalStateException("Invalid format for serialized
histogram");
}
- input.getDouble();
- input.getDouble();
+ double min = encoded.getDouble();
+ double max = encoded.getDouble();
double compression;
+ double encodedCompression;
int mainCapacity;
+ int bufferCapacity = 0;
int centroidCount;
int centroidSize;
if (encoding == VERBOSE_ENCODING) {
- compression = input.getDouble();
- mainCapacity = getDefaultCentroidCapacity(compression);
- centroidCount = input.getInt();
+ encodedCompression = encoded.getDouble();
+ compression = normalizeCompression(encodedCompression);
+ mainCapacity = Math.max(getDefaultCentroidCapacity(compression),
getLegacyDefaultCentroidCapacity(compression));
+ centroidCount = encoded.getInt();
centroidSize = VERBOSE_CENTROID_SIZE;
} else {
- compression = input.getFloat();
- mainCapacity = input.getShort();
- int bufferCapacity = input.getShort();
+ encodedCompression = encoded.getFloat();
+ compression = normalizeCompression(encodedCompression);
+ mainCapacity = encoded.getShort();
+ bufferCapacity = encoded.getShort();
checkSmallArrayCapacity(mainCapacity);
checkSmallArrayCapacity(bufferCapacity);
if (mainCapacity == -1) {
mainCapacity = getDefaultCentroidCapacity(compression);
}
- centroidCount = input.getShort();
+ centroidCount = encoded.getShort();
centroidSize = SMALL_CENTROID_SIZE;
}
if (centroidCount < 0) {
throw new IllegalArgumentException("Invalid negative TDigest centroid
count: " + centroidCount);
}
- if (centroidCount > input.remaining() / centroidSize) {
+ if (centroidCount > encoded.remaining() / centroidSize) {
throw new BufferUnderflowException();
}
- int readableMainCapacity = mainCapacity;
- if (encoding == VERBOSE_ENCODING && compression > 0.0 &&
Double.isFinite(compression)) {
- readableMainCapacity = Math.max(mainCapacity,
getLegacyDefaultCentroidCapacity(compression));
+ if (checkCapacity && centroidCount > mainCapacity) {
+ throw new IllegalArgumentException("TDigest centroid count exceeds
capacity: " + centroidCount);
}
- if (centroidCount > readableMainCapacity) {
- throw new ArrayIndexOutOfBoundsException(
- "Index " + readableMainCapacity + " out of bounds for length " +
readableMainCapacity);
- }
- }
-
- private static void checkSmallArrayCapacity(int capacity) {
- if (capacity < -1) {
- throw new NegativeArraySizeException(Integer.toString(capacity));
+ if (centroidCount > 0 && min > max) {
+ throw new IllegalArgumentException("Invalid TDigest extrema: " + min +
", " + max);
}
+ int centroidOffset = encoded.position();
+ int encodedLength = centroidOffset + centroidCount * centroidSize;
+ return new SerializedTDigestMetadata(encoding, min, max, compression,
centroidCount, mainCapacity,
+ bufferCapacity, centroidOffset, centroidSize, encodedLength,
Double.NaN, false,
+ Double.isNaN(min) || Double.isNaN(max), false, false, false, false,
encodedCompression);
}
- private static int normalizeBoundaries(double[] means, double[] weights, int
count, double min, double max) {
- if (count == 1) {
- double weight = weights[0];
- if (weight <= 1.0) {
- return count;
+ /// Validates centroids in one pass and optionally decodes their means and
weights into reusable arrays.
+ ///
+ /// The header must describe this exact input, and each supplied array must
hold all encoded centroids. Endpoint
+ /// rounding inherited from compact re-encoding or old merge arithmetic is
restored to the double header bounds.
+ public static SerializedTDigestMetadata inspectSerialized(ByteBuffer input,
SerializedTDigestMetadata header,
+ double[] means, double[] weights) {
+ ByteBuffer encoded = input.slice().order(ByteOrder.BIG_ENDIAN);
+ encoded.position(header.centroidOffset());
+ int centroidCount = header.centroidCount();
+ int centroidSize = header.centroidSize();
+ double min = header.min();
+ double max = header.max();
+ double totalWeight = 0.0;
+ double previousMean = Double.NEGATIVE_INFINITY;
+ boolean hasNonFiniteMeans = false;
+ boolean needsLegacyFallback = header.needsLegacyFallback();
+ boolean unorderedMeans = false;
+ boolean fractionalWeights = false;
+ boolean hasZeroWeightCentroids = false;
+ int positiveCentroidCount = 0;
+ double firstPositiveWeight = 0.0;
+ double lastPositiveWeight = 0.0;
+ boolean withinBounds = true;
+ double roundedMin = (double) (float) min;
+ double roundedMax = (double) (float) max;
+ double lowerAdjacent = Math.nextDown(min);
+ double upperAdjacent = Math.nextUp(max);
+ for (int i = 0; i < centroidCount; i++) {
+ double weight = centroidSize == VERBOSE_CENTROID_SIZE ?
encoded.getDouble() : encoded.getFloat();
+ double mean = centroidSize == VERBOSE_CENTROID_SIZE ?
encoded.getDouble() : encoded.getFloat();
+ if (!Double.isFinite(weight)) {
+ throw new IllegalArgumentException("Invalid TDigest centroid weight: "
+ weight);
+ }
+ if (means != null) {
+ means[i] = Math.max(min, Math.min(mean, max));
+ }
+ if (weights != null) {
+ weights[i] = weight;
+ }
+ // A zero-mass centroid has no distribution to merge, even when its
unused mean is NaN or stale.
+ if (weight == 0.0) {
+ hasZeroWeightCentroids = true;
+ continue;
+ }
+ if (weight > 0.0) {
+ if (positiveCentroidCount++ == 0) {
+ firstPositiveWeight = weight;
+ }
+ lastPositiveWeight = weight;
+ }
+ fractionalWeights |= weight != Math.rint(weight);
+ // A previous Pinot boundary repair emitted a negative residual for
singleton weights between one and two.
+ // Old tdunning merges could also overflow finite means to NaN or
infinity. Preserve those stored bytes,
+ // but never normalize or sort them into a plausible distribution.
+ needsLegacyFallback |= weight < 0.0 || Double.isNaN(mean)
Review Comment:
Fixed in
[4818fb8fbc](https://github.com/apache/pinot/commit/4818fb8fbc79d81de17ce0c5bd0070eb97f300df).
SMALL means whose float cast overflows are clamped to the finite double header
extrema before corruption/non-finite classification, including the star-tree
wrapper. Actual 3.2/3.3 compact payloads with positive and negative 1e39
outliers now retain numeric medians, mass and bounds; genuine invalid verbose
infinity remains degraded.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -1194,7 +1243,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:
The SPI now documents this mixed-version limitation in
[4818fb8fbc](https://github.com/apache/pinot/commit/4818fb8fbc79d81de17ce0c5bd0070eb97f300df).
Actual 3.3 with assertions reproduces its singleton-1.5 unit-endpoint failure,
while 3.2 accepts it. There is no mass-preserving representation with two unit
endpoints when total mass is below two; rewriting 1.5 to an integer would
change results. Pinot retains fractional mass and its own reader supports it.
In-tree producers use integral weights; recompression of these exceptional
fractional endpoint states by 3.3 remains unsupported.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TDigestUtils.java:
##########
@@ -86,44 +80,29 @@ 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();
+ return ((PercentileTDigestAccumulator) tDigest).serialize();
}
-
- int defaultCapacity = getDefaultCentroidCapacity(tDigest.compression());
- int legacyCapacity =
getLegacyDefaultCentroidCapacity(tDigest.compression());
- // centroidCount() drains pending values at the digest's internal
(two-level) compression, but does not perform
- // the final, destructive compression used by asBytes(). It therefore
gives us a safe bound for digests
- // deserialized from the compact encoding with an explicitly enlarged
main array while still allowing asBytes()
- // to run once.
- int currentCentroidCount = tDigest.centroidCount();
- long sizeBound = Math.min(Math.max(0L, tDigest.size()),
Math.max(defaultCapacity, legacyCapacity));
- int maxCentroids = Math.max(currentCentroidCount, (int) sizeBound);
+ // centroidCount flushes buffered input. Boundary repair can add at most
two centroids; compression and
+ // total weight are not allocation sizes, so extreme historical settings
need no enormous scratch array.
+ int maxCentroids = Math.addExact(tDigest.centroidCount(), 2);
Review Comment:
Fixed in
[4818fb8fbc](https://github.com/apache/pinot/commit/4818fb8fbc79d81de17ce0c5bd0070eb97f300df).
Generic serialization allocates from maxSerializedByteSize() as well as the
existing legacy bound. SPI writers explicitly advance ByteBuffer.position(); a
non-advancing writer now fails with a clear contract error. Regressions cover a
conforming writer whose actual bound exceeds centroidCount()+2 and an
absolute-put writer that violates the position contract.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/PercentileTDigestAccumulator.java:
##########
@@ -216,17 +243,27 @@ public void add(TDigest other) {
addAccumulator((PercentileTDigestAccumulator) other);
return;
}
- if (other instanceof MergingDigest && other.size() > Integer.MAX_VALUE) {
- // Centroid.count() exposes an int even though MergingDigest stores
double weights internally. Preserve large
- // pre-aggregated weights by ingesting the serialized double-precision
representation instead.
- addSerializedTDigest(TDigestUtils.serialize(other));
+ if (!other.hasValidStatistics()) {
+ materializePendingSerializedTDigest();
+ SerializedTDigestInput input = getSerializedTDigestInput();
+ input.reset(ByteBuffer.wrap(TDigestUtils.serialize(other)), false);
+ addSerializedTDigest(input);
return;
}
-
- Collection<Centroid> centroids = other.centroids();
materializePendingSerializedTDigest();
- for (Centroid centroid : centroids) {
- bufferIncomingCentroid(centroid.mean(), centroid.count());
+ if (_legacyDegraded) {
+ _legacySerializedDirty = true;
+ _totalWeight += other.getTotalWeight();
Review Comment:
Fixed in
[4818fb8fbc](https://github.com/apache/pinot/commit/4818fb8fbc79d81de17ce0c5bd0070eb97f300df).
Degraded merges/adds validate the resulting total is finite before mutating
state, preserving finite historical negative-weight compatibility. Regression
coverage rejects overflow at the merge/add and verifies the pre-existing state
remains usable.
--
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]
