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

First of a multi-part split of column default value support (#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 #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,27 @@ 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));
+  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));
+
+  std::shared_ptr<const Literal> initial_default;
+  if (initial_default_json.has_value()) {
+    ICEBERG_ASSIGN_OR_RAISE(Literal literal,
+                            LiteralFromJson(*initial_default_json, type.get()));
+    initial_default = std::make_shared<const Literal>(std::move(literal));
+  }
+  std::shared_ptr<const Literal> write_default;
+  if (write_default_json.has_value()) {
+    ICEBERG_ASSIGN_OR_RAISE(Literal literal,
+                            LiteralFromJson(*write_default_json, type.get()));
+    write_default = std::make_shared<const Literal>(std::move(literal));
+  }
 
   return std::make_unique<SchemaField>(field_id, std::move(name), std::move(type),
-                                       !required, doc);
+                                       !required, doc, std::move(initial_default),
+                                       std::move(write_default));
 }
 
 Result<std::unique_ptr<Schema>> SchemaFromJson(const nlohmann::json& json) {

src/iceberg/schema.cc

@@ -116,9 +116,15 @@ std::shared_ptr<Type> ReassignTypeIds(const std::shared_ptr<Type>& type,
 SchemaField ReassignField(const SchemaField& field, int32_t new_id,
                           const Schema::GetId& get_id, Schema::IdMap& ids_to_reassigned,
                           Schema::IdMap& ids_to_original) {
-  return {new_id, std::string(field.name()),
+  // Reassigning IDs only rewrites the field ID and nested type IDs; share the field's
+  // (immutable) default values rather than copying them.
+  return {new_id,
+          std::string(field.name()),
           ReassignTypeIds(field.type(), get_id, ids_to_reassigned, ids_to_original),
-          field.optional(), std::string(field.doc())};
+          field.optional(),
+          std::string(field.doc()),
+          field.initial_default_ptr(),
+          field.write_default_ptr()};
 }
 
 std::vector<SchemaField> ReassignIds(std::vector<SchemaField> fields,
@@ -447,7 +453,21 @@ Status Schema::Validate(int32_t format_version) const {
       }
     }
 
-    // TODO(GuoTao.yu): Check default values when they are supported
+    // Only the initial-default is gated on format version: it changes how existing
+    // data files are read (rows written before the column existed materialize this
+    // value), so it requires the v3 reader contract. A write-default only affects
+    // values written going forward and does not reinterpret existing data.
+    if (field.initial_default().has_value() &&
+        format_version < TableMetadata::kMinFormatVersionDefaultValues) {
+      return InvalidSchema(
+          "Invalid initial default for {}: non-null default ({}) is not supported "
+          "until v{}",
+          field.name(), field.initial_default()->get(),
+          TableMetadata::kMinFormatVersionDefaultValues);
+    }
+    if (field.initial_default().has_value() || field.write_default().has_value()) {
+      ICEBERG_RETURN_UNEXPECTED(field.Validate());
+    }
   }
 
   return {};

src/iceberg/schema_field.cc

@@ -21,19 +21,26 @@
 
 #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 {
 
 SchemaField::SchemaField(int32_t field_id, std::string_view name,
-                         std::shared_ptr<Type> type, bool optional, std::string_view doc)
+                         std::shared_ptr<Type> type, bool optional, std::string_view doc,
+                         std::shared_ptr<const Literal> initial_default,
+                         std::shared_ptr<const Literal> write_default)
     : field_id_(field_id),
       name_(name),
       type_(std::move(type)),
       optional_(optional),
-      doc_(doc) {}
+      doc_(doc),
+      initial_default_(std::move(initial_default)),
+      write_default_(std::move(write_default)) {}
 
 SchemaField SchemaField::MakeOptional(int32_t field_id, std::string_view name,
                                       std::shared_ptr<Type> type, std::string_view doc) {
@@ -55,13 +62,76 @@ 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_);
+}
+
+const std::shared_ptr<const Literal>& SchemaField::initial_default_ptr() const {
+  return initial_default_;
+}
+
+const std::shared_ptr<const Literal>& SchemaField::write_default_ptr() const {
+  return 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::WithWriteDefault(Literal write_default) const {
+  SchemaField copy = *this;
+  copy.write_default_ = std::make_shared<const Literal>(std::move(write_default));
+  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 +142,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>
 
@@ -46,8 +48,14 @@ class ICEBERG_EXPORT SchemaField : public iceberg::util::Formattable {
   /// \param[in] type The field type.
   /// \param[in] optional Whether values of this field are required or nullable.
   /// \param[in] doc Optional documentation string for the field.
+  /// \param[in] initial_default The v3 `initial-default` value, or null if absent. The
+  /// field shares ownership of the (immutable) value.
+  /// \param[in] write_default The v3 `write-default` value, or null if absent. The field
+  /// shares ownership of the (immutable) value.
   SchemaField(int32_t field_id, std::string_view name, std::shared_ptr<Type> type,
-              bool optional, std::string_view doc = {});
+              bool optional, std::string_view doc = {},
+              std::shared_ptr<const Literal> initial_default = nullptr,
+              std::shared_ptr<const Literal> write_default = nullptr);
 
   /// \brief Construct an optional (nullable) field.
   static SchemaField MakeOptional(int32_t field_id, std::string_view name,
@@ -71,6 +79,32 @@ 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;
+
+  /// \brief Get the shared owning pointer to the `initial-default` value, or null if
+  /// absent. Prefer initial_default() for reading; this exists so a rebuilt field can
+  /// share the (immutable) value rather than copy it.
+  [[nodiscard]] const std::shared_ptr<const Literal>& initial_default_ptr() const;
+
+  /// \brief Get the shared owning pointer to the `write-default` value, or null if
+  /// absent. Prefer write_default() for reading; this exists so a rebuilt field can
+  /// share the (immutable) value rather than copy it.
+  [[nodiscard]] const std::shared_ptr<const Literal>& write_default_ptr() const;
+
   [[nodiscard]] std::string ToString() const override;
 
   Status Validate() const;
@@ -91,6 +125,18 @@ 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. To build a field with defaults
+  /// without an intermediate copy, pass them to the constructor instead.
+  [[nodiscard]] SchemaField WithInitialDefault(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. To build a field with defaults
+  /// without an intermediate copy, pass them to the constructor instead.
+  [[nodiscard]] SchemaField WithWriteDefault(Literal write_default) const;
+
  private:
   /// \brief Compare two fields for equality.
   [[nodiscard]] bool Equals(const SchemaField& other) const;
@@ -100,6 +146,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));