feat: add snapshot update

Author: zhjwpku

src/iceberg/CMakeLists.txt

@@ -80,6 +80,7 @@ set(ICEBERG_SOURCES
     transform_function.cc
     type.cc
     update/pending_update.cc
+    update/snapshot_update.cc
     update/update_partition_spec.cc
     update/update_properties.cc
     update/update_sort_order.cc

src/iceberg/table.h

@@ -50,7 +50,7 @@ class ICEBERG_EXPORT Table : public std::enable_shared_from_this<Table> {
 
   virtual ~Table();
 
-  /// \brief Return the identifier of this table
+  /// \brief Returns the identifier of this table
   const TableIdentifier& name() const { return identifier_; }
 
   /// \brief Returns the UUID of the table
@@ -59,40 +59,40 @@ class ICEBERG_EXPORT Table : public std::enable_shared_from_this<Table> {
   /// \brief Return the schema for this table, return NotFoundError if not found
   Result<std::shared_ptr<Schema>> schema() const;
 
-  /// \brief Return a map of schema for this table
+  /// \brief Returns a map of schema for this table
   Result<
       std::reference_wrapper<const std::unordered_map<int32_t, std::shared_ptr<Schema>>>>
   schemas() const;
 
-  /// \brief Return the partition spec for this table, return NotFoundError if not found
+  /// \brief Returns the partition spec for this table, return NotFoundError if not found
   Result<std::shared_ptr<PartitionSpec>> spec() const;
 
-  /// \brief Return a map of partition specs for this table
+  /// \brief Returns a map of partition specs for this table
   Result<std::reference_wrapper<
       const std::unordered_map<int32_t, std::shared_ptr<PartitionSpec>>>>
   specs() const;
 
-  /// \brief Return the sort order for this table, return NotFoundError if not found
+  /// \brief Returns the sort order for this table, return NotFoundError if not found
   Result<std::shared_ptr<SortOrder>> sort_order() const;
 
-  /// \brief Return a map of sort order IDs to sort orders for this table
+  /// \brief Returns a map of sort order IDs to sort orders for this table
   Result<std::reference_wrapper<
       const std::unordered_map<int32_t, std::shared_ptr<SortOrder>>>>
   sort_orders() const;
 
-  /// \brief Return a map of string properties for this table
+  /// \brief Returns the properties of this table
   const TableProperties& properties() const;
 
-  /// \brief Return the table's metadata file location
+  /// \brief Returns the table's metadata file location
   std::string_view metadata_file_location() const;
 
-  /// \brief Return the table's base location
+  /// \brief Returns the table's base location
   std::string_view location() const;
 
   /// \brief Returns the time when this table was last updated
   TimePointMs last_updated_ms() const;
 
-  /// \brief Return the table's current snapshot, return NotFoundError if not found
+  /// \brief Returns the table's current snapshot, return NotFoundError if not found
   Result<std::shared_ptr<Snapshot>> current_snapshot() const;
 
   /// \brief Get the snapshot of this table with the given id

src/iceberg/table_metadata.cc

@@ -591,6 +591,8 @@ class TableMetadataBuilder::Impl {
   Status RemoveSchemas(const std::unordered_set<int32_t>& schema_ids);
   Result<int32_t> AddSchema(const Schema& schema, int32_t new_last_column_id);
   void SetLocation(std::string_view location);
+  Status AddSnapshot(const Snapshot& snapshot);
+  Status SetBranchSnapshot(int64_t snapshot_id, const std::string& branch);
 
   Result<std::unique_ptr<TableMetadata>> Build();
 
@@ -1207,12 +1209,14 @@ TableMetadataBuilder& TableMetadataBuilder::AddSortOrder(
 
 TableMetadataBuilder& TableMetadataBuilder::AddSnapshot(
     std::shared_ptr<Snapshot> snapshot) {
-  throw IcebergError(std::format("{} not implemented", __FUNCTION__));
+  ICEBERG_BUILDER_RETURN_IF_ERROR(impl_->AddSnapshot(*snapshot));
+  return *this;
 }
 
 TableMetadataBuilder& TableMetadataBuilder::SetBranchSnapshot(int64_t snapshot_id,
                                                               const std::string& branch) {
-  throw IcebergError(std::format("{} not implemented", __FUNCTION__));
+  ICEBERG_BUILDER_RETURN_IF_ERROR(impl_->SetBranchSnapshot(snapshot_id, branch));
+  return *this;
 }
 
 TableMetadataBuilder& TableMetadataBuilder::SetRef(const std::string& name,

src/iceberg/table_properties.h

@@ -244,6 +244,9 @@ class ICEBERG_EXPORT TableProperties : public ConfigBase<TableProperties> {
   inline static Entry<int64_t> kDeleteTargetFileSizeBytes{
       "write.delete.target-file-size-bytes", int64_t{64} * 1024 * 1024};  // 64 MB
 
+  inline static Entry<bool> kSnapshotIdInheritanceEnabled{
+      "compatibility.snapshot-id-inheritance.enabled", false};
+
   // Garbage collection properties
 
   inline static Entry<bool> kGcEnabled{"gc.enabled", true};

src/iceberg/table_update.cc

@@ -134,7 +134,7 @@ void SetDefaultSortOrder::GenerateRequirements(TableUpdateContext& context) cons
 // AddSnapshot
 
 void AddSnapshot::ApplyTo(TableMetadataBuilder& builder) const {
-  throw IcebergError(std::format("{} not implemented", __FUNCTION__));
+  builder.AddSnapshot(snapshot_);
 }
 
 void AddSnapshot::GenerateRequirements(TableUpdateContext& context) const {

src/iceberg/transaction.cc

@@ -105,6 +105,11 @@ Status Transaction::Apply(PendingUpdate& update) {
         metadata_builder_->AddPartitionSpec(std::move(result.spec));
       }
     } break;
+    case PendingUpdate::Kind::kUpdateSnapshot: {
+      auto& update_snapshot = internal::checked_cast<SnapshotUpdate&>(update);
+      ICEBERG_ASSIGN_OR_RAISE(auto result, update_snapshot.Apply());
+      metadata_builder_->AddSnapshot(std::move(result.snapshot));
+    } break;
     default:
       return NotSupported("Unsupported pending update: {}",
                           static_cast<int32_t>(update.kind()));

src/iceberg/transaction.h

@@ -77,7 +77,9 @@ class ICEBERG_EXPORT Transaction : public std::enable_shared_from_this<Transacti
   /// \brief Apply the pending changes to current table.
   Status Apply(PendingUpdate& updates);
 
-  friend class PendingUpdate;  // Need to access the Apply method.
+  /// \brief Friends to access the Apply method.
+  friend class PendingUpdate;
+  friend class SnapshotUpdate;
 
  private:
   // The table that this transaction will update.

src/iceberg/type_fwd.h

@@ -183,6 +183,7 @@ class PendingUpdate;
 class UpdatePartitionSpec;
 class UpdateProperties;
 class UpdateSortOrder;
+class SnapshotUpdate;
 
 /// ----------------------------------------------------------------------------
 /// TODO: Forward declarations below are not added yet.

src/iceberg/update/meson.build

@@ -18,6 +18,7 @@
 install_headers(
     [
         'pending_update.h',
+        'snapshot_update.h',
         'update_partition_spec.h',
         'update_sort_order.h',
         'update_properties.h',

src/iceberg/update/pending_update.h

@@ -45,6 +45,7 @@ class ICEBERG_EXPORT PendingUpdate : public ErrorCollector {
     kUpdatePartitionSpec,
     kUpdateProperties,
     kUpdateSortOrder,
+    kUpdateSnapshot,
   };
 
   /// \brief Return the kind of this pending update.