Core: Add length arg to FileIO.newInputFile

api/src/main/java/org/apache/iceberg/io/FileIO.java

@@ -37,6 +37,13 @@ public interface FileIO extends Serializable, Closeable {
    */
   InputFile newInputFile(String path);
 
+  /**
+   * Get a {@link InputFile} instance to read bytes from the file at the given path, with a known file length.
+   */
+  default InputFile newInputFile(String path, long length) {
+    return newInputFile(path);
+  }
+
   /**
    * Get a {@link OutputFile} instance to write bytes to the file at the given path.
    */

aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java

@@ -119,6 +119,11 @@ public InputFile newInputFile(String path) {
     return S3InputFile.fromLocation(path, client(), awsProperties, metrics);
   }
 
+  @Override
+  public InputFile newInputFile(String path, long length) {
+    return S3InputFile.fromLocation(path, length, client(), awsProperties, metrics);
+  }
+
   @Override
   public OutputFile newOutputFile(String path) {
     return S3OutputFile.fromLocation(path, client(), awsProperties, metrics);

aws/src/main/java/org/apache/iceberg/aws/s3/S3InputFile.java

@@ -29,15 +29,23 @@
 
 public class S3InputFile extends BaseS3File implements InputFile, NativelyEncryptedFile {
   private NativeFileCryptoParameters nativeDecryptionParameters;
+  private Long length;
 
   public static S3InputFile fromLocation(String location, S3Client client, AwsProperties awsProperties,
       MetricsContext metrics) {
-    return new S3InputFile(client, new S3URI(location, awsProperties.s3BucketToAccessPointMapping()),
+    return new S3InputFile(client, new S3URI(location, awsProperties.s3BucketToAccessPointMapping()), null,
         awsProperties, metrics);
   }
 
-  S3InputFile(S3Client client, S3URI uri, AwsProperties awsProperties, MetricsContext metrics) {
+  public static S3InputFile fromLocation(String location, long length, S3Client client, AwsProperties awsProperties,
+                                         MetricsContext metrics) {
+    return new S3InputFile(client, new S3URI(location, awsProperties.s3BucketToAccessPointMapping()),
+        length > 0 ? length : null, awsProperties, metrics);
+  }
+
+  S3InputFile(S3Client client, S3URI uri, Long length, AwsProperties awsProperties, MetricsContext metrics) {
     super(client, uri, awsProperties, metrics);
+    this.length = length;
   }
 
   /**
@@ -47,7 +55,11 @@ public static S3InputFile fromLocation(String location, S3Client client, AwsProp
    */
   @Override
   public long getLength() {
-    return getObjectMetadata().contentLength();
+    if (length == null) {
+      this.length = getObjectMetadata().contentLength();
+    }
+
+    return length;
   }
 
   @Override

aws/src/main/java/org/apache/iceberg/aws/s3/S3OutputFile.java

@@ -70,7 +70,7 @@ public PositionOutputStream createOrOverwrite() {
 
   @Override
   public InputFile toInputFile() {
-    return new S3InputFile(client(), uri(), awsProperties(), metrics());
+    return new S3InputFile(client(), uri(), null, awsProperties(), metrics());
   }
 
   @Override

core/src/main/java/org/apache/iceberg/ManifestFiles.java

@@ -82,7 +82,7 @@ public static ManifestReader<DataFile> read(ManifestFile manifest, FileIO io) {
   public static ManifestReader<DataFile> read(ManifestFile manifest, FileIO io, Map<Integer, PartitionSpec> specsById) {
     Preconditions.checkArgument(manifest.content() == ManifestContent.DATA,
         "Cannot read a delete manifest with a ManifestReader: %s", manifest);
-    InputFile file = io.newInputFile(manifest.path());
+    InputFile file = io.newInputFile(manifest.path(), manifest.length());
     InheritableMetadata inheritableMetadata = InheritableMetadataFactory.fromManifest(manifest);
     return new ManifestReader<>(file, specsById, inheritableMetadata, FileType.DATA_FILES);
   }
@@ -133,7 +133,7 @@ public static ManifestReader<DeleteFile> readDeleteManifest(ManifestFile manifes
                                                               Map<Integer, PartitionSpec> specsById) {
     Preconditions.checkArgument(manifest.content() == ManifestContent.DELETES,
         "Cannot read a data manifest with a DeleteManifestReader: %s", manifest);
-    InputFile file = io.newInputFile(manifest.path());
+    InputFile file = io.newInputFile(manifest.path(), manifest.length());
     InheritableMetadata inheritableMetadata = InheritableMetadataFactory.fromManifest(manifest);
     return new ManifestReader<>(file, specsById, inheritableMetadata, FileType.DELETE_FILES);
   }

core/src/main/java/org/apache/iceberg/RemoveSnapshots.java

@@ -563,7 +563,7 @@ private Set<String> findFilesToDelete(Set<ManifestFile> manifestsToScan, Set<Man
   }
 
   private static final Schema MANIFEST_PROJECTION = ManifestFile.schema()
-      .select("manifest_path", "added_snapshot_id", "deleted_data_files_count");
+      .select("manifest_path", "manifest_length", "added_snapshot_id", "deleted_data_files_count");
 
   private CloseableIterable<ManifestFile> readManifestFiles(Snapshot snapshot) {
     if (snapshot.manifestListLocation() != null) {

core/src/main/java/org/apache/iceberg/hadoop/HadoopFileIO.java

@@ -67,6 +67,11 @@ public InputFile newInputFile(String path) {
     return HadoopInputFile.fromLocation(path, hadoopConf.get());
   }
 
+  @Override
+  public InputFile newInputFile(String path, long length) {
+    return HadoopInputFile.fromLocation(path, length, hadoopConf.get());
+  }
+
   @Override
   public OutputFile newOutputFile(String path) {
     return HadoopOutputFile.fromPath(new Path(path), hadoopConf.get());

core/src/main/java/org/apache/iceberg/hadoop/HadoopInputFile.java

@@ -61,7 +61,11 @@ public static HadoopInputFile fromLocation(CharSequence location, Configuration
   public static HadoopInputFile fromLocation(CharSequence location, long length,
                                              Configuration conf) {
     FileSystem fs = Util.getFs(new Path(location.toString()), conf);
-    return new HadoopInputFile(fs, location.toString(), length, conf);
+    if (length > 0) {
+      return new HadoopInputFile(fs, location.toString(), length, conf);
+    } else {
+      return new HadoopInputFile(fs, location.toString(), conf);
+    }
   }
 
   public static HadoopInputFile fromLocation(CharSequence location, FileSystem fs) {

core/src/main/java/org/apache/iceberg/io/ResolvingFileIO.java

@@ -63,6 +63,11 @@ public InputFile newInputFile(String location) {
     return io(location).newInputFile(location);
   }
 
+  @Override
+  public InputFile newInputFile(String location, long length) {
+    return io(location).newInputFile(location, length);
+  }
+
   @Override
   public OutputFile newOutputFile(String location) {
     return io(location).newOutputFile(location);

core/src/test/java/org/apache/iceberg/hadoop/TestCatalogUtilDropTable.java

@@ -60,6 +60,8 @@ public void dropTableDataDeletesExpectedFiles() {
     FileIO fileIO = Mockito.mock(FileIO.class);
     Mockito.when(fileIO.newInputFile(Mockito.anyString()))
         .thenAnswer(invocation -> table.io().newInputFile(invocation.getArgument(0)));
+    Mockito.when(fileIO.newInputFile(Mockito.anyString(), Mockito.anyLong()))
+        .thenAnswer(invocation -> table.io().newInputFile(invocation.getArgument(0), invocation.getArgument(1)));
 
     CatalogUtil.dropTableData(fileIO, tableMetadata);
     ArgumentCaptor<String> argumentCaptor = ArgumentCaptor.forClass(String.class);
@@ -90,6 +92,8 @@ public void dropTableDataDoNotThrowWhenDeletesFail() {
     FileIO fileIO = Mockito.mock(FileIO.class);
     Mockito.when(fileIO.newInputFile(Mockito.anyString()))
         .thenAnswer(invocation -> table.io().newInputFile(invocation.getArgument(0)));
+    Mockito.when(fileIO.newInputFile(Mockito.anyString(), Mockito.anyLong()))
+        .thenAnswer(invocation -> table.io().newInputFile(invocation.getArgument(0), invocation.getArgument(1)));
     Mockito.doThrow(new RuntimeException()).when(fileIO).deleteFile(ArgumentMatchers.anyString());
 
     CatalogUtil.dropTableData(fileIO, tableMetadata);

gcp/src/main/java/org/apache/iceberg/gcp/gcs/GCSFileIO.java

@@ -80,6 +80,11 @@ public InputFile newInputFile(String path) {
     return GCSInputFile.fromLocation(path, client(), gcpProperties, metrics);
   }
 
+  @Override
+  public InputFile newInputFile(String path, long length) {
+    return GCSInputFile.fromLocation(path, length, client(), gcpProperties, metrics);
+  }
+
   @Override
   public OutputFile newOutputFile(String path) {
     return GCSOutputFile.fromLocation(path, client(), gcpProperties, metrics);

gcp/src/main/java/org/apache/iceberg/gcp/gcs/GCSInputFile.java

@@ -27,19 +27,31 @@
 import org.apache.iceberg.metrics.MetricsContext;
 
 class GCSInputFile extends BaseGCSFile implements InputFile {
+  private Long length;
 
   static GCSInputFile fromLocation(String location, Storage storage,
+                                   GCPProperties gcpProperties, MetricsContext metrics) {
+    return new GCSInputFile(storage, BlobId.fromGsUtilUri(location), null, gcpProperties, metrics);
+  }
+
+  static GCSInputFile fromLocation(String location, long length, Storage storage,
       GCPProperties gcpProperties, MetricsContext metrics) {
-    return new GCSInputFile(storage, BlobId.fromGsUtilUri(location), gcpProperties, metrics);
+    return new GCSInputFile(
+        storage, BlobId.fromGsUtilUri(location), length > 0 ? length : null, gcpProperties, metrics);
   }
 
-  GCSInputFile(Storage storage, BlobId blobId, GCPProperties gcpProperties, MetricsContext metrics) {
+  GCSInputFile(Storage storage, BlobId blobId, Long length, GCPProperties gcpProperties, MetricsContext metrics) {
     super(storage, blobId, gcpProperties, metrics);
+    this.length = length;
   }
 
   @Override
   public long getLength() {
-    return getBlob().getSize();
+    if (length == null) {
+      this.length = getBlob().getSize();
+    }
+
+    return length;
   }
 
   @Override

gcp/src/main/java/org/apache/iceberg/gcp/gcs/GCSOutputFile.java

@@ -67,6 +67,6 @@ public PositionOutputStream createOrOverwrite() {
 
   @Override
   public InputFile toInputFile() {
-    return new GCSInputFile(storage(), blobId(), gcpProperties(), metrics());
+    return new GCSInputFile(storage(), blobId(), null, gcpProperties(), metrics());
   }
 }