diff --git a/cmake_modules/IcebergBuildUtils.cmake b/cmake_modules/IcebergBuildUtils.cmake
index 99f57d926..74b029708 100644
--- a/cmake_modules/IcebergBuildUtils.cmake
+++ b/cmake_modules/IcebergBuildUtils.cmake
@@ -157,6 +157,10 @@ function(add_iceberg_lib LIB_NAME)
                                                                 hidden
                                      VISIBILITY_INLINES_HIDDEN 1)
 
+    if(MSVC_TOOLCHAIN)
+      target_compile_options(${LIB_NAME}_shared PRIVATE /bigobj)
+    endif()
+
     install(TARGETS ${LIB_NAME}_shared
             EXPORT iceberg_targets
             ARCHIVE DESTINATION ${INSTALL_ARCHIVE_DIR}
@@ -220,6 +224,10 @@ function(add_iceberg_lib LIB_NAME)
       target_compile_definitions(${LIB_NAME}_static PUBLIC ${VISIBILITY_NAME}_STATIC)
     endif()
 
+    if(MSVC_TOOLCHAIN)
+      target_compile_options(${LIB_NAME}_static PRIVATE /bigobj)
+    endif()
+
     install(TARGETS ${LIB_NAME}_static
             EXPORT iceberg_targets
             ARCHIVE DESTINATION ${INSTALL_ARCHIVE_DIR}
diff --git a/src/iceberg/catalog/rest/http_client.cc b/src/iceberg/catalog/rest/http_client.cc
index 7804ebd58..84d458b98 100644
--- a/src/iceberg/catalog/rest/http_client.cc
+++ b/src/iceberg/catalog/rest/http_client.cc
@@ -135,11 +135,33 @@ Status HandleFailureResponse(const cpr::Response& response,
 }  // namespace
 
 void HttpClient::PrepareSession(
-    const std::string& path, const std::unordered_map<std::string, std::string>& params,
+    const std::string& path, HttpMethod method,
+    const std::unordered_map<std::string, std::string>& params,
     const std::unordered_map<std::string, std::string>& headers) {
   session_->SetUrl(cpr::Url{path});
   session_->SetParameters(GetParameters(params));
   session_->RemoveContent();
+  // clear lingering POST mode state from prior requests. CURLOPT_POST is implicitly set
+  // to 1 by POST requests, and this state is not reset by RemoveContent(), so we must
+  // manually enforce HTTP GET to clear it.
+  curl_easy_setopt(session_->GetCurlHolder()->handle, CURLOPT_HTTPGET, 1L);
+  switch (method) {
+    case HttpMethod::kGet:
+      session_->PrepareGet();
+      break;
+    case HttpMethod::kPost:
+      session_->PreparePost();
+      break;
+    case HttpMethod::kPut:
+      session_->PreparePut();
+      break;
+    case HttpMethod::kDelete:
+      session_->PrepareDelete();
+      break;
+    case HttpMethod::kHead:
+      session_->PrepareHead();
+      break;
+  }
   auto final_headers = MergeHeaders(default_headers_, headers);
   session_->SetHeader(final_headers);
 }
@@ -163,7 +185,7 @@ Result<HttpResponse> HttpClient::Get(
   cpr::Response response;
   {
     std::lock_guard guard(session_mutex_);
-    PrepareSession(path, params, headers);
+    PrepareSession(path, HttpMethod::kGet, params, headers);
     response = session_->Get();
   }
 
@@ -180,7 +202,7 @@ Result<HttpResponse> HttpClient::Post(
   cpr::Response response;
   {
     std::lock_guard guard(session_mutex_);
-    PrepareSession(path, /*params=*/{}, headers);
+    PrepareSession(path, HttpMethod::kPost, /*params=*/{}, headers);
     session_->SetBody(cpr::Body{body});
     response = session_->Post();
   }
@@ -205,7 +227,7 @@ Result<HttpResponse> HttpClient::PostForm(
     auto form_headers = headers;
     form_headers[kHeaderContentType] = kMimeTypeFormUrlEncoded;
 
-    PrepareSession(path, /*params=*/{}, form_headers);
+    PrepareSession(path, HttpMethod::kPost, /*params=*/{}, form_headers);
     std::vector<cpr::Pair> pair_list;
     pair_list.reserve(form_data.size());
     for (const auto& [key, val] : form_data) {
@@ -228,7 +250,7 @@ Result<HttpResponse> HttpClient::Head(
   cpr::Response response;
   {
     std::lock_guard guard(session_mutex_);
-    PrepareSession(path, /*params=*/{}, headers);
+    PrepareSession(path, HttpMethod::kHead, /*params=*/{}, headers);
     response = session_->Head();
   }
 
@@ -245,7 +267,7 @@ Result<HttpResponse> HttpClient::Delete(
   cpr::Response response;
   {
     std::lock_guard guard(session_mutex_);
-    PrepareSession(path, params, headers);
+    PrepareSession(path, HttpMethod::kDelete, params, headers);
     response = session_->Delete();
   }
 
diff --git a/src/iceberg/catalog/rest/http_client.h b/src/iceberg/catalog/rest/http_client.h
index a1401b631..84f8e5906 100644
--- a/src/iceberg/catalog/rest/http_client.h
+++ b/src/iceberg/catalog/rest/http_client.h
@@ -25,6 +25,7 @@
 #include <string>
 #include <unordered_map>
 
+#include "iceberg/catalog/rest/endpoint.h"
 #include "iceberg/catalog/rest/iceberg_rest_export.h"
 #include "iceberg/catalog/rest/type_fwd.h"
 #include "iceberg/result.h"
@@ -109,7 +110,7 @@ class ICEBERG_REST_EXPORT HttpClient {
                               const ErrorHandler& error_handler);
 
  private:
-  void PrepareSession(const std::string& path,
+  void PrepareSession(const std::string& path, HttpMethod method,
                       const std::unordered_map<std::string, std::string>& params,
                       const std::unordered_map<std::string, std::string>& headers);
 
diff --git a/src/iceberg/catalog/rest/json_internal.cc b/src/iceberg/catalog/rest/json_internal.cc
index b6bb970ee..b9c4caca6 100644
--- a/src/iceberg/catalog/rest/json_internal.cc
+++ b/src/iceberg/catalog/rest/json_internal.cc
@@ -31,6 +31,8 @@
 #include "iceberg/partition_spec.h"
 #include "iceberg/sort_order.h"
 #include "iceberg/table_identifier.h"
+#include "iceberg/table_requirement.h"
+#include "iceberg/table_update.h"
 #include "iceberg/util/json_util_internal.h"
 #include "iceberg/util/macros.h"
 
@@ -69,6 +71,8 @@ constexpr std::string_view kType = "type";
 constexpr std::string_view kCode = "code";
 constexpr std::string_view kStack = "stack";
 constexpr std::string_view kError = "error";
+constexpr std::string_view kIdentifier = "identifier";
+constexpr std::string_view kRequirements = "requirements";
 
 }  // namespace
 
@@ -390,6 +394,75 @@ Result<CreateTableRequest> CreateTableRequestFromJson(const nlohmann::json& json
   return request;
 }
 
+// CommitTableRequest serialization
+nlohmann::json ToJson(const CommitTableRequest& request) {
+  nlohmann::json json;
+  if (!request.identifier.name.empty()) {
+    json[kIdentifier] = ToJson(request.identifier);
+  }
+
+  nlohmann::json requirements_json = nlohmann::json::array();
+  for (const auto& req : request.requirements) {
+    requirements_json.push_back(ToJson(*req));
+  }
+  json[kRequirements] = std::move(requirements_json);
+
+  nlohmann::json updates_json = nlohmann::json::array();
+  for (const auto& update : request.updates) {
+    updates_json.push_back(ToJson(*update));
+  }
+  json[kUpdates] = std::move(updates_json);
+
+  return json;
+}
+
+Result<CommitTableRequest> CommitTableRequestFromJson(const nlohmann::json& json) {
+  CommitTableRequest request;
+  if (json.contains(kIdentifier)) {
+    ICEBERG_ASSIGN_OR_RAISE(auto identifier_json,
+                            GetJsonValue<nlohmann::json>(json, kIdentifier));
+    ICEBERG_ASSIGN_OR_RAISE(request.identifier, TableIdentifierFromJson(identifier_json));
+  }
+
+  ICEBERG_ASSIGN_OR_RAISE(auto requirements_json,
+                          GetJsonValue<nlohmann::json>(json, kRequirements));
+  for (const auto& req_json : requirements_json) {
+    ICEBERG_ASSIGN_OR_RAISE(auto requirement, TableRequirementFromJson(req_json));
+    request.requirements.push_back(std::move(requirement));
+  }
+
+  ICEBERG_ASSIGN_OR_RAISE(auto updates_json,
+                          GetJsonValue<nlohmann::json>(json, kUpdates));
+  for (const auto& update_json : updates_json) {
+    ICEBERG_ASSIGN_OR_RAISE(auto update, TableUpdateFromJson(update_json));
+    request.updates.push_back(std::move(update));
+  }
+
+  ICEBERG_RETURN_UNEXPECTED(request.Validate());
+  return request;
+}
+
+// CommitTableResponse serialization
+nlohmann::json ToJson(const CommitTableResponse& response) {
+  nlohmann::json json;
+  json[kMetadataLocation] = response.metadata_location;
+  if (response.metadata) {
+    json[kMetadata] = ToJson(*response.metadata);
+  }
+  return json;
+}
+
+Result<CommitTableResponse> CommitTableResponseFromJson(const nlohmann::json& json) {
+  CommitTableResponse response;
+  ICEBERG_ASSIGN_OR_RAISE(response.metadata_location,
+                          GetJsonValue<std::string>(json, kMetadataLocation));
+  ICEBERG_ASSIGN_OR_RAISE(auto metadata_json,
+                          GetJsonValue<nlohmann::json>(json, kMetadata));
+  ICEBERG_ASSIGN_OR_RAISE(response.metadata, TableMetadataFromJson(metadata_json));
+  ICEBERG_RETURN_UNEXPECTED(response.Validate());
+  return response;
+}
+
 #define ICEBERG_DEFINE_FROM_JSON(Model)                       \
   template <>                                                 \
   Result<Model> FromJson<Model>(const nlohmann::json& json) { \
@@ -409,5 +482,7 @@ ICEBERG_DEFINE_FROM_JSON(LoadTableResult)
 ICEBERG_DEFINE_FROM_JSON(RegisterTableRequest)
 ICEBERG_DEFINE_FROM_JSON(RenameTableRequest)
 ICEBERG_DEFINE_FROM_JSON(CreateTableRequest)
+ICEBERG_DEFINE_FROM_JSON(CommitTableRequest)
+ICEBERG_DEFINE_FROM_JSON(CommitTableResponse)
 
 }  // namespace iceberg::rest
diff --git a/src/iceberg/catalog/rest/json_internal.h b/src/iceberg/catalog/rest/json_internal.h
index e2a88b4c1..081a02054 100644
--- a/src/iceberg/catalog/rest/json_internal.h
+++ b/src/iceberg/catalog/rest/json_internal.h
@@ -56,6 +56,8 @@ ICEBERG_DECLARE_JSON_SERDE(LoadTableResult)
 ICEBERG_DECLARE_JSON_SERDE(RegisterTableRequest)
 ICEBERG_DECLARE_JSON_SERDE(RenameTableRequest)
 ICEBERG_DECLARE_JSON_SERDE(CreateTableRequest)
+ICEBERG_DECLARE_JSON_SERDE(CommitTableRequest)
+ICEBERG_DECLARE_JSON_SERDE(CommitTableResponse)
 
 #undef ICEBERG_DECLARE_JSON_SERDE
 
diff --git a/src/iceberg/catalog/rest/rest_catalog.cc b/src/iceberg/catalog/rest/rest_catalog.cc
index 2c28691b1..6fe829ffc 100644
--- a/src/iceberg/catalog/rest/rest_catalog.cc
+++ b/src/iceberg/catalog/rest/rest_catalog.cc
@@ -42,6 +42,8 @@
 #include "iceberg/schema.h"
 #include "iceberg/sort_order.h"
 #include "iceberg/table.h"
+#include "iceberg/table_requirement.h"
+#include "iceberg/table_update.h"
 #include "iceberg/util/macros.h"
 
 namespace iceberg::rest {
@@ -175,7 +177,7 @@ Result<std::vector<Namespace>> RestCatalog::ListNamespaces(const Namespace& ns)
     if (list_response.next_page_token.empty()) {
       return result;
     }
-    next_token = list_response.next_page_token;
+    next_token = std::move(list_response.next_page_token);
   }
   return result;
 }
@@ -244,9 +246,30 @@ Status RestCatalog::UpdateNamespaceProperties(
   return {};
 }
 
-Result<std::vector<TableIdentifier>> RestCatalog::ListTables(
-    [[maybe_unused]] const Namespace& ns) const {
-  return NotImplemented("Not implemented");
+Result<std::vector<TableIdentifier>> RestCatalog::ListTables(const Namespace& ns) const {
+  ICEBERG_ENDPOINT_CHECK(supported_endpoints_, Endpoint::ListTables());
+
+  ICEBERG_ASSIGN_OR_RAISE(auto path, paths_->Tables(ns));
+  std::vector<TableIdentifier> result;
+  std::string next_token;
+  while (true) {
+    std::unordered_map<std::string, std::string> params;
+    if (!next_token.empty()) {
+      params[kQueryParamPageToken] = next_token;
+    }
+    ICEBERG_ASSIGN_OR_RAISE(
+        const auto response,
+        client_->Get(path, params, /*headers=*/{}, *TableErrorHandler::Instance()));
+    ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(response.body()));
+    ICEBERG_ASSIGN_OR_RAISE(auto list_response, ListTablesResponseFromJson(json));
+    result.insert(result.end(), list_response.identifiers.begin(),
+                  list_response.identifiers.end());
+    if (list_response.next_page_token.empty()) {
+      return result;
+    }
+    next_token = std::move(list_response.next_page_token);
+  }
+  return result;
 }
 
 Result<std::shared_ptr<Table>> RestCatalog::CreateTable(
@@ -280,10 +303,33 @@ Result<std::shared_ptr<Table>> RestCatalog::CreateTable(
 }
 
 Result<std::shared_ptr<Table>> RestCatalog::UpdateTable(
-    [[maybe_unused]] const TableIdentifier& identifier,
-    [[maybe_unused]] const std::vector<std::unique_ptr<TableRequirement>>& requirements,
-    [[maybe_unused]] const std::vector<std::unique_ptr<TableUpdate>>& updates) {
-  return NotImplemented("Not implemented");
+    const TableIdentifier& identifier,
+    const std::vector<std::unique_ptr<TableRequirement>>& requirements,
+    const std::vector<std::unique_ptr<TableUpdate>>& updates) {
+  ICEBERG_ENDPOINT_CHECK(supported_endpoints_, Endpoint::UpdateTable());
+  ICEBERG_ASSIGN_OR_RAISE(auto path, paths_->Table(identifier));
+
+  CommitTableRequest request{.identifier = identifier};
+  request.requirements.reserve(requirements.size());
+  for (const auto& req : requirements) {
+    request.requirements.push_back(req->Clone());
+  }
+  request.updates.reserve(updates.size());
+  for (const auto& update : updates) {
+    request.updates.push_back(update->Clone());
+  }
+
+  ICEBERG_ASSIGN_OR_RAISE(auto json_request, ToJsonString(ToJson(request)));
+  ICEBERG_ASSIGN_OR_RAISE(
+      const auto response,
+      client_->Post(path, json_request, /*headers=*/{}, *TableErrorHandler::Instance()));
+
+  ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(response.body()));
+  ICEBERG_ASSIGN_OR_RAISE(auto commit_response, CommitTableResponseFromJson(json));
+
+  return Table::Make(identifier, std::move(commit_response.metadata),
+                     std::move(commit_response.metadata_location), file_io_,
+                     shared_from_this());
 }
 
 Result<std::shared_ptr<Transaction>> RestCatalog::StageCreateTable(
@@ -321,9 +367,17 @@ Result<bool> RestCatalog::TableExists(const TableIdentifier& identifier) const {
       client_->Head(path, /*headers=*/{}, *TableErrorHandler::Instance()));
 }
 
-Status RestCatalog::RenameTable([[maybe_unused]] const TableIdentifier& from,
-                                [[maybe_unused]] const TableIdentifier& to) {
-  return NotImplemented("Not implemented");
+Status RestCatalog::RenameTable(const TableIdentifier& from, const TableIdentifier& to) {
+  ICEBERG_ENDPOINT_CHECK(supported_endpoints_, Endpoint::RenameTable());
+  ICEBERG_ASSIGN_OR_RAISE(auto path, paths_->Rename());
+
+  RenameTableRequest request{.source = from, .destination = to};
+  ICEBERG_ASSIGN_OR_RAISE(auto json_request, ToJsonString(ToJson(request)));
+  ICEBERG_ASSIGN_OR_RAISE(
+      const auto response,
+      client_->Post(path, json_request, /*headers=*/{}, *TableErrorHandler::Instance()));
+
+  return {};
 }
 
 Result<std::string> RestCatalog::LoadTableInternal(
@@ -350,9 +404,25 @@ Result<std::shared_ptr<Table>> RestCatalog::LoadTable(const TableIdentifier& ide
 }
 
 Result<std::shared_ptr<Table>> RestCatalog::RegisterTable(
-    [[maybe_unused]] const TableIdentifier& identifier,
-    [[maybe_unused]] const std::string& metadata_file_location) {
-  return NotImplemented("Not implemented");
+    const TableIdentifier& identifier, const std::string& metadata_file_location) {
+  ICEBERG_ENDPOINT_CHECK(supported_endpoints_, Endpoint::RegisterTable());
+  ICEBERG_ASSIGN_OR_RAISE(auto path, paths_->Register(identifier.ns));
+
+  RegisterTableRequest request{
+      .name = identifier.name,
+      .metadata_location = metadata_file_location,
+  };
+
+  ICEBERG_ASSIGN_OR_RAISE(auto json_request, ToJsonString(ToJson(request)));
+  ICEBERG_ASSIGN_OR_RAISE(
+      const auto response,
+      client_->Post(path, json_request, /*headers=*/{}, *TableErrorHandler::Instance()));
+
+  ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(response.body()));
+  ICEBERG_ASSIGN_OR_RAISE(auto load_result, LoadTableResultFromJson(json));
+  return Table::Make(identifier, std::move(load_result.metadata),
+                     std::move(load_result.metadata_location), file_io_,
+                     shared_from_this());
 }
 
 }  // namespace iceberg::rest
diff --git a/src/iceberg/catalog/rest/types.cc b/src/iceberg/catalog/rest/types.cc
index 5c23e47b5..3416bfe35 100644
--- a/src/iceberg/catalog/rest/types.cc
+++ b/src/iceberg/catalog/rest/types.cc
@@ -23,6 +23,8 @@
 #include "iceberg/schema.h"
 #include "iceberg/sort_order.h"
 #include "iceberg/table_metadata.h"
+#include "iceberg/table_requirement.h"
+#include "iceberg/table_update.h"
 
 namespace iceberg::rest {
 
@@ -69,4 +71,49 @@ bool LoadTableResult::operator==(const LoadTableResult& other) const {
   return true;
 }
 
+bool CommitTableRequest::operator==(const CommitTableRequest& other) const {
+  if (identifier != other.identifier) {
+    return false;
+  }
+  if (requirements.size() != other.requirements.size()) {
+    return false;
+  }
+  if (updates.size() != other.updates.size()) {
+    return false;
+  }
+
+  for (size_t i = 0; i < requirements.size(); ++i) {
+    if (!requirements[i] != !other.requirements[i]) {
+      return false;
+    }
+    if (requirements[i] && !requirements[i]->Equals(*other.requirements[i])) {
+      return false;
+    }
+  }
+
+  for (size_t i = 0; i < updates.size(); ++i) {
+    if (!updates[i] != !other.updates[i]) {
+      return false;
+    }
+    if (updates[i] && !updates[i]->Equals(*other.updates[i])) {
+      return false;
+    }
+  }
+
+  return true;
+}
+
+bool CommitTableResponse::operator==(const CommitTableResponse& other) const {
+  if (metadata_location != other.metadata_location) {
+    return false;
+  }
+  if (!metadata != !other.metadata) {
+    return false;
+  }
+  if (metadata && *metadata != *other.metadata) {
+    return false;
+  }
+  return true;
+}
+
 }  // namespace iceberg::rest
diff --git a/src/iceberg/catalog/rest/types.h b/src/iceberg/catalog/rest/types.h
index 01fe330d3..93e7048a5 100644
--- a/src/iceberg/catalog/rest/types.h
+++ b/src/iceberg/catalog/rest/types.h
@@ -59,11 +59,12 @@ struct ICEBERG_REST_EXPORT ErrorResponse {
   /// \brief Validates the ErrorResponse.
   Status Validate() const {
     if (message.empty() || type.empty()) {
-      return Invalid("Invalid error response: missing required fields");
+      return ValidationFailed("Invalid error response: missing required fields");
     }
 
     if (code < 400 || code > 600) {
-      return Invalid("Invalid error response: code {} is out of range [400, 600]", code);
+      return ValidationFailed(
+          "Invalid error response: code {} is out of range [400, 600]", code);
     }
 
     // stack is optional, no validation needed
@@ -93,7 +94,7 @@ struct ICEBERG_REST_EXPORT UpdateNamespacePropertiesRequest {
   Status Validate() const {
     for (const auto& key : removals) {
       if (updates.contains(key)) {
-        return Invalid("Duplicate key to update and remove: {}", key);
+        return ValidationFailed("Duplicate key to update and remove: {}", key);
       }
     }
     return {};
@@ -111,11 +112,11 @@ struct ICEBERG_REST_EXPORT RegisterTableRequest {
   /// \brief Validates the RegisterTableRequest.
   Status Validate() const {
     if (name.empty()) {
-      return Invalid("Missing table name");
+      return ValidationFailed("Missing table name");
     }
 
     if (metadata_location.empty()) {
-      return Invalid("Empty metadata location");
+      return ValidationFailed("Empty metadata location");
     }
 
     return {};
@@ -152,10 +153,10 @@ struct ICEBERG_REST_EXPORT CreateTableRequest {
   /// \brief Validates the CreateTableRequest.
   Status Validate() const {
     if (name.empty()) {
-      return Invalid("Missing table name");
+      return ValidationFailed("Missing table name");
     }
     if (!schema) {
-      return Invalid("Missing schema");
+      return ValidationFailed("Missing schema");
     }
     return {};
   }
@@ -176,7 +177,7 @@ struct ICEBERG_REST_EXPORT LoadTableResult {
   /// \brief Validates the LoadTableResult.
   Status Validate() const {
     if (!metadata) {
-      return Invalid("Invalid metadata: null");
+      return ValidationFailed("Invalid metadata: null");
     }
     return {};
   }
@@ -246,4 +247,35 @@ struct ICEBERG_REST_EXPORT ListTablesResponse {
   bool operator==(const ListTablesResponse&) const = default;
 };
 
+/// \brief Request to commit changes to a table.
+struct ICEBERG_REST_EXPORT CommitTableRequest {
+  TableIdentifier identifier;
+  std::vector<std::shared_ptr<TableRequirement>> requirements;  // required
+  std::vector<std::shared_ptr<TableUpdate>> updates;            // required
+
+  /// \brief Validates the CommitTableRequest.
+  Status Validate() const { return {}; }
+
+  bool operator==(const CommitTableRequest& other) const;
+};
+
+/// \brief Response from committing changes to a table.
+struct ICEBERG_REST_EXPORT CommitTableResponse {
+  std::string metadata_location;            // required
+  std::shared_ptr<TableMetadata> metadata;  // required
+
+  /// \brief Validates the CommitTableResponse.
+  Status Validate() const {
+    if (metadata_location.empty()) {
+      return ValidationFailed("Invalid metadata location: empty");
+    }
+    if (!metadata) {
+      return ValidationFailed("Invalid metadata: null");
+    }
+    return {};
+  }
+
+  bool operator==(const CommitTableResponse& other) const;
+};
+
 }  // namespace iceberg::rest
diff --git a/src/iceberg/json_internal.cc b/src/iceberg/json_internal.cc
index a52f418e4..30b321444 100644
--- a/src/iceberg/json_internal.cc
+++ b/src/iceberg/json_internal.cc
@@ -40,6 +40,8 @@
 #include "iceberg/table_identifier.h"
 #include "iceberg/table_metadata.h"
 #include "iceberg/table_properties.h"
+#include "iceberg/table_requirement.h"
+#include "iceberg/table_update.h"
 #include "iceberg/transform.h"
 #include "iceberg/type.h"
 #include "iceberg/util/checked_cast.h"
@@ -170,6 +172,54 @@ constexpr std::string_view kFileSizeInBytes = "file-size-in-bytes";
 constexpr std::string_view kFileFooterSizeInBytes = "file-footer-size-in-bytes";
 constexpr std::string_view kBlobMetadata = "blob-metadata";
 
+// TableUpdate action constants
+constexpr std::string_view kAction = "action";
+constexpr std::string_view kActionAssignUUID = "assign-uuid";
+constexpr std::string_view kActionUpgradeFormatVersion = "upgrade-format-version";
+constexpr std::string_view kActionAddSchema = "add-schema";
+constexpr std::string_view kActionSetCurrentSchema = "set-current-schema";
+constexpr std::string_view kActionAddPartitionSpec = "add-spec";
+constexpr std::string_view kActionSetDefaultPartitionSpec = "set-default-spec";
+constexpr std::string_view kActionRemovePartitionSpecs = "remove-partition-specs";
+constexpr std::string_view kActionRemoveSchemas = "remove-schemas";
+constexpr std::string_view kActionAddSortOrder = "add-sort-order";
+constexpr std::string_view kActionSetDefaultSortOrder = "set-default-sort-order";
+constexpr std::string_view kActionAddSnapshot = "add-snapshot";
+constexpr std::string_view kActionRemoveSnapshots = "remove-snapshots";
+constexpr std::string_view kActionRemoveSnapshotRef = "remove-snapshot-ref";
+constexpr std::string_view kActionSetSnapshotRef = "set-snapshot-ref";
+constexpr std::string_view kActionSetProperties = "set-properties";
+constexpr std::string_view kActionRemoveProperties = "remove-properties";
+constexpr std::string_view kActionSetLocation = "set-location";
+
+// TableUpdate field constants
+constexpr std::string_view kUUID = "uuid";
+constexpr std::string_view kSpec = "spec";
+constexpr std::string_view kSpecIds = "spec-ids";
+constexpr std::string_view kSchemaIds = "schema-ids";
+constexpr std::string_view kSortOrder = "sort-order";
+constexpr std::string_view kSortOrderId = "sort-order-id";
+constexpr std::string_view kSnapshot = "snapshot";
+constexpr std::string_view kSnapshotIds = "snapshot-ids";
+constexpr std::string_view kRefName = "ref-name";
+constexpr std::string_view kUpdates = "updates";
+constexpr std::string_view kRemovals = "removals";
+
+// TableRequirement type constants
+constexpr std::string_view kRequirementAssertDoesNotExist = "assert-create";
+constexpr std::string_view kRequirementAssertUUID = "assert-table-uuid";
+constexpr std::string_view kRequirementAssertRefSnapshotID = "assert-ref-snapshot-id";
+constexpr std::string_view kRequirementAssertLastAssignedFieldId =
+    "assert-last-assigned-field-id";
+constexpr std::string_view kRequirementAssertCurrentSchemaID = "assert-current-schema-id";
+constexpr std::string_view kRequirementAssertLastAssignedPartitionId =
+    "assert-last-assigned-partition-id";
+constexpr std::string_view kRequirementAssertDefaultSpecID = "assert-default-spec-id";
+constexpr std::string_view kRequirementAssertDefaultSortOrderID =
+    "assert-default-sort-order-id";
+constexpr std::string_view kLastAssignedFieldId = "last-assigned-field-id";
+constexpr std::string_view kLastAssignedPartitionId = "last-assigned-partition-id";
+
 }  // namespace
 
 nlohmann::json ToJson(const SortField& sort_field) {
@@ -233,7 +283,7 @@ nlohmann::json ToJson(const SchemaField& field) {
 nlohmann::json ToJson(const Type& type) {
   switch (type.type_id()) {
     case TypeId::kStruct: {
-      const auto& struct_type = static_cast<const StructType&>(type);
+      const auto& struct_type = internal::checked_cast<const StructType&>(type);
       nlohmann::json json;
       json[kType] = kStruct;
       nlohmann::json fields_json = nlohmann::json::array();
@@ -245,7 +295,7 @@ nlohmann::json ToJson(const Type& type) {
       return json;
     }
     case TypeId::kList: {
-      const auto& list_type = static_cast<const ListType&>(type);
+      const auto& list_type = internal::checked_cast<const ListType&>(type);
       nlohmann::json json;
       json[kType] = kList;
 
@@ -256,7 +306,7 @@ nlohmann::json ToJson(const Type& type) {
       return json;
     }
     case TypeId::kMap: {
-      const auto& map_type = static_cast<const MapType&>(type);
+      const auto& map_type = internal::checked_cast<const MapType&>(type);
       nlohmann::json json;
       json[std::string(kType)] = kMap;
 
@@ -281,7 +331,7 @@ nlohmann::json ToJson(const Type& type) {
     case TypeId::kDouble:
       return "double";
     case TypeId::kDecimal: {
-      const auto& decimal_type = static_cast<const DecimalType&>(type);
+      const auto& decimal_type = internal::checked_cast<const DecimalType&>(type);
       return std::format("decimal({},{})", decimal_type.precision(),
                          decimal_type.scale());
     }
@@ -298,7 +348,7 @@ nlohmann::json ToJson(const Type& type) {
     case TypeId::kBinary:
       return "binary";
     case TypeId::kFixed: {
-      const auto& fixed_type = static_cast<const FixedType&>(type);
+      const auto& fixed_type = internal::checked_cast<const FixedType&>(type);
       return std::format("fixed[{}]", fixed_type.length());
     }
     case TypeId::kUuid:
@@ -308,7 +358,7 @@ nlohmann::json ToJson(const Type& type) {
 }
 
 nlohmann::json ToJson(const Schema& schema) {
-  nlohmann::json json = ToJson(static_cast<const Type&>(schema));
+  nlohmann::json json = ToJson(internal::checked_cast<const Type&>(schema));
   json[kSchemaId] = schema.schema_id();
   if (!schema.IdentifierFieldIds().empty()) {
     json[kIdentifierFieldIds] = schema.IdentifierFieldIds();
@@ -1214,4 +1264,361 @@ Result<Namespace> NamespaceFromJson(const nlohmann::json& json) {
   return ns;
 }
 
+nlohmann::json ToJson(const TableUpdate& update) {
+  nlohmann::json json;
+  switch (update.kind()) {
+    case TableUpdate::Kind::kAssignUUID: {
+      const auto& u = internal::checked_cast<const table::AssignUUID&>(update);
+      json[kAction] = kActionAssignUUID;
+      json[kUUID] = u.uuid();
+      break;
+    }
+    case TableUpdate::Kind::kUpgradeFormatVersion: {
+      const auto& u = internal::checked_cast<const table::UpgradeFormatVersion&>(update);
+      json[kAction] = kActionUpgradeFormatVersion;
+      json[kFormatVersion] = u.format_version();
+      break;
+    }
+    case TableUpdate::Kind::kAddSchema: {
+      const auto& u = internal::checked_cast<const table::AddSchema&>(update);
+      json[kAction] = kActionAddSchema;
+      if (u.schema()) {
+        json[kSchema] = ToJson(*u.schema());
+      } else {
+        json[kSchema] = nlohmann::json::value_t::null;
+      }
+      json[kLastColumnId] = u.last_column_id();
+      break;
+    }
+    case TableUpdate::Kind::kSetCurrentSchema: {
+      const auto& u = internal::checked_cast<const table::SetCurrentSchema&>(update);
+      json[kAction] = kActionSetCurrentSchema;
+      json[kSchemaId] = u.schema_id();
+      break;
+    }
+    case TableUpdate::Kind::kAddPartitionSpec: {
+      const auto& u = internal::checked_cast<const table::AddPartitionSpec&>(update);
+      json[kAction] = kActionAddPartitionSpec;
+      if (u.spec()) {
+        json[kSpec] = ToJson(*u.spec());
+      } else {
+        json[kSpec] = nlohmann::json::value_t::null;
+      }
+      break;
+    }
+    case TableUpdate::Kind::kSetDefaultPartitionSpec: {
+      const auto& u =
+          internal::checked_cast<const table::SetDefaultPartitionSpec&>(update);
+      json[kAction] = kActionSetDefaultPartitionSpec;
+      json[kSpecId] = u.spec_id();
+      break;
+    }
+    case TableUpdate::Kind::kRemovePartitionSpecs: {
+      const auto& u = internal::checked_cast<const table::RemovePartitionSpecs&>(update);
+      json[kAction] = kActionRemovePartitionSpecs;
+      json[kSpecIds] = u.spec_ids();
+      break;
+    }
+    case TableUpdate::Kind::kRemoveSchemas: {
+      const auto& u = internal::checked_cast<const table::RemoveSchemas&>(update);
+      json[kAction] = kActionRemoveSchemas;
+      json[kSchemaIds] = u.schema_ids();
+      break;
+    }
+    case TableUpdate::Kind::kAddSortOrder: {
+      const auto& u = internal::checked_cast<const table::AddSortOrder&>(update);
+      json[kAction] = kActionAddSortOrder;
+      if (u.sort_order()) {
+        json[kSortOrder] = ToJson(*u.sort_order());
+      } else {
+        json[kSortOrder] = nlohmann::json::value_t::null;
+      }
+      break;
+    }
+    case TableUpdate::Kind::kSetDefaultSortOrder: {
+      const auto& u = internal::checked_cast<const table::SetDefaultSortOrder&>(update);
+      json[kAction] = kActionSetDefaultSortOrder;
+      json[kSortOrderId] = u.sort_order_id();
+      break;
+    }
+    case TableUpdate::Kind::kAddSnapshot: {
+      const auto& u = internal::checked_cast<const table::AddSnapshot&>(update);
+      json[kAction] = kActionAddSnapshot;
+      if (u.snapshot()) {
+        json[kSnapshot] = ToJson(*u.snapshot());
+      } else {
+        json[kSnapshot] = nlohmann::json::value_t::null;
+      }
+      break;
+    }
+    case TableUpdate::Kind::kRemoveSnapshots: {
+      const auto& u = internal::checked_cast<const table::RemoveSnapshots&>(update);
+      json[kAction] = kActionRemoveSnapshots;
+      json[kSnapshotIds] = u.snapshot_ids();
+      break;
+    }
+    case TableUpdate::Kind::kRemoveSnapshotRef: {
+      const auto& u = internal::checked_cast<const table::RemoveSnapshotRef&>(update);
+      json[kAction] = kActionRemoveSnapshotRef;
+      json[kRefName] = u.ref_name();
+      break;
+    }
+    case TableUpdate::Kind::kSetSnapshotRef: {
+      const auto& u = internal::checked_cast<const table::SetSnapshotRef&>(update);
+      json[kAction] = kActionSetSnapshotRef;
+      json[kRefName] = u.ref_name();
+      json[kSnapshotId] = u.snapshot_id();
+      json[kType] = ToString(u.type());
+      if (u.min_snapshots_to_keep().has_value()) {
+        json[kMinSnapshotsToKeep] = u.min_snapshots_to_keep().value();
+      }
+      if (u.max_snapshot_age_ms().has_value()) {
+        json[kMaxSnapshotAgeMs] = u.max_snapshot_age_ms().value();
+      }
+      if (u.max_ref_age_ms().has_value()) {
+        json[kMaxRefAgeMs] = u.max_ref_age_ms().value();
+      }
+      break;
+    }
+    case TableUpdate::Kind::kSetProperties: {
+      const auto& u = internal::checked_cast<const table::SetProperties&>(update);
+      json[kAction] = kActionSetProperties;
+      json[kUpdates] = u.updated();
+      break;
+    }
+    case TableUpdate::Kind::kRemoveProperties: {
+      const auto& u = internal::checked_cast<const table::RemoveProperties&>(update);
+      json[kAction] = kActionRemoveProperties;
+      json[kRemovals] = std::vector<std::string>(u.removed().begin(), u.removed().end());
+      break;
+    }
+    case TableUpdate::Kind::kSetLocation: {
+      const auto& u = internal::checked_cast<const table::SetLocation&>(update);
+      json[kAction] = kActionSetLocation;
+      json[kLocation] = u.location();
+      break;
+    }
+  }
+  return json;
+}
+
+nlohmann::json ToJson(const TableRequirement& requirement) {
+  nlohmann::json json;
+  switch (requirement.kind()) {
+    case TableRequirement::Kind::kAssertDoesNotExist:
+      json[kType] = kRequirementAssertDoesNotExist;
+      break;
+    case TableRequirement::Kind::kAssertUUID: {
+      const auto& r = internal::checked_cast<const table::AssertUUID&>(requirement);
+      json[kType] = kRequirementAssertUUID;
+      json[kUUID] = r.uuid();
+      break;
+    }
+    case TableRequirement::Kind::kAssertRefSnapshotID: {
+      const auto& r =
+          internal::checked_cast<const table::AssertRefSnapshotID&>(requirement);
+      json[kType] = kRequirementAssertRefSnapshotID;
+      json[kRefName] = r.ref_name();
+      if (r.snapshot_id().has_value()) {
+        json[kSnapshotId] = r.snapshot_id().value();
+      } else {
+        json[kSnapshotId] = nlohmann::json::value_t::null;
+      }
+      break;
+    }
+    case TableRequirement::Kind::kAssertLastAssignedFieldId: {
+      const auto& r =
+          internal::checked_cast<const table::AssertLastAssignedFieldId&>(requirement);
+      json[kType] = kRequirementAssertLastAssignedFieldId;
+      json[kLastAssignedFieldId] = r.last_assigned_field_id();
+      break;
+    }
+    case TableRequirement::Kind::kAssertCurrentSchemaID: {
+      const auto& r =
+          internal::checked_cast<const table::AssertCurrentSchemaID&>(requirement);
+      json[kType] = kRequirementAssertCurrentSchemaID;
+      json[kCurrentSchemaId] = r.schema_id();
+      break;
+    }
+    case TableRequirement::Kind::kAssertLastAssignedPartitionId: {
+      const auto& r = internal::checked_cast<const table::AssertLastAssignedPartitionId&>(
+          requirement);
+      json[kType] = kRequirementAssertLastAssignedPartitionId;
+      json[kLastAssignedPartitionId] = r.last_assigned_partition_id();
+      break;
+    }
+    case TableRequirement::Kind::kAssertDefaultSpecID: {
+      const auto& r =
+          internal::checked_cast<const table::AssertDefaultSpecID&>(requirement);
+      json[kType] = kRequirementAssertDefaultSpecID;
+      json[kDefaultSpecId] = r.spec_id();
+      break;
+    }
+    case TableRequirement::Kind::kAssertDefaultSortOrderID: {
+      const auto& r =
+          internal::checked_cast<const table::AssertDefaultSortOrderID&>(requirement);
+      json[kType] = kRequirementAssertDefaultSortOrderID;
+      json[kDefaultSortOrderId] = r.sort_order_id();
+      break;
+    }
+  }
+  return json;
+}
+
+Result<std::unique_ptr<TableUpdate>> TableUpdateFromJson(const nlohmann::json& json) {
+  ICEBERG_ASSIGN_OR_RAISE(auto action, GetJsonValue<std::string>(json, kAction));
+
+  if (action == kActionAssignUUID) {
+    ICEBERG_ASSIGN_OR_RAISE(auto uuid, GetJsonValue<std::string>(json, kUUID));
+    return std::make_unique<table::AssignUUID>(std::move(uuid));
+  }
+  if (action == kActionUpgradeFormatVersion) {
+    ICEBERG_ASSIGN_OR_RAISE(auto format_version,
+                            GetJsonValue<int8_t>(json, kFormatVersion));
+    return std::make_unique<table::UpgradeFormatVersion>(format_version);
+  }
+  if (action == kActionAddSchema) {
+    ICEBERG_ASSIGN_OR_RAISE(auto schema_json,
+                            GetJsonValue<nlohmann::json>(json, kSchema));
+    ICEBERG_ASSIGN_OR_RAISE(auto parsed_schema, SchemaFromJson(schema_json));
+    ICEBERG_ASSIGN_OR_RAISE(auto last_column_id,
+                            GetJsonValue<int32_t>(json, kLastColumnId));
+    return std::make_unique<table::AddSchema>(std::move(parsed_schema), last_column_id);
+  }
+  if (action == kActionSetCurrentSchema) {
+    ICEBERG_ASSIGN_OR_RAISE(auto schema_id, GetJsonValue<int32_t>(json, kSchemaId));
+    return std::make_unique<table::SetCurrentSchema>(schema_id);
+  }
+  if (action == kActionAddPartitionSpec) {
+    ICEBERG_ASSIGN_OR_RAISE(auto spec_json, GetJsonValue<nlohmann::json>(json, kSpec));
+    ICEBERG_ASSIGN_OR_RAISE(auto spec_id_opt,
+                            GetJsonValueOptional<int32_t>(spec_json, kSpecId));
+    // TODO(Feiyang Li): add fromJson for UnboundPartitionSpec and then use it here
+    return NotImplemented("FromJson of TableUpdate::AddPartitionSpec is not implemented");
+  }
+  if (action == kActionSetDefaultPartitionSpec) {
+    ICEBERG_ASSIGN_OR_RAISE(auto spec_id, GetJsonValue<int32_t>(json, kSpecId));
+    return std::make_unique<table::SetDefaultPartitionSpec>(spec_id);
+  }
+  if (action == kActionRemovePartitionSpecs) {
+    ICEBERG_ASSIGN_OR_RAISE(auto spec_ids,
+                            GetJsonValue<std::vector<int32_t>>(json, kSpecIds));
+    return std::make_unique<table::RemovePartitionSpecs>(std::move(spec_ids));
+  }
+  if (action == kActionRemoveSchemas) {
+    ICEBERG_ASSIGN_OR_RAISE(auto schema_ids_vec,
+                            GetJsonValue<std::vector<int32_t>>(json, kSchemaIds));
+    std::unordered_set<int32_t> schema_ids(schema_ids_vec.begin(), schema_ids_vec.end());
+    return std::make_unique<table::RemoveSchemas>(std::move(schema_ids));
+  }
+  if (action == kActionAddSortOrder) {
+    ICEBERG_ASSIGN_OR_RAISE(auto sort_order_json,
+                            GetJsonValue<nlohmann::json>(json, kSortOrder));
+    // TODO(Feiyang Li): add fromJson for UnboundSortOrder and then use it here
+    return NotImplemented("FromJson of TableUpdate::AddSortOrder is not implemented");
+  }
+  if (action == kActionSetDefaultSortOrder) {
+    ICEBERG_ASSIGN_OR_RAISE(auto sort_order_id,
+                            GetJsonValue<int32_t>(json, kSortOrderId));
+    return std::make_unique<table::SetDefaultSortOrder>(sort_order_id);
+  }
+  if (action == kActionAddSnapshot) {
+    ICEBERG_ASSIGN_OR_RAISE(auto snapshot_json,
+                            GetJsonValue<nlohmann::json>(json, kSnapshot));
+    ICEBERG_ASSIGN_OR_RAISE(auto snapshot, SnapshotFromJson(snapshot_json));
+    return std::make_unique<table::AddSnapshot>(std::move(snapshot));
+  }
+  if (action == kActionRemoveSnapshots) {
+    ICEBERG_ASSIGN_OR_RAISE(auto snapshot_ids,
+                            GetJsonValue<std::vector<int64_t>>(json, kSnapshotIds));
+    return std::make_unique<table::RemoveSnapshots>(std::move(snapshot_ids));
+  }
+  if (action == kActionRemoveSnapshotRef) {
+    ICEBERG_ASSIGN_OR_RAISE(auto ref_name, GetJsonValue<std::string>(json, kRefName));
+    return std::make_unique<table::RemoveSnapshotRef>(std::move(ref_name));
+  }
+  if (action == kActionSetSnapshotRef) {
+    ICEBERG_ASSIGN_OR_RAISE(auto ref_name, GetJsonValue<std::string>(json, kRefName));
+    ICEBERG_ASSIGN_OR_RAISE(auto snapshot_id, GetJsonValue<int64_t>(json, kSnapshotId));
+    ICEBERG_ASSIGN_OR_RAISE(
+        auto type,
+        GetJsonValue<std::string>(json, kType).and_then(SnapshotRefTypeFromString));
+    ICEBERG_ASSIGN_OR_RAISE(auto min_snapshots,
+                            GetJsonValueOptional<int32_t>(json, kMinSnapshotsToKeep));
+    ICEBERG_ASSIGN_OR_RAISE(auto max_snapshot_age,
+                            GetJsonValueOptional<int64_t>(json, kMaxSnapshotAgeMs));
+    ICEBERG_ASSIGN_OR_RAISE(auto max_ref_age,
+                            GetJsonValueOptional<int64_t>(json, kMaxRefAgeMs));
+    return std::make_unique<table::SetSnapshotRef>(std::move(ref_name), snapshot_id, type,
+                                                   min_snapshots, max_snapshot_age,
+                                                   max_ref_age);
+  }
+  if (action == kActionSetProperties) {
+    using StringMap = std::unordered_map<std::string, std::string>;
+    ICEBERG_ASSIGN_OR_RAISE(auto updates, GetJsonValue<StringMap>(json, kUpdates));
+    return std::make_unique<table::SetProperties>(std::move(updates));
+  }
+  if (action == kActionRemoveProperties) {
+    ICEBERG_ASSIGN_OR_RAISE(auto removals_vec,
+                            GetJsonValue<std::vector<std::string>>(json, kRemovals));
+    std::unordered_set<std::string> removals(
+        std::make_move_iterator(removals_vec.begin()),
+        std::make_move_iterator(removals_vec.end()));
+    return std::make_unique<table::RemoveProperties>(std::move(removals));
+  }
+  if (action == kActionSetLocation) {
+    ICEBERG_ASSIGN_OR_RAISE(auto location, GetJsonValue<std::string>(json, kLocation));
+    return std::make_unique<table::SetLocation>(std::move(location));
+  }
+
+  return JsonParseError("Unknown table update action: {}", action);
+}
+
+Result<std::unique_ptr<TableRequirement>> TableRequirementFromJson(
+    const nlohmann::json& json) {
+  ICEBERG_ASSIGN_OR_RAISE(auto type, GetJsonValue<std::string>(json, kType));
+
+  if (type == kRequirementAssertDoesNotExist) {
+    return std::make_unique<table::AssertDoesNotExist>();
+  }
+  if (type == kRequirementAssertUUID) {
+    ICEBERG_ASSIGN_OR_RAISE(auto uuid, GetJsonValue<std::string>(json, kUUID));
+    return std::make_unique<table::AssertUUID>(std::move(uuid));
+  }
+  if (type == kRequirementAssertRefSnapshotID) {
+    ICEBERG_ASSIGN_OR_RAISE(auto ref_name, GetJsonValue<std::string>(json, kRefName));
+    ICEBERG_ASSIGN_OR_RAISE(auto snapshot_id_opt,
+                            GetJsonValueOptional<int64_t>(json, kSnapshotId));
+    return std::make_unique<table::AssertRefSnapshotID>(std::move(ref_name),
+                                                        snapshot_id_opt);
+  }
+  if (type == kRequirementAssertLastAssignedFieldId) {
+    ICEBERG_ASSIGN_OR_RAISE(auto last_assigned_field_id,
+                            GetJsonValue<int32_t>(json, kLastAssignedFieldId));
+    return std::make_unique<table::AssertLastAssignedFieldId>(last_assigned_field_id);
+  }
+  if (type == kRequirementAssertCurrentSchemaID) {
+    ICEBERG_ASSIGN_OR_RAISE(auto schema_id,
+                            GetJsonValue<int32_t>(json, kCurrentSchemaId));
+    return std::make_unique<table::AssertCurrentSchemaID>(schema_id);
+  }
+  if (type == kRequirementAssertLastAssignedPartitionId) {
+    ICEBERG_ASSIGN_OR_RAISE(auto last_assigned_partition_id,
+                            GetJsonValue<int32_t>(json, kLastAssignedPartitionId));
+    return std::make_unique<table::AssertLastAssignedPartitionId>(
+        last_assigned_partition_id);
+  }
+  if (type == kRequirementAssertDefaultSpecID) {
+    ICEBERG_ASSIGN_OR_RAISE(auto spec_id, GetJsonValue<int32_t>(json, kDefaultSpecId));
+    return std::make_unique<table::AssertDefaultSpecID>(spec_id);
+  }
+  if (type == kRequirementAssertDefaultSortOrderID) {
+    ICEBERG_ASSIGN_OR_RAISE(auto sort_order_id,
+                            GetJsonValue<int32_t>(json, kDefaultSortOrderId));
+    return std::make_unique<table::AssertDefaultSortOrderID>(sort_order_id);
+  }
+
+  return JsonParseError("Unknown table requirement type: {}", type);
+}
+
 }  // namespace iceberg
diff --git a/src/iceberg/json_internal.h b/src/iceberg/json_internal.h
index 8ca0a676e..d55252ca0 100644
--- a/src/iceberg/json_internal.h
+++ b/src/iceberg/json_internal.h
@@ -77,43 +77,43 @@ ICEBERG_EXPORT Result<std::unique_ptr<SortOrder>> SortOrderFromJson(
 
 /// \brief Convert an Iceberg Schema to JSON.
 ///
-/// \param[in] schema The Iceberg schema to convert.
+/// \param schema The Iceberg schema to convert.
 /// \return The JSON representation of the schema.
 ICEBERG_EXPORT nlohmann::json ToJson(const Schema& schema);
 
 /// \brief Convert an Iceberg Schema to JSON.
 ///
-/// \param[in] schema The Iceberg schema to convert.
+/// \param schema The Iceberg schema to convert.
 /// \return The JSON string of the schema.
 ICEBERG_EXPORT Result<std::string> ToJsonString(const Schema& schema);
 
 /// \brief Convert JSON to an Iceberg Schema.
 ///
-/// \param[in] json The JSON representation of the schema.
+/// \param json The JSON representation of the schema.
 /// \return The Iceberg schema or an error if the conversion fails.
 ICEBERG_EXPORT Result<std::unique_ptr<Schema>> SchemaFromJson(const nlohmann::json& json);
 
 /// \brief Convert an Iceberg Type to JSON.
 ///
-/// \param[in] type The Iceberg type to convert.
+/// \param type The Iceberg type to convert.
 /// \return The JSON representation of the type.
 ICEBERG_EXPORT nlohmann::json ToJson(const Type& type);
 
 /// \brief Convert JSON to an Iceberg Type.
 ///
-/// \param[in] json The JSON representation of the type.
+/// \param json The JSON representation of the type.
 /// \return The Iceberg type or an error if the conversion fails.
 ICEBERG_EXPORT Result<std::unique_ptr<Type>> TypeFromJson(const nlohmann::json& json);
 
 /// \brief Convert an Iceberg SchemaField to JSON.
 ///
-/// \param[in] field The Iceberg field to convert.
+/// \param field The Iceberg field to convert.
 /// \return The JSON representation of the field.
 ICEBERG_EXPORT nlohmann::json ToJson(const SchemaField& field);
 
 /// \brief Convert JSON to an Iceberg SchemaField.
 ///
-/// \param[in] json The JSON representation of the field.
+/// \param json The JSON representation of the field.
 /// \return The Iceberg field or an error if the conversion fails.
 ICEBERG_EXPORT Result<std::unique_ptr<SchemaField>> FieldFromJson(
     const nlohmann::json& json);
@@ -185,26 +185,26 @@ ICEBERG_EXPORT Result<std::unique_ptr<PartitionSpec>> PartitionSpecFromJson(
 
 /// \brief Serializes a `SnapshotRef` object to JSON.
 ///
-/// \param[in] snapshot_ref The `SnapshotRef` object to be serialized.
+/// \param snapshot_ref The `SnapshotRef` object to be serialized.
 /// \return A JSON object representing the `SnapshotRef`.
 ICEBERG_EXPORT nlohmann::json ToJson(const SnapshotRef& snapshot_ref);
 
 /// \brief Deserializes a JSON object into a `SnapshotRef` object.
 ///
-/// \param[in] json The JSON object representing a `SnapshotRef`.
+/// \param json The JSON object representing a `SnapshotRef`.
 /// \return A `SnapshotRef` object or an error if the conversion fails.
 ICEBERG_EXPORT Result<std::unique_ptr<SnapshotRef>> SnapshotRefFromJson(
     const nlohmann::json& json);
 
 /// \brief Serializes a `Snapshot` object to JSON.
 ///
-/// \param[in] snapshot The `Snapshot` object to be serialized.
+/// \param snapshot The `Snapshot` object to be serialized.
 /// \return A JSON object representing the `snapshot`.
 ICEBERG_EXPORT nlohmann::json ToJson(const Snapshot& snapshot);
 
 /// \brief Deserializes a JSON object into a `Snapshot` object.
 ///
-/// \param[in] json The JSON representation of the snapshot.
+/// \param json The JSON representation of the snapshot.
 /// \return A `Snapshot` object or an error if the conversion fails.
 ICEBERG_EXPORT Result<std::unique_ptr<Snapshot>> SnapshotFromJson(
     const nlohmann::json& json);
@@ -295,66 +295,92 @@ ICEBERG_EXPORT Result<std::string> ToJsonString(const nlohmann::json& json);
 
 /// \brief Serializes a `MappedField` object to JSON.
 ///
-/// \param[in] field The `MappedField` object to be serialized.
+/// \param field The `MappedField` object to be serialized.
 /// \return A JSON object representing the `MappedField`.
 ICEBERG_EXPORT nlohmann::json ToJson(const MappedField& field);
 
 /// \brief Deserializes a JSON object into a `MappedField` object.
 ///
-/// \param[in] json The JSON object representing a `MappedField`.
+/// \param json The JSON object representing a `MappedField`.
 /// \return A `MappedField` object or an error if the conversion fails.
 ICEBERG_EXPORT Result<MappedField> MappedFieldFromJson(const nlohmann::json& json);
 
 /// \brief Serializes a `MappedFields` object to JSON.
 ///
-/// \param[in] mapped_fields The `MappedFields` object to be serialized.
+/// \param mapped_fields The `MappedFields` object to be serialized.
 /// \return A JSON object representing the `MappedFields`.
 ICEBERG_EXPORT nlohmann::json ToJson(const MappedFields& mapped_fields);
 
 /// \brief Deserializes a JSON object into a `MappedFields` object.
 ///
-/// \param[in] json The JSON object representing a `MappedFields`.
+/// \param json The JSON object representing a `MappedFields`.
 /// \return A `MappedFields` object or an error if the conversion fails.
 ICEBERG_EXPORT Result<std::unique_ptr<MappedFields>> MappedFieldsFromJson(
     const nlohmann::json& json);
 
 /// \brief Serializes a `NameMapping` object to JSON.
 ///
-/// \param[in] name_mapping The `NameMapping` object to be serialized.
+/// \param name_mapping The `NameMapping` object to be serialized.
 /// \return A JSON object representing the `NameMapping`.
 ICEBERG_EXPORT nlohmann::json ToJson(const NameMapping& name_mapping);
 
 /// \brief Deserializes a JSON object into a `NameMapping` object.
 ///
-/// \param[in] json The JSON object representing a `NameMapping`.
+/// \param json The JSON object representing a `NameMapping`.
 /// \return A `NameMapping` object or an error if the conversion fails.
 ICEBERG_EXPORT Result<std::unique_ptr<NameMapping>> NameMappingFromJson(
     const nlohmann::json& json);
 
 /// \brief Serializes a `TableIdentifier` object to JSON.
 ///
-/// \param[in] identifier The `TableIdentifier` object to be serialized.
+/// \param identifier The `TableIdentifier` object to be serialized.
 /// \return A JSON object representing the `TableIdentifier` in the form of key-value
 /// pairs.
 ICEBERG_EXPORT nlohmann::json ToJson(const TableIdentifier& identifier);
 
 /// \brief Deserializes a JSON object into a `TableIdentifier` object.
 ///
-/// \param[in] json The JSON object representing a `TableIdentifier`.
+/// \param json The JSON object representing a `TableIdentifier`.
 /// \return A `TableIdentifier` object or an error if the conversion fails.
 ICEBERG_EXPORT Result<TableIdentifier> TableIdentifierFromJson(
     const nlohmann::json& json);
 
 /// \brief Serializes a `Namespace` object to JSON.
 ///
-/// \param[in] ns The `Namespace` object to be serialized.
+/// \param ns The `Namespace` object to be serialized.
 /// \return A JSON array representing the namespace levels.
 ICEBERG_EXPORT nlohmann::json ToJson(const Namespace& ns);
 
 /// \brief Deserializes a JSON array into a `Namespace` object.
 ///
-/// \param[in] json The JSON array representing a `Namespace`.
+/// \param json The JSON array representing a `Namespace`.
 /// \return A `Namespace` object or an error if the conversion fails.
 ICEBERG_EXPORT Result<Namespace> NamespaceFromJson(const nlohmann::json& json);
 
+/// \brief Serializes a `TableUpdate` object to JSON.
+///
+/// \param update The `TableUpdate` object to be serialized.
+/// \return A JSON object representing the `TableUpdate`.
+ICEBERG_EXPORT nlohmann::json ToJson(const TableUpdate& update);
+
+/// \brief Deserializes a JSON object into a `TableUpdate` object.
+///
+/// \param json The JSON object representing a `TableUpdate`.
+/// \return A `TableUpdate` object or an error if the conversion fails.
+ICEBERG_EXPORT Result<std::unique_ptr<TableUpdate>> TableUpdateFromJson(
+    const nlohmann::json& json);
+
+/// \brief Serializes a `TableRequirement` object to JSON.
+///
+/// \param requirement The `TableRequirement` object to be serialized.
+/// \return A JSON object representing the `TableRequirement`.
+ICEBERG_EXPORT nlohmann::json ToJson(const TableRequirement& requirement);
+
+/// \brief Deserializes a JSON object into a `TableRequirement` object.
+///
+/// \param json The JSON object representing a `TableRequirement`.
+/// \return A `TableRequirement` object or an error if the conversion fails.
+ICEBERG_EXPORT Result<std::unique_ptr<TableRequirement>> TableRequirementFromJson(
+    const nlohmann::json& json);
+
 }  // namespace iceberg
diff --git a/src/iceberg/table_requirement.h b/src/iceberg/table_requirement.h
index 82779aec9..9b668d4a6 100644
--- a/src/iceberg/table_requirement.h
+++ b/src/iceberg/table_requirement.h
@@ -32,6 +32,7 @@
 #include "iceberg/iceberg_export.h"
 #include "iceberg/result.h"
 #include "iceberg/type_fwd.h"
+#include "iceberg/util/checked_cast.h"
 
 namespace iceberg {
 
@@ -63,6 +64,22 @@ class ICEBERG_EXPORT TableRequirement {
   /// \param base The base table metadata to validate against (may be nullptr)
   /// \return Status indicating success or failure with error details
   virtual Status Validate(const TableMetadata* base) const = 0;
+
+  /// \brief Check equality with another TableRequirement
+  ///
+  /// \param other The requirement to compare with
+  /// \return true if the requirements are equal, false otherwise
+  virtual bool Equals(const TableRequirement& other) const = 0;
+
+  /// \brief Create a deep copy of this requirement
+  ///
+  /// \return A unique_ptr to a new TableRequirement that is a copy of this one
+  virtual std::unique_ptr<TableRequirement> Clone() const = 0;
+
+  /// \brief Compare two TableRequirement instances for equality
+  friend bool operator==(const TableRequirement& lhs, const TableRequirement& rhs) {
+    return lhs.Equals(rhs);
+  }
 };
 
 namespace table {
@@ -78,6 +95,14 @@ class ICEBERG_EXPORT AssertDoesNotExist : public TableRequirement {
   Kind kind() const override { return Kind::kAssertDoesNotExist; }
 
   Status Validate(const TableMetadata* base) const override;
+
+  bool Equals(const TableRequirement& other) const override {
+    return other.kind() == Kind::kAssertDoesNotExist;
+  }
+
+  std::unique_ptr<TableRequirement> Clone() const override {
+    return std::make_unique<AssertDoesNotExist>();
+  }
 };
 
 /// \brief Requirement that the table UUID matches the expected value
@@ -94,6 +119,18 @@ class ICEBERG_EXPORT AssertUUID : public TableRequirement {
 
   Status Validate(const TableMetadata* base) const override;
 
+  bool Equals(const TableRequirement& other) const override {
+    if (other.kind() != Kind::kAssertUUID) {
+      return false;
+    }
+    const auto& other_uuid = internal::checked_cast<const AssertUUID&>(other);
+    return uuid_ == other_uuid.uuid_;
+  }
+
+  std::unique_ptr<TableRequirement> Clone() const override {
+    return std::make_unique<AssertUUID>(uuid_);
+  }
+
  private:
   std::string uuid_;
 };
@@ -116,6 +153,18 @@ class ICEBERG_EXPORT AssertRefSnapshotID : public TableRequirement {
 
   Status Validate(const TableMetadata* base) const override;
 
+  bool Equals(const TableRequirement& other) const override {
+    if (other.kind() != Kind::kAssertRefSnapshotID) {
+      return false;
+    }
+    const auto& other_ref = internal::checked_cast<const AssertRefSnapshotID&>(other);
+    return ref_name_ == other_ref.ref_name_ && snapshot_id_ == other_ref.snapshot_id_;
+  }
+
+  std::unique_ptr<TableRequirement> Clone() const override {
+    return std::make_unique<AssertRefSnapshotID>(ref_name_, snapshot_id_);
+  }
+
  private:
   std::string ref_name_;
   std::optional<int64_t> snapshot_id_;
@@ -136,6 +185,19 @@ class ICEBERG_EXPORT AssertLastAssignedFieldId : public TableRequirement {
 
   Status Validate(const TableMetadata* base) const override;
 
+  bool Equals(const TableRequirement& other) const override {
+    if (other.kind() != Kind::kAssertLastAssignedFieldId) {
+      return false;
+    }
+    const auto& other_field =
+        internal::checked_cast<const AssertLastAssignedFieldId&>(other);
+    return last_assigned_field_id_ == other_field.last_assigned_field_id_;
+  }
+
+  std::unique_ptr<TableRequirement> Clone() const override {
+    return std::make_unique<AssertLastAssignedFieldId>(last_assigned_field_id_);
+  }
+
  private:
   int32_t last_assigned_field_id_;
 };
@@ -154,6 +216,19 @@ class ICEBERG_EXPORT AssertCurrentSchemaID : public TableRequirement {
 
   Status Validate(const TableMetadata* base) const override;
 
+  bool Equals(const TableRequirement& other) const override {
+    if (other.kind() != Kind::kAssertCurrentSchemaID) {
+      return false;
+    }
+    const auto& other_schema =
+        internal::checked_cast<const AssertCurrentSchemaID&>(other);
+    return schema_id_ == other_schema.schema_id_;
+  }
+
+  std::unique_ptr<TableRequirement> Clone() const override {
+    return std::make_unique<AssertCurrentSchemaID>(schema_id_);
+  }
+
  private:
   int32_t schema_id_;
 };
@@ -173,6 +248,19 @@ class ICEBERG_EXPORT AssertLastAssignedPartitionId : public TableRequirement {
 
   Status Validate(const TableMetadata* base) const override;
 
+  bool Equals(const TableRequirement& other) const override {
+    if (other.kind() != Kind::kAssertLastAssignedPartitionId) {
+      return false;
+    }
+    const auto& other_partition =
+        internal::checked_cast<const AssertLastAssignedPartitionId&>(other);
+    return last_assigned_partition_id_ == other_partition.last_assigned_partition_id_;
+  }
+
+  std::unique_ptr<TableRequirement> Clone() const override {
+    return std::make_unique<AssertLastAssignedPartitionId>(last_assigned_partition_id_);
+  }
+
  private:
   int32_t last_assigned_partition_id_;
 };
@@ -191,6 +279,18 @@ class ICEBERG_EXPORT AssertDefaultSpecID : public TableRequirement {
 
   Status Validate(const TableMetadata* base) const override;
 
+  bool Equals(const TableRequirement& other) const override {
+    if (other.kind() != Kind::kAssertDefaultSpecID) {
+      return false;
+    }
+    const auto& other_spec = internal::checked_cast<const AssertDefaultSpecID&>(other);
+    return spec_id_ == other_spec.spec_id_;
+  }
+
+  std::unique_ptr<TableRequirement> Clone() const override {
+    return std::make_unique<AssertDefaultSpecID>(spec_id_);
+  }
+
  private:
   int32_t spec_id_;
 };
@@ -210,6 +310,19 @@ class ICEBERG_EXPORT AssertDefaultSortOrderID : public TableRequirement {
 
   Status Validate(const TableMetadata* base) const override;
 
+  bool Equals(const TableRequirement& other) const override {
+    if (other.kind() != Kind::kAssertDefaultSortOrderID) {
+      return false;
+    }
+    const auto& other_sort =
+        internal::checked_cast<const AssertDefaultSortOrderID&>(other);
+    return sort_order_id_ == other_sort.sort_order_id_;
+  }
+
+  std::unique_ptr<TableRequirement> Clone() const override {
+    return std::make_unique<AssertDefaultSortOrderID>(sort_order_id_);
+  }
+
  private:
   int32_t sort_order_id_;
 };
diff --git a/src/iceberg/table_update.cc b/src/iceberg/table_update.cc
index 91578977c..38ce0fbc9 100644
--- a/src/iceberg/table_update.cc
+++ b/src/iceberg/table_update.cc
@@ -20,6 +20,8 @@
 #include "iceberg/table_update.h"
 
 #include "iceberg/exception.h"
+#include "iceberg/schema.h"
+#include "iceberg/sort_order.h"
 #include "iceberg/table_metadata.h"
 #include "iceberg/table_requirements.h"
 
@@ -39,6 +41,18 @@ void AssignUUID::GenerateRequirements(TableUpdateContext& context) const {
   // AssignUUID does not generate additional requirements.
 }
 
+bool AssignUUID::Equals(const TableUpdate& other) const {
+  if (other.kind() != Kind::kAssignUUID) {
+    return false;
+  }
+  const auto& other_assign = static_cast<const AssignUUID&>(other);
+  return uuid_ == other_assign.uuid_;
+}
+
+std::unique_ptr<TableUpdate> AssignUUID::Clone() const {
+  return std::make_unique<AssignUUID>(uuid_);
+}
+
 // UpgradeFormatVersion
 
 void UpgradeFormatVersion::ApplyTo(TableMetadataBuilder& builder) const {
@@ -49,6 +63,18 @@ void UpgradeFormatVersion::GenerateRequirements(TableUpdateContext& context) con
   // UpgradeFormatVersion doesn't generate any requirements
 }
 
+bool UpgradeFormatVersion::Equals(const TableUpdate& other) const {
+  if (other.kind() != Kind::kUpgradeFormatVersion) {
+    return false;
+  }
+  const auto& other_upgrade = static_cast<const UpgradeFormatVersion&>(other);
+  return format_version_ == other_upgrade.format_version_;
+}
+
+std::unique_ptr<TableUpdate> UpgradeFormatVersion::Clone() const {
+  return std::make_unique<UpgradeFormatVersion>(format_version_);
+}
+
 // AddSchema
 
 void AddSchema::ApplyTo(TableMetadataBuilder& builder) const {
@@ -59,6 +85,24 @@ void AddSchema::GenerateRequirements(TableUpdateContext& context) const {
   context.RequireLastAssignedFieldIdUnchanged();
 }
 
+bool AddSchema::Equals(const TableUpdate& other) const {
+  if (other.kind() != Kind::kAddSchema) {
+    return false;
+  }
+  const auto& other_add = internal::checked_cast<const AddSchema&>(other);
+  if (!schema_ != !other_add.schema_) {
+    return false;
+  }
+  if (schema_ && !(*schema_ == *other_add.schema_)) {
+    return false;
+  }
+  return last_column_id_ == other_add.last_column_id_;
+}
+
+std::unique_ptr<TableUpdate> AddSchema::Clone() const {
+  return std::make_unique<AddSchema>(schema_, last_column_id_);
+}
+
 // SetCurrentSchema
 
 void SetCurrentSchema::ApplyTo(TableMetadataBuilder& builder) const {
@@ -69,6 +113,18 @@ void SetCurrentSchema::GenerateRequirements(TableUpdateContext& context) const {
   context.RequireCurrentSchemaIdUnchanged();
 }
 
+bool SetCurrentSchema::Equals(const TableUpdate& other) const {
+  if (other.kind() != Kind::kSetCurrentSchema) {
+    return false;
+  }
+  const auto& other_set = static_cast<const SetCurrentSchema&>(other);
+  return schema_id_ == other_set.schema_id_;
+}
+
+std::unique_ptr<TableUpdate> SetCurrentSchema::Clone() const {
+  return std::make_unique<SetCurrentSchema>(schema_id_);
+}
+
 // AddPartitionSpec
 
 void AddPartitionSpec::ApplyTo(TableMetadataBuilder& builder) const {
@@ -79,6 +135,24 @@ void AddPartitionSpec::GenerateRequirements(TableUpdateContext& context) const {
   context.RequireLastAssignedPartitionIdUnchanged();
 }
 
+bool AddPartitionSpec::Equals(const TableUpdate& other) const {
+  if (other.kind() != Kind::kAddPartitionSpec) {
+    return false;
+  }
+  const auto& other_add = internal::checked_cast<const AddPartitionSpec&>(other);
+  if (!spec_ != !other_add.spec_) {
+    return false;
+  }
+  if (spec_ && *spec_ != *other_add.spec_) {
+    return false;
+  }
+  return true;
+}
+
+std::unique_ptr<TableUpdate> AddPartitionSpec::Clone() const {
+  return std::make_unique<AddPartitionSpec>(spec_);
+}
+
 // SetDefaultPartitionSpec
 
 void SetDefaultPartitionSpec::ApplyTo(TableMetadataBuilder& builder) const {
@@ -89,6 +163,18 @@ void SetDefaultPartitionSpec::GenerateRequirements(TableUpdateContext& context)
   context.RequireDefaultSpecIdUnchanged();
 }
 
+bool SetDefaultPartitionSpec::Equals(const TableUpdate& other) const {
+  if (other.kind() != Kind::kSetDefaultPartitionSpec) {
+    return false;
+  }
+  const auto& other_set = static_cast<const SetDefaultPartitionSpec&>(other);
+  return spec_id_ == other_set.spec_id_;
+}
+
+std::unique_ptr<TableUpdate> SetDefaultPartitionSpec::Clone() const {
+  return std::make_unique<SetDefaultPartitionSpec>(spec_id_);
+}
+
 // RemovePartitionSpecs
 
 void RemovePartitionSpecs::ApplyTo(TableMetadataBuilder& builder) const {
@@ -100,6 +186,18 @@ void RemovePartitionSpecs::GenerateRequirements(TableUpdateContext& context) con
   context.RequireNoBranchesChanged();
 }
 
+bool RemovePartitionSpecs::Equals(const TableUpdate& other) const {
+  if (other.kind() != Kind::kRemovePartitionSpecs) {
+    return false;
+  }
+  const auto& other_remove = static_cast<const RemovePartitionSpecs&>(other);
+  return spec_ids_ == other_remove.spec_ids_;
+}
+
+std::unique_ptr<TableUpdate> RemovePartitionSpecs::Clone() const {
+  return std::make_unique<RemovePartitionSpecs>(spec_ids_);
+}
+
 // RemoveSchemas
 
 void RemoveSchemas::ApplyTo(TableMetadataBuilder& builder) const {
@@ -111,6 +209,18 @@ void RemoveSchemas::GenerateRequirements(TableUpdateContext& context) const {
   context.RequireNoBranchesChanged();
 }
 
+bool RemoveSchemas::Equals(const TableUpdate& other) const {
+  if (other.kind() != Kind::kRemoveSchemas) {
+    return false;
+  }
+  const auto& other_remove = static_cast<const RemoveSchemas&>(other);
+  return schema_ids_ == other_remove.schema_ids_;
+}
+
+std::unique_ptr<TableUpdate> RemoveSchemas::Clone() const {
+  return std::make_unique<RemoveSchemas>(schema_ids_);
+}
+
 // AddSortOrder
 
 void AddSortOrder::ApplyTo(TableMetadataBuilder& builder) const {
@@ -121,6 +231,24 @@ void AddSortOrder::GenerateRequirements(TableUpdateContext& context) const {
   // AddSortOrder doesn't generate any requirements
 }
 
+bool AddSortOrder::Equals(const TableUpdate& other) const {
+  if (other.kind() != Kind::kAddSortOrder) {
+    return false;
+  }
+  const auto& other_add = internal::checked_cast<const AddSortOrder&>(other);
+  if (!sort_order_ != !other_add.sort_order_) {
+    return false;
+  }
+  if (sort_order_ && !(*sort_order_ == *other_add.sort_order_)) {
+    return false;
+  }
+  return true;
+}
+
+std::unique_ptr<TableUpdate> AddSortOrder::Clone() const {
+  return std::make_unique<AddSortOrder>(sort_order_);
+}
+
 // SetDefaultSortOrder
 
 void SetDefaultSortOrder::ApplyTo(TableMetadataBuilder& builder) const {
@@ -131,6 +259,18 @@ void SetDefaultSortOrder::GenerateRequirements(TableUpdateContext& context) cons
   context.RequireDefaultSortOrderIdUnchanged();
 }
 
+bool SetDefaultSortOrder::Equals(const TableUpdate& other) const {
+  if (other.kind() != Kind::kSetDefaultSortOrder) {
+    return false;
+  }
+  const auto& other_set = static_cast<const SetDefaultSortOrder&>(other);
+  return sort_order_id_ == other_set.sort_order_id_;
+}
+
+std::unique_ptr<TableUpdate> SetDefaultSortOrder::Clone() const {
+  return std::make_unique<SetDefaultSortOrder>(sort_order_id_);
+}
+
 // AddSnapshot
 
 void AddSnapshot::ApplyTo(TableMetadataBuilder& builder) const {
@@ -141,6 +281,24 @@ void AddSnapshot::GenerateRequirements(TableUpdateContext& context) const {
   // AddSnapshot doesn't generate any requirements
 }
 
+bool AddSnapshot::Equals(const TableUpdate& other) const {
+  if (other.kind() != Kind::kAddSnapshot) {
+    return false;
+  }
+  const auto& other_add = internal::checked_cast<const AddSnapshot&>(other);
+  if (!snapshot_ != !other_add.snapshot_) {
+    return false;
+  }
+  if (snapshot_ && *snapshot_ != *other_add.snapshot_) {
+    return false;
+  }
+  return true;
+}
+
+std::unique_ptr<TableUpdate> AddSnapshot::Clone() const {
+  return std::make_unique<AddSnapshot>(snapshot_);
+}
+
 // RemoveSnapshots
 
 void RemoveSnapshots::ApplyTo(TableMetadataBuilder& builder) const {}
@@ -149,6 +307,18 @@ void RemoveSnapshots::GenerateRequirements(TableUpdateContext& context) const {
   // RemoveSnapshots doesn't generate any requirements
 }
 
+bool RemoveSnapshots::Equals(const TableUpdate& other) const {
+  if (other.kind() != Kind::kRemoveSnapshots) {
+    return false;
+  }
+  const auto& other_remove = static_cast<const RemoveSnapshots&>(other);
+  return snapshot_ids_ == other_remove.snapshot_ids_;
+}
+
+std::unique_ptr<TableUpdate> RemoveSnapshots::Clone() const {
+  return std::make_unique<RemoveSnapshots>(snapshot_ids_);
+}
+
 // RemoveSnapshotRef
 
 void RemoveSnapshotRef::ApplyTo(TableMetadataBuilder& builder) const {
@@ -159,6 +329,18 @@ void RemoveSnapshotRef::GenerateRequirements(TableUpdateContext& context) const
   // RemoveSnapshotRef doesn't generate any requirements
 }
 
+bool RemoveSnapshotRef::Equals(const TableUpdate& other) const {
+  if (other.kind() != Kind::kRemoveSnapshotRef) {
+    return false;
+  }
+  const auto& other_remove = static_cast<const RemoveSnapshotRef&>(other);
+  return ref_name_ == other_remove.ref_name_;
+}
+
+std::unique_ptr<TableUpdate> RemoveSnapshotRef::Clone() const {
+  return std::make_unique<RemoveSnapshotRef>(ref_name_);
+}
+
 // SetSnapshotRef
 
 void SetSnapshotRef::ApplyTo(TableMetadataBuilder& builder) const {
@@ -178,6 +360,24 @@ void SetSnapshotRef::GenerateRequirements(TableUpdateContext& context) const {
   }
 }
 
+bool SetSnapshotRef::Equals(const TableUpdate& other) const {
+  if (other.kind() != Kind::kSetSnapshotRef) {
+    return false;
+  }
+  const auto& other_set = static_cast<const SetSnapshotRef&>(other);
+  return ref_name_ == other_set.ref_name_ && snapshot_id_ == other_set.snapshot_id_ &&
+         type_ == other_set.type_ &&
+         min_snapshots_to_keep_ == other_set.min_snapshots_to_keep_ &&
+         max_snapshot_age_ms_ == other_set.max_snapshot_age_ms_ &&
+         max_ref_age_ms_ == other_set.max_ref_age_ms_;
+}
+
+std::unique_ptr<TableUpdate> SetSnapshotRef::Clone() const {
+  return std::make_unique<SetSnapshotRef>(ref_name_, snapshot_id_, type_,
+                                          min_snapshots_to_keep_, max_snapshot_age_ms_,
+                                          max_ref_age_ms_);
+}
+
 // SetProperties
 
 void SetProperties::ApplyTo(TableMetadataBuilder& builder) const {
@@ -188,6 +388,18 @@ void SetProperties::GenerateRequirements(TableUpdateContext& context) const {
   // SetProperties doesn't generate any requirements
 }
 
+bool SetProperties::Equals(const TableUpdate& other) const {
+  if (other.kind() != Kind::kSetProperties) {
+    return false;
+  }
+  const auto& other_set = static_cast<const SetProperties&>(other);
+  return updated_ == other_set.updated_;
+}
+
+std::unique_ptr<TableUpdate> SetProperties::Clone() const {
+  return std::make_unique<SetProperties>(updated_);
+}
+
 // RemoveProperties
 
 void RemoveProperties::ApplyTo(TableMetadataBuilder& builder) const {
@@ -198,6 +410,18 @@ void RemoveProperties::GenerateRequirements(TableUpdateContext& context) const {
   // RemoveProperties doesn't generate any requirements
 }
 
+bool RemoveProperties::Equals(const TableUpdate& other) const {
+  if (other.kind() != Kind::kRemoveProperties) {
+    return false;
+  }
+  const auto& other_remove = static_cast<const RemoveProperties&>(other);
+  return removed_ == other_remove.removed_;
+}
+
+std::unique_ptr<TableUpdate> RemoveProperties::Clone() const {
+  return std::make_unique<RemoveProperties>(removed_);
+}
+
 // SetLocation
 
 void SetLocation::ApplyTo(TableMetadataBuilder& builder) const {
@@ -208,4 +432,16 @@ void SetLocation::GenerateRequirements(TableUpdateContext& context) const {
   // SetLocation doesn't generate any requirements
 }
 
+bool SetLocation::Equals(const TableUpdate& other) const {
+  if (other.kind() != Kind::kSetLocation) {
+    return false;
+  }
+  const auto& other_set = static_cast<const SetLocation&>(other);
+  return location_ == other_set.location_;
+}
+
+std::unique_ptr<TableUpdate> SetLocation::Clone() const {
+  return std::make_unique<SetLocation>(location_);
+}
+
 }  // namespace iceberg::table
diff --git a/src/iceberg/table_update.h b/src/iceberg/table_update.h
index a6bdb9e5d..875243195 100644
--- a/src/iceberg/table_update.h
+++ b/src/iceberg/table_update.h
@@ -83,6 +83,22 @@ class ICEBERG_EXPORT TableUpdate {
   ///
   /// \param context The context containing base metadata and operation state
   virtual void GenerateRequirements(TableUpdateContext& context) const = 0;
+
+  /// \brief Check equality with another TableUpdate
+  ///
+  /// \param other The update to compare with
+  /// \return true if the updates are equal, false otherwise
+  virtual bool Equals(const TableUpdate& other) const = 0;
+
+  /// \brief Create a deep copy of this update
+  ///
+  /// \return A unique_ptr to a new TableUpdate that is a copy of this one
+  virtual std::unique_ptr<TableUpdate> Clone() const = 0;
+
+  /// \brief Compare two TableUpdate instances for equality
+  friend bool operator==(const TableUpdate& lhs, const TableUpdate& rhs) {
+    return lhs.Equals(rhs);
+  }
 };
 
 namespace table {
@@ -100,6 +116,10 @@ class ICEBERG_EXPORT AssignUUID : public TableUpdate {
 
   Kind kind() const override { return Kind::kAssignUUID; }
 
+  bool Equals(const TableUpdate& other) const override;
+
+  std::unique_ptr<TableUpdate> Clone() const override;
+
  private:
   std::string uuid_;
 };
@@ -118,6 +138,10 @@ class ICEBERG_EXPORT UpgradeFormatVersion : public TableUpdate {
 
   Kind kind() const override { return Kind::kUpgradeFormatVersion; }
 
+  bool Equals(const TableUpdate& other) const override;
+
+  std::unique_ptr<TableUpdate> Clone() const override;
+
  private:
   int8_t format_version_;
 };
@@ -138,6 +162,10 @@ class ICEBERG_EXPORT AddSchema : public TableUpdate {
 
   Kind kind() const override { return Kind::kAddSchema; }
 
+  bool Equals(const TableUpdate& other) const override;
+
+  std::unique_ptr<TableUpdate> Clone() const override;
+
  private:
   std::shared_ptr<Schema> schema_;
   int32_t last_column_id_;
@@ -156,6 +184,10 @@ class ICEBERG_EXPORT SetCurrentSchema : public TableUpdate {
 
   Kind kind() const override { return Kind::kSetCurrentSchema; }
 
+  bool Equals(const TableUpdate& other) const override;
+
+  std::unique_ptr<TableUpdate> Clone() const override;
+
  private:
   int32_t schema_id_;
 };
@@ -174,6 +206,10 @@ class ICEBERG_EXPORT AddPartitionSpec : public TableUpdate {
 
   Kind kind() const override { return Kind::kAddPartitionSpec; }
 
+  bool Equals(const TableUpdate& other) const override;
+
+  std::unique_ptr<TableUpdate> Clone() const override;
+
  private:
   std::shared_ptr<PartitionSpec> spec_;
 };
@@ -191,6 +227,10 @@ class ICEBERG_EXPORT SetDefaultPartitionSpec : public TableUpdate {
 
   Kind kind() const override { return Kind::kSetDefaultPartitionSpec; }
 
+  bool Equals(const TableUpdate& other) const override;
+
+  std::unique_ptr<TableUpdate> Clone() const override;
+
  private:
   int32_t spec_id_;
 };
@@ -209,6 +249,10 @@ class ICEBERG_EXPORT RemovePartitionSpecs : public TableUpdate {
 
   Kind kind() const override { return Kind::kRemovePartitionSpecs; }
 
+  bool Equals(const TableUpdate& other) const override;
+
+  std::unique_ptr<TableUpdate> Clone() const override;
+
  private:
   std::vector<int32_t> spec_ids_;
 };
@@ -227,6 +271,10 @@ class ICEBERG_EXPORT RemoveSchemas : public TableUpdate {
 
   Kind kind() const override { return Kind::kRemoveSchemas; }
 
+  bool Equals(const TableUpdate& other) const override;
+
+  std::unique_ptr<TableUpdate> Clone() const override;
+
  private:
   std::unordered_set<int32_t> schema_ids_;
 };
@@ -245,6 +293,10 @@ class ICEBERG_EXPORT AddSortOrder : public TableUpdate {
 
   Kind kind() const override { return Kind::kAddSortOrder; }
 
+  bool Equals(const TableUpdate& other) const override;
+
+  std::unique_ptr<TableUpdate> Clone() const override;
+
  private:
   std::shared_ptr<SortOrder> sort_order_;
 };
@@ -262,6 +314,10 @@ class ICEBERG_EXPORT SetDefaultSortOrder : public TableUpdate {
 
   Kind kind() const override { return Kind::kSetDefaultSortOrder; }
 
+  bool Equals(const TableUpdate& other) const override;
+
+  std::unique_ptr<TableUpdate> Clone() const override;
+
  private:
   int32_t sort_order_id_;
 };
@@ -280,6 +336,10 @@ class ICEBERG_EXPORT AddSnapshot : public TableUpdate {
 
   Kind kind() const override { return Kind::kAddSnapshot; }
 
+  bool Equals(const TableUpdate& other) const override;
+
+  std::unique_ptr<TableUpdate> Clone() const override;
+
  private:
   std::shared_ptr<Snapshot> snapshot_;
 };
@@ -298,6 +358,10 @@ class ICEBERG_EXPORT RemoveSnapshots : public TableUpdate {
 
   Kind kind() const override { return Kind::kRemoveSnapshots; }
 
+  bool Equals(const TableUpdate& other) const override;
+
+  std::unique_ptr<TableUpdate> Clone() const override;
+
  private:
   std::vector<int64_t> snapshot_ids_;
 };
@@ -315,6 +379,10 @@ class ICEBERG_EXPORT RemoveSnapshotRef : public TableUpdate {
 
   Kind kind() const override { return Kind::kRemoveSnapshotRef; }
 
+  bool Equals(const TableUpdate& other) const override;
+
+  std::unique_ptr<TableUpdate> Clone() const override;
+
  private:
   std::string ref_name_;
 };
@@ -350,6 +418,10 @@ class ICEBERG_EXPORT SetSnapshotRef : public TableUpdate {
 
   Kind kind() const override { return Kind::kSetSnapshotRef; }
 
+  bool Equals(const TableUpdate& other) const override;
+
+  std::unique_ptr<TableUpdate> Clone() const override;
+
  private:
   std::string ref_name_;
   int64_t snapshot_id_;
@@ -373,6 +445,10 @@ class ICEBERG_EXPORT SetProperties : public TableUpdate {
 
   Kind kind() const override { return Kind::kSetProperties; }
 
+  bool Equals(const TableUpdate& other) const override;
+
+  std::unique_ptr<TableUpdate> Clone() const override;
+
  private:
   std::unordered_map<std::string, std::string> updated_;
 };
@@ -391,6 +467,10 @@ class ICEBERG_EXPORT RemoveProperties : public TableUpdate {
 
   Kind kind() const override { return Kind::kRemoveProperties; }
 
+  bool Equals(const TableUpdate& other) const override;
+
+  std::unique_ptr<TableUpdate> Clone() const override;
+
  private:
   std::unordered_set<std::string> removed_;
 };
@@ -408,6 +488,10 @@ class ICEBERG_EXPORT SetLocation : public TableUpdate {
 
   Kind kind() const override { return Kind::kSetLocation; }
 
+  bool Equals(const TableUpdate& other) const override;
+
+  std::unique_ptr<TableUpdate> Clone() const override;
+
  private:
   std::string location_;
 };
diff --git a/src/iceberg/test/CMakeLists.txt b/src/iceberg/test/CMakeLists.txt
index a32bbe4de..8c6e60948 100644
--- a/src/iceberg/test/CMakeLists.txt
+++ b/src/iceberg/test/CMakeLists.txt
@@ -49,6 +49,10 @@ function(add_iceberg_test test_name)
     target_link_libraries(${test_name} PRIVATE iceberg_static GTest::gmock_main)
   endif()
 
+  if(MSVC_TOOLCHAIN)
+    target_compile_options(${test_name} PRIVATE /bigobj)
+  endif()
+
   add_test(NAME ${test_name} COMMAND ${test_name})
 endfunction()
 
@@ -186,6 +190,9 @@ if(ICEBERG_BUILD_REST)
     target_include_directories(${test_name} PRIVATE "${CMAKE_BINARY_DIR}/iceberg/test/")
     target_sources(${test_name} PRIVATE ${ARG_SOURCES})
     target_link_libraries(${test_name} PRIVATE GTest::gmock_main iceberg_rest_static)
+    if(MSVC_TOOLCHAIN)
+      target_compile_options(${test_name} PRIVATE /bigobj)
+    endif()
     add_test(NAME ${test_name} COMMAND ${test_name})
   endfunction()
 
diff --git a/src/iceberg/test/json_internal_test.cc b/src/iceberg/test/json_internal_test.cc
index a23cc6802..f88527ff4 100644
--- a/src/iceberg/test/json_internal_test.cc
+++ b/src/iceberg/test/json_internal_test.cc
@@ -31,6 +31,8 @@
 #include "iceberg/snapshot.h"
 #include "iceberg/sort_field.h"
 #include "iceberg/sort_order.h"
+#include "iceberg/table_requirement.h"
+#include "iceberg/table_update.h"
 #include "iceberg/test/matchers.h"
 #include "iceberg/transform.h"
 #include "iceberg/util/formatter.h"  // IWYU pragma: keep
@@ -293,4 +295,340 @@ TEST(JsonInternalTest, NameMapping) {
   TestJsonConversion(*mapping, expected_json);
 }
 
+// TableUpdate JSON Serialization/Deserialization Tests
+TEST(JsonInternalTest, TableUpdateAssignUUID) {
+  table::AssignUUID update("550e8400-e29b-41d4-a716-446655440000");
+  nlohmann::json expected =
+      R"({"action":"assign-uuid","uuid":"550e8400-e29b-41d4-a716-446655440000"})"_json;
+
+  EXPECT_EQ(ToJson(update), expected);
+  auto parsed = TableUpdateFromJson(expected);
+  ASSERT_THAT(parsed, IsOk());
+  EXPECT_EQ(*internal::checked_cast<table::AssignUUID*>(parsed.value().get()), update);
+}
+
+TEST(JsonInternalTest, TableUpdateUpgradeFormatVersion) {
+  table::UpgradeFormatVersion update(2);
+  nlohmann::json expected =
+      R"({"action":"upgrade-format-version","format-version":2})"_json;
+
+  EXPECT_EQ(ToJson(update), expected);
+  auto parsed = TableUpdateFromJson(expected);
+  ASSERT_THAT(parsed, IsOk());
+  EXPECT_EQ(*internal::checked_cast<table::UpgradeFormatVersion*>(parsed.value().get()),
+            update);
+}
+
+TEST(JsonInternalTest, TableUpdateAddSchema) {
+  auto schema = std::make_shared<Schema>(
+      std::vector<SchemaField>{SchemaField(1, "id", int64(), false),
+                               SchemaField(2, "name", string(), true)},
+      /*schema_id=*/1);
+  table::AddSchema update(schema, 2);
+
+  auto json = ToJson(update);
+  EXPECT_EQ(json["action"], "add-schema");
+  EXPECT_EQ(json["last-column-id"], 2);
+  EXPECT_TRUE(json.contains("schema"));
+
+  auto parsed = TableUpdateFromJson(json);
+  ASSERT_THAT(parsed, IsOk());
+  auto* actual = internal::checked_cast<table::AddSchema*>(parsed.value().get());
+  EXPECT_EQ(actual->last_column_id(), update.last_column_id());
+  EXPECT_EQ(*actual->schema(), *update.schema());
+}
+
+TEST(JsonInternalTest, TableUpdateSetCurrentSchema) {
+  table::SetCurrentSchema update(1);
+  nlohmann::json expected = R"({"action":"set-current-schema","schema-id":1})"_json;
+
+  EXPECT_EQ(ToJson(update), expected);
+  auto parsed = TableUpdateFromJson(expected);
+  ASSERT_THAT(parsed, IsOk());
+  EXPECT_EQ(*internal::checked_cast<table::SetCurrentSchema*>(parsed.value().get()),
+            update);
+}
+
+TEST(JsonInternalTest, TableUpdateSetDefaultPartitionSpec) {
+  table::SetDefaultPartitionSpec update(2);
+  nlohmann::json expected = R"({"action":"set-default-spec","spec-id":2})"_json;
+
+  EXPECT_EQ(ToJson(update), expected);
+  auto parsed = TableUpdateFromJson(expected);
+  ASSERT_THAT(parsed, IsOk());
+  EXPECT_EQ(
+      *internal::checked_cast<table::SetDefaultPartitionSpec*>(parsed.value().get()),
+      update);
+}
+
+TEST(JsonInternalTest, TableUpdateRemovePartitionSpecs) {
+  table::RemovePartitionSpecs update({1, 2, 3});
+  nlohmann::json expected =
+      R"({"action":"remove-partition-specs","spec-ids":[1,2,3]})"_json;
+
+  EXPECT_EQ(ToJson(update), expected);
+  auto parsed = TableUpdateFromJson(expected);
+  ASSERT_THAT(parsed, IsOk());
+  EXPECT_EQ(*internal::checked_cast<table::RemovePartitionSpecs*>(parsed.value().get()),
+            update);
+}
+
+TEST(JsonInternalTest, TableUpdateRemoveSchemas) {
+  table::RemoveSchemas update({1, 2});
+
+  auto json = ToJson(update);
+  EXPECT_EQ(json["action"], "remove-schemas");
+  EXPECT_THAT(json["schema-ids"].get<std::vector<int32_t>>(),
+              testing::UnorderedElementsAre(1, 2));
+
+  auto parsed = TableUpdateFromJson(json);
+  ASSERT_THAT(parsed, IsOk());
+  EXPECT_EQ(*internal::checked_cast<table::RemoveSchemas*>(parsed.value().get()), update);
+}
+
+TEST(JsonInternalTest, TableUpdateSetDefaultSortOrder) {
+  table::SetDefaultSortOrder update(1);
+  nlohmann::json expected =
+      R"({"action":"set-default-sort-order","sort-order-id":1})"_json;
+
+  EXPECT_EQ(ToJson(update), expected);
+  auto parsed = TableUpdateFromJson(expected);
+  ASSERT_THAT(parsed, IsOk());
+  EXPECT_EQ(*internal::checked_cast<table::SetDefaultSortOrder*>(parsed.value().get()),
+            update);
+}
+
+TEST(JsonInternalTest, TableUpdateAddSnapshot) {
+  auto snapshot = std::make_shared<Snapshot>(
+      Snapshot{.snapshot_id = 123456789,
+               .parent_snapshot_id = 987654321,
+               .sequence_number = 5,
+               .timestamp_ms = TimePointMsFromUnixMs(1234567890000).value(),
+               .manifest_list = "/path/to/manifest-list.avro",
+               .summary = {{SnapshotSummaryFields::kOperation, DataOperation::kAppend}},
+               .schema_id = 1});
+  table::AddSnapshot update(snapshot);
+
+  auto json = ToJson(update);
+  EXPECT_EQ(json["action"], "add-snapshot");
+  EXPECT_TRUE(json.contains("snapshot"));
+
+  auto parsed = TableUpdateFromJson(json);
+  ASSERT_THAT(parsed, IsOk());
+  auto* actual = internal::checked_cast<table::AddSnapshot*>(parsed.value().get());
+  EXPECT_EQ(*actual->snapshot(), *update.snapshot());
+}
+
+TEST(JsonInternalTest, TableUpdateRemoveSnapshots) {
+  table::RemoveSnapshots update({111, 222, 333});
+  nlohmann::json expected =
+      R"({"action":"remove-snapshots","snapshot-ids":[111,222,333]})"_json;
+
+  EXPECT_EQ(ToJson(update), expected);
+  auto parsed = TableUpdateFromJson(expected);
+  ASSERT_THAT(parsed, IsOk());
+  EXPECT_EQ(*internal::checked_cast<table::RemoveSnapshots*>(parsed.value().get()),
+            update);
+}
+
+TEST(JsonInternalTest, TableUpdateRemoveSnapshotRef) {
+  table::RemoveSnapshotRef update("my-branch");
+  nlohmann::json expected =
+      R"({"action":"remove-snapshot-ref","ref-name":"my-branch"})"_json;
+
+  EXPECT_EQ(ToJson(update), expected);
+  auto parsed = TableUpdateFromJson(expected);
+  ASSERT_THAT(parsed, IsOk());
+  EXPECT_EQ(*internal::checked_cast<table::RemoveSnapshotRef*>(parsed.value().get()),
+            update);
+}
+
+TEST(JsonInternalTest, TableUpdateSetSnapshotRefBranch) {
+  table::SetSnapshotRef update("main", 123456789, SnapshotRefType::kBranch, 5, 86400000,
+                               604800000);
+
+  auto json = ToJson(update);
+  EXPECT_EQ(json["action"], "set-snapshot-ref");
+  EXPECT_EQ(json["ref-name"], "main");
+  EXPECT_EQ(json["snapshot-id"], 123456789);
+  EXPECT_EQ(json["type"], "branch");
+
+  auto parsed = TableUpdateFromJson(json);
+  ASSERT_THAT(parsed, IsOk());
+  EXPECT_EQ(*internal::checked_cast<table::SetSnapshotRef*>(parsed.value().get()),
+            update);
+}
+
+TEST(JsonInternalTest, TableUpdateSetSnapshotRefTag) {
+  table::SetSnapshotRef update("release-1.0", 987654321, SnapshotRefType::kTag);
+
+  auto json = ToJson(update);
+  EXPECT_EQ(json["action"], "set-snapshot-ref");
+  EXPECT_EQ(json["type"], "tag");
+
+  auto parsed = TableUpdateFromJson(json);
+  ASSERT_THAT(parsed, IsOk());
+  EXPECT_EQ(*internal::checked_cast<table::SetSnapshotRef*>(parsed.value().get()),
+            update);
+}
+
+TEST(JsonInternalTest, TableUpdateSetProperties) {
+  table::SetProperties update({{"key1", "value1"}, {"key2", "value2"}});
+
+  auto json = ToJson(update);
+  EXPECT_EQ(json["action"], "set-properties");
+  EXPECT_TRUE(json.contains("updates"));
+
+  auto parsed = TableUpdateFromJson(json);
+  ASSERT_THAT(parsed, IsOk());
+  EXPECT_EQ(*internal::checked_cast<table::SetProperties*>(parsed.value().get()), update);
+}
+
+TEST(JsonInternalTest, TableUpdateRemoveProperties) {
+  table::RemoveProperties update({"key1", "key2"});
+
+  auto json = ToJson(update);
+  EXPECT_EQ(json["action"], "remove-properties");
+  EXPECT_TRUE(json.contains("removals"));
+
+  auto parsed = TableUpdateFromJson(json);
+  ASSERT_THAT(parsed, IsOk());
+  EXPECT_EQ(*internal::checked_cast<table::RemoveProperties*>(parsed.value().get()),
+            update);
+}
+
+TEST(JsonInternalTest, TableUpdateSetLocation) {
+  table::SetLocation update("s3://bucket/warehouse/table");
+  nlohmann::json expected =
+      R"({"action":"set-location","location":"s3://bucket/warehouse/table"})"_json;
+
+  EXPECT_EQ(ToJson(update), expected);
+  auto parsed = TableUpdateFromJson(expected);
+  ASSERT_THAT(parsed, IsOk());
+  EXPECT_EQ(*internal::checked_cast<table::SetLocation*>(parsed.value().get()), update);
+}
+
+TEST(JsonInternalTest, TableUpdateUnknownAction) {
+  nlohmann::json json = R"({"action":"unknown-action"})"_json;
+  auto result = TableUpdateFromJson(json);
+  EXPECT_THAT(result, IsError(ErrorKind::kJsonParseError));
+  EXPECT_THAT(result, HasErrorMessage("Unknown table update action"));
+}
+
+// TableRequirement JSON Serialization/Deserialization Tests
+TEST(TableRequirementJsonTest, TableRequirementAssertDoesNotExist) {
+  table::AssertDoesNotExist req;
+  nlohmann::json expected = R"({"type":"assert-create"})"_json;
+
+  EXPECT_EQ(ToJson(req), expected);
+  auto parsed = TableRequirementFromJson(expected);
+  ASSERT_THAT(parsed, IsOk());
+  EXPECT_EQ(parsed.value()->kind(), TableRequirement::Kind::kAssertDoesNotExist);
+}
+
+TEST(TableRequirementJsonTest, TableRequirementAssertUUID) {
+  table::AssertUUID req("550e8400-e29b-41d4-a716-446655440000");
+  nlohmann::json expected =
+      R"({"type":"assert-table-uuid","uuid":"550e8400-e29b-41d4-a716-446655440000"})"_json;
+
+  EXPECT_EQ(ToJson(req), expected);
+  auto parsed = TableRequirementFromJson(expected);
+  ASSERT_THAT(parsed, IsOk());
+  EXPECT_EQ(*internal::checked_cast<table::AssertUUID*>(parsed.value().get()), req);
+}
+
+TEST(TableRequirementJsonTest, TableRequirementAssertRefSnapshotID) {
+  table::AssertRefSnapshotID req("main", 123456789);
+  nlohmann::json expected =
+      R"({"type":"assert-ref-snapshot-id","ref-name":"main","snapshot-id":123456789})"_json;
+
+  EXPECT_EQ(ToJson(req), expected);
+  auto parsed = TableRequirementFromJson(expected);
+  ASSERT_THAT(parsed, IsOk());
+  EXPECT_EQ(*internal::checked_cast<table::AssertRefSnapshotID*>(parsed.value().get()),
+            req);
+}
+
+TEST(TableRequirementJsonTest, TableRequirementAssertRefSnapshotIDWithNull) {
+  table::AssertRefSnapshotID req("main", std::nullopt);
+  nlohmann::json expected =
+      R"({"type":"assert-ref-snapshot-id","ref-name":"main","snapshot-id":null})"_json;
+
+  EXPECT_EQ(ToJson(req), expected);
+  auto parsed = TableRequirementFromJson(expected);
+  ASSERT_THAT(parsed, IsOk());
+  EXPECT_EQ(*internal::checked_cast<table::AssertRefSnapshotID*>(parsed.value().get()),
+            req);
+}
+
+TEST(TableRequirementJsonTest, TableRequirementAssertLastAssignedFieldId) {
+  table::AssertLastAssignedFieldId req(100);
+  nlohmann::json expected =
+      R"({"type":"assert-last-assigned-field-id","last-assigned-field-id":100})"_json;
+
+  EXPECT_EQ(ToJson(req), expected);
+  auto parsed = TableRequirementFromJson(expected);
+  ASSERT_THAT(parsed, IsOk());
+  EXPECT_EQ(
+      *internal::checked_cast<table::AssertLastAssignedFieldId*>(parsed.value().get()),
+      req);
+}
+
+TEST(TableRequirementJsonTest, TableRequirementAssertCurrentSchemaID) {
+  table::AssertCurrentSchemaID req(1);
+  nlohmann::json expected =
+      R"({"type":"assert-current-schema-id","current-schema-id":1})"_json;
+
+  EXPECT_EQ(ToJson(req), expected);
+  auto parsed = TableRequirementFromJson(expected);
+  ASSERT_THAT(parsed, IsOk());
+  EXPECT_EQ(*internal::checked_cast<table::AssertCurrentSchemaID*>(parsed.value().get()),
+            req);
+}
+
+TEST(TableRequirementJsonTest, TableRequirementAssertLastAssignedPartitionId) {
+  table::AssertLastAssignedPartitionId req(1000);
+  nlohmann::json expected =
+      R"({"type":"assert-last-assigned-partition-id","last-assigned-partition-id":1000})"_json;
+
+  EXPECT_EQ(ToJson(req), expected);
+  auto parsed = TableRequirementFromJson(expected);
+  ASSERT_THAT(parsed, IsOk());
+  EXPECT_EQ(*internal::checked_cast<table::AssertLastAssignedPartitionId*>(
+                parsed.value().get()),
+            req);
+}
+
+TEST(TableRequirementJsonTest, TableRequirementAssertDefaultSpecID) {
+  table::AssertDefaultSpecID req(0);
+  nlohmann::json expected =
+      R"({"type":"assert-default-spec-id","default-spec-id":0})"_json;
+
+  EXPECT_EQ(ToJson(req), expected);
+  auto parsed = TableRequirementFromJson(expected);
+  ASSERT_THAT(parsed, IsOk());
+  EXPECT_EQ(*internal::checked_cast<table::AssertDefaultSpecID*>(parsed.value().get()),
+            req);
+}
+
+TEST(TableRequirementJsonTest, TableRequirementAssertDefaultSortOrderID) {
+  table::AssertDefaultSortOrderID req(0);
+  nlohmann::json expected =
+      R"({"type":"assert-default-sort-order-id","default-sort-order-id":0})"_json;
+
+  EXPECT_EQ(ToJson(req), expected);
+  auto parsed = TableRequirementFromJson(expected);
+  ASSERT_THAT(parsed, IsOk());
+  EXPECT_EQ(
+      *internal::checked_cast<table::AssertDefaultSortOrderID*>(parsed.value().get()),
+      req);
+}
+
+TEST(TableRequirementJsonTest, TableRequirementUnknownType) {
+  nlohmann::json json = R"({"type":"unknown-type"})"_json;
+  auto result = TableRequirementFromJson(json);
+  EXPECT_THAT(result, IsError(ErrorKind::kJsonParseError));
+  EXPECT_THAT(result, HasErrorMessage("Unknown table requirement type"));
+}
+
 }  // namespace iceberg
diff --git a/src/iceberg/test/rest_catalog_test.cc b/src/iceberg/test/rest_catalog_test.cc
index 529690402..7f04de0ae 100644
--- a/src/iceberg/test/rest_catalog_test.cc
+++ b/src/iceberg/test/rest_catalog_test.cc
@@ -21,6 +21,7 @@
 
 #include <unistd.h>
 
+#include <algorithm>
 #include <chrono>
 #include <memory>
 #include <print>
@@ -45,6 +46,8 @@
 #include "iceberg/sort_order.h"
 #include "iceberg/table.h"
 #include "iceberg/table_identifier.h"
+#include "iceberg/table_requirement.h"
+#include "iceberg/table_update.h"
 #include "iceberg/test/matchers.h"
 #include "iceberg/test/std_io.h"
 #include "iceberg/test/test_resource.h"
@@ -407,6 +410,45 @@ TEST_F(RestCatalogIntegrationTest, CreateTable) {
               HasErrorMessage("Table already exists: test_create_table.apple.ios.t1"));
 }
 
+TEST_F(RestCatalogIntegrationTest, ListTables) {
+  auto catalog_result = CreateCatalog();
+  ASSERT_THAT(catalog_result, IsOk());
+  auto& catalog = catalog_result.value();
+
+  // Create namespace
+  Namespace ns{.levels = {"test_list_tables"}};
+  auto status = catalog->CreateNamespace(ns, {});
+  EXPECT_THAT(status, IsOk());
+
+  // Initially no tables
+  auto list_result = catalog->ListTables(ns);
+  ASSERT_THAT(list_result, IsOk());
+  EXPECT_TRUE(list_result.value().empty());
+
+  // Create tables
+  auto schema = CreateDefaultSchema();
+  auto partition_spec = PartitionSpec::Unpartitioned();
+  auto sort_order = SortOrder::Unsorted();
+  std::unordered_map<std::string, std::string> table_properties;
+
+  TableIdentifier table1{.ns = ns, .name = "table1"};
+  auto create_result = catalog->CreateTable(table1, schema, partition_spec, sort_order,
+                                            "", table_properties);
+  ASSERT_THAT(create_result, IsOk());
+
+  TableIdentifier table2{.ns = ns, .name = "table2"};
+  create_result = catalog->CreateTable(table2, schema, partition_spec, sort_order, "",
+                                       table_properties);
+  ASSERT_THAT(create_result, IsOk());
+
+  // List and varify tables
+  list_result = catalog->ListTables(ns);
+  ASSERT_THAT(list_result, IsOk());
+  EXPECT_THAT(list_result.value(), testing::UnorderedElementsAre(
+                                       testing::Field(&TableIdentifier::name, "table1"),
+                                       testing::Field(&TableIdentifier::name, "table2")));
+}
+
 TEST_F(RestCatalogIntegrationTest, LoadTable) {
   auto catalog_result = CreateCatalog();
   ASSERT_THAT(catalog_result, IsOk());
@@ -484,4 +526,117 @@ TEST_F(RestCatalogIntegrationTest, DropTable) {
   EXPECT_FALSE(load_result.value());
 }
 
+TEST_F(RestCatalogIntegrationTest, RenameTable) {
+  auto catalog_result = CreateCatalog();
+  ASSERT_THAT(catalog_result, IsOk());
+  auto& catalog = catalog_result.value();
+
+  // Create namespace
+  Namespace ns{.levels = {"test_rename_table"}};
+  auto status = catalog->CreateNamespace(ns, {});
+  EXPECT_THAT(status, IsOk());
+
+  // Create table
+  auto schema = CreateDefaultSchema();
+  auto partition_spec = PartitionSpec::Unpartitioned();
+  auto sort_order = SortOrder::Unsorted();
+
+  TableIdentifier old_table_id{.ns = ns, .name = "old_table"};
+  std::unordered_map<std::string, std::string> table_properties;
+  auto table_result = catalog->CreateTable(old_table_id, schema, partition_spec,
+                                           sort_order, "", table_properties);
+  ASSERT_THAT(table_result, IsOk());
+
+  // Rename table
+  TableIdentifier new_table_id{.ns = ns, .name = "new_table"};
+  status = catalog->RenameTable(old_table_id, new_table_id);
+  ASSERT_THAT(status, IsOk());
+
+  // Verify old table no longer exists
+  auto exists_result = catalog->TableExists(old_table_id);
+  ASSERT_THAT(exists_result, IsOk());
+  EXPECT_FALSE(exists_result.value());
+
+  // Verify new table exists
+  exists_result = catalog->TableExists(new_table_id);
+  ASSERT_THAT(exists_result, IsOk());
+  EXPECT_TRUE(exists_result.value());
+}
+
+TEST_F(RestCatalogIntegrationTest, UpdateTable) {
+  auto catalog_result = CreateCatalog();
+  ASSERT_THAT(catalog_result, IsOk());
+  auto& catalog = catalog_result.value();
+
+  // Create namespace
+  Namespace ns{.levels = {"test_update_table"}};
+  auto status = catalog->CreateNamespace(ns, {});
+  EXPECT_THAT(status, IsOk());
+
+  // Create table
+  auto schema = CreateDefaultSchema();
+  auto partition_spec = PartitionSpec::Unpartitioned();
+  auto sort_order = SortOrder::Unsorted();
+
+  TableIdentifier table_id{.ns = ns, .name = "t1"};
+  std::unordered_map<std::string, std::string> table_properties;
+  auto table_result = catalog->CreateTable(table_id, schema, partition_spec, sort_order,
+                                           "", table_properties);
+  ASSERT_THAT(table_result, IsOk());
+  auto& table = table_result.value();
+
+  // Update table properties
+  std::vector<std::unique_ptr<TableRequirement>> requirements;
+  requirements.push_back(std::make_unique<table::AssertUUID>(table->uuid()));
+
+  std::vector<std::unique_ptr<TableUpdate>> updates;
+  updates.push_back(std::make_unique<table::SetProperties>(
+      std::unordered_map<std::string, std::string>{{"key1", "value1"}}));
+
+  auto update_result = catalog->UpdateTable(table_id, requirements, updates);
+  ASSERT_THAT(update_result, IsOk());
+  auto& updated_table = update_result.value();
+
+  // Verify the property was set
+  auto& props = updated_table->metadata()->properties.configs();
+  EXPECT_EQ(props.at("key1"), "value1");
+}
+
+TEST_F(RestCatalogIntegrationTest, RegisterTable) {
+  auto catalog_result = CreateCatalog();
+  ASSERT_THAT(catalog_result, IsOk());
+  auto& catalog = catalog_result.value();
+
+  // Create namespace
+  Namespace ns{.levels = {"test_register_table"}};
+  auto status = catalog->CreateNamespace(ns, {});
+  EXPECT_THAT(status, IsOk());
+
+  // Create table
+  auto schema = CreateDefaultSchema();
+  auto partition_spec = PartitionSpec::Unpartitioned();
+  auto sort_order = SortOrder::Unsorted();
+
+  TableIdentifier table_id{.ns = ns, .name = "t1"};
+  std::unordered_map<std::string, std::string> table_properties;
+  auto table_result = catalog->CreateTable(table_id, schema, partition_spec, sort_order,
+                                           "", table_properties);
+  ASSERT_THAT(table_result, IsOk());
+  auto& table = table_result.value();
+  std::string metadata_location(table->metadata_file_location());
+
+  // Drop table (without purge, to keep metadata file)
+  status = catalog->DropTable(table_id, /*purge=*/false);
+  ASSERT_THAT(status, IsOk());
+
+  // Register table with new name
+  TableIdentifier new_table_id{.ns = ns, .name = "t2"};
+  auto register_result = catalog->RegisterTable(new_table_id, metadata_location);
+  ASSERT_THAT(register_result, IsOk());
+  auto& registered_table = register_result.value();
+
+  EXPECT_EQ(table->metadata_file_location(), registered_table->metadata_file_location());
+  EXPECT_NE(table->name(), registered_table->name());
+}
+
 }  // namespace iceberg::rest
diff --git a/src/iceberg/test/rest_json_internal_test.cc b/src/iceberg/test/rest_json_internal_test.cc
index 60facf43c..24c099fc2 100644
--- a/src/iceberg/test/rest_json_internal_test.cc
+++ b/src/iceberg/test/rest_json_internal_test.cc
@@ -30,6 +30,8 @@
 #include "iceberg/sort_order.h"
 #include "iceberg/table_identifier.h"
 #include "iceberg/table_metadata.h"
+#include "iceberg/table_requirement.h"
+#include "iceberg/table_update.h"
 #include "iceberg/test/matchers.h"
 
 namespace iceberg::rest {
@@ -1178,4 +1180,204 @@ INSTANTIATE_TEST_SUITE_P(
       return info.param.test_name;
     });
 
+DECLARE_ROUNDTRIP_TEST(CommitTableRequest)
+
+INSTANTIATE_TEST_SUITE_P(
+    CommitTableRequestCases, CommitTableRequestTest,
+    ::testing::Values(
+        // Full request with identifier, requirements, and updates
+        CommitTableRequestParam{
+            .test_name = "FullRequest",
+            .expected_json_str =
+                R"({"identifier":{"namespace":["ns1"],"name":"table1"},"requirements":[{"type":"assert-table-uuid","uuid":"2cc52516-5e73-41f2-b139-545d41a4e151"},{"type":"assert-create"}],"updates":[{"action":"assign-uuid","uuid":"2cc52516-5e73-41f2-b139-545d41a4e151"},{"action":"set-current-schema","schema-id":23}]})",
+            .model = {.identifier = TableIdentifier{Namespace{{"ns1"}}, "table1"},
+                      .requirements = {std::make_shared<table::AssertUUID>(
+                                           "2cc52516-5e73-41f2-b139-545d41a4e151"),
+                                       std::make_shared<table::AssertDoesNotExist>()},
+                      .updates = {std::make_shared<table::AssignUUID>(
+                                      "2cc52516-5e73-41f2-b139-545d41a4e151"),
+                                  std::make_shared<table::SetCurrentSchema>(23)}}},
+        // Request without identifier (identifier optional)
+        CommitTableRequestParam{
+            .test_name = "WithoutIdentifier",
+            .expected_json_str =
+                R"({"requirements":[{"type":"assert-table-uuid","uuid":"2cc52516-5e73-41f2-b139-545d41a4e151"},{"type":"assert-create"}],"updates":[{"action":"assign-uuid","uuid":"2cc52516-5e73-41f2-b139-545d41a4e151"},{"action":"set-current-schema","schema-id":23}]})",
+            .model = {.requirements = {std::make_shared<table::AssertUUID>(
+                                           "2cc52516-5e73-41f2-b139-545d41a4e151"),
+                                       std::make_shared<table::AssertDoesNotExist>()},
+                      .updates = {std::make_shared<table::AssignUUID>(
+                                      "2cc52516-5e73-41f2-b139-545d41a4e151"),
+                                  std::make_shared<table::SetCurrentSchema>(23)}}},
+        // Request with empty requirements and updates
+        CommitTableRequestParam{
+            .test_name = "EmptyRequirementsAndUpdates",
+            .expected_json_str =
+                R"({"identifier":{"namespace":["ns1"],"name":"table1"},"requirements":[],"updates":[]})",
+            .model = {.identifier = TableIdentifier{Namespace{{"ns1"}}, "table1"}}}),
+    [](const ::testing::TestParamInfo<CommitTableRequestParam>& info) {
+      return info.param.test_name;
+    });
+
+DECLARE_DESERIALIZE_TEST(CommitTableRequest)
+
+INSTANTIATE_TEST_SUITE_P(
+    CommitTableRequestDeserializeCases, CommitTableRequestDeserializeTest,
+    ::testing::Values(
+        // Identifier field is missing (should deserialize to empty identifier)
+        CommitTableRequestDeserializeParam{
+            .test_name = "MissingIdentifier",
+            .json_str = R"({"requirements":[],"updates":[]})",
+            .expected_model = {}}),
+    [](const ::testing::TestParamInfo<CommitTableRequestDeserializeParam>& info) {
+      return info.param.test_name;
+    });
+
+DECLARE_INVALID_TEST(CommitTableRequest)
+
+INSTANTIATE_TEST_SUITE_P(
+    CommitTableRequestInvalidCases, CommitTableRequestInvalidTest,
+    ::testing::Values(
+        // Invalid table identifier - missing name field
+        CommitTableRequestInvalidParam{
+            .test_name = "InvalidTableIdentifier",
+            .invalid_json_str = R"({"identifier":{},"requirements":[],"updates":[]})",
+            .expected_error_message = "Missing 'name'"},
+        // Invalid table identifier - wrong type for name
+        CommitTableRequestInvalidParam{
+            .test_name = "InvalidIdentifierNameType",
+            .invalid_json_str =
+                R"({"identifier":{"namespace":["ns1"],"name":23},"requirements":[],"updates":[]})",
+            .expected_error_message = "type must be string, but is number"},
+        // Invalid requirements - non-object value in requirements array
+        CommitTableRequestInvalidParam{
+            .test_name = "InvalidRequirementsNonObject",
+            .invalid_json_str =
+                R"({"identifier":{"namespace":["ns1"],"name":"table1"},"requirements":[23],"updates":[]})",
+            .expected_error_message = "Missing 'type' in"},
+        // Invalid requirements - missing type field
+        CommitTableRequestInvalidParam{
+            .test_name = "InvalidRequirementsMissingType",
+            .invalid_json_str =
+                R"({"identifier":{"namespace":["ns1"],"name":"table1"},"requirements":[{}],"updates":[]})",
+            .expected_error_message = "Missing 'type'"},
+        // Invalid requirements - assert-table-uuid missing uuid field
+        CommitTableRequestInvalidParam{
+            .test_name = "InvalidRequirementsMissingUUID",
+            .invalid_json_str =
+                R"({"identifier":{"namespace":["ns1"],"name":"table1"},"requirements":[{"type":"assert-table-uuid"}],"updates":[]})",
+            .expected_error_message = "Missing 'uuid'"},
+        // Invalid updates - non-object value in updates array
+        CommitTableRequestInvalidParam{
+            .test_name = "InvalidUpdatesNonObject",
+            .invalid_json_str =
+                R"({"identifier":{"namespace":["ns1"],"name":"table1"},"requirements":[],"updates":[23]})",
+            .expected_error_message = "Missing 'action' in"},
+        // Invalid updates - missing action field
+        CommitTableRequestInvalidParam{
+            .test_name = "InvalidUpdatesMissingAction",
+            .invalid_json_str =
+                R"({"identifier":{"namespace":["ns1"],"name":"table1"},"requirements":[],"updates":[{}]})",
+            .expected_error_message = "Missing 'action'"},
+        // Invalid updates - assign-uuid missing uuid field
+        CommitTableRequestInvalidParam{
+            .test_name = "InvalidUpdatesMissingUUID",
+            .invalid_json_str =
+                R"({"identifier":{"namespace":["ns1"],"name":"table1"},"requirements":[],"updates":[{"action":"assign-uuid"}]})",
+            .expected_error_message = "Missing 'uuid'"},
+        // Missing required requirements field
+        CommitTableRequestInvalidParam{
+            .test_name = "MissingRequirements",
+            .invalid_json_str =
+                R"({"identifier":{"namespace":["ns1"],"name":"table1"},"updates":[]})",
+            .expected_error_message = "Missing 'requirements'"},
+        // Missing required updates field
+        CommitTableRequestInvalidParam{
+            .test_name = "MissingUpdates",
+            .invalid_json_str =
+                R"({"identifier":{"namespace":["ns1"],"name":"table1"},"requirements":[]})",
+            .expected_error_message = "Missing 'updates'"},
+        // Empty JSON object
+        CommitTableRequestInvalidParam{
+            .test_name = "EmptyJson",
+            .invalid_json_str = R"({})",
+            .expected_error_message = "Missing 'requirements'"}),
+    [](const ::testing::TestParamInfo<CommitTableRequestInvalidParam>& info) {
+      return info.param.test_name;
+    });
+
+DECLARE_ROUNDTRIP_TEST(CommitTableResponse)
+
+INSTANTIATE_TEST_SUITE_P(
+    CommitTableResponseCases, CommitTableResponseTest,
+    ::testing::Values(
+        // Full response with metadata location and metadata
+        CommitTableResponseParam{
+            .test_name = "FullResponse",
+            .expected_json_str =
+                R"({"metadata-location":"s3://bucket/metadata/v2.json","metadata":{"current-schema-id":1,"current-snapshot-id":null,"default-sort-order-id":0,"default-spec-id":0,"format-version":2,"last-column-id":1,"last-partition-id":0,"last-sequence-number":0,"last-updated-ms":0,"location":"s3://bucket/test","metadata-log":[],"partition-specs":[{"fields":[],"spec-id":0}],"partition-statistics":[],"properties":{},"refs":{},"schemas":[{"fields":[{"id":1,"name":"id","required":true,"type":"int"}],"schema-id":1,"type":"struct"}],"snapshot-log":[],"snapshots":[],"sort-orders":[{"fields":[],"order-id":0}],"statistics":[],"table-uuid":"test-uuid-1234"}})",
+            .model = {.metadata_location = "s3://bucket/metadata/v2.json",
+                      .metadata = MakeSimpleTableMetadata()}}),
+    [](const ::testing::TestParamInfo<CommitTableResponseParam>& info) {
+      return info.param.test_name;
+    });
+
+DECLARE_DESERIALIZE_TEST(CommitTableResponse)
+
+INSTANTIATE_TEST_SUITE_P(
+    CommitTableResponseDeserializeCases, CommitTableResponseDeserializeTest,
+    ::testing::Values(
+        // Standard response with all fields
+        CommitTableResponseDeserializeParam{
+            .test_name = "StandardResponse",
+            .json_str =
+                R"({"metadata-location":"s3://bucket/metadata/v2.json","metadata":{"format-version":2,"table-uuid":"test-uuid-1234","location":"s3://bucket/test","last-sequence-number":0,"last-updated-ms":0,"last-column-id":1,"schemas":[{"type":"struct","schema-id":1,"fields":[{"id":1,"name":"id","type":"int","required":true}]}],"current-schema-id":1,"partition-specs":[{"spec-id":0,"fields":[]}],"default-spec-id":0,"last-partition-id":0,"sort-orders":[{"order-id":0,"fields":[]}],"default-sort-order-id":0,"properties":{}}})",
+            .expected_model = {.metadata_location = "s3://bucket/metadata/v2.json",
+                               .metadata = MakeSimpleTableMetadata()}}),
+    [](const ::testing::TestParamInfo<CommitTableResponseDeserializeParam>& info) {
+      return info.param.test_name;
+    });
+
+DECLARE_INVALID_TEST(CommitTableResponse)
+
+INSTANTIATE_TEST_SUITE_P(
+    CommitTableResponseInvalidCases, CommitTableResponseInvalidTest,
+    ::testing::Values(
+        // Missing required metadata-location field
+        CommitTableResponseInvalidParam{
+            .test_name = "MissingMetadataLocation",
+            .invalid_json_str =
+                R"({"metadata":{"format-version":2,"table-uuid":"test","location":"s3://test","last-sequence-number":0,"last-column-id":1,"last-updated-ms":0,"schemas":[{"type":"struct","schema-id":1,"fields":[{"id":1,"name":"id","type":"int","required":true}]}],"current-schema-id":1,"partition-specs":[{"spec-id":0,"fields":[]}],"default-spec-id":0,"last-partition-id":0,"sort-orders":[{"order-id":0,"fields":[]}],"default-sort-order-id":0}})",
+            .expected_error_message = "Missing 'metadata-location'"},
+        // Missing required metadata field
+        CommitTableResponseInvalidParam{
+            .test_name = "MissingMetadata",
+            .invalid_json_str = R"({"metadata-location":"s3://bucket/metadata/v2.json"})",
+            .expected_error_message = "Missing 'metadata'"},
+        // Null metadata field
+        CommitTableResponseInvalidParam{
+            .test_name = "NullMetadata",
+            .invalid_json_str =
+                R"({"metadata-location":"s3://bucket/metadata/v2.json","metadata":null})",
+            .expected_error_message = "Missing 'metadata'"},
+        // Wrong type for metadata-location field
+        CommitTableResponseInvalidParam{
+            .test_name = "WrongMetadataLocationType",
+            .invalid_json_str =
+                R"({"metadata-location":123,"metadata":{"format-version":2,"table-uuid":"test","location":"s3://test","last-sequence-number":0,"last-column-id":1,"last-updated-ms":0,"schemas":[{"type":"struct","schema-id":1,"fields":[{"id":1,"name":"id","type":"int","required":true}]}],"current-schema-id":1,"partition-specs":[{"spec-id":0,"fields":[]}],"default-spec-id":0,"last-partition-id":0,"sort-orders":[{"order-id":0,"fields":[]}],"default-sort-order-id":0}})",
+            .expected_error_message = "type must be string, but is number"},
+        // Wrong type for metadata field
+        CommitTableResponseInvalidParam{
+            .test_name = "WrongMetadataType",
+            .invalid_json_str =
+                R"({"metadata-location":"s3://bucket/metadata/v2.json","metadata":"invalid"})",
+            .expected_error_message = "Cannot parse metadata from a non-object"},
+        // Empty JSON object
+        CommitTableResponseInvalidParam{
+            .test_name = "EmptyJson",
+            .invalid_json_str = R"({})",
+            .expected_error_message = "Missing 'metadata-location'"}),
+    [](const ::testing::TestParamInfo<CommitTableResponseInvalidParam>& info) {
+      return info.param.test_name;
+    });
+
 }  // namespace iceberg::rest