Skip to content

[HUDI-1740] Fix insert-overwrite API archival #2784

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
vinothchandar merged 21 commits into from
May 21, 2021
Merged

[HUDI-1740] Fix insert-overwrite API archival #2784

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
21 commits
Select commit Hold shift + click to select a range
80a46cb
[HUDI-1740] fix problem of archiving replace commits
yorkzero831 Apr 5, 2021
d3d5b40
Merge branch 'master' of https://github.com/apache/hudi into HUDI-174…
zherenyu831 Apr 6, 2021
49cfe0d
[HUDI-1740] Fix problem when getting empty replacecommit.requested
zherenyu831 Apr 6, 2021
5572b9f
[HUDI-1740] Fix insert overwrite API archival
ssdong Apr 7, 2021
e2e7870
[HUDI-1740] Ignore null RequestedReplaceMeta when archiving
zherenyu831 Apr 8, 2021
b2c037a
[HUDI-1740] revert logic of in-flight commit
zherenyu831 Apr 8, 2021
e491d1c
[HUDI-1740] Fix format problem
zherenyu831 Apr 8, 2021
917fa7c
[HUDI-1740] fix failed test
zherenyu831 Apr 8, 2021
09ac4b4
Improved the logic of handling empty and non-empty requested/inflight…
ssdong Apr 10, 2021
cc724b8
Remove unused imports
ssdong Apr 10, 2021
ab27482
Comments resolved
ssdong Apr 12, 2021
de8fbc5
Trigger CI build
ssdong Apr 13, 2021
d938e0a
Trigger CI build
ssdong Apr 13, 2021
7d05046
Merge branch 'master' into HUDI-1740-insert-overwrite-api-archival
ssdong Apr 15, 2021
af963ac
Use Option and update log message
ssdong Apr 19, 2021
2327417
Trigger CI build
ssdong Apr 19, 2021
a285561
Fix unit test
ssdong Apr 19, 2021
39c062d
Trigger CI build
ssdong Apr 19, 2021
6bfc9fb
Trigger CI build
ssdong Apr 19, 2021
de884c6
Merge branch 'master' into HUDI-1740-insert-overwrite-api-archival
ssdong Apr 30, 2021
e47c38f
Fix unit tests and typo
ssdong May 1, 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
5 changes: 5 additions & 0 deletions ...client/hudi-client-common/src/main/java/org/apache/hudi/client/ReplaceArchivalHelper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,6 +68,11 @@ public static org.apache.hudi.avro.model.HoodieReplaceCommitMetadata convertRepl
public static boolean deleteReplacedFileGroups(HoodieEngineContext context, HoodieTableMetaClient metaClient,
TableFileSystemView fileSystemView,
HoodieInstant instant, List<String> replacedPartitions) {
// There is no file id to be replaced in the very first replace commit file for insert overwrite operation
if (replacedPartitions.isEmpty()) {
LOG.warn("Found empty partitionToReplaceFileIds");
Comment thread
ssdong marked this conversation as resolved.
Outdated
return true;
}
context.setJobStatus(ReplaceArchivalHelper.class.getSimpleName(), "Delete replaced file groups");
List<Boolean> f = context.map(replacedPartitions, partition -> {
Stream<FileSlice> fileSlices = fileSystemView.getReplacedFileGroupsBeforeOrOn(instant.getTimestamp(), partition)
Expand Down
32 changes: 21 additions & 11 deletions ...udi-client-common/src/main/java/org/apache/hudi/client/utils/MetadataConversionUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,14 +37,18 @@
import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.common.table.timeline.TimelineMetadataUtils;
import org.apache.hudi.common.util.CleanerUtils;
import org.apache.hudi.common.util.ClusteringUtils;
import org.apache.hudi.common.util.CompactionUtils;
import org.apache.hudi.common.util.Option;
import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;

/**
* Helper class to convert between different action related payloads and {@link HoodieArchivedMetaEntry}.
*/
public class MetadataConversionUtils {

private static final Logger LOG = LogManager.getLogger(MetadataConversionUtils.class);

public static HoodieArchivedMetaEntry createMetaWrapper(HoodieInstant hoodieInstant, HoodieTableMetaClient metaClient) throws IOException {
HoodieArchivedMetaEntry archivedMetaWrapper = new HoodieArchivedMetaEntry();
archivedMetaWrapper.setCommitTime(hoodieInstant.getTimestamp());
Expand Down Expand Up @@ -72,9 +76,14 @@ public static HoodieArchivedMetaEntry createMetaWrapper(HoodieInstant hoodieInst
HoodieReplaceCommitMetadata replaceCommitMetadata = HoodieReplaceCommitMetadata
.fromBytes(metaClient.getActiveTimeline().getInstantDetails(hoodieInstant).get(), HoodieReplaceCommitMetadata.class);
archivedMetaWrapper.setHoodieReplaceCommitMetadata(ReplaceArchivalHelper.convertReplaceCommitMetadata(replaceCommitMetadata));
} else if (hoodieInstant.isInflight()) {
// inflight replacecommit files have the same meta data body as HoodieCommitMetadata
Comment thread
ssdong marked this conversation as resolved.
// so we could re-use it without further creating an inflight extension
HoodieCommitMetadata inflightCommitMetadata = HoodieCommitMetadata
.fromBytes(metaClient.getActiveTimeline().getInstantDetails(hoodieInstant).get(), HoodieCommitMetadata.class);
archivedMetaWrapper.setHoodieInflightReplaceMetadata(convertCommitMetadata(inflightCommitMetadata));
} else {
HoodieRequestedReplaceMetadata requestedReplaceMetadata =
ClusteringUtils.getRequestedReplaceMetadata(metaClient, hoodieInstant).get();

@ssdong ssdong Apr 10, 2021 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Using ClusteringUtils.getRequestedReplaceMetadata is confusing to the reader even though it produces the same result. Part of this method's purpose is to ignore the inflight replacecommit file and delegate it to requested replacecommit file to retrieve the clustering plan. However, what we merely need is a fetch&deserialization method for requested metadata. Hence, I have created a separate method for this single purpose.

HoodieRequestedReplaceMetadata requestedReplaceMetadata = getRequestedReplaceMetadata(metaClient, hoodieInstant).get();
archivedMetaWrapper.setHoodieRequestedReplaceMetadata(requestedReplaceMetadata);
}
archivedMetaWrapper.setActionType(ActionType.replacecommit.name());
Expand Down Expand Up @@ -105,14 +114,15 @@ public static HoodieArchivedMetaEntry createMetaWrapper(HoodieInstant hoodieInst
return archivedMetaWrapper;
}

public static HoodieArchivedMetaEntry createMetaWrapper(HoodieInstant hoodieInstant,
HoodieCommitMetadata hoodieCommitMetadata) {
HoodieArchivedMetaEntry archivedMetaWrapper = new HoodieArchivedMetaEntry();
archivedMetaWrapper.setCommitTime(hoodieInstant.getTimestamp());
archivedMetaWrapper.setActionState(hoodieInstant.getState().name());
archivedMetaWrapper.setHoodieCommitMetadata(convertCommitMetadata(hoodieCommitMetadata));
archivedMetaWrapper.setActionType(ActionType.commit.name());
return archivedMetaWrapper;
public static Option<HoodieRequestedReplaceMetadata> getRequestedReplaceMetadata(HoodieTableMetaClient metaClient, HoodieInstant pendingReplaceInstant) throws IOException {
final HoodieInstant requestedInstant = HoodieTimeline.getReplaceCommitRequestedInstant(pendingReplaceInstant.getTimestamp());

Option<byte[]> content = metaClient.getActiveTimeline().getInstantDetails(requestedInstant);
if (!content.isPresent() || content.get().length == 0) {
LOG.warn("No content found in requested file for instant " + pendingReplaceInstant);
return Option.of(new HoodieRequestedReplaceMetadata());
Comment thread
ssdong marked this conversation as resolved.
Outdated
}
return Option.of(TimelineMetadataUtils.deserializeRequestedReplaceMetadata(content.get()));
}

public static org.apache.hudi.avro.model.HoodieCommitMetadata convertCommitMetadata(
Expand Down
5 changes: 2 additions & 3 deletions ...ient/hudi-client-common/src/main/java/org/apache/hudi/table/HoodieTimelineArchiveLog.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -296,7 +296,6 @@ private boolean deleteAllInstantsOlderorEqualsInAuxMetaFolder(HoodieInstant thre

public void archive(HoodieEngineContext context, List<HoodieInstant> instants) throws HoodieCommitException {
try {
HoodieTimeline commitTimeline = metaClient.getActiveTimeline().getAllCommitsTimeline().filterCompletedInstants();
Schema wrapperSchema = HoodieArchivedMetaEntry.getClassSchema();
LOG.info("Wrapper schema " + wrapperSchema.toString());
List<IndexedRecord> records = new ArrayList<>();
Expand All @@ -308,7 +307,7 @@ public void archive(HoodieEngineContext context, List<HoodieInstant> instants) t
}
try {
deleteAnyLeftOverMarkerFiles(context, hoodieInstant);
records.add(convertToAvroRecord(commitTimeline, hoodieInstant));
records.add(convertToAvroRecord(hoodieInstant));
if (records.size() >= this.config.getCommitArchivalBatchSize()) {
writeToFile(wrapperSchema, records);
}
Expand Down Expand Up @@ -365,7 +364,7 @@ private void writeToFile(Schema wrapperSchema, List<IndexedRecord> records) thro
}
}

private IndexedRecord convertToAvroRecord(HoodieTimeline commitTimeline, HoodieInstant hoodieInstant)
private IndexedRecord convertToAvroRecord(HoodieInstant hoodieInstant)
throws IOException {
return MetadataConversionUtils.createMetaWrapper(hoodieInstant, metaClient);
}
Expand Down
10 changes: 10 additions & 0 deletions ...t/hudi-client-common/src/test/java/org/apache/hudi/utils/TestMetadataConversionUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,6 +87,16 @@ public void testCompletedReplace() throws Exception {
assertEquals(metaEntry.getHoodieReplaceCommitMetadata().getOperationType(), WriteOperationType.INSERT_OVERWRITE.toString());
}

@Test
public void testInflightReplace() throws Exception {
String newCommitTime = HoodieTestTable.makeNewCommitTime();
createReplace(newCommitTime, WriteOperationType.INSERT_OVERWRITE_TABLE);
HoodieArchivedMetaEntry metaEntry = MetadataConversionUtils.createMetaWrapper(
new HoodieInstant(State.INFLIGHT, HoodieTimeline.REPLACE_COMMIT_ACTION, newCommitTime), metaClient);
assertEquals(metaEntry.getActionState(), State.INFLIGHT.toString());
assertEquals(metaEntry.getHoodieInflightReplaceMetadata().getOperationType(), WriteOperationType.INSERT_OVERWRITE_TABLE.toString());
}

@Test
public void testCompletedCommitOrDeltaCommit() throws Exception {
String newCommitTime = HoodieTestTable.makeNewCommitTime();
Expand Down
18 changes: 18 additions & 0 deletions ...ient/hudi-spark-client/src/test/java/org/apache/hudi/io/TestHoodieTimelineArchiveLog.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -240,6 +240,8 @@ public void testArchiveTableWithReplacedFiles() throws Exception {

int numCommits = 4;
int commitInstant = 100;
// The first replace commit file have empty replace file id
createReplaceMetadataWithoutReplaceFileId(String.valueOf(commitInstant));
for (int i = 0; i < numCommits; i++) {
createReplaceMetadata(String.valueOf(commitInstant));
commitInstant += 100;
Expand Down Expand Up @@ -493,6 +495,22 @@ public void testConvertCommitMetadata() {
assertEquals(expectedCommitMetadata.getOperationType(), WriteOperationType.INSERT.toString());
}

private void createReplaceMetadataWithoutReplaceFileId(String instantTime) throws Exception {
String baseFile = "file-" + instantTime;

// create replace instant without a previous replace commit
HoodieRequestedReplaceMetadata requestedReplaceMetadata = HoodieRequestedReplaceMetadata.newBuilder()
.setOperationType(WriteOperationType.INSERT_OVERWRITE_TABLE.toString())
.setVersion(1)
.setExtraMetadata(Collections.emptyMap())
.build();
HoodieReplaceCommitMetadata replaceCommitMetadata = new HoodieReplaceCommitMetadata();
replaceCommitMetadata.setOperationType(WriteOperationType.INSERT_OVERWRITE_TABLE);
HoodieTestTable.of(metaClient)
.addReplaceCommit(instantTime, requestedReplaceMetadata, replaceCommitMetadata)
.withBaseFilesInPartition(HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH, baseFile);
}

private void createReplaceMetadata(String instantTime) throws Exception {
String fileId1 = "file-" + instantTime + "-1";
String fileId2 = "file-" + instantTime + "-2";
Expand Down
8 changes: 8 additions & 0 deletions hudi-common/src/main/avro/HoodieArchivedMetaEntry.avsc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -112,6 +112,14 @@
"HoodieRequestedReplaceMetadata"
],
"default": null
},
{
"name":"HoodieInflightReplaceMetadata",
"type":[
"null",
"HoodieCommitMetadata"
],
"default": null
}
]
}
2 changes: 1 addition & 1 deletion 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 @@ -245,7 +245,7 @@ public final void reset() {
bootstrapIndex = null;

// Initialize with new Hoodie timeline.
init(metaClient, getTimeline());
init(metaClient, metaClient.reloadActiveTimeline());

@satishkotha satishkotha Apr 7, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

IIUC, this is breaking some fundamental assumptions. There are many places where we pass "trimmed" timeline for time-travel queries etc. You are replacing that with all instants from active timeline, which is not desired.

Is this change needed if we handle empty partitionToReplaceFileIds in archival?

@zherenyu831 zherenyu831 Apr 8, 2021

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.

@satishkotha
Since this part will called after archival, and the archived commits still in the timeline.
In the post process hudi will try to load the byte from them, and will cause IO error

@zherenyu831 zherenyu831 Apr 8, 2021

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.

I think the root problem is why we are calling reset() when close the timeline

@satishkotha satishkotha Apr 9, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

@bvaradar mentioned this is by design. Balaji, could you please help resolve this?

@ssdong ssdong Apr 10, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

I've put getTimeline() back since it should be irrelevant to the archival issue though I am still curious to learn why this is needed.

} finally {
writeLock.unlock();
}
Expand Down
4 changes: 2 additions & 2 deletions hudi-common/src/test/java/org/apache/hudi/common/testutils/FileCreateUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -160,8 +160,8 @@ public static void createRequestedReplaceCommit(String basePath, String instantT
createMetaFile(basePath, instantTime, HoodieTimeline.REQUESTED_REPLACE_COMMIT_EXTENSION, serializeRequestedReplaceMetadata(requestedReplaceMetadata).get());
}

public static void createInflightReplaceCommit(String basePath, String instantTime) throws IOException {
createMetaFile(basePath, instantTime, HoodieTimeline.INFLIGHT_REPLACE_COMMIT_EXTENSION);
public static void createInflightReplaceCommit(String basePath, String instantTime, HoodieReplaceCommitMetadata metadata) throws IOException {
createMetaFile(basePath, instantTime, HoodieTimeline.INFLIGHT_REPLACE_COMMIT_EXTENSION, metadata.toJsonString().getBytes(StandardCharsets.UTF_8));
}

public static void createCleanFile(String basePath, String instantTime, HoodieCleanMetadata metadata) throws IOException {
Expand Down
2 changes: 1 addition & 1 deletion hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestTable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -179,7 +179,7 @@ public HoodieTestTable addDeltaCommit(String instantTime) throws Exception {

public HoodieTestTable addReplaceCommit(String instantTime, HoodieRequestedReplaceMetadata requestedReplaceMetadata, HoodieReplaceCommitMetadata metadata) throws Exception {
createRequestedReplaceCommit(basePath, instantTime, requestedReplaceMetadata);
createInflightReplaceCommit(basePath, instantTime);
createInflightReplaceCommit(basePath, instantTime, metadata);
createReplaceCommit(basePath, instantTime, metadata);
currentInstantTime = instantTime;
metaClient = HoodieTableMetaClient.reload(metaClient);
Expand Down