fix: NewSnapshotManager set auto_commit to false and restore true in SnapshotManager::Commit

Author: zhjwpku

src/iceberg/table.cc

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

src/iceberg/test/snapshot_manager_test.cc

@@ -479,6 +479,7 @@ TEST_F(SnapshotManagerTest, CreateReferencesAndRollback) {
     ICEBERG_UNWRAP_OR_FAIL(auto reloaded, catalog_->LoadTable(table_ident_));
     ICEBERG_UNWRAP_OR_FAIL(auto manager2, reloaded->NewSnapshotManager());
     manager2->RollbackTo(oldest_snapshot_id_);
+    EXPECT_THAT(manager2->Commit(), IsOk());
   }
 
   {
@@ -501,7 +502,6 @@ TEST_F(SnapshotManagerTest, SnapshotManagerThroughTransaction) {
   ICEBERG_UNWRAP_OR_FAIL(auto manager, SnapshotManager::Make(txn));
 
   manager->RollbackTo(oldest_snapshot_id_);
-  EXPECT_THAT(manager->Commit(), IsOk());
   EXPECT_THAT(txn->Commit(), IsOk());
 
   ICEBERG_UNWRAP_OR_FAIL(auto reloaded, catalog_->LoadTable(table_ident_));

src/iceberg/transaction.h

@@ -55,13 +55,22 @@ class ICEBERG_EXPORT Transaction : public std::enable_shared_from_this<Transacti
   /// \return the location of the metadata file
   std::string MetadataFileLocation(std::string_view filename) const;
 
+  /// \brief Enable auto-commit for this transaction.
+  void EnableAutoCommit() { auto_commit_ = true; }
+
+  /// \brief Disable auto-commit for this transaction.
+  void DisableAutoCommit() { auto_commit_ = false; }
+
   /// \brief Apply the pending changes from all actions and commit.
   ///
   /// \return Updated table if the transaction was committed successfully, or an error:
   /// - ValidationFailed: if any update cannot be applied to the current table metadata.
   /// - CommitFailed: if the updates cannot be committed due to conflicts.
   Result<std::shared_ptr<Table>> Commit();
 
+  /// \brief Returns true if this transaction has been committed.
+  bool is_committed() const { return committed_; }
+
   /// \brief Create a new UpdatePartitionSpec to update the partition spec of this table
   /// and commit the changes.
   Result<std::shared_ptr<UpdatePartitionSpec>> NewUpdatePartitionSpec();
@@ -140,7 +149,7 @@ class ICEBERG_EXPORT Transaction : public std::enable_shared_from_this<Transacti
   const Kind kind_;
   // Whether to auto-commit the transaction when updates are applied.
   // This is useful when a temporary transaction is created for a single operation.
-  const bool auto_commit_;
+  bool auto_commit_;
   // To make the state simple, we require updates are added and committed in order.
   bool last_update_committed_ = true;
   // Tracks if transaction has been committed to prevent double-commit

src/iceberg/update/snapshot_manager.cc

@@ -46,26 +46,30 @@ SnapshotManager::SnapshotManager(std::shared_ptr<Transaction> transaction)
 SnapshotManager::~SnapshotManager() = default;
 
 SnapshotManager& SnapshotManager::Cherrypick(int64_t snapshot_id) {
+  CommitIfRefUpdatesExist();
   // TODO(anyone): Implement cherrypick operation
   ICEBERG_BUILDER_CHECK(false, "Cherrypick operation not yet implemented");
   return *this;
 }
 
 SnapshotManager& SnapshotManager::SetCurrentSnapshot(int64_t snapshot_id) {
+  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(int64_t timestamp_ms) {
+  CommitIfRefUpdatesExist();
   ICEBERG_BUILDER_ASSIGN_OR_RETURN(auto set_snapshot, transaction_->NewSetSnapshot());
   set_snapshot->RollbackToTime(timestamp_ms);
   ICEBERG_BUILDER_RETURN_IF_ERROR(set_snapshot->Commit());
   return *this;
 }
 
 SnapshotManager& SnapshotManager::RollbackTo(int64_t snapshot_id) {
+  CommitIfRefUpdatesExist();
   ICEBERG_BUILDER_ASSIGN_OR_RETURN(auto set_snapshot, transaction_->NewSetSnapshot());
   set_snapshot->RollbackTo(snapshot_id);
   ICEBERG_BUILDER_RETURN_IF_ERROR(set_snapshot->Commit());
@@ -168,8 +172,13 @@ SnapshotManager& SnapshotManager::SetMaxRefAgeMs(const std::string& name,
 }
 
 Status SnapshotManager::Commit() {
+  transaction_->EnableAutoCommit();
   ICEBERG_RETURN_UNEXPECTED(CheckErrors());
-  return CommitIfRefUpdatesExist();
+  ICEBERG_RETURN_UNEXPECTED(CommitIfRefUpdatesExist());
+  if (!transaction_->is_committed()) {
+    ICEBERG_RETURN_UNEXPECTED(transaction_->Commit());
+  }
+  return {};
 }
 
 Result<std::shared_ptr<UpdateSnapshotReference>>