feat(snapshot): add snapshot file listing API

include/paimon/api.h

@@ -20,24 +20,25 @@
 
 #pragma once
 
-#include "paimon/commit_context.h"           // IWYU pragma: export
-#include "paimon/defs.h"                     // IWYU pragma: export
-#include "paimon/factories/factory.h"        // IWYU pragma: export
-#include "paimon/file_store_commit.h"        // IWYU pragma: export
-#include "paimon/file_store_write.h"         // IWYU pragma: export
-#include "paimon/fs/file_system_factory.h"   // IWYU pragma: export
-#include "paimon/memory/memory_pool.h"       // IWYU pragma: export
-#include "paimon/predicate/predicate.h"      // IWYU pragma: export
-#include "paimon/read_context.h"             // IWYU pragma: export
-#include "paimon/reader/batch_reader.h"      // IWYU pragma: export
-#include "paimon/record_batch.h"             // IWYU pragma: export
-#include "paimon/result.h"                   // IWYU pragma: export
-#include "paimon/scan_context.h"             // IWYU pragma: export
-#include "paimon/statistics_mode.h"          // IWYU pragma: export
-#include "paimon/status.h"                   // IWYU pragma: export
-#include "paimon/table/source/table_read.h"  // IWYU pragma: export
-#include "paimon/table/source/table_scan.h"  // IWYU pragma: export
-#include "paimon/write_context.h"            // IWYU pragma: export
+#include "paimon/commit_context.h"               // IWYU pragma: export
+#include "paimon/defs.h"                         // IWYU pragma: export
+#include "paimon/factories/factory.h"            // IWYU pragma: export
+#include "paimon/file_store_commit.h"            // IWYU pragma: export
+#include "paimon/file_store_write.h"             // IWYU pragma: export
+#include "paimon/fs/file_system_factory.h"       // IWYU pragma: export
+#include "paimon/memory/memory_pool.h"           // IWYU pragma: export
+#include "paimon/predicate/predicate.h"          // IWYU pragma: export
+#include "paimon/read_context.h"                 // IWYU pragma: export
+#include "paimon/reader/batch_reader.h"          // IWYU pragma: export
+#include "paimon/record_batch.h"                 // IWYU pragma: export
+#include "paimon/result.h"                       // IWYU pragma: export
+#include "paimon/scan_context.h"                 // IWYU pragma: export
+#include "paimon/snapshot/snapshot_file_scan.h"  // IWYU pragma: export
+#include "paimon/statistics_mode.h"              // IWYU pragma: export
+#include "paimon/status.h"                       // IWYU pragma: export
+#include "paimon/table/source/table_read.h"      // IWYU pragma: export
+#include "paimon/table/source/table_scan.h"      // IWYU pragma: export
+#include "paimon/write_context.h"                // IWYU pragma: export
 
 // IWYU pragma: begin_exports
 #include "paimon/realtime/realtime_context.h"

include/paimon/snapshot/snapshot_file_scan.h

