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 26b1ebc4e938 fix(record-index-bootstrap): sort record index keys by
UTF-8 bytes to match HFile sorting (#18941)
26b1ebc4e938 is described below
commit 26b1ebc4e9384d15c5599a4d97d522f781063b35
Author: Nada <[email protected]>
AuthorDate: Fri Jul 24 10:10:19 2026 -0400
fix(record-index-bootstrap): sort record index keys by UTF-8 bytes to match
HFile sorting (#18941)
* fix(HUDI-8898): sort record index keys by UTF-8 bytes to match HFile
ordering
* test(HUDI-8898): add record index bootstrap test for binary record keys
* Address review: fix UTF-8 key ordering in RI multi-slice lookup and MDT
compaction paths
* test(HUDI-8898): add TestStringUtils coverage for UTF-8 comparator
* Fix HoodieSortedMergeHandle to sort keys by UTF-8 bytes, not UTF-16
* Fix checkstyle: avoid escaped unicode chars in test keys
* Address review: note per-call UTF-8 encoding cost in compareUtf8Bytes
Javadoc
* Address review: restore record lambda param name in
JavaHoodieMetadataBulkInsertPartitioner
* Address review: sort Flink MDT bulk-insert keys by UTF-8 bytes, re-sort
filterRowKeys candidates, document unpaired-surrogate caveat
* Address review: merge duplicate binary-key bootstrap tests into a
parameterized test and correct the failure-mode comments to the read-side seek
miss
* Address review: add focused unit tests for Java and Spark MDT
partitioners, sorted buffer merge order, and comparator serialization
* Address review: add secondary index test with non-ASCII secondary key
values
* Address review: sort native-log MDT delta-commit keys by UTF-8 bytes
across engines, cover log-file lookups and assert write statuses in binary-key
tests
---------
Co-authored-by: voon <[email protected]>
---
.../java/org/apache/hudi/io/BaseCreateHandle.java | 7 +-
.../hudi/io/HoodieInlineLogAppendHandle.java | 5 +-
.../apache/hudi/io/HoodieSortedMergeHandle.java | 5 +-
.../apache/hudi/client/HoodieFlinkWriteClient.java | 5 +-
.../JavaHoodieMetadataBulkInsertPartitioner.java | 3 +-
...estJavaHoodieMetadataBulkInsertPartitioner.java | 100 ++++++++++++
.../SparkHoodieMetadataBulkInsertPartitioner.java | 2 +-
.../commit/BaseSparkCommitActionExecutor.java | 5 +-
...stSparkHoodieMetadataBulkInsertPartitioner.java | 61 +++++++
.../SortedKeyBasedFileGroupRecordBuffer.java | 12 +-
.../apache/hudi/common/util/HoodieRecordUtils.java | 6 +-
.../io/storage/HoodieNativeAvroHFileReader.java | 8 +-
.../hudi/metadata/HoodieBackedTableMetadata.java | 15 +-
.../TestSortedKeyBasedFileGroupRecordBuffer.java | 35 ++++
.../hudi/common/util/TestHoodieRecordUtils.java | 9 +-
.../org/apache/hudi/common/util/StringUtils.java | 36 +++++
.../apache/hudi/common/util/TestStringUtils.java | 64 ++++++++
.../hudi/functional/TestHoodieBackedMetadata.java | 176 +++++++++++++++++++++
.../java/org/apache/hudi/io/TestMergeHandle.java | 59 +++++++
.../functional/TestSecondaryIndexPruning.scala | 72 +++++++++
20 files changed, 664 insertions(+), 21 deletions(-)
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/BaseCreateHandle.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/BaseCreateHandle.java
index b4732a55b38a..94bfa6cd2e4d 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/BaseCreateHandle.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/BaseCreateHandle.java
@@ -30,6 +30,7 @@ import org.apache.hudi.common.model.IOType;
import org.apache.hudi.common.model.MetadataValues;
import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.StringUtils;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.core.io.storage.HoodieFileWriter;
import org.apache.hudi.exception.HoodieException;
@@ -131,8 +132,10 @@ public abstract class BaseCreateHandle<T, I, K, O> extends
HoodieWriteHandle<T,
public void write() {
Iterator<String> keyIterator;
if (hoodieTable.requireSortedRecords()) {
- // Sorting the keys limits the amount of extra memory required for
writing sorted records
- keyIterator = recordMap.keySet().stream().sorted().iterator();
+ // Sorting the keys limits the amount of extra memory required for
writing sorted records.
+ // requireSortedRecords() is true only for HFile base files, which order
keys by UTF-8 bytes,
+ // not String (UTF-16) order, so sort with the matching comparator.
+ keyIterator =
recordMap.keySet().stream().sorted(StringUtils.UTF8_LEXICOGRAPHIC_COMPARATOR).iterator();
} else {
keyIterator = recordMap.keySet().stream().iterator();
}
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieInlineLogAppendHandle.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieInlineLogAppendHandle.java
index 4db5bf2dc697..59709e74b1b0 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieInlineLogAppendHandle.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieInlineLogAppendHandle.java
@@ -40,6 +40,7 @@ import org.apache.hudi.common.util.DefaultSizeEstimator;
import org.apache.hudi.common.util.Lazy;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.SizeEstimator;
+import org.apache.hudi.common.util.StringUtils;
import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.exception.HoodieAppendException;
@@ -331,7 +332,9 @@ public class HoodieInlineLogAppendHandle<T, I, K, O>
extends HoodieAppendHandle<
case HFILE_DATA_BLOCK:
// Not supporting positions in HFile data blocks
header.remove(HeaderMetadataType.BASE_FILE_INSTANT_TIME_OF_RECORD_POSITIONS);
- records.sort(Comparator.comparing(HoodieRecord::getRecordKey));
+ // HFile orders keys by their raw UTF-8 bytes, so sort by UTF-8 bytes
rather than
+ // String (UTF-16) order to keep non-ASCII / binary keys consistent
with the writer.
+ records.sort(Comparator.comparing(HoodieRecord::getRecordKey,
StringUtils.UTF8_LEXICOGRAPHIC_COMPARATOR));
Map<String, String> hfileParams = new HashMap<>();
hfileParams.put(HFILE_COMPRESSION_ALGORITHM_NAME.key(),
writeConfig.getHFileCompressionAlgorithm());
hfileParams.put(HFILE_WITH_BLOOM_FILTER_ENABLED.key(),
Boolean.toString(writeConfig.hfileBloomFilterEnabled()));
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieSortedMergeHandle.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieSortedMergeHandle.java
index 9456d5ce586b..7cc74c40afee 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieSortedMergeHandle.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieSortedMergeHandle.java
@@ -24,6 +24,7 @@ import org.apache.hudi.common.model.HoodieBaseFile;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.StringUtils;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.exception.HoodieUpsertException;
import org.apache.hudi.keygen.BaseKeyGenerator;
@@ -47,7 +48,7 @@ import java.util.Queue;
@NotThreadSafe
public class HoodieSortedMergeHandle<T, I, K, O> extends
HoodieWriteMergeHandle<T, I, K, O> {
- private final Queue<String> newRecordKeysSorted = new PriorityQueue<>();
+ private final Queue<String> newRecordKeysSorted = new
PriorityQueue<>(StringUtils.UTF8_LEXICOGRAPHIC_COMPARATOR);
public HoodieSortedMergeHandle(HoodieWriteConfig config, String instantTime,
HoodieTable<T, I, K, O> hoodieTable,
Iterator<HoodieRecord<T>> recordItr, String
partitionPath, String fileId, TaskContextSupplier taskContextSupplier,
@@ -78,7 +79,7 @@ public class HoodieSortedMergeHandle<T, I, K, O> extends
HoodieWriteMergeHandle<
// To maintain overall sorted order across updates and inserts, write any
new inserts whose keys are less than
// the oldRecord's key.
- while (!newRecordKeysSorted.isEmpty() &&
newRecordKeysSorted.peek().compareTo(key) <= 0) {
+ while (!newRecordKeysSorted.isEmpty() &&
StringUtils.compareUtf8Bytes(newRecordKeysSorted.peek(), key) <= 0) {
String keyToPreWrite = newRecordKeysSorted.remove();
if (keyToPreWrite.equals(key)) {
// will be handled as an update later
diff --git
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkWriteClient.java
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkWriteClient.java
index 90632c823a3f..cd677fa34fea 100644
---
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkWriteClient.java
+++
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkWriteClient.java
@@ -35,6 +35,7 @@ import org.apache.hudi.common.model.TableServiceType;
import org.apache.hudi.common.model.WriteOperationType;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.StringUtils;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.exception.HoodieNotSupportedException;
import org.apache.hudi.index.FlinkHoodieIndexFactory;
@@ -337,7 +338,9 @@ public class HoodieFlinkWriteClient<T>
Map<String, List<HoodieRecord<T>>> preppedRecordsByFileId =
preppedRecords.stream().parallel()
.collect(Collectors.groupingBy(r ->
r.getCurrentLocation().getFileId()));
return preppedRecordsByFileId.values().stream().parallel().map(records -> {
- records.sort(Comparator.comparing(HoodieRecord::getRecordKey));
+ // Only used for the metadata table, whose base files are HFiles ordered
by raw UTF-8 bytes,
+ // so sort by UTF-8 bytes rather than String (UTF-16) order for
non-ASCII / binary keys.
+ records.sort(Comparator.comparing(HoodieRecord::getRecordKey,
StringUtils.UTF8_LEXICOGRAPHIC_COMPARATOR));
HoodieWriteMetadata<List<WriteStatus>> result;
BucketInfo bucketInfo = new BucketInfo(BucketType.INSERT,
records.get(0).getCurrentLocation().getFileId(),
records.get(0).getPartitionPath());
try (AutoCloseableWriteHandle closeableHandle = new
AutoCloseableWriteHandle(bucketInfo, records.iterator(), instantTime, table,
true)) {
diff --git
a/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/metadata/JavaHoodieMetadataBulkInsertPartitioner.java
b/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/metadata/JavaHoodieMetadataBulkInsertPartitioner.java
index 0d81cc91fcff..e2a2137a1252 100644
---
a/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/metadata/JavaHoodieMetadataBulkInsertPartitioner.java
+++
b/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/metadata/JavaHoodieMetadataBulkInsertPartitioner.java
@@ -19,6 +19,7 @@
package org.apache.hudi.metadata;
import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.util.StringUtils;
import org.apache.hudi.table.BulkInsertPartitioner;
import java.util.Comparator;
@@ -39,7 +40,7 @@ public class JavaHoodieMetadataBulkInsertPartitioner<T>
if (records.isEmpty()) {
return records;
}
- records.sort(Comparator.comparing(record ->
record.getKey().getRecordKey()));
+ records.sort(Comparator.comparing(record ->
record.getKey().getRecordKey(), StringUtils.UTF8_LEXICOGRAPHIC_COMPARATOR));
fileId =
HoodieTableMetadataUtil.getFileGroupPrefix(records.get(0).getCurrentLocation().getFileId());
return records;
}
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/main/java/org/apache/hudi/metadata/SparkHoodieMetadataBulkInsertPartitioner.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/metadata/SparkHoodieMetadataBulkInsertPartitioner.java
index 3fd5f346ce11..38ff2f4a77c1 100644
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/metadata/SparkHoodieMetadataBulkInsertPartitioner.java
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/metadata/SparkHoodieMetadataBulkInsertPartitioner.java
@@ -68,7 +68,7 @@ public class SparkHoodieMetadataBulkInsertPartitioner
implements BulkInsertParti
@Override
public JavaRDD<HoodieRecord> repartitionRecords(JavaRDD<HoodieRecord>
records, int outputSparkPartitions) {
Comparator<Tuple2<Integer, String>> keyComparator =
- (Comparator<Tuple2<Integer, String>> & Serializable)(t1, t2) ->
t1._2.compareTo(t2._2);
+ (Comparator<Tuple2<Integer, String>> & Serializable)(t1, t2) ->
StringUtils.compareUtf8Bytes(t1._2, t2._2);
// Partition the records by their file group
JavaRDD<HoodieRecord> partitionedRDD = records
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/BaseSparkCommitActionExecutor.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/BaseSparkCommitActionExecutor.java
index d253210f4b56..06079a511bf8 100644
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/BaseSparkCommitActionExecutor.java
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/BaseSparkCommitActionExecutor.java
@@ -40,6 +40,7 @@ import org.apache.hudi.common.util.CommitUtils;
import org.apache.hudi.common.util.HoodieTimer;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.ReflectionUtils;
+import org.apache.hudi.common.util.StringUtils;
import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.data.HoodieJavaPairRDD;
@@ -326,10 +327,12 @@ public abstract class BaseSparkCommitActionExecutor<T>
extends
if (table.requireSortedRecords()) {
// Partition and sort within each partition as a single step. This is
faster than partitioning first and then
// applying a sort.
+ // requireSortedRecords() is true only for HFile base files, which order
keys by UTF-8 bytes,
+ // not String (UTF-16) order, so sort with the matching comparator.
Comparator<Tuple2<HoodieKey, Option<HoodieRecordLocation>>> comparator =
(Comparator<Tuple2<HoodieKey, Option<HoodieRecordLocation>>> & Serializable)
(t1, t2) -> {
HoodieKey key1 = t1._1;
HoodieKey key2 = t2._1;
- return key1.getRecordKey().compareTo(key2.getRecordKey());
+ return StringUtils.compareUtf8Bytes(key1.getRecordKey(),
key2.getRecordKey());
};
partitionedRDD =
mappedRDD.repartitionAndSortWithinPartitions(partitioner, comparator);
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/main/java/org/apache/hudi/common/table/read/buffer/SortedKeyBasedFileGroupRecordBuffer.java
b/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/SortedKeyBasedFileGroupRecordBuffer.java
index 8fa9aa1dad91..083a7762bec2 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/SortedKeyBasedFileGroupRecordBuffer.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/table/read/buffer/SortedKeyBasedFileGroupRecordBuffer.java
@@ -27,6 +27,7 @@ import org.apache.hudi.common.table.PartialUpdateMode;
import org.apache.hudi.common.table.read.BufferedRecord;
import org.apache.hudi.common.table.read.UpdateProcessor;
import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.StringUtils;
import org.apache.hudi.common.util.ValidationUtils;
import java.io.IOException;
@@ -58,14 +59,18 @@ class SortedKeyBasedFileGroupRecordBuffer<T> extends
KeyBasedFileGroupRecordBuff
@Override
protected void initializeLogRecordIterator() {
- logRecordIterator =
records.values().stream().sorted(Comparator.comparing(BufferedRecord::getRecordKey)).iterator();
+ // This buffer is only used when the base file format is HFile
(requireSortedRecords()), which orders
+ // keys by UTF-8 bytes, not String (UTF-16) order, so sort with the
matching comparator.
+ logRecordIterator = records.values().stream()
+ .sorted(Comparator.comparing(BufferedRecord::getRecordKey,
StringUtils.UTF8_LEXICOGRAPHIC_COMPARATOR))
+ .iterator();
}
@Override
protected boolean hasNextBaseRecord(T baseRecord) throws IOException {
String recordKey =
readerContext.getRecordContext().getRecordKey(baseRecord, readerSchema);
int comparison = 0;
- while (!getLogRecordKeysSorted().isEmpty() && (comparison =
getLogRecordKeysSorted().peek().compareTo(recordKey)) <= 0) {
+ while (!getLogRecordKeysSorted().isEmpty() && (comparison =
StringUtils.compareUtf8Bytes(getLogRecordKeysSorted().peek(), recordKey)) <= 0)
{
String nextLogRecordKey = getLogRecordKeysSorted().poll();
if (comparison == 0) {
break; // Log record key matches the base record key, exit loop after
removing the key from the queue of log record keys
@@ -102,7 +107,8 @@ class SortedKeyBasedFileGroupRecordBuffer<T> extends
KeyBasedFileGroupRecordBuff
private Queue<String> getLogRecordKeysSorted() {
if (logRecordKeysSorted == null) {
- logRecordKeysSorted =
records.keySet().stream().map(Object::toString).collect(Collectors.toCollection(PriorityQueue::new));
+ logRecordKeysSorted = records.keySet().stream().map(Object::toString)
+ .collect(Collectors.toCollection(() -> new
PriorityQueue<>(StringUtils.UTF8_LEXICOGRAPHIC_COMPARATOR)));
}
return logRecordKeysSorted;
}
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/util/HoodieRecordUtils.java
b/hudi-common/src/main/java/org/apache/hudi/common/util/HoodieRecordUtils.java
index 1718092eba68..27337de679cb 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/util/HoodieRecordUtils.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/util/HoodieRecordUtils.java
@@ -224,12 +224,14 @@ public class HoodieRecordUtils {
}
/**
- * Returns an iterator over the input records sorted by record key.
+ * Returns an iterator over the input records sorted by record key in UTF-8
byte order. Callers
+ * use this to feed HFile-backed writers ({@code requireSortedRecords()}),
and HFiles order keys
+ * by their raw UTF-8 bytes, not String (UTF-16) order.
*/
public static <T> Iterator<HoodieRecord<T>>
sortRecordsByRecordKey(Iterator<HoodieRecord<T>> records) {
List<HoodieRecord<T>> sortedRecords = new ArrayList<>();
records.forEachRemaining(sortedRecords::add);
- sortedRecords.sort(Comparator.comparing(HoodieRecord::getRecordKey));
+ sortedRecords.sort(Comparator.comparing(HoodieRecord::getRecordKey,
StringUtils.UTF8_LEXICOGRAPHIC_COMPARATOR));
return sortedRecords.iterator();
}
diff --git
a/hudi-common/src/main/java/org/apache/hudi/core/io/storage/HoodieNativeAvroHFileReader.java
b/hudi-common/src/main/java/org/apache/hudi/core/io/storage/HoodieNativeAvroHFileReader.java
index 23e83474eb6d..0cf5eee29b66 100644
---
a/hudi-common/src/main/java/org/apache/hudi/core/io/storage/HoodieNativeAvroHFileReader.java
+++
b/hudi-common/src/main/java/org/apache/hudi/core/io/storage/HoodieNativeAvroHFileReader.java
@@ -31,6 +31,7 @@ import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.schema.HoodieSchemaField;
import org.apache.hudi.common.util.Lazy;
import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.StringUtils;
import org.apache.hudi.common.util.ValidationUtils;
import org.apache.hudi.common.util.collection.ClosableIterator;
import org.apache.hudi.common.util.collection.CloseableMappingIterator;
@@ -136,8 +137,11 @@ public class HoodieNativeAvroHFileReader extends
HoodieAvroHFileReaderImplBase {
public Set<Pair<String, Long>> filterRowKeys(Set<String> candidateRowKeys) {
try (HFileReader reader = readerFactory.createHFileReader()) {
reader.seekTo();
- // candidateRowKeys must be sorted
- return (candidateRowKeys instanceof TreeSet ? candidateRowKeys : new
TreeSet<>(candidateRowKeys))
+ // candidateRowKeys must be sorted by UTF-8 bytes to match HFile
ordering because the reader
+ // only seeks forward.
+ TreeSet<String> sortedRowKeys = new
TreeSet<>(StringUtils.UTF8_LEXICOGRAPHIC_COMPARATOR);
+ sortedRowKeys.addAll(candidateRowKeys);
+ return sortedRowKeys
.stream()
.filter(k -> {
try {
diff --git
a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadata.java
b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadata.java
index 188ff275dad3..46cf00a5326a 100644
---
a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadata.java
+++
b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadata.java
@@ -56,6 +56,7 @@ import org.apache.hudi.common.table.timeline.HoodieInstant;
import org.apache.hudi.common.table.view.HoodieTableFileSystemView;
import org.apache.hudi.common.util.ConfigUtils;
import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.StringUtils;
import org.apache.hudi.common.util.ValidationUtils;
import org.apache.hudi.common.util.collection.ClosableIterator;
import org.apache.hudi.common.util.collection.ClosableSortedDedupingIterator;
@@ -229,9 +230,9 @@ public class HoodieBackedTableMetadata extends
BaseTableMetadata {
boolean shouldLoadInMemory) {
// Apply key encoding
List<String> sortedKeyPrefixes = new ArrayList<>(rawKeys.map(key ->
key.encode()).collectAsList());
- // Sort the prefixes so that keys are looked up in order
- // Sort must come after encoding.
- Collections.sort(sortedKeyPrefixes);
+ // Sort the prefixes so that keys are looked up in order. Sort must come
after encoding.
+ // Sort by UTF-8 bytes to match the HFile order; the reader seeks forward
without rewinding.
+ sortedKeyPrefixes.sort(StringUtils.UTF8_LEXICOGRAPHIC_COMPARATOR);
// NOTE: Since we partition records to a particular file-group by full
key, we will have
// to scan all file-groups for all key-prefixes as each of these
might contain some
@@ -256,7 +257,10 @@ public class HoodieBackedTableMetadata extends
BaseTableMetadata {
private static TreeSet<String>
getDistinctSortedKeysForSingleSlice(HoodieData<String> keys) {
List<String> keysList = keys.collectAsList();
- return new TreeSet<>(keysList);
+ // Order by UTF-8 bytes to match the HFile order used for point lookups.
+ TreeSet<String> sortedKeys = new
TreeSet<>(StringUtils.UTF8_LEXICOGRAPHIC_COMPARATOR);
+ sortedKeys.addAll(keysList);
+ return sortedKeys;
}
/**
@@ -314,6 +318,9 @@ public class HoodieBackedTableMetadata extends
BaseTableMetadata {
}
distinctSortedKeyIter.forEachRemaining(keysList::add);
}
+ // The shuffle above repartitions/sorts by String (UTF-16) order,
but the HFile reader below
+ // does a forward-only seek in UTF-8 byte order. Re-sort so the two
agree.
+ keysList.sort(StringUtils.UTF8_LEXICOGRAPHIC_COMPARATOR);
FileSlice fileSlice =
fileSlices.get(mappingFunction.apply(keysList.get(0), numFileSlices));
return lookupRecordsItr(partitionName, keysList, fileSlice,
!isSecondaryIndex);
};
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 329b3e382ba8..4dde1b53c37c 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-common/src/test/java/org/apache/hudi/common/util/TestHoodieRecordUtils.java
b/hudi-common/src/test/java/org/apache/hudi/common/util/TestHoodieRecordUtils.java
index 2219f8043760..c76c194be133 100644
---
a/hudi-common/src/test/java/org/apache/hudi/common/util/TestHoodieRecordUtils.java
+++
b/hudi-common/src/test/java/org/apache/hudi/common/util/TestHoodieRecordUtils.java
@@ -71,9 +71,16 @@ class TestHoodieRecordUtils {
@Test
void sortRecordsByRecordKey() {
+ // 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 sort feeds
HFile-backed writers, so
+ // it must produce UTF-8 byte order.
+ String bmpPrivateUseKey = new String(Character.toChars(0xE000)) + "key";
+ String supplementaryKey = new String(Character.toChars(0x20000)) + "key";
List<HoodieRecord<DefaultHoodieRecordPayload>> records = Arrays.asList(
record("key3"),
+ record(supplementaryKey),
record("key1"),
+ record(bmpPrivateUseKey),
record("key2"));
Iterator<HoodieRecord<DefaultHoodieRecordPayload>> sortedRecords =
@@ -81,7 +88,7 @@ class TestHoodieRecordUtils {
List<String> sortedKeys = new ArrayList<>();
sortedRecords.forEachRemaining(record ->
sortedKeys.add(record.getRecordKey()));
- assertEquals(Arrays.asList("key1", "key2", "key3"), sortedKeys);
+ assertEquals(Arrays.asList("key1", "key2", "key3", bmpPrivateUseKey,
supplementaryKey), sortedKeys);
}
@Test
diff --git a/hudi-io/src/main/java/org/apache/hudi/common/util/StringUtils.java
b/hudi-io/src/main/java/org/apache/hudi/common/util/StringUtils.java
index 4f3a651d7914..bd8af56094dc 100644
--- a/hudi-io/src/main/java/org/apache/hudi/common/util/StringUtils.java
+++ b/hudi-io/src/main/java/org/apache/hudi/common/util/StringUtils.java
@@ -21,10 +21,12 @@ package org.apache.hudi.common.util;
import javax.annotation.Nullable;
+import java.io.Serializable;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
@@ -120,6 +122,40 @@ public class StringUtils {
return str.getBytes(StandardCharsets.UTF_8);
}
+ /**
+ * Serializable comparator ordering strings by their unsigned UTF-8 byte
representation. See
+ * {@link #compareUtf8Bytes(String, String)} for the rationale and
null-handling contract.
+ */
+ public static final Comparator<String> UTF8_LEXICOGRAPHIC_COMPARATOR =
+ (Comparator<String> & Serializable) StringUtils::compareUtf8Bytes;
+
+ /**
+ * Compares two strings by their unsigned UTF-8 byte order, matching the
ordering HFiles enforce
+ * (HBase's {@code CellComparatorImpl}). Unlike {@link
String#compareTo(String)} (UTF-16 code unit
+ * order), this stays consistent with HFile ordering for non-ASCII / binary
keys.
+ *
+ * <p>Neither argument may be {@code null}; like {@link
String#compareTo(String)}, a {@code null}
+ * argument throws {@link NullPointerException}.
+ *
+ * <p>Assumes well-formed UTF-16 input: {@code String#getBytes(UTF_8)}
replaces unpaired surrogates
+ * with {@code '?'}, so strings differing only in unpaired surrogates
compare equal.
+ *
+ * <p>Note: encodes both strings to UTF-8 on every call; for very large
sorts consider
+ * pre-encoding keys to byte arrays once and comparing those.
+ */
+ public static int compareUtf8Bytes(String s1, String s2) {
+ byte[] b1 = getUTF8Bytes(s1);
+ byte[] b2 = getUTF8Bytes(s2);
+ int len = Math.min(b1.length, b2.length);
+ for (int i = 0; i < len; i++) {
+ int cmp = (b1[i] & 0xFF) - (b2[i] & 0xFF);
+ if (cmp != 0) {
+ return cmp;
+ }
+ }
+ return b1.length - b2.length;
+ }
+
public static String fromUTF8Bytes(byte[] bytes) {
return fromUTF8Bytes(bytes, 0, bytes.length);
}
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 3be3bfe3f998..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;
@@ -283,4 +288,63 @@ public class TestStringUtils {
assertEquals("abc", StringUtils.stripEnd("abc", ""));
assertEquals("abc", StringUtils.stripEnd("abcabab", "ab"));
}
+
+ @Test
+ public void testCompareUtf8BytesAsciiMatchesStringCompareTo() {
+ // Pure ASCII bytes equal their UTF-16 code unit values, so UTF-8 byte
order and
+ // String.compareTo order coincide.
+ assertEquals(Integer.signum("apple".compareTo("banana")),
Integer.signum(StringUtils.compareUtf8Bytes("apple", "banana")));
+ assertEquals(Integer.signum("banana".compareTo("apple")),
Integer.signum(StringUtils.compareUtf8Bytes("banana", "apple")));
+ assertEquals(0, StringUtils.compareUtf8Bytes("apple", "apple"));
+ }
+
+ @Test
+ public void
testCompareUtf8BytesSupplementaryPairFlipsOrderVsStringCompareTo() {
+ // U+E000 (BMP private-use, UTF-8 lead byte 0xEE) vs U+20000
(supplementary plane, UTF-8 lead byte
+ // 0xF0). In UTF-16, U+20000 is encoded as a surrogate pair starting with
0xD840, which is < 0xE000,
+ // so String.compareTo orders U+20000 first. In UTF-8 byte order 0xF0 >
0xEE, flipping the order --
+ // exactly the pathological shape that breaks HFile's forward-only seek
under String.compareTo.
+ String bmpPrivateUse = new String(Character.toChars(0xE000));
+ String supplementary = new String(Character.toChars(0x20000));
+
+ assertTrue(bmpPrivateUse.compareTo(supplementary) > 0,
+ "String.compareTo should order U+E000 after U+20000 (UTF-16 code unit
order)");
+ assertTrue(StringUtils.compareUtf8Bytes(bmpPrivateUse, supplementary) < 0,
+ "compareUtf8Bytes should order U+E000 before U+20000 (UTF-8 byte
order)");
+ }
+
+ @Test
+ public void testCompareUtf8BytesEmptyPrefixAndIdenticalStrings() {
+ assertTrue(StringUtils.compareUtf8Bytes("", "a") < 0);
+ assertTrue(StringUtils.compareUtf8Bytes("a", "") > 0);
+ assertEquals(0, StringUtils.compareUtf8Bytes("", ""));
+ assertTrue(StringUtils.compareUtf8Bytes("ab", "abc") < 0);
+ assertTrue(StringUtils.compareUtf8Bytes("abc", "ab") > 0);
+ assertEquals(0, StringUtils.compareUtf8Bytes("abc", "abc"));
+ }
+
+ @Test
+ @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));
+ assertTrue(deserialized.compare(bmpPrivateUse, supplementary) < 0,
+ "Deserialized comparator should order U+E000 before U+20000 (UTF-8
byte order)");
+ }
}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/TestHoodieBackedMetadata.java
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/TestHoodieBackedMetadata.java
index f3b3bbf3c4c2..ec0d405fad91 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/TestHoodieBackedMetadata.java
+++
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/TestHoodieBackedMetadata.java
@@ -39,6 +39,7 @@ import org.apache.hudi.common.engine.HoodieEngineContext;
import org.apache.hudi.common.fs.ConsistencyGuardConfig;
import org.apache.hudi.common.fs.FSUtils;
import org.apache.hudi.common.model.FileSlice;
+import org.apache.hudi.common.model.HoodieAvroIndexedRecord;
import org.apache.hudi.common.model.HoodieBaseFile;
import org.apache.hudi.common.model.HoodieCleaningPolicy;
import org.apache.hudi.common.model.HoodieCommitMetadata;
@@ -162,6 +163,7 @@ import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
import java.util.stream.Collectors;
+import java.util.stream.Stream;
import static java.util.Arrays.asList;
import static java.util.Collections.emptyList;
@@ -1944,6 +1946,180 @@ public class TestHoodieBackedMetadata extends
TestHoodieMetadataBase {
testTableOperationsForMetaIndexImpl(writeConfig);
}
+ /**
+ * Record index bootstrap over binary / non-ASCII record keys must succeed.
The RI HFile orders
+ * keys by their raw UTF-8 bytes, so the bulk-insert partitioner must sort
by UTF-8 bytes too;
+ * sorting by {@link String#compareTo(String)} (UTF-16) lays the HFile
entries out of order
+ * relative to their UTF-8 bytes.
+ *
+ * <p>The failure is read-side, not write-side: the native {@code
HFileWriterImpl.append} does no
+ * key-order validation, so a mis-sorted HFile is still written
successfully. The forward-only
+ * HFile reader ({@code HFileReaderImpl.seekTo}) then either throws {@code
IllegalStateException}
+ * on a backward seek or silently misses keys, which the read-back count
assertion at the end of
+ * this test catches.
+ *
+ * <p>{@code riFileGroupCount == 1} covers the single-slice lookup path;
{@code riFileGroupCount == 4}
+ * covers the multi-slice {@code mapGroupsByKey} lookup path in
+ * {@code HoodieBackedTableMetadata#lookupIndexRecords}, which repartitions
keys in String/UTF-16
+ * order before doing a forward-only HFile seek in UTF-8 order.
+ */
+ @ParameterizedTest
+ @ValueSource(ints = {1, 4})
+ public void testRecordIndexBootstrapWithBinaryRecordKeys(int
riFileGroupCount) throws Exception {
+ init(COPY_ON_WRITE, true);
+ HoodieSparkEngineContext engineContext = new HoodieSparkEngineContext(jsc);
+
+ // First commit with the record index disabled: write base files with
binary record keys.
+ List<HoodieRecord> records =
generateRecordsWithBinaryKeys(WriteClientTestUtils.createNewInstantTime(), 0,
200);
+ HoodieWriteConfig firstConfig = getWriteConfigBuilder(true, true,
false).build();
+ String firstCommitTime = WriteClientTestUtils.createNewInstantTime();
+ try (SparkRDDWriteClient client = new SparkRDDWriteClient(engineContext,
firstConfig)) {
+ WriteClientTestUtils.startCommitWithTime(client, firstCommitTime);
+ List<WriteStatus> writeStatuses = client.insert(jsc.parallelize(records,
1), firstCommitTime).collect();
+ assertNoWriteErrors(writeStatuses);
+ client.commit(firstCommitTime, jsc.parallelize(writeStatuses));
+ }
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+
assertFalse(metaClient.getTableConfig().isMetadataPartitionAvailable(RECORD_INDEX));
+
+ // Enable the record index. The next commit triggers the bootstrap,
reading the binary keys from
+ // the base files above. One file group puts all keys in a single HFile;
more than one file group
+ // exercises the multi-slice lookup path on read-back.
+ HoodieWriteConfig riConfig = getWriteConfigBuilder(false, true, false)
+ .withMetadataConfig(HoodieMetadataConfig.newBuilder()
+ .enable(true)
+ .withEnableGlobalRecordLevelIndex(true)
+ .withRecordIndexFileGroupCount(riFileGroupCount, riFileGroupCount)
+ .build())
+ .build();
+
+ String secondCommitTime = WriteClientTestUtils.createNewInstantTime();
+ // Disjoint key range so the bootstrapped keys are not mutated.
+ List<HoodieRecord> secondBatch =
generateRecordsWithBinaryKeys(secondCommitTime, 1000, 20);
+ try (SparkRDDWriteClient client = new SparkRDDWriteClient(engineContext,
riConfig)) {
+ WriteClientTestUtils.startCommitWithTime(client, secondCommitTime);
+ // Without the fix the mis-sorted record-index HFile is still written;
the failure surfaces on
+ // read-back below, so the write itself is expected to succeed here.
+ List<WriteStatus> writeStatuses =
client.insert(jsc.parallelize(secondBatch, 1), secondCommitTime).collect();
+ assertNoWriteErrors(writeStatuses);
+ client.commit(secondCommitTime, jsc.parallelize(writeStatuses));
+ }
+
+ // The record index partition should exist and resolve every key: the
bootstrapped keys live in
+ // the record-index base HFiles, while the second-batch keys still sit in
un-compacted metadata
+ // log files at this point, so the lookup covers the log-side seek path
with binary keys too.
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+
assertTrue(metaClient.getTableConfig().isMetadataPartitionAvailable(RECORD_INDEX));
+ HoodieTableMetadata metadataReader =
metaClient.getTableFormat().getMetadataFactory().create(
+ context, storage, riConfig.getMetadataConfig(),
riConfig.getBasePath());
+ List<String> allKeys = Stream.concat(records.stream(),
secondBatch.stream())
+ .map(HoodieRecord::getRecordKey).collect(Collectors.toList());
+ // With more than one file group, readRecordIndexLocationsWithKeys
triggers the mapGroupsByKey
+ // multi-slice path.
+ HoodiePairData<String, HoodieRecordGlobalLocation> recordIndexData =
metadataReader
+ .readRecordIndexLocationsWithKeys(HoodieListData.eager(allKeys));
+ try {
+ Map<String, HoodieRecordGlobalLocation> result =
HoodieDataUtils.dedupeAndCollectAsMap(recordIndexData);
+ assertEquals(allKeys.size(), result.size(),
+ "Record index should resolve every binary key, bootstrapped or still
in a metadata log file.");
+ } finally {
+ recordIndexData.unpersistWithDependencies();
+ }
+ }
+
+ /**
+ * Generates {@code count} records with binary record keys interleaving
U+E000 (BMP) and U+20000
+ * (supplementary) prefixes, whose UTF-16 char order is the reverse of their
UTF-8 byte order.
+ */
+ private List<HoodieRecord> generateRecordsWithBinaryKeys(String commitTime,
int startIndex, int count) {
+ List<HoodieRecord> baseRecords = dataGen.generateInserts(commitTime,
count);
+ String[] binaryPrefixes = {new String(Character.toChars(0xE000)), new
String(Character.toChars(0x20000))};
+ List<HoodieRecord> binaryRecords = new ArrayList<>(count);
+ for (int i = 0; i < count; i++) {
+ HoodieRecord baseRecord = baseRecords.get(i);
+ int index = startIndex + i;
+ String binaryKey = binaryPrefixes[index % binaryPrefixes.length] +
String.format("%08d", index);
+ binaryRecords.add(new HoodieAvroIndexedRecord(
+ new HoodieKey(binaryKey, baseRecord.getPartitionPath()),
+ (IndexedRecord) baseRecord.getData()));
+ }
+ return binaryRecords;
+ }
+
+ /**
+ * Same as {@link #testRecordIndexBootstrapWithBinaryRecordKeys(int)} but
forces an MDT compaction after
+ * bootstrap, exercising the {@code BaseCreateHandle} / {@code
SortedKeyBasedFileGroupRecordBuffer}
+ * sort-order paths hit when the record-index base HFile is rewritten.
+ */
+ @Test
+ public void testRecordIndexBootstrapWithBinaryRecordKeysAfterCompaction()
throws Exception {
+ init(COPY_ON_WRITE, true);
+ HoodieSparkEngineContext engineContext = new HoodieSparkEngineContext(jsc);
+
+ List<HoodieRecord> records =
generateRecordsWithBinaryKeys(WriteClientTestUtils.createNewInstantTime(), 0,
200);
+ HoodieWriteConfig firstConfig = getWriteConfigBuilder(true, true,
false).build();
+ String firstCommitTime = WriteClientTestUtils.createNewInstantTime();
+ try (SparkRDDWriteClient client = new SparkRDDWriteClient(engineContext,
firstConfig)) {
+ WriteClientTestUtils.startCommitWithTime(client, firstCommitTime);
+ List<WriteStatus> writeStatuses = client.insert(jsc.parallelize(records,
1), firstCommitTime).collect();
+ assertNoWriteErrors(writeStatuses);
+ client.commit(firstCommitTime, jsc.parallelize(writeStatuses));
+ }
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+
assertFalse(metaClient.getTableConfig().isMetadataPartitionAvailable(RECORD_INDEX));
+
+ // A single delta commit is enough to trigger compaction on the very next
delta commit.
+ HoodieWriteConfig riConfig = getWriteConfigBuilder(false, true, false)
+ .withMetadataConfig(HoodieMetadataConfig.newBuilder()
+ .enable(true)
+ .withEnableGlobalRecordLevelIndex(true)
+ .withRecordIndexFileGroupCount(1, 1)
+ .withMaxNumDeltaCommitsBeforeCompaction(1)
+ .build())
+ .build();
+
+ List<HoodieRecord> allKeys = new ArrayList<>(records);
+ try (SparkRDDWriteClient client = new SparkRDDWriteClient(engineContext,
riConfig)) {
+ // Bootstrap: writes the initial record-index HFile from the binary keys
above.
+ String secondCommitTime = WriteClientTestUtils.createNewInstantTime();
+ List<HoodieRecord> secondBatch =
generateRecordsWithBinaryKeys(secondCommitTime, 1000, 20);
+ WriteClientTestUtils.startCommitWithTime(client, secondCommitTime);
+ // The mis-sorted bootstrap write succeeds; key ordering is validated by
the read-back below.
+ List<WriteStatus> secondWriteStatuses =
client.insert(jsc.parallelize(secondBatch, 1), secondCommitTime).collect();
+ assertNoWriteErrors(secondWriteStatuses);
+ client.commit(secondCommitTime, jsc.parallelize(secondWriteStatuses));
+ allKeys.addAll(secondBatch);
+
+ // The next delta commit on the record index partition triggers
compaction, rewriting the base
+ // HFile via BaseCreateHandle / SortedKeyBasedFileGroupRecordBuffer.
+ String thirdCommitTime = WriteClientTestUtils.createNewInstantTime();
+ List<HoodieRecord> thirdBatch =
generateRecordsWithBinaryKeys(thirdCommitTime, 2000, 20);
+ WriteClientTestUtils.startCommitWithTime(client, thirdCommitTime);
+ // The compaction rewrite of the base HFile also succeeds; ordering is
validated on read-back.
+ List<WriteStatus> thirdWriteStatuses =
client.insert(jsc.parallelize(thirdBatch, 1), thirdCommitTime).collect();
+ assertNoWriteErrors(thirdWriteStatuses);
+ client.commit(thirdCommitTime, jsc.parallelize(thirdWriteStatuses));
+ allKeys.addAll(thirdBatch);
+ }
+
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+ HoodieTableMetadata metadataReader =
metaClient.getTableFormat().getMetadataFactory().create(
+ context, storage, riConfig.getMetadataConfig(),
riConfig.getBasePath());
+ assertTrue(metadataReader.getLatestCompactionTime().isPresent(),
+ "Record index partition should have been compacted by now.");
+
+ List<String> allRecordKeys =
allKeys.stream().map(HoodieRecord::getRecordKey).collect(Collectors.toList());
+ HoodiePairData<String, HoodieRecordGlobalLocation> recordIndexData =
metadataReader
+ .readRecordIndexLocationsWithKeys(HoodieListData.eager(allRecordKeys));
+ try {
+ Map<String, HoodieRecordGlobalLocation> result =
HoodieDataUtils.dedupeAndCollectAsMap(recordIndexData);
+ assertEquals(allRecordKeys.size(), result.size(),
+ "Record index should resolve every binary key after the record index
partition has been compacted.");
+ } finally {
+ recordIndexData.unpersistWithDependencies();
+ }
+ }
+
/**
* First attempt at bootstrap failed but the file slices get created. The
next bootstrap should continue successfully.
*/
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/io/TestMergeHandle.java
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/io/TestMergeHandle.java
index 2e6db79cb3ef..2e1cee6b9f14 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/io/TestMergeHandle.java
+++
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/io/TestMergeHandle.java
@@ -57,6 +57,7 @@ import org.apache.hudi.common.util.DateTimeUtils;
import org.apache.hudi.common.util.HoodieRecordUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.ParquetUtils;
+import org.apache.hudi.common.util.StringUtils;
import org.apache.hudi.common.util.collection.ClosableIterator;
import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.config.HoodieWriteConfig;
@@ -77,6 +78,7 @@ import org.junit.jupiter.params.provider.ValueSource;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
@@ -198,6 +200,57 @@ public class TestMergeHandle extends BaseTestHandle {
validateSecondaryIndexStatsContent(writeStatus, numUpdates, numDeletes);
}
+ @Test
+ public void testSortedMergeHandleWritesBinaryKeysInUtf8Order() throws
Exception {
+ // Drives HoodieSortedMergeHandle directly against a Parquet base file
(requireSortedRecords() is
+ // false), validating comparator ordering only; does not cover the
HoodieMergeHandleFactory
+ // selection path for HFILE base-format tables.
+ // delete and recreate
+ metaClient.getStorage().deleteDirectory(metaClient.getBasePath());
+ Properties properties = new Properties();
+ properties.put(KeyGeneratorOptions.RECORDKEY_FIELD_NAME.key(), "_row_key");
+ properties.put(KeyGeneratorOptions.PARTITIONPATH_FIELD_NAME.key(),
"partition_path");
+ properties.put(HoodieWriteConfig.PRECOMBINE_FIELD_NAME.key(),
ORDERING_FIELD);
+ initMetaClient(getTableType(), properties);
+
+ HoodieWriteConfig config = getHoodieWriteConfigBuilder().build();
+ HoodieSparkTable.create(config, new HoodieLocalEngineContext(storageConf),
metaClient);
+
+ String partitionPath = HoodieTestDataGenerator.DEFAULT_PARTITION_PATHS[0];
+ HoodieTestDataGenerator dataGenerator = new HoodieTestDataGenerator(new
String[] {partitionPath});
+
+ // These two keys sort in opposite order under UTF-16 (String#compareTo)
vs UTF-8 bytes (HFile/MDT order).
+ String supplementaryKey = "😀_record";
+ String bmpHighKey = new String(Character.toChars(0xFFFD)) + "_record";
+ assertTrue(supplementaryKey.compareTo(bmpHighKey) < 0);
+ assertTrue(StringUtils.compareUtf8Bytes(bmpHighKey, supplementaryKey) < 0);
+
+ // Base file has the UTF-8-smaller key.
+ List<HoodieRecord> baseRecords =
withRowKey(dataGenerator.generateInserts("000", 1), bmpHighKey, partitionPath);
+ SparkRDDWriteClient client = getHoodieWriteClient(config);
+ String instantTime = client.startCommit();
+ JavaRDD<WriteStatus> statuses = client.upsert(jsc.parallelize(baseRecords,
1), instantTime);
+ client.commit(instantTime, statuses, Option.empty(), COMMIT_ACTION,
Collections.emptyMap(), Option.empty());
+
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+ HoodieSparkCopyOnWriteTable table = (HoodieSparkCopyOnWriteTable)
HoodieSparkCopyOnWriteTable.create(config, context, metaClient);
+ HoodieFileGroup fileGroup =
table.getFileSystemView().getAllFileGroups(partitionPath).collect(Collectors.toList()).get(0);
+ String fileId = fileGroup.getFileGroupId().getFileId();
+
+ // Merge the UTF-8-larger key in directly via HoodieSortedMergeHandle.
+ List<HoodieRecord> newRecords =
withRowKey(dataGenerator.generateInserts("001", 1), supplementaryKey,
partitionPath);
+ HoodieSortedMergeHandle mergeHandle = new HoodieSortedMergeHandle(
+ config, "001", table, newRecords.iterator(), partitionPath, fileId,
new LocalTaskContextSupplier(), Option.empty());
+ mergeHandle.doMerge();
+ WriteStatus writeStatus = (WriteStatus) mergeHandle.close().get(0);
+
+ String fullPath = metaClient.getBasePath() + "/" +
writeStatus.getStat().getPath();
+ List<GenericRecord> actualRecords = new
ParquetUtils().readAvroRecords(metaClient.getStorage(), new
StoragePath(fullPath));
+ List<String> actualKeysInOrder = actualRecords.stream().map(r ->
r.get("_row_key").toString()).collect(Collectors.toList());
+ // bmpHighKey must come first when sorted by UTF-8 bytes.
+ assertEquals(Arrays.asList(bmpHighKey, supplementaryKey),
actualKeysInOrder);
+ }
+
@Test
void testWriteFailures() throws Exception {
// delete and recreate
@@ -603,6 +656,12 @@ public class TestMergeHandle extends BaseTestHandle {
}).collect(Collectors.toList());
}
+ private List<HoodieRecord> withRowKey(List<HoodieRecord> records, String
rowKey, String partitionPath) {
+ GenericRecord genericRecord = (GenericRecord) ((SerializableIndexedRecord)
records.get(0).getData()).getData();
+ genericRecord.put("_row_key", rowKey);
+ return getHoodieRecords(OverwriteWithLatestAvroPayload.class.getName(),
Collections.singletonList(genericRecord), partitionPath, false);
+ }
+
private void setCurLocation(List<HoodieRecord> records, String fileId,
String instantTime) {
records.forEach(record -> record.setCurrentLocation(new
HoodieRecordLocation(instantTime, fileId)));
}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestSecondaryIndexPruning.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestSecondaryIndexPruning.scala
index 7d6640b6ea6d..370b0f7467c5 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestSecondaryIndexPruning.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestSecondaryIndexPruning.scala
@@ -230,6 +230,78 @@ class TestSecondaryIndexPruning extends
SparkClientFunctionalTestHarnessScala {
}
}
+ @Test
+ def testSecondaryIndexWithNonAsciiSecondaryKeyValues(): Unit = {
+ var hudiOpts = commonOpts
+ hudiOpts = hudiOpts ++ Map(
+ DataSourceWriteOptions.TABLE_TYPE.key -> COW_TABLE_TYPE_OPT_VAL,
+ DataSourceReadOptions.ENABLE_DATA_SKIPPING.key -> "true")
+ tableName += "test_secondary_index_non_ascii_partitioned_cow"
+
+ // These two secondary values have UTF-16 order reversed vs their raw
UTF-8 byte order:
+ // U+E000 encodes to bytes EE 80 80 while U+20000 encodes to F0 A0 80 80,
so U+E000 sorts
+ // before U+20000 by UTF-8 bytes; but in UTF-16 the U+20000 surrogate pair
(D840 DC00)
+ // sorts before the single U+E000 code unit. The fix orders metadata keys
by UTF-8 bytes,
+ // matching HFile, so both lookups must still resolve to the correct row.
+ val bmpVal = new String(Character.toChars(0xE000)) + "acme"
+ val astralVal = new String(Character.toChars(0x20000)) + "acme"
+ val asciiVal = "acme"
+
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | ts bigint,
+ | record_key_col string,
+ | not_record_key_col string,
+ | partition_key_col string
+ |) using hudi
+ | options (
+ | primaryKey ='record_key_col',
+ | type = 'cow',
+ | hoodie.metadata.enable = 'true',
+ | hoodie.metadata.record.index.enable = 'true',
+ | hoodie.datasource.write.recordkey.field = 'record_key_col',
+ | hoodie.enable.data.skipping = 'true',
+ | hoodie.datasource.write.payload.class =
"org.apache.hudi.common.model.OverwriteWithLatestAvroPayload"
+ | )
+ | partitioned by(partition_key_col)
+ | location '$basePath'
+ """.stripMargin)
+ // small file limit 0 so each insert lands in its own file, giving data
skipping something to prune
+ withSQLConf("hoodie.parquet.small.file.limit" -> "0") {
+ spark.sql(s"insert into $tableName values(1, 'row1', '$bmpVal', 'p1')")
+ spark.sql(s"insert into $tableName values(2, 'row2', '$astralVal',
'p2')")
+ spark.sql(s"insert into $tableName values(3, 'row3', '$asciiVal', 'p3')")
+ // create secondary index on the column holding the non-ascii values
+ spark.sql(s"create index idx_not_record_key_col on $tableName
(not_record_key_col)")
+ metaClient = HoodieTableMetaClient.builder()
+ .setBasePath(basePath)
+ .setConf(HoodieTestUtils.getDefaultStorageConf)
+ .build()
+
assert(metaClient.getTableConfig.getMetadataPartitions.contains("secondary_index_idx_not_record_key_col"))
+ // non-ascii secondary values must be preserved verbatim in the
secondary index records
+ checkAnswer(s"select key from hudi_metadata('$basePath') where type=7")(
+ Seq(bmpVal + SECONDARY_INDEX_RECORD_KEY_SEPARATOR + "row1"),
+ Seq(astralVal + SECONDARY_INDEX_RECORD_KEY_SEPARATOR + "row2"),
+ Seq(asciiVal + SECONDARY_INDEX_RECORD_KEY_SEPARATOR + "row3")
+ )
+ withSQLConf("hoodie.metadata.enable" -> "true",
+ "hoodie.enable.data.skipping" -> "true",
+ "hoodie.fileIndex.dataSkippingFailureMode" -> "strict") {
+ // each non-ascii equality predicate must resolve to exactly its own
row via the SI prefix lookup
+ checkAnswer(s"select ts, record_key_col, not_record_key_col,
partition_key_col from $tableName where not_record_key_col = '$bmpVal'")(
+ Seq(1, "row1", bmpVal, "p1")
+ )
+ checkAnswer(s"select ts, record_key_col, not_record_key_col,
partition_key_col from $tableName where not_record_key_col = '$astralVal'")(
+ Seq(2, "row2", astralVal, "p2")
+ )
+ // data skipping must prune files using the non-ascii secondary keys
+ verifyFilePruning(hudiOpts, EqualTo(attribute("not_record_key_col"),
Literal(bmpVal)))
+ verifyFilePruning(hudiOpts, EqualTo(attribute("not_record_key_col"),
Literal(astralVal)))
+ }
+ }
+ }
+
@Test
def testCreateAndDropSecondaryIndex(): Unit = {
var hudiOpts = commonOpts
