leaves12138 commented on code in PR #8826:
URL: https://github.com/apache/paimon/pull/8826#discussion_r3674303381
##########
paimon-python/pypaimon/ray/data_evolution_merge_join.py:
##########
@@ -598,14 +627,34 @@ 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:
+ 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)
Review Comment:
When `on_group_result` raises (for example, the first incremental commit
hits a deterministic `CommitConflictError`), this loop exits immediately even
though Ray may already have materialized several group results in the same
batch. Only the messages passed to the failing callback are known and aborted;
the remaining blobs are dropped, so their staged files are left orphaned. I
reproduced this with three file groups, `num_partitions=1`, and
`max_groups_per_commit=1`: forcing `FileStoreCommit.commit` to raise
`CommitConflictError` left two new Parquet files after the call returned.
Please drain and abort every uncommitted group result, or restructure the flow
so a commit failure cannot abandon buffered outputs, and add a regression test
for this case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
