fix(schema): preserve doc-only schema evolution

src/iceberg/schema_field.cc

@@ -243,7 +243,7 @@ bool DefaultEquals(const std::shared_ptr<const Literal>& lhs,
 
 bool SchemaField::Equals(const SchemaField& other) const {
   return field_id_ == other.field_id_ && name_ == other.name_ && *type_ == *other.type_ &&
-         optional_ == other.optional_ &&
+         optional_ == other.optional_ && doc_ == other.doc_ &&
          DefaultEquals(initial_default_, other.initial_default_) &&
          DefaultEquals(write_default_, other.write_default_);
 }

src/iceberg/test/metadata_serde_test.cc

@@ -142,7 +142,7 @@ TEST(MetadataSerdeTest, DeserializeV1Valid) {
 
   auto expected_schema = std::make_shared<Schema>(
       std::vector<SchemaField>{SchemaField::MakeRequired(1, "x", int64()),
-                               SchemaField::MakeRequired(2, "y", int64()),
+                               SchemaField::MakeRequired(2, "y", int64(), "comment"),
                                SchemaField::MakeRequired(3, "z", int64())});
 
   auto expected_spec_result = PartitionSpec::Make(
@@ -188,11 +188,12 @@ TEST(MetadataSerdeTest, DeserializeV2Valid) {
 
   ICEBERG_UNWRAP_OR_FAIL(
       std::shared_ptr<Schema> expected_schema_2,
-      Schema::Make(std::vector<SchemaField>{SchemaField::MakeRequired(1, "x", int64()),
-                                            SchemaField::MakeRequired(2, "y", int64()),
-                                            SchemaField::MakeRequired(3, "z", int64())},
-                   /*schema_id=*/1,
-                   /*identifier_field_ids=*/std::vector<int32_t>{1, 2}));
+      Schema::Make(
+          std::vector<SchemaField>{SchemaField::MakeRequired(1, "x", int64()),
+                                   SchemaField::MakeRequired(2, "y", int64(), "comment"),
+                                   SchemaField::MakeRequired(3, "z", int64())},
+          /*schema_id=*/1,
+          /*identifier_field_ids=*/std::vector<int32_t>{1, 2}));
 
   auto expected_spec_result = PartitionSpec::Make(
       /*spec_id=*/0,

src/iceberg/test/schema_field_test.cc

@@ -86,6 +86,21 @@ TEST(SchemaFieldTest, Equality) {
   ASSERT_EQ(field6, field1);
 }
 
+TEST(SchemaFieldTest, EqualityWithDoc) {
+  for (bool optional : {false, true}) {
+    SchemaField original(1, "foo", int32(), optional, "old");
+    SchemaField same(1, "foo", int32(), optional, "old");
+    SchemaField updated(1, "foo", int32(), optional, "new");
+    SchemaField without_doc(1, "foo", int32(), optional);
+
+    EXPECT_EQ(original, same);
+    EXPECT_NE(original, updated);
+    EXPECT_NE(updated, original);
+    EXPECT_NE(original, without_doc);
+    EXPECT_NE(without_doc, original);
+  }
+}
+
 TEST(SchemaFieldTest, WithDoc) {
   {
     SchemaField field(/*field_id=*/1, /*name=*/"foo", int32(),

src/iceberg/test/schema_test.cc

@@ -98,6 +98,52 @@ TEST(SchemaTest, Equality) {
   ASSERT_EQ(schema5, schema1);
 }
 
+TEST(SchemaTest, SameSchemaWithDoc) {
+  using iceberg::int64;
+  using iceberg::ListType;
+  using iceberg::MapType;
+  using iceberg::Schema;
+  using iceberg::SchemaField;
+  using iceberg::string;
+  using iceberg::StructType;
+  using iceberg::Type;
+  auto make_schemas = [](std::string_view doc, int32_t schema_id) {
+    auto nested = SchemaField::MakeRequired(2, "nested", int64(), doc);
+    auto element = SchemaField::MakeOptional(2, "element", int64(), doc);
+    auto key = SchemaField::MakeRequired(2, "key", string());
+    auto value = SchemaField::MakeOptional(3, "value", int64());
+    auto key_with_doc = SchemaField::MakeRequired(2, "key", string(), doc);
+    auto value_with_doc = SchemaField::MakeOptional(3, "value", int64(), doc);
+    std::vector<std::shared_ptr<Type>> types{
+        std::make_shared<StructType>(std::vector<SchemaField>{nested}),
+        std::make_shared<ListType>(element),
+        std::make_shared<MapType>(key_with_doc, value),
+        std::make_shared<MapType>(key, value_with_doc)};
+    std::vector<std::unique_ptr<Schema>> schemas;
+    schemas.push_back(std::make_unique<Schema>(
+        std::vector<SchemaField>{SchemaField::MakeRequired(1, "c", int64(), doc)},
+        schema_id));
+    for (auto& type : types) {
+      schemas.push_back(std::make_unique<Schema>(
+          std::vector<SchemaField>{SchemaField::MakeRequired(1, "c", type)}, schema_id));
+    }
+    return schemas;
+  };
+
+  const auto original = make_schemas("old", 0);
+  const auto same = make_schemas("old", 1);
+  const auto updated = make_schemas("new", 0);
+  const auto without_doc = make_schemas("", 0);
+  for (size_t i = 0; i < original.size(); ++i) {
+    SCOPED_TRACE(i);
+    EXPECT_TRUE(original[i]->SameSchema(*same[i]));
+    EXPECT_FALSE(original[i]->SameSchema(*updated[i]));
+    EXPECT_FALSE(updated[i]->SameSchema(*original[i]));
+    EXPECT_FALSE(original[i]->SameSchema(*without_doc[i]));
+    EXPECT_NE(*original[i], *updated[i]);
+  }
+}
+
 TEST(SchemaTest, ValidateRejectsV3TypesBeforeFormatV3) {
   iceberg::Schema timestamp_ns_schema(
       {iceberg::SchemaField(1, "timestamp_ns", iceberg::timestamp_ns(), false)});

src/iceberg/test/table_metadata_builder_test.cc

@@ -20,11 +20,14 @@
 #include <memory>
 #include <ranges>
 #include <string>
+#include <utility>
 #include <vector>
 
 #include <gmock/gmock.h>
 #include <gtest/gtest.h>
+#include <nlohmann/json.hpp>
 
+#include "iceberg/json_serde_internal.h"
 #include "iceberg/partition_spec.h"
 #include "iceberg/result.h"
 #include "iceberg/schema.h"
@@ -792,6 +795,59 @@ TEST(TableMetadataBuilderTest, SetCurrentSchemaBasic) {
   EXPECT_EQ(metadata->current_schema_id, 0);
 }
 
+TEST(TableMetadataBuilderTest, SetCurrentSchemaDocOnlyChange) {
+  auto make_schema = [](const std::string& doc) {
+    return std::make_shared<Schema>(
+        std::vector<SchemaField>{SchemaField::MakeRequired(1, "c", int64(), doc)}, 0);
+  };
+  const std::vector<std::pair<std::string, std::string>> doc_changes{
+      {"old", "new"}, {"", "new"}, {"old", ""}};
+  for (const auto& [old_doc, new_doc] : doc_changes) {
+    SCOPED_TRACE(old_doc + " -> " + new_doc);
+    auto base = CreateBaseMetadata();
+    base->schemas = {make_schema(old_doc)};
+    base->last_column_id = 1;
+    auto new_schema = make_schema(new_doc);
+    auto builder = TableMetadataBuilder::BuildFrom(base.get());
+    builder->SetCurrentSchema(new_schema, base->last_column_id);
+
+    ICEBERG_UNWRAP_OR_FAIL(auto metadata, builder->Build());
+    ICEBERG_UNWRAP_OR_FAIL(auto current_schema, metadata->Schema());
+    EXPECT_EQ(current_schema->fields()[0].doc(), new_doc);
+    EXPECT_NE(metadata->current_schema_id, base->current_schema_id);
+    EXPECT_EQ(metadata->last_column_id, base->last_column_id);
+    ASSERT_EQ(metadata->schemas.size(), 2);
+    EXPECT_EQ(metadata->schemas[0]->fields()[0].doc(), old_doc);
+    EXPECT_EQ(metadata->schemas[1]->schema_id(), metadata->current_schema_id);
+    EXPECT_EQ(metadata->schemas[1]->fields()[0].doc(), new_doc);
+
+    ICEBERG_UNWRAP_OR_FAIL(auto json, ToJson(*metadata));
+    ICEBERG_UNWRAP_OR_FAIL(auto restored, TableMetadataFromJson(json));
+    ICEBERG_UNWRAP_OR_FAIL(auto restored_schema, restored->Schema());
+    EXPECT_EQ(restored_schema->fields()[0].doc(), new_doc);
+    EXPECT_EQ(restored->current_schema_id, metadata->current_schema_id);
+    ASSERT_EQ(restored->schemas.size(), 2);
+    EXPECT_EQ(restored->schemas[0]->fields()[0].doc(), old_doc);
+
+    // Repeating the new doc must reuse its schema ID.
+    builder = TableMetadataBuilder::BuildFrom(metadata.get());
+    builder->SetCurrentSchema(make_schema(new_doc), metadata->last_column_id);
+    ICEBERG_UNWRAP_OR_FAIL(auto unchanged, builder->Build());
+    EXPECT_EQ(unchanged->schemas.size(), 2);
+    EXPECT_EQ(unchanged->current_schema_id, metadata->current_schema_id);
+    EXPECT_TRUE(builder->changes().empty());
+
+    // Restoring the old doc must reuse the historical schema, without adding a third.
+    builder = TableMetadataBuilder::BuildFrom(metadata.get());
+    builder->SetCurrentSchema(make_schema(old_doc), metadata->last_column_id);
+    ICEBERG_UNWRAP_OR_FAIL(auto reverted, builder->Build());
+    EXPECT_EQ(reverted->schemas.size(), 2);
+    EXPECT_EQ(reverted->current_schema_id, base->current_schema_id);
+    ICEBERG_UNWRAP_OR_FAIL(auto reverted_schema, reverted->Schema());
+    EXPECT_EQ(reverted_schema->fields()[0].doc(), old_doc);
+  }
+}
+
 TEST(TableMetadataBuilderTest, SetCurrentSchemaWithInvalidLastColumnId) {
   auto base = CreateBaseMetadata();
   auto builder = TableMetadataBuilder::BuildFrom(base.get());