From 0a25d45c8a3a6fc674f99fce6633ad7bbd127ea9 Mon Sep 17 00:00:00 2001 From: huanghsiang_cheng Date: Wed, 10 Jun 2026 22:55:29 -0700 Subject: [PATCH 01/10] Support selecting metadata column: _spec_id --- crates/iceberg/src/arrow/reader/pipeline.rs | 14 +- .../src/arrow/record_batch_transformer.rs | 21 ++- crates/iceberg/src/inspect/manifests.rs | 4 +- crates/iceberg/src/scan/context.rs | 15 ++- crates/iceberg/src/scan/mod.rs | 125 ++++++++++++++++-- 5 files changed, 158 insertions(+), 21 deletions(-) diff --git a/crates/iceberg/src/arrow/reader/pipeline.rs b/crates/iceberg/src/arrow/reader/pipeline.rs index ef38bc8b7c..9a5cf1e3f3 100644 --- a/crates/iceberg/src/arrow/reader/pipeline.rs +++ b/crates/iceberg/src/arrow/reader/pipeline.rs @@ -37,7 +37,9 @@ use crate::arrow::record_batch_transformer::RecordBatchTransformerBuilder; use crate::arrow::scan_metrics::{CountingFileRead, ScanMetrics, ScanResult}; use crate::error::Result; use crate::io::{FileIO, FileMetadata, FileRead}; -use crate::metadata_columns::{RESERVED_FIELD_ID_FILE, is_metadata_field}; +use crate::metadata_columns::{ + RESERVED_FIELD_ID_FILE, RESERVED_FIELD_ID_SPEC_ID, is_metadata_field, +}; use crate::scan::{ArrowRecordBatchStream, FileScanTask, FileScanTaskStream}; use crate::spec::Datum; use crate::{Error, ErrorKind}; @@ -248,6 +250,16 @@ impl FileScanTaskReader { record_batch_transformer_builder.with_constant(RESERVED_FIELD_ID_FILE, file_datum); } + if task + .project_field_ids() + .contains(&RESERVED_FIELD_ID_SPEC_ID) + && let Some(partition_spec) = &task.partition_spec + { + let spec_id_datum = Datum::int(partition_spec.spec_id()); + record_batch_transformer_builder = record_batch_transformer_builder + .with_constant(RESERVED_FIELD_ID_SPEC_ID, spec_id_datum); + } + if let (Some(partition_spec), Some(partition_data)) = (task.partition_spec.clone(), task.partition.clone()) { diff --git a/crates/iceberg/src/arrow/record_batch_transformer.rs b/crates/iceberg/src/arrow/record_batch_transformer.rs index 439358435c..b1ffb7b52c 100644 --- a/crates/iceberg/src/arrow/record_batch_transformer.rs +++ b/crates/iceberg/src/arrow/record_batch_transformer.rs @@ -383,11 +383,7 @@ impl RecordBatchTransformer { .get(field_id) .ok_or(Error::new(ErrorKind::Unexpected, "field not found"))? .0; - let datum = constant_fields.get(field_id).ok_or(Error::new( - ErrorKind::Unexpected, - "constant field not found", - ))?; - let arrow_type = datum_to_arrow_type_with_ree(datum); + let arrow_type = field.data_type().clone(); // Use the type from constant_fields (REE for constants) let constant_field = Field::new(field.name(), arrow_type, field.is_nullable()) @@ -486,7 +482,20 @@ impl RecordBatchTransformer { // they exist in the Parquet file. This is per Iceberg spec rule #1: partition metadata // is authoritative and should be preferred over file data. if let Some(datum) = constant_fields.get(field_id) { - let arrow_type = datum_to_arrow_type_with_ree(datum); + let arrow_type = if get_metadata_field(*field_id).is_ok() { + datum_to_arrow_type_with_ree(datum) + } else { + field_id_to_mapped_schema_map + .get(field_id) + .ok_or(Error::new( + ErrorKind::Unexpected, + "could not find field in schema", + ))? + .0 + .data_type() + .clone() + }; + return Ok(ColumnSource::Add { value: Some(datum.literal().clone()), target_type: arrow_type, diff --git a/crates/iceberg/src/inspect/manifests.rs b/crates/iceberg/src/inspect/manifests.rs index a985786460..e5f3e1ffa6 100644 --- a/crates/iceberg/src/inspect/manifests.rs +++ b/crates/iceberg/src/inspect/manifests.rs @@ -366,12 +366,12 @@ mod tests { -- child 2: "lower_bound" (Utf8) StringArray [ - "100", + "1", ] -- child 3: "upper_bound" (Utf8) StringArray [ - "300", + "1", ] ], ]"#]], diff --git a/crates/iceberg/src/scan/context.rs b/crates/iceberg/src/scan/context.rs index 75672d9cbb..6a6b2ebd65 100644 --- a/crates/iceberg/src/scan/context.rs +++ b/crates/iceberg/src/scan/context.rs @@ -28,8 +28,8 @@ use crate::scan::{ PartitionFilterCache, }; use crate::spec::{ - ManifestContentType, ManifestEntryRef, ManifestFile, ManifestList, NameMapping, SchemaRef, - SnapshotRef, TableMetadataRef, + ManifestContentType, ManifestEntryRef, ManifestFile, ManifestList, NameMapping, + PartitionSpecRef, SchemaRef, SnapshotRef, TableMetadataRef, }; use crate::{Error, ErrorKind, Result}; @@ -48,6 +48,7 @@ pub(crate) struct ManifestFileContext { delete_file_index: DeleteFileIndex, name_mapping: Option>, case_sensitive: bool, + partition_spec: Option, } /// Wraps a [`ManifestEntryRef`] alongside the objects that are needed @@ -63,6 +64,7 @@ pub(crate) struct ManifestEntryContext { pub delete_file_index: DeleteFileIndex, pub name_mapping: Option>, pub case_sensitive: bool, + pub partition_spec: Option, } impl ManifestFileContext { @@ -80,6 +82,7 @@ impl ManifestFileContext { delete_file_index, name_mapping, case_sensitive, + partition_spec, } = self; let manifest = object_cache.get_manifest(&manifest_file).await?; @@ -96,6 +99,7 @@ impl ManifestFileContext { delete_file_index: delete_file_index.clone(), name_mapping: name_mapping.clone(), case_sensitive, + partition_spec: partition_spec.clone(), }; sender @@ -135,8 +139,7 @@ impl ManifestEntryContext { ) .with_deletes(deletes) .with_partition(Some(self.manifest_entry.data_file.partition.clone())) - // TODO: Pass actual PartitionSpec through context chain for native flow - .with_partition_spec(None) + .with_partition_spec(self.partition_spec.clone()) .with_name_mapping(self.name_mapping) .with_case_sensitive(self.case_sensitive) .build()) @@ -284,6 +287,10 @@ impl PlanContext { delete_file_index, name_mapping: self.name_mapping.clone(), case_sensitive: self.case_sensitive, + partition_spec: self + .table_metadata + .partition_spec_by_id(manifest_file.partition_spec_id) + .cloned(), } } } diff --git a/crates/iceberg/src/scan/mod.rs b/crates/iceberg/src/scan/mod.rs index 86092313c6..62db8ad909 100644 --- a/crates/iceberg/src/scan/mod.rs +++ b/crates/iceberg/src/scan/mod.rs @@ -625,8 +625,9 @@ pub mod tests { use std::sync::Arc; use arrow_array::cast::AsArray; + use arrow_array::types::Int32Type; use arrow_array::{ - Array, ArrayRef, BooleanArray, Float64Array, Int32Array, Int64Array, RecordBatch, + Array, ArrayRef, BooleanArray, Float64Array, Int32Array, Int64Array, RecordBatch, RunArray, StringArray, }; use futures::{TryStreamExt, stream}; @@ -641,7 +642,7 @@ pub mod tests { use crate::arrow::ArrowReaderBuilder; use crate::expr::{BoundPredicate, Reference}; use crate::io::{FileIO, OutputFile}; - use crate::metadata_columns::RESERVED_COL_NAME_FILE; + use crate::metadata_columns::{RESERVED_COL_NAME_FILE, RESERVED_COL_NAME_SPEC_ID}; use crate::scan::FileScanTask; use crate::spec::{ DEFAULT_SCHEMA_NAME_MAPPING, DataContentType, DataFileBuilder, DataFileFormat, Datum, @@ -862,7 +863,7 @@ pub mod tests { .file_format(DataFileFormat::Parquet) .file_size_in_bytes(parquet_file_size) .record_count(1) - .partition(Struct::from_iter([Some(Literal::long(100))])) + .partition(Struct::from_iter([Some(Literal::long(1))])) .key_metadata(None) .build() .unwrap(), @@ -885,7 +886,7 @@ pub mod tests { .file_format(DataFileFormat::Parquet) .file_size_in_bytes(parquet_file_size) .record_count(1) - .partition(Struct::from_iter([Some(Literal::long(200))])) + .partition(Struct::from_iter([Some(Literal::long(1))])) .build() .unwrap(), ) @@ -907,7 +908,7 @@ pub mod tests { .file_format(DataFileFormat::Parquet) .file_size_in_bytes(parquet_file_size) .record_count(1) - .partition(Struct::from_iter([Some(Literal::long(300))])) + .partition(Struct::from_iter([Some(Literal::long(1))])) .build() .unwrap(), ) @@ -2025,8 +2026,6 @@ pub mod tests { #[tokio::test] async fn test_select_with_file_column() { - use arrow_array::cast::AsArray; - let mut fixture = TableTestFixture::new(); fixture.setup_manifest_files().await; @@ -2070,7 +2069,7 @@ pub mod tests { // Decode the RunArray to verify it contains the file path let run_array = file_col .as_any() - .downcast_ref::>() + .downcast_ref::>() .expect("_file column should be a RunArray"); let values = run_array.values(); @@ -2369,4 +2368,114 @@ pub mod tests { // Assert it finished (didn't timeout) assert!(result.is_ok(), "Scan timed out - deadlock detected"); } + + #[tokio::test] + async fn test_select_with_spec_id_column() { + let mut fixture = TableTestFixture::new(); + fixture.setup_manifest_files().await; + + // Select regular columns plus the _spec_id column + let table_scan = fixture + .table + .scan() + .select(["x", RESERVED_COL_NAME_SPEC_ID, "z"]) + .with_row_selection_enabled(true) + .build() + .unwrap(); + + let batch_stream = table_scan.to_arrow().await.unwrap(); + let batches: Vec<_> = batch_stream.try_collect().await.unwrap(); + + // Verify we have 3 columns: x, _spec_id, and z + assert_eq!(batches[0].num_columns(), 3); + + // Verify the x column exists and has correct data + let col1 = batches[0].column_by_name("x").unwrap(); + let int64_arr = col1.as_any().downcast_ref::().unwrap(); + assert_eq!(int64_arr.value(0), 1); + + // Verify the _spec_id column exists + let spec_id_col = batches[0].column_by_name(RESERVED_COL_NAME_SPEC_ID); + assert!( + spec_id_col.is_some(), + "_spec_id column should be present in the batch" + ); + + // Verify the _spec_id data type + let spec_id_col = spec_id_col.unwrap(); + assert!( + matches!( + spec_id_col.data_type(), + arrow_schema::DataType::RunEndEncoded(_, _) + ), + "_spec_id column should use RunEndEncoded type" + ); + + // Decode the RunArray to verify it contains the spec id + let run_array = spec_id_col + .as_any() + .downcast_ref::>() + .expect("_spec_id column should be a RunArray"); + + let values = run_array.values(); + let int_values = values.as_primitive::(); + assert_eq!(int_values.len(), 1, "Should have a single _spec_id"); + + let spec_id = int_values.value(0); + assert_eq!(spec_id, 0, "_spec_id should be 0, got: {spec_id}"); + + // Verify 'z' column exists + assert!(batches[0].column_by_name("z").is_some()); + } + + #[tokio::test] + async fn test_select_with_spec_id_column_from_unpartitioned_table() { + let mut fixture = TableTestFixture::new_unpartitioned(); + fixture.setup_unpartitioned_manifest_files().await; + + // Select regular columns plus the _spec_id column + let table_scan = fixture + .table + .scan() + .select(["x", RESERVED_COL_NAME_SPEC_ID]) + .with_row_selection_enabled(true) + .build() + .unwrap(); + + let batch_stream = table_scan.to_arrow().await.unwrap(); + let batches: Vec<_> = batch_stream.try_collect().await.unwrap(); + + // Verify we have 2 columns: x and _spec_id + assert_eq!(batches[0].num_columns(), 2); + + // Verify the _spec_id column exists + let spec_id_col = batches[0].column_by_name(RESERVED_COL_NAME_SPEC_ID); + assert!( + spec_id_col.is_some(), + "_spec_id column should be present in the batch" + ); + + // Verify the _spec_id data type + let spec_id_col = spec_id_col.unwrap(); + assert!( + matches!( + spec_id_col.data_type(), + arrow_schema::DataType::RunEndEncoded(_, _) + ), + "_spec_id column should use RunEndEncoded type" + ); + + // Decode the RunArray to verify it contains the spec id + let run_array = spec_id_col + .as_any() + .downcast_ref::>() + .expect("_spec_id column should be a RunArray"); + + let values = run_array.values(); + let int_values = values.as_primitive::(); + assert_eq!(int_values.len(), 1, "Should have a single _spec_id"); + + let spec_id = int_values.value(0); + assert_eq!(spec_id, 0, "_spec_id should be 0, got: {spec_id}"); + } } From bcc3b6d02460ae56db3fb85b83bdbd30cea708d8 Mon Sep 17 00:00:00 2001 From: huanghsiang_cheng Date: Mon, 22 Jun 2026 22:06:02 -0700 Subject: [PATCH 02/10] Fix CI --- crates/iceberg/src/arrow/value.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/crates/iceberg/src/arrow/value.rs b/crates/iceberg/src/arrow/value.rs index d07233c420..3b008b2d9d 100644 --- a/crates/iceberg/src/arrow/value.rs +++ b/crates/iceberg/src/arrow/value.rs @@ -834,6 +834,9 @@ pub(crate) fn create_primitive_array_repeated( let vals: Vec> = vec![None; num_rows]; Arc::new(Date32Array::from(vals)) } + (DataType::Int64, Some(PrimitiveLiteral::Int(value))) => { + Arc::new(Int64Array::from(vec![i64::from(*value); num_rows])) + } (DataType::Int64, Some(PrimitiveLiteral::Long(value))) => { Arc::new(Int64Array::from(vec![*value; num_rows])) } @@ -969,7 +972,7 @@ pub(crate) fn create_primitive_array_repeated( (dt, _) => { return Err(Error::new( ErrorKind::Unexpected, - format!("unexpected target column type {dt}"), + format!("unexpected target column type {dt}, prim_lit {:?}", prim_lit), )); } }) From 69e871490a6f6359b66c9ff012f2d446f5cc1c0c Mon Sep 17 00:00:00 2001 From: huanghsiang_cheng Date: Wed, 1 Jul 2026 23:03:25 -0700 Subject: [PATCH 03/10] style: inline w/ debug format --- crates/iceberg/src/arrow/value.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/iceberg/src/arrow/value.rs b/crates/iceberg/src/arrow/value.rs index 3b008b2d9d..c2c0d650d8 100644 --- a/crates/iceberg/src/arrow/value.rs +++ b/crates/iceberg/src/arrow/value.rs @@ -972,7 +972,7 @@ pub(crate) fn create_primitive_array_repeated( (dt, _) => { return Err(Error::new( ErrorKind::Unexpected, - format!("unexpected target column type {dt}, prim_lit {:?}", prim_lit), + format!("unexpected target column type {dt}, prim_lit {prim_lit:?}"), )); } }) From cf26ad174d2e3f093418d956e90f8da623fc9510 Mon Sep 17 00:00:00 2001 From: huanghsiang_cheng Date: Thu, 2 Jul 2026 11:18:59 -0700 Subject: [PATCH 04/10] Partition spec is always present --- crates/iceberg/src/arrow/reader/pipeline.rs | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/crates/iceberg/src/arrow/reader/pipeline.rs b/crates/iceberg/src/arrow/reader/pipeline.rs index 9a5cf1e3f3..e63d130c33 100644 --- a/crates/iceberg/src/arrow/reader/pipeline.rs +++ b/crates/iceberg/src/arrow/reader/pipeline.rs @@ -253,9 +253,8 @@ impl FileScanTaskReader { if task .project_field_ids() .contains(&RESERVED_FIELD_ID_SPEC_ID) - && let Some(partition_spec) = &task.partition_spec { - let spec_id_datum = Datum::int(partition_spec.spec_id()); + let spec_id_datum = Datum::int(task.partition_spec.clone().unwrap().spec_id()); record_batch_transformer_builder = record_batch_transformer_builder .with_constant(RESERVED_FIELD_ID_SPEC_ID, spec_id_datum); } From aee857e1acaf139f4ae64536f1cbb78b8f711f6d Mon Sep 17 00:00:00 2001 From: huanghsiang_cheng Date: Thu, 2 Jul 2026 15:46:40 -0700 Subject: [PATCH 05/10] Test partition evolution --- crates/iceberg/src/scan/mod.rs | 202 ++++++++++++++++++ ...table_metadata_v2_partition_evolution.json | 128 +++++++++++ 2 files changed, 330 insertions(+) create mode 100644 crates/iceberg/testdata/example_table_metadata_v2_partition_evolution.json diff --git a/crates/iceberg/src/scan/mod.rs b/crates/iceberg/src/scan/mod.rs index 62db8ad909..7049f53038 100644 --- a/crates/iceberg/src/scan/mod.rs +++ b/crates/iceberg/src/scan/mod.rs @@ -821,6 +821,40 @@ pub mod tests { } } + pub fn new_with_partition_evolution() -> Self { + let table = Self::new().table; + let table_location = table.metadata().location.clone(); + + let manifest_list1_location = + format!("{}/metadata/manifests_list_1.avro", table_location); + let manifest_list2_location = + format!("{}/metadata/manifests_list_2.avro", table_location); + let manifest_list3_location = + format!("{}/metadata/manifests_list_3.avro", table_location); + let table_metadata1_location = format!("{}/metadata/v1.json", table_location); + + let new_table_metadata = { + let template_json_str = fs::read_to_string(format!( + "{}/testdata/example_table_metadata_v2_partition_evolution.json", + env!("CARGO_MANIFEST_DIR") + )) + .unwrap(); + let metadata_json = render_template(&template_json_str, context! { + table_location => &table_location, + manifest_list_1_location => &manifest_list1_location, + manifest_list_2_location => &manifest_list2_location, + manifest_list_3_location => &manifest_list3_location, + table_metadata_1_location => &table_metadata1_location, + }); + Arc::new(serde_json::from_str::(&metadata_json).unwrap()) + }; + + Self { + table_location, + table: table.with_metadata(new_table_metadata), + } + } + fn next_manifest_file(&self) -> OutputFile { self.table .file_io() @@ -938,6 +972,124 @@ pub mod tests { manifest_list_write.close().await.unwrap(); } + pub async fn setup_manifest_files_with_partition_evolution(&mut self) { + let current_snapshot = self.table.metadata().current_snapshot().unwrap(); + let parent_snapshot = current_snapshot + .parent_snapshot(self.table.metadata()) + .unwrap(); + let current_schema = current_snapshot.schema(self.table.metadata()).unwrap(); + let current_partition_spec = self.table.metadata().default_partition_spec(); + + // Write the data files first, then use the file size in the manifest entries + let parquet_file_size = self.write_parquet_data_files(); + + let mut writer = ManifestWriterBuilder::new( + self.next_manifest_file(), + Some(current_snapshot.snapshot_id()), + None, + current_schema.clone(), + current_partition_spec.as_ref().clone(), + ) + .build_v2_data(); + writer + .add_entry( + ManifestEntry::builder() + .status(ManifestStatus::Added) + .data_file( + DataFileBuilder::default() + .partition_spec_id(1) + .content(DataContentType::Data) + .file_path(format!("{}/1.parquet", &self.table_location)) + .file_format(DataFileFormat::Parquet) + .file_size_in_bytes(parquet_file_size) + .record_count(1) + .partition(Struct::from_iter([ + Some(Literal::long(1)), + Some(Literal::string("apach")), + Some(Literal::int(27)), + ])) + .key_metadata(None) + .build() + .unwrap(), + ) + .build(), + ) + .unwrap(); + writer + .add_delete_entry( + ManifestEntry::builder() + .status(ManifestStatus::Deleted) + .snapshot_id(parent_snapshot.snapshot_id()) + .sequence_number(parent_snapshot.sequence_number()) + .file_sequence_number(parent_snapshot.sequence_number()) + .data_file( + DataFileBuilder::default() + .partition_spec_id(1) + .content(DataContentType::Data) + .file_path(format!("{}/2.parquet", &self.table_location)) + .file_format(DataFileFormat::Parquet) + .file_size_in_bytes(parquet_file_size) + .record_count(1) + .partition(Struct::from_iter([ + Some(Literal::long(1)), + Some(Literal::string("icebe")), + Some(Literal::int(5)), + ])) + .build() + .unwrap(), + ) + .build(), + ) + .unwrap(); + writer + .add_existing_entry( + ManifestEntry::builder() + .status(ManifestStatus::Existing) + .snapshot_id(parent_snapshot.snapshot_id()) + .sequence_number(parent_snapshot.sequence_number()) + .file_sequence_number(parent_snapshot.sequence_number()) + .data_file( + DataFileBuilder::default() + .partition_spec_id(1) + .content(DataContentType::Data) + .file_path(format!("{}/3.parquet", &self.table_location)) + .file_format(DataFileFormat::Parquet) + .file_size_in_bytes(parquet_file_size) + .record_count(1) + .partition(Struct::from_iter([ + Some(Literal::long(1)), + Some(Literal::string("apach")), + Some(Literal::int(19)), + ])) + .build() + .unwrap(), + ) + .build(), + ) + .unwrap(); + let data_file_manifest = writer.write_manifest_file().await.unwrap(); + + // Write to manifest list + let manifest_list_writer = self + .table + .file_io() + .new_output(current_snapshot.manifest_list()) + .unwrap() + .writer() + .await + .unwrap(); + let mut manifest_list_write = ManifestListWriter::v2( + manifest_list_writer, + current_snapshot.snapshot_id(), + current_snapshot.parent_snapshot_id(), + current_snapshot.sequence_number(), + ); + manifest_list_write + .add_manifests(vec![data_file_manifest].into_iter()) + .unwrap(); + manifest_list_write.close().await.unwrap(); + } + /// Writes identical Parquet data files (1.parquet, 2.parquet, 3.parquet) /// and returns the file size in bytes. fn write_parquet_data_files(&self) -> u64 { @@ -2478,4 +2630,54 @@ pub mod tests { let spec_id = int_values.value(0); assert_eq!(spec_id, 0, "_spec_id should be 0, got: {spec_id}"); } + + #[tokio::test] + async fn test_select_with_spec_id_column_with_partition_evolution() { + let mut fixture = TableTestFixture::new_with_partition_evolution(); + fixture + .setup_manifest_files_with_partition_evolution() + .await; + + // Select regular columns plus the _spec_id column + let table_scan = fixture + .table + .scan() + .select(["x", RESERVED_COL_NAME_SPEC_ID, "z"]) + .with_row_selection_enabled(true) + .build() + .unwrap(); + + let batch_stream = table_scan.to_arrow().await.unwrap(); + let batches: Vec<_> = batch_stream.try_collect().await.unwrap(); + + // Verify the _spec_id column exists + let spec_id_col = batches[0].column_by_name(RESERVED_COL_NAME_SPEC_ID); + assert!( + spec_id_col.is_some(), + "_spec_id column should be present in the batch" + ); + + // Verify the _spec_id data type + let spec_id_col = spec_id_col.unwrap(); + assert!( + matches!( + spec_id_col.data_type(), + arrow_schema::DataType::RunEndEncoded(_, _) + ), + "_spec_id column should use RunEndEncoded type" + ); + + // Decode the RunArray to verify it contains the spec id + let run_array = spec_id_col + .as_any() + .downcast_ref::>() + .expect("_spec_id column should be a RunArray"); + + let values = run_array.values(); + let int_values = values.as_primitive::(); + assert_eq!(int_values.len(), 1, "Should have a single _spec_id"); + + let spec_id = int_values.value(0); + assert_eq!(spec_id, 2, "_spec_id should be 2, got: {spec_id}"); + } } diff --git a/crates/iceberg/testdata/example_table_metadata_v2_partition_evolution.json b/crates/iceberg/testdata/example_table_metadata_v2_partition_evolution.json new file mode 100644 index 0000000000..fbb3e9d040 --- /dev/null +++ b/crates/iceberg/testdata/example_table_metadata_v2_partition_evolution.json @@ -0,0 +1,128 @@ +{ + "format-version": 2, + "table-uuid": "9c12d441-03fe-4693-9a96-a0705ddf69c1", + "location": "{{ table_location }}", + "last-sequence-number": 34, + "last-updated-ms": 1602638573590, + "last-column-id": 3, + "current-schema-id": 1, + "schemas": [ + { + "type": "struct", + "schema-id": 0, + "fields": [ + {"id": 1, "name": "x", "required": true, "type": "long"} + ]}, + { + "type": "struct", + "schema-id": 1, + "identifier-field-ids": [1, 2], + "fields": [ + {"id": 1, "name": "x", "required": true, "type": "long"}, + {"id": 2, "name": "y", "required": true, "type": "long", "doc": "comment"}, + {"id": 3, "name": "z", "required": true, "type": "long"}, + {"id": 4, "name": "a", "required": true, "type": "string"}, + {"id": 5, "name": "dbl", "required": true, "type": "double"}, + {"id": 6, "name": "i32", "required": true, "type": "int"}, + {"id": 7, "name": "i64", "required": true, "type": "long"}, + {"id": 8, "name": "bool", "required": true, "type": "boolean"} + ] + } + ], + "default-spec-id": 2, + "partition-specs": [ + { + "spec-id": 0, + "fields": [ + {"name": "x", "transform": "identity", "source-id": 1, "field-id": 1000} + ] + }, + { + "spec-id": 1, + "fields": [ + { + "name": "x", + "transform": "identity", + "source-id": 1, + "field-id": 1000 + }, + { + "name": "a_trunc_5", + "transform": "truncate[5]", + "source-id": 4, + "field-id": 1001 + } + ] + }, + { + "spec-id": 2, + "fields": [ + { + "name": "x", + "transform": "identity", + "source-id": 1, + "field-id": 1000 + }, + { + "name": "a_trunc_5", + "transform": "truncate[5]", + "source-id": 4, + "field-id": 1001 + }, + { + "name": "z_bucket_32", + "transform": "bucket[32]", + "source-id": 3, + "field-id": 1002 + } + ] + } + ], + "last-partition-id": 1002, + "default-sort-order-id": 3, + "sort-orders": [ + { + "order-id": 3, + "fields": [ + {"transform": "identity", "source-id": 2, "direction": "asc", "null-order": "nulls-first"}, + {"transform": "bucket[4]", "source-id": 3, "direction": "desc", "null-order": "nulls-last"} + ] + } + ], + "properties": {"read.split.target.size": "134217728"}, + "current-snapshot-id": 3059729675574597004, + "snapshots": [ + { + "snapshot-id": 3051729675574597004, + "timestamp-ms": 1515100955770, + "sequence-number": 0, + "summary": {"operation": "append"}, + "manifest-list": "{{ manifest_list_1_location }}" + }, + { + "snapshot-id": 3055729675574597004, + "parent-snapshot-id": 3051729675574597004, + "timestamp-ms": 1555100955770, + "sequence-number": 1, + "summary": {"operation": "append"}, + "manifest-list": "{{ manifest_list_2_location }}", + "schema-id": 1 + }, + { + "snapshot-id": 3059729675574597004, + "parent-snapshot-id": 3055729675574597004, + "timestamp-ms": 1555100955771, + "sequence-number": 2, + "summary": {"operation": "append"}, + "manifest-list": "{{ manifest_list_3_location }}", + "schema-id": 1 + } + ], + "snapshot-log": [ + {"snapshot-id": 3051729675574597004, "timestamp-ms": 1515100955770}, + {"snapshot-id": 3055729675574597004, "timestamp-ms": 1555100955770}, + {"snapshot-id": 3059729675574597004, "timestamp-ms": 1555100955770} + ], + "metadata-log": [{"metadata-file": "{{ table_metadata_1_location }}", "timestamp-ms": 1515100}], + "refs": {"test": {"snapshot-id": 3051729675574597004, "type": "tag", "max-ref-age-ms": 10000000}} +} \ No newline at end of file From edaaf7986146ddad58cb28a7509b3d489971b15e Mon Sep 17 00:00:00 2001 From: huanghsiang_cheng Date: Mon, 6 Jul 2026 11:28:08 -0700 Subject: [PATCH 06/10] Filter field ids which are not present in a data file --- .../src/arrow/record_batch_transformer.rs | 54 +++++++++++-------- crates/iceberg/src/inspect/manifests.rs | 4 +- crates/iceberg/src/scan/mod.rs | 6 +-- 3 files changed, 37 insertions(+), 27 deletions(-) diff --git a/crates/iceberg/src/arrow/record_batch_transformer.rs b/crates/iceberg/src/arrow/record_batch_transformer.rs index b1ffb7b52c..e7850e9372 100644 --- a/crates/iceberg/src/arrow/record_batch_transformer.rs +++ b/crates/iceberg/src/arrow/record_batch_transformer.rs @@ -477,29 +477,39 @@ impl RecordBatchTransformer { projected_iceberg_field_ids .iter() .map(|field_id| { - // Check if this is a constant field (metadata/virtual or identity-partitioned) - // Constant fields always use their pre-computed constant values, regardless of whether - // they exist in the Parquet file. This is per Iceberg spec rule #1: partition metadata - // is authoritative and should be preferred over file data. + // Check if this is a constant field (metadata/virtual or identity-partitioned). + // + // Metadata/virtual fields (like _file, _spec_id) never exist in the data file, + // so they always use their pre-computed constant value. + // + // For identity-partitioned fields, the Iceberg spec's "Column Projection" rules + // only apply to "field ids which are not present in a data file". When the column + // IS present in the Parquet file, it must be read from the file; the partition + // metadata constant is only a fallback for when the column is absent (e.g. add_files). if let Some(datum) = constant_fields.get(field_id) { - let arrow_type = if get_metadata_field(*field_id).is_ok() { - datum_to_arrow_type_with_ree(datum) - } else { - field_id_to_mapped_schema_map - .get(field_id) - .ok_or(Error::new( - ErrorKind::Unexpected, - "could not find field in schema", - ))? - .0 - .data_type() - .clone() - }; - - return Ok(ColumnSource::Add { - value: Some(datum.literal().clone()), - target_type: arrow_type, - }); + let is_metadata_field = get_metadata_field(*field_id).is_ok(); + let present_in_file = field_id_to_source_schema_map.contains_key(field_id); + + if is_metadata_field || !present_in_file { + let arrow_type = if is_metadata_field { + datum_to_arrow_type_with_ree(datum) + } else { + field_id_to_mapped_schema_map + .get(field_id) + .ok_or(Error::new( + ErrorKind::Unexpected, + "could not find field in schema", + ))? + .0 + .data_type() + .clone() + }; + + return Ok(ColumnSource::Add { + value: Some(datum.literal().clone()), + target_type: arrow_type, + }); + } } let (target_field, _) = diff --git a/crates/iceberg/src/inspect/manifests.rs b/crates/iceberg/src/inspect/manifests.rs index e5f3e1ffa6..a985786460 100644 --- a/crates/iceberg/src/inspect/manifests.rs +++ b/crates/iceberg/src/inspect/manifests.rs @@ -366,12 +366,12 @@ mod tests { -- child 2: "lower_bound" (Utf8) StringArray [ - "1", + "100", ] -- child 3: "upper_bound" (Utf8) StringArray [ - "1", + "300", ] ], ]"#]], diff --git a/crates/iceberg/src/scan/mod.rs b/crates/iceberg/src/scan/mod.rs index 7049f53038..621dba7bd3 100644 --- a/crates/iceberg/src/scan/mod.rs +++ b/crates/iceberg/src/scan/mod.rs @@ -897,7 +897,7 @@ pub mod tests { .file_format(DataFileFormat::Parquet) .file_size_in_bytes(parquet_file_size) .record_count(1) - .partition(Struct::from_iter([Some(Literal::long(1))])) + .partition(Struct::from_iter([Some(Literal::long(100))])) .key_metadata(None) .build() .unwrap(), @@ -920,7 +920,7 @@ pub mod tests { .file_format(DataFileFormat::Parquet) .file_size_in_bytes(parquet_file_size) .record_count(1) - .partition(Struct::from_iter([Some(Literal::long(1))])) + .partition(Struct::from_iter([Some(Literal::long(200))])) .build() .unwrap(), ) @@ -942,7 +942,7 @@ pub mod tests { .file_format(DataFileFormat::Parquet) .file_size_in_bytes(parquet_file_size) .record_count(1) - .partition(Struct::from_iter([Some(Literal::long(1))])) + .partition(Struct::from_iter([Some(Literal::long(300))])) .build() .unwrap(), ) From c8ea392dae9808d024e3842858fdeb91a33e92d6 Mon Sep 17 00:00:00 2001 From: huanghsiang_cheng Date: Mon, 6 Jul 2026 13:09:40 -0700 Subject: [PATCH 07/10] Fix typo checks --- crates/iceberg/src/scan/mod.rs | 6 +++--- .../example_table_metadata_v2_partition_evolution.json | 8 ++++---- 2 files changed, 7 insertions(+), 7 deletions(-) diff --git a/crates/iceberg/src/scan/mod.rs b/crates/iceberg/src/scan/mod.rs index 621dba7bd3..0b1b918f8e 100644 --- a/crates/iceberg/src/scan/mod.rs +++ b/crates/iceberg/src/scan/mod.rs @@ -1005,7 +1005,7 @@ pub mod tests { .record_count(1) .partition(Struct::from_iter([ Some(Literal::long(1)), - Some(Literal::string("apach")), + Some(Literal::string("apa")), Some(Literal::int(27)), ])) .key_metadata(None) @@ -1032,7 +1032,7 @@ pub mod tests { .record_count(1) .partition(Struct::from_iter([ Some(Literal::long(1)), - Some(Literal::string("icebe")), + Some(Literal::string("ice")), Some(Literal::int(5)), ])) .build() @@ -1058,7 +1058,7 @@ pub mod tests { .record_count(1) .partition(Struct::from_iter([ Some(Literal::long(1)), - Some(Literal::string("apach")), + Some(Literal::string("apa")), Some(Literal::int(19)), ])) .build() diff --git a/crates/iceberg/testdata/example_table_metadata_v2_partition_evolution.json b/crates/iceberg/testdata/example_table_metadata_v2_partition_evolution.json index fbb3e9d040..70743c394f 100644 --- a/crates/iceberg/testdata/example_table_metadata_v2_partition_evolution.json +++ b/crates/iceberg/testdata/example_table_metadata_v2_partition_evolution.json @@ -47,8 +47,8 @@ "field-id": 1000 }, { - "name": "a_trunc_5", - "transform": "truncate[5]", + "name": "a_trunc_3", + "transform": "truncate[3]", "source-id": 4, "field-id": 1001 } @@ -64,8 +64,8 @@ "field-id": 1000 }, { - "name": "a_trunc_5", - "transform": "truncate[5]", + "name": "a_trunc_3", + "transform": "truncate[3]", "source-id": 4, "field-id": 1001 }, From 24508e2f646dff263ea0fe46de103bb443c8ff31 Mon Sep 17 00:00:00 2001 From: huanghsiang_cheng Date: Mon, 6 Jul 2026 13:33:43 -0700 Subject: [PATCH 08/10] Fix CI --- crates/iceberg/src/scan/mod.rs | 1 - 1 file changed, 1 deletion(-) diff --git a/crates/iceberg/src/scan/mod.rs b/crates/iceberg/src/scan/mod.rs index b0b54d9a62..f2ea4122fd 100644 --- a/crates/iceberg/src/scan/mod.rs +++ b/crates/iceberg/src/scan/mod.rs @@ -985,7 +985,6 @@ pub mod tests { let mut writer = ManifestWriterBuilder::new( self.next_manifest_file(), Some(current_snapshot.snapshot_id()), - None, current_schema.clone(), current_partition_spec.as_ref().clone(), ) From 5987a95c4740ed6b10c87e4b0449de3d67c50753 Mon Sep 17 00:00:00 2001 From: huanghsiang_cheng Date: Mon, 6 Jul 2026 15:33:29 -0700 Subject: [PATCH 09/10] refactor: simplify control flow --- .../src/arrow/record_batch_transformer.rs | 60 +++++++------------ 1 file changed, 22 insertions(+), 38 deletions(-) diff --git a/crates/iceberg/src/arrow/record_batch_transformer.rs b/crates/iceberg/src/arrow/record_batch_transformer.rs index e7850e9372..56f4f58a64 100644 --- a/crates/iceberg/src/arrow/record_batch_transformer.rs +++ b/crates/iceberg/src/arrow/record_batch_transformer.rs @@ -359,45 +359,29 @@ impl RecordBatchTransformer { let fields: Result> = projected_iceberg_field_ids .iter() .map(|field_id| { - // Check if this is a constant field - if constant_fields.contains_key(field_id) { - // For metadata/virtual fields (like _file), get name from metadata_columns - // For partition fields, get name from schema (they exist in schema) - if let Ok(iceberg_field) = get_metadata_field(*field_id) { - // This is a metadata/virtual field - convert Iceberg field to Arrow - let datum = constant_fields.get(field_id).ok_or(Error::new( - ErrorKind::Unexpected, - "constant field not found", - ))?; - let arrow_type = datum_to_arrow_type_with_ree(datum); - let arrow_field = - Field::new(&iceberg_field.name, arrow_type, !iceberg_field.required) - .with_metadata(HashMap::from([( - PARQUET_FIELD_ID_META_KEY.to_string(), - iceberg_field.id.to_string(), - )])); - Ok(Arc::new(arrow_field)) - } else { - // This is a partition constant field (exists in schema but uses constant value) - let field = &field_id_to_mapped_schema_map - .get(field_id) - .ok_or(Error::new(ErrorKind::Unexpected, "field not found"))? - .0; - let arrow_type = field.data_type().clone(); - // Use the type from constant_fields (REE for constants) - let constant_field = - Field::new(field.name(), arrow_type, field.is_nullable()) - .with_metadata(field.metadata().clone()); - Ok(Arc::new(constant_field)) - } - } else { - // Regular field - use schema as-is - Ok(field_id_to_mapped_schema_map - .get(field_id) - .ok_or(Error::new(ErrorKind::Unexpected, "field not found"))? - .0 - .clone()) + // Metadata/virtual fields (like _file, _spec_id) don't exist in the table + // schema, so build their Arrow field from the metadata column definition and + // the pre-computed constant's type. + if let Some(datum) = constant_fields.get(field_id) + && let Ok(iceberg_field) = get_metadata_field(*field_id) + { + let arrow_type = datum_to_arrow_type_with_ree(datum); + let arrow_field = + Field::new(&iceberg_field.name, arrow_type, !iceberg_field.required) + .with_metadata(HashMap::from([( + PARQUET_FIELD_ID_META_KEY.to_string(), + iceberg_field.id.to_string(), + )])); + return Ok(Arc::new(arrow_field)); } + + // Regular fields and identity-partitioned constant fields both exist in the + // table schema, so use the mapped Arrow field as-is. + Ok(field_id_to_mapped_schema_map + .get(field_id) + .ok_or(Error::new(ErrorKind::Unexpected, "field not found"))? + .0 + .clone()) }) .collect(); From bfd0977a034d7c298a3c95f643c5654666ef6847 Mon Sep 17 00:00:00 2001 From: huanghsiang_cheng Date: Mon, 6 Jul 2026 16:19:51 -0700 Subject: [PATCH 10/10] Make sure we read from Parquet, not partition values --- crates/iceberg/src/scan/mod.rs | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/crates/iceberg/src/scan/mod.rs b/crates/iceberg/src/scan/mod.rs index f2ea4122fd..e5f2bf9979 100644 --- a/crates/iceberg/src/scan/mod.rs +++ b/crates/iceberg/src/scan/mod.rs @@ -1002,7 +1002,7 @@ pub mod tests { .file_size_in_bytes(parquet_file_size) .record_count(1) .partition(Struct::from_iter([ - Some(Literal::long(1)), + Some(Literal::long(100)), Some(Literal::string("apa")), Some(Literal::int(27)), ])) @@ -1029,7 +1029,7 @@ pub mod tests { .file_size_in_bytes(parquet_file_size) .record_count(1) .partition(Struct::from_iter([ - Some(Literal::long(1)), + Some(Literal::long(200)), Some(Literal::string("ice")), Some(Literal::int(5)), ])) @@ -1055,7 +1055,7 @@ pub mod tests { .file_size_in_bytes(parquet_file_size) .record_count(1) .partition(Struct::from_iter([ - Some(Literal::long(1)), + Some(Literal::long(300)), Some(Literal::string("apa")), Some(Literal::int(19)), ])) @@ -2645,6 +2645,11 @@ pub mod tests { let batch_stream = table_scan.to_arrow().await.unwrap(); let batches: Vec<_> = batch_stream.try_collect().await.unwrap(); + // Verify the x column exists and has correct data + let col1 = batches[0].column_by_name("x").unwrap(); + let int64_arr = col1.as_any().downcast_ref::().unwrap(); + assert_eq!(int64_arr.value(0), 1); + // Verify the _spec_id column exists let spec_id_col = batches[0].column_by_name(RESERVED_COL_NAME_SPEC_ID); assert!(