fix(logging): address commit lifecycle review feedback

Author: kamcheungting-db

src/iceberg/table_metadata.cc

@@ -38,7 +38,6 @@
 #include "iceberg/exception.h"
 #include "iceberg/file_io.h"
 #include "iceberg/json_serde_internal.h"
-#include "iceberg/logging/log_macros.h"
 #include "iceberg/metrics_config.h"
 #include "iceberg/partition_field.h"
 #include "iceberg/partition_spec.h"
@@ -1107,8 +1106,6 @@ Status TableMetadataBuilder::Impl::AddSnapshot(std::shared_ptr<Snapshot> snapsho
     metadata_.next_row_id += add_rows.value();
   }
 
-  ICEBERG_LOG_DEBUG("Added snapshot {} (sequence number {}) to table metadata",
-                    snapshot->snapshot_id, snapshot->sequence_number);
   return {};
 }
 

src/iceberg/test/fast_append_test.cc

@@ -34,6 +34,7 @@
 
 #include "iceberg/avro/avro_register.h"
 #include "iceberg/constants.h"
+#include "iceberg/logging/log_level.h"
 #include "iceberg/manifest/manifest_entry.h"
 #include "iceberg/manifest/manifest_reader.h"
 #include "iceberg/manifest/manifest_writer.h"
@@ -46,6 +47,7 @@
 #include "iceberg/table_metadata.h"
 #include "iceberg/table_properties.h"
 #include "iceberg/test/executor.h"
+#include "iceberg/test/logging_test_helpers.h"
 #include "iceberg/test/matchers.h"
 #include "iceberg/test/mock_catalog.h"
 #include "iceberg/test/update_test_base.h"
@@ -178,6 +180,31 @@ TEST_F(FastAppendTest, AppendDataFile) {
   EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kManifestsReplaced), "0");
 }
 
