Skip to content

fix(scan): apply equality deletes from an all-void partition spec globally #3449

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
JeonDaehong wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

fix(scan): apply equality deletes from an all-void partition spec globally #3449

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
83 changes: 81 additions & 2 deletions crates/iceberg/src/delete_file_index.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -248,7 +248,12 @@ impl PopulatedDeleteFileIndex {
}
DataContentType::EqualityDeletes => {
// The spec states that "Equality delete files stored with an unpartitioned spec are applied as global deletes".
if partition.fields().is_empty() {
// As in Java, a spec whose fields are all void transforms is unpartitioned.
let unpartitioned = match &arc_ctx.partition_spec {
Some(spec) => spec.is_unpartitioned(),
None => partition.fields().is_empty(),
};
if unpartitioned {
global_equality_deletes.push(arc_ctx);
} else {
eq_deletes_by_partition
Expand Down Expand Up @@ -386,7 +391,7 @@ mod tests {
use crate::ErrorKind;
use crate::spec::{
DataContentType, DataFileBuilder, DataFileFormat, Datum, Literal, ManifestEntry,
ManifestStatus, Struct,
ManifestStatus, NestedField, PartitionSpec, PrimitiveType, Schema, Struct, Transform, Type,
};

#[test]
Expand All @@ -408,6 +413,7 @@ mod tests {
.map(|entry| DeleteFileContext {
manifest_entry: entry.into(),
partition_spec_id: 0,
partition_spec: None,
})
.collect();

Expand Down Expand Up @@ -506,6 +512,7 @@ mod tests {
.map(|entry| DeleteFileContext {
manifest_entry: entry.into(),
partition_spec_id: spec_id,
partition_spec: None,
})
.collect();

Expand Down Expand Up @@ -592,6 +599,49 @@ mod tests {
assert!(actual_paths_to_apply_for_different_spec.is_empty());
}

#[test]
fn test_void_spec_equality_delete_is_global() {
// Dropping a v1 table's only partition field leaves it as a void transform. Like Java,
// such a spec is unpartitioned, so its equality deletes also apply to data files written
// under the earlier, partitioned spec.
let schema = Schema::builder()
.with_fields(vec![
NestedField::required(1, "x", Type::Primitive(PrimitiveType::Long)).into(),
])
.build()
.unwrap();
let void_spec = PartitionSpec::builder(schema)
.with_spec_id(1)
.add_partition_field("x", "x_void", Transform::Void)
.unwrap()
.build()
.unwrap();

let eq_delete = build_partitioned_eq_delete(&Struct::from_iter([None::<Literal>]), 1);
let index = PopulatedDeleteFileIndex::new(vec![DeleteFileContext {
manifest_entry: build_added_manifest_entry(5, &eq_delete).into(),
partition_spec_id: 1,
partition_spec: Some(Arc::new(void_spec)),
}])
.unwrap();

let old_spec_file =
build_partitioned_data_file(&Struct::from_iter([Some(Literal::long(100))]), 0);
let deletes = index
.get_deletes_for_data_file(&old_spec_file, Some(1))
.unwrap();
assert_eq!(deletes.len(), 1);
assert_eq!(deletes[0].file_path(), eq_delete.file_path());

// The sequence number rule still applies.
assert!(
index
.get_deletes_for_data_file(&old_spec_file, Some(5))
.unwrap()
.is_empty()
);
}

#[test]
fn test_partition_scoped_deletes_distinguish_signed_zero() {
let neg_zero = Struct::from_iter([Some(Literal::double(-0.0))]);
Expand All @@ -606,6 +656,7 @@ mod tests {
.map(|delete| DeleteFileContext {
manifest_entry: build_added_manifest_entry(4, delete).into(),
partition_spec_id: spec_id,
partition_spec: None,
})
.collect();

Expand Down Expand Up @@ -633,6 +684,7 @@ mod tests {
let index = PopulatedDeleteFileIndex::new(vec![DeleteFileContext {
manifest_entry: build_added_manifest_entry(5, &pos_delete).into(),
partition_spec_id: 0,
partition_spec: None,
}])
.unwrap();

Expand Down Expand Up @@ -662,6 +714,7 @@ mod tests {
let index = PopulatedDeleteFileIndex::new(vec![DeleteFileContext {
manifest_entry: build_added_manifest_entry(5, &pos_delete).into(),
partition_spec_id: 0,
partition_spec: None,
}])
.unwrap();

Expand Down Expand Up @@ -692,6 +745,7 @@ mod tests {
let index = PopulatedDeleteFileIndex::new(vec![DeleteFileContext {
manifest_entry: build_added_manifest_entry(5, &pos_delete).into(),
partition_spec_id: 0,
partition_spec: None,
}])
.unwrap();

Expand Down Expand Up @@ -722,6 +776,7 @@ mod tests {
let index = PopulatedDeleteFileIndex::new(vec![DeleteFileContext {
manifest_entry: build_added_manifest_entry(5, &pos_delete).into(),
partition_spec_id: 0,
partition_spec: None,
}])
.unwrap();

Expand All @@ -742,6 +797,7 @@ mod tests {
let index = PopulatedDeleteFileIndex::new(vec![DeleteFileContext {
manifest_entry: build_added_manifest_entry(5, &pos_delete).into(),
partition_spec_id: 0,
partition_spec: None,
}])
.unwrap();

Expand Down Expand Up @@ -792,14 +848,17 @@ mod tests {
DeleteFileContext {
manifest_entry: build_added_manifest_entry(5, &path_delete).into(),
partition_spec_id: 0,
partition_spec: None,
},
DeleteFileContext {
manifest_entry: build_added_manifest_entry(5, &partition_delete).into(),
partition_spec_id: 0,
partition_spec: None,
},
DeleteFileContext {
manifest_entry: build_added_manifest_entry(5, &eq_delete).into(),
partition_spec_id: 0,
partition_spec: None,
},
])
.unwrap();
Expand Down Expand Up @@ -833,10 +892,12 @@ mod tests {
DeleteFileContext {
manifest_entry: build_added_manifest_entry(5, &first_delete).into(),
partition_spec_id: 0,
partition_spec: None,
},
DeleteFileContext {
manifest_entry: build_added_manifest_entry(6, &second_delete).into(),
partition_spec_id: 0,
partition_spec: None,
},
])
.unwrap();
Expand Down Expand Up @@ -877,6 +938,7 @@ mod tests {
let index = PopulatedDeleteFileIndex::new(vec![DeleteFileContext {
manifest_entry: build_added_manifest_entry(5, &pos_delete).into(),
partition_spec_id: 1,
partition_spec: None,
}])
.unwrap();

Expand Down Expand Up @@ -911,6 +973,7 @@ mod tests {
let index = PopulatedDeleteFileIndex::new(vec![DeleteFileContext {
manifest_entry: build_added_manifest_entry(5, &pos_delete).into(),
partition_spec_id: 0,
partition_spec: None,
}])
.unwrap();

Expand Down Expand Up @@ -988,6 +1051,7 @@ mod tests {
let ctx = DeleteFileContext {
manifest_entry: build_added_manifest_entry(5, &dv).into(),
partition_spec_id: 0,
partition_spec: None,
};

let task = FileScanTaskDeleteFile::try_from(&ctx).unwrap();
Expand Down Expand Up @@ -1015,10 +1079,12 @@ mod tests {
DeleteFileContext {
manifest_entry: build_added_manifest_entry(5, &dv).into(),
partition_spec_id: spec_id,
partition_spec: None,
},
DeleteFileContext {
manifest_entry: build_added_manifest_entry(5, &pos_del).into(),
partition_spec_id: spec_id,
partition_spec: None,
},
];

Expand Down Expand Up @@ -1046,6 +1112,7 @@ mod tests {
let contexts = vec![DeleteFileContext {
manifest_entry: build_added_manifest_entry(3, &dv).into(),
partition_spec_id: spec_id,
partition_spec: None,
}];

let index = PopulatedDeleteFileIndex::new(contexts).unwrap();
Expand Down Expand Up @@ -1077,6 +1144,7 @@ mod tests {
let contexts = vec![DeleteFileContext {
manifest_entry: build_added_manifest_entry(5, &dv).into(),
partition_spec_id: spec_id,
partition_spec: None,
}];

let index = PopulatedDeleteFileIndex::new(contexts).unwrap();
Expand All @@ -1103,6 +1171,7 @@ mod tests {
let contexts = vec![DeleteFileContext {
manifest_entry: build_added_manifest_entry(5, &dv).into(),
partition_spec_id: 2,
partition_spec: None,
}];

let index = PopulatedDeleteFileIndex::new(contexts).unwrap();
Expand Down Expand Up @@ -1130,10 +1199,12 @@ mod tests {
DeleteFileContext {
manifest_entry: build_added_manifest_entry(5, &dv).into(),
partition_spec_id: spec_id,
partition_spec: None,
},
DeleteFileContext {
manifest_entry: build_added_manifest_entry(5, &pos_del).into(),
partition_spec_id: 0,
partition_spec: None,
},
];

Expand Down Expand Up @@ -1165,10 +1236,12 @@ mod tests {
DeleteFileContext {
manifest_entry: build_added_manifest_entry(5, &dv).into(),
partition_spec_id: spec_id,
partition_spec: None,
},
DeleteFileContext {
manifest_entry: build_added_manifest_entry(5, &eq_del).into(),
partition_spec_id: 0,
partition_spec: None,
},
];

Expand Down Expand Up @@ -1204,6 +1277,7 @@ mod tests {
let contexts = vec![DeleteFileContext {
manifest_entry: build_added_manifest_entry(5, &malformed_dv).into(),
partition_spec_id: 0,
partition_spec: None,
}];

let err = PopulatedDeleteFileIndex::new(contexts).unwrap_err();
Expand Down Expand Up @@ -1245,6 +1319,7 @@ mod tests {
let contexts = vec![DeleteFileContext {
manifest_entry: build_added_manifest_entry(5, &malformed_dv).into(),
partition_spec_id: 0,
partition_spec: None,
}];

let delete_file_index = PopulatedDeleteFileIndex::new(contexts).unwrap();
Expand All @@ -1269,10 +1344,12 @@ mod tests {
DeleteFileContext {
manifest_entry: build_added_manifest_entry(5, &dv_1).into(),
partition_spec_id: spec_id,
partition_spec: None,
},
DeleteFileContext {
manifest_entry: build_added_manifest_entry(6, &dv_2).into(),
partition_spec_id: spec_id,
partition_spec: None,
},
];

Expand All @@ -1294,11 +1371,13 @@ mod tests {
tx.try_send(DeleteFileContext {
manifest_entry: build_added_manifest_entry(5, &dv_1).into(),
partition_spec_id: spec_id,
partition_spec: None,
})
.unwrap();
tx.try_send(DeleteFileContext {
manifest_entry: build_added_manifest_entry(6, &dv_2).into(),
partition_spec_id: spec_id,
partition_spec: None,
})
.unwrap();
drop(tx);
Expand Down
1 change: 1 addition & 0 deletions crates/iceberg/src/scan/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -668,6 +668,7 @@ impl TableScan {
.send(DeleteFileContext {
manifest_entry: manifest_entry_context.manifest_entry.clone(),
partition_spec_id: manifest_entry_context.partition_spec_id,
partition_spec: manifest_entry_context.partition_spec.clone(),
})
.await?;

Expand Down
7 changes: 5 additions & 2 deletions crates/iceberg/src/scan/task.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,8 +24,8 @@ use typed_builder::TypedBuilder;
use crate::error::invalid_data;
use crate::expr::BoundPredicate;
use crate::spec::{
DataContentType, DataFileFormat, ManifestEntryRef, NameMapping, PartitionSpec, Schema,
SchemaRef, SortOrderRef, Struct, StructType,
DataContentType, DataFileFormat, ManifestEntryRef, NameMapping, PartitionSpec,
PartitionSpecRef, Schema, SchemaRef, SortOrderRef, Struct, StructType,
};
use crate::{Error, Result};

Expand Down Expand Up @@ -307,6 +307,8 @@ impl From<FileScanTask> for Result<FileScanTask> {
pub(crate) struct DeleteFileContext {
pub(crate) manifest_entry: ManifestEntryRef,
pub(crate) partition_spec_id: i32,
/// The delete file's partition spec, or `None` if the table metadata does not have it.
pub(crate) partition_spec: Option<PartitionSpecRef>,
}

impl TryFrom<&DeleteFileContext> for FileScanTaskDeleteFile {
Expand Down Expand Up @@ -1098,6 +1100,7 @@ mod tests {
.build(),
),
partition_spec_id: 0,
partition_spec: None,
}
}

Expand Down
Loading