JanKaul commented on code in PR #3046: URL: https://github.com/apache/iceberg-rust/pull/3046#discussion_r3842337390
########## crates/iceberg/src/transaction/merging.rs: ########## @@ -0,0 +1,398 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Shared engine for snapshot-producing operations that both add and delete files. +//! +//! [`MergingSnapshotProducer`] is the Rust equivalent of Java's +//! `MergingSnapshotProducer`. It handles manifest filtering, new manifest +//! creation, summary computation, and delegates the final snapshot commit to +//! [`SnapshotProducer`]. + +use std::collections::{HashMap, HashSet}; + +use uuid::Uuid; + +use crate::error::Result; +use crate::io::OutputFile; +use crate::spec::{ + DataContentType, DataFile, DataFileFormat, FormatVersion, ManifestContentType, ManifestEntry, + ManifestFile, ManifestStatus, ManifestWriter, ManifestWriterBuilder, Operation, PartitionSpec, + SchemaRef, SnapshotSummaryCollector, Summary, update_snapshot_summaries, +}; +use crate::table::Table; +use crate::transaction::ActionCommit; +use crate::transaction::snapshot::SnapshotProducer; +use crate::{Error, ErrorKind}; + +/// Create a manifest writer that handles encryption when available. +fn new_manifest_writer( + table: &Table, + output_file: OutputFile, + snapshot_id: Option<i64>, + content: ManifestContentType, + schema: SchemaRef, + partition_spec: PartitionSpec, +) -> Result<ManifestWriter> { + let builder = if let Some(em) = table.encryption_manager() { + ManifestWriterBuilder::new_from_encrypted( + em.encrypt(output_file), + snapshot_id, + schema, + partition_spec, + )? + } else { + ManifestWriterBuilder::new(output_file, snapshot_id, schema, partition_spec) + }; + + match table.metadata().format_version() { + FormatVersion::V1 => Ok(builder.build_v1()), + FormatVersion::V2 => match content { + ManifestContentType::Data => Ok(builder.build_v2_data()), + ManifestContentType::Deletes => Ok(builder.build_v2_deletes()), + }, + FormatVersion::V3 => match content { + ManifestContentType::Data => Ok(builder.build_v3_data()), + ManifestContentType::Deletes => Ok(builder.build_v3_deletes()), + }, + } +} + +/// Filters existing manifests by removing entries for deleted data files. +/// +/// This is equivalent to Java's `ManifestFilterManager`. When a rewrite or +/// overwrite operation deletes files, the filter manager rewrites affected +/// manifests so that deleted entries are dropped and surviving entries are +/// re-emitted with [`ManifestStatus::Existing`]. +pub(crate) struct ManifestFilterManager { + deleted_file_paths: HashSet<String>, + fail_missing_delete_paths: bool, +} + +impl ManifestFilterManager { + pub(crate) fn new(fail_missing_delete_paths: bool) -> Self { + Self { + deleted_file_paths: HashSet::new(), + fail_missing_delete_paths, + } + } + + pub(crate) fn add_delete(&mut self, path: String) { + self.deleted_file_paths.insert(path); + } + + /// Filter `manifests` by removing entries whose file path is in the delete + /// set. Returns the surviving manifests plus a [`SnapshotSummaryCollector`] + /// that recorded metrics for every removed file. + pub(crate) async fn filter_manifests( + &self, + table: &Table, + manifests: Vec<ManifestFile>, + snapshot_id: i64, + ) -> Result<(Vec<ManifestFile>, SnapshotSummaryCollector)> { + if self.deleted_file_paths.is_empty() { + return Ok((manifests, SnapshotSummaryCollector::default())); + } + + let mut result: Vec<ManifestFile> = Vec::with_capacity(manifests.len()); + let mut removed_collector = SnapshotSummaryCollector::default(); + let mut found_paths: HashSet<String> = HashSet::new(); + + for manifest_file in &manifests { + // Only filter data manifests; pass delete manifests through unchanged. + if manifest_file.content != ManifestContentType::Data { + result.push(manifest_file.clone()); + continue; + } + + let manifest = table.manifest_reader().read(manifest_file).await?; + + // Check whether this manifest contains any files we want to delete. + let has_deletes = manifest + .entries() + .iter() + .any(|e| e.is_alive() && self.deleted_file_paths.contains(e.file_path())); + + if !has_deletes { + // Manifest is unaffected — pass through verbatim. + result.push(manifest_file.clone()); + continue; + } + + // Resolve the partition spec for this manifest. After partition + // evolution, old manifests may carry a non-default spec — we must + // use the manifest's own spec so the rewritten manifest stays valid. + let manifest_spec = table + .metadata() + .partition_spec_by_id(manifest_file.partition_spec_id) + .ok_or_else(|| { + Error::new( + ErrorKind::DataInvalid, + format!( + "Manifest references unknown partition spec {}", + manifest_file.partition_spec_id, + ), + ) + })? + .clone(); + let schema = table.metadata().current_schema().clone(); + + // Rewrite: keep surviving entries as EXISTING, drop deleted ones. + let mut surviving_entries: Vec<ManifestEntry> = Vec::new(); + for entry in manifest.entries() { + if entry.is_alive() && self.deleted_file_paths.contains(entry.file_path()) { + // Record removal metrics. + found_paths.insert(entry.file_path().to_string()); + removed_collector.remove_file( + entry.data_file(), + schema.clone(), + manifest_spec.clone(), + ); + } else if entry.is_alive() { + // Surviving entry — re-emit as EXISTING with original ids preserved. + let existing = ManifestEntry::builder() + .status(ManifestStatus::Existing) + .snapshot_id( + entry + .snapshot_id() + .unwrap_or(manifest_file.added_snapshot_id), + ) + .sequence_number(entry.sequence_number().unwrap_or(0)) + .file_sequence_number(entry.file_sequence_number.unwrap_or(0)) + .data_file(entry.data_file().clone()) + .build(); + surviving_entries.push(existing); + } + // Already-deleted entries (status == Deleted) are dropped. + } + + if surviving_entries.is_empty() { + // Manifest is now empty — omit entirely. + continue; + } + + // Write the filtered manifest using the manifest's own partition spec. + let new_manifest_path = format!( + "{}/{}-m-filter-{}.{}", + table.metadata().metadata_location()?, Review Comment: The manifest naming here deviates from the standard naming scheme `<commit-uuid>-m<counter>.avro`. I personally would stay with the standard scheme and don't invent any new concepts. Technically it would be fine. ########## crates/iceberg/src/transaction/merging.rs: ########## @@ -0,0 +1,398 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Shared engine for snapshot-producing operations that both add and delete files. +//! +//! [`MergingSnapshotProducer`] is the Rust equivalent of Java's +//! `MergingSnapshotProducer`. It handles manifest filtering, new manifest +//! creation, summary computation, and delegates the final snapshot commit to +//! [`SnapshotProducer`]. + +use std::collections::{HashMap, HashSet}; + +use uuid::Uuid; + +use crate::error::Result; +use crate::io::OutputFile; +use crate::spec::{ + DataContentType, DataFile, DataFileFormat, FormatVersion, ManifestContentType, ManifestEntry, + ManifestFile, ManifestStatus, ManifestWriter, ManifestWriterBuilder, Operation, PartitionSpec, + SchemaRef, SnapshotSummaryCollector, Summary, update_snapshot_summaries, +}; +use crate::table::Table; +use crate::transaction::ActionCommit; +use crate::transaction::snapshot::SnapshotProducer; +use crate::{Error, ErrorKind}; + +/// Create a manifest writer that handles encryption when available. +fn new_manifest_writer( + table: &Table, + output_file: OutputFile, + snapshot_id: Option<i64>, + content: ManifestContentType, + schema: SchemaRef, + partition_spec: PartitionSpec, +) -> Result<ManifestWriter> { + let builder = if let Some(em) = table.encryption_manager() { + ManifestWriterBuilder::new_from_encrypted( + em.encrypt(output_file), + snapshot_id, + schema, + partition_spec, + )? + } else { + ManifestWriterBuilder::new(output_file, snapshot_id, schema, partition_spec) + }; + + match table.metadata().format_version() { + FormatVersion::V1 => Ok(builder.build_v1()), + FormatVersion::V2 => match content { + ManifestContentType::Data => Ok(builder.build_v2_data()), + ManifestContentType::Deletes => Ok(builder.build_v2_deletes()), + }, + FormatVersion::V3 => match content { + ManifestContentType::Data => Ok(builder.build_v3_data()), + ManifestContentType::Deletes => Ok(builder.build_v3_deletes()), + }, + } +} + +/// Filters existing manifests by removing entries for deleted data files. +/// +/// This is equivalent to Java's `ManifestFilterManager`. When a rewrite or +/// overwrite operation deletes files, the filter manager rewrites affected +/// manifests so that deleted entries are dropped and surviving entries are +/// re-emitted with [`ManifestStatus::Existing`]. +pub(crate) struct ManifestFilterManager { + deleted_file_paths: HashSet<String>, + fail_missing_delete_paths: bool, +} + +impl ManifestFilterManager { + pub(crate) fn new(fail_missing_delete_paths: bool) -> Self { + Self { + deleted_file_paths: HashSet::new(), + fail_missing_delete_paths, + } + } + + pub(crate) fn add_delete(&mut self, path: String) { + self.deleted_file_paths.insert(path); + } + + /// Filter `manifests` by removing entries whose file path is in the delete + /// set. Returns the surviving manifests plus a [`SnapshotSummaryCollector`] + /// that recorded metrics for every removed file. + pub(crate) async fn filter_manifests( + &self, + table: &Table, + manifests: Vec<ManifestFile>, + snapshot_id: i64, + ) -> Result<(Vec<ManifestFile>, SnapshotSummaryCollector)> { + if self.deleted_file_paths.is_empty() { + return Ok((manifests, SnapshotSummaryCollector::default())); + } + + let mut result: Vec<ManifestFile> = Vec::with_capacity(manifests.len()); + let mut removed_collector = SnapshotSummaryCollector::default(); + let mut found_paths: HashSet<String> = HashSet::new(); + + for manifest_file in &manifests { + // Only filter data manifests; pass delete manifests through unchanged. + if manifest_file.content != ManifestContentType::Data { + result.push(manifest_file.clone()); + continue; + } + + let manifest = table.manifest_reader().read(manifest_file).await?; + + // Check whether this manifest contains any files we want to delete. + let has_deletes = manifest + .entries() + .iter() + .any(|e| e.is_alive() && self.deleted_file_paths.contains(e.file_path())); + + if !has_deletes { + // Manifest is unaffected — pass through verbatim. + result.push(manifest_file.clone()); + continue; + } + + // Resolve the partition spec for this manifest. After partition + // evolution, old manifests may carry a non-default spec — we must + // use the manifest's own spec so the rewritten manifest stays valid. + let manifest_spec = table + .metadata() + .partition_spec_by_id(manifest_file.partition_spec_id) + .ok_or_else(|| { + Error::new( + ErrorKind::DataInvalid, + format!( + "Manifest references unknown partition spec {}", + manifest_file.partition_spec_id, + ), + ) + })? + .clone(); + let schema = table.metadata().current_schema().clone(); + + // Rewrite: keep surviving entries as EXISTING, drop deleted ones. + let mut surviving_entries: Vec<ManifestEntry> = Vec::new(); + for entry in manifest.entries() { + if entry.is_alive() && self.deleted_file_paths.contains(entry.file_path()) { + // Record removal metrics. + found_paths.insert(entry.file_path().to_string()); + removed_collector.remove_file( + entry.data_file(), + schema.clone(), + manifest_spec.clone(), + ); + } else if entry.is_alive() { + // Surviving entry — re-emit as EXISTING with original ids preserved. + let existing = ManifestEntry::builder() + .status(ManifestStatus::Existing) + .snapshot_id( + entry + .snapshot_id() + .unwrap_or(manifest_file.added_snapshot_id), + ) + .sequence_number(entry.sequence_number().unwrap_or(0)) + .file_sequence_number(entry.file_sequence_number.unwrap_or(0)) Review Comment: I think the sequence_number here shouldn't silently be set to 0 here. True, this should always be a `Some()` here but it's still better to throw an explicit error. -- 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]
