[core] Support guava cache in CacheManager

paimon-benchmark/paimon-micro-benchmarks/src/test/java/org/apache/paimon/benchmark/cache/CacheManagerBenchmark.java

@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.paimon.benchmark.cache;
+
+import org.apache.paimon.benchmark.Benchmark;
+import org.apache.paimon.io.cache.Cache;
+import org.apache.paimon.io.cache.CacheKey;
+import org.apache.paimon.io.cache.CacheManager;
+import org.apache.paimon.options.MemorySize;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+import java.io.RandomAccessFile;
+import java.nio.file.Path;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Benchmark for measure the performance for cache. */
+public class CacheManagerBenchmark {
+    @TempDir Path tempDir;
+
+    @Test
+    public void testCache() throws Exception {
+        Benchmark benchmark =
+                new Benchmark("cache-benchmark", 100)
+                        .setNumWarmupIters(1)
+                        .setOutputPerIteration(true);
+        File file1 = new File(tempDir.toFile(), "cache-benchmark1");
+        assertThat(file1.createNewFile()).isTrue();
+        CacheKey key1 = CacheKey.forPageIndex(new RandomAccessFile(file1, "r"), 0, 0);
+
+        File file2 = new File(tempDir.toFile(), "cache-benchmark2");
+        assertThat(file2.createNewFile()).isTrue();
+        CacheKey key2 = CacheKey.forPageIndex(new RandomAccessFile(file2, "r"), 0, 0);
+
+        for (Cache.CacheType cacheType : Cache.CacheType.values()) {
+            CacheManager cacheManager = new CacheManager(cacheType, MemorySize.ofBytes(10));
+            benchmark.addCase(
+                    String.format("cache-%s", cacheType.toString()),
+                    5,
+                    () -> {
+                        try {
+                            final int count = 10;
+                            for (int i = 0; i < count; i++) {
+                                cacheManager.getPage(
+                                        i < count / 2 ? key1 : key2,
+                                        key -> {
+                                            try {
+                                                Thread.sleep(1000);
+                                            } catch (InterruptedException e) {
+                                                throw new RuntimeException(e);
+                                            }
+                                            return new byte[6];
+                                        },
+                                        key -> {});
+                            }
+                        } catch (Exception e) {
+                            throw new RuntimeException(e);
+                        }
+                    });
+        }
+        benchmark.run();
+    }
+}

paimon-common/src/main/java/org/apache/paimon/io/cache/Cache.java

@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.paimon.io.cache;
+
+import org.apache.paimon.memory.MemorySegment;
+
+import javax.annotation.Nullable;
+
+import java.util.Map;
+import java.util.function.Function;
+
+/** Cache interface in paimon which supports caffeine and guava caches. */
+public interface Cache {
+    @Nullable
+    CacheValue get(CacheKey key, Function<CacheKey, CacheValue> supplier);
+
+    void put(CacheKey key, CacheValue value);
+
+    void invalidate(CacheKey key);
+
+    void invalidateAll();
+
+    Map<CacheKey, CacheValue> asMap();
+
+    /** Value for cache. */
+    class CacheValue {
+
+        final MemorySegment segment;
+        final CacheCallback callback;
+
+        CacheValue(MemorySegment segment, CacheCallback callback) {
+            this.segment = segment;
+            this.callback = callback;
+        }
+    }
+
+    /** Type for cache. */
+    enum CacheType {
+        CAFFEINE,
+        GUAVA;
+    }
+}

paimon-common/src/main/java/org/apache/paimon/io/cache/CacheBuilder.java

@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.paimon.io.cache;
+
+import org.apache.paimon.options.MemorySize;
+
+import org.apache.paimon.shade.caffeine2.com.github.benmanes.caffeine.cache.Caffeine;
+import org.apache.paimon.shade.caffeine2.com.github.benmanes.caffeine.cache.RemovalCause;
+import org.apache.paimon.shade.guava30.com.google.common.cache.RemovalNotification;
+
+/** Cache builder builds cache from cache type. */
+public abstract class CacheBuilder {
+    protected MemorySize memorySize;
+
+    CacheBuilder maximumWeight(MemorySize memorySize) {
+        this.memorySize = memorySize;
+        return this;
+    }
+
+    public abstract Cache build();
+
+    public static CacheBuilder newBuilder(Cache.CacheType type) {
+        switch (type) {
+            case CAFFEINE:
+                return new CaffeineCacheBuilder();
+            case GUAVA:
+                return new GuavaCacheBuilder();
+            default:
+                throw new UnsupportedOperationException("Unsupported CacheType: " + type);
+        }
+    }
+
+    static class CaffeineCacheBuilder extends CacheBuilder {
+        @Override
+        public Cache build() {
+            return new CaffeineCache(
+                    Caffeine.newBuilder()
+                            .weigher(CacheBuilder::weigh)
+                            .maximumWeight(memorySize.getBytes())
+                            .removalListener(this::onRemoval)
+                            .executor(Runnable::run)
+                            .build());
+        }
+
+        private void onRemoval(CacheKey key, Cache.CacheValue value, RemovalCause cause) {
+            if (value != null) {
+                value.callback.onRemoval(key);
+            }
+        }
+    }
+
+    static class GuavaCacheBuilder extends CacheBuilder {
+        @Override
+        public Cache build() {
+            return new GuavaCache(
+                    org.apache.paimon.shade.guava30.com.google.common.cache.CacheBuilder
+                            .newBuilder()
+                            .weigher(CacheBuilder::weigh)
+                            .maximumWeight(memorySize.getBytes())
+                            .removalListener(this::onRemoval)
+                            .build());
+        }
+
+        private void onRemoval(RemovalNotification<CacheKey, Cache.CacheValue> notification) {
+            if (notification.getValue() != null) {
+                notification.getValue().callback.onRemoval(notification.getKey());
+            }
+        }
+    }
+
+    private static int weigh(CacheKey cacheKey, Cache.CacheValue cacheValue) {
+        return cacheValue.segment.size();
+    }
+}

