rebase changes

apache · Jan 8, 2023 · 149f406 · 149f406
1 parent b3d2e07
commit 149f406
Show file tree

Hide file tree

Showing 5 changed files with 10 additions and 8 deletions.
diff --git a/core/src/main/java/org/apache/iceberg/MicroBatches.java b/core/src/main/java/org/apache/iceberg/MicroBatches.java
@@ -120,7 +120,7 @@ public MicroBatch generate(
 
       return generateMicroBatch(
           MicroBatchesUtil.skippedManifestIndexesFromSnapshot(
-              snapshot, startFileIndex, scanAllFiles),
+              io, snapshot, startFileIndex, scanAllFiles),
           startFileIndex,
           endFileIndex,
           targetSizeInBytes,

diff --git a/core/src/main/java/org/apache/iceberg/MicroBatchesUtil.java b/core/src/main/java/org/apache/iceberg/MicroBatchesUtil.java
@@ -32,13 +32,13 @@ public class MicroBatchesUtil {
   private MicroBatchesUtil() {}
 
   public static List<Pair<ManifestFile, Integer>> skippedManifestIndexesFromSnapshot(
-      Snapshot snapshot, long startFileIndex, boolean scanAllFiles) {
+      FileIO io, Snapshot snapshot, long startFileIndex, boolean scanAllFiles) {
     //  Preconditions.checkArgument(startFileIndex >= 0, "startFileIndex is unexpectedly smaller
     // than 0");
     List<ManifestFile> manifests =
         scanAllFiles
-            ? snapshot.dataManifests()
-            : snapshot.dataManifests().stream()
+            ? snapshot.dataManifests(io)
+            : snapshot.dataManifests(io).stream()
                 .filter(m -> m.snapshotId().equals(snapshot.snapshotId()))
                 .collect(Collectors.toList());
 

diff --git a/spark/v3.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java b/spark/v3.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java
@@ -212,7 +212,7 @@ private List<FileScanTask> planFiles(StreamingOffset startOffset, StreamingOffse
               .specsById(table.specs())
               .generate(
                   currentOffset.position(),
-                  Iterables.size(currentSnapshot.addedFiles()),
+                  Iterables.size(currentSnapshot.addedDataFiles(table.io())),
                   Long.MAX_VALUE,
                   currentOffset.shouldScanAllFiles());
 

diff --git a/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java b/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java
@@ -350,7 +350,7 @@ public Offset latestOffset(Offset startOffset, ReadLimit limit) {
       // generate manifest index for the curSnapshot
       List<Pair<ManifestFile, Integer>> indexedManifests =
           MicroBatchesUtil.skippedManifestIndexesFromSnapshot(
-              curSnapshot, startPosOfSnapOffset, scanAllFiles);
+              table.io(), curSnapshot, startPosOfSnapOffset, scanAllFiles);
       // this is under assumption we will be able to add at-least 1 file in the new offset
       for (int idx = 0; idx < indexedManifests.size() && isOk; idx++) {
         // be rest assured curPos >= startFileIndex
@@ -410,7 +410,9 @@ private long addedFilesCount(Snapshot snapshot) {
         PropertyUtil.propertyAsLong(snapshot.summary(), SnapshotSummary.ADDED_FILES_PROP, -1);
     // If snapshotSummary doesn't have SnapshotSummary.ADDED_FILES_PROP,
     // iterate through addedFiles iterator to find addedFilesCount.
-    return addedFilesCount == -1 ? Iterables.size(snapshot.addedFiles()) : addedFilesCount;
+    return addedFilesCount == -1
+        ? Iterables.size(snapshot.addedDataFiles(table.io()))
+        : addedFilesCount;
   }
 
   @Override

diff --git a/spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java b/spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java
@@ -222,7 +222,7 @@ private List<FileScanTask> planFiles(StreamingOffset startOffset, StreamingOffse
               .specsById(table.specs())
               .generate(
                   currentOffset.position(),
-                  Iterables.size(currentSnapshot.addedFiles()),
+                  Iterables.size(currentSnapshot.addedDataFiles(table.io())),
                   Long.MAX_VALUE,
                   currentOffset.shouldScanAllFiles());
       fileScanTasks.addAll(latestMicroBatch.tasks());