fix

Author: linguoxuan

src/iceberg/table.cc

@@ -87,7 +87,6 @@ Status Table::Refresh() {
   ICEBERG_ASSIGN_OR_RAISE(auto refreshed_table, catalog_->LoadTable(identifier_));
   if (metadata_location_ != refreshed_table->metadata_file_location()) {
     metadata_ = std::move(refreshed_table->metadata_);
-    metadata_location_ = std::string(refreshed_table->metadata_file_location());
     io_ = std::move(refreshed_table->io_);
     metadata_cache_ = std::make_unique<TableMetadataCache>(metadata_.get());
   }

src/iceberg/transaction.cc

@@ -347,32 +347,22 @@ Result<std::shared_ptr<Table>> Transaction::Commit() {
     return ctx_->table;
   }
 
-  Result<std::shared_ptr<Table>> commit_result;
-  if (!CanRetry()) {
-    std::vector<std::unique_ptr<TableRequirement>> requirements;
-    switch (ctx_->kind) {
-      case TransactionKind::kCreate: {
-        ICEBERG_ASSIGN_OR_RAISE(requirements, TableRequirements::ForCreateTable(updates));
-      } break;
-      case TransactionKind::kUpdate: {
-        ICEBERG_ASSIGN_OR_RAISE(
-            requirements,
-            TableRequirements::ForUpdateTable(*ctx_->metadata_builder->base(), updates));
-      } break;
-    }
-    commit_result =
-        ctx_->table->catalog()->UpdateTable(ctx_->table->name(), requirements, updates);
-  } else {
-    const auto& props = ctx_->table->properties();
-    int32_t num_retries = props.Get(TableProperties::kCommitNumRetries);
-    int32_t min_wait_ms = props.Get(TableProperties::kCommitMinRetryWaitMs);
-    int32_t max_wait_ms = props.Get(TableProperties::kCommitMaxRetryWaitMs);
-    int32_t total_timeout_ms = props.Get(TableProperties::kCommitTotalRetryTimeMs);
-
-    commit_result =
-        MakeCommitRetryRunner(num_retries, min_wait_ms, max_wait_ms, total_timeout_ms)
-            .Run([this]() -> Result<std::shared_ptr<Table>> { return CommitOnce(); });
-  }
+  const auto& props = ctx_->table->properties();
+  int32_t num_retries =
+      CanRetry() ? static_cast<int32_t>(props.Get(TableProperties::kCommitNumRetries))
+                 : 0;
+  int32_t min_wait_ms = props.Get(TableProperties::kCommitMinRetryWaitMs);
+  int32_t max_wait_ms = props.Get(TableProperties::kCommitMaxRetryWaitMs);
+  int32_t total_timeout_ms = props.Get(TableProperties::kCommitTotalRetryTimeMs);
+
+  bool is_first_attempt = true;
+  auto commit_result =
+      MakeCommitRetryRunner(num_retries, min_wait_ms, max_wait_ms, total_timeout_ms)
+          .Run([this, &is_first_attempt]() -> Result<std::shared_ptr<Table>> {
+            auto result = CommitOnce(is_first_attempt);
+            is_first_attempt = false;
+            return result;
+          });
 
   for (const auto& update : pending_updates_) {
     std::ignore = update->Finalize(commit_result.has_value()
@@ -389,27 +379,31 @@ Result<std::shared_ptr<Table>> Transaction::Commit() {
   return ctx_->table;
 }
 
-Result<std::shared_ptr<Table>> Transaction::CommitOnce() {
-  auto refresh_result = ctx_->table->Refresh();
-  if (!refresh_result.has_value()) {
-    return std::unexpected(refresh_result.error());
-  }
+Result<std::shared_ptr<Table>> Transaction::CommitOnce(bool is_first_attempt) {
+  std::vector<std::unique_ptr<TableRequirement>> requirements;
 
-  if (ctx_->metadata_builder->base() != ctx_->table->metadata().get()) {
-    ctx_->metadata_builder =
-        TableMetadataBuilder::BuildFrom(ctx_->table->metadata().get());
-    for (const auto& update : pending_updates_) {
-      auto commit_status = update->Commit();
-      if (!commit_status.has_value()) {
-        return std::unexpected(commit_status.error());
+  switch (ctx_->kind) {
+    case TransactionKind::kCreate: {
+      ICEBERG_ASSIGN_OR_RAISE(requirements, TableRequirements::ForCreateTable(
+                                                ctx_->metadata_builder->changes()));
+    } break;
+    case TransactionKind::kUpdate: {
+      if (!is_first_attempt) {
+        ICEBERG_RETURN_UNEXPECTED(ctx_->table->Refresh());
       }
-    }
+      if (ctx_->metadata_builder->base() != ctx_->table->metadata().get()) {
+        ctx_->metadata_builder =
+            TableMetadataBuilder::BuildFrom(ctx_->table->metadata().get());
+        for (const auto& update : pending_updates_) {
+          ICEBERG_RETURN_UNEXPECTED(Apply(*update));
+        }
+      }
+      ICEBERG_ASSIGN_OR_RAISE(requirements, TableRequirements::ForUpdateTable(
+                                                *ctx_->metadata_builder->base(),
+                                                ctx_->metadata_builder->changes()));
+    } break;
   }
 
-  ICEBERG_ASSIGN_OR_RAISE(auto requirements, TableRequirements::ForUpdateTable(
-                                                 *ctx_->metadata_builder->base(),
-                                                 ctx_->metadata_builder->changes()));
-
   return ctx_->table->catalog()->UpdateTable(ctx_->table->name(), requirements,
                                              ctx_->metadata_builder->changes());
 }

src/iceberg/transaction.h

@@ -138,8 +138,8 @@ class ICEBERG_EXPORT Transaction : public std::enable_shared_from_this<Transacti
   Status ApplyUpdateSortOrder(UpdateSortOrder& update);
   Status ApplyUpdateStatistics(UpdateStatistics& update);
 
-  /// \brief Perform a single commit attempt for UPDATE transactions
-  Result<std::shared_ptr<Table>> CommitOnce();
+  /// \brief Perform a single commit attempt
+  Result<std::shared_ptr<Table>> CommitOnce(bool is_first_attempt);
 
   /// \brief Whether this transaction can retry after a commit conflict.
   bool CanRetry() const;

src/iceberg/update/snapshot_update.cc

@@ -218,8 +218,7 @@ Result<std::vector<ManifestFile>> SnapshotUpdate::WriteDeleteManifests(
 }
 
 int64_t SnapshotUpdate::SnapshotId() {
-  while (!snapshot_id_.has_value() ||
-         base().SnapshotById(snapshot_id_.value()).has_value()) {
+  if (!snapshot_id_.has_value()) {
     snapshot_id_ = SnapshotUtil::GenerateSnapshotId(base());
   }
   return snapshot_id_.value();

src/iceberg/update/update_partition_spec.h

@@ -100,6 +100,8 @@ class ICEBERG_EXPORT UpdatePartitionSpec : public PendingUpdate {
 
   Kind kind() const final { return Kind::kUpdatePartitionSpec; }
 
+  bool IsRetryable() const override { return false; }
+
   struct ApplyResult {
     std::shared_ptr<PartitionSpec> spec;
     bool set_as_default;

src/iceberg/update/update_partition_statistics.h

@@ -63,6 +63,8 @@ class ICEBERG_EXPORT UpdatePartitionStatistics : public PendingUpdate {
 
   Kind kind() const final { return Kind::kUpdatePartitionStatistics; }
 
+  bool IsRetryable() const override { return false; }
+
   struct ApplyResult {
     std::vector<std::pair<int64_t, std::shared_ptr<PartitionStatisticsFile>>> to_set;
     std::vector<int64_t> to_remove;

src/iceberg/update/update_schema.h

@@ -334,6 +334,8 @@ class ICEBERG_EXPORT UpdateSchema : public PendingUpdate {
 
   Kind kind() const final { return Kind::kUpdateSchema; }
 
+  bool IsRetryable() const override { return false; }
+
   struct ApplyResult {
     std::shared_ptr<Schema> schema;
     int32_t new_last_column_id;

src/iceberg/update/update_statistics.h

@@ -62,6 +62,8 @@ class ICEBERG_EXPORT UpdateStatistics : public PendingUpdate {
 
   Kind kind() const final { return Kind::kUpdateStatistics; }
 
+  bool IsRetryable() const override { return false; }
+
   struct ApplyResult {
     std::vector<std::pair<int64_t, std::shared_ptr<StatisticsFile>>> to_set;
     std::vector<int64_t> to_remove;

src/iceberg/util/retry_util.h

@@ -21,7 +21,6 @@
 
 #include <algorithm>
 #include <chrono>
-#include <functional>
 #include <random>
 #include <thread>
 #include <vector>
@@ -51,26 +50,48 @@ class RetryRunner {
   explicit RetryRunner(RetryConfig config = {}) : config_(std::move(config)) {}
 
   /// \brief Specify error types that should trigger a retry.
+  ///
+  /// When set, only errors matching one of these kinds will be retried.
+  /// All other errors will stop retries immediately.
+  ///
+  /// \note OnlyRetryOn takes priority over StopRetryOn. If OnlyRetryOn is set,
+  /// StopRetryOn is ignored.
   RetryRunner& OnlyRetryOn(std::initializer_list<ErrorKind> error_kinds) {
     only_retry_on_ = std::vector<ErrorKind>(error_kinds);
     return *this;
   }
 
-  /// \brief Specify a single error type that should trigger a retry
+  /// \brief Specify a single error type that should trigger a retry.
+  ///
+  /// \note OnlyRetryOn takes priority over StopRetryOn. If OnlyRetryOn is set,
+  /// StopRetryOn is ignored.
   RetryRunner& OnlyRetryOn(ErrorKind error_kind) {
     only_retry_on_ = std::vector<ErrorKind>{error_kind};
     return *this;
   }
 
-  /// \brief Specify error types that should stop retries immediately
+  /// \brief Specify error types that should stop retries immediately.
+  ///
+  /// When set, errors matching one of these kinds will not be retried.
+  /// All other errors will be retried.
+  ///
+  /// \note OnlyRetryOn takes priority over StopRetryOn. If OnlyRetryOn is set,
+  /// StopRetryOn is ignored.
   RetryRunner& StopRetryOn(std::initializer_list<ErrorKind> error_kinds) {
     stop_retry_on_ = std::vector<ErrorKind>(error_kinds);
     return *this;
   }
 
   /// \brief Run a task that returns a Result<T>
+  ///
+  /// TODO: Replace attempt_counter with a metrics reporter once it is available.
   template <typename F, typename T = typename std::invoke_result_t<F>::value_type>
   Result<T> Run(F&& task, int32_t* attempt_counter = nullptr) {
+    if (config_.num_retries < 0) {
+      return InvalidArgument("num_retries must be non-negative, got {}",
+                             config_.num_retries);
+    }
+
     auto start_time = std::chrono::steady_clock::now();
     int32_t attempt = 0;
     int32_t max_attempts = config_.num_retries + 1;
@@ -103,7 +124,7 @@ class RetryRunner {
       }
 
       int32_t delay_ms = CalculateDelay(attempt);
-      Sleep(delay_ms);
+      std::this_thread::sleep_for(std::chrono::milliseconds(delay_ms));
     }
   }
 
@@ -137,11 +158,6 @@ class RetryRunner {
     return std::max(1, delay_ms);
   }
 
-  /// \brief Sleep for the specified duration
-  void Sleep(int32_t ms) const {
-    std::this_thread::sleep_for(std::chrono::milliseconds(ms));
-  }
-
   RetryConfig config_;
   std::vector<ErrorKind> only_retry_on_;
   std::vector<ErrorKind> stop_retry_on_;