[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,11 @@ 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. It stops on concurrent commits,
+  except overwrites outside the current row-ID scope.
 
 ## Read By Row Id
 

paimon-python/pypaimon/ray/data_evolution_merge_into.py

@@ -35,6 +35,7 @@
     distributed_delete_apply,
     distributed_update_apply,
     distributed_write_collect_msgs,
+    raise_group_apply_error,
 )
 from pypaimon.ray.data_evolution_merge_transform import (
     LiteralValue,
@@ -281,14 +282,18 @@ def _build_datasets(
     # snapshot the caller observed; otherwise concurrent commits in between
     # would mix data from different snapshots.
     base_snapshot_id = base_snapshot.id if base_snapshot is not None else None
+    # An existing but empty snapshot (0 rows) is still an empty target: matching
+    # against it joins a zero-block dataset (ArrowInvalid on some Arrow versions).
+    target_is_empty = (
+        base_snapshot is None or base_snapshot.total_record_count == 0)
 
     update_ds = None
     delete_ds = None
     insert_ds = None
     update_cols_union: List[str] = []
 
     if ctx.is_self_merge:
-        if matched_specs and base_snapshot is not None:
+        if matched_specs and not target_is_empty:
             update_cols_union = _union_update_cols(matched_specs)
             if update_cols_union:
                 update_ds = build_self_merge_update_ds(
@@ -317,7 +322,7 @@ def _build_datasets(
     # Mirror Spark: matched/not-matched run as two independent joins
     # (inner / left_anti). One unified left_outer join would force
     # joined.materialize() to feed both branches, which can OOM on large merges.
-    if matched_specs and base_snapshot is not None:
+    if matched_specs and not target_is_empty:
         update_cols_union = _union_update_cols(matched_specs)
         if update_cols_union:
             update_ds = build_matched_update_ds(
@@ -362,7 +367,7 @@ def _build_datasets(
             catalog_options=ctx.catalog_options,
             num_partitions=num_partitions,
             snapshot_id=base_snapshot_id,
-            target_empty=base_snapshot is None,
+            target_empty=target_is_empty,
             ray_remote_args=ray_remote_args,
         )
 
@@ -388,7 +393,8 @@ def _execute_and_commit(
 
     try:
         if update_ds is not None:
-            update_msgs, num_updated, update_row_ids = distributed_update_apply(
+            (update_msgs, num_updated, update_row_ids,
+             update_group_error) = distributed_update_apply(
                 update_ds, table, update_cols_union,
                 num_partitions=num_partitions,
                 ray_remote_args=ray_remote_args,
@@ -398,7 +404,12 @@ def _execute_and_commit(
                 ),
                 collect_row_ids=collect_action_row_ids,
             )
+            # Keep the successful groups' files in pending_msgs so the except
+            # below aborts them: merge_into commits atomically, so a failed
+            # group must abort the whole batch rather than commit part of it.
             pending_msgs.extend(update_msgs)
+            if update_group_error is not None:
+                raise_group_apply_error(update_group_error)
 
         if delete_ds is not None:
             delete_msgs, num_deleted, delete_row_ids = distributed_delete_apply(
@@ -522,14 +533,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
 
@@ -32,6 +32,39 @@
 )
 
 
+class GroupApplyError(RuntimeError):
+    """A target file group failed while applying a distributed update. Groups
+    that succeeded are already committed/aborted by the time this is raised."""
+
+
+def _encode_group_error(exc: BaseException) -> Dict[str, str]:
+    """Worker-side exception as a text-only payload (type/message/traceback).
+
+    Never carries the instance: dumps succeeding on the worker does not mean
+    loads succeeds on the driver, and a failed unpickle there would drop the
+    groups that succeeded. A dict of strings always round-trips.
+    """
+    import traceback as _traceback
+    return {
+        "type": type(exc).__name__,
+        "message": str(exc),
+        "traceback": "".join(_traceback.format_exception(
+            type(exc), exc, exc.__traceback__)),
+    }
+
+
+def raise_group_apply_error(payload: Dict[str, Any]) -> None:
+    """Reconstruct and raise the failure carried by a ``group_error`` payload."""
+    raise GroupApplyError(
+        "update_by_row_id file group failed: "
+        "{type}: {message}\n{traceback}".format(
+            type=payload.get("type", "Exception"),
+            message=payload.get("message", ""),
+            traceback=payload.get("traceback", ""),
+        )
+    )
+
+
 def _map_kwargs(
     ray_remote_args: Optional[Dict[str, Any]],
 ) -> Dict[str, Any]:
@@ -445,7 +478,20 @@ def distributed_update_apply(
     ray_remote_args: Optional[Dict[str, Any]] = None,
     base_snapshot_id: Optional[int] = None,
     collect_row_ids: bool = False,
-) -> Tuple[list, int, list]:
+    on_group_result: Optional[Callable[[list, int, list], None]] = None,
+) -> Tuple[list, int, list, Optional[dict]]:
+    """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.
+
+    Returns ``(msgs, num_updated, row_ids, group_error)``. ``group_error`` is
+    ``None`` on success, else a payload for the first failed group (see
+    ``raise_group_apply_error``); returned rather than raised so the caller can
+    flush/abort the successful groups before surfacing it.
+    """
     import numpy as np
     import pickle
     import uuid
@@ -481,7 +527,8 @@ def distributed_update_apply(
     )
     sorted_first_row_ids = list(planner.first_row_ids)
     if not sorted_first_row_ids:
-        return [], 0, []
+        # No target file groups (e.g. empty snapshot): keep the 4-tuple contract.
+        return [], 0, [], None
 
     # Pin commit-time conflict check to the snapshot the join was built on,
     # so concurrent commits between read and planner are detected.
@@ -547,66 +594,103 @@ def _assign_frid(batch: pa.Table) -> pa.Table:
     captured_table = table
     captured_cols = cols
 
+    def _group_result(msgs_blob, n_updated, row_ids_blob, error_blob):
+        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_blob": pa.array([error_blob], type=pa.binary()),
+        })
+
     def _apply_group(group: pa.Table) -> pa.Table:
         if group.num_rows == 0:
             return pa.Table.from_pydict({
                 "msgs_blob": pa.array([], type=pa.binary()),
                 "n_updated": pa.array([], type=pa.int64()),
                 "row_ids_blob": pa.array([], type=pa.binary()),
+                "error_blob": pa.array([], type=pa.binary()),
             })
 
-        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),
-        )
-        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()
-            ),
-        })
+        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."
+                )
 
