apache · gripleaf · Sep 16, 2026
diff --git a/docs/source/api/scan.rst b/docs/source/api/scan.rst
@@ -47,13 +47,82 @@ Decimal literals are rescaled to the bucket field's type only when the conversio
 is exact. NaN literals and decimals that cannot be represented exactly disable
 inferred bucket pruning.
 
+Sharing table metadata
+======================
+
+``TableScanResources`` retains schema metadata across scans of the same managed table and
+branch. Create it once with a file system and pass it to each ``ScanContextBuilder``:
+
+.. code-block:: cpp
+
+   PAIMON_ASSIGN_OR_RAISE(
+       std::shared_ptr<paimon::TableScanResources> resources,
+       paimon::TableScanResources::Create(table_path, file_system, "main"));
+
+   // Repeat for each query, reusing resources.
+   paimon::ScanContextBuilder builder(table_path);
+   builder.WithTableResources(resources).SetPredicate(predicate);
+   PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<paimon::ScanContext> context, builder.Finish());
+   PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<paimon::TableScan> scan,
+                          paimon::TableScan::Create(std::move(context)));
+
+Include ``paimon/table/source/table_scan_resources.h`` to create the resource object.
+At ``Finish()``, the resources supply the context's file system when none was explicitly set.
+An explicit ``WithFileSystem()`` must point to the same instance as the resources' file system;
+a different instance is rejected regardless of builder call order. The resources also supply
+the default branch; conflicting explicit branches are rejected. The physical table path must match,
+including when scanning its ``$ro`` or ``$audit_log`` system table. Format tables and global system tables do not use these resources.
+``Finish()`` resets the builder's resource setting, like ``WithCache()`` and ``WithExecutor()``.
+
+Every new scan still checks for the latest schema ID. Previously loaded schema versions, Arrow
+schemas, and partition and primary-key field information are reused. Existing scans retain their
+original schema. Snapshot selection, filters, streaming progress, executors and
+scan metrics remain independent. ``SetTableSchema()`` keeps its existing
+behavior: on main it bypasses the shared schema cache; on other branches it is ignored.
+
+Successfully loaded snapshots are cached by their full file paths, with an LRU limit of 20 entries
+per resource object. The whole snapshot cache is replaced on the first access after it reaches
+30 minutes of age. Reads and writes do not extend this deadline; recently added entries are
+discarded along with older ones. There is no per-entry TTL or background refresh thread.
+Snapshot caching is enabled by these resources; ordinary snapshot managers have no cache unless
+one is explicitly supplied.
+Latest and earliest snapshot discovery and existence checks still use the catalog or file system
+as appropriate. Read or parse failures are not cached.
+
+Snapshot deletion through a manager replaces its entire injected snapshot cache before and after
+the deletion attempt. Explicit invalidation also discards the whole cache. In-flight loads can
+finish against the old cache but cannot populate the replacement used by subsequent callers.
+Historical commit lookup, timestamp searches, and retained-snapshot publication checks read files
+directly so cached metadata cannot hide missing or replaced files. Other clients' deletions do not
+immediately invalidate this process's cache. A cached snapshot can remain available after its
+metadata file expires; a cache hit does not establish that the snapshot or its data is still readable.
+
+The resources can be shared by concurrent scans when the supplied file system supports concurrent
+use. Schema versions and schema-derived resources have no entry-count limit and are retained for
+the resource object's lifetime. Eviction releases the cache's references; active scans retain
+the metadata they need.
+The snapshot cache limit bounds entry counts, not bytes or metadata held by active scans.
+There is no background refresh.
+The caller must recreate the resources after fast-forward, deleting and recreating a table or
+branch, or changing file system access configuration.
+Fast-forward can replace schema and snapshot contents under existing IDs; discovering the latest ID
+does not refresh their cached contents. After fast-forward completes, create a new
+``TableScanResources`` and use it for subsequent scans; cache hits in existing resources may still
+return the old contents, and existing scans retain their original metadata. This follows
+Paimon's `fast-forward cache refresh requirement
+<https://paimon.apache.org/docs/1.3/maintenance/manage-branches/#fast-forward>`_.
+A metadata cache does not pin snapshots or prevent their data files from expiring.
+
 Interface
 =========
 
 .. doxygenclass:: paimon::TableScan
    :members:
    :undoc-members:
 
