feat: implement TableMetadataBuilder with AssignUUID (#268)

This commit implements the core TableMetadataBuilder pattern following
the established design from the previous session.

Pattern established:
1. Builder validates input and collects errors (no exceptions)
2. Changes are tracked for serialization
3. Requirements generated for optimistic concurrency
4. Build() validates metadata consistency before returning

Author: gty404

src/iceberg/partition_spec.h

@@ -47,6 +47,7 @@ class ICEBERG_EXPORT PartitionSpec : public util::Formattable {
   /// \brief The start ID for partition field.  It is only used to generate
   /// partition field id for v1 metadata where it is tracked.
   static constexpr int32_t kLegacyPartitionDataIdStart = 1000;
+  static constexpr int32_t kInvalidPartitionFieldId = -1;
 
   /// \brief Create a new partition spec.
   ///

src/iceberg/schema.h

@@ -46,6 +46,7 @@ namespace iceberg {
 class ICEBERG_EXPORT Schema : public StructType {
  public:
   static constexpr int32_t kInitialSchemaId = 0;
+  static constexpr int32_t kInvalidColumnId = -1;
 
   explicit Schema(std::vector<SchemaField> fields,
                   std::optional<int32_t> schema_id = std::nullopt);

src/iceberg/table_metadata.cc

@@ -20,6 +20,7 @@
 #include "iceberg/table_metadata.h"
 
 #include <algorithm>
+#include <chrono>
 #include <format>
 #include <string>
 
@@ -36,9 +37,14 @@
 #include "iceberg/table_update.h"
 #include "iceberg/util/gzip_internal.h"
 #include "iceberg/util/macros.h"
+#include "iceberg/util/uuid.h"
 
 namespace iceberg {
 
+namespace {
+const TimePointMs kInvalidLastUpdatedMs = TimePointMs::min();
+}
+
 std::string ToString(const SnapshotLogEntry& entry) {
   return std::format("SnapshotLogEntry[timestampMillis={},snapshotId={}]",
                      entry.timestamp_ms, entry.snapshot_id);
@@ -201,13 +207,46 @@ Status TableMetadataUtil::Write(FileIO& io, const std::string& location,
 
 // TableMetadataBuilder implementation
 
-struct TableMetadataBuilder::Impl {};
+struct TableMetadataBuilder::Impl {
+  // Base metadata (nullptr for new tables)
+  const TableMetadata* base;
+
+  // Working metadata copy
+  TableMetadata metadata;
+
+  // Change tracking
+  std::vector<std::unique_ptr<TableUpdate>> changes;
+
+  // Error collection (since methods return *this and cannot throw)
+  std::vector<Error> errors;
+
+  // Metadata location tracking
+  std::optional<std::string> metadata_location;
+  std::optional<std::string> previous_metadata_location;
+
+  // Constructor for new table
+  explicit Impl(int8_t format_version) : base(nullptr), metadata{} {
+    metadata.format_version = format_version;
+    metadata.last_sequence_number = TableMetadata::kInitialSequenceNumber;
+    metadata.last_updated_ms = kInvalidLastUpdatedMs;
+    metadata.last_column_id = Schema::kInvalidColumnId;
+    metadata.default_spec_id = PartitionSpec::kInitialSpecId;
+    metadata.last_partition_id = PartitionSpec::kInvalidPartitionFieldId;
+    metadata.current_snapshot_id = Snapshot::kInvalidSnapshotId;
+    metadata.default_sort_order_id = SortOrder::kInitialSortOrderId;
+    metadata.next_row_id = TableMetadata::kInitialRowId;
+  }
+
+  // Constructor from existing metadata
+  explicit Impl(const TableMetadata* base_metadata)
+      : base(base_metadata), metadata(*base_metadata) {}
+};
 
 TableMetadataBuilder::TableMetadataBuilder(int8_t format_version)
-    : impl_(std::make_unique<Impl>()) {}
+    : impl_(std::make_unique<Impl>(format_version)) {}
 
 TableMetadataBuilder::TableMetadataBuilder(const TableMetadata* base)
-    : impl_(std::make_unique<Impl>()) {}
+    : impl_(std::make_unique<Impl>(base)) {}
 
 TableMetadataBuilder::~TableMetadataBuilder() = default;
 
@@ -238,12 +277,35 @@ TableMetadataBuilder& TableMetadataBuilder::SetPreviousMetadataLocation(
 }
 
 TableMetadataBuilder& TableMetadataBuilder::AssignUUID() {
-  throw IcebergError(std::format("{} not implemented", __FUNCTION__));
+  if (impl_->metadata.table_uuid.empty()) {
+    // Generate a random UUID
+    return AssignUUID(Uuid::GenerateV4().ToString());
+  }
+
+  return *this;
 }
 
 TableMetadataBuilder& TableMetadataBuilder::AssignUUID(std::string_view uuid) {
-  throw IcebergError(std::format("{} not implemented", __FUNCTION__));
-  ;
+  std::string uuid_str(uuid);
+
+  // Validation: UUID cannot be empty
+  if (uuid_str.empty()) {
+    impl_->errors.emplace_back(ErrorKind::kInvalidArgument, "Cannot assign empty UUID");
+    return *this;
+  }
+
+  // Check if UUID is already set to the same value (no-op)
+  if (StringUtils::EqualsIgnoreCase(impl_->metadata.table_uuid, uuid_str)) {
+    return *this;
+  }
+
+  // Update the metadata
+  impl_->metadata.table_uuid = uuid_str;
+
+  // Record the change
+  impl_->changes.push_back(std::make_unique<table::AssignUUID>(std::move(uuid_str)));
+
+  return *this;
 }
 
 TableMetadataBuilder& TableMetadataBuilder::UpgradeFormatVersion(
@@ -377,12 +439,29 @@ TableMetadataBuilder& TableMetadataBuilder::RemoveEncryptionKey(std::string_view
   throw IcebergError(std::format("{} not implemented", __FUNCTION__));
 }
 
-TableMetadataBuilder& TableMetadataBuilder::DiscardChanges() {
-  throw IcebergError(std::format("{} not implemented", __FUNCTION__));
-}
-
 Result<std::unique_ptr<TableMetadata>> TableMetadataBuilder::Build() {
-  return NotImplemented("TableMetadataBuilder::Build not implemented");
+  // 1. Check for accumulated errors
+  if (!impl_->errors.empty()) {
+    std::string error_msg = "Failed to build TableMetadata due to validation errors:\n";
+    for (const auto& [kind, message] : impl_->errors) {
+      error_msg += "  - " + message + "\n";
+    }
+    return CommitFailed("{}", error_msg);
+  }
+
+  // 2. Validate metadata consistency through TableMetadata#Validate
+
+  // 3. Update last_updated_ms if there are changes
+  if (impl_->metadata.last_updated_ms == kInvalidLastUpdatedMs) {
+    impl_->metadata.last_updated_ms =
+        TimePointMs{std::chrono::duration_cast<std::chrono::milliseconds>(
+            std::chrono::system_clock::now().time_since_epoch())};
+  }
+
+  // 4. Create and return the TableMetadata
+  auto result = std::make_unique<TableMetadata>(std::move(impl_->metadata));
+
+  return result;
 }
 
 }  // namespace iceberg

src/iceberg/table_metadata.h

@@ -377,13 +377,6 @@ class ICEBERG_EXPORT TableMetadataBuilder {
   /// \return Reference to this builder for method chaining
   TableMetadataBuilder& RemoveEncryptionKey(std::string_view key_id);
 
-  /// \brief Discard all accumulated changes
-  ///
-  /// This is useful when you want to reset the builder state without
-  /// creating a new builder instance.
-  /// \return Reference to this builder for method chaining
-  TableMetadataBuilder& DiscardChanges();
-
   /// \brief Build the TableMetadata object
   ///
   /// \return A Result containing the constructed TableMetadata or an error

src/iceberg/table_requirement.cc

@@ -20,15 +20,28 @@
 #include "iceberg/table_requirement.h"
 
 #include "iceberg/table_metadata.h"
+#include "iceberg/util/string_util.h"
 
 namespace iceberg::table {
 
 Status AssertDoesNotExist::Validate(const TableMetadata* base) const {
-  return NotImplemented("AssertTableDoesNotExist::Validate not implemented");
+  return NotImplemented("AssertDoesNotExist::Validate not implemented");
 }
 
 Status AssertUUID::Validate(const TableMetadata* base) const {
-  return NotImplemented("AssertTableUUID::Validate not implemented");
+  // Validate that the table UUID matches the expected value
+
+  if (base == nullptr) {
+    return CommitFailed("Requirement failed: current table metadata is missing");
+  }
+
+  if (!StringUtils::EqualsIgnoreCase(base->table_uuid, uuid_)) {
+    return CommitFailed(
+        "Requirement failed: table UUID does not match (expected='{}', actual='{}')",
+        uuid_, base->table_uuid);
+  }
+
+  return {};
 }
 
 Status AssertRefSnapshotID::Validate(const TableMetadata* base) const {

src/iceberg/table_requirements.cc

@@ -26,11 +26,11 @@
 namespace iceberg {
 
 void TableUpdateContext::AddRequirement(std::unique_ptr<TableRequirement> requirement) {
-  throw IcebergError("TableUpdateContext::AddRequirement not implemented");
+  requirements_.emplace_back(std::move(requirement));
 }
 
 Result<std::vector<std::unique_ptr<TableRequirement>>> TableUpdateContext::Build() {
-  return NotImplemented("TableUpdateContext::Build not implemented");
+  return std::move(requirements_);
 }
 
 Result<std::vector<std::unique_ptr<TableRequirement>>> TableRequirements::ForCreateTable(

src/iceberg/table_update.cc

@@ -21,18 +21,32 @@
 
 #include "iceberg/exception.h"
 #include "iceberg/table_metadata.h"
+#include "iceberg/table_requirement.h"
 #include "iceberg/table_requirements.h"
 
 namespace iceberg::table {
 
 // AssignUUID
 
 void AssignUUID::ApplyTo(TableMetadataBuilder& builder) const {
-  throw IcebergError(std::format("{} not implemented", __FUNCTION__));
+  builder.AssignUUID(uuid_);
 }
 
 Status AssignUUID::GenerateRequirements(TableUpdateContext& context) const {
-  return NotImplemented("AssignTableUUID::GenerateRequirements not implemented");
+  // AssignUUID operation generates a requirement to assert the table's UUID
+  // if a base metadata exists (i.e., this is an update operation)
+
+  const TableMetadata* base = context.base();
+
+  if (base != nullptr && !base->table_uuid.empty()) {
+    // For table updates, assert that the current UUID matches what we expect
+    context.AddRequirement(std::make_unique<AssertUUID>(base->table_uuid));
+  }
+
+  // Note: For table creation (base == nullptr), no UUID requirement is needed
+  // as the table doesn't exist yet
+
+  return {};
 }
 
 // UpgradeFormatVersion
@@ -42,8 +56,7 @@ void UpgradeFormatVersion::ApplyTo(TableMetadataBuilder& builder) const {
 }
 
 Status UpgradeFormatVersion::GenerateRequirements(TableUpdateContext& context) const {
-  return NotImplemented(
-      "UpgradeTableFormatVersion::GenerateRequirements not implemented");
+  return NotImplemented("UpgradeFormatVersion::GenerateRequirements not implemented");
 }
 
 // AddSchema

src/iceberg/test/CMakeLists.txt

@@ -82,7 +82,8 @@ add_iceberg_test(table_test
                  test_common.cc
                  json_internal_test.cc
                  table_test.cc
-                 schema_json_test.cc)
+                 schema_json_test.cc
+                 table_metadata_builder_test.cc)
 
 add_iceberg_test(expression_test
                  SOURCES

src/iceberg/test/matchers.h

@@ -23,6 +23,7 @@
 #include <gtest/gtest.h>
 
 #include "iceberg/result.h"
+#include "iceberg/util/macros.h"
 
 /*
  * \brief Define custom matchers for expected<T, Error> values
@@ -210,4 +211,17 @@ auto ErrorIs(MatcherT&& matcher) {
       ResultMatcher<std::decay_t<MatcherT>>(false, std::forward<MatcherT>(matcher)));
 }
 
+// Evaluate `rexpr` which should return a Result<T, Error>.
+// On success: assign the contained value to `lhs`.
+// On failure: fail the test with the error message.
+#define ICEBERG_UNWRAP_OR_FAIL_IMPL(result_name, lhs, rexpr)  \
+  auto&& result_name = (rexpr);                               \
+  ASSERT_TRUE(result_name.has_value())                        \
+      << "Operation failed: " << result_name.error().message; \
+  lhs = std::move(result_name.value());
+
+#define ICEBERG_UNWRAP_OR_FAIL(lhs, rexpr)                                             \
+  ICEBERG_UNWRAP_OR_FAIL_IMPL(ICEBERG_ASSIGN_OR_RAISE_NAME(result_, __COUNTER__), lhs, \
+                              rexpr)
+
 }  // namespace iceberg

src/iceberg/test/meson.build

@@ -47,6 +47,7 @@ iceberg_tests = {
         'sources': files(
             'json_internal_test.cc',
             'schema_json_test.cc',
+            'table_metadata_builder_test.cc',
             'table_test.cc',
             'test_common.cc',
         ),