yihua commented on code in PR #8856:
URL: https://github.com/apache/hudi/pull/8856#discussion_r1267028523
##########
hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java:
##########
@@ -850,59 +851,58 @@ public static HoodieData<HoodieRecord>
convertFilesToBloomFilterRecords(HoodieEn
String instantTime) {
HoodieData<HoodieRecord> allRecordsRDD = engineContext.emptyHoodieData();
- List<Pair<String, List<String>>> partitionToDeletedFilesList =
partitionToDeletedFiles.entrySet()
- .stream().map(e -> Pair.of(e.getKey(),
e.getValue())).collect(Collectors.toList());
- int parallelism = Math.max(Math.min(partitionToDeletedFilesList.size(),
recordsGenerationParams.getBloomIndexParallelism()), 1);
- HoodieData<Pair<String, List<String>>> partitionToDeletedFilesRDD =
engineContext.parallelize(partitionToDeletedFilesList, parallelism);
+ List<Pair<String, String>> partitionToDeletedFilesList =
partitionToDeletedFiles.entrySet().stream().flatMap(entry -> {
+ List<String> filesList = entry.getValue();
+ return filesList.stream().map(file -> Pair.of(entry.getKey(), file));
+ }).collect(Collectors.toList());
- HoodieData<HoodieRecord> deletedFilesRecordsRDD =
partitionToDeletedFilesRDD.flatMap(partitionToDeletedFilesPair -> {
- final String partitionName = partitionToDeletedFilesPair.getLeft();
- final List<String> deletedFileList =
partitionToDeletedFilesPair.getRight();
- return deletedFileList.stream().flatMap(deletedFile -> {
- if (!FSUtils.isBaseFile(new Path(deletedFile))) {
- return Stream.empty();
- }
+ int parallelism = Math.max(Math.min(partitionToDeletedFilesList.size(),
recordsGenerationParams.getBloomIndexParallelism()), 1);
+ HoodieData<Pair<String, String>> partitionToDeletedFilesRDD =
engineContext.parallelize(partitionToDeletedFilesList, parallelism);
- final String partition = getPartitionIdentifier(partitionName);
- return
Stream.<HoodieRecord>of(HoodieMetadataPayload.createBloomFilterMetadataRecord(
- partition, deletedFile, instantTime, StringUtils.EMPTY_STRING,
ByteBuffer.allocate(0), true));
- }).iterator();
- });
+ HoodieData<HoodieRecord> deletedFilesRecordsRDD =
partitionToDeletedFilesRDD.map(partitionToDeletedFilePair -> {
+ String partitionName = partitionToDeletedFilePair.getLeft();
+ String deletedFile = partitionToDeletedFilePair.getRight();
+ if (!FSUtils.isBaseFile(new Path(deletedFile))) {
+ return null;
+ }
+ final String partition = getPartitionIdentifier(partitionName);
+ return (HoodieRecord)
(HoodieMetadataPayload.createBloomFilterMetadataRecord(
+ partition, deletedFile, instantTime, StringUtils.EMPTY_STRING,
ByteBuffer.allocate(0), true));
+ }).filter(Objects::nonNull);
allRecordsRDD = allRecordsRDD.union(deletedFilesRecordsRDD);
- List<Pair<String, Map<String, Long>>> partitionToAppendedFilesList =
partitionToAppendedFiles.entrySet()
- .stream().map(entry -> Pair.of(entry.getKey(),
entry.getValue())).collect(Collectors.toList());
+ List<Pair<String, String>> partitionToAppendedFilesList =
partitionToAppendedFiles.entrySet().stream().flatMap(entry -> {
+ Set<String> filesSet = entry.getValue().keySet();
+ return filesSet.stream().map(file -> Pair.of(entry.getKey(), file));
Review Comment:
```suggestion
return entry.getValue().keySet().stream().map(file ->
Pair.of(entry.getKey(), file));
```
##########
hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java:
##########
@@ -850,59 +851,58 @@ public static HoodieData<HoodieRecord>
convertFilesToBloomFilterRecords(HoodieEn
String instantTime) {
HoodieData<HoodieRecord> allRecordsRDD = engineContext.emptyHoodieData();
- List<Pair<String, List<String>>> partitionToDeletedFilesList =
partitionToDeletedFiles.entrySet()
- .stream().map(e -> Pair.of(e.getKey(),
e.getValue())).collect(Collectors.toList());
- int parallelism = Math.max(Math.min(partitionToDeletedFilesList.size(),
recordsGenerationParams.getBloomIndexParallelism()), 1);
- HoodieData<Pair<String, List<String>>> partitionToDeletedFilesRDD =
engineContext.parallelize(partitionToDeletedFilesList, parallelism);
+ List<Pair<String, String>> partitionToDeletedFilesList =
partitionToDeletedFiles.entrySet().stream().flatMap(entry -> {
+ List<String> filesList = entry.getValue();
+ return filesList.stream().map(file -> Pair.of(entry.getKey(), file));
Review Comment:
```suggestion
return entry.getValue().stream().map(file -> Pair.of(entry.getKey(),
file));
```
##########
hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java:
##########
@@ -927,48 +927,44 @@ public static HoodieData<HoodieRecord>
convertFilesToColumnStatsRecords(HoodieEn
return engineContext.emptyHoodieData();
}
- final List<Pair<String, List<String>>> partitionToDeletedFilesList =
partitionToDeletedFiles.entrySet().stream()
- .map(e -> Pair.of(e.getKey(), e.getValue()))
- .collect(Collectors.toList());
+ List<Pair<String, String>> partitionToDeletedFilesList =
partitionToDeletedFiles.entrySet().stream().flatMap(entry -> {
+ List<String> filesList = entry.getValue();
+ return filesList.stream().map(file -> Pair.of(entry.getKey(), file));
+ }).collect(Collectors.toList());
int deletedFilesTargetParallelism =
Math.max(Math.min(partitionToDeletedFilesList.size(),
recordsGenerationParams.getColumnStatsIndexParallelism()), 1);
- final HoodieData<Pair<String, List<String>>> partitionToDeletedFilesRDD =
+ final HoodieData<Pair<String, String>> partitionToDeletedFilesRDD =
engineContext.parallelize(partitionToDeletedFilesList,
deletedFilesTargetParallelism);
HoodieData<HoodieRecord> deletedFilesRecordsRDD =
partitionToDeletedFilesRDD.flatMap(partitionToDeletedFilesPair -> {
- final String partitionPath = partitionToDeletedFilesPair.getLeft();
- final String partitionId = getPartitionIdentifier(partitionPath);
- final List<String> deletedFileList =
partitionToDeletedFilesPair.getRight();
-
- return deletedFileList.stream().flatMap(deletedFile -> {
- final String filePathWithPartition = partitionPath + "/" + deletedFile;
- return getColumnStatsRecords(partitionId, filePathWithPartition,
dataTableMetaClient, columnsToIndex, true);
- }).iterator();
+ String partitionPath = partitionToDeletedFilesPair.getLeft();
+ String partitionId = getPartitionIdentifier(partitionPath);
+ String deletedFile = partitionToDeletedFilesPair.getRight();
+ String filePathWithPartition = partitionPath + "/" + deletedFile;
+ return getColumnStatsRecords(partitionId, filePathWithPartition,
dataTableMetaClient, columnsToIndex, true).iterator();
});
allRecordsRDD = allRecordsRDD.union(deletedFilesRecordsRDD);
- final List<Pair<String, Map<String, Long>>> partitionToAppendedFilesList =
partitionToAppendedFiles.entrySet().stream()
- .map(entry -> Pair.of(entry.getKey(), entry.getValue()))
- .collect(Collectors.toList());
+ List<Pair<String, String>> partitionToAppendedFilesList =
partitionToAppendedFiles.entrySet().stream().flatMap(entry -> {
+ Set<String> filesSet = entry.getValue().keySet();
+ return filesSet.stream().map(file -> Pair.of(entry.getKey(), file));
Review Comment:
```suggestion
return entry.getValue().keySet().stream().map(file ->
Pair.of(entry.getKey(), file));
```
##########
hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java:
##########
@@ -927,48 +927,44 @@ public static HoodieData<HoodieRecord>
convertFilesToColumnStatsRecords(HoodieEn
return engineContext.emptyHoodieData();
}
- final List<Pair<String, List<String>>> partitionToDeletedFilesList =
partitionToDeletedFiles.entrySet().stream()
- .map(e -> Pair.of(e.getKey(), e.getValue()))
- .collect(Collectors.toList());
+ List<Pair<String, String>> partitionToDeletedFilesList =
partitionToDeletedFiles.entrySet().stream().flatMap(entry -> {
+ List<String> filesList = entry.getValue();
+ return filesList.stream().map(file -> Pair.of(entry.getKey(), file));
Review Comment:
```suggestion
return entry.getValue().stream().map(file -> Pair.of(entry.getKey(),
file));
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
