Skip to content

perf(scan): skip deleted entries before partition promotion #3516

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

Merged
laskoviymishka merged 5 commits into from
Oct 28, 2026
Merged

perf(scan): skip deleted entries before partition promotion #3516

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
b9a6614
perf(scan): skip deleted entries before partition promotion
anoopj Oct 23, 2026
f5c0301
Merge branch 'main' into followup-3483-manifest-promotion
anoopj Oct 26, 2026
e3cdb62
Merge branch 'main' into followup-3483-manifest-promotion
anoopj Oct 26, 2026
a86d560
Merge remote-tracking branch 'origin/main' into followup-3483-manifes…
anoopj Oct 27, 2026
48aa270
Merge branch 'main' into followup-3483-manifest-promotion
anoopj Oct 27, 2026
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
7 changes: 5 additions & 2 deletions crates/iceberg/src/scan/context.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -97,8 +97,7 @@ fn partition_types_differ(current: &StructType, written: &StructType) -> bool {
}

impl ManifestFileContext {
/// Consumes this [`ManifestFileContext`], fetching its Manifest from FileIO and then
/// streaming its constituent [`ManifestEntries`] to the channel provided in the context
/// Fetches the manifest and streams its live entries to the context's channel.
pub(crate) async fn fetch_manifest_and_stream_manifest_entries(self) -> Result<()> {
let ManifestFileContext {
object_cache,
Expand Down Expand Up @@ -135,6 +134,10 @@ impl ManifestFileContext {
};

for manifest_entry in manifest.entries() {
if !manifest_entry.is_alive() {
continue;
}

let manifest_entry = match promotion_target.as_ref() {
Some((current, written)) => {
let mut entry = manifest_entry.as_ref().clone();
Expand Down
45 changes: 4 additions & 41 deletions crates/iceberg/src/scan/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -638,11 +638,6 @@ impl TableScan {
F: Fn(ManifestEntryContext) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<T>> + Send + 'static,
{
// skip processing this manifest entry if it has been marked as deleted
if !manifest_entry_context.manifest_entry.is_alive() {
return Ok(());
}

// abort the plan if we encounter a manifest entry for a delete file
if manifest_entry_context.manifest_entry.content_type() != DataContentType::Data {
return Err(Error::new(
Expand Down Expand Up @@ -695,11 +690,6 @@ impl TableScan {
manifest_entry_context: ManifestEntryContext,
mut delete_file_ctx_tx: Sender<DeleteFileContext>,
) -> Result<()> {
// skip processing this manifest entry if it has been marked as deleted
if !manifest_entry_context.manifest_entry.is_alive() {
return Ok(());
}

// abort the plan if we encounter a manifest entry that is not for a delete file
if manifest_entry_context.manifest_entry.content_type() == DataContentType::Data {
return Err(Error::new(
Expand Down Expand Up @@ -1736,43 +1726,15 @@ mod tests {
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);
fixture.drop_partition_source(1);
let table = fixture.table;

let tasks: Vec<FileScanTask> = table
.scan()
Expand All @@ -1794,7 +1756,8 @@ mod tests {
// 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);
fixture.drop_partition_source(1);
let table = fixture.table;

let tasks: Vec<FileScanTask> = table
.scan()
Expand Down
174 changes: 58 additions & 116 deletions crates/iceberg/src/test_utils/scan.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,9 +43,9 @@ use crate::metadata_columns::{
};
use crate::spec::{
DataContentType, DataFileBuilder, DataFileFormat, FormatVersion, Literal, ManifestEntry,
ManifestListWriter, ManifestStatus, ManifestWriterBuilder, NestedField, PartitionSpec,
PrimitiveType, Schema, Struct, StructType, TableMetadata, TableMetadataBuilder, Transform,
Type, UnboundPartitionSpec,
ManifestFile, ManifestListWriter, ManifestStatus, ManifestWriterBuilder, NestedField,
PartitionSpec, PrimitiveType, Schema, SchemaRef, Struct, StructType, TableMetadata,
TableMetadataBuilder, Transform, Type, UnboundPartitionSpec,
};
use crate::table::Table;
use crate::test_utils::test_runtime;
Expand Down Expand Up @@ -607,24 +607,15 @@ 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,
) {
async fn write_unpromoted_data_manifest(
&self,
current_schema: &Schema,
partition: Struct,
) -> (SchemaRef, ManifestFile) {
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
Expand All @@ -641,13 +632,17 @@ impl TableTestFixture {

// 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(
let 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();
);
let mut data_writer = match self.table.metadata().format_version() {
FormatVersion::V1 => data_writer.build_v1(),
FormatVersion::V2 => data_writer.build_v2_data(),
FormatVersion::V3 => data_writer.build_v3_data(),
};
data_writer
.add_existing_entry(
ManifestEntry::builder()
Expand All @@ -657,13 +652,13 @@ impl TableTestFixture {
.file_sequence_number(parent_snapshot.sequence_number())
.data_file(
DataFileBuilder::default()
.partition_spec_id(0)
.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))]))
.partition(partition)
.build()
.unwrap(),
)
Expand All @@ -672,6 +667,25 @@ impl TableTestFixture {
.unwrap();
let data_manifest = data_writer.write_manifest_file().await.unwrap();

(written_schema, data_manifest)
}

/// Writes manifests for int-partitioned data and an equality delete.
/// `delete_before_promotion` selects an int delete partition; otherwise it is long.
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 current_schema = current_snapshot.schema(self.table.metadata()).unwrap();
let current_partition_spec = self.table.metadata().default_partition_spec();
let (written_schema, data_manifest) = self
.write_unpromoted_data_manifest(
&current_schema,
Struct::from_iter([Some(Literal::int(7))]),
)
.await;

// 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`.
Expand Down Expand Up @@ -759,54 +773,16 @@ impl TableTestFixture {
.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();
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(),
let (_, data_manifest) = self
.write_unpromoted_data_manifest(
&current_schema,
Struct::from_iter([Some(Literal::int(7))]),
)
.unwrap();
let data_manifest = data_writer.write_manifest_file().await.unwrap();
.await;

let mut dv_writer = ManifestWriterBuilder::new(
self.next_manifest_file(),
Expand Down Expand Up @@ -895,59 +871,14 @@ impl TableTestFixture {
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();
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(),
let (_, data_manifest) = self
.write_unpromoted_data_manifest(
&current_schema,
Struct::from_iter([Some(Literal::int(7)), Some(Literal::date(20_000))]),
)
.unwrap();
let data_manifest = data_writer.write_manifest_file().await.unwrap();
.await;

// Equality-delete manifest, Added in the current snapshot (sequence number above
// the data file's) under the current `long` schema.
Expand Down Expand Up @@ -1004,15 +935,26 @@ impl TableTestFixture {

// 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`.
self.drop_partition_source(9);
}

/// Drops a source column after making the default partition spec unpartitioned.
pub(crate) fn drop_partition_source(&mut self, source_id: i32) {
let current_schema = self.table.metadata().current_schema();
let evolved_schema = Schema::builder()
.with_fields(
current_schema
.as_struct()
.fields()
.iter()
.filter(|field| field.id != 9)
.filter(|field| field.id != source_id)
.cloned(),
)
.with_identifier_field_ids(
current_schema
.identifier_field_ids()
.filter(|id| *id != source_id),
)
.build()
.unwrap();
let metadata = TableMetadataBuilder::new_from_metadata(self.table.metadata().clone(), None)
Expand Down
Loading