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

Part 1 of a multi-part split of #730 (column default values, item 2 of
#637). The full
end-to-end implementation is in #731, kept open as the proof-of-concept;
this series
lands it in reviewable pieces.

This PR is the **schema foundation** — representing, serializing and
validating v3
column default values. It is purely additive and changes no read or
write behavior on
its own.

## What's in this PR

- **`SchemaField`** carries `initial-default` / `write-default`, stored
as
`std::shared_ptr<const Literal>` (immutable payload shared across
copies, like the
adjacent `type_`; the C++ analog of Java's `final Literal<?>`). They are
set via the
constructor. Getters return `std::optional<std::reference_wrapper<const
Literal>>` for
reading (the `Schema::FindFieldByName` idiom); `initial_default_ptr()` /
`write_default_ptr()` expose the shared pointer so a rebuilt field (e.g.
ID
  reassignment) shares the value instead of copying it.
- **JSON serde**: parse/write `initial-default` / `write-default` using
the existing
  single-value serialization (all primitive types).
- **`Schema::Validate`**: version-gates the `initial-default` to format
v3
(`kMinFormatVersionDefaultValues`) — it reinterprets how existing data
files are read,
so it requires the v3 reader contract. The `write-default` only affects
values written
going forward and is **not** version-gated (matching Java's
`Schema.checkCompatibility`,
which gates only the initial default). Both defaults are otherwise
validated to be
  non-null primitive literals matching the field type.
- **Generic projection**: a column missing from a data file with an
`initial-default`
maps to `FieldProjection::Kind::kDefault` carrying the literal (the
per-format readers
  consume this in the follow-up PRs).

## Follow-ups (stacked on this PR)

- read path — Parquet (`literal_util` + parquet
projection/materialization)
- read path — Avro
- schema evolution (`UpdateSchema` add/update column defaults)

## Testing

Added tests

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"
@@ -49,6 +51,7 @@
 #include "iceberg/util/json_util_internal.h"
 #include "iceberg/util/macros.h"
 #include "iceberg/util/string_util.h"
+#include "iceberg/util/temporal_util.h"
 #include "iceberg/util/timepoint.h"
 
 namespace iceberg {
@@ -324,6 +327,12 @@ Result<nlohmann::json> ToJson(const SchemaField& field) {
   if (!field.doc().empty()) {
     json[kDoc] = field.doc();
   }
+  if (field.initial_default() != nullptr) {
+    ICEBERG_ASSIGN_OR_RAISE(json[kInitialDefault], ToJson(*field.initial_default()));
+  }
+  if (field.write_default() != nullptr) {
+    ICEBERG_ASSIGN_OR_RAISE(json[kWriteDefault], ToJson(*field.write_default()));
+  }
   return json;
 }
 
@@ -337,7 +346,6 @@ Result<nlohmann::json> ToJson(const Type& type) {
       for (const auto& field : struct_type.fields()) {
         ICEBERG_ASSIGN_OR_RAISE(auto field_json, ToJson(field));
         fields_json.push_back(std::move(field_json));
-        // TODO(gangwu): add default values
       }
       json[kFields] = fields_json;
       return json;
@@ -628,16 +636,66 @@ Result<std::unique_ptr<Type>> TypeFromJson(const nlohmann::json& json) {
   }
 }
 
+namespace {
+
+// The spec's JSON single-value form for `timestamptz` / `timestamptz_ns` default
+// values requires a UTC offset. The shared timestamp parser accepts any offset and
+// silently normalizes to UTC, which would let C++ accept default metadata that Java
+// rejects and then rewrite the offset on serialization. Enforce UTC for these
+// defaults at parse time, where the original offset is still visible.
+Status ValidateTimestamptzDefaultIsUtc(const Type& type, const nlohmann::json& value) {
+  const auto type_id = type.type_id();
+  if (type_id != TypeId::kTimestampTz && type_id != TypeId::kTimestampTzNs) {
+    return {};
+  }
+  if (!value.is_string()) {
+    return JsonParseError("Invalid timestamptz default {} for {}: expected a string",
+                          SafeDumpJson(value), type.ToString());
+  }
+  const auto str = value.get<std::string>();
+  ICEBERG_ASSIGN_OR_RAISE(bool is_utc, TemporalUtils::IsUtcOffset(str));
+  if (!is_utc) {
+    return JsonParseError(
+        "Invalid timestamptz default '{}' for {}: default values must use a UTC offset",
+        str, type.ToString());
+  }
+  return {};
+}
+
+}  // namespace
+
 Result<std::unique_ptr<SchemaField>> FieldFromJson(const nlohmann::json& json) {
   ICEBERG_ASSIGN_OR_RAISE(
       auto type, GetJsonValue<nlohmann::json>(json, kType).and_then(TypeFromJson));
   ICEBERG_ASSIGN_OR_RAISE(auto field_id, GetJsonValue<int32_t>(json, kId));
   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(auto initial_default_json,
+                          GetJsonValueOptional<nlohmann::json>(json, kInitialDefault));
+  ICEBERG_ASSIGN_OR_RAISE(auto write_default_json,
+                          GetJsonValueOptional<nlohmann::json>(json, kWriteDefault));
+
+  std::shared_ptr<const Literal> initial_default;
+  if (initial_default_json.has_value()) {
+    ICEBERG_RETURN_UNEXPECTED(
+        ValidateTimestamptzDefaultIsUtc(*type, *initial_default_json));
+    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_RETURN_UNEXPECTED(
+        ValidateTimestamptzDefaultIsUtc(*type, *write_default_json));
+    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(),
+          field.write_default()};
 }
 
 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() != nullptr &&
+        format_version < TableMetadata::kMinFormatVersionDefaultValues) {
+      return InvalidSchema(
+          "Invalid initial default for {}: non-null default ({}) is not supported "
+          "until v{}",
+          field.name(), *field.initial_default(),
+          TableMetadata::kMinFormatVersionDefaultValues);
+    }
+    if (field.initial_default() != nullptr || field.write_default() != nullptr) {
+      ICEBERG_RETURN_UNEXPECTED(field.Validate());
+    }
   }
 
   return {};

