[iceberg] branch master updated: API: Avoid hasNext in CloseableIterable.concat (#5306)

[bl...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

blue pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iceberg.git


The following commit(s) were added to refs/heads/master by this push:
     new 9f6f687f6 API: Avoid hasNext in CloseableIterable.concat (#5306)
9f6f687f6 is described below

commit 9f6f687f6c649a700e1a603abf1965a22f7f8430
Author: Eduard Tudenhöfner <et...@gmail.com>
AuthorDate: Wed Jul 20 04:43:51 2022 +0200

    API: Avoid hasNext in CloseableIterable.concat (#5306)
    
    Some iterables need to load the first item to check whether there is a next item, which can cause an extra open call.
---
 .../org/apache/iceberg/io/CloseableIterable.java   | 19 +++++-----
 .../apache/iceberg/io/TestCloseableIterable.java   | 40 +++++++++++++++++++++-
 2 files changed, 47 insertions(+), 12 deletions(-)

diff --git a/api/src/main/java/org/apache/iceberg/io/CloseableIterable.java b/api/src/main/java/org/apache/iceberg/io/CloseableIterable.java
index f38366b3c..cc4be871f 100644
--- a/api/src/main/java/org/apache/iceberg/io/CloseableIterable.java
+++ b/api/src/main/java/org/apache/iceberg/io/CloseableIterable.java
@@ -118,12 +118,7 @@ public interface CloseableIterable<T> extends Iterable<T>, Closeable {
   }
 
   static <E> CloseableIterable<E> concat(Iterable<CloseableIterable<E>> iterable) {
-    Iterator<CloseableIterable<E>> iterables = iterable.iterator();
-    if (!iterables.hasNext()) {
-      return empty();
-    } else {
-      return new ConcatCloseableIterable<>(iterable);
-    }
+    return new ConcatCloseableIterable<>(iterable);
   }
 
   class ConcatCloseableIterable<E> extends CloseableGroup implements CloseableIterable<E> {
@@ -148,8 +143,6 @@ public interface CloseableIterable<T> extends Iterable<T>, Closeable {
 
       private ConcatCloseableIterator(Iterable<CloseableIterable<E>> inputs) {
         this.iterables = inputs.iterator();
-        this.currentIterable = iterables.next();
-        this.currentIterator = currentIterable.iterator();
       }
 
       @Override
@@ -158,13 +151,15 @@ public interface CloseableIterable<T> extends Iterable<T>, Closeable {
           return false;
         }
 
-        if (currentIterator.hasNext()) {
+        if (null != currentIterator && currentIterator.hasNext()) {
           return true;
         }
 
         while (iterables.hasNext()) {
           try {
-            currentIterable.close();
+            if (null != currentIterable) {
+              currentIterable.close();
+            }
           } catch (IOException e) {
             throw new RuntimeIOException(e, "Failed to close iterable");
           }
@@ -178,7 +173,9 @@ public interface CloseableIterable<T> extends Iterable<T>, Closeable {
         }
 
         try {
-          currentIterable.close();
+          if (null != currentIterable) {
+            currentIterable.close();
+          }
         } catch (IOException e) {
           throw new RuntimeIOException(e, "Failed to close iterable");
         }
diff --git a/api/src/test/java/org/apache/iceberg/io/TestCloseableIterable.java b/api/src/test/java/org/apache/iceberg/io/TestCloseableIterable.java
index 842d61a18..8eef8fc03 100644
--- a/api/src/test/java/org/apache/iceberg/io/TestCloseableIterable.java
+++ b/api/src/test/java/org/apache/iceberg/io/TestCloseableIterable.java
@@ -20,11 +20,15 @@
 package org.apache.iceberg.io;
 
 import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
 import java.util.NoSuchElementException;
+import java.util.concurrent.atomic.AtomicInteger;
 import org.apache.iceberg.AssertHelpers;
 import org.apache.iceberg.io.TestableCloseableIterable.TestableCloseableIterator;
 import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.assertj.core.api.Assertions;
 import org.junit.Assert;
 import org.junit.Test;
 
@@ -60,7 +64,7 @@ public class TestCloseableIterable {
   }
 
   @Test
-  public void testConcateWithEmptyIterables() {
+  public void testConcatWithEmptyIterables() {
     CloseableIterable<Integer> iter = CloseableIterable.combine(Lists.newArrayList(1, 2, 3), () -> { });
     CloseableIterable<Integer> empty = CloseableIterable.empty();
 
@@ -82,4 +86,38 @@ public class TestCloseableIterable {
         NoSuchElementException.class,
         () -> Iterables.getLast(concat5));
   }
+
+  @Test
+  public void testConcatWithEmpty() {
+    AtomicInteger counter = new AtomicInteger(0);
+    CloseableIterable.concat(Collections.emptyList()).forEach(c -> counter.incrementAndGet());
+    Assertions.assertThat(counter.get()).isEqualTo(0);
+  }
+
+  @Test
+  public void concatShouldOnlyEvaluateItemsOnce() throws IOException {
+    AtomicInteger counter = new AtomicInteger(0);
+    List<Integer> items = Lists.newArrayList(1, 2, 3, 4, 5);
+    Iterable<Integer> iterable = Iterables.filter(items, item -> {
+      counter.incrementAndGet();
+      return true;
+    });
+
+    Iterable<CloseableIterable<Integer>> transform =
+        Iterables.transform(iterable, item -> new CloseableIterable<Integer>() {
+          @Override
+          public void close() {
+          }
+
+          @Override
+          public CloseableIterator<Integer> iterator() {
+            return CloseableIterator.withClose(Collections.singletonList(item).iterator());
+          }
+        });
+
+    try (CloseableIterable<Integer> concat = CloseableIterable.concat(transform)) {
+      concat.forEach(c -> c++);
+    }
+    Assertions.assertThat(counter.get()).isEqualTo(items.size());
+  }
 }