Support structured streaming read for Iceberg

Author: XuQianJin-Stars

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

@@ -21,11 +21,11 @@
 
 import java.io.IOException;
 import java.util.Collections;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
 import org.apache.iceberg.io.CloseableIterable;
-import org.apache.iceberg.io.CloseableIterator;
 import org.apache.iceberg.io.FileIO;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
@@ -198,7 +198,7 @@ private MicroBatch generateMicroBatch(List<Pair<ManifestFile, Integer>> indexedM
 
         try (CloseableIterable<FileScanTask> taskIterable = open(indexedManifests.get(idx).first(),
             scanAllFiles)) {
-          CloseableIterator<FileScanTask> taskIter = taskIterable.iterator();
+          Iterator<FileScanTask> taskIter = taskIterable.iterator();
           while (taskIter.hasNext()) {
             FileScanTask task = taskIter.next();
             if (currentFileIndex >= startFileIndex) {

spark2/src/main/java/org/apache/iceberg/spark/source/Reader.java

@@ -329,33 +329,7 @@ public Statistics estimateStatistics() {
 
   @Override
   public boolean enableBatchRead() {
-    if (readUsingBatch == null) {
-      boolean allParquetFileScanTasks =
-          tasks().stream()
-              .allMatch(combinedScanTask -> !combinedScanTask.isDataTask() && combinedScanTask.files()
-                  .stream()
-                  .allMatch(fileScanTask -> fileScanTask.file().format().equals(
-                      FileFormat.PARQUET)));
-
-      boolean allOrcFileScanTasks =
-          tasks().stream()
-              .allMatch(combinedScanTask -> !combinedScanTask.isDataTask() && combinedScanTask.files()
-                  .stream()
-                  .allMatch(fileScanTask -> fileScanTask.file().format().equals(
-                      FileFormat.ORC)));
-
-      boolean atLeastOneColumn = lazySchema().columns().size() > 0;
-
-      boolean onlyPrimitives = lazySchema().columns().stream().allMatch(c -> c.type().isPrimitiveType());
-
-      boolean hasNoDeleteFiles = tasks().stream().noneMatch(TableScanUtil::hasDeletes);
-
-      boolean batchReadsEnabled = batchReadsEnabled(allParquetFileScanTasks, allOrcFileScanTasks);
-
-      this.readUsingBatch = batchReadsEnabled && hasNoDeleteFiles && (allOrcFileScanTasks ||
-          (allParquetFileScanTasks && atLeastOneColumn && onlyPrimitives));
-    }
-    return readUsingBatch;
+    return readUsingBatch == null ? checkEnableBatchRead(tasks()) : readUsingBatch;
   }
 
   private boolean batchReadsEnabled(boolean isParquetOnly, boolean isOrcOnly) {

spark2/src/main/java/org/apache/iceberg/spark/source/StreamingReader.java

@@ -21,6 +21,7 @@
 
 import java.util.Collections;
 import java.util.List;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.stream.Collectors;
 import org.apache.commons.collections.CollectionUtils;
@@ -80,7 +81,9 @@ class StreamingReader extends Reader implements MicroBatchReader {
   private final long splitOpenFileCost;
   private Boolean readUsingBatch = null;
 
-  // Used to cache the pending batches for this streaming batch interval.
+  /**
+   * Used to cache the pending batches for this streaming batch interval.
+   */
   private Pair<StreamingOffset, List<MicroBatch>> cachedPendingBatches = null;
 
   StreamingReader(Table table, Broadcast<FileIO> io, Broadcast<EncryptionManager> encryptionManager,
@@ -324,7 +327,7 @@ List<MicroBatch> getChangesWithRateLimit(StreamingOffset startOffset, long maxSi
 
       // If the current request size is already satisfied, or none of the DataFile size can satisfy the current left
       // size, break the current loop.
-      if (currentLeftSize <= 0L || !lastBatch.lastIndexOfSnapshot()) {
+      if (currentLeftSize <= 0L || !Objects.requireNonNull(lastBatch).lastIndexOfSnapshot()) {
         break;
       }
     }