feat: add snapshot encryption key-id

src/iceberg/json_serde.cc

@@ -476,6 +476,7 @@ nlohmann::json ToJson(const Snapshot& snapshot) {
   if (snapshot.first_row_id.has_value()) {
     SetOptionalField(json, kAddedRows, snapshot.added_rows);
   }
+  SetOptionalField(json, kKeyId, snapshot.key_id);
   return json;
 }
 
@@ -893,11 +894,13 @@ Result<std::unique_ptr<Snapshot>> SnapshotFromJson(const nlohmann::json& json) {
   }
 
   ICEBERG_ASSIGN_OR_RAISE(auto schema_id, GetJsonValueOptional<int32_t>(json, kSchemaId));
+  ICEBERG_ASSIGN_OR_RAISE(auto key_id, GetJsonValueOptional<std::string>(json, kKeyId));
 
   return std::make_unique<Snapshot>(
       snapshot_id, parent_snapshot_id,
       sequence_number.value_or(TableMetadata::kInitialSequenceNumber), timestamp_ms,
-      manifest_list, std::move(summary), schema_id, first_row_id, added_rows);
+      manifest_list, std::move(summary), schema_id, first_row_id, added_rows,
+      std::move(key_id));
 }
 
 nlohmann::json ToJson(const BlobMetadata& blob_metadata) {

src/iceberg/snapshot.cc

@@ -173,7 +173,7 @@ bool Snapshot::Equals(const Snapshot& other) const {
          parent_snapshot_id == other.parent_snapshot_id &&
          sequence_number == other.sequence_number && timestamp_ms == other.timestamp_ms &&
          schema_id == other.schema_id && first_row_id == other.first_row_id &&
-         added_rows == other.added_rows;
+         added_rows == other.added_rows && key_id == other.key_id;
 }
 
 Result<std::unique_ptr<Snapshot>> Snapshot::Make(

src/iceberg/snapshot.h

@@ -412,6 +412,9 @@ struct ICEBERG_EXPORT Snapshot {
   std::optional<int64_t> first_row_id;
   /// The upper bound of rows with assigned row IDs in this snapshot.
   std::optional<int64_t> added_rows;
+  /// ID of the encryption key (in table metadata `encryption-keys`) that holds the
+  /// encrypted key metadata of the manifest list. Set only for encrypted tables.
+  std::optional<std::string> key_id;
 
   /// \brief Create a new Snapshot instance with validation on the inputs.
   static Result<std::unique_ptr<Snapshot>> Make(

src/iceberg/test/json_serde_test.cc

@@ -406,6 +406,28 @@ TEST(JsonInternalTest, SnapshotFromJsonReadsTopLevelRowLineageFields) {
   EXPECT_EQ(json["summary"]["added-rows"], "26");
 }
 
+TEST(JsonInternalTest, SnapshotEncryptionKeyId) {
+  nlohmann::json snapshot_json =
+      R"({"snapshot-id":1234567890,
+          "sequence-number":1,
+          "timestamp-ms":1234567890123,
+          "manifest-list":"/path/to/manifest_list",
+          "summary":{"operation":"append"},
+          "key-id":"C9PJpnsZpsbZarERjyOitg=="})"_json;
+
+  ICEBERG_UNWRAP_OR_FAIL(auto snapshot, SnapshotFromJson(snapshot_json));
+  ASSERT_TRUE(snapshot->key_id.has_value());
+  EXPECT_EQ(snapshot->key_id.value(), "C9PJpnsZpsbZarERjyOitg==");
+  EXPECT_EQ(ToJson(*snapshot)["key-id"], "C9PJpnsZpsbZarERjyOitg==");
+
+  // Snapshots of unencrypted tables have no key-id
+  snapshot_json.erase("key-id");
+  ICEBERG_UNWRAP_OR_FAIL(auto plain, SnapshotFromJson(snapshot_json));
+  EXPECT_FALSE(plain->key_id.has_value());
+  EXPECT_FALSE(ToJson(*plain).contains("key-id"));
+  EXPECT_NE(*plain, *snapshot);
+}
+
 // FIXME: disable it for now since Iceberg Spark plugin generates
 // custom summary keys.
 TEST(JsonInternalTest, DISABLED_SnapshotFromJsonWithInvalidSummary) {