From 4a9aa75641686c997d9d5dc60697cd20f051e7f2 Mon Sep 17 00:00:00 2001 From: osipovartem Date: Thu, 15 Oct 2026 06:55:49 +0300 Subject: [PATCH 1/2] Preserve source columns across Iceberg partition spec changes --- datafusion_iceberg/src/pruning_statistics.rs | 84 ++- datafusion_iceberg/src/table/mod.rs | 696 ++++++++++++++++--- datafusion_iceberg/tests/equality_delete.rs | 45 ++ 3 files changed, 692 insertions(+), 133 deletions(-) diff --git a/datafusion_iceberg/src/pruning_statistics.rs b/datafusion_iceberg/src/pruning_statistics.rs index e53b112d..24906645 100644 --- a/datafusion_iceberg/src/pruning_statistics.rs +++ b/datafusion_iceberg/src/pruning_statistics.rs @@ -262,6 +262,18 @@ impl<'table, 'manifests> PruneDataFiles<'table, 'manifests> { files, } } + + fn can_prune_column(entry: &ManifestEntry, column_id: i32) -> bool { + match entry.data_file().content() { + Content::Data => true, + Content::EqualityDeletes => entry + .data_file() + .equality_ids() + .as_ref() + .is_some_and(|ids| ids.contains(&column_id)), + Content::PositionDeletes => false, + } + } } impl PruningStatistics for PruneDataFiles<'_, '_> { @@ -273,19 +285,21 @@ impl PruningStatistics for PruneDataFiles<'_, '_> { .field_with_name(&column.name) .ok()? .data_type(); - let min_values = - self.files - .iter() - .map(|manifest| match &manifest.1.data_file().lower_bounds() { - Some(map) => map.get(&column_id).and_then(|value| { - value - .clone() - .cast(&field.field_type) - .ok() - .map(Value::into_any) - }), - None => None, - }); + let min_values = self.files.iter().map(|manifest| { + if !Self::can_prune_column(&manifest.1, column_id) { + return None; + } + match &manifest.1.data_file().lower_bounds() { + 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 { @@ -296,19 +310,21 @@ impl PruningStatistics for PruneDataFiles<'_, '_> { .field_with_name(&column.name) .ok()? .data_type(); - let max_values = - self.files - .iter() - .map(|manifest| match &manifest.1.data_file().upper_bounds() { - Some(map) => map.get(&column_id).and_then(|value| { - value - .clone() - .cast(&field.field_type) - .ok() - .map(Value::into_any) - }), - None => None, - }); + let max_values = self.files.iter().map(|manifest| { + if !Self::can_prune_column(&manifest.1, column_id) { + return None; + } + match &manifest.1.data_file().upper_bounds() { + 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() } fn num_containers(&self) -> usize { @@ -316,13 +332,15 @@ impl PruningStatistics for PruneDataFiles<'_, '_> { } fn null_counts(&self, column: &Column) -> Option { let column_id = self.schema.fields().get_name(&column.name)?.id; - let null_counts = - self.files.iter().map( - |manifest| match &manifest.1.data_file().null_value_counts() { - Some(map) => map.get(&{ column_id }).copied(), - None => None, - }, - ); + let null_counts = self.files.iter().map(|manifest| { + if !Self::can_prune_column(&manifest.1, column_id) { + return None; + } + match &manifest.1.data_file().null_value_counts() { + Some(map) => map.get(&{ column_id }).copied(), + None => None, + } + }); ScalarValue::iter_to_array(null_counts.map(ScalarValue::Int64)).ok() } fn contained( diff --git a/datafusion_iceberg/src/table/mod.rs b/datafusion_iceberg/src/table/mod.rs index d400799c..3c705531 100644 --- a/datafusion_iceberg/src/table/mod.rs +++ b/datafusion_iceberg/src/table/mod.rs @@ -29,7 +29,6 @@ use lru::LruCache; use object_store::path::Path; use object_store::ObjectStoreExt; use object_store::{ObjectMeta, ObjectStore}; -use std::collections::BTreeMap; use std::thread::available_parallelism; use std::{ collections::{HashMap, HashSet}, @@ -108,7 +107,8 @@ use datafusion::{ }; use iceberg_rust::spec::{ manifest::DataFile, - partition::Transform, + manifest_list::{Content as ManifestContent, ManifestListEntry}, + partition::{PartitionSpec, Transform}, schema::Schema, sort::{NullOrder, SortDirection, SortOrder}, table_metadata::FormatVersion, @@ -723,7 +723,7 @@ async fn table_scan( }) .transpose()?; - let mut table_partition_cols = datafusion_partition_columns(partition_fields)?; + let pruning_partition_cols = datafusion_partition_columns(partition_fields)?; let mut file_schema_builder = SchemaBuilder::from(TryInto::::try_into(schema.fields()).unwrap()); @@ -752,23 +752,6 @@ async fn table_scan( .cloned() .unwrap_or((0..arrow_schema.fields().len()).collect_vec()); - let mut data_file_path_scan_index = None; - if enable_data_file_path_column { - data_file_path_scan_index = Some(file_schema.fields().len() + table_partition_cols.len()); - table_partition_cols.push(Field::new(DATA_FILE_PATH_COLUMN, DataType::Utf8, false)); - } - - let manifest_file_path_partition_index = enable_manifest_file_path_column.then(|| { - let index = table_partition_cols.len(); - table_partition_cols.push(Field::new(MANIFEST_FILE_PATH_COLUMN, DataType::Utf8, false)); - index - }); - - // All files have to be grouped according to their partition values. This is done by using a HashMap with the partition values as the key. - // This way data files with the same partition value are mapped to the same vector. - let mut data_file_groups: HashMap> = HashMap::new(); - let mut delete_file_groups: HashMap = HashMap::new(); - let manifests = if scan_changed_data_files { table .change_manifests(snapshot_range.0, snapshot_range.1) @@ -786,11 +769,36 @@ async fn table_scan( ) }); + // Partition tuples in historical manifests follow their own spec. They cannot be + // exposed as current-spec columns: a bucket value would replace the source field. + // The mixed-spec path reads source fields from Parquet while retaining partition + // metadata for safe pruning of manifests written under the current spec. + let include_partition_values = !has_historical_data_manifests( + &manifests, + table.metadata().default_spec_id, + scan_changed_data_files, + ); + let mut table_partition_cols = if include_partition_values { + pruning_partition_cols.clone() + } else { + Vec::new() + }; + let mut data_file_path_scan_index = None; + if enable_data_file_path_column { + data_file_path_scan_index = Some(file_schema.fields().len() + table_partition_cols.len()); + table_partition_cols.push(Field::new(DATA_FILE_PATH_COLUMN, DataType::Utf8, false)); + } + let manifest_file_path_partition_index = enable_manifest_file_path_column.then(|| { + let index = table_partition_cols.len(); + table_partition_cols.push(Field::new(MANIFEST_FILE_PATH_COLUMN, DataType::Utf8, false)); + index + }); + // Prune data & delete file and insert them into the according map let (content_file_iter, mut statistics) = if let Some(physical_predicate) = physical_predicate.clone() { - let partition_schema = Arc::new(ArrowSchema::new(table_partition_cols.clone())); + let partition_schema = Arc::new(ArrowSchema::new(pruning_partition_cols)); let partition_predicates = conjunction( filters .iter() @@ -930,71 +938,16 @@ async fn table_scan( (itertools::Either::Right(iter), statistics) }; - if partition_fields.is_empty() { - let mut data_files = Vec::new(); - let mut equality_deletes = Vec::new(); - let mut position_deletes = Vec::new(); - content_file_iter - .filter(|manifest| scan_manifest_entry(&manifest.1, scan_changed_data_files)) - .for_each(|manifest| match manifest.1.data_file().content() { - Content::Data => data_files.push(manifest), - Content::EqualityDeletes => equality_deletes.push(manifest), - Content::PositionDeletes => position_deletes.push(manifest), - }); - if !data_files.is_empty() { - data_file_groups.insert( - Struct { - fields: Vec::new(), - lookup: BTreeMap::new(), - }, - data_files, - ); - } - if !equality_deletes.is_empty() || !position_deletes.is_empty() { - delete_file_groups.insert( - Struct { - fields: Vec::new(), - lookup: BTreeMap::new(), - }, - PartitionDeleteFileIndex { - equality_deletes, - position_deletes, - }, - ); - } - } else { - content_file_iter.for_each(|manifest| { - if scan_manifest_entry(&manifest.1, scan_changed_data_files) { - match manifest.1.data_file().content() { - Content::Data => { - data_file_groups - .entry(manifest.1.data_file().partition().clone()) - .or_default() - .push(manifest); - } - Content::EqualityDeletes => { - delete_file_groups - .entry(manifest.1.data_file().partition().clone()) - .or_default() - .equality_deletes - .push(manifest); - } - Content::PositionDeletes => { - delete_file_groups - .entry(manifest.1.data_file().partition().clone()) - .or_default() - .position_deletes - .push(manifest); - } - } - } - }); - } - - if excluded_data_file_paths.is_some_and(|paths| !paths.is_empty()) { - // Delete files cannot affect a partition with no retained data files. - delete_file_groups.retain(|partition, _| data_file_groups.contains_key(partition)); - } + let spec_ids_by_manifest = manifests + .iter() + .map(|manifest| (manifest.manifest_path.as_str(), manifest.partition_spec_id)) + .collect::>(); + let (mut data_file_groups, delete_file_groups) = group_content_files( + content_file_iter, + &spec_ids_by_manifest, + &table.metadata().partition_specs, + scan_changed_data_files, + )?; let has_position_deletes = delete_file_groups .values() @@ -1241,6 +1194,7 @@ async fn table_scan( schema, &data_manifest.1, last_updated_ms, + include_partition_values, include_data_file_path_column, PartitionedFileMetadataColumns { first_row_id: enable_row_lineage, @@ -1307,6 +1261,7 @@ async fn table_scan( &delete_schema, &delete_manifest.1, last_updated_ms, + include_partition_values, enable_data_file_path_column, PartitionedFileMetadataColumns::default(), manifest_path, @@ -1409,6 +1364,7 @@ async fn table_scan( schema, &x.1, last_updated_ms, + include_partition_values, include_data_file_path_column, PartitionedFileMetadataColumns { first_row_id: enable_row_lineage, @@ -1503,6 +1459,7 @@ async fn table_scan( &schema, &entry, last_updated_ms, + include_partition_values, include_data_file_path_column, PartitionedFileMetadataColumns { first_row_id: enable_row_lineage, @@ -1920,21 +1877,26 @@ fn generate_partitioned_file( schema: &Schema, manifest: &ManifestEntry, last_updated_ms: i64, + include_partition_values: bool, enable_data_file_path: bool, metadata_columns: PartitionedFileMetadataColumns, manifest_file_path: Option, ) -> Result { let manifest_statistics = manifest_statistics(schema, manifest); - let mut partition_values = manifest - .data_file() - .partition() - .iter() - .map(|x| { - x.as_ref() - .map(value_to_scalarvalue) - .unwrap_or(Ok(ScalarValue::Null)) - }) - .collect::, _>>()?; + let mut partition_values = if include_partition_values { + manifest + .data_file() + .partition() + .iter() + .map(|x| { + x.as_ref() + .map(value_to_scalarvalue) + .unwrap_or(Ok(ScalarValue::Null)) + }) + .collect::, _>>()? + } else { + Vec::new() + }; if enable_data_file_path { partition_values.push(ScalarValue::Utf8(Some( @@ -2729,6 +2691,119 @@ struct PartitionDeleteFileIndex { pub position_deletes: Vec<(ManifestPath, ManifestEntry)>, } +type PartitionKey = (i32, Struct); +type PartitionDataFileGroups = HashMap>; +type PartitionDeleteFileGroups = HashMap; + +fn has_historical_data_manifests( + manifests: &[ManifestListEntry], + current_spec_id: i32, + scan_changed_data_files: bool, +) -> bool { + manifests.iter().any(|manifest| { + manifest.content == ManifestContent::Data + && manifest.partition_spec_id != current_spec_id + && (manifest.added_files_count != Some(0) + || manifest.existing_files_count != Some(0) + || (scan_changed_data_files && manifest.deleted_files_count != Some(0))) + }) +} + +fn group_content_files( + content_files: impl IntoIterator, + spec_ids_by_manifest: &HashMap<&str, i32>, + partition_specs: &HashMap, + scan_changed_data_files: bool, +) -> Result<(PartitionDataFileGroups, PartitionDeleteFileGroups), DataFusionError> { + let mut data_file_groups = HashMap::>::new(); + let mut delete_file_groups = HashMap::::new(); + let mut global_equality_deletes = Vec::new(); + + for manifest in content_files { + if !scan_manifest_entry(&manifest.1, scan_changed_data_files) { + continue; + } + let spec_id = *spec_ids_by_manifest + .get(manifest.0.as_str()) + .ok_or_else(|| { + DataFusionError::Execution(format!( + "Missing partition spec for manifest {}", + manifest.0 + )) + })?; + let spec = partition_specs.get(&spec_id).ok_or_else(|| { + DataFusionError::Execution(format!("Unknown partition spec {spec_id}")) + })?; + let partition = (spec_id, manifest.1.data_file().partition().clone()); + match manifest.1.data_file().content() { + Content::Data => data_file_groups + .entry(partition) + .or_default() + .push(manifest), + Content::EqualityDeletes if spec.fields().is_empty() => { + global_equality_deletes.push(manifest); + } + Content::EqualityDeletes => delete_file_groups + .entry(partition) + .or_default() + .equality_deletes + .push(manifest), + Content::PositionDeletes => delete_file_groups + .entry(partition) + .or_default() + .position_deletes + .push(manifest), + } + } + + delete_file_groups.retain(|partition, _| data_file_groups.contains_key(partition)); + // Global equality deletes need an anti-join per data group. When there are no + // local deletes, coalesce groups to bound repeated delete scans while keeping + // enough independent groups to read large tables in parallel. + const MAX_GLOBAL_DELETE_APPLICATIONS: usize = 1024; + const MAX_GLOBAL_DELETE_SCAN_GROUPS: usize = 32; + if !global_equality_deletes.is_empty() + && delete_file_groups.is_empty() + && data_file_groups.len() > 1 + { + let group_count = data_file_groups + .len() + .min(MAX_GLOBAL_DELETE_SCAN_GROUPS) + .min((MAX_GLOBAL_DELETE_APPLICATIONS / global_equality_deletes.len()).max(1)); + let mut groups = Vec::<(PartitionKey, Vec<_>)>::with_capacity(group_count); + for (index, (key, files)) in data_file_groups.drain().enumerate() { + if index < group_count { + groups.push((key, files)); + } else { + groups[index % group_count].1.extend(files); + } + } + data_file_groups.extend(groups); + } + // Mixed global and local deletes cannot be coalesced without broadening + // local delete scope. Fail explicitly instead of building an unbounded plan. + if data_file_groups.len() > 1 + && global_equality_deletes + .len() + .checked_mul(data_file_groups.len()) + .is_none_or(|count| count > MAX_GLOBAL_DELETE_APPLICATIONS) + { + return Err(DataFusionError::NotImplemented(format!( + "More than {MAX_GLOBAL_DELETE_APPLICATIONS} global equality-delete applications in one scan" + ))); + } + if !global_equality_deletes.is_empty() { + for partition in data_file_groups.keys() { + delete_file_groups + .entry(partition.clone()) + .or_default() + .equality_deletes + .extend(global_equality_deletes.iter().cloned()); + } + } + Ok((data_file_groups, delete_file_groups)) +} + #[cfg(test)] mod tests { @@ -2748,6 +2823,8 @@ mod tests { catalog::tabular::Tabular, object_store::ObjectStoreBuilder, spec::{ + manifest::{Content, DataFile, FileFormat, ManifestEntry, Status}, + manifest_list::{Content as ManifestContent, ManifestListEntry}, namespace::Namespace, partition::{PartitionField, Transform}, row_lineage::{ @@ -2758,7 +2835,7 @@ mod tests { snapshot::SnapshotBuilder, types::{PrimitiveType, StructField, Type}, util, - values::Value, + values::{Struct, Value}, }, }; use iceberg_rust::{ @@ -2774,13 +2851,16 @@ mod tests { use object_store::ObjectStoreExt; use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder; - use std::sync::Arc; + use std::{ + collections::{HashMap, HashSet}, + sync::Arc, + }; use crate::{catalog::catalog::IcebergCatalog, table::fake_object_store_url, DataFusionTable}; use super::{ - position_delete_sequence_filter, DataSourceExec, ExecutionPlan, ParquetSource, - TableProvider, + group_content_files, has_historical_data_manifests, position_delete_sequence_filter, + DataSourceExec, ExecutionPlan, ParquetSource, TableProvider, }; use datafusion::datasource::physical_plan::FileScanConfig; @@ -4693,6 +4773,422 @@ mod tests { } } + #[tokio::test] + async fn historical_bucket_partition_does_not_replace_identity_source_value() { + let catalog: Arc = Arc::new( + SqlCatalog::new("sqlite://", "test", ObjectStoreBuilder::memory()) + .await + .unwrap(), + ); + let schema = Schema::builder() + .with_struct_field(StructField::new( + 1, + "id", + true, + Type::Primitive(PrimitiveType::Long), + None, + )) + .build() + .unwrap(); + let old_spec = PartitionSpec::builder() + .with_partition_field(PartitionField::new( + 1, + 1000, + "id_bucket", + Transform::Bucket(16), + )) + .build() + .unwrap(); + let table = Table::builder() + .with_name("bucket_to_identity") + .with_location("/test/bucket_to_identity") + .with_schema(schema) + .with_partition_spec(old_spec) + .build(&["test".to_owned()], catalog) + .await + .unwrap(); + let ctx = SessionContext::new(); + let provider = Arc::new(DataFusionTable::from(table)); + ctx.register_table("bucket_to_identity", provider.clone()) + .unwrap(); + ctx.sql("INSERT INTO bucket_to_identity VALUES (1), (2), (3), (4)") + .await + .unwrap() + .collect() + .await + .unwrap(); + + let committed = match provider.tabular.read().unwrap().clone() { + Tabular::Table(table) => table, + _ => panic!("expected Iceberg table"), + }; + let identifier = committed.identifier().clone(); + let catalog = committed.catalog(); + let object_store = committed.object_store(); + let mut metadata = committed.into_metadata(); + metadata.partition_specs.insert( + 1, + PartitionSpec::builder() + .with_spec_id(1) + .with_partition_field(PartitionField::new(1, 1001, "id", Transform::Identity)) + .build() + .unwrap(), + ); + metadata.default_spec_id = 1; + let evolved = Table::new(identifier, catalog, object_store, metadata) + .await + .unwrap(); + let ctx = SessionContext::new(); + ctx.register_table( + "bucket_to_identity", + Arc::new(DataFusionTable::from(evolved)), + ) + .unwrap(); + + for (sql, expected) in [ + ( + "SELECT id FROM bucket_to_identity ORDER BY id", + vec![1, 2, 3, 4], + ), + ("SELECT id FROM bucket_to_identity WHERE id = 1", vec![1]), + ] { + let batches = ctx.sql(sql).await.unwrap().collect().await.unwrap(); + let values = batches + .iter() + .flat_map(|batch| { + batch + .column(0) + .as_any() + .downcast_ref::() + .unwrap() + .values() + .iter() + .copied() + }) + .collect::>(); + assert_eq!(values, expected, "{sql}"); + } + } + + #[test] + fn mixed_spec_deletes_match_spec_and_partition_or_are_global() { + let partition = Struct::from_iter([("part".to_owned(), Some(Value::Int(4)))]); + let empty_partition = Struct::from_iter(Vec::<(String, Option)>::new()); + let entry = |name: &str, content: Content, partition: Struct| { + let file = DataFile::builder() + .with_content(content) + .with_file_path(format!("/{name}.parquet")) + .with_file_format(FileFormat::Parquet) + .with_partition(partition) + .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(None) + .with_upper_bounds(None) + .build() + .unwrap(); + ( + name.to_owned(), + ManifestEntry::builder() + .with_format_version(super::FormatVersion::V2) + .with_status(Status::Added) + .with_data_file(file) + .build() + .unwrap(), + ) + }; + let files = vec![ + entry("data-old", Content::Data, partition.clone()), + entry("data-new", Content::Data, partition.clone()), + entry("eq-old", Content::EqualityDeletes, partition.clone()), + entry("pos-new", Content::PositionDeletes, partition.clone()), + entry("eq-global", Content::EqualityDeletes, empty_partition), + ]; + let spec_ids_by_manifest = HashMap::from([ + ("data-old", 0), + ("data-new", 1), + ("eq-old", 0), + ("pos-new", 1), + ("eq-global", 2), + ]); + let specs = HashMap::from([ + ( + 0, + PartitionSpec::builder() + .with_spec_id(0) + .with_partition_field(PartitionField::new( + 1, + 1000, + "part", + Transform::Bucket(16), + )) + .build() + .unwrap(), + ), + ( + 1, + PartitionSpec::builder() + .with_spec_id(1) + .with_partition_field(PartitionField::new(1, 1001, "part", Transform::Identity)) + .build() + .unwrap(), + ), + ( + 2, + PartitionSpec::builder() + .with_spec_id(2) + .with_fields(Vec::new()) + .build() + .unwrap(), + ), + ]); + let mut global_only = files + .iter() + .filter(|(path, _)| ["data-old", "data-new", "eq-global"].contains(&path.as_str())) + .cloned() + .collect::>(); + for value in 100..138 { + global_only.push(entry( + "data-old", + Content::Data, + Struct::from_iter([("part".to_owned(), Some(Value::Int(value)))]), + )); + } + let (global_data, global_deletes) = + group_content_files(global_only, &spec_ids_by_manifest, &specs, false).unwrap(); + assert_eq!(global_data.len(), 32); + assert_eq!(global_deletes.len(), 32); + assert_eq!(global_data.values().map(Vec::len).sum::(), 40); + + let (data, deletes) = + group_content_files(files, &spec_ids_by_manifest, &specs, false).unwrap(); + assert_eq!(data.len(), 2); + let old = &deletes[&(0, partition.clone())]; + assert_eq!(old.position_deletes.len(), 0); + assert_eq!( + old.equality_deletes + .iter() + .map(|(path, _)| path.as_str()) + .collect::>(), + HashSet::from(["eq-old", "eq-global"]) + ); + let current = &deletes[&(1, partition)]; + assert_eq!(current.position_deletes.len(), 1); + assert_eq!(current.position_deletes[0].0, "pos-new"); + assert_eq!(current.equality_deletes.len(), 1); + assert_eq!(current.equality_deletes[0].0, "eq-global"); + } + + #[test] + fn global_delete_plan_fanout_is_bounded() { + let file = |content, partition| { + DataFile::builder() + .with_content(content) + .with_file_path("/test.parquet".to_owned()) + .with_file_format(FileFormat::Parquet) + .with_partition(partition) + .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(None) + .with_upper_bounds(None) + .build() + .unwrap() + }; + let entry = |file| { + ManifestEntry::builder() + .with_format_version(super::FormatVersion::V2) + .with_status(Status::Added) + .with_data_file(file) + .build() + .unwrap() + }; + let data = entry(file(Content::Data, Struct::from_iter(Vec::new()))); + let local = entry(file( + Content::PositionDeletes, + Struct::from_iter(Vec::new()), + )); + let global = entry(file( + Content::EqualityDeletes, + Struct::from_iter(Vec::new()), + )); + let single_group = std::iter::once(("data-a".to_owned(), data.clone())) + .chain(std::iter::repeat_n( + ("global".to_owned(), global.clone()), + 1025, + )) + .collect::>(); + let specs = [0, 1] + .map(|id| { + ( + id, + PartitionSpec::builder() + .with_spec_id(id) + .with_fields(Vec::new()) + .build() + .unwrap(), + ) + }) + .into_iter() + .collect::>(); + let ids = HashMap::from([("data-a", 0), ("data-b", 1), ("local", 0), ("global", 0)]); + assert!(group_content_files(single_group, &ids, &specs, false).is_ok()); + let files = [ + ("data-a".to_owned(), data.clone()), + ("data-b".to_owned(), data), + ("local".to_owned(), local), + ] + .into_iter() + .chain(std::iter::repeat_n(("global".to_owned(), global), 513)) + .collect::>(); + let error = group_content_files(files, &ids, &specs, false).unwrap_err(); + assert!(error + .to_string() + .contains("global equality-delete applications")); + } + + #[test] + #[ignore = "manual mixed-spec scan planning benchmark"] + fn benchmark_global_delete_grouping() { + use std::time::Instant; + + let file = |content, partition| { + DataFile::builder() + .with_content(content) + .with_file_path("/test.parquet".to_owned()) + .with_file_format(FileFormat::Parquet) + .with_partition(partition) + .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(None) + .with_upper_bounds(None) + .build() + .unwrap() + }; + let entry = |file| { + ManifestEntry::builder() + .with_format_version(super::FormatVersion::V2) + .with_status(Status::Added) + .with_data_file(file) + .build() + .unwrap() + }; + let data = (0..10_000) + .map(|value| { + ( + "data".to_owned(), + entry(file( + Content::Data, + Struct::from_iter([("part".to_owned(), Some(Value::Int(value)))]), + )), + ) + }) + .collect::>(); + let global = (0..4) + .map(|_| { + ( + "global".to_owned(), + entry(file( + Content::EqualityDeletes, + Struct::from_iter(Vec::<(String, Option)>::new()), + )), + ) + }) + .collect::>(); + let specs = HashMap::from([ + ( + 0, + PartitionSpec::builder() + .with_spec_id(0) + .with_partition_field(PartitionField::new(1, 1000, "part", Transform::Identity)) + .build() + .unwrap(), + ), + ( + 1, + PartitionSpec::builder() + .with_spec_id(1) + .with_fields(Vec::new()) + .build() + .unwrap(), + ), + ]); + let ids = HashMap::from([("data", 0), ("global", 1)]); + for with_global in [false, true] { + let start = Instant::now(); + for _ in 0..20 { + let files = data + .iter() + .cloned() + .chain(global.iter().filter(|_| with_global).cloned()); + let (groups, deletes) = group_content_files(files, &ids, &specs, false).unwrap(); + if with_global { + assert_eq!(groups.len(), 32); + assert_eq!(deletes.len(), 32); + } else { + assert_eq!(groups.len(), 10_000); + assert!(deletes.is_empty()); + } + std::hint::black_box((groups, deletes)); + } + eprintln!( + "10k files, global={with_global}, 20 plans: {:?}", + start.elapsed() + ); + } + } + + #[test] + fn deleted_only_historical_manifest_keeps_current_partition_fast_path() { + let deleted_only = ManifestListEntry { + format_version: super::FormatVersion::V2, + manifest_path: "old".to_owned(), + manifest_length: 0, + partition_spec_id: 0, + content: ManifestContent::Data, + sequence_number: 0, + min_sequence_number: 0, + added_snapshot_id: 0, + added_files_count: Some(0), + existing_files_count: Some(0), + deleted_files_count: Some(1), + added_rows_count: Some(0), + existing_rows_count: Some(0), + deleted_rows_count: Some(1), + partitions: None, + key_metadata: None, + first_row_id: None, + }; + assert!(!has_historical_data_manifests( + &[deleted_only.clone()], + 1, + false + )); + assert!(has_historical_data_manifests( + &[deleted_only.clone()], + 1, + true + )); + let mut live = deleted_only.clone(); + live.existing_files_count = Some(1); + assert!(has_historical_data_manifests(&[live.clone()], 1, false)); + live.content = ManifestContent::Deletes; + assert!(!has_historical_data_manifests(&[live], 1, true)); + } + #[tokio::test] async fn bucketed_long_string_remains_visible_after_insert() { let catalog: Arc = Arc::new( diff --git a/datafusion_iceberg/tests/equality_delete.rs b/datafusion_iceberg/tests/equality_delete.rs index d5bb4500..772560a5 100644 --- a/datafusion_iceberg/tests/equality_delete.rs +++ b/datafusion_iceberg/tests/equality_delete.rs @@ -21,6 +21,7 @@ use iceberg_rust::{ partition::{PartitionField, PartitionSpec, Transform}, schema::Schema, types::{PrimitiveType, StructField, Type}, + values::Value, }, table::Table, }; @@ -174,6 +175,43 @@ pub async fn test_equality_delete() { .await .unwrap(); + // A delete writer may report metrics for non-key columns. They cannot be + // used to prune an equality delete: the row matched by its key can have a + // different value in the data file. + let files = files + .into_iter() + .map(|file| { + let mut lower = file.lower_bounds().clone().unwrap_or_default(); + lower.insert(5, Value::Int(999)); + let mut upper = file.upper_bounds().clone().unwrap_or_default(); + upper.insert(5, Value::Int(999)); + iceberg_rust::spec::manifest::DataFile::builder() + .with_content(file.content().clone()) + .with_file_path(file.file_path().clone()) + .with_file_format(file.file_format().clone()) + .with_partition(file.partition().clone()) + .with_record_count(*file.record_count()) + .with_file_size_in_bytes(*file.file_size_in_bytes()) + .with_column_sizes(file.column_sizes().clone()) + .with_value_counts(file.value_counts().clone()) + .with_null_value_counts(file.null_value_counts().clone()) + .with_nan_value_counts(file.nan_value_counts().clone()) + .with_distinct_counts(file.distinct_counts().clone()) + .with_lower_bounds(Some(lower)) + .with_upper_bounds(Some(upper)) + .with_key_metadata(file.key_metadata().clone()) + .with_split_offsets(file.split_offsets().clone()) + .with_equality_ids(file.equality_ids().clone()) + .with_sort_order_id(*file.sort_order_id()) + .with_first_row_id(*file.first_row_id()) + .with_referenced_data_file(file.referenced_data_file().clone()) + .with_content_offset(*file.content_offset()) + .with_content_size_in_bytes(*file.content_size_in_bytes()) + .build() + .unwrap() + }) + .collect::>(); + // Load the latest table version, which includes the inserted rows let Tabular::Table(mut table) = catalog .clone() @@ -204,6 +242,13 @@ pub async fn test_equality_delete() { ]; assert_batches_eq!(expected, &batches); + let batches = run_query( + "select id from warehouse.test.orders where amount = 1 order by id", + &ctx, + ) + .await; + assert_batches_eq!(["+----+", "| id |", "+----+", "| 2 |", "+----+"], &batches); + // Test that projecting a column that is not included in equality deletes works run_query( "INSERT INTO warehouse.test.orders (id, customer_id, product_id, date, amount) VALUES From 115439ada7c449d62776534481984d66756a5e3c Mon Sep 17 00:00:00 2001 From: osipovartem Date: Thu, 15 Oct 2026 07:30:32 +0300 Subject: [PATCH 2/2] Avoid cloning manifest entries in scan tests --- datafusion_iceberg/src/table/mod.rs | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/datafusion_iceberg/src/table/mod.rs b/datafusion_iceberg/src/table/mod.rs index 3c705531..86f8017f 100644 --- a/datafusion_iceberg/src/table/mod.rs +++ b/datafusion_iceberg/src/table/mod.rs @@ -5173,18 +5173,22 @@ mod tests { first_row_id: None, }; assert!(!has_historical_data_manifests( - &[deleted_only.clone()], + std::slice::from_ref(&deleted_only), 1, false )); assert!(has_historical_data_manifests( - &[deleted_only.clone()], + std::slice::from_ref(&deleted_only), 1, true )); - let mut live = deleted_only.clone(); + let mut live = deleted_only; live.existing_files_count = Some(1); - assert!(has_historical_data_manifests(&[live.clone()], 1, false)); + assert!(has_historical_data_manifests( + std::slice::from_ref(&live), + 1, + false + )); live.content = ManifestContent::Deletes; assert!(!has_historical_data_manifests(&[live], 1, true)); }