apache / apache/hudi

HoodieListData and HoodieListPairData needs to support notion of partitions

Open
#17,063 1 comment 0 reactions 0 assignees View on GitHub
from-jira priority:high type:bug
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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.