lanyuanxiaoyao/hudi
1
0
Fork 0
Code Activity

[HUDI-4255] Make the flink merge and replace handle intermediate file visible (#5866)

Browse Source
This commit is contained in:
Danny Chan
2022-06-15 14:23:23 +08:00
committed by GitHub
parent 25bbff64cf
commit 0811bb38fb
3 changed files with 12 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandle.java
View File

@@ -195,8 +195,10 @@ public class HoodieMergeHandle<T extends HoodieRecordPayload, I, K, O> extends H
writeStatus.getStat().setFileId(fileId);
setWriteStatusPath();
// Create Marker file
createMarkerFile(partitionPath, newFileName);
// Create Marker file,
// uses name of `newFilePath` instead of `newFileName`
// in case the sub-class may roll over the file handle name.
createMarkerFile(partitionPath, newFilePath.getName());
// Create the writer for writing the new version file
fileWriter = createNewFileWriter(instantTime, newFilePath, hoodieTable, config,

4
hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/FlinkMergeAndReplaceHandle.java
View File

@@ -137,10 +137,8 @@ public class FlinkMergeAndReplaceHandle<T extends HoodieRecordPayload, I, K, O>
* Use the writeToken + "-" + rollNumber as the new writeToken of a mini-batch write.
*/
protected String newFileNameWithRollover(int rollNumber) {
// make the intermediate file as hidden
final String fileID = "." + this.fileId;
return FSUtils.makeBaseFileName(instantTime, writeToken + "-" + rollNumber,
fileID, hoodieTable.getBaseFileExtension());
this.fileId, hoodieTable.getBaseFileExtension());
}
@Override

8
hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/FlinkMergeHandle.java
View File

@@ -158,11 +158,17 @@ public class FlinkMergeHandle<T extends HoodieRecordPayload, I, K, O>
* Use the writeToken + "-" + rollNumber as the new writeToken of a mini-batch write.
*/
protected String newFileNameWithRollover(int rollNumber) {
// make the intermediate file as hidden
return FSUtils.makeBaseFileName(instantTime, writeToken + "-" + rollNumber,
this.fileId, hoodieTable.getBaseFileExtension());
}
@Override
protected void setWriteStatusPath() {
// if there was rollover, should set up the path as the initial new file path.
Path path = rolloverPaths.size() > 0 ? rolloverPaths.get(0) : newFilePath;
writeStatus.getStat().setPath(new Path(config.getBasePath()), path);
}
@Override
public List<WriteStatus> close() {
try {