Re: [PR] feat: add json serde for REST Catalog request/response models [iceberg-cpp]

["wgtmac (via GitHub)" <gi...@apache.org>]

wgtmac commented on code in PR #251:
URL: https://github.com/apache/iceberg-cpp/pull/251#discussion_r2485349537


##########
src/iceberg/catalog/rest/json_internal.cc:
##########
@@ -0,0 +1,357 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/catalog/rest/json_internal.h"
+
+#include <ranges>
+#include <string>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+#include <nlohmann/json.hpp>
+
+#include "iceberg/catalog/rest/types.h"
+#include "iceberg/json_internal.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/schema.h"
+#include "iceberg/sort_order.h"
+#include "iceberg/table_metadata.h"
+#include "iceberg/util/json_util_internal.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg::rest {
+
+namespace {
+
+// REST API JSON field constants
+constexpr std::string_view kNamespace = "namespace";
+constexpr std::string_view kNamespaces = "namespaces";
+constexpr std::string_view kProperties = "properties";
+constexpr std::string_view kRemovals = "removals";
+constexpr std::string_view kUpdates = "updates";
+constexpr std::string_view kUpdated = "updated";
+constexpr std::string_view kRemoved = "removed";
+constexpr std::string_view kMissing = "missing";
+constexpr std::string_view kNextPageToken = "next-page-token";
+constexpr std::string_view kName = "name";
+constexpr std::string_view kLocation = "location";
+constexpr std::string_view kSchema = "schema";
+constexpr std::string_view kPartitionSpec = "partition-spec";
+constexpr std::string_view kWriteOrder = "write-order";
+constexpr std::string_view kStageCreate = "stage-create";
+constexpr std::string_view kMetadataLocation = "metadata-location";
+constexpr std::string_view kOverwrite = "overwrite";
+constexpr std::string_view kSource = "source";
+constexpr std::string_view kDestination = "destination";
+constexpr std::string_view kMetadata = "metadata";
+constexpr std::string_view kConfig = "config";
+constexpr std::string_view kIdentifiers = "identifiers";
+
+}  // namespace
+
+// CreateNamespaceRequest
+nlohmann::json ToJson(const CreateNamespaceRequest& request) {
+  nlohmann::json json;
+  json[kNamespace] = request.namespace_.levels;
+  if (!request.properties.empty()) {
+    json[kProperties] = request.properties;
+  }
+  return json;
+}
+
+Result<CreateNamespaceRequest> CreateNamespaceRequestFromJson(
+    const nlohmann::json& json) {
+  CreateNamespaceRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(request.namespace_.levels,
+                          GetJsonValue<std::vector<std::string>>(json, kNamespace));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.properties,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kProperties)));

Review Comment:
   ```suggestion
         GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
                                                                             kProperties));
   ```
   
   Why adding an extra pair of parenthesis? 



##########
src/iceberg/catalog/rest/json_internal.cc:
##########
@@ -0,0 +1,357 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/catalog/rest/json_internal.h"
+
+#include <ranges>
+#include <string>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+#include <nlohmann/json.hpp>
+
+#include "iceberg/catalog/rest/types.h"
+#include "iceberg/json_internal.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/schema.h"
+#include "iceberg/sort_order.h"
+#include "iceberg/table_metadata.h"
+#include "iceberg/util/json_util_internal.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg::rest {
+
+namespace {
+
+// REST API JSON field constants
+constexpr std::string_view kNamespace = "namespace";
+constexpr std::string_view kNamespaces = "namespaces";
+constexpr std::string_view kProperties = "properties";
+constexpr std::string_view kRemovals = "removals";
+constexpr std::string_view kUpdates = "updates";
+constexpr std::string_view kUpdated = "updated";
+constexpr std::string_view kRemoved = "removed";
+constexpr std::string_view kMissing = "missing";
+constexpr std::string_view kNextPageToken = "next-page-token";
+constexpr std::string_view kName = "name";
+constexpr std::string_view kLocation = "location";
+constexpr std::string_view kSchema = "schema";
+constexpr std::string_view kPartitionSpec = "partition-spec";
+constexpr std::string_view kWriteOrder = "write-order";
+constexpr std::string_view kStageCreate = "stage-create";
+constexpr std::string_view kMetadataLocation = "metadata-location";
+constexpr std::string_view kOverwrite = "overwrite";
+constexpr std::string_view kSource = "source";
+constexpr std::string_view kDestination = "destination";
+constexpr std::string_view kMetadata = "metadata";
+constexpr std::string_view kConfig = "config";
+constexpr std::string_view kIdentifiers = "identifiers";
+
+}  // namespace
+
+// CreateNamespaceRequest

Review Comment:
   ```suggestion
   ```



##########
src/iceberg/catalog/rest/json_internal.cc:
##########
@@ -0,0 +1,357 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/catalog/rest/json_internal.h"
+
+#include <ranges>
+#include <string>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+#include <nlohmann/json.hpp>
+
+#include "iceberg/catalog/rest/types.h"
+#include "iceberg/json_internal.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/schema.h"
+#include "iceberg/sort_order.h"
+#include "iceberg/table_metadata.h"
+#include "iceberg/util/json_util_internal.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg::rest {
+
+namespace {
+
+// REST API JSON field constants
+constexpr std::string_view kNamespace = "namespace";
+constexpr std::string_view kNamespaces = "namespaces";
+constexpr std::string_view kProperties = "properties";
+constexpr std::string_view kRemovals = "removals";
+constexpr std::string_view kUpdates = "updates";
+constexpr std::string_view kUpdated = "updated";
+constexpr std::string_view kRemoved = "removed";
+constexpr std::string_view kMissing = "missing";
+constexpr std::string_view kNextPageToken = "next-page-token";
+constexpr std::string_view kName = "name";
+constexpr std::string_view kLocation = "location";
+constexpr std::string_view kSchema = "schema";
+constexpr std::string_view kPartitionSpec = "partition-spec";
+constexpr std::string_view kWriteOrder = "write-order";
+constexpr std::string_view kStageCreate = "stage-create";
+constexpr std::string_view kMetadataLocation = "metadata-location";
+constexpr std::string_view kOverwrite = "overwrite";
+constexpr std::string_view kSource = "source";
+constexpr std::string_view kDestination = "destination";
+constexpr std::string_view kMetadata = "metadata";
+constexpr std::string_view kConfig = "config";
+constexpr std::string_view kIdentifiers = "identifiers";
+
+}  // namespace
+
+// CreateNamespaceRequest
+nlohmann::json ToJson(const CreateNamespaceRequest& request) {
+  nlohmann::json json;
+  json[kNamespace] = request.namespace_.levels;
+  if (!request.properties.empty()) {
+    json[kProperties] = request.properties;
+  }
+  return json;
+}
+
+Result<CreateNamespaceRequest> CreateNamespaceRequestFromJson(
+    const nlohmann::json& json) {
+  CreateNamespaceRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(request.namespace_.levels,
+                          GetJsonValue<std::vector<std::string>>(json, kNamespace));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.properties,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kProperties)));
+  return request;
+}
+
+// UpdateNamespacePropertiesRequest

Review Comment:
   ```suggestion
   ```