src/iceberg/schema_field.cc

@@ -21,19 +21,39 @@
 
 #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 {
 
+namespace {
+
+// A null default value is modeled as the absence of a default (matching Java), so it is
+// not stored.
+std::shared_ptr<const Literal> DropNullDefault(std::shared_ptr<const Literal> value) {
+  if (value != nullptr && value->IsNull()) {
+    return nullptr;
+  }
+  return value;
+}
+
+}  // namespace
+
 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_(DropNullDefault(std::move(initial_default))),
+      write_default_(DropNullDefault(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 +75,74 @@ bool SchemaField::optional() const { return optional_; }
 
 std::string_view SchemaField::doc() const { return doc_; }
 
+const std::shared_ptr<const Literal>& SchemaField::initial_default() const {
+  return initial_default_;
+}
+
+const std::shared_ptr<const Literal>& SchemaField::write_default() const {
+  return write_default_;
+}
+
+namespace {
+
+Status ValidateDefault(const SchemaField& field, const Literal& value,
+                       std::string_view kind) {
+  // A null default is modeled as absence and dropped at construction, so it never reaches
+  // here; only the out-of-range cast sentinels need rejecting.
+  if (value.IsAboveMax() || value.IsBelowMin()) {
+    return InvalidSchema("Invalid {} value for {}: value is out of range", kind,
+                         field.name());
+  }
+  if (field.type() == nullptr) {
+    return InvalidSchema("Invalid {} value for {}: field has no type", kind,
+                         field.name());
+  }
+  // The spec requires unknown/variant/geometry/geography columns to default to null, so a
+  // non-null default on them is invalid (a null default was already dropped as absence).
+  switch (field.type()->type_id()) {
+    case TypeId::kUnknown:
+    case TypeId::kVariant:
+    case TypeId::kGeometry:
+    case TypeId::kGeography:
+      return InvalidSchema("Invalid {} value for {}: type {} cannot have a default value",
+                           kind, field.name(), *field.type());
+    default:
+      break;
+  }
+  // Defaults are otherwise only supported on primitive fields. The spec also permits JSON
+  // single-value defaults for struct/list/map (e.g. an empty struct `{}` whose sub-field
+  // defaults live in field metadata); that matches the current Java model's gap and is
+  // left as a follow-up.
+  if (!field.type()->is_primitive()) {
+    return InvalidSchema(
+        "Invalid {} value for {}: default values are only supported for primitive types",
+        kind, field.name());
+  }
+  // Defaults are stored verbatim (no implicit cast), so a default whose literal type does
+  // not match the field type is invalid.
+  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 +153,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

@@ -46,8 +46,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 +77,14 @@ class ICEBERG_EXPORT SchemaField : public iceberg::util::Formattable {
   /// \brief Get the field documentation.
   std::string_view doc() const;
 
+  /// \brief Get the owning pointer to the default value for this field used when reading
+  /// rows written before the field existed (v3 `initial-default`), or null if absent.
+  const std::shared_ptr<const Literal>& initial_default() const;
+
+  /// \brief Get the owning pointer to the default value for this field used when a writer
+  /// does not supply a value (v3 `write-default`), or null if absent.
+  const std::shared_ptr<const Literal>& write_default() const;
+
   [[nodiscard]] std::string ToString() const override;
 
   Status Validate() const;
@@ -100,6 +114,9 @@ class ICEBERG_EXPORT SchemaField : public iceberg::util::Formattable {
   std::shared_ptr<Type> type_;
   bool optional_;
   std::string doc_;
+  // Immutable default values, shared (not deep-copied) across field copies, like `type_`.
+  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() != nullptr) {
+      // Rows written before the field existed assume its `initial-default` value.
+      child_projection.kind = FieldProjection::Kind::kDefault;
+      child_projection.from = *expected_field.initial_default();
     } 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));