Skip to content

Commit 2cd893d

Browse files
nsivabalannsivabalan
committed
Fixing test failures
1 parent 054e64f commit 2cd893d

5 files changed

Lines changed: 23 additions & 24 deletions

File tree

‎hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/clean/CleanPlanner.java‎

Lines changed: 16 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -230,24 +230,24 @@ private Map<String, Pair<Boolean, List<CleanFileInfo>>> getFilesToCleanKeepingLa
230230
LOG.info("Cleaning " + partitionPaths + ", retaining latest " + config.getCleanerFileVersionsRetained()
231231
+ " file versions. ");
232232
Map<String, Pair<Boolean, List<CleanFileInfo>>> map = new HashMap<>();
233-
List<CleanFileInfo> deletePaths = new ArrayList<>();
234233
// Collect all the datafiles savepointed by all the savepoints
235234
List<String> savepointedFiles = hoodieTable.getSavepointTimestamps().stream()
236235
.flatMap(this::getSavepointedDataFiles)
237236
.collect(Collectors.toList());
238237

239238
// In this scenario, we will assume that once replaced a file group automatically becomes eligible for cleaning completely
240239
// In other words, the file versions only apply to the active file groups.
241-
List<Pair<String, List<HoodieFileGroup>>> fileGroups = fileSystemView.getAllFileGroups(partitionPaths).collect(Collectors.toList());
242-
for (Pair<String, List<HoodieFileGroup>> pairFileGroup : fileGroups) {
243-
244-
deletePaths.addAll(getReplacedFilesEligibleToClean(savepointedFiles, pairFileGroup.getLeft(), Option.empty()));
240+
List<Pair<String, List<HoodieFileGroup>>> fileGroupsPerPartition = fileSystemView.getAllFileGroups(partitionPaths).collect(Collectors.toList());
241+
for (Pair<String, List<HoodieFileGroup>> partitionFileGroupList : fileGroupsPerPartition) {
242+
List<CleanFileInfo> deletePaths = new ArrayList<>(getReplacedFilesEligibleToClean(savepointedFiles, partitionFileGroupList.getLeft(), Option.empty()));
245243
boolean toDeletePartition = false;
246-
for (HoodieFileGroup fileGroup : pairFileGroup.getRight()) {
244+
for (HoodieFileGroup fileGroup : partitionFileGroupList.getRight()) {
247245
int keepVersions = config.getCleanerFileVersionsRetained();
248246
// do not cleanup slice required for pending compaction
249247
Iterator<FileSlice> fileSliceIterator =
250-
fileGroup.getAllFileSlices().filter(fs -> !isFileSliceNeededForPendingCompaction(fs)).iterator();
248+
fileGroup.getAllFileSlices()
249+
.filter(fs -> !isFileSliceNeededForPendingCompaction(fs))
250+
.iterator();
251251
if (isFileGroupInPendingCompaction(fileGroup)) {
252252
// We have already saved the last version of file-groups for pending compaction Id
253253
keepVersions--;
@@ -270,10 +270,10 @@ private Map<String, Pair<Boolean, List<CleanFileInfo>>> getFilesToCleanKeepingLa
270270
}
271271
}
272272
// if there are no valid file groups for the partition, mark it to be deleted
273-
if (fileGroups.isEmpty()) {
273+
if (partitionFileGroupList.getValue().isEmpty()) {
274274
toDeletePartition = true;
275275
}
276-
map.put(pairFileGroup.getLeft(), Pair.of(toDeletePartition, deletePaths));
276+
map.put(partitionFileGroupList.getLeft(), Pair.of(toDeletePartition, deletePaths));
277277
}
278278
return map;
279279
}
@@ -301,7 +301,6 @@ private Map<String, Pair<Boolean, List<CleanFileInfo>>> getFilesToCleanKeepingLa
301301
*/
302302
private Map<String, Pair<Boolean, List<CleanFileInfo>>> getFilesToCleanKeepingLatestCommits(List<String> partitionPaths, int commitsRetained, HoodieCleaningPolicy policy) {
303303
LOG.info("Cleaning " + partitionPaths + ", retaining latest " + commitsRetained + " commits. ");
304-
List<CleanFileInfo> deletePaths = new ArrayList<>();
305304
Map<String, Pair<Boolean, List<CleanFileInfo>>> cleanFileInfoPerPartitionMap = new HashMap<>();
306305

307306
// Collect all the datafiles savepointed by all the savepoints
@@ -315,12 +314,12 @@ private Map<String, Pair<Boolean, List<CleanFileInfo>>> getFilesToCleanKeepingLa
315314
Option<HoodieInstant> earliestCommitToRetainOption = getEarliestCommitToRetain();
316315
HoodieInstant earliestCommitToRetain = earliestCommitToRetainOption.get();
317316
// add active files
318-
List<Pair<String, List<HoodieFileGroup>>> fileGroups = fileSystemView.getAllFileGroups(partitionPaths).collect(Collectors.toList());
319-
for (Pair<String, List<HoodieFileGroup>> pairFileGroup : fileGroups) {
320-
317+
List<Pair<String, List<HoodieFileGroup>>> fileGroupsPerPartition = fileSystemView.getAllFileGroups(partitionPaths).collect(Collectors.toList());
318+
for (Pair<String, List<HoodieFileGroup>> partitionFileGroupList : fileGroupsPerPartition) {
319+
List<CleanFileInfo> deletePaths = new ArrayList<>(getReplacedFilesEligibleToClean(savepointedFiles, partitionFileGroupList.getLeft(), earliestCommitToRetainOption));
321320
// all replaced file groups before earliestCommitToRetain are eligible to clean
322-
deletePaths.addAll(getReplacedFilesEligibleToClean(savepointedFiles, pairFileGroup.getLeft(), earliestCommitToRetainOption));
323-
for (HoodieFileGroup fileGroup : pairFileGroup.getRight()) {
321+
deletePaths.addAll(getReplacedFilesEligibleToClean(savepointedFiles, partitionFileGroupList.getLeft(), earliestCommitToRetainOption));
322+
for (HoodieFileGroup fileGroup : partitionFileGroupList.getRight()) {
324323
List<FileSlice> fileSliceList = fileGroup.getAllFileSlices().collect(Collectors.toList());
325324

326325
if (fileSliceList.isEmpty()) {
@@ -391,10 +390,10 @@ private Map<String, Pair<Boolean, List<CleanFileInfo>>> getFilesToCleanKeepingLa
391390
}
392391
}
393392
// if there are no valid file groups for the partition, mark it to be deleted
394-
if (fileGroups.isEmpty()) {
393+
if (partitionFileGroupList.getValue().isEmpty()) {
395394
toDeletePartition = true;
396395
}
397-
cleanFileInfoPerPartitionMap.put(pairFileGroup.getLeft(), Pair.of(toDeletePartition, deletePaths));
396+
cleanFileInfoPerPartitionMap.put(partitionFileGroupList.getLeft(), Pair.of(toDeletePartition, deletePaths));
398397
}
399398
}
400399
return cleanFileInfoPerPartitionMap;

‎hudi-common/src/main/java/org/apache/hudi/common/table/view/AbstractTableFileSystemView.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -737,8 +737,8 @@ public final Stream<HoodieFileGroup> getAllFileGroups(String partitionStr) {
737737
}
738738

739739
@Override
740-
public final Stream<Pair<String, List<HoodieFileGroup>>> getAllFileGroups(List<String> partitionStr) {
741-
return getAllFileGroupsIncludingReplaced(partitionStr)
740+
public final Stream<Pair<String, List<HoodieFileGroup>>> getAllFileGroups(List<String> partitionPaths) {
741+
return getAllFileGroupsIncludingReplaced(partitionPaths)
742742
.map(pair -> Pair.of(pair.getLeft(), pair.getRight().stream().filter(fg -> !isFileGroupReplaced(fg)).collect(Collectors.toList())));
743743
}
744744

‎hudi-common/src/main/java/org/apache/hudi/common/table/view/PriorityBasedFileSystemView.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -205,8 +205,8 @@ public Stream<HoodieFileGroup> getAllFileGroups(String partitionPath) {
205205
}
206206

207207
@Override
208-
public Stream<Pair<String, List<HoodieFileGroup>>> getAllFileGroups(List<String> partitionPath) {
209-
return execute(partitionPath, preferredView::getAllFileGroups, secondaryView::getAllFileGroups);
208+
public Stream<Pair<String, List<HoodieFileGroup>>> getAllFileGroups(List<String> partitionPaths) {
209+
return execute(partitionPaths, preferredView::getAllFileGroups, secondaryView::getAllFileGroups);
210210
}
211211

212212
@Override

‎hudi-common/src/main/java/org/apache/hudi/common/table/view/RemoteHoodieTableFileSystemView.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -380,9 +380,9 @@ public Stream<HoodieFileGroup> getAllFileGroups(String partitionPath) {
380380
}
381381

382382
@Override
383-
public Stream<Pair<String, List<HoodieFileGroup>>> getAllFileGroups(List<String> partitionPathList) {
383+
public Stream<Pair<String, List<HoodieFileGroup>>> getAllFileGroups(List<String> partitionPaths) {
384384
ArrayList<Pair<String, List<HoodieFileGroup>>> fileGroupPerPartitionList = new ArrayList<>();
385-
for (String partitionPath : partitionPathList) {
385+
for (String partitionPath : partitionPaths) {
386386
Stream<HoodieFileGroup> fileGroup = getAllFileGroups(partitionPath);
387387
fileGroupPerPartitionList.add(Pair.of(partitionPath, fileGroup.collect(Collectors.toList())));
388388
}

‎hudi-common/src/main/java/org/apache/hudi/common/table/view/TableFileSystemView.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -149,7 +149,7 @@ interface SliceView extends SliceViewWithLatestSlice {
149149
*/
150150
Stream<HoodieFileGroup> getAllFileGroups(String partitionPath);
151151

152-
Stream<Pair<String, List<HoodieFileGroup>>> getAllFileGroups(List<String> partitionPath);
152+
Stream<Pair<String, List<HoodieFileGroup>>> getAllFileGroups(List<String> partitionPaths);
153153

154154
/**
155155
* Return Pending Compaction Operations.

0 commit comments

Comments
 (0)