Skip to content

[HUDI-1500] Support replace commit in DeltaSync with commit metadata preserved #3802

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
nsivabalan merged 3 commits into from
Oct 29, 2021
Merged

[HUDI-1500] Support replace commit in DeltaSync with commit metadata preserved #3802

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
e906d36
[HUDI-1500] Support replace commit in DeltaSync with commit metadata …
codope Oct 14, 2021
b63edfa
Test inline clustering with metadata preserved
codope Oct 25, 2021
481d388
Revert testAsyncClusteringService change
codope Oct 27, 2021
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
2 changes: 1 addition & 1 deletion ...lient/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieClusteringConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -133,7 +133,7 @@ public class HoodieClusteringConfig extends HoodieConfig {

public static final ConfigProperty<Boolean> PRESERVE_COMMIT_METADATA = ConfigProperty
.key("hoodie.clustering.preserve.commit.metadata")
.defaultValue(false)
.defaultValue(true)
.sinceVersion("0.9.0")
.withDocumentation("When rewriting data, preserves existing hoodie_commit_time");

Expand Down
4 changes: 1 addition & 3 deletions hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/DeltaSync.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -323,9 +323,7 @@ public Pair<SchemaProvider, Pair<String, JavaRDD<HoodieRecord>>> readFromSource(
// Retrieve the previous round checkpoints, if any
Option<String> resumeCheckpointStr = Option.empty();
if (commitTimelineOpt.isPresent()) {
// TODO: now not support replace action HUDI-1500
Option<HoodieInstant> lastCommit = commitTimelineOpt.get()
.filter(instant -> !instant.getAction().equals(HoodieTimeline.REPLACE_COMMIT_ACTION)).lastInstant();
Option<HoodieInstant> lastCommit = commitTimelineOpt.get().lastInstant();
if (lastCommit.isPresent()) {
HoodieCommitMetadata commitMetadata = HoodieCommitMetadata
.fromBytes(commitTimelineOpt.get().getInstantDetails(lastCommit.get()).get(), HoodieCommitMetadata.class);
Expand Down
21 changes: 15 additions & 6 deletions ...utilities/src/test/java/org/apache/hudi/utilities/functional/TestHoodieDeltaStreamer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -963,8 +963,9 @@ private void deltaStreamerTestRunner(HoodieDeltaStreamer ds, Function<Boolean, B
deltaStreamerTestRunner(ds, null, condition);
}

@Test
public void testInlineClustering() throws Exception {
@ParameterizedTest
@ValueSource(strings = {"true", "false"})
public void testInlineClustering(String preserveCommitMetadata) throws Exception {
String tableBasePath = dfsBasePath + "/inlineClustering";
// Keep it higher than batch-size to test continuous mode
int totalRecords = 3000;
Expand All @@ -973,7 +974,7 @@ public void testInlineClustering() throws Exception {
HoodieDeltaStreamer.Config cfg = TestHelpers.makeConfig(tableBasePath, WriteOperationType.UPSERT);
cfg.continuousMode = true;
cfg.tableType = HoodieTableType.MERGE_ON_READ.name();
cfg.configs.addAll(getAsyncServicesConfigs(totalRecords, "false", "true", "2", "", ""));
cfg.configs.addAll(getAsyncServicesConfigs(totalRecords, "false", "true", "2", "", "", preserveCommitMetadata));
HoodieDeltaStreamer ds = new HoodieDeltaStreamer(cfg, jsc);
deltaStreamerTestRunner(ds, cfg, (r) -> {
TestHelpers.assertAtLeastNCommits(2, tableBasePath, dfs);
Expand Down Expand Up @@ -1064,6 +1065,13 @@ public void testCleanerDeleteReplacedDataWithArchive(Boolean asyncClean) throws
}
}

private List<String> getAsyncServicesConfigs(int totalRecords, String autoClean, String inlineCluster, String inlineClusterMaxCommit,
String asyncCluster, String asyncClusterMaxCommit, String preserveCommitMetadata) {
List<String> configs = getAsyncServicesConfigs(totalRecords, autoClean, inlineCluster, inlineClusterMaxCommit, asyncCluster, asyncClusterMaxCommit);
configs.add(String.format("%s=%s", HoodieClusteringConfig.PRESERVE_COMMIT_METADATA.key(), preserveCommitMetadata));
return configs;
}

private List<String> getAsyncServicesConfigs(int totalRecords, String autoClean, String inlineCluster,
String inlineClusterMaxCommit, String asyncCluster, String asyncClusterMaxCommit) {
List<String> configs = new ArrayList<>();
Expand Down Expand Up @@ -1156,8 +1164,9 @@ public void testAsyncClusteringService() throws Exception {
});
}

@Test
public void testAsyncClusteringServiceWithCompaction() throws Exception {
@ParameterizedTest
@ValueSource(strings = {"true", "false"})
public void testAsyncClusteringServiceWithCompaction(String preserveCommitMetadata) throws Exception {
String tableBasePath = dfsBasePath + "/asyncClusteringCompaction";
// Keep it higher than batch-size to test continuous mode
int totalRecords = 3000;
Expand All @@ -1166,7 +1175,7 @@ public void testAsyncClusteringServiceWithCompaction() throws Exception {
HoodieDeltaStreamer.Config cfg = TestHelpers.makeConfig(tableBasePath, WriteOperationType.INSERT);
cfg.continuousMode = true;
cfg.tableType = HoodieTableType.MERGE_ON_READ.name();
cfg.configs.addAll(getAsyncServicesConfigs(totalRecords, "false", "", "", "true", "2"));
cfg.configs.addAll(getAsyncServicesConfigs(totalRecords, "false", "", "", "true", "2", preserveCommitMetadata));
HoodieDeltaStreamer ds = new HoodieDeltaStreamer(cfg, jsc);
deltaStreamerTestRunner(ds, cfg, (r) -> {
TestHelpers.assertAtLeastNCommits(2, tableBasePath, dfs);
Expand Down