feat: implement TableMetadataBuilder with AssignUUID

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/table_metadata.cc

@@ -20,6 +20,7 @@
 #include "iceberg/table_metadata.h"
 
 #include <algorithm>
+#include <chrono>
 #include <format>
 #include <string>
 
@@ -36,6 +37,7 @@
 #include "iceberg/table_update.h"
 #include "iceberg/util/gzip_internal.h"
 #include "iceberg/util/macros.h"
+#include "iceberg/util/uuid.h"
 
 namespace iceberg {
 
@@ -201,13 +203,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<Status> 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 = TimePointMs{std::chrono::milliseconds(0)};
+    metadata.last_column_id = 0;
+    metadata.default_spec_id = TableMetadata::kInitialSpecId;
+    metadata.last_partition_id = 0;
+    metadata.current_snapshot_id = TableMetadata::kInvalidSnapshotId;
+    metadata.default_sort_order_id = TableMetadata::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 +273,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 null or empty
+  if (uuid_str.empty()) {
+    impl_->errors.emplace_back(InvalidArgument("Cannot assign null or 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>(uuid_str));
+
+  return *this;
 }
 
 TableMetadataBuilder& TableMetadataBuilder::UpgradeFormatVersion(
@@ -378,11 +436,50 @@ TableMetadataBuilder& TableMetadataBuilder::RemoveEncryptionKey(std::string_view
 }
 
 TableMetadataBuilder& TableMetadataBuilder::DiscardChanges() {
-  throw IcebergError(std::format("{} not implemented", __FUNCTION__));
+  // Clear all changes and errors
+  impl_->changes.clear();
+  impl_->errors.clear();
+
+  // Reset metadata to base state
+  if (impl_->base != nullptr) {
+    impl_->metadata = *impl_->base;
+  } else {
+    // Reset to initial state for new table
+    *impl_ = Impl(impl_->metadata.format_version);
+  }
+
+  return *this;
 }
 
 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& error : impl_->errors) {
+      error_msg += "  - " + error.error().message + "\n";
+    }
+    return CommitFailed("{}", error_msg);
+  }
+
+  // 2. Validate metadata consistency
+
+  // Validate UUID exists for format version > 1
+  if (impl_->metadata.format_version > 1 && impl_->metadata.table_uuid.empty()) {
+    return InvalidArgument("UUID is required for format version {}",
+                           impl_->metadata.format_version);
+  }
+
+  // 3. Update last_updated_ms if there are changes
+  if (!impl_->changes.empty() && impl_->base != nullptr) {
+    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

@@ -73,6 +73,9 @@ struct ICEBERG_EXPORT TableMetadata {
   static constexpr int64_t kInitialSequenceNumber = 0;
   static constexpr int64_t kInvalidSequenceNumber = -1;
   static constexpr int64_t kInitialRowId = 0;
+  static constexpr int32_t kInitialSpecId = 0;
+  static constexpr int32_t kInitialSortOrderId = 1;
+  static constexpr int64_t kInvalidSnapshotId = -1;
 
   /// An integer version number for the format
   int8_t format_version;

src/iceberg/table_requirement.cc

@@ -20,15 +20,28 @@
 #include "iceberg/table_requirement.h"
 
 #include "iceberg/table_metadata.h"
+#include "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

@@ -84,6 +84,8 @@ add_iceberg_test(table_test
                  table_test.cc
                  schema_json_test.cc)
 
+add_iceberg_test(table_metadata_builder_test SOURCES table_metadata_builder_test.cc)
+
 add_iceberg_test(expression_test
                  SOURCES
                  expression_test.cc

src/iceberg/test/meson.build

@@ -80,6 +80,9 @@ iceberg_tests = {
         ),
     },
     'roaring_test': {'sources': files('roaring_test.cc')},
+    'table_metadata_builder_test': {
+        'sources': files('table_metadata_builder_test.cc'),
+    },
 }
 
 if get_option('rest').enabled()