Dell: Preserve known ECS input file lengths

dell/src/main/java/org/apache/iceberg/dell/ecs/EcsFileIO.java

@@ -58,6 +58,11 @@ public InputFile newInputFile(String path) {
     return EcsInputFile.fromLocation(path, client(), dellProperties, metrics);
   }
 
+  @Override
+  public InputFile newInputFile(String path, long length) {
+    return EcsInputFile.fromLocation(path, length, client(), dellProperties, metrics);
+  }
+
   @Override
   public OutputFile newOutputFile(String path) {
     return EcsOutputFile.fromLocation(path, client(), dellProperties, metrics);

dell/src/main/java/org/apache/iceberg/dell/ecs/EcsInputFile.java

@@ -20,12 +20,15 @@
 
 import com.emc.object.s3.S3Client;
 import org.apache.iceberg.dell.DellProperties;
+import org.apache.iceberg.exceptions.ValidationException;
 import org.apache.iceberg.io.InputFile;
 import org.apache.iceberg.io.SeekableInputStream;
 import org.apache.iceberg.metrics.MetricsContext;
 
 class EcsInputFile extends BaseEcsFile implements InputFile {
 
+  private Long length;
+
   public static EcsInputFile fromLocation(String location, S3Client client) {
     return new EcsInputFile(
         client, new EcsURI(location), new DellProperties(), MetricsContext.nullMetrics());
@@ -42,18 +45,42 @@ static EcsInputFile fromLocation(
     return new EcsInputFile(client, new EcsURI(location), dellProperties, metrics);
   }
 
+  static EcsInputFile fromLocation(
+      String location,
+      long length,
+      S3Client client,
+      DellProperties dellProperties,
+      MetricsContext metrics) {
+    return new EcsInputFile(client, new EcsURI(location), dellProperties, length, metrics);
+  }
+
   EcsInputFile(S3Client client, EcsURI uri, DellProperties dellProperties, MetricsContext metrics) {
     super(client, uri, dellProperties, metrics);
   }
 
+  EcsInputFile(
+      S3Client client,
+      EcsURI uri,
+      DellProperties dellProperties,
+      long length,
+      MetricsContext metrics) {
+    super(client, uri, dellProperties, metrics);
+    ValidationException.check(length >= 0, "Invalid file length: %s", length);
+    this.length = length;
+  }
+
   /**
    * Note: this may be stale if file was deleted since metadata is cached for size/existence checks.
    *
    * @return content length
    */
   @Override
   public long getLength() {
-    return getObjectMetadata().getContentLength();
+    if (length == null) {
+      this.length = getObjectMetadata().getContentLength();
+    }
+
+    return length;
   }
 
   @Override

dell/src/test/java/org/apache/iceberg/dell/ecs/TestEcsInputFile.java

@@ -25,6 +25,7 @@
 import java.io.InputStream;
 import java.nio.charset.StandardCharsets;
 import org.apache.iceberg.dell.mock.ecs.EcsS3MockRule;
+import org.apache.iceberg.io.InputFile;
 import org.apache.iceberg.relocated.com.google.common.io.ByteStreams;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
@@ -58,4 +59,15 @@ public void testFileRead() throws IOException {
           .isEqualTo("0123456789");
     }
   }
+
+  @Test
+  public void knownLengthAvoidsMetadataRequest() {
+    String location = new EcsURI(rule.bucket(), rule.randomObjectName()).toString();
+    EcsFileIO fileIO = new EcsFileIO();
+    fileIO.initialize(rule.clientProperties());
+
+    InputFile inputFile = fileIO.newInputFile(location, 10L);
+
+    assertThat(inputFile.getLength()).as("File length should use the known value").isEqualTo(10L);
+  }
 }