This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-3516-17b303c0cad70e8249f6284a973b48d51cc3c0d3 in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
commit af1717ae9287c256b987d0418251443052ead66b Author: Anoop Johnson <[email protected]> AuthorDate: Wed Oct 28 10:20:50 2026 +0000 perf(scan): skip deleted entries before partition promotion (#3516) Discard deleted manifest entries before copying their metadata or dispatching scan work. Live entries still widen before partition pruning and delete matching, consistent with the spec and Java's live-entry scan. Share the promotion fixtures' old-schema data manifest and source-drop setup while preserving their schema, format, and sequence-number cases. Follow-up to #3483. --- crates/iceberg/src/scan/context.rs | 7 +- crates/iceberg/src/scan/mod.rs | 45 +-------- crates/iceberg/src/test_utils/scan.rs | 174 ++++++++++++---------------------- 3 files changed, 67 insertions(+), 159 deletions(-) diff --git a/crates/iceberg/src/scan/context.rs b/crates/iceberg/src/scan/context.rs index 0e8b0aa50..695b7f85b 100644 --- a/crates/iceberg/src/scan/context.rs +++ b/crates/iceberg/src/scan/context.rs @@ -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, @@ -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(); diff --git a/crates/iceberg/src/scan/mod.rs b/crates/iceberg/src/scan/mod.rs index 636abb4d7..005b5ea57 100644 --- a/crates/iceberg/src/scan/mod.rs +++ b/crates/iceberg/src/scan/mod.rs @@ -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( @@ -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( @@ -1736,35 +1726,6 @@ 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) @@ -1772,7 +1733,8 @@ mod tests { // 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() @@ -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() diff --git a/crates/iceberg/src/test_utils/scan.rs b/crates/iceberg/src/test_utils/scan.rs index a6023a9bf..42038724f 100644 --- a/crates/iceberg/src/test_utils/scan.rs +++ b/crates/iceberg/src/test_utils/scan.rs @@ -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; @@ -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 @@ -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() @@ -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(), ) @@ -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( + ¤t_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`. @@ -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( + ¤t_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(), @@ -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( + ¤t_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. @@ -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)
