feat(schema): represent, serialize and validate v3 column default values

First of a multi-part split of column default value support (apache#730) — the
schema foundation the read and evolution paths build on. Purely additive;
no read/write behavior change on its own.

- SchemaField carries `initial-default` / `write-default` (immutable
  std::shared_ptr<const Literal>) with copy-preserving WithInitialDefault /
  WithWriteDefault modifiers; getters return optional<reference_wrapper>.
- JSON serde reads/writes `initial-default` / `write-default` via the
  existing single-value serialization.
- Schema::Validate rejects default values below format v3 and validates
  they are non-null primitive literals matching the field type.
- Generic schema projection maps a column missing from a data file with an
  initial-default to FieldProjection::Kind::kDefault.

Read-path application (Parquet/Avro) and schema evolution follow in separate
PRs. See apache#731 for the full end-to-end proof-of-concept.

Author: huan233usc

src/iceberg/json_serde.cc

@@ -27,6 +27,8 @@
 #include <nlohmann/json.hpp>
 
 #include "iceberg/constants.h"
+#include "iceberg/expression/json_serde_internal.h"
+#include "iceberg/expression/literal.h"
 #include "iceberg/json_serde_internal.h"
 #include "iceberg/name_mapping.h"
 #include "iceberg/partition_field.h"
@@ -298,6 +300,15 @@ nlohmann::json ToJson(const SchemaField& field) {
   if (!field.doc().empty()) {
     json[kDoc] = field.doc();
   }
+  // Defaults are validated to be primitive literals matching the field type, so
+  // single-value serialization cannot fail here.
+  if (field.initial_default().has_value()) {
+    ICEBERG_ASSIGN_OR_THROW(json[kInitialDefault],
+                            ToJson(field.initial_default()->get()));
+  }
+  if (field.write_default().has_value()) {
+    ICEBERG_ASSIGN_OR_THROW(json[kWriteDefault], ToJson(field.write_default()->get()));
+  }
   return json;
 }
 
@@ -310,7 +321,6 @@ nlohmann::json ToJson(const Type& type) {
       nlohmann::json fields_json = nlohmann::json::array();
       for (const auto& field : struct_type.fields()) {
         fields_json.push_back(ToJson(field));
-        // TODO(gangwu): add default values
       }
       json[kFields] = fields_json;
       return json;
@@ -552,9 +562,23 @@ Result<std::unique_ptr<SchemaField>> FieldFromJson(const nlohmann::json& json) {
   ICEBERG_ASSIGN_OR_RAISE(auto name, GetJsonValue<std::string>(json, kName));
   ICEBERG_ASSIGN_OR_RAISE(auto required, GetJsonValue<bool>(json, kRequired));
   ICEBERG_ASSIGN_OR_RAISE(auto doc, GetJsonValueOrDefault<std::string>(json, kDoc));
-
-  return std::make_unique<SchemaField>(field_id, std::move(name), std::move(type),
-                                       !required, doc);
+  ICEBERG_ASSIGN_OR_RAISE(std::optional<nlohmann::json> initial_default_json,
+                          GetJsonValueOptional<nlohmann::json>(json, kInitialDefault));
+  ICEBERG_ASSIGN_OR_RAISE(std::optional<nlohmann::json> write_default_json,
+                          GetJsonValueOptional<nlohmann::json>(json, kWriteDefault));
+
+  SchemaField field(field_id, std::move(name), std::move(type), !required, doc);
+  if (initial_default_json.has_value()) {
+    ICEBERG_ASSIGN_OR_RAISE(Literal literal,
+                            LiteralFromJson(*initial_default_json, field.type().get()));
+    field = field.WithInitialDefault(std::move(literal));
+  }
+  if (write_default_json.has_value()) {
+    ICEBERG_ASSIGN_OR_RAISE(Literal literal,
+                            LiteralFromJson(*write_default_json, field.type().get()));
+    field = field.WithWriteDefault(std::move(literal));
+  }
+  return std::make_unique<SchemaField>(std::move(field));
 }
 
 Result<std::unique_ptr<Schema>> SchemaFromJson(const nlohmann::json& json) {

src/iceberg/schema.cc

@@ -447,7 +447,15 @@ Status Schema::Validate(int32_t format_version) const {
       }
     }
 
-    // TODO(GuoTao.yu): Check default values when they are supported
+    // Column default values (both initial-default and write-default) require v3+.
+    if (field.initial_default().has_value() || field.write_default().has_value()) {
+      if (format_version < TableMetadata::kMinFormatVersionDefaultValues) {
+        return InvalidSchema(
+            "Invalid default value for {}: default values are not supported until v{}",
+            field.name(), TableMetadata::kMinFormatVersionDefaultValues);
+      }
+      ICEBERG_RETURN_UNEXPECTED(field.Validate());
+    }
   }
 
   return {};

src/iceberg/schema_field.cc

@@ -21,9 +21,12 @@
 
 #include <format>
 #include <string_view>
+#include <utility>
 
+#include "iceberg/expression/literal.h"
 #include "iceberg/type.h"
 #include "iceberg/util/formatter.h"  // IWYU pragma: keep
+#include "iceberg/util/macros.h"
 
 namespace iceberg {
 
@@ -55,13 +58,86 @@ bool SchemaField::optional() const { return optional_; }
 
 std::string_view SchemaField::doc() const { return doc_; }
 
+std::optional<std::reference_wrapper<const Literal>> SchemaField::initial_default()
+    const {
+  if (initial_default_ == nullptr) {
+    return std::nullopt;
+  }
+  return std::cref(*initial_default_);
+}
+
+std::optional<std::reference_wrapper<const Literal>> SchemaField::write_default() const {
+  if (write_default_ == nullptr) {
+    return std::nullopt;
+  }
+  return std::cref(*write_default_);
+}
+
+SchemaField SchemaField::WithInitialDefault(Literal initial_default) const {
+  SchemaField copy = *this;
+  copy.initial_default_ = std::make_shared<const Literal>(std::move(initial_default));
+  return copy;
+}
+
+SchemaField SchemaField::WithInitialDefault(
+    std::optional<std::reference_wrapper<const Literal>> initial_default) const {
+  SchemaField copy = *this;
+  copy.initial_default_ = initial_default.has_value()
+                              ? std::make_shared<const Literal>(initial_default->get())
+                              : nullptr;
+  return copy;
+}
+
+SchemaField SchemaField::WithWriteDefault(Literal write_default) const {
+  SchemaField copy = *this;
+  copy.write_default_ = std::make_shared<const Literal>(std::move(write_default));
+  return copy;
+}
+
+SchemaField SchemaField::WithWriteDefault(
+    std::optional<std::reference_wrapper<const Literal>> write_default) const {
+  SchemaField copy = *this;
+  copy.write_default_ = write_default.has_value()
+                            ? std::make_shared<const Literal>(write_default->get())
+                            : nullptr;
+  return copy;
+}
+
+namespace {
+
+Status ValidateDefault(const SchemaField& field, const Literal& value,
+                       std::string_view kind) {
+  if (value.IsNull() || value.IsAboveMax() || value.IsBelowMin()) {
+    return InvalidSchema("Invalid {} value for {}: must be a non-null value", kind,
+                         field.name());
+  }
+  if (field.type() == nullptr || !field.type()->is_primitive()) {
+    return InvalidSchema("Invalid {} value for {}: {} (must be null)", kind, field.name(),
+                         value);
+  }
+  if (*value.type() != *field.type()) {
+    return InvalidSchema("{} of field {} has type {} but expected {}", kind, field.name(),
+                         *value.type(), *field.type());
+  }
+  return {};
+}
+
+}  // namespace
+
 Status SchemaField::Validate() const {
   if (name_.empty()) [[unlikely]] {
     return InvalidSchema("SchemaField cannot have empty name");
   }
   if (type_ == nullptr) [[unlikely]] {
     return InvalidSchema("SchemaField cannot have null type");
   }
+  if (initial_default_ != nullptr) {
+    ICEBERG_RETURN_UNEXPECTED(
+        ValidateDefault(*this, *initial_default_, "initial-default"));
+  }
+  if (write_default_ != nullptr) {
+    ICEBERG_RETURN_UNEXPECTED(ValidateDefault(*this, *write_default_, "write-default"));
+  }
   return {};
 }
 
@@ -72,9 +148,23 @@ std::string SchemaField::ToString() const {
   return result;
 }
 
+namespace {
+
+bool DefaultEquals(const std::shared_ptr<const Literal>& lhs,
+                   const std::shared_ptr<const Literal>& rhs) {
+  if (lhs == nullptr || rhs == nullptr) {
+    return lhs == rhs;
+  }
+  return *lhs == *rhs;
+}
+
+}  // namespace
+
 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_ &&
+         DefaultEquals(initial_default_, other.initial_default_) &&
+         DefaultEquals(write_default_, other.write_default_);
 }
 
 }  // namespace iceberg

