Schema Evolution RecordBatch processor

Cargo.toml

@@ -39,6 +39,7 @@ rust-version = "1.77.1"
 anyhow = "1.0.72"
 apache-avro = "0.17"
 array-init = "2"
+arrow = { version = "52" }
 arrow-arith = { version = "52" }
 arrow-array = { version = "52" }
 arrow-ord = { version = "52" }

crates/iceberg/Cargo.toml

@@ -44,6 +44,7 @@ tokio = ["dep:tokio"]
 anyhow = { workspace = true }
 apache-avro = { workspace = true }
 array-init = { workspace = true }
+arrow = { workspace = true }
 arrow-arith = { workspace = true }
 arrow-array = { workspace = true }
 arrow-ord = { workspace = true }

crates/iceberg/src/arrow/mod.rs

@@ -20,4 +20,6 @@
 mod schema;
 pub use schema::*;
 mod reader;
+pub(crate) mod record_batch_evolution_processor;
+
 pub use reader::*;

crates/iceberg/src/arrow/reader.rs

@@ -38,6 +38,7 @@ use parquet::arrow::{ParquetRecordBatchStreamBuilder, ProjectionMask, PARQUET_FI
 use parquet::file::metadata::ParquetMetaData;
 use parquet::schema::types::{SchemaDescriptor, Type as ParquetType};
 
+use crate::arrow::record_batch_evolution_processor::RecordBatchEvolutionProcessor;
 use crate::arrow::{arrow_schema_to_schema, get_arrow_datum};
 use crate::error::Result;
 use crate::expr::visitors::bound_predicate_visitor::{visit, BoundPredicateVisitor};
@@ -194,6 +195,11 @@ impl ArrowReader {
         )?;
         record_batch_stream_builder = record_batch_stream_builder.with_projection(projection_mask);
 
+        // create a RecordBatchEvolutionProcessor if our task schema contains columns
+        // not present in the parquet file or whose types have been promoted
+        let mut record_batch_evolution_processor =
+            RecordBatchEvolutionProcessor::build(task.schema_ref(), task.project_field_ids());
+
         if let Some(batch_size) = batch_size {
             record_batch_stream_builder = record_batch_stream_builder.with_batch_size(batch_size);
         }
@@ -233,8 +239,10 @@ impl ArrowReader {
         // Build the batch stream and send all the RecordBatches that it generates
         // to the requester.
         let mut record_batch_stream = record_batch_stream_builder.build()?;
+
         while let Some(batch) = record_batch_stream.try_next().await? {
-            tx.send(Ok(batch)).await?
+            tx.send(record_batch_evolution_processor.process_record_batch(batch))
+                .await?
         }
 
         Ok(())