laskoviymishka commented on code in PR #3300: URL: https://github.com/apache/iceberg-rust/pull/3300#discussion_r4326567251
########## crates/iceberg/src/transaction/overwrite.rs: ########## @@ -0,0 +1,767 @@ +// 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 check duplicate files. + 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. Review Comment: The strict match is part of the public API now, and a caller with a rebuilt `DataFile` only finds out at commit time. I'd write it down here: ```suggestion /// Specify data files to be removed from the table in this overwrite. /// /// Each file must be the `DataFile` as committed: it is matched by path against the /// current snapshot's live data files, and its `record_count`, `file_size_in_bytes`, /// `partition_spec_id` and `partition` must equal the table's entry. Duplicate paths /// collapse to the first occurrence. /// /// # Errors /// /// The commit fails with `ErrorKind::DataInvalid` if a file is not a data file, is not /// live in the current snapshot, or its metadata differs from the table's entry. ``` ########## crates/iceberg/src/transaction/overwrite.rs: ########## @@ -0,0 +1,767 @@ +// 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 check duplicate files. + 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. + 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?; Review Comment: With the default `check_duplicate`, an overwrite that deletes `x` and adds a new `x` at the same path fails with "Cannot add files that are already referenced by table" — `validate_duplicate_files` doesn't know `x` is being removed in this same snapshot. I'd exclude the paths in `deleted_files` from the check so path-reusing writers don't have to turn it off for every file; at minimum, document it on `with_check_duplicate` and pin it with a test. ########## crates/iceberg/src/transaction/overwrite.rs: ########## @@ -0,0 +1,767 @@ +// 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 check duplicate files. + 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. + 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(), Review Comment: These files feed `summary()`'s removed-file loop, which renders each partition path with `current_schema()` against the file's own historical spec. If an identity- or truncate-partitioned `int` source has since been promoted to `long`, `partition_type` returns `Long` while the stored literal is still `Int`, and `Datum`'s Display falls into `(_, _) => unreachable!()`. So a plain overwrite-delete on a valid evolved table panics. A dropped decimal or uuid identity source gets there too, via `(Unknown, Int128)` / `(Unknown, UInt128)`. That code lives in main, but this is the first caller that sends old-spec files through it, so I'd fix it here. Either summarize removed files with the schema of the manifest they were matched in (`existing_manifest` already has `manifest.metadata().schema`), or make `partition_to_path` handle a widened type instead of unwrapping into a panic. Either is fine, but it needs a test that promotes the partition source and then deletes a file written under the old spec. ########## crates/iceberg/src/transaction/overwrite.rs: ########## @@ -0,0 +1,767 @@ +// 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 check duplicate files. + 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. + 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(); + if requested.record_count != actual.record_count + || requested.file_size_in_bytes != actual.file_size_in_bytes + || requested.partition_spec_id != actual.partition_spec_id + || requested.partition != actual.partition + { + return Err(invalid_data!( + "Delete file metadata does not match the table: {}", Review Comment: This only names the path, so the caller has to diff two `DataFile`s to work out which of the four fields is off. I'd report the first field that differs with both values, e.g. `record_count: requested 99, table has 1`. ########## crates/iceberg/src/transaction/overwrite.rs: ########## @@ -0,0 +1,767 @@ +// 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 check duplicate files. + 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. + 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(); + if requested.record_count != actual.record_count + || requested.file_size_in_bytes != actual.file_size_in_bytes + || requested.partition_spec_id != actual.partition_spec_id + || requested.partition != actual.partition + { + return Err(invalid_data!( + "Delete file metadata does not match the table: {}", + entry.file_path() + )); + } + deletes.push(requested.file_path.as_str()); + } + } + 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, Operation, PrimitiveType, 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; + use crate::transaction::{AddColumn, ApplyTransactionAction, Transaction}; + use crate::{ErrorKind, TableUpdate}; + + fn test_data_file(path: &str, partition_spec_id: i32) -> DataFile { + DataFileBuilder::default() + .content(DataContentType::Data) + .file_path(path.to_string()) + .file_format(DataFileFormat::Parquet) + .file_size_in_bytes(100) + .record_count(1) + .partition_spec_id(partition_spec_id) + .partition(Struct::from_iter([Some(Literal::long(300))])) + .build() + .unwrap() + } + + async fn current_manifest_entries(table: &Table) -> Vec<(ManifestStatus, String)> { + let snapshot = table.metadata().current_snapshot().unwrap(); + let manifest_list = table.manifest_list_reader(snapshot).load().await.unwrap(); + + let mut entries = vec![]; + for manifest_file in manifest_list.entries() { + let manifest = table.manifest_reader().read(manifest_file).await.unwrap(); + for entry in manifest.entries() { + entries.push((entry.status(), entry.file_path().to_string())); + } + } + entries + } + + #[tokio::test] + async fn test_overwrite_with_deleted_files() { + let catalog = new_memory_catalog().await; + let table = make_v3_minimal_table_in_catalog(&catalog).await; + let spec_id = table.metadata().default_partition_spec_id(); + + let original_file1 = test_data_file("test/original1.parquet", spec_id); + let original_file2 = test_data_file("test/original2.parquet", spec_id); + let tx = Transaction::new(&table); + let action = tx + .fast_append() + .add_data_files(vec![original_file1.clone(), original_file2.clone()]); + let tx = action.apply(tx).unwrap(); + let table = tx.commit(&catalog).await.unwrap(); + + let replacement_file = test_data_file("test/replacement.parquet", spec_id); + let tx = Transaction::new(&table); + let action = tx + .overwrite() + .add_data_files(vec![replacement_file.clone()]) + .delete_data_files(vec![original_file1.clone(), original_file2.clone()]); + let tx = action.apply(tx).unwrap(); + let table = tx.commit(&catalog).await.unwrap(); + + let snapshot = table.metadata().current_snapshot().unwrap(); + assert_eq!(snapshot.summary().operation, Operation::Overwrite); + let summary = &snapshot.summary().additional_properties; + assert_eq!(summary.get("deleted-data-files").unwrap(), "2"); + assert_eq!(summary.get("deleted-records").unwrap(), "2"); + + let entries = current_manifest_entries(&table).await; + assert_eq!(3, entries.len(), "{entries:?}"); + for entry in [ + (ManifestStatus::Deleted, "test/original1.parquet"), + (ManifestStatus::Deleted, "test/original2.parquet"), + (ManifestStatus::Added, "test/replacement.parquet"), + ] { + assert!( + entries.contains(&(entry.0, entry.1.to_string())), + "{entries:?}" + ); + } + } + + #[tokio::test] + async fn test_second_overwrite_does_not_resurrect_a_deleted_file() { + let catalog = new_memory_catalog().await; + let table = make_v3_minimal_table_in_catalog(&catalog).await; + let spec_id = table.metadata().default_partition_spec_id(); + + let file_a = test_data_file("test/a.parquet", spec_id); + let file_b = test_data_file("test/b.parquet", spec_id); + let tx = Transaction::new(&table); + let action = tx + .fast_append() + .add_data_files(vec![file_a.clone(), file_b.clone()]); + let table = action.apply(tx).unwrap().commit(&catalog).await.unwrap(); + + let tx = Transaction::new(&table); + let action = tx + .overwrite() + .add_data_files(vec![test_data_file("test/c.parquet", spec_id)]) + .delete_data_files(vec![file_a.clone()]); + let table = action.apply(tx).unwrap().commit(&catalog).await.unwrap(); + + let tx = Transaction::new(&table); + let action = tx + .overwrite() + .add_data_files(vec![test_data_file("test/d.parquet", spec_id)]) + .delete_data_files(vec![file_b.clone()]); + let table = action.apply(tx).unwrap().commit(&catalog).await.unwrap(); + + let entries = current_manifest_entries(&table).await; + assert!( + !entries.iter().any( + |(status, path)| path == "test/a.parquet" && *status != ManifestStatus::Deleted + ), + "the first overwrite's delete must not come back, entries: {entries:?}" + ); + } + + #[tokio::test] + async fn test_rewritten_manifest_keeps_the_schema_and_spec_it_was_written_with() { + let catalog = new_memory_catalog().await; + let table = make_v3_minimal_table_in_catalog(&catalog).await; + let original_spec_id = table.metadata().default_partition_spec_id(); + let original_schema_id = table.metadata().current_schema_id(); + + let file_a = test_data_file("test/a.parquet", original_spec_id); + let tx = Transaction::new(&table); + let action = tx.fast_append().add_data_files(vec![file_a.clone()]); + let table = action.apply(tx).unwrap().commit(&catalog).await.unwrap(); + + let tx = Transaction::new(&table); + let action = tx.update_schema().add_column(AddColumn::optional( + "extra", + Type::Primitive(PrimitiveType::Int), + )); + let table = action.apply(tx).unwrap().commit(&catalog).await.unwrap(); + + // There is no transaction action for partition evolution yet, so the spec is replaced + // on the metadata directly. + let evolved = table + .metadata() + .clone() + .into_builder(None) + .add_default_partition_spec( + UnboundPartitionSpec::builder() + .add_partition_field( + UnboundPartitionField::builder() + .source_ids(vec![2]) + .name("y") + .transform(Transform::Identity) + .build() + .unwrap(), + ) + .unwrap() + .build(), + ) + .unwrap() + .build() + .unwrap() + .metadata; + let table = table.with_metadata(Arc::new(evolved)); + assert_ne!(original_schema_id, table.metadata().current_schema_id()); + assert_ne!( + original_spec_id, + table.metadata().default_partition_spec_id() + ); + + let tx = Transaction::new(&table); + let action = tx + .overwrite() + .add_data_files(vec![test_data_file( + "test/b.parquet", + table.metadata().default_partition_spec_id(), + )]) + .delete_data_files(vec![file_a.clone()]); + let mut action_commit = commit_with_fresh_state(action, &table).await.unwrap(); + let updates = action_commit.take_updates(); + let new_snapshot: SnapshotRef = if let TableUpdate::AddSnapshot { snapshot } = &updates[0] { + SnapshotRef::new(snapshot.clone()) + } else { + unreachable!() + }; + + let manifest_list = table + .manifest_list_reader(&new_snapshot) + .load() + .await + .unwrap(); + let mut rewritten = None; + for manifest_file in manifest_list.entries() { + let manifest = table.manifest_reader().read(manifest_file).await.unwrap(); + if manifest + .entries() + .iter() + .any(|entry| entry.status() == ManifestStatus::Deleted) + { + rewritten = Some((manifest_file.clone(), manifest)); + } + } + let (rewritten_file, rewritten_manifest) = + rewritten.expect("the manifest holding test/a.parquet must have been rewritten"); + + assert_eq!(original_schema_id, rewritten_manifest.metadata().schema_id); + assert_eq!(original_spec_id, rewritten_file.partition_spec_id); + } + + #[tokio::test] + async fn test_overwrite_counts_a_duplicated_delete_path_once() { + let catalog = new_memory_catalog().await; + let table = make_v3_minimal_table_in_catalog(&catalog).await; + let spec_id = table.metadata().default_partition_spec_id(); + + let file_a = test_data_file("test/a.parquet", spec_id); + let file_b = test_data_file("test/b.parquet", spec_id); + let tx = Transaction::new(&table); + let action = tx + .fast_append() + .add_data_files(vec![file_a.clone(), file_b.clone()]); + let table = action.apply(tx).unwrap().commit(&catalog).await.unwrap(); + + let tx = Transaction::new(&table); + let action = tx + .overwrite() + .add_data_files(vec![test_data_file("test/c.parquet", spec_id)]) + .delete_data_files(vec![file_a.clone(), file_a.clone()]); + let table = action.apply(tx).unwrap().commit(&catalog).await.unwrap(); + + let summary = table.metadata().current_snapshot().unwrap().summary(); + assert_eq!( + summary.additional_properties.get("deleted-data-files"), + Some(&"1".to_string()) + ); + assert_eq!( + summary.additional_properties.get("deleted-records"), + Some(&"1".to_string()) + ); + assert_eq!( + summary.additional_properties.get("total-data-files"), + Some(&"2".to_string()) + ); + } + + #[tokio::test] + async fn test_delete_only_overwrite_reports_a_delete_operation() { + let catalog = new_memory_catalog().await; + let table = make_v3_minimal_table_in_catalog(&catalog).await; + let spec_id = table.metadata().default_partition_spec_id(); + + let file_a = test_data_file("test/a.parquet", spec_id); + let file_b = test_data_file("test/b.parquet", spec_id); + let tx = Transaction::new(&table); + let action = tx + .fast_append() + .add_data_files(vec![file_a.clone(), file_b.clone()]); + let table = action.apply(tx).unwrap().commit(&catalog).await.unwrap(); + + let tx = Transaction::new(&table); + let action = tx.overwrite().delete_data_files(vec![file_a.clone()]); + let table = action.apply(tx).unwrap().commit(&catalog).await.unwrap(); + + let snapshot = table.metadata().current_snapshot().unwrap(); + assert_eq!(snapshot.summary().operation, Operation::Delete); + assert_eq!( + snapshot + .summary() + .additional_properties + .get("deleted-data-files"), + Some(&"1".to_string()) + ); + + let entries = current_manifest_entries(&table).await; + assert!( + entries.contains(&(ManifestStatus::Deleted, "test/a.parquet".to_string())), + "the deleted file must be a tombstone, entries: {entries:?}" + ); + assert!( + entries.contains(&(ManifestStatus::Existing, "test/b.parquet".to_string())), + "the untouched file must stay live, entries: {entries:?}" + ); + } + + #[tokio::test] + async fn test_overwrite_rejects_deleting_a_file_the_table_does_not_have() { + let catalog = new_memory_catalog().await; + let table = make_v3_minimal_table_in_catalog(&catalog).await; + let spec_id = table.metadata().default_partition_spec_id(); + + let file_a = test_data_file("test/a.parquet", spec_id); + let tx = Transaction::new(&table); + let action = tx.fast_append().add_data_files(vec![file_a.clone()]); + let table = action.apply(tx).unwrap().commit(&catalog).await.unwrap(); + + let tx = Transaction::new(&table); + let action = tx + .overwrite() + .add_data_files(vec![test_data_file("test/b.parquet", spec_id)]) + .delete_data_files(vec![ + file_a.clone(), + test_data_file("test/never-committed.parquet", spec_id), + ]); + let err = commit_with_fresh_state(action, &table).await.err().unwrap(); + assert_eq!(err.kind(), ErrorKind::DataInvalid); + assert!( + err.to_string() + .contains("Missing required files to delete: test/never-committed.parquet"), + "{err}" + ); + } + + #[tokio::test] + async fn test_overwrite_rejects_incorrect_delete_metrics() { + let catalog = new_memory_catalog().await; + let table = make_v3_minimal_table_in_catalog(&catalog).await; + let spec_id = table.metadata().default_partition_spec_id(); + let file = test_data_file("test/a.parquet", spec_id); + let tx = Transaction::new(&table); + let table = tx + .fast_append() + .add_data_files(vec![file.clone()]) + .apply(tx) + .unwrap() + .commit(&catalog) + .await + .unwrap(); + for incorrect in [ + DataFile { + record_count: 99, + ..file.clone() + }, + DataFile { + file_size_in_bytes: 999, + ..file.clone() + }, + DataFile { + partition: Struct::from_iter([Some(Literal::long(999))]), + ..file.clone() + }, + DataFile { + partition_spec_id: 99, + ..file.clone() + }, + DataFile { + partition: Struct::from_iter([Some(Literal::string("invalid"))]), + ..file.clone() + }, + ] { + let action = Transaction::new(&table) + .overwrite() + .delete_data_files(vec![incorrect]); + let err = commit_with_fresh_state(action, &table).await.err().unwrap(); + assert_eq!(err.kind(), ErrorKind::DataInvalid); + assert!( + err.message() + .contains("Delete file metadata does not match") + ); + } + } + + #[tokio::test] + async fn test_overwrite_rejects_non_data_deletions() { + let catalog = new_memory_catalog().await; + let table = make_v3_minimal_table_in_catalog(&catalog).await; + let spec_id = table.metadata().default_partition_spec_id(); + for content in [ + DataContentType::PositionDeletes, + DataContentType::EqualityDeletes, + ] { + let file = DataFile { + content, + ..test_data_file("test/deletes.parquet", spec_id) + }; + let action = Transaction::new(&table) + .overwrite() + .delete_data_files(vec![file]); + let err = commit_with_fresh_state(action, &table).await.err().unwrap(); + assert_eq!(err.kind(), ErrorKind::DataInvalid); + assert!(err.message().contains("Only data content type is allowed")); + } + } + + #[tokio::test] + async fn test_overwrite_preserves_v3_row_lineage() { + let catalog = new_memory_catalog().await; + let table = make_v3_minimal_table_in_catalog(&catalog).await; + let spec_id = table.metadata().default_partition_spec_id(); + let a = test_data_file("test/a.parquet", spec_id); + let b = test_data_file("test/b.parquet", spec_id); + let c = test_data_file("test/c.parquet", spec_id); + let tx = Transaction::new(&table); + let table = tx + .fast_append() + .add_data_files(vec![a, b.clone(), c]) + .apply(tx) + .unwrap() + .commit(&catalog) + .await + .unwrap(); + let snapshot = table.metadata().current_snapshot().unwrap(); + let manifests = table.manifest_list_reader(snapshot).load().await.unwrap(); + let manifest = table + .manifest_reader() + .read(&manifests.entries()[0]) + .await + .unwrap(); + let original_ids: Vec<_> = manifest + .entries() + .iter() + .map(|e| (e.file_path().to_string(), e.data_file().first_row_id)) + .collect(); + assert!(original_ids.iter().all(|(_, id)| id.is_some())); + let next_row_id = table.metadata().next_row_id(); + let replacement = test_data_file("test/d.parquet", spec_id); + let tx = Transaction::new(&table); + let table = tx + .overwrite() + .delete_data_files(vec![b]) + .add_data_files(vec![replacement]) + .apply(tx) + .unwrap() + .commit(&catalog) + .await + .unwrap(); + let manifests = table + .manifest_list_reader(table.metadata().current_snapshot().unwrap()) + .load() + .await + .unwrap(); + let mut actual_ids = vec![]; + for file in manifests.entries() { + let manifest = table.manifest_reader().read(file).await.unwrap(); + actual_ids.extend( + manifest + .entries() + .iter() + .filter(|e| e.is_alive()) + .map(|e| (e.file_path().to_string(), e.data_file().first_row_id)), + ); + } + for (path, id) in original_ids.iter().filter(|(p, _)| p != "test/b.parquet") { + assert!(actual_ids.contains(&(path.clone(), *id))); + } + // The rewritten manifest reserves IDs for its existing rows; their explicit IDs stay unchanged. + assert!(actual_ids.contains(&( + "test/d.parquet".to_string(), + Some((next_row_id + 2).try_into().unwrap()) + ))); + assert_eq!(table.metadata().next_row_id(), next_row_id + 3); + } + + #[tokio::test] + async fn test_overwrite_rejects_a_path_in_a_delete_manifest() { Review Comment: This covers the rejection side only; nothing asserts that the `Deletes` manifest actually survives an overwrite. I'd reuse this fixture: overwrite-delete `test/a.parquet` instead, then assert `test/delete-manifest.avro` is still in the new manifest list with its `added_snapshot_id` and sequence numbers unchanged. -- 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]
