Disable cache sharing when cache is disabled in JDBC connectors

lib/trino-collect/src/main/java/io/trino/collect/cache/EmptyCache.java

@@ -0,0 +1,238 @@
+/*
+ * 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.collect.cache;
+
+import com.google.common.cache.AbstractLoadingCache;
+import com.google.common.cache.CacheLoader;
+import com.google.common.cache.CacheStats;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.util.concurrent.UncheckedExecutionException;
+
+import javax.annotation.CheckForNull;
+
+import java.util.Collection;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutionException;
+
+import static java.util.Objects.requireNonNull;
+
+class EmptyCache<K, V>
+        extends AbstractLoadingCache<K, V>
+{
+    private final CacheLoader<? super K, V> loader;
+    private final StatsCounter statsCounter;
+
+    EmptyCache(CacheLoader<? super K, V> loader, boolean recordStats)
+    {
+        this.loader = requireNonNull(loader, "loader is null");
+        this.statsCounter = recordStats ? new SimpleStatsCounter() : new NoopStatsCounter();
+    }
+
+    @CheckForNull
+    @Override
+    public V getIfPresent(Object key)
+    {
+        statsCounter.recordMisses(1);
+        return null;
+    }
+
+    @Override
+    public V get(K key)
+            throws ExecutionException
+    {
+        return get(key, () -> loader.load(key));
+    }
+
+    @Override
+    public V get(K key, Callable<? extends V> valueLoader)
+            throws ExecutionException
+    {
+        statsCounter.recordMisses(1);
+        try {
+            V value = valueLoader.call();
+            statsCounter.recordLoadSuccess(1);
+            return value;
+        }
+        catch (RuntimeException e) {
+            statsCounter.recordLoadException(1);
+            throw new UncheckedExecutionException(e);
+        }
+        catch (Exception e) {
+            statsCounter.recordLoadException(1);
+            throw new ExecutionException(e);
+        }
+    }
+
+    @Override
+    public void put(K key, V value)
+    {
+        // Cache, even if configured to evict everything immediately, should allow writes.
+    }
+
+    @Override
+    public void refresh(K key) {}
+
+    @Override
+    public void invalidate(Object key) {}
+
+    @Override
+    public void invalidateAll(Iterable<?> keys) {}
+
+    @Override
+    public void invalidateAll() {}
+
+    @Override
+    public long size()
+    {
+        return 0;
+    }
+
+    @Override
+    public CacheStats stats()
+    {
+        return statsCounter.snapshot();
+    }
+
+    @Override
+    public ConcurrentMap<K, V> asMap()
+    {
+        return new ConcurrentMap<K, V>()
+        {
+            @Override
+            public V putIfAbsent(K key, V value)
+            {
+                // Cache, even if configured to evict everything immediately, should allow writes.
+                return value;
+            }
+
+            @Override
+            public boolean remove(Object key, Object value)
+            {
+                return false;
+            }
+
+            @Override
+            public boolean replace(K key, V oldValue, V newValue)
+            {
+                return false;
+            }
+
+            @Override
+            public V replace(K key, V value)
+            {
+                return null;
+            }
+
+            @Override
+            public int size()
+            {
+                return 0;
+            }
+
+            @Override
+            public boolean isEmpty()
+            {
+                return true;
+            }
+
+            @Override
+            public boolean containsKey(Object key)
+            {
+                return false;
+            }
+
+            @Override
+            public boolean containsValue(Object value)
+            {
+                return false;
+            }
+
+            @Override
+            public V get(Object key)
+            {
+                return null;
+            }
+
+            @Override
+            public V put(K key, V value)
+            {
+                // Cache, even if configured to evict everything immediately, should allow writes.
+                return null;
+            }
+
+            @Override
+            public V remove(Object key)
+            {
+                return null;
+            }
+
+            @Override
+            public void putAll(Map<? extends K, ? extends V> m)
+            {
+                // Cache, even if configured to evict everything immediately, should allow writes.
+            }
+
+            @Override
+            public void clear() {}
+
+            @Override
+            public Set<K> keySet()
+            {
+                return ImmutableSet.of();
+            }
+
+            @Override
+            public Collection<V> values()
+            {
+                return ImmutableSet.of();
+            }
+
+            @Override
+            public Set<Entry<K, V>> entrySet()
+            {
+                return ImmutableSet.of();
+            }
+        };
+    }
+
+    private static class NoopStatsCounter
+            implements StatsCounter
+    {
+        private static final CacheStats EMPTY_STATS = new SimpleStatsCounter().snapshot();
+
+        @Override
+        public void recordHits(int count) {}
+
+        @Override
+        public void recordMisses(int count) {}
+
+        @Override
+        public void recordLoadSuccess(long loadTime) {}
+
+        @Override
+        public void recordLoadException(long loadTime) {}
+
+        @Override
+        public void recordEviction() {}
+
+        @Override
+        public CacheStats snapshot()
+        {
+            return EMPTY_STATS;
+        }
+    }
+}

lib/trino-collect/src/main/java/io/trino/collect/cache/EvictableCacheBuilder.java

@@ -51,6 +51,7 @@ public static EvictableCacheBuilder<Object, Object> newBuilder()
     private Optional<Long> maximumWeight = Optional.empty();
     private Optional<Weigher<? super Token<K>, ? super V>> weigher = Optional.empty();
     private boolean recordStats;
+    private Optional<DisabledCacheImplementation> disabledCacheImplementation = Optional.empty();
 
     private EvictableCacheBuilder() {}
 
@@ -109,6 +110,27 @@ public EvictableCacheBuilder<K, V> recordStats()
         return this;
     }
 
+    /**
+     * Choose a behavior for case when caching is disabled that may allow data and failure sharing between concurrent callers.
+     */
+    public EvictableCacheBuilder<K, V> shareResultsAndFailuresEvenIfDisabled()
+    {
+        checkState(!disabledCacheImplementation.isPresent(), "disabledCacheImplementation already set");
+        disabledCacheImplementation = Optional.of(DisabledCacheImplementation.GUAVA);
+        return this;
+    }
+
+    /**
+     * Choose a behavior for case when caching is disabled that prevents data and failure sharing between concurrent callers.
+     * Note: disabled cache won't report any statistics.
+     */
+    public EvictableCacheBuilder<K, V> shareNothingWhenDisabled()
+    {
+        checkState(!disabledCacheImplementation.isPresent(), "disabledCacheImplementation already set");
+        disabledCacheImplementation = Optional.of(DisabledCacheImplementation.NOOP);
+        return this;
+    }
+
     @CheckReturnValue
     public <K1 extends K, V1 extends V> Cache<K1, V1> build()
     {
@@ -119,15 +141,26 @@ public <K1 extends K, V1 extends V> Cache<K1, V1> build()
     public <K1 extends K, V1 extends V> LoadingCache<K1, V1> build(CacheLoader<? super K1, V1> loader)
     {
         if (cacheDisabled()) {
-            // Disabled cache is always empty, so doesn't exhibit invalidation problems.
-            // Avoid overhead of EvictableCache wrapper.
-            CacheBuilder<Object, Object> cacheBuilder = CacheBuilder.newBuilder()
-                    .maximumSize(0)
-                    .expireAfterWrite(0, SECONDS);
-            if (recordStats) {
-                cacheBuilder.recordStats();
+            // Silently providing a behavior different from Guava's could be surprising, so require explicit choice.
+            DisabledCacheImplementation disabledCacheImplementation = this.disabledCacheImplementation.orElseThrow(() -> new IllegalStateException(
+                    "Even when cache is disabled, the loads are synchronized and both load results and failures are shared between threads. " +
+                            "This is rarely desired, thus builder caller is expected to either opt-in into this behavior with shareResultsAndFailuresEvenIfDisabled(), " +
+                            "or choose not to share results (and failures) between concurrent invocations with shareNothingWhenDisabled()."));
+            switch (disabledCacheImplementation) {
+                case NOOP:
+                    return new EmptyCache<>(loader, recordStats);
+                case GUAVA:
+                    // Disabled cache is always empty, so doesn't exhibit invalidation problems.
+                    // Avoid overhead of EvictableCache wrapper.
+                    CacheBuilder<Object, Object> cacheBuilder = CacheBuilder.newBuilder()
+                            .maximumSize(0)
+                            .expireAfterWrite(0, SECONDS);
+                    if (recordStats) {
+                        cacheBuilder.recordStats();
+                    }
+                    return buildUnsafeCache(cacheBuilder, loader);
             }
-            return buildUnsafeCache(cacheBuilder, loader);
+            throw new UnsupportedOperationException("Unsupported option: " + disabledCacheImplementation);
         }
 
         if (!(maximumSize.isPresent() || maximumWeight.isPresent() || expireAfterWrite.isPresent())) {
@@ -191,4 +224,10 @@ private static Duration toDuration(long duration, TimeUnit unit)
         // Saturated conversion, as in com.google.common.cache.CacheBuilder.toNanosSaturated
         return Duration.ofNanos(unit.toNanos(duration));
     }
+
+    private enum DisabledCacheImplementation
+    {
+        NOOP,
+        GUAVA,
+    }
 }