Skip to content

[FLINK-41110][core] Compare UTF-8 bytes to find VARIANT object fields #29580

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
raminqaf wants to merge 1 commit into from
Closed

[FLINK-41110][core] Compare UTF-8 bytes to find VARIANT object fields #29580

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 6 additions & 1 deletion flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariant.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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);
Expand All @@ -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.
Comment on lines +368 to +369

@raminqaf raminqaf Oct 26, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This will be fixed in: https://issues.apache.org/jira/browse/FLINK-41180

@raminqaf raminqaf Oct 26, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

And this is the follow up PR: #29607

int cmp = getMetadataKey(metadata, id).compareTo(key);
if (cmp < 0) {
low = mid + 1;
Expand Down
3 changes: 2 additions & 1 deletion flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariantInternalBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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;
}
Expand Down
36 changes: 34 additions & 2 deletions flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariantUtil.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -850,6 +850,31 @@ public static <T> T handleArray(byte[] value, int pos, ArrayHandler<T> 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.
Expand All @@ -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;
}

/**
Expand Down
34 changes: 34 additions & 0 deletions flink-core/src/test/java/org/apache/flink/types/variant/BinaryVariantTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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]))
Expand Down