[HUDI-1319] Make async operations work with metadata table #2332
+162
−269
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -119,8 +119,8 @@ public class HoodieBackedTableMetadataWriter implements HoodieTableMetadataWrite | |
| enabled = true; | ||
|
|
||
| // Inline compaction and auto clean is required as we dont expose this table outside | ||
| ValidationUtils.checkArgument(!this.metadataWriteConfig.isAutoClean(), "Cleaning is controlled internally for Metadata table."); | ||
| ValidationUtils.checkArgument(!this.metadataWriteConfig.isInlineCompaction(), "Compaction is controlled internally for metadata table."); | ||
| // Metadata Table cannot have metadata listing turned on. (infinite loop, much?) | ||
| ValidationUtils.checkArgument(this.metadataWriteConfig.shouldAutoCommit(), "Auto commit is required for Metadata Table"); | ||
| ValidationUtils.checkArgument(!this.metadataWriteConfig.useFileListingMetadata(), "File listing cannot be used for Metadata Table"); | ||
|
|
@@ -148,7 +148,7 @@ public class HoodieBackedTableMetadataWriter implements HoodieTableMetadataWrite | |
| // may have occurred on the table. Hence, calling this always ensures that the metadata is brought in sync | ||
| // with the active timeline. | ||
| HoodieTimer timer = new HoodieTimer().startTimer(); | ||
| syncFromInstants(datasetMetaClient); | ||
| metrics.ifPresent(m -> m.updateMetrics(HoodieMetadataMetrics.SYNC_STR, timer.endTimer())); | ||
| } | ||
| } else { | ||
|
|
@@ -184,12 +184,14 @@ private HoodieWriteConfig createMetadataWriteConfig(HoodieWriteConfig writeConfi | |
| .forTable(tableName) | ||
| .withCompactionConfig(HoodieCompactionConfig.newBuilder() | ||
| .withAsyncClean(writeConfig.isMetadataAsyncClean()) | ||
| // we will trigger cleaning manually, to control the instant times | ||
| .withAutoClean(false) | ||
| .withCleanerParallelism(parallelism) | ||
| .withCleanerPolicy(HoodieCleaningPolicy.KEEP_LATEST_COMMITS) | ||
| .retainCommits(writeConfig.getMetadataCleanerCommitsRetained()) | ||
| .archiveCommitsWith(writeConfig.getMetadataMinCommitsToKeep(), writeConfig.getMetadataMaxCommitsToKeep()) | ||
| // we will trigger compaction manually, to control the instant times | ||
| .withInlineCompaction(false) | ||
| .withMaxNumDeltaCommitsBeforeCompaction(writeConfig.getMetadataCompactDeltaCommitMax()).build()) | ||
| .withParallelism(parallelism, parallelism) | ||
| .withDeleteParallelism(parallelism) | ||
|
|
@@ -376,7 +378,7 @@ private void bootstrapFromFilesystem(JavaSparkContext jsc, HoodieTableMetaClient | |
| * | ||
| * @param datasetMetaClient {@code HoodieTableMetaClient} for the dataset | ||
| */ | ||
| private void syncFromInstants(HoodieTableMetaClient datasetMetaClient) { | ||
| ValidationUtils.checkState(enabled, "Metadata table cannot be synced as it is not enabled"); | ||
|
|
||
| try { | ||
|
|
@@ -391,56 +393,35 @@ private void syncFromInstants(JavaSparkContext jsc, HoodieTableMetaClient datase | |
| final HoodieActiveTimeline timeline = datasetMetaClient.getActiveTimeline(); | ||
| for (HoodieInstant instant : instantsToSync) { | ||
| LOG.info("Syncing instant " + instant + " to metadata table"); | ||
| ValidationUtils.checkArgument(instant.isCompleted(), "Only completed instants can be synced."); | ||
|
|
||
| switch (instant.getAction()) { | ||
| case HoodieTimeline.CLEAN_ACTION: | ||
| HoodieCleanMetadata cleanMetadata = CleanerUtils.getCleanerMetadata(datasetMetaClient, instant); | ||
| update(cleanMetadata, instant.getTimestamp()); | ||
| break; | ||
| case HoodieTimeline.DELTA_COMMIT_ACTION: | ||
| case HoodieTimeline.COMMIT_ACTION: | ||
| case HoodieTimeline.COMPACTION_ACTION: | ||
| HoodieCommitMetadata commitMetadata = HoodieCommitMetadata.fromBytes( | ||
| timeline.getInstantDetails(instant).get(), HoodieCommitMetadata.class); | ||
| update(commitMetadata, instant.getTimestamp()); | ||
| break; | ||
| case HoodieTimeline.ROLLBACK_ACTION: | ||
| HoodieRollbackMetadata rollbackMetadata = TimelineMetadataUtils.deserializeHoodieRollbackMetadata( | ||
| timeline.getInstantDetails(instant).get()); | ||
| update(rollbackMetadata, instant.getTimestamp()); | ||
| break; | ||
| case HoodieTimeline.RESTORE_ACTION: | ||
| HoodieRestoreMetadata restoreMetadata = TimelineMetadataUtils.deserializeHoodieRestoreMetadata( | ||
| timeline.getInstantDetails(instant).get()); | ||
| update(restoreMetadata, instant.getTimestamp()); | ||
| break; | ||
| case HoodieTimeline.SAVEPOINT_ACTION: | ||
| // Nothing to be done here | ||
| break; | ||
| default: | ||
| throw new HoodieException("Unknown type of action " + instant.getAction()); | ||
| } | ||
| } | ||
| // re-init the table metadata, for any future writes. | ||
|
|
@@ -472,8 +453,7 @@ public void update(HoodieCommitMetadata commitMetadata, String instantTime) { | |
| writeStats.forEach(hoodieWriteStat -> { | ||
| String pathWithPartition = hoodieWriteStat.getPath(); | ||
| if (pathWithPartition == null) { | ||
| throw new HoodieMetadataException("Unable to find path in write stat to update metadata table " + hoodieWriteStat); | ||
| } | ||
|
|
||
| int offset = partition.equals(NON_PARTITIONED_NAME) ? 0 : partition.length() + 1; | ||
|
|
@@ -509,13 +489,6 @@ public void update(HoodieCleanerPlan cleanerPlan, String instantTime) { | |
| return; | ||
| } | ||
|
|
||
| List<HoodieRecord> records = new LinkedList<>(); | ||
| int[] fileDeleteCount = {0}; | ||
| cleanerPlan.getFilePathsToBeDeletedPerPartition().forEach((partition, deletedPathInfo) -> { | ||
|
|
@@ -725,6 +698,13 @@ private synchronized void commit(JavaSparkContext jsc, JavaRDD<HoodieRecord> rec | |
| throw new HoodieMetadataException("Failed to commit metadata table records at instant " + instantTime); | ||
| } | ||
| }); | ||
| // trigger cleaning, compaction, with suffixes based on the same instant time. This ensures that any future | ||
| // delta commits synced over will not have an instant time lesser than the last completed instant on the | ||
| // metadata table. | ||
| writeClient.clean(instantTime + "001"); | ||
|
Member
There was a problem hiding this comment. This is a good idea. I hope it does not break some function which may depend on yyyyMMddHHmmss format of the instantTime.
Member
Author
There was a problem hiding this comment. @prashantwason our tests use 001,002 etc. So code should be fine. but may be some metrics reporting code might have some issue. let me double check. good call
Member
Author
There was a problem hiding this comment.
Seems like the parser can handle this |
||
| if (writeClient.scheduleCompactionAtInstant(instantTime + "002", Option.empty())) { | ||
| writeClient.compact(instantTime + "002"); | ||
| } | ||
| } | ||
|
|
||
| // Update total size of the metadata and count of base/log files | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Clarification: Since we are not updating the metadata table here, for compactions the metadata table will be updated the next time the HoodieWriteClient is created.
But for commit operations the metadata table (may) be updated right after the commit completes in "postCommit" function. Right?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
yes correct. even for compaction, once the compaction commits, it will trigger a sync. I think we should ensure the sync only happens for commit operations (i.e only one allowed at the moment). I ll double check this.