HoodieListData and HoodieListPairData needs to support notion of partitions
- Dominant language
- Java
- Stars
- 6.2k
- Forks
- 2.5k
- Avg merge
- 2d 8h
- Merged PRs (30d)
- 111
Description
as of today all the places that involves map partitions underneath, the JavaRDD and HoodieListData/HoodieListPairData deviates.
We need to have map partitions capability for the 2 classes. We also need partitioner interface and the same idea as RDD on how partitioning works. Maybe we also need to abstract the partitioner.
We already have code that requires Map partition in index lookup and today we do hacky things to achieve it
{code:java}
@Override
protected Map> getRecordsByKeys(List keys, String partitionName) {
if (keys.isEmpty()) {
return Collections.emptyMap();
}
Map> result;
// Load the file slices for the partition. Each file slice is a shard which saves a portion of the keys.
List partitionFileSlices = partitionFileSliceMap.computeIfAbsent(partitionName,
k -> HoodieTableMetadataUtil.getPartitionLatestMergedFileSlices(metadataMetaClient, getMetadataFileSystemView(), partitionName));
final int numFileSlices = partitionFileSlices.size();
checkState(numFileSlices > 0, "Number of file slices for partition " + partitionName + " should be > 0");
// Lookup keys from each file slice
if (numFileSlices == 1) {
// Optimization for a single slice for smaller metadata table partitions
result = lookupKeys(partitionName, keys, partitionFileSlices.get(0));
} else {
// Parallel lookup for large sized partitions with many file slices
// Partition the keys by the file slice which contains it
ArrayList> partitionedKeys = partitionKeysByFileSlices(keys, numFileSlices); <---- We do partition by in MetadataTable Class
result = new HashMap<>(keys.size());
getEngineContext().setJobStatus(this.getClass().getSimpleName(), "Reading keys from metadata table partition " + partitionName);
getEngineContext().map(partitionedKeys, keysList -> {
if (keysList.isEmpty()) {
return Collections.>emptyMap();
}
int shardIndex = HoodieTableMetadataUtil.mapRecordKeyToFileGroupIndex(keysList.get(0), numFileSlices);
return lookupKeys(partitionName, keysList, partitionFileSlices.get(shardIndex));
}, partitionedKeys.size()).forEach(result::putAll);
}
return result;
}
private static ArrayList> partitionKeysByFileSlices(List keys, int numFileSlices) {
ArrayList> partitionedKeys = new ArrayList<>(numFileSlices);
for (int i = 0; i < numFileSlices; ++i) {
partitionedKeys.add(new ArrayList<>());
}
keys.forEach(key -> {
int shardIndex = HoodieTableMetadataUtil.mapRecordKeyToFileGroupIndex(key, numFileSlices);
partitionedKeys.get(shardIndex).add(key);
});
return partitionedKeys;
} {code}
## JIRA info
- Link: https://issues.apache.org/jira/browse/HUDI-9542
- Type: Bug
Contributor guide
No contributing guide indexed for this repository
Research direction
Start with HoodieListData and HoodieListPairData, then inspect the existing getEngineContext().map usage in index lookup and compare it with JavaRDD partitioning. Determine the required map-partitions and partitioner semantics; done means both classes support partition-aware operations consistently with RDD behavior and the index lookup no longer needs its workaround.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java, spark
- Domain
- data-engineering, distributed-systems
- Issue type
- Bug
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Active
- Clarity
- Needs clarification
- Newbie friendliness
- 35/100