blackmwk commented on code in PR #2620: URL: https://github.com/apache/iceberg-rust/pull/2620#discussion_r3543127839
########## crates/iceberg/src/transaction/delete_aware.rs: ########## @@ -0,0 +1,275 @@ +// 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. + +use std::collections::HashSet; +use std::future::Future; + +use futures::SinkExt; +use futures::future::try_join_all; +use once_cell::sync::Lazy; + +use crate::delete_file_index::DeleteFileIndex; +use crate::error::Result; +use crate::scan::DeleteFileContext; +use crate::spec::{ + DataContentType, DataFile, FormatVersion, INITIAL_SEQUENCE_NUMBER, ManifestContentType, + ManifestFile, Operation, +}; +use crate::table::Table; +use crate::transaction::manifest_filter::ManifestFilterManager; +use crate::transaction::snapshot::SnapshotProduceOperation; +use crate::util::snapshot::ancestors_between; +use crate::{Error, ErrorKind}; + +/// Operations whose snapshots may add delete files. +static VALIDATE_ADDED_DELETE_FILES_OPERATIONS: Lazy<HashSet<Operation>> = + Lazy::new(|| HashSet::from([Operation::Overwrite, Operation::Delete])); + +/// An additive sub-trait of [`SnapshotProduceOperation`] implemented only by delete-class +/// operations (delete, overwrite, rewrite). +/// +/// Append-only operations implement only the base [`SnapshotProduceOperation`] and carry +/// none of the validation/filtering surface defined here. A `DeleteAwareOperation` adds two +/// capabilities on top of the base trait: +/// +/// - **Write-time conflict validation** via [`validate`](DeleteAwareOperation::validate), +/// invoked by the action against the refreshed base table before any snapshot is written. +/// - **Manifest filtering** via the owned [`ManifestFilterManager`]s reached through +/// [`data_filter`](DeleteAwareOperation::data_filter) and +/// [`delete_filter`](DeleteAwareOperation::delete_filter), which rewrite carried-forward +/// manifests to drop the files the operation removes. +/// +/// The filter managers are owned fields on the operation, populated incrementally as the +/// action builds the operation. Because the operation is rebuilt on each commit attempt, the +/// managers are reconstructed deterministically per attempt from the action's stored inputs. +#[allow(unused)] +pub(crate) trait DeleteAwareOperation: SnapshotProduceOperation { Review Comment: Why we need this trait? Almost all operations has deletes ########## crates/iceberg/src/transaction/rewrite_files.rs: ########## @@ -0,0 +1,455 @@ +// 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. + +use std::collections::HashMap; +use std::sync::Arc; + +use async_trait::async_trait; +use uuid::Uuid; + +use crate::error::Result; +use crate::spec::{ + DataContentType, DataFile, ManifestContentType, ManifestEntry, ManifestFile, Operation, +}; +use crate::table::Table; +use crate::transaction::delete_aware::DeleteAwareOperation; +use crate::transaction::manifest_filter::ManifestFilterManager; +use crate::transaction::snapshot::{ + DefaultManifestProcess, SnapshotProduceOperation, SnapshotProducer, +}; +use crate::transaction::{ActionCommit, TransactionAction}; +use crate::{Error, ErrorKind}; + +/// A transaction action that rewrites a set of existing files with a new, equivalent set +/// (i.e. compaction). +/// +/// `RewriteFiles` adds new files and deletes old ones in a single `Replace` snapshot. Added +/// and deleted files are bucketed by [`DataContentType`] as they come in: data files land in +/// the data buckets, while positional/equality delete files land in the delete buckets (this +/// split mirrors `#1606`). +/// +/// The action stores its inputs and rebuilds a fresh [`RewriteFilesOperation`] on each commit +/// attempt via [`build_operation`](RewriteFilesAction::build_operation), so the operation's +/// filter managers are repopulated deterministically and survive `do_commit` retries. +pub struct RewriteFilesAction { + // below are properties used to create SnapshotProducer when commit + commit_uuid: Option<Uuid>, + key_metadata: Option<Vec<u8>>, + snapshot_properties: HashMap<String, String>, + // Added/deleted files split by content type, as in #1606. + added_data_files: Vec<DataFile>, + added_delete_files: Vec<DataFile>, + deleted_data_files: Vec<DataFile>, + deleted_delete_files: Vec<DataFile>, + // When set, only positional deletes count as conflicts during validation. + data_sequence_number: Option<i64>, + // The snapshot the rewrite started from; validation scans for new deletes since this id. + starting_snapshot_id: Option<i64>, +} + +impl RewriteFilesAction { + pub(crate) fn new() -> Self { + Self { + commit_uuid: None, + key_metadata: None, + snapshot_properties: HashMap::default(), + added_data_files: vec![], + added_delete_files: vec![], + deleted_data_files: vec![], + deleted_delete_files: vec![], + data_sequence_number: None, + starting_snapshot_id: None, + } + } + + /// Add files produced by the rewrite, bucketing each by its content type: data files go + /// to the data bucket; positional/equality delete files go to the delete bucket. + pub fn add_data_files(mut self, files: impl IntoIterator<Item = DataFile>) -> Result<Self> { + for f in files { + match f.content_type() { + DataContentType::Data => self.added_data_files.push(f), + DataContentType::PositionDeletes | DataContentType::EqualityDeletes => { + self.added_delete_files.push(f) + } + } + } + Ok(self) + } + + /// Mark files removed by the rewrite, bucketing each by its content type: data files go + /// to the data bucket; positional/equality delete files go to the delete bucket. + pub fn delete_files(mut self, files: impl IntoIterator<Item = DataFile>) -> Result<Self> { + for f in files { + match f.content_type() { + DataContentType::Data => self.deleted_data_files.push(f), + DataContentType::PositionDeletes | DataContentType::EqualityDeletes => { + self.deleted_delete_files.push(f) + } + } + } + Ok(self) + } + + /// Set the snapshot the rewrite started from. Conflict validation scans for new delete + /// files added since this snapshot. + pub fn set_starting_snapshot_id(mut self, id: i64) -> Self { + self.starting_snapshot_id = Some(id); + self + } + + /// Pin the data sequence number for the rewrite. When set, only positional deletes are + /// treated as conflicts during validation. + pub fn set_data_sequence_number(mut self, seq: i64) -> Self { + self.data_sequence_number = Some(seq); + self + } + + /// Set commit UUID for the snapshot. + pub fn set_commit_uuid(mut self, commit_uuid: Uuid) -> Self { + self.commit_uuid = Some(commit_uuid); + self + } + + /// Set key metadata for manifest files. + pub fn set_key_metadata(mut self, key_metadata: Vec<u8>) -> Self { + self.key_metadata = Some(key_metadata); + self + } + + /// Set snapshot summary properties. + pub fn set_snapshot_properties(mut self, snapshot_properties: HashMap<String, String>) -> Self { + self.snapshot_properties = snapshot_properties; + self + } + + /// Build a fresh [`RewriteFilesOperation`] from the action's stored inputs. + /// + /// A new operation is built per commit attempt; the filter managers are repopulated + /// deterministically from the deleted-file vectors (so the operation survives `do_commit` + /// retries). Removed files become filter-manager removals via + /// [`ManifestFilterManager::delete_file`] — data files into the data filter, delete files + /// into the delete filter (`DataFile` covers both kinds). + fn build_operation(&self) -> RewriteFilesOperation { + let mut op = RewriteFilesOperation { + deleted_data_files: self.deleted_data_files.clone(), + deleted_delete_files: self.deleted_delete_files.clone(), + starting_snapshot_id: self.starting_snapshot_id, + data_sequence_number: self.data_sequence_number, + data_filter: ManifestFilterManager::default(), + delete_filter: ManifestFilterManager::default(), + }; + + // Removed files become filter-manager removals (DataFile covers both kinds). + for f in &self.deleted_data_files { + op.data_filter.delete_file(f.clone()); + } + for f in &self.deleted_delete_files { + op.delete_filter.delete_file(f.clone()); + } + + op + } +} + +/// The operation driven by [`SnapshotProducer`](crate::transaction::snapshot::SnapshotProducer) +/// for a `RewriteFiles` commit. +/// +/// Rebuilt per commit attempt by [`RewriteFilesAction::build_operation`]. The +/// `SnapshotProduceOperation` and `DeleteAwareOperation` implementations are wired in +/// subsequent subtasks (5.2–5.4). +pub(crate) struct RewriteFilesOperation { + deleted_data_files: Vec<DataFile>, + deleted_delete_files: Vec<DataFile>, + starting_snapshot_id: Option<i64>, + data_sequence_number: Option<i64>, + data_filter: ManifestFilterManager, + delete_filter: ManifestFilterManager, +} + +impl SnapshotProduceOperation for RewriteFilesOperation { + fn operation(&self) -> Operation { + Operation::Replace + } + + async fn delete_entries( + &self, + _snapshot_produce: &SnapshotProducer<'_>, + ) -> Result<Vec<ManifestEntry>> { + // Removed files are dropped by rewriting the carried-forward manifests in + // `existing_manifest` via the filter managers, so no separate delete entries are + // emitted here. + Ok(vec![]) + } + + async fn existing_manifest( + &mut self, + snapshot_produce: &mut SnapshotProducer<'_>, + ) -> Result<Vec<ManifestFile>> { + // No current snapshot means there are no manifests to carry forward. + let Some(snapshot) = snapshot_produce.table.metadata().current_snapshot() else { + return Ok(vec![]); + }; + + let manifests = snapshot_produce + .table + .manifest_list_reader(snapshot) + .load() + .await? + .entries() + .to_vec(); + + // Partition the carried-forward manifests by content type so each filter manager + // rewrites only the manifests it owns: data manifests through `data_filter`, + // positional/equality delete manifests through `delete_filter`. + let (data, deletes): (Vec<ManifestFile>, Vec<ManifestFile>) = manifests + .into_iter() + .partition(|m| m.content == ManifestContentType::Data); + + let mut out = self + .data_filter + .filter_manifests(snapshot_produce, data) + .await?; + out.extend( + self.delete_filter + .filter_manifests(snapshot_produce, deletes) + .await?, + ); + + Ok(out) + } +} + +impl DeleteAwareOperation for RewriteFilesOperation { + async fn validate(&self, base: &Table, parent_snapshot_id: Option<i64>) -> Result<()> { + // Precondition: a rewrite must remove at least one file. (Intent from #1606.) + if self.deleted_data_files.is_empty() && self.deleted_delete_files.is_empty() { + return Err(Error::new( + ErrorKind::DataInvalid, + "Files to delete cannot be empty", + )); + } + + // When data files are being rewritten, ensure no new row-level deletes have appeared + // for them since the starting snapshot. When a data sequence number is pinned, only + // positional deletes count as conflicts (`ignore_equality_deletes`), following #1606. + if !self.deleted_data_files.is_empty() { + self.validate_no_new_deletes_for_data_files( + base, + self.starting_snapshot_id, + parent_snapshot_id, + &self.deleted_data_files, + self.data_sequence_number.is_some(), + ) + .await?; + } + + Ok(()) + } + + fn data_filter(&mut self) -> &mut ManifestFilterManager { + &mut self.data_filter + } + + fn delete_filter(&mut self) -> &mut ManifestFilterManager { + &mut self.delete_filter + } +} + +#[async_trait] +impl TransactionAction for RewriteFilesAction { + async fn commit(self: Arc<Self>, table: &Table) -> Result<ActionCommit> { + // Rebuild a fresh operation for this attempt; the filter managers are repopulated + // deterministically from the action's stored inputs (survives `do_commit` retries). + let op = self.build_operation(); + + // STEP 1 — conflict validation before any write. Scans the refreshed base table for + // concurrent changes (e.g. new row-level deletes for rewritten data files) since the + // starting snapshot. + op.validate(table, table.metadata().current_snapshot_id()) + .await?; + + // STEP 2 — produce the `Replace` snapshot. The producer drives the operation by `&mut` + // so the filter managers can rewrite the carried-forward manifests inside + // `existing_manifest`. + let snapshot_producer = SnapshotProducer::new( + table, + self.commit_uuid.unwrap_or_else(Uuid::now_v7), + self.key_metadata.clone(), + self.snapshot_properties.clone(), + self.added_data_files.clone(), + ); + + snapshot_producer.commit(op, DefaultManifestProcess).await Review Comment: This is actually incorrect, what if there are conflicts? Do we need to repeat this try logic everywhere? -- 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]
