diff --git a/docs/layouts/shortcodes/generated/catalog_configuration.html b/docs/layouts/shortcodes/generated/catalog_configuration.html
index 63f7adda1e0d..a67b55dc0ba5 100644
--- a/docs/layouts/shortcodes/generated/catalog_configuration.html
+++ b/docs/layouts/shortcodes/generated/catalog_configuration.html
@@ -36,7 +36,7 @@
             <td><h5>cache-enabled</h5></td>
             <td style="word-wrap: break-word;">true</td>
             <td>Boolean</td>
-            <td>Controls whether the catalog will cache databases, tables and manifests.</td>
+            <td>Controls whether the catalog will cache databases, tables, manifests and partitions.</td>
         </tr>
         <tr>
             <td><h5>cache.expiration-interval</h5></td>
@@ -74,6 +74,12 @@
             <td>Integer</td>
             <td>Controls the max number for snapshots per table in the catalog are cached.</td>
         </tr>
+        <tr>
+            <td><h5>cache.cache-stats-enabled</h5></td>
+            <td style="word-wrap: break-word;">20</td>
+            <td>Integer</td>
+            <td>Controls whether the catalog cache stats are enabled.</td>
+        </tr>
         <tr>
             <td><h5>client-pool-size</h5></td>
             <td style="word-wrap: break-word;">2</td>
diff --git a/paimon-common/src/main/java/org/apache/paimon/options/CatalogOptions.java b/paimon-common/src/main/java/org/apache/paimon/options/CatalogOptions.java
index bb8cfae68284..f240f5f286d9 100644
--- a/paimon-common/src/main/java/org/apache/paimon/options/CatalogOptions.java
+++ b/paimon-common/src/main/java/org/apache/paimon/options/CatalogOptions.java
@@ -87,7 +87,7 @@ public class CatalogOptions {
                     .booleanType()
                     .defaultValue(true)
                     .withDescription(
-                            "Controls whether the catalog will cache databases, tables and manifests.");
+                            "Controls whether the catalog will cache databases, tables, manifests and partitions.");
 
     public static final ConfigOption<Duration> CACHE_EXPIRATION_INTERVAL_MS =
             key("cache.expiration-interval")
@@ -128,6 +128,12 @@ public class CatalogOptions {
                     .withDescription(
                             "Controls the max number for snapshots per table in the catalog are cached.");
 
+    public static final ConfigOption<Boolean> CACHE_STATS_ENABLED =
+            key("cache.cache-stats-enabled")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription("Controls whether the catalog cache stats are enabled.");
+
     public static final ConfigOption<Boolean> ALLOW_UPPER_CASE =
             ConfigOptions.key("allow-upper-case")
                     .booleanType()
diff --git a/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java b/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java
index 1caff252a654..c61ff9e14a26 100644
--- a/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java
+++ b/paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java
@@ -100,7 +100,7 @@ protected AbstractFileStore(
         this.catalogEnvironment = catalogEnvironment;
         this.writeManifestCache =
                 SegmentsCache.create(
-                        options.pageSize(), options.writeManifestCache(), Long.MAX_VALUE);
+                        options.pageSize(), options.writeManifestCache(), Long.MAX_VALUE, false);
     }
 
     @Override
diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/CachingCatalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/CachingCatalog.java
index 82d503b7a272..73e98c829247 100644
--- a/paimon-core/src/main/java/org/apache/paimon/catalog/CachingCatalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/catalog/CachingCatalog.java
@@ -20,6 +20,7 @@
 
 import org.apache.paimon.fs.Path;
 import org.apache.paimon.manifest.PartitionEntry;
+import org.apache.paimon.operation.metrics.CatalogCacheStats;
 import org.apache.paimon.options.MemorySize;
 import org.apache.paimon.options.Options;
 import org.apache.paimon.schema.SchemaChange;
@@ -48,6 +49,7 @@
 import static org.apache.paimon.options.CatalogOptions.CACHE_MANIFEST_SMALL_FILE_THRESHOLD;
 import static org.apache.paimon.options.CatalogOptions.CACHE_PARTITION_MAX_NUM;
 import static org.apache.paimon.options.CatalogOptions.CACHE_SNAPSHOT_MAX_NUM_PER_TABLE;