##########
src/iceberg/catalog/rest/json_internal.cc:
##########
@@ -0,0 +1,357 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/catalog/rest/json_internal.h"
+
+#include <ranges>
+#include <string>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+#include <nlohmann/json.hpp>
+
+#include "iceberg/catalog/rest/types.h"
+#include "iceberg/json_internal.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/schema.h"
+#include "iceberg/sort_order.h"
+#include "iceberg/table_metadata.h"
+#include "iceberg/util/json_util_internal.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg::rest {
+
+namespace {
+
+// REST API JSON field constants
+constexpr std::string_view kNamespace = "namespace";
+constexpr std::string_view kNamespaces = "namespaces";
+constexpr std::string_view kProperties = "properties";
+constexpr std::string_view kRemovals = "removals";
+constexpr std::string_view kUpdates = "updates";
+constexpr std::string_view kUpdated = "updated";
+constexpr std::string_view kRemoved = "removed";
+constexpr std::string_view kMissing = "missing";
+constexpr std::string_view kNextPageToken = "next-page-token";
+constexpr std::string_view kName = "name";
+constexpr std::string_view kLocation = "location";
+constexpr std::string_view kSchema = "schema";
+constexpr std::string_view kPartitionSpec = "partition-spec";
+constexpr std::string_view kWriteOrder = "write-order";
+constexpr std::string_view kStageCreate = "stage-create";
+constexpr std::string_view kMetadataLocation = "metadata-location";
+constexpr std::string_view kOverwrite = "overwrite";
+constexpr std::string_view kSource = "source";
+constexpr std::string_view kDestination = "destination";
+constexpr std::string_view kMetadata = "metadata";
+constexpr std::string_view kConfig = "config";
+constexpr std::string_view kIdentifiers = "identifiers";
+
+}  // namespace
+
+// CreateNamespaceRequest
+nlohmann::json ToJson(const CreateNamespaceRequest& request) {
+  nlohmann::json json;
+  json[kNamespace] = request.namespace_.levels;
+  if (!request.properties.empty()) {
+    json[kProperties] = request.properties;
+  }
+  return json;
+}
+
+Result<CreateNamespaceRequest> CreateNamespaceRequestFromJson(
+    const nlohmann::json& json) {
+  CreateNamespaceRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(request.namespace_.levels,
+                          GetJsonValue<std::vector<std::string>>(json, kNamespace));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.properties,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kProperties)));
+  return request;
+}
+
+// UpdateNamespacePropertiesRequest
+nlohmann::json ToJson(const UpdateNamespacePropertiesRequest& request) {
+  // Initialize as an empty object so that when all optional fields are absent we return
+  // {} instead of null
+  nlohmann::json json = nlohmann::json::object();
+  if (!request.removals.empty()) {
+    json[kRemovals] = request.removals;
+  }
+  if (!request.updates.empty()) {
+    json[kUpdates] = request.updates;
+  }
+  return json;
+}
+
+Result<UpdateNamespacePropertiesRequest> UpdateNamespacePropertiesRequestFromJson(
+    const nlohmann::json& json) {
+  UpdateNamespacePropertiesRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.removals, GetJsonValueOrDefault<std::vector<std::string>>(json, kRemovals));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.updates,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kUpdates)));
+  return request;
+}
+
+// CreateTableRequest
+nlohmann::json ToJson(const CreateTableRequest& request) {
+  nlohmann::json json;
+  json[kName] = request.name;
+  if (!request.location.empty()) {
+    json[kLocation] = request.location;
+  }
+  json[kSchema] = ToJson(*request.schema);
+  if (request.partition_spec) {
+    json[kPartitionSpec] = ToJson(*request.partition_spec);
+  }
+  if (request.write_order) {
+    json[kWriteOrder] = ToJson(*request.write_order);
+  }
+  SetOptionalField(json, kStageCreate, request.stage_create);
+  if (!request.properties.empty()) {
+    json[kProperties] = request.properties;
+  }
+  return json;
+}
+
+Result<CreateTableRequest> CreateTableRequestFromJson(const nlohmann::json& json) {
+  CreateTableRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(request.name, GetJsonValue<std::string>(json, kName));
+  ICEBERG_ASSIGN_OR_RAISE(request.location,
+                          GetJsonValueOrDefault<std::string>(json, kLocation, ""));
+  ICEBERG_ASSIGN_OR_RAISE(auto schema_json, GetJsonValue<nlohmann::json>(json, kSchema));
+  ICEBERG_ASSIGN_OR_RAISE(request.schema, SchemaFromJson(schema_json));
+  if (json.contains(kPartitionSpec)) {
+    ICEBERG_ASSIGN_OR_RAISE(auto partition_spec_json,
+                            GetJsonValue<nlohmann::json>(json, kPartitionSpec));
+    ICEBERG_ASSIGN_OR_RAISE(request.partition_spec,
+                            PartitionSpecFromJson(request.schema, partition_spec_json));
+  } else {
+    request.partition_spec = nullptr;
+  }
+  if (json.contains(kWriteOrder)) {
+    ICEBERG_ASSIGN_OR_RAISE(auto write_order_json,
+                            GetJsonValue<nlohmann::json>(json, kWriteOrder));
+    ICEBERG_ASSIGN_OR_RAISE(request.write_order, SortOrderFromJson(write_order_json));
+  } else {
+    request.write_order = nullptr;
+  }
+  ICEBERG_ASSIGN_OR_RAISE(request.stage_create,
+                          GetJsonValueOptional<bool>(json, kStageCreate));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.properties,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kProperties)));
+  return request;
+}
+
+// RegisterTableRequest
+nlohmann::json ToJson(const RegisterTableRequest& request) {
+  nlohmann::json json;
+  json[kName] = request.name;
+  json[kMetadataLocation] = request.metadata_location;
+  if (request.overwrite) {
+    json[kOverwrite] = request.overwrite;
+  }
+  return json;
+}
+
+Result<RegisterTableRequest> RegisterTableRequestFromJson(const nlohmann::json& json) {
+  RegisterTableRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(request.name, GetJsonValue<std::string>(json, kName));
+  ICEBERG_ASSIGN_OR_RAISE(request.metadata_location,
+                          GetJsonValue<std::string>(json, kMetadataLocation));
+  // Default to false if not present
+  ICEBERG_ASSIGN_OR_RAISE(auto overwrite_opt,
+                          GetJsonValueOptional<bool>(json, kOverwrite));
+  request.overwrite = overwrite_opt.value_or(false);
+  return request;
+}
+
+// RenameTableRequest
+nlohmann::json ToJson(const RenameTableRequest& request) {
+  nlohmann::json json;
+  json[kSource] = ToJson(request.source);
+  json[kDestination] = ToJson(request.destination);
+  return json;
+}
+
+Result<RenameTableRequest> RenameTableRequestFromJson(const nlohmann::json& json) {
+  RenameTableRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(auto source_json, GetJsonValue<nlohmann::json>(json, kSource));
+  ICEBERG_ASSIGN_OR_RAISE(request.source, TableIdentifierFromJson(source_json));
+  ICEBERG_ASSIGN_OR_RAISE(auto dest_json,
+                          GetJsonValue<nlohmann::json>(json, kDestination));
+  ICEBERG_ASSIGN_OR_RAISE(request.destination, TableIdentifierFromJson(dest_json));
+  return request;
+}
+
+// LoadTableResult (used by CreateTableResponse, LoadTableResponse)
+nlohmann::json ToJson(const LoadTableResult& result) {
+  nlohmann::json json;
+  if (!result.metadata_location.empty()) {
+    json[kMetadataLocation] = result.metadata_location;
+  }
+  json[kMetadata] = ToJson(*result.metadata);
+  if (!result.config.empty()) {
+    json[kConfig] = result.config;
+  }
+  return json;
+}
+
+Result<LoadTableResult> LoadTableResultFromJson(const nlohmann::json& json) {

Review Comment:
   We need `StorageCredential`. At least you need to add a TODO to `types.h`, `ToJson` and `LoadTableResultFromJson`.



##########
src/iceberg/catalog/rest/json_internal.cc:
##########
@@ -0,0 +1,357 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/catalog/rest/json_internal.h"
+
+#include <ranges>
+#include <string>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+#include <nlohmann/json.hpp>
+
+#include "iceberg/catalog/rest/types.h"
+#include "iceberg/json_internal.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/schema.h"
+#include "iceberg/sort_order.h"
+#include "iceberg/table_metadata.h"
+#include "iceberg/util/json_util_internal.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg::rest {
+
+namespace {
+
+// REST API JSON field constants
+constexpr std::string_view kNamespace = "namespace";
+constexpr std::string_view kNamespaces = "namespaces";
+constexpr std::string_view kProperties = "properties";
+constexpr std::string_view kRemovals = "removals";
+constexpr std::string_view kUpdates = "updates";
+constexpr std::string_view kUpdated = "updated";
+constexpr std::string_view kRemoved = "removed";
+constexpr std::string_view kMissing = "missing";
+constexpr std::string_view kNextPageToken = "next-page-token";
+constexpr std::string_view kName = "name";
+constexpr std::string_view kLocation = "location";
+constexpr std::string_view kSchema = "schema";
+constexpr std::string_view kPartitionSpec = "partition-spec";
+constexpr std::string_view kWriteOrder = "write-order";
+constexpr std::string_view kStageCreate = "stage-create";
+constexpr std::string_view kMetadataLocation = "metadata-location";
+constexpr std::string_view kOverwrite = "overwrite";
+constexpr std::string_view kSource = "source";
+constexpr std::string_view kDestination = "destination";
+constexpr std::string_view kMetadata = "metadata";
+constexpr std::string_view kConfig = "config";
+constexpr std::string_view kIdentifiers = "identifiers";
+
+}  // namespace
+
+// CreateNamespaceRequest
+nlohmann::json ToJson(const CreateNamespaceRequest& request) {
+  nlohmann::json json;
+  json[kNamespace] = request.namespace_.levels;
+  if (!request.properties.empty()) {
+    json[kProperties] = request.properties;
+  }
+  return json;
+}
+
+Result<CreateNamespaceRequest> CreateNamespaceRequestFromJson(
+    const nlohmann::json& json) {
+  CreateNamespaceRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(request.namespace_.levels,
+                          GetJsonValue<std::vector<std::string>>(json, kNamespace));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.properties,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kProperties)));
+  return request;
+}
+
+// UpdateNamespacePropertiesRequest
+nlohmann::json ToJson(const UpdateNamespacePropertiesRequest& request) {
+  // Initialize as an empty object so that when all optional fields are absent we return
+  // {} instead of null
+  nlohmann::json json = nlohmann::json::object();
+  if (!request.removals.empty()) {
+    json[kRemovals] = request.removals;
+  }
+  if (!request.updates.empty()) {
+    json[kUpdates] = request.updates;
+  }
+  return json;
+}
+
+Result<UpdateNamespacePropertiesRequest> UpdateNamespacePropertiesRequestFromJson(
+    const nlohmann::json& json) {
+  UpdateNamespacePropertiesRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.removals, GetJsonValueOrDefault<std::vector<std::string>>(json, kRemovals));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.updates,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kUpdates)));
+  return request;
+}
+
+// CreateTableRequest
+nlohmann::json ToJson(const CreateTableRequest& request) {
+  nlohmann::json json;
+  json[kName] = request.name;
+  if (!request.location.empty()) {
+    json[kLocation] = request.location;
+  }
+  json[kSchema] = ToJson(*request.schema);
+  if (request.partition_spec) {
+    json[kPartitionSpec] = ToJson(*request.partition_spec);
+  }
+  if (request.write_order) {
+    json[kWriteOrder] = ToJson(*request.write_order);
+  }
+  SetOptionalField(json, kStageCreate, request.stage_create);
+  if (!request.properties.empty()) {
+    json[kProperties] = request.properties;
+  }
+  return json;
+}
+
+Result<CreateTableRequest> CreateTableRequestFromJson(const nlohmann::json& json) {
+  CreateTableRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(request.name, GetJsonValue<std::string>(json, kName));
+  ICEBERG_ASSIGN_OR_RAISE(request.location,
+                          GetJsonValueOrDefault<std::string>(json, kLocation, ""));
+  ICEBERG_ASSIGN_OR_RAISE(auto schema_json, GetJsonValue<nlohmann::json>(json, kSchema));
+  ICEBERG_ASSIGN_OR_RAISE(request.schema, SchemaFromJson(schema_json));
+  if (json.contains(kPartitionSpec)) {
+    ICEBERG_ASSIGN_OR_RAISE(auto partition_spec_json,
+                            GetJsonValue<nlohmann::json>(json, kPartitionSpec));
+    ICEBERG_ASSIGN_OR_RAISE(request.partition_spec,
+                            PartitionSpecFromJson(request.schema, partition_spec_json));
+  } else {
+    request.partition_spec = nullptr;
+  }
+  if (json.contains(kWriteOrder)) {
+    ICEBERG_ASSIGN_OR_RAISE(auto write_order_json,
+                            GetJsonValue<nlohmann::json>(json, kWriteOrder));
+    ICEBERG_ASSIGN_OR_RAISE(request.write_order, SortOrderFromJson(write_order_json));
+  } else {
+    request.write_order = nullptr;
+  }
+  ICEBERG_ASSIGN_OR_RAISE(request.stage_create,
+                          GetJsonValueOptional<bool>(json, kStageCreate));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.properties,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kProperties)));
+  return request;
+}
+
+// RegisterTableRequest
+nlohmann::json ToJson(const RegisterTableRequest& request) {
+  nlohmann::json json;
+  json[kName] = request.name;
+  json[kMetadataLocation] = request.metadata_location;
+  if (request.overwrite) {
+    json[kOverwrite] = request.overwrite;
+  }
+  return json;
+}
+
+Result<RegisterTableRequest> RegisterTableRequestFromJson(const nlohmann::json& json) {
+  RegisterTableRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(request.name, GetJsonValue<std::string>(json, kName));
+  ICEBERG_ASSIGN_OR_RAISE(request.metadata_location,
+                          GetJsonValue<std::string>(json, kMetadataLocation));
+  // Default to false if not present
+  ICEBERG_ASSIGN_OR_RAISE(auto overwrite_opt,
+                          GetJsonValueOptional<bool>(json, kOverwrite));
+  request.overwrite = overwrite_opt.value_or(false);

Review Comment:
   ```suggestion
     ICEBERG_ASSIGN_OR_RAISE(request.overwrite,
                             GetJsonValueOrDefault<bool>(json, kOverwrite, false));
   ```



##########
src/iceberg/catalog/rest/json_internal.cc:
##########
@@ -0,0 +1,357 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/catalog/rest/json_internal.h"
+
+#include <ranges>
+#include <string>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+#include <nlohmann/json.hpp>
+
+#include "iceberg/catalog/rest/types.h"
+#include "iceberg/json_internal.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/schema.h"
+#include "iceberg/sort_order.h"
+#include "iceberg/table_metadata.h"
+#include "iceberg/util/json_util_internal.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg::rest {
+
+namespace {
+
+// REST API JSON field constants
+constexpr std::string_view kNamespace = "namespace";
+constexpr std::string_view kNamespaces = "namespaces";
+constexpr std::string_view kProperties = "properties";
+constexpr std::string_view kRemovals = "removals";
+constexpr std::string_view kUpdates = "updates";
+constexpr std::string_view kUpdated = "updated";
+constexpr std::string_view kRemoved = "removed";
+constexpr std::string_view kMissing = "missing";
+constexpr std::string_view kNextPageToken = "next-page-token";
+constexpr std::string_view kName = "name";
+constexpr std::string_view kLocation = "location";
+constexpr std::string_view kSchema = "schema";
+constexpr std::string_view kPartitionSpec = "partition-spec";
+constexpr std::string_view kWriteOrder = "write-order";
+constexpr std::string_view kStageCreate = "stage-create";
+constexpr std::string_view kMetadataLocation = "metadata-location";
+constexpr std::string_view kOverwrite = "overwrite";
+constexpr std::string_view kSource = "source";
+constexpr std::string_view kDestination = "destination";
+constexpr std::string_view kMetadata = "metadata";
+constexpr std::string_view kConfig = "config";
+constexpr std::string_view kIdentifiers = "identifiers";
+
+}  // namespace
+
+// CreateNamespaceRequest
+nlohmann::json ToJson(const CreateNamespaceRequest& request) {
+  nlohmann::json json;
+  json[kNamespace] = request.namespace_.levels;
+  if (!request.properties.empty()) {
+    json[kProperties] = request.properties;
+  }
+  return json;
+}
+
+Result<CreateNamespaceRequest> CreateNamespaceRequestFromJson(
+    const nlohmann::json& json) {
+  CreateNamespaceRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(request.namespace_.levels,
+                          GetJsonValue<std::vector<std::string>>(json, kNamespace));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.properties,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kProperties)));
+  return request;
+}
+
+// UpdateNamespacePropertiesRequest
+nlohmann::json ToJson(const UpdateNamespacePropertiesRequest& request) {
+  // Initialize as an empty object so that when all optional fields are absent we return
+  // {} instead of null
+  nlohmann::json json = nlohmann::json::object();
+  if (!request.removals.empty()) {
+    json[kRemovals] = request.removals;
+  }
+  if (!request.updates.empty()) {
+    json[kUpdates] = request.updates;
+  }
+  return json;
+}
+
+Result<UpdateNamespacePropertiesRequest> UpdateNamespacePropertiesRequestFromJson(
+    const nlohmann::json& json) {
+  UpdateNamespacePropertiesRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.removals, GetJsonValueOrDefault<std::vector<std::string>>(json, kRemovals));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.updates,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kUpdates)));
+  return request;
+}
+
+// CreateTableRequest
+nlohmann::json ToJson(const CreateTableRequest& request) {
+  nlohmann::json json;
+  json[kName] = request.name;
+  if (!request.location.empty()) {
+    json[kLocation] = request.location;
+  }
+  json[kSchema] = ToJson(*request.schema);
+  if (request.partition_spec) {
+    json[kPartitionSpec] = ToJson(*request.partition_spec);
+  }
+  if (request.write_order) {
+    json[kWriteOrder] = ToJson(*request.write_order);
+  }
+  SetOptionalField(json, kStageCreate, request.stage_create);
+  if (!request.properties.empty()) {
+    json[kProperties] = request.properties;
+  }
+  return json;
+}
+
+Result<CreateTableRequest> CreateTableRequestFromJson(const nlohmann::json& json) {
+  CreateTableRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(request.name, GetJsonValue<std::string>(json, kName));
+  ICEBERG_ASSIGN_OR_RAISE(request.location,
+                          GetJsonValueOrDefault<std::string>(json, kLocation, ""));
+  ICEBERG_ASSIGN_OR_RAISE(auto schema_json, GetJsonValue<nlohmann::json>(json, kSchema));
+  ICEBERG_ASSIGN_OR_RAISE(request.schema, SchemaFromJson(schema_json));
+  if (json.contains(kPartitionSpec)) {
+    ICEBERG_ASSIGN_OR_RAISE(auto partition_spec_json,
+                            GetJsonValue<nlohmann::json>(json, kPartitionSpec));
+    ICEBERG_ASSIGN_OR_RAISE(request.partition_spec,
+                            PartitionSpecFromJson(request.schema, partition_spec_json));
+  } else {
+    request.partition_spec = nullptr;
+  }
+  if (json.contains(kWriteOrder)) {
+    ICEBERG_ASSIGN_OR_RAISE(auto write_order_json,
+                            GetJsonValue<nlohmann::json>(json, kWriteOrder));
+    ICEBERG_ASSIGN_OR_RAISE(request.write_order, SortOrderFromJson(write_order_json));
+  } else {
+    request.write_order = nullptr;
+  }
+  ICEBERG_ASSIGN_OR_RAISE(request.stage_create,
+                          GetJsonValueOptional<bool>(json, kStageCreate));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.properties,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kProperties)));
+  return request;
+}
+
+// RegisterTableRequest
+nlohmann::json ToJson(const RegisterTableRequest& request) {
+  nlohmann::json json;
+  json[kName] = request.name;
+  json[kMetadataLocation] = request.metadata_location;
+  if (request.overwrite) {
+    json[kOverwrite] = request.overwrite;
+  }
+  return json;
+}
+
+Result<RegisterTableRequest> RegisterTableRequestFromJson(const nlohmann::json& json) {
+  RegisterTableRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(request.name, GetJsonValue<std::string>(json, kName));
+  ICEBERG_ASSIGN_OR_RAISE(request.metadata_location,
+                          GetJsonValue<std::string>(json, kMetadataLocation));
+  // Default to false if not present
+  ICEBERG_ASSIGN_OR_RAISE(auto overwrite_opt,
+                          GetJsonValueOptional<bool>(json, kOverwrite));
+  request.overwrite = overwrite_opt.value_or(false);
+  return request;
+}
+
+// RenameTableRequest
+nlohmann::json ToJson(const RenameTableRequest& request) {
+  nlohmann::json json;
+  json[kSource] = ToJson(request.source);
+  json[kDestination] = ToJson(request.destination);
+  return json;
+}
+
+Result<RenameTableRequest> RenameTableRequestFromJson(const nlohmann::json& json) {
+  RenameTableRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(auto source_json, GetJsonValue<nlohmann::json>(json, kSource));
+  ICEBERG_ASSIGN_OR_RAISE(request.source, TableIdentifierFromJson(source_json));
+  ICEBERG_ASSIGN_OR_RAISE(auto dest_json,
+                          GetJsonValue<nlohmann::json>(json, kDestination));
+  ICEBERG_ASSIGN_OR_RAISE(request.destination, TableIdentifierFromJson(dest_json));
+  return request;
+}
+
+// LoadTableResult (used by CreateTableResponse, LoadTableResponse)
+nlohmann::json ToJson(const LoadTableResult& result) {
+  nlohmann::json json;
+  if (!result.metadata_location.empty()) {
+    json[kMetadataLocation] = result.metadata_location;
+  }
+  json[kMetadata] = ToJson(*result.metadata);
+  if (!result.config.empty()) {
+    json[kConfig] = result.config;
+  }
+  return json;
+}
+
+Result<LoadTableResult> LoadTableResultFromJson(const nlohmann::json& json) {
+  LoadTableResult result;
+  ICEBERG_ASSIGN_OR_RAISE(result.metadata_location, GetJsonValueOrDefault<std::string>(
+                                                        json, kMetadataLocation, ""));
+  ICEBERG_ASSIGN_OR_RAISE(auto metadata_json,
+                          GetJsonValue<nlohmann::json>(json, kMetadata));
+  ICEBERG_ASSIGN_OR_RAISE(result.metadata, TableMetadataFromJson(metadata_json));
+  ICEBERG_ASSIGN_OR_RAISE(
+      result.config, (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(
+                         json, kConfig)));

Review Comment:
   ```suggestion
         result.config, GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(
                            json, kConfig));
   ```



##########
src/iceberg/catalog/rest/json_internal.cc:
##########
@@ -0,0 +1,357 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/catalog/rest/json_internal.h"
+
+#include <ranges>
+#include <string>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+#include <nlohmann/json.hpp>
+
+#include "iceberg/catalog/rest/types.h"
+#include "iceberg/json_internal.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/schema.h"
+#include "iceberg/sort_order.h"
+#include "iceberg/table_metadata.h"
+#include "iceberg/util/json_util_internal.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg::rest {
+
+namespace {
+
+// REST API JSON field constants
+constexpr std::string_view kNamespace = "namespace";
+constexpr std::string_view kNamespaces = "namespaces";
+constexpr std::string_view kProperties = "properties";
+constexpr std::string_view kRemovals = "removals";
+constexpr std::string_view kUpdates = "updates";
+constexpr std::string_view kUpdated = "updated";
+constexpr std::string_view kRemoved = "removed";
+constexpr std::string_view kMissing = "missing";
+constexpr std::string_view kNextPageToken = "next-page-token";
+constexpr std::string_view kName = "name";
+constexpr std::string_view kLocation = "location";
+constexpr std::string_view kSchema = "schema";
+constexpr std::string_view kPartitionSpec = "partition-spec";
+constexpr std::string_view kWriteOrder = "write-order";
+constexpr std::string_view kStageCreate = "stage-create";
+constexpr std::string_view kMetadataLocation = "metadata-location";
+constexpr std::string_view kOverwrite = "overwrite";
+constexpr std::string_view kSource = "source";
+constexpr std::string_view kDestination = "destination";
+constexpr std::string_view kMetadata = "metadata";
+constexpr std::string_view kConfig = "config";
+constexpr std::string_view kIdentifiers = "identifiers";
+
+}  // namespace
+
+// CreateNamespaceRequest
+nlohmann::json ToJson(const CreateNamespaceRequest& request) {
+  nlohmann::json json;
+  json[kNamespace] = request.namespace_.levels;
+  if (!request.properties.empty()) {
+    json[kProperties] = request.properties;
+  }
+  return json;
+}
+
+Result<CreateNamespaceRequest> CreateNamespaceRequestFromJson(
+    const nlohmann::json& json) {
+  CreateNamespaceRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(request.namespace_.levels,
+                          GetJsonValue<std::vector<std::string>>(json, kNamespace));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.properties,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kProperties)));
+  return request;
+}
+
+// UpdateNamespacePropertiesRequest
+nlohmann::json ToJson(const UpdateNamespacePropertiesRequest& request) {
+  // Initialize as an empty object so that when all optional fields are absent we return
+  // {} instead of null
+  nlohmann::json json = nlohmann::json::object();
+  if (!request.removals.empty()) {
+    json[kRemovals] = request.removals;
+  }
+  if (!request.updates.empty()) {
+    json[kUpdates] = request.updates;
+  }
+  return json;
+}
+
+Result<UpdateNamespacePropertiesRequest> UpdateNamespacePropertiesRequestFromJson(
+    const nlohmann::json& json) {
+  UpdateNamespacePropertiesRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.removals, GetJsonValueOrDefault<std::vector<std::string>>(json, kRemovals));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.updates,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kUpdates)));
+  return request;
+}
+
+// CreateTableRequest
+nlohmann::json ToJson(const CreateTableRequest& request) {
+  nlohmann::json json;
+  json[kName] = request.name;
+  if (!request.location.empty()) {
+    json[kLocation] = request.location;
+  }
+  json[kSchema] = ToJson(*request.schema);
+  if (request.partition_spec) {
+    json[kPartitionSpec] = ToJson(*request.partition_spec);
+  }
+  if (request.write_order) {
+    json[kWriteOrder] = ToJson(*request.write_order);
+  }
+  SetOptionalField(json, kStageCreate, request.stage_create);
+  if (!request.properties.empty()) {
+    json[kProperties] = request.properties;
+  }
+  return json;
+}
+
+Result<CreateTableRequest> CreateTableRequestFromJson(const nlohmann::json& json) {
+  CreateTableRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(request.name, GetJsonValue<std::string>(json, kName));
+  ICEBERG_ASSIGN_OR_RAISE(request.location,
+                          GetJsonValueOrDefault<std::string>(json, kLocation, ""));
+  ICEBERG_ASSIGN_OR_RAISE(auto schema_json, GetJsonValue<nlohmann::json>(json, kSchema));
+  ICEBERG_ASSIGN_OR_RAISE(request.schema, SchemaFromJson(schema_json));
+  if (json.contains(kPartitionSpec)) {
+    ICEBERG_ASSIGN_OR_RAISE(auto partition_spec_json,
+                            GetJsonValue<nlohmann::json>(json, kPartitionSpec));
+    ICEBERG_ASSIGN_OR_RAISE(request.partition_spec,
+                            PartitionSpecFromJson(request.schema, partition_spec_json));
+  } else {
+    request.partition_spec = nullptr;
+  }
+  if (json.contains(kWriteOrder)) {
+    ICEBERG_ASSIGN_OR_RAISE(auto write_order_json,
+                            GetJsonValue<nlohmann::json>(json, kWriteOrder));
+    ICEBERG_ASSIGN_OR_RAISE(request.write_order, SortOrderFromJson(write_order_json));
+  } else {
+    request.write_order = nullptr;
+  }
+  ICEBERG_ASSIGN_OR_RAISE(request.stage_create,
+                          GetJsonValueOptional<bool>(json, kStageCreate));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.properties,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kProperties)));
+  return request;
+}
+
+// RegisterTableRequest
+nlohmann::json ToJson(const RegisterTableRequest& request) {
+  nlohmann::json json;
+  json[kName] = request.name;
+  json[kMetadataLocation] = request.metadata_location;
+  if (request.overwrite) {
+    json[kOverwrite] = request.overwrite;
+  }
+  return json;
+}
+
+Result<RegisterTableRequest> RegisterTableRequestFromJson(const nlohmann::json& json) {
+  RegisterTableRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(request.name, GetJsonValue<std::string>(json, kName));
+  ICEBERG_ASSIGN_OR_RAISE(request.metadata_location,
+                          GetJsonValue<std::string>(json, kMetadataLocation));
+  // Default to false if not present
+  ICEBERG_ASSIGN_OR_RAISE(auto overwrite_opt,
+                          GetJsonValueOptional<bool>(json, kOverwrite));
+  request.overwrite = overwrite_opt.value_or(false);
+  return request;
+}
+
+// RenameTableRequest
+nlohmann::json ToJson(const RenameTableRequest& request) {
+  nlohmann::json json;
+  json[kSource] = ToJson(request.source);
+  json[kDestination] = ToJson(request.destination);
+  return json;
+}
+
+Result<RenameTableRequest> RenameTableRequestFromJson(const nlohmann::json& json) {
+  RenameTableRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(auto source_json, GetJsonValue<nlohmann::json>(json, kSource));
+  ICEBERG_ASSIGN_OR_RAISE(request.source, TableIdentifierFromJson(source_json));
+  ICEBERG_ASSIGN_OR_RAISE(auto dest_json,
+                          GetJsonValue<nlohmann::json>(json, kDestination));
+  ICEBERG_ASSIGN_OR_RAISE(request.destination, TableIdentifierFromJson(dest_json));
+  return request;
+}
+
+// LoadTableResult (used by CreateTableResponse, LoadTableResponse)
+nlohmann::json ToJson(const LoadTableResult& result) {
+  nlohmann::json json;
+  if (!result.metadata_location.empty()) {
+    json[kMetadataLocation] = result.metadata_location;
+  }
+  json[kMetadata] = ToJson(*result.metadata);
+  if (!result.config.empty()) {
+    json[kConfig] = result.config;
+  }
+  return json;
+}
+
+Result<LoadTableResult> LoadTableResultFromJson(const nlohmann::json& json) {
+  LoadTableResult result;
+  ICEBERG_ASSIGN_OR_RAISE(result.metadata_location, GetJsonValueOrDefault<std::string>(
+                                                        json, kMetadataLocation, ""));
+  ICEBERG_ASSIGN_OR_RAISE(auto metadata_json,
+                          GetJsonValue<nlohmann::json>(json, kMetadata));
+  ICEBERG_ASSIGN_OR_RAISE(result.metadata, TableMetadataFromJson(metadata_json));
+  ICEBERG_ASSIGN_OR_RAISE(
+      result.config, (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(
+                         json, kConfig)));
+  return result;
+}
+
+// ListNamespacesResponse
+nlohmann::json ToJson(const ListNamespacesResponse& response) {
+  nlohmann::json json;
+  if (!response.next_page_token.empty()) {
+    json[kNextPageToken] = response.next_page_token;
+  }
+  nlohmann::json namespaces = nlohmann::json::array();
+  for (const auto& ns : response.namespaces) {
+    namespaces.push_back(ToJson(ns));
+  }
+  json[kNamespaces] = std::move(namespaces);
+  return json;
+}
+
+Result<ListNamespacesResponse> ListNamespacesResponseFromJson(
+    const nlohmann::json& json) {
+  ListNamespacesResponse response;
+  ICEBERG_ASSIGN_OR_RAISE(response.next_page_token,
+                          GetJsonValueOrDefault<std::string>(json, kNextPageToken, ""));
+  ICEBERG_ASSIGN_OR_RAISE(auto namespaces_json,
+                          GetJsonValue<nlohmann::json>(json, kNamespaces));
+  for (const auto& ns_json : namespaces_json) {
+    ICEBERG_ASSIGN_OR_RAISE(auto ns, NamespaceFromJson(ns_json));
+    response.namespaces.push_back(std::move(ns));
+  }
+  return response;
+}
+
+// CreateNamespaceResponse
+nlohmann::json ToJson(const CreateNamespaceResponse& response) {
+  nlohmann::json json;
+  json[kNamespace] = response.namespace_.levels;
+  if (!response.properties.empty()) {
+    json[kProperties] = response.properties;
+  }
+  return json;
+}
+
+Result<CreateNamespaceResponse> CreateNamespaceResponseFromJson(
+    const nlohmann::json& json) {
+  CreateNamespaceResponse response;
+  ICEBERG_ASSIGN_OR_RAISE(response.namespace_.levels,
+                          GetJsonValue<std::vector<std::string>>(json, kNamespace));
+  ICEBERG_ASSIGN_OR_RAISE(
+      response.properties,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kProperties)));

