hameizi commented on code in PR #10541:
URL: https://github.com/apache/paimon/pull/10541#discussion_r4269068017
##########
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:
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.
--
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]
