fix: align rest table update serde with java/spec

Author: wgtmac

src/iceberg/catalog/rest/json_serde.cc

@@ -903,13 +903,21 @@ Result<CommitTableRequest> CommitTableRequestFromJson(const nlohmann::json& json
 
   ICEBERG_ASSIGN_OR_RAISE(auto requirements_json,
                           GetJsonValue<nlohmann::json>(json, kRequirements));
+  if (!requirements_json.is_array()) {
+    return JsonParseError("Expected '{}' to be an array, got {}", kRequirements,
+                          SafeDumpJson(requirements_json));
+  }
   for (const auto& req_json : requirements_json) {
     ICEBERG_ASSIGN_OR_RAISE(auto requirement, TableRequirementFromJson(req_json));
     request.requirements.push_back(std::move(requirement));
   }
 
   ICEBERG_ASSIGN_OR_RAISE(auto updates_json,
                           GetJsonValue<nlohmann::json>(json, kUpdates));
+  if (!updates_json.is_array()) {
+    return JsonParseError("Expected '{}' to be an array, got {}", kUpdates,
+                          SafeDumpJson(updates_json));
+  }
   for (const auto& update_json : updates_json) {
     ICEBERG_ASSIGN_OR_RAISE(auto update, TableUpdateFromJson(update_json));
     request.updates.push_back(std::move(update));

src/iceberg/encryption/encrypted_key.h

@@ -0,0 +1,43 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+/// \file iceberg/encryption/encrypted_key.h
+/// Encrypted table key metadata.
+
+#include <optional>
+#include <string>
+#include <unordered_map>
+
+#include "iceberg/iceberg_export.h"
+
+namespace iceberg {
+
+/// \brief Represents an encrypted table key entry.
+struct ICEBERG_EXPORT EncryptedKey {
+  std::string key_id;
+  std::string encrypted_key_metadata;
+  std::optional<std::string> encrypted_by_id = std::nullopt;
+  std::unordered_map<std::string, std::string> properties;
+
+  friend bool operator==(const EncryptedKey&, const EncryptedKey&) = default;
+};
+
+}  // namespace iceberg

src/iceberg/encryption/meson.build

@@ -0,0 +1,18 @@
+# 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.
+
+install_headers(['encrypted_key.h'], subdir: 'iceberg/encryption')

src/iceberg/json_serde.cc

@@ -43,6 +43,7 @@
 #include "iceberg/table_update.h"
 #include "iceberg/transform.h"
 #include "iceberg/type.h"
+#include "iceberg/util/base64.h"
 #include "iceberg/util/checked_cast.h"
 #include "iceberg/util/formatter.h"  // IWYU pragma: keep
 #include "iceberg/util/json_util_internal.h"
@@ -90,6 +91,8 @@ constexpr std::string_view kValueId = "value-id";
 constexpr std::string_view kRequired = "required";
 constexpr std::string_view kElementRequired = "element-required";
 constexpr std::string_view kValueRequired = "value-required";
+constexpr std::string_view kEncryptedKeyMetadata = "encrypted-key-metadata";
+constexpr std::string_view kEncryptedById = "encrypted-by-id";
 
 // Snapshot constants
 constexpr std::string_view kSpecId = "spec-id";
@@ -166,6 +169,7 @@ constexpr std::string_view kRefs = "refs";
 constexpr std::string_view kStatistics = "statistics";
 constexpr std::string_view kPartitionStatistics = "partition-statistics";
 constexpr std::string_view kNextRowId = "next-row-id";
+constexpr std::string_view kEncryptionKeys = "encryption-keys";
 constexpr std::string_view kMetadataFile = "metadata-file";
 constexpr std::string_view kStatisticsPath = "statistics-path";
 constexpr std::string_view kFileSizeInBytes = "file-size-in-bytes";
@@ -196,6 +200,8 @@ constexpr std::string_view kActionRemoveStatistics = "remove-statistics";
 constexpr std::string_view kActionSetPartitionStatistics = "set-partition-statistics";
 constexpr std::string_view kActionRemovePartitionStatistics =
     "remove-partition-statistics";
+constexpr std::string_view kActionAddEncryptionKey = "add-encryption-key";
+constexpr std::string_view kActionRemoveEncryptionKey = "remove-encryption-key";
 
 // TableUpdate field constants
 constexpr std::string_view kUUID = "uuid";
@@ -210,6 +216,9 @@ constexpr std::string_view kRefName = "ref-name";
 constexpr std::string_view kRef = "ref";
 constexpr std::string_view kUpdates = "updates";
 constexpr std::string_view kRemovals = "removals";
+constexpr std::string_view kUpdated = "updated";
+constexpr std::string_view kRemoved = "removed";
+constexpr std::string_view kEncryptionKey = "encryption-key";
 
 // TableRequirement type constants
 constexpr std::string_view kRequirementAssertDoesNotExist = "assert-create";
@@ -851,6 +860,41 @@ Result<MetadataLogEntry> MetadataLogEntryFromJson(const nlohmann::json& json) {
   return metadata_log_entry;
 }
 
+nlohmann::json ToJson(const EncryptedKey& encrypted_key) {
+  nlohmann::json json;
+  json[kKeyId] = encrypted_key.key_id;
+  json[kEncryptedKeyMetadata] = Base64::Encode(encrypted_key.encrypted_key_metadata);
+  SetOptionalField(json, kEncryptedById, encrypted_key.encrypted_by_id);
+  if (!encrypted_key.properties.empty()) {
+    json[kProperties] = encrypted_key.properties;
+  }
+  return json;
+}
+
+Result<EncryptedKey> EncryptedKeyFromJson(const nlohmann::json& json) {
+  using StringMap = std::unordered_map<std::string, std::string>;
+
+  if (!json.is_object()) {
+    return JsonParseError("Invalid encryption key, must be non-null object: {}",
+                          SafeDumpJson(json));
+  }
+
+  ICEBERG_ASSIGN_OR_RAISE(auto key_id, GetJsonValue<std::string>(json, kKeyId));
+  ICEBERG_ASSIGN_OR_RAISE(auto encoded_metadata,
+                          GetJsonValue<std::string>(json, kEncryptedKeyMetadata));
+  ICEBERG_ASSIGN_OR_RAISE(auto encrypted_key_metadata, Base64::Decode(encoded_metadata));
+  ICEBERG_ASSIGN_OR_RAISE(auto encrypted_by_id,
+                          GetJsonValueOptional<std::string>(json, kEncryptedById));
+  ICEBERG_ASSIGN_OR_RAISE(auto properties,
+                          GetJsonValueOrDefault<StringMap>(json, kProperties));
+  return EncryptedKey{
+      .key_id = std::move(key_id),
+      .encrypted_key_metadata = std::move(encrypted_key_metadata),
+      .encrypted_by_id = std::move(encrypted_by_id),
+      .properties = std::move(properties),
+  };
+}
+
 nlohmann::json ToJson(const TableMetadata& table_metadata) {
   nlohmann::json json;
 
@@ -917,6 +961,9 @@ nlohmann::json ToJson(const TableMetadata& table_metadata) {
   json[kSnapshots] = ToJsonList(table_metadata.snapshots);
   json[kStatistics] = ToJsonList(table_metadata.statistics);
   json[kPartitionStatistics] = ToJsonList(table_metadata.partition_statistics);
+  if (!table_metadata.encryption_keys.empty()) {
+    json[kEncryptionKeys] = ToJsonList(table_metadata.encryption_keys);
+  }
   json[kSnapshotLog] = ToJsonList(table_metadata.snapshot_log);
   json[kMetadataLog] = ToJsonList(table_metadata.metadata_log);
 
@@ -1182,6 +1229,9 @@ Result<std::unique_ptr<TableMetadata>> TableMetadataFromJson(const nlohmann::jso
       table_metadata->partition_statistics,
       FromJsonList<PartitionStatisticsFile>(json, kPartitionStatistics,
                                             PartitionStatisticsFileFromJson));
+  ICEBERG_ASSIGN_OR_RAISE(
+      table_metadata->encryption_keys,
+      FromJsonList<EncryptedKey>(json, kEncryptionKeys, EncryptedKeyFromJson));
   ICEBERG_ASSIGN_OR_RAISE(
       table_metadata->snapshot_log,
       FromJsonList<SnapshotLogEntry>(json, kSnapshotLog, SnapshotLogEntryFromJson));
@@ -1486,6 +1536,18 @@ nlohmann::json ToJson(const TableUpdate& update) {
       json[kSnapshotId] = u.snapshot_id();
       break;
     }
+    case TableUpdate::Kind::kAddEncryptionKey: {
+      const auto& u = internal::checked_cast<const table::AddEncryptionKey&>(update);
+      json[kAction] = kActionAddEncryptionKey;
+      json[kEncryptionKey] = ToJson(u.key());
+      break;
+    }
+    case TableUpdate::Kind::kRemoveEncryptionKey: {
+      const auto& u = internal::checked_cast<const table::RemoveEncryptionKey&>(update);
+      json[kAction] = kActionRemoveEncryptionKey;
+      json[kKeyId] = u.key_id();
+      break;
+    }
   }
   return json;
 }
@@ -1506,7 +1568,6 @@ nlohmann::json ToJson(const TableRequirement& requirement) {
       const auto& r =
           internal::checked_cast<const table::AssertRefSnapshotID&>(requirement);
       json[kType] = kRequirementAssertRefSnapshotID;
-      // REST spec names this field "ref", not "ref-name".
       json[kRef] = r.ref_name();
       if (r.snapshot_id().has_value()) {
         json[kSnapshotId] = r.snapshot_id().value();
@@ -1570,8 +1631,10 @@ Result<std::unique_ptr<TableUpdate>> TableUpdateFromJson(const nlohmann::json& j
     ICEBERG_ASSIGN_OR_RAISE(auto schema_json,
                             GetJsonValue<nlohmann::json>(json, kSchema));
     ICEBERG_ASSIGN_OR_RAISE(auto parsed_schema, SchemaFromJson(schema_json));
-    ICEBERG_ASSIGN_OR_RAISE(auto last_column_id,
-                            GetJsonValue<int32_t>(json, kLastColumnId));
+    ICEBERG_ASSIGN_OR_RAISE(auto highest_field_id, parsed_schema->HighestFieldId());
+    ICEBERG_ASSIGN_OR_RAISE(
+        auto last_column_id,
+        GetJsonValueOrDefault<int32_t>(json, kLastColumnId, highest_field_id));
     return std::make_unique<table::AddSchema>(std::move(parsed_schema), last_column_id);
   }
   if (action == kActionSetCurrentSchema) {
@@ -1650,12 +1713,17 @@ Result<std::unique_ptr<TableUpdate>> TableUpdateFromJson(const nlohmann::json& j
   }
   if (action == kActionSetProperties) {
     using StringMap = std::unordered_map<std::string, std::string>;
-    ICEBERG_ASSIGN_OR_RAISE(auto updates, GetJsonValue<StringMap>(json, kUpdates));
+    ICEBERG_ASSIGN_OR_RAISE(auto updates,
+                            json.contains(kUpdates) || !json.contains(kUpdated)
+                                ? GetJsonValue<StringMap>(json, kUpdates)
+                                : GetJsonValue<StringMap>(json, kUpdated));
     return std::make_unique<table::SetProperties>(std::move(updates));
   }
   if (action == kActionRemoveProperties) {
     ICEBERG_ASSIGN_OR_RAISE(auto removals_vec,
-                            GetJsonValue<std::vector<std::string>>(json, kRemovals));
+                            json.contains(kRemovals) || !json.contains(kRemoved)
+                                ? GetJsonValue<std::vector<std::string>>(json, kRemovals)
+                                : GetJsonValue<std::vector<std::string>>(json, kRemoved));
     std::unordered_set<std::string> removals(
         std::make_move_iterator(removals_vec.begin()),
         std::make_move_iterator(removals_vec.end()));
@@ -1688,6 +1756,17 @@ Result<std::unique_ptr<TableUpdate>> TableUpdateFromJson(const nlohmann::json& j
     ICEBERG_ASSIGN_OR_RAISE(auto snapshot_id, GetJsonValue<int64_t>(json, kSnapshotId));
     return std::make_unique<table::RemovePartitionStatistics>(snapshot_id);
   }
+  if (action == kActionAddEncryptionKey) {
+    if (!json.contains(kEncryptionKey)) {
+      return JsonParseError("Invalid encryption key, must be non-null object: null");
+    }
+    ICEBERG_ASSIGN_OR_RAISE(auto key, EncryptedKeyFromJson(json.at(kEncryptionKey)));
+    return std::make_unique<table::AddEncryptionKey>(std::move(key));
+  }
+  if (action == kActionRemoveEncryptionKey) {
+    ICEBERG_ASSIGN_OR_RAISE(auto key_id, GetJsonValue<std::string>(json, kKeyId));
+    return std::make_unique<table::RemoveEncryptionKey>(std::move(key_id));
+  }
 
   return JsonParseError("Unknown table update action: {}", action);
 }

src/iceberg/json_serde_internal.h

@@ -28,6 +28,7 @@
 
 #include <nlohmann/json_fwd.hpp>
 
+#include "iceberg/encryption/encrypted_key.h"
 #include "iceberg/result.h"
 #include "iceberg/statistics_file.h"
 #include "iceberg/table_metadata.h"
@@ -283,6 +284,18 @@ ICEBERG_EXPORT nlohmann::json ToJson(const MetadataLogEntry& metadata_log_entry)
 ICEBERG_EXPORT Result<MetadataLogEntry> MetadataLogEntryFromJson(
     const nlohmann::json& json);
 
+/// \brief Serializes an `EncryptedKey` object to JSON.
+///
+/// \param encrypted_key The `EncryptedKey` object to be serialized.
+/// \return A JSON object representing the `EncryptedKey`.
+ICEBERG_EXPORT nlohmann::json ToJson(const EncryptedKey& encrypted_key);
+
+/// \brief Deserializes a JSON object into an `EncryptedKey` object.
+///
+/// \param json The JSON object representing an `EncryptedKey`.
+/// \return An `EncryptedKey` object or an error if the conversion fails.
+ICEBERG_EXPORT Result<EncryptedKey> EncryptedKeyFromJson(const nlohmann::json& json);
+
 /// \brief Serializes a `TableMetadata` object to JSON.
 ///
 /// \param table_metadata The `TableMetadata` object to be serialized.

src/iceberg/meson.build

@@ -287,6 +287,7 @@ install_headers(
 subdir('catalog')
 subdir('data')
 subdir('deletes')
+subdir('encryption')
 subdir('expression')
 subdir('manifest')
 subdir('metrics')