[GOBBLIN-2226] Construct iceberg data files during commit step #4140
Conversation
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
Blazer-007
reviewed
Sep 5, 2025
Comment on lines
28
to
30
| public String getBase64EncodedDataFile() { | ||
| return this.base64EncodedDataFile; | ||
| } |
Comment on lines
193
to
205
| private Map<Path, FileStatus> calcSrcFileStatusByDestFilePath( | ||
| Map<Path, DataFile> destDataFileBySrcPath, | ||
| Map<Path, Path> destPathToSrcPath) throws IOException { | ||
| Function<Path, FileStatus> getFileStatus = CheckedExceptionFunction.wrapToTunneled(this.sourceFs::getFileStatus); | ||
| Map<Path, FileStatus> srcFileStatusByDestFilePath = new ConcurrentHashMap<>(); | ||
| final Map<Path, FileStatus> srcFileStatusByDestFilePath = new ConcurrentHashMap<>(); | ||
| try { | ||
| srcFileStatusByDestFilePath = destDataFileBySrcPath.entrySet() | ||
| destDataFileBySrcPath.entrySet() | ||
| .parallelStream() | ||
| .collect(Collectors.toConcurrentMap(entry -> new Path(entry.getValue().path().toString()), | ||
| entry -> getFileStatus.apply(entry.getKey()))); | ||
| .forEach(entry -> { | ||
| Path destPath = new Path(entry.getValue().path().toString()); | ||
| destPathToSrcPath.put(destPath, entry.getKey()); | ||
| srcFileStatusByDestFilePath.put(destPath, getFileStatus.apply(entry.getKey())); | ||
| }); |
There was a problem hiding this comment.
nit : Let's move this logic to caller itself and see if we can parallelize that for-loop to reduce runtime
Comment on lines
358
to
369
| List<DataFile> icebergDataFiles = new ArrayList<>(); | ||
| for (WorkUnitState wus : statesHelper.getNonPostPublishStates()) { | ||
| if (wus.getWorkingState() == WorkingState.SUCCESSFUL) { | ||
| wus.setWorkingState(WorkUnitState.WorkingState.COMMITTED); | ||
| } | ||
| CopyEntity copyEntity = CopySource.deserializeCopyEntity(wus); | ||
| if (copyEntity instanceof CopyableFile) { | ||
| if (copyEntity instanceof CopyableFile || copyEntity instanceof IcebergPartitionCopyableFile) { | ||
| if (copyEntity instanceof IcebergPartitionCopyableFile) { | ||
| icebergDataFiles.add(((IcebergPartitionCopyableFile) copyEntity).getDataFile()); |
There was a problem hiding this comment.
Lets add a comment to describe why this is done
Comment on lines
+39
to
+58
|
|
||
| public DataFile getDataFile() { | ||
| return SerializationUtil.deserializeFromBase64(base64EncodedDataFile); | ||
| } |
There was a problem hiding this comment.
nit : lets add java doc here too
thisisArjit
force-pushed
the
iceberg-data-files
branch
3 times, most recently
from
734354f to
71bb739
Compare
thisisArjit
changed the title
thisisArjit
force-pushed
the
iceberg-data-files
branch
from
71bb739 to
e0ae58f
Compare
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
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.
Dear Gobblin maintainers,
Please accept this PR. I understand that it will not be reviewed until I have checked off all the steps below!
JIRA
Description
Instead of adding all the data files to Post publish step for iceberg partition copy, this PR adds Data file to each Iceberg partition copyable file during work unit generation & then adds all the data files to post publish step during commit step, hence avoiding serialisation & deserialisation of all the data files at once
Changes:
During WU generation, Copyable file is created and serialised. This PR adds an extension to Copyable file, IcebergPartitionCopyable file which contains the corresponding data file
During Commit step, in Iceberg post commit step, all the data files are collected from all the iceberg copyable file & the then gets added to IcebergOverwritePartitionsStep
Tests
Commits