diff --git a/crates/iceberg/src/arrow/record_batch_transformer.rs b/crates/iceberg/src/arrow/record_batch_transformer.rs index b7b8d609a7..22dd795d32 100644 --- a/crates/iceberg/src/arrow/record_batch_transformer.rs +++ b/crates/iceberg/src/arrow/record_batch_transformer.rs @@ -59,11 +59,11 @@ fn constants_map( for (pos, field) in partition_spec.fields().iter().enumerate() { // Only identity transforms should use constant values from partition metadata if matches!(field.transform, Transform::Identity) { - // Get the field from schema to extract its type - let iceberg_field = schema.field_by_id(field.source_id).ok_or(Error::new( - ErrorKind::Unexpected, - format!("Field {} not found in schema", field.source_id), - ))?; + // The source column may have been dropped from the schema after the spec was + // created. It cannot be projected in that case, so no constant is needed. + let Some(iceberg_field) = schema.field_by_id(field.source_id) else { + continue; + }; // Ensure the field type is primitive let prim_type = match &*iceberg_field.field_type { diff --git a/crates/iceberg/src/scan/context.rs b/crates/iceberg/src/scan/context.rs index 8462eb89f4..4bd7e31de1 100644 --- a/crates/iceberg/src/scan/context.rs +++ b/crates/iceberg/src/scan/context.rs @@ -175,9 +175,23 @@ impl PlanContext { .await } + /// Returns the partition filter for a manifest, or an always-true filter when the + /// manifest's spec cannot be resolved against the snapshot schema. Files of such + /// manifests are not partition-pruned but still receive the row filter. fn get_partition_filter(&self, manifest_file: &ManifestFile) -> Result> { let partition_spec_id = manifest_file.partition_spec_id; + // Historical specs may reference source columns that were later dropped from the + // schema, in which case the partition type cannot be resolved. A missing spec is + // reported by the partition filter cache instead. + let resolvable = self + .table_metadata + .partition_spec_by_id(partition_spec_id) + .is_none_or(|spec| spec.partition_type(&self.snapshot_schema).is_ok()); + if !resolvable { + return Ok(Arc::new(BoundPredicate::AlwaysTrue)); + } + let partition_filter = self.partition_filter_cache.get( partition_spec_id, &self.table_metadata, diff --git a/crates/iceberg/src/scan/mod.rs b/crates/iceberg/src/scan/mod.rs index b377493729..50e3772e1b 100644 --- a/crates/iceberg/src/scan/mod.rs +++ b/crates/iceberg/src/scan/mod.rs @@ -646,8 +646,10 @@ pub mod tests { use crate::scan::FileScanTask; use crate::spec::{ DEFAULT_SCHEMA_NAME_MAPPING, DataContentType, DataFileBuilder, DataFileFormat, Datum, - Literal, ManifestEntry, ManifestListWriter, ManifestStatus, ManifestWriterBuilder, - NestedField, PartitionSpec, PrimitiveType, Schema, Struct, StructType, TableMetadata, Type, + Literal, MAIN_BRANCH, ManifestEntry, ManifestListWriter, ManifestStatus, + ManifestWriterBuilder, NestedField, Operation, PartitionSpec, PrimitiveType, Schema, + Snapshot, Struct, StructType, Summary, TableMetadata, TableMetadataBuilder, Type, + UnboundPartitionSpec, }; use crate::table::Table; use crate::test_utils::test_runtime; @@ -1643,6 +1645,98 @@ pub mod tests { ); } + #[tokio::test] + async fn test_filtered_scan_with_dropped_partition_source_column() { + let mut fixture = TableTestFixture::new(); + fixture.setup_manifest_files().await; + + // baseline: the same filtered scan against the table before evolution + let baseline = scan_y_gte_5(&fixture.table).await; + assert!(!baseline.is_empty()); + assert!(baseline.iter().all(|y| *y >= 5)); + + // Evolve the table so that the manifests reference a historical spec whose source + // column is no longer in the current schema: make an unpartitioned spec the + // default, then drop the original spec's source column from the schema. + let current_schema = fixture.table.metadata().current_schema(); + let evolved_schema = Schema::builder() + .with_fields( + current_schema + .as_struct() + .fields() + .iter() + .filter(|field| field.id != 1) + .cloned(), + ) + .with_identifier_field_ids(vec![2]) + .build() + .unwrap(); + let evolved = + TableMetadataBuilder::new_from_metadata(fixture.table.metadata().clone(), None) + .add_default_partition_spec(UnboundPartitionSpec::builder().build()) + .unwrap() + .add_current_schema(evolved_schema) + .unwrap() + .build() + .unwrap() + .metadata; + + // a commit after the evolution carries the previous manifests forward: the new + // snapshot uses the evolved schema while its manifests still use historical spec 0 + let parent = evolved.current_snapshot().unwrap().clone(); + let snapshot = Snapshot::builder() + .with_snapshot_id(parent.snapshot_id() + 1) + .with_parent_snapshot_id(Some(parent.snapshot_id())) + .with_sequence_number(evolved.last_sequence_number() + 1) + .with_timestamp_ms(evolved.last_updated_ms + 1) + .with_schema_id(evolved.current_schema_id()) + .with_manifest_list(parent.manifest_list()) + .with_summary(Summary { + operation: Operation::Append, + additional_properties: HashMap::new(), + }) + .build(); + let metadata = TableMetadataBuilder::new_from_metadata(evolved, None) + .set_branch_snapshot(snapshot, MAIN_BRANCH) + .unwrap() + .build() + .unwrap() + .metadata; + let table = fixture.table.clone().with_metadata(Arc::new(metadata)); + + // planning and reading must succeed, and the results must match the table before + // evolution: no rows wrongly pruned and none returned unfiltered + let evolved = scan_y_gte_5(&table).await; + assert_eq!(evolved, baseline); + } + + async fn scan_y_gte_5(table: &Table) -> Vec { + let table_scan = table + .scan() + .select(["y"]) + .with_filter(Reference::new("y").greater_than_or_equal_to(Datum::long(5))) + .build() + .unwrap(); + let batches: Vec<_> = table_scan + .to_arrow() + .await + .unwrap() + .try_collect() + .await + .unwrap(); + + let mut values: Vec = batches + .iter() + .flat_map(|batch| { + let col = batch.column_by_name("y").unwrap(); + let arr = col.as_any().downcast_ref::().unwrap(); + (0..arr.len()).map(|i| arr.value(i)).collect::>() + }) + .collect(); + values.sort_unstable(); + values + } + #[tokio::test] async fn test_open_parquet_no_deletions() { let mut fixture = TableTestFixture::new();