BUG 35114254 - [34625628->14.1.1.0.13] ENH: Stream query results to client by partition to avoid hitting 2GB buffer limit (14.1.1.0 cl 99033 --> 14.1.1.0/ce)

back out cl's 98748, 98799, and 98806
RQ: job.9.20230310195716.229
#nobug80

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

Author: fryp

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

@@ -8,31 +8,27 @@
 
 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;
 
@@ -80,81 +76,6 @@ 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)
@@ -180,7 +101,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 =
@@ -205,7 +126,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)
         {
@@ -231,7 +152,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

@@ -18,7 +18,6 @@
 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;
@@ -60,11 +59,6 @@ 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<>();
@@ -388,28 +382,6 @@ 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.
      *

prj/coherence-core/src/main/java/com/tangosol/net/AsyncNamedCache.java

@@ -32,16 +32,13 @@
 import java.util.Map;
 import java.util.Random;
 import java.util.Set;
-import java.util.TreeSet;
 
 import java.util.concurrent.CompletableFuture;
 
 import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 import java.util.function.IntSupplier;
 
-import java.util.stream.Collectors;
-
 /**
  * Asynchronous {@link NamedCache}.
  *
@@ -106,7 +103,7 @@ public default CompletableFuture<Map<K, V>> getAll(Collection<? extends K> colKe
      * @return a {@link CompletableFuture} for a Map of keys to values for the
      *         specified filter
      */
-    public default CompletableFuture<Map<K, V>> getAll(Filter<?> filter)
+    public default CompletableFuture<Map<K, V>> getAll(Filter filter)
         {
         return invokeAll(filter, CacheProcessors.get());
         }
@@ -234,7 +231,7 @@ public default CompletableFuture<Void> removeAll(Collection<? extends K> colKeys
      *
      * @return a {@link CompletableFuture}
      */
-    public default CompletableFuture<Void> removeAll(Filter<?> filter)
+    public default CompletableFuture<Void> removeAll(Filter filter)
         {
         return invokeAll(filter, CacheProcessors.removeBlind()).thenAccept(ANY);
         }
@@ -260,7 +257,7 @@ public default CompletableFuture<Set<K>> keySet()
      *
      * @return a set of keys for entries that satisfy the specified criteria
      */
-    public default CompletableFuture<Set<K>> keySet(Filter<?> filter)
+    public default CompletableFuture<Set<K>> keySet(Filter filter)
         {
         return invokeAll(filter, CacheProcessors.nop())
                 .thenApply(Map::keySet);
@@ -291,7 +288,7 @@ public default CompletableFuture<Void> keySet(Consumer<? super K> callback)
      * @return a {@link CompletableFuture} that can be used to determine whether
      *         the operation completed
      */
-    public default CompletableFuture<Void> keySet(Filter<?> filter,
+    public default CompletableFuture<Void> keySet(Filter filter,
                                                   Consumer<? super K> callback)
         {
         return invokeAll(filter, CacheProcessors.nop(),
@@ -319,9 +316,9 @@ public default CompletableFuture<Set<Map.Entry<K, V>>> entrySet()
      *
      * @return a set of entries that satisfy the specified criteria
      */
-    public default CompletableFuture<Set<Map.Entry<K, V>>> entrySet(Filter<?> filter)
+    public default CompletableFuture<Set<Map.Entry<K, V>>> entrySet(Filter filter)
         {
-        return invokeAll(filter, CacheProcessors.binaryGet())
+        return invokeAll(filter, CacheProcessors.get())
                 .thenApply(Map::entrySet);
         }
 
@@ -341,14 +338,13 @@ public default CompletableFuture<Set<Map.Entry<K, V>>> entrySet(Filter<?> filter
      *
      * @return a set of entries that satisfy the specified criteria
      */
-    @SuppressWarnings("unchecked")
-    default CompletableFuture<Set<Map.Entry<K, V>>> entrySet(Filter<?> filter, Comparator<?> comparator)
+    public default CompletableFuture<Set<Map.Entry<K, V>>> entrySet(Filter filter, Comparator comparator)
         {
-        return entrySet(filter)
-                .thenApply(setResult ->
+        return invokeAll(filter, CacheProcessors.get())
+                .thenApply(mapResult ->
                     {
-                    int               cEntries = setResult.size();
-                    Map.Entry<K, V>[] aEntries = setResult.toArray(new Map.Entry[cEntries]);
+                    int         cEntries = mapResult.size();
+                    Map.Entry[] aEntries = mapResult.entrySet().toArray(new Map.Entry[cEntries]);
 
                     Arrays.sort(aEntries, new EntryComparator(
                             comparator == null ? SafeComparator.INSTANCE : comparator));
@@ -393,9 +389,10 @@ public default CompletableFuture<Void> entrySet(Consumer<? super Map.Entry<? ext
      * @return a {@link CompletableFuture} that can be used to determine whether
      *         the operation completed
      */
-    public default CompletableFuture<Void> entrySet(Filter<?> filter, BiConsumer<? super K, ? super V> callback)
+    public default CompletableFuture<Void> entrySet(Filter filter, BiConsumer<? super K, ? super V> callback)
         {
-        return entrySet(filter, entry -> callback.accept(entry.getKey(), entry.getValue()));
+        return invokeAll(filter, CacheProcessors.get(),
+                entry -> callback.accept(entry.getKey(), entry.getValue()));
         }
 
     /**
@@ -409,10 +406,10 @@ public default CompletableFuture<Void> entrySet(Filter<?> filter, BiConsumer<? s
      * @return a {@link CompletableFuture} that can be used to determine whether
      *         the operation completed
      */
-    public default CompletableFuture<Void> entrySet(Filter<?> filter,
+    public default CompletableFuture<Void> entrySet(Filter filter,
                                                     Consumer<? super Map.Entry<? extends K, ? extends V>> callback)
         {
-        return invokeAll(filter, CacheProcessors.binaryGet(), callback);
+        return invokeAll(filter, CacheProcessors.get(), callback);
         }
 
     /**
@@ -435,10 +432,10 @@ public default CompletableFuture<Collection<V>> values()
      * @return a collection of values for entries that satisfy the specified
      *         criteria
      */
-    public default CompletableFuture<Collection<V>> values(Filter<?> filter)
+    public default CompletableFuture<Collection<V>> values(Filter filter)
         {
-        return entrySet(filter)
-                .thenApply(entries -> entries.stream().map(Map.Entry::getValue).collect(Collectors.toList()));
+        return invokeAll(filter, CacheProcessors.get())
+                .thenApply(Map::values);
         }
 
     /**
@@ -454,10 +451,15 @@ public default CompletableFuture<Collection<V>> values(Filter<?> filter)
      * @return a collection of values for entries that satisfy the specified
      *         criteria
      */
-    public default CompletableFuture<Collection<V>> values(Filter<?> filter, Comparator<? super V> comparator)
+    public default CompletableFuture<Collection<V>> values(Filter filter, Comparator<? super V> comparator)
         {
-        return entrySet(filter)
-                .thenApply(entries -> entries.stream().map(Map.Entry::getValue).collect(Collectors.toCollection(() -> new TreeSet<>(comparator))));
+        return values(filter)
+                .thenApply(colValues ->
+                    {
+                    List<V> values = new ArrayList<>(colValues);
+                    values.sort(comparator);
+                    return values;
+                    });
         }
 
     /**
@@ -485,9 +487,10 @@ public default CompletableFuture<Void> values(Consumer<? super V> callback)
      * @return a {@link CompletableFuture} that can be used to determine whether
      *         the operation completed
      */
-    public default CompletableFuture<Void> values(Filter<?> filter, Consumer<? super V> callback)
+    public default CompletableFuture<Void> values(Filter filter, Consumer<? super V> callback)
         {
-        return entrySet(filter, entry -> callback.accept(entry.getValue()));
+        return invokeAll(filter, CacheProcessors.get(),
+                entry -> callback.accept(entry.getValue()));
         }
 
     // ---- Asynchronous InvocableMap methods -------------------------------
@@ -553,7 +556,7 @@ public <R> CompletableFuture<Map<K, R>> invokeAll(Collection<? extends K> collKe
      * @return a {@link CompletableFuture} that can be used to obtain the result
      *         of the invocation for each entry
      */
-    public <R> CompletableFuture<Map<K, R>> invokeAll(Filter<?> filter,
+    public <R> CompletableFuture<Map<K, R>> invokeAll(Filter filter,
             InvocableMap.EntryProcessor<K, V, R> processor);
 
     /**
@@ -672,7 +675,7 @@ public default <R> CompletableFuture<Void> invokeAll(Collection<? extends K> col
      * @return a {@link CompletableFuture} that can be used to determine if the
      *         operation completed successfully
      */
-    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);
 
@@ -696,7 +699,7 @@ public <R> CompletableFuture<Void> invokeAll(Filter<?> filter,
      * @return a {@link CompletableFuture} that can be used to determine if the
      *         operation completed successfully
      */
-    public default <R> CompletableFuture<Void> invokeAll(Filter<?> filter,
+    public default <R> CompletableFuture<Void> invokeAll(Filter filter,
             InvocableMap.EntryProcessor<K, V, R> processor, BiConsumer<? super K,? super R> callback)
         {
         return invokeAll(filter, processor,
@@ -747,7 +750,7 @@ public <R> CompletableFuture<R> aggregate(Collection<? extends K> collKeys,
      * @return a {@link CompletableFuture} that can be used to obtain the result
      *         of the aggregation
      */
-    public <R> CompletableFuture<R> aggregate(Filter<?> filter,
+    public <R> CompletableFuture<R> aggregate(Filter filter,
         InvocableMap.EntryAggregator<? super K, ? super V, R> aggregator);
 
     // ---- Asynchronous Map methods ----------------------------------------
@@ -1047,7 +1050,7 @@ public default CompletableFuture<Map<K, Void>> replaceAll(Collection<? extends K
      * @param filter   the filter that should be used to select entries
      * @param function the function to apply to each entry
      */
-    public default CompletableFuture<Map<K, Void>> replaceAll(Filter<?> filter,
+    public default CompletableFuture<Map<K, Void>> replaceAll(Filter filter,
             Remote.BiFunction<? super K, ? super V, ? extends V> function)
         {
         return invokeAll(filter, CacheProcessors.replace(function));