dannycjones commented on code in PR #2620: URL: https://github.com/apache/iceberg-rust/pull/2620#discussion_r3452576004
########## 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])); Review Comment: nit: still open for me, we discussed here https://github.com/apache/iceberg-rust/pull/2590/changes#r3360509392 I want to make sure its purpose is abundantly clear, this is my attempt. (Maybe I have misunderstood anyway.) Plus, I'd like it to be a bit more idiomatic if possible. ```rust /// Returns if the given operation may have introduced deletes that apply to new data files. /// /// Some operations may add new delete files but they do not affect the logical content - for example, a replace may rewrite delete files by applying deletes and retaining the deletes not applied, or by rewriting equality deletes as positional/vector deletes. /// In this case, we do not consider them as conflicting. fn may_add_conflicting_deletes(operation: Operation) -> boolean { match operation { Operation::Overwrite | Operation::Delete => true, Operation::Append | Operation::Replace => false, } } ``` ########## 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. + +This document defines the trait/struct structure for that shape. Concrete operations +(`OverwriteFiles`, `RowDelta`, `RewriteFiles`) and the `DeleteFileIndex` conflict-scoping +gap are out of scope here and left as placeholders. Review Comment: I found this a bit confusing. How's this? ```suggestion This document defines the trait/struct structure for that shape. Concrete operations (`OverwriteFiles`, `RowDelta`, `RewriteFiles`) as well as in-depth delete file conflict resolution (`DeleteFileIndex`) are out of scope here and left as placeholders. ``` ########## 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: does this make sense for a `replace` operation? should we allow replace where there are no snapshots? my expectation here is instead to return an Iceberg error. ########## 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: I don't mind disturbing the append-only path if it results in a better API long-term. ########## 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. + +This document defines the trait/struct structure for that shape. Concrete operations +(`OverwriteFiles`, `RowDelta`, `RewriteFiles`) and the `DeleteFileIndex` conflict-scoping +gap are out of scope here and left as placeholders. + +## 2. Existing Architecture and Components + +The snapshot-production machinery on `main` lives in `transaction/snapshot.rs` and +`transaction/append.rs`: + +- **`SnapshotProducer`** (`snapshot.rs`) — a concrete struct that *drives* one operation + plus one manifest process and writes the new snapshot (id generation, manifest writing, + manifest-list writing, summary). Its `commit<OP, MP>(...)` method does **not** validate. +- **`SnapshotProduceOperation`** (`snapshot.rs`) — a plain trait with no supertrait, + exposing `operation()`, `delete_entries()`, and `existing_manifest()`. `existing_manifest` + today only *selects* which manifests to carry forward. +- **`ManifestProcess` + `DefaultManifestProcess`** (`snapshot.rs`) — a seam for + merge/compaction of the manifest set. `DefaultManifestProcess` is a no-op pass-through. +- **`FastAppendOperation`** (`append.rs`) — the only `SnapshotProduceOperation` impl; it + implements the base trait and nothing else. +- **`TransactionAction`** (`action.rs`) — `commit(self: Arc<Self>, table: &Table)`. The + action is the object `Arc`-cloned and reused across `Transaction::do_commit` retries; the + producer and operation are rebuilt from scratch on each attempt. + +There is no validator trait, no `validate.rs`, no filter manager, and no producer-side +validation call anywhere on `main`. + +## 3. Proposed Architecture + +```mermaid +classDiagram + direction TB + + class SnapshotProducer { + <<struct — exists>> + +commit(&mut op, process) ActionCommit + } + class SnapshotProduceOperation { + <<trait — exists>> + +operation() Operation + +existing_manifest(&mut self, sp) Vec~ManifestFile~ + } + class ManifestProcess { + <<trait — exists>> + } + class FastAppendOperation { + <<exists — base trait only>> + } + + class DeleteAwareOperation { + <<trait — NEW : SnapshotProduceOperation>> + +validate(base, parent) + +validation_history(...) default + +validate_no_new_deletes_for_data_files(...) default + +data_filter(&mut self) &mut ManifestFilterManager + +delete_filter(&mut self) &mut ManifestFilterManager + } + class ManifestFilterManager { + <<struct — NEW>> + +delete_file(file) + +filter_manifests(sp, manifests) + } + class OverwriteFilesOperation { + <<NEW — placeholder>> + } + class RowDeltaOperation { + <<NEW — placeholder>> + } + + SnapshotProducer ..> SnapshotProduceOperation : drives (&mut) + SnapshotProducer ..> ManifestProcess : drives + FastAppendOperation ..|> SnapshotProduceOperation + DeleteAwareOperation --|> SnapshotProduceOperation : sub-trait + DeleteAwareOperation ..> ManifestFilterManager : owns (data + delete) + OverwriteFilesOperation ..|> DeleteAwareOperation + RowDeltaOperation ..|> DeleteAwareOperation +``` + +Key points: + +- The new `DeleteAwareOperation` is an **additive sub-trait** of the existing + `SnapshotProduceOperation`. Append-only operations implement only the base trait and carry + none of the validation/filtering surface. +- `validate` is invoked by the **action** before it hands off to `SnapshotProducer::commit`, + so validation precedes every write. The producer stays oblivious to validation. +- The producer drives the operation by `&mut` so the filter managers can mutate during + manifest production. This is the one base-trait signature change: `existing_manifest` + takes `&mut self`; `FastAppendOperation` absorbs it trivially. + +## 4. Proposed Components + +### 4.1 `DeleteAwareOperation` (new trait) + +A sub-trait of `SnapshotProduceOperation`, implemented only by delete-class operations. It +provides three things: + +- **`validate`** — the per-operation conflict check (real polymorphism; each operation + implements its own). +- **Reusable validation helpers with default implementations** — `validation_history` and + `validate_no_new_deletes_for_data_files`. A single `validate()` is not enough: each + delete-class operation composes the same lower-level checks (scan the snapshots added since + the validation window, find conflicting deletes for a set of data files). Putting these on + the trait as defaults lets each `validate` reuse them instead of re-deriving the logic. + (See `#2590` for the reference shape of these helpers.) Review Comment: Should this actually be moved out to a separate struct? It sounds to me so far that these only have one implementation, and I'd prefer the API of the trait be simpler so that the responsibilities of operation implementations are clear. ########## docs/rfcs/conflict-detection.md: ########## Review Comment: Rename file 0003_conflict_detection.md? I think this is a worthwhile RFC to commit to the repo. ########## 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: What do you think we should strive for? To align closely to the iceberg-java structure, or to well-define the requirements and strive for something idiomatic? ########## 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. + +This document defines the trait/struct structure for that shape. Concrete operations +(`OverwriteFiles`, `RowDelta`, `RewriteFiles`) and the `DeleteFileIndex` conflict-scoping +gap are out of scope here and left as placeholders. + +## 2. Existing Architecture and Components + +The snapshot-production machinery on `main` lives in `transaction/snapshot.rs` and +`transaction/append.rs`: + +- **`SnapshotProducer`** (`snapshot.rs`) — a concrete struct that *drives* one operation + plus one manifest process and writes the new snapshot (id generation, manifest writing, + manifest-list writing, summary). Its `commit<OP, MP>(...)` method does **not** validate. +- **`SnapshotProduceOperation`** (`snapshot.rs`) — a plain trait with no supertrait, + exposing `operation()`, `delete_entries()`, and `existing_manifest()`. `existing_manifest` + today only *selects* which manifests to carry forward. +- **`ManifestProcess` + `DefaultManifestProcess`** (`snapshot.rs`) — a seam for + merge/compaction of the manifest set. `DefaultManifestProcess` is a no-op pass-through. +- **`FastAppendOperation`** (`append.rs`) — the only `SnapshotProduceOperation` impl; it + implements the base trait and nothing else. +- **`TransactionAction`** (`action.rs`) — `commit(self: Arc<Self>, table: &Table)`. The + action is the object `Arc`-cloned and reused across `Transaction::do_commit` retries; the + producer and operation are rebuilt from scratch on each attempt. + +There is no validator trait, no `validate.rs`, no filter manager, and no producer-side +validation call anywhere on `main`. + +## 3. Proposed Architecture + +```mermaid +classDiagram + direction TB + + class SnapshotProducer { + <<struct — exists>> + +commit(&mut op, process) ActionCommit + } + class SnapshotProduceOperation { + <<trait — exists>> + +operation() Operation + +existing_manifest(&mut self, sp) Vec~ManifestFile~ + } + class ManifestProcess { + <<trait — exists>> + } + class FastAppendOperation { + <<exists — base trait only>> + } + + class DeleteAwareOperation { + <<trait — NEW : SnapshotProduceOperation>> + +validate(base, parent) + +validation_history(...) default + +validate_no_new_deletes_for_data_files(...) default + +data_filter(&mut self) &mut ManifestFilterManager + +delete_filter(&mut self) &mut ManifestFilterManager + } + class ManifestFilterManager { + <<struct — NEW>> + +delete_file(file) + +filter_manifests(sp, manifests) + } + class OverwriteFilesOperation { + <<NEW — placeholder>> + } + class RowDeltaOperation { + <<NEW — placeholder>> + } + + SnapshotProducer ..> SnapshotProduceOperation : drives (&mut) + SnapshotProducer ..> ManifestProcess : drives + FastAppendOperation ..|> SnapshotProduceOperation + DeleteAwareOperation --|> SnapshotProduceOperation : sub-trait + DeleteAwareOperation ..> ManifestFilterManager : owns (data + delete) + OverwriteFilesOperation ..|> DeleteAwareOperation + RowDeltaOperation ..|> DeleteAwareOperation +``` + +Key points: + +- The new `DeleteAwareOperation` is an **additive sub-trait** of the existing + `SnapshotProduceOperation`. Append-only operations implement only the base trait and carry + none of the validation/filtering surface. +- `validate` is invoked by the **action** before it hands off to `SnapshotProducer::commit`, + so validation precedes every write. The producer stays oblivious to validation. +- The producer drives the operation by `&mut` so the filter managers can mutate during + manifest production. This is the one base-trait signature change: `existing_manifest` + takes `&mut self`; `FastAppendOperation` absorbs it trivially. + +## 4. Proposed Components + +### 4.1 `DeleteAwareOperation` (new trait) + +A sub-trait of `SnapshotProduceOperation`, implemented only by delete-class operations. It +provides three things: + +- **`validate`** — the per-operation conflict check (real polymorphism; each operation + implements its own). +- **Reusable validation helpers with default implementations** — `validation_history` and + `validate_no_new_deletes_for_data_files`. A single `validate()` is not enough: each + delete-class operation composes the same lower-level checks (scan the snapshots added since + the validation window, find conflicting deletes for a set of data files). Putting these on + the trait as defaults lets each `validate` reuse them instead of re-deriving the logic. + (See `#2590` for the reference shape of these helpers.) +- **Filter accessors** — `data_filter` / `delete_filter` reach the operation's owned filter + managers so the above can be wired without each implementor re-plumbing them. Review Comment: Is the re-plumbing really so bad? Again, motivation is to keep traits simple. ########## 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: My vote would be for the latter, although I think a reference to equivalent Java structures can be helpful. i.e. implementation of this trait is equivalent to X class in Java. ########## 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. + +This document defines the trait/struct structure for that shape. Concrete operations +(`OverwriteFiles`, `RowDelta`, `RewriteFiles`) and the `DeleteFileIndex` conflict-scoping +gap are out of scope here and left as placeholders. + +## 2. Existing Architecture and Components + +The snapshot-production machinery on `main` lives in `transaction/snapshot.rs` and +`transaction/append.rs`: + +- **`SnapshotProducer`** (`snapshot.rs`) — a concrete struct that *drives* one operation + plus one manifest process and writes the new snapshot (id generation, manifest writing, + manifest-list writing, summary). Its `commit<OP, MP>(...)` method does **not** validate. +- **`SnapshotProduceOperation`** (`snapshot.rs`) — a plain trait with no supertrait, + exposing `operation()`, `delete_entries()`, and `existing_manifest()`. `existing_manifest` + today only *selects* which manifests to carry forward. +- **`ManifestProcess` + `DefaultManifestProcess`** (`snapshot.rs`) — a seam for + merge/compaction of the manifest set. `DefaultManifestProcess` is a no-op pass-through. +- **`FastAppendOperation`** (`append.rs`) — the only `SnapshotProduceOperation` impl; it + implements the base trait and nothing else. +- **`TransactionAction`** (`action.rs`) — `commit(self: Arc<Self>, table: &Table)`. The + action is the object `Arc`-cloned and reused across `Transaction::do_commit` retries; the + producer and operation are rebuilt from scratch on each attempt. + +There is no validator trait, no `validate.rs`, no filter manager, and no producer-side +validation call anywhere on `main`. + +## 3. Proposed Architecture + +```mermaid +classDiagram + direction TB + + class SnapshotProducer { + <<struct — exists>> + +commit(&mut op, process) ActionCommit + } + class SnapshotProduceOperation { + <<trait — exists>> + +operation() Operation + +existing_manifest(&mut self, sp) Vec~ManifestFile~ + } + class ManifestProcess { + <<trait — exists>> + } + class FastAppendOperation { + <<exists — base trait only>> + } + + class DeleteAwareOperation { + <<trait — NEW : SnapshotProduceOperation>> + +validate(base, parent) + +validation_history(...) default + +validate_no_new_deletes_for_data_files(...) default + +data_filter(&mut self) &mut ManifestFilterManager + +delete_filter(&mut self) &mut ManifestFilterManager + } + class ManifestFilterManager { + <<struct — NEW>> + +delete_file(file) + +filter_manifests(sp, manifests) + } + class OverwriteFilesOperation { + <<NEW — placeholder>> + } + class RowDeltaOperation { + <<NEW — placeholder>> + } + + SnapshotProducer ..> SnapshotProduceOperation : drives (&mut) + SnapshotProducer ..> ManifestProcess : drives + FastAppendOperation ..|> SnapshotProduceOperation + DeleteAwareOperation --|> SnapshotProduceOperation : sub-trait + DeleteAwareOperation ..> ManifestFilterManager : owns (data + delete) + OverwriteFilesOperation ..|> DeleteAwareOperation + RowDeltaOperation ..|> DeleteAwareOperation +``` + +Key points: + +- The new `DeleteAwareOperation` is an **additive sub-trait** of the existing + `SnapshotProduceOperation`. Append-only operations implement only the base trait and carry + none of the validation/filtering surface. +- `validate` is invoked by the **action** before it hands off to `SnapshotProducer::commit`, + so validation precedes every write. The producer stays oblivious to validation. +- The producer drives the operation by `&mut` so the filter managers can mutate during + manifest production. This is the one base-trait signature change: `existing_manifest` + takes `&mut self`; `FastAppendOperation` absorbs it trivially. + +## 4. Proposed Components + +### 4.1 `DeleteAwareOperation` (new trait) + +A sub-trait of `SnapshotProduceOperation`, implemented only by delete-class operations. It +provides three things: + +- **`validate`** — the per-operation conflict check (real polymorphism; each operation + implements its own). +- **Reusable validation helpers with default implementations** — `validation_history` and + `validate_no_new_deletes_for_data_files`. A single `validate()` is not enough: each + delete-class operation composes the same lower-level checks (scan the snapshots added since + the validation window, find conflicting deletes for a set of data files). Putting these on + the trait as defaults lets each `validate` reuse them instead of re-deriving the logic. + (See `#2590` for the reference shape of these helpers.) +- **Filter accessors** — `data_filter` / `delete_filter` reach the operation's owned filter + managers so the above can be wired without each implementor re-plumbing them. + +```rust +// NEW — does not exist on main +trait DeleteAwareOperation: SnapshotProduceOperation { + /// Per-operation conflict check against the refreshed base table, run before any write. + /// Implemented per operation; composes the helpers below. + async fn validate(&self, base: &Table, parent_snapshot_id: Option<i64>) -> Result<()>; + + /// Default helper: collect manifests + snapshot ids between two points, filtered by + /// operation set and manifest content type. Reused by the checks below. + async fn validation_history( + &self, + base: &Table, + from_snapshot_id: Option<i64>, + to_snapshot_id: i64, + matching_operations: &HashSet<Operation>, + content_type: ManifestContentType, + ) -> Result<(Vec<ManifestFile>, HashSet<i64>)> { + // default implementation + } + + /// Default helper: fail if any delete added since `from_snapshot_id` targets the + /// given data files. + async fn validate_no_new_deletes_for_data_files( + &self, + base: &Table, + from_snapshot_id: Option<i64>, + to_snapshot_id: Option<i64>, + data_files: &[DataFile], + ) -> Result<()> { + // default implementation, built on validation_history + } + + /// Accessors to the operation's owned managers. Built up as the operation is + /// constructed and mutated during manifest production. + fn data_filter(&mut self) -> &mut ManifestFilterManager; + fn delete_filter(&mut self) -> &mut ManifestFilterManager; +} +``` + +The managers are **owned fields on the operation**, populated incrementally as the action +builds the operation, e.g.: + +```rust +struct RowDelta { + data_filter: ManifestFilterManager, + delete_filter: ManifestFilterManager, + // ... added files, conflict-detection config, etc. +} + +impl RowDelta { + fn remove_rows(&mut self, file: DataFile) { + self.delete_filter().delete_file(file); + } +} + +impl DeleteAwareOperation for RowDelta { + async fn validate(&self, base: &Table, parent: Option<i64>) -> Result<()> { + // composes the default helpers, e.g. + self.validate_no_new_deletes_for_data_files(base, parent, None, &self.targets) + .await + } + fn data_filter(&mut self) -> &mut ManifestFilterManager { &mut self.data_filter } + fn delete_filter(&mut self) -> &mut ManifestFilterManager { &mut self.delete_filter } +} +``` + +Because the operation is rebuilt each commit attempt, the managers are reconstructed +deterministically per attempt from the action's stored inputs — no stale per-attempt state. + +### 4.2 `ManifestFilterManager` (new struct) + +A concrete struct (not a trait) that an operation owns — one for data manifests, one for +delete manifests. It accumulates which files/entries to drop and rewrites the affected +manifests during manifest production. Its only durable state is within a single filtering +pass, which matches the operation's per-attempt lifetime. Review Comment: How do we believe this should interact with `trait ManifestProcess`? I see that `SnapshotProduceOperation::existing_manifest` already uses it, but I'm wondering if there's a better way to factor this. Don't have an opinion right now. -- 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]
