jopdorp commented on code in PR #3300: URL: https://github.com/apache/iceberg-rust/pull/3300#discussion_r4422232173
########## crates/iceberg/src/transaction/overwrite.rs: ########## @@ -0,0 +1,957 @@ +// 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::{DataFile, ManifestEntry, ManifestFile, Operation}; +use crate::table::Table; +use crate::transaction::action::CommitStatus; +use crate::transaction::snapshot::{ + DefaultManifestProcess, SnapshotProduceOperation, SnapshotProducer, +}; +use crate::transaction::{ActionCommit, TransactionAction}; + +/// OverwriteAction is a transaction action for overwriting data files in the table. +/// +/// Creates a snapshot with `Operation::Overwrite` semantics — adds new data files and +/// optionally removes existing data files by rewriting affected manifests with those +/// entries marked as `ManifestStatus::Deleted`. +#[derive(Clone)] +pub struct OverwriteAction { + check_duplicate: bool, + commit_uuid: Option<Uuid>, + snapshot_properties: HashMap<String, String>, + added_data_files: Vec<DataFile>, + 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![], + 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. + pub fn add_data_files(mut self, data_files: impl IntoIterator<Item = DataFile>) -> Self { + self.added_data_files.extend(data_files); + 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 + } +} + +#[async_trait] +impl TransactionAction for OverwriteAction { + type State = (); + + fn new_state(&self) -> Self::State {} + + async fn commit(&self, _state: &mut (), table: &Table) -> Result<ActionCommit> { + let mut deleted_files = HashMap::new(); + for file in &self.deleted_data_files { + if file.content_type() != crate::spec::DataContentType::Data { + return Err(invalid_data!("Only data content type is allowed")); + } + deleted_files.entry(file.file_path.as_str()).or_insert(file); + } + let snapshot_producer = SnapshotProducer::new( + table, + self.commit_uuid.unwrap_or_else(Uuid::now_v7), + self.snapshot_properties.clone(), + self.added_data_files.clone(), + self.deleted_data_files.clone(), + ); + + snapshot_producer.validate_added_data_files()?; + + if self.check_duplicate { + snapshot_producer.validate_duplicate_files().await?; + } + + let operation = OverwriteOperation { + has_added_data_files: !self.added_data_files.is_empty(), + deleted_files, + }; + snapshot_producer + .commit(operation, DefaultManifestProcess) + .await + } + + async fn cleanup(self, _state: (), _table: &Table, _status: CommitStatus) {} +} + +struct OverwriteOperation<'a> { + has_added_data_files: bool, + deleted_files: HashMap<&'a str, &'a DataFile>, +} + +impl SnapshotProduceOperation for OverwriteOperation<'_> { + fn operation(&self) -> Operation { + match (self.has_added_data_files, !self.deleted_files.is_empty()) { + (true, true) => Operation::Overwrite, + (false, true) => Operation::Delete, + // Also a properties-only commit, the workaround from #1548. + _ => Operation::Append, + } + } + + // Only the listed files are replaced, so the table totals carry over. + fn truncate_full_table(&self) -> bool { + false + } + + async fn delete_entries( + &self, + _snapshot_produce: &SnapshotProducer<'_>, + ) -> Result<Vec<ManifestEntry>> { + Ok(vec![]) + } + + async fn existing_manifest( + &self, + snapshot_produce: &SnapshotProducer<'_>, + ) -> Result<Vec<ManifestFile>> { + let table = snapshot_produce.table; + let mut manifests = vec![]; + let mut matched: HashSet<&str> = HashSet::new(); + + if let Some(snapshot) = table.metadata().current_snapshot() { + let manifest_list = table.manifest_list_reader(snapshot).load().await?; + for manifest_file in manifest_list.entries() { + // Delete-only manifests record which files were removed and must survive + // until `expire_snapshots` cleans them up (see #2148). + if !manifest_file.has_added_files() + && !manifest_file.has_existing_files() + && !manifest_file.has_deleted_files() + { + continue; + } + if manifest_file.content == crate::spec::ManifestContentType::Deletes + || self.deleted_files.is_empty() + { + manifests.push(manifest_file.clone()); + continue; + } + + let manifest = table.manifest_reader().read(manifest_file).await?; + let mut deletes = vec![]; + for entry in manifest.entries().iter().filter(|entry| entry.is_alive()) { + if let Some(requested) = self.deleted_files.get(entry.file_path()) { + let actual = entry.data_file(); + let mismatch = if requested.record_count != actual.record_count { + format!( + "record_count: requested {}, table has {}", + requested.record_count, actual.record_count + ) + } else if requested.file_size_in_bytes != actual.file_size_in_bytes { + format!( + "file_size_in_bytes: requested {}, table has {}", + requested.file_size_in_bytes, actual.file_size_in_bytes + ) + } else if requested.partition_spec_id != actual.partition_spec_id { + format!( + "partition_spec_id: requested {}, table has {}", + requested.partition_spec_id, actual.partition_spec_id + ) + } else if requested.partition != actual.partition { + format!( + "partition: requested {:?}, table has {:?}", + requested.partition, actual.partition + ) + } else { + deletes.push(requested.file_path.as_str()); + continue; + }; + return Err(invalid_data!( + "Delete file metadata does not match the table: {} ({mismatch})", + entry.file_path() + )); + } + } + if deletes.is_empty() { + manifests.push(manifest_file.clone()); + continue; + } + matched.extend(deletes); + + let mut writer = snapshot_produce.new_manifest_writer( + 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 manifest.entries().iter().filter(|entry| entry.is_alive()) { + if self.deleted_files.contains_key(entry.file_path()) { + writer.add_delete_entry(entry.as_ref().clone())?; Review Comment: The removed files now come from the tombstones the rewrite writes, one per live copy, so the summary counts each copy like Java and the assert holds. That landed with moving onto main's snapshot building blocks in the merge 5a2119b04, and e5b660cf2 pins deleting a path that two fast_append commits put in two manifests. ########## crates/iceberg/src/transaction/overwrite.rs: ########## @@ -0,0 +1,957 @@ +// 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::{DataFile, ManifestEntry, ManifestFile, Operation}; +use crate::table::Table; +use crate::transaction::action::CommitStatus; +use crate::transaction::snapshot::{ + DefaultManifestProcess, SnapshotProduceOperation, SnapshotProducer, +}; +use crate::transaction::{ActionCommit, TransactionAction}; + +/// OverwriteAction is a transaction action for overwriting data files in the table. +/// +/// Creates a snapshot with `Operation::Overwrite` semantics — adds new data files and +/// optionally removes existing data files by rewriting affected manifests with those +/// entries marked as `ManifestStatus::Deleted`. Review Comment: Took your wording in cf6cb1fd0. ########## crates/iceberg/src/transaction/overwrite.rs: ########## @@ -0,0 +1,957 @@ +// 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::{DataFile, ManifestEntry, ManifestFile, Operation}; +use crate::table::Table; +use crate::transaction::action::CommitStatus; +use crate::transaction::snapshot::{ + DefaultManifestProcess, SnapshotProduceOperation, SnapshotProducer, +}; +use crate::transaction::{ActionCommit, TransactionAction}; + +/// OverwriteAction is a transaction action for overwriting data files in the table. +/// +/// Creates a snapshot with `Operation::Overwrite` semantics — adds new data files and +/// optionally removes existing data files by rewriting affected manifests with those +/// entries marked as `ManifestStatus::Deleted`. +#[derive(Clone)] +pub struct OverwriteAction { + check_duplicate: bool, + commit_uuid: Option<Uuid>, + snapshot_properties: HashMap<String, String>, + added_data_files: Vec<DataFile>, + 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![], + 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. + pub fn add_data_files(mut self, data_files: impl IntoIterator<Item = DataFile>) -> Self { + self.added_data_files.extend(data_files); + 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 + } +} + +#[async_trait] +impl TransactionAction for OverwriteAction { + type State = (); + + fn new_state(&self) -> Self::State {} + + async fn commit(&self, _state: &mut (), table: &Table) -> Result<ActionCommit> { + let mut deleted_files = HashMap::new(); + for file in &self.deleted_data_files { + if file.content_type() != crate::spec::DataContentType::Data { + return Err(invalid_data!("Only data content type is allowed")); Review Comment: The error names the file and its content type now, and both types moved into the `use crate::spec::{…}` import, in 5a2119b04. ########## crates/iceberg/src/transaction/overwrite.rs: ########## @@ -0,0 +1,957 @@ +// 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::{DataFile, ManifestEntry, ManifestFile, Operation}; +use crate::table::Table; +use crate::transaction::action::CommitStatus; +use crate::transaction::snapshot::{ + DefaultManifestProcess, SnapshotProduceOperation, SnapshotProducer, +}; +use crate::transaction::{ActionCommit, TransactionAction}; + +/// OverwriteAction is a transaction action for overwriting data files in the table. +/// +/// Creates a snapshot with `Operation::Overwrite` semantics — adds new data files and +/// optionally removes existing data files by rewriting affected manifests with those +/// entries marked as `ManifestStatus::Deleted`. +#[derive(Clone)] +pub struct OverwriteAction { + check_duplicate: bool, + commit_uuid: Option<Uuid>, + snapshot_properties: HashMap<String, String>, + added_data_files: Vec<DataFile>, + 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![], + 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. + pub fn add_data_files(mut self, data_files: impl IntoIterator<Item = DataFile>) -> Self { + self.added_data_files.extend(data_files); + 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 + } +} + +#[async_trait] +impl TransactionAction for OverwriteAction { + type State = (); + + fn new_state(&self) -> Self::State {} + + async fn commit(&self, _state: &mut (), table: &Table) -> Result<ActionCommit> { + let mut deleted_files = HashMap::new(); + for file in &self.deleted_data_files { + if file.content_type() != crate::spec::DataContentType::Data { + return Err(invalid_data!("Only data content type is allowed")); + } + deleted_files.entry(file.file_path.as_str()).or_insert(file); + } + let snapshot_producer = SnapshotProducer::new( + table, + self.commit_uuid.unwrap_or_else(Uuid::now_v7), + self.snapshot_properties.clone(), + self.added_data_files.clone(), + self.deleted_data_files.clone(), + ); + + snapshot_producer.validate_added_data_files()?; + + if self.check_duplicate { + snapshot_producer.validate_duplicate_files().await?; + } + + let operation = OverwriteOperation { + has_added_data_files: !self.added_data_files.is_empty(), + deleted_files, + }; + snapshot_producer + .commit(operation, DefaultManifestProcess) + .await + } + + async fn cleanup(self, _state: (), _table: &Table, _status: CommitStatus) {} +} + +struct OverwriteOperation<'a> { + has_added_data_files: bool, + deleted_files: HashMap<&'a str, &'a DataFile>, +} + +impl SnapshotProduceOperation for OverwriteOperation<'_> { + fn operation(&self) -> Operation { + match (self.has_added_data_files, !self.deleted_files.is_empty()) { + (true, true) => Operation::Overwrite, + (false, true) => Operation::Delete, + // Also a properties-only commit, the workaround from #1548. + _ => Operation::Append, + } + } + + // Only the listed files are replaced, so the table totals carry over. + fn truncate_full_table(&self) -> bool { + false + } + + async fn delete_entries( + &self, + _snapshot_produce: &SnapshotProducer<'_>, + ) -> Result<Vec<ManifestEntry>> { + Ok(vec![]) + } + + async fn existing_manifest( + &self, + snapshot_produce: &SnapshotProducer<'_>, + ) -> Result<Vec<ManifestFile>> { + let table = snapshot_produce.table; + let mut manifests = vec![]; + let mut matched: HashSet<&str> = HashSet::new(); + + if let Some(snapshot) = table.metadata().current_snapshot() { + let manifest_list = table.manifest_list_reader(snapshot).load().await?; + for manifest_file in manifest_list.entries() { + // Delete-only manifests record which files were removed and must survive + // until `expire_snapshots` cleans them up (see #2148). + if !manifest_file.has_added_files() + && !manifest_file.has_existing_files() + && !manifest_file.has_deleted_files() + { + continue; + } + if manifest_file.content == crate::spec::ManifestContentType::Deletes + || self.deleted_files.is_empty() + { + manifests.push(manifest_file.clone()); + continue; + } + + let manifest = table.manifest_reader().read(manifest_file).await?; Review Comment: We left a TODO to read them concurrently like `validate_files_not_live` does, in 5a2119b04. ########## crates/iceberg/src/transaction/overwrite.rs: ########## @@ -0,0 +1,957 @@ +// 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::{DataFile, ManifestEntry, ManifestFile, Operation}; +use crate::table::Table; +use crate::transaction::action::CommitStatus; +use crate::transaction::snapshot::{ + DefaultManifestProcess, SnapshotProduceOperation, SnapshotProducer, +}; +use crate::transaction::{ActionCommit, TransactionAction}; + +/// OverwriteAction is a transaction action for overwriting data files in the table. +/// +/// Creates a snapshot with `Operation::Overwrite` semantics — adds new data files and +/// optionally removes existing data files by rewriting affected manifests with those +/// entries marked as `ManifestStatus::Deleted`. +#[derive(Clone)] +pub struct OverwriteAction { + check_duplicate: bool, + commit_uuid: Option<Uuid>, + snapshot_properties: HashMap<String, String>, + added_data_files: Vec<DataFile>, + 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![], + 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. + pub fn add_data_files(mut self, data_files: impl IntoIterator<Item = DataFile>) -> Self { + self.added_data_files.extend(data_files); + 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 + } +} + +#[async_trait] +impl TransactionAction for OverwriteAction { + type State = (); + + fn new_state(&self) -> Self::State {} + + async fn commit(&self, _state: &mut (), table: &Table) -> Result<ActionCommit> { + let mut deleted_files = HashMap::new(); + for file in &self.deleted_data_files { + if file.content_type() != crate::spec::DataContentType::Data { + return Err(invalid_data!("Only data content type is allowed")); + } + deleted_files.entry(file.file_path.as_str()).or_insert(file); + } + let snapshot_producer = SnapshotProducer::new( + table, + self.commit_uuid.unwrap_or_else(Uuid::now_v7), + self.snapshot_properties.clone(), + self.added_data_files.clone(), + self.deleted_data_files.clone(), + ); + + snapshot_producer.validate_added_data_files()?; + + if self.check_duplicate { + snapshot_producer.validate_duplicate_files().await?; + } + + let operation = OverwriteOperation { + has_added_data_files: !self.added_data_files.is_empty(), + deleted_files, + }; + snapshot_producer + .commit(operation, DefaultManifestProcess) + .await + } + + async fn cleanup(self, _state: (), _table: &Table, _status: CommitStatus) {} +} + +struct OverwriteOperation<'a> { + has_added_data_files: bool, + deleted_files: HashMap<&'a str, &'a DataFile>, +} + +impl SnapshotProduceOperation for OverwriteOperation<'_> { + fn operation(&self) -> Operation { + match (self.has_added_data_files, !self.deleted_files.is_empty()) { + (true, true) => Operation::Overwrite, + (false, true) => Operation::Delete, + // Also a properties-only commit, the workaround from #1548. + _ => Operation::Append, + } + } + + // Only the listed files are replaced, so the table totals carry over. + fn truncate_full_table(&self) -> bool { + false + } + + async fn delete_entries( + &self, + _snapshot_produce: &SnapshotProducer<'_>, + ) -> Result<Vec<ManifestEntry>> { + Ok(vec![]) + } + + async fn existing_manifest( + &self, + snapshot_produce: &SnapshotProducer<'_>, + ) -> Result<Vec<ManifestFile>> { + let table = snapshot_produce.table; + let mut manifests = vec![]; + let mut matched: HashSet<&str> = HashSet::new(); + + if let Some(snapshot) = table.metadata().current_snapshot() { + let manifest_list = table.manifest_list_reader(snapshot).load().await?; + for manifest_file in manifest_list.entries() { + // Delete-only manifests record which files were removed and must survive + // until `expire_snapshots` cleans them up (see #2148). + if !manifest_file.has_added_files() + && !manifest_file.has_existing_files() + && !manifest_file.has_deleted_files() + { + continue; + } + if manifest_file.content == crate::spec::ManifestContentType::Deletes + || self.deleted_files.is_empty() + { + manifests.push(manifest_file.clone()); + continue; + } + + let manifest = table.manifest_reader().read(manifest_file).await?; + let mut deletes = vec![]; + for entry in manifest.entries().iter().filter(|entry| entry.is_alive()) { + if let Some(requested) = self.deleted_files.get(entry.file_path()) { + let actual = entry.data_file(); + let mismatch = if requested.record_count != actual.record_count { Review Comment: The comparison is a `metadata_mismatch` helper returning `Option<String>` now, and the live entries are filtered once, in 5a2119b04. ########## crates/iceberg/src/transaction/overwrite.rs: ########## @@ -0,0 +1,957 @@ +// 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::{DataFile, ManifestEntry, ManifestFile, Operation}; +use crate::table::Table; +use crate::transaction::action::CommitStatus; +use crate::transaction::snapshot::{ + DefaultManifestProcess, SnapshotProduceOperation, SnapshotProducer, +}; +use crate::transaction::{ActionCommit, TransactionAction}; + +/// OverwriteAction is a transaction action for overwriting data files in the table. +/// +/// Creates a snapshot with `Operation::Overwrite` semantics — adds new data files and +/// optionally removes existing data files by rewriting affected manifests with those +/// entries marked as `ManifestStatus::Deleted`. +#[derive(Clone)] +pub struct OverwriteAction { + check_duplicate: bool, + commit_uuid: Option<Uuid>, + snapshot_properties: HashMap<String, String>, + added_data_files: Vec<DataFile>, + 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![], + 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. + pub fn add_data_files(mut self, data_files: impl IntoIterator<Item = DataFile>) -> Self { + self.added_data_files.extend(data_files); + 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 + } +} + +#[async_trait] +impl TransactionAction for OverwriteAction { + type State = (); + + fn new_state(&self) -> Self::State {} + + async fn commit(&self, _state: &mut (), table: &Table) -> Result<ActionCommit> { + let mut deleted_files = HashMap::new(); + for file in &self.deleted_data_files { + if file.content_type() != crate::spec::DataContentType::Data { + return Err(invalid_data!("Only data content type is allowed")); + } + deleted_files.entry(file.file_path.as_str()).or_insert(file); + } + let snapshot_producer = SnapshotProducer::new( + table, + self.commit_uuid.unwrap_or_else(Uuid::now_v7), + self.snapshot_properties.clone(), + self.added_data_files.clone(), + self.deleted_data_files.clone(), + ); + + snapshot_producer.validate_added_data_files()?; + + if self.check_duplicate { + snapshot_producer.validate_duplicate_files().await?; + } + + let operation = OverwriteOperation { + has_added_data_files: !self.added_data_files.is_empty(), + deleted_files, + }; + snapshot_producer + .commit(operation, DefaultManifestProcess) + .await + } + + async fn cleanup(self, _state: (), _table: &Table, _status: CommitStatus) {} +} + +struct OverwriteOperation<'a> { + has_added_data_files: bool, + deleted_files: HashMap<&'a str, &'a DataFile>, +} + +impl SnapshotProduceOperation for OverwriteOperation<'_> { + fn operation(&self) -> Operation { + match (self.has_added_data_files, !self.deleted_files.is_empty()) { + (true, true) => Operation::Overwrite, + (false, true) => Operation::Delete, + // Also a properties-only commit, the workaround from #1548. + _ => Operation::Append, + } + } + + // Only the listed files are replaced, so the table totals carry over. + fn truncate_full_table(&self) -> bool { + false + } + + async fn delete_entries( + &self, + _snapshot_produce: &SnapshotProducer<'_>, + ) -> Result<Vec<ManifestEntry>> { + Ok(vec![]) + } + + async fn existing_manifest( + &self, + snapshot_produce: &SnapshotProducer<'_>, + ) -> Result<Vec<ManifestFile>> { + let table = snapshot_produce.table; + let mut manifests = vec![]; + let mut matched: HashSet<&str> = HashSet::new(); + + if let Some(snapshot) = table.metadata().current_snapshot() { + let manifest_list = table.manifest_list_reader(snapshot).load().await?; + for manifest_file in manifest_list.entries() { + // Delete-only manifests record which files were removed and must survive + // until `expire_snapshots` cleans them up (see #2148). + if !manifest_file.has_added_files() + && !manifest_file.has_existing_files() + && !manifest_file.has_deleted_files() + { + continue; + } + if manifest_file.content == crate::spec::ManifestContentType::Deletes + || self.deleted_files.is_empty() + { + manifests.push(manifest_file.clone()); + continue; + } + + let manifest = table.manifest_reader().read(manifest_file).await?; + let mut deletes = vec![]; + for entry in manifest.entries().iter().filter(|entry| entry.is_alive()) { + if let Some(requested) = self.deleted_files.get(entry.file_path()) { + let actual = entry.data_file(); + let mismatch = if requested.record_count != actual.record_count { + format!( + "record_count: requested {}, table has {}", + requested.record_count, actual.record_count + ) + } else if requested.file_size_in_bytes != actual.file_size_in_bytes { + format!( + "file_size_in_bytes: requested {}, table has {}", + requested.file_size_in_bytes, actual.file_size_in_bytes + ) + } else if requested.partition_spec_id != actual.partition_spec_id { + format!( + "partition_spec_id: requested {}, table has {}", + requested.partition_spec_id, actual.partition_spec_id + ) + } else if requested.partition != actual.partition { + format!( + "partition: requested {:?}, table has {:?}", + requested.partition, actual.partition + ) + } else { + deletes.push(requested.file_path.as_str()); + continue; + }; + return Err(invalid_data!( + "Delete file metadata does not match the table: {} ({mismatch})", + entry.file_path() + )); + } + } + if deletes.is_empty() { + manifests.push(manifest_file.clone()); + continue; + } + matched.extend(deletes); + + let mut writer = snapshot_produce.new_manifest_writer( + 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 manifest.entries().iter().filter(|entry| entry.is_alive()) { + if self.deleted_files.contains_key(entry.file_path()) { + 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 mut missing: Vec<&str> = self + .deleted_files + .keys() + .copied() + .filter(|path| !matched.contains(path)) + .collect(); + if !missing.is_empty() { + missing.sort_unstable(); + return Err(invalid_data!( + "Missing required files to delete: {}", + missing.join(", ") + )); + } + + Ok(manifests) + } +} + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use crate::memory::tests::new_memory_catalog; + use crate::spec::{ + DataContentType, DataFile, DataFileBuilder, DataFileFormat, Literal, ManifestListWriter, + ManifestStatus, ManifestWriterBuilder, NestedField, Operation, PrimitiveType, Schema, + SnapshotRef, Struct, Transform, Type, UnboundPartitionField, UnboundPartitionSpec, + }; + use crate::table::Table; + use crate::test_utils::commit_with_fresh_state; + use crate::transaction::tests::make_v3_minimal_table_in_catalog; Review Comment: `test_second_overwrite_does_not_resurrect_a_deleted_file` runs on a V2 table now, in cf6cb1fd0. -- 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]
