Handle task memory requiremetns which exceed node sizes

If task memory requirements exceed memory capacity of any node in the
cluster treat such task as if it would request full node.

core/trino-main/src/main/java/io/trino/execution/scheduler/FullNodeCapableNodeAllocatorService.java

@@ -53,6 +53,7 @@
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Supplier;
 
 import static com.google.common.base.Preconditions.checkState;
@@ -65,6 +66,7 @@
 import static io.airlift.concurrent.Threads.daemonThreadsNamed;
 import static io.trino.execution.scheduler.FallbackToFullNodePartitionMemoryEstimator.FULL_NODE_MEMORY;
 import static io.trino.spi.StandardErrorCode.NO_NODES_AVAILABLE;
+import static java.lang.Math.max;
 import static java.lang.Thread.currentThread;
 import static java.util.Comparator.comparing;
 import static java.util.Objects.requireNonNull;
@@ -110,6 +112,7 @@ public class FullNodeCapableNodeAllocatorService
     private final AtomicBoolean stopped = new AtomicBoolean();
     private final Semaphore processSemaphore = new Semaphore(0);
     private final ConcurrentMap<String, Long> nodePoolSizes = new ConcurrentHashMap<>();
+    private final AtomicLong maxNodePoolSize = new AtomicLong(FULL_NODE_MEMORY.toBytes());
 
     @Inject
     public FullNodeCapableNodeAllocatorService(
@@ -142,13 +145,20 @@ private void refreshNodePoolSizes()
             }
         }
 
+        long tmpMaxNodePoolSize = 0;
         for (Map.Entry<String, Optional<MemoryInfo>> entry : workerMemoryInfo.entrySet()) {
             Optional<MemoryInfo> memoryInfo = entry.getValue();
             if (memoryInfo.isEmpty()) {
                 continue;
             }
-            nodePoolSizes.put(entry.getKey(), memoryInfo.get().getPool().getMaxBytes());
+            long nodePoolSize = memoryInfo.get().getPool().getMaxBytes();
+            nodePoolSizes.put(entry.getKey(), nodePoolSize);
+            tmpMaxNodePoolSize = max(tmpMaxNodePoolSize, nodePoolSize);
         }
+        if (tmpMaxNodePoolSize == 0) {
+            tmpMaxNodePoolSize = FULL_NODE_MEMORY.toBytes();
+        }
+        maxNodePoolSize.set(tmpMaxNodePoolSize);
     }
 
     private Optional<Long> getNodePoolSize(InternalNode internalNode)
@@ -209,6 +219,13 @@ private void processSharedPendingAcquires()
                     iterator.remove();
                     continue;
                 }
+                if (pendingAcquire.getNodeRequirements().getMemory().toBytes() > maxNodePoolSize.get()) {
+                    // nodes in the cluster shrank and what used to be a request for a shared node now is a request for full node
+                    iterator.remove();
+                    detachedFullNodePendingAcquires.add(pendingAcquire);
+                    continue;
+                }
+
                 try {
                     Candidates candidates = selectCandidates(pendingAcquire.getNodeRequirements(), pendingAcquire.getNodeSelector());
                     if (candidates.isEmpty()) {
@@ -659,6 +676,6 @@ public void release()
 
     private boolean isFullNode(NodeRequirements requirements)
     {
-        return requirements.getMemory().compareTo(FULL_NODE_MEMORY) >= 0;
+        return requirements.getMemory().toBytes() >= maxNodePoolSize.get();
     }
 }

core/trino-main/src/test/java/io/trino/execution/scheduler/TestFullNodeCapableNodeAllocator.java

@@ -77,6 +77,8 @@ public class TestFullNodeCapableNodeAllocator
     private static final NodeRequirements FULL_NODE_3_REQUIREMENTS = new NodeRequirements(Optional.empty(), Set.of(NODE_3_ADDRESS), FULL_NODE_MEMORY);
     private static final NodeRequirements FULL_NODE_CATALOG_1_REQUIREMENTS = new NodeRequirements(Optional.of(CATALOG_1), Set.of(), FULL_NODE_MEMORY);
     private static final NodeRequirements FULL_NODE_CATALOG_2_REQUIREMENTS = new NodeRequirements(Optional.of(CATALOG_2), Set.of(), FULL_NODE_MEMORY);
+    // not using FULL_NODE_MEMORY marker but with memory requirements exceeding any node in cluster
+    private static final NodeRequirements EFFECTIVELY_FULL_NODE_REQUIREMENTS = new NodeRequirements(Optional.empty(), Set.of(), DataSize.of(65, GIGABYTE));
 
     // none of the tests should require periodic execution of routine which processes pending acquisitions
     private static final long TEST_TIMEOUT = FullNodeCapableNodeAllocatorService.PROCESS_PENDING_ACQUIRES_DELAY_SECONDS * 1000 / 2;
@@ -254,19 +256,32 @@ public void testRemoveAcquiredSharedNode()
     @Test(timeOut = TEST_TIMEOUT)
     public void testAllocateFullSimple()
             throws Exception
+    {
+        testAllocateFullSimple(FULL_NODE_REQUIREMENTS);
+    }
+
+    @Test(timeOut = TEST_TIMEOUT)
+    public void testEffectivelyFullNodeSimple()
+            throws Exception
+    {
+        testAllocateFullSimple(EFFECTIVELY_FULL_NODE_REQUIREMENTS);
+    }
+
+    private void testAllocateFullSimple(NodeRequirements fullNodeRequirements)
+            throws Exception
     {
         TestingNodeSupplier nodeSupplier = TestingNodeSupplier.create(basicNodesMap(NODE_1, NODE_2));
         setupNodeAllocatorService(nodeSupplier, 3);
 
         try (NodeAllocator nodeAllocator = nodeAllocatorService.getNodeAllocator(Q1_SESSION)) {
             // allocate 2 full nodes should not block
-            NodeAllocator.NodeLease acquire1 = nodeAllocator.acquire(FULL_NODE_REQUIREMENTS);
+            NodeAllocator.NodeLease acquire1 = nodeAllocator.acquire(fullNodeRequirements);
             assertAcquired(acquire1);
-            NodeAllocator.NodeLease acquire2 = nodeAllocator.acquire(FULL_NODE_REQUIREMENTS);
+            NodeAllocator.NodeLease acquire2 = nodeAllocator.acquire(fullNodeRequirements);
             assertAcquired(acquire2);
 
             // trying to allocate third full node should block
-            NodeAllocator.NodeLease acquire3 = nodeAllocator.acquire(FULL_NODE_REQUIREMENTS);
+            NodeAllocator.NodeLease acquire3 = nodeAllocator.acquire(fullNodeRequirements);
             assertNotAcquired(acquire3);
 
             // third acquisition should unblock if one of old ones is released
@@ -288,7 +303,7 @@ public void testAllocateFullSimple()
             });
 
             // shared acquisition should block full acquisition
-            NodeAllocator.NodeLease acquire5 = nodeAllocator.acquire(FULL_NODE_REQUIREMENTS);
+            NodeAllocator.NodeLease acquire5 = nodeAllocator.acquire(fullNodeRequirements);
             assertNotAcquired(acquire5);
 
             // and when shared acquisition is gone full node should be acquired