feat: Add RowDeltaAction for row-level deletes via delete files (#28)

Add a new RowDeltaAction transaction action that supports adding both
data files and delete files (equality and positional) in a single atomic
operation. This enables row-level deletes using Iceberg's merge-on-read
strategy.

Changes:
- Add RowDeltaAction in transaction/row_delta.rs with full test coverage
- Extend SnapshotProducer to support writing delete manifests alongside
  data manifests
- Add Transaction::row_delta() method for creating row delta actions
- Operation is Delete when only delete files are added, Overwrite when
  both data and delete files are added
- Validates format version >= 2 for delete files
- Fix pre-existing test compilation errors (missing limit field)

Author: phillipleblanc

crates/iceberg/src/arrow/caching_delete_file_loader.rs

@@ -940,6 +940,7 @@ mod tests {
             partition_spec: None,
             name_mapping: None,
             case_sensitive: false,
+            limit: None,
         };
 
         // Load the deletes - should handle both types without error

crates/iceberg/src/arrow/reader.rs

@@ -2411,6 +2411,7 @@ message schema {
             partition_spec: None,
             name_mapping: None,
             case_sensitive: false,
+            limit: None,
         };
 
         // Task 2: read the second and third row groups
@@ -2428,6 +2429,7 @@ message schema {
             partition_spec: None,
             name_mapping: None,
             case_sensitive: false,
+            limit: None,
         };
 
         let tasks1 = Box::pin(futures::stream::iter(vec![Ok(task1)])) as FileScanTaskStream;
@@ -2556,6 +2558,7 @@ message schema {
                 partition_spec: None,
                 name_mapping: None,
                 case_sensitive: false,
+                limit: None,
             })]
             .into_iter(),
         )) as FileScanTaskStream;
@@ -2728,6 +2731,7 @@ message schema {
             partition_spec: None,
             name_mapping: None,
             case_sensitive: false,
+            limit: None,
         };
 
         let tasks = Box::pin(futures::stream::iter(vec![Ok(task)])) as FileScanTaskStream;
@@ -2946,6 +2950,7 @@ message schema {
             partition_spec: None,
             name_mapping: None,
             case_sensitive: false,
+            limit: None,
         };
 
         let tasks = Box::pin(futures::stream::iter(vec![Ok(task)])) as FileScanTaskStream;
@@ -3157,6 +3162,7 @@ message schema {
             partition_spec: None,
             name_mapping: None,
             case_sensitive: false,
+            limit: None,
         };
 
         let tasks = Box::pin(futures::stream::iter(vec![Ok(task)])) as FileScanTaskStream;
@@ -3261,6 +3267,7 @@ message schema {
                 partition_spec: None,
                 name_mapping: None,
                 case_sensitive: false,
+                limit: None,
             })]
             .into_iter(),
         )) as FileScanTaskStream;
@@ -3359,6 +3366,7 @@ message schema {
                 partition_spec: None,
                 name_mapping: None,
                 case_sensitive: false,
+                limit: None,
             })]
             .into_iter(),
         )) as FileScanTaskStream;
@@ -3446,6 +3454,7 @@ message schema {
                 partition_spec: None,
                 name_mapping: None,
                 case_sensitive: false,
+                limit: None,
             })]
             .into_iter(),
         )) as FileScanTaskStream;
@@ -3547,6 +3556,7 @@ message schema {
                 partition_spec: None,
                 name_mapping: None,
                 case_sensitive: false,
+                limit: None,
             })]
             .into_iter(),
         )) as FileScanTaskStream;
@@ -3677,6 +3687,7 @@ message schema {
                 partition_spec: None,
                 name_mapping: None,
                 case_sensitive: false,
+                limit: None,
             })]
             .into_iter(),
         )) as FileScanTaskStream;
@@ -3774,6 +3785,7 @@ message schema {
                 partition_spec: None,
                 name_mapping: None,
                 case_sensitive: false,
+                limit: None,
             })]
             .into_iter(),
         )) as FileScanTaskStream;
@@ -3884,6 +3896,7 @@ message schema {
                 partition_spec: None,
                 name_mapping: None,
                 case_sensitive: false,
+                limit: None,
             })]
             .into_iter(),
         )) as FileScanTaskStream;
@@ -4024,6 +4037,7 @@ message schema {
                 partition_spec: Some(partition_spec),
                 name_mapping: None,
                 case_sensitive: false,
+                limit: None,
             })]
             .into_iter(),
         )) as FileScanTaskStream;

crates/iceberg/src/transaction/mod.rs

@@ -54,6 +54,7 @@ mod action;
 
 pub use action::*;
 mod append;
+mod row_delta;
 mod snapshot;
 mod sort_order;
 mod update_location;
@@ -71,6 +72,7 @@ use crate::spec::TableProperties;
 use crate::table::Table;
 use crate::transaction::action::BoxedTransactionAction;
 use crate::transaction::append::FastAppendAction;
+use crate::transaction::row_delta::RowDeltaAction;
 use crate::transaction::sort_order::ReplaceSortOrderAction;
 use crate::transaction::update_location::UpdateLocationAction;
 use crate::transaction::update_properties::UpdatePropertiesAction;
@@ -141,6 +143,11 @@ impl Transaction {
         FastAppendAction::new()
     }
 
+    /// Creates a row delta action for adding both data files and delete files.
+    pub fn row_delta(&self) -> RowDeltaAction {
+        RowDeltaAction::new()
+    }
+
     /// Creates replace sort order action.
     pub fn replace_sort_order(&self) -> ReplaceSortOrderAction {
         ReplaceSortOrderAction::new()