hameizi commented on code in PR #10541:
URL: https://github.com/apache/paimon/pull/10541#discussion_r4269065376
##########
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)
+ self._table = catalog.get_table(table_identifier)
Review 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.
--
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]
