add validator, write added delete files to manifest

Author: CTTY

crates/iceberg/src/transaction/append.rs

@@ -27,6 +27,7 @@ use crate::table::Table;
 use crate::transaction::snapshot::{
     DefaultManifestProcess, SnapshotProduceOperation, SnapshotProducer,
 };
+use crate::transaction::validate::SnapshotValidator;
 use crate::transaction::{ActionCommit, TransactionAction};
 
 /// FastAppendAction is a transaction action for fast append data files to the table.
@@ -113,6 +114,8 @@ impl TransactionAction for FastAppendAction {
 
 struct FastAppendOperation;
 
+impl SnapshotValidator for FastAppendOperation {}
+
 impl SnapshotProduceOperation for FastAppendOperation {
     fn operation(&self) -> Operation {
         Operation::Append

crates/iceberg/src/transaction/mod.rs

@@ -63,6 +63,7 @@ mod update_location;
 mod update_properties;
 mod update_statistics;
 mod upgrade_format_version;
+mod validate;
 
 use std::sync::Arc;
 use std::time::Duration;

crates/iceberg/src/transaction/rewrite_files.rs

@@ -26,9 +26,10 @@ use super::{ActionCommit, TransactionAction};
 use crate::error::{Error, ErrorKind, Result};
 use crate::spec::{
     DataContentType, DataFile, ManifestContentType, ManifestEntry, ManifestEntryRef, ManifestFile,
-    ManifestStatus, Operation,
+    ManifestStatus, Operation, SnapshotRef,
 };
 use crate::table::Table;
+use crate::transaction::validate::SnapshotValidator;
 
 /// Transaction action for rewriting files.
 pub struct RewriteFilesAction {
@@ -41,7 +42,12 @@ pub struct RewriteFilesAction {
     deleted_delete_files: Vec<DataFile>,
 }
 
-pub struct RewriteFilesOperation;
+pub struct RewriteFilesOperation {
+    added_data_files: Vec<DataFile>,
+    added_delete_files: Vec<DataFile>,
+    deleted_data_files: Vec<DataFile>,
+    deleted_delete_files: Vec<DataFile>,
+}
 
 impl RewriteFilesAction {
     pub fn new() -> Self {
@@ -122,13 +128,16 @@ impl TransactionAction for RewriteFilesAction {
             self.deleted_delete_files.clone(),
         );
 
-        // todo need to figure out validation
-        // 1. validate replace and added files
-        // 2. validate no new deletes using the starting snapshot id
+        let rewrite_operation = RewriteFilesOperation {
+            added_data_files: self.added_data_files.clone(),
+            added_delete_files: self.added_delete_files.clone(),
+            deleted_data_files: self.deleted_data_files.clone(),
+            deleted_delete_files: self.deleted_delete_files.clone(),
+        };
 
         // todo should be able to configure merge manifest process
         snapshot_producer
-            .commit(RewriteFilesOperation, DefaultManifestProcess)
+            .commit(rewrite_operation, DefaultManifestProcess)
             .await
     }
 }
@@ -168,6 +177,36 @@ fn copy_with_deleted_status(entry: &ManifestEntryRef) -> Result<ManifestEntry> {
     Ok(builder.build())
 }
 
+impl SnapshotValidator for RewriteFilesOperation {
+    fn validate(&self, _table: &Table, _snapshot: Option<&SnapshotRef>) -> Result<()> {
+        // Validate replaced and added files
+        if self.deleted_data_files.is_empty() && self.deleted_delete_files.is_empty() {
+            return Err(Error::new(
+                ErrorKind::DataInvalid,
+                "Files to delete cannot be empty",
+            ));
+        }
+        if self.deleted_data_files.is_empty() && !self.added_data_files.is_empty() {
+            return Err(Error::new(
+                ErrorKind::DataInvalid,
+                "Data files to add must be empty because there's no data file to be rewritten",
+            ));
+        }
+        if self.deleted_delete_files.is_empty() && !self.added_delete_files.is_empty() {
+            return Err(Error::new(
+                ErrorKind::DataInvalid,
+                "Delete files to add must be empty because there's no delete file to be rewritten",
+            ));
+        }
+
+        // todo add use_starting_seq_number to help the validation
+        // todo validate no new deletes since the current base
+        // if there are replaced data files, there cannot be any new row-level deletes for those data files
+
+        Ok(())
+    }
+}
+
 impl SnapshotProduceOperation for RewriteFilesOperation {
     fn operation(&self) -> Operation {
         Operation::Replace

crates/iceberg/src/transaction/snapshot.rs

@@ -31,11 +31,12 @@ use crate::spec::{
 };
 use crate::table::Table;
 use crate::transaction::ActionCommit;
+use crate::transaction::validate::SnapshotValidator;
 use crate::{Error, ErrorKind, TableRequirement, TableUpdate};
 
 const META_ROOT_PATH: &str = "metadata";
 
-pub(crate) trait SnapshotProduceOperation: Send + Sync {
+pub(crate) trait SnapshotProduceOperation: Send + Sync + SnapshotValidator {
     fn operation(&self) -> Operation;
     fn delete_entries(
         &self,
@@ -265,8 +266,12 @@ impl<'a> SnapshotProducer<'a> {
     }
 
     // Write manifest file for added data files and return the ManifestFile for ManifestList.
-    async fn write_added_manifest(&mut self) -> Result<ManifestFile> {
-        let added_data_files = std::mem::take(&mut self.added_data_files);
+    async fn write_added_manifest(&mut self, content_type: ManifestContentType) -> Result<ManifestFile> {
+        let added_data_files =  match content_type {
+            ManifestContentType::Data => std::mem::take(&mut self.added_data_files),
+            ManifestContentType::Deletes => std::mem::take(&mut self.added_delete_files),
+        };
+
         if added_data_files.is_empty() {
             return Err(Error::new(
                 ErrorKind::PreconditionFailed,
@@ -289,7 +294,7 @@ impl<'a> SnapshotProducer<'a> {
             }
         });
         let mut writer = self.new_manifest_writer(
-            ManifestContentType::Data,
+            content_type,
             self.table.metadata().default_partition_spec_id(),
         )?;
         for entry in manifest_entries {
@@ -373,7 +378,11 @@ impl<'a> SnapshotProducer<'a> {
 
         // Process added entries.
         if !self.added_data_files.is_empty() {
-            let added_manifest = self.write_added_manifest().await?;
+            let added_manifest = self.write_added_manifest(ManifestContentType::Data).await?;
+            manifest_files.push(added_manifest);
+        }
+        if !self.added_delete_files.is_empty() {
+            let added_manifest = self.write_added_manifest(ManifestContentType::Deletes).await?;
             manifest_files.push(added_manifest);
         }
 
@@ -455,6 +464,10 @@ impl<'a> SnapshotProducer<'a> {
         snapshot_produce_operation: OP,
         process: MP,
     ) -> Result<ActionCommit> {
+        // Validate to avoid conflicts
+        snapshot_produce_operation
+            .validate(self.table, self.table.metadata().current_snapshot())?;
+
         let new_manifests = self
             .manifest_file(&snapshot_produce_operation, &process)
             .await?;

crates/iceberg/src/transaction/validate.rs

@@ -0,0 +1,42 @@
+// 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::HashSet;
+
+use crate::error::Result;
+use crate::spec::{ManifestContentType, ManifestFile, Operation, SnapshotRef};
+use crate::table::Table;
+
+pub(crate) trait SnapshotValidator {
+    fn validate(&self, _table: &Table, _snapshot: Option<&SnapshotRef>) -> Result<()> {
+        // todo: add default implementation
+        Ok(())
+    }
+
+    #[allow(dead_code)]
+    async fn validation_history(
+        &self,
+        _base: &Table,
+        _from_snapshot: Option<&SnapshotRef>,
+        _to_snapshot: &SnapshotRef,
+        _matching_operations: HashSet<Operation>,
+        _manifest_content_type: ManifestContentType,
+    ) -> Result<(Vec<ManifestFile>, HashSet<i64>)> {
+        // todo: add default implementation
+        Ok((vec![], HashSet::new()))
+    }
+}