Skip to content

fix: gate manifests table file counts by manifest content type #2813

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

Open
gmhelmold wants to merge 1 commit into
base: main
Choose a base branch
from
+318 −14
Open

fix: gate manifests table file counts by manifest content type #2813

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
184 changes: 170 additions & 14 deletions crates/iceberg/src/inspect/manifests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -335,16 +359,121 @@ mod tests {
],
added_delete_files_count: PrimitiveArray<Int32>
[
1,
0,
],
existing_delete_files_count: PrimitiveArray<Int32>
[
1,
0,
],
deleted_delete_files_count: PrimitiveArray<Int32>
[
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::<Vec<_>>().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<Int32>
[
0,
1,
],
path: (skipped),
length: (skipped),
partition_spec_id: PrimitiveArray<Int32>
[
0,
0,
],
added_snapshot_id: PrimitiveArray<Int64>
[
3055729675574597004,
3055729675574597004,
],
added_data_files_count: PrimitiveArray<Int32>
[
1,
0,
],
existing_data_files_count: PrimitiveArray<Int32>
[
1,
0,
],
deleted_data_files_count: PrimitiveArray<Int32>
[
1,
0,
],
added_delete_files_count: PrimitiveArray<Int32>
[
0,
1,
],
existing_delete_files_count: PrimitiveArray<Int32>
[
0,
0,
],
deleted_delete_files_count: PrimitiveArray<Int32>
[
0,
0,
],
partition_summaries: ListArray
[
StructArray
Expand Down Expand Up @@ -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"],
Expand Down
148 changes: 148 additions & 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 @@ -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
Expand Down Expand Up @@ -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 {
Expand Down
Loading