BUG 35114255 - [34625628->14.1.1.0.13-CE] ENH: Stream query results to client by partition to avoid hitting 2GB buffer limit (v14.1.1.0 -> ce/v14.1.1.0 @ 98748)

[git-p4: depot-paths = "//dev/coherence-ce/release/coherence-ce-v14.1.1.0/": change = 98756]

Author: aseovic

prj/coherence-core/src/main/java/com/tangosol/internal/util/DefaultAsyncNamedCache.java

@@ -1,34 +1,38 @@
 /*
- * Copyright (c) 2000, 2020, Oracle and/or its affiliates.
+ * Copyright (c) 2000, 2023, Oracle and/or its affiliates.
  *
  * Licensed under the Universal Permissive License v 1.0 as shown at
- * http://oss.oracle.com/licenses/upl.
+ * https://oss.oracle.com/licenses/upl.
  */
 package com.tangosol.internal.util;
 
 import com.oracle.coherence.common.util.Options;
 
+import com.tangosol.internal.util.processor.CacheProcessors;
+
 import com.tangosol.net.AsyncNamedCache;
 import com.tangosol.net.CacheService;
 import com.tangosol.net.Member;
 import com.tangosol.net.NamedCache;
 import com.tangosol.net.PartitionedService;
+import com.tangosol.net.partition.PartitionSet;
 
 import com.tangosol.util.Filter;
+import com.tangosol.util.ImmutableArrayList;
 import com.tangosol.util.InvocableMap;
-
-import com.tangosol.internal.util.processor.CacheProcessors;
 import com.tangosol.util.InvocableMap.StreamingAggregator;
-
 import com.tangosol.util.aggregator.AsynchronousAggregator;
-
+import com.tangosol.util.filter.PartitionedFilter;
 import com.tangosol.util.processor.AsynchronousProcessor;
 import com.tangosol.util.processor.SingleEntryAsynchronousProcessor;
 import com.tangosol.util.processor.StreamingAsynchronousProcessor;
 
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
+import java.util.Set;
 
 import java.util.concurrent.CompletableFuture;
 
@@ -76,6 +80,81 @@ public NamedCache<K, V> getNamedCache()
         return m_cache;
         }
 
+    @SuppressWarnings("unchecked")
+    @Override
+    public CompletableFuture<Set<Map.Entry<K, V>>> entrySet(Filter<?> filter)
+        {
+        // optimized implementation that runs query against individual partitions
+        // in parallel and aggregates the results
+
+        if (m_cache.getCacheService() instanceof PartitionedService && !(filter instanceof PartitionedFilter))
+            {
+            int          cParts = ((PartitionedService) m_cache.getCacheService()).getPartitionCount();
+            PartitionSet parts  = new PartitionSet(cParts);
+
+            List<CompletableFuture<Void>> futures     = new ArrayList<>(cParts);
+            List<Map.Entry<K, V>>         listEntries = new ArrayList<>();
+
+            for (int i = 0; i < cParts; i++)
+                {
+                parts.add(i);
+                futures.add(invokeAll(new PartitionedFilter<>(filter, parts), new AsynchronousProcessor<>(CacheProcessors.binaryGet(), i))
+                                    .thenAccept(results -> listEntries.addAll(results.entrySet())));
+                parts.remove(i);
+                }
+
+            CompletableFuture<Set<Map.Entry<K, V>>> result = new CompletableFuture<>();
+            CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()]))
+                    .whenComplete((v, err) ->
+                        {
+                        if (err != null)
+                            {
+                            result.completeExceptionally(err);
+                            }
+                        else
+                            {
+                            result.complete(new ImmutableArrayList(listEntries).getSet());
+                            }
+                    });
+
+            return result;
+            }
+        else
+            {
+            return AsyncNamedCache.super.entrySet(filter);
+            }
+        }
+
+    @Override
+    public CompletableFuture<Void> entrySet(Filter<?> filter, Consumer<? super Map.Entry<? extends K, ? extends V>> callback)
+        {
+        // optimized implementation that runs query against individual partitions
+        // in parallel and aggregates the results
+
+        if (m_cache.getCacheService() instanceof PartitionedService && !(filter instanceof PartitionedFilter))
+            {
+            int          cParts  = ((PartitionedService) m_cache.getCacheService()).getPartitionCount();
+            PartitionSet parts   = new PartitionSet(cParts);
+
+            List<CompletableFuture<Void>> futures = new ArrayList<>(cParts);
+
+            for (int i = 0; i < cParts; i++)
+                {
+                parts.add(i);
+                futures.add(invokeAll(new PartitionedFilter<>(filter, parts),
+                                      new StreamingAsynchronousProcessor<>(CacheProcessors.binaryGet(), i, callback),
+                                      callback)); // needed to ensure the streaming invokeAll is called!
+                parts.remove(i);
+                }
+
+            return CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()]));
+            }
+        else
+            {
+            return AsyncNamedCache.super.entrySet(filter, callback);
+            }
+        }
+
     @Override
     public <R> CompletableFuture<R> invoke(K key,
                                 InvocableMap.EntryProcessor<K, V, R> processor)