Review Comment:
   ```suggestion
         GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
                                                                              kProperties));
   ```



##########
src/iceberg/catalog/rest/json_internal.cc:
##########
@@ -0,0 +1,357 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/catalog/rest/json_internal.h"
+
+#include <ranges>
+#include <string>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+#include <nlohmann/json.hpp>
+
+#include "iceberg/catalog/rest/types.h"
+#include "iceberg/json_internal.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/schema.h"
+#include "iceberg/sort_order.h"
+#include "iceberg/table_metadata.h"
+#include "iceberg/util/json_util_internal.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg::rest {
+
+namespace {
+
+// REST API JSON field constants
+constexpr std::string_view kNamespace = "namespace";
+constexpr std::string_view kNamespaces = "namespaces";
+constexpr std::string_view kProperties = "properties";
+constexpr std::string_view kRemovals = "removals";
+constexpr std::string_view kUpdates = "updates";
+constexpr std::string_view kUpdated = "updated";
+constexpr std::string_view kRemoved = "removed";
+constexpr std::string_view kMissing = "missing";
+constexpr std::string_view kNextPageToken = "next-page-token";
+constexpr std::string_view kName = "name";
+constexpr std::string_view kLocation = "location";
+constexpr std::string_view kSchema = "schema";
+constexpr std::string_view kPartitionSpec = "partition-spec";
+constexpr std::string_view kWriteOrder = "write-order";
+constexpr std::string_view kStageCreate = "stage-create";
+constexpr std::string_view kMetadataLocation = "metadata-location";
+constexpr std::string_view kOverwrite = "overwrite";
+constexpr std::string_view kSource = "source";
+constexpr std::string_view kDestination = "destination";
+constexpr std::string_view kMetadata = "metadata";
+constexpr std::string_view kConfig = "config";
+constexpr std::string_view kIdentifiers = "identifiers";
+
+}  // namespace
+
+// CreateNamespaceRequest
+nlohmann::json ToJson(const CreateNamespaceRequest& request) {
+  nlohmann::json json;
+  json[kNamespace] = request.namespace_.levels;
+  if (!request.properties.empty()) {
+    json[kProperties] = request.properties;
+  }
+  return json;
+}
+
+Result<CreateNamespaceRequest> CreateNamespaceRequestFromJson(
+    const nlohmann::json& json) {
+  CreateNamespaceRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(request.namespace_.levels,
+                          GetJsonValue<std::vector<std::string>>(json, kNamespace));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.properties,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kProperties)));
+  return request;
+}
+
+// UpdateNamespacePropertiesRequest
+nlohmann::json ToJson(const UpdateNamespacePropertiesRequest& request) {
+  // Initialize as an empty object so that when all optional fields are absent we return
+  // {} instead of null
+  nlohmann::json json = nlohmann::json::object();
+  if (!request.removals.empty()) {
+    json[kRemovals] = request.removals;
+  }
+  if (!request.updates.empty()) {
+    json[kUpdates] = request.updates;
+  }
+  return json;
+}
+
+Result<UpdateNamespacePropertiesRequest> UpdateNamespacePropertiesRequestFromJson(
+    const nlohmann::json& json) {
+  UpdateNamespacePropertiesRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.removals, GetJsonValueOrDefault<std::vector<std::string>>(json, kRemovals));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.updates,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kUpdates)));
+  return request;
+}
+
+// CreateTableRequest
+nlohmann::json ToJson(const CreateTableRequest& request) {
+  nlohmann::json json;
+  json[kName] = request.name;
+  if (!request.location.empty()) {
+    json[kLocation] = request.location;
+  }
+  json[kSchema] = ToJson(*request.schema);

Review Comment:
   Perhaps you can also add `SetNullableField` and `SetArrayField` just like `SetOptionalField`.



##########
src/iceberg/test/rest_json_internal_test.cc:
##########
@@ -0,0 +1,801 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include <algorithm>
+#include <string>
+#include <unordered_map>
+#include <vector>
+
+#include <gmock/gmock.h>
+#include <gtest/gtest.h>
+#include <nlohmann/json.hpp>
+
+#include "iceberg/catalog/rest/json_internal.h"
+#include "iceberg/catalog/rest/types.h"
+#include "iceberg/table_identifier.h"
+
+namespace iceberg::rest {
+
+namespace {
+
+// Helper to compare two TableIdentifier objects
+bool TableIdentifierEqual(const TableIdentifier& a, const TableIdentifier& b) {
+  return a.ns.levels == b.ns.levels && a.name == b.name;
+}
+
+// Helper to compare string vectors (order-independent)
+bool StringVectorEqual(const std::vector<std::string>& a,
+                       const std::vector<std::string>& b) {
+  return std::ranges::is_permutation(a, b);
+}
+
+}  // namespace
+
+struct CreateNamespaceRequestParam {
+  std::string test_name;
+  std::string expected_json_str;
+  Namespace namespace_;
+  std::unordered_map<std::string, std::string> properties;
+};
+
+class CreateNamespaceRequestTest
+    : public ::testing::TestWithParam<CreateNamespaceRequestParam> {
+ protected:
+  void TestRoundTrip() {
+    const auto& param = GetParam();
+
+    // Build original object
+    CreateNamespaceRequest original;
+    original.namespace_ = param.namespace_;
+    original.properties = param.properties;
+
+    // ToJson and verify JSON string
+    auto json = ToJson(original);
+    auto expected_json = nlohmann::json::parse(param.expected_json_str);
+    EXPECT_EQ(json, expected_json) << "ToJson mismatch";
+
+    // FromJson and verify object equality
+    auto result = CreateNamespaceRequestFromJson(expected_json);
+    ASSERT_TRUE(result.has_value()) << result.error().message;
+    auto& parsed = result.value();
+
+    EXPECT_EQ(parsed.namespace_.levels, original.namespace_.levels);
+    EXPECT_EQ(parsed.properties, original.properties);

Review Comment:
   ```suggestion
       EXPECT_EQ(parsed, original);
   ```
   
   Can we directly compare `CreateNamespaceRequest`s?



##########
src/iceberg/catalog/rest/json_internal.cc:
##########
@@ -0,0 +1,357 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/catalog/rest/json_internal.h"
+
+#include <ranges>
+#include <string>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+#include <nlohmann/json.hpp>
+
+#include "iceberg/catalog/rest/types.h"
+#include "iceberg/json_internal.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/schema.h"
+#include "iceberg/sort_order.h"
+#include "iceberg/table_metadata.h"
+#include "iceberg/util/json_util_internal.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg::rest {
+
+namespace {
+
+// REST API JSON field constants
+constexpr std::string_view kNamespace = "namespace";
+constexpr std::string_view kNamespaces = "namespaces";
+constexpr std::string_view kProperties = "properties";
+constexpr std::string_view kRemovals = "removals";
+constexpr std::string_view kUpdates = "updates";
+constexpr std::string_view kUpdated = "updated";
+constexpr std::string_view kRemoved = "removed";
+constexpr std::string_view kMissing = "missing";
+constexpr std::string_view kNextPageToken = "next-page-token";
+constexpr std::string_view kName = "name";
+constexpr std::string_view kLocation = "location";
+constexpr std::string_view kSchema = "schema";
+constexpr std::string_view kPartitionSpec = "partition-spec";
+constexpr std::string_view kWriteOrder = "write-order";
+constexpr std::string_view kStageCreate = "stage-create";
+constexpr std::string_view kMetadataLocation = "metadata-location";
+constexpr std::string_view kOverwrite = "overwrite";
+constexpr std::string_view kSource = "source";
+constexpr std::string_view kDestination = "destination";
+constexpr std::string_view kMetadata = "metadata";
+constexpr std::string_view kConfig = "config";
+constexpr std::string_view kIdentifiers = "identifiers";
+
+}  // namespace
+
+// CreateNamespaceRequest
+nlohmann::json ToJson(const CreateNamespaceRequest& request) {
+  nlohmann::json json;
+  json[kNamespace] = request.namespace_.levels;
+  if (!request.properties.empty()) {
+    json[kProperties] = request.properties;
+  }
+  return json;
+}
+
+Result<CreateNamespaceRequest> CreateNamespaceRequestFromJson(
+    const nlohmann::json& json) {
+  CreateNamespaceRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(request.namespace_.levels,
+                          GetJsonValue<std::vector<std::string>>(json, kNamespace));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.properties,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kProperties)));
+  return request;
+}
+
+// UpdateNamespacePropertiesRequest
+nlohmann::json ToJson(const UpdateNamespacePropertiesRequest& request) {
+  // Initialize as an empty object so that when all optional fields are absent we return
+  // {} instead of null
+  nlohmann::json json = nlohmann::json::object();
+  if (!request.removals.empty()) {
+    json[kRemovals] = request.removals;
+  }
+  if (!request.updates.empty()) {
+    json[kUpdates] = request.updates;
+  }
+  return json;
+}
+
+Result<UpdateNamespacePropertiesRequest> UpdateNamespacePropertiesRequestFromJson(
+    const nlohmann::json& json) {
+  UpdateNamespacePropertiesRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.removals, GetJsonValueOrDefault<std::vector<std::string>>(json, kRemovals));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.updates,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,

Review Comment:
   ditto



##########
src/iceberg/catalog/rest/json_internal.cc:
##########
@@ -0,0 +1,357 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/catalog/rest/json_internal.h"
+
+#include <ranges>
+#include <string>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+#include <nlohmann/json.hpp>
+
+#include "iceberg/catalog/rest/types.h"
+#include "iceberg/json_internal.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/schema.h"
+#include "iceberg/sort_order.h"
+#include "iceberg/table_metadata.h"
+#include "iceberg/util/json_util_internal.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg::rest {
+
+namespace {
+
+// REST API JSON field constants
+constexpr std::string_view kNamespace = "namespace";
+constexpr std::string_view kNamespaces = "namespaces";
+constexpr std::string_view kProperties = "properties";
+constexpr std::string_view kRemovals = "removals";
+constexpr std::string_view kUpdates = "updates";
+constexpr std::string_view kUpdated = "updated";
+constexpr std::string_view kRemoved = "removed";
+constexpr std::string_view kMissing = "missing";
+constexpr std::string_view kNextPageToken = "next-page-token";
+constexpr std::string_view kName = "name";
+constexpr std::string_view kLocation = "location";
+constexpr std::string_view kSchema = "schema";
+constexpr std::string_view kPartitionSpec = "partition-spec";
+constexpr std::string_view kWriteOrder = "write-order";
+constexpr std::string_view kStageCreate = "stage-create";
+constexpr std::string_view kMetadataLocation = "metadata-location";
+constexpr std::string_view kOverwrite = "overwrite";
+constexpr std::string_view kSource = "source";
+constexpr std::string_view kDestination = "destination";
+constexpr std::string_view kMetadata = "metadata";
+constexpr std::string_view kConfig = "config";
+constexpr std::string_view kIdentifiers = "identifiers";
+
+}  // namespace
+
+// CreateNamespaceRequest
+nlohmann::json ToJson(const CreateNamespaceRequest& request) {
+  nlohmann::json json;
+  json[kNamespace] = request.namespace_.levels;
+  if (!request.properties.empty()) {
+    json[kProperties] = request.properties;
+  }
+  return json;
+}
+
+Result<CreateNamespaceRequest> CreateNamespaceRequestFromJson(
+    const nlohmann::json& json) {
+  CreateNamespaceRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(request.namespace_.levels,
+                          GetJsonValue<std::vector<std::string>>(json, kNamespace));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.properties,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kProperties)));
+  return request;
+}
+
+// UpdateNamespacePropertiesRequest
+nlohmann::json ToJson(const UpdateNamespacePropertiesRequest& request) {
+  // Initialize as an empty object so that when all optional fields are absent we return
+  // {} instead of null
+  nlohmann::json json = nlohmann::json::object();
+  if (!request.removals.empty()) {
+    json[kRemovals] = request.removals;
+  }
+  if (!request.updates.empty()) {
+    json[kUpdates] = request.updates;
+  }
+  return json;
+}
+
+Result<UpdateNamespacePropertiesRequest> UpdateNamespacePropertiesRequestFromJson(
+    const nlohmann::json& json) {
+  UpdateNamespacePropertiesRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.removals, GetJsonValueOrDefault<std::vector<std::string>>(json, kRemovals));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.updates,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kUpdates)));
+  return request;
+}
+
+// CreateTableRequest
+nlohmann::json ToJson(const CreateTableRequest& request) {
+  nlohmann::json json;
+  json[kName] = request.name;
+  if (!request.location.empty()) {
+    json[kLocation] = request.location;
+  }
+  json[kSchema] = ToJson(*request.schema);
+  if (request.partition_spec) {
+    json[kPartitionSpec] = ToJson(*request.partition_spec);
+  }
+  if (request.write_order) {
+    json[kWriteOrder] = ToJson(*request.write_order);
+  }
+  SetOptionalField(json, kStageCreate, request.stage_create);
+  if (!request.properties.empty()) {
+    json[kProperties] = request.properties;
+  }
+  return json;
+}
+
+Result<CreateTableRequest> CreateTableRequestFromJson(const nlohmann::json& json) {
+  CreateTableRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(request.name, GetJsonValue<std::string>(json, kName));
+  ICEBERG_ASSIGN_OR_RAISE(request.location,
+                          GetJsonValueOrDefault<std::string>(json, kLocation, ""));

Review Comment:
   ```suggestion
                             GetJsonValueOrDefault<std::string>(json, kLocation));
   ```



##########
src/iceberg/catalog/rest/meson.build:
##########
@@ -46,4 +46,7 @@ iceberg_rest_dep = declare_dependency(
 meson.override_dependency('iceberg-rest', iceberg_rest_dep)
 pkg.generate(iceberg_rest_lib)
 
-install_headers(['rest_catalog.h', 'types.h'], subdir: 'iceberg/catalog/rest')
+install_headers(
+    ['rest_catalog.h', 'types.h', 'json_internal.h'],

Review Comment:
   ```suggestion
       ['rest_catalog.h', 'types.h'],
   ```
   
   Internal headers are not supposed to be installed.



##########
src/iceberg/catalog/rest/json_internal.cc:
##########
@@ -0,0 +1,357 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/catalog/rest/json_internal.h"
+
+#include <ranges>
+#include <string>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+#include <nlohmann/json.hpp>
+
+#include "iceberg/catalog/rest/types.h"
+#include "iceberg/json_internal.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/schema.h"
+#include "iceberg/sort_order.h"
+#include "iceberg/table_metadata.h"
+#include "iceberg/util/json_util_internal.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg::rest {
+
+namespace {
+
+// REST API JSON field constants
+constexpr std::string_view kNamespace = "namespace";
+constexpr std::string_view kNamespaces = "namespaces";
+constexpr std::string_view kProperties = "properties";
+constexpr std::string_view kRemovals = "removals";
+constexpr std::string_view kUpdates = "updates";
+constexpr std::string_view kUpdated = "updated";
+constexpr std::string_view kRemoved = "removed";
+constexpr std::string_view kMissing = "missing";
+constexpr std::string_view kNextPageToken = "next-page-token";
+constexpr std::string_view kName = "name";
+constexpr std::string_view kLocation = "location";
+constexpr std::string_view kSchema = "schema";
+constexpr std::string_view kPartitionSpec = "partition-spec";
+constexpr std::string_view kWriteOrder = "write-order";
+constexpr std::string_view kStageCreate = "stage-create";
+constexpr std::string_view kMetadataLocation = "metadata-location";
+constexpr std::string_view kOverwrite = "overwrite";
+constexpr std::string_view kSource = "source";
+constexpr std::string_view kDestination = "destination";
+constexpr std::string_view kMetadata = "metadata";
+constexpr std::string_view kConfig = "config";
+constexpr std::string_view kIdentifiers = "identifiers";
+
+}  // namespace
+
+// CreateNamespaceRequest
+nlohmann::json ToJson(const CreateNamespaceRequest& request) {
+  nlohmann::json json;
+  json[kNamespace] = request.namespace_.levels;
+  if (!request.properties.empty()) {
+    json[kProperties] = request.properties;
+  }
+  return json;
+}
+
+Result<CreateNamespaceRequest> CreateNamespaceRequestFromJson(
+    const nlohmann::json& json) {
+  CreateNamespaceRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(request.namespace_.levels,
+                          GetJsonValue<std::vector<std::string>>(json, kNamespace));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.properties,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kProperties)));
+  return request;
+}
+
+// UpdateNamespacePropertiesRequest
+nlohmann::json ToJson(const UpdateNamespacePropertiesRequest& request) {
+  // Initialize as an empty object so that when all optional fields are absent we return
+  // {} instead of null
+  nlohmann::json json = nlohmann::json::object();
+  if (!request.removals.empty()) {
+    json[kRemovals] = request.removals;
+  }
+  if (!request.updates.empty()) {
+    json[kUpdates] = request.updates;
+  }
+  return json;
+}
+
+Result<UpdateNamespacePropertiesRequest> UpdateNamespacePropertiesRequestFromJson(
+    const nlohmann::json& json) {
+  UpdateNamespacePropertiesRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.removals, GetJsonValueOrDefault<std::vector<std::string>>(json, kRemovals));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.updates,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kUpdates)));
+  return request;
+}
+
+// CreateTableRequest
+nlohmann::json ToJson(const CreateTableRequest& request) {
+  nlohmann::json json;
+  json[kName] = request.name;
+  if (!request.location.empty()) {
+    json[kLocation] = request.location;
+  }
+  json[kSchema] = ToJson(*request.schema);

Review Comment:
   request.schema is a pointer and we need to check if it is nullptr before dereferencing it.



##########
src/iceberg/catalog/rest/json_internal.cc:
##########
@@ -0,0 +1,357 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/catalog/rest/json_internal.h"
+
+#include <ranges>
+#include <string>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+#include <nlohmann/json.hpp>
+
+#include "iceberg/catalog/rest/types.h"
+#include "iceberg/json_internal.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/schema.h"
+#include "iceberg/sort_order.h"
+#include "iceberg/table_metadata.h"
+#include "iceberg/util/json_util_internal.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg::rest {
+
+namespace {
+
+// REST API JSON field constants
+constexpr std::string_view kNamespace = "namespace";
+constexpr std::string_view kNamespaces = "namespaces";
+constexpr std::string_view kProperties = "properties";
+constexpr std::string_view kRemovals = "removals";
+constexpr std::string_view kUpdates = "updates";
+constexpr std::string_view kUpdated = "updated";
+constexpr std::string_view kRemoved = "removed";
+constexpr std::string_view kMissing = "missing";
+constexpr std::string_view kNextPageToken = "next-page-token";
+constexpr std::string_view kName = "name";
+constexpr std::string_view kLocation = "location";
+constexpr std::string_view kSchema = "schema";
+constexpr std::string_view kPartitionSpec = "partition-spec";
+constexpr std::string_view kWriteOrder = "write-order";
+constexpr std::string_view kStageCreate = "stage-create";
+constexpr std::string_view kMetadataLocation = "metadata-location";
+constexpr std::string_view kOverwrite = "overwrite";
+constexpr std::string_view kSource = "source";
+constexpr std::string_view kDestination = "destination";
+constexpr std::string_view kMetadata = "metadata";
+constexpr std::string_view kConfig = "config";
+constexpr std::string_view kIdentifiers = "identifiers";
+
+}  // namespace
+
+// CreateNamespaceRequest
+nlohmann::json ToJson(const CreateNamespaceRequest& request) {
+  nlohmann::json json;
+  json[kNamespace] = request.namespace_.levels;
+  if (!request.properties.empty()) {
+    json[kProperties] = request.properties;
+  }
+  return json;
+}
+
+Result<CreateNamespaceRequest> CreateNamespaceRequestFromJson(
+    const nlohmann::json& json) {
+  CreateNamespaceRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(request.namespace_.levels,
+                          GetJsonValue<std::vector<std::string>>(json, kNamespace));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.properties,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kProperties)));
+  return request;
+}
+
+// UpdateNamespacePropertiesRequest
+nlohmann::json ToJson(const UpdateNamespacePropertiesRequest& request) {
+  // Initialize as an empty object so that when all optional fields are absent we return
+  // {} instead of null
+  nlohmann::json json = nlohmann::json::object();
+  if (!request.removals.empty()) {
+    json[kRemovals] = request.removals;
+  }
+  if (!request.updates.empty()) {
+    json[kUpdates] = request.updates;
+  }
+  return json;
+}
+
+Result<UpdateNamespacePropertiesRequest> UpdateNamespacePropertiesRequestFromJson(
+    const nlohmann::json& json) {
+  UpdateNamespacePropertiesRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.removals, GetJsonValueOrDefault<std::vector<std::string>>(json, kRemovals));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.updates,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kUpdates)));
+  return request;
+}
+
+// CreateTableRequest
+nlohmann::json ToJson(const CreateTableRequest& request) {
+  nlohmann::json json;
+  json[kName] = request.name;
+  if (!request.location.empty()) {
+    json[kLocation] = request.location;
+  }
+  json[kSchema] = ToJson(*request.schema);
+  if (request.partition_spec) {
+    json[kPartitionSpec] = ToJson(*request.partition_spec);
+  }
+  if (request.write_order) {
+    json[kWriteOrder] = ToJson(*request.write_order);
+  }
+  SetOptionalField(json, kStageCreate, request.stage_create);
+  if (!request.properties.empty()) {
+    json[kProperties] = request.properties;
+  }
+  return json;
+}
+
+Result<CreateTableRequest> CreateTableRequestFromJson(const nlohmann::json& json) {
+  CreateTableRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(request.name, GetJsonValue<std::string>(json, kName));
+  ICEBERG_ASSIGN_OR_RAISE(request.location,
+                          GetJsonValueOrDefault<std::string>(json, kLocation, ""));
+  ICEBERG_ASSIGN_OR_RAISE(auto schema_json, GetJsonValue<nlohmann::json>(json, kSchema));
+  ICEBERG_ASSIGN_OR_RAISE(request.schema, SchemaFromJson(schema_json));
+  if (json.contains(kPartitionSpec)) {
+    ICEBERG_ASSIGN_OR_RAISE(auto partition_spec_json,
+                            GetJsonValue<nlohmann::json>(json, kPartitionSpec));
+    ICEBERG_ASSIGN_OR_RAISE(request.partition_spec,
+                            PartitionSpecFromJson(request.schema, partition_spec_json));
+  } else {
+    request.partition_spec = nullptr;
+  }
+  if (json.contains(kWriteOrder)) {
+    ICEBERG_ASSIGN_OR_RAISE(auto write_order_json,
+                            GetJsonValue<nlohmann::json>(json, kWriteOrder));
+    ICEBERG_ASSIGN_OR_RAISE(request.write_order, SortOrderFromJson(write_order_json));
+  } else {
+    request.write_order = nullptr;
+  }
+  ICEBERG_ASSIGN_OR_RAISE(request.stage_create,
+                          GetJsonValueOptional<bool>(json, kStageCreate));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.properties,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kProperties)));
+  return request;
+}
+
+// RegisterTableRequest

