diff --git a/datafusion_iceberg/src/pruning_statistics.rs b/datafusion_iceberg/src/pruning_statistics.rs index 521b92e5..ed58c9c3 100644 --- a/datafusion_iceberg/src/pruning_statistics.rs +++ b/datafusion_iceberg/src/pruning_statistics.rs @@ -48,16 +48,19 @@ use iceberg_rust::{ pub(crate) struct PruneManifests<'table, 'manifests> { partition_fields: &'table [BoundPartitionField<'table>], + partition_spec_id: i32, files: &'manifests [ManifestListEntry], } impl<'table, 'manifests> PruneManifests<'table, 'manifests> { pub(crate) fn new( partition_fields: &'table [BoundPartitionField<'table>], + partition_spec_id: i32, files: &'manifests [ManifestListEntry], ) -> Self { Self { partition_fields, + partition_spec_id, files, } } @@ -74,13 +77,14 @@ impl PruningStatistics for PruneManifests<'_, '_> { .field_type() .tranform(partition_field.transform()) .ok()?; - let min_values = self.files.iter().filter_map(|manifest| { - manifest.partitions.as_ref().and_then(|partitions| { - partitions[index] - .lower_bound - .as_ref() - .map(|min| Some(min.clone().into_any())) - }) + let min_values = self.files.iter().map(|manifest| { + (manifest.partition_spec_id == self.partition_spec_id) + .then_some(manifest) + .and_then(|manifest| manifest.partitions.as_ref()) + .and_then(|partitions| partitions.get(index)) + .and_then(|partition| partition.lower_bound.as_ref()) + .and_then(|min| min.clone().cast(&data_type).ok()) + .map(Value::into_any) }); any_iter_to_array(min_values, &(&data_type).try_into().ok()?).ok() } @@ -94,13 +98,14 @@ impl PruningStatistics for PruneManifests<'_, '_> { .field_type() .tranform(partition_field.transform()) .ok()?; - let max_values = self.files.iter().filter_map(|manifest| { - manifest.partitions.as_ref().and_then(|partitions| { - partitions[index] - .upper_bound - .as_ref() - .map(|max| Some(max.clone().into_any())) - }) + let max_values = self.files.iter().map(|manifest| { + (manifest.partition_spec_id == self.partition_spec_id) + .then_some(manifest) + .and_then(|manifest| manifest.partitions.as_ref()) + .and_then(|partitions| partitions.get(index)) + .and_then(|partition| partition.upper_bound.as_ref()) + .and_then(|max| max.clone().cast(&data_type).ok()) + .map(Value::into_any) }); any_iter_to_array(max_values, &(&data_type).try_into().ok()?).ok() } @@ -113,14 +118,12 @@ impl PruningStatistics for PruneManifests<'_, '_> { .iter() .enumerate() .find(|(_, field)| field.source_name() == column.name())?; - let contains_null = self.files.iter().filter_map(|manifest| { - manifest.partitions.as_ref().map(|partitions| { - if !partitions[index].contains_null { - Some(0) - } else { - None - } - }) + let contains_null = self.files.iter().map(|manifest| { + (manifest.partition_spec_id == self.partition_spec_id) + .then_some(manifest) + .and_then(|manifest| manifest.partitions.as_ref()) + .and_then(|partitions| partitions.get(index)) + .and_then(|partition| (!partition.contains_null).then_some(0)) }); ScalarValue::iter_to_array(contains_null.map(ScalarValue::Int32)).ok() } @@ -169,7 +172,8 @@ impl<'table, 'manifests> PruneDataFiles<'table, 'manifests> { impl PruningStatistics for PruneDataFiles<'_, '_> { fn min_values(&self, column: &Column) -> Option { - let column_id = self.schema.fields().get_name(&column.name)?.id; + let field = self.schema.fields().get_name(&column.name)?; + let column_id = field.id; let datatype = self .arrow_schema .field_with_name(&column.name) @@ -179,15 +183,20 @@ impl PruningStatistics for PruneDataFiles<'_, '_> { self.files .iter() .map(|manifest| match &manifest.1.data_file().lower_bounds() { - Some(map) => map - .get(&{ column_id }) - .map(|value| value.clone().into_any()), + Some(map) => map.get(&column_id).and_then(|value| { + value + .clone() + .cast(&field.field_type) + .ok() + .map(Value::into_any) + }), None => None, }); any_iter_to_array(min_values, datatype).ok() } fn max_values(&self, column: &Column) -> Option { - let column_id = self.schema.fields().get_name(&column.name)?.id; + let field = self.schema.fields().get_name(&column.name)?; + let column_id = field.id; let datatype = self .arrow_schema .field_with_name(&column.name) @@ -197,9 +206,13 @@ impl PruningStatistics for PruneDataFiles<'_, '_> { self.files .iter() .map(|manifest| match &manifest.1.data_file().upper_bounds() { - Some(map) => map - .get(&{ column_id }) - .map(|value| value.clone().into_any()), + Some(map) => map.get(&column_id).and_then(|value| { + value + .clone() + .cast(&field.field_type) + .ok() + .map(Value::into_any) + }), None => None, }); any_iter_to_array(max_values, datatype).ok() @@ -510,13 +523,120 @@ fn value_to_scalarvalue(value: Value) -> Result { mod tests { use super::*; use datafusion::arrow::array::{ - Array, Date32Array, Decimal128Array, TimestampMicrosecondArray, + Array, Date32Array, Decimal128Array, Int64Array, TimestampMicrosecondArray, }; use datafusion::arrow::datatypes::Field; use datafusion::common::config::ConfigOptions; use iceberg_rust::spec::decimal::decimal_from_i128_with_scale; + use iceberg_rust::spec::{ + manifest::{Content, DataFile, FileFormat, Status}, + manifest_list::{Content as ManifestContent, FieldSummary}, + partition::PartitionField, + table_metadata::FormatVersion, + types::{PrimitiveType, StructField, StructType, Type}, + values::Struct, + }; use std::sync::Arc; + #[test] + fn manifest_pruning_does_not_compare_different_partition_specs() { + let source = StructField::new(2, "b", false, Type::Primitive(PrimitiveType::Long), None); + let partition = PartitionField::new(2, 1000, "b", Transform::Identity); + let fields = [BoundPartitionField::new(&partition, &source)]; + let entry = |spec_id, lower: Value| ManifestListEntry { + format_version: FormatVersion::V2, + manifest_path: format!("/{spec_id}.avro"), + manifest_length: 1, + partition_spec_id: spec_id, + content: ManifestContent::Data, + sequence_number: 1, + min_sequence_number: 1, + added_snapshot_id: 1, + added_files_count: Some(1), + existing_files_count: Some(0), + deleted_files_count: Some(0), + added_rows_count: Some(1), + existing_rows_count: Some(0), + deleted_rows_count: Some(0), + partitions: Some(vec![FieldSummary { + contains_null: false, + contains_nan: None, + lower_bound: Some(lower.clone()), + upper_bound: Some(lower), + }]), + key_metadata: None, + first_row_id: None, + }; + let manifests = vec![ + entry(0, Value::Int(1000)), + entry(1, Value::LongInt(3)), + entry(1, Value::Int(-42)), + ]; + let pruning = PruneManifests::new(&fields, 1, &manifests); + let minimums = pruning.min_values(&Column::from_name("b")).unwrap(); + let minimums = minimums.as_any().downcast_ref::().unwrap(); + assert_eq!(minimums.len(), 3); + assert!(minimums.is_null(0)); + assert_eq!(minimums.value(1), 3); + assert_eq!(minimums.value(2), -42); + let maximums = pruning.max_values(&Column::from_name("b")).unwrap(); + let maximums = maximums.as_any().downcast_ref::().unwrap(); + assert!(maximums.is_null(0)); + assert_eq!(maximums.value(2), -42); + let null_counts = pruning.null_counts(&Column::from_name("b")).unwrap(); + assert!(null_counts.is_null(0)); + } + + #[test] + fn data_file_pruning_promotes_old_numeric_bounds_and_keeps_unknowns() { + let schema = Schema::from_struct_type( + StructType::new(vec![StructField::new( + 1, + "id", + false, + Type::Primitive(PrimitiveType::Long), + None, + )]), + 1, + None, + ); + let arrow_schema = ArrowSchema::new(vec![Field::new("id", DataType::Int64, true)]); + let entry = |lower: Value| { + let file = DataFile::builder() + .with_content(Content::Data) + .with_file_path("/data.parquet".into()) + .with_file_format(FileFormat::Parquet) + .with_partition(Struct::from_iter(Vec::<(String, Option)>::new())) + .with_record_count(1) + .with_file_size_in_bytes(1) + .with_column_sizes(None) + .with_value_counts(None) + .with_null_value_counts(None) + .with_nan_value_counts(None) + .with_distinct_counts(None) + .with_lower_bounds(Some(std::collections::HashMap::from([(1, lower)]))) + .with_upper_bounds(None) + .build() + .unwrap(); + ManifestEntry::builder() + .with_format_version(FormatVersion::V2) + .with_status(Status::Added) + .with_data_file(file) + .build() + .unwrap() + }; + let files = vec![ + ("old".into(), entry(Value::Int(-42))), + ("unknown".into(), entry(Value::String("invalid".into()))), + ]; + let pruning = PruneDataFiles::new(&schema, &arrow_schema, &files); + let min_values = pruning.min_values(&Column::from_name("id")).unwrap(); + let min_values = min_values.as_any().downcast_ref::().unwrap(); + assert_eq!(min_values.len(), 2); + assert_eq!(min_values.value(0), -42); + assert!(min_values.is_null(1)); + } + /// Helper: invoke `DateTransform` directly with a transform name and scalar value. fn invoke_date_transform( transform_name: &str, diff --git a/datafusion_iceberg/src/statistics.rs b/datafusion_iceberg/src/statistics.rs index d1094356..2325cc1a 100644 --- a/datafusion_iceberg/src/statistics.rs +++ b/datafusion_iceberg/src/statistics.rs @@ -119,6 +119,7 @@ pub(crate) fn manifest_statistics(schema: &Schema, manifest: &ManifestEntry) -> } fn convert_value_to_scalar_value(value: Value, field_type: &Type) -> Result { + let value = value.cast(field_type)?; match value { Value::Boolean(b) => Ok(ScalarValue::Boolean(Some(b))), Value::Int(i) => Ok(ScalarValue::Int32(Some(i))), @@ -231,4 +232,38 @@ mod tests { assert_eq!(scalar, ScalarValue::Decimal128(Some(mantissa), 38, 0)); } + + #[test] + fn promotes_numeric_bounds_to_scan_schema() { + assert_eq!( + convert_value_to_scalar_value(Value::Int(-42), &Type::Primitive(PrimitiveType::Long)) + .unwrap(), + ScalarValue::Int64(Some(-42)) + ); + assert_eq!( + convert_value_to_scalar_value( + Value::try_from_bytes( + &1.25_f32.to_le_bytes(), + &Type::Primitive(PrimitiveType::Float), + ) + .unwrap(), + &Type::Primitive(PrimitiveType::Double), + ) + .unwrap(), + ScalarValue::Float64(Some(1.25)) + ); + } + + #[test] + fn preserves_decimal_bound_with_declared_precision() { + let field_type = Type::Primitive(PrimitiveType::Decimal { + precision: 10, + scale: 2, + }); + let value = Value::Decimal(decimal_from_i128_with_scale(1234, 2).unwrap()); + assert_eq!( + convert_value_to_scalar_value(value, &field_type).unwrap(), + ScalarValue::Decimal128(Some(1234), 10, 2) + ); + } } diff --git a/datafusion_iceberg/src/table/mod.rs b/datafusion_iceberg/src/table/mod.rs index e46c7144..95b8b1ff 100644 --- a/datafusion_iceberg/src/table/mod.rs +++ b/datafusion_iceberg/src/table/mod.rs @@ -759,8 +759,11 @@ async fn table_scan( let pruning_predicate = PruningPredicateBuilder::new() .with_file_schema(partition_schema.clone()) .try_build(physical_partition_predicate)?; - let manifests_to_prune = - pruning_predicate.prune(&PruneManifests::new(partition_fields, &manifests))?; + let manifests_to_prune = pruning_predicate.prune(&PruneManifests::new( + partition_fields, + table.metadata().default_spec_id, + &manifests, + ))?; table .datafiles( diff --git a/iceberg-rust-spec/src/spec/manifest.rs b/iceberg-rust-spec/src/spec/manifest.rs index afc2de80..b279d75e 100644 --- a/iceberg-rust-spec/src/spec/manifest.rs +++ b/iceberg-rust-spec/src/spec/manifest.rs @@ -671,12 +671,27 @@ impl AvroMap { .into_iter() .filter_map(|(k, v)| { let field = schema.get(k as usize)?; - Some(Value::try_from_bytes(&v, &field.field_type).map(|val| (k, val))) + Some(decode_bound(&v, &field.field_type).map(|val| (k, val))) }) .collect() } } +pub(crate) fn decode_bound(bytes: &[u8], data_type: &Type) -> Result { + match (data_type, bytes.len()) { + (Type::Primitive(PrimitiveType::Long), 4) => Ok(Value::LongInt(i64::from( + i32::from_le_bytes(bytes.try_into()?), + ))), + (Type::Primitive(PrimitiveType::Double), 4) => Ok(Value::Double( + ordered_float::OrderedFloat(f64::from(f32::from_le_bytes(bytes.try_into()?))), + )), + (Type::Primitive(PrimitiveType::Timestamp), 4) => { + Value::Date(i32::from_le_bytes(bytes.try_into()?)).cast(data_type) + } + _ => Value::try_from_bytes(bytes, data_type), + } +} + impl From> for AvroMap { fn from(value: HashMap) -> Self { AvroMap(HashMap::from_iter( @@ -2181,6 +2196,92 @@ mod tests { ); } + #[test] + fn numeric_bounds_decode_against_manifest_schema() { + let old_schema = StructType::new(vec![ + StructField::new(1, "id", false, Type::Primitive(PrimitiveType::Int), None), + StructField::new( + 2, + "score", + false, + Type::Primitive(PrimitiveType::Float), + None, + ), + ]); + let promoted_schema = StructType::new(vec![ + StructField::new(1, "id", false, Type::Primitive(PrimitiveType::Long), None), + StructField::new( + 2, + "score", + false, + Type::Primitive(PrimitiveType::Double), + None, + ), + ]); + let bounds: AvroMap = HashMap::from([ + (1, Value::Int(-42)), + (2, Value::Float(ordered_float::OrderedFloat(1.25))), + ]) + .into(); + assert_eq!(bounds.0[&1].len(), 4); + assert_eq!(bounds.0[&2].len(), 4); + + let old = bounds.clone().into_value_map(&old_schema).unwrap(); + assert_eq!(old.get(&1), Some(&Value::Int(-42))); + assert_eq!( + old.get(&2), + Some(&Value::Float(ordered_float::OrderedFloat(1.25))) + ); + let round_trip: AvroMap = old.into(); + assert_eq!(round_trip.0, bounds.0); + + let promoted = bounds.into_value_map(&promoted_schema).unwrap(); + assert_eq!(promoted.get(&1), Some(&Value::LongInt(-42))); + assert_eq!( + promoted.get(&2), + Some(&Value::Double(ordered_float::OrderedFloat(1.25))) + ); + } + + #[test] + fn numeric_bounds_reject_unknown_width_and_skip_unknown_fields() { + let schema = StructType::new(vec![StructField::new( + 1, + "id", + false, + Type::Primitive(PrimitiveType::Long), + None, + )]); + let bounds = AvroMap(HashMap::from([ + (1, ByteBuf::from((-7_i32).to_le_bytes())), + (999, ByteBuf::from([0_u8, 1_u8, 2_u8])), + ])); + let decoded = bounds.into_value_map(&schema).unwrap(); + assert_eq!(decoded.len(), 1); + assert_eq!(decoded.get(&1), Some(&Value::LongInt(-7))); + assert!(decode_bound(&[0, 1, 2], &Type::Primitive(PrimitiveType::Long)).is_err()); + assert!(decode_bound(&[0, 1, 2], &Type::Primitive(PrimitiveType::Double)).is_err()); + assert!(decode_bound(&[0, 1, 2], &Type::Primitive(PrimitiveType::Timestamp)).is_err()); + } + + #[test] + fn date_bound_promotes_to_microsecond_timestamp() { + let days = 19_000_i32; + assert_eq!( + decode_bound( + &days.to_le_bytes(), + &Type::Primitive(PrimitiveType::Timestamp) + ) + .unwrap(), + Value::Timestamp(i64::from(days) * 86_400_000_000) + ); + assert!(decode_bound( + &i32::MAX.to_le_bytes(), + &Type::Primitive(PrimitiveType::Timestamp), + ) + .is_err()); + } + fn row_id_entry( status: Status, content: Content, diff --git a/iceberg-rust-spec/src/spec/manifest_list.rs b/iceberg-rust-spec/src/spec/manifest_list.rs index 91112c6e..2a5beb68 100644 --- a/iceberg-rust-spec/src/spec/manifest_list.rs +++ b/iceberg-rust-spec/src/spec/manifest_list.rs @@ -379,11 +379,7 @@ impl ManifestListEntry { } } - candidates.ok_or_else(|| { - Error::NotFound(format!( - "Schema containing all source fields for partition spec {partition_spec_id}" - )) - }) + Ok(candidates.unwrap_or_default()) } pub fn try_from_enum( @@ -438,6 +434,7 @@ impl ManifestListEntry { deleted_rows_count: Some(entry.deleted_rows_count), partitions: entry .partitions + .filter(|partitions| partitions.len() == partition_types.len()) .map(|v| { v.into_iter() .zip(partition_types.iter()) @@ -485,6 +482,7 @@ impl ManifestListEntry { deleted_rows_count: Some(entry.deleted_rows_count), partitions: entry .partitions + .filter(|partitions| partitions.len() == partition_types.len()) .map(|v| { v.into_iter() .zip(partition_types.iter()) @@ -532,6 +530,7 @@ impl ManifestListEntry { deleted_rows_count: entry.deleted_rows_count, partitions: entry .partitions + .filter(|partitions| partitions.len() == partition_types.len()) .map(|v| { v.into_iter() .zip(partition_types.iter()) @@ -632,7 +631,7 @@ impl FieldSummary { .ok_or_else(|| Error::InvalidFormat("partition field type candidates".to_string()))?; let mut last_error = None; for data_type in data_type_candidates { - match Value::try_from_bytes(bytes, data_type) { + match super::manifest::decode_bound(bytes, data_type) { Ok(value) if data_type == target_type => return Ok(value), Ok(value) => match value.promote_iceberg(data_type, target_type) { Ok(value) => return Ok(value), @@ -1369,12 +1368,27 @@ mod tests { Some(Value::Int(42)) ); } + + let mut metadata = table_metadata(HashMap::new()); + metadata.schemas.remove(&1); + let record = apache_avro::Reader::new(&bytes[..]) + .unwrap() + .next() + .unwrap(); + let decoded = avro_value_to_manifest_list_entry_for_format_version( + record, + &metadata, + FormatVersion::V1, + ) + .unwrap(); + assert!(decoded.partitions.is_none()); } fn decode_historical_bound_after_snapshot_expiration( source_type: PrimitiveType, target_type: PrimitiveType, value: Value, + retain_source_schema: bool, ) -> Value { let schema = |schema_id, primitive_type| { Schema::builder() @@ -1391,7 +1405,7 @@ mod tests { .build() .unwrap() }; - let metadata = TableMetadataBuilder::default() + let mut metadata = TableMetadataBuilder::default() .format_version(FormatVersion::V3) .location("/") .current_schema_id(2) @@ -1415,6 +1429,9 @@ mod tests { )])) .build() .unwrap(); + if !retain_source_schema { + metadata.schemas.remove(&1); + } let entry = ManifestListEntry { format_version: FormatVersion::V1, manifest_path: "expired-snapshot-manifest.avro".to_string(), @@ -1463,6 +1480,16 @@ mod tests { PrimitiveType::Int, PrimitiveType::Long, Value::Int(42), + true, + ), + Value::LongInt(42) + ); + assert_eq!( + decode_historical_bound_after_snapshot_expiration( + PrimitiveType::Int, + PrimitiveType::Long, + Value::Int(42), + false, ), Value::LongInt(42) ); @@ -1477,11 +1504,21 @@ mod tests { &Type::Primitive(PrimitiveType::Double), ) .unwrap(); + assert_eq!( + decode_historical_bound_after_snapshot_expiration( + PrimitiveType::Float, + PrimitiveType::Double, + float.clone(), + true, + ), + expected_double + ); assert_eq!( decode_historical_bound_after_snapshot_expiration( PrimitiveType::Float, PrimitiveType::Double, float, + false, ), expected_double ); @@ -1498,6 +1535,7 @@ mod tests { scale: 2, }, decimal.clone(), + true, ), decimal ); diff --git a/iceberg-rust-spec/src/spec/values.rs b/iceberg-rust-spec/src/spec/values.rs index 45a0c824..7dbe84bd 100644 --- a/iceberg-rust-spec/src/spec/values.rs +++ b/iceberg-rust-spec/src/spec/values.rs @@ -45,13 +45,16 @@ use uuid::Uuid; use crate::error::Error; use super::{ - decimal::{decimal_from_i128_with_scale, decimal_scale, decimal_to_be_bytes_min, Decimal}, + decimal::{ + decimal_from_i128_with_scale, decimal_mantissa, decimal_scale, decimal_to_be_bytes_min, + Decimal, + }, partition::{PartitionField, Transform}, types::{PrimitiveType, StructType, Type}, }; #[cfg(test)] -use super::decimal::{decimal_from_str_exact, decimal_mantissa}; +use super::decimal::decimal_from_str_exact; pub static YEARS_BEFORE_UNIX_EPOCH: i32 = 1970; @@ -788,6 +791,25 @@ impl Value { (Value::Int(input), Type::Primitive(PrimitiveType::Long)) => { Ok(Value::LongInt(input as i64)) } + (Value::Float(input), Type::Primitive(PrimitiveType::Double)) => { + Ok(Value::Double(OrderedFloat(f64::from(input.0)))) + } + ( + Value::Decimal(input), + Type::Primitive(PrimitiveType::Decimal { precision, scale }), + ) if decimal_scale(&input) == *scale + && decimal_mantissa(&input)? + .unsigned_abs() + .checked_ilog10() + .unwrap_or(0) + < *precision => + { + Ok(Value::Decimal(input)) + } + (Value::Date(input), Type::Primitive(PrimitiveType::Timestamp)) => i64::from(input) + .checked_mul(86_400_000_000) + .map(Value::Timestamp) + .ok_or_else(|| Error::InvalidFormat("date exceeds timestamp range".into())), (Value::Int(input), Type::Primitive(PrimitiveType::Date)) => Ok(Value::Date(input)), (Value::LongInt(input), Type::Primitive(PrimitiveType::Time)) => { Ok(Value::Time(input)) @@ -820,6 +842,11 @@ impl Value { Type::Primitive(PrimitiveType::Float), Type::Primitive(PrimitiveType::Double), ) => Ok(Value::Double(OrderedFloat(f64::from(input.0)))), + ( + Value::Date(input), + Type::Primitive(PrimitiveType::Date), + Type::Primitive(PrimitiveType::Timestamp), + ) => Value::Date(input).cast(target_type), ( value @ Value::Decimal(_), Type::Primitive(PrimitiveType::Decimal { @@ -1862,11 +1889,11 @@ mod tests { promoted.get("id_partition"), Some(&Some(Value::Double(OrderedFloat(f64::from(34.11_f32))))) ); - assert!( + assert_eq!( Value::Float(OrderedFloat(34.11)) .cast(&Type::Primitive(PrimitiveType::Double)) - .is_err(), - "partition promotion must not widen the generic Value::cast API" + .unwrap(), + Value::Double(OrderedFloat(f64::from(34.11_f32))) ); } @@ -1999,12 +2026,59 @@ mod tests { } #[test] - fn test_float_value_rejects_every_non_float_target_type() { + fn test_float_value_rejects_targets_outside_double() { let value = Value::Float(OrderedFloat(34.11_f32)); - let targets = all_other_primitive_types(&[PrimitiveType::Float]); + let targets = all_other_primitive_types(&[PrimitiveType::Float, PrimitiveType::Double]); assert_invalid_casts(&value, &targets); } + #[test] + fn test_numeric_value_promotions() { + assert_eq!( + Value::Int(-42) + .cast(&Type::Primitive(PrimitiveType::Long)) + .unwrap(), + Value::LongInt(-42) + ); + assert_eq!( + Value::Float(OrderedFloat(1.25)) + .cast(&Type::Primitive(PrimitiveType::Double)) + .unwrap(), + Value::Double(OrderedFloat(1.25)) + ); + assert_eq!( + Value::Date(19_000) + .cast(&Type::Primitive(PrimitiveType::Timestamp)) + .unwrap(), + Value::Timestamp(19_000_i64 * 86_400_000_000) + ); + assert!(Value::Date(i32::MAX) + .cast(&Type::Primitive(PrimitiveType::Timestamp)) + .is_err()); + } + + #[test] + fn decimal_cast_rejects_value_exceeding_target_precision() { + let input = Value::Decimal(decimal_from_i128_with_scale(123_456, 0).unwrap()); + assert!(input + .clone() + .cast(&Type::Primitive(PrimitiveType::Decimal { + precision: 3, + scale: 0, + })) + .is_err()); + assert_eq!( + input + .clone() + .cast(&Type::Primitive(PrimitiveType::Decimal { + precision: 6, + scale: 0, + })) + .unwrap(), + input + ); + } + #[test] fn test_double_value_rejects_every_non_double_target_type() { let value = Value::Double(OrderedFloat(34.11_f64)); @@ -2013,9 +2087,9 @@ mod tests { } #[test] - fn test_date_value_rejects_every_non_date_target_type() { + fn test_date_value_rejects_targets_outside_timestamp() { let value = Value::Date(17396); // 2017-08-18 - let targets = all_other_primitive_types(&[PrimitiveType::Date]); + let targets = all_other_primitive_types(&[PrimitiveType::Date, PrimitiveType::Timestamp]); assert_invalid_casts(&value, &targets); } diff --git a/iceberg-rust/src/table/manifest.rs b/iceberg-rust/src/table/manifest.rs index 96e3dbb1..a3d5cf40 100644 --- a/iceberg-rust/src/table/manifest.rs +++ b/iceberg-rust/src/table/manifest.rs @@ -346,7 +346,6 @@ impl<'schema, 'metadata> ManifestWriter<'schema, 'metadata> { table_metadata: &'metadata TableMetadata, ) -> Result { let inherited_snapshot_id = manifest.added_snapshot_id; - let current_schema = table_metadata.current_schema()?; let mut writer = AvroWriter::new(schema, Vec::new()); let mut existing_files = 0; let mut existing_rows = 0; @@ -402,14 +401,11 @@ impl<'schema, 'metadata> ManifestWriter<'schema, 'metadata> { }, )?; - writer.extend(manifest_reader.filter_map(|entry| { - let mut entry = entry.ok()?; + for entry in manifest_reader { + let mut entry = entry?; if *entry.status() == Status::Deleted { - return None; + continue; } - entry - .data_file_mut() - .promote_bounds_to_schema(current_schema); *entry.status_mut() = Status::Existing; if entry.sequence_number().is_none() { *entry.sequence_number_mut() = Some(manifest.sequence_number); @@ -419,8 +415,8 @@ impl<'schema, 'metadata> ManifestWriter<'schema, 'metadata> { } existing_files += 1; existing_rows += entry.data_file().record_count(); - to_value(entry).ok() - }))?; + writer.append(to_value(entry)?)?; + } manifest.sequence_number = table_metadata.last_sequence_number + 1; manifest.added_snapshot_id = snapshot_id; @@ -493,7 +489,6 @@ impl<'schema, 'metadata> ManifestWriter<'schema, 'metadata> { table_metadata: &'metadata TableMetadata, ) -> Result<(Self, FilteredManifestStats), Error> { let inherited_snapshot_id = manifest.added_snapshot_id; - let current_schema = table_metadata.current_schema()?; let manifest_reader = ManifestReader::new(bytes)?; let mut next_inherited_row_id = manifest.first_row_id; @@ -564,9 +559,6 @@ impl<'schema, 'metadata> ManifestWriter<'schema, 'metadata> { &mut next_inherited_row_id, entry.data_file_mut(), )?; - entry - .data_file_mut() - .promote_bounds_to_schema(current_schema); if entry.sequence_number().is_none() { *entry.sequence_number_mut() = Some(manifest.sequence_number); }