You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by at...@apache.org on 2026/07/21 16:53:13 UTC
(iceberg-rust) branch main updated: Skip partition pruning for manifests whose spec references a dropped source column (#2845)
This is an automated email from the ASF dual-hosted git repository.
laskoviymishka pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
The following commit(s) were added to refs/heads/main by this push:
new f019ee147 Skip partition pruning for manifests whose spec references a dropped source column (#2845)
f019ee147 is described below
commit f019ee147c42283580f9533db13a8a62a322666a
Author: Anoop Johnson <an...@apache.org>
AuthorDate: Tue Jul 21 09:53:06 2026 -0700
Skip partition pruning for manifests whose spec references a dropped source column (#2845)
## Which issue does this PR close?
Closes #2844.
## What changes are included in this PR?
When a scan has a predicate, planning resolves every manifest's spec
partition type against the snapshot schema. Manifests written under a
historical spec whose source column was later dropped (which is a legit
possibility in v2+) fail that resolution, so one such manifest fails the
whole scan, even when the predicate only references live columns.
This PR is the stopgap described in #2844. The longer-term fix is
deriving partition types from transforms where possible (as iceberg-java
does in apache/iceberg#17262), which would also restore pruning on a
historical spec's still-live partition fields.
## Are these changes tested?
yes. added tests
---
.../iceberg/src/arrow/record_batch_transformer.rs | 10 +--
crates/iceberg/src/scan/context.rs | 14 ++++
crates/iceberg/src/scan/mod.rs | 98 +++++++++++++++++++++-
3 files changed, 115 insertions(+), 7 deletions(-)
diff --git a/crates/iceberg/src/arrow/record_batch_transformer.rs b/crates/iceberg/src/arrow/record_batch_transformer.rs
index 12f384ff5..76ea4fe28 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 8462eb89f..4bd7e31de 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<Arc<BoundPredicate>> {
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 3fd9dfacf..1bad4500a 100644
--- a/crates/iceberg/src/scan/mod.rs
+++ b/crates/iceberg/src/scan/mod.rs
@@ -651,8 +651,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;
@@ -1842,6 +1844,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<i64> {
+ 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<i64> = batches
+ .iter()
+ .flat_map(|batch| {
+ let col = batch.column_by_name("y").unwrap();
+ let arr = col.as_any().downcast_ref::<Int64Array>().unwrap();
+ (0..arr.len()).map(|i| arr.value(i)).collect::<Vec<_>>()
+ })
+ .collect();
+ values.sort_unstable();
+ values
+ }
+
#[tokio::test]
async fn test_open_parquet_no_deletions() {
let mut fixture = TableTestFixture::new();