Use LRUCache in TableMetadataCache

We recently discovered that LRUCache, based on LinkedHashMap, performs
almost twice as fast as the Caffein max size cache. Let's replace the
caffeine cache to optimise the performance.

Author: Alex Borodin

flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableMetadataCache.java

@@ -18,8 +18,6 @@
  */
 package org.apache.iceberg.flink.sink.dynamic;
 
-import com.github.benmanes.caffeine.cache.Cache;
-import com.github.benmanes.caffeine.cache.Caffeine;
 import java.util.Map;
 import java.util.Set;
 import org.apache.flink.annotation.Internal;
@@ -53,18 +51,18 @@ class TableMetadataCache {
   private final Catalog catalog;
   private final long refreshMs;
   private final int inputSchemasPerTableCacheMaximumSize;
-  private final Cache<TableIdentifier, CacheItem> cache;
+  private final LRUCache<TableIdentifier, CacheItem> tableCache;
 
   TableMetadataCache(
       Catalog catalog, int maximumSize, long refreshMs, int inputSchemasPerTableCacheMaximumSize) {
     this.catalog = catalog;
     this.refreshMs = refreshMs;
     this.inputSchemasPerTableCacheMaximumSize = inputSchemasPerTableCacheMaximumSize;
-    this.cache = Caffeine.newBuilder().maximumSize(maximumSize).build();
+    this.tableCache = new LRUCache<>(maximumSize, ignored -> {});
   }
 
   Tuple2<Boolean, Exception> exists(TableIdentifier identifier) {
-    CacheItem cached = cache.getIfPresent(identifier);
+    CacheItem cached = tableCache.get(identifier);
     if (cached != null && Boolean.TRUE.equals(cached.tableExists)) {
       return EXISTS;
     } else if (needsRefresh(cached, true)) {
@@ -87,7 +85,7 @@ PartitionSpec spec(TableIdentifier identifier, PartitionSpec spec) {
   }
 
   void update(TableIdentifier identifier, Table table) {
-    cache.put(
+    tableCache.put(
         identifier,
         new CacheItem(
             true,
@@ -98,7 +96,7 @@ void update(TableIdentifier identifier, Table table) {
   }
 
   private String branch(TableIdentifier identifier, String branch, boolean allowRefresh) {
-    CacheItem cached = cache.getIfPresent(identifier);
+    CacheItem cached = tableCache.get(identifier);
     if (cached != null && cached.tableExists && cached.branches.contains(branch)) {
       return branch;
     }
@@ -113,7 +111,7 @@ private String branch(TableIdentifier identifier, String branch, boolean allowRe
 
   private ResolvedSchemaInfo schema(
       TableIdentifier identifier, Schema input, boolean allowRefresh) {
-    CacheItem cached = cache.getIfPresent(identifier);
+    CacheItem cached = tableCache.get(identifier);
     Schema compatible = null;
     if (cached != null && cached.tableExists) {
       // This only works if the {@link Schema#equals(Object)} returns true for the old schema
@@ -164,7 +162,7 @@ private ResolvedSchemaInfo schema(
   }
 
   private PartitionSpec spec(TableIdentifier identifier, PartitionSpec spec, boolean allowRefresh) {
-    CacheItem cached = cache.getIfPresent(identifier);
+    CacheItem cached = tableCache.get(identifier);
     if (cached != null && cached.tableExists) {
       for (PartitionSpec tableSpec : cached.specs.values()) {
         if (PartitionSpecEvolution.checkCompatibility(tableSpec, spec)) {
@@ -188,7 +186,7 @@ private Tuple2<Boolean, Exception> refreshTable(TableIdentifier identifier) {
       return EXISTS;
     } catch (NoSuchTableException e) {
       LOG.debug("Table doesn't exist {}", identifier, e);
-      cache.put(identifier, new CacheItem(false, null, null, null, 1));
+      tableCache.put(identifier, new CacheItem(false, null, null, null, 1));
       return Tuple2.of(false, e);
     }
   }
@@ -199,7 +197,7 @@ private boolean needsRefresh(CacheItem cacheItem, boolean allowRefresh) {
   }
 
   public void invalidate(TableIdentifier identifier) {
-    cache.invalidate(identifier);
+    tableCache.remove(identifier);
   }
 
   /** Handles timeout for missing items only. Caffeine performance causes noticeable delays. */
@@ -268,7 +266,7 @@ DataConverter recordConverter() {
   }
 
   @VisibleForTesting
-  Cache<TableIdentifier, CacheItem> getInternalCache() {
-    return cache;
+  LRUCache<TableIdentifier, CacheItem> getInternalCache() {
+    return tableCache;
   }
 }

flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableMetadataCache.java

@@ -90,7 +90,6 @@ void testCachingDisabled() {
     TableMetadataCache cache = new TableMetadataCache(catalog, 0, Long.MAX_VALUE, 10);
 
     // Cleanup routine doesn't run after every write
-    cache.getInternalCache().cleanUp();
-    assertThat(cache.getInternalCache().estimatedSize()).isEqualTo(0);
+    assertThat(cache.getInternalCache().size()).isEqualTo(0);
   }
 }

flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java

@@ -85,11 +85,11 @@ void testBranchCreationAndCaching() {
 
     catalog.createTable(tableIdentifier, SCHEMA);
     tableUpdater.update(tableIdentifier, "myBranch", SCHEMA, PartitionSpec.unpartitioned());
-    TableMetadataCache.CacheItem cacheItem = cache.getInternalCache().getIfPresent(tableIdentifier);
+    TableMetadataCache.CacheItem cacheItem = cache.getInternalCache().get(tableIdentifier);
     assertThat(cacheItem).isNotNull();
 
     tableUpdater.update(tableIdentifier, "myBranch", SCHEMA, PartitionSpec.unpartitioned());
-    assertThat(cache.getInternalCache().getIfPresent(tableIdentifier)).isEqualTo(cacheItem);
+    assertThat(cache.getInternalCache().get(tableIdentifier)).isEqualTo(cacheItem);
   }
 
   @Test
@@ -153,7 +153,6 @@ void testLastResultInvalidation() {
         .isEqualTo(CompareSchemasVisitor.Result.SAME);
 
     // Last result cache should be cleared
-    assertThat(cache.getInternalCache().getIfPresent(tableIdentifier).inputSchemas().get(SCHEMA2))
-        .isNull();
+    assertThat(cache.getInternalCache().get(tableIdentifier).inputSchemas().get(SCHEMA2)).isNull();
   }
 }