diff --git a/docs/content.zh/docs/sql/reference/data-types.md b/docs/content.zh/docs/sql/reference/data-types.md index 600dec9ef102d1..703185f11de6aa 100644 --- a/docs/content.zh/docs/sql/reference/data-types.md +++ b/docs/content.zh/docs/sql/reference/data-types.md @@ -1695,6 +1695,8 @@ A scalar value can also be cast to a `VARIANT` with `CAST` or `TRY_CAST`. Only a stored with microseconds, and above with nanoseconds, even when the value has no digits below a microsecond. Nanoseconds only cover 1677-09-21 to 2262-04-11, so for a precision above 6 a value outside that range fails the cast, and `TRY_CAST` returns `NULL`. +- A `TIME` holds the milliseconds of one day. Only a source or a function can produce a value + outside that range, but such a value fails the cast, and `TRY_CAST` returns `NULL`. - A `NaN` or infinite `FLOAT` or `DOUBLE` is stored as is, although `PARSE_JSON` rejects them. A `VARIANT` is not limited to JSON, but JSON has no literal for them. So `JSON_STRING` fails on such a variant, and so does writing it to a sink with the `json` or `raw` format, which fails the job. @@ -1707,7 +1709,7 @@ A scalar value can also be cast to a `VARIANT` with `CAST` or `TRY_CAST`. Only a `CAST`. - A constructed type casts element by element when its element, field, or value type casts to `VARIANT`, for example `ARRAY` to `ARRAY`. A `NULL` element stays a SQL `NULL`. A - whole `ARRAY`, `MAP`, or `ROW` does not cast into a single `VARIANT` yet. + whole constructed value can also become a single `VARIANT`, see below. - As a `MAP` key or a `MULTISET` element, two `VARIANT` values match only when their binary encodings match. So a `1` cast from `INT` does not match a `1` cast from `BIGINT` or parsed by `PARSE_JSON('1')`. @@ -1720,7 +1722,40 @@ CAST(NULL AS VARIANT) -- NULL CAST(CAST('NaN' AS DOUBLE) AS VARIANT) -- NaN, stored as a DOUBLE CAST(INTERVAL '2' DAY AS VARIANT) -- fails at validation CAST(ARRAY[1, NULL] AS ARRAY) -- [1, NULL], each element a VARIANT, the NULL stays SQL NULL -CAST(ARRAY[1, 2] AS VARIANT) -- fails at validation, not supported yet +``` + +A whole `ARRAY`, `MAP`, `ROW`, or `STRUCTURED` value can also be cast into a single `VARIANT`. An +`ARRAY` becomes a variant array, and a `MAP`, `ROW`, or `STRUCTURED` value a variant object. Each +leaf is stored by the rules above, so the cast is supported only when every leaf type casts to +`VARIANT`. + +- A `ROW` or `STRUCTURED` value is keyed by its field names. +- A `MAP` needs a character string key, which becomes the object key. A `NULL` key fails the cast. + If a key appears twice, the last value is kept. +- A variant object sorts its keys by their UTF-8 bytes, so neither the field order of a `ROW` nor + the entry order of a `MAP` is kept. +- A `NULL` element, field, or map value becomes a variant null, so an array keeps its length and an + object keeps its keys. A cast element by element differs here: `ARRAY` to `ARRAY` + keeps a SQL `NULL`, while `ARRAY` to `VARIANT` turns it into a variant null. +- A nested `VARIANT` is embedded as is. +- The whole value must fit into the 16 MiB of a `VARIANT`. Any `ARRAY` or `MAP` can exceed it, and so + can a `ROW` with a nested `VARIANT` or with fields whose declared sizes add up to more. The cast + then fails, and `TRY_CAST` returns `NULL` for the whole value. +- Casting the `VARIANT` back to the original type returns the original value, since a cast to `ROW` + matches fields by name. There are two exceptions: + - A SQL `NULL` in a `VARIANT` field, in an `ARRAY`, or as a value of a + `MAP` comes back as a variant null. + - A `MAP` that holds a key twice comes back with only the last value for that key. + +The following examples use `r` for a column of type `ROW` that holds +`('ada', 7)`: + +```sql +CAST(ARRAY[1, NULL] AS VARIANT) -- [1, null], the NULL becomes a variant null +CAST(MAP['b', 2, 'a', 1] AS VARIANT) -- {"a": 1, "b": 2} +CAST(r AS VARIANT) -- {"id": 7, "name": "ada"} +CAST(MAP[1, 'a'] AS VARIANT) -- fails at validation, a MAP key must be a character string +CAST(ARRAY[INTERVAL '1' DAY] AS VARIANT) -- fails at validation ``` A `VARIANT` has no equality and no order, because one value has many valid binary encodings. The @@ -1949,15 +1984,15 @@ COALESCE(TRY_CAST('non-number' AS INT), 0) --- 结果返回数字 0 的 INT 格 | `FLOAT` | Y | N | Y | Y | Y | Y | Y | Y | Y | Y | N | N | N | N | N | N | N | N | N | N | N | Y | N | | `DOUBLE` | Y | N | Y | Y | Y | Y | Y | Y | Y | Y | N | N | N | N | N | N | N | N | N | N | N | Y | N | | `DATE` | Y | N | N | N | N | N | N | N | N | N | Y | N | Y | Y | N | N | N | N | N | N | N | Y | N | -| `TIME` | Y | N | N | N | N | N | N | N | N | N | N | Y | Y | Y | N | N | N | N | N | N | N | Y | N | +| `TIME` | Y | N | N | N | N | N | N | N | N | N | N | Y | Y | Y | N | N | N | N | N | N | N | ! | N | | `TIMESTAMP` | Y | N | N | N | N | N | N | N | N | N | Y | Y | Y | Y | N | N | N | N | N | N | N | ! | N | | `TIMESTAMP_LTZ` | Y | N | N | N | N | N | N | N | N | N | Y | Y | Y | Y | N | N | N | N | N | N | N | ! | N | | `INTERVAL` | Y | N | N | N | N | N | Y⁵ | Y⁶ | N | N | N | N | N | N | Y | N | N | N | N | N | N | N | N | -| `ARRAY` | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | N | !³ | N | N | N | N | N | N | N | +| `ARRAY` | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | N | !³ | N | N | N | N | N | !⁸ | N | | `MULTISET` | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | !³ | N | N | N | N | N | N | -| `MAP` | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | !³ | N | N | N | N | N | -| `ROW` | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | !³ | N | N | N | N | -| `STRUCTURED` | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | !³ | N | N | N | +| `MAP` | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | !³ | N | N | N | !⁸ | N | +| `ROW` | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | !³ | N | N | !⁸ | N | +| `STRUCTURED` | Y | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | !³ | N | !⁸ | N | | `RAW` | Y | ! | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | Y⁴ | N | N | | `VARIANT` | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | ! | N | !³ | N | !³ | !³ | !³ | N | Y | N | | `BITMAP` | Y | Y⁷ | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | N | @@ -1971,6 +2006,7 @@ COALESCE(TRY_CAST('non-number' AS INT), 0) --- 结果返回数字 0 的 INT 格 5. 支持转换，当且仅当用使用 `INTERVAL` 做“月”到“年”的转换。 6. 支持转换，当且仅当用使用 `INTERVAL` 做“天”到“时间”的转换。 7. 仅支持转换到无界的 `VARBINARY`（`BYTES`），因为裁剪或填充会破坏序列化的位图数据。 +8. Supported iff every leaf type casts to `VARIANT` and every `MAP` key is a character string. Always fallible for `ARRAY` and `MAP`. Fallible for `ROW` and `STRUCTURED` iff a field is fallible, a field is a `VARIANT`, or the declared sizes of the fields can exceed the 16 MiB of a `VARIANT`. 请注意：无论是 `CAST` 还是 `TRY_CAST`，当输入为 `NULL` ，输出也为 `NULL`。 diff --git a/docs/content/docs/sql/reference/data-types.md b/docs/content/docs/sql/reference/data-types.md index f3235b83f5b7ae..1ac289a0402784 100644 --- a/docs/content/docs/sql/reference/data-types.md +++ b/docs/content/docs/sql/reference/data-types.md @@ -1740,8 +1740,8 @@ leaf is stored by the rules above, so the cast is supported only when every leaf - A `ROW` or `STRUCTURED` value is keyed by its field names. - A `MAP` needs a character string key, which becomes the object key. A `NULL` key fails the cast. If a key appears twice, the last value is kept. -- A variant object sorts its keys, so neither the field order of a `ROW` nor the entry order of a - `MAP` is kept. +- A variant object sorts its keys by their UTF-8 bytes, so neither the field order of a `ROW` nor + the entry order of a `MAP` is kept. - A `NULL` element, field, or map value becomes a variant null, so an array keeps its length and an object keeps its keys. A cast element by element differs here: `ARRAY` to `ARRAY` keeps a SQL `NULL`, while `ARRAY` to `VARIANT` turns it into a variant null. diff --git a/flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariant.java b/flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariant.java index 254a742ededbeb..2542748620523d 100644 --- a/flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariant.java +++ b/flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariant.java @@ -37,10 +37,14 @@ import static org.apache.flink.types.variant.BinaryVariantUtil.VERSION; import static org.apache.flink.types.variant.BinaryVariantUtil.VERSION_MASK; import static org.apache.flink.types.variant.BinaryVariantUtil.checkIndex; +import static org.apache.flink.types.variant.BinaryVariantUtil.compareMetadataKey; +import static org.apache.flink.types.variant.BinaryVariantUtil.encodeKey; import static org.apache.flink.types.variant.BinaryVariantUtil.getMetadataKey; import static org.apache.flink.types.variant.BinaryVariantUtil.handleArray; import static org.apache.flink.types.variant.BinaryVariantUtil.handleObject; +import static org.apache.flink.types.variant.BinaryVariantUtil.hasCodePointFromE000; import static org.apache.flink.types.variant.BinaryVariantUtil.malformedVariant; +import static org.apache.flink.types.variant.BinaryVariantUtil.metadataKeyEquals; import static org.apache.flink.types.variant.BinaryVariantUtil.microsToInstant; import static org.apache.flink.types.variant.BinaryVariantUtil.nanosToInstant; import static org.apache.flink.types.variant.BinaryVariantUtil.readUnsigned; @@ -337,48 +341,67 @@ private void checkType(Type expected, Type actual) { // Find the field value whose key is equal to `key`. Return null if the key is not found. // It is only legal to call it when `getType()` is `Type.OBJECT`. private BinaryVariant getFieldByKey(String key) { + byte[] keyBytes = encodeKey(key); return handleObject( value, pos, (size, idSize, offsetSize, idStart, offsetStart, dataStart) -> { + int index; // Use linear search for a short list. Switch to binary search when the length // reaches `BINARY_SEARCH_THRESHOLD`. if (size < BINARY_SEARCH_THRESHOLD) { - for (int i = 0; i < size; ++i) { - int id = readUnsigned(value, idStart + idSize * i, idSize); - if (key.equals(getMetadataKey(metadata, id))) { - int offset = - readUnsigned( - value, offsetStart + offsetSize * i, offsetSize); - return new BinaryVariant(value, metadata, dataStart + offset); - } - } + index = findFieldLinear(keyBytes, size, idSize, idStart); } else { - int low = 0; - int high = size - 1; - while (low <= high) { - // Use unsigned right shift to compute the middle of `low` and `high`. - // This is not only a performance optimization, because it can properly - // handle the case where `low + high` overflows int. - int mid = (low + high) >>> 1; - int id = readUnsigned(value, idStart + idSize * mid, idSize); - int cmp = getMetadataKey(metadata, id).compareTo(key); - if (cmp < 0) { - low = mid + 1; - } else if (cmp > 0) { - high = mid - 1; - } else { - int offset = - readUnsigned( - value, offsetStart + offsetSize * mid, offsetSize); - return new BinaryVariant(value, metadata, dataStart + offset); - } + index = findFieldBinary(keyBytes, size, idSize, idStart); + // Objects that earlier Flink versions wrote are sorted in UTF-16 order + // instead. Only a miss for a key with a code point from U+E000 up can be + // wrong there, see `hasCodePointFromE000`. Such a miss scans all fields, + // because an object does not record which order it uses. + if (index < 0 && hasCodePointFromE000(keyBytes)) { + index = findFieldLinear(keyBytes, size, idSize, idStart); } } - return null; + if (index < 0) { + return null; + } + int offset = readUnsigned(value, offsetStart + offsetSize * index, offsetSize); + return new BinaryVariant(value, metadata, dataStart + offset); }); } + // Return the index of the field whose key has the UTF-8 bytes `key`, or -1 if there is none. + private int findFieldLinear(byte[] key, int size, int idSize, int idStart) { + for (int i = 0; i < size; ++i) { + int id = readUnsigned(value, idStart + idSize * i, idSize); + if (metadataKeyEquals(metadata, id, key)) { + return i; + } + } + return -1; + } + + // Like `findFieldLinear`, for fields sorted by the unsigned UTF-8 bytes of their keys. + private int findFieldBinary(byte[] key, int size, int idSize, int idStart) { + int low = 0; + int high = size - 1; + while (low <= high) { + // Use unsigned right shift to compute the middle of `low` and `high`. This is not only + // a performance optimization, because it can properly handle the case where + // `low + high` overflows int. + int mid = (low + high) >>> 1; + int id = readUnsigned(value, idStart + idSize * mid, idSize); + int cmp = compareMetadataKey(metadata, id, key); + if (cmp < 0) { + low = mid + 1; + } else if (cmp > 0) { + high = mid - 1; + } else { + return mid; + } + } + return -1; + } + // Get the array element at the `index` slot. Return null if `index` is out of the bound of // `[0, arraySize())`. // It is only legal to call it when `getType()` is `Type.ARRAY`. diff --git a/flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariantBuilder.java b/flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariantBuilder.java index 4de45e319fc051..9a96c283e60c32 100644 --- a/flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariantBuilder.java +++ b/flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariantBuilder.java @@ -186,9 +186,7 @@ public VariantObjectBuilder(boolean allowDuplicateKeys) { @Override public VariantObjectBuilder add(String key, Variant value) { - int id = builder.addKey(key); - entries.add( - new BinaryVariantInternalBuilder.FieldEntry(key, id, builder.getWritePos())); + builder.addField(entries, 0, key); builder.appendVariant((BinaryVariant) value); return this; } diff --git a/flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariantInternalBuilder.java b/flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariantInternalBuilder.java index 3ed2e634412152..e922a2bdec83ed 100644 --- a/flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariantInternalBuilder.java +++ b/flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariantInternalBuilder.java @@ -33,7 +33,6 @@ import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.Arrays; -import java.util.Collections; import java.util.Comparator; import java.util.HashMap; import java.util.UUID; @@ -74,6 +73,7 @@ import static org.apache.flink.types.variant.BinaryVariantUtil.VERSION; import static org.apache.flink.types.variant.BinaryVariantUtil.arrayHeader; import static org.apache.flink.types.variant.BinaryVariantUtil.checkIndex; +import static org.apache.flink.types.variant.BinaryVariantUtil.encodeKey; import static org.apache.flink.types.variant.BinaryVariantUtil.fitsVariantDecimal; import static org.apache.flink.types.variant.BinaryVariantUtil.getMetadataKey; import static org.apache.flink.types.variant.BinaryVariantUtil.handleArray; @@ -477,7 +477,7 @@ public int addKey(String key) { } else { id = dictionaryKeys.size(); dictionary.put(key, id); - dictionaryKeys.add(key.getBytes(StandardCharsets.UTF_8)); + dictionaryKeys.add(encodeKey(key)); } return id; } @@ -491,7 +491,7 @@ public int getWritePos() { // Add a field to the object that starts at `start`. Call it right before writing the value, // since the field records the current write position as its offset. public void addField(ArrayList fields, int start, String key) { - fields.add(new FieldEntry(key, addKey(key), writePos - start)); + fields.add(new FieldEntry(addKey(key), writePos - start)); } // Finish writing a variant object after all of its fields have already been written. The @@ -504,20 +504,21 @@ public void addField(ArrayList fields, int start, String key) { // record the offset of the field. The offset is computed as `getWritePos() - start`. // 3. The caller calls `finishWritingObject` to finish writing a variant object. // - // This function is responsible to sort the fields by key. If there are duplicate field keys: + // This function is responsible to sort the fields by the unsigned UTF-8 bytes of their keys, + // as the variant spec requires. Fields whose keys have the same bytes are duplicates: // - when `allowDuplicateKeys` is true, the field with the greatest offset value (the last // appended one) is kept. // - otherwise, throw an exception. public void finishWritingObject(int start, ArrayList fields) { int size = fields.size(); - Collections.sort(fields); + fields.sort(keyOrder); int maxId = size == 0 ? 0 : fields.get(0).id; if (allowDuplicateKeys) { int distinctPos = 0; // Maintain a list of distinct keys in-place. for (int i = 1; i < size; ++i) { maxId = Math.max(maxId, fields.get(i).id); - if (fields.get(i).id == fields.get(i - 1).id) { + if (keyOrder.compare(fields.get(i - 1), fields.get(i)) == 0) { // Found a duplicate key. Keep the field with a greater offset. if (fields.get(distinctPos).offset < fields.get(i).offset) { fields.set( @@ -553,13 +554,12 @@ public void finishWritingObject(int start, ArrayList fields) { } writePos = start + currentOffset; // Change back to the sort order by field keys to meet the variant spec. - Collections.sort(fields); + fields.sort(keyOrder); } } else { for (int i = 1; i < size; ++i) { maxId = Math.max(maxId, fields.get(i).id); - String key = fields.get(i).key; - if (key.equals(fields.get(i - 1).key)) { + if (keyOrder.compare(fields.get(i - 1), fields.get(i)) == 0) { throw VARIANT_DUPLICATE_KEY_EXCEPTION; } } @@ -697,24 +697,17 @@ private void checkCapacity(int additional) { // Temporarily store the information of a field. We need to collect all fields in an JSON // object, // sort them by their keys, and build the variant object in sorted order. - public static final class FieldEntry implements Comparable { - final String key; + public static final class FieldEntry { final int id; final int offset; - public FieldEntry(String key, int id, int offset) { - this.key = key; + FieldEntry(int id, int offset) { this.id = id; this.offset = offset; } FieldEntry withNewOffset(int newOffset) { - return new FieldEntry(key, id, newOffset); - } - - @Override - public int compareTo(FieldEntry other) { - return key.compareTo(other.key); + return new FieldEntry(id, newOffset); } } @@ -834,5 +827,9 @@ private boolean tryParseDecimal(String input) { private final HashMap dictionary = new HashMap<>(); // Store all keys in `dictionary` in the order of id. private final ArrayList dictionaryKeys = new ArrayList<>(); + // Order fields by the stored UTF-8 bytes of their keys. Keys that differ as Strings but are + // stored as the same bytes, such as an unpaired surrogate and '?', compare as equal. + private final Comparator keyOrder = + (a, b) -> Arrays.compareUnsigned(dictionaryKeys.get(a.id), dictionaryKeys.get(b.id)); private final boolean allowDuplicateKeys; } diff --git a/flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariantUtil.java b/flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariantUtil.java index 546b5d4bcdde32..23cdf8c6779620 100644 --- a/flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariantUtil.java +++ b/flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariantUtil.java @@ -86,7 +86,7 @@ public class BinaryVariantUtil { * Object value. The content contains a size, a list of field ids, a list of field offsets, and * the actual field data. The length of the id list is {@code size}, while the length of the * offset list is {@code size + 1}, where the last offset represent the total size of the field - * data. The fields in an object must be sorted by the field name in alphabetical order. + * data. The fields in an object must be sorted by the unsigned UTF-8 bytes of their names. * Duplicate field names in one object are not allowed. * *

We use 5 bits in the type info to specify the integer type of the object header: it should @@ -850,6 +850,58 @@ public static T handleArray(byte[] value, int pos, ArrayHandler handler) * considered a malformed variant because it is read from the corresponding variant value. */ public static String getMetadataKey(byte[] metadata, int id) { + long range = metadataKeyRange(metadata, id); + return new String(metadata, keyStart(range), keyLength(range), StandardCharsets.UTF_8); + } + + /** Encodes an object key the way the variant metadata stores it. */ + static byte[] encodeKey(String key) { + return key.getBytes(StandardCharsets.UTF_8); + } + + /** Whether the key at {@code id} in the variant metadata is the UTF-8 encoded {@code key}. */ + static boolean metadataKeyEquals(byte[] metadata, int id, byte[] key) { + long range = metadataKeyRange(metadata, id); + int start = keyStart(range); + int length = keyLength(range); + return length == key.length + && Arrays.equals(metadata, start, start + length, key, 0, length); + } + + /** + * Compares the key at {@code id} in the variant metadata with the UTF-8 encoded {@code key} by + * their unsigned bytes, the order of object fields in the variant spec. + */ + static int compareMetadataKey(byte[] metadata, int id, byte[] key) { + long range = metadataKeyRange(metadata, id); + int start = keyStart(range); + return Arrays.compareUnsigned( + metadata, start, start + keyLength(range), key, 0, key.length); + } + + /** + * Whether the UTF-8 encoded {@code key} has a code point from U+E000 up, which encodes to a + * lead byte of 0xEE or above. UTF-16 order differs from UTF-8 byte order only between a code + * point from U+E000 to U+FFFF and one from U+10000 up, whose UTF-16 surrogates sort below + * U+E000. So a binary search for a key without such a code point finds the same field in either + * order, and a miss is final. + */ + static boolean hasCodePointFromE000(byte[] key) { + for (byte b : key) { + if ((b & 0xFF) >= 0xEE) { + return true; + } + } + return false; + } + + /** + * Locates the key at {@code id} in the variant metadata. The range is packed into a long so + * that a lookup allocates nothing. + * + * @return the start index of the key in the high 32 bits and its length in the low 32 bits + */ + private static long metadataKeyRange(byte[] metadata, int id) { checkIndex(0, metadata.length); // Extracts the highest 2 bits in the metadata header to determine the integer size of the // offset list. @@ -867,8 +919,15 @@ public static String getMetadataKey(byte[] metadata, int id) { throw malformedVariant(); } checkIndex(stringStart + nextOffset - 1, metadata.length); - return new String( - metadata, stringStart + offset, nextOffset - offset, StandardCharsets.UTF_8); + return ((long) (stringStart + offset) << 32) | (nextOffset - offset); + } + + private static int keyStart(long range) { + return (int) (range >>> 32); + } + + private static int keyLength(long range) { + return (int) range; } /** diff --git a/flink-core/src/test/java/org/apache/flink/types/variant/BinaryVariantInternalBuilderTest.java b/flink-core/src/test/java/org/apache/flink/types/variant/BinaryVariantInternalBuilderTest.java index 2043381dc766f2..a6b5d153a6486e 100644 --- a/flink-core/src/test/java/org/apache/flink/types/variant/BinaryVariantInternalBuilderTest.java +++ b/flink-core/src/test/java/org/apache/flink/types/variant/BinaryVariantInternalBuilderTest.java @@ -219,7 +219,7 @@ void testContainerOfExactlyTheSizeLimitFails() { final ArrayList fields = new ArrayList<>( Collections.singletonList( - new BinaryVariantInternalBuilder.FieldEntry("a", id, 0))); + new BinaryVariantInternalBuilder.FieldEntry(id, 0))); assertThatThrownBy(() -> objectBuilder.finishWritingObject(0, fields)) .isSameAs(BinaryVariantInternalBuilder.VARIANT_SIZE_LIMIT_EXCEPTION); } diff --git a/flink-core/src/test/java/org/apache/flink/types/variant/BinaryVariantTest.java b/flink-core/src/test/java/org/apache/flink/types/variant/BinaryVariantTest.java index e3725c1d202f55..8729209bbf4295 100644 --- a/flink-core/src/test/java/org/apache/flink/types/variant/BinaryVariantTest.java +++ b/flink-core/src/test/java/org/apache/flink/types/variant/BinaryVariantTest.java @@ -23,7 +23,9 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; import org.junit.jupiter.params.provider.CsvSource; +import org.junit.jupiter.params.provider.MethodSource; import org.junit.jupiter.params.provider.ValueSource; import java.math.BigDecimal; @@ -35,13 +37,22 @@ import java.time.ZoneOffset; import java.time.temporal.ChronoUnit; import java.util.Collections; +import java.util.List; import java.util.UUID; +import java.util.stream.Stream; +import static org.apache.flink.types.variant.BinaryVariantUtil.BINARY_SEARCH_THRESHOLD; +import static org.apache.flink.types.variant.BinaryVariantUtil.handleObject; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; class BinaryVariantTest { + // U+E000 is EE 80 80 in UTF-8 and U+1F600 is F0 9F 98 80, so UTF-8 byte order puts U+E000 + // first. UTF-16 order compares E000 with the surrogate D83D and puts U+1F600 first. + private static final String BMP_KEY = "\uE000"; + private static final String SUPPLEMENTARY_KEY = "\uD83D\uDE00"; + private BinaryVariantBuilder builder; @BeforeEach @@ -315,28 +326,52 @@ void testObjectVariant() { } } - @Test - void testDuplicatedKeyObjectVariant() { + @ParameterizedTest(name = "{0}") + @MethodSource("duplicateKeys") + void testDuplicatedKeyObjectVariant( + final String description, + final String key, + final String duplicate, + final List fieldNames) { assertThatThrownBy( () -> builder.object(false) - .add("k", builder.of((byte) 10)) - .add("k", builder.of("hello")) + .add(key, builder.of((byte) 10)) + .add(duplicate, builder.of("hello")) .build()) .isInstanceOf(RuntimeException.class) .hasMessage("VARIANT_DUPLICATE_KEY"); Variant variant = builder.object(true) - .add("k", builder.of((byte) 10)) - .add("k", builder.of("hello")) + .add(key, builder.of((byte) 10)) + .add(duplicate, builder.of("hello")) .add("k1", builder.of(10)) .build(); - assertThat(variant.getField("k").getString()).isEqualTo("hello"); + assertThat(variant.getFieldNames()).containsExactlyElementsOf(fieldNames); + assertThat(variant.getField(key).getString()).isEqualTo("hello"); assertThat(variant.getField("k1").getInt()).isEqualTo(10); } + private static Stream duplicateKeys() { + return Stream.of( + Arguments.of("same key", "k", "k", List.of("k", "k1")), + // UTF-8 has no unpaired surrogates, so the builder stores both keys as "k?". + Arguments.of("same UTF-8 bytes", "k?", "k\uD800", List.of("k1", "k?"))); + } + + @Test + void testObjectFieldsAreSortedByUtf8Bytes() { + final Variant object = + builder.object() + .add(SUPPLEMENTARY_KEY, builder.of(1)) + .add(BMP_KEY, builder.of(2)) + .build(); + + assertThat(object.getFieldNames()).containsExactly(BMP_KEY, SUPPLEMENTARY_KEY); + } + @Test void testToJsonScalar() { Instant instant = Instant.EPOCH; @@ -488,6 +523,69 @@ void testNonAsciiStringsAndFieldNames() { + "\"}}"); } + @ParameterizedTest(name = "{0} fields") + @ValueSource(ints = {1, BINARY_SEARCH_THRESHOLD}) + void testGetFieldWithUnpairedSurrogate(final int size) { + // UTF-8 has no unpaired surrogates. The builder writes one as '?', and a lookup encodes its + // key the same way. + final Variant object = + objectWithFillerFields(builder, size - 1).add("a\uD800", builder.of(100)).build(); + + assertThat(object.getFieldNames()).contains("a?"); + assertThat(object.getField("a\uD800").getInt()).isEqualTo(100); + assertThat(object.getField("a?").getInt()).isEqualTo(100); + } + + @ParameterizedTest(name = "{0}") + @MethodSource("objectsInBothKeyOrders") + void testGetFieldWithBinarySearch(final String keyOrder, final Variant object) { + assertThat(object.getField("é").getInt()).isEqualTo(100); + assertThat(object.getField("中").getInt()).isEqualTo(200); + assertThat(object.getField(BMP_KEY).getInt()).isEqualTo(300); + assertThat(object.getField(SUPPLEMENTARY_KEY).getInt()).isEqualTo(400); + assertThat(object.getField("\uE001")).isNull(); + } + + private static Stream objectsInBothKeyOrders() { + final BinaryVariantBuilder builder = new BinaryVariantBuilder(); + final BinaryVariant object = + (BinaryVariant) + objectWithFillerFields(builder, BINARY_SEARCH_THRESHOLD - 4) + .add("é", builder.of(100)) + .add("中", builder.of(200)) + .add(BMP_KEY, builder.of(300)) + .add(SUPPLEMENTARY_KEY, builder.of(400)) + .build(); + // The other keys sort before the last two in both orders, so swapping the last two fields + // turns UTF-8 byte order into UTF-16 order. + return Stream.of( + Arguments.of("UTF-8 byte order", object), + Arguments.of("UTF-16 order", swapLastTwoFields(object))); + } + + @Test + void testGetFieldWithEmptyKey() { + final Variant object = + builder.object().add("", builder.of(1)).add("b", builder.of(2)).build(); + + assertThat(object.getField("").getInt()).isEqualTo(1); + assertThat(object.getField("b").getInt()).isEqualTo(2); + } + + @Test + void testGetFieldWithMalformedUtf8Key() { + final BinaryVariant object = + (BinaryVariant) builder.object().add("a", builder.of(1)).build(); + // The key is the last metadata byte. 0xFF never occurs in UTF-8, so no lookup key encodes + // to it, even though getFieldNames decodes it to U+FFFD. + final byte[] metadata = object.getMetadata().clone(); + metadata[metadata.length - 1] = (byte) 0xFF; + final Variant malformed = new BinaryVariant(object.getValue(), metadata); + + assertThat(malformed.getFieldNames()).containsExactly("�"); + assertThat(malformed.getField("�")).isNull(); + } + @Test void testVariantException() { assertThatThrownBy(() -> new BinaryVariant(new byte[0], new byte[0])) @@ -555,4 +653,37 @@ void testJavaSerialization() throws Exception { assertThat(((BinaryVariant) subVariant).getPos()).isGreaterThan(0); assertThat(CommonTestUtils.createCopySerializable(subVariant)).isEqualTo(subVariant); } + + /** Starts an object with {@code count} fields named k00, k01, and so on. */ + private static BinaryVariantBuilder.VariantObjectBuilder objectWithFillerFields( + final BinaryVariantBuilder builder, final int count) { + final BinaryVariantBuilder.VariantObjectBuilder object = builder.object(); + for (int i = 0; i < count; i++) { + object.add(String.format("k%02d", i), builder.of(i)); + } + return object; + } + + /** Swaps the ids and offsets of the last two fields, which keeps each value with its key. */ + private static BinaryVariant swapLastTwoFields(final BinaryVariant object) { + final byte[] value = object.getValue().clone(); + handleObject( + value, + 0, + (size, idSize, offsetSize, idStart, offsetStart, dataStart) -> { + swapBytes(value, idStart + idSize * (size - 2), idSize); + swapBytes(value, offsetStart + offsetSize * (size - 2), offsetSize); + return null; + }); + return new BinaryVariant(value, object.getMetadata()); + } + + /** Swaps the {@code length} bytes at {@code start} with the {@code length} bytes after them. */ + private static void swapBytes(final byte[] bytes, final int start, final int length) { + for (int i = start; i < start + length; i++) { + final byte b = bytes[i]; + bytes[i] = bytes[i + length]; + bytes[i + length] = b; + } + } }