[SUPPORT] using MDT to filter fileSlices for MOR table
- 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.