feat: add delta writer for row-level changes

Cargo.toml

@@ -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"

crates/catalog/rest/src/catalog.rs

@@ -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/iceberg/Cargo.toml

@@ -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/src/transaction/append.rs

@@ -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/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,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()