laskoviymishka commented on code in PR #3300: URL: https://github.com/apache/iceberg-rust/pull/3300#discussion_r4431185075
########## crates/iceberg/src/transaction/overwrite.rs: ########## @@ -0,0 +1,1102 @@ +// 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, HashSet}; +use std::sync::Arc; + +use async_trait::async_trait; +use uuid::Uuid; + +use crate::error::{Result, invalid_data}; +use crate::spec::{DataContentType, DataFile, ManifestContentType, Operation}; +use crate::table::Table; +use crate::transaction::action::CommitStatus; +use crate::transaction::snapshot::simple::write_added_manifest; +use crate::transaction::snapshot::{ + SnapshotCommitInput, current_manifests, generate_snapshot_id, manifest_list_path, + manifest_path, new_manifest_writer, validate_files_not_live, validate_partition_value, + write_snapshot_commit, +}; +use crate::transaction::{ActionCommit, TransactionAction}; +use crate::{Error, ErrorKind}; + +/// Adds data files and removes listed data files by path. +/// +/// The operation is `Overwrite` when files are both added and removed, otherwise `Delete` or +/// `Append`. There is no row-filter mode, and files committed concurrently are not checked. +#[derive(Clone)] +pub struct OverwriteAction { + check_duplicate: bool, + commit_uuid: Option<Uuid>, + snapshot_properties: HashMap<String, String>, + added_data_files: Vec<DataFile>, + added_paths: HashSet<String>, + deleted_data_files: Vec<DataFile>, +} + +impl OverwriteAction { + pub(crate) fn new() -> Self { + Self { + check_duplicate: true, + commit_uuid: None, + snapshot_properties: HashMap::default(), + added_data_files: vec![], + added_paths: HashSet::new(), + deleted_data_files: vec![], + } + } + + /// Set whether to reject added paths already live in the table (enabled by default). + /// Paths deleted in the same overwrite may be re-added. + pub fn with_check_duplicate(mut self, v: bool) -> Self { + self.check_duplicate = v; + self + } + + /// Add data files to the snapshot. Duplicate paths keep the first occurrence. + pub fn add_data_files(mut self, data_files: impl IntoIterator<Item = DataFile>) -> Self { + for data_file in data_files { + if self.added_paths.insert(data_file.file_path.clone()) { + self.added_data_files.push(data_file); + } + } + self + } + + /// Specify data files to be removed from the table in this overwrite. + /// + /// Files are matched by path against the current snapshot's live data files. + /// Their `record_count`, `file_size_in_bytes`, `partition_spec_id` and `partition` + /// must match the committed entry. Duplicate paths keep the first occurrence. + /// + /// # Errors + /// + /// The commit fails with [`crate::ErrorKind::DataInvalid`] for non-data files, + /// paths that are not live in the current snapshot, or mismatched metadata. + pub fn delete_data_files(mut self, data_files: impl IntoIterator<Item = DataFile>) -> Self { + self.deleted_data_files.extend(data_files); + 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 snapshot summary properties. + pub fn set_snapshot_properties(mut self, snapshot_properties: HashMap<String, String>) -> Self { + self.snapshot_properties = snapshot_properties; + self + } + + fn validate(&self, table: &Table) -> Result<()> { + // TODO: Allowing snapshot property setup with no data files is a workaround. + // For details, please refer to https://github.com/apache/iceberg-rust/issues/1548 + if self.added_data_files.is_empty() + && self.deleted_data_files.is_empty() + && self.snapshot_properties.is_empty() + { + return Err(Error::new( + ErrorKind::PreconditionFailed, + "No added or deleted data files or added snapshot properties found", + )); + } + if let Some(data_file) = self + .added_data_files + .iter() + .chain(&self.deleted_data_files) + .find(|f| f.content_type() != DataContentType::Data) + { + return Err(invalid_data!( + "Only data content type is allowed for overwrite, got {:?} for {}", + data_file.content_type(), + data_file.file_path() + )); + } + let metadata = table.metadata(); + for data_file in &self.added_data_files { + if metadata.default_partition_spec_id() != data_file.partition_spec_id { + return Err(invalid_data!( + "Data file partition spec id does not match table default partition spec id" + )); + } + validate_partition_value(data_file.partition(), metadata.default_partition_type())?; + } + Ok(()) + } +} + +#[async_trait] +impl TransactionAction for OverwriteAction { + type State = OverwriteState; + + fn new_state(&self) -> OverwriteState { + OverwriteState { + commit_uuid: self.commit_uuid.unwrap_or_else(Uuid::now_v7), + snapshot_id: None, + attempt: 0, + manifest_counter: 0, + } + } + + async fn commit(&self, state: &mut OverwriteState, table: &Table) -> Result<ActionCommit> { + self.validate(table)?; + let mut deleted_files = HashMap::new(); + for file in &self.deleted_data_files { + deleted_files.entry(file.file_path()).or_insert(file); + } + if self.check_duplicate { + let added: Vec<DataFile> = self Review Comment: Could [`validate_files_not_live`](https://github.com/apache/iceberg-rust/blob/36fd78505453a2148c5992512ea743831a71007a/crates/iceberg/src/transaction/snapshot/mod.rs#L181-L189) take paths instead of `&[DataFile]`? It only reads `file_path()`, but with the default `check_duplicate` this clones every added `DataFile` on each attempt, even when nothing is deleted. FastAppend [passes its slice](https://github.com/apache/iceberg-rust/blob/36fd78505453a2148c5992512ea743831a71007a/crates/iceberg/src/transaction/append.rs#L133) without a clone, so both call sites could hand over borrowed `&str`s. ########## crates/iceberg/src/transaction/overwrite.rs: ########## @@ -0,0 +1,1102 @@ +// 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, HashSet}; +use std::sync::Arc; + +use async_trait::async_trait; +use uuid::Uuid; + +use crate::error::{Result, invalid_data}; +use crate::spec::{DataContentType, DataFile, ManifestContentType, Operation}; +use crate::table::Table; +use crate::transaction::action::CommitStatus; +use crate::transaction::snapshot::simple::write_added_manifest; +use crate::transaction::snapshot::{ + SnapshotCommitInput, current_manifests, generate_snapshot_id, manifest_list_path, + manifest_path, new_manifest_writer, validate_files_not_live, validate_partition_value, + write_snapshot_commit, +}; +use crate::transaction::{ActionCommit, TransactionAction}; +use crate::{Error, ErrorKind}; + +/// Adds data files and removes listed data files by path. +/// +/// The operation is `Overwrite` when files are both added and removed, otherwise `Delete` or +/// `Append`. There is no row-filter mode, and files committed concurrently are not checked. +#[derive(Clone)] +pub struct OverwriteAction { + check_duplicate: bool, + commit_uuid: Option<Uuid>, + snapshot_properties: HashMap<String, String>, + added_data_files: Vec<DataFile>, + added_paths: HashSet<String>, + deleted_data_files: Vec<DataFile>, +} + +impl OverwriteAction { + pub(crate) fn new() -> Self { + Self { + check_duplicate: true, + commit_uuid: None, + snapshot_properties: HashMap::default(), + added_data_files: vec![], + added_paths: HashSet::new(), + deleted_data_files: vec![], + } + } + + /// Set whether to reject added paths already live in the table (enabled by default). + /// Paths deleted in the same overwrite may be re-added. + pub fn with_check_duplicate(mut self, v: bool) -> Self { + self.check_duplicate = v; + self + } + + /// Add data files to the snapshot. Duplicate paths keep the first occurrence. + pub fn add_data_files(mut self, data_files: impl IntoIterator<Item = DataFile>) -> Self { + for data_file in data_files { + if self.added_paths.insert(data_file.file_path.clone()) { + self.added_data_files.push(data_file); + } + } + self + } + + /// Specify data files to be removed from the table in this overwrite. + /// + /// Files are matched by path against the current snapshot's live data files. + /// Their `record_count`, `file_size_in_bytes`, `partition_spec_id` and `partition` + /// must match the committed entry. Duplicate paths keep the first occurrence. + /// + /// # Errors + /// + /// The commit fails with [`crate::ErrorKind::DataInvalid`] for non-data files, + /// paths that are not live in the current snapshot, or mismatched metadata. + pub fn delete_data_files(mut self, data_files: impl IntoIterator<Item = DataFile>) -> Self { + self.deleted_data_files.extend(data_files); + 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 snapshot summary properties. + pub fn set_snapshot_properties(mut self, snapshot_properties: HashMap<String, String>) -> Self { + self.snapshot_properties = snapshot_properties; + self + } + + fn validate(&self, table: &Table) -> Result<()> { + // TODO: Allowing snapshot property setup with no data files is a workaround. + // For details, please refer to https://github.com/apache/iceberg-rust/issues/1548 + if self.added_data_files.is_empty() + && self.deleted_data_files.is_empty() + && self.snapshot_properties.is_empty() + { + return Err(Error::new( + ErrorKind::PreconditionFailed, + "No added or deleted data files or added snapshot properties found", + )); + } + if let Some(data_file) = self + .added_data_files + .iter() + .chain(&self.deleted_data_files) + .find(|f| f.content_type() != DataContentType::Data) + { + return Err(invalid_data!( + "Only data content type is allowed for overwrite, got {:?} for {}", + data_file.content_type(), + data_file.file_path() + )); + } + let metadata = table.metadata(); + for data_file in &self.added_data_files { + if metadata.default_partition_spec_id() != data_file.partition_spec_id { + return Err(invalid_data!( + "Data file partition spec id does not match table default partition spec id" + )); + } + validate_partition_value(data_file.partition(), metadata.default_partition_type())?; + } + Ok(()) + } +} + +#[async_trait] +impl TransactionAction for OverwriteAction { + type State = OverwriteState; + + fn new_state(&self) -> OverwriteState { + OverwriteState { + commit_uuid: self.commit_uuid.unwrap_or_else(Uuid::now_v7), + snapshot_id: None, + attempt: 0, + manifest_counter: 0, + } + } + + async fn commit(&self, state: &mut OverwriteState, table: &Table) -> Result<ActionCommit> { + self.validate(table)?; + let mut deleted_files = HashMap::new(); + for file in &self.deleted_data_files { + deleted_files.entry(file.file_path()).or_insert(file); + } + if self.check_duplicate { + let added: Vec<DataFile> = self + .added_data_files + .iter() + .filter(|f| !deleted_files.contains_key(f.file_path())) + .cloned() + .collect(); + validate_files_not_live(table, &added).await?; + } + + let snapshot_id = *state + .snapshot_id + .get_or_insert_with(|| generate_snapshot_id(table)); + let mut manifests = vec![]; + // One per tombstone written, so a path live in several manifests counts each copy. + let mut removed_files = vec![]; + + for manifest_file in current_manifests(table).await? { + if manifest_file.content == ManifestContentType::Deletes || deleted_files.is_empty() { + manifests.push(manifest_file); + continue; + } + + let manifest = table.manifest_reader().read(&manifest_file).await?; Review Comment: Could this read go through the object cache, like [`validate_files_not_live`](https://github.com/apache/iceberg-rust/blob/36fd78505453a2148c5992512ea743831a71007a/crates/iceberg/src/transaction/snapshot/mod.rs#L176-L204) does? When the commit also adds files and `check_duplicate` is on, that pass has already cached these manifests, and this loop fetches each one from storage again, on every retry too. The concurrent-read TODO from 5a2119b04 was dropped in [cc0c9e491](https://github.com/apache/iceberg-rust/commit/cc0c9e491fc70814ea5cae441e5e5b5f8bdfb8c1), so HEAD has neither concurrent reads nor the TODO. Could it come back with this? ```suggestion // TODO: read these manifests concurrently, as `validate_files_not_live` does. let manifest = table.object_cache().get_manifest(&manifest_file).await?; ``` ########## crates/iceberg/src/transaction/overwrite.rs: ########## @@ -0,0 +1,1102 @@ +// 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, HashSet}; +use std::sync::Arc; + +use async_trait::async_trait; +use uuid::Uuid; + +use crate::error::{Result, invalid_data}; +use crate::spec::{DataContentType, DataFile, ManifestContentType, Operation}; +use crate::table::Table; +use crate::transaction::action::CommitStatus; +use crate::transaction::snapshot::simple::write_added_manifest; +use crate::transaction::snapshot::{ + SnapshotCommitInput, current_manifests, generate_snapshot_id, manifest_list_path, + manifest_path, new_manifest_writer, validate_files_not_live, validate_partition_value, + write_snapshot_commit, +}; +use crate::transaction::{ActionCommit, TransactionAction}; +use crate::{Error, ErrorKind}; + +/// Adds data files and removes listed data files by path. +/// +/// The operation is `Overwrite` when files are both added and removed, otherwise `Delete` or +/// `Append`. There is no row-filter mode, and files committed concurrently are not checked. +#[derive(Clone)] +pub struct OverwriteAction { + check_duplicate: bool, + commit_uuid: Option<Uuid>, + snapshot_properties: HashMap<String, String>, + added_data_files: Vec<DataFile>, + added_paths: HashSet<String>, + deleted_data_files: Vec<DataFile>, +} + +impl OverwriteAction { + pub(crate) fn new() -> Self { + Self { + check_duplicate: true, + commit_uuid: None, + snapshot_properties: HashMap::default(), + added_data_files: vec![], + added_paths: HashSet::new(), + deleted_data_files: vec![], + } + } + + /// Set whether to reject added paths already live in the table (enabled by default). + /// Paths deleted in the same overwrite may be re-added. + pub fn with_check_duplicate(mut self, v: bool) -> Self { + self.check_duplicate = v; + self + } + + /// Add data files to the snapshot. Duplicate paths keep the first occurrence. + pub fn add_data_files(mut self, data_files: impl IntoIterator<Item = DataFile>) -> Self { + for data_file in data_files { + if self.added_paths.insert(data_file.file_path.clone()) { + self.added_data_files.push(data_file); + } + } + self + } + + /// Specify data files to be removed from the table in this overwrite. + /// + /// Files are matched by path against the current snapshot's live data files. + /// Their `record_count`, `file_size_in_bytes`, `partition_spec_id` and `partition` + /// must match the committed entry. Duplicate paths keep the first occurrence. + /// + /// # Errors + /// + /// The commit fails with [`crate::ErrorKind::DataInvalid`] for non-data files, + /// paths that are not live in the current snapshot, or mismatched metadata. + pub fn delete_data_files(mut self, data_files: impl IntoIterator<Item = DataFile>) -> Self { + self.deleted_data_files.extend(data_files); + 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 snapshot summary properties. + pub fn set_snapshot_properties(mut self, snapshot_properties: HashMap<String, String>) -> Self { + self.snapshot_properties = snapshot_properties; + self + } + + fn validate(&self, table: &Table) -> Result<()> { + // TODO: Allowing snapshot property setup with no data files is a workaround. + // For details, please refer to https://github.com/apache/iceberg-rust/issues/1548 + if self.added_data_files.is_empty() + && self.deleted_data_files.is_empty() + && self.snapshot_properties.is_empty() + { + return Err(Error::new( + ErrorKind::PreconditionFailed, + "No added or deleted data files or added snapshot properties found", + )); + } + if let Some(data_file) = self + .added_data_files + .iter() + .chain(&self.deleted_data_files) + .find(|f| f.content_type() != DataContentType::Data) + { + return Err(invalid_data!( + "Only data content type is allowed for overwrite, got {:?} for {}", + data_file.content_type(), + data_file.file_path() + )); + } + let metadata = table.metadata(); + for data_file in &self.added_data_files { + if metadata.default_partition_spec_id() != data_file.partition_spec_id { + return Err(invalid_data!( + "Data file partition spec id does not match table default partition spec id" + )); + } + validate_partition_value(data_file.partition(), metadata.default_partition_type())?; + } + Ok(()) + } +} + +#[async_trait] +impl TransactionAction for OverwriteAction { + type State = OverwriteState; + + fn new_state(&self) -> OverwriteState { + OverwriteState { + commit_uuid: self.commit_uuid.unwrap_or_else(Uuid::now_v7), + snapshot_id: None, + attempt: 0, + manifest_counter: 0, + } + } + + async fn commit(&self, state: &mut OverwriteState, table: &Table) -> Result<ActionCommit> { + self.validate(table)?; + let mut deleted_files = HashMap::new(); + for file in &self.deleted_data_files { + deleted_files.entry(file.file_path()).or_insert(file); + } + if self.check_duplicate { + let added: Vec<DataFile> = self + .added_data_files + .iter() + .filter(|f| !deleted_files.contains_key(f.file_path())) + .cloned() + .collect(); + validate_files_not_live(table, &added).await?; + } + + let snapshot_id = *state + .snapshot_id + .get_or_insert_with(|| generate_snapshot_id(table)); + let mut manifests = vec![]; + // One per tombstone written, so a path live in several manifests counts each copy. + let mut removed_files = vec![]; + + for manifest_file in current_manifests(table).await? { + if manifest_file.content == ManifestContentType::Deletes || deleted_files.is_empty() { + manifests.push(manifest_file); + continue; + } + + let manifest = table.manifest_reader().read(&manifest_file).await?; + let live: Vec<_> = manifest.entries().iter().filter(|e| e.is_alive()).collect(); + let mut has_deletes = false; + for entry in &live { + let Some(requested) = deleted_files.get(entry.file_path()) else { + continue; + }; + if let Some(mismatch) = metadata_mismatch(requested, entry.data_file()) { + return Err(invalid_data!( + "Delete file metadata does not match the table: {} ({mismatch})", + entry.file_path() + )); + } + has_deletes = true; + } + if !has_deletes { + manifests.push(manifest_file); + continue; + } + + let mut writer = new_manifest_writer( + table, + state.next_manifest_path(table)?, + snapshot_id, + manifest_file.content, + manifest.metadata().schema.clone(), + Arc::new(manifest.metadata().partition_spec.clone()), + )?; + // Like Java, only live entries are carried into the rewrite. + for entry in live { + if deleted_files.contains_key(entry.file_path()) { + removed_files.push(entry.data_file().clone()); + writer.add_delete_entry(entry.as_ref().clone())?; + } else { + writer.add_existing_entry(entry.as_ref().clone())?; + } + } + manifests.push(writer.write_manifest_file().await?); + } + + let removed_paths: HashSet<&str> = removed_files.iter().map(|f| f.file_path()).collect(); + let mut missing: Vec<&str> = deleted_files + .keys() + .copied() + .filter(|path| !removed_paths.contains(path)) + .collect(); + if !missing.is_empty() { + missing.sort_unstable(); + return Err(invalid_data!( + "Missing required files to delete: {}", + missing.join(", ") + )); + } + + if !self.added_data_files.is_empty() { + manifests.push( + write_added_manifest( + table, + state.next_manifest_path(table)?, + snapshot_id, + table.metadata().default_partition_spec_id(), + &self.added_data_files, + ) + .await?, + ); + } + + let operation = match (!self.added_data_files.is_empty(), !removed_files.is_empty()) { + (true, true) => Operation::Overwrite, + (false, true) => Operation::Delete, + // Also a properties-only commit, the workaround from #1548. + _ => Operation::Append, + }; + let manifest_list_path = + manifest_list_path(table, snapshot_id, state.attempt, state.commit_uuid)?; + state.attempt += 1; + write_snapshot_commit(table, SnapshotCommitInput { + snapshot_id, + manifest_list_path, + operation, + properties: self.snapshot_properties.clone(), + added_files: &self.added_data_files, + removed_files: &removed_files, + manifests, + // Only the listed files are replaced, so the table totals carry over. + truncate_full_table: false, + }) + .await + } + + async fn cleanup(self, _state: OverwriteState, _table: &Table, _status: CommitStatus) {} Review Comment: Could we match every requested path before writing the first rewritten manifest? As written, a missing path alongside matched ones, or a metadata mismatch in a later manifest, returns after [earlier rewrites are written](https://github.com/apache/iceberg-rust/blob/36fd78505453a2148c5992512ea743831a71007a/crates/iceberg/src/transaction/overwrite.rs#L186-L238), and nothing removes them. Separately, could `cleanup` carry FastAppend's [`TODO(#3417)`](https://github.com/apache/iceberg-rust/blob/36fd78505453a2148c5992512ea743831a71007a/crates/iceberg/src/transaction/append.rs#L138-L141)? ```suggestion async fn cleanup(self, _state: OverwriteState, _table: &Table, _status: CommitStatus) { // TODO(#3417): remove this attempt's manifests once terminal cleanup is // wired up. } ``` -- 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]
