CTTY commented on code in PR #2620: URL: https://github.com/apache/iceberg-rust/pull/2620#discussion_r3547810025
########## 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: I think we can put this in SnapshotProduceOperation ########## 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: I think this is a limitation with the current design: there is no cache that survives across multiple retries. Currently, we assume `TransactionAction`s are immutable. But it sounds like we should store caches/states in them directly to function as `MergingSnapshotProducer`? My current (oversimplified) mental model is: 1. Adding a `Mutex<MergeCache>` in actions like `RewriteFilesAction` to provide a cache with interior mutability 2. When calling `TransactionAction::commit`, the action/cache should pass states to `SnapshotProducer` so they can be committed Basically we don't add a java-like `MergingSnapshotProducer`, we just store the mutable diff in actions as `MergeCache`. Would love to hear more opinions on this direction! ########## docs/rfcs/conflict-detection.md: ########## @@ -0,0 +1,444 @@ +<!-- + 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. +--> + +# Design Document: Snapshot Conflict Validation + +## 1. Problem Statement + +Snapshot production on `main` only supports append-only operations. There is no +machinery for the operations that **remove existing files** from a table — delete, +overwrite, and rewrite. Those operations need two capabilities that append does not: + +1. **Write-time conflict validation** — before writing a new snapshot, the commit must + check the refreshed base table for concurrent changes that would make the pending + operation incorrect (e.g. a concurrent commit that added delete files for data files + this operation is about to rewrite). +2. **Manifest filtering** — instead of carrying existing manifests forward verbatim, a + delete-class operation must rewrite manifests to drop the files it removes. + +iceberg-java packages all of this into a single `MergingSnapshotProducer` base class via +implementation inheritance. Rust has no implementation inheritance, and the append path is +already factored differently. We need a shape that adds validation + filtering for +delete-class operations **without disturbing the append-only path** and without recreating +a Java-style inheritance chain. Review Comment: Yes, I want it to be as rust idiomatic as possible. A mapping to explain which traits/struct in rust functions as X class in java makes sense to me ########## 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![]); + }; Review Comment: I expect it to be a no-op. But haven't got a chance to check how this is like in java yet -- 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]
