ZdravDim commented on code in PR #58455:
URL: https://github.com/apache/spark/pull/58455#discussion_r4335404677
##########
common/variant/src/main/java/org/apache/spark/types/variant/VariantBuilder.java:
##########
@@ -555,18 +583,372 @@ private void appendVariantImpl(byte[] value, byte[]
metadata, int pos) {
int offset = readUnsigned(value, offsetStart + offsetSize * i,
offsetSize);
int elementPos = dataStart + offset;
offsets.add(writePos - start);
- appendVariantImpl(value, metadata, elementPos);
+ appendVariantImpl(value, metadata, elementPos,
needCanonicalization);
}
finishWritingArray(start, offsets);
return null;
});
break;
+ default:
+ if (needCanonicalization) {
+ appendCanonicalizedScalar(value, pos);
+ } else {
+ shallowAppendVariantImpl(value, pos);
+ }
+ break;
+ }
+ }
+
+ // Canonicalize and append a single scalar value: integers re-emitted at the
smallest int width,
+ // integer-valued decimals promoted to the integer encoding, decimal
trailing zeros stripped,
+ // -0.0 mapped to +0.0, and short strings short-encoded -- so e.g. `1.0`,
`1`, and a wide-encoded
+ // `1` all produce byte-equal output. The read-side check
(`isValueCanonical`) mirrors these
+ // rules through the same shared helpers, so the write and read sides cannot
drift.
+ private void appendCanonicalizedScalar(byte[] value, int pos) {
+ switch (VariantUtil.getType(value, pos)) {
+ case LONG:
+ appendLong(VariantUtil.getLong(value, pos));
+ break;
+ case DECIMAL: {
+ BigDecimal bd = VariantUtil.getDecimal(value, pos);
+ if (decimalPromotesToLong(bd)) {
+ appendLong(bd.longValue());
+ } else {
+ // Fractional, or too large for a long: emit as a decimal (negative
scale coerced to 0).
+ appendDecimal(canonicalDecimalForm(bd));
+ }
+ break;
+ }
+ case FLOAT:
+ appendFloat(canonicalizeFloat(VariantUtil.getFloat(value, pos)));
+ break;
+ case DOUBLE:
+ appendDouble(canonicalizeDouble(VariantUtil.getDouble(value, pos)));
+ break;
+ case STRING:
+ appendString(VariantUtil.getString(value, pos));
+ break;
default:
shallowAppendVariantImpl(value, pos);
break;
}
}
+ // Return a canonical Variant.
+ // Two Variants are semantically equal iff their canonical forms are
byte-equal,
+ // so canonicalizing lets the byte-equality machinery (hash aggregate
bucketing,
+ // hash partitioning) group and compare Variants by value rather than by
+ // their incidental physical encoding.
+ //
+ // Numeric equality follows the parquet-format spec's equivalence classes:
+ // - Exact Numeric (every int width and every decimal) unify by value: an
integer-valued
+ // decimal in signed 64-bit range becomes the minimal-width int; other
decimals use the
+ // minimal width with trailing zeros stripped. So 1, 1.0 and 1.00 are
equal.
+ // - float and double are each their own class (only -0.0 -> +0.0 and NaN
-> canonical
+ // NaN), never unified with each other or with Exact Numeric. So the
double 1e0 != 1.
+ //
+ // The metadata dictionary is rebuilt with its keys sorted by unsigned UTF-8
bytes
+ // (VariantUtil.compareKeys, the same order finishWritingObject uses for
object fields)
+ // and unused entries stripped, with field ids remapped to the sorted
positions.
+ public static Variant canonicalize(Variant v) {
+ // Fast path: an already-canonical input is returned unchanged.
isCanonical rejects sub-variants
+ // (pos != 0) -- a view into a parent's shared value/metadata -- so those
fall through to the
+ // rebuild below, which reads the element at v.pos and produces a
standalone canonical Variant.
+ if (isCanonical(v)) {
+ return v;
+ }
+ return doCanonicalize(v);
+ }
+
+ static Variant doCanonicalize(Variant v) {
+ VariantBuilder builder = new VariantBuilder(/* allowDuplicateKeys */
false);
+ builder.buildCanonicalized(v.value, v.metadata, v.pos);
+ return builder.result();
+ }
+
+ private void buildCanonicalized(byte[] value, byte[] metadata, int pos) {
+ int numKeys = getMetadataNumKeys(metadata);
+ // Check that numKeys is not malformed before sizing `seen` from it (a
huge value would OOM).
+ int metaOffsetSize = ((metadata[0] >> 6) & 0x3) + 1;
+ if (1L + ((long) numKeys + 2) * metaOffsetSize > metadata.length) {
+ throw malformedVariant();
+ }
+ boolean[] seen = new boolean[numKeys];
+ ArrayList<byte[]> keys = new ArrayList<>();
+ collectAllObjectKeys(value, metadata, pos, seen, keys);
+ keys.sort((a, b) -> compareKeys(a, b));
+ for (byte[] key : keys) {
+ addKey(new String(key, StandardCharsets.UTF_8));
+ }
+ appendVariantImpl(value, metadata, pos, /* needCanonicalization */ true);
+ }
+
+ private void collectAllObjectKeys(
+ byte[] value, byte[] metadata, int pos, boolean[] seen,
ArrayList<byte[]> keys) {
+ checkIndex(pos, value.length);
+ int basicType = value[pos] & BASIC_TYPE_MASK;
+ switch (basicType) {
+ case OBJECT:
+ handleObject(value, pos, (size, idSize, offsetSize, idStart,
offsetStart, dataStart) -> {
+ for (int i = 0; i < size; ++i) {
+ int id = readUnsigned(value, idStart + idSize * i, idSize);
+ int offset = readUnsigned(value, offsetStart + offsetSize * i,
offsetSize);
+ int elementPos = dataStart + offset;
+ if (id >= seen.length) {
+ throw malformedVariant();
+ }
+ if (!seen[id]) {
+ seen[id] = true;
+ keys.add(getMetadataKeyBytes(metadata, id));
+ }
+ collectAllObjectKeys(value, metadata, elementPos, seen, keys);
+ }
+ return null;
+ });
+ break;
+ case ARRAY:
+ handleArray(value, pos, (size, offsetSize, offsetStart, dataStart) -> {
+ for (int i = 0; i < size; ++i) {
+ int offset = readUnsigned(value, offsetStart + offsetSize * i,
offsetSize);
+ int elementPos = dataStart + offset;
+ collectAllObjectKeys(value, metadata, elementPos, seen, keys);
+ }
+ return null;
+ });
+ break;
+ default:
+ break;
+ }
+ }
+
+ // Return true iff `v` is ALREADY in the exact byte form that
`buildCanonicalized` would produce
+ // -- i.e. calling `canonicalize(v)` is a no-op. A sub-variant (pos != 0) is
never canonical (its
+ // value/metadata are a view into a parent). Intended as a read-side fast
+ // path so already-canonical Variants skip the allocation-heavy rebuild
(dictionary sort +
+ // re-serialize).
+ //
+ // Checked here:
+ // - metadata dictionary: keys sorted + deduped, minimal offset width, no
unused keys;
+ // - objects/arrays: field ids ascending, offsets exact-cumulative,
id/offset widths minimal;
+ // - scalars:
+ // - minimal int width
+ // - decimal integer-promoted, trailing-zero-free, minimal width
+ // - float/double the exact bytes appendFloat/appendDouble emit
+ // - string short-encoded when it fits.
+ public static boolean isCanonical(Variant v) {
+ if (v.pos != 0) {
+ return false;
+ }
+ byte[] value = v.value;
+ byte[] metadata = v.metadata;
+ checkIndex(0, metadata.length);
+ int metaOffsetSize = ((metadata[0] >> 6) & 0x3) + 1;
+ int numKeys = readUnsigned(metadata, 1, metaOffsetSize);
+ // Start of the dictionary string region: header + dict-size field +
(numKeys + 1) offsets.
+ int stringStart = 1 + (numKeys + 2) * metaOffsetSize;
+
+ if (readUnsigned(metadata, 1 + metaOffsetSize, metaOffsetSize) != 0) {
+ return false;
+ }
+
+ if (numKeys > 1) {
+ // Dictionary keys must be strictly ascending in unsigned-UTF-8 order.
+ int prevOffset = readUnsigned(metadata, 1 + metaOffsetSize,
metaOffsetSize);
+ int curOffset = readUnsigned(metadata, 1 + 2 * metaOffsetSize,
metaOffsetSize);
+ if (prevOffset > curOffset) {
+ throw malformedVariant();
+ }
+ checkIndex(stringStart + curOffset - 1, metadata.length);
+ for (int id = 1; id < numKeys; ++id) {
+ int nextOffset = readUnsigned(metadata, 1 + (id + 2) * metaOffsetSize,
metaOffsetSize);
+ if (curOffset > nextOffset) {
+ throw malformedVariant();
+ }
+ checkIndex(stringStart + nextOffset - 1, metadata.length);
+ if (Arrays.compareUnsigned(
+ metadata, stringStart + prevOffset, stringStart + curOffset,
+ metadata, stringStart + curOffset, stringStart + nextOffset) >= 0)
{
+ return false;
+ }
+ prevOffset = curOffset;
+ curOffset = nextOffset;
+ }
+ }
+ int lastOffset = readUnsigned(metadata, 1 + (numKeys + 1) *
metaOffsetSize, metaOffsetSize);
+ // No trailing bytes: canonicalize emits exactly this length (header +
dict-size field +
+ // (numKeys + 1) offsets + the lastOffset string bytes)
+ if (metadata.length != stringStart + lastOffset) {
+ return false;
+ }
+ // numKeys > lastOffset is possible if we have an empty key.
+ long maxSize = Math.max(lastOffset, numKeys);
+ if ((metadata[0] & 0xFF) != (VERSION | ((minIntWidth(maxSize) - 1) << 6)))
{
+ return false;
+ }
+ boolean[] referenced = new boolean[numKeys];
+ if (!isValueCanonical(value, metadata, 0, referenced)) {
+ return false;
+ }
+ // No trailing bytes after the value (canonicalize emits an exact-length
value).
+ if (VariantUtil.valueSize(value, 0) != value.length) {
+ return false;
+ }
+ for (int id = 0; id < numKeys; ++id) {
+ if (!referenced[id]) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ // Value-traversal half of `isCanonical`. Returns false on the first
non-canonical node.
+ private static boolean isValueCanonical(
+ byte[] value, byte[] metadata, int pos, boolean[] referenced) {
+ switch (VariantUtil.getType(value, pos)) {
+ case OBJECT:
+ return handleObject(value, pos,
+ (size, idSize, offsetSize, idStart, offsetStart, dataStart) -> {
+ // Field ids strictly ascending: with a sorted dictionary, canon
emits fields in key
+ // order, and key order == id order. This also rejects duplicate
keys (equal ids).
+ int prevId = -1;
+ int dataSize = 0;
+ for (int i = 0; i < size; ++i) {
+ int id = readUnsigned(value, idStart + idSize * i, idSize);
+ if (id <= prevId || id >= referenced.length) {
+ return false;
+ }
+ prevId = id;
+ referenced[id] = true;
+ // Offsets must be the exact running data size (no gaps or
padding).
+ int offset = readUnsigned(value, offsetStart + offsetSize * i,
offsetSize);
+ if (offset != dataSize) {
+ return false;
+ }
+ int elementPos = dataStart + offset;
+ if (!isValueCanonical(value, metadata, elementPos, referenced)) {
+ return false;
+ }
+ dataSize += VariantUtil.valueSize(value, elementPos);
+ }
+ int lastOffset = readUnsigned(value, offsetStart + offsetSize *
size, offsetSize);
+ if (lastOffset != dataSize) {
+ return false;
+ }
+ // Canonical iff the header byte is exactly what the writer would
emit.
+ int maxId = size == 0 ? 0 : prevId;
+ byte canonicalHeader =
+ objectHeader(size > U8_MAX, minIntWidth(maxId),
minIntWidth(dataSize));
+ return value[pos] == canonicalHeader;
+ });
+ case ARRAY:
+ return handleArray(value, pos, (size, offsetSize, offsetStart,
dataStart) -> {
+ int dataSize = 0;
+ for (int i = 0; i < size; ++i) {
+ int offset = readUnsigned(value, offsetStart + offsetSize * i,
offsetSize);
+ if (offset != dataSize) {
+ return false;
+ }
+ int elementPos = dataStart + offset;
+ if (!isValueCanonical(value, metadata, elementPos, referenced)) {
+ return false;
+ }
+ dataSize += VariantUtil.valueSize(value, elementPos);
+ }
+ int lastOffset = readUnsigned(value, offsetStart + offsetSize *
size, offsetSize);
+ if (lastOffset != dataSize) {
+ return false;
+ }
+ // Canonical iff the header byte is exactly what the writer would
emit.
+ byte canonicalHeader = arrayHeader(size > U8_MAX,
minIntWidth(dataSize));
+ return value[pos] == canonicalHeader;
+ });
+ case LONG:
+ // Must use the smallest int width `appendLong` would pick.
+ return VariantUtil.getTypeInfo(value, pos)
+ == minimalLongTypeInfo(VariantUtil.getLong(value, pos));
+ case DECIMAL: {
+ BigDecimal onDisk = VariantUtil.getDecimalWithOriginalScale(value,
pos);
+ BigDecimal stripped = onDisk.stripTrailingZeros();
+ // A decimal that promotes to a long is stored as the wrong type ->
not canonical.
+ if (decimalPromotesToLong(stripped)) {
+ return false;
+ }
+ BigDecimal canonForm = canonicalDecimalForm(stripped);
+ return onDisk.scale() == canonForm.scale()
+ && VariantUtil.getTypeInfo(value, pos) ==
minimalDecimalTypeInfo(canonForm);
+ }
+ case FLOAT: {
+ // Canonical iff the stored 4 bytes are exactly what `appendFloat`
would write.
+ int rawBits = (int) VariantUtil.readLong(value, pos + 1, 4);
+ float f = Float.intBitsToFloat(rawBits);
+ return rawBits == Float.floatToIntBits(canonicalizeFloat(f));
+ }
+ case DOUBLE: {
+ // Same as FLOAT, over 8 bytes.
+ long rawBits = VariantUtil.readLong(value, pos + 1, 8);
+ double d = Double.longBitsToDouble(rawBits);
+ return rawBits == Double.doubleToLongBits(canonicalizeDouble(d));
+ }
+ case STRING:
+ // A short string is always canonical (its length <=
MAX_SHORT_STR_SIZE by construction). A
+ // LONG_STR is canonical only when its length exceeds the short-string
cap.
+ if ((value[pos] & BASIC_TYPE_MASK) == SHORT_STR) {
Review Comment:
This returns `true` for every short string without checking its bytes, but
the rebuild re-encodes them, so SHORT_STR `[FF]` passes `isCanonical` while
`doCanonicalize` turns it into `0d ef bf bd`.
##########
common/variant/src/main/java/org/apache/spark/types/variant/VariantBuilder.java:
##########
@@ -527,21 +530,46 @@ public void appendVariant(Variant v) {
}
private void appendVariantImpl(byte[] value, byte[] metadata, int pos) {
+ appendVariantImpl(value, metadata, pos, /* needCanonicalization */ false);
+ }
+
+ // Shared re-emit walk for `appendVariant` and `canonicalize`. Object/array
structure is rebuilt
+ // identically in both modes. When `needCanonicalization` is true, scalar
values are
+ // canonicalized as they are re-emitted (see `appendCanonicalizedScalar`);
otherwise they are
+ // copied byte-for-byte, which is the behavior `appendVariant` and its
callers rely on.
+ private void appendVariantImpl(
+ byte[] value, byte[] metadata, int pos, boolean needCanonicalization) {
checkIndex(pos, value.length);
int basicType = value[pos] & BASIC_TYPE_MASK;
switch (basicType) {
case OBJECT:
handleObject(value, pos, (size, idSize, offsetSize, idStart,
offsetStart, dataStart) -> {
ArrayList<FieldEntry> fields = new ArrayList<>(size);
int start = writePos;
+ // Canonicalize lays field data in UTF-8 key order
(finishWritingObject orders the
+ // id/offset lists, not the data); otherwise an object stored
differently -- e.g. the
+ // UTF-16 order older Spark used -- stays non-canonical. A plain
copy keeps stored order.
+ Integer[] order = new Integer[size];
Review Comment:
This `Integer[] order` is now created and filled for every object, even in
plain copy mode, where it is always just `0, 1, 2, ...`. Could we create it
only when canonicalizing?
##########
common/variant/src/main/java/org/apache/spark/types/variant/VariantUtil.java:
##########
@@ -719,4 +731,23 @@ public static String getMetadataKey(byte[] metadata, int
id) {
// explicitly rather than relying on the platform-dependent JVM default
charset.
return new String(metadata, stringStart + offset, nextOffset - offset,
StandardCharsets.UTF_8);
}
+
+ // Like `getMetadataKey`, but returns the key's raw stored bytes instead of
decoding to a
+ // `String`.
+ public static byte[] getMetadataKeyBytes(byte[] metadata, int id) {
Review Comment:
Only the last line differs from `getMetadataKey`, should we create a helper
for it?
##########
common/variant/src/main/java/org/apache/spark/types/variant/VariantBuilder.java:
##########
@@ -555,18 +583,372 @@ private void appendVariantImpl(byte[] value, byte[]
metadata, int pos) {
int offset = readUnsigned(value, offsetStart + offsetSize * i,
offsetSize);
int elementPos = dataStart + offset;
offsets.add(writePos - start);
- appendVariantImpl(value, metadata, elementPos);
+ appendVariantImpl(value, metadata, elementPos,
needCanonicalization);
}
finishWritingArray(start, offsets);
return null;
});
break;
+ default:
+ if (needCanonicalization) {
+ appendCanonicalizedScalar(value, pos);
+ } else {
+ shallowAppendVariantImpl(value, pos);
+ }
+ break;
+ }
+ }
+
+ // Canonicalize and append a single scalar value: integers re-emitted at the
smallest int width,
+ // integer-valued decimals promoted to the integer encoding, decimal
trailing zeros stripped,
+ // -0.0 mapped to +0.0, and short strings short-encoded -- so e.g. `1.0`,
`1`, and a wide-encoded
+ // `1` all produce byte-equal output. The read-side check
(`isValueCanonical`) mirrors these
+ // rules through the same shared helpers, so the write and read sides cannot
drift.
+ private void appendCanonicalizedScalar(byte[] value, int pos) {
+ switch (VariantUtil.getType(value, pos)) {
+ case LONG:
+ appendLong(VariantUtil.getLong(value, pos));
+ break;
+ case DECIMAL: {
+ BigDecimal bd = VariantUtil.getDecimal(value, pos);
+ if (decimalPromotesToLong(bd)) {
+ appendLong(bd.longValue());
+ } else {
+ // Fractional, or too large for a long: emit as a decimal (negative
scale coerced to 0).
+ appendDecimal(canonicalDecimalForm(bd));
+ }
+ break;
+ }
+ case FLOAT:
+ appendFloat(canonicalizeFloat(VariantUtil.getFloat(value, pos)));
+ break;
+ case DOUBLE:
+ appendDouble(canonicalizeDouble(VariantUtil.getDouble(value, pos)));
+ break;
+ case STRING:
+ appendString(VariantUtil.getString(value, pos));
+ break;
default:
shallowAppendVariantImpl(value, pos);
break;
}
}
+ // Return a canonical Variant.
+ // Two Variants are semantically equal iff their canonical forms are
byte-equal,
+ // so canonicalizing lets the byte-equality machinery (hash aggregate
bucketing,
+ // hash partitioning) group and compare Variants by value rather than by
+ // their incidental physical encoding.
+ //
+ // Numeric equality follows the parquet-format spec's equivalence classes:
+ // - Exact Numeric (every int width and every decimal) unify by value: an
integer-valued
+ // decimal in signed 64-bit range becomes the minimal-width int; other
decimals use the
+ // minimal width with trailing zeros stripped. So 1, 1.0 and 1.00 are
equal.
+ // - float and double are each their own class (only -0.0 -> +0.0 and NaN
-> canonical
+ // NaN), never unified with each other or with Exact Numeric. So the
double 1e0 != 1.
+ //
+ // The metadata dictionary is rebuilt with its keys sorted by unsigned UTF-8
bytes
+ // (VariantUtil.compareKeys, the same order finishWritingObject uses for
object fields)
+ // and unused entries stripped, with field ids remapped to the sorted
positions.
+ public static Variant canonicalize(Variant v) {
+ // Fast path: an already-canonical input is returned unchanged.
isCanonical rejects sub-variants
+ // (pos != 0) -- a view into a parent's shared value/metadata -- so those
fall through to the
+ // rebuild below, which reads the element at v.pos and produces a
standalone canonical Variant.
+ if (isCanonical(v)) {
+ return v;
+ }
+ return doCanonicalize(v);
+ }
+
+ static Variant doCanonicalize(Variant v) {
+ VariantBuilder builder = new VariantBuilder(/* allowDuplicateKeys */
false);
+ builder.buildCanonicalized(v.value, v.metadata, v.pos);
+ return builder.result();
+ }
+
+ private void buildCanonicalized(byte[] value, byte[] metadata, int pos) {
+ int numKeys = getMetadataNumKeys(metadata);
+ // Check that numKeys is not malformed before sizing `seen` from it (a
huge value would OOM).
+ int metaOffsetSize = ((metadata[0] >> 6) & 0x3) + 1;
+ if (1L + ((long) numKeys + 2) * metaOffsetSize > metadata.length) {
+ throw malformedVariant();
+ }
+ boolean[] seen = new boolean[numKeys];
+ ArrayList<byte[]> keys = new ArrayList<>();
+ collectAllObjectKeys(value, metadata, pos, seen, keys);
+ keys.sort((a, b) -> compareKeys(a, b));
+ for (byte[] key : keys) {
+ addKey(new String(key, StandardCharsets.UTF_8));
+ }
+ appendVariantImpl(value, metadata, pos, /* needCanonicalization */ true);
+ }
+
+ private void collectAllObjectKeys(
+ byte[] value, byte[] metadata, int pos, boolean[] seen,
ArrayList<byte[]> keys) {
+ checkIndex(pos, value.length);
+ int basicType = value[pos] & BASIC_TYPE_MASK;
+ switch (basicType) {
+ case OBJECT:
+ handleObject(value, pos, (size, idSize, offsetSize, idStart,
offsetStart, dataStart) -> {
+ for (int i = 0; i < size; ++i) {
+ int id = readUnsigned(value, idStart + idSize * i, idSize);
+ int offset = readUnsigned(value, offsetStart + offsetSize * i,
offsetSize);
+ int elementPos = dataStart + offset;
+ if (id >= seen.length) {
+ throw malformedVariant();
+ }
+ if (!seen[id]) {
+ seen[id] = true;
+ keys.add(getMetadataKeyBytes(metadata, id));
+ }
+ collectAllObjectKeys(value, metadata, elementPos, seen, keys);
+ }
+ return null;
+ });
+ break;
+ case ARRAY:
+ handleArray(value, pos, (size, offsetSize, offsetStart, dataStart) -> {
+ for (int i = 0; i < size; ++i) {
+ int offset = readUnsigned(value, offsetStart + offsetSize * i,
offsetSize);
+ int elementPos = dataStart + offset;
+ collectAllObjectKeys(value, metadata, elementPos, seen, keys);
+ }
+ return null;
+ });
+ break;
+ default:
+ break;
+ }
+ }
+
+ // Return true iff `v` is ALREADY in the exact byte form that
`buildCanonicalized` would produce
+ // -- i.e. calling `canonicalize(v)` is a no-op. A sub-variant (pos != 0) is
never canonical (its
+ // value/metadata are a view into a parent). Intended as a read-side fast
+ // path so already-canonical Variants skip the allocation-heavy rebuild
(dictionary sort +
+ // re-serialize).
+ //
+ // Checked here:
+ // - metadata dictionary: keys sorted + deduped, minimal offset width, no
unused keys;
+ // - objects/arrays: field ids ascending, offsets exact-cumulative,
id/offset widths minimal;
+ // - scalars:
+ // - minimal int width
+ // - decimal integer-promoted, trailing-zero-free, minimal width
+ // - float/double the exact bytes appendFloat/appendDouble emit
+ // - string short-encoded when it fits.
+ public static boolean isCanonical(Variant v) {
+ if (v.pos != 0) {
+ return false;
+ }
+ byte[] value = v.value;
+ byte[] metadata = v.metadata;
+ checkIndex(0, metadata.length);
+ int metaOffsetSize = ((metadata[0] >> 6) & 0x3) + 1;
+ int numKeys = readUnsigned(metadata, 1, metaOffsetSize);
+ // Start of the dictionary string region: header + dict-size field +
(numKeys + 1) offsets.
+ int stringStart = 1 + (numKeys + 2) * metaOffsetSize;
+
+ if (readUnsigned(metadata, 1 + metaOffsetSize, metaOffsetSize) != 0) {
+ return false;
+ }
+
+ if (numKeys > 1) {
+ // Dictionary keys must be strictly ascending in unsigned-UTF-8 order.
+ int prevOffset = readUnsigned(metadata, 1 + metaOffsetSize,
metaOffsetSize);
+ int curOffset = readUnsigned(metadata, 1 + 2 * metaOffsetSize,
metaOffsetSize);
+ if (prevOffset > curOffset) {
Review Comment:
This check is always false, since it always checks `if (0 >
curOffset)`because we made sure `prevOffset = 0` at line 750.
##########
common/variant/src/main/java/org/apache/spark/types/variant/VariantBuilder.java:
##########
@@ -555,18 +583,372 @@ private void appendVariantImpl(byte[] value, byte[]
metadata, int pos) {
int offset = readUnsigned(value, offsetStart + offsetSize * i,
offsetSize);
int elementPos = dataStart + offset;
offsets.add(writePos - start);
- appendVariantImpl(value, metadata, elementPos);
+ appendVariantImpl(value, metadata, elementPos,
needCanonicalization);
}
finishWritingArray(start, offsets);
return null;
});
break;
+ default:
+ if (needCanonicalization) {
+ appendCanonicalizedScalar(value, pos);
+ } else {
+ shallowAppendVariantImpl(value, pos);
+ }
+ break;
+ }
+ }
+
+ // Canonicalize and append a single scalar value: integers re-emitted at the
smallest int width,
+ // integer-valued decimals promoted to the integer encoding, decimal
trailing zeros stripped,
+ // -0.0 mapped to +0.0, and short strings short-encoded -- so e.g. `1.0`,
`1`, and a wide-encoded
+ // `1` all produce byte-equal output. The read-side check
(`isValueCanonical`) mirrors these
+ // rules through the same shared helpers, so the write and read sides cannot
drift.
+ private void appendCanonicalizedScalar(byte[] value, int pos) {
+ switch (VariantUtil.getType(value, pos)) {
+ case LONG:
+ appendLong(VariantUtil.getLong(value, pos));
+ break;
+ case DECIMAL: {
+ BigDecimal bd = VariantUtil.getDecimal(value, pos);
+ if (decimalPromotesToLong(bd)) {
+ appendLong(bd.longValue());
+ } else {
+ // Fractional, or too large for a long: emit as a decimal (negative
scale coerced to 0).
+ appendDecimal(canonicalDecimalForm(bd));
+ }
+ break;
+ }
+ case FLOAT:
+ appendFloat(canonicalizeFloat(VariantUtil.getFloat(value, pos)));
+ break;
+ case DOUBLE:
+ appendDouble(canonicalizeDouble(VariantUtil.getDouble(value, pos)));
+ break;
+ case STRING:
+ appendString(VariantUtil.getString(value, pos));
+ break;
default:
shallowAppendVariantImpl(value, pos);
break;
}
}
+ // Return a canonical Variant.
+ // Two Variants are semantically equal iff their canonical forms are
byte-equal,
+ // so canonicalizing lets the byte-equality machinery (hash aggregate
bucketing,
+ // hash partitioning) group and compare Variants by value rather than by
+ // their incidental physical encoding.
+ //
+ // Numeric equality follows the parquet-format spec's equivalence classes:
+ // - Exact Numeric (every int width and every decimal) unify by value: an
integer-valued
+ // decimal in signed 64-bit range becomes the minimal-width int; other
decimals use the
+ // minimal width with trailing zeros stripped. So 1, 1.0 and 1.00 are
equal.
+ // - float and double are each their own class (only -0.0 -> +0.0 and NaN
-> canonical
+ // NaN), never unified with each other or with Exact Numeric. So the
double 1e0 != 1.
+ //
+ // The metadata dictionary is rebuilt with its keys sorted by unsigned UTF-8
bytes
+ // (VariantUtil.compareKeys, the same order finishWritingObject uses for
object fields)
+ // and unused entries stripped, with field ids remapped to the sorted
positions.
+ public static Variant canonicalize(Variant v) {
Review Comment:
Shoud this be next to other public methods?
--
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]
