JunRuiLee commented on code in PR #10541:
URL: https://github.com/apache/paimon/pull/10541#discussion_r4258491583
##########
paimon-python/pypaimon/ray/ray_paimon.py:
##########
@@ -567,10 +569,26 @@ def write_paimon(
hash_fixed_precluster: Pre-clustering mode. ``"auto"`` follows
table options, ``"off"`` disables it, and ``"map_groups"``
explicitly enables HASH_FIXED grouping.
+ commit_mode: ``"two_phase"`` (default) has the write tasks return
Review Comment:
Could we document that rolling back the target table while an actor-mode
write is active, including retries, is unsupported? This would align with
Paimon's Flink recovery guidance: stop writers before rolling back the table,
then restore the job from the matching savepoint.
##########
paimon-python/pypaimon/write/ray_datasink.py:
##########
@@ -552,8 +701,533 @@ def _write_group(group: pa.Table) -> pa.Table:
table,
overwrite=overwrite,
static_partition=static_partition,
+ options=options,
)
coordinator.on_write_start()
return _consume_write_results(
messages, coordinator, message_col, error_col
)
+
+
+class PaimonCommitterActor:
+ """Per-write-job singleton that serializes all Paimon commits.
+
+ Created by :meth:`PaimonActorDatasink.on_write_start` on the driver with
+ ``num_cpus=0``. Write tasks RPC their commit messages here and block on
+ the ack, so commits execute one at a time (Ray actors run methods
+ serially) and intra-job optimistic-lock (CAS) contention is eliminated.
+
+ The table is built once at actor init; a fresh builder and TableCommit
+ are created per commit request. ``BatchTableCommit`` is strictly one-shot,
+ so reuse would fail outright.
+
+ **Within-job commit dedup via a stable commit_user.** The write task
+ computes a deterministic ``commit_user``
+ (``{job_id}-{write_uuid}-{task_idx}``) and sends it, with the
+ driver-pinned ``base_snapshot_id``, in the RPC args. If the actor dies
+ mid-commit, Ray re-issues the *same* actor task with the *same* args, so
+ the replayed commit carries the same ``commit_user``; the first-attempt
+ dedup scan then finds the already-landed snapshot (same ``commit_user`` +
+ ``commit_identifier`` + ``commit_kind``)
+ and skips it instead of committing the same files twice. Without this the
+ fresh builder's random commit_user would make every replay a duplicate.
+ This does not extend across jobs -- see :class:`PaimonActorDatasink`.
+
+ **The scan is skipped where it provably has nothing to find.** That scan
+ reads every snapshot from the write's fixed bound up to the current latest,
+ so as the write commits, each commit scans one more snapshot than the last:
+ over a write of T commits it costs O(T^2) snapshot reads, all serialized
+ through this one actor. But every commit this write makes goes through
here,
+ so an instance alive since before the first of them knows, without looking,
+ that a commit_user it has never handled has nothing on the table to be a
+ replay of. Only that direction is taken from memory. The reverse -- "I
+ remember committing this, so skip it" -- is deliberately *not*, because
+ memory is not an authority on the table: a rollback can remove a snapshot
+ this instance remembers writing, so what memory holds and what the table
+ holds can differ, and only one of the two decides. So an already-attempted
+ commit_user keeps going through the filesystem dedup path unchanged. Memory
+ only ever withholds work the filesystem would have found nothing in.
+ """
+
+ def __init__(
+ self,
+ catalog_options: Dict[str, str],
+ table_identifier: str,
+ options: Optional[Dict[str, str]] = None,
+ ):
+ from pypaimon.catalog.catalog_factory import CatalogFactory
+
+ catalog = CatalogFactory.create(catalog_options)
Review Comment:
Reloading the table from the catalog loses dynamic overrides already applied
to the input table through table.copy(...). The writers retain those overrides,
but the committer only receives the explicit options argument. Could we
preserve the input table's dynamic overrides when constructing the committer?
--
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]