paimon-common/src/main/java/org/apache/paimon/io/cache/CacheManager.java

@@ -22,10 +22,6 @@
 import org.apache.paimon.memory.MemorySegment;
 import org.apache.paimon.options.MemorySize;
 
-import org.apache.paimon.shade.caffeine2.com.github.benmanes.caffeine.cache.Cache;
-import org.apache.paimon.shade.caffeine2.com.github.benmanes.caffeine.cache.Caffeine;
-import org.apache.paimon.shade.caffeine2.com.github.benmanes.caffeine.cache.RemovalCause;
-
 import java.io.IOException;
 
 import static org.apache.paimon.utils.Preconditions.checkNotNull;
@@ -39,68 +35,45 @@ public class CacheManager {
      */
     public static final int REFRESH_COUNT = 10;
 
-    private final Cache<CacheKey, CacheValue> cache;
+    private final Cache cache;
 
     private int fileReadCount;
 
     public CacheManager(MemorySize maxMemorySize) {
-        this.cache =
-                Caffeine.newBuilder()
-                        .weigher(this::weigh)
-                        .maximumWeight(maxMemorySize.getBytes())
-                        .removalListener(this::onRemoval)
-                        .executor(Runnable::run)
-                        .build();
+        this(Cache.CacheType.GUAVA, maxMemorySize);
+    }
+
+    public CacheManager(Cache.CacheType cacheType, MemorySize maxMemorySize) {
+        this.cache = CacheBuilder.newBuilder(cacheType).maximumWeight(maxMemorySize).build();
         this.fileReadCount = 0;
     }
 
     @VisibleForTesting
-    public Cache<CacheKey, ?> cache() {
+    public Cache cache() {
         return cache;
     }
 
     public MemorySegment getPage(CacheKey key, CacheReader reader, CacheCallback callback) {
-        CacheValue value =
+        Cache.CacheValue value =
                 cache.get(
                         key,
                         k -> {
                             this.fileReadCount++;
                             try {
-                                return new CacheValue(MemorySegment.wrap(reader.read(k)), callback);
+                                return new Cache.CacheValue(
+                                        MemorySegment.wrap(reader.read(key)), callback);
                             } catch (IOException e) {
                                 throw new RuntimeException(e);
                             }
                         });
-
         return checkNotNull(value, String.format("Cache result for key(%s) is null", key)).segment;
     }
 
     public void invalidPage(CacheKey key) {
         cache.invalidate(key);
     }
 
-    private int weigh(CacheKey cacheKey, CacheValue cacheValue) {
-        return cacheValue.segment.size();
-    }
-
-    private void onRemoval(CacheKey key, CacheValue value, RemovalCause cause) {
-        if (value != null) {
-            value.callback.onRemoval(key);
-        }
-    }
-
     public int fileReadCount() {
         return fileReadCount;
     }
-
-    private static class CacheValue {
-
-        private final MemorySegment segment;
-        private final CacheCallback callback;
-
-        private CacheValue(MemorySegment segment, CacheCallback callback) {
-            this.segment = segment;
-            this.callback = callback;
-        }
-    }
 }

paimon-common/src/main/java/org/apache/paimon/io/cache/CaffeineCache.java

@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.paimon.io.cache;
+
+import javax.annotation.Nullable;
+
+import java.util.Map;
+import java.util.function.Function;
+
+/** Caffeine cache implementation. */
+public class CaffeineCache implements Cache {
+    private final org.apache.paimon.shade.caffeine2.com.github.benmanes.caffeine.cache.Cache<
+                    CacheKey, CacheValue>
+            cache;
+
+    public CaffeineCache(
+            org.apache.paimon.shade.caffeine2.com.github.benmanes.caffeine.cache.Cache<
+                            CacheKey, CacheValue>
+                    cache) {
+        this.cache = cache;
+    }
+
+    @Nullable
+    @Override
+    public CacheValue get(CacheKey key, Function<CacheKey, CacheValue> supplier) {
+        return this.cache.get(key, supplier);
+    }
+
+    @Override
+    public void put(CacheKey key, CacheValue value) {
+        this.cache.put(key, value);
+    }
+
+    @Override
+    public void invalidate(CacheKey key) {
+        this.cache.invalidate(key);
+    }
+
+    @Override
+    public void invalidateAll() {
+        this.cache.invalidateAll();
+    }
+
+    @Override
+    public Map<CacheKey, CacheValue> asMap() {
+        return this.cache.asMap();
+    }
+}