feat: add row delta update

src/iceberg/CMakeLists.txt

@@ -106,6 +106,7 @@ set(ICEBERG_SOURCES
     update/merge_append.cc
     update/merging_snapshot_update.cc
     update/pending_update.cc
+    update/row_delta.cc
     update/set_snapshot.cc
     update/snapshot_manager.cc
     update/snapshot_update.cc

src/iceberg/meson.build

@@ -131,6 +131,7 @@ iceberg_sources = files(
     'update/merge_append.cc',
     'update/merging_snapshot_update.cc',
     'update/pending_update.cc',
+    'update/row_delta.cc',
     'update/set_snapshot.cc',
     'update/snapshot_manager.cc',
     'update/snapshot_update.cc',

src/iceberg/table.cc

@@ -35,6 +35,7 @@
 #include "iceberg/update/expire_snapshots.h"
 #include "iceberg/update/fast_append.h"
 #include "iceberg/update/merge_append.h"
+#include "iceberg/update/row_delta.h"
 #include "iceberg/update/set_snapshot.h"
 #include "iceberg/update/snapshot_manager.h"
 #include "iceberg/update/update_location.h"
@@ -231,6 +232,12 @@ Result<std::shared_ptr<DeleteFiles>> Table::NewDeleteFiles() {
   return DeleteFiles::Make(name().name, std::move(ctx));
 }
 
+Result<std::shared_ptr<RowDelta>> Table::NewRowDelta() {
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto ctx, TransactionContext::Make(shared_from_this(), TransactionKind::kUpdate));
+  return RowDelta::Make(name().name, std::move(ctx));
+}
+
 Result<std::shared_ptr<UpdateStatistics>> Table::NewUpdateStatistics() {
   ICEBERG_ASSIGN_OR_RAISE(
       auto ctx, TransactionContext::Make(shared_from_this(), TransactionKind::kUpdate));
@@ -334,6 +341,14 @@ Result<std::shared_ptr<MergeAppend>> StaticTable::NewMergeAppend() {
   return NotSupported("Cannot create a merge append for a static table");
 }
 
+Result<std::shared_ptr<DeleteFiles>> StaticTable::NewDeleteFiles() {
+  return NotSupported("Cannot create delete files for a static table");
+}
+
+Result<std::shared_ptr<RowDelta>> StaticTable::NewRowDelta() {
+  return NotSupported("Cannot create a row delta for a static table");
+}
+
 Result<std::shared_ptr<SnapshotManager>> StaticTable::NewSnapshotManager() {
   return NotSupported("Cannot create a snapshot manager for a static table");
 }

src/iceberg/table.h

@@ -182,6 +182,9 @@ class ICEBERG_EXPORT Table : public std::enable_shared_from_this<Table> {
   /// \brief Create a new DeleteFiles to delete data files and commit the changes.
   virtual Result<std::shared_ptr<DeleteFiles>> NewDeleteFiles();
 
+  /// \brief Create a new RowDelta to add rows and row-level deletes.
+  virtual Result<std::shared_ptr<RowDelta>> NewRowDelta();
+
   /// \brief Create a new SnapshotManager to manage snapshots and snapshot references.
   virtual Result<std::shared_ptr<SnapshotManager>> NewSnapshotManager();
 
@@ -251,6 +254,10 @@ class ICEBERG_EXPORT StaticTable : public Table {
 
   Result<std::shared_ptr<MergeAppend>> NewMergeAppend() override;
 
+  Result<std::shared_ptr<DeleteFiles>> NewDeleteFiles() override;
+
+  Result<std::shared_ptr<RowDelta>> NewRowDelta() override;
+
   Result<std::shared_ptr<SnapshotManager>> NewSnapshotManager() override;
 
  private:

src/iceberg/test/CMakeLists.txt

@@ -230,6 +230,7 @@ if(ICEBERG_BUILD_BUNDLE)
                    merge_append_test.cc
                    merging_snapshot_update_test.cc
                    name_mapping_update_test.cc
+                   row_delta_test.cc
                    snapshot_manager_test.cc
                    transaction_test.cc
                    update_location_test.cc