Review Comment:
   ```suggestion
   ```



##########
src/iceberg/catalog/rest/json_internal.cc:
##########
@@ -0,0 +1,357 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/catalog/rest/json_internal.h"
+
+#include <ranges>
+#include <string>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+#include <nlohmann/json.hpp>
+
+#include "iceberg/catalog/rest/types.h"
+#include "iceberg/json_internal.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/schema.h"
+#include "iceberg/sort_order.h"
+#include "iceberg/table_metadata.h"
+#include "iceberg/util/json_util_internal.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg::rest {
+
+namespace {
+
+// REST API JSON field constants
+constexpr std::string_view kNamespace = "namespace";
+constexpr std::string_view kNamespaces = "namespaces";
+constexpr std::string_view kProperties = "properties";
+constexpr std::string_view kRemovals = "removals";
+constexpr std::string_view kUpdates = "updates";
+constexpr std::string_view kUpdated = "updated";
+constexpr std::string_view kRemoved = "removed";
+constexpr std::string_view kMissing = "missing";
+constexpr std::string_view kNextPageToken = "next-page-token";
+constexpr std::string_view kName = "name";
+constexpr std::string_view kLocation = "location";
+constexpr std::string_view kSchema = "schema";
+constexpr std::string_view kPartitionSpec = "partition-spec";
+constexpr std::string_view kWriteOrder = "write-order";
+constexpr std::string_view kStageCreate = "stage-create";
+constexpr std::string_view kMetadataLocation = "metadata-location";
+constexpr std::string_view kOverwrite = "overwrite";
+constexpr std::string_view kSource = "source";
+constexpr std::string_view kDestination = "destination";
+constexpr std::string_view kMetadata = "metadata";
+constexpr std::string_view kConfig = "config";
+constexpr std::string_view kIdentifiers = "identifiers";
+
+}  // namespace
+
+// CreateNamespaceRequest
+nlohmann::json ToJson(const CreateNamespaceRequest& request) {
+  nlohmann::json json;
+  json[kNamespace] = request.namespace_.levels;
+  if (!request.properties.empty()) {
+    json[kProperties] = request.properties;
+  }
+  return json;
+}
+
+Result<CreateNamespaceRequest> CreateNamespaceRequestFromJson(
+    const nlohmann::json& json) {
+  CreateNamespaceRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(request.namespace_.levels,
+                          GetJsonValue<std::vector<std::string>>(json, kNamespace));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.properties,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kProperties)));
+  return request;
+}
+
+// UpdateNamespacePropertiesRequest
+nlohmann::json ToJson(const UpdateNamespacePropertiesRequest& request) {
+  // Initialize as an empty object so that when all optional fields are absent we return
+  // {} instead of null
+  nlohmann::json json = nlohmann::json::object();
+  if (!request.removals.empty()) {
+    json[kRemovals] = request.removals;
+  }
+  if (!request.updates.empty()) {
+    json[kUpdates] = request.updates;
+  }
+  return json;
+}
+
+Result<UpdateNamespacePropertiesRequest> UpdateNamespacePropertiesRequestFromJson(
+    const nlohmann::json& json) {
+  UpdateNamespacePropertiesRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.removals, GetJsonValueOrDefault<std::vector<std::string>>(json, kRemovals));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.updates,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kUpdates)));
+  return request;
+}
+
+// CreateTableRequest
+nlohmann::json ToJson(const CreateTableRequest& request) {
+  nlohmann::json json;
+  json[kName] = request.name;
+  if (!request.location.empty()) {
+    json[kLocation] = request.location;
+  }
+  json[kSchema] = ToJson(*request.schema);
+  if (request.partition_spec) {
+    json[kPartitionSpec] = ToJson(*request.partition_spec);
+  }
+  if (request.write_order) {
+    json[kWriteOrder] = ToJson(*request.write_order);
+  }
+  SetOptionalField(json, kStageCreate, request.stage_create);
+  if (!request.properties.empty()) {
+    json[kProperties] = request.properties;
+  }
+  return json;
+}
+
+Result<CreateTableRequest> CreateTableRequestFromJson(const nlohmann::json& json) {
+  CreateTableRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(request.name, GetJsonValue<std::string>(json, kName));
+  ICEBERG_ASSIGN_OR_RAISE(request.location,
+                          GetJsonValueOrDefault<std::string>(json, kLocation, ""));
+  ICEBERG_ASSIGN_OR_RAISE(auto schema_json, GetJsonValue<nlohmann::json>(json, kSchema));
+  ICEBERG_ASSIGN_OR_RAISE(request.schema, SchemaFromJson(schema_json));
+  if (json.contains(kPartitionSpec)) {
+    ICEBERG_ASSIGN_OR_RAISE(auto partition_spec_json,
+                            GetJsonValue<nlohmann::json>(json, kPartitionSpec));
+    ICEBERG_ASSIGN_OR_RAISE(request.partition_spec,
+                            PartitionSpecFromJson(request.schema, partition_spec_json));
+  } else {
+    request.partition_spec = nullptr;
+  }
+  if (json.contains(kWriteOrder)) {
+    ICEBERG_ASSIGN_OR_RAISE(auto write_order_json,
+                            GetJsonValue<nlohmann::json>(json, kWriteOrder));
+    ICEBERG_ASSIGN_OR_RAISE(request.write_order, SortOrderFromJson(write_order_json));
+  } else {
+    request.write_order = nullptr;
+  }
+  ICEBERG_ASSIGN_OR_RAISE(request.stage_create,
+                          GetJsonValueOptional<bool>(json, kStageCreate));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.properties,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kProperties)));

Review Comment:
   ```suggestion
         GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
                                                                             kProperties));
   ```



##########
src/iceberg/catalog/rest/json_internal.cc:
##########
@@ -0,0 +1,357 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/catalog/rest/json_internal.h"
+
+#include <ranges>
+#include <string>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+#include <nlohmann/json.hpp>
+
+#include "iceberg/catalog/rest/types.h"
+#include "iceberg/json_internal.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/schema.h"
+#include "iceberg/sort_order.h"
+#include "iceberg/table_metadata.h"
+#include "iceberg/util/json_util_internal.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg::rest {
+
+namespace {
+
+// REST API JSON field constants
+constexpr std::string_view kNamespace = "namespace";
+constexpr std::string_view kNamespaces = "namespaces";
+constexpr std::string_view kProperties = "properties";
+constexpr std::string_view kRemovals = "removals";
+constexpr std::string_view kUpdates = "updates";
+constexpr std::string_view kUpdated = "updated";
+constexpr std::string_view kRemoved = "removed";
+constexpr std::string_view kMissing = "missing";
+constexpr std::string_view kNextPageToken = "next-page-token";
+constexpr std::string_view kName = "name";
+constexpr std::string_view kLocation = "location";
+constexpr std::string_view kSchema = "schema";
+constexpr std::string_view kPartitionSpec = "partition-spec";
+constexpr std::string_view kWriteOrder = "write-order";
+constexpr std::string_view kStageCreate = "stage-create";
+constexpr std::string_view kMetadataLocation = "metadata-location";
+constexpr std::string_view kOverwrite = "overwrite";
+constexpr std::string_view kSource = "source";
+constexpr std::string_view kDestination = "destination";
+constexpr std::string_view kMetadata = "metadata";
+constexpr std::string_view kConfig = "config";
+constexpr std::string_view kIdentifiers = "identifiers";
+
+}  // namespace
+
+// CreateNamespaceRequest
+nlohmann::json ToJson(const CreateNamespaceRequest& request) {
+  nlohmann::json json;
+  json[kNamespace] = request.namespace_.levels;
+  if (!request.properties.empty()) {
+    json[kProperties] = request.properties;
+  }
+  return json;
+}
+
+Result<CreateNamespaceRequest> CreateNamespaceRequestFromJson(
+    const nlohmann::json& json) {
+  CreateNamespaceRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(request.namespace_.levels,
+                          GetJsonValue<std::vector<std::string>>(json, kNamespace));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.properties,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kProperties)));
+  return request;
+}
+
+// UpdateNamespacePropertiesRequest
+nlohmann::json ToJson(const UpdateNamespacePropertiesRequest& request) {
+  // Initialize as an empty object so that when all optional fields are absent we return
+  // {} instead of null
+  nlohmann::json json = nlohmann::json::object();
+  if (!request.removals.empty()) {
+    json[kRemovals] = request.removals;
+  }
+  if (!request.updates.empty()) {
+    json[kUpdates] = request.updates;
+  }
+  return json;
+}
+
+Result<UpdateNamespacePropertiesRequest> UpdateNamespacePropertiesRequestFromJson(
+    const nlohmann::json& json) {
+  UpdateNamespacePropertiesRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.removals, GetJsonValueOrDefault<std::vector<std::string>>(json, kRemovals));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.updates,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kUpdates)));
+  return request;
+}
+
+// CreateTableRequest
+nlohmann::json ToJson(const CreateTableRequest& request) {
+  nlohmann::json json;
+  json[kName] = request.name;
+  if (!request.location.empty()) {
+    json[kLocation] = request.location;
+  }
+  json[kSchema] = ToJson(*request.schema);
+  if (request.partition_spec) {
+    json[kPartitionSpec] = ToJson(*request.partition_spec);
+  }
+  if (request.write_order) {
+    json[kWriteOrder] = ToJson(*request.write_order);
+  }
+  SetOptionalField(json, kStageCreate, request.stage_create);
+  if (!request.properties.empty()) {
+    json[kProperties] = request.properties;
+  }
+  return json;
+}
+
+Result<CreateTableRequest> CreateTableRequestFromJson(const nlohmann::json& json) {
+  CreateTableRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(request.name, GetJsonValue<std::string>(json, kName));
+  ICEBERG_ASSIGN_OR_RAISE(request.location,
+                          GetJsonValueOrDefault<std::string>(json, kLocation, ""));
+  ICEBERG_ASSIGN_OR_RAISE(auto schema_json, GetJsonValue<nlohmann::json>(json, kSchema));
+  ICEBERG_ASSIGN_OR_RAISE(request.schema, SchemaFromJson(schema_json));
+  if (json.contains(kPartitionSpec)) {
+    ICEBERG_ASSIGN_OR_RAISE(auto partition_spec_json,
+                            GetJsonValue<nlohmann::json>(json, kPartitionSpec));
+    ICEBERG_ASSIGN_OR_RAISE(request.partition_spec,
+                            PartitionSpecFromJson(request.schema, partition_spec_json));
+  } else {
+    request.partition_spec = nullptr;
+  }
+  if (json.contains(kWriteOrder)) {
+    ICEBERG_ASSIGN_OR_RAISE(auto write_order_json,
+                            GetJsonValue<nlohmann::json>(json, kWriteOrder));
+    ICEBERG_ASSIGN_OR_RAISE(request.write_order, SortOrderFromJson(write_order_json));
+  } else {
+    request.write_order = nullptr;
+  }
+  ICEBERG_ASSIGN_OR_RAISE(request.stage_create,
+                          GetJsonValueOptional<bool>(json, kStageCreate));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.properties,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kProperties)));
+  return request;
+}
+
+// RegisterTableRequest
+nlohmann::json ToJson(const RegisterTableRequest& request) {
+  nlohmann::json json;
+  json[kName] = request.name;
+  json[kMetadataLocation] = request.metadata_location;
+  if (request.overwrite) {
+    json[kOverwrite] = request.overwrite;
+  }
+  return json;
+}
+
+Result<RegisterTableRequest> RegisterTableRequestFromJson(const nlohmann::json& json) {
+  RegisterTableRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(request.name, GetJsonValue<std::string>(json, kName));
+  ICEBERG_ASSIGN_OR_RAISE(request.metadata_location,
+                          GetJsonValue<std::string>(json, kMetadataLocation));
+  // Default to false if not present
+  ICEBERG_ASSIGN_OR_RAISE(auto overwrite_opt,
+                          GetJsonValueOptional<bool>(json, kOverwrite));
+  request.overwrite = overwrite_opt.value_or(false);
+  return request;
+}
+
+// RenameTableRequest
+nlohmann::json ToJson(const RenameTableRequest& request) {
+  nlohmann::json json;
+  json[kSource] = ToJson(request.source);
+  json[kDestination] = ToJson(request.destination);
+  return json;
+}
+
+Result<RenameTableRequest> RenameTableRequestFromJson(const nlohmann::json& json) {
+  RenameTableRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(auto source_json, GetJsonValue<nlohmann::json>(json, kSource));
+  ICEBERG_ASSIGN_OR_RAISE(request.source, TableIdentifierFromJson(source_json));
+  ICEBERG_ASSIGN_OR_RAISE(auto dest_json,
+                          GetJsonValue<nlohmann::json>(json, kDestination));
+  ICEBERG_ASSIGN_OR_RAISE(request.destination, TableIdentifierFromJson(dest_json));
+  return request;
+}
+
+// LoadTableResult (used by CreateTableResponse, LoadTableResponse)
+nlohmann::json ToJson(const LoadTableResult& result) {
+  nlohmann::json json;
+  if (!result.metadata_location.empty()) {
+    json[kMetadataLocation] = result.metadata_location;
+  }
+  json[kMetadata] = ToJson(*result.metadata);
+  if (!result.config.empty()) {
+    json[kConfig] = result.config;
+  }
+  return json;
+}
+
+Result<LoadTableResult> LoadTableResultFromJson(const nlohmann::json& json) {
+  LoadTableResult result;
+  ICEBERG_ASSIGN_OR_RAISE(result.metadata_location, GetJsonValueOrDefault<std::string>(
+                                                        json, kMetadataLocation, ""));
+  ICEBERG_ASSIGN_OR_RAISE(auto metadata_json,
+                          GetJsonValue<nlohmann::json>(json, kMetadata));
+  ICEBERG_ASSIGN_OR_RAISE(result.metadata, TableMetadataFromJson(metadata_json));
+  ICEBERG_ASSIGN_OR_RAISE(
+      result.config, (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(
+                         json, kConfig)));
+  return result;
+}
+
+// ListNamespacesResponse
+nlohmann::json ToJson(const ListNamespacesResponse& response) {
+  nlohmann::json json;
+  if (!response.next_page_token.empty()) {
+    json[kNextPageToken] = response.next_page_token;
+  }
+  nlohmann::json namespaces = nlohmann::json::array();
+  for (const auto& ns : response.namespaces) {
+    namespaces.push_back(ToJson(ns));
+  }
+  json[kNamespaces] = std::move(namespaces);
+  return json;
+}
+
+Result<ListNamespacesResponse> ListNamespacesResponseFromJson(
+    const nlohmann::json& json) {
+  ListNamespacesResponse response;
+  ICEBERG_ASSIGN_OR_RAISE(response.next_page_token,
+                          GetJsonValueOrDefault<std::string>(json, kNextPageToken, ""));
+  ICEBERG_ASSIGN_OR_RAISE(auto namespaces_json,
+                          GetJsonValue<nlohmann::json>(json, kNamespaces));
+  for (const auto& ns_json : namespaces_json) {
+    ICEBERG_ASSIGN_OR_RAISE(auto ns, NamespaceFromJson(ns_json));
+    response.namespaces.push_back(std::move(ns));
+  }
+  return response;
+}
+
+// CreateNamespaceResponse
+nlohmann::json ToJson(const CreateNamespaceResponse& response) {
+  nlohmann::json json;
+  json[kNamespace] = response.namespace_.levels;
+  if (!response.properties.empty()) {
+    json[kProperties] = response.properties;
+  }
+  return json;
+}
+
+Result<CreateNamespaceResponse> CreateNamespaceResponseFromJson(
+    const nlohmann::json& json) {
+  CreateNamespaceResponse response;
+  ICEBERG_ASSIGN_OR_RAISE(response.namespace_.levels,
+                          GetJsonValue<std::vector<std::string>>(json, kNamespace));
+  ICEBERG_ASSIGN_OR_RAISE(
+      response.properties,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kProperties)));
+  return response;
+}
+
+// GetNamespaceResponse
+nlohmann::json ToJson(const GetNamespaceResponse& response) {
+  nlohmann::json json;
+  json[kNamespace] = response.namespace_.levels;
+  if (!response.properties.empty()) {
+    json[kProperties] = response.properties;
+  }
+  return json;
+}
+
+Result<GetNamespaceResponse> GetNamespaceResponseFromJson(const nlohmann::json& json) {
+  GetNamespaceResponse response;
+  ICEBERG_ASSIGN_OR_RAISE(response.namespace_.levels,
+                          GetJsonValue<std::vector<std::string>>(json, kNamespace));
+  ICEBERG_ASSIGN_OR_RAISE(
+      response.properties,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kProperties)));
+  return response;
+}
+
+// UpdateNamespacePropertiesResponse
+nlohmann::json ToJson(const UpdateNamespacePropertiesResponse& response) {
+  nlohmann::json json;
+  json[kUpdated] = response.updated;
+  json[kRemoved] = response.removed;
+  if (!response.missing.empty()) {
+    json[kMissing] = response.missing;
+  }
+  return json;
+}
+
+Result<UpdateNamespacePropertiesResponse> UpdateNamespacePropertiesResponseFromJson(
+    const nlohmann::json& json) {
+  UpdateNamespacePropertiesResponse response;
+  ICEBERG_ASSIGN_OR_RAISE(response.updated,
+                          GetJsonValue<std::vector<std::string>>(json, kUpdated));
+  ICEBERG_ASSIGN_OR_RAISE(response.removed,
+                          GetJsonValue<std::vector<std::string>>(json, kRemoved));
+  ICEBERG_ASSIGN_OR_RAISE(
+      response.missing, GetJsonValueOrDefault<std::vector<std::string>>(json, kMissing));
+  return response;
+}
+
+// ListTablesResponse

