You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by si...@apache.org on 2025/07/24 01:26:19 UTC
(hudi) branch master updated: [HUDI-9548] Secondary index lookup should do escape -> hash -> sort for read write (#13521)
This is an automated email from the ASF dual-hosted git repository.
sivabalan 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 ff60d3213651 [HUDI-9548] Secondary index lookup should do escape -> hash -> sort for read write (#13521)
ff60d3213651 is described below
commit ff60d32136516bb8f652471db9d2373b00ded458
Author: Davis-Zhang-Onehouse <16...@users.noreply.github.com>
AuthorDate: Wed Jul 23 18:26:12 2025 -0700
[HUDI-9548] Secondary index lookup should do escape -> hash -> sort for read write (#13521)
---
.../hudi/client/TestJavaHoodieBackedMetadata.java | 7 +-
.../client/utils/SparkMetadataWriterUtils.java | 3 +-
.../hudi/common/table/view/NoOpTableMetadata.java | 2 +-
.../apache/hudi/metadata/BaseTableMetadata.java | 10 ++-
.../metadata/FileSystemBackedTableMetadata.java | 2 +-
.../hudi/metadata/HoodieBackedTableMetadata.java | 94 ++++++++++++++--------
.../apache/hudi/metadata/HoodieTableMetadata.java | 2 +-
.../hudi/metadata/HoodieTableMetadataUtil.java | 8 +-
.../hudi/metadata/SecondaryIndexKeyUtils.java | 8 +-
.../hudi/metadata/TestHoodieTableMetadataUtil.java | 24 ++++--
.../apache/hudi/source/stats/FileStatsIndex.java | 4 +-
.../org/apache/hudi/ColumnStatsIndexSupport.scala | 2 +-
.../org/apache/hudi/ExpressionIndexSupport.scala | 4 +-
.../apache/hudi/PartitionStatsIndexSupport.scala | 2 +-
.../hudi/functional/TestHoodieBackedMetadata.java | 9 ++-
15 files changed, 115 insertions(+), 66 deletions(-)
diff --git a/hudi-client/hudi-java-client/src/test/java/org/apache/hudi/client/TestJavaHoodieBackedMetadata.java b/hudi-client/hudi-java-client/src/test/java/org/apache/hudi/client/TestJavaHoodieBackedMetadata.java
index cd55d027d6f5..0c9e4df761da 100644
--- a/hudi-client/hudi-java-client/src/test/java/org/apache/hudi/client/TestJavaHoodieBackedMetadata.java
+++ b/hudi-client/hudi-java-client/src/test/java/org/apache/hudi/client/TestJavaHoodieBackedMetadata.java
@@ -164,6 +164,7 @@ import static org.apache.hudi.common.testutils.HoodieTestUtils.INSTANT_FILE_NAME
import static org.apache.hudi.common.testutils.HoodieTestUtils.INSTANT_GENERATOR;
import static org.apache.hudi.config.HoodieCompactionConfig.INLINE_COMPACT_NUM_DELTA_COMMITS;
import static org.apache.hudi.metadata.HoodieTableMetadata.getMetadataTableBasePath;
+import static org.apache.hudi.metadata.HoodieTableMetadataUtil.IDENTITY_ENCODING;
import static org.apache.hudi.metadata.HoodieTableMetadataUtil.deleteMetadataTable;
import static org.apache.hudi.metadata.MetadataPartitionType.COLUMN_STATS;
import static org.apache.hudi.metadata.MetadataPartitionType.FILES;
@@ -1473,7 +1474,7 @@ public class TestJavaHoodieBackedMetadata extends TestHoodieMetadataBase {
ColumnIndexID columnIndexID = new ColumnIndexID(HoodieRecord.RECORD_KEY_METADATA_FIELD);
List<HoodieRecord<HoodieMetadataPayload>> result = tableMetadata.getRecordsByKeyPrefixes(
HoodieListData.lazy(Collections.singletonList(columnIndexID.asBase64EncodedString())),
- MetadataPartitionType.COLUMN_STATS.getPartitionPath(), true, Option.empty()).collectAsList();
+ MetadataPartitionType.COLUMN_STATS.getPartitionPath(), true, IDENTITY_ENCODING).collectAsList();
// there are 3 partitions in total and 2 commits. total entries should be 6.
assertEquals(result.size(), 6);
@@ -1485,7 +1486,7 @@ public class TestJavaHoodieBackedMetadata extends TestHoodieMetadataBase {
PartitionIndexID partitionIndexID = new PartitionIndexID(HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH);
result = tableMetadata.getRecordsByKeyPrefixes(
HoodieListData.lazy(Collections.singletonList(columnIndexID.asBase64EncodedString().concat(partitionIndexID.asBase64EncodedString()))),
- MetadataPartitionType.COLUMN_STATS.getPartitionPath(), true, Option.empty()).collectAsList();
+ MetadataPartitionType.COLUMN_STATS.getPartitionPath(), true, IDENTITY_ENCODING).collectAsList();
// 1 partition and 2 commits. total entries should be 2.
assertEquals(result.size(), 2);
result.forEach(entry -> {
@@ -1505,7 +1506,7 @@ public class TestJavaHoodieBackedMetadata extends TestHoodieMetadataBase {
columnIndexID = new ColumnIndexID(HoodieRecord.COMMIT_TIME_METADATA_FIELD);
result = tableMetadata.getRecordsByKeyPrefixes(
HoodieListData.lazy(Collections.singletonList(columnIndexID.asBase64EncodedString().concat(partitionIndexID.asBase64EncodedString()))),
- MetadataPartitionType.COLUMN_STATS.getPartitionPath(), true, Option.empty()).collectAsList();
+ MetadataPartitionType.COLUMN_STATS.getPartitionPath(), true, IDENTITY_ENCODING).collectAsList();
// 1 partition and 2 commits. total entries should be 2.
assertEquals(result.size(), 2);
diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/utils/SparkMetadataWriterUtils.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/utils/SparkMetadataWriterUtils.java
index 3920ea2c50ae..cddb24c23981 100644
--- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/utils/SparkMetadataWriterUtils.java
+++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/utils/SparkMetadataWriterUtils.java
@@ -108,6 +108,7 @@ import static org.apache.hudi.metadata.HoodieMetadataPayload.COLUMN_STATS_FIELD_
import static org.apache.hudi.metadata.HoodieMetadataPayload.COLUMN_STATS_FIELD_VALUE_COUNT;
import static org.apache.hudi.metadata.HoodieMetadataPayload.createBloomFilterMetadataRecord;
import static org.apache.hudi.metadata.HoodieMetadataPayload.createColumnStatsRecords;
+import static org.apache.hudi.metadata.HoodieTableMetadataUtil.IDENTITY_ENCODING;
import static org.apache.hudi.metadata.HoodieTableMetadataUtil.PARTITION_NAME_BLOOM_FILTERS;
import static org.apache.hudi.metadata.HoodieTableMetadataUtil.PARTITION_NAME_COLUMN_STATS;
import static org.apache.hudi.metadata.HoodieTableMetadataUtil.getFileSystemViewForMetadataTable;
@@ -401,7 +402,7 @@ public class SparkMetadataWriterUtils {
// Fetch EI column stat records for above files
List<HoodieColumnRangeMetadata<Comparable>> partitionColumnMetadata =
tableMetadata.getRecordsByKeyPrefixes(
- HoodieListData.lazy(HoodieTableMetadataUtil.generateKeyPrefixes(validColumnsToIndex, partitionName)), indexPartition, false, Option.empty())
+ HoodieListData.lazy(HoodieTableMetadataUtil.generateKeyPrefixes(validColumnsToIndex, partitionName)), indexPartition, false, IDENTITY_ENCODING)
// schema and properties are ignored in getInsertValue, so simply pass as null
.map(record -> record.getData().getInsertValue(null, null))
.filter(Option::isPresent)
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/view/NoOpTableMetadata.java b/hudi-common/src/main/java/org/apache/hudi/common/table/view/NoOpTableMetadata.java
index 6f0b9d0695c3..6c7cc33f934f 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/view/NoOpTableMetadata.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/view/NoOpTableMetadata.java
@@ -118,7 +118,7 @@ class NoOpTableMetadata implements HoodieTableMetadata {
public HoodieData<HoodieRecord<HoodieMetadataPayload>> getRecordsByKeyPrefixes(HoodieData<String> keyPrefixes,
String partitionName,
boolean shouldLoadInMemory,
- Option<SerializableFunctionUnchecked<String, String>> keyEncoder) {
+ SerializableFunctionUnchecked<String, String> keyEncoder) {
throw new HoodieMetadataException("Unsupported operation: getRecordsByKeyPrefixes!");
}
diff --git a/hudi-common/src/main/java/org/apache/hudi/metadata/BaseTableMetadata.java b/hudi-common/src/main/java/org/apache/hudi/metadata/BaseTableMetadata.java
index 0b071b5bfd5c..1519e38de244 100644
--- a/hudi-common/src/main/java/org/apache/hudi/metadata/BaseTableMetadata.java
+++ b/hudi-common/src/main/java/org/apache/hudi/metadata/BaseTableMetadata.java
@@ -66,6 +66,8 @@ import java.util.Set;
import java.util.function.Function;
import java.util.stream.Collectors;
+import static org.apache.hudi.metadata.HoodieTableMetadataUtil.IDENTITY_ENCODING;
+
/**
* Abstract class for implementing common table metadata operations.
*/
@@ -210,7 +212,7 @@ public abstract class BaseTableMetadata extends AbstractHoodieTableMetadata {
List<String> partitionIDFileIDStringsList = new ArrayList<>(partitionIDFileIDStrings);
Map<String, HoodieRecord<HoodieMetadataPayload>> hoodieRecords =
HoodieDataUtils.dedupeAndCollectAsMap(
- getRecordsByKeys(HoodieListData.eager(partitionIDFileIDStringsList), metadataPartitionName, Option.empty()));
+ getRecordsByKeys(HoodieListData.eager(partitionIDFileIDStringsList), metadataPartitionName, IDENTITY_ENCODING));
metrics.ifPresent(m -> m.updateMetrics(HoodieMetadataMetrics.LOOKUP_BLOOM_FILTERS_METADATA_STR, timer.endTimer()));
metrics.ifPresent(m -> m.setMetric(HoodieMetadataMetrics.LOOKUP_BLOOM_FILTERS_FILE_COUNT_STR, partitionIDFileIDStringsList.size()));
@@ -330,7 +332,7 @@ public abstract class BaseTableMetadata extends AbstractHoodieTableMetadata {
Map<String, HoodieRecord<HoodieMetadataPayload>> partitionIdRecordPairs =
HoodieDataUtils.dedupeAndCollectAsMap(
getRecordsByKeys(HoodieListData.eager(new ArrayList<>(partitionIdToPathMap.keySet())),
- MetadataPartitionType.FILES.getPartitionPath(), Option.empty()));
+ MetadataPartitionType.FILES.getPartitionPath(), IDENTITY_ENCODING));
metrics.ifPresent(
m -> m.updateMetrics(HoodieMetadataMetrics.LOOKUP_FILES_STR, timer.endTimer()));
@@ -387,7 +389,7 @@ public abstract class BaseTableMetadata extends AbstractHoodieTableMetadata {
Map<String, HoodieRecord<HoodieMetadataPayload>> hoodieRecords =
HoodieDataUtils.dedupeAndCollectAsMap(
getRecordsByKeys(
- HoodieListData.eager(columnStatKeylist), MetadataPartitionType.COLUMN_STATS.getPartitionPath(), Option.empty()));
+ HoodieListData.eager(columnStatKeylist), MetadataPartitionType.COLUMN_STATS.getPartitionPath(), IDENTITY_ENCODING));
metrics.ifPresent(m -> m.updateMetrics(HoodieMetadataMetrics.LOOKUP_COLUMN_STATS_METADATA_STR, timer.endTimer()));
Map<Pair<String, String>, List<HoodieMetadataColumnStats>> fileToColumnStatMap = new HashMap<>();
for (final Map.Entry<String, HoodieRecord<HoodieMetadataPayload>> entry : hoodieRecords.entrySet()) {
@@ -436,7 +438,7 @@ public abstract class BaseTableMetadata extends AbstractHoodieTableMetadata {
* @return A collection of pairs (key -> record)
*/
public abstract HoodiePairData<String, HoodieRecord<HoodieMetadataPayload>> getRecordsByKeys(
- HoodieData<String> keys, String partitionName, Option<SerializableFunctionUnchecked<String, String>> keyEncodingFn);
+ HoodieData<String> keys, String partitionName, SerializableFunctionUnchecked<String, String> keyEncodingFn);
/**
* Returns a collection of pairs (secondary-key -> set-of-record-keys) for the provided secondary keys.
diff --git a/hudi-common/src/main/java/org/apache/hudi/metadata/FileSystemBackedTableMetadata.java b/hudi-common/src/main/java/org/apache/hudi/metadata/FileSystemBackedTableMetadata.java
index 06097655e7ca..6b01180696af 100644
--- a/hudi-common/src/main/java/org/apache/hudi/metadata/FileSystemBackedTableMetadata.java
+++ b/hudi-common/src/main/java/org/apache/hudi/metadata/FileSystemBackedTableMetadata.java
@@ -314,7 +314,7 @@ public class FileSystemBackedTableMetadata extends AbstractHoodieTableMetadata {
public HoodieData<HoodieRecord<HoodieMetadataPayload>> getRecordsByKeyPrefixes(HoodieData<String> keyPrefixes,
String partitionName,
boolean shouldLoadInMemory,
- Option<SerializableFunctionUnchecked<String, String>> keyEncoder) {
+ SerializableFunctionUnchecked<String, String> keyEncoder) {
throw new HoodieMetadataException("Unsupported operation: getRecordsByKeyPrefixes!");
}
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 79964298cf1b..8d199f3ecff5 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
@@ -95,12 +95,12 @@ import static org.apache.hudi.common.config.HoodieMemoryConfig.SPILLABLE_MAP_BAS
import static org.apache.hudi.common.config.HoodieMetadataConfig.DEFAULT_METADATA_ENABLE_FULL_SCAN_LOG_FILES;
import static org.apache.hudi.common.util.ValidationUtils.checkState;
import static org.apache.hudi.metadata.HoodieMetadataPayload.KEY_FIELD_NAME;
+import static org.apache.hudi.metadata.HoodieTableMetadataUtil.IDENTITY_ENCODING;
import static org.apache.hudi.metadata.HoodieTableMetadataUtil.PARTITION_NAME_BLOOM_FILTERS;
import static org.apache.hudi.metadata.HoodieTableMetadataUtil.PARTITION_NAME_COLUMN_STATS;
import static org.apache.hudi.metadata.HoodieTableMetadataUtil.PARTITION_NAME_FILES;
import static org.apache.hudi.metadata.HoodieTableMetadataUtil.existingIndexVersionOrDefault;
import static org.apache.hudi.metadata.HoodieTableMetadataUtil.getFileSystemViewForMetadataTable;
-import static org.apache.hudi.metadata.SecondaryIndexKeyUtils.unescapeSpecialChars;
/**
* Table metadata provided by an internal DFS backed Hudi metadata table.
@@ -168,7 +168,7 @@ public class HoodieBackedTableMetadata extends BaseTableMetadata {
@Override
protected Option<HoodieRecord<HoodieMetadataPayload>> getRecordByKey(String key, String partitionName) {
List<HoodieRecord<HoodieMetadataPayload>> records = getRecordsByKeys(
- HoodieListData.eager(Collections.singletonList(key)), partitionName, Option.empty())
+ HoodieListData.eager(Collections.singletonList(key)), partitionName, IDENTITY_ENCODING)
.values().collectAsList();
ValidationUtils.checkArgument(records.size() <= 1, "Found more than 1 record for record key " + key);
return records.isEmpty() ? Option.empty() : Option.ofNullable(records.get(0));
@@ -213,12 +213,10 @@ public class HoodieBackedTableMetadata extends BaseTableMetadata {
public HoodieData<HoodieRecord<HoodieMetadataPayload>> getRecordsByKeyPrefixes(HoodieData<String> keyPrefixes,
String partitionName,
boolean shouldLoadInMemory,
- Option<SerializableFunctionUnchecked<String, String>> keyEncodingFn) {
+ SerializableFunctionUnchecked<String, String> keyEncodingFn) {
ValidationUtils.checkState(keyPrefixes instanceof HoodieListData, "getRecordsByKeyPrefixes only support HoodieListData at the moment");
- // Apply key encoding if present
- List<String> sortedKeyPrefixes = keyEncodingFn.isPresent()
- ? new ArrayList<>(keyPrefixes.map(k -> keyEncodingFn.get().apply(k)).collectAsList())
- : new ArrayList<>(keyPrefixes.collectAsList());
+ // Apply key encoding
+ List<String> sortedKeyPrefixes = new ArrayList<>(keyPrefixes.map(keyEncodingFn::apply).collectAsList());
// Sort the prefixes so that keys are looked up in order
// Sort must come after encoding.
Collections.sort(sortedKeyPrefixes);
@@ -263,12 +261,29 @@ public class HoodieBackedTableMetadata extends BaseTableMetadata {
return Predicates.startsWithAny(null, right);
}
+ /**
+ * All keys to be looked up go through the following steps:
+ * 1. [encode] escape/encode the key if needed
+ * 2. [hash to file group] compute the hash of the key to
+ * 3. [lookup within file groups] lookup the key in the file group
+ * 4. the record is returned
+ */
+ /**
+ * Performs lookup of records in the metadata table.
+ *
+ * @param keys The keys to look up in the metadata table
+ * @param partitionName The name of the metadata partition to search in
+ * @param fileSlices The list of file slices to search through
+ * @param isSecondaryIndex Whether this lookup is for a secondary index
+ * @param keyEncodingFn Optional function to encode keys before lookup
+ * @return Pair data containing the looked up records keyed by their original keys
+ */
private HoodiePairData<String, HoodieRecord<HoodieMetadataPayload>> doLookup(HoodieData<String> keys, String partitionName, List<FileSlice> fileSlices,
- boolean isSecondaryIndex, Option<SerializableFunctionUnchecked<String, String>> keyEncodingFn) {
+ boolean isSecondaryIndex, SerializableFunctionUnchecked<String, String> keyEncodingFn) {
final int numFileSlices = fileSlices.size();
if (numFileSlices == 1) {
- List<String> keysList = keyEncodingFn.isPresent() ? keys.map(k -> keyEncodingFn.get().apply(k)).collectAsList() : keys.collectAsList();
+ List<String> keysList = keys.map(keyEncodingFn::apply).collectAsList();
TreeSet<String> distinctSortedKeys = new TreeSet<>(keysList);
return lookupKeyRecordPairs(partitionName, new ArrayList<>(distinctSortedKeys), fileSlices.get(0), !isSecondaryIndex);
}
@@ -277,12 +292,14 @@ public class HoodieBackedTableMetadata extends BaseTableMetadata {
// SI write path concatenates secKey$recordKey, the secKey needs extracted for hashing;
// SI read path gives secKey only, no need for secKey extraction.
SerializableBiFunction<String, Integer, Integer> mappingFunction = HoodieTableMetadataUtil::mapRecordKeyToFileGroupIndex;
- keys = repartitioningIfNeeded(keys, partitionName, numFileSlices, mappingFunction);
+ keys = repartitioningIfNeeded(keys, partitionName, numFileSlices, mappingFunction, keyEncodingFn);
HoodiePairData<Integer, String> persistedInitialPairData = keys
- // Tag key with file group index and apply key encoding
- .mapToPair(recordKey -> new ImmutablePair<>(
- mappingFunction.apply(recordKey, numFileSlices),
- keyEncodingFn.isPresent() ? keyEncodingFn.get().apply(recordKey) : recordKey));
+ // Tag key with file group index
+ .mapToPair(recordKey -> {
+ String encodedKey = keyEncodingFn.apply(recordKey);
+ // Always encode the key before apply mapping.
+ return new ImmutablePair<>(mappingFunction.apply(encodedKey, numFileSlices), encodedKey);
+ });
persistedInitialPairData.persist("MEMORY_AND_DISK_SER");
// Use the new processValuesOfTheSameShards API instead of explicit rangeBasedRepartitionForEachKey
SerializableFunction<Iterator<String>, Iterator<Pair<String, HoodieRecord<HoodieMetadataPayload>>>> processFunction =
@@ -295,7 +312,7 @@ public class HoodieBackedTableMetadata extends BaseTableMetadata {
}
distinctSortedKeyIter.forEachRemaining(keysList::add);
}
- FileSlice fileSlice = fileSlices.get(mappingFunction.apply(unescapeSpecialChars(keysList.get(0)), numFileSlices));
+ FileSlice fileSlice = fileSlices.get(mappingFunction.apply(keysList.get(0), numFileSlices));
return lookupKeyRecordPairsItr(partitionName, keysList, fileSlice, !isSecondaryIndex);
};
@@ -307,12 +324,19 @@ public class HoodieBackedTableMetadata extends BaseTableMetadata {
return result.filter((String k, HoodieRecord<HoodieMetadataPayload> v) -> !v.getData().isDeleted());
}
+ /**
+ * All keys to be looked up go through the following steps:
+ * 1. [encode] escape/encode the key if needed
+ * 2. [hash to file group] compute the hash of the key to
+ * 3. [lookup within file groups] lookup the key in the file group
+ * 4. the record is returned
+ */
private HoodieData<HoodieRecord<HoodieMetadataPayload>> doLookupIndexRecords(HoodieData<String> keys, String partitionName, List<FileSlice> fileSlices,
- boolean isSecondaryIndex, Option<SerializableFunctionUnchecked<String, String>> keyEncodingFn) {
+ boolean isSecondaryIndex, SerializableFunctionUnchecked<String, String> keyEncodingFn) {
final int numFileSlices = fileSlices.size();
if (numFileSlices == 1) {
- List<String> keysList = keyEncodingFn.isPresent() ? keys.map(k -> keyEncodingFn.get().apply(k)).collectAsList() : keys.collectAsList();
+ List<String> keysList = keys.map(keyEncodingFn::apply).collectAsList();
TreeSet<String> distinctSortedKeys = new TreeSet<>(keysList);
return lookupRecords(partitionName, new ArrayList<>(distinctSortedKeys), fileSlices.get(0), !isSecondaryIndex);
}
@@ -321,12 +345,14 @@ public class HoodieBackedTableMetadata extends BaseTableMetadata {
// SI write path concatenates secKey$recordKey, the secKey needs extracted for hashing;
// SI read path gives secKey only, no need for secKey extraction.
SerializableBiFunction<String, Integer, Integer> mappingFunction = HoodieTableMetadataUtil::mapRecordKeyToFileGroupIndex;
- keys = repartitioningIfNeeded(keys, partitionName, numFileSlices, mappingFunction);
+ keys = repartitioningIfNeeded(keys, partitionName, numFileSlices, mappingFunction, keyEncodingFn);
HoodiePairData<Integer, String> persistedInitialPairData = keys
// Tag key with file group index
- .mapToPair(recordKey -> new ImmutablePair<>(
- mappingFunction.apply(recordKey, numFileSlices),
- keyEncodingFn.isPresent() ? keyEncodingFn.get().apply(recordKey) : recordKey));
+ .mapToPair(recordKey -> {
+ String encodedKey = keyEncodingFn.apply(recordKey);
+ // Always encode the key before apply mapping.
+ return new ImmutablePair<>(mappingFunction.apply(encodedKey, numFileSlices), encodedKey);
+ });
persistedInitialPairData.persist("MEMORY_AND_DISK_SER");
// Use the new processValuesOfTheSameShards API instead of explicit rangeBasedRepartitionForEachKey
@@ -340,7 +366,7 @@ public class HoodieBackedTableMetadata extends BaseTableMetadata {
}
distinctSortedKeyIter.forEachRemaining(keysList::add);
}
- FileSlice fileSlice = fileSlices.get(mappingFunction.apply(unescapeSpecialChars(keysList.get(0)), numFileSlices));
+ FileSlice fileSlice = fileSlices.get(mappingFunction.apply(keysList.get(0), numFileSlices));
return lookupRecordsItr(partitionName, keysList, fileSlice, !isSecondaryIndex);
};
List<Integer> keySpace = IntStream.range(0, numFileSlices).boxed().collect(Collectors.toList());
@@ -367,7 +393,7 @@ public class HoodieBackedTableMetadata extends BaseTableMetadata {
"Record index is not initialized in MDT");
// TODO [HUDI-9544]: Metric does not work for rdd based API due to lazy evaluation.
- return getRecordsByKeys(recordKeys, MetadataPartitionType.RECORD_INDEX.getPartitionPath(), Option.empty())
+ return getRecordsByKeys(recordKeys, MetadataPartitionType.RECORD_INDEX.getPartitionPath(), IDENTITY_ENCODING)
.mapToPair((Pair<String, HoodieRecord<HoodieMetadataPayload>> p) -> Pair.of(p.getLeft(), p.getRight().getData().getRecordGlobalLocation()));
}
@@ -386,7 +412,7 @@ public class HoodieBackedTableMetadata extends BaseTableMetadata {
// The caller is required to check for record index existence in MDT before calling this method.
ValidationUtils.checkState(dataMetaClient.getTableConfig().isMetadataPartitionAvailable(MetadataPartitionType.RECORD_INDEX),
"Record index is not initialized in MDT");
- return readIndexRecords(recordKeys, MetadataPartitionType.RECORD_INDEX.getPartitionPath(), Option.empty())
+ return readIndexRecords(recordKeys, MetadataPartitionType.RECORD_INDEX.getPartitionPath(), IDENTITY_ENCODING)
.map(r -> r.getData().getRecordGlobalLocation());
}
@@ -432,7 +458,7 @@ public class HoodieBackedTableMetadata extends BaseTableMetadata {
@Override
public HoodiePairData<String, HoodieRecord<HoodieMetadataPayload>> getRecordsByKeys(
- HoodieData<String> keys, String partitionName, Option<SerializableFunctionUnchecked<String, String>> keyEncodingFn) {
+ HoodieData<String> keys, String partitionName, SerializableFunctionUnchecked<String, String> keyEncodingFn) {
List<FileSlice> fileSlices = partitionFileSliceMap.computeIfAbsent(partitionName,
k -> HoodieTableMetadataUtil.getPartitionLatestMergedFileSlices(metadataMetaClient, getMetadataFileSystemView(), partitionName));
checkState(!fileSlices.isEmpty(), "No file slices found for partition: " + partitionName);
@@ -442,7 +468,7 @@ public class HoodieBackedTableMetadata extends BaseTableMetadata {
}
public HoodieData<String> getRecordKeysFromSecondaryKeysV2(HoodieData<String> secondaryKeys, String partitionName) {
- return readIndexRecords(secondaryKeys, partitionName, Option.of(SecondaryIndexKeyUtils::escapeSpecialChars)).map(
+ return readIndexRecords(secondaryKeys, partitionName, SecondaryIndexKeyUtils::escapeSpecialChars).map(
hoodieRecord -> SecondaryIndexKeyUtils.getRecordKeyFromSecondaryIndexKey(hoodieRecord.getRecordKey()));
}
@@ -469,7 +495,7 @@ public class HoodieBackedTableMetadata extends BaseTableMetadata {
protected HoodieData<HoodieRecord<HoodieMetadataPayload>> readIndexRecords(HoodieData<String> keys,
String partitionName,
- Option<SerializableFunctionUnchecked<String, String>> keyEncodingFn) {
+ SerializableFunctionUnchecked<String, String> keyEncodingFn) {
List<FileSlice> fileSlices = partitionFileSliceMap.computeIfAbsent(partitionName,
k -> HoodieTableMetadataUtil.getPartitionLatestMergedFileSlices(metadataMetaClient, getMetadataFileSystemView(), partitionName));
checkState(!fileSlices.isEmpty(), "No file slices found for partition: " + partitionName);
@@ -481,16 +507,18 @@ public class HoodieBackedTableMetadata extends BaseTableMetadata {
// When testing we noticed that the parallelism can be very low which hurts the performance. so we should start with a reasonable
// level of parallelism in that case.
private HoodieData<String> repartitioningIfNeeded(
- HoodieData<String> keys, String partitionName, int numFileSlices, SerializableBiFunction<String, Integer, Integer> mappingFunction) {
+ HoodieData<String> keys, String partitionName, int numFileSlices, SerializableBiFunction<String, Integer, Integer> mappingFunction,
+ SerializableFunctionUnchecked<String, String> keyEncodingFn) {
if (keys instanceof HoodieListData) {
- int parallelism = (int) keys.map(k -> mappingFunction.apply(k, numFileSlices)).distinct().count();
+ int parallelism;
+ parallelism = (int) keys.map(k -> mappingFunction.apply(keyEncodingFn.apply(k), numFileSlices)).distinct().count();
// In case of empty lookup set, we should avoid RDD with 0 partitions.
parallelism = Math.max(parallelism, 1);
- LOG.info("getRecordFast repartition HoodieListData to JavaRDD: exit, partitionName {}, num partitions: {}",
+ LOG.info("Repartitioning keys for partition {} from list data with parallelism: {}",
partitionName, parallelism);
keys = getEngineContext().parallelize(keys.collectAsList(), parallelism);
} else if (keys.getNumPartitions() < metadataConfig.getRepartitionMinPartitionsThreshold()) {
- LOG.info("getRecordFast repartition HoodieNonListData. partitionName {}, num partitions: {}", partitionName, metadataConfig.getRepartitionDefaultPartitions());
+ LOG.info("Repartitioning keys for partition {} to {} partitions", partitionName, metadataConfig.getRepartitionDefaultPartitions());
keys = keys.repartition(metadataConfig.getRepartitionDefaultPartitions());
}
return keys;
@@ -732,7 +760,7 @@ public class HoodieBackedTableMetadata extends BaseTableMetadata {
return HoodieListPairData.eager(Collections.emptyList());
}
- Map<String, Set<String>> res = getRecordsByKeyPrefixes(keys, partitionName, false, Option.of(SecondaryIndexKeyUtils::escapeSpecialChars))
+ Map<String, Set<String>> res = getRecordsByKeyPrefixes(keys, partitionName, false, SecondaryIndexKeyUtils::escapeSpecialChars)
.map(record -> {
if (!record.getData().isDeleted()) {
return SecondaryIndexKeyUtils.getSecondaryKeyRecordKeyPair(record.getRecordKey());
@@ -761,7 +789,7 @@ public class HoodieBackedTableMetadata extends BaseTableMetadata {
if (secondaryKeys.isEmpty()) {
return HoodieListPairData.eager(Collections.emptyList());
}
- return readIndexRecords(secondaryKeys, partitionName, Option.of(SecondaryIndexKeyUtils::escapeSpecialChars))
+ return readIndexRecords(secondaryKeys, partitionName, SecondaryIndexKeyUtils::escapeSpecialChars)
.filter(hoodieRecord -> !hoodieRecord.getData().isDeleted())
.mapToPair(hoodieRecord -> SecondaryIndexKeyUtils.getRecordKeySecondaryKeyPair(hoodieRecord.getRecordKey()))
.groupByKey()
diff --git a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadata.java b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadata.java
index 1e6e5c466319..8035d31af5c2 100644
--- a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadata.java
+++ b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadata.java
@@ -265,7 +265,7 @@ public interface HoodieTableMetadata extends Serializable, AutoCloseable {
HoodieData<HoodieRecord<HoodieMetadataPayload>> getRecordsByKeyPrefixes(HoodieData<String> keyPrefixes,
String partitionName,
boolean shouldLoadInMemory,
- Option<SerializableFunctionUnchecked<String, String>> keyEncoder);
+ SerializableFunctionUnchecked<String, String> keyEncoder);
/**
* Get the instant time to which the metadata is synced w.r.t data timeline.
diff --git a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java
index 568c54975342..535dedf148b1 100644
--- a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java
+++ b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java
@@ -55,6 +55,7 @@ import org.apache.hudi.common.engine.HoodieReaderContext;
import org.apache.hudi.common.engine.ReaderContextFactory;
import org.apache.hudi.common.fs.FSUtils;
import org.apache.hudi.common.function.SerializableBiFunction;
+import org.apache.hudi.common.function.SerializableFunctionUnchecked;
import org.apache.hudi.common.function.SerializablePairFunction;
import org.apache.hudi.common.model.EmptyHoodieRecordPayload;
import org.apache.hudi.common.model.FileSlice;
@@ -210,6 +211,7 @@ public class HoodieTableMetadataUtil {
public static final String PARTITION_NAME_EXPRESSION_INDEX_PREFIX = "expr_index_";
public static final String PARTITION_NAME_SECONDARY_INDEX = "secondary_index";
public static final String PARTITION_NAME_SECONDARY_INDEX_PREFIX = "secondary_index_";
+ public static final SerializableFunctionUnchecked<String, String> IDENTITY_ENCODING = key -> key;
private static final Set<Schema.Type> SUPPORTED_TYPES_PARTITION_STATS = new HashSet<>(Arrays.asList(
Schema.Type.INT, Schema.Type.LONG, Schema.Type.FLOAT, Schema.Type.DOUBLE, Schema.Type.STRING, Schema.Type.BOOLEAN, Schema.Type.NULL, Schema.Type.BYTES));
@@ -1343,7 +1345,7 @@ public class HoodieTableMetadataUtil {
* (either all containing the secondary index separator or none containing it).
* <p>
* For secondary index partitions (version >= 2), if the record keys contain the secondary index separator,
- * the secondary key portion is used for hashing. Otherwise, the full record key is used.
+ * the unescaped secondary key portion is used for hashing. Otherwise, the full record key is used.
*
* @param needsSecondaryKeyExtraction Whether to extract secondary key from composite keys (should be determined by caller)
*
@@ -1352,7 +1354,7 @@ public class HoodieTableMetadataUtil {
public static SerializableBiFunction<String, Integer, Integer> getSecondaryKeyToFileGroupMappingFunction(boolean needsSecondaryKeyExtraction) {
if (needsSecondaryKeyExtraction) {
return (recordKey, numFileGroups) -> {
- String secondaryKey = SecondaryIndexKeyUtils.getSecondaryKeyFromSecondaryIndexKey(recordKey);
+ String secondaryKey = SecondaryIndexKeyUtils.getUnescapedSecondaryKeyFromSecondaryIndexKey(recordKey);
return mapRecordKeyToFileGroupIndex(secondaryKey, numFileGroups);
};
}
@@ -2716,7 +2718,7 @@ public class HoodieTableMetadataUtil {
// Fetch metadata table COLUMN_STATS partition records for above files
List<HoodieColumnRangeMetadata<Comparable>> partitionColumnMetadata = tableMetadata
.getRecordsByKeyPrefixes(
- HoodieListData.lazy(generateKeyPrefixes(colsToIndex, partitionName)), MetadataPartitionType.COLUMN_STATS.getPartitionPath(), false, Option.empty())
+ HoodieListData.lazy(generateKeyPrefixes(colsToIndex, partitionName)), MetadataPartitionType.COLUMN_STATS.getPartitionPath(), false, IDENTITY_ENCODING)
// schema and properties are ignored in getInsertValue, so simply pass as null
.map(record -> ((HoodieMetadataPayload)record.getData()).getColumnStatMetadata())
.filter(Option::isPresent)
diff --git a/hudi-common/src/main/java/org/apache/hudi/metadata/SecondaryIndexKeyUtils.java b/hudi-common/src/main/java/org/apache/hudi/metadata/SecondaryIndexKeyUtils.java
index 19a3bded29b1..e6c759b5fde1 100644
--- a/hudi-common/src/main/java/org/apache/hudi/metadata/SecondaryIndexKeyUtils.java
+++ b/hudi-common/src/main/java/org/apache/hudi/metadata/SecondaryIndexKeyUtils.java
@@ -53,10 +53,16 @@ public class SecondaryIndexKeyUtils {
}
public static String getSecondaryKeyFromSecondaryIndexKey(String key) {
+ // the payload key is in the format of "secondaryKey$primaryKey"
+ // we need to extract the secondary key from the payload key
+ return unescapeSpecialChars(getUnescapedSecondaryKeyFromSecondaryIndexKey(key));
+ }
+
+ public static String getUnescapedSecondaryKeyFromSecondaryIndexKey(String key) {
// the payload key is in the format of "secondaryKey$primaryKey"
// we need to extract the secondary key from the payload key
int delimiterIndex = getSecondaryIndexKeySeparatorPosition(key);
- return unescapeSpecialChars(key.substring(0, delimiterIndex));
+ return key.substring(0, delimiterIndex);
}
public static String constructSecondaryIndexKey(String secondaryKey, String recordKey) {
diff --git a/hudi-common/src/test/java/org/apache/hudi/metadata/TestHoodieTableMetadataUtil.java b/hudi-common/src/test/java/org/apache/hudi/metadata/TestHoodieTableMetadataUtil.java
index e7f50dbb820e..5900fdafbdde 100644
--- a/hudi-common/src/test/java/org/apache/hudi/metadata/TestHoodieTableMetadataUtil.java
+++ b/hudi-common/src/test/java/org/apache/hudi/metadata/TestHoodieTableMetadataUtil.java
@@ -22,31 +22,39 @@ import org.apache.hudi.common.function.SerializableBiFunction;
import org.junit.jupiter.api.Test;
+import static org.apache.hudi.metadata.SecondaryIndexKeyUtils.constructSecondaryIndexKey;
+import static org.apache.hudi.metadata.SecondaryIndexKeyUtils.escapeSpecialChars;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
public class TestHoodieTableMetadataUtil {
@Test
- public void testGetRecordKeyToFileGroupIndexFunctionOptimized() {
+ public void testGetRecordKeyToFileGroupIndexFunction() {
// Test with secondary key format
String compositeKey = "secondaryKey$recordKey";
- SerializableBiFunction<String, Integer, Integer> optimizedFunction =
+ SerializableBiFunction<String, Integer, Integer> hashOnSecKeyOnly =
HoodieTableMetadataUtil.getSecondaryKeyToFileGroupMappingFunction(true);
+ SerializableBiFunction<String, Integer, Integer> hashOnFullKey =
+ HoodieTableMetadataUtil.getSecondaryKeyToFileGroupMappingFunction(false);
- int result1 = optimizedFunction.apply(compositeKey, 10);
- int result2 = optimizedFunction.apply("anotherSecondaryKey$anotherRecordKey", 10);
+ int result1 = hashOnSecKeyOnly.apply(compositeKey, 10);
+ int result2 = hashOnSecKeyOnly.apply("anotherSecondaryKey$anotherRecordKey", 10);
// Both should hash the secondary key portion
assertNotEquals(result1, result2);
// Test with regular key format
- optimizedFunction = HoodieTableMetadataUtil.getSecondaryKeyToFileGroupMappingFunction(false);
-
- int result3 = optimizedFunction.apply("simpleKey", 10);
- int result4 = optimizedFunction.apply("anotherSimpleKey", 10);
+ int result3 = hashOnFullKey.apply("simpleKey", 10);
+ int result4 = hashOnFullKey.apply("anotherSimpleKey", 10);
// Both should hash the full key
assertNotEquals(result3, result4);
+
+ // Hash on Sec key only <=> Hash full key if key equals to UNESCAPED sec key
+ assertEquals(hashOnSecKeyOnly.apply("secKey$recKey", 10), hashOnFullKey.apply("secKey", 10));
+ assertEquals(hashOnSecKeyOnly.apply(constructSecondaryIndexKey("seckey", "reckey"), 10), hashOnFullKey.apply("seckey", 10));
+ assertEquals(hashOnSecKeyOnly.apply(constructSecondaryIndexKey("$", "reckey"), 10), hashOnFullKey.apply(escapeSpecialChars("$"), 10));
}
}
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/stats/FileStatsIndex.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/stats/FileStatsIndex.java
index 073491639cfc..0c1c296feab4 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/stats/FileStatsIndex.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/stats/FileStatsIndex.java
@@ -24,7 +24,6 @@ import org.apache.hudi.common.data.HoodieData;
import org.apache.hudi.common.data.HoodieListData;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.table.HoodieTableMetaClient;
-import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.ValidationUtils;
import org.apache.hudi.common.util.VisibleForTesting;
import org.apache.hudi.common.util.collection.Pair;
@@ -64,6 +63,7 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;
import static org.apache.hudi.common.util.ValidationUtils.checkState;
+import static org.apache.hudi.metadata.HoodieTableMetadataUtil.IDENTITY_ENCODING;
import static org.apache.hudi.source.stats.ColumnStatsSchemas.COL_STATS_DATA_TYPE;
import static org.apache.hudi.source.stats.ColumnStatsSchemas.COL_STATS_TARGET_POS;
import static org.apache.hudi.source.stats.ColumnStatsSchemas.METADATA_DATA_TYPE;
@@ -397,7 +397,7 @@ public class FileStatsIndex implements ColumnStatsIndex {
HoodieData<HoodieRecord<HoodieMetadataPayload>> records =
getMetadataTable().getRecordsByKeyPrefixes(
- HoodieListData.lazy(encodedTargetColumnNames), getIndexPartitionName(), false, Option.empty());
+ HoodieListData.lazy(encodedTargetColumnNames), getIndexPartitionName(), false, IDENTITY_ENCODING);
org.apache.hudi.util.AvroToRowDataConverters.AvroToRowDataConverter converter =
AvroToRowDataConverters.createRowConverter((RowType) METADATA_DATA_TYPE.getLogicalType());
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/ColumnStatsIndexSupport.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/ColumnStatsIndexSupport.scala
index 60d4df93c18f..204a50ca2a03 100644
--- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/ColumnStatsIndexSupport.scala
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/ColumnStatsIndexSupport.scala
@@ -358,7 +358,7 @@ class ColumnStatsIndexSupport(spark: SparkSession,
val metadataRecords: HoodieData[HoodieRecord[HoodieMetadataPayload]] =
metadataTable.getRecordsByKeyPrefixes(
HoodieListData.eager(keyPrefixes.toSeq.asJava), HoodieTableMetadataUtil.PARTITION_NAME_COLUMN_STATS, shouldReadInMemory,
- toJavaOption(Option.empty[SerializableFunctionUnchecked[String, String]]))
+ HoodieTableMetadataUtil.IDENTITY_ENCODING)
val columnStatsRecords: HoodieData[HoodieMetadataColumnStats] =
//TODO: [HUDI-8303] Explicit conversion might not be required for Scala 2.12+
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/ExpressionIndexSupport.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/ExpressionIndexSupport.scala
index 0dab9f6b3f65..f81bda62ac38 100644
--- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/ExpressionIndexSupport.scala
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/ExpressionIndexSupport.scala
@@ -360,7 +360,7 @@ class ExpressionIndexSupport(spark: SparkSession,
val metadataRecords: HoodieData[HoodieRecord[HoodieMetadataPayload]] =
metadataTable.getRecordsByKeyPrefixes(
HoodieListData.eager(keyPrefixes.toSeq.asJava), HoodieTableMetadataUtil.PARTITION_NAME_COLUMN_STATS, shouldReadInMemory,
- toJavaOption(Option.empty[SerializableFunctionUnchecked[String, String]]))
+ HoodieTableMetadataUtil.IDENTITY_ENCODING)
val columnStatsRecords: HoodieData[HoodieMetadataColumnStats] =
//TODO: [HUDI-8303] Explicit conversion might not be required for Scala 2.12+
@@ -564,7 +564,7 @@ class ExpressionIndexSupport(spark: SparkSession,
val metadataRecords: HoodieData[HoodieRecord[HoodieMetadataPayload]] =
metadataTable.getRecordsByKeyPrefixes(
HoodieListData.eager(keyPrefixes.toSeq.asJava), indexPartition, shouldReadInMemory,
- toJavaOption(Option.empty[SerializableFunctionUnchecked[String, String]]))
+ HoodieTableMetadataUtil.IDENTITY_ENCODING)
val columnStatsRecords: HoodieData[HoodieMetadataColumnStats] =
//TODO: [HUDI-8303] Explicit conversion might not be required for Scala 2.12+
metadataRecords.map(JFunction.toJavaSerializableFunction(record => {
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/PartitionStatsIndexSupport.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/PartitionStatsIndexSupport.scala
index f00cd9c20f48..5b5fcfeb12d4 100644
--- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/PartitionStatsIndexSupport.scala
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/PartitionStatsIndexSupport.scala
@@ -73,7 +73,7 @@ class PartitionStatsIndexSupport(spark: SparkSession,
val metadataRecords: HoodieData[HoodieRecord[HoodieMetadataPayload]] =
metadataTable.getRecordsByKeyPrefixes(
HoodieListData.eager(encodedTargetColumnNames.asJava), HoodieTableMetadataUtil.PARTITION_NAME_PARTITION_STATS, shouldReadInMemory,
- toJavaOption(Option.empty[SerializableFunctionUnchecked[String, String]]))
+ HoodieTableMetadataUtil.IDENTITY_ENCODING)
val columnStatsRecords: HoodieData[HoodieMetadataColumnStats] =
//TODO: [HUDI-8303] Explicit conversion might not be required for Scala 2.12+
metadataRecords.map(JFunction.toJavaSerializableFunction(record => {
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 8f3492e75189..3d0661b5e22a 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
@@ -193,6 +193,7 @@ import static org.apache.hudi.config.HoodieCompactionConfig.INLINE_COMPACT_NUM_D
import static org.apache.hudi.io.storage.HoodieSparkIOFactory.getHoodieSparkIOFactory;
import static org.apache.hudi.metadata.HoodieTableMetadata.SOLO_COMMIT_TIMESTAMP;
import static org.apache.hudi.metadata.HoodieTableMetadata.getMetadataTableBasePath;
+import static org.apache.hudi.metadata.HoodieTableMetadataUtil.IDENTITY_ENCODING;
import static org.apache.hudi.metadata.HoodieTableMetadataUtil.deleteMetadataTable;
import static org.apache.hudi.metadata.MetadataPartitionType.BLOOM_FILTERS;
import static org.apache.hudi.metadata.MetadataPartitionType.COLUMN_STATS;
@@ -1999,7 +2000,7 @@ public class TestHoodieBackedMetadata extends TestHoodieMetadataBase {
ColumnIndexID columnIndexID = new ColumnIndexID(HoodieRecord.RECORD_KEY_METADATA_FIELD);
List<HoodieRecord<HoodieMetadataPayload>> result = tableMetadata.getRecordsByKeyPrefixes(
HoodieListData.lazy(Collections.singletonList(columnIndexID.asBase64EncodedString())),
- MetadataPartitionType.COLUMN_STATS.getPartitionPath(), true, Option.empty()).collectAsList();
+ MetadataPartitionType.COLUMN_STATS.getPartitionPath(), true, IDENTITY_ENCODING).collectAsList();
// there are 3 partitions in total and 2 commits. total entries should be 6.
assertEquals(result.size(), 6);
@@ -2008,7 +2009,7 @@ public class TestHoodieBackedMetadata extends TestHoodieMetadataBase {
PartitionIndexID partitionIndexID = new PartitionIndexID(HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH);
result = tableMetadata.getRecordsByKeyPrefixes(
HoodieListData.lazy(Collections.singletonList(columnIndexID.asBase64EncodedString().concat(partitionIndexID.asBase64EncodedString()))),
- MetadataPartitionType.COLUMN_STATS.getPartitionPath(), true, Option.empty()).collectAsList();
+ MetadataPartitionType.COLUMN_STATS.getPartitionPath(), true, IDENTITY_ENCODING).collectAsList();
// 1 partition and 2 commits. total entries should be 2.
assertEquals(result.size(), 2);
result.forEach(entry -> {
@@ -2027,7 +2028,7 @@ public class TestHoodieBackedMetadata extends TestHoodieMetadataBase {
columnIndexID = new ColumnIndexID(HoodieRecord.COMMIT_TIME_METADATA_FIELD);
result = tableMetadata.getRecordsByKeyPrefixes(
HoodieListData.lazy(Collections.singletonList(columnIndexID.asBase64EncodedString().concat(partitionIndexID.asBase64EncodedString()))),
- MetadataPartitionType.COLUMN_STATS.getPartitionPath(), true, Option.empty()).collectAsList();
+ MetadataPartitionType.COLUMN_STATS.getPartitionPath(), true, IDENTITY_ENCODING).collectAsList();
// 1 partition and 2 commits. total entries should be 2.
assertEquals(result.size(), 2);
@@ -3016,7 +3017,7 @@ public class TestHoodieBackedMetadata extends TestHoodieMetadataBase {
HoodieTableMetadata tableMetadata = metadata(client, storage);
assertTrue(tableMetadata.getRecordsByKeyPrefixes(
HoodieListData.lazy(Collections.singletonList(HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH)),
- FILES.getPartitionPath(), false, Option.empty()).isEmpty());
+ FILES.getPartitionPath(), false, IDENTITY_ENCODING).isEmpty());
assertTrue(tableMetadata.getAllPartitionPaths().contains(HoodieTestDataGenerator.DEFAULT_SECOND_PARTITION_PATH));
assertFalse(tableMetadata.getAllPartitionPaths().contains(HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH));
// above upsert would have triggered clean