[python][ray] Incrementally commit update_by_row_id file groups

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,9 @@ 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 after every worker
+  finishes. For example, `64` commits after each 64 `_FIRST_ROW_ID` file groups.
 
 **Returns:** `{"num_updated": <rows>}`.
 
@@ -548,6 +552,10 @@ 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 commit mode overlaps later worker writes with earlier commits, but is not
+  atomic across the whole operation. If a later group fails, earlier committed windows
+  remain visible and the table is partially updated. Inspect or reconcile that state,
+  then rerun the intended update as appropriate.
 
 ## Read By Row Id
 

paimon-python/pypaimon/ray/data_evolution_merge_into.py

@@ -522,14 +522,27 @@ def _require_ray_join() -> None:
 
 def _reraise_inner(err: BaseException) -> None:
     """Unwrap Ray's RayTaskError so callers see the worker-side exception."""
+    try:
+        from ray.exceptions import RayTaskError
+    except ImportError:
+        raise err
+
+    if not isinstance(err, RayTaskError):
+        raise err
+
     inner = err
-    cause = getattr(err, "cause", None) or getattr(err, "__cause__", None)
-    while cause is not None:
+    seen = {id(err)}
+    cause = getattr(err, "cause", None)
+    while cause is not None and id(cause) not in seen:
+        seen.add(id(cause))
         inner = cause
-        cause = getattr(inner, "cause", None) or getattr(inner, "__cause__", None)
+        cause = (
+            getattr(inner, "cause", None)
+            if isinstance(inner, RayTaskError) else None
+        )
     if inner is err:
         raise err
-    raise inner from err
+    raise inner from None
 
 
 def _validate_disjoint_action_row_ids(update_row_ids, delete_row_ids) -> None:

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
 
@@ -445,7 +445,15 @@ 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]:
+    """Apply updates grouped by target file and collect their commit messages.
+
+    When ``on_group_result`` is set, it is called on the driver once for every
+    completed ``_FIRST_ROW_ID`` group. Commit messages are delivered to the
+    callback instead of being retained in the returned list, which lets callers
+    commit completed file groups incrementally while the remaining groups run.
+    """
     import numpy as np
     import pickle
     import uuid
@@ -598,14 +606,31 @@ def _apply_group(group: pa.Table) -> pa.Table:
     all_msgs: list = []
     num_updated = 0
     action_row_ids = []
-    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():
+    iter_batches_kwargs = {"batch_format": "pyarrow"}
+    if on_group_result is not None:
+        # Yield each group immediately.
+        iter_batches_kwargs["batch_size"] = 1
+        iter_batches_kwargs["prefetch_batches"] = 0
+    for batch in msgs_ds.iter_batches(**iter_batches_kwargs):
+        message_blobs = batch.column("msgs_blob").to_pylist()
+        updated_counts = batch.column("n_updated").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 in zip(
+                message_blobs, updated_counts, row_id_blobs):
+            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)
     return all_msgs, num_updated, action_row_ids