laskoviymishka commented on code in PR #3300:
URL: https://github.com/apache/iceberg-rust/pull/3300#discussion_r4394334523
##########
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:
Could we open an issue for this arm? With the scale gone, a dropped decimal
source renders `123` for 1.23. That's harmless for `changed-partition-count`,
but it lands in the `partitions.*` summary keys once
[`write.summary.partition-limit`](https://github.com/apache/iceberg-rust/blob/b106b0af66388e8aa296e6c7bfd8c190396e9e98/crates/iceberg/src/spec/table_properties.rs#L383)
is above 0. We settled on the Display arms last round, so this is only about
tracking it.
##########
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:
Could the doc say which operation this reports and what it doesn't check?
[`operation()`](https://github.com/apache/iceberg-rust/blob/b106b0af66388e8aa296e6c7bfd8c190396e9e98/crates/iceberg/src/transaction/overwrite.rs#L144-L152)
returns `Delete` or `Append` for one-sided commits, and there's no row-filter
mode or validation of concurrent commits. Java's default `OverwriteFiles`
[skips that validation
too](https://github.com/apache/iceberg/blob/2856631da4d81e79ece85c3694bfd81a86ca612e/core/src/main/java/org/apache/iceberg/BaseOverwriteFiles.java#L34-L38),
so it's scope to state rather than a gap.
```suggestion
/// Adds data files and removes listed data files by path.
///
/// The operation is `Overwrite` when files are both added and removed,
otherwise `Delete` or
/// `Append`. There is no row-filter mode, and files committed concurrently
are not checked.
```
##########
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:
Could this error name the file and its content type? When several files go
to `delete_data_files`, there's no telling which one was rejected. Also,
`crate::spec::DataContentType` here and `crate::spec::ManifestContentType` at
L185 could move into the `use crate::spec::{…}` import, though `snapshot.rs`
writes the same path inline today.
##########
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:
Could the deleted-file count follow the tombstones we actually write here,
the way Java does? As I read it, a path live in two Data manifests gets a
Deleted entry in each, while
[`deleted_data_files`](https://github.com/apache/iceberg-rust/blob/b106b0af66388e8aa296e6c7bfd8c190396e9e98/crates/iceberg/src/transaction/snapshot.rs#L142-L143)
is deduped to one. So the
[`debug_assert_eq!`](https://github.com/apache/iceberg-rust/blob/b106b0af66388e8aa296e6c7bfd8c190396e9e98/crates/iceberg/src/transaction/snapshot.rs#L398-L408)
panics in debug builds, and release reports one deleted file while the totals
drift by one per extra copy.
Two `fast_append().with_check_duplicate(false)` commits of one path create
that state, and since b106b0af6 a delete plus re-add of it gets there too.
Java's
[`ManifestFilterManager`](https://github.com/apache/iceberg/blob/2114bf631e49af532d66e2ce148ee49dd1dd1f1f/core/src/main/java/org/apache/iceberg/ManifestFilterManager.java#L483-L548)
deletes every live copy and [counts each
one](https://github.com/apache/iceberg/blob/2114bf631e49af532d66e2ce148ee49dd1dd1f1f/core/src/main/java/org/apache/iceberg/ManifestFilterManager.java#L250-L265),
with no assert. A test for this case would pin it.
##########
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:
thanks, excluding the deleted paths reads right, and
`test_overwrite_readds_a_deleted_path_by_default` pins it.
##########
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:
Could this check move into a small helper that returns `Option<String>`? The
`continue` in the last `else` of the `let` initializer is easy to misread as
falling through to the `return Err`, and the loop at L239 then filters the live
entries a second time. The behavior is right as written. This is only about
readability.
##########
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:
Could both partition cases build the expected message from the two
`Struct`s, e.g. `format!("partition: requested {:?}, table has {:?}", …)`? This
line hard-codes the derived `Debug` text, which changes on any rename in
`Struct` or `Literal`. Meanwhile the `Literal::string` case at L752 only checks
the `partition: requested` prefix.
##########
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:
Could one of these tests run on a V2 table? Every test here uses
`make_v3_minimal_table_in_catalog`, but V2 rewrites go through a different
[manifest
builder](https://github.com/apache/iceberg-rust/blob/b106b0af66388e8aa296e6c7bfd8c190396e9e98/crates/iceberg/src/transaction/snapshot.rs#L284-L294).
As I read it, nothing in the rewrite is V2-specific, so this is about
coverage, not a suspected bug. The promotion test's catalog create at L610
shows how to pick the format version.
##########
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:
Could we keep the ordering note [`main`
had](https://github.com/apache/iceberg-rust/blob/f83bec42298a1cf51a42ea2aee26c041a15db5b4/crates/iceberg/src/transaction/snapshot.rs#L517-L522)?
`summary()` still has to run before `write_added_manifest`, which
[takes](https://github.com/apache/iceberg-rust/blob/b106b0af66388e8aa296e6c7bfd8c190396e9e98/crates/iceberg/src/transaction/snapshot.rs#L327-L328)
`added_data_files`, and since the move nothing here says so.
```suggestion
// Must run before `write_added_manifest`, which takes
`added_data_files`.
let summary = self.summary(snapshot_produce_operation).map_err(|err|
{
```
##########
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:
Could this assert the whole entry list? It also passes when `a.parquet` is
missing entirely. As I read the rewrite loop, missing is the expected end
state, since the second rewrite carries only live entries and drops `a`'s
tombstone. An `assert_eq!` against the expected list (no `a.parquet`,
`b.parquet` Deleted, `c` and `d` live) would pin the delete and the tombstone
drop together, like the new re-add test does.
##########
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:
Could these reads run concurrently, the way
[`validate_duplicate_files`](https://github.com/apache/iceberg-rust/blob/b106b0af66388e8aa296e6c7bfd8c190396e9e98/crates/iceberg/src/transaction/snapshot.rs#L195-L221)
does, or at least carry a TODO? Here every non-empty Data manifest is read one
after another, with no pruning and no early exit once every delete is matched.
With `check_duplicate` on, the same commit has already read each manifest once
in that concurrent pass.
--
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]