+.. doxygenclass:: paimon::TableScanResources
+   :members:
+
 .. doxygenclass:: paimon::ScanContextBuilder
    :members:
    :undoc-members:

diff --git a/include/paimon/scan_context.h b/include/paimon/scan_context.h
@@ -33,6 +33,7 @@
 #include "paimon/visibility.h"
 namespace paimon {
 class ScanContextBuilder;
+class TableScanResources;
 class ScanFilter;
 class Executor;
 class FormatTable;
@@ -56,7 +57,8 @@ class PAIMON_EXPORT ScanContext {
                 const std::optional<std::string>& table_schema,
                 const std::map<std::string, std::string>& options,
                 const std::shared_ptr<Cache>& cache,
-                const std::shared_ptr<FormatTable>& format_table);
+                const std::shared_ptr<FormatTable>& format_table,
+                const std::shared_ptr<TableScanResources>& table_resources);
 
     ~ScanContext();
 
@@ -99,6 +101,10 @@ class PAIMON_EXPORT ScanContext {
         return specific_file_system_;
     }
 
+    const std::shared_ptr<TableScanResources>& GetTableResources() const {
+        return table_resources_;
+    }
+
     const std::optional<std::string>& GetSpecificTableSchema() const {
         return table_schema_;
     }
@@ -127,6 +133,7 @@ class PAIMON_EXPORT ScanContext {
     std::map<std::string, std::string> options_;
     std::shared_ptr<Cache> cache_;
     std::shared_ptr<FormatTable> format_table_;
+    std::shared_ptr<TableScanResources> table_resources_;
 };
 
 /// Filter configuration for table scan operations
