feat: add SnapshotManager

Author: zhjwpku

src/iceberg/CMakeLists.txt

@@ -71,6 +71,7 @@ set(ICEBERG_SOURCES
     schema_util.cc
     snapshot.cc
     sort_field.cc
+    update/snapshot_manager.cc
     sort_order.cc
     statistics_file.cc
     table.cc

src/iceberg/table.cc

@@ -32,6 +32,7 @@
 #include "iceberg/table_scan.h"
 #include "iceberg/transaction.h"
 #include "iceberg/update/expire_snapshots.h"
+#include "iceberg/update/snapshot_manager.h"
 #include "iceberg/update/update_partition_spec.h"
 #include "iceberg/update/update_partition_statistics.h"
 #include "iceberg/update/update_properties.h"
@@ -222,6 +223,10 @@ Result<std::shared_ptr<UpdatePartitionStatistics>> Table::NewUpdatePartitionStat
   return transaction->NewUpdatePartitionStatistics();
 }
 
+Result<std::shared_ptr<SnapshotManager>> Table::NewSnapshotManager() {
+  return SnapshotManager::Make(name().ToString(), shared_from_this());
+}
+
 Result<std::shared_ptr<StagedTable>> StagedTable::Make(
     TableIdentifier identifier, std::shared_ptr<TableMetadata> metadata,
     std::string metadata_location, std::shared_ptr<FileIO> io,

src/iceberg/table.h

@@ -168,6 +168,9 @@ class ICEBERG_EXPORT Table : public std::enable_shared_from_this<Table> {
   /// \brief Create a new FastAppend to append data files and commit the changes.
   virtual Result<std::shared_ptr<FastAppend>> NewFastAppend();
 
+  /// \brief Create a new SnapshotManager to manage snapshots and snapshot references.
+  virtual Result<std::shared_ptr<SnapshotManager>> NewSnapshotManager();
+
  protected:
   Table(TableIdentifier identifier, std::shared_ptr<TableMetadata> metadata,
         std::string metadata_location, std::shared_ptr<FileIO> io,

src/iceberg/test/CMakeLists.txt

@@ -179,6 +179,7 @@ if(ICEBERG_BUILD_BUNDLE)
                    expire_snapshots_test.cc
                    fast_append_test.cc
                    set_snapshot_test.cc
+                   snapshot_manager_test.cc
                    transaction_test.cc
                    update_location_test.cc
                    update_partition_spec_test.cc

src/iceberg/test/snapshot_manager_test.cc

@@ -19,6 +19,7 @@
 
 #pragma once
 
+#include <chrono>
 #include <format>
 #include <memory>
 #include <string>
@@ -28,11 +29,16 @@
 
 #include "iceberg/arrow/arrow_fs_file_io_internal.h"
 #include "iceberg/catalog/memory/in_memory_catalog.h"
+#include "iceberg/constants.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/schema.h"
+#include "iceberg/sort_order.h"
 #include "iceberg/table.h"
 #include "iceberg/table_identifier.h"
 #include "iceberg/table_metadata.h"
 #include "iceberg/test/matchers.h"
 #include "iceberg/test/test_resource.h"
+#include "iceberg/type.h"
 #include "iceberg/util/uuid.h"
 
 namespace iceberg {
@@ -43,6 +49,7 @@ class UpdateTestBase : public ::testing::Test {
   void SetUp() override {
     InitializeFileIO();
     RegisterTableFromResource("TableMetadataV2Valid.json");
+    RegisterEmptyTable();
   }
 
   /// \brief Initialize file IO and create necessary directories.
@@ -56,6 +63,7 @@ class UpdateTestBase : public ::testing::Test {
         static_cast<arrow::ArrowFileSystemFileIO&>(*file_io_).fs());
     ASSERT_TRUE(arrow_fs != nullptr);
     ASSERT_TRUE(arrow_fs->CreateDir(table_location_ + "/metadata").ok());
+    ASSERT_TRUE(arrow_fs->CreateDir(empty_table_location_ + "/metadata").ok());
   }
 
   /// \brief Register a table from a metadata resource file.
@@ -78,11 +86,60 @@ class UpdateTestBase : public ::testing::Test {
                            catalog_->RegisterTable(table_ident_, metadata_location));
   }
 
+  /// \brief Register an empty table with no snapshots.
+  ///
+  /// Creates a minimal table metadata with no snapshots (current_snapshot_id = -1).
+  /// This is useful for testing operations on empty tables.
+  void RegisterEmptyTable() {
+    // Drop existing table if it exists
+    std::ignore = catalog_->DropTable(empty_table_ident_, /*purge=*/false);
+
+    // Create minimal schema with a single field
+    auto field = SchemaField::MakeRequired(1, "id", int64());
+    auto schema = std::make_shared<Schema>(std::vector<SchemaField>{field}, 0);
+
+    // Create empty table metadata
+    auto metadata = std::make_unique<TableMetadata>();
+    metadata->format_version = 2;
+    metadata->table_uuid = Uuid::GenerateV7().ToString();
+    metadata->location = empty_table_location_;
+    metadata->last_sequence_number = 0;
+    metadata->last_updated_ms = TimePointMs{std::chrono::milliseconds(1000)};
+    metadata->last_column_id = 1;
+    metadata->current_schema_id = 0;
+    metadata->schemas.push_back(schema);
+    metadata->partition_specs.push_back(PartitionSpec::Unpartitioned());
+    metadata->default_spec_id = PartitionSpec::kInitialSpecId;
+    metadata->last_partition_id = 0;
+    metadata->current_snapshot_id = kInvalidSnapshotId;
+    metadata->sort_orders.push_back(SortOrder::Unsorted());
+    metadata->default_sort_order_id = SortOrder::kUnsortedOrderId;
+    metadata->next_row_id = TableMetadata::kInitialRowId;
+
+    // Write table metadata to the table location.
+    auto metadata_location =
+        std::format("{}/metadata/00001-{}.metadata.json", empty_table_location_,
+                    Uuid::GenerateV7().ToString());
+    ASSERT_THAT(TableMetadataUtil::Write(*file_io_, metadata_location, *metadata),
+                IsOk());
+
+    // Register the table in the catalog.
+    ICEBERG_UNWRAP_OR_FAIL(
+        auto registered_table,
+        catalog_->RegisterTable(empty_table_ident_, metadata_location));
+
+    // Reload the table to ensure it's in a clean state without any transaction state.
+    ICEBERG_UNWRAP_OR_FAIL(empty_table_, catalog_->LoadTable(empty_table_ident_));
+  }
+
   const TableIdentifier table_ident_{.name = "test_table"};
   const std::string table_location_{"/warehouse/test_table"};
+  const TableIdentifier empty_table_ident_{.name = "empty_table"};
+  const std::string empty_table_location_{"/warehouse/empty_table"};
   std::shared_ptr<FileIO> file_io_;
   std::shared_ptr<InMemoryCatalog> catalog_;
   std::shared_ptr<Table> table_;
+  std::shared_ptr<Table> empty_table_;
 };
 
 }  // namespace iceberg

src/iceberg/test/update_test_base.h

@@ -36,6 +36,7 @@
 #include "iceberg/update/fast_append.h"
 #include "iceberg/update/pending_update.h"
 #include "iceberg/update/set_snapshot.h"
+#include "iceberg/update/snapshot_manager.h"
 #include "iceberg/update/snapshot_update.h"
 #include "iceberg/update/update_location.h"
 #include "iceberg/update/update_partition_spec.h"
@@ -428,4 +429,11 @@ Transaction::NewUpdateSnapshotReference() {
   return update_ref;
 }
 
+Result<std::shared_ptr<SnapshotManager>> Transaction::NewSnapshotManager() {
+  ICEBERG_ASSIGN_OR_RAISE(std::shared_ptr<SnapshotManager> snapshot_manager,
+                          SnapshotManager::Make(shared_from_this()));
+  ICEBERG_RETURN_UNEXPECTED(AddUpdate(snapshot_manager));
+  return snapshot_manager;
+}
+
 }  // namespace iceberg

