Skip to content

[HUDI-4290] Fix fetchLatestBaseFiles to filter replaced filegroups #5941

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
codope merged 2 commits into from
Jun 23, 2022
Merged

[HUDI-4290] Fix fetchLatestBaseFiles to filter replaced filegroups #5941

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
0bf33ad
[HUDI-4290] Fix fetchLatestBaseFiles to filter replaced filegroups
codope Jun 22, 2022
42f683e
Separate out incremental sync fsview test with clustering
codope Jun 23, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 4 additions & 2 deletions hudi-common/src/main/java/org/apache/hudi/common/table/view/AbstractTableFileSystemView.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,8 +18,6 @@

package org.apache.hudi.common.table.view;

import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hudi.common.bootstrap.index.BootstrapIndex;
import org.apache.hudi.common.fs.FSUtils;
import org.apache.hudi.common.model.BootstrapBaseFileMapping;
Expand All @@ -41,6 +39,9 @@
import org.apache.hudi.common.util.ValidationUtils;
import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.exception.HoodieIOException;

import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;

Expand Down Expand Up @@ -973,6 +974,7 @@ Stream<FileSlice> fetchAllFileSlices(String partitionPath) {
*/
public Stream<HoodieBaseFile> fetchLatestBaseFiles(final String partitionPath) {
return fetchAllStoredFileGroups(partitionPath)
.filter(fg -> !isFileGroupReplaced(fg))

@yihua yihua Jun 22, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good catch! I see that getLatestBaseFiles(String partitionStr) filters out the replaced file groups. Should that API be used in Presto Hive connector? Also, should we audit all similar APIs regarding compaction and clustering?

Still, to be on par with fetchLatestBaseFiles(), this needs to be fixed anyway.

@codope codope Jun 23, 2022 •
edited
Loading

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is needed when metadata is not enabled in presto-hive connector then we let presto do the listing and just fetch base files using this API after adding to the view. getLatestBaseFiles(String partitionStr) does additional work, which we did not want for perf reasons.

should we audit all similar APIs regarding compaction and clustering?

I did check all other public APIs in this view class. This is the only one.

.map(fg -> Pair.of(fg.getFileGroupId(), getLatestBaseFile(fg)))
.filter(p -> p.getValue().isPresent())
.map(p -> addBootstrapBaseFileIfPresent(p.getKey(), p.getValue().get()));
Expand Down
20 changes: 17 additions & 3 deletions ...common/src/test/java/org/apache/hudi/common/table/view/TestHoodieTableFileSystemView.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,9 +18,6 @@

package org.apache.hudi.common.table.view;

import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsAction;
import org.apache.hudi.avro.model.HoodieClusteringPlan;
import org.apache.hudi.avro.model.HoodieCompactionPlan;
import org.apache.hudi.avro.model.HoodieFSPermission;
Expand Down Expand Up @@ -61,6 +58,9 @@
import org.apache.hudi.common.util.collection.ImmutablePair;
import org.apache.hudi.common.util.collection.Pair;

import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsAction;
import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;
import org.junit.jupiter.api.BeforeEach;
Expand Down Expand Up @@ -1437,6 +1437,12 @@ public void testReplaceFileIdIsExcludedInView() throws IOException {
assertFalse(roView.getLatestBaseFiles(partitionPath2)
.anyMatch(dfile -> dfile.getFileId().equals(fileId3) || dfile.getFileId().equals(fileId4)),
"No commit, should not find any data file");
assertFalse(((IncrementalTimelineSyncFileSystemView) fsView).fetchLatestBaseFiles(partitionPath1)
.anyMatch(dfile -> dfile.getFileId().equals(fileId1) || dfile.getFileId().equals(fileId2)),
"No commit, should not find any data file");
assertFalse(((IncrementalTimelineSyncFileSystemView) fsView).fetchLatestBaseFiles(partitionPath2)
.anyMatch(dfile -> dfile.getFileId().equals(fileId3) || dfile.getFileId().equals(fileId4)),
"No commit, should not find any data file");

// Only one commit
String commitTime1 = "1";
Expand Down Expand Up @@ -1472,6 +1478,14 @@ public void testReplaceFileIdIsExcludedInView() throws IOException {
.filter(dfile -> dfile.getFileId().equals(fileId3)).count());
assertEquals(0, roView.getLatestBaseFiles(partitionPath2)
.filter(dfile -> dfile.getFileId().equals(fileId4)).count());
assertEquals(0, ((IncrementalTimelineSyncFileSystemView) fsView).fetchLatestBaseFiles(partitionPath1)
.filter(dfile -> dfile.getFileId().equals(fileId1)).count());
assertEquals(fileName2, ((IncrementalTimelineSyncFileSystemView) fsView).fetchLatestBaseFiles(partitionPath1)
.filter(dfile -> dfile.getFileId().equals(fileId2)).findFirst().get().getFileName());
assertEquals(0, ((IncrementalTimelineSyncFileSystemView) fsView).fetchLatestBaseFiles(partitionPath2)
.filter(dfile -> dfile.getFileId().equals(fileId3)).count());
assertEquals(0, ((IncrementalTimelineSyncFileSystemView) fsView).fetchLatestBaseFiles(partitionPath2)
.filter(dfile -> dfile.getFileId().equals(fileId4)).count());

// ensure replacedFileGroupsBefore works with all instants
List<HoodieFileGroup> replacedOnInstant1 = fsView.getReplacedFileGroupsBeforeOrOn("0", partitionPath1).collect(Collectors.toList());
Expand Down