Review Comment:
   ```suggestion
   ```



##########
src/iceberg/catalog/rest/json_internal.cc:
##########
@@ -0,0 +1,357 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/catalog/rest/json_internal.h"
+
+#include <ranges>
+#include <string>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+#include <nlohmann/json.hpp>
+
+#include "iceberg/catalog/rest/types.h"
+#include "iceberg/json_internal.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/schema.h"
+#include "iceberg/sort_order.h"
+#include "iceberg/table_metadata.h"
+#include "iceberg/util/json_util_internal.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg::rest {
+
+namespace {
+
+// REST API JSON field constants
+constexpr std::string_view kNamespace = "namespace";
+constexpr std::string_view kNamespaces = "namespaces";
+constexpr std::string_view kProperties = "properties";
+constexpr std::string_view kRemovals = "removals";
+constexpr std::string_view kUpdates = "updates";
+constexpr std::string_view kUpdated = "updated";
+constexpr std::string_view kRemoved = "removed";
+constexpr std::string_view kMissing = "missing";
+constexpr std::string_view kNextPageToken = "next-page-token";
+constexpr std::string_view kName = "name";
+constexpr std::string_view kLocation = "location";
+constexpr std::string_view kSchema = "schema";
+constexpr std::string_view kPartitionSpec = "partition-spec";
+constexpr std::string_view kWriteOrder = "write-order";
+constexpr std::string_view kStageCreate = "stage-create";
+constexpr std::string_view kMetadataLocation = "metadata-location";
+constexpr std::string_view kOverwrite = "overwrite";
+constexpr std::string_view kSource = "source";
+constexpr std::string_view kDestination = "destination";
+constexpr std::string_view kMetadata = "metadata";
+constexpr std::string_view kConfig = "config";
+constexpr std::string_view kIdentifiers = "identifiers";
+
+}  // namespace
+
+// CreateNamespaceRequest
+nlohmann::json ToJson(const CreateNamespaceRequest& request) {
+  nlohmann::json json;
+  json[kNamespace] = request.namespace_.levels;
+  if (!request.properties.empty()) {
+    json[kProperties] = request.properties;
+  }
+  return json;
+}
+
+Result<CreateNamespaceRequest> CreateNamespaceRequestFromJson(
+    const nlohmann::json& json) {
+  CreateNamespaceRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(request.namespace_.levels,
+                          GetJsonValue<std::vector<std::string>>(json, kNamespace));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.properties,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kProperties)));
+  return request;
+}
+
+// UpdateNamespacePropertiesRequest
+nlohmann::json ToJson(const UpdateNamespacePropertiesRequest& request) {
+  // Initialize as an empty object so that when all optional fields are absent we return
+  // {} instead of null
+  nlohmann::json json = nlohmann::json::object();
+  if (!request.removals.empty()) {
+    json[kRemovals] = request.removals;
+  }
+  if (!request.updates.empty()) {
+    json[kUpdates] = request.updates;
+  }
+  return json;
+}
+
+Result<UpdateNamespacePropertiesRequest> UpdateNamespacePropertiesRequestFromJson(
+    const nlohmann::json& json) {
+  UpdateNamespacePropertiesRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.removals, GetJsonValueOrDefault<std::vector<std::string>>(json, kRemovals));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.updates,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kUpdates)));
+  return request;
+}
+
+// CreateTableRequest
+nlohmann::json ToJson(const CreateTableRequest& request) {
+  nlohmann::json json;
+  json[kName] = request.name;
+  if (!request.location.empty()) {
+    json[kLocation] = request.location;
+  }
+  json[kSchema] = ToJson(*request.schema);
+  if (request.partition_spec) {
+    json[kPartitionSpec] = ToJson(*request.partition_spec);
+  }
+  if (request.write_order) {
+    json[kWriteOrder] = ToJson(*request.write_order);
+  }
+  SetOptionalField(json, kStageCreate, request.stage_create);
+  if (!request.properties.empty()) {
+    json[kProperties] = request.properties;
+  }
+  return json;
+}
+
+Result<CreateTableRequest> CreateTableRequestFromJson(const nlohmann::json& json) {
+  CreateTableRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(request.name, GetJsonValue<std::string>(json, kName));
+  ICEBERG_ASSIGN_OR_RAISE(request.location,
+                          GetJsonValueOrDefault<std::string>(json, kLocation, ""));
+  ICEBERG_ASSIGN_OR_RAISE(auto schema_json, GetJsonValue<nlohmann::json>(json, kSchema));
+  ICEBERG_ASSIGN_OR_RAISE(request.schema, SchemaFromJson(schema_json));
+  if (json.contains(kPartitionSpec)) {
+    ICEBERG_ASSIGN_OR_RAISE(auto partition_spec_json,
+                            GetJsonValue<nlohmann::json>(json, kPartitionSpec));
+    ICEBERG_ASSIGN_OR_RAISE(request.partition_spec,
+                            PartitionSpecFromJson(request.schema, partition_spec_json));
+  } else {
+    request.partition_spec = nullptr;
+  }
+  if (json.contains(kWriteOrder)) {
+    ICEBERG_ASSIGN_OR_RAISE(auto write_order_json,
+                            GetJsonValue<nlohmann::json>(json, kWriteOrder));
+    ICEBERG_ASSIGN_OR_RAISE(request.write_order, SortOrderFromJson(write_order_json));
+  } else {
+    request.write_order = nullptr;
+  }
+  ICEBERG_ASSIGN_OR_RAISE(request.stage_create,
+                          GetJsonValueOptional<bool>(json, kStageCreate));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.properties,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kProperties)));
+  return request;
+}
+
+// RegisterTableRequest
+nlohmann::json ToJson(const RegisterTableRequest& request) {
+  nlohmann::json json;
+  json[kName] = request.name;
+  json[kMetadataLocation] = request.metadata_location;
+  if (request.overwrite) {
+    json[kOverwrite] = request.overwrite;
+  }
+  return json;
+}
+
+Result<RegisterTableRequest> RegisterTableRequestFromJson(const nlohmann::json& json) {
+  RegisterTableRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(request.name, GetJsonValue<std::string>(json, kName));
+  ICEBERG_ASSIGN_OR_RAISE(request.metadata_location,
+                          GetJsonValue<std::string>(json, kMetadataLocation));
+  // Default to false if not present
+  ICEBERG_ASSIGN_OR_RAISE(auto overwrite_opt,
+                          GetJsonValueOptional<bool>(json, kOverwrite));
+  request.overwrite = overwrite_opt.value_or(false);
+  return request;
+}
+
+// RenameTableRequest
+nlohmann::json ToJson(const RenameTableRequest& request) {
+  nlohmann::json json;
+  json[kSource] = ToJson(request.source);
+  json[kDestination] = ToJson(request.destination);
+  return json;
+}
+
+Result<RenameTableRequest> RenameTableRequestFromJson(const nlohmann::json& json) {
+  RenameTableRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(auto source_json, GetJsonValue<nlohmann::json>(json, kSource));
+  ICEBERG_ASSIGN_OR_RAISE(request.source, TableIdentifierFromJson(source_json));
+  ICEBERG_ASSIGN_OR_RAISE(auto dest_json,
+                          GetJsonValue<nlohmann::json>(json, kDestination));
+  ICEBERG_ASSIGN_OR_RAISE(request.destination, TableIdentifierFromJson(dest_json));
+  return request;
+}
+
+// LoadTableResult (used by CreateTableResponse, LoadTableResponse)
+nlohmann::json ToJson(const LoadTableResult& result) {
+  nlohmann::json json;
+  if (!result.metadata_location.empty()) {
+    json[kMetadataLocation] = result.metadata_location;
+  }
+  json[kMetadata] = ToJson(*result.metadata);
+  if (!result.config.empty()) {
+    json[kConfig] = result.config;
+  }
+  return json;
+}
+
+Result<LoadTableResult> LoadTableResultFromJson(const nlohmann::json& json) {
+  LoadTableResult result;
+  ICEBERG_ASSIGN_OR_RAISE(result.metadata_location, GetJsonValueOrDefault<std::string>(
+                                                        json, kMetadataLocation, ""));
+  ICEBERG_ASSIGN_OR_RAISE(auto metadata_json,
+                          GetJsonValue<nlohmann::json>(json, kMetadata));
+  ICEBERG_ASSIGN_OR_RAISE(result.metadata, TableMetadataFromJson(metadata_json));
+  ICEBERG_ASSIGN_OR_RAISE(
+      result.config, (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(
+                         json, kConfig)));
+  return result;
+}
+
+// ListNamespacesResponse
+nlohmann::json ToJson(const ListNamespacesResponse& response) {
+  nlohmann::json json;
+  if (!response.next_page_token.empty()) {
+    json[kNextPageToken] = response.next_page_token;
+  }
+  nlohmann::json namespaces = nlohmann::json::array();
+  for (const auto& ns : response.namespaces) {
+    namespaces.push_back(ToJson(ns));
+  }
+  json[kNamespaces] = std::move(namespaces);
+  return json;
+}
+
+Result<ListNamespacesResponse> ListNamespacesResponseFromJson(
+    const nlohmann::json& json) {
+  ListNamespacesResponse response;
+  ICEBERG_ASSIGN_OR_RAISE(response.next_page_token,
+                          GetJsonValueOrDefault<std::string>(json, kNextPageToken, ""));
+  ICEBERG_ASSIGN_OR_RAISE(auto namespaces_json,
+                          GetJsonValue<nlohmann::json>(json, kNamespaces));
+  for (const auto& ns_json : namespaces_json) {
+    ICEBERG_ASSIGN_OR_RAISE(auto ns, NamespaceFromJson(ns_json));
+    response.namespaces.push_back(std::move(ns));
+  }
+  return response;
+}
+
+// CreateNamespaceResponse

Review Comment:
   ```suggestion
   ```



##########
src/iceberg/catalog/rest/json_internal.cc:
##########
@@ -0,0 +1,357 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/catalog/rest/json_internal.h"
+
+#include <ranges>
+#include <string>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+#include <nlohmann/json.hpp>
+
+#include "iceberg/catalog/rest/types.h"
+#include "iceberg/json_internal.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/schema.h"
+#include "iceberg/sort_order.h"
+#include "iceberg/table_metadata.h"
+#include "iceberg/util/json_util_internal.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg::rest {
+
+namespace {
+
+// REST API JSON field constants
+constexpr std::string_view kNamespace = "namespace";
+constexpr std::string_view kNamespaces = "namespaces";
+constexpr std::string_view kProperties = "properties";
+constexpr std::string_view kRemovals = "removals";
+constexpr std::string_view kUpdates = "updates";
+constexpr std::string_view kUpdated = "updated";
+constexpr std::string_view kRemoved = "removed";
+constexpr std::string_view kMissing = "missing";
+constexpr std::string_view kNextPageToken = "next-page-token";
+constexpr std::string_view kName = "name";
+constexpr std::string_view kLocation = "location";
+constexpr std::string_view kSchema = "schema";
+constexpr std::string_view kPartitionSpec = "partition-spec";
+constexpr std::string_view kWriteOrder = "write-order";
+constexpr std::string_view kStageCreate = "stage-create";
+constexpr std::string_view kMetadataLocation = "metadata-location";
+constexpr std::string_view kOverwrite = "overwrite";
+constexpr std::string_view kSource = "source";
+constexpr std::string_view kDestination = "destination";
+constexpr std::string_view kMetadata = "metadata";
+constexpr std::string_view kConfig = "config";
+constexpr std::string_view kIdentifiers = "identifiers";
+
+}  // namespace
+
+// CreateNamespaceRequest
+nlohmann::json ToJson(const CreateNamespaceRequest& request) {
+  nlohmann::json json;
+  json[kNamespace] = request.namespace_.levels;
+  if (!request.properties.empty()) {
+    json[kProperties] = request.properties;
+  }
+  return json;
+}
+
+Result<CreateNamespaceRequest> CreateNamespaceRequestFromJson(
+    const nlohmann::json& json) {
+  CreateNamespaceRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(request.namespace_.levels,
+                          GetJsonValue<std::vector<std::string>>(json, kNamespace));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.properties,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kProperties)));
+  return request;
+}
+
+// UpdateNamespacePropertiesRequest
+nlohmann::json ToJson(const UpdateNamespacePropertiesRequest& request) {
+  // Initialize as an empty object so that when all optional fields are absent we return
+  // {} instead of null
+  nlohmann::json json = nlohmann::json::object();
+  if (!request.removals.empty()) {
+    json[kRemovals] = request.removals;
+  }
+  if (!request.updates.empty()) {
+    json[kUpdates] = request.updates;
+  }
+  return json;
+}
+
+Result<UpdateNamespacePropertiesRequest> UpdateNamespacePropertiesRequestFromJson(
+    const nlohmann::json& json) {
+  UpdateNamespacePropertiesRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.removals, GetJsonValueOrDefault<std::vector<std::string>>(json, kRemovals));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.updates,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kUpdates)));
+  return request;
+}
+
+// CreateTableRequest
+nlohmann::json ToJson(const CreateTableRequest& request) {
+  nlohmann::json json;
+  json[kName] = request.name;
+  if (!request.location.empty()) {
+    json[kLocation] = request.location;
+  }
+  json[kSchema] = ToJson(*request.schema);
+  if (request.partition_spec) {
+    json[kPartitionSpec] = ToJson(*request.partition_spec);
+  }
+  if (request.write_order) {
+    json[kWriteOrder] = ToJson(*request.write_order);
+  }
+  SetOptionalField(json, kStageCreate, request.stage_create);
+  if (!request.properties.empty()) {
+    json[kProperties] = request.properties;
+  }
+  return json;
+}
+
+Result<CreateTableRequest> CreateTableRequestFromJson(const nlohmann::json& json) {
+  CreateTableRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(request.name, GetJsonValue<std::string>(json, kName));
+  ICEBERG_ASSIGN_OR_RAISE(request.location,
+                          GetJsonValueOrDefault<std::string>(json, kLocation, ""));
+  ICEBERG_ASSIGN_OR_RAISE(auto schema_json, GetJsonValue<nlohmann::json>(json, kSchema));
+  ICEBERG_ASSIGN_OR_RAISE(request.schema, SchemaFromJson(schema_json));
+  if (json.contains(kPartitionSpec)) {
+    ICEBERG_ASSIGN_OR_RAISE(auto partition_spec_json,
+                            GetJsonValue<nlohmann::json>(json, kPartitionSpec));
+    ICEBERG_ASSIGN_OR_RAISE(request.partition_spec,
+                            PartitionSpecFromJson(request.schema, partition_spec_json));
+  } else {
+    request.partition_spec = nullptr;
+  }
+  if (json.contains(kWriteOrder)) {
+    ICEBERG_ASSIGN_OR_RAISE(auto write_order_json,
+                            GetJsonValue<nlohmann::json>(json, kWriteOrder));
+    ICEBERG_ASSIGN_OR_RAISE(request.write_order, SortOrderFromJson(write_order_json));
+  } else {
+    request.write_order = nullptr;
+  }
+  ICEBERG_ASSIGN_OR_RAISE(request.stage_create,
+                          GetJsonValueOptional<bool>(json, kStageCreate));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.properties,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kProperties)));
+  return request;
+}
+
+// RegisterTableRequest
+nlohmann::json ToJson(const RegisterTableRequest& request) {
+  nlohmann::json json;
+  json[kName] = request.name;
+  json[kMetadataLocation] = request.metadata_location;
+  if (request.overwrite) {
+    json[kOverwrite] = request.overwrite;
+  }
+  return json;
+}
+
+Result<RegisterTableRequest> RegisterTableRequestFromJson(const nlohmann::json& json) {
+  RegisterTableRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(request.name, GetJsonValue<std::string>(json, kName));
+  ICEBERG_ASSIGN_OR_RAISE(request.metadata_location,
+                          GetJsonValue<std::string>(json, kMetadataLocation));
+  // Default to false if not present
+  ICEBERG_ASSIGN_OR_RAISE(auto overwrite_opt,
+                          GetJsonValueOptional<bool>(json, kOverwrite));
+  request.overwrite = overwrite_opt.value_or(false);
+  return request;
+}
+
+// RenameTableRequest
+nlohmann::json ToJson(const RenameTableRequest& request) {
+  nlohmann::json json;
+  json[kSource] = ToJson(request.source);
+  json[kDestination] = ToJson(request.destination);
+  return json;
+}
+
+Result<RenameTableRequest> RenameTableRequestFromJson(const nlohmann::json& json) {
+  RenameTableRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(auto source_json, GetJsonValue<nlohmann::json>(json, kSource));
+  ICEBERG_ASSIGN_OR_RAISE(request.source, TableIdentifierFromJson(source_json));
+  ICEBERG_ASSIGN_OR_RAISE(auto dest_json,
+                          GetJsonValue<nlohmann::json>(json, kDestination));
+  ICEBERG_ASSIGN_OR_RAISE(request.destination, TableIdentifierFromJson(dest_json));
+  return request;
+}
+
+// LoadTableResult (used by CreateTableResponse, LoadTableResponse)
+nlohmann::json ToJson(const LoadTableResult& result) {
+  nlohmann::json json;
+  if (!result.metadata_location.empty()) {
+    json[kMetadataLocation] = result.metadata_location;
+  }
+  json[kMetadata] = ToJson(*result.metadata);
+  if (!result.config.empty()) {
+    json[kConfig] = result.config;
+  }
+  return json;
+}
+
+Result<LoadTableResult> LoadTableResultFromJson(const nlohmann::json& json) {
+  LoadTableResult result;
+  ICEBERG_ASSIGN_OR_RAISE(result.metadata_location, GetJsonValueOrDefault<std::string>(
+                                                        json, kMetadataLocation, ""));
+  ICEBERG_ASSIGN_OR_RAISE(auto metadata_json,
+                          GetJsonValue<nlohmann::json>(json, kMetadata));
+  ICEBERG_ASSIGN_OR_RAISE(result.metadata, TableMetadataFromJson(metadata_json));
+  ICEBERG_ASSIGN_OR_RAISE(
+      result.config, (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(
+                         json, kConfig)));
+  return result;
+}
+
+// ListNamespacesResponse

