[python][ray] Incrementally commit update_by_row_id file groups#8826
[python][ray] Incrementally commit update_by_row_id file groups#8826XiaoHongbo-Hope wants to merge 25 commits into
Conversation
1187b47 to
9e0ef29
Compare
9e0ef29 to
8612798
Compare
| 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. |
There was a problem hiding this comment.
batch_size=1 does not yield each group immediately; it only splits blocks after map_groups emits them. Ray buffers multiple group results within the same task, so if a later group fails, earlier completed groups never reach on_group_result and are not committed. This reproduces with num_partitions=1, three groups, and max_groups_per_commit=1: when the third group fails, the snapshot does not advance.
There was a problem hiding this comment.
batch_size=1does not yield each group immediately; it only splits blocks aftermap_groupsemits them. Ray buffers multiple group results within the same task, so if a later group fails, earlier completed groups never reachon_group_resultand are not committed. This reproduces withnum_partitions=1, three groups, andmax_groups_per_commit=1: when the third group fails, the snapshot does not advance.
Fixed as of ac42026: _apply_group returns each group’s failure as data instead of
raising, so completed groups in the same task are still committed before the error is
surfaced.
Zouxxyy
left a comment
Zouxxyy
left a comment
There was a problem hiding this comment.
Two findings from this review:
| # No target file groups (e.g. an existing but empty snapshot). Match the | ||
| # 4-tuple contract so callers that unpack (msgs, num, row_ids, err) -- | ||
| # including merge_into's NOT MATCHED insert path -- don't crash. | ||
| return [], 0, [], None |
There was a problem hiding this comment.
Returning a 4-tuple fixes the unpacking crash, but it does not make an existing-but-empty snapshot follow the empty-target path. _build_datasets still passes target_empty=base_snapshot is None; for an empty snapshot, base_snapshot is non-null, so the NOT MATCHED insert path builds a left-anti join against a zero-block target. The new test fails on Python 3.10 and 3.11 with ArrowInvalid. Please also treat base_snapshot.total_record_count == 0 as empty, and skip the matched branches, so this regression passes the supported CI matrix.
There was a problem hiding this comment.
Returning a 4-tuple fixes the unpacking crash, but it does not make an existing-but-empty snapshot follow the empty-target path.
_build_datasetsstill passestarget_empty=base_snapshot is None; for an empty snapshot,base_snapshotis non-null, so the NOT MATCHED insert path builds a left-anti join against a zero-block target. The new test fails on Python 3.10 and 3.11 withArrowInvalid. Please also treatbase_snapshot.total_record_count == 0as empty, and skip the matched branches, so this regression passes the supported CI matrix.
Fixed
| exc_info=commit_exc, | ||
| ) | ||
| clean_up_rejected_commit() | ||
| raise commit_exc |
There was a problem hiding this comment.
Once this branch classifies the rejection as deterministic, the staged CommitMessage files are safe to abort. Rethrowing the raw exception loses that signal: TableCommit._commit only aborts on CommitConflictError, while merge_into and RayDataSink stop aborting after commit starts, so 400/403/404 rejections can leave data, changelog, and index files orphaned. Please propagate a safe-to-abort rejection type or marker to the layer that owns the messages, while preserving the original cause, and add a caller-level test showing that no manual abort() is required.
There was a problem hiding this comment.
Once this branch classifies the rejection as deterministic, the staged
CommitMessagefiles are safe to abort. Rethrowing the raw exception loses that signal:TableCommit._commitonly aborts onCommitConflictError, whilemerge_intoandRayDataSinkstop aborting after commit starts, so 400/403/404 rejections can leave data, changelog, and index files orphaned. Please propagate a safe-to-abort rejection type or marker to the layer that owns the messages, while preserving the original cause, and add a caller-level test showing that no manualabort()is required.
Fixed
3 participants
Purpose
For large update_by_row_id jobs, a single final commit makes late failures expensive because all uncommitted file groups must be rewritten.
Add optional max_groups_per_commit windows while keeping the default single atomic commit. Incremental mode may retain earlier windows after a later failure.
The follow-up also bounds conflict planning by row range, keeps explicit REST rejections deterministic, and releases later uncommitted groups while draining failures.
Tests