API: Pass FileIO into Snapshot methods that read metadata

.palantir/revapi.yml

@@ -27,6 +27,21 @@ acceptedBreaks:
     - code: "java.method.addedToInterface"
       new: "method ThisT org.apache.iceberg.SnapshotUpdate<ThisT>::scanManifestsWith(java.util.concurrent.ExecutorService)"
       justification: "Accept all changes prior to introducing API compatibility checks"
+    - code: "java.method.addedToInterface"
+      new: "method java.lang.Iterable<org.apache.iceberg.DataFile> org.apache.iceberg.Snapshot::addedFiles(org.apache.iceberg.io.FileIO)"
+      justification: "Allow adding a new method to the interface - old method is deprecated"
+    - code: "java.method.addedToInterface"
+      new: "method java.lang.Iterable<org.apache.iceberg.DataFile> org.apache.iceberg.Snapshot::deletedFiles(org.apache.iceberg.io.FileIO)"
+      justification: "Allow adding a new method to the interface - old method is deprecated"
+    - code: "java.method.addedToInterface"
+      new: "method java.util.List<org.apache.iceberg.ManifestFile> org.apache.iceberg.Snapshot::allManifests(org.apache.iceberg.io.FileIO)"
+      justification: "Allow adding a new method to the interface - old method is deprecated"
+    - code: "java.method.addedToInterface"
+      new: "method java.util.List<org.apache.iceberg.ManifestFile> org.apache.iceberg.Snapshot::dataManifests(org.apache.iceberg.io.FileIO)"
+      justification: "Allow adding a new method to the interface - old method is deprecated"
+    - code: "java.method.addedToInterface"
+      new: "method java.util.List<org.apache.iceberg.ManifestFile> org.apache.iceberg.Snapshot::deleteManifests(org.apache.iceberg.io.FileIO)"
+      justification: "Allow adding a new method to the interface - old method is deprecated"
     - code: "java.method.addedToInterface"
       new: "method long org.apache.iceberg.actions.ExpireSnapshots.Result::deletedEqualityDeleteFilesCount()"
       justification: "Interface is backward compatible, very unlikely anyone implements this Result bean interface"

api/src/main/java/org/apache/iceberg/Snapshot.java

@@ -22,6 +22,7 @@
 import java.io.Serializable;
 import java.util.List;
 import java.util.Map;
+import org.apache.iceberg.io.FileIO;
 
 /**
  * A snapshot of the data in a table at a point in time.
@@ -68,23 +69,54 @@ public interface Snapshot extends Serializable {
    * Return all {@link ManifestFile} instances for either data or delete manifests in this snapshot.
    *
    * @return a list of ManifestFile
+   * @deprecated since 0.14.0, will be removed in 1.0.0; Use {@link Snapshot#allManifests(FileIO)} instead.
    */
+  @Deprecated
   List<ManifestFile> allManifests();
 
+  /**
+   * Return all {@link ManifestFile} instances for either data or delete manifests in this snapshot.
+   *
+   * @param io a {@link FileIO} instance used for reading files from storage
+   * @return a list of ManifestFile
+   */
+  List<ManifestFile> allManifests(FileIO io);
+
   /**
    * Return a {@link ManifestFile} for each data manifest in this snapshot.
    *
    * @return a list of ManifestFile
+   * @deprecated since 0.14.0, will be removed in 1.0.0; Use {@link Snapshot#dataManifests(FileIO)} instead.
    */
+  @Deprecated
   List<ManifestFile> dataManifests();
 
+  /**
+   * Return a {@link ManifestFile} for each data manifest in this snapshot.
+   *
+   * @param io a {@link FileIO} instance used for reading files from storage
+   * @return a list of ManifestFile
+   */
+  List<ManifestFile> dataManifests(FileIO io);
+
   /**
    * Return a {@link ManifestFile} for each delete manifest in this snapshot.
    *
    * @return a list of ManifestFile
+   * @deprecated since 0.14.0, will be removed in 1.0.0; Use {@link Snapshot#deleteManifests(FileIO)} instead.
    */
+  @Deprecated
   List<ManifestFile> deleteManifests();
 
+
+  /**
+   * Return a {@link ManifestFile} for each delete manifest in this snapshot.
+   *
+   * @param io a {@link FileIO} instance used for reading files from storage
+   * @return a list of ManifestFile
+   */
+  List<ManifestFile> deleteManifests(FileIO io);
+
   /**
    * Return the name of the {@link DataOperations data operation} that produced this snapshot.
    *
@@ -107,19 +139,45 @@ public interface Snapshot extends Serializable {
    * record_count, and file_size_in_bytes. Other columns will be null.
    *
    * @return all files added to the table in this snapshot.
+   * @deprecated since 0.14.0, will be removed in 1.0.0; Use {@link Snapshot#addedFiles(FileIO)} instead.
    */
+  @Deprecated
   Iterable<DataFile> addedFiles();
 
