jopdorp commented on code in PR #3300: URL: https://github.com/apache/iceberg-rust/pull/3300#discussion_r4422233953
########## 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; + use crate::transaction::{AddColumn, ApplyTransactionAction, Transaction}; + use crate::{Catalog, ErrorKind, TableCommit, TableCreation, 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:?}" + ); Review Comment: It asserts the whole entry list now, with `a` gone, `b` Deleted and `c`, `d` live, 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")); + } + 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; + use crate::transaction::{AddColumn, ApplyTransactionAction, Transaction}; + use crate::{Catalog, ErrorKind, TableCommit, TableCreation, 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_deletes_files_after_partition_source_promotion() { + let catalog = new_memory_catalog().await; + let base = make_v3_minimal_table_in_catalog(&catalog).await; + for transform in [Transform::Identity, Transform::Truncate(10)] { + let schema = Schema::builder() + .with_fields([NestedField::required(1, "x", PrimitiveType::Int.into()).into()]) + .build() + .unwrap(); + let spec = UnboundPartitionSpec::builder() + .add_partition_field( + UnboundPartitionField::builder() + .source_ids(vec![1]) + .name("p") + .transform(transform) + .build() + .unwrap(), + ) + .unwrap() + .build(); + let table = catalog + .create_table( + base.identifier().namespace(), + TableCreation::builder() + .name(format!("promoted-{transform}")) + .schema(schema) + .partition_spec(spec) + .format_version(crate::spec::FormatVersion::V3) + .build(), + ) + .await + .unwrap(); + let file = DataFile { + partition: Struct::from_iter([Some(Literal::int(300))]), + ..test_data_file( + "test/a.parquet", + table.metadata().default_partition_spec_id(), + ) + }; + let tx = Transaction::new(&table); + let table = tx + .fast_append() + .add_data_files([file.clone()]) + .apply(tx) + .unwrap() + .commit(&catalog) + .await + .unwrap(); + let schema = Schema::builder() + .with_fields([NestedField::required(1, "x", PrimitiveType::Long.into()).into()]) + .build() + .unwrap(); + let table = catalog + .update_table( + TableCommit::builder() + .ident(table.identifier().clone()) + .requirements(vec![]) + .updates(vec![ + TableUpdate::AddSchema { schema }, + TableUpdate::SetCurrentSchema { schema_id: -1 }, + ]) + .build(), + ) + .await + .unwrap(); + let tx = Transaction::new(&table); + let table = tx + .overwrite() + .delete_data_files([file]) + .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["deleted-records"], + "1" + ); + assert_eq!(current_manifest_entries(&table).await, [( + ManifestStatus::Deleted, + "test/a.parquet".to_string() + )]); + } + } + + #[tokio::test] + async fn test_overwrite_readds_a_deleted_path_by_default() { + let catalog = new_memory_catalog().await; + let table = make_v3_minimal_table_in_catalog(&catalog).await; + let file = test_data_file( + "test/a.parquet", + table.metadata().default_partition_spec_id(), + ); + let tx = Transaction::new(&table); + let table = tx + .fast_append() + .add_data_files([file.clone()]) + .apply(tx) + .unwrap() + .commit(&catalog) + .await + .unwrap(); + let tx = Transaction::new(&table); + let table = tx + .overwrite() + .delete_data_files([file.clone()]) + .add_data_files([file]) + .apply(tx) + .unwrap() + .commit(&catalog) + .await + .unwrap(); + assert_eq!(current_manifest_entries(&table).await, [ + (ManifestStatus::Deleted, "test/a.parquet".to_string()), + (ManifestStatus::Added, "test/a.parquet".to_string()), + ]); + } + + #[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, mismatch) in [ + ( + DataFile { + record_count: 99, + ..file.clone() + }, + "record_count: requested 99, table has 1", + ), + ( + DataFile { + file_size_in_bytes: 999, + ..file.clone() + }, + "file_size_in_bytes: requested 999, table has 100", + ), + ( + DataFile { + partition: Struct::from_iter([Some(Literal::long(999))]), + ..file.clone() + }, + "partition: requested Struct { fields: [Some(Primitive(Long(999)))] }, table has Struct { fields: [Some(Primitive(Long(300)))] }", Review Comment: Both partition cases build the expected message from the two structs now, in cf6cb1fd0. ########## crates/iceberg/src/transaction/snapshot.rs: ########## @@ -367,6 +381,9 @@ impl<'a> SnapshotProducer<'a> { } let existing_manifests = snapshot_produce_operation.existing_manifest(self).await?; + let summary = self.summary(snapshot_produce_operation).map_err(|err| { Review Comment: Main dropped the old `SnapshotProducer` and `write_snapshot_commit` builds the summary from explicit inputs, so there's no ordering left to note. Overwrite runs on those building blocks since the merge 5a2119b04. ########## crates/iceberg/src/spec/values/datum.rs: ########## @@ -334,6 +337,8 @@ impl Display for Datum { ) => { write!(f, "{}", decimal_from_i128_with_scale(*val, *scale)) } + // The scale lives on the dropped source type, so only the unscaled value is left. + (PrimitiveType::Unknown, PrimitiveLiteral::Int128(val)) => write!(f, "{val}"), Review Comment: Opened #3586 to track it. -- 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]
