fix: review comments

Author: zhjwpku

src/iceberg/table.cc

@@ -224,7 +224,10 @@ Result<std::shared_ptr<UpdatePartitionStatistics>> Table::NewUpdatePartitionStat
 }
 
 Result<std::shared_ptr<SnapshotManager>> Table::NewSnapshotManager() {
-  return SnapshotManager::Make(name().ToString(), shared_from_this());
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto transaction, Transaction::Make(shared_from_this(), Transaction::Kind::kUpdate,
+                                          /*auto_commit=*/true));
+  return SnapshotManager::Make(std::move(transaction));
 }
 
 Result<std::shared_ptr<StagedTable>> StagedTable::Make(

src/iceberg/test/snapshot_manager_test.cc

@@ -63,9 +63,7 @@ Transaction::~Transaction() = default;
 
 Result<std::shared_ptr<Transaction>> Transaction::Make(std::shared_ptr<Table> table,
                                                        Kind kind, bool auto_commit) {
-  if (!table || !table->catalog()) [[unlikely]] {
-    return InvalidArgument("Table and catalog cannot be null");
-  }
+  ICEBERG_PRECHECK(table && table->catalog(), "Table and catalog cannot be null");
 
   std::unique_ptr<TableMetadataBuilder> metadata_builder;
   if (kind == Kind::kCreate) {
@@ -94,9 +92,11 @@ std::string Transaction::MetadataFileLocation(std::string_view filename) const {
 }
 
 Status Transaction::AddUpdate(const std::shared_ptr<PendingUpdate>& update) {
-  if (!last_update_committed_) {
-    return InvalidArgument("Cannot add update when previous update is not committed");
-  }
+  ICEBERG_PRECHECK(update->kind() != PendingUpdate::Kind::kSnapshotManager,
+                   "SnapshotManager updates should not be added to the transaction");
+  ICEBERG_CHECK(last_update_committed_,
+                "Cannot add update when previous update is not committed");
+
   pending_updates_.emplace_back(std::weak_ptr<PendingUpdate>(update));
   last_update_committed_ = false;
   return {};
@@ -302,13 +302,9 @@ Status Transaction::ApplyUpdatePartitionStatistics(UpdatePartitionStatistics& up
 }
 
 Result<std::shared_ptr<Table>> Transaction::Commit() {
-  if (committed_) {
-    return Invalid("Transaction already committed");
-  }
-  if (!last_update_committed_) {
-    return InvalidArgument(
-        "Cannot commit transaction when previous update is not committed");
-  }
+  ICEBERG_CHECK(!committed_, "Transaction already committed");
+  ICEBERG_CHECK(last_update_committed_,
+                "Cannot commit transaction when previous update is not committed");
 
   const auto& updates = metadata_builder_->changes();
   if (updates.empty()) {
@@ -432,7 +428,8 @@ Transaction::NewUpdateSnapshotReference() {
 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));
+  // SnapshotManager has its own commit logic, so it is not added to the pending updates.
+  // This differs from the Java implementation.
   return snapshot_manager;
 }
 

src/iceberg/transaction.cc

@@ -184,7 +184,6 @@ class TableProperties;
 /// \brief Table update.
 class TableMetadataBuilder;
 class TableUpdate;
-class SnapshotManager;
 class TableRequirement;
 class TableUpdateContext;
 class Transaction;
@@ -194,6 +193,7 @@ class ExpireSnapshots;
 class FastAppend;
 class PendingUpdate;
 class SetSnapshot;
+class SnapshotManager;
 class SnapshotUpdate;
 class UpdateLocation;
 class UpdatePartitionSpec;

src/iceberg/type_fwd.h

@@ -44,6 +44,7 @@ class ICEBERG_EXPORT PendingUpdate : public ErrorCollector {
   enum class Kind : uint8_t {
     kExpireSnapshots,
     kSetSnapshot,
+    kSnapshotManager,
     kUpdateLocation,
     kUpdatePartitionSpec,
     kUpdatePartitionStatistics,

src/iceberg/update/pending_update.h

@@ -34,64 +34,38 @@
 
 namespace iceberg {
 
-Result<std::shared_ptr<SnapshotManager>> SnapshotManager::Make(
-    const std::string& table_name, std::shared_ptr<Table> table) {
-  if (table == nullptr) {
-    return InvalidArgument("Table cannot be null");
-  }
-  if (table->metadata() == nullptr) {
-    return InvalidArgument("Cannot manage snapshots: table {} does not exist",
-                           table_name);
-  }
-  // Create a transaction first
-  ICEBERG_ASSIGN_OR_RAISE(auto transaction,
-                          Transaction::Make(table, Transaction::Kind::kUpdate,
-                                            /*auto_commit=*/false));
-  auto manager = std::shared_ptr<SnapshotManager>(
-      new SnapshotManager(std::move(transaction), /*is_external=*/false));
-  return manager;
-}
-
 Result<std::shared_ptr<SnapshotManager>> SnapshotManager::Make(
     std::shared_ptr<Transaction> transaction) {
-  if (transaction == nullptr) {
-    return InvalidArgument("Invalid input transaction: null");
-  }
-  return std::shared_ptr<SnapshotManager>(
-      new SnapshotManager(std::move(transaction), /*is_external=*/true));
+  ICEBERG_PRECHECK(transaction != nullptr, "Invalid input transaction: null");
+  return std::shared_ptr<SnapshotManager>(new SnapshotManager(std::move(transaction)));
 }
 
-SnapshotManager::SnapshotManager(std::shared_ptr<Transaction> transaction,
-                                 bool is_external)
-    : PendingUpdate(transaction), is_external_transaction_(is_external) {}
+SnapshotManager::SnapshotManager(std::shared_ptr<Transaction> transaction)
+    : PendingUpdate(transaction) {}
 
 SnapshotManager::~SnapshotManager() = default;
 
 SnapshotManager& SnapshotManager::Cherrypick(int64_t snapshot_id) {
-  ICEBERG_BUILDER_RETURN_IF_ERROR(CommitIfRefUpdatesExist());
   // TODO(anyone): Implement cherrypick operation
   ICEBERG_BUILDER_CHECK(false, "Cherrypick operation not yet implemented");
   return *this;
 }
 
 SnapshotManager& SnapshotManager::SetCurrentSnapshot(int64_t snapshot_id) {
-  ICEBERG_BUILDER_RETURN_IF_ERROR(CommitIfRefUpdatesExist());
   ICEBERG_BUILDER_ASSIGN_OR_RETURN(auto set_snapshot, transaction_->NewSetSnapshot());
   set_snapshot->SetCurrentSnapshot(snapshot_id);
   ICEBERG_BUILDER_RETURN_IF_ERROR(set_snapshot->Commit());
   return *this;
 }
 
 SnapshotManager& SnapshotManager::RollbackToTime(TimePointMs timestamp_ms) {
-  ICEBERG_BUILDER_RETURN_IF_ERROR(CommitIfRefUpdatesExist());
   ICEBERG_BUILDER_ASSIGN_OR_RETURN(auto set_snapshot, transaction_->NewSetSnapshot());
   set_snapshot->RollbackToTime(UnixMsFromTimePointMs(timestamp_ms));
   ICEBERG_BUILDER_RETURN_IF_ERROR(set_snapshot->Commit());
   return *this;
 }
 
 SnapshotManager& SnapshotManager::RollbackTo(int64_t snapshot_id) {
-  ICEBERG_BUILDER_RETURN_IF_ERROR(CommitIfRefUpdatesExist());
   ICEBERG_BUILDER_ASSIGN_OR_RETURN(auto set_snapshot, transaction_->NewSetSnapshot());
   set_snapshot->RollbackTo(snapshot_id);
   ICEBERG_BUILDER_RETURN_IF_ERROR(set_snapshot->Commit());
@@ -101,9 +75,8 @@ SnapshotManager& SnapshotManager::RollbackTo(int64_t snapshot_id) {
 SnapshotManager& SnapshotManager::CreateBranch(const std::string& name) {
   if (base().current_snapshot_id != kInvalidSnapshotId) {
     ICEBERG_BUILDER_ASSIGN_OR_RETURN(auto current_snapshot, base().Snapshot());
-    if (current_snapshot != nullptr) {
-      return CreateBranch(name, current_snapshot->snapshot_id);
-    }
+    ICEBERG_DCHECK(current_snapshot != nullptr, "Current snapshot should not be null");
+    return CreateBranch(name, current_snapshot->snapshot_id);
   }
   const auto& current_refs = base().refs;
   ICEBERG_BUILDER_CHECK(!base().refs.contains(name), "Ref {} already exists", name);
@@ -197,11 +170,8 @@ SnapshotManager& SnapshotManager::SetMaxRefAgeMs(const std::string& name,
 Result<std::shared_ptr<Snapshot>> SnapshotManager::Apply() { return base().Snapshot(); }
 
 Status SnapshotManager::Commit() {
-  ICEBERG_RETURN_UNEXPECTED(CommitIfRefUpdatesExist());
-  if (!is_external_transaction_) {
-    ICEBERG_RETURN_UNEXPECTED(transaction_->Commit());
-  }
-  return {};
+  ICEBERG_RETURN_UNEXPECTED(CheckErrors());
+  return CommitIfRefUpdatesExist();
 }
 
 Result<std::shared_ptr<UpdateSnapshotReference>>

src/iceberg/update/snapshot_manager.cc

@@ -37,14 +37,6 @@ namespace iceberg {
 /// snapshots, and managing branches and tags.
 class ICEBERG_EXPORT SnapshotManager : public PendingUpdate {
  public:
-  /// \brief Create a SnapshotManager for a table.
-  ///
-  /// \param table_name The name of the table
-  /// \param table The table to manage snapshots for
-  /// \return A new SnapshotManager instance, or an error if the table doesn't exist
-  static Result<std::shared_ptr<SnapshotManager>> Make(const std::string& table_name,
-                                                       std::shared_ptr<Table> table);
-
   /// \brief Create a SnapshotManager from an existing transaction.
   ///
   /// \param transaction The transaction to use
@@ -194,17 +186,14 @@ class ICEBERG_EXPORT SnapshotManager : public PendingUpdate {
   /// \brief Constructor for creating a SnapshotManager with a transaction.
   ///
   /// \param transaction The transaction to use
-  /// \param is_external Whether this is an external transaction (true) or created
-  /// internally (false)
-  SnapshotManager(std::shared_ptr<Transaction> transaction, bool is_external);
+  explicit SnapshotManager(std::shared_ptr<Transaction> transaction);
 
   /// \brief Get or create the UpdateSnapshotReference operation.
   Result<std::shared_ptr<UpdateSnapshotReference>> UpdateSnapshotReferencesOperation();
 
   /// \brief Commit any pending reference updates if they exist.
   Status CommitIfRefUpdatesExist();
 
-  bool is_external_transaction_;
   std::shared_ptr<UpdateSnapshotReference> update_snapshot_references_operation_;
 };