From 2e3589c541041642b4d16c620c1eb93b0b61df6a Mon Sep 17 00:00:00 2001 From: Ramin Gharib Date: Thu, 22 Oct 2026 18:19:31 +0200 Subject: [PATCH 1/3] [FLINK-41110][core] Compare UTF-8 bytes to find VARIANT object fields Variant.getField decoded each key it compared into a new String. Finding a field in a 13-key object could create 13 throwaway strings. For objects below BINARY_SEARCH_THRESHOLD fields, getField now encodes the wanted key once and compares it with the UTF-8 bytes of the metadata dictionary. Larger objects still decode each key and compare Strings, because the binary search depends on the order in which the builder sorts the keys. The wanted key is encoded the same way the builder encodes the keys it writes. So below the threshold, getField("a\uD800") now finds the field stored as "a?". Larger objects still miss such a key. Metadata keys that are not valid UTF-8 no longer match the U+FFFD name that getFieldNames lists for them, in objects below the threshold. Flink does not write such keys, and distinct malformed keys already decoded to the same name. JMH on JDK 17, 4 lookups per row of 13 keys: 285 ns to 132 ns per row, 1368 to 296 bytes allocated. Generated-by: Claude Code 2.1.309 (Claude Opus 5.5) (cherry picked from commit 82834bb220ba01a948786273cda6eb59ab57a7ba) --- .../flink/types/variant/BinaryVariant.java | 7 +++- .../variant/BinaryVariantInternalBuilder.java | 3 +- .../types/variant/BinaryVariantUtil.java | 36 +++++++++++++++++-- .../types/variant/BinaryVariantTest.java | 34 ++++++++++++++++++ 4 files changed, 76 insertions(+), 4 deletions(-) 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..90e61c159df304 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,12 @@ 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.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.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; @@ -344,9 +346,10 @@ private BinaryVariant getFieldByKey(String key) { // Use linear search for a short list. Switch to binary search when the length // reaches `BINARY_SEARCH_THRESHOLD`. if (size < BINARY_SEARCH_THRESHOLD) { + byte[] keyBytes = encodeKey(key); for (int i = 0; i < size; ++i) { int id = readUnsigned(value, idStart + idSize * i, idSize); - if (key.equals(getMetadataKey(metadata, id))) { + if (metadataKeyEquals(metadata, id, keyBytes)) { int offset = readUnsigned( value, offsetStart + offsetSize * i, offsetSize); @@ -362,6 +365,8 @@ private BinaryVariant getFieldByKey(String key) { // handle the case where `low + high` overflows int. int mid = (low + high) >>> 1; int id = readUnsigned(value, idStart + idSize * mid, idSize); + // Compare decoded Strings here. The builder sorts fields by + // String.compareTo, which is not UTF-8 byte order. int cmp = getMetadataKey(metadata, id).compareTo(key); if (cmp < 0) { low = mid + 1; 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..4ac26cb2b46440 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 @@ -74,6 +74,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 +478,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; } 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..c0812989fa6336 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 @@ -850,6 +850,31 @@ 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); + } + + /** + * 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 +892,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/BinaryVariantTest.java b/flink-core/src/test/java/org/apache/flink/types/variant/BinaryVariantTest.java index e3725c1d202f55..4dd84d1d0db9df 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 @@ -488,6 +488,40 @@ void testNonAsciiStringsAndFieldNames() { + "\"}}"); } + @Test + void testGetFieldWithUnpairedSurrogate() { + // UTF-8 has no unpaired surrogates. The builder writes one as '?', and a lookup encodes its + // key the same way. + final Variant object = builder.object().add("a\uD800", builder.of(1)).build(); + + assertThat(object.getFieldNames()).containsExactly("a?"); + assertThat(object.getField("a\uD800").getInt()).isEqualTo(1); + assertThat(object.getField("a?").getInt()).isEqualTo(1); + } + + @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])) From 72f751e668fd5196c22df2dea716bad8e8a24921 Mon Sep 17 00:00:00 2001 From: Ramin Gharib Date: Mon, 26 Oct 2026 09:47:20 +0100 Subject: [PATCH 2/3] [hotfix][docs] Sync the zh VARIANT data type docs with en The zh page keeps the VARIANT section in English, but it missed later changes: the TIME bullet, the cast of a whole ARRAY, MAP, ROW or STRUCTURED value into a VARIANT, the matching cells of the cast matrix, and note 8. Copy them from the en page. Generated-by: Claude Code 2.1.310 (Claude Opus 5.5) (cherry picked from commit e9939d53403f4146f6542a684457a99bf33ac1c0) --- .../docs/sql/reference/data-types.md | 50 ++++++++++++++++--- 1 file changed, 43 insertions(+), 7 deletions(-) diff --git a/docs/content.zh/docs/sql/reference/data-types.md b/docs/content.zh/docs/sql/reference/data-types.md index 600dec9ef102d1..5bd96e1a6a93ee 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, 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`。 From 2d0a4444fe95ee404548b82c5953821b59ad6030 Mon Sep 17 00:00:00 2001 From: Ramin Gharib Date: Mon, 26 Oct 2026 09:47:29 +0100 Subject: [PATCH 3/3] [FLINK-41180][core] Sort VARIANT object fields by UTF-8 bytes The variant spec orders the fields of an object by the unsigned UTF-8 bytes of their names. The builder sorted them with String.compareTo, which is UTF-16 order, and getField binary-searched objects of 32 or more fields in the same order. The two orders differ when, at the first differing character, one name has a code point from U+E000 to U+FFFF and the other one from U+10000 up. In a spec-ordered object of 32 fields with the keys U+FF61 and U+1F600, getField(U+1F600) returned null. The builder now sorts fields by their stored key bytes, and getField binary-searches in the same order. Objects written in UTF-16 order stay readable: after a miss for a key with a code point from U+E000 up, getField scans the fields linearly. For any other key both orders agree, so the miss stands. Keys that differ as Strings but are stored as the same bytes, such as "a\uD800" and "a?", are now duplicates. Such an object held two fields with the same name before. getField("a\uD800") now finds the field in objects of any size. Objects with keys from U+E000 up can have a different field order now, which shows in getFieldNames and in CAST to STRING. A Flink version without this change can miss such keys in objects of 32 or more fields. Such an object stored before the upgrade is not byte-equal to the same object built after it. Generated-by: Claude Code 2.1.310 (Claude Opus 5.5) (cherry picked from commit dab40d31eba000411f919fe9b77afc6d77bfa1a7) --- .../docs/sql/reference/data-types.md | 4 +- docs/content/docs/sql/reference/data-types.md | 4 +- .../flink/types/variant/BinaryVariant.java | 82 +++++++----- .../types/variant/BinaryVariantBuilder.java | 4 +- .../variant/BinaryVariantInternalBuilder.java | 32 ++--- .../types/variant/BinaryVariantUtil.java | 29 ++++- .../BinaryVariantInternalBuilderTest.java | 2 +- .../types/variant/BinaryVariantTest.java | 123 ++++++++++++++++-- 8 files changed, 208 insertions(+), 72 deletions(-) diff --git a/docs/content.zh/docs/sql/reference/data-types.md b/docs/content.zh/docs/sql/reference/data-types.md index 5bd96e1a6a93ee..703185f11de6aa 100644 --- a/docs/content.zh/docs/sql/reference/data-types.md +++ b/docs/content.zh/docs/sql/reference/data-types.md @@ -1732,8 +1732,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/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 90e61c159df304..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,12 @@ 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; @@ -339,51 +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) { - byte[] keyBytes = encodeKey(key); - for (int i = 0; i < size; ++i) { - int id = readUnsigned(value, idStart + idSize * i, idSize); - if (metadataKeyEquals(metadata, id, keyBytes)) { - 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); - // Compare decoded Strings here. The builder sorts fields by - // String.compareTo, which is not UTF-8 byte order. - 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 4ac26cb2b46440..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; @@ -492,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 @@ -505,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( @@ -554,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; } } @@ -698,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); } } @@ -835,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 c0812989fa6336..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 @@ -868,6 +868,33 @@ static boolean metadataKeyEquals(byte[] metadata, int id, byte[] key) { && 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. 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 4dd84d1d0db9df..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,15 +523,44 @@ void testNonAsciiStringsAndFieldNames() { + "\"}}"); } - @Test - void testGetFieldWithUnpairedSurrogate() { + @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 = builder.object().add("a\uD800", builder.of(1)).build(); + 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); + } - assertThat(object.getFieldNames()).containsExactly("a?"); - assertThat(object.getField("a\uD800").getInt()).isEqualTo(1); - assertThat(object.getField("a?").getInt()).isEqualTo(1); + @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 @@ -589,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; + } + } }