diff --git a/crates/iceberg/src/delete_file_index.rs b/crates/iceberg/src/delete_file_index.rs index e58bb671b6..5933375229 100644 --- a/crates/iceberg/src/delete_file_index.rs +++ b/crates/iceberg/src/delete_file_index.rs @@ -23,9 +23,10 @@ use futures::StreamExt; use futures::channel::mpsc::{Sender, channel}; use tokio::sync::Notify; +use crate::metadata_columns::RESERVED_FIELD_ID_DELETE_FILE_PATH; use crate::runtime::Runtime; use crate::scan::{DeleteFileContext, FileScanTaskDeleteFile}; -use crate::spec::{DataContentType, DataFile, Struct}; +use crate::spec::{DataContentType, DataFile, PrimitiveLiteral, Struct}; /// Index of delete files #[derive(Debug, Clone)] @@ -42,11 +43,12 @@ enum DeleteFileIndexState { #[derive(Debug)] struct PopulatedDeleteFileIndex { global_equality_deletes: Vec>, - eq_deletes_by_partition: HashMap>>, - pos_deletes_by_partition: HashMap>>, - // TODO: do we need this? - // pos_deletes_by_path: HashMap>>, - + /// Equality deletes keyed by partition spec id, then by partition value. + eq_deletes_by_partition: HashMap>>>, + /// Position deletes that name no single data file, keyed the same way. + pos_deletes_by_partition: HashMap>>>, + /// Position deletes keyed by the data file path they apply to. + pos_deletes_by_path: HashMap>>, // TODO: Deletion Vector support } @@ -113,6 +115,40 @@ impl DeleteFileIndex { } } +/// The single data file a position delete file applies to, `None` if it is not tied +/// to a single data file. +fn position_delete_target(data_file: &DataFile) -> Option { + // data files is named directly + if let Some(path) = data_file.referenced_data_file() { + return Some(path); + } + + // lower and upper bound of reserved field are equal, so all rows + // have the same value + let lower = data_file + .lower_bounds() + .get(&RESERVED_FIELD_ID_DELETE_FILE_PATH)?; + let upper = data_file + .upper_bounds() + .get(&RESERVED_FIELD_ID_DELETE_FILE_PATH)?; + if lower != upper { + return None; + } + + match lower.literal() { + PrimitiveLiteral::String(path) => Some(path.clone()), + _ => None, + } +} + +/// Whether a position delete file's sequence number lets it apply to a data file whose +/// own sequence number is `data_file_seq_num`. +fn position_delete_applies(delete_seq_num: Option, data_file_seq_num: Option) -> bool { + data_file_seq_num + .map(|seq| delete_seq_num >= Some(seq)) + .unwrap_or(true) +} + impl PopulatedDeleteFileIndex { /// Creates a new populated delete file index from a list of delete file contexts, which /// allows for fast lookup when determining which delete files apply to a given data file. @@ -120,11 +156,18 @@ impl PopulatedDeleteFileIndex { /// 1. The partition information is extracted from each delete file's manifest entry. /// 2. If the partition is empty and the delete file is not a positional delete, /// it is added to the `global_equality_deletes` vector - /// 3. Otherwise, the delete file is added to one of two hash maps based on its content type. + /// 3. A positional delete that names a single data file is keyed by that path. + /// 4. Any other delete file is keyed by partition, in the map for its content type. fn new(files: Vec) -> PopulatedDeleteFileIndex { - let mut eq_deletes_by_partition: HashMap>> = - HashMap::default(); - let mut pos_deletes_by_partition: HashMap>> = + let mut eq_deletes_by_partition: HashMap< + i32, + HashMap>>, + > = HashMap::default(); + let mut pos_deletes_by_partition: HashMap< + i32, + HashMap>>, + > = HashMap::default(); + let mut pos_deletes_by_path: HashMap>> = HashMap::default(); let mut global_equality_deletes: Vec> = vec![]; @@ -132,6 +175,14 @@ impl PopulatedDeleteFileIndex { files.into_iter().for_each(|ctx| { let arc_ctx = Arc::new(ctx); + if arc_ctx.manifest_entry.sequence_number().is_none() { + tracing::warn!( + delete_file = arc_ctx.manifest_entry.data_file().file_path(), + status = ?arc_ctx.manifest_entry.status(), + "delete file manifest entry has no data sequence number. It will be skipped for data files with a known sequence number" + ); + } + let partition = arc_ctx.manifest_entry.data_file().partition(); // The spec states that "Equality delete files stored with an unpartitioned spec are applied as global deletes". @@ -144,23 +195,49 @@ impl PopulatedDeleteFileIndex { } let destination_map = match arc_ctx.manifest_entry.content_type() { - DataContentType::PositionDeletes => &mut pos_deletes_by_partition, + DataContentType::PositionDeletes => { + if let Some(path) = position_delete_target(arc_ctx.manifest_entry.data_file()) { + pos_deletes_by_path.entry(path).or_default().push(arc_ctx); + return; + } + &mut pos_deletes_by_partition + } DataContentType::EqualityDeletes => &mut eq_deletes_by_partition, _ => unreachable!(), }; destination_map + .entry(arc_ctx.partition_spec_id) + .or_default() .entry(partition.clone()) - .and_modify(|entry| { - entry.push(arc_ctx.clone()); - }) - .or_insert(vec![arc_ctx.clone()]); + .or_default() + .push(arc_ctx); }); + // A large number of delete files that are attributed globally or partition-scoped + // _could_ explain slow reads or memory problems, so make a DEBUG trace available. + tracing::debug!( + pos_deletes_attributed = pos_deletes_by_path.values().map(Vec::len).sum::(), + pos_deletes_target_files = pos_deletes_by_path.len(), + pos_deletes_partition_scoped = pos_deletes_by_partition + .values() + .flat_map(HashMap::values) + .map(Vec::len) + .sum::(), + eq_deletes_partition_scoped = eq_deletes_by_partition + .values() + .flat_map(HashMap::values) + .map(Vec::len) + .sum::(), + eq_deletes_global = global_equality_deletes.len(), + "delete file index populated" + ); + PopulatedDeleteFileIndex { global_equality_deletes, eq_deletes_by_partition, pos_deletes_by_partition, + pos_deletes_by_path, } } @@ -182,7 +259,11 @@ impl PopulatedDeleteFileIndex { }) .for_each(|delete| results.push(delete.as_ref().into())); - if let Some(deletes) = self.eq_deletes_by_partition.get(data_file.partition()) { + if let Some(deletes) = self + .eq_deletes_by_partition + .get(&data_file.partition_spec_id) + .and_then(|by_partition| by_partition.get(data_file.partition())) + { deletes .iter() // filter that returns true if the provided delete file's sequence number is **greater than** `seq_num` @@ -190,24 +271,28 @@ impl PopulatedDeleteFileIndex { seq_num .map(|seq_num| delete.manifest_entry.sequence_number() > Some(seq_num)) .unwrap_or_else(|| true) - && data_file.partition_spec_id == delete.partition_spec_id }) .for_each(|delete| results.push(delete.as_ref().into())); } - // TODO: the spec states that: - // "The data file's file_path is equal to the delete file's referenced_data_file if it is non-null". - // we're not yet doing that here. The referenced data file's name will also be present in the positional - // delete file's file path column. - if let Some(deletes) = self.pos_deletes_by_partition.get(data_file.partition()) { + if let Some(deletes) = self.pos_deletes_by_path.get(data_file.file_path()) { deletes .iter() - // filter that returns true if the provided delete file's sequence number is **greater than or equal to** `seq_num` .filter(|&delete| { - seq_num - .map(|seq_num| delete.manifest_entry.sequence_number() >= Some(seq_num)) - .unwrap_or_else(|| true) - && data_file.partition_spec_id == delete.partition_spec_id + position_delete_applies(delete.manifest_entry.sequence_number(), seq_num) + }) + .for_each(|delete| results.push(delete.as_ref().into())); + } + + if let Some(deletes) = self + .pos_deletes_by_partition + .get(&data_file.partition_spec_id) + .and_then(|by_partition| by_partition.get(data_file.partition())) + { + deletes + .iter() + .filter(|&delete| { + position_delete_applies(delete.manifest_entry.sequence_number(), seq_num) }) .for_each(|delete| results.push(delete.as_ref().into())); } @@ -222,8 +307,8 @@ mod tests { use super::*; use crate::spec::{ - DataContentType, DataFileBuilder, DataFileFormat, Literal, ManifestEntry, ManifestStatus, - Struct, + DataContentType, DataFileBuilder, DataFileFormat, Datum, Literal, ManifestEntry, + ManifestStatus, Struct, }; #[test] @@ -444,7 +529,6 @@ mod tests { .file_format(DataFileFormat::Parquet) .content(DataContentType::PositionDeletes) .record_count(1) - .referenced_data_file(Some("/some-data-file.parquet".to_string())) .partition(partition.clone()) .partition_spec_id(spec_id) .file_size_in_bytes(100) @@ -485,4 +569,223 @@ mod tests { .data_file(file.clone()) .build() } + + /// A position delete naming `data_file_path` through the optional spec field. + fn build_pos_delete_referencing(data_file_path: &str) -> DataFile { + DataFileBuilder::default() + .file_path(format!("{}-pos-delete.parquet", Uuid::new_v4())) + .file_format(DataFileFormat::Parquet) + .content(DataContentType::PositionDeletes) + .record_count(1) + .referenced_data_file(Some(data_file_path.to_string())) + .partition(Struct::empty()) + .partition_spec_id(0) + .file_size_in_bytes(100) + .build() + .unwrap() + } + + /// A position delete pinned by the reserved `file_path` column bounds + fn build_pos_delete_with_bounds( + lower: &str, + upper: &str, + partition: &Struct, + spec_id: i32, + ) -> DataFile { + DataFileBuilder::default() + .file_path(format!("{}-pos-delete.parquet", Uuid::new_v4())) + .file_format(DataFileFormat::Parquet) + .content(DataContentType::PositionDeletes) + .record_count(1) + .lower_bounds(HashMap::from([( + RESERVED_FIELD_ID_DELETE_FILE_PATH, + Datum::string(lower), + )])) + .upper_bounds(HashMap::from([( + RESERVED_FIELD_ID_DELETE_FILE_PATH, + Datum::string(upper), + )])) + .partition(partition.clone()) + .partition_spec_id(spec_id) + .file_size_in_bytes(100) + .build() + .unwrap() + } + + /// Build `PopulatedDeleteFileIndex` from delete file manifest entries + fn index_of(entries: Vec) -> PopulatedDeleteFileIndex { + PopulatedDeleteFileIndex::new( + entries + .into_iter() + .map(|entry| { + let partition_spec_id = entry.data_file().partition_spec_id; + DeleteFileContext { + manifest_entry: entry.into(), + partition_spec_id, + } + }) + .collect(), + ) + } + + #[test] + fn test_position_delete_attributed_by_referenced_data_file() { + let data_file = build_unpartitioned_data_file(); + let other_data_file = build_unpartitioned_data_file(); + let index = index_of(vec![build_added_manifest_entry( + 2, + &build_pos_delete_referencing(data_file.file_path()), + )]); + + assert_eq!( + index.get_deletes_for_data_file(&data_file, Some(1)).len(), + 1 + ); + assert!( + index + .get_deletes_for_data_file(&other_data_file, Some(1)) + .is_empty() + ); + } + + #[test] + fn test_position_delete_attributed_by_file_path_bounds() { + let data_file = build_unpartitioned_data_file(); + let other_data_file = build_unpartitioned_data_file(); + let path = data_file.file_path(); + let index = index_of(vec![build_added_manifest_entry( + 2, + &build_pos_delete_with_bounds(path, path, &Struct::empty(), 0), + )]); + + assert_eq!( + index.get_deletes_for_data_file(&data_file, Some(1)).len(), + 1 + ); + assert!( + index + .get_deletes_for_data_file(&other_data_file, Some(1)) + .is_empty() + ); + } + + #[test] + fn test_position_delete_with_unequal_bounds_stays_partition_scoped() { + let data_file = build_unpartitioned_data_file(); + let other_data_file = build_unpartitioned_data_file(); + let index = index_of(vec![build_added_manifest_entry( + 2, + &build_pos_delete_with_bounds("a-data.parquet", "z-data.parquet", &Struct::empty(), 0), + )]); + + // The delete spans several data files, so every file in the partition has to + // consider it. + assert_eq!( + index.get_deletes_for_data_file(&data_file, Some(1)).len(), + 1 + ); + assert_eq!( + index + .get_deletes_for_data_file(&other_data_file, Some(1)) + .len(), + 1 + ); + } + + #[test] + fn test_position_delete_without_bounds_stays_partition_scoped() { + let data_file = build_unpartitioned_data_file(); + let other_data_file = build_unpartitioned_data_file(); + let index = index_of(vec![build_added_manifest_entry( + 2, + &build_unpartitioned_pos_delete(), + )]); + + assert_eq!( + index.get_deletes_for_data_file(&data_file, Some(1)).len(), + 1 + ); + assert_eq!( + index + .get_deletes_for_data_file(&other_data_file, Some(1)) + .len(), + 1 + ); + } + + #[test] + fn test_position_delete_by_path_ignores_partition_spec_id() { + let partition = Struct::from_iter([Some(Literal::long(100))]); + let data_file = build_partitioned_data_file(&partition, 1); + let path = data_file.file_path(); + + // different spec_id on pos. delete, but path bound matches => should tie to data file + let attributed = index_of(vec![build_added_manifest_entry( + 2, + &build_pos_delete_with_bounds(path, path, &partition, 0), + )]); + assert_eq!( + attributed + .get_deletes_for_data_file(&data_file, Some(1)) + .len(), + 1 + ); + + // No path bounds => goes to partition bucket => spec_id does not match => should not be tied + // to data file + let mismatched_spec = index_of(vec![build_added_manifest_entry( + 2, + &build_partitioned_pos_delete(&partition, 0), + )]); + assert!( + mismatched_spec + .get_deletes_for_data_file(&data_file, Some(1)) + .is_empty() + ); + + // With same spec_id, it should be tied to data file + let matching_spec = index_of(vec![build_added_manifest_entry( + 2, + &build_partitioned_pos_delete(&partition, 1), + )]); + assert_eq!( + matching_spec + .get_deletes_for_data_file(&data_file, Some(1)) + .len(), + 1 + ); + } + + #[test] + fn test_position_delete_from_earlier_commit_does_not_apply() { + let data_file = build_unpartitioned_data_file(); + let path = data_file.file_path(); + + let by_path = index_of(vec![build_added_manifest_entry( + 1, + &build_pos_delete_referencing(path), + )]); + assert!( + by_path + .get_deletes_for_data_file(&data_file, Some(2)) + .is_empty(), + "position delete at seq 1 must not apply to a data file at seq 2" + ); + } + + #[test] + fn test_position_delete_from_same_commit_applies() { + let data_file = build_unpartitioned_data_file(); + let path = data_file.file_path(); + + let by_path = index_of(vec![build_added_manifest_entry( + 2, + &build_pos_delete_referencing(path), + )]); + assert_eq!( + by_path.get_deletes_for_data_file(&data_file, Some(2)).len(), + 1, + "position delete at seq 2 must apply to a data file at seq 2 (same commit)" + ); + } }