+import static org.apache.paimon.options.CatalogOptions.CACHE_STATS_ENABLED;
 import static org.apache.paimon.utils.Preconditions.checkNotNull;
 
 /** A {@link Catalog} to cache databases and tables and manifests. */
@@ -63,6 +65,8 @@ public class CachingCatalog extends DelegateCatalog {
     // partition cache will affect data latency
     @Nullable protected final Cache<Identifier, List<PartitionEntry>> partitionCache;
 
+    private final CatalogCacheStats catalogCacheStats;
+
     public CachingCatalog(Catalog wrapped) {
         this(
                 wrapped,
@@ -70,7 +74,8 @@ public CachingCatalog(Catalog wrapped) {
                 CACHE_MANIFEST_SMALL_FILE_MEMORY.defaultValue(),
                 CACHE_MANIFEST_SMALL_FILE_THRESHOLD.defaultValue().getBytes(),
                 CACHE_PARTITION_MAX_NUM.defaultValue(),
-                CACHE_SNAPSHOT_MAX_NUM_PER_TABLE.defaultValue());
+                CACHE_SNAPSHOT_MAX_NUM_PER_TABLE.defaultValue(),
+                CACHE_STATS_ENABLED.defaultValue());
     }
 
     public CachingCatalog(
@@ -79,7 +84,8 @@ public CachingCatalog(
             MemorySize manifestMaxMemory,
             long manifestCacheThreshold,
             long cachedPartitionMaxNum,
-            int snapshotMaxNumPerTable) {
+            int snapshotMaxNumPerTable,
+            boolean cacheStatsEnabled) {
         this(
                 wrapped,
                 expirationInterval,
@@ -87,6 +93,7 @@ public CachingCatalog(
                 manifestCacheThreshold,
                 cachedPartitionMaxNum,
                 snapshotMaxNumPerTable,
+                cacheStatsEnabled,
                 Ticker.systemTicker());
     }
 
@@ -97,6 +104,7 @@ public CachingCatalog(
             long manifestCacheThreshold,
             long cachedPartitionMaxNum,
             int snapshotMaxNumPerTable,
+            boolean cacheStatsEnabled,
             Ticker ticker) {
         super(wrapped);
         if (expirationInterval.isZero() || expirationInterval.isNegative()) {
@@ -121,7 +129,8 @@ public CachingCatalog(
                         .expireAfterAccess(expirationInterval)
                         .ticker(ticker)
                         .build();
-        this.manifestCache = SegmentsCache.create(manifestMaxMemory, manifestCacheThreshold);
+        this.manifestCache =
+                SegmentsCache.create(manifestMaxMemory, manifestCacheThreshold, cacheStatsEnabled);
         this.partitionCache =
                 cachedPartitionMaxNum == 0
                         ? null
@@ -135,6 +144,7 @@ public CachingCatalog(
                                 .maximumWeight(cachedPartitionMaxNum)
                                 .ticker(ticker)
                                 .build();
+        this.catalogCacheStats = cacheStatsEnabled ? new CatalogCacheStats() : null;
     }
 
     public static Catalog tryToCreate(Catalog catalog, Options options) {
@@ -156,7 +166,8 @@ public static Catalog tryToCreate(Catalog catalog, Options options) {
                 manifestMaxMemory,
                 manifestThreshold,
                 options.get(CACHE_PARTITION_MAX_NUM),
-                options.get(CACHE_SNAPSHOT_MAX_NUM_PER_TABLE));
+                options.get(CACHE_SNAPSHOT_MAX_NUM_PER_TABLE),
+                options.get(CACHE_STATS_ENABLED));
     }
 
     @Override
@@ -305,6 +316,23 @@ public void invalidateTable(Identifier identifier) {
         }
     }
 
+    public CatalogCacheStats getCacheStats() {
+        catalogCacheStats.setDatabaseCacheSize(databaseCache.estimatedSize());
+        catalogCacheStats.setTableCacheSize(tableCache.estimatedSize());
+        if (manifestCache != null) {
+            catalogCacheStats.setManifestCacheSize(manifestCache.getSegmentCacheSize());
+        }
+        if (partitionCache != null) {
+            int partitionSize = 0;
+            for (Map.Entry<Identifier, List<PartitionEntry>> entry :
+                    partitionCache.asMap().entrySet()) {
+                partitionSize += entry.getValue().size();
+            }
+            catalogCacheStats.setPartitionCacheSize(partitionSize);
+        }
+        return catalogCacheStats;
+    }
+
     // ================================== refresh ================================================
     // following caches will affect the latency of table, so refresh method is provided for engine
 
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java b/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java
index c73a92062b80..dbf8d94f5701 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java
@@ -393,6 +393,10 @@ private ManifestsReader.Result readManifests() {
         return manifestsReader.read(specifiedSnapshot, scanMode);
     }
 
+    public ScanMetrics getScanMetrics() {
+        return scanMetrics;
+    }
+
     // ------------------------------------------------------------------------
     // Start Thread Safe Methods: The following methods need to be thread safe because they will be
     // called by multiple threads
@@ -425,19 +429,18 @@ private List<ManifestEntry> readManifest(
             ManifestFileMeta manifest,
             @Nullable Filter<InternalRow> additionalFilter,
             @Nullable Filter<ManifestEntry> additionalTFilter) {
+        ManifestFile manifestFile = manifestFileFactory.create();
         List<ManifestEntry> entries =
-                manifestFileFactory
-                        .create()
-                        .read(
-                                manifest.fileName(),
-                                manifest.fileSize(),
-                                createCacheRowFilter(),
-                                createEntryRowFilter().and(additionalFilter),
-                                entry ->
-                                        (additionalTFilter == null || additionalTFilter.test(entry))
-                                                && (manifestEntryFilter == null
-                                                        || manifestEntryFilter.test(entry))
-                                                && filterByStats(entry));
+                manifestFile.read(
+                        manifest.fileName(),
+                        manifest.fileSize(),
+                        createCacheRowFilter(),
+                        createEntryRowFilter().and(additionalFilter),
+                        entry ->
+                                (additionalTFilter == null || additionalTFilter.test(entry))
+                                        && (manifestEntryFilter == null
+                                                || manifestEntryFilter.test(entry))
+                                        && filterByStats(entry));
         if (dropStats) {
             List<ManifestEntry> copied = new ArrayList<>(entries.size());
             for (ManifestEntry entry : entries) {
@@ -445,6 +448,13 @@ private List<ManifestEntry> readManifest(
             }
             entries = copied;
         }
+        if (scanMetrics != null && manifestFile.hitCache() != null) {
+            if (manifestFile.hitCache()) {
+                this.scanMetrics.getObjectCacheStats().increaseHitObject();
+            } else {
+                this.scanMetrics.getObjectCacheStats().increaseMissedObject();
+            }
+        }
         return entries;
     }
 
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/metrics/CatalogCacheStats.java b/paimon-core/src/main/java/org/apache/paimon/operation/metrics/CatalogCacheStats.java
new file mode 100644
index 000000000000..aa69a409c6f9
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/metrics/CatalogCacheStats.java
@@ -0,0 +1,61 @@
+/*
+ * 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.operation.metrics;
+
+/** Statistics for cache of a caching catalog. */
+public class CatalogCacheStats {
+    private long databaseCacheSize;
+    private long tableCacheSize;
+    private long manifestCacheSize;
+    private long partitionCacheSize;
+
+    public CatalogCacheStats() {}
+
+    public long getDatabaseCacheSize() {
+        return databaseCacheSize;
+    }
+
+    public void setDatabaseCacheSize(long databaseCacheSize) {
+        this.databaseCacheSize = databaseCacheSize;
+    }
+
+    public long getTableCacheSize() {
+        return tableCacheSize;
+    }
+
+    public void setTableCacheSize(long tableCacheSize) {
+        this.tableCacheSize = tableCacheSize;
+    }
+
+    public long getManifestCacheSize() {
+        return manifestCacheSize;
+    }
+
+    public void setManifestCacheSize(long manifestCacheSize) {
+        this.manifestCacheSize = manifestCacheSize;
+    }
+
+    public long getPartitionCacheSize() {
+        return partitionCacheSize;
+    }
+
+    public void setPartitionCacheSize(long partitionCacheSize) {
+        this.partitionCacheSize = partitionCacheSize;
+    }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/metrics/ObjectCacheStats.java b/paimon-core/src/main/java/org/apache/paimon/operation/metrics/ObjectCacheStats.java
new file mode 100644
index 000000000000..4aab9878d8d8
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/metrics/ObjectCacheStats.java
@@ -0,0 +1,48 @@
+/*
+ * 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.operation.metrics;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+/** Statistics for object cache of a caching catalog. */
+public class ObjectCacheStats {
+    private final AtomicLong hitObject;
+    private final AtomicLong missedObject;
+
+    public ObjectCacheStats() {
+        this.hitObject = new AtomicLong(0);
+        this.missedObject = new AtomicLong(0);
+    }
+
+    public AtomicLong getHitObject() {
+        return hitObject;
+    }
+
+    public void increaseHitObject() {
+        this.hitObject.incrementAndGet();
+    }
+
+    public AtomicLong getMissedObject() {
+        return missedObject;
+    }
+
+    public void increaseMissedObject() {
+        this.missedObject.incrementAndGet();
+    }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/metrics/ScanMetrics.java b/paimon-core/src/main/java/org/apache/paimon/operation/metrics/ScanMetrics.java
index 96f0aec1c0b2..2ee6834c8ac1 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/metrics/ScanMetrics.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/metrics/ScanMetrics.java
@@ -30,9 +30,14 @@ public class ScanMetrics {
     public static final String GROUP_NAME = "scan";
 
     private final MetricGroup metricGroup;
+    private Histogram durationHistogram;
+
+    private ScanStats latestScan;
+    private ObjectCacheStats objectCacheStats;
 
     public ScanMetrics(MetricRegistry registry, String tableName) {
         this.metricGroup = registry.tableMetricGroup(GROUP_NAME, tableName);
+        this.objectCacheStats = new ObjectCacheStats();
         registerGenericScanMetrics();
     }
 
@@ -41,10 +46,6 @@ public MetricGroup getMetricGroup() {
         return metricGroup;
     }
 
-    private Histogram durationHistogram;
-
-    private ScanStats latestScan;
-
     public static final String LAST_SCAN_DURATION = "lastScanDuration";
     public static final String SCAN_DURATION = "scanDuration";
     public static final String LAST_SCANNED_MANIFESTS = "lastScannedManifests";
@@ -72,4 +73,16 @@ public void reportScan(ScanStats scanStats) {
         latestScan = scanStats;
         durationHistogram.update(scanStats.getDuration());
     }
+
+    public ScanStats getLatestScan() {
+        return latestScan;
+    }
+
+    public ObjectCacheStats getObjectCacheStats() {
+        return objectCacheStats;
+    }
+
+    public void setObjectCacheStats(ObjectCacheStats objectCacheStats) {
+        this.objectCacheStats = objectCacheStats;
+    }
 }
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/metrics/ScanStats.java b/paimon-core/src/main/java/org/apache/paimon/operation/metrics/ScanStats.java
index 700619c3680f..0ad9514e5d04 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/metrics/ScanStats.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/metrics/ScanStats.java
@@ -38,22 +38,22 @@ public ScanStats(
     }
 
     @VisibleForTesting
-    protected long getScannedManifests() {
+    public long getScannedManifests() {
         return scannedManifests;
     }
 
     @VisibleForTesting
-    protected long getSkippedTableFiles() {
+    public long getSkippedTableFiles() {
         return skippedTableFiles;
     }
 
     @VisibleForTesting
-    protected long getResultedTableFiles() {
+    public long getResultedTableFiles() {
         return resultedTableFiles;
     }
 
     @VisibleForTesting
-    protected long getDuration() {
+    public long getDuration() {
         return duration;
     }
 }
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataTableScan.java b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataTableScan.java
index 24c6943f546f..e542ff7ae20a 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataTableScan.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataTableScan.java
@@ -26,6 +26,7 @@
 import org.apache.paimon.data.BinaryRow;
 import org.apache.paimon.metrics.MetricRegistry;
 import org.apache.paimon.operation.FileStoreScan;
+import org.apache.paimon.operation.metrics.ScanMetrics;
 import org.apache.paimon.table.source.snapshot.CompactedStartingScanner;
 import org.apache.paimon.table.source.snapshot.ContinuousCompactorStartingScanner;
 import org.apache.paimon.table.source.snapshot.ContinuousFromSnapshotFullStartingScanner;
@@ -112,6 +113,10 @@ public CoreOptions options() {
         return options;
     }
 
+    public ScanMetrics getScanMetrics() {
+        return snapshotReader.getScanMetrics();
+    }
+
     protected StartingScanner createStartingScanner(boolean isStreaming) {
         SnapshotManager snapshotManager = snapshotReader.snapshotManager();
         CoreOptions.StreamScanMode type =
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReader.java b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReader.java
index f3e0a92b8fc7..82095dbefd41 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReader.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReader.java
@@ -27,6 +27,7 @@
 import org.apache.paimon.manifest.PartitionEntry;
 import org.apache.paimon.metrics.MetricRegistry;
 import org.apache.paimon.operation.ManifestsReader;
+import org.apache.paimon.operation.metrics.ScanMetrics;
 import org.apache.paimon.predicate.Predicate;
 import org.apache.paimon.table.source.DataSplit;
 import org.apache.paimon.table.source.ScanMode;
@@ -110,6 +111,8 @@ public interface SnapshotReader {
 
     Iterator<ManifestEntry> readFileIterator();
 
+    ScanMetrics getScanMetrics();
+
     /** Result plan of this scan. */
     interface Plan extends TableScan.Plan {
 
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java
index ce01bdba9447..89ac82086f18 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java
@@ -33,6 +33,7 @@
 import org.apache.paimon.manifest.ManifestFileMeta;
 import org.apache.paimon.manifest.PartitionEntry;
 import org.apache.paimon.metrics.MetricRegistry;
+import org.apache.paimon.operation.AbstractFileStoreScan;
 import org.apache.paimon.operation.DefaultValueAssigner;
 import org.apache.paimon.operation.FileStoreScan;
 import org.apache.paimon.operation.ManifestsReader;
@@ -276,6 +277,14 @@ public SnapshotReader dropStats() {
         return this;
     }
 
+    @Override
+    public ScanMetrics getScanMetrics() {
+        if (scan instanceof AbstractFileStoreScan) {
+            return ((AbstractFileStoreScan) scan).getScanMetrics();
+        }
+        return null;
+    }
+
     @Override
     public SnapshotReader withShard(int indexOfThisSubtask, int numberOfParallelSubtasks) {
         if (splitGenerator.alwaysRawConvertible()) {
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java b/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java
index 1cb967f8d1e2..80d5bc187236 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java
@@ -34,6 +34,7 @@
 import org.apache.paimon.manifest.PartitionEntry;
 import org.apache.paimon.metrics.MetricRegistry;
 import org.apache.paimon.operation.ManifestsReader;
+import org.apache.paimon.operation.metrics.ScanMetrics;
 import org.apache.paimon.predicate.LeafPredicate;
 import org.apache.paimon.predicate.Predicate;
 import org.apache.paimon.predicate.PredicateBuilder;
@@ -406,6 +407,11 @@ public List<BucketEntry> bucketEntries() {
         public Iterator<ManifestEntry> readFileIterator() {
             return wrapped.readFileIterator();
         }
+
+        @Override
+        public ScanMetrics getScanMetrics() {
+            return wrapped.getScanMetrics();
+        }
     }
 
     private class AuditLogBatchScan implements DataTableScan {
diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/ObjectsCache.java b/paimon-core/src/main/java/org/apache/paimon/utils/ObjectsCache.java
index 1c9d9664f22f..b74e63f4179c 100644
--- a/paimon-core/src/main/java/org/apache/paimon/utils/ObjectsCache.java
+++ b/paimon-core/src/main/java/org/apache/paimon/utils/ObjectsCache.java
@@ -47,6 +47,7 @@ public class ObjectsCache<K, V> {
     private final ThreadLocal<InternalRowSerializer> formatSerializer;
     private final FunctionWithIOException<K, Long> fileSizeFunction;
     private final BiFunctionWithIOE<K, Long, CloseableIterator<InternalRow>> reader;
+    private boolean hitCache;
 
     public ObjectsCache(
             SegmentsCache<K> cache,
@@ -71,8 +72,14 @@ public List<V> read(
             throws IOException {
         Segments segments = cache.getIfPresents(key);
         if (segments != null) {
+            if (cacheStatsEnabled()) {
+                hitCache = true;
+            }
             return readFromSegments(segments, readFilter, readVFilter);
         } else {
+            if (cacheStatsEnabled()) {
+                hitCache = false;
+            }
             if (fileSize == null) {
                 fileSize = fileSizeFunction.apply(key);
             }
@@ -130,4 +137,12 @@ private Segments readSegments(K key, @Nullable Long fileSize, Filter<InternalRow
             throw new RuntimeException(e);
         }
     }
+
+    public boolean hitCache() {
+        return hitCache;
+    }
+
+    public boolean cacheStatsEnabled() {
+        return cache.cacheStatsEnabled();
+    }
 }
diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/ObjectsFile.java b/paimon-core/src/main/java/org/apache/paimon/utils/ObjectsFile.java
index b0bea8e66a82..6e5de0ae3391 100644
--- a/paimon-core/src/main/java/org/apache/paimon/utils/ObjectsFile.java
+++ b/paimon-core/src/main/java/org/apache/paimon/utils/ObjectsFile.java
@@ -207,4 +207,11 @@ public static <V> List<V> readFromIterator(
             throw new RuntimeException(e);
         }
     }
+
+    public Boolean hitCache() {
+        if (cache != null && cache.cacheStatsEnabled()) {
+            return cache.hitCache();
+        }
+        return null;
+    }
 }
diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/SegmentsCache.java b/paimon-core/src/main/java/org/apache/paimon/utils/SegmentsCache.java
index d5c2178c8b6a..2d1ad82f1aea 100644
--- a/paimon-core/src/main/java/org/apache/paimon/utils/SegmentsCache.java
+++ b/paimon-core/src/main/java/org/apache/paimon/utils/SegmentsCache.java
@@ -37,8 +37,13 @@ public class SegmentsCache<T> {
     private final Cache<T, Segments> cache;
     private final MemorySize maxMemorySize;
     private final long maxElementSize;
+    private final boolean cacheStatsEnabled;
 
-    public SegmentsCache(int pageSize, MemorySize maxMemorySize, long maxElementSize) {
+    public SegmentsCache(
+            int pageSize,
+            MemorySize maxMemorySize,
+            long maxElementSize,
+            boolean cacheStatsEnabled) {
         this.pageSize = pageSize;
         this.cache =
                 Caffeine.newBuilder()
@@ -49,6 +54,7 @@ public SegmentsCache(int pageSize, MemorySize maxMemorySize, long maxElementSize
                         .build();
         this.maxMemorySize = maxMemorySize;
         this.maxElementSize = maxElementSize;
+        this.cacheStatsEnabled = cacheStatsEnabled;
     }
 
     public int pageSize() {
@@ -63,6 +69,10 @@ public long maxElementSize() {
         return maxElementSize;
     }
 
+    public boolean cacheStatsEnabled() {
+        return cacheStatsEnabled;
+    }
+
     @Nullable
     public Segments getIfPresents(T key) {
         return cache.getIfPresent(key);
@@ -77,17 +87,29 @@ private int weigh(T cacheKey, Segments segments) {
     }
 
     @Nullable
-    public static <T> SegmentsCache<T> create(MemorySize maxMemorySize, long maxElementSize) {
-        return create((int) PAGE_SIZE.defaultValue().getBytes(), maxMemorySize, maxElementSize);
+    public static <T> SegmentsCache<T> create(
+            MemorySize maxMemorySize, long maxElementSize, boolean cacheStatsEnabled) {
+        return create(
+                (int) PAGE_SIZE.defaultValue().getBytes(),
+                maxMemorySize,
+                maxElementSize,
+                cacheStatsEnabled);
     }
 
     @Nullable
     public static <T> SegmentsCache<T> create(
-            int pageSize, MemorySize maxMemorySize, long maxElementSize) {
+            int pageSize,
+            MemorySize maxMemorySize,
+            long maxElementSize,
+            boolean cacheStatsEnabled) {
         if (maxMemorySize.getBytes() == 0) {
             return null;
         }
 
-        return new SegmentsCache<>(pageSize, maxMemorySize, maxElementSize);
+        return new SegmentsCache<>(pageSize, maxMemorySize, maxElementSize, cacheStatsEnabled);
+    }
+
+    public long getSegmentCacheSize() {
+        return cache.estimatedSize();
     }
 }
diff --git a/paimon-core/src/test/java/org/apache/paimon/catalog/CachingCatalogTest.java b/paimon-core/src/test/java/org/apache/paimon/catalog/CachingCatalogTest.java
index 4792e33c932b..ad9c70191141 100644
--- a/paimon-core/src/test/java/org/apache/paimon/catalog/CachingCatalogTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/catalog/CachingCatalogTest.java
@@ -353,7 +353,8 @@ private void innerTestManifestCache(long manifestCacheThreshold) throws Exceptio
                         MemorySize.ofMebiBytes(1),
                         manifestCacheThreshold,
                         0L,
-                        10);
+                        10,
+                        false);
         Identifier tableIdent = new Identifier("db", "tbl");
         catalog.dropTable(tableIdent, true);
         catalog.createTable(tableIdent, DEFAULT_TABLE_SCHEMA, false);
diff --git a/paimon-core/src/test/java/org/apache/paimon/catalog/TestableCachingCatalog.java b/paimon-core/src/test/java/org/apache/paimon/catalog/TestableCachingCatalog.java
index 1d4a9b0e8a58..0c1a53536aa1 100644
--- a/paimon-core/src/test/java/org/apache/paimon/catalog/TestableCachingCatalog.java
+++ b/paimon-core/src/test/java/org/apache/paimon/catalog/TestableCachingCatalog.java
@@ -45,6 +45,7 @@ public TestableCachingCatalog(Catalog catalog, Duration expirationInterval, Tick
                 Long.MAX_VALUE,
                 Long.MAX_VALUE,
                 Integer.MAX_VALUE,
+                false,
                 ticker);
         this.cacheExpirationInterval = expirationInterval;
     }
diff --git a/paimon-core/src/test/java/org/apache/paimon/utils/ObjectsCacheTest.java b/paimon-core/src/test/java/org/apache/paimon/utils/ObjectsCacheTest.java
index 9d3275e3ab48..9216c32e7e2e 100644
--- a/paimon-core/src/test/java/org/apache/paimon/utils/ObjectsCacheTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/utils/ObjectsCacheTest.java
@@ -44,7 +44,7 @@ public void test() throws IOException {
         Map<String, List<String>> map = new HashMap<>();
         ObjectsCache<String, String> cache =
                 new ObjectsCache<>(
-                        new SegmentsCache<>(1024, MemorySize.ofKibiBytes(5), Long.MAX_VALUE),
+                        new SegmentsCache<>(1024, MemorySize.ofKibiBytes(5), Long.MAX_VALUE, true),
                         new StringSerializer(),
                         RowType.of(DataTypes.STRING()),
                         k -> 1L,
@@ -62,6 +62,7 @@ public void test() throws IOException {
                 cache.read(
                         "k1", null, Filter.alwaysTrue(), Filter.alwaysTrue(), Filter.alwaysTrue());
         assertThat(values).isEmpty();
+        assertThat(cache.hitCache()).isFalse();
 
         // test values
         List<String> expect = Arrays.asList("v1", "v2", "v3");
@@ -70,12 +71,14 @@ public void test() throws IOException {
                 cache.read(
                         "k2", null, Filter.alwaysTrue(), Filter.alwaysTrue(), Filter.alwaysTrue());
         assertThat(values).containsExactlyElementsOf(expect);
+        assertThat(cache.hitCache()).isFalse();
 
         // test cache
         values =
                 cache.read(
                         "k2", null, Filter.alwaysTrue(), Filter.alwaysTrue(), Filter.alwaysTrue());
         assertThat(values).containsExactlyElementsOf(expect);
+        assertThat(cache.hitCache()).isTrue();
 
         // test filter
         values =