+TEST_F(FastAppendTest, StageOnlyCommitLogNamesAddedSnapshot) {
+  auto capturing = std::make_shared<CapturingLogger>();
+  capturing->SetLevel(LogLevel::kTrace);
+  ScopedDefaultLogger guard(capturing);
+
+  ICEBERG_UNWRAP_OR_FAIL(auto fast_append, table_->NewFastAppend());
+  fast_append->StageOnly();
+  fast_append->AppendFile(file_a_);
+  ASSERT_THAT(fast_append->Commit(), IsOk());
+  ASSERT_THAT(table_->Refresh(), IsOk());
+
+  ASSERT_FALSE(table_->metadata()->snapshots.empty());
+  const auto snapshot_id = table_->metadata()->snapshots.back()->snapshot_id;
+  bool found = false;
+  for (const auto& record : capturing->records()) {
+    if (record.level == LogLevel::kInfo &&
+        record.message.find(std::format("committed snapshot {}", snapshot_id)) !=
+            std::string::npos) {
+      found = true;
+      break;
+    }
+  }
+  EXPECT_TRUE(found) << "expected the staged snapshot in the commit success log";
+}
+
 TEST_F(FastAppendTest, AppendMultipleDataFiles) {
   std::shared_ptr<FastAppend> fast_append;
   ICEBERG_UNWRAP_OR_FAIL(fast_append, table_->NewFastAppend());

src/iceberg/test/table_metadata_builder_test.cc

@@ -24,7 +24,6 @@
 #include <gmock/gmock.h>
 #include <gtest/gtest.h>
 
-#include "iceberg/logging/log_level.h"
 #include "iceberg/partition_spec.h"
 #include "iceberg/result.h"
 #include "iceberg/schema.h"
@@ -34,7 +33,6 @@
 #include "iceberg/table_metadata.h"
 #include "iceberg/table_properties.h"
 #include "iceberg/table_update.h"
-#include "iceberg/test/logging_test_helpers.h"
 #include "iceberg/test/matchers.h"
 #include "iceberg/transform.h"
 #include "iceberg/type.h"
@@ -1187,29 +1185,6 @@ TEST(TableMetadataBuilderTest, RemoveSchemasAfterSchemaChange) {
   ASSERT_THAT(builder->Build(), HasErrorMessage("Cannot remove current schema: 1"));
 }
 
-// Adding a snapshot to the builder emits a DEBUG record naming the snapshot.
-TEST(TableMetadataBuilderTest, AddSnapshotEmitsDebugLog) {
-  auto capturing = std::make_shared<CapturingLogger>();
-  capturing->SetLevel(LogLevel::kTrace);
-  ScopedDefaultLogger guard(capturing);
-
-  auto base = CreateBaseMetadata();
-  auto builder = TableMetadataBuilder::BuildFrom(base.get());
-  builder->AddSnapshot(
-      std::make_shared<Snapshot>(Snapshot{.snapshot_id = 42, .sequence_number = 7}));
-  ICEBERG_UNWRAP_OR_FAIL(auto metadata, builder->Build());
-
-  bool found = false;
-  for (const auto& record : capturing->records()) {
-    if (record.level == LogLevel::kDebug &&
-        record.message.find("Added snapshot 42") != std::string::npos) {
-      found = true;
-      break;
-    }
-  }
-  EXPECT_TRUE(found) << "expected a DEBUG record naming the added snapshot";
-}
-
 TEST(TableMetadataBuilderTest, RemoveSnapshotRef) {
   auto base = CreateBaseMetadata();
   auto builder = TableMetadataBuilder::BuildFrom(base.get());

src/iceberg/test/transaction_test.cc

@@ -22,7 +22,9 @@
 #include "iceberg/expression/expressions.h"
 #include "iceberg/expression/term.h"
 #include "iceberg/logging/log_level.h"
+#include "iceberg/snapshot.h"
 #include "iceberg/sort_order.h"
+#include "iceberg/table_metadata.h"
 #include "iceberg/test/logging_test_helpers.h"
 #include "iceberg/test/matchers.h"
 #include "iceberg/test/mock_catalog.h"
@@ -227,9 +229,68 @@ TEST_F(TransactionRetryTest, CommitRetryEmitsRetryAndSuccessLogs) {
       << "expected a success INFO";
 }
 
-// A commit that exhausts its retries emits an ERROR with the attempt count and the
-// final error.
-TEST_F(TransactionRetryTest, CommitRetryExhaustedEmitsErrorLog) {
+// A metadata-only retry must not attribute a snapshot committed concurrently by
+// another writer to this transaction.
+TEST_F(TransactionRetryTest, MetadataOnlyRetryDoesNotLogConcurrentSnapshot) {
+  auto capturing = std::make_shared<CapturingLogger>();
+  capturing->SetLevel(LogLevel::kTrace);
+  ScopedDefaultLogger guard(capturing);
+
+  constexpr int64_t kConcurrentSnapshotId = 987654321;
+  auto metadata_builder = TableMetadataBuilder::BuildFrom(mock_table_->metadata().get());
+  auto concurrent_snapshot = std::make_shared<Snapshot>(Snapshot{
+      .snapshot_id = kConcurrentSnapshotId,
+      .parent_snapshot_id = mock_table_->metadata()->current_snapshot_id,
+      .sequence_number = mock_table_->metadata()->last_sequence_number + 1,
+      .timestamp_ms = TimePointMs{},
+      .manifest_list = "concurrent-manifest-list.avro",
+      .summary = {{SnapshotSummaryFields::kOperation, "append"}},
+  });
+  metadata_builder->SetBranchSnapshot(concurrent_snapshot,
+                                      std::string(SnapshotRef::kMainBranch));
+  ICEBERG_UNWRAP_OR_FAIL(auto concurrent_metadata, metadata_builder->Build());
+  auto concurrent_metadata_ptr =
+      std::shared_ptr<TableMetadata>(std::move(concurrent_metadata));
+  const std::string concurrent_metadata_location = "concurrent.metadata.json";
+
+  ON_CALL(*mock_catalog_, LoadTable(::testing::_))
+      .WillByDefault([this, concurrent_metadata_ptr, &concurrent_metadata_location](
+                         const TableIdentifier&) -> Result<std::shared_ptr<Table>> {
+        return Table::Make(mock_table_->name(), concurrent_metadata_ptr,
+                           concurrent_metadata_location, mock_table_->io(),
+                           mock_catalog_);
+      });
+
+  int update_call_count = 0;
+  ON_CALL(*mock_catalog_, UpdateTable(::testing::_, ::testing::_, ::testing::_))
+      .WillByDefault(
+          [this, concurrent_metadata_ptr, &concurrent_metadata_location,
+           &update_call_count](const TableIdentifier&,
+                               const std::vector<std::unique_ptr<TableRequirement>>&,
+                               const std::vector<std::unique_ptr<TableUpdate>>&)
+              -> Result<std::shared_ptr<Table>> {
+            if (++update_call_count == 1) {
+              return CommitFailed("conflict on first attempt");
+            }
+            return Table::Make(mock_table_->name(), concurrent_metadata_ptr,
+                               concurrent_metadata_location, mock_table_->io(),
+                               mock_catalog_);
+          });
+
+  ICEBERG_UNWRAP_OR_FAIL(auto txn, mock_table_->NewTransaction());
+  ICEBERG_UNWRAP_OR_FAIL(auto update, txn->NewUpdateProperties());
+  update->Set("retry.test", "value");
+  ASSERT_THAT(update->Commit(), IsOk());
+  ASSERT_THAT(txn->Commit(), IsOk());
+
+  EXPECT_FALSE(HasRecord(capturing->records(), LogLevel::kInfo,
+                         std::to_string(kConcurrentSnapshotId)))
+      << "metadata-only commit attributed the concurrent snapshot to itself";
+}
+
+// A commit that exhausts its retries returns the final error without emitting a
+// generic ERROR log. Genuine retry attempts still emit WARN records.
+TEST_F(TransactionRetryTest, CommitRetryExhaustedDoesNotEmitErrorLog) {
   auto capturing = std::make_shared<CapturingLogger>();
   capturing->SetLevel(LogLevel::kTrace);
   ScopedDefaultLogger guard(capturing);
@@ -249,10 +310,8 @@ TEST_F(TransactionRetryTest, CommitRetryExhaustedEmitsErrorLog) {
   EXPECT_THAT(txn->Commit(), IsError(ErrorKind::kCommitFailed));
 
   auto records = capturing->records();
-  EXPECT_TRUE(HasRecord(records, LogLevel::kError, "failed after 5 attempt(s)"))
-      << "expected a final ERROR with the attempt count";
-  EXPECT_TRUE(HasRecord(records, LogLevel::kError, "always conflicts"))
-      << "final ERROR should carry the last error";
+  EXPECT_FALSE(HasRecord(records, LogLevel::kError, ""))
+      << "the final commit error should be propagated without a generic ERROR log";
   // Retries 2..5 each log a WARN.
   EXPECT_TRUE(
       HasRecord(records, LogLevel::kWarn, "Retrying transaction commit (attempt 5)"));
@@ -288,7 +347,11 @@ TEST_F(TransactionRetryTest, CommitSuccessEmitsInfoLog) {
   EXPECT_FALSE(HasRecord(records, LogLevel::kWarn, "Retrying transaction commit"));
 }
 
-TEST_F(TransactionRetryTest, CommitNonRetryableErrorStopsImmediately) {
+TEST_F(TransactionRetryTest, CommitStateUnknownStopsImmediatelyWithoutErrorLog) {
+  auto capturing = std::make_shared<CapturingLogger>();
+  capturing->SetLevel(LogLevel::kTrace);
+  ScopedDefaultLogger guard(capturing);
+
   int update_call_count = 0;
   ON_CALL(*mock_catalog_, UpdateTable(::testing::_, ::testing::_, ::testing::_))
       .WillByDefault(
@@ -308,6 +371,8 @@ TEST_F(TransactionRetryTest, CommitNonRetryableErrorStopsImmediately) {
   auto result = txn->Commit();
   EXPECT_THAT(result, IsError(ErrorKind::kCommitStateUnknown));
   EXPECT_EQ(update_call_count, 1);  // Should not retry
+  EXPECT_FALSE(HasRecord(capturing->records(), LogLevel::kError, ""))
+      << "an unknown commit state must not be logged as a confirmed failure";
 }
 
 TEST_F(TransactionRetryTest, CreateTransactionDoesNotRetry) {

src/iceberg/transaction.cc

@@ -19,6 +19,7 @@
 #include "iceberg/transaction.h"
 
 #include <format>
+#include <iterator>
 #include <memory>
 #include <string>
 
@@ -377,9 +378,6 @@ Result<std::shared_ptr<Table>> Transaction::Commit() {
   int32_t max_wait_ms = props.Get(TableProperties::kCommitMaxRetryWaitMs);
   int32_t total_timeout_ms = props.Get(TableProperties::kCommitTotalRetryTimeMs);
 
-  // Snapshot id before the commit, to detect whether this commit advanced it (a
-  // data commit) versus a metadata-only commit that adds no snapshot.
-  const int64_t base_current_snapshot_id = ctx_->table->metadata()->current_snapshot_id;
   bool is_first_attempt = true;
   int32_t attempt = 0;
   std::string last_error;
@@ -403,27 +401,42 @@ Result<std::shared_ptr<Table>> Transaction::Commit() {
           });
 
   if (commit_result.has_value()) {
-    // Name the resulting snapshot only when this commit produced one (current
-    // snapshot advanced); metadata-only commits report a plain success.
+    // The builder contains only changes made by the successful attempt. Inspecting
+    // AddSnapshot changes avoids attributing a concurrent writer's snapshot to this
+    // transaction and also detects snapshots committed with StageOnly or ToBranch.
     std::string detail;
-    if (auto snapshot = commit_result.value()->metadata()->Snapshot();
-        snapshot.has_value() &&
-        snapshot.value()->snapshot_id != base_current_snapshot_id) {
-      const auto& summary = snapshot.value()->summary;
-      auto op = summary.find(SnapshotSummaryFields::kOperation);
-      detail =
-          std::format(": committed snapshot {} (op={})", snapshot.value()->snapshot_id,
-                      op != summary.end() ? op->second : "unknown");
+    const auto& changes = ctx_->metadata_builder->changes();
+    size_t added_snapshot_count = 0;
+    for (const auto& change : changes) {
+      added_snapshot_count += change->kind() == TableUpdate::Kind::kAddSnapshot;
+    }
+    if (added_snapshot_count > 0) {
+      detail.reserve(32 + added_snapshot_count * 48);
+      std::format_to(std::back_inserter(detail), ": committed snapshot{} ",
+                     added_snapshot_count == 1 ? "" : "s");
+
+      size_t appended_snapshot_count = 0;
+      for (const auto& change : changes) {
+        if (change->kind() != TableUpdate::Kind::kAddSnapshot) {
+          continue;
+        }
+        const auto& snapshot =
+            internal::checked_cast<const table::AddSnapshot&>(*change).snapshot();
+        if (appended_snapshot_count++ > 0) {
+          detail += ", ";
+        }
+        const auto& summary = snapshot->summary;
+        auto op = summary.find(SnapshotSummaryFields::kOperation);
+        std::format_to(std::back_inserter(detail), "{} (op={})", snapshot->snapshot_id,
+                       op != summary.end() ? op->second : "unknown");
+      }
     }
     if (attempt > 1) {
       ICEBERG_LOG_INFO("Transaction commit succeeded after {} attempts{}", attempt,
                        detail);
     } else {
       ICEBERG_LOG_INFO("Transaction commit succeeded{}", detail);
     }
-  } else {
-    ICEBERG_LOG_ERROR("Transaction commit failed after {} attempt(s): {}", attempt,
-                      commit_result.error().message);
   }
 
   Result<const TableMetadata*> finalize_result =