Extract Streamable interface

Author: wendigo

core/trino-main/src/main/java/io/trino/memory/ClusterMemoryManager.java

@@ -42,6 +42,7 @@
 import io.trino.spi.QueryId;
 import io.trino.spi.TrinoException;
 import io.trino.spi.memory.MemoryPoolInfo;
+import io.trino.util.Streamable;
 import jakarta.annotation.PreDestroy;
 import org.weakref.jmx.JmxException;
 import org.weakref.jmx.MBeanExporter;
@@ -60,7 +61,6 @@
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Consumer;
 import java.util.function.Function;
-import java.util.function.Supplier;
 import java.util.stream.Stream;
 
 import static com.google.common.base.Preconditions.checkState;
@@ -174,7 +174,7 @@ public synchronized void addChangeListener(Consumer<MemoryPoolInfo> listener)
         changeListeners.add(listener);
     }
 
-    public synchronized void process(Supplier<Stream<QueryExecution>> allQueries, Function<QueryId, Optional<QueryExecution>> executionInfoSupplier)
+    public synchronized void process(Streamable<QueryExecution> allQueries, Function<QueryId, Optional<QueryExecution>> executionInfoSupplier)
     {
         // TODO revocable memory reservations can also leak and may need to be detected in the future
         // We are only concerned about the leaks in the memory pool.
@@ -186,7 +186,7 @@ public synchronized void process(Supplier<Stream<QueryExecution>> allQueries, Fu
         long totalUserMemoryBytes = 0L;
         long totalMemoryBytes = 0L;
         int queriesCount = 0;
-        Iterator<QueryExecution> iterator = allQueries.get()
+        Iterator<QueryExecution> iterator = allQueries.stream()
                 .filter(query -> query.getState() == QueryState.RUNNING)
                 .iterator();
 
@@ -233,7 +233,7 @@ public synchronized void process(Supplier<Stream<QueryExecution>> allQueries, Fu
 
         if (!lowMemoryKillers.isEmpty() && outOfMemory && !queryKilled) {
             if (isLastKillTargetGone()) {
-                callOomKiller(allQueries.get(), executionInfoSupplier);
+                callOomKiller(allQueries.stream(), executionInfoSupplier);
             }
             else {
                 log.debug("Last killed target is still not gone: %s", lastKillTarget);

core/trino-main/src/main/java/io/trino/security/AccessControlUtil.java

@@ -13,10 +13,10 @@
  */
 package io.trino.security;
 
-import com.google.common.base.Supplier;
 import io.trino.SessionRepresentation;
 import io.trino.server.BasicQueryInfo;
 import io.trino.spi.security.Identity;
+import io.trino.util.Streamable;
 
 import java.util.Collection;
 import java.util.stream.Stream;
@@ -35,10 +35,10 @@ public static void checkCanViewQueryOwnedBy(Identity identity, Identity queryOwn
         accessControl.checkCanViewQueryOwnedBy(identity, queryOwner);
     }
 
-    public static Stream<BasicQueryInfo> filterQueries(Supplier<Stream<BasicQueryInfo>> queries, Identity identity, AccessControl accessControl)
+    public static Stream<BasicQueryInfo> filterQueries(Streamable<BasicQueryInfo> queries, Identity identity, AccessControl accessControl)
     {
         Collection<Identity> owners = queries
-                .get()
+                .stream()
                 .map(BasicQueryInfo::getSession)
                 .map(SessionRepresentation::toIdentity)
                 .filter(owner -> !owner.getUser().equals(identity.getUser()))
@@ -47,7 +47,7 @@ public static Stream<BasicQueryInfo> filterQueries(Supplier<Stream<BasicQueryInf
         Collection<Identity> allowedOwners = accessControl.filterQueriesOwnedBy(identity, owners);
 
         return queries
-                .get()
+                .stream()
                 .filter(queryInfo -> {
                     Identity queryIdentity = queryInfo.getSession().toIdentity();
                     return queryIdentity.getUser().equals(identity.getUser()) || allowedOwners.contains(queryIdentity);

core/trino-main/src/main/java/io/trino/util/Streamable.java

@@ -0,0 +1,21 @@
+/*
+ * 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 io.trino.util;
+
+import java.util.stream.Stream;
+
+public interface Streamable<T>
+{
+    Stream<T> stream();
+}