src/iceberg/transaction.cc

@@ -105,6 +105,9 @@ class ICEBERG_EXPORT Transaction : public std::enable_shared_from_this<Transacti
   /// and tags) and commit the changes.
   Result<std::shared_ptr<UpdateSnapshotReference>> NewUpdateSnapshotReference();
 
+  /// \brief Create a new SnapshotManager to manage snapshots.
+  Result<std::shared_ptr<SnapshotManager>> NewSnapshotManager();
+
  private:
   Transaction(std::shared_ptr<Table> table, Kind kind, bool auto_commit,
               std::unique_ptr<TableMetadataBuilder> metadata_builder);

src/iceberg/transaction.h

@@ -184,6 +184,7 @@ class TableProperties;
 /// \brief Table update.
 class TableMetadataBuilder;
 class TableUpdate;
+class SnapshotManager;
 class TableRequirement;
 class TableUpdateContext;
 class Transaction;

src/iceberg/type_fwd.h

@@ -87,6 +87,11 @@ FastAppend& FastAppend::AppendManifest(const ManifestFile& manifest) {
   return *this;
 }
 
+FastAppend& FastAppend::ToBranch(const std::string& branch) {
+  ICEBERG_BUILDER_RETURN_IF_ERROR(SetTargetBranch(branch));
+  return *this;
+}
+
 std::string FastAppend::operation() { return DataOperation::kAppend; }
 
 Result<std::vector<ManifestFile>> FastAppend::Apply(