Pass file size to Iceberg commit task

Intel-bigdata · Dec 28, 2021 · 1d71890 · 1d71890
1 parent a3b7569
commit 1d71890
Show file tree

Hide file tree

Showing 4 changed files with 14 additions and 6 deletions.
diff --git a/presto-iceberg/src/main/java/com/facebook/presto/iceberg/CommitTaskData.java b/presto-iceberg/src/main/java/com/facebook/presto/iceberg/CommitTaskData.java
@@ -23,16 +23,19 @@
 public class CommitTaskData
 {
     private final String path;
+    private final long fileSizeInBytes;
     private final MetricsWrapper metrics;
     private final Optional<String> partitionDataJson;
 
     @JsonCreator
     public CommitTaskData(
             @JsonProperty("path") String path,
+            @JsonProperty("fileSizeInBytes") long fileSizeInBytes,
             @JsonProperty("metrics") MetricsWrapper metrics,
             @JsonProperty("partitionDataJson") Optional<String> partitionDataJson)
     {
         this.path = requireNonNull(path, "path is null");
+        this.fileSizeInBytes = fileSizeInBytes;
         this.metrics = requireNonNull(metrics, "metrics is null");
         this.partitionDataJson = requireNonNull(partitionDataJson, "partitionDataJson is null");
     }
@@ -43,6 +46,12 @@ public String getPath()
         return path;
     }
 
+    @JsonProperty
+    public long getFileSizeInBytes()
+    {
+        return fileSizeInBytes;
+    }
+
     @JsonProperty
     public MetricsWrapper getMetrics()
     {

diff --git a/presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergMetadata.java b/presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergMetadata.java
@@ -447,10 +447,9 @@ public Optional<ConnectorOutputMetadata> finishInsert(ConnectorSession session,
 
         AppendFiles appendFiles = transaction.newFastAppend();
         for (CommitTaskData task : commitTasks) {
-            HdfsContext context = new HdfsContext(session, table.getSchemaName(), table.getTableName());
-
             DataFiles.Builder builder = DataFiles.builder(icebergTable.spec())
-                    .withInputFile(new HdfsInputFile(new Path(task.getPath()), hdfsEnvironment, context))
+                    .withPath(task.getPath())
+                    .withFileSizeInBytes(task.getFileSizeInBytes())
                     .withFormat(table.getFileFormat())
                     .withMetrics(task.getMetrics().metrics());
 

diff --git a/presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergNativeMetadata.java b/presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergNativeMetadata.java
@@ -59,7 +59,6 @@
 import org.apache.iceberg.exceptions.AlreadyExistsException;
 import org.apache.iceberg.exceptions.NamespaceNotEmptyException;
 import org.apache.iceberg.exceptions.NoSuchTableException;
-import org.apache.iceberg.io.FileIO;
 import org.apache.iceberg.types.Type;
 import org.apache.iceberg.types.TypeUtil;
 import org.apache.iceberg.types.Types;
@@ -411,10 +410,10 @@ public Optional<ConnectorOutputMetadata> finishInsert(ConnectorSession session,
                 .toArray(Type[]::new);
 
         AppendFiles appendFiles = transaction.newFastAppend();
-        FileIO io = transaction.table().io();
         for (CommitTaskData task : commitTasks) {
             DataFiles.Builder builder = DataFiles.builder(icebergTable.spec())
-                    .withInputFile(io.newInputFile(task.getPath()))
+                    .withPath(task.getPath())
+                    .withFileSizeInBytes(task.getFileSizeInBytes())
                     .withFormat(table.getFileFormat())
                     .withMetrics(task.getMetrics().metrics());
 

diff --git a/presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergPageSink.java b/presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergPageSink.java
@@ -159,6 +159,7 @@ public CompletableFuture<Collection<Slice>> finish()
 
             CommitTaskData task = new CommitTaskData(
                     context.getPath().toString(),
+                    context.writer.getFileSizeInBytes(),
                     new MetricsWrapper(context.writer.getMetrics()),
                     context.getPartitionData().map(PartitionData::toJson));