Core: Expose the stats of the manifest file content cache

For observability purposes clients could use the stats of the
manifest file content cache to see for instance the cache
hit/miss ratio so that users can fine tune the configuration
of the cache.

Author: Gabor Kaszab

core/src/main/java/org/apache/iceberg/ManifestFiles.java

@@ -33,6 +33,7 @@
 import org.apache.iceberg.io.FileIO;
 import org.apache.iceberg.io.InputFile;
 import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.metrics.CacheMetrics;
 import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
@@ -86,6 +87,11 @@ public static void dropCache(FileIO fileIO) {
     CONTENT_CACHES.cleanUp();
   }
 
+  /** Get statistics of the manifest file content cache for a FileIO. */
+  public static CacheMetrics contentCacheStats(FileIO io) {
+    return CacheMetrics.of(contentCache(io).stats());
+  }
+
   /**
    * Returns a {@link CloseableIterable} of file paths in the {@link ManifestFile}.
    *

core/src/main/java/org/apache/iceberg/metrics/CacheMetrics.java

@@ -0,0 +1,51 @@
+/*
+ * 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.iceberg.metrics;
+
+import com.github.benmanes.caffeine.cache.stats.CacheStats;
+import org.immutables.value.Value;
+
+@Value.Immutable
+public abstract class CacheMetrics {
+  public abstract long hitCount();
+
+  public abstract long missCount();
+
+  public abstract long loadSuccessCount();
+
+  public abstract long loadFailureCount();
+
+  public abstract long totalLoadTime();
+
+  public abstract long evictionCount();
+
+  public abstract long evictionWeight();
+
+  public static CacheMetrics of(CacheStats stats) {
+    return ImmutableCacheMetrics.builder()
+        .hitCount(stats.hitCount())
+        .missCount(stats.missCount())
+        .loadSuccessCount(stats.loadSuccessCount())
+        .loadFailureCount(stats.loadFailureCount())
+        .totalLoadTime(stats.totalLoadTime())
+        .evictionCount(stats.evictionCount())
+        .evictionWeight(stats.evictionWeight())
+        .build();
+  }
+}

core/src/test/java/org/apache/iceberg/TestManifestCaching.java

@@ -36,6 +36,7 @@
 import org.apache.iceberg.hadoop.HadoopFileIO;
 import org.apache.iceberg.io.ContentCache;
 import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.metrics.CacheMetrics;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.types.Types;
@@ -78,16 +79,16 @@ public void testPlanWithCache() throws Exception {
     assertThat(cache.estimatedCacheSize())
         .as("All manifest files should be cached")
         .isEqualTo(numFiles);
-    assertThat(cache.stats().loadCount())
+    assertThat(ManifestFiles.contentCacheStats(table.io()).loadSuccessCount())
         .as("All manifest files should be recently loaded")
         .isEqualTo(numFiles);
-    long missCount = cache.stats().missCount();
+    long missCount = ManifestFiles.contentCacheStats(table.io()).missCount();
 
     // planFiles and verify that cache size still the same
     TableScan scan2 = table.newScan();
     assertThat(scan2.planFiles()).hasSize(numFiles);
     assertThat(cache.estimatedCacheSize()).isEqualTo(numFiles);
-    assertThat(cache.stats().missCount())
+    assertThat(ManifestFiles.contentCacheStats(table.io()).missCount())
         .as("All manifest file reads should hit cache")
         .isEqualTo(missCount);
 
@@ -115,10 +116,14 @@ public void testPlanWithSmallCache() throws Exception {
     assertThat(cache.maxTotalBytes()).isEqualTo(1);
     assertThat(scan.planFiles()).hasSize(numFiles);
     assertThat(cache.estimatedCacheSize()).isEqualTo(0);
-    assertThat(cache.stats().loadCount())
+    assertThat(ManifestFiles.contentCacheStats(table.io()).loadSuccessCount())
         .as("File should not be loaded through cache")
         .isEqualTo(0);
-    assertThat(cache.stats().requestCount()).as("Cache should not serve file").isEqualTo(0);
+    assertThat(
+            ManifestFiles.contentCacheStats(table.io()).hitCount()
+                + ManifestFiles.contentCacheStats(table.io()).missCount())
+        .as("Cache should not serve file")
+        .isEqualTo(0);
     ManifestFiles.dropCache(scan.table().io());
   }
 
@@ -199,7 +204,7 @@ public void testWeakFileIOReferenceCleanUp() {
     assertThat(cache1).isSameAs(firstCache);
     assertThat(cacheN).isSameAs(lastCache);
     assertThat(manifestCache.stats().loadCount()).isEqualTo(maxIO);
-    assertThat(manifestCache.stats().evictionCount()).isEqualTo(maxIO - 2);
+    assertThat(CacheMetrics.of(manifestCache.stats()).evictionCount()).isEqualTo(maxIO - 2);
   }
 
   /**

core/src/test/java/org/apache/iceberg/metrics/TestCacheMetrics.java

@@ -0,0 +1,70 @@
+/*
+ * 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.iceberg.metrics;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.Caffeine;
+import com.github.benmanes.caffeine.cache.Weigher;
+import org.junit.jupiter.api.Test;
+
+public class TestCacheMetrics {
+  @Test
+  public void testNoInputStats() {
+    CacheMetrics cacheMetrics = CacheMetrics.of(Caffeine.newBuilder().build().stats());
+
+    assertThat(cacheMetrics.hitCount()).isZero();
+    assertThat(cacheMetrics.missCount()).isZero();
+    assertThat(cacheMetrics.loadSuccessCount()).isZero();
+    assertThat(cacheMetrics.loadFailureCount()).isZero();
+    assertThat(cacheMetrics.totalLoadTime()).isZero();
+    assertThat(cacheMetrics.evictionCount()).isZero();
+    assertThat(cacheMetrics.evictionWeight()).isZero();
+  }
+
+  @Test
+  public void testCacheMetricsFromCaffeineCache() {
+    int maxTotalWeight = 300;
+
+    Cache<Integer, Integer> inputCache =
+        Caffeine.newBuilder()
+            .maximumWeight(maxTotalWeight)
+            .weigher((Weigher<Integer, Integer>) (key, value) -> value * 100)
+            .recordStats()
+            .build();
+
+    inputCache.get(1, key -> key);
+    inputCache.get(1, key -> key);
+    inputCache.get(2, key -> key);
+    inputCache.get(3, key -> key); // This evicts the other entries due to max weight
+
+    inputCache.cleanUp();
+
+    CacheMetrics cacheMetrics = CacheMetrics.of(inputCache.stats());
+
+    assertThat(cacheMetrics.hitCount()).isOne();
+    assertThat(cacheMetrics.missCount()).isEqualTo(3);
+    assertThat(cacheMetrics.loadSuccessCount()).isEqualTo(3);
+    assertThat(cacheMetrics.loadFailureCount()).isZero();
+    assertThat(cacheMetrics.totalLoadTime()).isGreaterThan(0);
+    assertThat(cacheMetrics.evictionCount()).isEqualTo(2);
+    assertThat(cacheMetrics.evictionWeight()).isEqualTo(300);
+  }
+}