This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new dfc210631e99 fix(flink): use UTF-8 ordering for LSM record keys
(#19548)
dfc210631e99 is described below
commit dfc210631e999da6e1c0f3d64fbec94aec660441
Author: Shuo Cheng <[email protected]>
AuthorDate: Tue Aug 11 14:05:16 2026 +0800
fix(flink): use UTF-8 ordering for LSM record keys (#19548)
---
.../read/lsm/TestLsmFileGroupRecordIterator.java | 13 ++--
.../hudi/sink/utils/RecordKeySortComparator.java | 11 ++-
.../hudi/sink/utils/RecordKeySortKeyComputer.java | 68 +++++++++++++++--
.../sink/utils/TestRecordKeySortKeyComputer.java | 86 ++++++++++++++++------
4 files changed, 138 insertions(+), 40 deletions(-)
diff --git
a/hudi-common/src/test/java/org/apache/hudi/common/table/read/lsm/TestLsmFileGroupRecordIterator.java
b/hudi-common/src/test/java/org/apache/hudi/common/table/read/lsm/TestLsmFileGroupRecordIterator.java
index edd5a83284b4..0bd825460b40 100644
---
a/hudi-common/src/test/java/org/apache/hudi/common/table/read/lsm/TestLsmFileGroupRecordIterator.java
+++
b/hudi-common/src/test/java/org/apache/hudi/common/table/read/lsm/TestLsmFileGroupRecordIterator.java
@@ -198,17 +198,18 @@ class TestLsmFileGroupRecordIterator {
@Test
void testLoserTreeUsesUtf8Ordering() {
- String bmpPrivateUseKey = new String(Character.toChars(0xE000));
- String supplementaryKey = new String(Character.toChars(0x20000));
+ String fullWidthExclamationKey = "！";
+ String emojiKey = "😀";
LsmFileGroupRecordIterator.LoserTree<String> loserTree =
new LsmFileGroupRecordIterator.LoserTree<>(
Arrays.asList(
- sortedRunReader(0, record(bmpPrivateUseKey, "bmp")),
- sortedRunReader(1, record(supplementaryKey,
"supplementary"))));
+ sortedRunReader(0, record(fullWidthExclamationKey,
"full-width-exclamation")),
+ sortedRunReader(1, record(emojiKey, "emoji"))));
+ // UTF-16 orders the emoji first, while UTF-8 bytes order the full-width
character first.
assertEquals(Arrays.asList(
- bmpPrivateUseKey + ":bmp",
- supplementaryKey + ":supplementary"), drain(loserTree));
+ fullWidthExclamationKey + ":full-width-exclamation",
+ emojiKey + ":emoji"), drain(loserTree));
}
@Test
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/RecordKeySortComparator.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/RecordKeySortComparator.java
index 1abd372ab5c6..305f7f1903e8 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/RecordKeySortComparator.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/RecordKeySortComparator.java
@@ -18,6 +18,7 @@
package org.apache.hudi.sink.utils;
+import org.apache.hudi.common.util.StringUtils;
import org.apache.hudi.sink.bulk.RowDataKeyGen;
import org.apache.flink.table.data.RowData;
@@ -26,9 +27,9 @@ import
org.apache.flink.table.runtime.generated.RecordComparator;
/**
* Compares rows by compact keys that preserve encoded Hudi record-key
ordering.
*
- * <p>LSM sorted runs must use the same ordering as the common LSM reader,
which compares the
- * encoded record-key strings. Comparing the typed primary-key fields directly
is not equivalent;
- * for example, integer keys {@code 2} and {@code 10} have the opposite order
after encoding.
+ * <p>LSM sorted runs must use the same unsigned UTF-8 byte ordering as the
common LSM reader.
+ * Comparing the typed primary-key fields directly is not equivalent; for
example, integer keys
+ * {@code 2} and {@code 10} have the opposite order after encoding.
*/
public class RecordKeySortComparator implements RecordComparator {
private final RowDataKeyGen keyGen;
@@ -39,6 +40,8 @@ public class RecordKeySortComparator implements
RecordComparator {
@Override
public int compare(RowData record1, RowData record2) {
- return
keyGen.getRecordKeyForComparison(record1).compareTo(keyGen.getRecordKeyForComparison(record2));
+ return StringUtils.compareUtf8Bytes(
+ keyGen.getRecordKeyForComparison(record1),
+ keyGen.getRecordKeyForComparison(record2));
}
}
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/RecordKeySortKeyComputer.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/RecordKeySortKeyComputer.java
index 932167b4dede..2fdb7053699d 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/RecordKeySortKeyComputer.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/RecordKeySortKeyComputer.java
@@ -20,7 +20,6 @@ package org.apache.hudi.sink.utils;
import org.apache.hudi.sink.bulk.RowDataKeyGen;
-import org.apache.flink.api.common.typeutils.base.StringComparator;
import org.apache.flink.core.memory.MemorySegment;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.runtime.generated.NormalizedKeyComputer;
@@ -28,10 +27,9 @@ import
org.apache.flink.table.runtime.generated.NormalizedKeyComputer;
/**
* Computes a normalized prefix for sorting rows by their encoded Hudi record
key.
*
- * <p>The common LSM reader compares encoded record keys with {@link
String#compareTo(String)},
- * whose order is based on UTF-16 code units. This computer therefore uses
Flink's normalized-key
- * encoding for Java strings, which preserves that order while still using
only one byte for the
- * common ASCII case.
+ * <p>The common LSM reader compares encoded record keys by their unsigned
UTF-8 byte
+ * representation. This computer copies a prefix of those bytes into the
normalized key so its
+ * ordering is consistent with the reader.
*
* <p>The normalized prefix is eight bytes for one record-key field and
sixteen bytes for two or
* more fields. It is packed into one or two big-endian {@code long}s so the
sort hot path can
@@ -47,7 +45,6 @@ public class RecordKeySortKeyComputer implements
NormalizedKeyComputer {
private final RowDataKeyGen keyGen;
private final int numKeyBytes;
- private final StringComparator stringComparator = new StringComparator(true);
/**
* Creates a normalized-key computer using up to sixteen bytes of the
encoded record-key prefix.
@@ -62,8 +59,65 @@ public class RecordKeySortKeyComputer implements
NormalizedKeyComputer {
@Override
public void putKey(RowData record, MemorySegment target, int offset) {
- stringComparator.putNormalizedKey(
+ int bytesWritten = putUtf8Prefix(
keyGen.getRecordKeyForComparison(record), target, offset, numKeyBytes);
+ for (int i = bytesWritten; i < numKeyBytes; i++) {
+ target.put(offset + i, (byte) 0);
+ }
+ }
+
+ /**
+ * Encodes at most {@code maxBytes} of the UTF-8 prefix directly into the
target segment.
+ *
+ * <p>This writes the exact first {@code maxBytes} UTF-8 bytes, including a
partial multi-byte
+ * sequence at the prefix boundary, without materializing the complete
encoded record key.
+ *
+ * <p>Production record keys contain well-formed UTF-16 because they are
derived from UTF-8 input.
+ * An unpaired surrogate is encoded with the same replacement byte used by
+ * {@link String#getBytes(java.nio.charset.Charset)}.
+ */
+ private static int putUtf8Prefix(
+ String value, MemorySegment target, int offset, int maxBytes) {
+ int bytesWritten = 0;
+ for (int i = 0; i < value.length() && bytesWritten < maxBytes; i++) {
+ char current = value.charAt(i);
+ int encoded;
+ int encodedBytes;
+ // Pack the encoded bytes in big-endian order into the least-significant
encodedBytes bytes.
+ if (current < 0x80) {
+ encoded = current;
+ encodedBytes = 1;
+ } else if (current < 0x800) {
+ encoded = (0xC0 | current >>> 6) << 8
+ | (0x80 | current & 0x3F);
+ encodedBytes = 2;
+ } else if (Character.isHighSurrogate(current)
+ && i + 1 < value.length()
+ && Character.isLowSurrogate(value.charAt(i + 1))) {
+ int codePoint = Character.toCodePoint(current, value.charAt(++i));
+ encoded = (0xF0 | codePoint >>> 18) << 24
+ | (0x80 | codePoint >>> 12 & 0x3F) << 16
+ | (0x80 | codePoint >>> 6 & 0x3F) << 8
+ | (0x80 | codePoint & 0x3F);
+ encodedBytes = 4;
+ } else if (Character.isSurrogate(current)) {
+ encoded = '?';
+ encodedBytes = 1;
+ } else {
+ encoded = (0xE0 | current >>> 12) << 16
+ | (0x80 | current >>> 6 & 0x3F) << 8
+ | (0x80 | current & 0x3F);
+ encodedBytes = 3;
+ }
+
+ // Writing byte by byte is intentional: the normalized prefix may end
inside this sequence.
+ for (int shift = (encodedBytes - 1) * Byte.SIZE;
+ shift >= 0 && bytesWritten < maxBytes;
+ shift -= Byte.SIZE) {
+ target.put(offset + bytesWritten++, (byte) (encoded >>> shift));
+ }
+ }
+ return bytesWritten;
}
@Override
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/TestRecordKeySortKeyComputer.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/TestRecordKeySortKeyComputer.java
index 8bb29ceabb2e..fb0d998448f5 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/TestRecordKeySortKeyComputer.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/TestRecordKeySortKeyComputer.java
@@ -18,6 +18,7 @@
package org.apache.hudi.sink.utils;
+import org.apache.hudi.common.util.StringUtils;
import org.apache.hudi.configuration.FlinkOptions;
import org.apache.hudi.sink.buffer.HeapMemorySegmentPool;
import org.apache.hudi.sink.bulk.RowDataKeyGen;
@@ -40,12 +41,13 @@ import org.apache.flink.table.types.logical.VarCharType;
import org.apache.flink.util.MutableObjectIterator;
import org.junit.jupiter.api.Test;
+import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
-import java.util.Collections;
import java.util.List;
import java.util.Random;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -85,7 +87,7 @@ class TestRecordKeySortKeyComputer {
}
@Test
- void testNormalizedKeyPreservesJavaStringOrder() {
+ void testNormalizedKeyPreservesUtf8Order() {
RowType rowType = rowType(new String[] {"key"}, new LogicalType[] {new
VarCharType()});
RowDataKeyGen keyGen = keyGen(rowType, "key");
RecordKeySortKeyComputer computer = new RecordKeySortKeyComputer(keyGen,
1);
@@ -94,15 +96,44 @@ class TestRecordKeySortKeyComputer {
assertComparison(computer, comparator, stringRow("10"), stringRow("2"),
false);
assertComparison(computer, comparator, stringRow("abc"),
stringRow("abcd"), false);
assertComparison(computer, comparator, stringRow("\u0000a"),
stringRow("\u0000b"), false);
- String supplementaryCharacter = new String(Character.toChars(0x1F600));
- String privateUseCharacter = String.valueOf((char) 0xE000);
+ String supplementaryCharacter = new String(Character.toChars(0x20000));
+ String privateUseCharacter = new String(Character.toChars(0xE000));
+ assertTrue(StringUtils.compareUtf8Bytes(privateUseCharacter,
supplementaryCharacter) < 0);
assertComparison(computer, comparator,
- stringRow(supplementaryCharacter), stringRow(privateUseCharacter),
false);
+ stringRow(privateUseCharacter), stringRow(supplementaryCharacter),
false);
assertComparison(computer, comparator, stringRow("abcdefgh1"),
stringRow("abcdefgh2"), true);
}
@Test
- void testNormalizedKeyCompareMatchesStringCompareToAtEncodingBoundaries() {
+ void testNormalizedKeyContainsExactUtf8Prefix() {
+ RowType rowType = rowType(new String[] {"key"}, new LogicalType[] {new
VarCharType()});
+ RowDataKeyGen keyGen = keyGen(rowType, "key");
+ RecordKeySortKeyComputer computer = new RecordKeySortKeyComputer(keyGen,
1);
+ List<String> values = Arrays.asList(
+ "ascii",
+ "12345678suffix",
+ "1234567" + new String(Character.toChars(0x20000)),
+ "123456" + new String(Character.toChars(0x20000)),
+ stringFromCodeUnits(0x007F, 0x0080, 0x07FF, 0x0800),
+ new String(Character.toChars(0xE000)),
+ new String(Character.toChars(0x10FFFF)));
+
+ for (String value : values) {
+ MemorySegment normalizedKey =
+
MemorySegmentFactory.allocateUnpooledSegment(computer.getNumKeyBytes());
+ computer.putKey(stringRow(value), normalizedKey, 0);
+
+ byte[] expected = Arrays.copyOf(
+
keyGen.getRecordKeyForComparison(stringRow(value)).getBytes(StandardCharsets.UTF_8),
+ computer.getNumKeyBytes());
+ byte[] actual = new byte[computer.getNumKeyBytes()];
+ normalizedKey.get(0, actual);
+ assertArrayEquals(expected, actual, () -> "Unexpected UTF-8 prefix for "
+ printable(value));
+ }
+ }
+
+ @Test
+ void testNormalizedKeyCompareMatchesUtf8OrderAtEncodingBoundaries() {
List<String> values = Arrays.asList(
stringFromCodeUnits(0x0000),
stringFromCodeUnits(0x0000) + "a",
@@ -116,26 +147,29 @@ class TestRecordKeySortKeyComputer {
stringFromCodeUnits(0x1FFF),
stringFromCodeUnits(0x2000),
stringFromCodeUnits(0xD7FF),
- stringFromCodeUnits(0xD800),
stringFromCodeUnits(0xD800, 0xDC00),
stringFromCodeUnits(0xD83D, 0xDE00),
stringFromCodeUnits(0xE000),
stringFromCodeUnits(0xFFFF));
- assertNormalizedKeyComparisonsMatchStringCompareTo(values);
+ assertNormalizedKeyComparisonsMatchUtf8Order(values);
}
@Test
- void testNormalizedKeyCompareMatchesStringCompareToForRandomUtf16Strings() {
+ void testNormalizedKeyCompareMatchesUtf8OrderForRandomStrings() {
Random random = new Random(42);
List<String> values = new ArrayList<>();
for (int i = 0; i < 1_000; i++) {
int length = random.nextInt(23) + 1;
- char[] chars = new char[length];
+ StringBuilder builder = new StringBuilder();
for (int j = 0; j < length; j++) {
- chars[j] = (char) random.nextInt(Character.MAX_VALUE + 1);
+ int codePoint;
+ do {
+ codePoint = random.nextInt(Character.MAX_CODE_POINT + 1);
+ } while (codePoint >= Character.MIN_SURROGATE && codePoint <=
Character.MAX_SURROGATE);
+ builder.appendCodePoint(codePoint);
}
- values.add(new String(chars));
+ values.add(builder.toString());
}
RowType rowType = rowType(new String[] {"key"}, new LogicalType[] {new
VarCharType()});
@@ -145,7 +179,7 @@ class TestRecordKeySortKeyComputer {
for (int i = 0; i < values.size(); i++) {
String left = values.get(i);
String right = values.get((i * 31 + 17) % values.size());
- assertNormalizedComparisonMatchesStringCompareTo(
+ assertNormalizedComparisonMatchesUtf8Order(
computer, comparator, keyGen, left, right);
}
}
@@ -161,14 +195,17 @@ class TestRecordKeySortKeyComputer {
stringRow("aa", "a"),
stringRow("a,b", "c"),
stringRow("a", "b,c"),
+ stringRow(new String(Character.toChars(0xE000)), "a"),
+ stringRow(new String(Character.toChars(0x20000)), "a"),
stringRow("", "a"),
GenericRowData.of(null, StringData.fromString("a")));
for (RowData left : rows) {
for (RowData right : rows) {
- int recordKeyResult =
keyGen.getRecordKey(left).compareTo(keyGen.getRecordKey(right));
- int comparisonKeyResult = keyGen.getRecordKeyForComparison(left)
- .compareTo(keyGen.getRecordKeyForComparison(right));
+ int recordKeyResult = StringUtils.compareUtf8Bytes(
+ keyGen.getRecordKey(left), keyGen.getRecordKey(right));
+ int comparisonKeyResult = StringUtils.compareUtf8Bytes(
+ keyGen.getRecordKeyForComparison(left),
keyGen.getRecordKeyForComparison(right));
assertEquals(Integer.signum(recordKeyResult),
Integer.signum(comparisonKeyResult));
}
}
@@ -186,7 +223,9 @@ class TestRecordKeySortKeyComputer {
assertBufferSort(complexRowType, "key1,key2", Arrays.asList(
stringRow("same-prefix-2", "a"),
stringRow("same-prefix-1", "z"),
- stringRow("same-prefix-1", "a")));
+ stringRow("same-prefix-1", "a"),
+ stringRow(new String(Character.toChars(0xE000)), "a"),
+ stringRow(new String(Character.toChars(0x20000)), "a")));
}
private static void assertComparison(
@@ -207,20 +246,20 @@ class TestRecordKeySortKeyComputer {
assertEquals(Integer.signum(comparator.compare(left, right)),
Integer.signum(result));
}
- private static void
assertNormalizedKeyComparisonsMatchStringCompareTo(List<String> values) {
+ private static void
assertNormalizedKeyComparisonsMatchUtf8Order(List<String> values) {
RowType rowType = rowType(new String[] {"key"}, new LogicalType[] {new
VarCharType()});
RowDataKeyGen keyGen = keyGen(rowType, "key");
RecordKeySortKeyComputer computer = new RecordKeySortKeyComputer(keyGen,
1);
RecordKeySortComparator comparator = new RecordKeySortComparator(keyGen);
for (String left : values) {
for (String right : values) {
- assertNormalizedComparisonMatchesStringCompareTo(
+ assertNormalizedComparisonMatchesUtf8Order(
computer, comparator, keyGen, left, right);
}
}
}
- private static void assertNormalizedComparisonMatchesStringCompareTo(
+ private static void assertNormalizedComparisonMatchesUtf8Order(
RecordKeySortKeyComputer computer,
RecordKeySortComparator comparator,
RowDataKeyGen keyGen,
@@ -234,8 +273,9 @@ class TestRecordKeySortKeyComputer {
computer.putKey(leftRow, leftKey, 0);
computer.putKey(rightRow, rightKey, 0);
- int expected = keyGen.getRecordKeyForComparison(leftRow)
- .compareTo(keyGen.getRecordKeyForComparison(rightRow));
+ int expected = StringUtils.compareUtf8Bytes(
+ keyGen.getRecordKeyForComparison(leftRow),
+ keyGen.getRecordKeyForComparison(rightRow));
int normalizedResult = computer.compareKey(leftKey, 0, rightKey, 0);
if (normalizedResult != 0) {
assertEquals(Integer.signum(expected), Integer.signum(normalizedResult),
@@ -279,7 +319,7 @@ class TestRecordKeySortKeyComputer {
assertTrue(buffer.write(row));
expected.add(keyGen.getRecordKey(row));
}
- Collections.sort(expected);
+ expected.sort(StringUtils.UTF8_LEXICOGRAPHIC_COMPARATOR);
new QuickSort().sort(buffer);
MutableObjectIterator<BinaryRowData> iterator = buffer.getIterator();
