feat(avro): apply column default values when reading missing fields (#800)

## What

Part 3 of 4 of Iceberg v3 column default-value support (POC #731), built
on the
schema layer (#746) and the Parquet read path (#792).

When a column is present in the read (table) schema but absent from an
Avro data
file — because the column was added after those rows were written — fill
it with
the column's v3 `initial-default` instead of `null`.

## Changes

- **Avro projection** (`avro_schema_util.cc`): when a field is missing
from the
  file and carries an `initial-default`, project it as
`FieldProjection::Kind::kDefault`, mirroring the generic / Parquet
paths.
- **Avro decode** (`avro_data_util.cc`, `avro_direct_decoder.cc`):
materialize
the `kDefault` branch through an Avro-local `AppendDefaultToBuilder`
helper.
  It reuses the shared `ToArrowScalar` conversion, while keeping Avro's
row-by-row `ArrayBuilder` append behavior out of the shared Arrow
utility.

## Tests

- `avro_data_test`: `AppendDefaultToBuilder` appends a value and casts
to the
builder type; `AppendDatumToBuilder` fills missing required and optional
  default fields.
- `avro_test`: end-to-end — write an Avro file with an old schema, then
read it
through `ReaderFactoryRegistry` with an evolved schema carrying defaults
  (`ReadMissingFieldsWithDefaults`).

## Stack

1. #746 — schema: represent / serialize / validate (merged)
2. #792 — read path: Parquet (merged)
3. **this PR** — read path: Avro
4. schema evolution: `addColumn` / `updateColumnDefault` (#793)

Author: huan233usc

src/iceberg/avro/avro_data_util.cc

@@ -17,12 +17,15 @@
  * under the License.
  */
 
+#include <span>
+
 #include <arrow/array/builder_binary.h>
 #include <arrow/array/builder_decimal.h>
 #include <arrow/array/builder_nested.h>
 #include <arrow/array/builder_primitive.h>
 #include <arrow/extension_type.h>
 #include <arrow/json/from_string.h>
+#include <arrow/scalar.h>
 #include <arrow/type.h>
 #include <arrow/util/decimal.h>
 #include <avro/Generic.hh>
@@ -31,6 +34,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"
@@ -88,6 +92,8 @@ 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(AppendDefaultToBuilder(field_projection, field_builder));
     } else if (field_projection.kind == FieldProjection::Kind::kMetadata) {
       int32_t field_id = expected_field.field_id();
       if (field_id == MetadataColumns::kFilePathColumnId) {
@@ -466,6 +472,10 @@ Status AppendFieldToBuilder(const ::avro::NodePtr& avro_node,
     return {};
   }
 
+  if (projection.kind == FieldProjection::Kind::kDefault) {
+    return AppendDefaultToBuilder(projection, array_builder);
+  }
+
   const bool is_row_lineage =
       MetadataColumns::IsRowLineageColumn(projected_field.field_id());
 
@@ -497,6 +507,148 @@ Status AppendFieldToBuilder(const ::avro::NodePtr& avro_node,
 
 }  // namespace
 
+namespace {
+
+Result<std::shared_ptr<::arrow::Scalar>> MakeDefaultScalar(
+    const Literal& literal, const std::shared_ptr<::arrow::DataType>& builder_type) {
+  // The builder's own memory pool is not exposed, so the small scalar buffer uses the
+  // default pool.
+  ICEBERG_ASSIGN_OR_RAISE(std::shared_ptr<::arrow::Scalar> scalar,
+                          arrow::ToArrowScalar(literal, ::arrow::default_memory_pool()));
+
+  // For an extension builder (e.g. `arrow.uuid`) target its storage type: ToArrowScalar
+  // yields the storage scalar (fixed_size_binary(16) for uuid) and Scalar::CastTo has no
+  // kernel that targets an extension type. This mirrors MakeDefaultArray's extension
+  // handling.
+  std::shared_ptr<::arrow::DataType> target_type = builder_type;
+  if (target_type->id() == ::arrow::Type::EXTENSION) {
+    target_type = internal::checked_cast<const ::arrow::ExtensionType&>(*target_type)
+                      .storage_type();
+  }
+
+  if (!scalar->type->Equals(*target_type)) {
+    ICEBERG_ARROW_ASSIGN_OR_RETURN(scalar, scalar->CastTo(target_type));
+  }
+  return scalar;
+}
+
+Status PrepareStructDefaultScalars(std::span<FieldProjection> projections,
+                                   ::arrow::ArrayBuilder* builder);
+
+// Recurse into whatever nested builder this projection describes, so a default cached for
+// a struct field is found at any nesting depth (e.g. `list<list<struct<...>>>`) instead
+// of only when the collection's child is an immediate struct.
+Status PrepareNestedDefaultScalars(FieldProjection& projection,
+                                   ::arrow::ArrayBuilder* builder) {
+  if (projection.kind != FieldProjection::Kind::kProjected ||
+      projection.children.empty()) {
+    return {};
+  }
+
+  switch (builder->type()->id()) {
+    case ::arrow::Type::STRUCT:
+      return PrepareStructDefaultScalars(projection.children, builder);
+    case ::arrow::Type::LIST: {
+      // List projections store a single child for the element.
+      auto* list_builder = internal::checked_cast<::arrow::ListBuilder*>(builder);
+      return PrepareNestedDefaultScalars(projection.children[0],
+                                         list_builder->value_builder());
+    }
+    case ::arrow::Type::LARGE_LIST: {
+      auto* list_builder = internal::checked_cast<::arrow::LargeListBuilder*>(builder);
+      return PrepareNestedDefaultScalars(projection.children[0],
+                                         list_builder->value_builder());
+    }
+    case ::arrow::Type::MAP: {
+      auto* map_builder = internal::checked_cast<::arrow::MapBuilder*>(builder);
+      if (projection.children.size() >= 1) {
+        ICEBERG_RETURN_UNEXPECTED(PrepareNestedDefaultScalars(
+            projection.children[0], map_builder->key_builder()));
+      }
+      if (projection.children.size() >= 2) {
+        ICEBERG_RETURN_UNEXPECTED(PrepareNestedDefaultScalars(
+            projection.children[1], map_builder->item_builder()));
+      }
+      return {};
+    }
+    default:
+      return {};
+  }
+}
+
+Status PrepareStructDefaultScalars(std::span<FieldProjection> projections,
+                                   ::arrow::ArrayBuilder* builder) {
+  auto* struct_builder = internal::checked_cast<::arrow::StructBuilder*>(builder);
+  if (static_cast<size_t>(struct_builder->num_fields()) != projections.size()) {
+    return InvalidArgument(
+        "Inconsistent number of struct builder fields ({}) and projections ({})",
+        struct_builder->num_fields(), projections.size());
+  }
+
+  for (size_t i = 0; i < projections.size(); ++i) {
+    auto& field_projection = projections[i];
+    auto* field_builder = struct_builder->field_builder(static_cast<int>(i));
+
+    if (field_projection.kind == FieldProjection::Kind::kDefault) {
+      // Get-or-create the single Avro attributes container: another Avro attribute may
+      // have created it already, so guard on `default_scalar` being unset rather than on
+      // the container's presence (otherwise a pre-existing container would skip
+      // preparation and silently fall back to per-row scalar rebuilding).
+      std::shared_ptr<AvroExtraAttributes> attrs;
+      if (field_projection.attributes == nullptr) {
+        attrs = std::make_shared<AvroExtraAttributes>();
+        field_projection.attributes = attrs;
+      } else {
+        // Avro attaches only AvroExtraAttributes; checked_pointer_cast asserts that
+        // invariant in debug rather than silently reinterpreting a foreign attributes
+        // type.
+        attrs = internal::checked_pointer_cast<AvroExtraAttributes>(
+            field_projection.attributes);
+      }
+      if (attrs->default_scalar == nullptr) {
+        ICEBERG_ASSIGN_OR_RAISE(
+            attrs->default_scalar,
+            MakeDefaultScalar(std::get<Literal>(field_projection.from),
+                              field_builder->type()));
+      }
+      continue;
+    }
+
+    ICEBERG_RETURN_UNEXPECTED(
+        PrepareNestedDefaultScalars(field_projection, field_builder));
+  }
+  return {};
+}
+
+}  // namespace
+
+Status PrepareDefaultScalars(SchemaProjection& projection,
+                             ::arrow::ArrayBuilder* root_builder) {
+  return PrepareStructDefaultScalars(projection.fields, root_builder);
+}
+
+Status AppendDefaultToBuilder(const Literal& literal, ::arrow::ArrayBuilder* builder) {
+  ICEBERG_ASSIGN_OR_RAISE(std::shared_ptr<::arrow::Scalar> scalar,
+                          MakeDefaultScalar(literal, builder->type()));
+  ICEBERG_ARROW_RETURN_NOT_OK(builder->AppendScalar(*scalar));
+  return {};
+}
+
+Status AppendDefaultToBuilder(const FieldProjection& projection,
+                              ::arrow::ArrayBuilder* builder) {
+  // Avro projections carry a single attributes type, so once one is attached it is an
+  // AvroExtraAttributes; use checked_cast instead of a per-row dynamic_cast.
+  if (projection.attributes != nullptr) {
+    const auto& attrs =
+        internal::checked_cast<const AvroExtraAttributes&>(*projection.attributes);
+    if (attrs.default_scalar != nullptr) {
+      ICEBERG_ARROW_RETURN_NOT_OK(builder->AppendScalar(*attrs.default_scalar));
+      return {};
+    }
+  }
+  return AppendDefaultToBuilder(std::get<Literal>(projection.from), builder);
+}
+
 Status AppendDatumToBuilder(const ::avro::NodePtr& avro_node,
                             const ::avro::GenericDatum& avro_datum,
                             const SchemaProjection& projection,

src/iceberg/avro/avro_data_util_internal.h

@@ -19,14 +19,51 @@
 
 #pragma once
 
+#include <memory>
+
 #include <arrow/array/builder_base.h>
+#include <arrow/scalar.h>
 #include <avro/GenericDatum.hh>
 
 #include "iceberg/arrow/metadata_column_util_internal.h"
+#include "iceberg/expression/literal.h"
 #include "iceberg/schema_util.h"
 
 namespace iceberg::avro {
 
+/// \brief Avro-specific per-field projection attributes.
+///
+/// `FieldProjection` has a single attributes slot, so all Avro-side attributes live in
+/// one container (mirroring `ParquetExtraAttributes`) rather than separate subclasses
+/// that could not coexist. `default_scalar` is the Arrow scalar for a `kDefault` field,
+/// materialized once (see `PrepareDefaultScalars`) so row-by-row decode only needs
+/// `AppendScalar` instead of repeating `ToArrowScalar` / `CastTo` per row.
+struct AvroExtraAttributes : FieldProjection::ExtraAttributes {
+  std::shared_ptr<::arrow::Scalar> default_scalar;
+};
+
+/// \brief Precompute cast Arrow scalars for every `kDefault` field under `projection`.
+///
+/// Walks `root_builder` in lockstep with the projection so each default is cast to the
+/// builder's Arrow type once per scan. Safe to call repeatedly; existing
+/// `AvroExtraAttributes` entries are left unchanged.
+Status PrepareDefaultScalars(SchemaProjection& projection,
+                             ::arrow::ArrayBuilder* root_builder);
+
+/// \brief Append a literal once to `builder` while decoding Avro row-by-row.
+///
+/// Used to materialize `FieldProjection::Kind::kDefault`. Shares `ToArrowScalar` with
+/// Parquet's batch path (`MakeDefaultArray`); the append shape stays Avro-local because
+/// Avro builds Arrow arrays via per-row `ArrayBuilder`s rather than whole-column arrays.
+/// Prefer the `FieldProjection` overload after `PrepareDefaultScalars` so the scalar is
+/// reused across rows.
+Status AppendDefaultToBuilder(const Literal& literal, ::arrow::ArrayBuilder* builder);
+
+/// \brief Append a `kDefault` projection, reusing a scalar cached on
+/// `projection.attributes` when present.
+Status AppendDefaultToBuilder(const FieldProjection& projection,
+                              ::arrow::ArrayBuilder* builder);
+
 /// \brief Append an Avro datum to an Arrow array builder.
 ///
 /// This function handles schema evolution by using the provided projection to map

src/iceberg/avro/avro_direct_decoder.cc

@@ -29,6 +29,7 @@
 #include <avro/Types.hh>
 
 #include "iceberg/arrow/arrow_status_internal.h"
+#include "iceberg/avro/avro_data_util_internal.h"
 #include "iceberg/avro/avro_direct_decoder_internal.h"
 #include "iceberg/avro/avro_schema_util_internal.h"
 #include "iceberg/metadata_columns.h"
@@ -209,6 +210,8 @@ Status DecodeStructToBuilder(const ::avro::NodePtr& avro_node, ::avro::Decoder&
     auto* field_builder = struct_builder->field_builder(static_cast<int>(proj_idx));
     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(AppendDefaultToBuilder(field_projection, field_builder));
     } else if (field_projection.kind == FieldProjection::Kind::kMetadata) {
       int32_t field_id = expected_field.field_id();
       if (field_id == MetadataColumns::kFilePathColumnId) {

src/iceberg/avro/avro_reader.cc

@@ -434,6 +434,8 @@ class AvroReader::Impl {
                            builder_result.status().message());
     }
     context_->builder_ = builder_result.MoveValueUnsafe();
+    ICEBERG_RETURN_UNEXPECTED(
+        PrepareDefaultScalars(projection_, context_->builder_.get()));
     backend_->InitReadContext(backend_->GetReaderSchema());
 
     return {};

src/iceberg/avro/avro_schema_util.cc

@@ -752,6 +752,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() != 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 {