Fix eager loading of Hive partitions

Remove defensive copy of Hive partitions to avoid eagerly loading all
the partitions

presto-hive/src/main/java/com/facebook/presto/hive/BackgroundHiveSplitLoader.java

@@ -19,7 +19,6 @@
 import com.facebook.presto.spi.TupleDomain;
 import com.google.common.base.Throwables;
 import com.google.common.collect.ImmutableList;
-import com.google.common.collect.Iterables;
 import io.airlift.units.DataSize;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.BlockLocation;
@@ -47,9 +46,7 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.Properties;
-import java.util.Queue;
 import java.util.concurrent.ConcurrentLinkedDeque;
-import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.Executor;
 import java.util.concurrent.atomic.AtomicInteger;
 
@@ -82,7 +79,7 @@ public class BackgroundHiveSplitLoader
     private final Executor executor;
     private final ConnectorSession session;
     private final AtomicInteger outstandingTasks = new AtomicInteger();
-    private final Queue<HivePartitionMetadata> partitions = new ConcurrentLinkedQueue<>();
+    private final ConcurrentLazyQueue<HivePartitionMetadata> partitions;
     private final Deque<HiveFileIterator> fileIterators = new ConcurrentLinkedDeque<>();
     private final AtomicInteger remainingInitialSplits;
 
@@ -120,7 +117,7 @@ public BackgroundHiveSplitLoader(
         this.recursiveDirWalkerEnabled = recursiveDirWalkerEnabled;
         this.forceLocalScheduling = forceLocalScheduling;
         this.executor = executor;
-        Iterables.addAll(this.partitions, partitions);
+        this.partitions = new ConcurrentLazyQueue<>(partitions);
     }
 
     @Override

presto-hive/src/main/java/com/facebook/presto/hive/ConcurrentLazyQueue.java

@@ -0,0 +1,42 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.facebook.presto.hive;
+
+import javax.annotation.concurrent.GuardedBy;
+
+import java.util.Iterator;
+
+public class ConcurrentLazyQueue<E>
+{
+    @GuardedBy("this")
+    private final Iterator<E> iterator;
+
+    public ConcurrentLazyQueue(Iterable<E> iterable)
+    {
+        this.iterator = iterable.iterator();
+    }
+
+    public synchronized boolean isEmpty()
+    {
+        return !iterator.hasNext();
+    }
+
+    public synchronized E poll()
+    {
+        if (!iterator.hasNext()) {
+            return null;
+        }
+        return iterator.next();
+    }
+}