Review Comment:
   ```suggestion
   ```



##########
src/iceberg/catalog/rest/json_internal.cc:
##########
@@ -0,0 +1,357 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/catalog/rest/json_internal.h"
+
+#include <ranges>
+#include <string>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+#include <nlohmann/json.hpp>
+
+#include "iceberg/catalog/rest/types.h"
+#include "iceberg/json_internal.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/schema.h"
+#include "iceberg/sort_order.h"
+#include "iceberg/table_metadata.h"
+#include "iceberg/util/json_util_internal.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg::rest {
+
+namespace {
+
+// REST API JSON field constants
+constexpr std::string_view kNamespace = "namespace";
+constexpr std::string_view kNamespaces = "namespaces";
+constexpr std::string_view kProperties = "properties";
+constexpr std::string_view kRemovals = "removals";
+constexpr std::string_view kUpdates = "updates";
+constexpr std::string_view kUpdated = "updated";
+constexpr std::string_view kRemoved = "removed";
+constexpr std::string_view kMissing = "missing";
+constexpr std::string_view kNextPageToken = "next-page-token";
+constexpr std::string_view kName = "name";
+constexpr std::string_view kLocation = "location";
+constexpr std::string_view kSchema = "schema";
+constexpr std::string_view kPartitionSpec = "partition-spec";
+constexpr std::string_view kWriteOrder = "write-order";
+constexpr std::string_view kStageCreate = "stage-create";
+constexpr std::string_view kMetadataLocation = "metadata-location";
+constexpr std::string_view kOverwrite = "overwrite";
+constexpr std::string_view kSource = "source";
+constexpr std::string_view kDestination = "destination";
+constexpr std::string_view kMetadata = "metadata";
+constexpr std::string_view kConfig = "config";
+constexpr std::string_view kIdentifiers = "identifiers";
+
+}  // namespace
+
+// CreateNamespaceRequest
+nlohmann::json ToJson(const CreateNamespaceRequest& request) {
+  nlohmann::json json;
+  json[kNamespace] = request.namespace_.levels;
+  if (!request.properties.empty()) {
+    json[kProperties] = request.properties;
+  }
+  return json;
+}
+
+Result<CreateNamespaceRequest> CreateNamespaceRequestFromJson(
+    const nlohmann::json& json) {
+  CreateNamespaceRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(request.namespace_.levels,
+                          GetJsonValue<std::vector<std::string>>(json, kNamespace));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.properties,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kProperties)));
+  return request;
+}
+
+// UpdateNamespacePropertiesRequest
+nlohmann::json ToJson(const UpdateNamespacePropertiesRequest& request) {
+  // Initialize as an empty object so that when all optional fields are absent we return
+  // {} instead of null
+  nlohmann::json json = nlohmann::json::object();
+  if (!request.removals.empty()) {
+    json[kRemovals] = request.removals;
+  }
+  if (!request.updates.empty()) {
+    json[kUpdates] = request.updates;
+  }
+  return json;
+}
+
+Result<UpdateNamespacePropertiesRequest> UpdateNamespacePropertiesRequestFromJson(
+    const nlohmann::json& json) {
+  UpdateNamespacePropertiesRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.removals, GetJsonValueOrDefault<std::vector<std::string>>(json, kRemovals));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.updates,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kUpdates)));
+  return request;
+}
+
+// CreateTableRequest
+nlohmann::json ToJson(const CreateTableRequest& request) {
+  nlohmann::json json;
+  json[kName] = request.name;
+  if (!request.location.empty()) {
+    json[kLocation] = request.location;
+  }
+  json[kSchema] = ToJson(*request.schema);
+  if (request.partition_spec) {
+    json[kPartitionSpec] = ToJson(*request.partition_spec);
+  }
+  if (request.write_order) {
+    json[kWriteOrder] = ToJson(*request.write_order);
+  }
+  SetOptionalField(json, kStageCreate, request.stage_create);
+  if (!request.properties.empty()) {
+    json[kProperties] = request.properties;
+  }
+  return json;
+}
+
+Result<CreateTableRequest> CreateTableRequestFromJson(const nlohmann::json& json) {
+  CreateTableRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(request.name, GetJsonValue<std::string>(json, kName));
+  ICEBERG_ASSIGN_OR_RAISE(request.location,
+                          GetJsonValueOrDefault<std::string>(json, kLocation, ""));
+  ICEBERG_ASSIGN_OR_RAISE(auto schema_json, GetJsonValue<nlohmann::json>(json, kSchema));
+  ICEBERG_ASSIGN_OR_RAISE(request.schema, SchemaFromJson(schema_json));
+  if (json.contains(kPartitionSpec)) {
+    ICEBERG_ASSIGN_OR_RAISE(auto partition_spec_json,
+                            GetJsonValue<nlohmann::json>(json, kPartitionSpec));
+    ICEBERG_ASSIGN_OR_RAISE(request.partition_spec,
+                            PartitionSpecFromJson(request.schema, partition_spec_json));
+  } else {
+    request.partition_spec = nullptr;
+  }
+  if (json.contains(kWriteOrder)) {
+    ICEBERG_ASSIGN_OR_RAISE(auto write_order_json,
+                            GetJsonValue<nlohmann::json>(json, kWriteOrder));
+    ICEBERG_ASSIGN_OR_RAISE(request.write_order, SortOrderFromJson(write_order_json));
+  } else {
+    request.write_order = nullptr;
+  }
+  ICEBERG_ASSIGN_OR_RAISE(request.stage_create,
+                          GetJsonValueOptional<bool>(json, kStageCreate));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.properties,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kProperties)));
+  return request;
+}
+
+// RegisterTableRequest
+nlohmann::json ToJson(const RegisterTableRequest& request) {
+  nlohmann::json json;
+  json[kName] = request.name;
+  json[kMetadataLocation] = request.metadata_location;
+  if (request.overwrite) {
+    json[kOverwrite] = request.overwrite;
+  }
+  return json;
+}
+
+Result<RegisterTableRequest> RegisterTableRequestFromJson(const nlohmann::json& json) {
+  RegisterTableRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(request.name, GetJsonValue<std::string>(json, kName));
+  ICEBERG_ASSIGN_OR_RAISE(request.metadata_location,
+                          GetJsonValue<std::string>(json, kMetadataLocation));
+  // Default to false if not present
+  ICEBERG_ASSIGN_OR_RAISE(auto overwrite_opt,
+                          GetJsonValueOptional<bool>(json, kOverwrite));
+  request.overwrite = overwrite_opt.value_or(false);
+  return request;
+}
+
+// RenameTableRequest
+nlohmann::json ToJson(const RenameTableRequest& request) {
+  nlohmann::json json;
+  json[kSource] = ToJson(request.source);
+  json[kDestination] = ToJson(request.destination);
+  return json;
+}
+
+Result<RenameTableRequest> RenameTableRequestFromJson(const nlohmann::json& json) {
+  RenameTableRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(auto source_json, GetJsonValue<nlohmann::json>(json, kSource));
+  ICEBERG_ASSIGN_OR_RAISE(request.source, TableIdentifierFromJson(source_json));
+  ICEBERG_ASSIGN_OR_RAISE(auto dest_json,
+                          GetJsonValue<nlohmann::json>(json, kDestination));
+  ICEBERG_ASSIGN_OR_RAISE(request.destination, TableIdentifierFromJson(dest_json));
+  return request;
+}
+
+// LoadTableResult (used by CreateTableResponse, LoadTableResponse)
+nlohmann::json ToJson(const LoadTableResult& result) {
+  nlohmann::json json;
+  if (!result.metadata_location.empty()) {
+    json[kMetadataLocation] = result.metadata_location;
+  }
+  json[kMetadata] = ToJson(*result.metadata);
+  if (!result.config.empty()) {
+    json[kConfig] = result.config;
+  }
+  return json;
+}
+
+Result<LoadTableResult> LoadTableResultFromJson(const nlohmann::json& json) {
+  LoadTableResult result;
+  ICEBERG_ASSIGN_OR_RAISE(result.metadata_location, GetJsonValueOrDefault<std::string>(
+                                                        json, kMetadataLocation, ""));
+  ICEBERG_ASSIGN_OR_RAISE(auto metadata_json,
+                          GetJsonValue<nlohmann::json>(json, kMetadata));
+  ICEBERG_ASSIGN_OR_RAISE(result.metadata, TableMetadataFromJson(metadata_json));
+  ICEBERG_ASSIGN_OR_RAISE(
+      result.config, (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(
+                         json, kConfig)));
+  return result;
+}
+
+// ListNamespacesResponse
+nlohmann::json ToJson(const ListNamespacesResponse& response) {
+  nlohmann::json json;
+  if (!response.next_page_token.empty()) {
+    json[kNextPageToken] = response.next_page_token;
+  }
+  nlohmann::json namespaces = nlohmann::json::array();
+  for (const auto& ns : response.namespaces) {
+    namespaces.push_back(ToJson(ns));
+  }
+  json[kNamespaces] = std::move(namespaces);
+  return json;
+}
+
+Result<ListNamespacesResponse> ListNamespacesResponseFromJson(
+    const nlohmann::json& json) {
+  ListNamespacesResponse response;
+  ICEBERG_ASSIGN_OR_RAISE(response.next_page_token,
+                          GetJsonValueOrDefault<std::string>(json, kNextPageToken, ""));
+  ICEBERG_ASSIGN_OR_RAISE(auto namespaces_json,
+                          GetJsonValue<nlohmann::json>(json, kNamespaces));
+  for (const auto& ns_json : namespaces_json) {
+    ICEBERG_ASSIGN_OR_RAISE(auto ns, NamespaceFromJson(ns_json));
+    response.namespaces.push_back(std::move(ns));
+  }
+  return response;
+}
+
+// CreateNamespaceResponse
+nlohmann::json ToJson(const CreateNamespaceResponse& response) {
+  nlohmann::json json;
+  json[kNamespace] = response.namespace_.levels;
+  if (!response.properties.empty()) {
+    json[kProperties] = response.properties;
+  }
+  return json;
+}
+
+Result<CreateNamespaceResponse> CreateNamespaceResponseFromJson(
+    const nlohmann::json& json) {
+  CreateNamespaceResponse response;
+  ICEBERG_ASSIGN_OR_RAISE(response.namespace_.levels,
+                          GetJsonValue<std::vector<std::string>>(json, kNamespace));
+  ICEBERG_ASSIGN_OR_RAISE(
+      response.properties,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kProperties)));
+  return response;
+}
+
+// GetNamespaceResponse

Review Comment:
   ```suggestion
   ```



