(beam) branch master updated: Attach file extension to Iceberg writes (#32254)

[ahmedabualsaud]

This is an automated email from the ASF dual-hosted git repository.

ahmedabualsaud pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git


The following commit(s) were added to refs/heads/master by this push:
     new ec3cec201bf Attach file extension to Iceberg writes (#32254)
ec3cec201bf is described below

commit ec3cec201bfc52dff7f3619cb66c8dc4469a9e47
Author: Ahmed Abualsaud <65791736+ahmedab...@users.noreply.github.com>
AuthorDate: Tue Aug 20 16:04:36 2024 -0700

    Attach file extension to Iceberg writes (#32254)
---
 .../src/main/java/org/apache/beam/sdk/io/iceberg/RecordWriter.java  | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

diff --git 
a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/RecordWriter.java
 
b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/RecordWriter.java
index 576d1e32c46..6bbb103e17c 100644
--- 
a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/RecordWriter.java
+++ 
b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/RecordWriter.java
@@ -57,10 +57,12 @@ class RecordWriter {
     this.table = table;
     this.fileFormat = fileFormat;
     if (table.spec().isUnpartitioned()) {
-      absoluteFilename = table.locationProvider().newDataLocation(filename);
+      absoluteFilename =
+          
fileFormat.addExtension(table.locationProvider().newDataLocation(filename));
     } else {
       absoluteFilename =
-          table.locationProvider().newDataLocation(table.spec(), partitionKey, 
filename);
+          fileFormat.addExtension(
+              table.locationProvider().newDataLocation(table.spec(), 
partitionKey, filename));
     }
     OutputFile outputFile = table.io().newOutputFile(absoluteFilename);