+  /**
+   * Return all files added to the table in this snapshot.
+   * <p>
+   * The files returned include the following columns: file_path, file_format, partition,
+   * record_count, and file_size_in_bytes. Other columns will be null.
+   *
+   * @param io a {@link FileIO} instance used for reading files from storage
+   * @return all files added to the table in this snapshot.
+   */
+  Iterable<DataFile> addedFiles(FileIO io);
+
   /**
    * Return all files deleted from the table in this snapshot.
    * <p>
    * The files returned include the following columns: file_path, file_format, partition,
    * record_count, and file_size_in_bytes. Other columns will be null.
    *
    * @return all files deleted from the table in this snapshot.
+   * @deprecated since 0.14.0, will be removed in 1.0.0; Use {@link Snapshot#deletedFiles(FileIO)} instead.
    */
+  @Deprecated
   Iterable<DataFile> deletedFiles();
 
+  /**
+   * Return all files deleted from the table in this snapshot.
+   * <p>
+   * The files returned include the following columns: file_path, file_format, partition,
+   * record_count, and file_size_in_bytes. Other columns will be null.
+   *
+   * @param io a {@link FileIO} instance used for reading files from storage
+   * @return all files deleted from the table in this snapshot.
+   */
+  Iterable<DataFile> deletedFiles(FileIO io);
+
   /**
    * Return the location of this snapshot's manifest list, or null if it is not separate.
    *

aws/src/integration/java/org/apache/iceberg/aws/glue/TestGlueCatalogLock.java

@@ -97,7 +97,8 @@ public void testParallelCommitMultiThreadSingleCommit() {
 
     table.refresh();
     Assert.assertEquals("Commits should all succeed sequentially", nThreads, table.history().size());
-    Assert.assertEquals("Should have all manifests", nThreads, table.currentSnapshot().allManifests().size());
+    Assert.assertEquals("Should have all manifests", nThreads,
+        table.currentSnapshot().allManifests(table.io()).size());
   }
 
   @Test
@@ -138,7 +139,7 @@ public void testParallelCommitMultiThreadMultiCommit() {
 
     table.refresh();
     Assert.assertEquals("Commits should all succeed sequentially", 20, table.history().size());
-    Assert.assertEquals("should have 20 manifests", 20, table.currentSnapshot().allManifests().size());
+    Assert.assertEquals("should have 20 manifests", 20, table.currentSnapshot().allManifests(table.io()).size());
   }
 
 }

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

@@ -66,7 +66,7 @@ protected TableScan newRefinedScan(TableOperations ops, Table table, Schema sche
 
     @Override
     protected CloseableIterable<ManifestFile> manifests() {
-      return reachableManifests(Snapshot::dataManifests);
+      return reachableManifests(snapshot -> snapshot.dataManifests(tableOps().io()));
     }
   }
 }

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

@@ -66,7 +66,7 @@ protected TableScan newRefinedScan(TableOperations ops, Table table, Schema sche
 
     @Override
     protected CloseableIterable<ManifestFile> manifests() {
-      return reachableManifests(Snapshot::deleteManifests);
+      return reachableManifests(snapshot -> snapshot.deleteManifests(tableOps().io()));
     }
   }
 }

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

@@ -84,7 +84,8 @@ protected TableScan newRefinedScan(TableOperations ops, Table table, Schema sche
 
     @Override
     protected CloseableIterable<FileScanTask> doPlanFiles() {
-      CloseableIterable<ManifestFile> manifests = reachableManifests(Snapshot::allManifests);
+      CloseableIterable<ManifestFile> manifests =
+          reachableManifests(snapshot -> snapshot.allManifests(tableOps().io()));
 
       String schemaString = SchemaParser.toJson(schema());
       String specString = PartitionSpecParser.toJson(PartitionSpec.unpartitioned());

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

@@ -66,7 +66,7 @@ protected TableScan newRefinedScan(TableOperations ops, Table table, Schema sche
 
     @Override
     protected CloseableIterable<ManifestFile> manifests() {
-      return reachableManifests(Snapshot::allManifests);
+      return reachableManifests(snapshot -> snapshot.allManifests(tableOps().io()));
     }
   }
 }

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

@@ -125,7 +125,7 @@ protected CloseableIterable<FileScanTask> doPlanFiles() {
         } else {
           return StaticDataTask.of(
               io.newInputFile(tableOps().current().metadataFileLocation()),
-              MANIFEST_FILE_SCHEMA, schema(), snap.allManifests(),
+              MANIFEST_FILE_SCHEMA, schema(), snap.allManifests(io),
               manifest -> ManifestsTable.manifestFileToRow(specs.get(manifest.partitionSpecId()), manifest)
           );
         }

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

@@ -162,7 +162,7 @@ private ManifestFile copyManifest(ManifestFile manifest) {
 
   @Override
   public List<ManifestFile> apply(TableMetadata base) {
-    List<ManifestFile> currentManifests = base.currentSnapshot().dataManifests();
+    List<ManifestFile> currentManifests = base.currentSnapshot().dataManifests(ops.io());
     Set<ManifestFile> currentManifestSet = ImmutableSet.copyOf(currentManifests);
 
     validateDeletedManifests(currentManifestSet);
@@ -183,7 +183,7 @@ public List<ManifestFile> apply(TableMetadata base) {
     List<ManifestFile> apply = Lists.newArrayList();
     Iterables.addAll(apply, newManifestsWithMetadata);
     apply.addAll(keptManifests);
-    apply.addAll(base.currentSnapshot().deleteManifests());
+    apply.addAll(base.currentSnapshot().deleteManifests(ops.io()));
 
     return apply;
   }