diff --git a/docs/docs/pypaimon/ray-data.md b/docs/docs/pypaimon/ray-data.md
index 88c0a7143ac4..13becf26491c 100644
--- a/docs/docs/pypaimon/ray-data.md
+++ b/docs/docs/pypaimon/ray-data.md
@@ -524,6 +524,7 @@ metrics = update_by_row_id(
     source=ray_dataset,          # ray.data.Dataset / pa.Table / pandas, carrying _ROW_ID
     catalog_options={"warehouse": "/path/to/warehouse"},
     update_cols=["feature"],     # non-blob columns to overwrite
+    max_groups_per_commit=64,    # optional incremental commit window
 )
 print(metrics)   # {"num_updated": 50}
 ```
@@ -537,6 +538,8 @@ print(metrics)   # {"num_updated": 50}
 - `num_partitions`: parallelism for grouping the update rows by target file;
   defaults to `max(1, cluster_cpus * 2)`.
 - `ray_remote_args`: Ray remote options applied to the update tasks.
+- `max_groups_per_commit`: optional positive number of completed target file groups
+  per commit. By default, all groups are committed together.
 
 **Returns:** `{"num_updated": <rows>}`.
 
@@ -548,6 +551,13 @@ print(metrics)   # {"num_updated": 50}
 - Partition columns cannot be updated (in-place rewrite can't move a row across partitions).
 - Deletion-vectors-enabled tables are not supported yet: a DV-deleted row still lives
   in its data file, so it can't be told apart from a live row without reading the target.
+- Incremental commits are not atomic across the whole operation. Ordinary exceptions
+  raised while applying a group are reported as results so other groups can finish;
+  completed groups are flushed before the error is raised.
+- Ray task retry options in `ray_remote_args` do not retry group exceptions reported
+  as results, so a transient group failure can leave a partial update. Worker loss and
+  other task-level failures which bypass Python exception handling may still lose
+  results buffered by that task.
 
 ## Read By Row Id
 
diff --git a/paimon-python/pypaimon/ray/data_evolution_merge_join.py b/paimon-python/pypaimon/ray/data_evolution_merge_join.py
index 359af8e6efa2..4f1a64064e8e 100644
--- a/paimon-python/pypaimon/ray/data_evolution_merge_join.py
+++ b/paimon-python/pypaimon/ray/data_evolution_merge_join.py
@@ -16,7 +16,7 @@
 # limitations under the License.
 ################################################################################
 
-from typing import Any, Dict, List, Optional, Sequence, Tuple
+from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple
 
 import pyarrow as pa
 
@@ -32,6 +32,21 @@
 )
 
 
+class GroupApplyError(RuntimeError):
+    """A file group failed after other groups may have completed."""
+
+
+def _group_error_text(error: BaseException) -> str:
+    import traceback
+
+    return "update_by_row_id file group failed: {}: {}\n{}".format(
+        type(error).__name__,
+        str(error),
+        "".join(traceback.format_exception(
+            type(error), error, error.__traceback__)),
+    )
+
+
 def _map_kwargs(
     ray_remote_args: Optional[Dict[str, Any]],
 ) -> Dict[str, Any]:
@@ -445,6 +460,7 @@ def distributed_update_apply(
     ray_remote_args: Optional[Dict[str, Any]] = None,
     base_snapshot_id: Optional[int] = None,
     collect_row_ids: bool = False,
+    on_group_result: Optional[Callable[[list, int, list], None]] = None,
 ) -> Tuple[list, int, list]:
     import numpy as np
     import pickle
@@ -546,6 +562,15 @@ def _assign_frid(batch: pa.Table) -> pa.Table:
 
     captured_table = table
     captured_cols = cols
+    capture_group_errors = on_group_result is not None
+
+    def _group_result(msgs_blob, n_updated, row_ids_blob, error):
+        return pa.Table.from_pydict({
+            "msgs_blob": pa.array([msgs_blob], type=pa.binary()),
+            "n_updated": pa.array([n_updated], type=pa.int64()),
+            "row_ids_blob": pa.array([row_ids_blob], type=pa.binary()),
+            "error": pa.array([error], type=pa.string()),
+        })
 
     def _apply_group(group: pa.Table) -> pa.Table:
         if group.num_rows == 0:
@@ -553,39 +578,43 @@ def _apply_group(group: pa.Table) -> pa.Table:
                 "msgs_blob": pa.array([], type=pa.binary()),
                 "n_updated": pa.array([], type=pa.int64()),
                 "row_ids_blob": pa.array([], type=pa.binary()),
+                "error": pa.array([], type=pa.string()),
             })
 
-        if (
-            pc.count_distinct(group.column(row_id_name)).as_py()
-            != group.num_rows
-        ):
-            raise ValueError(
-                "MERGE matched multiple source rows to the same "
-                "target _ROW_ID. Deduplicate the source before "
-                "merging."
-            )
+        try:
+            if (
+                pc.count_distinct(group.column(row_id_name)).as_py()
+                != group.num_rows
+            ):
+                raise ValueError(
+                    "MERGE matched multiple source rows to the same "
+                    "target _ROW_ID. Deduplicate the source before "
+                    "merging."
+                )
 
-        for_update = group.drop_columns([frid_col])
-        row_ids = (
-            for_update.column(row_id_name).to_pylist()
-            if collect_row_ids else []
-        )
-        worker = TableUpdateByRowId(
-            captured_table,
-            "_merge_into_shard_" + uuid.uuid4().hex[:8],
-            BATCH_COMMIT_IDENTIFIER,
-            _precomputed_files_info=ray.get(precomputed_info_ref),
+            for_update = group.drop_columns([frid_col])
+            row_ids = (
+                for_update.column(row_id_name).to_pylist()
+                if collect_row_ids else []
+            )
+            worker = TableUpdateByRowId(
+                captured_table,
+                "_merge_into_shard_" + uuid.uuid4().hex[:8],
+                BATCH_COMMIT_IDENTIFIER,
+                _precomputed_files_info=ray.get(precomputed_info_ref),
+            )
+            msgs = worker.update_columns(for_update, list(captured_cols))
+        except Exception as error:
+            if not capture_group_errors:
+                raise
+            return _group_result(b"", 0, b"", _group_error_text(error))
+
+        return _group_result(
+            pickle.dumps(msgs),
+            for_update.num_rows,
+            pickle.dumps(row_ids),
+            None,
         )
-        msgs = worker.update_columns(for_update, list(captured_cols))
-        return pa.Table.from_pydict({
-            "msgs_blob": [pickle.dumps(msgs)],
-            "n_updated": pa.array(
-                [for_update.num_rows], type=pa.int64()
-            ),
-            "row_ids_blob": pa.array(
-                [pickle.dumps(row_ids)], type=pa.binary()
-            ),
-        })
 
     # One group per target data file; bounded by file count and num_partitions.
     group_partitions = max(
@@ -598,14 +627,36 @@ def _apply_group(group: pa.Table) -> pa.Table:
     all_msgs: list = []
     num_updated = 0
     action_row_ids = []
+    group_error = None
     for batch in msgs_ds.iter_batches(batch_format="pyarrow"):
-        for blob in batch.column("msgs_blob").to_pylist():
-            all_msgs.extend(pickle.loads(blob))
-        for n in batch.column("n_updated").to_pylist():
+        message_blobs = batch.column("msgs_blob").to_pylist()
+        updated_counts = batch.column("n_updated").to_pylist()
+        errors = batch.column("error").to_pylist()
+        row_id_blobs = (
+            batch.column("row_ids_blob").to_pylist()
+            if collect_row_ids else [None] * len(message_blobs)
+        )
+        for blob, n, row_ids_blob, error in zip(
+                message_blobs, updated_counts, row_id_blobs, errors):
+            if error is not None:
+                # Keep the first failure as the primary error, but continue
+                # draining results so successful groups can still be committed.
+                if group_error is None:
+                    group_error = error
+                continue
+            group_msgs = pickle.loads(blob)
+            group_row_ids = (
+                pickle.loads(row_ids_blob) if collect_row_ids else []
+            )
+            if on_group_result is None:
+                all_msgs.extend(group_msgs)
+            else:
+                on_group_result(group_msgs, n, group_row_ids)
             num_updated += n
-        if collect_row_ids:
-            for blob in batch.column("row_ids_blob").to_pylist():
-                action_row_ids.extend(pickle.loads(blob))
+            if collect_row_ids:
+                action_row_ids.extend(group_row_ids)
+    if group_error is not None:
+        raise GroupApplyError(group_error)
     return all_msgs, num_updated, action_row_ids
 
 
diff --git a/paimon-python/pypaimon/ray/update_by_row_id.py b/paimon-python/pypaimon/ray/update_by_row_id.py
index 41b968c4fdc8..2335d3a67526 100644
--- a/paimon-python/pypaimon/ray/update_by_row_id.py
+++ b/paimon-python/pypaimon/ray/update_by_row_id.py
@@ -34,7 +34,10 @@
     _require_ray_join,
     _resolve_num_partitions,
 )
-from pypaimon.ray.data_evolution_merge_join import distributed_update_apply
+from pypaimon.ray.data_evolution_merge_join import (
+    GroupApplyError,
+    distributed_update_apply,
+)
 from pypaimon.ray.data_evolution_merge_transform import build_update_schema
 from pypaimon.schema.data_types import is_blob_file_field
 
@@ -56,6 +59,7 @@ def update_by_row_id(
     update_cols: List[str],
     num_partitions: Optional[int] = None,
     ray_remote_args: Optional[Dict[str, Any]] = None,
+    max_groups_per_commit: Optional[int] = None,
 ) -> Dict[str, int]:
     """Update ``update_cols`` of a data-evolution table by ``_ROW_ID``.
 
@@ -65,6 +69,14 @@ def update_by_row_id(
     the target is never fully read and there is no join against it. Requires
     ``ray >= 2.50`` and a target with ``data-evolution.enabled`` + ``row-tracking.enabled``.
 
+    By default, all file groups are committed atomically. Set
+    ``max_groups_per_commit`` to commit completed groups in smaller windows.
+    Ordinary exceptions raised while applying a group are returned as results
+    in incremental mode, so completed groups are flushed before the error is
+    raised and Ray task retry options in ``ray_remote_args`` do not retry the
+    group. Worker loss and other task-level failures which bypass Python
+    exception handling may still lose results buffered by that task.
+
     Returns ``{"num_updated": <rows>}``.
     """
     from pypaimon.catalog.catalog_factory import CatalogFactory
@@ -74,6 +86,11 @@ def update_by_row_id(
     _require_ray_join()
     if not update_cols:
         raise ValueError("update_cols must be non-empty.")
+    if max_groups_per_commit is not None:
+        if (isinstance(max_groups_per_commit, bool)
+                or not isinstance(max_groups_per_commit, int)
+                or max_groups_per_commit <= 0):
+            raise ValueError("max_groups_per_commit must be a positive integer.")
     update_cols = list(dict.fromkeys(update_cols))  # de-dup, keep order
     num_partitions = _resolve_num_partitions(num_partitions)
 
@@ -139,22 +156,120 @@ def _project_cast(batch: pa.Table) -> pa.Table:
             raise ValueError(
                 f"target '{target}' has no rows; every _ROW_ID in the source is foreign.")
         return {"num_updated": 0}
+    incremental_committer = (
+        _IncrementalUpdateCommitter(table, max_groups_per_commit)
+        if max_groups_per_commit is not None else None
+    )
     try:
+        apply_kwargs = {
+            "num_partitions": num_partitions,
+            "ray_remote_args": ray_remote_args,
+            "base_snapshot_id": base.id,
+        }
+        if incremental_committer is not None:
+            apply_kwargs["on_group_result"] = incremental_committer.add_group
         msgs, num_updated, _ = distributed_update_apply(
-            update_ds, table, update_cols,
-            num_partitions=num_partitions,
-            ray_remote_args=ray_remote_args,
-            base_snapshot_id=base.id,
+            update_ds, table, update_cols, **apply_kwargs
         )
+        if incremental_committer is not None:
+            incremental_committer.finish()
+    except GroupApplyError:
+        if incremental_committer is not None:
+            try:
+                incremental_committer.finish()
+            except Exception:
+                incremental_committer.abort_pending()
+                raise
+        raise
     except Exception as e:
+        if incremental_committer is not None:
+            incremental_committer.abort_pending()
+            if incremental_committer._commit_failed:
+                raise
         _reraise_inner(e)
         raise  # _reraise_inner always raises; keeps msgs/num_updated defined for linters
+    finally:
+        if incremental_committer is not None:
+            incremental_committer.close()
 
-    if msgs:
+    if incremental_committer is None and msgs:
         _commit_update_messages(table, msgs)
     return {"num_updated": num_updated}
 
 
+class _IncrementalUpdateCommitter:
+
+    def __init__(self, table, max_groups_per_commit: int):
+        self._table = table
+        self._max_groups_per_commit = max_groups_per_commit
+        self._pending_messages: list = []
+        self._pending_groups = 0
+        self._table_commit = None
+        self._next_commit_identifier = 1
+        self._commit_failed = False
+        self._deferred_commit_error = None
+
+    def add_group(self, commit_messages, _num_updated, _row_ids) -> None:
+        self._pending_messages.extend(commit_messages)
+        self._pending_groups += 1
+        if (self._deferred_commit_error is None
+                and self._pending_groups >= self._max_groups_per_commit):
+            try:
+                self._commit_pending()
+            except Exception as error:
+                # Keep draining materialized Ray results so their staged files
+                # are known to the driver and can be aborted by the caller.
+                self._deferred_commit_error = error
+
+    def finish(self) -> None:
+        if self._deferred_commit_error is not None:
+            raise self._deferred_commit_error
+        self._commit_pending()
+
+    def _commit_pending(self) -> None:
+        if self._pending_groups == 0:
+            return
+        if not self._pending_messages:
+            self._pending_groups = 0
+            return
+
+        try:
+            if self._table_commit is None:
+                self._table_commit = (
+                    self._table.new_stream_write_builder().new_commit()
+                )
+
+            messages = self._pending_messages
+            self._pending_messages = []
+            self._pending_groups = 0
+            commit_identifier = self._next_commit_identifier
+            self._table_commit.commit(messages, commit_identifier)
+        except Exception:
+            self._commit_failed = True
+            raise
+        self._next_commit_identifier += 1
+
+    def abort_pending(self) -> None:
+        if not self._pending_messages:
+            return
+        messages = self._pending_messages
+        self._pending_messages = []
+        self._pending_groups = 0
+        _abort_pending_update_messages(self._table, messages)
+
+    def close(self) -> None:
+        if self._table_commit is None:
+            return
+        try:
+            self._table_commit.close()
+        except Exception as close_error:
+            logger.warning(
+                "Failed to close incremental update_by_row_id commit: %s",
+                close_error,
+                exc_info=close_error,
+            )
+
+
 def _commit_update_messages(table, commit_messages) -> None:
     pending_msgs: list = list(commit_messages)
     commit_started = False
diff --git a/paimon-python/pypaimon/tests/ray_update_by_row_id_test.py b/paimon-python/pypaimon/tests/ray_update_by_row_id_test.py
index 613fa50d0319..45a412afe001 100644
--- a/paimon-python/pypaimon/tests/ray_update_by_row_id_test.py
+++ b/paimon-python/pypaimon/tests/ray_update_by_row_id_test.py
@@ -88,6 +88,16 @@ def _rowid_by_id(self, target):
         tab = self._read(target, ["_ROW_ID", "id"])
         return dict(zip(tab.column("id").to_pylist(), tab.column("_ROW_ID").to_pylist()))
 
+    @staticmethod
+    def _data_files_under(table):
+        table_path = table.file_io.to_filesystem_path(table.table_path)
+        return {
+            os.path.join(root, file_name)
+            for root, _, files in os.walk(table_path)
+            for file_name in files
+            if file_name.endswith(".parquet")
+        }
+
     def test_update_by_row_id_basic(self):
         target = self._create()
         self._write(target, pa.Table.from_pydict(
@@ -130,6 +140,277 @@ def test_updates_correct_row_across_files(self):
         self.assertEqual(got[21], 999)
         self.assertTrue(all(v == 0 for k, v in got.items() if k != 21))
 
+    def test_incrementally_commits_file_group_windows(self):
+        from pypaimon.write.table_commit import StreamTableCommit
+
+        target = self._create()
+        chunks = [[start, start + 1] for start in range(10, 60, 10)]
+        for chunk in chunks:
+            self._write(target, pa.Table.from_pydict(
+                {"id": chunk, "name": ["x", "x"], "age": [0, 0]},
+                schema=self.pa_schema,
+            ))
+
+        table = self.catalog.get_table(target)
+        base_snapshot_id = table.snapshot_manager().get_latest_snapshot().id
+        row_ids = self._rowid_by_id(target)
+        updated_ids = [chunk[0] for chunk in chunks]
+        source = pa.table(
+            {
+                "_ROW_ID": [row_ids[row_id] for row_id in updated_ids],
+                "age": updated_ids,
+            },
+            schema=pa.schema([("_ROW_ID", pa.int64()), ("age", pa.int32())]),
+        )
+        commits = []
+        original_commit = StreamTableCommit.commit
+
+        def record_commit(stream_commit, messages, commit_identifier):
+            commits.append((len(messages), commit_identifier))
+            return original_commit(
+                stream_commit, messages, commit_identifier)
+
+        with mock.patch.object(StreamTableCommit, "commit", record_commit):
+            stats = update_by_row_id(
+                target,
+                ray.data.from_arrow(source).repartition(4),
+                self.catalog_options,
+                update_cols=["age"],
+                num_partitions=4,
+                max_groups_per_commit=2,
+            )
+
+        self.assertEqual({"num_updated": 5}, stats)
+        self.assertEqual([(2, 1), (2, 2), (1, 3)], commits)
+        self.assertEqual(
+            base_snapshot_id + 3,
+            table.snapshot_manager().get_latest_snapshot().id,
+        )
+        result = self._read(target).sort_by("id").to_pydict()
+        ages = dict(zip(result["id"], result["age"]))
+        self.assertEqual(
+            {row_id: row_id for row_id in updated_ids},
+            {row_id: ages[row_id] for row_id in updated_ids},
+        )
+
+    def test_group_failure_preserves_completed_groups(self):
+        for max_groups, expected_snapshots in [(1, 2), (5, 1)]:
+            with self.subTest(max_groups_per_commit=max_groups):
+                target = self._create()
+                for row_id in range(1, 4):
+                    self._write(target, pa.Table.from_pydict(
+                        {"id": [row_id], "name": ["a"], "age": [0]},
+                        schema=self.pa_schema,
+                    ))
+
+                table = self.catalog.get_table(target)
+                base_snapshot_id = (
+                    table.snapshot_manager().get_latest_snapshot().id
+                )
+                row_ids = self._rowid_by_id(target)
+                source = pa.table(
+                    {
+                        "_ROW_ID": [
+                            row_ids[1], row_ids[2], row_ids[3], row_ids[3],
+                        ],
+                        "age": [100, 200, 300, 301],
+                    },
+                    schema=pa.schema([
+                        ("_ROW_ID", pa.int64()),
+                        ("age", pa.int32()),
+                    ]),
+                )
+
+                with self.assertRaisesRegex(RuntimeError, "Deduplicate"):
+                    update_by_row_id(
+                        target,
+                        ray.data.from_arrow(source),
+                        self.catalog_options,
+                        update_cols=["age"],
+                        num_partitions=1,
+                        max_groups_per_commit=max_groups,
+                    )
+
+                self.assertEqual(
+                    base_snapshot_id + expected_snapshots,
+                    table.snapshot_manager().get_latest_snapshot().id,
+                )
+                self.assertEqual(
+                    [100, 200, 0],
+                    self._read(target).sort_by("id")["age"].to_pylist(),
+                )
+
+    def test_atomic_group_failure_commits_nothing(self):
+        target = self._create()
+        for row_id in range(1, 4):
+            self._write(target, pa.Table.from_pydict(
+                {"id": [row_id], "name": ["a"], "age": [0]},
+                schema=self.pa_schema,
+            ))
+
+        table = self.catalog.get_table(target)
+        base_snapshot_id = table.snapshot_manager().get_latest_snapshot().id
+        row_ids = self._rowid_by_id(target)
+        source = pa.table(
+            {
+                "_ROW_ID": [
+                    row_ids[1], row_ids[2], row_ids[3], row_ids[3],
+                ],
+                "age": [100, 200, 300, 301],
+            },
+            schema=pa.schema([
+                ("_ROW_ID", pa.int64()),
+                ("age", pa.int32()),
+            ]),
+        )
+
+        with self.assertRaisesRegex(ValueError, "Deduplicate"):
+            update_by_row_id(
+                target,
+                ray.data.from_arrow(source),
+                self.catalog_options,
+                update_cols=["age"],
+                num_partitions=1,
+            )
+
+        self.assertEqual(
+            base_snapshot_id,
+            table.snapshot_manager().get_latest_snapshot().id,
+        )
+        self.assertEqual(
+            [0, 0, 0],
+            self._read(target).sort_by("id")["age"].to_pylist(),
+        )
+
+    def test_incremental_committer_batches_and_aborts_pending_groups(self):
+        import importlib
+
+        module = importlib.import_module("pypaimon.ray.update_by_row_id")
+        commits = []
+        aborted = []
+        close_calls = []
+
+        class FakeCommit:
+            def commit(self, messages, commit_identifier):
+                commits.append((list(messages), commit_identifier))
+
+            def close(self):
+                close_calls.append(True)
+
+        class FakeBuilder:
+            def new_commit(self):
+                return FakeCommit()
+
+        class FakeTable:
+            def new_stream_write_builder(self):
+                return FakeBuilder()
+
+        committer = module._IncrementalUpdateCommitter(FakeTable(), 2)
+        with mock.patch.object(
+                module,
+                "_abort_pending_update_messages",
+                side_effect=lambda table, messages: aborted.append(list(messages))):
+            committer.add_group(["group-1"], 1, [])
+            committer.add_group(["group-2"], 1, [])
+            committer.add_group(["group-3"], 1, [])
+            committer.abort_pending()
+            committer.close()
+
+        self.assertEqual(
+            [(["group-1", "group-2"], 1)],
+            commits,
+        )
+        self.assertEqual([["group-3"]], aborted)
+        self.assertEqual([True], close_calls)
+
+    def test_incremental_commit_failure_aborts_later_groups(self):
+        import importlib
+
+        module = importlib.import_module("pypaimon.ray.update_by_row_id")
+        aborted = []
+        commit_calls = []
+
+        class FakeCommit:
+            def commit(self, messages, commit_identifier):
+                commit_calls.append((list(messages), commit_identifier))
+                raise RuntimeError("commit failed")
+
+            def close(self):
+                pass
+
+        class FakeBuilder:
+            def new_commit(self):
+                return FakeCommit()
+
+        class FakeTable:
+            def new_stream_write_builder(self):
+                return FakeBuilder()
+
+        committer = module._IncrementalUpdateCommitter(FakeTable(), 1)
+        with mock.patch.object(
+                module,
+                "_abort_pending_update_messages",
+                side_effect=lambda table, messages: aborted.append(list(messages))):
+            committer.add_group(["group-1"], 1, [])
+            committer.add_group(["group-2"], 1, [])
+            with self.assertRaisesRegex(RuntimeError, "commit failed"):
+                committer.finish()
+            committer.abort_pending()
+            committer.close()
+
+        self.assertEqual([(["group-1"], 1)], commit_calls)
+        self.assertEqual([["group-2"]], aborted)
+
+    def test_incremental_commit_conflict_aborts_buffered_group_files(self):
+        from pypaimon.write.commit.conflict_detection import CommitConflictError
+        from pypaimon.write.file_store_commit import FileStoreCommit
+
+        target = self._create()
+        for row_id in range(1, 4):
+            self._write(target, pa.Table.from_pydict(
+                {"id": [row_id], "name": ["a"], "age": [0]},
+                schema=self.pa_schema,
+            ))
+
+        table = self.catalog.get_table(target)
+        base_snapshot_id = table.snapshot_manager().get_latest_snapshot().id
+        files_before = self._data_files_under(table)
+        row_ids = self._rowid_by_id(target)
+        source = pa.table(
+            {
+                "_ROW_ID": [row_ids[1], row_ids[2], row_ids[3]],
+                "age": [100, 200, 300],
+            },
+            schema=pa.schema([
+                ("_ROW_ID", pa.int64()),
+                ("age", pa.int32()),
+            ]),
+        )
+
+        with mock.patch.object(
+                FileStoreCommit,
+                "commit",
+                side_effect=CommitConflictError("forced conflict")):
+            with self.assertRaisesRegex(CommitConflictError, "forced conflict"):
+                update_by_row_id(
+                    target,
+                    ray.data.from_arrow(source),
+                    self.catalog_options,
+                    update_cols=["age"],
+                    num_partitions=1,
+                    max_groups_per_commit=1,
+                )
+
+        self.assertEqual(files_before, self._data_files_under(table))
+        self.assertEqual(
+            base_snapshot_id,
+            table.snapshot_manager().get_latest_snapshot().id,
+        )
+        self.assertEqual(
+            [0, 0, 0],
+            self._read(target).sort_by("id")["age"].to_pylist(),
+        )
+
     def test_pins_base_snapshot_for_conflict_detection(self):
         # The update pins its base snapshot and threads it to distributed_update_apply,
         # which uses it for commit-time conflict detection against concurrent writers.
@@ -183,6 +464,20 @@ def test_commit_failure_does_not_abort_after_commit_started(self):
         self.assertEqual(recorder["abort_calls"], 0)
         self.assertEqual(recorder["close_calls"], 1)
 
+    def test_incremental_driver_commit_failure_is_not_unwrapped(self):
+        retry_error = ValueError("retry failed")
+        commit_error = RuntimeError("commit failed")
+        commit_error.__cause__ = retry_error
+
+        with self.assertRaises(RuntimeError) as raised:
+            self._run_with_fake_commit(
+                commit_error=commit_error,
+                incremental=True,
+            )
+
+        self.assertIs(commit_error, raised.exception)
+        self.assertIs(retry_error, raised.exception.__cause__)
+
     def test_close_failure_after_success_warns_and_returns_stats(self):
         close_error = RuntimeError("close failed")
 
@@ -324,8 +619,30 @@ def test_rejects_unknown_and_empty_update_cols(self):
         with self.assertRaises(ValueError):
             update_by_row_id(target, src, self.catalog_options, update_cols=[])
 
+    def test_rejects_invalid_max_groups_per_commit(self):
+        target = self._create()
+        self._write(target, pa.Table.from_pydict(
+            {"id": [1], "name": ["a"], "age": [1]}, schema=self.pa_schema))
+        source = pa.table(
+            {"_ROW_ID": [0], "age": [9]},
+            schema=pa.schema([("_ROW_ID", pa.int64()), ("age", pa.int32())]),
+        )
+
+        for value in [0, -1, True, 1.5, "2"]:
+            with self.subTest(value=value):
+                with self.assertRaisesRegex(
+                        ValueError, "must be a positive integer"):
+                    update_by_row_id(
+                        target,
+                        source,
+                        self.catalog_options,
+                        update_cols=["age"],
+                        max_groups_per_commit=value,
+                    )
+
     def _run_with_fake_commit(self, *, recorder=None, new_commit_errors=None,
-                              commit_error=None, close_error=None):
+                              commit_error=None, close_error=None,
+                              incremental=False):
         import importlib
 
         m = importlib.import_module("pypaimon.ray.update_by_row_id")
@@ -354,7 +671,7 @@ def deletion_vectors_enabled(self):
                 return False
 
         class FakeCommit:
-            def commit(self, msgs):
+            def commit(self, msgs, *args):
                 recorder["commit_calls"] += 1
                 recorder["commit_msgs"] = list(msgs)
                 if recorder["commit_error"] is not None:
@@ -396,6 +713,9 @@ def snapshot_manager(self):
             def new_batch_write_builder(self):
                 return FakeWriteBuilder()
 
+            def new_stream_write_builder(self):
+                return FakeWriteBuilder()
+
         class FakeCatalog:
             def get_table(self, target):
                 return FakeTable()
@@ -410,6 +730,13 @@ def map_batches(self, fn, batch_format=None):
         parser_path = (
             "pypaimon.schema.data_types.PyarrowFieldParser.from_paimon_schema"
         )
+
+        def fake_apply(*args, **kwargs):
+            if incremental:
+                kwargs["on_group_result"](recorder["msgs"], 3, [])
+                return [], 3, []
+            return recorder["msgs"], 3, []
+
         with mock.patch(
                 "pypaimon.catalog.catalog_factory.CatalogFactory.create",
                 return_value=FakeCatalog()), \
@@ -425,12 +752,13 @@ def map_batches(self, fn, batch_format=None):
                                ("age", pa.int32()),
                            ])), \
                 mock.patch.object(m, "distributed_update_apply",
-                                  return_value=(recorder["msgs"], 3, [])):
+                                  side_effect=fake_apply):
             recorder["result"] = m.update_by_row_id(
                 "default.fake",
                 FakeSource(),
                 self.catalog_options,
                 update_cols=["age"],
+                max_groups_per_commit=1 if incremental else None,
             )
         return recorder