feat: add delta writer for row-level changes

This commit adds support for row-level changes (inserts, updates, and deletes) 
to Iceberg tables through a new DeltaWriter implementation. This enables CDC 
(Change Data Capture), upsert operations, and efficient row-level deletions.

Key Components:

1. RowDeltaAction Transaction
   - New transaction type for applying row-level changes atomically
   - Supports both data files and delete files in a single transaction
   - Located in transaction/row_delta.rs

2. Position Delete Writer
   - Writes position delete files for efficient row-level deletions
   - Deletes rows by file path and row position
   - Located in writer/base_writer/position_delete_writer.rs

3. Delta Writer (Combined Writer)
   - Orchestrates data file, position delete, and equality delete writers
   - Intelligently routes operations based on row tracking
   - Memory-bounded row tracking with configurable limits
   - Falls back to equality deletes for older/evicted rows
   - Located in writer/combined_writer/delta_writer.rs

Input Format:
The DeltaWriter expects RecordBatch with an operations column:
- Value 1 = Insert/Update (write to data file)
- Value -1 = Delete (write to delete file)

Memory Management:
- Tracks recently written rows for efficient position deletes
- Configurable max_seen_rows limit (default: 100,000)
- FIFO eviction when limit is reached
- Can be disabled (set to 0) to use only equality deletes

Author: DAlperin

Cargo.lock

@@ -47,6 +47,7 @@ arrow-array = "57.1"
 arrow-buffer = "57.1"
 arrow-cast = "57.1"
 arrow-ord = "57.1"
+arrow-row = "57.1"
 arrow-schema = "57.1"
 arrow-select = "57.1"
 arrow-string = "57.1"

Cargo.toml

@@ -716,6 +716,13 @@ impl Catalog for RestCatalog {
             })
             .build()?;
 
+        println!(
+            "Create table request: {:?}",
+            request
+                .body()
+                .map(|b| String::from_utf8_lossy(b.as_bytes().unwrap_or(&[])))
+        );
+
         let http_response = context.client.query_catalog(request).await?;
 
         let response = match http_response.status() {

crates/catalog/rest/src/catalog.rs

@@ -49,6 +49,7 @@ arrow-array = { workspace = true }
 arrow-buffer = { workspace = true }
 arrow-cast = { workspace = true }
 arrow-ord = { workspace = true }
+arrow-row = { workspace = true }
 arrow-schema = { workspace = true }
 arrow-select = { workspace = true }
 arrow-string = { workspace = true }

crates/iceberg/Cargo.toml

@@ -90,10 +90,19 @@ impl TransactionAction for FastAppendAction {
             self.key_metadata.clone(),
             self.snapshot_properties.clone(),
             self.added_data_files.clone(),
+            vec![], // fast append doesn't support delete files
         );
 
-        // validate added files
-        snapshot_producer.validate_added_data_files()?;
+        // validate added files - ensure they are Data content type
+        for data_file in &self.added_data_files {
+            if data_file.content_type() != crate::spec::DataContentType::Data {
+                return Err(crate::Error::new(
+                    crate::ErrorKind::DataInvalid,
+                    "Only data content type is allowed for fast append",
+                ));
+            }
+        }
+        snapshot_producer.validate_added_data_files(&self.added_data_files)?;
 
         // Checks duplicate files
         if self.check_duplicate {

crates/iceberg/src/transaction/append.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,16 @@ impl Transaction {
         FastAppendAction::new()
     }
 
+    /// Creates a row delta action for row-level changes.
+    ///
+    /// Use this action for:
+    /// - CDC (Change Data Capture) ingestion
+    /// - Upsert operations
+    /// - Adding delete files (position or equality deletes)
+    pub fn row_delta(&self) -> RowDeltaAction {
+        RowDeltaAction::new()
+    }
+
     /// Creates replace sort order action.
     pub fn replace_sort_order(&self) -> ReplaceSortOrderAction {
         ReplaceSortOrderAction::new()