-    # One group per target data file; bounded by file count and num_partitions.
-    group_partitions = max(
-        1, min(len(captured_sorted), num_partitions)
-    )
+            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 group_error:
+            # Return the failure as data, not raise: a raised error kills the
+            # whole Ray task and discards its other (completed) groups. Strings
+            # only, so the payload always unpickles on the driver.
+            return _group_result(
+                b"", 0, b"", pickle.dumps(_encode_group_error(group_error)))
+
+        return _group_result(
+            pickle.dumps(msgs), for_update.num_rows,
+            pickle.dumps(row_ids), None)
+
+    # Bound the shuffle to num_partitions so a sparse source over a huge table
+    # does not spawn one partition per untouched file group.
+    group_partitions = max(1, min(len(captured_sorted), num_partitions))
     msgs_ds = with_frid.groupby(
         frid_col, num_partitions=group_partitions
     ).map_groups(_apply_group, **map_kwargs)
 
     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():
+    group_error_payload = None
+    iter_batches_kwargs = {"batch_format": "pyarrow"}
+    if on_group_result is not None:
+        # Deliver each group as soon as its block lands so it commits early.
+        # (Failure isolation comes from _apply_group, not from batch_size.)
+        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()
+        error_blobs = batch.column("error_blob").to_pylist()
+        row_id_blobs = (
+            batch.column("row_ids_blob").to_pylist()
+            if collect_row_ids else [None] * len(message_blobs)
+        )
+        for blob, n, err_blob, row_ids_blob in zip(
+                message_blobs, updated_counts, error_blobs, row_id_blobs):
+            if err_blob is not None:
+                # Report the first failure only after all successes are delivered.
+                if group_error_payload is None:
+                    group_error_payload = pickle.loads(err_blob)
+                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))
-    return all_msgs, num_updated, action_row_ids
+            if collect_row_ids:
+                action_row_ids.extend(group_row_ids)
+    # Failure is returned, not raised: the caller flushes/aborts the successful
+    # groups first, then surfaces it. See raise_group_apply_error.
+    return all_msgs, num_updated, action_row_ids, group_error_payload
 
 
 def _read_output_schema(table, read_cols: Sequence[str]) -> "pa.Schema":