Repository navigation
Conversation
Ray writes collect every task's commit messages at the driver and commit
them once after the last task finishes. That is all-or-nothing, but no rows
are visible until the whole write completes, and the write tasks contend on
the optimistic lock when they do commit.
Add commit_mode='actor' as an opt-in alternative. Each write task sends its
commit messages to a per-write singleton PaimonCommitterActor and blocks on
the ack. Ray runs actor methods serially, so commits execute one at a time --
no intra-job CAS contention -- and each task's rows become visible as that
task returns, which also keeps any progress mechanism keyed off write()
returning ordered after the commit.
The mode is append-only. It rejects overwrite/static_partition,
hash_fixed_precluster='map_groups' and POSTPONE tables rather than silently
downgrading an actor request to two-phase semantics. Primary-key tables are
already rejected by ray/shuffle.py, and serializing commits does not change
that: their sequence numbers come from a baseline each writer reads when it
is created and are baked into the data files before prepare_commit, so two
parallel tasks writing one bucket can overlap regardless of commit order.
Two trade-offs are deliberate and are documented on the public API. A
partial write stays committed if a later task fails -- on_write_failed does
not roll back -- and the returned snapshot id is the table's latest rather
than this write's own, since the write no longer has a single commit of its
own to report.
Retry safety rests on a stable per-task commit_user
({job_id}-{write_uuid}-{task_idx}) and a duplicate-commit scan bounded by a
snapshot id plus that snapshot file's mtime, both pinned once before any task
runs. A bound re-read per attempt would move forward and let a retried task
step over its own earlier commit. This covers an actor dying mid-commit (Ray
replays the same actor task with the same args) and a whole write task being
retried. When the scan cannot establish whether the commit already landed it
refuses to commit rather than risk writing it twice. The mtime is required
because snapshot ids are reused after a rollback, and because a snapshot's
own timeMillis is stamped by whichever client committed it and so is not
monotonic across writers.
Also fix a race in SnapshotManager.get_latest_snapshot(), which is shared
read-layer code rather than specific to this mode. Resolving the latest
snapshot id and reading that id's file are two steps, and the snapshot can
be expired in between -- another writer commits past it and retention drops
it -- so the read raised FileNotFoundError at the caller even though the
snapshot that replaced it was readable. Re-resolve instead of letting the
vanished file surface, retrying only FileNotFoundError so that permission
and parse errors still propagate, and fail closed with the new
SnapshotNotResolvable rather than returning None, which callers such as the
immutable-option guard would read as "the table is empty".
Actor mode was only reachable by importing write_paimon_dataset from pypaimon.write.ray_datasink; write_paimon, the documented entry point, had no way to select it or to pass dynamic table options. Forward both kwargs and document what actor mode trades away: a partial write stays committed if a later task fails, and the returned snapshot id is the table's latest rather than this write's own.
… argument The three commit.dedup-* options were read from CoreOptions, which flow through table.copy() and are inherited by every writer built from that table. The dedup bound describes one writer at one instant, so putting it in table properties causes cross-task contamination in the actor mode: a base_snapshot pinned for task 0 would be inherited by tasks 1..N as though they had also had their bounds validated. Replace them with a CommitDedup value object passed as a keyword-only argument to new_commit(dedup=...). The object is consumed by FileStoreCommit and never written to any property store. The three CoreOptions entries and their accessors are removed. Also log a WARNING when first-attempt dedup is requested on a table with commit.native.enabled=true: the native committer returns before the Python duplicate-commit scan is reached, so a replayed commit on that path can publish the same files twice. The scan is not implemented in the native path; the warning makes the hazard visible without blocking the write.
df077b6 to
8c13c57
Compare
JingsongLi
left a comment
JingsongLi
left a comment
There was a problem hiding this comment.
Requirement fit: SUPPORTED. Implementation: FINDINGS. Incremental visibility has end-to-end value, but the retry safety issues below block production use.
Verified on this head: 109 focused Ray/snapshot/commit tests passed (Ray 2.59.0, PyArrow 19.0.1), flake8 passed, and additional real-table/real-Ray/REST-native reproductions confirmed the four findings. The supplied-table branch case was also checked and works; the options finding concerns other copied options.
| # window above it was not expired out from under us -- whatever was | ||
| # dropped below it could never have held this write's commit. | ||
| return base_id + 1, True | ||
| if base_millis is None and base_id <= end_id: |
There was a problem hiding this comment.
[P1] Treat a missing anchor as unknown history
A missing base file with latest.id >= base_id does not prove that IDs above the base were allocated without rollback: the recreated base can itself expire. I reproduced pinning base 3, rollback_to(1), task commit at reused ID 2, other commits at 3/4, then retention removing snapshot metadata 1..3 while the task files remain live in snapshot 4. A whole-task retry with fresh files and the same commit_user scans only snapshot 4 here, publishes snapshot 5, and changes rows [1,42,43,44] to [1,42,42,43,44]. The control with the regenerated anchor still present correctly deduplicates. Please use durable history identity or conservatively scan/fail closed when the anchor is absent; do not declare this candidate window intact.
There was a problem hiding this comment.
Fixed. A missing anchor now marks the dedup evidence as incomplete, even when the latest snapshot ID is at or above the pinned bound. The scan can still acknowledge a matching committed task, but if it finds no match and the evidence remains
incomplete, it raises DedupEvidenceMissing instead of authorizing another publication.
I added a regression covering the reported sequence: rollback, reuse of snapshot IDs, expiration of the recreated anchor, and a whole-task retry with fresh files. It asserts that the latest snapshot remains 4 and row 42 appears only once.
Retry and rescan authorization also validate the identity of previously observed snapshots, rather than relying solely on snapshot ID comparisons, to detect rollback followed by regrowth.
| # never reached. A caller only asks for first-attempt dedup because | ||
| # it may re-issue this commit with the same commit_user, and on the | ||
| # native path that re-issue commits the files a second time. | ||
| logger.warning( |
There was a problem hiding this comment.
[P1] Enforce dedup safety on the native actor route
Actor mode accepts commit.native.enabled=true, but the native dispatch returns before FileStoreCommit can honor CommitDedup. A warning leaves automatic actor/whole-task retry unsafe. I reproduced this on the repository REST mock server with the real Rust runtime (paimon-rust-2 head c078e7c77b1ba56c71d2501a56ac78520bc8a2ca): two actor requests for freshly prepared equivalent files, the same stable-task-0 user and the same pinned base produced snapshot 2 with two copies of row 42; Python fallback was patched to fail, so both publications used native commit. Ordinary native commit_with_identifier disables initial committed-identifier filtering. Please force a committer that honors dedup for actor mode, or reject this combination until native dedup is implemented.
There was a problem hiding this comment.
Fixed by rejecting this combination rather than leaving it as a warning or silently falling back. Both the datasink and committer actor reject effective commit.native.enabled=true during initialization, before native publication can occur.
This also covers the actor fast path that legitimately skips first-attempt dedup.
TableCommit additionally rejects native commit when first-attempt dedup is requested, as defense in depth. Regression tests cover table-level configuration, dynamic overrides, direct actor construction, and the TableCommit guard. Native
commits without the dedup requirement remain unchanged.
| from pypaimon.catalog.catalog_factory import CatalogFactory | ||
|
|
||
| catalog = CatalogFactory.create(catalog_options) | ||
| self._table = catalog.get_table(table_identifier) |
There was a problem hiding this comment.
[P2] Preserve the supplied table’s effective options in the actor
The task writer uses self.table, but this reload retains only catalog options plus the separate options= argument. Overrides already applied through table.copy(...) disappear from the committer. With a supplied table.copy({row-tracking.enabled: true, commit.metadata.source: copied}), my real Ray comparison showed two_phase assigning first_row_id=0, next_row_id=1 and paimon.commit.metadata.source=copied; actor wrote first_row_id=None, next_row_id=None and omitted that metadata. The driver still reports success. Please carry the supplied table’s cumulative effective options/resolved state to the actor so writer and committer operate with the same configuration.
There was a problem hiding this comment.
Fixed. The actor previously received only the separate options argument, so preserving branch selection alone did not address other overrides already applied through table.copy(...).
The datasink now forwards the supplied table’s cumulative dynamic overrides, merged with explicit write options, and the actor reapplies them to the catalog-loaded table. This carries copied settings such as row-tracking.enabled and
commit.metadata.source into the committer as well as the writer. Explicit write options take precedence over earlier copied overrides.
| # and the reported snapshot id would be some other branch's. | ||
| latest_snapshot = self._target_table().snapshot_manager().get_latest_snapshot() | ||
|
|
||
| if latest_snapshot: |
There was a problem hiding this comment.
[P2] Return None when no actor task committed data
The existing snapshot is enough to enter this branch even when every task returned an empty result. A real Ray write of an empty Arrow dataset to a populated table returned PaimonWriteResult(row_count=0, snapshot_id=1), although no snapshot was committed. This violates the public write_paimon_dataset/write_paimon return contract (None when no data/snapshot was committed) and reports a pre-existing snapshot as this write’s result. Track whether any task actually committed and return None for the no-commit case.
There was a problem hiding this comment.
Fixed. Completion now determines whether this write committed anything from the successful task result entries, not from the existence of a latest table snapshot. Empty tasks return no entries, so an entirely empty write returns None without
reporting a pre-existing snapshot as its result.
Added a real-Ray regression that writes an empty Arrow dataset to a populated table and asserts both that the result is None and that the existing snapshot remains unchanged.
Preserve copied dynamic options and reject native actor commits. Validate snapshot history and deduplicate before checking recovery files. Fail closed on incomplete evidence, return None for empty writes, and add regression coverage and recovery documentation.
3 participants
Purpose
Add
commit_mode='actor'as an opt-in alternative to the existing Raytwo-phase write path.
With
two_phase, write tasks return commit messages to the driver, andpublication happens after all tasks finish. With
actor, each task sendsits commit messages to a per-write singleton committer actor and waits for
acknowledgement. This serializes commits within the write and makes
successfully committed task results visible before the entire write finishes.
The default remains
two_phase.Changes
PaimonActorDatasinkandPaimonCommitterActorfor incremental,serialized commits.
table.copy(...),merged with explicit write options. Explicit options take precedence, and
both writers and the committer receive the overrides.
commit_mode='actor'with effectivecommit.native.enabled=trueduring datasink and actor initialization. Native actor dedup is not
implemented; a warning is insufficient for retry safety.
Nonewhen no task committed anything, including empty writes toan already populated table.
snapshot ID and reading its file. Retry only
FileNotFoundError, withbounded relocation attempts, and fail closed if the snapshot cannot
be resolved.
Retry and recovery
Each task has a stable, independent commit identity derived from the Ray
job, write invocation, and task index. Actor reconstruction and whole-task
retries within the same write reuse that identity.
Task indices are not a committed high watermark: tasks may finish out of
order, so committing task N does not imply that earlier tasks committed.
Recovery follows one authorization path:
A confirmed replay is acknowledged before file checks, so compaction
removing its original files does not invalidate an already committed task.
For a task that still needs publication, missing data files, sidecars,
changelog files, or added index files cause recovery to fail.
Missing snapshot metadata is not proof that a task never committed.
Transient missing metadata receives bounded retries; if no committed result
can be established and the evidence remains incomplete, recovery fails
closed instead of risking duplicate publication.
Previously observed snapshots are checked by identity, not only by ID,
so rollback followed by reuse of snapshot IDs is not treated as unchanged
history. Unknown commit outcomes preserve prepared files; recovery failures
do not automatically abort or delete them.
Scope and operational contract
writes, POSTPONE tables, and
hash_fixed_precluster='map_groups'areunsupported.
Failure handling does not roll back those commits.
table/branch's latest snapshot, potentially produced by another writer,
rather than necessarily a snapshot created by this write.
retries are active. Detected history changes fail the commit, but these
checks are not atomic fencing against concurrent history changes.
Stop all writers before rollback/reset, then restart from a clean state;
do not replay stale task receipts afterward.
introduce a cross-job checkpoint protocol or guarantee cross-job
exactly-once delivery.
Tests
Regression coverage includes:
actor replay, and whole-task retry dedup.
and commit metadata propagation.
replacement during rescans, and transient/permanent missing metadata.
and data/sidecar/changelog/index path handling.
populated tables.
Latest local review run: 334 passed, 54 skipped, 7 failed, 38 subtests.
The 7 failures are existing Lance/Vortex cases in the append-only reader
suite that raise
ModuleNotFoundErrorbecause the optional dependenciesare unavailable. Those cases lack dependency skip guards and are outside
this change's scope.
Project-configured flake8 passes.