@@ -101,7 +180,7 @@ public <R> CompletableFuture<Map<K, R>> invokeAll(Collection<? extends K> collKe
         }
 
     @Override
-    public <R> CompletableFuture<Map<K, R>> invokeAll(Filter filter,
+    public <R> CompletableFuture<Map<K, R>> invokeAll(Filter<?> filter,
                                            InvocableMap.EntryProcessor<K, V, R> processor)
         {
         AsynchronousProcessor<K, V, R> asyncProcessor =
@@ -126,7 +205,7 @@ public <R> CompletableFuture<Void> invokeAll(Collection<? extends K> collKeys,
         }
 
     @Override
-    public <R> CompletableFuture<Void> invokeAll(Filter filter,
+    public <R> CompletableFuture<Void> invokeAll(Filter<?> filter,
                                                  InvocableMap.EntryProcessor<K, V, R> processor,
                                                  Consumer<? super Map.Entry<? extends K, ? extends R>> callback)
         {
@@ -152,7 +231,7 @@ public <R> CompletableFuture<R> aggregate(
 
     @Override
     public <R> CompletableFuture<R> aggregate(
-            Filter filter, InvocableMap.EntryAggregator<? super K, ? super V, R> aggregator)
+            Filter<?> filter, InvocableMap.EntryAggregator<? super K, ? super V, R> aggregator)
         {
         AsynchronousAggregator<? super K, ? super V, ?, R> asyncAggregator =
                 instantiateAsyncAggregator(aggregator);

prj/coherence-core/src/main/java/com/tangosol/internal/util/processor/CacheProcessors.java

@@ -1,8 +1,8 @@
 /*
- * Copyright (c) 2000, 2020, Oracle and/or its affiliates.
+ * Copyright (c) 2000, 2023, Oracle and/or its affiliates.
  *
  * Licensed under the Universal Permissive License v 1.0 as shown at
- * http://oss.oracle.com/licenses/upl.
+ * https://oss.oracle.com/licenses/upl.
  */
 package com.tangosol.internal.util.processor;
 
@@ -18,6 +18,7 @@
 import com.tangosol.net.NamedCache;
 
 import com.tangosol.util.BinaryEntry;
+import com.tangosol.util.ConverterCollections;
 import com.tangosol.util.ExternalizableHelper;
 import com.tangosol.util.InvocableMap;
 import com.tangosol.util.LiteMap;
@@ -59,6 +60,11 @@ public static <K, V> InvocableMap.EntryProcessor<K, V, V> get()
         return new Get<>();
         }
 
+    public static <K, V> InvocableMap.EntryProcessor<K, V, V> binaryGet()
+        {
+        return new BinaryGet<>();
+        }
+
     public static <K, V> InvocableMap.EntryProcessor<K, V, Optional<V>> getOrDefault()
         {
         return new GetOrDefault<>();
@@ -382,6 +388,28 @@ public Map<K, V> processAll(Set<? extends InvocableMap.Entry<K, V>> setEntries)
             }
         }
 
+    /**
+     * Get entry processor that avoids value deserialization by returning
+     * Binary value directly, assuming it will be converted on the client
+     * via {@link ConverterCollections.ConverterEntry} or similar.
+     *
+     * @param <K>  the type of the Map entry key
+     * @param <V>  the type of the Map entry value
+     */
+    public static class BinaryGet<K, V>
+            extends Get<K, V>
+        {
+        @SuppressWarnings("unchecked")
+        @Override
+        public V process(InvocableMap.Entry<K, V> entry)
+            {
+            // cast below is a hack to make compiler happy,
+            // and only works because of type erasure, as both
+            // V and Binary erase into Object
+            return (V) ((BinaryEntry) entry).getBinaryValue();
+            }
+        }
+
     /**
      * GetOrDefault entry processor.
      *