Merge pull request JanKaul#172 from JanKaul/fix/commit-empty-recordbatch-stream

Fix/commit empty recordbatch stream

Author: JanKaul

iceberg-rust/src/table/transaction/mod.rs

@@ -344,6 +344,10 @@ impl<'table> TableTransaction<'table> {
             updates.extend(update);
         }
 
+        if updates.is_empty() {
+            return Ok(());
+        }
+
         let new_table = catalog
             .clone()
             .update_table(CommitTable {

iceberg-rust/src/table/transaction/operation.rs

@@ -106,10 +106,8 @@ impl Operation {
                 let old_snapshot = table_metadata.current_snapshot(branch.as_deref())?;
 
                 let snapshot_operation = match (data_files.len(), delete_files.len()) {
-                    (0, 0) => Err(Error::InvalidFormat(
-                        "Empty data and delete files".to_string(),
-                    )),
-                    (_, 0) => Ok(SnapshotOperation::Append),
+                    (0, 0) => return Ok((None, Vec::new())),
+                    (_, 0) => Ok::<_, Error>(SnapshotOperation::Append),
                     (0, _) => Ok(SnapshotOperation::Delete),
                     (_, _) => Ok(SnapshotOperation::Overwrite),
                 }?;
@@ -279,18 +277,17 @@ impl Operation {
                         (selected_manifest_opt, selected_manifest_bytes_opt)
                     {
                         let manifest_bytes = manifest_bytes.await??;
-                        let manifest_reader = ManifestReader::new(&*manifest_bytes)?
-                            .map(|entry| {
-                                let mut entry = entry?;
-                                *entry.status_mut() = Status::Existing;
-                                if entry.sequence_number().is_none() {
-                                    *entry.sequence_number_mut() = Some(manifest.sequence_number);
-                                }
-                                if entry.snapshot_id().is_none() {
-                                    *entry.snapshot_id_mut() = Some(manifest.added_snapshot_id);
-                                }
-                                Ok(entry)
-                            });
+                        let manifest_reader = ManifestReader::new(&*manifest_bytes)?.map(|entry| {
+                            let mut entry = entry?;
+                            *entry.status_mut() = Status::Existing;
+                            if entry.sequence_number().is_none() {
+                                *entry.sequence_number_mut() = Some(manifest.sequence_number);
+                            }
+                            if entry.snapshot_id().is_none() {
+                                *entry.snapshot_id_mut() = Some(manifest.added_snapshot_id);
+                            }
+                            Ok(entry)
+                        });
 
                         split_datafiles(
                             new_datafile_iter.chain(manifest_reader),
@@ -459,7 +456,8 @@ impl Operation {
                     let bounding_partition_values = files
                         .iter()
                         .try_fold(None, |acc, x| {
-                            let node = partition_struct_to_vec(x.partition(), &partition_column_names)?;
+                            let node =
+                                partition_struct_to_vec(x.partition(), &partition_column_names)?;
                             let Some(mut acc) = acc else {
                                 return Ok::<_, Error>(Some(Rectangle::new(node.clone(), node)));
                             };