This is an automated email from the ASF dual-hosted git repository. raminqaf pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit 82834bb220ba01a948786273cda6eb59ab57a7ba Author: Ramin Gharib <[email protected]> AuthorDate: Thu Oct 22 18:19:31 2026 +0200 [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) --- .../apache/flink/types/variant/BinaryVariant.java | 7 ++++- .../variant/BinaryVariantInternalBuilder.java | 3 +- .../flink/types/variant/BinaryVariantUtil.java | 36 ++++++++++++++++++++-- .../flink/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 254a742eded..90e61c159df 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.SIZE_LIMIT; 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 @@ public final class BinaryVariant implements Variant { // 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 @@ public final class BinaryVariant implements Variant { // 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 3ed2e634412..4ac26cb2b46 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.U8_MAX; 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 class BinaryVariantInternalBuilder { } 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 546b5d4bcdd..c0812989fa6 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 class BinaryVariantUtil { * 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 class BinaryVariantUtil { 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 e3725c1d202..4dd84d1d0db 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 @@ class BinaryVariantTest { + "\"}}"); } + @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]))
