JingsongLi commented on code in PR #10541:
URL: https://github.com/apache/paimon/pull/10541#discussion_r4261063266
##########
paimon-python/pypaimon/write/file_store_commit.py:
##########
@@ -823,53 +871,234 @@ def _write_manifest_files(self, commit_entries,
base_name):
return self.manifest_file_manager.rolling_write(
commit_entries, self.manifest_target_size, base_name)
+ def _build_snapshot_properties(
+ self,
+ snapshot_properties: Optional[Dict[str, str]]
+ ) -> Optional[Dict[str, str]]:
+ """Merge commit.metadata.* options into the snapshot's properties.
+
+ Keys under ``commit.metadata.`` are re-prefixed with ``paimon.`` and
+ written into the committed snapshot, so a caller can tag a commit with
+ its own metadata without the table schema knowing about it. The options
+ are read from ``table.options``, which already carries whatever
+ ``WriteBuilder.with_options()`` applied via ``table.copy()``.
+
+ Properties passed explicitly to ``commit()`` are the most specific
+ source and win over anything derived from options.
+ """
+ properties = {}
+ for key, value in self.table.options.options.to_map().items():
+ if key.startswith(CoreOptions.COMMIT_METADATA_PREFIX) and value is
not None:
+ properties[CoreOptions.SNAPSHOT_COMMIT_PREFIX + key] =
str(value)
+ if snapshot_properties:
+ properties.update(snapshot_properties)
+ return properties or None
+
+ def _dedup_start_after_rollback(self, end_id: int) -> Tuple[int, bool]:
+ """Scan start when the base snapshot id can no longer be trusted.
+
+ Reached when the file the id bound named is gone or has been rewritten,
+ which means the id line was rolled back and ids were reused -- our own
+ commit may now sit *below* the bound. Snapshot files are written in
+ increasing id order, so their mtimes are non-decreasing in id even
+ across a rollback; walking down from the current latest until a
snapshot
+ older than the write's start therefore finds the lowest id this write
+ could have reached.
+ """
+ base_millis = self.dedup.base_file_millis
+ for snapshot_id in range(end_id, 0, -1):
+ file_millis =
self.snapshot_manager.snapshot_file_millis(snapshot_id)
+ if file_millis is None:
+ # Cannot place this id relative to the write's start, so it
+ # cannot be ruled out as ours. Start here and let the scan
+ # report it unreadable rather than silently stepping over it.
+ return snapshot_id, False
+ if file_millis < base_millis:
+ # Written before this write started, as is everything below it.
+ return snapshot_id + 1, True
+ return 1, True # Snapshot.FIRST_SNAPSHOT_ID
+
+ def _dedup_scan_window(self, retry_result, end_id: int) -> Tuple[int,
bool]:
+ """Lowest snapshot id our own commit could occupy, and whether the
+ history reaching back that far is still intact.
+
+ The second return value is False when snapshots that could have held
our
+ commit are no longer readable. The caller must not treat that as "not a
+ duplicate".
+ """
+ if isinstance(retry_result, CommitFailRetryResult):
+ # Optimistic-lock retry: the failed attempt already saw a snapshot,
+ # so only snapshots after it can hold our (re-)committed result.
+ # That bound is this process's own observation from moments ago, so
+ # the window is intact by construction.
+ if retry_result.latest_snapshot is not None:
+ return retry_result.latest_snapshot.id + 1, True
+ return 1, True # Snapshot.FIRST_SNAPSHOT_ID
+
+ base_id = self.dedup.base_snapshot_id
+ if base_id is None:
+ return 1, True
+ if base_id == 0:
+ # The branch had no snapshot when this write started, so every
+ # snapshot on it is a candidate and none is missing from below.
+ return 1, True
+ if self.dedup.base_file_millis is None:
+ # Bound without the mtime that validates it: cannot tell a normal
id
+ # line from a rolled-back one, so scan everything.
+ return 1, base_id <= end_id
+
+ base_millis = self.snapshot_manager.snapshot_file_millis(base_id)
+ if base_millis == self.dedup.base_file_millis:
+ # The bound still names the same file, so ids above it were handed
+ # out normally and our commit, if any, is among them. Expiry trims
+ # the oldest snapshots first, so a surviving base also proves the
+ # 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:
Review 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.
##########
paimon-python/pypaimon/write/table_commit.py:
##########
@@ -56,7 +58,21 @@ def __init__(self, table, commit_user: str,
static_partition: Optional[dict]):
self._native_callbacks_registered = False
self.file_store_commit = FileStoreCommit(
snapshot_commit, table, commit_user,
- commit_callbacks=self._commit_callbacks)
+ commit_callbacks=self._commit_callbacks, dedup=dedup)
+ if (dedup is not None and dedup.on_first_attempt
+ and table.options.native_commit_enabled()):
+ # The native committer has no duplicate-commit scan, and _commit
+ # returns as soon as it publishes, so the Python dedup below is
+ # 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(
Review 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.
##########
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)
+ self._options = options
+ self._table_identifier = table_identifier
+ # commit_users whose commit this instance has started. Only ever used
to
+ # withhold the claim below -- never to answer "already committed",
which
+ # would be a memory of the table that a rollback can silently outdate.
+ # Recording an *attempt* is safe against that: it only ever sends work
+ # back to the filesystem.
+ self._attempted_users: Set[str] = set()
+ # Whether this instance has been alive since before this write's first
+ # commit, and so has seen every commit the write made. A restarted
+ # instance has not, and must ask the filesystem instead.
+ self._sees_every_commit = _is_first_actor_incarnation()
+
+ def _target_table(self):
+ """The table the commits actually go to.
+
+ ``with_options`` applies the dynamic options through ``table.copy()``,
+ and options like ``branch`` change which snapshot directory that table
+ resolves to. Anything derived for the commit -- above all the dedup
+ bound -- has to come from this table, not the un-copied one, or the
+ bound is an id on a different branch than the scan walks.
+ """
+ return self._table.copy(self._options) if self._options else
self._table
+
+ def dedup_base(self) -> Dict[str, Optional[int]]:
+ """Fixed lower bound for this write's duplicate-commit scan.
+
+ The branch's latest snapshot id, with the filesystem mtime of that
+ snapshot's file. Read on the table the commits actually go to, so the
+ bound is an id on the same branch the scan will walk.
+
+ Taken once per write and shipped to every task, so every attempt of
+ every task scans from the same place: a bound re-read per attempt moves
+ forward and lets a retry step over its own earlier commit.
+
+ The id and the mtime are two reads, and the snapshot can be expired out
+ from under the second one (another writer commits past it, retention
+ drops it). Both halves have to describe the same, still-present
+ snapshot, so an incomplete pair is re-read rather than frozen: a bound
+ whose mtime is missing can no longer tell a normal id line from a
+ rolled-back one, which degrades the scan to the whole history and makes
+ the expired ids below the bound look like missing evidence -- refusing
+ a first commit that has not even been attempted yet.
+ """
+ manager = self._target_table().snapshot_manager()
+ for attempt in range(DEDUP_ANCHOR_READ_ATTEMPTS + 1):
+ if attempt:
+ time.sleep(DEDUP_ANCHOR_READ_WAIT_SECONDS)
+ # get_latest_snapshot relocates past a candidate expired mid-read
+ # itself, so a snapshot vanishing between the id and the file does
+ # not surface here. What is left for this loop is the pair below.
+ latest = manager.get_latest_snapshot()
+ if latest is None:
+ # No snapshot to bound against. 0 rather than None: it says
"the
+ # branch started empty", which makes every snapshot a candidate
+ # and nothing missing from below -- as opposed to "no bound was
+ # taken".
+ return {"snapshot_id": 0, "file_millis": None}
+ file_millis = manager.snapshot_file_millis(latest.id)
+ if file_millis is not None:
+ return {"snapshot_id": latest.id, "file_millis": file_millis}
+ # The snapshot that read as latest is already gone. Re-reading gets
+ # the one that replaced it, which is newer and therefore still a
+ # valid lower bound: this write has not committed anything yet, so
+ # no snapshot that could be ours exists at or below it.
+ logger.info(
+ f"Dedup bound snapshot {latest.id} of {self._table_identifier}
"
+ f"disappeared while being read (attempt {attempt + 1});
re-reading"
+ )
+ raise RuntimeError(
+ f"Cannot establish a duplicate-commit bound for
{self._table_identifier}: "
+ f"the branch's latest snapshot was expired out from under every
one of "
+ f"{DEDUP_ANCHOR_READ_ATTEMPTS + 1} reads. Committing with an
incomplete "
+ f"bound would make the dedup scan unable to tell a replay from a
first "
+ f"commit. Retry when the table is not being expired this
aggressively."
+ )
+
+ def commit(
+ self,
+ commit_messages: List["CommitMessage"],
+ message: Optional[str] = None,
+ commit_user: Optional[str] = None,
+ dedup_base: Optional[Dict[str, Optional[int]]] = None,
+ ) -> None:
+ non_empty_messages = [m for m in commit_messages if not m.is_empty()]
+ if not non_empty_messages:
+ return
+
+ options = dict(self._options) if self._options else {}
+ # Enable the first-attempt duplicate scan so a replayed commit (same
+ # stable commit_user, resent after a mid-commit actor crash) is
+ # recognised and skipped rather than committed a second time.
+ #
+ # Unless this instance can already account for the commit_user: it has
+ # handled every commit of this write, and has never started one for
this
+ # commit_user, so no snapshot on the table can be a replay of it and
+ # there is nothing for the scan to find. Skipping it is what keeps the
+ # write's total dedup cost linear in its number of commits rather than
+ # quadratic. Note this gate only governs the *first* attempt -- an
+ # optimistic-lock retry inside the commit still dedups against the
+ # snapshot its failed attempt saw, whatever this is set to.
+ can_vouch = (
+ self._sees_every_commit
+ and commit_user is not None
+ and commit_user not in self._attempted_users
+ )
+ # A runtime argument to new_commit, not an option: it describes this
one
+ # commit attempt, whereas options reach the committer through
+ # table.copy() and would be inherited by every other writer built from
+ # that table.
+ dedup = CommitDedup(
+ on_first_attempt=not can_vouch,
+ base_snapshot_id=(
+ dedup_base.get("snapshot_id") if dedup_base else None),
+ base_file_millis=(
+ dedup_base.get("file_millis") if dedup_base else None),
+ )
+
+ builder = self._table.new_batch_write_builder()
+ # with_options returns a new builder; discarding it silently drops the
+ # options (commit.metadata.* would never land).
+ builder = builder.with_options(options)
+ if commit_user is not None:
+ # with_options rebuilt the builder (regenerating a random
+ # commit_user), so the stable per-task commit_user must be applied
+ # after it. This is what lets _is_duplicate_commit match a replay.
+ builder.commit_user = commit_user
+
+ # Create commit and pass message via snapshot_properties if provided
+ table_commit = builder.new_commit(dedup=dedup)
+ snapshot_properties = {}
+ if message:
+ snapshot_properties['commit.message'] = message
+
+ try:
+ if commit_user is not None:
+ # Recorded before the attempt, not after: a commit that raises
+ # may still have landed -- the retry loop gives up with
+ # "uncertain of success" after the snapshot is durable -- so a
+ # later request for this commit_user must go to the filesystem
+ # rather than be claimed as never attempted.
+ self._attempted_users.add(commit_user)
+ table_commit.commit(non_empty_messages,
snapshot_properties=snapshot_properties)
+ logger.info(
+ f"Actor committed {len(non_empty_messages)} commit messages "
+ f"for table {self._table_identifier}"
+ )
+ finally:
+ try:
+ table_commit.close()
+ except Exception as e:
+ logger.warning(f"Error closing table_commit: {e}", exc_info=e)
+
+
+@DeveloperAPI
+class PaimonActorDatasink(_DatasinkBase):
+ """Ray Datasink that serializes commits through a per-job singleton actor.
+
+ The write path is identical to the two-phase :class:`PaimonDatasink` (write
+ files + prepare_commit inside each task), but instead of collecting commit
+ messages at the driver the task sends its CommitMessages to a
+ :class:`PaimonCommitterActor` over RPC and synchronously waits for the
+ commit ack. Commits therefore happen one at a time -- no intra-job
+ optimistic-lock contention -- while each task still returns only after its
+ own data is committed, so progress/checkpoint mechanisms keyed off
+ ``write()`` returning stay ordered after the commit, and data remains
+ incrementally visible.
+
+ Trade-offs (intentional):
+
+ * **Append-only tables only.** Primary-key tables are rejected by
+ ``write_paimon_dataset``. Their sequence numbers are derived from a
+ baseline each writer reads when it is *created* and are baked into the
+ data files before ``prepare_commit``, so two parallel Ray tasks writing
+ the same bucket can assign overlapping sequence numbers regardless of
+ how the commits are ordered. Serializing commits does not address this;
+ use ``hash_fixed_precluster='map_groups'`` with
+ ``commit_mode='two_phase'``, which keeps each bucket in one task.
+ * **No overwrite support.** Constructing with ``overwrite=True`` raises
+ ``ValueError``.
+ * **No abort / rollback on failure.** Snapshots committed before a job
+ failure stay visible; ``on_write_failed`` only logs.
+ * **Within-job commit dedup via a stable per-task commit_user.** Each
+ write task derives ``commit_user = {job_id}-{write_uuid}-{task_idx}``
+ and sends it with its CommitMessages, alongside the ``base_snapshot_id``
+ pinned by ``on_write_start``. Two retry scopes are covered: if the actor
+ dies mid-commit Ray replays the same actor task with the same args, and
+ if a whole write task is retried it regenerates its data files but keeps
+ the same commit_user. In both cases the committer's first-attempt dedup
+ scan finds the already-landed snapshot and skips it, so at most one
+ file-set per task is committed; files from the discarded attempt are
+ unreferenced orphans.
+ * **No cross-job exactly-once.** ``commit_user`` embeds the Ray job id, so
+ re-running a failed job writes and commits again. Recovering from a job
+ failure is the caller's responsibility.
+ """
+
+ def __init__(
+ self,
+ table: "Table",
+ catalog_options: Dict[str, str],
+ table_identifier: str,
+ overwrite: bool = False,
+ static_partition: Optional[Dict[str, Any]] = None,
+ postpone_bucket_plan=None,
+ options: Optional[Dict[str, str]] = None,
+ ):
+ if overwrite or static_partition is not None:
+ raise ValueError(
+ "PaimonActorDatasink does not support overwrite=True or "
+ "static_partition; atomic overwrite requires a single "
+ "coordinated commit. Use the default PaimonDatasink "
+ "(two-phase mode) for overwrite."
+ )
+ self.table = table
+ self._catalog_options = catalog_options
+ self._table_identifier = table_identifier
+ self._options = options
+ self._postpone_bucket_plan = postpone_bucket_plan
+ self._table_name = table.identifier.get_full_name()
+ # Unique per datasink instance: two write_paimon calls in one job
+ # must not share an actor (they may carry different options). The
+ # per-job anonymous namespace already isolates concurrent jobs.
+ self._actor_name = (
+ f"paimon-committer-{table_identifier}-{uuid.uuid4().hex[:8]}"
+ )
+ # Unique per datasink instance (per write_paimon call): combined with
+ # job_id and task_idx it makes each write task's commit_user unique, so
+ # two write_paimon calls or two jobs never share a commit_user (which
+ # would let the dedup scan mistake one task's commit for another's).
+ self._write_uuid = uuid.uuid4().hex
+ self._actor = None
+ # Dedup scan lower bound, pinned once by on_write_start (see there for
+ # why it must not be recomputed per task attempt).
+ self._dedup_base = None
+
+ def __getstate__(self) -> dict:
+ return self.__dict__.copy()
+
+ def __setstate__(self, state: dict) -> None:
+ self.__dict__.update(state)
+ if not hasattr(self, '_table_name'):
+ self._table_name = self.table.identifier.get_full_name()
+ if not hasattr(self, '_actor'):
+ self._actor = None
+ if not hasattr(self, '_write_uuid'):
+ self._write_uuid = uuid.uuid4().hex
+ if not hasattr(self, 'static_partition'):
+ self.static_partition = None
+ if not hasattr(self, '_postpone_bucket_plan'):
+ self._postpone_bucket_plan = None
+ if not hasattr(self, '_dedup_base'):
+ self._dedup_base = None
+
+ def _target_table(self):
+ """The table the writers and committer actually operate on.
+
+ ``options`` is applied via ``table.copy()`` in every write/commit
+ builder, and options like ``branch`` change which snapshot directory
+ the result resolves to. Driver-side reads about the written data have
+ to go through here, not ``self.table``, or they report on a different
+ branch than the one being written.
+ """
+ return self.table.copy(self._options) if self._options else self.table
+
+ def on_write_start(self, schema=None) -> None:
+ # Created on the driver BEFORE the write closure is serialized (the
+ # closure is ray.put lazily at first task submission, after on_start),
+ # so the handle stored on self reaches every write task. write() also
+ # falls back to ray.get_actor for paths that serialize earlier (e.g.
+ # ActorPoolStrategy with initial_size >= 1).
+ self._actor = ray.remote(PaimonCommitterActor).options(
+ name=self._actor_name,
+ num_cpus=0,
+ max_restarts=3,
+ # If the actor dies mid-commit, Ray replays the in-flight commit on
+ # the restarted actor with the SAME args (same stable commit_user).
+ # The first-attempt dedup scan then skips the already-landed
+ # snapshot, so retrying is safe; 0 would instead fail the blocked
+ # write task and (Ray Data does not retry it) fail the whole job.
+ max_task_retries=3,
+ ).remote(
+ catalog_options=self._catalog_options,
+ table_identifier=self._table_identifier,
+ options=self._options,
+ )
+
+ # Pin the dedup scan's lower bound ONCE, before any write task runs. It
+ # must not be read inside write(): a whole-task retry would then
observe
+ # its own earlier commit as the latest snapshot and set the bound above
+ # it, so the scan range (bound, latest] would exclude that commit, the
+ # duplicate would go undetected and the task's files would be committed
+ # twice. Pinned here, every attempt of every task scans from the same
+ # pre-write snapshot and finds the replay.
+ #
+ # Read inside the actor, which holds the options-applied table: the
+ # bound has to be an id on the branch the commits go to, and the scan
+ # also needs the snapshot file's mtime to tell that id line apart from
+ # one that was rolled back and re-used the same ids.
+ self._dedup_base = ray.get(self._actor.dedup_base.remote())
+
+ logger.info(
+ f"Starting actor-commit write job for table {self._table_name} "
+ f"(committer actor: {self._actor_name})"
+ )
+
+ def write(
+ self,
+ blocks: Iterable[Block],
+ ctx: TaskContext,
+ ) -> List[Any]:
+ table_write = None
+ writer_closed = False
+ try:
+ writer_builder = (
+ self.table.new_postpone_fixed_bucket_write_builder()
+ if self._postpone_bucket_plan is not None
+ else self.table.new_batch_write_builder()
+ )
+ if self._options:
+ writer_builder = writer_builder.with_options(self._options)
+
+ if self._postpone_bucket_plan is not None:
+ writer_builder.with_bucket_plan(self._postpone_bucket_plan)
+ table_write = writer_builder.new_write()
+
+ table_schema = self.table.table_schema
+ from pypaimon.schema.data_types import PyarrowFieldParser
+ target_pa_schema =
PyarrowFieldParser.from_paimon_schema(table_schema.fields)
+
+ total_rows = 0
+ for block in blocks:
+ block_arrow: pa.Table =
BlockAccessor.for_block(block).to_arrow()
+
+ if block_arrow.num_rows == 0:
+ continue
+
+ total_rows += block_arrow.num_rows
+ block_arrow = _cast_binary_to_table_schema(block_arrow,
target_pa_schema)
+
+ table_write.write_arrow(block_arrow)
+
+ commit_messages = table_write.prepare_commit()
+ non_empty_messages = [m for m in commit_messages if not
m.is_empty()]
+
+ # Close the writer BEFORE publishing the commit. A failing close
+ # runs the writer's abort path, which deletes the very data files
+ # these commit messages reference. After a successful commit that
+ # would leave a snapshot pointing at deleted files while the task
+ # still reported success (reads then fail with FileNotFoundError).
+ # Closing first turns the same failure into a plain task failure
+ # with nothing committed, which Ray can safely retry.
+ table_write.close()
+ writer_closed = True
+
+ if not non_empty_messages:
+ logger.info(
+ f"Worker has no data to commit for table
{self._table_name}"
+ )
+ return []
+
+ commit_tag = _build_commit_tag(ctx)
+ # Stable per-task commit_user: unique across tasks/jobs, but
+ # identical when Ray replays this actor commit (same RPC args), so
+ # the committer's dedup scan can recognise a re-issued commit.
+ commit_user = f"{_current_job_id() or
'nojob'}-{self._write_uuid}-{ctx.task_idx}"
+ # Pinned by on_write_start. Deliberately NOT read here: see
+ # on_write_start for why a per-attempt read breaks dedup on a
+ # whole-task retry. None means on_write_start never ran, so leave
+ # it unset and let the committer scan from the first snapshot --
+ # slower, but still correct, since commit_user is unique to this
+ # task.
+ dedup_base = self._dedup_base
+ actor = self._actor
+ if actor is None:
+ actor = ray.get_actor(self._actor_name)
+ # Synchronous: write() returns only after this task's data is
+ # committed, keeping downstream checkpoint recording correctly
+ # ordered after the commit.
+ ray.get(actor.commit.remote(
+ non_empty_messages,
+ message=f"[{commit_tag}]",
+ commit_user=commit_user,
+ dedup_base=dedup_base,
+ ))
+
+ logger.info(
+ f"Worker {ctx.task_idx} committed {len(non_empty_messages)} "
+ f"commit messages ({total_rows} rows) for table
{self._table_name} via actor"
+ )
+ return [{"row_count": total_rows}]
+ finally:
+ # Only reached with the writer still open when we failed before
+ # the commit. Abort-on-close may delete this task's files, which
+ # is correct here because nothing was committed; swallow so the
+ # original failure is the one that propagates.
+ if table_write is not None and not writer_closed:
+ try:
+ table_write.close()
+ except Exception as e:
+ logger.warning(
+ f"Error closing table_write: {e}", exc_info=e
+ )
+
+ def _shutdown_actor(self) -> None:
+ if self._actor is None:
+ return
+ try:
+ ray.kill(self._actor)
+ except Exception as e:
+ logger.warning(
+ f"Error killing committer actor {self._actor_name}: {e}",
+ exc_info=e,
+ )
+ self._actor = None
+
+ def on_write_complete(self, write_result: Any) -> None:
+ # All commits already happened via the actor during the write tasks.
+ self._shutdown_actor()
+ if hasattr(write_result, "write_returns"):
+ write_returns = write_result.write_returns
+ elif isinstance(write_result, list):
+ write_returns = write_result
+ else:
+ write_returns = []
+
+ # Set commit_result for compatibility with write_paimon_dataset return
value
+ # Actor mode commits incrementally, so we report the final state.
+ # Read through _target_table: with options like branch, self.table
+ # resolves to a different snapshot directory than the one just written,
+ # and the reported snapshot id would be some other branch's.
+ latest_snapshot =
self._target_table().snapshot_manager().get_latest_snapshot()
+
+ if latest_snapshot:
Review 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.
##########
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:
[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.
--
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]