##########
src/iceberg/catalog/rest/json_internal.cc:
##########
@@ -0,0 +1,357 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/catalog/rest/json_internal.h"
+
+#include <ranges>
+#include <string>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+#include <nlohmann/json.hpp>
+
+#include "iceberg/catalog/rest/types.h"
+#include "iceberg/json_internal.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/schema.h"
+#include "iceberg/sort_order.h"
+#include "iceberg/table_metadata.h"
+#include "iceberg/util/json_util_internal.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg::rest {
+
+namespace {
+
+// REST API JSON field constants
+constexpr std::string_view kNamespace = "namespace";
+constexpr std::string_view kNamespaces = "namespaces";
+constexpr std::string_view kProperties = "properties";
+constexpr std::string_view kRemovals = "removals";
+constexpr std::string_view kUpdates = "updates";
+constexpr std::string_view kUpdated = "updated";
+constexpr std::string_view kRemoved = "removed";
+constexpr std::string_view kMissing = "missing";
+constexpr std::string_view kNextPageToken = "next-page-token";
+constexpr std::string_view kName = "name";
+constexpr std::string_view kLocation = "location";
+constexpr std::string_view kSchema = "schema";
+constexpr std::string_view kPartitionSpec = "partition-spec";
+constexpr std::string_view kWriteOrder = "write-order";
+constexpr std::string_view kStageCreate = "stage-create";
+constexpr std::string_view kMetadataLocation = "metadata-location";
+constexpr std::string_view kOverwrite = "overwrite";
+constexpr std::string_view kSource = "source";
+constexpr std::string_view kDestination = "destination";
+constexpr std::string_view kMetadata = "metadata";
+constexpr std::string_view kConfig = "config";
+constexpr std::string_view kIdentifiers = "identifiers";
+
+}  // namespace
+
+// CreateNamespaceRequest
+nlohmann::json ToJson(const CreateNamespaceRequest& request) {
+  nlohmann::json json;
+  json[kNamespace] = request.namespace_.levels;
+  if (!request.properties.empty()) {
+    json[kProperties] = request.properties;
+  }
+  return json;
+}
+
+Result<CreateNamespaceRequest> CreateNamespaceRequestFromJson(
+    const nlohmann::json& json) {
+  CreateNamespaceRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(request.namespace_.levels,
+                          GetJsonValue<std::vector<std::string>>(json, kNamespace));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.properties,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kProperties)));
+  return request;
+}
+
+// UpdateNamespacePropertiesRequest
+nlohmann::json ToJson(const UpdateNamespacePropertiesRequest& request) {
+  // Initialize as an empty object so that when all optional fields are absent we return
+  // {} instead of null
+  nlohmann::json json = nlohmann::json::object();
+  if (!request.removals.empty()) {
+    json[kRemovals] = request.removals;
+  }
+  if (!request.updates.empty()) {
+    json[kUpdates] = request.updates;
+  }
+  return json;
+}
+
+Result<UpdateNamespacePropertiesRequest> UpdateNamespacePropertiesRequestFromJson(
+    const nlohmann::json& json) {
+  UpdateNamespacePropertiesRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.removals, GetJsonValueOrDefault<std::vector<std::string>>(json, kRemovals));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.updates,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kUpdates)));
+  return request;
+}
+
+// CreateTableRequest
+nlohmann::json ToJson(const CreateTableRequest& request) {
+  nlohmann::json json;
+  json[kName] = request.name;
+  if (!request.location.empty()) {
+    json[kLocation] = request.location;
+  }
+  json[kSchema] = ToJson(*request.schema);
+  if (request.partition_spec) {
+    json[kPartitionSpec] = ToJson(*request.partition_spec);
+  }
+  if (request.write_order) {
+    json[kWriteOrder] = ToJson(*request.write_order);
+  }
+  SetOptionalField(json, kStageCreate, request.stage_create);
+  if (!request.properties.empty()) {
+    json[kProperties] = request.properties;
+  }
+  return json;
+}
+
+Result<CreateTableRequest> CreateTableRequestFromJson(const nlohmann::json& json) {
+  CreateTableRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(request.name, GetJsonValue<std::string>(json, kName));
+  ICEBERG_ASSIGN_OR_RAISE(request.location,
+                          GetJsonValueOrDefault<std::string>(json, kLocation, ""));
+  ICEBERG_ASSIGN_OR_RAISE(auto schema_json, GetJsonValue<nlohmann::json>(json, kSchema));
+  ICEBERG_ASSIGN_OR_RAISE(request.schema, SchemaFromJson(schema_json));
+  if (json.contains(kPartitionSpec)) {
+    ICEBERG_ASSIGN_OR_RAISE(auto partition_spec_json,
+                            GetJsonValue<nlohmann::json>(json, kPartitionSpec));
+    ICEBERG_ASSIGN_OR_RAISE(request.partition_spec,
+                            PartitionSpecFromJson(request.schema, partition_spec_json));
+  } else {
+    request.partition_spec = nullptr;
+  }
+  if (json.contains(kWriteOrder)) {
+    ICEBERG_ASSIGN_OR_RAISE(auto write_order_json,
+                            GetJsonValue<nlohmann::json>(json, kWriteOrder));
+    ICEBERG_ASSIGN_OR_RAISE(request.write_order, SortOrderFromJson(write_order_json));
+  } else {
+    request.write_order = nullptr;
+  }
+  ICEBERG_ASSIGN_OR_RAISE(request.stage_create,
+                          GetJsonValueOptional<bool>(json, kStageCreate));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.properties,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kProperties)));
+  return request;
+}
+
+// RegisterTableRequest
+nlohmann::json ToJson(const RegisterTableRequest& request) {
+  nlohmann::json json;
+  json[kName] = request.name;
+  json[kMetadataLocation] = request.metadata_location;
+  if (request.overwrite) {
+    json[kOverwrite] = request.overwrite;
+  }
+  return json;
+}
+
+Result<RegisterTableRequest> RegisterTableRequestFromJson(const nlohmann::json& json) {
+  RegisterTableRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(request.name, GetJsonValue<std::string>(json, kName));
+  ICEBERG_ASSIGN_OR_RAISE(request.metadata_location,
+                          GetJsonValue<std::string>(json, kMetadataLocation));
+  // Default to false if not present
+  ICEBERG_ASSIGN_OR_RAISE(auto overwrite_opt,
+                          GetJsonValueOptional<bool>(json, kOverwrite));
+  request.overwrite = overwrite_opt.value_or(false);
+  return request;
+}
+
+// RenameTableRequest
+nlohmann::json ToJson(const RenameTableRequest& request) {
+  nlohmann::json json;
+  json[kSource] = ToJson(request.source);
+  json[kDestination] = ToJson(request.destination);
+  return json;
+}
+
+Result<RenameTableRequest> RenameTableRequestFromJson(const nlohmann::json& json) {
+  RenameTableRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(auto source_json, GetJsonValue<nlohmann::json>(json, kSource));
+  ICEBERG_ASSIGN_OR_RAISE(request.source, TableIdentifierFromJson(source_json));
+  ICEBERG_ASSIGN_OR_RAISE(auto dest_json,
+                          GetJsonValue<nlohmann::json>(json, kDestination));
+  ICEBERG_ASSIGN_OR_RAISE(request.destination, TableIdentifierFromJson(dest_json));
+  return request;
+}
+
+// LoadTableResult (used by CreateTableResponse, LoadTableResponse)
+nlohmann::json ToJson(const LoadTableResult& result) {
+  nlohmann::json json;
+  if (!result.metadata_location.empty()) {
+    json[kMetadataLocation] = result.metadata_location;
+  }
+  json[kMetadata] = ToJson(*result.metadata);
+  if (!result.config.empty()) {
+    json[kConfig] = result.config;
+  }
+  return json;
+}
+
+Result<LoadTableResult> LoadTableResultFromJson(const nlohmann::json& json) {
+  LoadTableResult result;
+  ICEBERG_ASSIGN_OR_RAISE(result.metadata_location, GetJsonValueOrDefault<std::string>(
+                                                        json, kMetadataLocation, ""));
+  ICEBERG_ASSIGN_OR_RAISE(auto metadata_json,
+                          GetJsonValue<nlohmann::json>(json, kMetadata));
+  ICEBERG_ASSIGN_OR_RAISE(result.metadata, TableMetadataFromJson(metadata_json));
+  ICEBERG_ASSIGN_OR_RAISE(
+      result.config, (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(
+                         json, kConfig)));
+  return result;
+}
+
+// ListNamespacesResponse
+nlohmann::json ToJson(const ListNamespacesResponse& response) {
+  nlohmann::json json;
+  if (!response.next_page_token.empty()) {
+    json[kNextPageToken] = response.next_page_token;
+  }
+  nlohmann::json namespaces = nlohmann::json::array();
+  for (const auto& ns : response.namespaces) {
+    namespaces.push_back(ToJson(ns));
+  }
+  json[kNamespaces] = std::move(namespaces);
+  return json;
+}
+
+Result<ListNamespacesResponse> ListNamespacesResponseFromJson(
+    const nlohmann::json& json) {
+  ListNamespacesResponse response;
+  ICEBERG_ASSIGN_OR_RAISE(response.next_page_token,
+                          GetJsonValueOrDefault<std::string>(json, kNextPageToken, ""));
+  ICEBERG_ASSIGN_OR_RAISE(auto namespaces_json,
+                          GetJsonValue<nlohmann::json>(json, kNamespaces));
+  for (const auto& ns_json : namespaces_json) {
+    ICEBERG_ASSIGN_OR_RAISE(auto ns, NamespaceFromJson(ns_json));
+    response.namespaces.push_back(std::move(ns));
+  }
+  return response;
+}
+
+// CreateNamespaceResponse
+nlohmann::json ToJson(const CreateNamespaceResponse& response) {
+  nlohmann::json json;
+  json[kNamespace] = response.namespace_.levels;
+  if (!response.properties.empty()) {
+    json[kProperties] = response.properties;
+  }
+  return json;
+}
+
+Result<CreateNamespaceResponse> CreateNamespaceResponseFromJson(
+    const nlohmann::json& json) {
+  CreateNamespaceResponse response;
+  ICEBERG_ASSIGN_OR_RAISE(response.namespace_.levels,
+                          GetJsonValue<std::vector<std::string>>(json, kNamespace));
+  ICEBERG_ASSIGN_OR_RAISE(
+      response.properties,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kProperties)));
+  return response;
+}
+
+// GetNamespaceResponse
+nlohmann::json ToJson(const GetNamespaceResponse& response) {
+  nlohmann::json json;
+  json[kNamespace] = response.namespace_.levels;
+  if (!response.properties.empty()) {
+    json[kProperties] = response.properties;
+  }
+  return json;
+}
+
+Result<GetNamespaceResponse> GetNamespaceResponseFromJson(const nlohmann::json& json) {
+  GetNamespaceResponse response;
+  ICEBERG_ASSIGN_OR_RAISE(response.namespace_.levels,
+                          GetJsonValue<std::vector<std::string>>(json, kNamespace));
+  ICEBERG_ASSIGN_OR_RAISE(
+      response.properties,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kProperties)));
+  return response;
+}
+
+// UpdateNamespacePropertiesResponse

Review Comment:
   ```suggestion
   ```



##########
src/iceberg/catalog/rest/json_internal.cc:
##########
@@ -0,0 +1,357 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/catalog/rest/json_internal.h"
+
+#include <ranges>
+#include <string>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+#include <nlohmann/json.hpp>
+
+#include "iceberg/catalog/rest/types.h"
+#include "iceberg/json_internal.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/schema.h"
+#include "iceberg/sort_order.h"
+#include "iceberg/table_metadata.h"
+#include "iceberg/util/json_util_internal.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg::rest {
+
+namespace {
+
+// REST API JSON field constants
+constexpr std::string_view kNamespace = "namespace";
+constexpr std::string_view kNamespaces = "namespaces";
+constexpr std::string_view kProperties = "properties";
+constexpr std::string_view kRemovals = "removals";
+constexpr std::string_view kUpdates = "updates";
+constexpr std::string_view kUpdated = "updated";
+constexpr std::string_view kRemoved = "removed";
+constexpr std::string_view kMissing = "missing";
+constexpr std::string_view kNextPageToken = "next-page-token";
+constexpr std::string_view kName = "name";
+constexpr std::string_view kLocation = "location";
+constexpr std::string_view kSchema = "schema";
+constexpr std::string_view kPartitionSpec = "partition-spec";
+constexpr std::string_view kWriteOrder = "write-order";
+constexpr std::string_view kStageCreate = "stage-create";
+constexpr std::string_view kMetadataLocation = "metadata-location";
+constexpr std::string_view kOverwrite = "overwrite";
+constexpr std::string_view kSource = "source";
+constexpr std::string_view kDestination = "destination";
+constexpr std::string_view kMetadata = "metadata";
+constexpr std::string_view kConfig = "config";
+constexpr std::string_view kIdentifiers = "identifiers";
+
+}  // namespace
+
+// CreateNamespaceRequest
+nlohmann::json ToJson(const CreateNamespaceRequest& request) {
+  nlohmann::json json;
+  json[kNamespace] = request.namespace_.levels;
+  if (!request.properties.empty()) {
+    json[kProperties] = request.properties;
+  }
+  return json;
+}
+
+Result<CreateNamespaceRequest> CreateNamespaceRequestFromJson(
+    const nlohmann::json& json) {
+  CreateNamespaceRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(request.namespace_.levels,
+                          GetJsonValue<std::vector<std::string>>(json, kNamespace));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.properties,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kProperties)));
+  return request;
+}
+
+// UpdateNamespacePropertiesRequest
+nlohmann::json ToJson(const UpdateNamespacePropertiesRequest& request) {
+  // Initialize as an empty object so that when all optional fields are absent we return
+  // {} instead of null
+  nlohmann::json json = nlohmann::json::object();
+  if (!request.removals.empty()) {
+    json[kRemovals] = request.removals;
+  }
+  if (!request.updates.empty()) {
+    json[kUpdates] = request.updates;
+  }
+  return json;
+}
+
+Result<UpdateNamespacePropertiesRequest> UpdateNamespacePropertiesRequestFromJson(
+    const nlohmann::json& json) {
+  UpdateNamespacePropertiesRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.removals, GetJsonValueOrDefault<std::vector<std::string>>(json, kRemovals));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.updates,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kUpdates)));
+  return request;
+}
+
+// CreateTableRequest
+nlohmann::json ToJson(const CreateTableRequest& request) {
+  nlohmann::json json;
+  json[kName] = request.name;
+  if (!request.location.empty()) {
+    json[kLocation] = request.location;
+  }
+  json[kSchema] = ToJson(*request.schema);
+  if (request.partition_spec) {
+    json[kPartitionSpec] = ToJson(*request.partition_spec);
+  }
+  if (request.write_order) {
+    json[kWriteOrder] = ToJson(*request.write_order);
+  }
+  SetOptionalField(json, kStageCreate, request.stage_create);
+  if (!request.properties.empty()) {
+    json[kProperties] = request.properties;
+  }
+  return json;
+}
+
+Result<CreateTableRequest> CreateTableRequestFromJson(const nlohmann::json& json) {
+  CreateTableRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(request.name, GetJsonValue<std::string>(json, kName));
+  ICEBERG_ASSIGN_OR_RAISE(request.location,
+                          GetJsonValueOrDefault<std::string>(json, kLocation, ""));
+  ICEBERG_ASSIGN_OR_RAISE(auto schema_json, GetJsonValue<nlohmann::json>(json, kSchema));
+  ICEBERG_ASSIGN_OR_RAISE(request.schema, SchemaFromJson(schema_json));
+  if (json.contains(kPartitionSpec)) {
+    ICEBERG_ASSIGN_OR_RAISE(auto partition_spec_json,
+                            GetJsonValue<nlohmann::json>(json, kPartitionSpec));
+    ICEBERG_ASSIGN_OR_RAISE(request.partition_spec,
+                            PartitionSpecFromJson(request.schema, partition_spec_json));
+  } else {
+    request.partition_spec = nullptr;
+  }
+  if (json.contains(kWriteOrder)) {
+    ICEBERG_ASSIGN_OR_RAISE(auto write_order_json,
+                            GetJsonValue<nlohmann::json>(json, kWriteOrder));
+    ICEBERG_ASSIGN_OR_RAISE(request.write_order, SortOrderFromJson(write_order_json));
+  } else {
+    request.write_order = nullptr;
+  }
+  ICEBERG_ASSIGN_OR_RAISE(request.stage_create,
+                          GetJsonValueOptional<bool>(json, kStageCreate));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.properties,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kProperties)));
+  return request;
+}
+
+// RegisterTableRequest
+nlohmann::json ToJson(const RegisterTableRequest& request) {
+  nlohmann::json json;
+  json[kName] = request.name;
+  json[kMetadataLocation] = request.metadata_location;
+  if (request.overwrite) {
+    json[kOverwrite] = request.overwrite;
+  }
+  return json;
+}
+
+Result<RegisterTableRequest> RegisterTableRequestFromJson(const nlohmann::json& json) {
+  RegisterTableRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(request.name, GetJsonValue<std::string>(json, kName));
+  ICEBERG_ASSIGN_OR_RAISE(request.metadata_location,
+                          GetJsonValue<std::string>(json, kMetadataLocation));
+  // Default to false if not present
+  ICEBERG_ASSIGN_OR_RAISE(auto overwrite_opt,
+                          GetJsonValueOptional<bool>(json, kOverwrite));
+  request.overwrite = overwrite_opt.value_or(false);
+  return request;
+}
+
+// RenameTableRequest

Review Comment:
   ```suggestion
   ```



##########
src/iceberg/test/rest_json_internal_test.cc:
##########
@@ -0,0 +1,801 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include <algorithm>
+#include <string>
+#include <unordered_map>
+#include <vector>
+
+#include <gmock/gmock.h>
+#include <gtest/gtest.h>
+#include <nlohmann/json.hpp>
+
+#include "iceberg/catalog/rest/json_internal.h"
+#include "iceberg/catalog/rest/types.h"
+#include "iceberg/table_identifier.h"
+
+namespace iceberg::rest {
+
+namespace {
+
+// Helper to compare two TableIdentifier objects
+bool TableIdentifierEqual(const TableIdentifier& a, const TableIdentifier& b) {
+  return a.ns.levels == b.ns.levels && a.name == b.name;
+}
+
+// Helper to compare string vectors (order-independent)
+bool StringVectorEqual(const std::vector<std::string>& a,
+                       const std::vector<std::string>& b) {
+  return std::ranges::is_permutation(a, b);
+}
+
+}  // namespace

Review Comment:
   ```suggestion
   ```
   
   We don't need them. They are auto-generated by compilers.



##########
src/iceberg/catalog/rest/json_internal.cc:
##########
@@ -0,0 +1,357 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/catalog/rest/json_internal.h"
+
+#include <ranges>
+#include <string>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+#include <nlohmann/json.hpp>
+
+#include "iceberg/catalog/rest/types.h"
+#include "iceberg/json_internal.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/schema.h"
+#include "iceberg/sort_order.h"
+#include "iceberg/table_metadata.h"
+#include "iceberg/util/json_util_internal.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg::rest {
+
+namespace {
+
+// REST API JSON field constants
+constexpr std::string_view kNamespace = "namespace";
+constexpr std::string_view kNamespaces = "namespaces";
+constexpr std::string_view kProperties = "properties";
+constexpr std::string_view kRemovals = "removals";
+constexpr std::string_view kUpdates = "updates";
+constexpr std::string_view kUpdated = "updated";
+constexpr std::string_view kRemoved = "removed";
+constexpr std::string_view kMissing = "missing";
+constexpr std::string_view kNextPageToken = "next-page-token";
+constexpr std::string_view kName = "name";
+constexpr std::string_view kLocation = "location";
+constexpr std::string_view kSchema = "schema";
+constexpr std::string_view kPartitionSpec = "partition-spec";
+constexpr std::string_view kWriteOrder = "write-order";
+constexpr std::string_view kStageCreate = "stage-create";
+constexpr std::string_view kMetadataLocation = "metadata-location";
+constexpr std::string_view kOverwrite = "overwrite";
+constexpr std::string_view kSource = "source";
+constexpr std::string_view kDestination = "destination";
+constexpr std::string_view kMetadata = "metadata";
+constexpr std::string_view kConfig = "config";
+constexpr std::string_view kIdentifiers = "identifiers";
+
+}  // namespace
+
+// CreateNamespaceRequest
+nlohmann::json ToJson(const CreateNamespaceRequest& request) {
+  nlohmann::json json;
+  json[kNamespace] = request.namespace_.levels;
+  if (!request.properties.empty()) {
+    json[kProperties] = request.properties;
+  }
+  return json;
+}
+
+Result<CreateNamespaceRequest> CreateNamespaceRequestFromJson(
+    const nlohmann::json& json) {
+  CreateNamespaceRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(request.namespace_.levels,
+                          GetJsonValue<std::vector<std::string>>(json, kNamespace));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.properties,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kProperties)));
+  return request;
+}
+
+// UpdateNamespacePropertiesRequest
+nlohmann::json ToJson(const UpdateNamespacePropertiesRequest& request) {
+  // Initialize as an empty object so that when all optional fields are absent we return
+  // {} instead of null
+  nlohmann::json json = nlohmann::json::object();
+  if (!request.removals.empty()) {
+    json[kRemovals] = request.removals;
+  }
+  if (!request.updates.empty()) {
+    json[kUpdates] = request.updates;
+  }
+  return json;
+}
+
+Result<UpdateNamespacePropertiesRequest> UpdateNamespacePropertiesRequestFromJson(
+    const nlohmann::json& json) {
+  UpdateNamespacePropertiesRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.removals, GetJsonValueOrDefault<std::vector<std::string>>(json, kRemovals));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.updates,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kUpdates)));
+  return request;
+}
+
+// CreateTableRequest
+nlohmann::json ToJson(const CreateTableRequest& request) {
+  nlohmann::json json;
+  json[kName] = request.name;
+  if (!request.location.empty()) {
+    json[kLocation] = request.location;
+  }
+  json[kSchema] = ToJson(*request.schema);
+  if (request.partition_spec) {
+    json[kPartitionSpec] = ToJson(*request.partition_spec);
+  }
+  if (request.write_order) {
+    json[kWriteOrder] = ToJson(*request.write_order);
+  }
+  SetOptionalField(json, kStageCreate, request.stage_create);
+  if (!request.properties.empty()) {
+    json[kProperties] = request.properties;
+  }
+  return json;
+}
+
+Result<CreateTableRequest> CreateTableRequestFromJson(const nlohmann::json& json) {
+  CreateTableRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(request.name, GetJsonValue<std::string>(json, kName));
+  ICEBERG_ASSIGN_OR_RAISE(request.location,
+                          GetJsonValueOrDefault<std::string>(json, kLocation, ""));
+  ICEBERG_ASSIGN_OR_RAISE(auto schema_json, GetJsonValue<nlohmann::json>(json, kSchema));
+  ICEBERG_ASSIGN_OR_RAISE(request.schema, SchemaFromJson(schema_json));
+  if (json.contains(kPartitionSpec)) {
+    ICEBERG_ASSIGN_OR_RAISE(auto partition_spec_json,
+                            GetJsonValue<nlohmann::json>(json, kPartitionSpec));
+    ICEBERG_ASSIGN_OR_RAISE(request.partition_spec,
+                            PartitionSpecFromJson(request.schema, partition_spec_json));
+  } else {
+    request.partition_spec = nullptr;
+  }
+  if (json.contains(kWriteOrder)) {
+    ICEBERG_ASSIGN_OR_RAISE(auto write_order_json,
+                            GetJsonValue<nlohmann::json>(json, kWriteOrder));
+    ICEBERG_ASSIGN_OR_RAISE(request.write_order, SortOrderFromJson(write_order_json));
+  } else {
+    request.write_order = nullptr;
+  }
+  ICEBERG_ASSIGN_OR_RAISE(request.stage_create,
+                          GetJsonValueOptional<bool>(json, kStageCreate));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.properties,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kProperties)));
+  return request;
+}
+
+// RegisterTableRequest
+nlohmann::json ToJson(const RegisterTableRequest& request) {
+  nlohmann::json json;
+  json[kName] = request.name;
+  json[kMetadataLocation] = request.metadata_location;
+  if (request.overwrite) {
+    json[kOverwrite] = request.overwrite;
+  }
+  return json;
+}
+
+Result<RegisterTableRequest> RegisterTableRequestFromJson(const nlohmann::json& json) {
+  RegisterTableRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(request.name, GetJsonValue<std::string>(json, kName));
+  ICEBERG_ASSIGN_OR_RAISE(request.metadata_location,
+                          GetJsonValue<std::string>(json, kMetadataLocation));
+  // Default to false if not present
+  ICEBERG_ASSIGN_OR_RAISE(auto overwrite_opt,
+                          GetJsonValueOptional<bool>(json, kOverwrite));
+  request.overwrite = overwrite_opt.value_or(false);
+  return request;
+}
+
+// RenameTableRequest
+nlohmann::json ToJson(const RenameTableRequest& request) {
+  nlohmann::json json;
+  json[kSource] = ToJson(request.source);
+  json[kDestination] = ToJson(request.destination);
+  return json;
+}
+
+Result<RenameTableRequest> RenameTableRequestFromJson(const nlohmann::json& json) {
+  RenameTableRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(auto source_json, GetJsonValue<nlohmann::json>(json, kSource));
+  ICEBERG_ASSIGN_OR_RAISE(request.source, TableIdentifierFromJson(source_json));
+  ICEBERG_ASSIGN_OR_RAISE(auto dest_json,
+                          GetJsonValue<nlohmann::json>(json, kDestination));
+  ICEBERG_ASSIGN_OR_RAISE(request.destination, TableIdentifierFromJson(dest_json));
+  return request;
+}
+
+// LoadTableResult (used by CreateTableResponse, LoadTableResponse)
+nlohmann::json ToJson(const LoadTableResult& result) {
+  nlohmann::json json;
+  if (!result.metadata_location.empty()) {
+    json[kMetadataLocation] = result.metadata_location;
+  }
+  json[kMetadata] = ToJson(*result.metadata);
+  if (!result.config.empty()) {
+    json[kConfig] = result.config;
+  }
+  return json;
+}
+
+Result<LoadTableResult> LoadTableResultFromJson(const nlohmann::json& json) {
+  LoadTableResult result;
+  ICEBERG_ASSIGN_OR_RAISE(result.metadata_location, GetJsonValueOrDefault<std::string>(
+                                                        json, kMetadataLocation, ""));
+  ICEBERG_ASSIGN_OR_RAISE(auto metadata_json,
+                          GetJsonValue<nlohmann::json>(json, kMetadata));
+  ICEBERG_ASSIGN_OR_RAISE(result.metadata, TableMetadataFromJson(metadata_json));
+  ICEBERG_ASSIGN_OR_RAISE(
+      result.config, (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(
+                         json, kConfig)));
+  return result;
+}
+
+// ListNamespacesResponse
+nlohmann::json ToJson(const ListNamespacesResponse& response) {
+  nlohmann::json json;
+  if (!response.next_page_token.empty()) {
+    json[kNextPageToken] = response.next_page_token;
+  }
+  nlohmann::json namespaces = nlohmann::json::array();
+  for (const auto& ns : response.namespaces) {
+    namespaces.push_back(ToJson(ns));
+  }
+  json[kNamespaces] = std::move(namespaces);
+  return json;
+}
+
+Result<ListNamespacesResponse> ListNamespacesResponseFromJson(
+    const nlohmann::json& json) {
+  ListNamespacesResponse response;
+  ICEBERG_ASSIGN_OR_RAISE(response.next_page_token,
+                          GetJsonValueOrDefault<std::string>(json, kNextPageToken, ""));
+  ICEBERG_ASSIGN_OR_RAISE(auto namespaces_json,
+                          GetJsonValue<nlohmann::json>(json, kNamespaces));
+  for (const auto& ns_json : namespaces_json) {
+    ICEBERG_ASSIGN_OR_RAISE(auto ns, NamespaceFromJson(ns_json));
+    response.namespaces.push_back(std::move(ns));
+  }
+  return response;
+}
+
+// CreateNamespaceResponse
+nlohmann::json ToJson(const CreateNamespaceResponse& response) {
+  nlohmann::json json;
+  json[kNamespace] = response.namespace_.levels;
+  if (!response.properties.empty()) {
+    json[kProperties] = response.properties;
+  }
+  return json;
+}
+
+Result<CreateNamespaceResponse> CreateNamespaceResponseFromJson(
+    const nlohmann::json& json) {
+  CreateNamespaceResponse response;
+  ICEBERG_ASSIGN_OR_RAISE(response.namespace_.levels,
+                          GetJsonValue<std::vector<std::string>>(json, kNamespace));
+  ICEBERG_ASSIGN_OR_RAISE(
+      response.properties,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kProperties)));
+  return response;
+}
+
+// GetNamespaceResponse
+nlohmann::json ToJson(const GetNamespaceResponse& response) {
+  nlohmann::json json;
+  json[kNamespace] = response.namespace_.levels;
+  if (!response.properties.empty()) {
+    json[kProperties] = response.properties;
+  }
+  return json;
+}
+
+Result<GetNamespaceResponse> GetNamespaceResponseFromJson(const nlohmann::json& json) {
+  GetNamespaceResponse response;
+  ICEBERG_ASSIGN_OR_RAISE(response.namespace_.levels,
+                          GetJsonValue<std::vector<std::string>>(json, kNamespace));
+  ICEBERG_ASSIGN_OR_RAISE(
+      response.properties,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kProperties)));