src/iceberg/schema_field.h

@@ -24,7 +24,9 @@
 /// type (e.g. a struct).
 
 #include <cstdint>
+#include <functional>
 #include <memory>
+#include <optional>
 #include <string>
 #include <string_view>
 
@@ -71,6 +73,22 @@ class ICEBERG_EXPORT SchemaField : public iceberg::util::Formattable {
   /// \brief Get the field documentation.
   std::string_view doc() const;
 
+  /// \brief Get the default value for this field used when reading rows written
+  /// before the field existed (v3 `initial-default`). Empty if absent.
+  ///
+  /// The returned reference is a non-owning view into a value owned by this field;
+  /// it remains valid for the lifetime of this SchemaField.
+  [[nodiscard]] std::optional<std::reference_wrapper<const Literal>> initial_default()
+      const;
+
+  /// \brief Get the default value for this field used when a writer does not
+  /// supply a value (v3 `write-default`). Empty if absent.
+  ///
+  /// The returned reference is a non-owning view into a value owned by this field;
+  /// it remains valid for the lifetime of this SchemaField.
+  [[nodiscard]] std::optional<std::reference_wrapper<const Literal>> write_default()
+      const;
+
   [[nodiscard]] std::string ToString() const override;
 
   Status Validate() const;
@@ -91,6 +109,30 @@ class ICEBERG_EXPORT SchemaField : public iceberg::util::Formattable {
     return copy;
   }
 
+  /// \brief Return a copy of this field with the given `initial-default` value.
+  ///
+  /// The returned field takes ownership of the value.
+  [[nodiscard]] SchemaField WithInitialDefault(Literal initial_default) const;
+
+  /// \brief Return a copy of this field with the `initial-default` value copied from
+  /// `initial_default`, or without one if it is empty.
+  ///
+  /// The referenced value is copied; no reference to it is retained.
+  [[nodiscard]] SchemaField WithInitialDefault(
+      std::optional<std::reference_wrapper<const Literal>> initial_default) const;
+
+  /// \brief Return a copy of this field with the given `write-default` value.
+  ///
+  /// The returned field takes ownership of the value.
+  [[nodiscard]] SchemaField WithWriteDefault(Literal write_default) const;
+
+  /// \brief Return a copy of this field with the `write-default` value copied from
+  /// `write_default`, or without one if it is empty.
+  ///
+  /// The referenced value is copied; no reference to it is retained.
+  [[nodiscard]] SchemaField WithWriteDefault(
+      std::optional<std::reference_wrapper<const Literal>> write_default) const;
+
  private:
   /// \brief Compare two fields for equality.
   [[nodiscard]] bool Equals(const SchemaField& other) const;
@@ -100,6 +142,11 @@ class ICEBERG_EXPORT SchemaField : public iceberg::util::Formattable {
   std::shared_ptr<Type> type_;
   bool optional_;
   std::string doc_;
+  // Default values are owned by this field and never mutated after being set; copies
+  // of the field share the same payload (reference-counted) instead of deep-copying,
+  // like `type_` above. Sharing is unobservable because the payload is immutable.
+  std::shared_ptr<const Literal> initial_default_;
+  std::shared_ptr<const Literal> write_default_;
 };
 
 }  // namespace iceberg

src/iceberg/schema_util.cc

@@ -172,10 +172,14 @@ Result<FieldProjection> ProjectNested(const Type& expected_type, const Type& sou
                                            iter->second.local_index, prune_source));
     } else if (MetadataColumns::IsMetadataColumn(field_id)) {
       child_projection.kind = FieldProjection::Kind::kMetadata;
+    } else if (expected_field.initial_default().has_value()) {
+      // Rows written before the field existed assume its `initial-default` value.
+      child_projection.kind = FieldProjection::Kind::kDefault;
+      child_projection.from = expected_field.initial_default()->get();
     } else if (expected_field.optional()) {
       child_projection.kind = FieldProjection::Kind::kNull;
     } else {
-      // TODO(gangwu): support default value for v3 and constant value
+      // TODO(gangwu): support constant value
       return InvalidSchema("Missing required field: {}", expected_field.ToString());
     }
     result.children.emplace_back(std::move(child_projection));