@@ -0,0 +1,76 @@
+/*
+ * 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 <cstdint>
+#include <map>
+#include <memory>
+#include <optional>
+#include <set>
+#include <string>
+
+#include "paimon/result.h"
+#include "paimon/visibility.h"
+
+namespace paimon {
+class Executor;
+class FileSystem;
+class MemoryPool;
+class ScanFilter;
+
+/// Lists the physical files required by a table snapshot.
+class PAIMON_EXPORT SnapshotFileScan {
+ public:
+    SnapshotFileScan() = delete;
+    ~SnapshotFileScan() = delete;
+
+    /// List the physical files required to materialize a snapshot.
+    ///
+    /// The result includes the snapshot and schema files, manifest lists and manifest files,
+    /// live data and changelog files, external file indexes, the index manifest and its live index
+    /// files, statistics, and real-time offset files. Mutable snapshot hints such as `LATEST` and
+    /// `EARLIEST` are not included.
+    ///
+    /// Partition and bucket filters apply only to files which belong to a partition or bucket.
+    /// Snapshot-level shared metadata is always returned. A manifest file is returned when it
+    /// cannot be pruned by manifest-level partition and bucket statistics, even if entry-level
+    /// filtering later removes all of its entries. Predicate filters are not supported.
+    ///
+    /// @param table_path Root path of the table.
+    /// @param branch Branch to list. An empty value selects the main branch. This parameter takes
+    ///               precedence over a branch configured in options.
+    /// @param snapshot_id Snapshot to list, or `std::nullopt` to use the latest snapshot.
+    /// @param scan_filter Optional partition and bucket filters. Partition maps use AND semantics,
+    ///                    while the vector uses OR semantics. Predicate filters are rejected.
+    /// @param options User options overriding options stored in the latest table schema.
+    /// @param file_system Optional file system implementation. If null, it is resolved from
+    /// options.
+    /// @param executor Optional executor used to read manifest files in parallel.
+    /// @param memory_pool Optional memory pool. If null, the default pool is used.
+    /// @return Physical file paths required by the selected snapshot and filters.
+    static Result<std::set<std::string>> ListFiles(
+        const std::string& table_path, const std::string& branch,
+        const std::optional<int64_t>& snapshot_id, const std::shared_ptr<ScanFilter>& scan_filter,
+        const std::map<std::string, std::string>& options,
+        const std::shared_ptr<FileSystem>& file_system, const std::shared_ptr<Executor>& executor,
+        const std::shared_ptr<MemoryPool>& memory_pool);
+};
+
+}  // namespace paimon

src/paimon/CMakeLists.txt

@@ -399,6 +399,7 @@ set(PAIMON_CORE_SRCS
     core/schema/schema_validation.cpp
     core/schema/table_schema.cpp
     core/snapshot.cpp
+    core/snapshot_file_scan.cpp
     core/snapshot_info.cpp
     core/stats/simple_stats_collector.cpp
     core/stats/simple_stats_converter.cpp
@@ -896,6 +897,7 @@ if(PAIMON_BUILD_TESTS)
                     core/schema/schema_validation_test.cpp
                     core/schema/arrow_schema_validator_test.cpp
                     core/schema/table_schema_test.cpp
+                    core/snapshot_file_scan_test.cpp
                     core/snapshot_test.cpp
                     core/stats/simple_stats_evolution_test.cpp
                     core/stats/simple_stats_collector_test.cpp

src/paimon/core/operation/file_store_scan.h

@@ -213,6 +213,13 @@ class FileStoreScan {
 
     Result<std::vector<PartitionEntry>> ReadPartitionEntries() const;
 
+    /// Merge raw manifest entries into the set of currently-live files.
+    ///
+    /// Entries are deduplicated by identifier. A Delete cancels the corresponding Add, and
+    /// lingering Delete entries are omitted from the result.
+    static Status MergeLiveEntries(const std::vector<ManifestEntry>& unmerged_entries,
+                                   std::vector<ManifestEntry>* live_entries);
+
  protected:
     /// @note Keep this thread-safe.
     virtual Result<bool> FilterByStats(const ManifestEntry& entry) const = 0;
@@ -273,13 +280,6 @@ class FileStoreScan {
                                          int32_t bucket,
                                          std::vector<ManifestEntry>* merged_entries) const;
 
-    /// Merge raw manifest entries into the set of currently-live files. Entries are deduplicated
-    /// by identifier (matching Add cancels a prior or following Delete), and lingering Delete
-    /// entries are dropped so the caller receives Add-only output, matching the semantics of
-    /// `ReadAndMergeFileEntries`.
-    static Status MergeLiveEntries(const std::vector<ManifestEntry>& unmerged_entries,
-                                   std::vector<ManifestEntry>* live_entries);
-
     Status ReadAndMergeFileEntries(const std::vector<ManifestFileMeta>& manifest_metas,
                                    std::vector<ManifestEntry>* merged_entries) const;