This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
The following commit(s) were added to refs/heads/main by this push:
new af1717ae9 perf(scan): skip deleted entries before partition promotion
(#3516)
af1717ae9 is described below
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)
