From 8f1ba46c6e7535004ad6c5bd4ac3a3c3095e7414 Mon Sep 17 00:00:00 2001 From: Anoop Johnson Date: Fri, 16 Oct 2026 17:50:56 -0700 Subject: [PATCH 1/8] fix(expr): widen promoted partition values before comparison A partition column promoted int->long (or float->double) keeps its originally-written values in manifests produced before the promotion. The entry partition tuple is decoded under the manifest's own narrower type, while the scan predicate is bound to the current wider type. `Datum::partial_cmp`/`PartialEq` require matching types, so the `ExpressionEvaluator` compared e.g. an int partition value against a long predicate literal, got no ordering, and silently pruned the data file, dropping matching rows from filtered scans. The manifest-list `ManifestEvaluator` is unaffected: it decodes `FieldSummary` bounds with `Datum::try_from_bytes`, which already widens a 4-byte int bound to long. Mirror that on the entry path by widening the partition value to the reference's type before comparing. Decimals already compare across precisions, so only int->long and float->double need widening. Co-authored-by: Isaac --- .../src/expr/visitors/expression_evaluator.rs | 138 +++++++++++++----- crates/iceberg/src/spec/values/datum.rs | 13 ++ 2 files changed, 115 insertions(+), 36 deletions(-) diff --git a/crates/iceberg/src/expr/visitors/expression_evaluator.rs b/crates/iceberg/src/expr/visitors/expression_evaluator.rs index d9c2e1b571..e9f24f70aa 100644 --- a/crates/iceberg/src/expr/visitors/expression_evaluator.rs +++ b/crates/iceberg/src/expr/visitors/expression_evaluator.rs @@ -64,6 +64,22 @@ impl<'a> ExpressionEvaluatorVisitor<'a> { fn new(partition: &'a Struct) -> Self { Self { partition } } + + /// Returns the partition value for `reference`, widened to the reference's + /// type so a value written under a narrower type (e.g. an `int` column later + /// promoted to `long`) compares correctly against the bound predicate. + fn partition_value(&self, reference: &BoundReference) -> Result> { + let Some(datum) = reference.accessor().get(self.partition)? else { + return Ok(None); + }; + + Ok(Some( + match reference.field().field_type.as_primitive_type() { + Some(primitive_type) => datum.promote_to(primitive_type), + None => datum, + }, + )) + } } impl BoundPredicateVisitor for ExpressionEvaluatorVisitor<'_> { @@ -130,7 +146,7 @@ impl BoundPredicateVisitor for ExpressionEvaluatorVisitor<'_> { literal: &Datum, _predicate: &BoundPredicate, ) -> Result { - match reference.accessor().get(self.partition)? { + match self.partition_value(reference)? { Some(datum) => Ok(&datum < literal), None => Ok(false), } @@ -142,7 +158,7 @@ impl BoundPredicateVisitor for ExpressionEvaluatorVisitor<'_> { literal: &Datum, _predicate: &BoundPredicate, ) -> Result { - match reference.accessor().get(self.partition)? { + match self.partition_value(reference)? { Some(datum) => Ok(&datum <= literal), None => Ok(false), } @@ -154,7 +170,7 @@ impl BoundPredicateVisitor for ExpressionEvaluatorVisitor<'_> { literal: &Datum, _predicate: &BoundPredicate, ) -> Result { - match reference.accessor().get(self.partition)? { + match self.partition_value(reference)? { Some(datum) => Ok(&datum > literal), None => Ok(false), } @@ -166,7 +182,7 @@ impl BoundPredicateVisitor for ExpressionEvaluatorVisitor<'_> { literal: &Datum, _predicate: &BoundPredicate, ) -> Result { - match reference.accessor().get(self.partition)? { + match self.partition_value(reference)? { Some(datum) => Ok(&datum >= literal), None => Ok(false), } @@ -178,7 +194,7 @@ impl BoundPredicateVisitor for ExpressionEvaluatorVisitor<'_> { literal: &Datum, _predicate: &BoundPredicate, ) -> Result { - match reference.accessor().get(self.partition)? { + match self.partition_value(reference)? { Some(datum) => Ok(&datum == literal), None => Ok(false), } @@ -190,7 +206,7 @@ impl BoundPredicateVisitor for ExpressionEvaluatorVisitor<'_> { literal: &Datum, _predicate: &BoundPredicate, ) -> Result { - match reference.accessor().get(self.partition)? { + match self.partition_value(reference)? { Some(datum) => Ok(&datum != literal), None => Ok(true), } @@ -227,7 +243,7 @@ impl BoundPredicateVisitor for ExpressionEvaluatorVisitor<'_> { literals: &FnvHashSet, _predicate: &BoundPredicate, ) -> Result { - match reference.accessor().get(self.partition)? { + match self.partition_value(reference)? { Some(datum) => Ok(literals.contains(&datum)), None => Ok(false), } @@ -239,7 +255,7 @@ impl BoundPredicateVisitor for ExpressionEvaluatorVisitor<'_> { literals: &FnvHashSet, _predicate: &BoundPredicate, ) -> Result { - match reference.accessor().get(self.partition)? { + match self.partition_value(reference)? { Some(datum) => Ok(!literals.contains(&datum)), None => Ok(true), } @@ -330,9 +346,7 @@ mod tests { Ok(ExpressionEvaluator::new(partition_filter)) } - fn create_data_file_float() -> DataFile { - let partition = Struct::from_iter([Some(Literal::float(1.0_f32))]); - + fn create_data_file(partition: Struct) -> DataFile { DataFile { content: DataContentType::Data, file_path: "/test/path".to_string(), @@ -358,32 +372,12 @@ mod tests { } } - fn create_data_file_string() -> DataFile { - let partition = Struct::from_iter([Some(Literal::string("test str"))]); + fn create_data_file_float() -> DataFile { + create_data_file(Struct::from_iter([Some(Literal::float(1.0_f32))])) + } - DataFile { - content: DataContentType::Data, - file_path: "/test/path".to_string(), - file_format: DataFileFormat::Parquet, - partition, - record_count: 1, - file_size_in_bytes: 1, - column_sizes: HashMap::new(), - value_counts: HashMap::new(), - null_value_counts: HashMap::new(), - nan_value_counts: HashMap::new(), - lower_bounds: HashMap::new(), - upper_bounds: HashMap::new(), - key_metadata: None, - split_offsets: None, - equality_ids: None, - sort_order_id: None, - partition_spec_id: 0, - first_row_id: None, - referenced_data_file: None, - content_offset: None, - content_size_in_bytes: None, - } + fn create_data_file_string() -> DataFile { + create_data_file(Struct::from_iter([Some(Literal::string("test str"))])) } #[test] @@ -885,4 +879,76 @@ mod tests { Ok(()) } + + #[test] + fn eval_promotes_int_partition_to_long_predicate() -> Result<()> { + // A column partitioned as `int` and later promoted to `long` keeps `int` + // partition values in manifests written before the promotion; a predicate + // bound to the promoted `long` type must still match them. + let case_sensitive = true; + let (partition_spec, schema) = create_partition_spec(PrimitiveType::Long)?; + + let predicate = Predicate::Binary(BinaryExpression::new( + PredicateOperator::GreaterThanOrEq, + Reference::new("a"), + Datum::long(5_i64), + )) + .bind(schema.clone(), case_sensitive)?; + + let expression_evaluator = + create_expression_evaluator(partition_spec, &schema, &predicate, case_sensitive)?; + + let data_file = create_data_file(Struct::from_iter([Some(Literal::int(5_i32))])); + + assert!(expression_evaluator.eval(&data_file)?); + + Ok(()) + } + + #[test] + fn eval_promotes_int_partition_in_set_predicate() -> Result<()> { + // The set-membership path matches on hash/equality, so it needs the same + // widening as the ordered comparisons. + let case_sensitive = true; + let (partition_spec, schema) = create_partition_spec(PrimitiveType::Long)?; + + let predicate = Predicate::Set(SetExpression::new( + PredicateOperator::In, + Reference::new("a"), + FnvHashSet::from_iter([Datum::long(5_i64), Datum::long(6_i64)]), + )) + .bind(schema.clone(), case_sensitive)?; + + let expression_evaluator = + create_expression_evaluator(partition_spec, &schema, &predicate, case_sensitive)?; + + let data_file = create_data_file(Struct::from_iter([Some(Literal::int(5_i32))])); + + assert!(expression_evaluator.eval(&data_file)?); + + Ok(()) + } + + #[test] + fn eval_promotes_float_partition_to_double_predicate() -> Result<()> { + // Same widening for a `float` column promoted to `double`. + let case_sensitive = true; + let (partition_spec, schema) = create_partition_spec(PrimitiveType::Double)?; + + let predicate = Predicate::Binary(BinaryExpression::new( + PredicateOperator::LessThan, + Reference::new("a"), + Datum::double(2.0_f64), + )) + .bind(schema.clone(), case_sensitive)?; + + let expression_evaluator = + create_expression_evaluator(partition_spec, &schema, &predicate, case_sensitive)?; + + let data_file = create_data_file(Struct::from_iter([Some(Literal::float(1.0_f32))])); + + assert!(expression_evaluator.eval(&data_file)?); + + Ok(()) + } } diff --git a/crates/iceberg/src/spec/values/datum.rs b/crates/iceberg/src/spec/values/datum.rs index 09c1e88313..1c44e952de 100644 --- a/crates/iceberg/src/spec/values/datum.rs +++ b/crates/iceberg/src/spec/values/datum.rs @@ -1229,6 +1229,19 @@ impl Datum { &self.r#type } + /// Returns a copy whose literal is widened to `target` for the value + /// promotions `int` -> `long` and `float` -> `double`, so a value written + /// under the narrower type compares against a predicate bound to the wider + /// one. Any other `target` returns an unchanged copy; decimals already + /// compare across precisions. + pub(crate) fn promote_to(&self, target: &PrimitiveType) -> Datum { + match (&self.literal, target) { + (PrimitiveLiteral::Int(value), PrimitiveType::Long) => Datum::long(*value), + (PrimitiveLiteral::Float(value), PrimitiveType::Double) => Datum::double(value.0), + _ => self.clone(), + } + } + /// Returns true if the Literal represents a primitive type /// that can be a NaN, and that it's value is NaN pub fn is_nan(&self) -> bool { From 958f3d3aa655bf143981490e4740faa90c2ef7c1 Mon Sep 17 00:00:00 2001 From: Anoop Johnson Date: Fri, 16 Oct 2026 17:50:58 -0700 Subject: [PATCH 2/8] fix(expr): widen promoted column bounds in metrics pruning Companion to the partition-tuple widening: InclusiveMetricsEvaluator compares a data file's column bounds against the scan predicate, but those bounds are decoded under the manifest's embedded (narrower) schema while the predicate is bound to the current (wider) type. For an int column promoted to long (or float to double), the inequality and IN checks compared mismatched types, got no ordering, and discarded files that could match, dropping rows; the eq check failed open and merely skipped valid pruning. Widen each stored bound to the reference's type before comparing, via the same `Datum::promote_to` helper, in the inequality, eq, and IN paths. --- .../visitors/inclusive_metrics_evaluator.rs | 107 +++++++++++++++++- 1 file changed, 104 insertions(+), 3 deletions(-) diff --git a/crates/iceberg/src/expr/visitors/inclusive_metrics_evaluator.rs b/crates/iceberg/src/expr/visitors/inclusive_metrics_evaluator.rs index 8ccd1364d9..7077de2cda 100644 --- a/crates/iceberg/src/expr/visitors/inclusive_metrics_evaluator.rs +++ b/crates/iceberg/src/expr/visitors/inclusive_metrics_evaluator.rs @@ -120,7 +120,8 @@ impl<'a> InclusiveMetricsEvaluator<'a> { }; if let Some(bound) = bound { - if cmp_fn(bound, datum) { + let bound = Self::promote_bound(bound, reference); + if cmp_fn(&bound, datum) { return ROWS_MIGHT_MATCH; } @@ -129,6 +130,16 @@ impl<'a> InclusiveMetricsEvaluator<'a> { ROWS_MIGHT_MATCH } + + /// Widens a stored bound to the reference's type so a bound written under a + /// narrower type (e.g. an `int` column later promoted to `long`) compares + /// against a predicate bound to the wider type. + fn promote_bound(bound: &Datum, reference: &BoundReference) -> Datum { + match reference.field().field_type.as_primitive_type() { + Some(primitive_type) => bound.promote_to(primitive_type), + None => bound.clone(), + } + } } impl BoundPredicateVisitor for InclusiveMetricsEvaluator<'_> { @@ -259,6 +270,7 @@ impl BoundPredicateVisitor for InclusiveMetricsEvaluator<'_> { } if let Some(lower_bound) = self.lower_bound(field_id) { + let lower_bound = Self::promote_bound(lower_bound, reference); if lower_bound.is_nan() { // NaN indicates unreliable bounds. // See the InclusiveMetricsEvaluator docs for more. @@ -269,6 +281,7 @@ impl BoundPredicateVisitor for InclusiveMetricsEvaluator<'_> { } if let Some(upper_bound) = self.upper_bound(field_id) { + let upper_bound = Self::promote_bound(upper_bound, reference); if upper_bound.is_nan() { // NaN indicates unreliable bounds. // See the InclusiveMetricsEvaluator docs for more. @@ -438,24 +451,26 @@ impl BoundPredicateVisitor for InclusiveMetricsEvaluator<'_> { } if let Some(lower_bound) = self.lower_bound(field_id) { + let lower_bound = Self::promote_bound(lower_bound, reference); if lower_bound.is_nan() { // NaN indicates unreliable bounds. See the InclusiveMetricsEvaluator docs for more. return ROWS_MIGHT_MATCH; } - if !literals.iter().any(|datum| datum.ge(lower_bound)) { + if !literals.iter().any(|datum| datum.ge(&lower_bound)) { // if all values are less than lower bound, rows cannot match. return ROWS_CANNOT_MATCH; } } if let Some(upper_bound) = self.upper_bound(field_id) { + let upper_bound = Self::promote_bound(upper_bound, reference); if upper_bound.is_nan() { // NaN indicates unreliable bounds. See the InclusiveMetricsEvaluator docs for more. return ROWS_MIGHT_MATCH; } - if !literals.iter().any(|datum| datum.le(upper_bound)) { + if !literals.iter().any(|datum| datum.le(&upper_bound)) { // if all values are greater than upper bound, rows cannot match. return ROWS_CANNOT_MATCH; } @@ -521,6 +536,92 @@ mod test { assert!(result); } + fn promotion_schema(primitive: PrimitiveType) -> SchemaRef { + Arc::new( + Schema::builder() + .with_fields(vec![Arc::new(NestedField::optional( + 1, + "a", + Type::Primitive(primitive), + ))]) + .build() + .unwrap(), + ) + } + + fn data_file_with_bounds(lower: Datum, upper: Datum) -> DataFile { + DataFile { + lower_bounds: HashMap::from([(1, lower)]), + upper_bounds: HashMap::from([(1, upper)]), + ..create_test_data_file() + } + } + + #[test] + fn eval_promotes_int_bounds_for_inequality() { + // A column promoted int->long keeps int-typed column bounds in + // pre-promotion manifests; a predicate bound to the promoted long type + // must still match a file whose range covers it, not silently prune it. + let schema = promotion_schema(PrimitiveType::Long); + let data_file = data_file_with_bounds(Datum::int(5), Datum::int(10)); + let predicate = Predicate::Binary(BinaryExpression::new( + GreaterThan, + Reference::new("a"), + Datum::long(4), + )) + .bind(schema, true) + .unwrap(); + + assert!(InclusiveMetricsEvaluator::eval(&predicate, &data_file, true).unwrap()); + } + + #[test] + fn eval_promotes_int_bounds_for_in() { + let schema = promotion_schema(PrimitiveType::Long); + let data_file = data_file_with_bounds(Datum::int(5), Datum::int(10)); + let predicate = Predicate::Set(SetExpression::new( + In, + Reference::new("a"), + FnvHashSet::from_iter([Datum::long(5), Datum::long(6)]), + )) + .bind(schema, true) + .unwrap(); + + assert!(InclusiveMetricsEvaluator::eval(&predicate, &data_file, true).unwrap()); + } + + #[test] + fn eval_promotes_int_bounds_for_eq() { + // With the bounds widened, `== 5` correctly prunes a file whose range is + // entirely above 5 (int bounds [6, 10]) instead of failing open. + let schema = promotion_schema(PrimitiveType::Long); + let data_file = data_file_with_bounds(Datum::int(6), Datum::int(10)); + let predicate = Predicate::Binary(BinaryExpression::new( + Eq, + Reference::new("a"), + Datum::long(5), + )) + .bind(schema, true) + .unwrap(); + + assert!(!InclusiveMetricsEvaluator::eval(&predicate, &data_file, true).unwrap()); + } + + #[test] + fn eval_promotes_float_bounds_for_inequality() { + let schema = promotion_schema(PrimitiveType::Double); + let data_file = data_file_with_bounds(Datum::float(1.5_f32), Datum::float(2.0_f32)); + let predicate = Predicate::Binary(BinaryExpression::new( + GreaterThan, + Reference::new("a"), + Datum::double(1.0), + )) + .bind(schema, true) + .unwrap(); + + assert!(InclusiveMetricsEvaluator::eval(&predicate, &data_file, true).unwrap()); + } + #[test] fn test_all_nulls() { let result = From 642d8dd66a7b75e0ba3f61bd74863e340e91d93d Mon Sep 17 00:00:00 2001 From: Anoop Johnson Date: Mon, 19 Oct 2026 08:57:07 -0700 Subject: [PATCH 3/8] fix(scan): widen promoted partition tuples at manifest read Replaces the comparison-time partition promotion with a reader-side projection, per review. The earlier comparison-time promotion only reached the pruning evaluators, so a promoted partition column still mismatched in the delete index: a data file's int partition never matched an eq delete or DV keyed by long, dropping or mis-applying deletes. Projecting the partition tuple fixes pruning and delete matching in one place. --- crates/iceberg/src/delete_file_index.rs | 75 +++++++++- .../src/expr/visitors/expression_evaluator.rs | 138 +++++------------- crates/iceberg/src/scan/context.rs | 41 +++++- crates/iceberg/src/scan/mod.rs | 49 +++++++ crates/iceberg/src/spec/manifest/data_file.rs | 117 ++++++++++++++- crates/iceberg/src/spec/values/datum.rs | 7 +- crates/iceberg/src/spec/values/primitive.rs | 26 ++++ crates/iceberg/src/test_utils/scan.rs | 111 ++++++++++++++ 8 files changed, 453 insertions(+), 111 deletions(-) diff --git a/crates/iceberg/src/delete_file_index.rs b/crates/iceberg/src/delete_file_index.rs index 01b4773f62..fb18d00f4c 100644 --- a/crates/iceberg/src/delete_file_index.rs +++ b/crates/iceberg/src/delete_file_index.rs @@ -386,9 +386,82 @@ mod tests { use crate::ErrorKind; use crate::spec::{ DataContentType, DataFileBuilder, DataFileFormat, Datum, Literal, ManifestEntry, - ManifestStatus, Struct, + ManifestStatus, NestedField, PrimitiveType, Struct, StructType, Type, }; + #[test] + fn promoted_data_file_partition_matches_delete_after_column_promotion() { + // The partition column was promoted int -> long. The eq delete, written + // after the promotion, carries a `long` partition value; the data file, + // written before it, carries an `int` one. They must match once the data + // file is widened to the current schema. + let spec_id = 1; + let eq_delete = + build_partitioned_eq_delete(&Struct::from_iter([Some(Literal::long(7))]), spec_id); + let index = PopulatedDeleteFileIndex::new(vec![DeleteFileContext { + manifest_entry: build_added_manifest_entry(4, &eq_delete).into(), + partition_spec_id: spec_id, + }]) + .unwrap(); + + let data_file = + build_partitioned_data_file(&Struct::from_iter([Some(Literal::int(7))]), spec_id); + + // The int partition does not match the long-keyed delete. + assert!( + index + .get_deletes_for_data_file(&data_file, Some(0)) + .unwrap() + .is_empty() + ); + + // Widening the data file's partition to the current (long) type makes it match. + let partition_type = StructType::new(vec![ + NestedField::required(1000, "b", Type::Primitive(PrimitiveType::Long)).into(), + ]); + let mut promoted = data_file.clone(); + promoted.partition = data_file.promoted_partition(&partition_type); + + assert_eq!( + index + .get_deletes_for_data_file(&promoted, Some(0)) + .unwrap() + .len(), + 1 + ); + } + + #[test] + fn promoted_data_file_partition_matches_delete_after_source_dropped() { + // The promoted partition source column was then dropped, so the spec + // resolves it to `Unknown`. Ceiling-widening still normalizes the data + // file's int partition to long so it matches the delete's long partition. + let spec_id = 1; + let eq_delete = + build_partitioned_eq_delete(&Struct::from_iter([Some(Literal::long(7))]), spec_id); + let index = PopulatedDeleteFileIndex::new(vec![DeleteFileContext { + manifest_entry: build_added_manifest_entry(4, &eq_delete).into(), + partition_spec_id: spec_id, + }]) + .unwrap(); + + let data_file = + build_partitioned_data_file(&Struct::from_iter([Some(Literal::int(7))]), spec_id); + let partition_type = StructType::new(vec![ + NestedField::required(1000, "b", Type::Primitive(PrimitiveType::Unknown)).into(), + ]); + let mut promoted = data_file.clone(); + promoted.partition = data_file.promoted_partition(&partition_type); + + assert_eq!( + index + .get_deletes_for_data_file(&promoted, Some(0)) + .unwrap() + .len(), + 1 + ); + } + #[test] fn test_delete_file_index_unpartitioned() { let deletes: Vec = vec![ diff --git a/crates/iceberg/src/expr/visitors/expression_evaluator.rs b/crates/iceberg/src/expr/visitors/expression_evaluator.rs index e9f24f70aa..d9c2e1b571 100644 --- a/crates/iceberg/src/expr/visitors/expression_evaluator.rs +++ b/crates/iceberg/src/expr/visitors/expression_evaluator.rs @@ -64,22 +64,6 @@ impl<'a> ExpressionEvaluatorVisitor<'a> { fn new(partition: &'a Struct) -> Self { Self { partition } } - - /// Returns the partition value for `reference`, widened to the reference's - /// type so a value written under a narrower type (e.g. an `int` column later - /// promoted to `long`) compares correctly against the bound predicate. - fn partition_value(&self, reference: &BoundReference) -> Result> { - let Some(datum) = reference.accessor().get(self.partition)? else { - return Ok(None); - }; - - Ok(Some( - match reference.field().field_type.as_primitive_type() { - Some(primitive_type) => datum.promote_to(primitive_type), - None => datum, - }, - )) - } } impl BoundPredicateVisitor for ExpressionEvaluatorVisitor<'_> { @@ -146,7 +130,7 @@ impl BoundPredicateVisitor for ExpressionEvaluatorVisitor<'_> { literal: &Datum, _predicate: &BoundPredicate, ) -> Result { - match self.partition_value(reference)? { + match reference.accessor().get(self.partition)? { Some(datum) => Ok(&datum < literal), None => Ok(false), } @@ -158,7 +142,7 @@ impl BoundPredicateVisitor for ExpressionEvaluatorVisitor<'_> { literal: &Datum, _predicate: &BoundPredicate, ) -> Result { - match self.partition_value(reference)? { + match reference.accessor().get(self.partition)? { Some(datum) => Ok(&datum <= literal), None => Ok(false), } @@ -170,7 +154,7 @@ impl BoundPredicateVisitor for ExpressionEvaluatorVisitor<'_> { literal: &Datum, _predicate: &BoundPredicate, ) -> Result { - match self.partition_value(reference)? { + match reference.accessor().get(self.partition)? { Some(datum) => Ok(&datum > literal), None => Ok(false), } @@ -182,7 +166,7 @@ impl BoundPredicateVisitor for ExpressionEvaluatorVisitor<'_> { literal: &Datum, _predicate: &BoundPredicate, ) -> Result { - match self.partition_value(reference)? { + match reference.accessor().get(self.partition)? { Some(datum) => Ok(&datum >= literal), None => Ok(false), } @@ -194,7 +178,7 @@ impl BoundPredicateVisitor for ExpressionEvaluatorVisitor<'_> { literal: &Datum, _predicate: &BoundPredicate, ) -> Result { - match self.partition_value(reference)? { + match reference.accessor().get(self.partition)? { Some(datum) => Ok(&datum == literal), None => Ok(false), } @@ -206,7 +190,7 @@ impl BoundPredicateVisitor for ExpressionEvaluatorVisitor<'_> { literal: &Datum, _predicate: &BoundPredicate, ) -> Result { - match self.partition_value(reference)? { + match reference.accessor().get(self.partition)? { Some(datum) => Ok(&datum != literal), None => Ok(true), } @@ -243,7 +227,7 @@ impl BoundPredicateVisitor for ExpressionEvaluatorVisitor<'_> { literals: &FnvHashSet, _predicate: &BoundPredicate, ) -> Result { - match self.partition_value(reference)? { + match reference.accessor().get(self.partition)? { Some(datum) => Ok(literals.contains(&datum)), None => Ok(false), } @@ -255,7 +239,7 @@ impl BoundPredicateVisitor for ExpressionEvaluatorVisitor<'_> { literals: &FnvHashSet, _predicate: &BoundPredicate, ) -> Result { - match self.partition_value(reference)? { + match reference.accessor().get(self.partition)? { Some(datum) => Ok(!literals.contains(&datum)), None => Ok(true), } @@ -346,7 +330,9 @@ mod tests { Ok(ExpressionEvaluator::new(partition_filter)) } - fn create_data_file(partition: Struct) -> DataFile { + fn create_data_file_float() -> DataFile { + let partition = Struct::from_iter([Some(Literal::float(1.0_f32))]); + DataFile { content: DataContentType::Data, file_path: "/test/path".to_string(), @@ -372,12 +358,32 @@ mod tests { } } - fn create_data_file_float() -> DataFile { - create_data_file(Struct::from_iter([Some(Literal::float(1.0_f32))])) - } - fn create_data_file_string() -> DataFile { - create_data_file(Struct::from_iter([Some(Literal::string("test str"))])) + let partition = Struct::from_iter([Some(Literal::string("test str"))]); + + DataFile { + content: DataContentType::Data, + file_path: "/test/path".to_string(), + file_format: DataFileFormat::Parquet, + partition, + record_count: 1, + file_size_in_bytes: 1, + column_sizes: HashMap::new(), + value_counts: HashMap::new(), + null_value_counts: HashMap::new(), + nan_value_counts: HashMap::new(), + lower_bounds: HashMap::new(), + upper_bounds: HashMap::new(), + key_metadata: None, + split_offsets: None, + equality_ids: None, + sort_order_id: None, + partition_spec_id: 0, + first_row_id: None, + referenced_data_file: None, + content_offset: None, + content_size_in_bytes: None, + } } #[test] @@ -879,76 +885,4 @@ mod tests { Ok(()) } - - #[test] - fn eval_promotes_int_partition_to_long_predicate() -> Result<()> { - // A column partitioned as `int` and later promoted to `long` keeps `int` - // partition values in manifests written before the promotion; a predicate - // bound to the promoted `long` type must still match them. - let case_sensitive = true; - let (partition_spec, schema) = create_partition_spec(PrimitiveType::Long)?; - - let predicate = Predicate::Binary(BinaryExpression::new( - PredicateOperator::GreaterThanOrEq, - Reference::new("a"), - Datum::long(5_i64), - )) - .bind(schema.clone(), case_sensitive)?; - - let expression_evaluator = - create_expression_evaluator(partition_spec, &schema, &predicate, case_sensitive)?; - - let data_file = create_data_file(Struct::from_iter([Some(Literal::int(5_i32))])); - - assert!(expression_evaluator.eval(&data_file)?); - - Ok(()) - } - - #[test] - fn eval_promotes_int_partition_in_set_predicate() -> Result<()> { - // The set-membership path matches on hash/equality, so it needs the same - // widening as the ordered comparisons. - let case_sensitive = true; - let (partition_spec, schema) = create_partition_spec(PrimitiveType::Long)?; - - let predicate = Predicate::Set(SetExpression::new( - PredicateOperator::In, - Reference::new("a"), - FnvHashSet::from_iter([Datum::long(5_i64), Datum::long(6_i64)]), - )) - .bind(schema.clone(), case_sensitive)?; - - let expression_evaluator = - create_expression_evaluator(partition_spec, &schema, &predicate, case_sensitive)?; - - let data_file = create_data_file(Struct::from_iter([Some(Literal::int(5_i32))])); - - assert!(expression_evaluator.eval(&data_file)?); - - Ok(()) - } - - #[test] - fn eval_promotes_float_partition_to_double_predicate() -> Result<()> { - // Same widening for a `float` column promoted to `double`. - let case_sensitive = true; - let (partition_spec, schema) = create_partition_spec(PrimitiveType::Double)?; - - let predicate = Predicate::Binary(BinaryExpression::new( - PredicateOperator::LessThan, - Reference::new("a"), - Datum::double(2.0_f64), - )) - .bind(schema.clone(), case_sensitive)?; - - let expression_evaluator = - create_expression_evaluator(partition_spec, &schema, &predicate, case_sensitive)?; - - let data_file = create_data_file(Struct::from_iter([Some(Literal::float(1.0_f32))])); - - assert!(expression_evaluator.eval(&data_file)?); - - Ok(()) - } } diff --git a/crates/iceberg/src/scan/context.rs b/crates/iceberg/src/scan/context.rs index 211ebf8f0c..63812a537c 100644 --- a/crates/iceberg/src/scan/context.rs +++ b/crates/iceberg/src/scan/context.rs @@ -73,6 +73,19 @@ pub(crate) struct ManifestEntryContext { pub sort_order: Option, } +/// Returns whether `current` widens any field of `written` at the same position +/// (differing primitive types), i.e. partition values written with `written` need +/// widening to `current` before they compare against predicates bound to the current +/// schema. +fn partition_types_differ(current: &StructType, written: &StructType) -> bool { + current.fields().len() != written.fields().len() + || current + .fields() + .iter() + .zip(written.fields()) + .any(|(c, w)| c.field_type.as_primitive_type() != w.field_type.as_primitive_type()) +} + impl ManifestFileContext { /// Consumes this [`ManifestFileContext`], fetching its Manifest from FileIO and then /// streaming its constituent [`ManifestEntries`] to the channel provided in the context @@ -95,7 +108,32 @@ impl ManifestFileContext { let manifest = object_cache.get_manifest(&manifest_file).await?; + // A manifest written under an older schema stores partition values in that + // schema's (narrower) types. When the current partition type widens the one the + // manifest was written with (e.g. an int source column promoted to long), widen + // each entry's partition tuple here so the pruning evaluator and the delete-file + // index compare like-typed values. The gate compares the partition types + // themselves, not schema ids, since v1 may omit them (defaulting to 0). Column + // bounds are widened lazily where the metrics evaluator compares them. + let promotion_target: Option = match partition_spec.as_ref() { + Some(spec) => { + let current = spec.partition_type(&snapshot_schema)?; + let written = spec.partition_type(manifest.metadata().schema())?; + partition_types_differ(¤t, &written).then_some(current) + } + None => None, + }; + for manifest_entry in manifest.entries() { + let manifest_entry = match promotion_target.as_ref() { + Some(partition_type) => { + let mut entry = manifest_entry.as_ref().clone(); + entry.data_file.partition = entry.data_file.promoted_partition(partition_type); + Arc::new(entry) + } + None => manifest_entry.clone(), + }; + // Carry the raw id through unresolved, then resolve it to an order for the // reader. The resolved order is `None` for the unsorted order (empty fields) so // that `Some` always means genuinely sorted, matching Java's `isSorted()` gate; @@ -108,8 +146,7 @@ impl ManifestFileContext { .cloned(); let manifest_entry_context = ManifestEntryContext { - // TODO: refactor to avoid the expensive ManifestEntry clone - manifest_entry: manifest_entry.clone(), + manifest_entry, expression_evaluator_cache: expression_evaluator_cache.clone(), field_ids: field_ids.clone(), partition_spec_id: manifest_file.partition_spec_id, diff --git a/crates/iceberg/src/scan/mod.rs b/crates/iceberg/src/scan/mod.rs index 6d23e94704..60027c9e9e 100644 --- a/crates/iceberg/src/scan/mod.rs +++ b/crates/iceberg/src/scan/mod.rs @@ -1195,6 +1195,55 @@ mod tests { ); } + #[tokio::test] + async fn test_scan_widens_old_partition_so_pruning_and_deletes_match() { + // Guards the reader-side normalization through a real scan: a data manifest + // written when the partition source `x` was `int` must have its `int(7)` + // partition widened to `long(7)` during planning, so (a) a `x = 7` (long) + // filter keeps the file and (b) the equality delete keyed by `long(7)` + // attaches. Without the normalization the int/long mismatch drops the delete + // (and prunes the file under the filter). + let mut fixture = TableTestFixture::new(); + fixture.setup_promoted_partition_with_eq_delete().await; + + let tasks: Vec = fixture + .table + .scan() + .build() + .unwrap() + .plan_files() + .await + .unwrap() + .try_collect() + .await + .unwrap(); + assert_eq!(tasks.len(), 1); + assert_eq!( + tasks[0].deletes().len(), + 1, + "equality delete must attach after widening the old int partition to long" + ); + + let tasks: Vec = fixture + .table + .scan() + .with_filter(Reference::new("x").equal_to(Datum::long(7))) + .build() + .unwrap() + .plan_files() + .await + .unwrap() + .try_collect() + .await + .unwrap(); + assert_eq!( + tasks.len(), + 1, + "filter on the promoted type must not prune the file" + ); + assert_eq!(tasks[0].deletes().len(), 1); + } + #[tokio::test] async fn test_plan_files_carries_row_lineage_into_file_scan_task() { let mut fixture = TableTestFixture::new(); diff --git a/crates/iceberg/src/spec/manifest/data_file.rs b/crates/iceberg/src/spec/manifest/data_file.rs index 94568ab44e..5362da3493 100644 --- a/crates/iceberg/src/spec/manifest/data_file.rs +++ b/crates/iceberg/src/spec/manifest/data_file.rs @@ -29,7 +29,9 @@ use super::{ }; use crate::Error; use crate::error::{Result, invalid_data}; -use crate::spec::{DEFAULT_PARTITION_SPEC_ID, Struct, StructType, Type}; +use crate::spec::{ + DEFAULT_PARTITION_SPEC_ID, Literal, NestedFieldRef, PrimitiveType, Struct, StructType, Type, +}; /// Data file carries data file path, partition tuple, metrics, … #[derive(Debug, PartialEq, Clone, Eq, Builder)] @@ -184,7 +186,36 @@ pub struct DataFile { pub(crate) content_size_in_bytes: Option, } +fn promote_partition_value(value: Option<&Literal>, field: &NestedFieldRef) -> Option { + match value { + Some(Literal::Primitive(primitive)) => { + let promoted = match field.field_type.as_primitive_type() { + // The partition source column was dropped, so there is no target type to + // widen to; widen to the promotion ceiling instead so this value and its + // deletes still normalize to the same type and the delete index matches. + Some(PrimitiveType::Unknown) => primitive.widen(), + Some(target) => primitive.promote_to(target), + None => None, + }; + promoted.map(Literal::Primitive).or_else(|| value.cloned()) + } + other => other.cloned(), + } +} + impl DataFile { + /// Returns this file's partition tuple widened to `partition_type` for fields + /// promoted (`int` -> `long`, `float` -> `double`) since the file's manifest + /// was written. Values whose type is unchanged are copied through. Column + /// bounds are widened separately, lazily, where they are compared. + pub(crate) fn promoted_partition(&self, partition_type: &StructType) -> Struct { + self.partition + .iter() + .zip(partition_type.fields()) + .map(|(value, field)| promote_partition_value(value, field)) + .collect() + } + /// Get the content type of the data file (data, equality deletes, or position deletes) pub fn content_type(&self) -> DataContentType { self.content @@ -417,7 +448,11 @@ impl std::fmt::Display for DataFileFormat { #[cfg(test)] mod test { - use crate::spec::DataContentType; + use crate::spec::{ + DataContentType, DataFileBuilder, DataFileFormat, Literal, NestedField, PrimitiveType, + Struct, StructType, Type, + }; + #[test] fn test_data_content_type_default() { assert_eq!(DataContentType::default(), DataContentType::Data); @@ -427,4 +462,82 @@ mod test { fn test_data_content_type_default_value() { assert_eq!(DataContentType::default() as i32, 0); } + + #[test] + fn promoted_partition_widens_int_and_float() { + // Partition columns promoted int -> long and float -> double; a file + // written before the promotion still carries the narrower types. + let partition_type = StructType::new(vec![ + NestedField::required(1000, "b", Type::Primitive(PrimitiveType::Long)).into(), + NestedField::required(1001, "c", Type::Primitive(PrimitiveType::Double)).into(), + ]); + + let data_file = DataFileBuilder::default() + .content(DataContentType::Data) + .file_path("x.parquet".to_string()) + .file_format(DataFileFormat::Parquet) + .partition(Struct::from_iter([ + Some(Literal::int(7)), + Some(Literal::float(1.5_f32)), + ])) + .record_count(1) + .file_size_in_bytes(1) + .partition_spec_id(0) + .build() + .unwrap(); + + assert_eq!( + data_file.promoted_partition(&partition_type), + Struct::from_iter([Some(Literal::long(7)), Some(Literal::double(1.5))]) + ); + } + + #[test] + fn promoted_partition_widens_dropped_source_to_ceiling() { + // A dropped partition source resolves to `Unknown`, so there is no target + // type; the value still widens to its ceiling so it keeps matching deletes + // written while the source still existed. + let partition_type = StructType::new(vec![ + NestedField::required(1000, "b", Type::Primitive(PrimitiveType::Unknown)).into(), + ]); + + let data_file = DataFileBuilder::default() + .content(DataContentType::Data) + .file_path("x.parquet".to_string()) + .file_format(DataFileFormat::Parquet) + .partition(Struct::from_iter([Some(Literal::int(7))])) + .record_count(1) + .file_size_in_bytes(1) + .partition_spec_id(0) + .build() + .unwrap(); + + assert_eq!( + data_file.promoted_partition(&partition_type), + Struct::from_iter([Some(Literal::long(7))]) + ); + } + + #[test] + fn promoted_partition_leaves_matching_types_unchanged() { + let partition_type = StructType::new(vec![ + NestedField::required(1000, "b", Type::Primitive(PrimitiveType::Long)).into(), + ]); + + let data_file = DataFileBuilder::default() + .content(DataContentType::Data) + .file_path("x.parquet".to_string()) + .file_format(DataFileFormat::Parquet) + .partition(Struct::from_iter([Some(Literal::long(7))])) + .record_count(1) + .file_size_in_bytes(1) + .partition_spec_id(0) + .build() + .unwrap(); + + assert_eq!( + data_file.promoted_partition(&partition_type), + data_file.partition + ); + } } diff --git a/crates/iceberg/src/spec/values/datum.rs b/crates/iceberg/src/spec/values/datum.rs index 1c44e952de..2a6c9a2907 100644 --- a/crates/iceberg/src/spec/values/datum.rs +++ b/crates/iceberg/src/spec/values/datum.rs @@ -1235,10 +1235,9 @@ impl Datum { /// one. Any other `target` returns an unchanged copy; decimals already /// compare across precisions. pub(crate) fn promote_to(&self, target: &PrimitiveType) -> Datum { - match (&self.literal, target) { - (PrimitiveLiteral::Int(value), PrimitiveType::Long) => Datum::long(*value), - (PrimitiveLiteral::Float(value), PrimitiveType::Double) => Datum::double(value.0), - _ => self.clone(), + match self.literal.promote_to(target) { + Some(literal) => Datum::new(target.clone(), literal), + None => self.clone(), } } diff --git a/crates/iceberg/src/spec/values/primitive.rs b/crates/iceberg/src/spec/values/primitive.rs index b76978bd5f..2204e22427 100644 --- a/crates/iceberg/src/spec/values/primitive.rs +++ b/crates/iceberg/src/spec/values/primitive.rs @@ -21,6 +21,8 @@ use std::cmp::Ordering; use ordered_float::{FloatCore, OrderedFloat}; +use crate::spec::PrimitiveType; + /// Values present in iceberg type /// /// `Float` and `Double` compare as Java's `Float.compare` and `Double.compare` do: `-0.0` is @@ -105,6 +107,30 @@ fn float_cmp(a: &OrderedFloat, b: &OrderedFloat) -> Ordering } impl PrimitiveLiteral { + /// Widens this literal to its promotion ceiling (`int` -> `long`, `float` -> + /// `double`), returning `None` for any other literal. Used when the target type + /// is unknown (the partition source column was dropped) so a value and its + /// deletes still normalize to the same type. + pub(crate) fn widen(&self) -> Option { + match self { + PrimitiveLiteral::Int(value) => Some(PrimitiveLiteral::Long(i64::from(*value))), + PrimitiveLiteral::Float(value) => { + Some(PrimitiveLiteral::Double(OrderedFloat(f64::from(value.0)))) + } + _ => None, + } + } + + /// Widens this literal to `target` for the value promotions `int` -> `long` + /// and `float` -> `double`, returning `None` when `target` does not promote it. + pub(crate) fn promote_to(&self, target: &PrimitiveType) -> Option { + match (self, target) { + (PrimitiveLiteral::Int(_), PrimitiveType::Long) + | (PrimitiveLiteral::Float(_), PrimitiveType::Double) => self.widen(), + _ => None, + } + } + /// Must follow the declaration order of the variants: `partial_cmp` uses it to order /// different variants the way the derived `PartialOrd` did. fn variant_index(&self) -> u8 { diff --git a/crates/iceberg/src/test_utils/scan.rs b/crates/iceberg/src/test_utils/scan.rs index 63f6418efd..009c0bf366 100644 --- a/crates/iceberg/src/test_utils/scan.rs +++ b/crates/iceberg/src/test_utils/scan.rs @@ -604,6 +604,117 @@ impl TableTestFixture { manifest_list_write.close().await.unwrap(); } + /// Writes a data manifest under an older schema where the identity partition + /// source `x` (id 1) is still `int` (so the partition value is stored as `int`), + /// plus an equality-delete manifest under the current schema where `x` is `long`. + /// Exercises the scan's reader-side widening: the old `int` partition must widen + /// to `long` so the delete, keyed by the current `long` partition, still matches + /// the data file. The data/delete parquet files themselves are not written, which + /// is enough for `plan_files` (it reads only manifests). + pub(crate) async fn setup_promoted_partition_with_eq_delete(&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(); + + // The schema the data manifest was written under: the current schema with the + // partition source `x` still `int`. + let mut fields = current_schema.as_struct().fields().to_vec(); + fields[0] = NestedField::required(1, "x", PrimitiveType::Int.into()).into(); + let written_schema = Arc::new( + Schema::builder() + .with_schema_id(0) + .with_fields(fields) + .build() + .unwrap(), + ); + + // Data manifest under the old (int) schema; an Existing entry from the parent + // snapshot so its data sequence number is below the equality delete's. + let mut data_writer = ManifestWriterBuilder::new( + self.next_manifest_file(), + Some(current_snapshot.snapshot_id()), + written_schema, + current_partition_spec.as_ref().clone(), + ) + .build_v2_data(); + data_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(0) + .content(DataContentType::Data) + .file_path(format!("{}/x7.parquet", &self.table_location)) + .file_format(DataFileFormat::Parquet) + .file_size_in_bytes(100) + .record_count(1) + .partition(Struct::from_iter([Some(Literal::int(7))])) + .build() + .unwrap(), + ) + .build(), + ) + .unwrap(); + let data_manifest = data_writer.write_manifest_file().await.unwrap(); + + // Equality-delete manifest under the current (long) schema; Added in the + // current snapshot so its sequence number is above the data file's. + let mut delete_writer = ManifestWriterBuilder::new( + self.next_manifest_file(), + Some(current_snapshot.snapshot_id()), + current_schema.clone(), + current_partition_spec.as_ref().clone(), + ) + .build_v2_deletes(); + delete_writer + .add_entry( + ManifestEntry::builder() + .status(ManifestStatus::Added) + .data_file( + DataFileBuilder::default() + .partition_spec_id(0) + .content(DataContentType::EqualityDeletes) + .equality_ids(Some(vec![2])) + .file_path(format!("{}/eq-delete.parquet", &self.table_location)) + .file_format(DataFileFormat::Parquet) + .file_size_in_bytes(100) + .record_count(1) + .partition(Struct::from_iter([Some(Literal::long(7))])) + .build() + .unwrap(), + ) + .build(), + ) + .unwrap(); + let delete_manifest = delete_writer.write_manifest_file().await.unwrap(); + + 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_manifest, delete_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 { From 91df4febe03230756ab856d5620caa0fa3d217f4 Mon Sep 17 00:00:00 2001 From: Anoop Johnson Date: Mon, 19 Oct 2026 13:14:35 -0700 Subject: [PATCH 4/8] fix(scan): tighten promotion gate, borrow unpromoted bounds, guard tuple length Addresses review feedback: - Gate `partition_types_differ` on the type pairs `promoted_partition` actually widens (int->long/unknown, float->double/unknown) instead of any primitive difference, so a decimal-precision widening or a dropped non-numeric source no longer deep-clones every manifest entry on every scan. - `InclusiveMetricsEvaluator::promote_bound` returns `Cow` and `Datum::promote_to` returns `Option`, so a bound that does not widen (e.g. a string or binary) is borrowed instead of cloned per data file during planning. - `DataFile::promoted_partition` returns `DataInvalid` when the stored tuple and the spec partition type have different lengths, instead of silently dropping trailing values (and the partition-scoped deletes they would match). - Drop the two delete-index unit tests that only re-exercised `promoted_partition` plus `Struct` equality, already covered by the data_file and scan-level tests. --- crates/iceberg/src/delete_file_index.rs | 75 +------------------ .../visitors/inclusive_metrics_evaluator.rs | 33 ++++---- crates/iceberg/src/scan/context.rs | 31 ++++++-- crates/iceberg/src/spec/manifest/data_file.rs | 25 +++++-- crates/iceberg/src/spec/values/datum.rs | 13 ++-- 5 files changed, 68 insertions(+), 109 deletions(-) diff --git a/crates/iceberg/src/delete_file_index.rs b/crates/iceberg/src/delete_file_index.rs index fb18d00f4c..01b4773f62 100644 --- a/crates/iceberg/src/delete_file_index.rs +++ b/crates/iceberg/src/delete_file_index.rs @@ -386,82 +386,9 @@ mod tests { use crate::ErrorKind; use crate::spec::{ DataContentType, DataFileBuilder, DataFileFormat, Datum, Literal, ManifestEntry, - ManifestStatus, NestedField, PrimitiveType, Struct, StructType, Type, + ManifestStatus, Struct, }; - #[test] - fn promoted_data_file_partition_matches_delete_after_column_promotion() { - // The partition column was promoted int -> long. The eq delete, written - // after the promotion, carries a `long` partition value; the data file, - // written before it, carries an `int` one. They must match once the data - // file is widened to the current schema. - let spec_id = 1; - let eq_delete = - build_partitioned_eq_delete(&Struct::from_iter([Some(Literal::long(7))]), spec_id); - let index = PopulatedDeleteFileIndex::new(vec![DeleteFileContext { - manifest_entry: build_added_manifest_entry(4, &eq_delete).into(), - partition_spec_id: spec_id, - }]) - .unwrap(); - - let data_file = - build_partitioned_data_file(&Struct::from_iter([Some(Literal::int(7))]), spec_id); - - // The int partition does not match the long-keyed delete. - assert!( - index - .get_deletes_for_data_file(&data_file, Some(0)) - .unwrap() - .is_empty() - ); - - // Widening the data file's partition to the current (long) type makes it match. - let partition_type = StructType::new(vec![ - NestedField::required(1000, "b", Type::Primitive(PrimitiveType::Long)).into(), - ]); - let mut promoted = data_file.clone(); - promoted.partition = data_file.promoted_partition(&partition_type); - - assert_eq!( - index - .get_deletes_for_data_file(&promoted, Some(0)) - .unwrap() - .len(), - 1 - ); - } - - #[test] - fn promoted_data_file_partition_matches_delete_after_source_dropped() { - // The promoted partition source column was then dropped, so the spec - // resolves it to `Unknown`. Ceiling-widening still normalizes the data - // file's int partition to long so it matches the delete's long partition. - let spec_id = 1; - let eq_delete = - build_partitioned_eq_delete(&Struct::from_iter([Some(Literal::long(7))]), spec_id); - let index = PopulatedDeleteFileIndex::new(vec![DeleteFileContext { - manifest_entry: build_added_manifest_entry(4, &eq_delete).into(), - partition_spec_id: spec_id, - }]) - .unwrap(); - - let data_file = - build_partitioned_data_file(&Struct::from_iter([Some(Literal::int(7))]), spec_id); - let partition_type = StructType::new(vec![ - NestedField::required(1000, "b", Type::Primitive(PrimitiveType::Unknown)).into(), - ]); - let mut promoted = data_file.clone(); - promoted.partition = data_file.promoted_partition(&partition_type); - - assert_eq!( - index - .get_deletes_for_data_file(&promoted, Some(0)) - .unwrap() - .len(), - 1 - ); - } - #[test] fn test_delete_file_index_unpartitioned() { let deletes: Vec = vec![ diff --git a/crates/iceberg/src/expr/visitors/inclusive_metrics_evaluator.rs b/crates/iceberg/src/expr/visitors/inclusive_metrics_evaluator.rs index 7077de2cda..f7abe7ab52 100644 --- a/crates/iceberg/src/expr/visitors/inclusive_metrics_evaluator.rs +++ b/crates/iceberg/src/expr/visitors/inclusive_metrics_evaluator.rs @@ -15,6 +15,8 @@ // specific language governing permissions and limitations // under the License. +use std::borrow::Cow; + use fnv::FnvHashSet; use crate::expr::visitors::bound_predicate_visitor::{BoundPredicateVisitor, visit}; @@ -120,8 +122,8 @@ impl<'a> InclusiveMetricsEvaluator<'a> { }; if let Some(bound) = bound { - let bound = Self::promote_bound(bound, reference); - if cmp_fn(&bound, datum) { + let bound = &*Self::promote_bound(bound, reference); + if cmp_fn(bound, datum) { return ROWS_MIGHT_MATCH; } @@ -133,12 +135,15 @@ impl<'a> InclusiveMetricsEvaluator<'a> { /// Widens a stored bound to the reference's type so a bound written under a /// narrower type (e.g. an `int` column later promoted to `long`) compares - /// against a predicate bound to the wider type. - fn promote_bound(bound: &Datum, reference: &BoundReference) -> Datum { - match reference.field().field_type.as_primitive_type() { - Some(primitive_type) => bound.promote_to(primitive_type), - None => bound.clone(), - } + /// against a predicate bound to the wider type. Borrows the bound when nothing + /// is promoted, so unpromoted bounds (e.g. strings/binaries) are not copied. + fn promote_bound<'b>(bound: &'b Datum, reference: &BoundReference) -> Cow<'b, Datum> { + reference + .field() + .field_type + .as_primitive_type() + .and_then(|primitive_type| bound.promote_to(primitive_type)) + .map_or(Cow::Borrowed(bound), Cow::Owned) } } @@ -270,7 +275,7 @@ impl BoundPredicateVisitor for InclusiveMetricsEvaluator<'_> { } if let Some(lower_bound) = self.lower_bound(field_id) { - let lower_bound = Self::promote_bound(lower_bound, reference); + let lower_bound = &*Self::promote_bound(lower_bound, reference); if lower_bound.is_nan() { // NaN indicates unreliable bounds. // See the InclusiveMetricsEvaluator docs for more. @@ -281,7 +286,7 @@ impl BoundPredicateVisitor for InclusiveMetricsEvaluator<'_> { } if let Some(upper_bound) = self.upper_bound(field_id) { - let upper_bound = Self::promote_bound(upper_bound, reference); + let upper_bound = &*Self::promote_bound(upper_bound, reference); if upper_bound.is_nan() { // NaN indicates unreliable bounds. // See the InclusiveMetricsEvaluator docs for more. @@ -451,26 +456,26 @@ impl BoundPredicateVisitor for InclusiveMetricsEvaluator<'_> { } if let Some(lower_bound) = self.lower_bound(field_id) { - let lower_bound = Self::promote_bound(lower_bound, reference); + let lower_bound = &*Self::promote_bound(lower_bound, reference); if lower_bound.is_nan() { // NaN indicates unreliable bounds. See the InclusiveMetricsEvaluator docs for more. return ROWS_MIGHT_MATCH; } - if !literals.iter().any(|datum| datum.ge(&lower_bound)) { + if !literals.iter().any(|datum| datum.ge(lower_bound)) { // if all values are less than lower bound, rows cannot match. return ROWS_CANNOT_MATCH; } } if let Some(upper_bound) = self.upper_bound(field_id) { - let upper_bound = Self::promote_bound(upper_bound, reference); + let upper_bound = &*Self::promote_bound(upper_bound, reference); if upper_bound.is_nan() { // NaN indicates unreliable bounds. See the InclusiveMetricsEvaluator docs for more. return ROWS_MIGHT_MATCH; } - if !literals.iter().any(|datum| datum.le(&upper_bound)) { + if !literals.iter().any(|datum| datum.le(upper_bound)) { // if all values are greater than upper bound, rows cannot match. return ROWS_CANNOT_MATCH; } diff --git a/crates/iceberg/src/scan/context.rs b/crates/iceberg/src/scan/context.rs index 63812a537c..d20deebd27 100644 --- a/crates/iceberg/src/scan/context.rs +++ b/crates/iceberg/src/scan/context.rs @@ -30,7 +30,8 @@ use crate::scan::{ }; use crate::spec::{ ManifestContentType, ManifestEntryRef, ManifestFile, ManifestList, NameMapping, - PartitionSpecRef, SchemaRef, SnapshotRef, SortOrderRef, StructType, TableMetadataRef, + PartitionSpecRef, PrimitiveType, SchemaRef, SnapshotRef, SortOrderRef, StructType, + TableMetadataRef, }; use crate::{Error, ErrorKind, Result}; @@ -77,13 +78,26 @@ pub(crate) struct ManifestEntryContext { /// (differing primitive types), i.e. partition values written with `written` need /// widening to `current` before they compare against predicates bound to the current /// schema. +/// Returns whether any `int` or `float` field of `written` is widened in `current` (to +/// `long` or `double`, or to `unknown` after its source column is dropped), i.e. partition +/// values written with `written` need widening before they compare against predicates +/// bound to the current schema. fn partition_types_differ(current: &StructType, written: &StructType) -> bool { - current.fields().len() != written.fields().len() - || current - .fields() - .iter() - .zip(written.fields()) - .any(|(c, w)| c.field_type.as_primitive_type() != w.field_type.as_primitive_type()) + current.fields().iter().zip(written.fields()).any(|(c, w)| { + matches!( + ( + w.field_type.as_primitive_type(), + c.field_type.as_primitive_type() + ), + ( + Some(PrimitiveType::Int), + Some(PrimitiveType::Long | PrimitiveType::Unknown) + ) | ( + Some(PrimitiveType::Float), + Some(PrimitiveType::Double | PrimitiveType::Unknown) + ) + ) + }) } impl ManifestFileContext { @@ -128,7 +142,8 @@ impl ManifestFileContext { let manifest_entry = match promotion_target.as_ref() { Some(partition_type) => { let mut entry = manifest_entry.as_ref().clone(); - entry.data_file.partition = entry.data_file.promoted_partition(partition_type); + entry.data_file.partition = + entry.data_file.promoted_partition(partition_type)?; Arc::new(entry) } None => manifest_entry.clone(), diff --git a/crates/iceberg/src/spec/manifest/data_file.rs b/crates/iceberg/src/spec/manifest/data_file.rs index 5362da3493..b7cc192949 100644 --- a/crates/iceberg/src/spec/manifest/data_file.rs +++ b/crates/iceberg/src/spec/manifest/data_file.rs @@ -208,12 +208,25 @@ impl DataFile { /// promoted (`int` -> `long`, `float` -> `double`) since the file's manifest /// was written. Values whose type is unchanged are copied through. Column /// bounds are widened separately, lazily, where they are compared. - pub(crate) fn promoted_partition(&self, partition_type: &StructType) -> Struct { - self.partition + /// + /// # Errors + /// Returns `DataInvalid` if the stored tuple and `partition_type` have different + /// lengths; they always should, since both come from the same spec. + pub(crate) fn promoted_partition(&self, partition_type: &StructType) -> Result { + if self.partition.fields().len() != partition_type.fields().len() { + return Err(invalid_data!( + "Partition tuple has {} values but the spec partition type has {} fields", + self.partition.fields().len(), + partition_type.fields().len() + )); + } + + Ok(self + .partition .iter() .zip(partition_type.fields()) .map(|(value, field)| promote_partition_value(value, field)) - .collect() + .collect()) } /// Get the content type of the data file (data, equality deletes, or position deletes) @@ -487,7 +500,7 @@ mod test { .unwrap(); assert_eq!( - data_file.promoted_partition(&partition_type), + data_file.promoted_partition(&partition_type).unwrap(), Struct::from_iter([Some(Literal::long(7)), Some(Literal::double(1.5))]) ); } @@ -513,7 +526,7 @@ mod test { .unwrap(); assert_eq!( - data_file.promoted_partition(&partition_type), + data_file.promoted_partition(&partition_type).unwrap(), Struct::from_iter([Some(Literal::long(7))]) ); } @@ -536,7 +549,7 @@ mod test { .unwrap(); assert_eq!( - data_file.promoted_partition(&partition_type), + data_file.promoted_partition(&partition_type).unwrap(), data_file.partition ); } diff --git a/crates/iceberg/src/spec/values/datum.rs b/crates/iceberg/src/spec/values/datum.rs index 2a6c9a2907..5c4339d8ba 100644 --- a/crates/iceberg/src/spec/values/datum.rs +++ b/crates/iceberg/src/spec/values/datum.rs @@ -1232,13 +1232,12 @@ impl Datum { /// Returns a copy whose literal is widened to `target` for the value /// promotions `int` -> `long` and `float` -> `double`, so a value written /// under the narrower type compares against a predicate bound to the wider - /// one. Any other `target` returns an unchanged copy; decimals already - /// compare across precisions. - pub(crate) fn promote_to(&self, target: &PrimitiveType) -> Datum { - match self.literal.promote_to(target) { - Some(literal) => Datum::new(target.clone(), literal), - None => self.clone(), - } + /// one. Returns `None` when `target` does not widen this value (decimals + /// already compare across precisions), so callers can avoid a copy. + pub(crate) fn promote_to(&self, target: &PrimitiveType) -> Option { + self.literal + .promote_to(target) + .map(|literal| Datum::new(target.clone(), literal)) } /// Returns true if the Literal represents a primitive type From 57a01abc00a792970ff9fcaa4ad7fa7f674f2bd6 Mon Sep 17 00:00:00 2001 From: Anoop Johnson Date: Mon, 19 Oct 2026 13:26:25 -0700 Subject: [PATCH 5/8] test(scan): add end-to-end promotion scan regressions for deletes Adds plan_files scan regressions over a promoted partition column, routing an old-schema manifest through the reader-side widening: - A v3 deletion vector written after the promotion against the older data file. Without the widening the DV partition check errors (long(7) vs int(7)); the test guards that the data file's partition widens so the DV attaches. - An equality delete written before the promotion against the older data file, so both sides widen to long. Confirms the pre-existing match is preserved. Parameterizes the equality-delete fixture on whether the delete predates the promotion, and adds a v3 deletion-vector fixture. --- crates/iceberg/src/scan/mod.rs | 48 +++++++- crates/iceberg/src/test_utils/scan.rs | 164 ++++++++++++++++++++++++-- 2 files changed, 200 insertions(+), 12 deletions(-) diff --git a/crates/iceberg/src/scan/mod.rs b/crates/iceberg/src/scan/mod.rs index 60027c9e9e..db2676ea46 100644 --- a/crates/iceberg/src/scan/mod.rs +++ b/crates/iceberg/src/scan/mod.rs @@ -1204,7 +1204,7 @@ mod tests { // attaches. Without the normalization the int/long mismatch drops the delete // (and prunes the file under the filter). let mut fixture = TableTestFixture::new(); - fixture.setup_promoted_partition_with_eq_delete().await; + fixture.setup_promoted_partition_with_eq_delete(false).await; let tasks: Vec = fixture .table @@ -1244,6 +1244,52 @@ mod tests { assert_eq!(tasks[0].deletes().len(), 1); } + #[tokio::test] + async fn test_scan_widens_old_partition_with_pre_promotion_delete() { + // Data file and equality delete both written before the promotion (int + // partition). They matched on main; after both sides widen to long they must + // still match, so the delete stays attached. + let mut fixture = TableTestFixture::new(); + fixture.setup_promoted_partition_with_eq_delete(true).await; + + let tasks: Vec = fixture + .table + .scan() + .build() + .unwrap() + .plan_files() + .await + .unwrap() + .try_collect() + .await + .unwrap(); + assert_eq!(tasks.len(), 1); + assert_eq!(tasks[0].deletes().len(), 1); + } + + #[tokio::test] + async fn test_scan_widens_old_partition_so_deletion_vector_matches() { + // A v3 deletion vector written after the promotion (long partition) references + // an older data file (int partition). The widening must make the partitions + // agree so the DV attaches, instead of failing the DV partition check. + let mut fixture = TableTestFixture::new(); + fixture.setup_promoted_partition_with_dv().await; + + let tasks: Vec = fixture + .table + .scan() + .build() + .unwrap() + .plan_files() + .await + .unwrap() + .try_collect() + .await + .unwrap(); + assert_eq!(tasks.len(), 1); + assert_eq!(tasks[0].deletes().len(), 1); + } + #[tokio::test] async fn test_plan_files_carries_row_lineage_into_file_scan_task() { let mut fixture = TableTestFixture::new(); diff --git a/crates/iceberg/src/test_utils/scan.rs b/crates/iceberg/src/test_utils/scan.rs index 009c0bf366..74fa3e6aff 100644 --- a/crates/iceberg/src/test_utils/scan.rs +++ b/crates/iceberg/src/test_utils/scan.rs @@ -606,12 +606,17 @@ impl TableTestFixture { /// Writes a data manifest under an older schema where the identity partition /// source `x` (id 1) is still `int` (so the partition value is stored as `int`), - /// plus an equality-delete manifest under the current schema where `x` is `long`. - /// Exercises the scan's reader-side widening: the old `int` partition must widen - /// to `long` so the delete, keyed by the current `long` partition, still matches - /// the data file. The data/delete parquet files themselves are not written, which - /// is enough for `plan_files` (it reads only manifests). - pub(crate) async fn setup_promoted_partition_with_eq_delete(&mut self) { + /// plus an equality-delete manifest. Exercises the scan's reader-side widening. + /// When `delete_before_promotion` is false the delete is written under the current + /// `long` schema (`long` partition), and the old `int` partition must widen to + /// match it. When true the delete is also written under the old `int` schema (`int` + /// partition), which matched on main and must keep matching once both sides widen. + /// The data/delete parquet files themselves are not written, which is enough for + /// `plan_files` (it reads only manifests). + pub(crate) async fn setup_promoted_partition_with_eq_delete( + &mut self, + delete_before_promotion: bool, + ) { let current_snapshot = self.table.metadata().current_snapshot().unwrap(); let parent_snapshot = current_snapshot .parent_snapshot(self.table.metadata()) @@ -636,7 +641,7 @@ impl TableTestFixture { let mut data_writer = ManifestWriterBuilder::new( self.next_manifest_file(), Some(current_snapshot.snapshot_id()), - written_schema, + written_schema.clone(), current_partition_spec.as_ref().clone(), ) .build_v2_data(); @@ -664,12 +669,21 @@ impl TableTestFixture { .unwrap(); let data_manifest = data_writer.write_manifest_file().await.unwrap(); - // Equality-delete manifest under the current (long) schema; Added in the - // current snapshot so its sequence number is above the data file's. + // Equality-delete manifest, Added in the current snapshot so its sequence + // number is above the data file's. Written under the old `int` schema or the + // current `long` schema depending on `delete_before_promotion`. + let (delete_schema, delete_partition) = if delete_before_promotion { + (written_schema, Struct::from_iter([Some(Literal::int(7))])) + } else { + ( + current_schema.clone(), + Struct::from_iter([Some(Literal::long(7))]), + ) + }; let mut delete_writer = ManifestWriterBuilder::new( self.next_manifest_file(), Some(current_snapshot.snapshot_id()), - current_schema.clone(), + delete_schema, current_partition_spec.as_ref().clone(), ) .build_v2_deletes(); @@ -686,7 +700,7 @@ impl TableTestFixture { .file_format(DataFileFormat::Parquet) .file_size_in_bytes(100) .record_count(1) - .partition(Struct::from_iter([Some(Literal::long(7))])) + .partition(delete_partition) .build() .unwrap(), ) @@ -715,6 +729,134 @@ impl TableTestFixture { manifest_list_write.close().await.unwrap(); } + /// Writes a v3 data manifest under an older schema where the identity partition + /// source `x` (id 1) is still `int`, plus a v3 deletion-vector manifest under the + /// current `long` schema whose DV references the data file with a `long` partition. + /// The reader-side widening must make the data file's `int` partition agree with + /// the DV's `long` partition so the DV partition check passes and the DV attaches; + /// on main the check errors. Parquet/Puffin files are not written (`plan_files` + /// reads only manifests). + pub(crate) async fn setup_promoted_partition_with_dv(&mut self) { + let metadata = TableMetadataBuilder::new_from_metadata( + self.table.metadata().clone(), + self.table.metadata_location().map(str::to_string), + ) + .upgrade_format_version(FormatVersion::V3) + .unwrap() + .build() + .unwrap() + .metadata; + self.table = Table::builder() + .metadata(metadata) + .identifier(self.table.identifier().clone()) + .file_io(self.table.file_io().clone()) + .metadata_location(self.table.metadata_location().unwrap().to_string()) + .runtime(test_runtime()) + .build() + .unwrap(); + + 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(); + + let mut fields = current_schema.as_struct().fields().to_vec(); + fields[0] = NestedField::required(1, "x", PrimitiveType::Int.into()).into(); + let written_schema = Arc::new( + Schema::builder() + .with_schema_id(0) + .with_fields(fields) + .build() + .unwrap(), + ); + + let data_file_path = format!("{}/x7.parquet", &self.table_location); + + let mut data_writer = ManifestWriterBuilder::new( + self.next_manifest_file(), + Some(current_snapshot.snapshot_id()), + written_schema, + current_partition_spec.as_ref().clone(), + ) + .build_v3_data(); + data_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(0) + .content(DataContentType::Data) + .file_path(data_file_path.clone()) + .file_format(DataFileFormat::Parquet) + .file_size_in_bytes(100) + .record_count(1) + .partition(Struct::from_iter([Some(Literal::int(7))])) + .build() + .unwrap(), + ) + .build(), + ) + .unwrap(); + let data_manifest = data_writer.write_manifest_file().await.unwrap(); + + let mut dv_writer = ManifestWriterBuilder::new( + self.next_manifest_file(), + Some(current_snapshot.snapshot_id()), + current_schema.clone(), + current_partition_spec.as_ref().clone(), + ) + .build_v3_deletes(); + dv_writer + .add_entry( + ManifestEntry::builder() + .status(ManifestStatus::Added) + .data_file( + DataFileBuilder::default() + .partition_spec_id(0) + .content(DataContentType::PositionDeletes) + .file_path(format!("{}/dv.puffin", &self.table_location)) + .file_format(DataFileFormat::Puffin) + .file_size_in_bytes(100) + .record_count(1) + .referenced_data_file(Some(data_file_path)) + .content_offset(Some(0)) + .content_size_in_bytes(Some(40)) + .partition(Struct::from_iter([Some(Literal::long(7))])) + .build() + .unwrap(), + ) + .build(), + ) + .unwrap(); + let dv_manifest = dv_writer.write_manifest_file().await.unwrap(); + + let manifest_list_writer = self + .table + .file_io() + .new_output(current_snapshot.manifest_list()) + .unwrap() + .writer() + .await + .unwrap(); + let mut manifest_list_write = ManifestListWriter::v3( + manifest_list_writer, + current_snapshot.snapshot_id(), + current_snapshot.parent_snapshot_id(), + current_snapshot.sequence_number(), + Some(42), + ); + manifest_list_write + .add_manifests(vec![data_manifest, dv_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 { From 02ad0ea576ced9c817bd0c100307422b1240fc7b Mon Sep 17 00:00:00 2001 From: Anoop Johnson Date: Mon, 19 Oct 2026 15:15:38 -0700 Subject: [PATCH 6/8] fix(scan): copy dropped partition sources through unchanged A dropped partition source resolves to the Unknown primitive type. Widening its stored value to the promotion ceiling keyed on the backing literal, so a date (stored as an int) was silently promoted to long. The promotion gate skips a manifest whose only type difference is a dropped source, so a data manifest that widened a dropped date diverged from the delete manifest that left it alone, detaching the delete. Copy dropped-source values through unchanged and restrict the gate to genuine int to long and float to double promotions on live columns. --- crates/iceberg/src/scan/context.rs | 20 ++----- crates/iceberg/src/spec/manifest/data_file.rs | 57 ++++++++++++++----- 2 files changed, 49 insertions(+), 28 deletions(-) diff --git a/crates/iceberg/src/scan/context.rs b/crates/iceberg/src/scan/context.rs index d20deebd27..6ad7a5fa4a 100644 --- a/crates/iceberg/src/scan/context.rs +++ b/crates/iceberg/src/scan/context.rs @@ -74,14 +74,9 @@ pub(crate) struct ManifestEntryContext { pub sort_order: Option, } -/// Returns whether `current` widens any field of `written` at the same position -/// (differing primitive types), i.e. partition values written with `written` need -/// widening to `current` before they compare against predicates bound to the current -/// schema. -/// Returns whether any `int` or `float` field of `written` is widened in `current` (to -/// `long` or `double`, or to `unknown` after its source column is dropped), i.e. partition -/// values written with `written` need widening before they compare against predicates -/// bound to the current schema. +/// Returns whether any `int` or `float` field of `written` is widened to `long` or +/// `double` in `current`, i.e. partition values written with `written` need widening +/// before they compare against predicates bound to the current schema. fn partition_types_differ(current: &StructType, written: &StructType) -> bool { current.fields().iter().zip(written.fields()).any(|(c, w)| { matches!( @@ -89,13 +84,8 @@ fn partition_types_differ(current: &StructType, written: &StructType) -> bool { w.field_type.as_primitive_type(), c.field_type.as_primitive_type() ), - ( - Some(PrimitiveType::Int), - Some(PrimitiveType::Long | PrimitiveType::Unknown) - ) | ( - Some(PrimitiveType::Float), - Some(PrimitiveType::Double | PrimitiveType::Unknown) - ) + (Some(PrimitiveType::Int), Some(PrimitiveType::Long)) + | (Some(PrimitiveType::Float), Some(PrimitiveType::Double)) ) }) } diff --git a/crates/iceberg/src/spec/manifest/data_file.rs b/crates/iceberg/src/spec/manifest/data_file.rs index b7cc192949..b89ceeb230 100644 --- a/crates/iceberg/src/spec/manifest/data_file.rs +++ b/crates/iceberg/src/spec/manifest/data_file.rs @@ -29,9 +29,7 @@ use super::{ }; use crate::Error; use crate::error::{Result, invalid_data}; -use crate::spec::{ - DEFAULT_PARTITION_SPEC_ID, Literal, NestedFieldRef, PrimitiveType, Struct, StructType, Type, -}; +use crate::spec::{DEFAULT_PARTITION_SPEC_ID, Literal, NestedFieldRef, Struct, StructType, Type}; /// Data file carries data file path, partition tuple, metrics, … #[derive(Debug, PartialEq, Clone, Eq, Builder)] @@ -190,10 +188,6 @@ fn promote_partition_value(value: Option<&Literal>, field: &NestedFieldRef) -> O match value { Some(Literal::Primitive(primitive)) => { let promoted = match field.field_type.as_primitive_type() { - // The partition source column was dropped, so there is no target type to - // widen to; widen to the promotion ceiling instead so this value and its - // deletes still normalize to the same type and the delete index matches. - Some(PrimitiveType::Unknown) => primitive.widen(), Some(target) => primitive.promote_to(target), None => None, }; @@ -211,7 +205,7 @@ impl DataFile { /// /// # Errors /// Returns `DataInvalid` if the stored tuple and `partition_type` have different - /// lengths; they always should, since both come from the same spec. + /// lengths; the lengths should always match, since both come from the same spec. pub(crate) fn promoted_partition(&self, partition_type: &StructType) -> Result { if self.partition.fields().len() != partition_type.fields().len() { return Err(invalid_data!( @@ -461,6 +455,7 @@ impl std::fmt::Display for DataFileFormat { #[cfg(test)] mod test { + use crate::ErrorKind; use crate::spec::{ DataContentType, DataFileBuilder, DataFileFormat, Literal, NestedField, PrimitiveType, Struct, StructType, Type, @@ -506,10 +501,11 @@ mod test { } #[test] - fn promoted_partition_widens_dropped_source_to_ceiling() { - // A dropped partition source resolves to `Unknown`, so there is no target - // type; the value still widens to its ceiling so it keeps matching deletes - // written while the source still existed. + fn promoted_partition_leaves_dropped_source_unchanged() { + // A dropped partition source resolves to `Unknown`, which has no promotion + // target, so the value is copied through verbatim. Widening it (e.g. a + // `date`, stored as an `int`, to `long`) would diverge from delete manifests + // that skip promotion, silently detaching their deletes. let partition_type = StructType::new(vec![ NestedField::required(1000, "b", Type::Primitive(PrimitiveType::Unknown)).into(), ]); @@ -527,7 +523,7 @@ mod test { assert_eq!( data_file.promoted_partition(&partition_type).unwrap(), - Struct::from_iter([Some(Literal::long(7))]) + data_file.partition ); } @@ -553,4 +549,39 @@ mod test { data_file.partition ); } + + #[test] + fn promoted_partition_rejects_tuple_length_mismatch() { + let data_file = |values: Vec>| { + DataFileBuilder::default() + .content(DataContentType::Data) + .file_path("x.parquet".to_string()) + .file_format(DataFileFormat::Parquet) + .partition(Struct::from_iter(values)) + .record_count(1) + .file_size_in_bytes(1) + .partition_spec_id(0) + .build() + .unwrap() + }; + let one_field = StructType::new(vec![ + NestedField::required(1000, "b", Type::Primitive(PrimitiveType::Long)).into(), + ]); + let two_fields = StructType::new(vec![ + NestedField::required(1000, "b", Type::Primitive(PrimitiveType::Long)).into(), + NestedField::required(1001, "c", Type::Primitive(PrimitiveType::Long)).into(), + ]); + + let longer = data_file(vec![Some(Literal::int(7)), Some(Literal::int(8))]); + assert_eq!( + longer.promoted_partition(&one_field).unwrap_err().kind(), + ErrorKind::DataInvalid + ); + + let shorter = data_file(vec![Some(Literal::int(7))]); + assert_eq!( + shorter.promoted_partition(&two_fields).unwrap_err().kind(), + ErrorKind::DataInvalid + ); + } } From 6fb4785a95d8ef968cd5531e82e60d8058969b1b Mon Sep 17 00:00:00 2001 From: Anoop Johnson Date: Mon, 19 Oct 2026 15:57:00 -0700 Subject: [PATCH 7/8] fix(scan): normalize dropped numeric partition sources by written type --- crates/iceberg/src/scan/context.rs | 24 ++- crates/iceberg/src/scan/mod.rs | 100 +++++++++++ crates/iceberg/src/spec/manifest/data_file.rs | 123 ++++++++++--- crates/iceberg/src/test_utils/scan.rs | 169 +++++++++++++++++- 4 files changed, 380 insertions(+), 36 deletions(-) diff --git a/crates/iceberg/src/scan/context.rs b/crates/iceberg/src/scan/context.rs index 6ad7a5fa4a..0e8b0aa503 100644 --- a/crates/iceberg/src/scan/context.rs +++ b/crates/iceberg/src/scan/context.rs @@ -74,9 +74,10 @@ pub(crate) struct ManifestEntryContext { pub sort_order: Option, } -/// Returns whether any `int` or `float` field of `written` is widened to `long` or -/// `double` in `current`, i.e. partition values written with `written` need widening -/// before they compare against predicates bound to the current schema. +/// Returns whether any `int` or `float` field of `written` is widened in `current`, to +/// `long`/`double` or to `unknown` once its source column is dropped, i.e. partition +/// values written with `written` need normalizing before they compare against predicates +/// bound to the current schema or against delete-file partitions. fn partition_types_differ(current: &StructType, written: &StructType) -> bool { current.fields().iter().zip(written.fields()).any(|(c, w)| { matches!( @@ -84,8 +85,13 @@ fn partition_types_differ(current: &StructType, written: &StructType) -> bool { w.field_type.as_primitive_type(), c.field_type.as_primitive_type() ), - (Some(PrimitiveType::Int), Some(PrimitiveType::Long)) - | (Some(PrimitiveType::Float), Some(PrimitiveType::Double)) + ( + Some(PrimitiveType::Int), + Some(PrimitiveType::Long | PrimitiveType::Unknown) + ) | ( + Some(PrimitiveType::Float), + Some(PrimitiveType::Double | PrimitiveType::Unknown) + ) ) }) } @@ -119,21 +125,21 @@ impl ManifestFileContext { // index compare like-typed values. The gate compares the partition types // themselves, not schema ids, since v1 may omit them (defaulting to 0). Column // bounds are widened lazily where the metrics evaluator compares them. - let promotion_target: Option = match partition_spec.as_ref() { + let promotion_target: Option<(StructType, StructType)> = match partition_spec.as_ref() { Some(spec) => { let current = spec.partition_type(&snapshot_schema)?; let written = spec.partition_type(manifest.metadata().schema())?; - partition_types_differ(¤t, &written).then_some(current) + partition_types_differ(¤t, &written).then_some((current, written)) } None => None, }; for manifest_entry in manifest.entries() { let manifest_entry = match promotion_target.as_ref() { - Some(partition_type) => { + Some((current, written)) => { let mut entry = manifest_entry.as_ref().clone(); entry.data_file.partition = - entry.data_file.promoted_partition(partition_type)?; + entry.data_file.promoted_partition(current, written)?; Arc::new(entry) } None => manifest_entry.clone(), diff --git a/crates/iceberg/src/scan/mod.rs b/crates/iceberg/src/scan/mod.rs index db2676ea46..5da08ebc31 100644 --- a/crates/iceberg/src/scan/mod.rs +++ b/crates/iceberg/src/scan/mod.rs @@ -1290,6 +1290,106 @@ mod tests { assert_eq!(tasks[0].deletes().len(), 1); } + /// Evolves `table` so its historical `identity(x)` spec outlives the source column: + /// makes the default spec unpartitioned and drops `x` (field id 1) from the schema, + /// so the manifests' spec resolves `x` to `unknown`. + fn drop_promoted_partition_source(table: Table) -> Table { + let evolved_schema = Schema::builder() + .with_fields( + table + .metadata() + .current_schema() + .as_struct() + .fields() + .iter() + .filter(|field| field.id != 1) + .cloned(), + ) + .with_identifier_field_ids(vec![2]) + .build() + .unwrap(); + let metadata = TableMetadataBuilder::new_from_metadata(table.metadata().clone(), None) + .add_default_partition_spec(UnboundPartitionSpec::builder().build()) + .unwrap() + .add_current_schema(evolved_schema) + .unwrap() + .build() + .unwrap() + .metadata; + table.with_metadata(Arc::new(metadata)) + } + + #[tokio::test] + async fn test_scan_keeps_eq_delete_after_dropping_promoted_source() { + // Partition source `x` promoted int -> long, then dropped. The data file (int) + // and its equality delete (long) must still normalize to the same long partition + // so the delete stays attached; otherwise it silently detaches. + let mut fixture = TableTestFixture::new(); + fixture.setup_promoted_partition_with_eq_delete(false).await; + let table = drop_promoted_partition_source(fixture.table); + + let tasks: Vec = table + .scan() + .build() + .unwrap() + .plan_files() + .await + .unwrap() + .try_collect() + .await + .unwrap(); + assert_eq!(tasks.len(), 1); + assert_eq!(tasks[0].deletes().len(), 1); + } + + #[tokio::test] + async fn test_scan_keeps_deletion_vector_after_dropping_promoted_source() { + // Same as above for a deletion vector: without widening the dropped numeric + // source the DV partition check errors (long(7) vs int(7)). + let mut fixture = TableTestFixture::new(); + fixture.setup_promoted_partition_with_dv().await; + let table = drop_promoted_partition_source(fixture.table); + + let tasks: Vec = table + .scan() + .build() + .unwrap() + .plan_files() + .await + .unwrap() + .try_collect() + .await + .unwrap(); + assert_eq!(tasks.len(), 1); + assert_eq!(tasks[0].deletes().len(), 1); + } + + #[tokio::test] + async fn test_scan_keeps_eq_delete_when_dropped_source_is_date() { + // A promoted numeric source `x` sits beside a dropped `date` source `d`. The gate + // fires on `x`'s int -> long widening, but `d` (stored as an int) must be copied + // through, not widened to long, or the data file's partition diverges from the + // equality delete's and detaches it. + let mut fixture = TableTestFixture::new(); + fixture + .setup_promoted_partition_with_eq_delete_and_dropped_date() + .await; + + let tasks: Vec = fixture + .table + .scan() + .build() + .unwrap() + .plan_files() + .await + .unwrap() + .try_collect() + .await + .unwrap(); + assert_eq!(tasks.len(), 1); + assert_eq!(tasks[0].deletes().len(), 1); + } + #[tokio::test] async fn test_plan_files_carries_row_lineage_into_file_scan_task() { let mut fixture = TableTestFixture::new(); diff --git a/crates/iceberg/src/spec/manifest/data_file.rs b/crates/iceberg/src/spec/manifest/data_file.rs index b89ceeb230..6138e52202 100644 --- a/crates/iceberg/src/spec/manifest/data_file.rs +++ b/crates/iceberg/src/spec/manifest/data_file.rs @@ -29,7 +29,9 @@ use super::{ }; use crate::Error; use crate::error::{Result, invalid_data}; -use crate::spec::{DEFAULT_PARTITION_SPEC_ID, Literal, NestedFieldRef, Struct, StructType, Type}; +use crate::spec::{ + DEFAULT_PARTITION_SPEC_ID, Literal, NestedFieldRef, PrimitiveType, Struct, StructType, Type, +}; /// Data file carries data file path, partition tuple, metrics, … #[derive(Debug, PartialEq, Clone, Eq, Builder)] @@ -184,10 +186,24 @@ pub struct DataFile { pub(crate) content_size_in_bytes: Option, } -fn promote_partition_value(value: Option<&Literal>, field: &NestedFieldRef) -> Option { +fn promote_partition_value( + value: Option<&Literal>, + current_field: &NestedFieldRef, + written_field: &NestedFieldRef, +) -> Option { match value { Some(Literal::Primitive(primitive)) => { - let promoted = match field.field_type.as_primitive_type() { + let promoted = match current_field.field_type.as_primitive_type() { + // The source column was dropped, so there is no current type to widen to. + // Widen a numeric to its promotion ceiling so this value and its deletes + // normalize the same way; keep every other type (e.g. a `date`, stored as + // an `int`) at its written type so deletes that skip normalization match. + Some(PrimitiveType::Unknown) => { + match written_field.field_type.as_primitive_type() { + Some(PrimitiveType::Int | PrimitiveType::Float) => primitive.widen(), + _ => None, + } + } Some(target) => primitive.promote_to(target), None => None, }; @@ -198,28 +214,40 @@ fn promote_partition_value(value: Option<&Literal>, field: &NestedFieldRef) -> O } impl DataFile { - /// Returns this file's partition tuple widened to `partition_type` for fields - /// promoted (`int` -> `long`, `float` -> `double`) since the file's manifest - /// was written. Values whose type is unchanged are copied through. Column - /// bounds are widened separately, lazily, where they are compared. + /// Returns this file's partition tuple normalized to the current schema: a field + /// promoted (`int` -> `long`, `float` -> `double`) since the manifest was written is + /// widened to its `current` type; a dropped numeric source, which has no `current` + /// type, is widened to its promotion ceiling by its `written` type; all other values + /// are copied through. Column bounds are widened separately, lazily, where they are + /// compared. + /// + /// `current` and `written` are the file's spec resolved against the current schema and + /// against the manifest's schema, so they match the stored tuple in length. /// /// # Errors - /// Returns `DataInvalid` if the stored tuple and `partition_type` have different - /// lengths; the lengths should always match, since both come from the same spec. - pub(crate) fn promoted_partition(&self, partition_type: &StructType) -> Result { - if self.partition.fields().len() != partition_type.fields().len() { + /// Returns `DataInvalid` if the stored tuple and `current` have different lengths; the + /// lengths should always match, since both come from the same spec. + pub(crate) fn promoted_partition( + &self, + current: &StructType, + written: &StructType, + ) -> Result { + if self.partition.fields().len() != current.fields().len() { return Err(invalid_data!( "Partition tuple has {} values but the spec partition type has {} fields", self.partition.fields().len(), - partition_type.fields().len() + current.fields().len() )); } Ok(self .partition .iter() - .zip(partition_type.fields()) - .map(|(value, field)| promote_partition_value(value, field)) + .zip(current.fields()) + .zip(written.fields()) + .map(|((value, current_field), written_field)| { + promote_partition_value(value, current_field, written_field) + }) .collect()) } @@ -475,10 +503,14 @@ mod test { fn promoted_partition_widens_int_and_float() { // Partition columns promoted int -> long and float -> double; a file // written before the promotion still carries the narrower types. - let partition_type = StructType::new(vec![ + let current = StructType::new(vec![ NestedField::required(1000, "b", Type::Primitive(PrimitiveType::Long)).into(), NestedField::required(1001, "c", Type::Primitive(PrimitiveType::Double)).into(), ]); + let written = StructType::new(vec![ + NestedField::required(1000, "b", Type::Primitive(PrimitiveType::Int)).into(), + NestedField::required(1001, "c", Type::Primitive(PrimitiveType::Float)).into(), + ]); let data_file = DataFileBuilder::default() .content(DataContentType::Data) @@ -495,20 +527,22 @@ mod test { .unwrap(); assert_eq!( - data_file.promoted_partition(&partition_type).unwrap(), + data_file.promoted_partition(¤t, &written).unwrap(), Struct::from_iter([Some(Literal::long(7)), Some(Literal::double(1.5))]) ); } #[test] - fn promoted_partition_leaves_dropped_source_unchanged() { - // A dropped partition source resolves to `Unknown`, which has no promotion - // target, so the value is copied through verbatim. Widening it (e.g. a - // `date`, stored as an `int`, to `long`) would diverge from delete manifests - // that skip promotion, silently detaching their deletes. - let partition_type = StructType::new(vec![ + fn promoted_partition_widens_dropped_numeric_source() { + // A dropped partition source resolves to `Unknown`, so there is no current + // type; a numeric still widens to its ceiling by its written type, keeping it + // aligned with deletes written while the source still existed. + let current = StructType::new(vec![ NestedField::required(1000, "b", Type::Primitive(PrimitiveType::Unknown)).into(), ]); + let written = StructType::new(vec![ + NestedField::required(1000, "b", Type::Primitive(PrimitiveType::Int)).into(), + ]); let data_file = DataFileBuilder::default() .content(DataContentType::Data) @@ -522,7 +556,36 @@ mod test { .unwrap(); assert_eq!( - data_file.promoted_partition(&partition_type).unwrap(), + data_file.promoted_partition(¤t, &written).unwrap(), + Struct::from_iter([Some(Literal::long(7))]) + ); + } + + #[test] + fn promoted_partition_preserves_dropped_date_source() { + // A `date` is stored as an `int`, so widening a dropped date the way a dropped + // int widens would turn it into a `long` and diverge from delete manifests that + // skip normalization. A dropped non-numeric source is copied through unchanged. + let current = StructType::new(vec![ + NestedField::required(1000, "d", Type::Primitive(PrimitiveType::Unknown)).into(), + ]); + let written = StructType::new(vec![ + NestedField::required(1000, "d", Type::Primitive(PrimitiveType::Date)).into(), + ]); + + let data_file = DataFileBuilder::default() + .content(DataContentType::Data) + .file_path("x.parquet".to_string()) + .file_format(DataFileFormat::Parquet) + .partition(Struct::from_iter([Some(Literal::date(20_000))])) + .record_count(1) + .file_size_in_bytes(1) + .partition_spec_id(0) + .build() + .unwrap(); + + assert_eq!( + data_file.promoted_partition(¤t, &written).unwrap(), data_file.partition ); } @@ -545,7 +608,9 @@ mod test { .unwrap(); assert_eq!( - data_file.promoted_partition(&partition_type).unwrap(), + data_file + .promoted_partition(&partition_type, &partition_type) + .unwrap(), data_file.partition ); } @@ -574,13 +639,19 @@ mod test { let longer = data_file(vec![Some(Literal::int(7)), Some(Literal::int(8))]); assert_eq!( - longer.promoted_partition(&one_field).unwrap_err().kind(), + longer + .promoted_partition(&one_field, &one_field) + .unwrap_err() + .kind(), ErrorKind::DataInvalid ); let shorter = data_file(vec![Some(Literal::int(7))]); assert_eq!( - shorter.promoted_partition(&two_fields).unwrap_err().kind(), + shorter + .promoted_partition(&two_fields, &two_fields) + .unwrap_err() + .kind(), ErrorKind::DataInvalid ); } diff --git a/crates/iceberg/src/test_utils/scan.rs b/crates/iceberg/src/test_utils/scan.rs index 74fa3e6aff..186171cd0c 100644 --- a/crates/iceberg/src/test_utils/scan.rs +++ b/crates/iceberg/src/test_utils/scan.rs @@ -44,7 +44,8 @@ use crate::metadata_columns::{ use crate::spec::{ DataContentType, DataFileBuilder, DataFileFormat, FormatVersion, Literal, ManifestEntry, ManifestListWriter, ManifestStatus, ManifestWriterBuilder, NestedField, PartitionSpec, - PrimitiveType, Schema, Struct, StructType, TableMetadata, TableMetadataBuilder, Type, + PrimitiveType, Schema, Struct, StructType, TableMetadata, TableMetadataBuilder, Transform, + Type, UnboundPartitionSpec, }; use crate::table::Table; use crate::test_utils::test_runtime; @@ -857,6 +858,172 @@ impl TableTestFixture { manifest_list_write.close().await.unwrap(); } + /// Writes a two-field spec `identity(x)`, `identity(d date)` where the data manifest + /// was written with `x` still `int` (so the current `long` schema widens it), then + /// drops the date source `d`. A dropped `date` is stored as an `int`, so if it were + /// widened like a dropped numeric the data file's partition would become `long` and + /// diverge from the equality delete's, detaching it. Parquet files are not written + /// (`plan_files` reads only manifests). + pub(crate) async fn setup_promoted_partition_with_eq_delete_and_dropped_date(&mut self) { + // Evolve to the two-field spec with the date source still present. + let mut fields = self + .table + .metadata() + .current_schema() + .as_struct() + .fields() + .to_vec(); + fields.push(NestedField::optional(9, "d", PrimitiveType::Date.into()).into()); + let date_schema = Schema::builder().with_fields(fields).build().unwrap(); + let date_spec = PartitionSpec::builder(date_schema.clone()) + .add_partition_field("x", "x", Transform::Identity) + .unwrap() + .add_partition_field("d", "d", Transform::Identity) + .unwrap() + .build() + .unwrap(); + let metadata = TableMetadataBuilder::new_from_metadata(self.table.metadata().clone(), None) + .add_current_schema(date_schema) + .unwrap() + .add_default_partition_spec(date_spec.into_unbound()) + .unwrap() + .build() + .unwrap() + .metadata; + self.table = self.table.clone().with_metadata(Arc::new(metadata)); + + let current_snapshot = self.table.metadata().current_snapshot().unwrap(); + let parent_snapshot = current_snapshot + .parent_snapshot(self.table.metadata()) + .unwrap(); + let current_schema = self.table.metadata().current_schema().clone(); + let current_partition_spec = self.table.metadata().default_partition_spec(); + + // The schema the data manifest was written under: the partition source `x` still + // `int`. + let mut fields = current_schema.as_struct().fields().to_vec(); + fields[0] = NestedField::required(1, "x", PrimitiveType::Int.into()).into(); + let written_schema = Arc::new( + Schema::builder() + .with_schema_id(0) + .with_fields(fields) + .build() + .unwrap(), + ); + + // Data manifest under the old (int) schema; an Existing entry from the parent + // snapshot so its sequence number is below the equality delete's. + let mut data_writer = ManifestWriterBuilder::new( + self.next_manifest_file(), + Some(current_snapshot.snapshot_id()), + written_schema, + current_partition_spec.as_ref().clone(), + ) + .build_v2_data(); + data_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(current_partition_spec.spec_id()) + .content(DataContentType::Data) + .file_path(format!("{}/x7.parquet", &self.table_location)) + .file_format(DataFileFormat::Parquet) + .file_size_in_bytes(100) + .record_count(1) + .partition(Struct::from_iter([ + Some(Literal::int(7)), + Some(Literal::date(20_000)), + ])) + .build() + .unwrap(), + ) + .build(), + ) + .unwrap(); + let data_manifest = data_writer.write_manifest_file().await.unwrap(); + + // Equality-delete manifest, Added in the current snapshot (sequence number above + // the data file's) under the current `long` schema. + let mut delete_writer = ManifestWriterBuilder::new( + self.next_manifest_file(), + Some(current_snapshot.snapshot_id()), + current_schema.clone(), + current_partition_spec.as_ref().clone(), + ) + .build_v2_deletes(); + delete_writer + .add_entry( + ManifestEntry::builder() + .status(ManifestStatus::Added) + .data_file( + DataFileBuilder::default() + .partition_spec_id(current_partition_spec.spec_id()) + .content(DataContentType::EqualityDeletes) + .equality_ids(Some(vec![2])) + .file_path(format!("{}/eq-delete.parquet", &self.table_location)) + .file_format(DataFileFormat::Parquet) + .file_size_in_bytes(100) + .record_count(1) + .partition(Struct::from_iter([ + Some(Literal::long(7)), + Some(Literal::date(20_000)), + ])) + .build() + .unwrap(), + ) + .build(), + ) + .unwrap(); + let delete_manifest = delete_writer.write_manifest_file().await.unwrap(); + + 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_manifest, delete_manifest].into_iter()) + .unwrap(); + manifest_list_write.close().await.unwrap(); + + // Drop the date source `d`: the manifests keep the two-field spec while the + // current schema loses `d`, so the spec resolves `d` to `unknown`. + let evolved_schema = Schema::builder() + .with_fields( + current_schema + .as_struct() + .fields() + .iter() + .filter(|field| field.id != 9) + .cloned(), + ) + .build() + .unwrap(); + let metadata = TableMetadataBuilder::new_from_metadata(self.table.metadata().clone(), None) + .add_default_partition_spec(UnboundPartitionSpec::builder().build()) + .unwrap() + .add_current_schema(evolved_schema) + .unwrap() + .build() + .unwrap() + .metadata; + self.table = self.table.clone().with_metadata(Arc::new(metadata)); + } + /// 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 { From db4e54454ab9372bf53f8a9c7657deec5f5a3af4 Mon Sep 17 00:00:00 2001 From: Anoop Johnson Date: Tue, 20 Oct 2026 21:51:33 -0700 Subject: [PATCH 8/8] Feedback --- crates/iceberg/src/scan/mod.rs | 22 ++++++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/crates/iceberg/src/scan/mod.rs b/crates/iceberg/src/scan/mod.rs index 5da08ebc31..7682a0a2e3 100644 --- a/crates/iceberg/src/scan/mod.rs +++ b/crates/iceberg/src/scan/mod.rs @@ -1223,6 +1223,11 @@ mod tests { 1, "equality delete must attach after widening the old int partition to long" ); + assert_eq!( + tasks[0].partition(), + Some(&Struct::from_iter([Some(Literal::long(7))])), + "the widened long(7) tuple must reach FileScanTask::partition()" + ); let tasks: Vec = fixture .table @@ -1242,6 +1247,23 @@ mod tests { "filter on the promoted type must not prune the file" ); assert_eq!(tasks[0].deletes().len(), 1); + + let tasks: Vec = fixture + .table + .scan() + .with_filter(Reference::new("x").equal_to(Datum::long(8))) + .build() + .unwrap() + .plan_files() + .await + .unwrap() + .try_collect() + .await + .unwrap(); + assert!( + tasks.is_empty(), + "a file in partition x=7 must be pruned by an x=8 filter" + ); } #[tokio::test]