aleksandr-chernousov-db commented on code in PR #58455:
URL: https://github.com/apache/spark/pull/58455#discussion_r4006279099
##########
common/variant/src/main/java/org/apache/spark/types/variant/VariantBuilder.java:
##########
@@ -532,18 +541,308 @@ 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, needNormalization);
}
finishWritingArray(start, offsets);
return null;
});
break;
+ default:
+ if (needNormalization) {
+ 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 scalar normalization rules that
the read-side check
+ // (`isValueCanonical`) must mirror are factored into shared helpers so the
two 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.
+ //
+ // The metadata dictionary is rebuilt with its keys sorted by (the same order
+ // finishWritingObject already uses for object fields, so the two stay
+ // consistent) and unused entries stripped, with field ids remapped to the
sorted positions.
+ public static Variant canonicalize(Variant v) {
+ // Fast path: a top-level (pos == 0) input that is already canonical is
returned unchanged. A
+ // sub-variant (pos != 0) is a view into a parent's shared value/metadata,
so it always takes
+ // the slow path, which reads the element at v.pos and rebuilds a
standalone canonical Variant.
+ if (v.pos == 0 && isCanonical(v.value, v.metadata)) {
+ return 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) {
+ ArrayList<String> keys = new ArrayList<>();
+ collectAllObjectKeys(value, metadata, pos, keys);
+ keys.sort((a, b) -> compareKeys(encodeKey(a), encodeKey(b)));
Review Comment:
Now storing encoded keys (actually, just don't decode them in the first
place)
--
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]
