hameizi opened a new pull request, #10541: URL: https://github.com/apache/paimon/pull/10541
### Purpose Ray writes currently collect all commit messages at the driver and commit once after the last task finishes, so nothing is visible until the whole write completes and the tasks contend on the optimistic lock. This adds `commit_mode='actor'` as an opt-in alternative: each write task sends its commit messages to a per-write singleton `PaimonCommitterActor` and waits for the ack. Ray runs actor methods serially, so commits are serialized instead of racing, and each task's rows become visible as it returns. Default stays `two_phase`. Notes: - The mode is append-only and rejects `overwrite`/`static_partition`, `hash_fixed_precluster='map_groups'` and POSTPONE tables up front, rather than silently falling back to coordinator commits. Primary-key tables are already rejected in `ray/shuffle.py`. - A partial write stays committed if a later task fails -- `on_write_failed` does not roll back. This is intentional for an incremental mode, and the returned `snapshot_id` is the table's latest rather than this write's own. - Retry safety: each task uses a stable `commit_user`, and the duplicate-commit scan is bounded by a snapshot id plus that snapshot file's mtime, pinned once before any task runs. When it cannot prove the commit is absent it refuses to commit instead of risking a double write. The bound is a `new_commit(dedup=...)` argument rather than a table option -- it describes one writer at one instant, so it must not be inherited through `table.options`. - `commit.native.enabled=true` bypasses this scan entirely (the native committer returns before the Python commit path), so a replayed commit there can publish the same files twice. This logs a warning; implementing the scan natively is out of scope here. Also fixes `SnapshotManager.get_latest_snapshot()`, which resolved the latest id and read its file as two non-atomic steps and so raised `FileNotFoundError` when the snapshot expired in between. It now re-resolves, retrying only `FileNotFoundError`, and fails closed rather than returning `None` (callers must not read "cannot inspect" as "table is empty"). Happy to split this into its own PR first if preferred. ### Tests `pypaimon/tests/ray_actor_commit_test.py` (new): actor-mode writes and incremental visibility, each rejection path, dynamic options reaching the writers, actor-replay and task-retry dedup, and the snapshot race interleaving. `snapshot_manager_test.py`: three tests for the relocate loop -- successful relocation, fail-closed after exhausting attempts, and non-`FileNotFoundError` propagation. `file_store_commit_test.py`: `test_commit_fail_retry_missing_snapshot_still_fails_closed` now asserts the whole scan window is walked rather than abandoned at the first unreadable id. `table_commit_test.py`: two tests for the native-commit warning -- raised when a commit requests first-attempt dedup on a native-commit table, silent otherwise. 109 tests pass locally (Ray 2.53.0, nothing skipped); `flake8 --config dev/cfg.ini pypaimon/` is clean. 🤖 Generated with [Claude Code](https://claude.com/claude-code) -- 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]
