✨ feat(data): Apply defaults in data writers

Pass the input schema through DataWriter so Parquet and Avro writers can align missing columns with write defaults.

Add an end-to-end test that distinguishes write-default from initial-default for both formats.

Refs #637

Author: huan233usc

src/iceberg/avro/avro_writer.cc

@@ -32,6 +32,7 @@
 
 #include "iceberg/arrow/arrow_io_internal.h"
 #include "iceberg/arrow/arrow_status_internal.h"
+#include "iceberg/arrow/write_schema_util_internal.h"
 #include "iceberg/avro/avro_data_util_internal.h"
 #include "iceberg/avro/avro_direct_encoder_internal.h"
 #include "iceberg/avro/avro_metrics.h"
@@ -181,6 +182,7 @@ class AvroWriter::Impl {
  public:
   Status Open(const WriterOptions& options) {
     write_schema_ = options.schema;
+    input_schema_ = options.input_schema;
 
     ::avro::NodePtr root;
     ICEBERG_RETURN_UNEXPECTED(ToAvroNodeVisitor{}.Visit(*write_schema_, &root));
@@ -229,6 +231,13 @@ class AvroWriter::Impl {
   }
 
   Status Write(ArrowArray* data) {
+    ArrowArray aligned_data{};
+    if (input_schema_ != nullptr) {
+      ICEBERG_ASSIGN_OR_RAISE(
+          aligned_data, arrow::AlignBatchForWrite(data, *input_schema_, *write_schema_,
+                                                  ::arrow::default_memory_pool()));
+      data = &aligned_data;
+    }
     ICEBERG_ARROW_ASSIGN_OR_RETURN(auto batch,
                                    ::arrow::ImportRecordBatch(data, arrow_schema_));
 
@@ -272,6 +281,8 @@ class AvroWriter::Impl {
  private:
   // The schema to write.
   std::shared_ptr<::iceberg::Schema> write_schema_;
+  // Schema of incoming Arrow batches. Null means batches already match write_schema_.
+  std::shared_ptr<::iceberg::Schema> input_schema_;
   // The avro schema to write.
   std::shared_ptr<::avro::ValidSchema> avro_schema_;
   // Arrow output stream of the Avro file to write

src/iceberg/data/data_writer.cc

@@ -36,6 +36,7 @@ class DataWriter::Impl {
     WriterOptions writer_options{
         .path = options.path,
         .schema = options.schema,
+        .input_schema = options.input_schema,
         .io = options.io,
         .properties = WriterProperties::FromMap(options.properties),
     };

src/iceberg/data/data_writer.h

@@ -42,6 +42,8 @@ namespace iceberg {
 struct ICEBERG_DATA_EXPORT DataWriterOptions {
   std::string path;
   std::shared_ptr<Schema> schema;
+  /// Schema of Arrow batches passed to Write. When null, batches must match `schema`.
+  std::shared_ptr<Schema> input_schema;
   std::shared_ptr<PartitionSpec> spec;
   PartitionValues partition;
   FileFormatType format = FileFormatType::kParquet;

src/iceberg/parquet/parquet_writer.cc

@@ -40,6 +40,7 @@
 
 #include "iceberg/arrow/arrow_io_internal.h"
 #include "iceberg/arrow/arrow_status_internal.h"
+#include "iceberg/arrow/write_schema_util_internal.h"
 #include "iceberg/parquet/parquet_metrics_internal.h"
 #include "iceberg/schema_internal.h"
 #include "iceberg/type.h"
@@ -239,6 +240,7 @@ class ParquetWriter::Impl {
  public:
   Status Open(const WriterOptions& options) {
     schema_ = options.schema;
+    input_schema_ = options.input_schema;
 
     ICEBERG_ASSIGN_OR_RAISE(auto compression, ParseCompression(options.properties));
     ICEBERG_ASSIGN_OR_RAISE(auto compression_level, ParseCodecLevel(options.properties));
@@ -283,6 +285,12 @@ class ParquetWriter::Impl {
   }
 
   Status Write(ArrowArray* array) {
+    ArrowArray aligned_array{};
+    if (input_schema_ != nullptr) {
+      ICEBERG_ASSIGN_OR_RAISE(aligned_array, arrow::AlignBatchForWrite(
+                                                 array, *input_schema_, *schema_, pool_));
+      array = &aligned_array;
+    }
     ICEBERG_ARROW_ASSIGN_OR_RETURN(auto batch,
                                    ::arrow::ImportRecordBatch(array, arrow_schema_));
 
@@ -344,6 +352,8 @@ class ParquetWriter::Impl {
   ::arrow::MemoryPool* pool_ = ::arrow::default_memory_pool();
   // Schema to write from the Iceberg table.
   std::shared_ptr<Schema> schema_;
+  // Schema of incoming Arrow batches. Null means batches already match schema_.
+  std::shared_ptr<Schema> input_schema_;
   // Schema to write from the Parquet file.
   std::shared_ptr<::arrow::Schema> arrow_schema_;
   // Parquet schema descriptor generated from the Arrow schema.

src/iceberg/test/data_writer_test.cc

@@ -29,6 +29,7 @@
 #include "iceberg/avro/avro_register.h"
 #include "iceberg/data/equality_delete_writer.h"
 #include "iceberg/data/position_delete_writer.h"
+#include "iceberg/expression/literal.h"
 #include "iceberg/file_format.h"
 #include "iceberg/file_reader.h"
 #include "iceberg/manifest/manifest_entry.h"
@@ -188,6 +189,44 @@ TEST_P(DataWriterFormatTest, WriteRowLineage) {
                                                                [3, 102, 7]])"));
 }
 
+TEST_P(DataWriterFormatTest, WriteMissingColumnUsesWriteDefault) {
+  auto [format, path] = GetParam();
+  auto input_schema = std::make_shared<Schema>(
+      std::vector<SchemaField>{SchemaField::MakeRequired(1, "id", int32())});
+  auto write_schema = std::make_shared<Schema>(std::vector<SchemaField>{
+      SchemaField::MakeRequired(1, "id", int32()),
+      SchemaField(2, "added", int32(), /*optional=*/false, /*doc=*/{},
+                  std::make_shared<const Literal>(Literal::Int(42)),
+                  std::make_shared<const Literal>(Literal::Int(7))),
+  });
+  DataWriterOptions options{
+      .path = path,
+      .schema = write_schema,
+      .input_schema = input_schema,
+      .spec = partition_spec_,
+      .partition = PartitionValues{},
+      .format = format,
+      .io = file_io_,
+      .properties = FormatProperties(format),
+  };
+
+  ICEBERG_UNWRAP_OR_FAIL(auto writer, DataWriter::Make(options));
+  auto input = CreateArray(*input_schema, R"([[1], [2]])");
+  ArrowArray arrow_array;
+  ASSERT_TRUE(::arrow::ExportArray(*input, &arrow_array).ok());
+  ASSERT_THAT(writer->Write(&arrow_array), IsOk());
+  ASSERT_THAT(writer->Close(), IsOk());
+
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto reader,
+      ReaderFactoryRegistry::Open(
+          format, {.path = path, .io = file_io_, .projection = write_schema}));
+
+  // The file was written after the column was added, so the missing input column uses
+  // write-default (7), not initial-default (42).
+  ASSERT_NO_FATAL_FAILURE(VerifyNextBatch(*reader, *write_schema, R"([[1, 7], [2, 7]])"));
+}
+
 INSTANTIATE_TEST_SUITE_P(
     FormatTypes, DataWriterFormatTest,
     ::testing::Values(std::make_pair(FileFormatType::kParquet, "test_data.parquet"),