CTTY commented on code in PR #2367:
URL: https://github.com/apache/iceberg-rust/pull/2367#discussion_r3140310235
##########
crates/iceberg/src/transaction/snapshot.rs:
##########
@@ -312,13 +325,69 @@ impl<'a> SnapshotProducer<'a> {
builder.build()
}
});
- let mut writer = self.new_manifest_writer(ManifestContentType::Data)?;
+ let mut writer = self.new_manifest_writer(
+ content_type,
+ self.table.metadata().default_partition_spec_id(),
+ )?;
for entry in manifest_entries {
writer.add_entry(entry)?;
}
writer.write_manifest_file().await
}
+ async fn write_deleted_manifest(
+ &mut self,
+ deleted_entries: Vec<ManifestEntry>,
+ ) -> Result<Vec<ManifestFile>> {
+ if deleted_entries.is_empty() {
+ Ok(Vec::new())
+ } else {
+ // Initialize partition groups
+ let mut partition_groups = HashMap::new();
+ for entry in deleted_entries {
+ partition_groups
+ .entry(entry.data_file().partition_spec_id)
+ .or_insert_with(Vec::new)
+ .push(entry);
+ }
+
+ // Write manifest files for each spec-entries pair
+ let mut deleted_manifests = Vec::new();
+ for (spec_id, entries) in partition_groups {
Review Comment:
do we need to use runtime here? I think probably no because this is rather
light weight
##########
crates/iceberg/src/transaction/snapshot.rs:
##########
@@ -289,8 +295,15 @@ impl<'a> SnapshotProducer<'a> {
}
// Write manifest file for added data files and return the ManifestFile
for ManifestList.
- async fn write_added_manifest(&mut self) -> Result<ManifestFile> {
- let added_data_files = std::mem::take(&mut self.added_data_files);
+ async fn write_added_manifest(
+ &mut self,
+ content_type: ManifestContentType,
+ ) -> Result<ManifestFile> {
+ let added_data_files = match content_type {
Review Comment:
the name should be `added_files`
##########
crates/iceberg/src/transaction/snapshot.rs:
##########
@@ -107,41 +107,39 @@ pub(crate) trait ManifestProcess: Send + Sync {
) -> Vec<ManifestFile>;
}
+#[derive(TypedBuilder)]
+#[builder(field_defaults(setter(prefix = "with_")))]
pub(crate) struct SnapshotProducer<'a> {
pub(crate) table: &'a Table,
+ #[builder(
+ setter(skip),
+ default_code = "SnapshotProducer::generate_unique_snapshot_id(table)"
+ )]
snapshot_id: i64,
commit_uuid: Uuid,
+ #[builder(default)]
key_metadata: Option<Vec<u8>>,
+ #[builder(default)]
snapshot_properties: HashMap<String, String>,
+ #[builder(default)]
added_data_files: Vec<DataFile>,
+ #[builder(default)]
+ added_delete_files: Vec<DataFile>,
+ #[builder(default)]
+ pub deleted_data_files: Vec<DataFile>,
+ #[builder(default)]
+ pub deleted_delete_files: Vec<DataFile>,
Review Comment:
removed_*_files are better names
##########
crates/iceberg/src/transaction/snapshot.rs:
##########
@@ -312,13 +325,69 @@ impl<'a> SnapshotProducer<'a> {
builder.build()
}
});
- let mut writer = self.new_manifest_writer(ManifestContentType::Data)?;
+ let mut writer = self.new_manifest_writer(
+ content_type,
+ self.table.metadata().default_partition_spec_id(),
+ )?;
for entry in manifest_entries {
writer.add_entry(entry)?;
}
writer.write_manifest_file().await
}
+ async fn write_deleted_manifest(
+ &mut self,
+ deleted_entries: Vec<ManifestEntry>,
+ ) -> Result<Vec<ManifestFile>> {
+ if deleted_entries.is_empty() {
+ Ok(Vec::new())
+ } else {
+ // Initialize partition groups
+ let mut partition_groups = HashMap::new();
+ for entry in deleted_entries {
+ partition_groups
+ .entry(entry.data_file().partition_spec_id)
+ .or_insert_with(Vec::new)
+ .push(entry);
+ }
+
+ // Write manifest files for each spec-entries pair
+ let mut deleted_manifests = Vec::new();
+ for (spec_id, entries) in partition_groups {
+ let mut data_manifest_writer: Option<ManifestWriter> = None;
+ let mut delete_manifest_writer: Option<ManifestWriter> = None;
+ for entry in entries {
+ match entry.data_file().content_type() {
+ DataContentType::Data => data_manifest_writer
+ .get_or_insert(
+
self.new_manifest_writer(ManifestContentType::Data, spec_id)?,
+ )
+ .add_entry(entry)?,
Review Comment:
Is this correct? need to double check. entry is deleted_entry
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
