Skip to content
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

[Iceberg] cleanup FileIO resources #33509

Merged
merged 6 commits into from
Jan 8, 2025
+11 −7
Merged

[Iceberg] cleanup FileIO resources #33509

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
d177a85
cleanup FileIO resources
ahmedabu98 Jan 6, 2025
97fffb0
trigger integration tests
ahmedabu98 Jan 6, 2025
52311f7
Merge branch 'master' into cleanup_fileio
ahmedabu98 Jan 6, 2025
d978acd
cleanup
ahmedabu98 Jan 8, 2025
bd6a251
Merge branch 'cleanup_fileio' of https://github.com/ahmedabu98/beam i…
ahmedabu98 Jan 8, 2025
f841d5c
Merge branch 'master' of https://github.com/ahmedabu98/beam into clea…
ahmedabu98 Jan 8, 2025
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 .github/trigger_files/IO_Iceberg_Integration_Tests.json
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
{
"comment": "Modify this file in a trivial way to cause this test suite to run",
"modification": 3
"modification": 1
}
10 changes: 5 additions & 5 deletions sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AppendFilesToTables.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -187,8 +187,7 @@ private void appendManifestFiles(Table table, Iterable<FileWriteResult> fileWrit
int specId = entry.getKey();
List<DataFile> files = entry.getValue();
PartitionSpec spec = Preconditions.checkStateNotNull(specs.get(specId));
ManifestWriter<DataFile> writer =
createManifestWriter(table.location(), uuid, spec, table.io());
ManifestWriter<DataFile> writer = createManifestWriter(table.location(), uuid, spec, table);
for (DataFile file : files) {
writer.add(file);
committedDataFileByteSize.update(file.fileSizeInBytes());
Expand All @@ -201,14 +200,15 @@ private void appendManifestFiles(Table table, Iterable<FileWriteResult> fileWrit
}

private ManifestWriter<DataFile> createManifestWriter(
String tableLocation, String uuid, PartitionSpec spec, FileIO io) {
String tableLocation, String uuid, PartitionSpec spec, Table table) {
Copy link
Contributor

Choose a reason for hiding this comment

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

if we are passing Table now we could drop tableLocation and spec

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Thx for the catch. I refactored this a lil to make it more clean.

btw spec here isn't a simple table.spec(). This part of the code deals with the edge case where we have a batch of data files that were written with different specs (e.g. user updates spec during pipeline runtime). We group data files by spec and create/append one ManifestFile per spec.

Copy link
Contributor

Choose a reason for hiding this comment

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

ah! nice makes sense

String location =
FileFormat.AVRO.addExtension(
String.format(
"%s/metadata/%s-%s-%s.manifest",
tableLocation, manifestFilePrefix, uuid, spec.specId()));
OutputFile outputFile = io.newOutputFile(location);
return ManifestFiles.write(spec, outputFile);
try (FileIO io = table.io()) {
return ManifestFiles.write(spec, io.newOutputFile(location));
}
}
}
}
6 changes: 5 additions & 1 deletion sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/RecordWriter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@
import org.apache.iceberg.data.Record;
import org.apache.iceberg.data.parquet.GenericParquetWriter;
import org.apache.iceberg.io.DataWriter;
import org.apache.iceberg.io.FileIO;
import org.apache.iceberg.io.OutputFile;
import org.apache.iceberg.parquet.Parquet;
import org.slf4j.Logger;
Expand Down Expand Up @@ -66,7 +67,10 @@ class RecordWriter {
fileFormat.addExtension(
table.locationProvider().newDataLocation(table.spec(), partitionKey, filename));
}
OutputFile outputFile = table.io().newOutputFile(absoluteFilename);
OutputFile outputFile;
try (FileIO io = table.io()) {
outputFile = io.newOutputFile(absoluteFilename);
}

switch (fileFormat) {
case AVRO:
Expand Down
Loading