You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by GitBox <gi...@apache.org> on 2022/04/11 04:19:47 UTC
[GitHub] [hudi] nsivabalan commented on a diff in pull request #4957: [HUDI-3406] Rollback incorrectly relying on FS listing instead of Com…
nsivabalan commented on code in PR #4957:
URL: https://github.com/apache/hudi/pull/4957#discussion_r846926292
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/ListingBasedRollbackStrategy.java:
##########
@@ -57,20 +83,260 @@ public ListingBasedRollbackStrategy(HoodieTable table,
@Override
public List<HoodieRollbackRequest> getRollbackRequests(HoodieInstant instantToRollback) {
try {
- List<ListingBasedRollbackRequest> rollbackRequests = null;
- if (table.getMetaClient().getTableType() == HoodieTableType.COPY_ON_WRITE) {
- rollbackRequests = RollbackUtils.generateRollbackRequestsByListingCOW(context,
- table.getMetaClient().getBasePath());
- } else {
- rollbackRequests = RollbackUtils
- .generateRollbackRequestsUsingFileListingMOR(instantToRollback, table, context);
- }
- List<HoodieRollbackRequest> listingBasedRollbackRequests = new ListingBasedRollbackHelper(table.getMetaClient(), config)
- .getRollbackRequestsForRollbackPlan(context, instantToRollback, rollbackRequests);
- return listingBasedRollbackRequests;
- } catch (IOException e) {
+ HoodieTableMetaClient metaClient = table.getMetaClient();
+ List<String> partitionPaths =
+ FSUtils.getAllPartitionPaths(context, table.getMetaClient().getBasePath(), false, false);
+ int numPartitions = Math.max(Math.min(partitionPaths.size(), config.getRollbackParallelism()), 1);
+
+ context.setJobStatus(this.getClass().getSimpleName(), "Creating Listing Rollback Plan");
+
+ HoodieTableType tableType = table.getMetaClient().getTableType();
+ String baseFileExtension = getBaseFileExtension(metaClient);
+ Option<HoodieCommitMetadata> commitMetadataOptional = getHoodieCommitMetadata(metaClient, instantToRollback);
+ Boolean isCommitMetadataCompleted = checkCommitMetadataCompleted(instantToRollback, commitMetadataOptional);
+
+ return context.flatMap(partitionPaths, partitionPath -> {
+ List<HoodieRollbackRequest> hoodieRollbackRequests = new ArrayList<>(partitionPaths.size());
+ FileStatus[] filesToDelete =
+ fetchFilesFromInstant(instantToRollback, partitionPath, metaClient.getBasePath(), baseFileExtension,
+ metaClient.getFs(), commitMetadataOptional, isCommitMetadataCompleted);
+
+ if (HoodieTableType.COPY_ON_WRITE == tableType) {
+ hoodieRollbackRequests.add(getHoodieRollbackRequest(partitionPath, filesToDelete));
+ } else if (HoodieTableType.MERGE_ON_READ == tableType) {
+ String commit = instantToRollback.getTimestamp();
+ HoodieActiveTimeline activeTimeline = table.getMetaClient().reloadActiveTimeline();
+ switch (instantToRollback.getAction()) {
+ case HoodieTimeline.COMMIT_ACTION:
+ case HoodieTimeline.REPLACE_COMMIT_ACTION:
+ hoodieRollbackRequests.add(getHoodieRollbackRequest(partitionPath, filesToDelete));
+ break;
+ case HoodieTimeline.COMPACTION_ACTION:
+ // If there is no delta commit present after the current commit (if compaction), no action, else we
+ // need to make sure that a compaction commit rollback also deletes any log files written as part of the
+ // succeeding deltacommit.
+ boolean higherDeltaCommits =
Review Comment:
should have been named noHigherDeltaCommits
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/ListingBasedRollbackStrategy.java:
##########
@@ -57,20 +83,260 @@ public ListingBasedRollbackStrategy(HoodieTable table,
@Override
public List<HoodieRollbackRequest> getRollbackRequests(HoodieInstant instantToRollback) {
try {
- List<ListingBasedRollbackRequest> rollbackRequests = null;
- if (table.getMetaClient().getTableType() == HoodieTableType.COPY_ON_WRITE) {
- rollbackRequests = RollbackUtils.generateRollbackRequestsByListingCOW(context,
- table.getMetaClient().getBasePath());
- } else {
- rollbackRequests = RollbackUtils
- .generateRollbackRequestsUsingFileListingMOR(instantToRollback, table, context);
- }
- List<HoodieRollbackRequest> listingBasedRollbackRequests = new ListingBasedRollbackHelper(table.getMetaClient(), config)
- .getRollbackRequestsForRollbackPlan(context, instantToRollback, rollbackRequests);
- return listingBasedRollbackRequests;
- } catch (IOException e) {
+ HoodieTableMetaClient metaClient = table.getMetaClient();
+ List<String> partitionPaths =
+ FSUtils.getAllPartitionPaths(context, table.getMetaClient().getBasePath(), false, false);
+ int numPartitions = Math.max(Math.min(partitionPaths.size(), config.getRollbackParallelism()), 1);
+
+ context.setJobStatus(this.getClass().getSimpleName(), "Creating Listing Rollback Plan");
+
+ HoodieTableType tableType = table.getMetaClient().getTableType();
+ String baseFileExtension = getBaseFileExtension(metaClient);
+ Option<HoodieCommitMetadata> commitMetadataOptional = getHoodieCommitMetadata(metaClient, instantToRollback);
+ Boolean isCommitMetadataCompleted = checkCommitMetadataCompleted(instantToRollback, commitMetadataOptional);
+
+ return context.flatMap(partitionPaths, partitionPath -> {
+ List<HoodieRollbackRequest> hoodieRollbackRequests = new ArrayList<>(partitionPaths.size());
+ FileStatus[] filesToDelete =
+ fetchFilesFromInstant(instantToRollback, partitionPath, metaClient.getBasePath(), baseFileExtension,
+ metaClient.getFs(), commitMetadataOptional, isCommitMetadataCompleted);
+
+ if (HoodieTableType.COPY_ON_WRITE == tableType) {
+ hoodieRollbackRequests.add(getHoodieRollbackRequest(partitionPath, filesToDelete));
+ } else if (HoodieTableType.MERGE_ON_READ == tableType) {
+ String commit = instantToRollback.getTimestamp();
+ HoodieActiveTimeline activeTimeline = table.getMetaClient().reloadActiveTimeline();
+ switch (instantToRollback.getAction()) {
+ case HoodieTimeline.COMMIT_ACTION:
+ case HoodieTimeline.REPLACE_COMMIT_ACTION:
+ hoodieRollbackRequests.add(getHoodieRollbackRequest(partitionPath, filesToDelete));
+ break;
+ case HoodieTimeline.COMPACTION_ACTION:
+ // If there is no delta commit present after the current commit (if compaction), no action, else we
+ // need to make sure that a compaction commit rollback also deletes any log files written as part of the
+ // succeeding deltacommit.
+ boolean higherDeltaCommits =
+ !activeTimeline.getDeltaCommitTimeline().filterCompletedInstants().findInstantsAfter(commit, 1)
+ .empty();
+ if (higherDeltaCommits) {
+ // Rollback of a compaction action with no higher deltacommit means that the compaction is scheduled
+ // and has not yet finished. In this scenario we should delete only the newly created base files
+ // and not corresponding base commit log files created with this as baseCommit since updates would
+ // have been written to the log files.
+ hoodieRollbackRequests.add(getHoodieRollbackRequest(partitionPath,
+ listFilesToBeDeleted(instantToRollback.getTimestamp(), baseFileExtension, partitionPath,
+ metaClient.getFs())));
+ } else {
+ // No deltacommits present after this compaction commit (inflight or requested). In this case, we
+ // can also delete any log files that were created with this compaction commit as base
Review Comment:
not sure the comment is clear. if there are no delta commits after the compaction commit that is being rolledback, why do we need to delete log files? there should not be any log files right.
Also, irrespective of whether there are any log files or not (if else condition from L121 to 134), we can just do
```
listFilesToBeDeleted(instantToRollback.getTimestamp(), baseFileExtension, partitionPath,
metaClient.getFs())
```
for compaction commit which will return all files w/ matching commit time and we should be good right? or am I missing anything here ?
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/ListingBasedRollbackStrategy.java:
##########
@@ -57,20 +83,260 @@ public ListingBasedRollbackStrategy(HoodieTable table,
@Override
public List<HoodieRollbackRequest> getRollbackRequests(HoodieInstant instantToRollback) {
try {
- List<ListingBasedRollbackRequest> rollbackRequests = null;
- if (table.getMetaClient().getTableType() == HoodieTableType.COPY_ON_WRITE) {
- rollbackRequests = RollbackUtils.generateRollbackRequestsByListingCOW(context,
- table.getMetaClient().getBasePath());
- } else {
- rollbackRequests = RollbackUtils
- .generateRollbackRequestsUsingFileListingMOR(instantToRollback, table, context);
- }
- List<HoodieRollbackRequest> listingBasedRollbackRequests = new ListingBasedRollbackHelper(table.getMetaClient(), config)
- .getRollbackRequestsForRollbackPlan(context, instantToRollback, rollbackRequests);
- return listingBasedRollbackRequests;
- } catch (IOException e) {
+ HoodieTableMetaClient metaClient = table.getMetaClient();
+ List<String> partitionPaths =
+ FSUtils.getAllPartitionPaths(context, table.getMetaClient().getBasePath(), false, false);
+ int numPartitions = Math.max(Math.min(partitionPaths.size(), config.getRollbackParallelism()), 1);
+
+ context.setJobStatus(this.getClass().getSimpleName(), "Creating Listing Rollback Plan");
+
+ HoodieTableType tableType = table.getMetaClient().getTableType();
+ String baseFileExtension = getBaseFileExtension(metaClient);
+ Option<HoodieCommitMetadata> commitMetadataOptional = getHoodieCommitMetadata(metaClient, instantToRollback);
+ Boolean isCommitMetadataCompleted = checkCommitMetadataCompleted(instantToRollback, commitMetadataOptional);
+
+ return context.flatMap(partitionPaths, partitionPath -> {
+ List<HoodieRollbackRequest> hoodieRollbackRequests = new ArrayList<>(partitionPaths.size());
+ FileStatus[] filesToDelete =
+ fetchFilesFromInstant(instantToRollback, partitionPath, metaClient.getBasePath(), baseFileExtension,
+ metaClient.getFs(), commitMetadataOptional, isCommitMetadataCompleted);
+
+ if (HoodieTableType.COPY_ON_WRITE == tableType) {
+ hoodieRollbackRequests.add(getHoodieRollbackRequest(partitionPath, filesToDelete));
+ } else if (HoodieTableType.MERGE_ON_READ == tableType) {
+ String commit = instantToRollback.getTimestamp();
+ HoodieActiveTimeline activeTimeline = table.getMetaClient().reloadActiveTimeline();
+ switch (instantToRollback.getAction()) {
+ case HoodieTimeline.COMMIT_ACTION:
+ case HoodieTimeline.REPLACE_COMMIT_ACTION:
+ hoodieRollbackRequests.add(getHoodieRollbackRequest(partitionPath, filesToDelete));
+ break;
+ case HoodieTimeline.COMPACTION_ACTION:
+ // If there is no delta commit present after the current commit (if compaction), no action, else we
+ // need to make sure that a compaction commit rollback also deletes any log files written as part of the
+ // succeeding deltacommit.
+ boolean higherDeltaCommits =
+ !activeTimeline.getDeltaCommitTimeline().filterCompletedInstants().findInstantsAfter(commit, 1)
+ .empty();
+ if (higherDeltaCommits) {
+ // Rollback of a compaction action with no higher deltacommit means that the compaction is scheduled
+ // and has not yet finished. In this scenario we should delete only the newly created base files
+ // and not corresponding base commit log files created with this as baseCommit since updates would
+ // have been written to the log files.
+ hoodieRollbackRequests.add(getHoodieRollbackRequest(partitionPath,
+ listFilesToBeDeleted(instantToRollback.getTimestamp(), baseFileExtension, partitionPath,
+ metaClient.getFs())));
+ } else {
+ // No deltacommits present after this compaction commit (inflight or requested). In this case, we
+ // can also delete any log files that were created with this compaction commit as base
+ // commit.
+ hoodieRollbackRequests.add(getHoodieRollbackRequest(partitionPath, filesToDelete));
+ }
+ break;
+ case HoodieTimeline.DELTA_COMMIT_ACTION:
+ // --------------------------------------------------------------------------------------------------
+ // (A) The following cases are possible if index.canIndexLogFiles and/or index.isGlobal
+ // --------------------------------------------------------------------------------------------------
+ // (A.1) Failed first commit - Inserts were written to log files and HoodieWriteStat has no entries. In
+ // this scenario we would want to delete these log files.
+ // (A.2) Failed recurring commit - Inserts/Updates written to log files. In this scenario,
+ // HoodieWriteStat will have the baseCommitTime for the first log file written, add rollback blocks.
+ // (A.3) Rollback triggered for first commit - Inserts were written to the log files but the commit is
+ // being reverted. In this scenario, HoodieWriteStat will be `null` for the attribute prevCommitTime and
+ // and hence will end up deleting these log files. This is done so there are no orphan log files
+ // lying around.
+ // (A.4) Rollback triggered for recurring commits - Inserts/Updates are being rolled back, the actions
+ // taken in this scenario is a combination of (A.2) and (A.3)
+ // ---------------------------------------------------------------------------------------------------
+ // (B) The following cases are possible if !index.canIndexLogFiles and/or !index.isGlobal
+ // ---------------------------------------------------------------------------------------------------
+ // (B.1) Failed first commit - Inserts were written to base files and HoodieWriteStat has no entries.
+ // In this scenario, we delete all the base files written for the failed commit.
+ // (B.2) Failed recurring commits - Inserts were written to base files and updates to log files. In
+ // this scenario, perform (A.1) and for updates written to log files, write rollback blocks.
+ // (B.3) Rollback triggered for first commit - Same as (B.1)
+ // (B.4) Rollback triggered for recurring commits - Same as (B.2) plus we need to delete the log files
+ // as well if the base base file gets deleted.
+ HoodieCommitMetadata commitMetadata = HoodieCommitMetadata.fromBytes(
+ table.getMetaClient().getCommitTimeline().getInstantDetails(instantToRollback).get(),
+ HoodieCommitMetadata.class);
+
+ // In case all data was inserts and the commit failed, delete the file belonging to that commit
+ // We do not know fileIds for inserts (first inserts are either log files or base files),
+ // delete all files for the corresponding failed commit, if present (same as COW)
+ hoodieRollbackRequests.add(getHoodieRollbackRequest(partitionPath, filesToDelete));
+
+ // append rollback blocks for updates and inserts as A.2 and B.2
+ if (commitMetadata.getPartitionToWriteStats().containsKey(partitionPath)) {
+ hoodieRollbackRequests.addAll(
+ getRollbackRequestToAppend(partitionPath, instantToRollback, commitMetadata, table));
+ }
+ break;
+ default:
+ throw new HoodieRollbackException("Unknown listing type, during rollback of " + instantToRollback);
+ }
+ } else {
+ throw new HoodieRollbackException(
+ String.format("Unsupported table type: %s, during listing rollback of %s", tableType, instantToRollback));
+ }
+ return hoodieRollbackRequests.stream();
+ }, numPartitions);
+ } catch (Exception e) {
LOG.error("Generating rollback requests failed for " + instantToRollback.getTimestamp(), e);
throw new HoodieRollbackException("Generating rollback requests failed for " + instantToRollback.getTimestamp(), e);
}
}
+
+ private String getBaseFileExtension(HoodieTableMetaClient metaClient) {
+ return metaClient.getTableConfig().getBaseFileFormat().getFileExtension();
+ }
+
+ @NotNull
+ private HoodieRollbackRequest getHoodieRollbackRequest(String partitionPath, FileStatus[] filesToDeletedStatus) {
+ List<String> filesToDelete = getFilesToBeDeleted(filesToDeletedStatus);
+ return new HoodieRollbackRequest(
+ partitionPath, EMPTY_STRING, EMPTY_STRING, filesToDelete, Collections.emptyMap());
+ }
+
+ @NotNull
+ private List<String> getFilesToBeDeleted(FileStatus[] dataFilesToDeletedStatus) {
+ return Arrays.stream(dataFilesToDeletedStatus).map(fileStatus -> {
+ String dataFileToBeDeleted = fileStatus.getPath().toString();
+ // strip scheme E.g: file:/var/folders
+ return dataFileToBeDeleted.substring(dataFileToBeDeleted.indexOf(":") + 1);
+ }).collect(Collectors.toList());
+ }
+
+ private FileStatus[] listFilesToBeDeleted(String commit, String basefileExtension, String partitionPath,
+ FileSystem fs) throws IOException {
+ LOG.info("Collecting files to be cleaned/rolledback up for path " + partitionPath + " and commit " + commit);
+ PathFilter filter = (path) -> {
+ if (path.toString().contains(basefileExtension)) {
+ String fileCommitTime = FSUtils.getCommitTime(path.getName());
+ return commit.equals(fileCommitTime);
+ }
+ return false;
+ };
+ return fs.listStatus(FSUtils.getPartitionPath(config.getBasePath(), partitionPath), filter);
+ }
+
+ private FileStatus[] fetchFilesFromInstant(HoodieInstant instantToRollback, String partitionPath, String basePath,
+ String baseFileExtension, HoodieWrapperFileSystem fs,
+ Option<HoodieCommitMetadata> commitMetadataOptional,
+ Boolean isCommitMetadataCompleted) throws IOException {
+ if (isCommitMetadataCompleted) {
+ return fetchFilesFromCommitMetadata(instantToRollback, partitionPath, basePath, commitMetadataOptional.get(),
+ baseFileExtension, fs);
+ } else {
+ return fetchFilesFromListFiles(instantToRollback, partitionPath, basePath, baseFileExtension, fs);
+ }
+ }
+
+ private FileStatus[] fetchFilesFromCommitMetadata(HoodieInstant instantToRollback, String partitionPath,
+ String basePath, HoodieCommitMetadata commitMetadata,
+ String baseFileExtension, HoodieWrapperFileSystem fs)
+ throws IOException {
+ SerializablePathFilter pathFilter = getSerializablePathFilter(baseFileExtension, instantToRollback.getTimestamp());
+ Path[] filePaths = getFilesFromCommitMetadata(basePath, commitMetadata, partitionPath);
+
+ return fs.listStatus(filePaths, pathFilter);
+ }
+
+ private FileStatus[] fetchFilesFromListFiles(HoodieInstant instantToRollback, String partitionPath, String basePath,
+ String baseFileExtension, HoodieWrapperFileSystem fs)
+ throws IOException {
+ SerializablePathFilter pathFilter = getSerializablePathFilter(baseFileExtension, instantToRollback.getTimestamp());
+ Path[] filePaths = listFilesToBeDeleted(basePath, partitionPath);
Review Comment:
shouldn't this be getFilesInPartitions() ?
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org