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