apache / apache/hudi

[SUPPORT] using MDT to filter fileSlices for MOR table

Open
#8,531 0 comments 0 reactions 1 assignee Claimed by @xushiyan View on GitHub
area:metadata-table priority:medium type:feature
Dominant language
Java
Stars
6.2k
Forks
2.5k
Avg merge
2d 8h
Merged PRs (30d)
111

Description

Because the metadataTable does not record information about log files, it is not used for file filtering when using the mor table. Can we first obtain the fileId set corresponding to the parquet file from the metadataTable and then use this set to filter all files from other indexes?

Code of [org.apache.hudi.MergeOnReadSnapshotRelation](https://github.com/apache/hudi/blob/master/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadSnapshotRelation.scala)
1. fileIndex.listFileSlices here not used metadataTable
```
protected override def collectFileSplits(partitionFilters: Seq[Expression], dataFilters: Seq[Expression]): List[HoodieMergeOnReadFileSplit] = {
val convertedPartitionFilters =
HoodieFileIndex.convertFilterForTimestampKeyGenerator(metaClient, partitionFilters)

if (globPaths.isEmpty) {
val fileSlices = fileIndex.listFileSlices(convertedPartitionFilters)
buildSplits(fileSlices.values.flatten.toSeq)
} else {
val fileSlices = listLatestFileSlices(globPaths, partitionFilters, dataFilters)
buildSplits(fileSlices)
}
}
```
2. can we call fileIndex.listFiles to get all base files and then use those file to filter unnecessary fileSlices?
```
protected override def collectFileSplits(partitionFilters: Seq[Expression], dataFilters: Seq[Expression]): List[HoodieMergeOnReadFileSplit] = {
val convertedPartitionFilters =
HoodieFileIndex.convertFilterForTimestampKeyGenerator(metaClient, partitionFilters)

if (globPaths.isEmpty) {
// find all the files matching the filters from metadataTable
val mdtSlices = fileIndex.listFiles(partitionFilters, dataFilters).flatMap(_.files).map(file=> file.getPath).toSet

// find all the files matching the filters from file system
val fileSlices = fileIndex.listFileSlices(convertedPartitionFilters)
.values.flatten.toSeq
// filter out the files which are not present in the metadata table
.filter(fileSlice => fileSlice.getBaseFile.isPresent && mdtSlices.contains(fileSlice.getBaseFile.get().getFileStatus.getPath))
buildSplits(fileSlices)
} else {
val fileSlices = listLatestFileSlices(globPaths, partitionFilters, dataFilters)
buildSplits(fileSlices)
}
}
```
In my test case, this change will effectively reducing the number of files that need to be queried

Considering that I am not yet familiar with this part of future planning, I hope someone can give me some suggestions

Contributor guide

No contributing guide indexed for this repository

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.