feat: support column default values (initial-default / write-default)

Implements Iceberg v3 column default values (#730, item 2 of #637):

- Schema model: SchemaField carries optional `initial-default` and
  `write-default` literals, with validation that defaults are primitive
  and match the field type.
- JSON serde: parse and write the two fields using single-value
  serialization (resolves the TODO in struct field serialization).
- Read path: Project() maps a missing column with an initial-default to
  FieldProjection::Kind::kDefault carrying the literal (for both required
  and optional columns, per spec), and the Parquet and Avro readers
  materialize it as a constant column via a new Literal-to-Arrow helper.
  This resolves the default-value TODO in schema projection.
- Schema evolution: Add*Column accept an optional default value (used as
  both initial-default and write-default); a required column with a
  default no longer needs AllowIncompatibleChanges(); RequireColumn()
  accepts columns added with a default in the same update (resolves the
  defaulted-add TODO); UpdateColumnDefault() updates the write-default of
  an existing column; doc/rename/type-promotion updates preserve defaults
  (promotion casts them to the new type).
- Format version gating: Schema::Validate() rejects schemas with default
  values below v3, using the existing kMinFormatVersionDefaultValues.

Writers consume complete Arrow arrays, so applying write-default to
omitted columns remains the engine's responsibility (as in Java); the
library stores, validates, and serializes it.

Author: huan233usc

src/iceberg/CMakeLists.txt

@@ -244,6 +244,7 @@ if(ICEBERG_BUILD_BUNDLE)
       arrow/arrow_io.cc
       arrow/s3/arrow_s3_file_io.cc
       arrow/arrow_register.cc
+      arrow/literal_util.cc
       arrow/metadata_column_util.cc
       avro/avro_data_util.cc
       avro/avro_direct_decoder.cc

src/iceberg/arrow/literal_util.cc

@@ -0,0 +1,185 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include <string>
+#include <utility>
+#include <vector>
+
+#include <arrow/array.h>
+#include <arrow/array/builder_base.h>
+#include <arrow/array/util.h>
+#include <arrow/buffer.h>
+#include <arrow/compute/api.h>
+#include <arrow/scalar.h>
+#include <arrow/type.h>
+
+#include "iceberg/arrow/arrow_status_internal.h"
+#include "iceberg/arrow/literal_util_internal.h"
+#include "iceberg/type.h"
+#include "iceberg/util/checked_cast.h"
+#include "iceberg/util/formatter.h"  // IWYU pragma: keep
+#include "iceberg/util/macros.h"
+
+namespace iceberg::arrow {
+
+namespace {
+
+Result<std::shared_ptr<::arrow::DataType>> ToArrowType(const PrimitiveType& type) {
+  switch (type.type_id()) {
+    case TypeId::kBoolean:
+      return ::arrow::boolean();
+    case TypeId::kInt:
+      return ::arrow::int32();
+    case TypeId::kLong:
+      return ::arrow::int64();
+    case TypeId::kFloat:
+      return ::arrow::float32();
+    case TypeId::kDouble:
+      return ::arrow::float64();
+    case TypeId::kDecimal: {
+      const auto& decimal_type = internal::checked_cast<const DecimalType&>(type);
+      return ::arrow::decimal128(decimal_type.precision(), decimal_type.scale());
+    }
+    case TypeId::kDate:
+      return ::arrow::date32();
+    case TypeId::kTime:
+      return ::arrow::time64(::arrow::TimeUnit::MICRO);
+    case TypeId::kTimestamp:
+      return ::arrow::timestamp(::arrow::TimeUnit::MICRO);
+    case TypeId::kTimestampTz:
+      return ::arrow::timestamp(::arrow::TimeUnit::MICRO, "UTC");
+    case TypeId::kTimestampNs:
+      return ::arrow::timestamp(::arrow::TimeUnit::NANO);
+    case TypeId::kTimestampTzNs:
+      return ::arrow::timestamp(::arrow::TimeUnit::NANO, "UTC");
+    case TypeId::kString:
+      return ::arrow::utf8();
+    case TypeId::kBinary:
+      return ::arrow::binary();
+    case TypeId::kFixed: {
+      const auto& fixed_type = internal::checked_cast<const FixedType&>(type);
+      return ::arrow::fixed_size_binary(static_cast<int32_t>(fixed_type.length()));
+    }
+    case TypeId::kUuid:
+      return ::arrow::fixed_size_binary(16);
+    default:
+      return NotSupported("Cannot convert {} to an Arrow type", type);
+  }
+}
+
+Result<std::shared_ptr<::arrow::Buffer>> ToArrowBuffer(
+    const std::vector<uint8_t>& bytes) {
+  ICEBERG_ARROW_ASSIGN_OR_RETURN(auto buffer, ::arrow::AllocateBuffer(bytes.size()));
+  std::memcpy(buffer->mutable_data(), bytes.data(), bytes.size());
+  return std::shared_ptr<::arrow::Buffer>(std::move(buffer));
+}
+
+}  // namespace
+
+Result<std::shared_ptr<::arrow::Scalar>> ToArrowScalar(const Literal& literal) {
+  if (literal.type() == nullptr) {
+    return InvalidArgument("Cannot convert a literal without type to an Arrow scalar");
+  }
+
+  ICEBERG_ASSIGN_OR_RAISE(auto arrow_type, ToArrowType(*literal.type()));
+  if (literal.IsNull()) {
+    return ::arrow::MakeNullScalar(std::move(arrow_type));
+  }
+
+  const auto& value = literal.value();
+  switch (literal.type()->type_id()) {
+    case TypeId::kBoolean:
+      return std::make_shared<::arrow::BooleanScalar>(std::get<bool>(value));
+    case TypeId::kInt:
+      return std::make_shared<::arrow::Int32Scalar>(std::get<int32_t>(value));
+    case TypeId::kLong:
+      return std::make_shared<::arrow::Int64Scalar>(std::get<int64_t>(value));
+    case TypeId::kFloat:
+      return std::make_shared<::arrow::FloatScalar>(std::get<float>(value));
+    case TypeId::kDouble:
+      return std::make_shared<::arrow::DoubleScalar>(std::get<double>(value));
+    case TypeId::kDecimal: {
+      const auto& decimal = std::get<Decimal>(value);
+      ::arrow::Decimal128 arrow_decimal(
+          static_cast<int64_t>(decimal.value() >> 64),
+          static_cast<uint64_t>(decimal.value() & ~uint64_t{0}));
+      return std::make_shared<::arrow::Decimal128Scalar>(arrow_decimal,
+                                                         std::move(arrow_type));
+    }
+    case TypeId::kDate:
+      return std::make_shared<::arrow::Date32Scalar>(std::get<int32_t>(value));
+    case TypeId::kTime:
+      return std::make_shared<::arrow::Time64Scalar>(std::get<int64_t>(value),
+                                                     std::move(arrow_type));
+    case TypeId::kTimestamp:
+    case TypeId::kTimestampTz:
+    case TypeId::kTimestampNs:
+    case TypeId::kTimestampTzNs:
+      return std::make_shared<::arrow::TimestampScalar>(std::get<int64_t>(value),
+                                                        std::move(arrow_type));
+    case TypeId::kString:
+      return std::make_shared<::arrow::StringScalar>(std::get<std::string>(value));
+    case TypeId::kBinary: {
+      ICEBERG_ASSIGN_OR_RAISE(auto buffer,
+                              ToArrowBuffer(std::get<std::vector<uint8_t>>(value)));
+      return std::make_shared<::arrow::BinaryScalar>(std::move(buffer));
+    }
+    case TypeId::kFixed: {
+      ICEBERG_ASSIGN_OR_RAISE(auto buffer,
+                              ToArrowBuffer(std::get<std::vector<uint8_t>>(value)));
+      return std::make_shared<::arrow::FixedSizeBinaryScalar>(std::move(buffer),
+                                                              std::move(arrow_type));
+    }
+    case TypeId::kUuid: {
+      const auto& uuid = std::get<Uuid>(value);
+      ICEBERG_ASSIGN_OR_RAISE(
+          auto buffer,
+          ToArrowBuffer(std::vector<uint8_t>(uuid.bytes().begin(), uuid.bytes().end())));
+      return std::make_shared<::arrow::FixedSizeBinaryScalar>(std::move(buffer),
+                                                              std::move(arrow_type));
+    }
+    default:
+      return NotSupported("Cannot convert {} literal to an Arrow scalar",
+                          *literal.type());
+  }
+}
+
+Result<std::shared_ptr<::arrow::Array>> MakeDefaultArray(
+    const Literal& literal, const std::shared_ptr<::arrow::DataType>& type,
+    int64_t num_rows, ::arrow::MemoryPool* pool) {
+  ICEBERG_ASSIGN_OR_RAISE(auto scalar, ToArrowScalar(literal));
+  ICEBERG_ARROW_ASSIGN_OR_RETURN(auto array,
+                                 ::arrow::MakeArrayFromScalar(*scalar, num_rows, pool));
+  if (!array->type()->Equals(*type)) {
+    ICEBERG_ARROW_ASSIGN_OR_RETURN(auto cast_result, ::arrow::compute::Cast(array, type));
+    return cast_result.make_array();
+  }
+  return array;
+}
+
+Status AppendDefaultToBuilder(const Literal& literal, ::arrow::ArrayBuilder* builder) {
+  ICEBERG_ASSIGN_OR_RAISE(auto scalar, ToArrowScalar(literal));
+  if (!scalar->type->Equals(*builder->type())) {
+    ICEBERG_ARROW_ASSIGN_OR_RETURN(scalar, scalar->CastTo(builder->type()));
+  }
+  ICEBERG_ARROW_RETURN_NOT_OK(builder->AppendScalar(*scalar));
+  return {};
+}
+
+}  // namespace iceberg::arrow

src/iceberg/arrow/literal_util_internal.h

@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#pragma once
+
+#include <cstdint>
+#include <memory>
+
+#include <arrow/type_fwd.h>
+
+#include "iceberg/expression/literal.h"
+#include "iceberg/result.h"
+
+namespace iceberg::arrow {
+
+/// \brief Convert a primitive literal to an Arrow scalar of its canonical Arrow type.
+///
+/// A null literal converts to a null scalar of the corresponding Arrow type.
+Result<std::shared_ptr<::arrow::Scalar>> ToArrowScalar(const Literal& literal);
+
+/// \brief Create an Arrow array of `num_rows` rows where every row holds the literal
+/// value, e.g. to materialize a missing column with a default value.
+///
+/// The array is cast to `type` when the literal's canonical Arrow type differs.
+Result<std::shared_ptr<::arrow::Array>> MakeDefaultArray(
+    const Literal& literal, const std::shared_ptr<::arrow::DataType>& type,
+    int64_t num_rows, ::arrow::MemoryPool* pool);
+
+/// \brief Append the literal value once to `builder`, e.g. to materialize a missing
+/// field with a default value while building rows.
+Status AppendDefaultToBuilder(const Literal& literal, ::arrow::ArrayBuilder* builder);
+
+}  // namespace iceberg::arrow

src/iceberg/avro/avro_data_util.cc

@@ -31,6 +31,7 @@
 #include <avro/Types.hh>
 
 #include "iceberg/arrow/arrow_status_internal.h"
+#include "iceberg/arrow/literal_util_internal.h"
 #include "iceberg/avro/avro_data_util_internal.h"
 #include "iceberg/avro/avro_schema_util_internal.h"
 #include "iceberg/metadata_columns.h"
@@ -87,6 +88,9 @@ Status AppendStructToBuilder(const ::avro::NodePtr& avro_node,
                                                      metadata_context, field_builder));
     } else if (field_projection.kind == FieldProjection::Kind::kNull) {
       ICEBERG_ARROW_RETURN_NOT_OK(field_builder->AppendNull());
+    } else if (field_projection.kind == FieldProjection::Kind::kDefault) {
+      ICEBERG_RETURN_UNEXPECTED(arrow::AppendDefaultToBuilder(
+          std::get<Literal>(field_projection.from), field_builder));
     } else if (field_projection.kind == FieldProjection::Kind::kMetadata) {
       int32_t field_id = expected_field.field_id();
       if (field_id == MetadataColumns::kFilePathColumnId) {
@@ -462,6 +466,11 @@ Status AppendFieldToBuilder(const ::avro::NodePtr& avro_node,
     return {};
   }
 
+  if (projection.kind == FieldProjection::Kind::kDefault) {
+    return arrow::AppendDefaultToBuilder(std::get<Literal>(projection.from),
+                                         array_builder);
+  }
+
   if (avro_node->type() == ::avro::AVRO_UNION) {
     size_t branch = avro_datum.unionBranch();
     if (avro_node->leafAt(branch)->type() == ::avro::AVRO_NULL) {

src/iceberg/avro/avro_schema_util.cc

@@ -730,6 +730,10 @@ Result<FieldProjection> ProjectStruct(const StructType& struct_type,
                                            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 {

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(auto initial_default_json,
+                          GetJsonValueOptional<nlohmann::json>(json, kInitialDefault));
+  ICEBERG_ASSIGN_OR_RAISE(auto 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(auto literal,
+                            LiteralFromJson(*initial_default_json, field.type().get()));
+    field = field.WithInitialDefault(std::move(literal));
+  }
+  if (write_default_json.has_value()) {
+    ICEBERG_ASSIGN_OR_RAISE(auto 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/parquet/parquet_data_util.cc

@@ -24,6 +24,7 @@
 #include <arrow/type.h>
 
 #include "iceberg/arrow/arrow_status_internal.h"
+#include "iceberg/arrow/literal_util_internal.h"
 #include "iceberg/metadata_columns.h"
 #include "iceberg/parquet/parquet_data_util_internal.h"
 #include "iceberg/schema.h"
@@ -119,6 +120,11 @@ Result<std::shared_ptr<::arrow::Array>> ProjectStructArray(
       ICEBERG_ASSIGN_OR_RAISE(
           projected_array,
           MakeNullArray(output_arrow_type, struct_array->length(), pool));
+    } else if (field_projection.kind == FieldProjection::Kind::kDefault) {
+      ICEBERG_ASSIGN_OR_RAISE(
+          projected_array,
+          arrow::MakeDefaultArray(std::get<Literal>(field_projection.from),
+                                  output_arrow_type, struct_array->length(), pool));
     } else if (field_projection.kind == FieldProjection::Kind::kMetadata) {
       int32_t field_id = projected_field.field_id();
       if (field_id == MetadataColumns::kFilePathColumnId) {

src/iceberg/schema.cc

@@ -447,7 +447,18 @@ Status Schema::Validate(int32_t format_version) const {
       }
     }
 
-    // TODO(GuoTao.yu): Check default values when they are supported
+    // Column default values require v3+.
+    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 {};