diff --git a/crates/iceberg/src/expr/visitors/inclusive_metrics_evaluator.rs b/crates/iceberg/src/expr/visitors/inclusive_metrics_evaluator.rs index 8ccd1364d9..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,6 +122,7 @@ impl<'a> InclusiveMetricsEvaluator<'a> { }; if let Some(bound) = bound { + let bound = &*Self::promote_bound(bound, reference); if cmp_fn(bound, datum) { return ROWS_MIGHT_MATCH; } @@ -129,6 +132,19 @@ 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. 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) + } } impl BoundPredicateVisitor for InclusiveMetricsEvaluator<'_> { @@ -259,6 +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); if lower_bound.is_nan() { // NaN indicates unreliable bounds. // See the InclusiveMetricsEvaluator docs for more. @@ -269,6 +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); if upper_bound.is_nan() { // NaN indicates unreliable bounds. // See the InclusiveMetricsEvaluator docs for more. @@ -438,6 +456,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. return ROWS_MIGHT_MATCH; @@ -450,6 +469,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. return ROWS_MIGHT_MATCH; @@ -521,6 +541,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 = diff --git a/crates/iceberg/src/scan/context.rs b/crates/iceberg/src/scan/context.rs index 211ebf8f0c..0e8b0aa503 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}; @@ -73,6 +74,28 @@ pub(crate) struct ManifestEntryContext { pub sort_order: Option, } +/// 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!( + ( + 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 { /// Consumes this [`ManifestFileContext`], fetching its Manifest from FileIO and then /// streaming its constituent [`ManifestEntries`] to the channel provided in the context @@ -95,7 +118,33 @@ 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<(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, written)) + } + None => None, + }; + for manifest_entry in manifest.entries() { + let manifest_entry = match promotion_target.as_ref() { + Some((current, written)) => { + let mut entry = manifest_entry.as_ref().clone(); + entry.data_file.partition = + entry.data_file.promoted_partition(current, written)?; + 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 +157,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..7682a0a2e3 100644 --- a/crates/iceberg/src/scan/mod.rs +++ b/crates/iceberg/src/scan/mod.rs @@ -1195,6 +1195,223 @@ 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(false).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" + ); + 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 + .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); + + 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] + 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); + } + + /// 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 94568ab44e..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, 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,71 @@ pub struct DataFile { pub(crate) content_size_in_bytes: Option, } +fn promote_partition_value( + value: Option<&Literal>, + current_field: &NestedFieldRef, + written_field: &NestedFieldRef, +) -> Option { + match value { + Some(Literal::Primitive(primitive)) => { + 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, + }; + promoted.map(Literal::Primitive).or_else(|| value.cloned()) + } + other => other.cloned(), + } +} + impl DataFile { + /// 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 `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(), + current.fields().len() + )); + } + + Ok(self + .partition + .iter() + .zip(current.fields()) + .zip(written.fields()) + .map(|((value, current_field), written_field)| { + promote_partition_value(value, current_field, written_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 +483,12 @@ impl std::fmt::Display for DataFileFormat { #[cfg(test)] mod test { - use crate::spec::DataContentType; + use crate::ErrorKind; + 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 +498,161 @@ 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 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) + .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(¤t, &written).unwrap(), + Struct::from_iter([Some(Literal::long(7)), Some(Literal::double(1.5))]) + ); + } + + #[test] + 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) + .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(¤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 + ); + } + + #[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, &partition_type) + .unwrap(), + 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, &one_field) + .unwrap_err() + .kind(), + ErrorKind::DataInvalid + ); + + let shorter = data_file(vec![Some(Literal::int(7))]); + assert_eq!( + shorter + .promoted_partition(&two_fields, &two_fields) + .unwrap_err() + .kind(), + ErrorKind::DataInvalid + ); + } } diff --git a/crates/iceberg/src/spec/values/datum.rs b/crates/iceberg/src/spec/values/datum.rs index 09c1e88313..5c4339d8ba 100644 --- a/crates/iceberg/src/spec/values/datum.rs +++ b/crates/iceberg/src/spec/values/datum.rs @@ -1229,6 +1229,17 @@ 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. 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 /// that can be a NaN, and that it's value is NaN pub fn is_nan(&self) -> bool { 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..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; @@ -604,6 +605,425 @@ 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. 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()) + .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.clone(), + 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, 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()), + delete_schema, + 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(delete_partition) + .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 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 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 {