Review Comment:
   ```suggestion
         GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
                                                                              kProperties));
   ```



##########
src/iceberg/test/rest_json_internal_test.cc:
##########
@@ -0,0 +1,801 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include <algorithm>
+#include <string>
+#include <unordered_map>
+#include <vector>
+
+#include <gmock/gmock.h>
+#include <gtest/gtest.h>
+#include <nlohmann/json.hpp>
+
+#include "iceberg/catalog/rest/json_internal.h"
+#include "iceberg/catalog/rest/types.h"
+#include "iceberg/table_identifier.h"
+
+namespace iceberg::rest {
+
+namespace {
+
+// Helper to compare two TableIdentifier objects
+bool TableIdentifierEqual(const TableIdentifier& a, const TableIdentifier& b) {
+  return a.ns.levels == b.ns.levels && a.name == b.name;
+}
+
+// Helper to compare string vectors (order-independent)
+bool StringVectorEqual(const std::vector<std::string>& a,
+                       const std::vector<std::string>& b) {
+  return std::ranges::is_permutation(a, b);
+}
+
+}  // namespace
+
+struct CreateNamespaceRequestParam {
+  std::string test_name;
+  std::string expected_json_str;
+  Namespace namespace_;
+  std::unordered_map<std::string, std::string> properties;
+};
+
+class CreateNamespaceRequestTest
+    : public ::testing::TestWithParam<CreateNamespaceRequestParam> {
+ protected:
+  void TestRoundTrip() {
+    const auto& param = GetParam();
+
+    // Build original object
+    CreateNamespaceRequest original;
+    original.namespace_ = param.namespace_;
+    original.properties = param.properties;
+
+    // ToJson and verify JSON string
+    auto json = ToJson(original);
+    auto expected_json = nlohmann::json::parse(param.expected_json_str);
+    EXPECT_EQ(json, expected_json) << "ToJson mismatch";
+
+    // FromJson and verify object equality
+    auto result = CreateNamespaceRequestFromJson(expected_json);
+    ASSERT_TRUE(result.has_value()) << result.error().message;
+    auto& parsed = result.value();
+
+    EXPECT_EQ(parsed.namespace_.levels, original.namespace_.levels);
+    EXPECT_EQ(parsed.properties, original.properties);
+  }
+};
+
+TEST_P(CreateNamespaceRequestTest, RoundTrip) { TestRoundTrip(); }
+
+INSTANTIATE_TEST_SUITE_P(
+    CreateNamespaceRequestCases, CreateNamespaceRequestTest,
+    ::testing::Values(
+        // Full request with properties
+        CreateNamespaceRequestParam{
+            .test_name = "FullRequest",
+            .expected_json_str =
+                R"({"namespace":["accounting","tax"],"properties":{"owner":"Hank"}})",
+            .namespace_ = Namespace{{"accounting", "tax"}},
+            .properties = {{"owner", "Hank"}},
+        },
+        // Request with empty properties (omit properties field when empty)
+        CreateNamespaceRequestParam{
+            .test_name = "EmptyProperties",
+            .expected_json_str = R"({"namespace":["accounting","tax"]})",
+            .namespace_ = Namespace{{"accounting", "tax"}},
+            .properties = {},
+        },
+        // Request with empty namespace
+        CreateNamespaceRequestParam{
+            .test_name = "EmptyNamespace",
+            .expected_json_str = R"({"namespace":[]})",
+            .namespace_ = Namespace{},
+            .properties = {},
+        }),
+    [](const ::testing::TestParamInfo<CreateNamespaceRequestParam>& info) {
+      return info.param.test_name;
+    });
+
+TEST(CreateNamespaceRequestTest, DeserializeWithoutDefaults) {
+  // Properties is null
+  std::string json_null_props = R"({"namespace":["accounting","tax"],"properties":null})";
+  auto result1 = CreateNamespaceRequestFromJson(nlohmann::json::parse(json_null_props));
+  ASSERT_TRUE(result1.has_value());
+  EXPECT_EQ(result1.value().namespace_.levels,
+            std::vector<std::string>({"accounting", "tax"}));
+  EXPECT_TRUE(result1.value().properties.empty());
+
+  // Properties is missing
+  std::string json_missing_props = R"({"namespace":["accounting","tax"]})";
+  auto result2 =
+      CreateNamespaceRequestFromJson(nlohmann::json::parse(json_missing_props));
+  ASSERT_TRUE(result2.has_value());
+  EXPECT_EQ(result2.value().namespace_.levels,
+            std::vector<std::string>({"accounting", "tax"}));
+  EXPECT_TRUE(result2.value().properties.empty());
+}
+
+TEST(CreateNamespaceRequestTest, InvalidRequests) {
+  // Incorrect type for namespace
+  std::string json_wrong_ns_type =
+      R"({"namespace":"accounting%1Ftax","properties":null})";
+  auto result1 =
+      CreateNamespaceRequestFromJson(nlohmann::json::parse(json_wrong_ns_type));
+  EXPECT_FALSE(result1.has_value());

Review Comment:
   Can you check the expected error message as well?
   
   ```
     EXPECT_THAT(result, IsError(ErrorKind::kJsonParseError));
     EXPECT_THAT(result, HasErrorMessage("..."));
   ```



##########
src/iceberg/catalog/rest/json_internal.cc:
##########
@@ -0,0 +1,357 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/catalog/rest/json_internal.h"
+
+#include <ranges>
+#include <string>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+#include <nlohmann/json.hpp>
+
+#include "iceberg/catalog/rest/types.h"
+#include "iceberg/json_internal.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/schema.h"
+#include "iceberg/sort_order.h"
+#include "iceberg/table_metadata.h"
+#include "iceberg/util/json_util_internal.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg::rest {
+
+namespace {
+
+// REST API JSON field constants
+constexpr std::string_view kNamespace = "namespace";
+constexpr std::string_view kNamespaces = "namespaces";
+constexpr std::string_view kProperties = "properties";
+constexpr std::string_view kRemovals = "removals";
+constexpr std::string_view kUpdates = "updates";
+constexpr std::string_view kUpdated = "updated";
+constexpr std::string_view kRemoved = "removed";
+constexpr std::string_view kMissing = "missing";
+constexpr std::string_view kNextPageToken = "next-page-token";
+constexpr std::string_view kName = "name";
+constexpr std::string_view kLocation = "location";
+constexpr std::string_view kSchema = "schema";
+constexpr std::string_view kPartitionSpec = "partition-spec";
+constexpr std::string_view kWriteOrder = "write-order";
+constexpr std::string_view kStageCreate = "stage-create";
+constexpr std::string_view kMetadataLocation = "metadata-location";
+constexpr std::string_view kOverwrite = "overwrite";
+constexpr std::string_view kSource = "source";
+constexpr std::string_view kDestination = "destination";
+constexpr std::string_view kMetadata = "metadata";
+constexpr std::string_view kConfig = "config";
+constexpr std::string_view kIdentifiers = "identifiers";
+
+}  // namespace
+
+// CreateNamespaceRequest
+nlohmann::json ToJson(const CreateNamespaceRequest& request) {
+  nlohmann::json json;
+  json[kNamespace] = request.namespace_.levels;
+  if (!request.properties.empty()) {
+    json[kProperties] = request.properties;
+  }
+  return json;
+}
+
+Result<CreateNamespaceRequest> CreateNamespaceRequestFromJson(
+    const nlohmann::json& json) {
+  CreateNamespaceRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(request.namespace_.levels,
+                          GetJsonValue<std::vector<std::string>>(json, kNamespace));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.properties,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kProperties)));
+  return request;
+}
+
+// UpdateNamespacePropertiesRequest
+nlohmann::json ToJson(const UpdateNamespacePropertiesRequest& request) {
+  // Initialize as an empty object so that when all optional fields are absent we return
+  // {} instead of null
+  nlohmann::json json = nlohmann::json::object();
+  if (!request.removals.empty()) {
+    json[kRemovals] = request.removals;
+  }
+  if (!request.updates.empty()) {
+    json[kUpdates] = request.updates;
+  }
+  return json;
+}
+
+Result<UpdateNamespacePropertiesRequest> UpdateNamespacePropertiesRequestFromJson(
+    const nlohmann::json& json) {
+  UpdateNamespacePropertiesRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.removals, GetJsonValueOrDefault<std::vector<std::string>>(json, kRemovals));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.updates,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kUpdates)));
+  return request;
+}
+
+// CreateTableRequest
+nlohmann::json ToJson(const CreateTableRequest& request) {
+  nlohmann::json json;
+  json[kName] = request.name;
+  if (!request.location.empty()) {
+    json[kLocation] = request.location;
+  }
+  json[kSchema] = ToJson(*request.schema);
+  if (request.partition_spec) {
+    json[kPartitionSpec] = ToJson(*request.partition_spec);
+  }
+  if (request.write_order) {
+    json[kWriteOrder] = ToJson(*request.write_order);
+  }
+  SetOptionalField(json, kStageCreate, request.stage_create);
+  if (!request.properties.empty()) {
+    json[kProperties] = request.properties;
+  }
+  return json;
+}
+
+Result<CreateTableRequest> CreateTableRequestFromJson(const nlohmann::json& json) {
+  CreateTableRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(request.name, GetJsonValue<std::string>(json, kName));
+  ICEBERG_ASSIGN_OR_RAISE(request.location,
+                          GetJsonValueOrDefault<std::string>(json, kLocation, ""));
+  ICEBERG_ASSIGN_OR_RAISE(auto schema_json, GetJsonValue<nlohmann::json>(json, kSchema));
+  ICEBERG_ASSIGN_OR_RAISE(request.schema, SchemaFromJson(schema_json));
+  if (json.contains(kPartitionSpec)) {
+    ICEBERG_ASSIGN_OR_RAISE(auto partition_spec_json,
+                            GetJsonValue<nlohmann::json>(json, kPartitionSpec));
+    ICEBERG_ASSIGN_OR_RAISE(request.partition_spec,
+                            PartitionSpecFromJson(request.schema, partition_spec_json));
+  } else {
+    request.partition_spec = nullptr;
+  }
+  if (json.contains(kWriteOrder)) {
+    ICEBERG_ASSIGN_OR_RAISE(auto write_order_json,
+                            GetJsonValue<nlohmann::json>(json, kWriteOrder));
+    ICEBERG_ASSIGN_OR_RAISE(request.write_order, SortOrderFromJson(write_order_json));
+  } else {
+    request.write_order = nullptr;
+  }
+  ICEBERG_ASSIGN_OR_RAISE(request.stage_create,
+                          GetJsonValueOptional<bool>(json, kStageCreate));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.properties,
+      (GetJsonValueOrDefault<std::unordered_map<std::string, std::string>>(json,
+                                                                           kProperties)));
+  return request;
+}
+
+// RegisterTableRequest
+nlohmann::json ToJson(const RegisterTableRequest& request) {
+  nlohmann::json json;
+  json[kName] = request.name;
+  json[kMetadataLocation] = request.metadata_location;
+  if (request.overwrite) {
+    json[kOverwrite] = request.overwrite;
+  }
+  return json;
+}
+
+Result<RegisterTableRequest> RegisterTableRequestFromJson(const nlohmann::json& json) {
+  RegisterTableRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(request.name, GetJsonValue<std::string>(json, kName));
+  ICEBERG_ASSIGN_OR_RAISE(request.metadata_location,
+                          GetJsonValue<std::string>(json, kMetadataLocation));
+  // Default to false if not present
+  ICEBERG_ASSIGN_OR_RAISE(auto overwrite_opt,
+                          GetJsonValueOptional<bool>(json, kOverwrite));
+  request.overwrite = overwrite_opt.value_or(false);
+  return request;
+}
+
+// RenameTableRequest
+nlohmann::json ToJson(const RenameTableRequest& request) {
+  nlohmann::json json;
+  json[kSource] = ToJson(request.source);
+  json[kDestination] = ToJson(request.destination);
+  return json;
+}
+
+Result<RenameTableRequest> RenameTableRequestFromJson(const nlohmann::json& json) {
+  RenameTableRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(auto source_json, GetJsonValue<nlohmann::json>(json, kSource));
+  ICEBERG_ASSIGN_OR_RAISE(request.source, TableIdentifierFromJson(source_json));
+  ICEBERG_ASSIGN_OR_RAISE(auto dest_json,
+                          GetJsonValue<nlohmann::json>(json, kDestination));
+  ICEBERG_ASSIGN_OR_RAISE(request.destination, TableIdentifierFromJson(dest_json));
+  return request;
+}
+
+// LoadTableResult (used by CreateTableResponse, LoadTableResponse)
+nlohmann::json ToJson(const LoadTableResult& result) {
+  nlohmann::json json;
+  if (!result.metadata_location.empty()) {
+    json[kMetadataLocation] = result.metadata_location;
+  }
+  json[kMetadata] = ToJson(*result.metadata);
+  if (!result.config.empty()) {
+    json[kConfig] = result.config;
+  }
+  return json;
+}
+
+Result<LoadTableResult> LoadTableResultFromJson(const nlohmann::json& json) {

Review Comment:
   Please also carefully check if any missing field that is in the spec but not added on our side.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org