diff --git a/crates/iceberg/src/inspect/manifests.rs b/crates/iceberg/src/inspect/manifests.rs index a985786460..85215a1545 100644 --- a/crates/iceberg/src/inspect/manifests.rs +++ b/crates/iceberg/src/inspect/manifests.rs @@ -29,7 +29,10 @@ use futures::{StreamExt, stream}; use crate::Result; use crate::arrow::schema_to_arrow_schema; use crate::scan::ArrowRecordBatchStream; -use crate::spec::{Datum, FieldSummary, ListType, NestedField, PrimitiveType, StructType, Type}; +use crate::spec::{ + Datum, FieldSummary, ListType, ManifestContentType, NestedField, PrimitiveType, StructType, + Type, +}; use crate::table::Table; /// Manifests table. @@ -168,17 +171,38 @@ impl<'a> ManifestsTable<'a> { length.append_value(manifest.manifest_length); partition_spec_id.append_value(manifest.partition_spec_id); added_snapshot_id.append_value(manifest.added_snapshot_id); - added_data_files_count.append_value(manifest.added_files_count.unwrap_or(0) as i32); - existing_data_files_count - .append_value(manifest.existing_files_count.unwrap_or(0) as i32); - deleted_data_files_count - .append_value(manifest.deleted_files_count.unwrap_or(0) as i32); - added_delete_files_count - .append_value(manifest.added_files_count.unwrap_or(0) as i32); - existing_delete_files_count - .append_value(manifest.existing_files_count.unwrap_or(0) as i32); - deleted_delete_files_count - .append_value(manifest.deleted_files_count.unwrap_or(0) as i32); + let is_data_manifest = manifest.content == ManifestContentType::Data; + let is_delete_manifest = manifest.content == ManifestContentType::Deletes; + added_data_files_count.append_value(if is_data_manifest { + manifest.added_files_count.unwrap_or(0) as i32 + } else { + 0 + }); + existing_data_files_count.append_value(if is_data_manifest { + manifest.existing_files_count.unwrap_or(0) as i32 + } else { + 0 + }); + deleted_data_files_count.append_value(if is_data_manifest { + manifest.deleted_files_count.unwrap_or(0) as i32 + } else { + 0 + }); + added_delete_files_count.append_value(if is_delete_manifest { + manifest.added_files_count.unwrap_or(0) as i32 + } else { + 0 + }); + existing_delete_files_count.append_value(if is_delete_manifest { + manifest.existing_files_count.unwrap_or(0) as i32 + } else { + 0 + }); + deleted_delete_files_count.append_value(if is_delete_manifest { + manifest.deleted_files_count.unwrap_or(0) as i32 + } else { + 0 + }); let spec = self .table @@ -335,16 +359,121 @@ mod tests { ], added_delete_files_count: PrimitiveArray [ - 1, + 0, ], existing_delete_files_count: PrimitiveArray [ - 1, + 0, ], deleted_delete_files_count: PrimitiveArray [ + 0, + ], + partition_summaries: ListArray + [ + StructArray + -- validity: + [ + valid, + ] + [ + -- child 0: "contains_null" (Boolean) + BooleanArray + [ + false, + ] + -- child 1: "contains_nan" (Boolean) + BooleanArray + [ + false, + ] + -- child 2: "lower_bound" (Utf8) + StringArray + [ + "100", + ] + -- child 3: "upper_bound" (Utf8) + StringArray + [ + "300", + ] + ], + ]"#]], + &["path", "length"], + Some("path"), + ); + } + + #[tokio::test] + async fn test_manifests_table_data_and_delete_manifest() { + let mut fixture = TableTestFixture::new(); + fixture.setup_manifest_files_with_delete().await; + + let record_batch = fixture.table.inspect().manifests().scan().await.unwrap(); + + check_record_batches( + record_batch.try_collect::>().await.unwrap(), + expect![[r#" + Field { "content": Int32, metadata: {"PARQUET:field_id": "14"} }, + Field { "path": Utf8, metadata: {"PARQUET:field_id": "1"} }, + Field { "length": Int64, metadata: {"PARQUET:field_id": "2"} }, + Field { "partition_spec_id": Int32, metadata: {"PARQUET:field_id": "3"} }, + Field { "added_snapshot_id": Int64, metadata: {"PARQUET:field_id": "4"} }, + Field { "added_data_files_count": Int32, metadata: {"PARQUET:field_id": "5"} }, + Field { "existing_data_files_count": Int32, metadata: {"PARQUET:field_id": "6"} }, + Field { "deleted_data_files_count": Int32, metadata: {"PARQUET:field_id": "7"} }, + Field { "added_delete_files_count": Int32, metadata: {"PARQUET:field_id": "15"} }, + Field { "existing_delete_files_count": Int32, metadata: {"PARQUET:field_id": "16"} }, + Field { "deleted_delete_files_count": Int32, metadata: {"PARQUET:field_id": "17"} }, + Field { "partition_summaries": List(non-null Struct("contains_null": non-null Boolean, metadata: {"PARQUET:field_id": "10"}, "contains_nan": Boolean, metadata: {"PARQUET:field_id": "11"}, "lower_bound": Utf8, metadata: {"PARQUET:field_id": "12"}, "upper_bound": Utf8, metadata: {"PARQUET:field_id": "13"}), metadata: {"PARQUET:field_id": "9"}), metadata: {"PARQUET:field_id": "8"} }"#]], + expect![[r#" + content: PrimitiveArray + [ + 0, 1, ], + path: (skipped), + length: (skipped), + partition_spec_id: PrimitiveArray + [ + 0, + 0, + ], + added_snapshot_id: PrimitiveArray + [ + 3055729675574597004, + 3055729675574597004, + ], + added_data_files_count: PrimitiveArray + [ + 1, + 0, + ], + existing_data_files_count: PrimitiveArray + [ + 1, + 0, + ], + deleted_data_files_count: PrimitiveArray + [ + 1, + 0, + ], + added_delete_files_count: PrimitiveArray + [ + 0, + 1, + ], + existing_delete_files_count: PrimitiveArray + [ + 0, + 0, + ], + deleted_delete_files_count: PrimitiveArray + [ + 0, + 0, + ], partition_summaries: ListArray [ StructArray @@ -373,6 +502,33 @@ mod tests { [ "300", ] + ], + StructArray + -- validity: + [ + valid, + ] + [ + -- child 0: "contains_null" (Boolean) + BooleanArray + [ + false, + ] + -- child 1: "contains_nan" (Boolean) + BooleanArray + [ + false, + ] + -- child 2: "lower_bound" (Utf8) + StringArray + [ + "100", + ] + -- child 3: "upper_bound" (Utf8) + StringArray + [ + "100", + ] ], ]"#]], &["path", "length"], diff --git a/crates/iceberg/src/scan/mod.rs b/crates/iceberg/src/scan/mod.rs index 9e12ffa4a9..1797ab78d4 100644 --- a/crates/iceberg/src/scan/mod.rs +++ b/crates/iceberg/src/scan/mod.rs @@ -831,6 +831,16 @@ pub mod tests { .unwrap() } + /// Like [`Self::next_manifest_file`], but with a caller-chosen (deterministic) + /// file name instead of a random UUID. Used where tests need a stable sort + /// order across manifests (e.g. sorting the manifests table by `path`). + fn named_manifest_file(&self, name: &str) -> OutputFile { + self.table + .file_io() + .new_output(format!("{}/metadata/{}.avro", self.table_location, name)) + .unwrap() + } + pub async fn setup_manifest_files(&mut self) { let current_snapshot = self.table.metadata().current_snapshot().unwrap(); let parent_snapshot = current_snapshot @@ -936,6 +946,144 @@ pub mod tests { manifest_list_write.close().await.unwrap(); } + /// Same as [`Self::setup_manifest_files`], but additionally writes a v2 + /// deletes manifest (one added `PositionDeletes` entry) and references + /// both manifests from the current snapshot's manifest list. + pub async fn setup_manifest_files_with_delete(&mut self) { + 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(); + + // Write the data files first, then use the file size in the manifest entries + let parquet_file_size = self.write_parquet_data_files(); + + let mut writer = ManifestWriterBuilder::new( + self.named_manifest_file("manifest-data"), + Some(current_snapshot.snapshot_id()), + current_schema.clone(), + current_partition_spec.as_ref().clone(), + ) + .build_v2_data(); + writer + .add_entry( + ManifestEntry::builder() + .status(ManifestStatus::Added) + .data_file( + DataFileBuilder::default() + .partition_spec_id(0) + .content(DataContentType::Data) + .file_path(format!("{}/1.parquet", &self.table_location)) + .file_format(DataFileFormat::Parquet) + .file_size_in_bytes(parquet_file_size) + .record_count(1) + .partition(Struct::from_iter([Some(Literal::long(100))])) + .key_metadata(None) + .build() + .unwrap(), + ) + .build(), + ) + .unwrap(); + writer + .add_delete_entry( + ManifestEntry::builder() + .status(ManifestStatus::Deleted) + .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(format!("{}/2.parquet", &self.table_location)) + .file_format(DataFileFormat::Parquet) + .file_size_in_bytes(parquet_file_size) + .record_count(1) + .partition(Struct::from_iter([Some(Literal::long(200))])) + .build() + .unwrap(), + ) + .build(), + ) + .unwrap(); + 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(format!("{}/3.parquet", &self.table_location)) + .file_format(DataFileFormat::Parquet) + .file_size_in_bytes(parquet_file_size) + .record_count(1) + .partition(Struct::from_iter([Some(Literal::long(300))])) + .build() + .unwrap(), + ) + .build(), + ) + .unwrap(); + let data_file_manifest = writer.write_manifest_file().await.unwrap(); + + // A v2 deletes manifest with a single added position-delete entry. + let mut deletes_writer = ManifestWriterBuilder::new( + self.named_manifest_file("manifest-deletes"), + Some(current_snapshot.snapshot_id()), + current_schema.clone(), + current_partition_spec.as_ref().clone(), + ) + .build_v2_deletes(); + deletes_writer + .add_entry( + ManifestEntry::builder() + .status(ManifestStatus::Added) + .data_file( + DataFileBuilder::default() + .partition_spec_id(0) + .content(DataContentType::PositionDeletes) + .file_path(format!("{}/4-deletes.parquet", &self.table_location)) + .file_format(DataFileFormat::Parquet) + .file_size_in_bytes(1024) + .record_count(1) + .partition(Struct::from_iter([Some(Literal::long(100))])) + .key_metadata(None) + .build() + .unwrap(), + ) + .build(), + ) + .unwrap(); + let deletes_manifest = deletes_writer.write_manifest_file().await.unwrap(); + + // Write to manifest list + let manifest_list_writer = self + .table + .file_io() + .new_output(current_snapshot.manifest_list()) + .unwrap() + .writer() + .await + .unwrap(); + let mut manifest_list_write = ManifestListWriter::v2( + manifest_list_writer, + current_snapshot.snapshot_id(), + current_snapshot.parent_snapshot_id(), + current_snapshot.sequence_number(), + ); + manifest_list_write + .add_manifests(vec![data_file_manifest, deletes_manifest].into_iter()) + .unwrap(); + manifest_list_write.close().await.unwrap(); + } + /// Writes identical Parquet data files (1.parquet, 2.parquet, 3.parquet) /// and returns the file size in bytes. fn write_parquet_data_files(&self) -> u64 {