@@ -219,7 +226,8 @@ class PAIMON_EXPORT ScanContextBuilder {
     /// This bypasses the global file system registry and uses the provided implementation directly.
     /// @param file_system The file system to use.
     /// @return Reference to this builder for method chaining.
-    /// @note If not set, use default file system (configured in `Options::FILE_SYSTEM`)
+    /// @note If not set, use the table resources' file system when provided, otherwise the
+    /// default file system (configured in `Options::FILE_SYSTEM`).
     ScanContextBuilder& WithFileSystem(const std::shared_ptr<FileSystem>& file_system);
 
     /// Set the table schema as a string to avoid schema loading I/O operations.
@@ -238,6 +246,14 @@ class PAIMON_EXPORT ScanContextBuilder {
     /// @return Reference to this builder for method chaining.
     ScanContextBuilder& WithCache(const std::shared_ptr<Cache>& cache);
 
+    /// Share metadata resources with other scans of the same managed table and branch.
+    /// Finish() fills the context's file system from the resources when none was explicitly set.
+    /// An explicit file system must point to the same instance as the resources' file system.
+    /// The resources also supply the default branch; conflicting explicit branches are rejected.
+    /// Passing nullptr disables sharing. Finish() resets this setting.
+    /// On main, SetTableSchema() continues to bypass the shared schema cache.
+    ScanContextBuilder& WithTableResources(const std::shared_ptr<TableScanResources>& resources);
+
     /// Build and return a `ScanContext` instance with input validation.
     /// @return Result containing the constructed `ScanContext` or an error status.
     Result<std::unique_ptr<ScanContext>> Finish();

diff --git a/include/paimon/table/source/table_scan_resources.h b/include/paimon/table/source/table_scan_resources.h
@@ -0,0 +1,68 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <memory>
+#include <string>
+
+#include "paimon/result.h"
+#include "paimon/visibility.h"
+
+namespace paimon {
+class FileSystem;
+class TableScanResourcesAccess;
+
+/// Metadata resources shared by scans of one managed table and branch.
+///
+/// Share this object through ScanContextBuilder::WithTableResources(). Each scan still discovers
+/// the latest schema ID and chooses its own snapshot. Schema versions and schema-derived resources
+/// are retained for this object's lifetime without an entry limit.
+/// Snapshots use a separate 20-entry LRU cache. On access, the whole snapshot
+/// cache is replaced once it is 30 minutes old; reads and writes do not extend that deadline.
+/// Eviction releases cache references; active scans retain the metadata they use. The snapshot
+/// cache limit counts entries, not bytes. Latest snapshot discovery and existence checks query the
+/// file system. A cached snapshot does not pin its metadata or data files. The caller must recreate
+/// this object after fast-forward, dropping and recreating a table or branch, or changing the file
+/// system's access configuration. Fast-forward can replace schema and snapshot contents under
+/// existing IDs; there is no immediate cross-client invalidation. Use the new resources for
+/// subsequent scans.
+///
+/// Concurrent scans may share these resources. The supplied file system must support concurrent
+/// use. Filters, executors, scan progress and scan metrics are not shared.
+class PAIMON_EXPORT TableScanResources {
+ public:
+    /// Creates resources without loading table metadata.
+    /// @param table_path Physical table root, without a system table suffix.
+    /// @param file_system Non-null file system used by scans sharing these resources.
+    /// @param branch Branch to scan; an empty branch is normalized to main.
+    static Result<std::shared_ptr<TableScanResources>> Create(
+        const std::string& table_path, const std::shared_ptr<FileSystem>& file_system,
+        const std::string& branch);
+
+    ~TableScanResources();
+
+ private:
+    friend class TableScanResourcesAccess;
+    class Impl;
+    explicit TableScanResources(std::unique_ptr<Impl>&& impl);
+
+    std::unique_ptr<Impl> impl_;
+};
+}  // namespace paimon
diff --git a/src/paimon/CMakeLists.txt b/src/paimon/CMakeLists.txt
@@ -449,6 +449,7 @@ set(PAIMON_CORE_SRCS
     core/table/source/startup_mode.cpp
     core/table/source/table_read.cpp
     core/table/source/table_scan.cpp
+    core/table/source/table_scan_resources.cpp
     core/table/source/data_evolution_batch_scan.cpp
     core/table/source/primary_key_sorted_index_scan.cpp
     core/table/source/primary_key_sorted_index_result.cpp
@@ -957,6 +958,7 @@ if(PAIMON_BUILD_TESTS)
                     core/table/source/snapshot/snapshot_reader_test.cpp
                     core/table/source/startup_mode_test.cpp
                     core/table/source/table_scan_test.cpp
+                    core/table/source/table_scan_resources_test.cpp
                     core/table/system/system_table_test.cpp
                     core/tag/tag_test.cpp
                     core/utils/blob_view_lookup_test.cpp

diff --git a/src/paimon/core/operation/expire_snapshots.cpp b/src/paimon/core/operation/expire_snapshots.cpp
@@ -151,7 +151,9 @@ Result<int32_t> ExpireSnapshots::ExpireUntil(int64_t earliest_snapshot_id, int64
     }
     std::vector<Snapshot> retained_snapshots;
     for (int64_t id = end_exclusive_id; id <= latest_snapshot_id; ++id) {
-        PAIMON_ASSIGN_OR_RAISE(Snapshot snapshot, snapshot_manager_->LoadSnapshot(id));
+        // Cached metadata cannot prove that the current file has been published.
+        PAIMON_ASSIGN_OR_RAISE(Snapshot snapshot,
+                               snapshot_manager_->LoadSnapshotFromFileSystem(id));
         retained_snapshots.push_back(std::move(snapshot));
     }
     if (latest.from_catalog && !(retained_snapshots.back() == latest.snapshot.value())) {
@@ -226,7 +228,7 @@ Result<int32_t> ExpireSnapshots::ExpireUntil(int64_t earliest_snapshot_id, int64
                 expired_offset_files.insert(offsets_path.value());
             }
         }
-        auto status = fs_->Delete(snapshot_manager_->SnapshotPath(id));
+        auto status = snapshot_manager_->DeleteSnapshot(id);
         // delete quietly will ignore any status error
         (void)status;
     }

diff --git a/src/paimon/core/operation/expire_snapshots_test.cpp b/src/paimon/core/operation/expire_snapshots_test.cpp
@@ -43,6 +43,7 @@
 #include "paimon/format/file_format.h"
 #include "paimon/fs/local/local_file_system.h"
 #include "paimon/memory/memory_pool.h"
+#include "paimon/testing/utils/snapshot_test_helper.h"
 #include "paimon/testing/utils/testharness.h"
 
 namespace paimon::test {
@@ -228,6 +229,31 @@ TEST_F(ExpireSnapshotsTest, TestInvalidInput) {
     }
 }
 
+TEST_F(ExpireSnapshotsTest, CachedRetainedSnapshotDoesNotProvePublication) {
+    auto cache = std::make_shared<SnapshotManager::SnapshotCache>();
+    auto manager = std::make_shared<SnapshotManager>(fs_, test_data_path_, "main", cache);
+    ASSERT_OK(fs_->Mkdirs(manager->SnapshotDirectory()));
+    for (int64_t id : {1, 2, 3}) {
+        ASSERT_OK_AND_ASSIGN(std::string json, BuildTestSnapshot(id).ToJsonString());
+        ASSERT_OK(fs_->WriteFile(manager->SnapshotPath(id), json, false));
+        ASSERT_OK(manager->LoadSnapshot(id));
+    }
+    ASSERT_OK(manager->CommitEarliestHint(1));
+    ASSERT_OK(manager->CommitLatestHint(3));
+    ASSERT_OK(fs_->Delete(manager->SnapshotPath(3)));
+    manager->SetSnapshotLoader([]() -> Result<std::optional<Snapshot>> {
+        return std::optional<Snapshot>(BuildTestSnapshot(3));
+    });
+    ASSERT_OK_AND_ASSIGN(CoreOptions options,
+                         CoreOptions::FromMap({{Options::SNAPSHOT_NUM_RETAINED_MIN, "2"},
+                                               {Options::SNAPSHOT_NUM_RETAINED_MAX, "2"}}));
+    ExpireSnapshots expire(manager, path_factory_, manifest_list_, manifest_file_, fs_,
+                           options.GetExpireConfig(), options.RealtimeEnabled(), executor_);
+    ASSERT_TRUE(expire.Expire().status().IsNotExist());
+    ASSERT_OK_AND_ASSIGN(bool exists, manager->SnapshotExists(1));
+    ASSERT_TRUE(exists);
+}
+
 TEST_F(ExpireSnapshotsTest, TestGetDataFileToDelete) {
     auto mgr = std::make_shared<SnapshotManager>(fs_, test_data_path_);
     ASSERT_OK_AND_ASSIGN(CoreOptions options, CoreOptions::FromMap({}));

diff --git a/src/paimon/core/operation/scan_context.cpp b/src/paimon/core/operation/scan_context.cpp
@@ -21,6 +21,8 @@
 #include <utility>
 
 #include "paimon/common/utils/path_util.h"
+#include "paimon/core/table/source/table_scan_resources_impl.h"
+#include "paimon/defs.h"
 #include "paimon/executor.h"
 #include "paimon/memory/memory_pool.h"
 #include "paimon/status.h"
@@ -40,7 +42,8 @@ ScanContext::ScanContext(const std::string& path, bool is_streaming_mode,
                          const std::optional<std::string>& table_schema,
                          const std::map<std::string, std::string>& options,
                          const std::shared_ptr<Cache>& cache,
-                         const std::shared_ptr<FormatTable>& format_table)
+                         const std::shared_ptr<FormatTable>& format_table,
+                         const std::shared_ptr<TableScanResources>& table_resources)
     : path_(path),
       is_streaming_mode_(is_streaming_mode),
       limit_(limit),
@@ -53,7 +56,8 @@ ScanContext::ScanContext(const std::string& path, bool is_streaming_mode,
       table_schema_(table_schema),
       options_(options),
       cache_(cache),
-      format_table_(format_table) {}
+      format_table_(format_table),
+      table_resources_(table_resources) {}
 
 ScanContext::~ScanContext() = default;
 
@@ -75,6 +79,7 @@ class ScanContextBuilder::Impl {
         table_schema_ = std::nullopt;
         options_.clear();
         cache_.reset();
+        table_resources_.reset();
     }
 
  private:
@@ -97,6 +102,7 @@ class ScanContextBuilder::Impl {
     std::optional<std::string> table_schema_;
     std::map<std::string, std::string> options_;
     std::shared_ptr<Cache> cache_;
+    std::shared_ptr<TableScanResources> table_resources_;
 };
 
 ScanContextBuilder::ScanContextBuilder(const std::string& path)
@@ -191,11 +197,20 @@ ScanContextBuilder& ScanContextBuilder::WithCache(const std::shared_ptr<Cache>&
     return *this;
 }
 
+ScanContextBuilder& ScanContextBuilder::WithTableResources(
+    const std::shared_ptr<TableScanResources>& resources) {
+    impl_->table_resources_ = resources;
+    return *this;
+}
+
 Result<std::unique_ptr<ScanContext>> ScanContextBuilder::Finish() {
     if (impl_->built_from_format_table_ && impl_->format_table_ == nullptr) {
         return Status::Invalid("cannot scan with null format table");
     }
     if (impl_->format_table_ != nullptr) {
+        if (impl_->table_resources_) {
+            return Status::Invalid("table scan resources cannot be used with a format table");
+        }
         // The table already answers both, and from a source this cannot see behind, so a second
         // answer is refused rather than silently dropped.
         if (impl_->table_schema_) {
@@ -213,15 +228,25 @@ Result<std::unique_ptr<ScanContext>> ScanContextBuilder::Finish() {
     if (impl_->path_.empty()) {
         return Status::Invalid("cannot scan with empty table path");
     }
+    auto options = impl_->options_;
+    std::shared_ptr<FileSystem> file_system = impl_->specific_file_system_;
+    if (impl_->table_resources_) {
+        auto& resources = TableScanResourcesAccess::Get(*impl_->table_resources_);
+        auto branch = options.emplace(Options::BRANCH, resources.branch_).first;
+        if (!file_system) {
+            file_system = resources.file_system_;
+        }
+        PAIMON_RETURN_NOT_OK(resources.Validate(impl_->path_, branch->second, file_system));
+    }
     std::shared_ptr<Executor> executor =
         impl_->executor_ ? impl_->executor_ : CreateDefaultExecutor();
     auto ctx = std::make_unique<ScanContext>(
         impl_->path_, impl_->is_streaming_mode_, impl_->limit_,
         std::make_shared<ScanFilter>(impl_->predicates_, impl_->partition_filters_,
                                      impl_->bucket_filter_),
         impl_->global_index_result_, impl_->realtime_context_, impl_->memory_pool_, executor,
-        impl_->specific_file_system_, impl_->table_schema_, impl_->options_, impl_->cache_,
-        impl_->format_table_);
+        file_system, impl_->table_schema_, options, impl_->cache_, impl_->format_table_,
+        impl_->table_resources_);
     impl_->Reset();
     return ctx;
 }