This is an automated email from the ASF dual-hosted git repository. voonhous pushed a commit to branch nada.attia/ri-bootstrap-binary-keys-oss in repository https://gitbox.apache.org/repos/asf/hudi.git
commit dc3aabdffe8730043e858485caa678113f14a757 Author: voon <[email protected]> AuthorDate: Fri Jul 24 19:15:39 2026 +0800 Address review: add focused unit tests for Java and Spark MDT partitioners, sorted buffer merge order, and comparator serialization --- ...estJavaHoodieMetadataBulkInsertPartitioner.java | 100 +++++++++++++++++++++ ...stSparkHoodieMetadataBulkInsertPartitioner.java | 61 +++++++++++++ .../TestSortedKeyBasedFileGroupRecordBuffer.java | 35 ++++++++ .../apache/hudi/common/util/TestStringUtils.java | 29 +++++- 4 files changed, 221 insertions(+), 4 deletions(-) diff --git a/hudi-client/hudi-java-client/src/test/java/org/apache/hudi/metadata/TestJavaHoodieMetadataBulkInsertPartitioner.java b/hudi-client/hudi-java-client/src/test/java/org/apache/hudi/metadata/TestJavaHoodieMetadataBulkInsertPartitioner.java new file mode 100644 index 000000000000..2f600bac0954 --- /dev/null +++ b/hudi-client/hudi-java-client/src/test/java/org/apache/hudi/metadata/TestJavaHoodieMetadataBulkInsertPartitioner.java @@ -0,0 +1,100 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.hudi.metadata; + +import org.apache.hudi.common.model.EmptyHoodieRecordPayload; +import org.apache.hudi.common.model.HoodieAvroRecord; +import org.apache.hudi.common.model.HoodieKey; +import org.apache.hudi.common.model.HoodieRecord; +import org.apache.hudi.common.model.HoodieRecordLocation; +import org.apache.hudi.common.util.StringUtils; + +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Tests {@link JavaHoodieMetadataBulkInsertPartitioner}, which sorts MDT/HFile record keys by raw + * UTF-8 bytes rather than String (UTF-16) order. + */ +class TestJavaHoodieMetadataBulkInsertPartitioner { + + @Test + void repartitionRecordsSortsBinaryKeysByUtf8Bytes() { + // U+E000 (UTF-8 lead byte 0xEE) sorts BEFORE U+20000 (UTF-8 lead byte 0xF0) in raw UTF-8 byte + // order, but AFTER it under String.compareTo (UTF-16). This is the pathological pair the + // partitioner's UTF-8 comparator must get right so HFile forward-only seeks stay valid. + String bmpPrivateUse = new String(Character.toChars(0xE000)); + String supplementary = new String(Character.toChars(0x20000)); + // All records share one file group so the partitioner's single-group assumption holds. + String fileId = "files-0000"; + + // Shuffled input mixing both prefixes plus ascii suffixes. + List<String> inputKeys = Arrays.asList( + supplementary + "-b", + "ascii-key", + bmpPrivateUse + "-a", + supplementary + "-a", + bmpPrivateUse + "-b"); + + List<HoodieRecord<EmptyHoodieRecordPayload>> records = new ArrayList<>(); + for (String key : inputKeys) { + HoodieRecord<EmptyHoodieRecordPayload> record = + new HoodieAvroRecord<>(new HoodieKey(key, ""), new EmptyHoodieRecordPayload()); + record.unseal(); + record.setCurrentLocation(new HoodieRecordLocation("001", fileId)); + record.seal(); + records.add(record); + } + + JavaHoodieMetadataBulkInsertPartitioner<EmptyHoodieRecordPayload> partitioner = + new JavaHoodieMetadataBulkInsertPartitioner<>(); + List<HoodieRecord<EmptyHoodieRecordPayload>> sorted = partitioner.repartitionRecords(records, 1); + + assertTrue(partitioner.arePartitionRecordsSorted(), "Records must be sorted"); + + List<String> actualKeys = new ArrayList<>(); + for (HoodieRecord<EmptyHoodieRecordPayload> record : sorted) { + actualKeys.add(record.getRecordKey()); + } + List<String> expectedKeys = new ArrayList<>(inputKeys); + expectedKeys.sort(StringUtils.UTF8_LEXICOGRAPHIC_COMPARATOR); + assertEquals(expectedKeys, actualKeys, "Records must be sorted by UTF-8 byte order"); + + // The divergent pair: every U+E000-prefixed key precedes every U+20000-prefixed key in UTF-8 + // byte order, the opposite of String.compareTo (UTF-16) order. + int lastBmpIndex = -1; + int firstSupplementaryIndex = actualKeys.size(); + for (int i = 0; i < actualKeys.size(); i++) { + if (actualKeys.get(i).startsWith(bmpPrivateUse)) { + lastBmpIndex = i; + } else if (actualKeys.get(i).startsWith(supplementary) && firstSupplementaryIndex == actualKeys.size()) { + firstSupplementaryIndex = i; + } + } + assertTrue(lastBmpIndex < firstSupplementaryIndex, + "All U+E000-prefixed keys should sort before U+20000-prefixed keys in UTF-8 order"); + } +} diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestSparkHoodieMetadataBulkInsertPartitioner.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestSparkHoodieMetadataBulkInsertPartitioner.java index aa46a177ac4e..0a8b7918d850 100644 --- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestSparkHoodieMetadataBulkInsertPartitioner.java +++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestSparkHoodieMetadataBulkInsertPartitioner.java @@ -19,8 +19,10 @@ package org.apache.hudi.client; +import org.apache.hudi.common.model.HoodieKey; import org.apache.hudi.common.model.HoodieRecord; import org.apache.hudi.common.model.HoodieRecordLocation; +import org.apache.hudi.common.util.StringUtils; import org.apache.hudi.metadata.DefaultMetadataTableFileGroupIndexParser; import org.apache.hudi.metadata.HoodieMetadataPayload; import org.apache.hudi.metadata.MetadataPartitionType; @@ -31,6 +33,7 @@ import org.apache.spark.api.java.JavaRDD; import org.junit.jupiter.api.Test; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collections; import java.util.List; import java.util.Map; @@ -108,4 +111,62 @@ class TestSparkHoodieMetadataBulkInsertPartitioner extends SparkClientFunctional Set<String> fileIDPrefixes = IntStream.of(0, 1, 2, 4).mapToObj(partitioner::getFileIdPfx).collect(Collectors.toSet()); assertEquals(fileIDPrefixes, recordsPerFileGroup.keySet(), "fileIDPrefixes should match the name of the MDT fileGroups"); } + + @Test + public void testPartitionerSortsBinaryKeysByUtf8Bytes() { + // U+E000 (UTF-8 lead byte 0xEE) sorts BEFORE U+20000 (UTF-8 lead byte 0xF0) in raw UTF-8 byte + // order, but AFTER it under String.compareTo (UTF-16). All records target a single MDT file group + // so the partitioner's only job here is the within-partition UTF-8 sort. + String fileGroupId = MetadataPartitionType.FILES.getFileIdPrefix() + "000"; + String bmpPrivateUse = new String(Character.toChars(0xE000)); + String supplementary = new String(Character.toChars(0x20000)); + + // Shuffled input mixing both prefixes plus an ascii key. + List<String> inputKeys = Arrays.asList( + supplementary + "-b", + "ascii-key", + bmpPrivateUse + "-a", + supplementary + "-a", + bmpPrivateUse + "-b"); + + List<HoodieRecord> records = new ArrayList<>(); + for (String key : inputKeys) { + // createPartitionListRecord fixes the record key, so start from it (for a valid MDT payload) + // and rebind an explicitly chosen HoodieKey via newInstance. + HoodieRecord r = HoodieMetadataPayload.createPartitionListRecord(Collections.EMPTY_LIST) + .newInstance(new HoodieKey(key, "")); + r.unseal(); + r.setCurrentLocation(new HoodieRecordLocation("001", fileGroupId)); + r.seal(); + records.add(r); + } + + SparkHoodieMetadataBulkInsertPartitioner partitioner = + new SparkHoodieMetadataBulkInsertPartitioner(new DefaultMetadataTableFileGroupIndexParser(1)); + JavaRDD<HoodieRecord> partitionedRecords = + partitioner.repartitionRecords(jsc().parallelize(records, records.size()), 0); + + // All records map to one file group, hence a single partition. + assertEquals(1, partitionedRecords.getNumPartitions(), "All records map to a single file group"); + assertTrue(partitioner.arePartitionRecordsSorted(), "Must be sorted"); + + List<String> actualKeys = partitionedRecords.map(r -> r.getRecordKey()).collect(); + List<String> expectedKeys = new ArrayList<>(inputKeys); + expectedKeys.sort(StringUtils.UTF8_LEXICOGRAPHIC_COMPARATOR); + assertEquals(expectedKeys, actualKeys, "Records must be sorted by UTF-8 byte order within the file group"); + + // The divergent pair: every U+E000-prefixed key precedes every U+20000-prefixed key in UTF-8 + // byte order, the opposite of String.compareTo (UTF-16) order. + int lastBmpIndex = -1; + int firstSupplementaryIndex = actualKeys.size(); + for (int i = 0; i < actualKeys.size(); i++) { + if (actualKeys.get(i).startsWith(bmpPrivateUse)) { + lastBmpIndex = i; + } else if (actualKeys.get(i).startsWith(supplementary) && firstSupplementaryIndex == actualKeys.size()) { + firstSupplementaryIndex = i; + } + } + assertTrue(lastBmpIndex < firstSupplementaryIndex, + "All U+E000-prefixed keys should sort before U+20000-prefixed keys in UTF-8 order"); + } } diff --git a/hudi-common/src/test/java/org/apache/hudi/common/table/read/buffer/TestSortedKeyBasedFileGroupRecordBuffer.java b/hudi-common/src/test/java/org/apache/hudi/common/table/read/buffer/TestSortedKeyBasedFileGroupRecordBuffer.java index 611128a96309..599358c74fc8 100644 --- a/hudi-common/src/test/java/org/apache/hudi/common/table/read/buffer/TestSortedKeyBasedFileGroupRecordBuffer.java +++ b/hudi-common/src/test/java/org/apache/hudi/common/table/read/buffer/TestSortedKeyBasedFileGroupRecordBuffer.java @@ -56,6 +56,7 @@ import java.util.stream.Stream; import static org.apache.hudi.common.model.DefaultHoodieRecordPayload.DELETE_KEY; import static org.apache.hudi.common.model.DefaultHoodieRecordPayload.DELETE_MARKER; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyList; import static org.mockito.Mockito.RETURNS_DEEP_STUBS; @@ -188,6 +189,40 @@ class TestSortedKeyBasedFileGroupRecordBuffer extends BaseTestFileGroupRecordBuf assertEquals(1, readStats.getNumDeletes()); } + @Test + void readBaseFileAndLogFileWithBinaryKeys() throws IOException { + // U+E000 (UTF-8 lead byte 0xEE) sorts BEFORE U+20000 (UTF-8 lead byte 0xF0) in raw UTF-8 byte + // order, but AFTER it under String.compareTo (UTF-16). The base-file record carries the + // U+E000-prefixed (UTF-8-smaller, UTF-16-larger) key and the log carries the U+20000-prefixed + // (UTF-8-larger, UTF-16-smaller) key, so a correct merge must emit them in UTF-8 byte order. + String bmpPrivateUse = new String(Character.toChars(0xE000)); + String supplementary = new String(Character.toChars(0x20000)); + TestRecord asciiA = new TestRecord("a", 0); + TestRecord asciiB = new TestRecord("b", 0); + TestRecord bmpRecord = new TestRecord(bmpPrivateUse + "-base", 0); + TestRecord supplementaryRecord = new TestRecord(supplementary + "-log", 0); + + HoodieReadStats readStats = new HoodieReadStats(); + HoodieReaderContext<TestRecord> mockReaderContext = mock(HoodieReaderContext.class, RETURNS_DEEP_STUBS); + SortedKeyBasedFileGroupRecordBuffer<TestRecord> fileGroupRecordBuffer = buildSortedKeyBasedFileGroupRecordBuffer(mockReaderContext, readStats); + + // Base-file records must already be in UTF-8 byte order: "a" (0x61) then the U+E000 key (0xEE...). + fileGroupRecordBuffer.setBaseFileIterator(ClosableIterator.wrap(Arrays.asList(asciiA, bmpRecord).iterator())); + + // Log records are supplied shuffled; the buffer sorts them by UTF-8 bytes before merging. + HoodieDataBlock dataBlock = mock(HoodieDataBlock.class); + when(dataBlock.getSchema()).thenReturn(HoodieTestDataGenerator.HOODIE_SCHEMA); + when(dataBlock.getEngineRecordIterator(mockReaderContext)).thenReturn( + ClosableIterator.wrap(Arrays.asList(supplementaryRecord, asciiB).iterator())); + fileGroupRecordBuffer.processDataBlock(dataBlock, Option.empty()); + + List<TestRecord> actualRecords = getActualRecordsForSortedKeyBased(fileGroupRecordBuffer); + // Expected UTF-8 byte order: "a", "b", U+E000 key, U+20000 key; nothing is dropped. + assertEquals(Arrays.asList(asciiA, asciiB, bmpRecord, supplementaryRecord), actualRecords); + // The U+E000-prefixed base record precedes the U+20000-prefixed log record (reverse of UTF-16). + assertTrue(actualRecords.indexOf(bmpRecord) < actualRecords.indexOf(supplementaryRecord)); + } + private SortedKeyBasedFileGroupRecordBuffer<TestRecord> buildSortedKeyBasedFileGroupRecordBuffer(HoodieReaderContext<TestRecord> mockReaderContext, HoodieReadStats readStats) { when(mockReaderContext.getSchemaHandler().getRequiredSchema()).thenReturn(HoodieTestDataGenerator.HOODIE_SCHEMA); when(mockReaderContext.getSchemaHandler().getInternalSchema()).thenReturn(InternalSchema.getEmptyInternalSchema()); diff --git a/hudi-io/src/test/java/org/apache/hudi/common/util/TestStringUtils.java b/hudi-io/src/test/java/org/apache/hudi/common/util/TestStringUtils.java index 00bb26aac3b7..265ebbf305ef 100644 --- a/hudi-io/src/test/java/org/apache/hudi/common/util/TestStringUtils.java +++ b/hudi-io/src/test/java/org/apache/hudi/common/util/TestStringUtils.java @@ -21,12 +21,17 @@ package org.apache.hudi.common.util; import org.junit.jupiter.api.Test; +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; import java.nio.ByteBuffer; import java.nio.charset.StandardCharsets; import java.security.SecureRandom; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; +import java.util.Comparator; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -319,11 +324,27 @@ public class TestStringUtils { } @Test - public void testUtf8LexicographicComparatorMatchesCompareUtf8Bytes() { + @SuppressWarnings("unchecked") + public void testUtf8LexicographicComparatorSerializableAndRejectsNull() throws Exception { + // Like String.compareTo, a null argument is rejected. + assertThrows(NullPointerException.class, () -> StringUtils.UTF8_LEXICOGRAPHIC_COMPARATOR.compare(null, "a")); + + // The comparator is declared as (Comparator<String> & Serializable) so Spark can capture it inside + // serialized closures. Round-trip it through Java serialization and confirm the deserialized + // instance still orders keys by UTF-8 bytes for the divergent U+E000 vs U+20000 pair (U+E000's + // UTF-8 lead byte 0xEE sorts before U+20000's 0xF0, the reverse of String.compareTo / UTF-16). + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + try (ObjectOutputStream oos = new ObjectOutputStream(baos)) { + oos.writeObject(StringUtils.UTF8_LEXICOGRAPHIC_COMPARATOR); + } + Comparator<String> deserialized; + try (ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(baos.toByteArray()))) { + deserialized = (Comparator<String>) ois.readObject(); + } + String bmpPrivateUse = new String(Character.toChars(0xE000)); String supplementary = new String(Character.toChars(0x20000)); - assertEquals(StringUtils.compareUtf8Bytes(bmpPrivateUse, supplementary), - StringUtils.UTF8_LEXICOGRAPHIC_COMPARATOR.compare(bmpPrivateUse, supplementary)); - assertThrows(NullPointerException.class, () -> StringUtils.UTF8_LEXICOGRAPHIC_COMPARATOR.compare(null, "a")); + assertTrue(deserialized.compare(bmpPrivateUse, supplementary) < 0, + "Deserialized comparator should order U+E000 before U+20000 (UTF-8 byte order)"); } }
