Add assertion for empty data files for append action

Author: dentiny

crates/iceberg/src/error.rs

@@ -28,6 +28,9 @@ pub type Result<T> = std::result::Result<T, Error>;
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 #[non_exhaustive]
 pub enum ErrorKind {
+    /// Function receives an invalid argument, which iceberg library cannot handle.
+    InvalidArgument,
+
     /// Iceberg don't know what happened here, and no actions other than
     /// just returning it back. For example, iceberg returns an internal
     /// service error.
@@ -76,6 +79,7 @@ impl From<ErrorKind> for &'static str {
             ErrorKind::TableNotFound => "TableNotFound",
             ErrorKind::NamespaceAlreadyExists => "NamespaceAlreadyExists",
             ErrorKind::NamespaceNotFound => "NamespaceNotFound",
+            ErrorKind::InvalidArgument => "InvalidArgument",
         }
     }
 }

crates/iceberg/src/transaction/append.rs

@@ -219,6 +219,14 @@ mod tests {
     use crate::transaction::Transaction;
     use crate::{TableRequirement, TableUpdate};
 
+    #[tokio::test]
+    async fn test_empty_data_append_action() {
+        let table = make_v2_minimal_table();
+        let tx = Transaction::new(&table);
+        let mut action = tx.fast_append(None, vec![]).unwrap();
+        assert(action.add_data_files(vec![]).is_err());
+    }
+
     #[tokio::test]
     async fn test_fast_append_action() {
         let table = make_v2_minimal_table();

crates/iceberg/src/transaction/snapshot.rs

@@ -129,6 +129,13 @@ impl<'a> SnapshotProduceAction<'a> {
         data_files: impl IntoIterator<Item = DataFile>,
     ) -> Result<&mut Self> {
         let data_files: Vec<DataFile> = data_files.into_iter().collect();
+        if data_files.is_empty() {
+            return Err(Error::new(
+                ErrorKind::DataInvalid,
+                "Doesn't add any data files when fast append",
+            ));
+        }
+
         for data_file in &data_files {
             if data_file.content_type() != crate::spec::DataContentType::Data {
                 return Err(Error::new(
@@ -172,6 +179,13 @@ impl<'a> SnapshotProduceAction<'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);
+        if added_data_files.is_empty() {
+            return Err(Error::new(
+                ErrorKind::Unexpected,
+                "No added data files found when write a manifest file",
+            ));
+        }
+
         let snapshot_id = self.snapshot_id;
         let format_version = self.tx.current_table.metadata().format_version();
         let manifest_entries = added_data_files.into_iter().map(|data_file| {