zhuxiangyi commented on code in PR #10105:
URL: https://github.com/apache/paimon/pull/10105#discussion_r4266531585
##########
paimon-core/src/main/java/org/apache/paimon/metastore/ChainTableOverwriteCommitCallback.java:
##########
@@ -52,52 +73,341 @@
*/
public class ChainTableOverwriteCommitCallback implements CommitCallback {
+ private static final Logger LOG =
+ LoggerFactory.getLogger(ChainTableOverwriteCommitCallback.class);
+
+ /**
+ * The latest snapshot of the snapshot branch when an overwrite of the
delta branch was
+ * published, 0 if it had none, kept in the properties of the overwrite's
snapshot.
+ */
+ public static final String SNAPSHOT_BRANCH_POSITION =
"chain-table.snapshot-branch.snapshot-id";
+
private transient FileStoreTable table;
private transient CoreOptions coreOptions;
+ private final String commitUser;
- public ChainTableOverwriteCommitCallback(FileStoreTable table) {
+ public ChainTableOverwriteCommitCallback(FileStoreTable table, String
commitUser) {
this.table = table;
this.coreOptions = table.coreOptions();
+ this.commitUser = commitUser;
+ }
+
+ /**
+ * Records where the snapshot branch stands as the overwrite is published:
what the overwrite
+ * supersedes there are the files that snapshot holds, and a {@link
#retry} of the overwrite has
+ * no other exact way to tell them from what the snapshot branch receives
afterwards, since the
+ * two branches commit independently and their clocks cannot order one
against the other.
+ *
+ * <p>A snapshot branch that cannot be read fails the overwrite before it
is published.
+ */
+ @Override
+ public void beforeOverwrite(ManifestCommittable committable) {
+ if (!ChainTableUtils.isScanFallbackDeltaBranch(coreOptions)) {
+ return;
+ }
+ Long latest = snapshotTable().snapshotManager().latestSnapshotId();
+ committable.addProperty(
+ SNAPSHOT_BRANCH_POSITION, String.valueOf(latest == null ? 0L :
latest));
}
@Override
public void call(Context context) {
+ if (!ChainTableUtils.isScanFallbackDeltaBranch(coreOptions)) {
+ return;
+ }
+ if (context.snapshot.commitKind() != CommitKind.OVERWRITE) {
+ return;
+ }
+ truncateSnapshotPartitions(context.deltaFiles);
+ }
+ /**
+ * The commit of this committable was published by an earlier attempt
whose callback may not
+ * have completed, for example because the snapshot branch was unreachable
right after the delta
+ * snapshot was written. Resolve that snapshot and redo the cleanup, which
is idempotent. The
+ * partitions are taken from the manifest changes of the snapshot rather
than from the
+ * committable, since an overwrite also clears partitions it wrote no new
file to.
+ */
+ @Override
+ public void retry(ManifestCommittable committable) {
if (!ChainTableUtils.isScanFallbackDeltaBranch(coreOptions)) {
return;
}
+ List<Snapshot> snapshots =
+ table.snapshotManager()
+ .findSnapshotsForIdentifiers(
+ commitUser,
Collections.singletonList(committable.identifier()));
+ if (snapshots.isEmpty()) {
+ LOG.warn(
+ "No snapshot of commit user {} with identifier {} in table
{}, "
+ + "cannot redo the snapshot branch cleanup of its
overwrite.",
+ commitUser,
+ committable.identifier(),
+ table.name());
+ return;
+ }
+ for (Snapshot snapshot : snapshots) {
+ if (snapshot.commitKind() != CommitKind.OVERWRITE) {
+ continue;
+ }
+ List<BinaryRow> overwritePartitions =
+ overwritePartitions(
+ table.store()
+ .newScan()
+ .withKind(ScanMode.DELTA)
+ .withSnapshot(snapshot.id())
+ .plan()
+ .files());
+ clearSnapshotFilesAsOf(overwritePartitions, snapshot);
+ }
+ }
- if (context.snapshot.commitKind() != CommitKind.OVERWRITE) {
+ /**
+ * Clear what the overwrite superseded in the given partitions of the
snapshot branch, and
+ * nothing that landed there since. That is what {@link #call} cleared at
the time; a replay
+ * that repeats it after later data arrived must not take that data with
it, whether the
+ * original cleanup had completed or not.
+ *
+ * <p>What the overwrite superseded are the files the snapshot branch held
in those partitions
+ * when the overwrite was published, and whatever compactions of the
snapshot branch have since
+ * rewritten from them alone. If a compaction has merged them with data
written after the
+ * overwrite, another commit such as a rescale has rewritten them, or the
snapshot branch no
+ * longer retains the history to tell, the cleanup cannot be done exactly,
and the retry fails
+ * rather than report a cleanup it did not do.
+ */
+ private void clearSnapshotFilesAsOf(List<BinaryRow> partitions, Snapshot
overwrite) {
+ if (partitions.isEmpty()) {
return;
}
+ String position =
+ overwrite.properties() == null
+ ? null
+ : overwrite.properties().get(SNAPSHOT_BRANCH_POSITION);
+ if (position == null) {
+ throw new IllegalStateException(
+ String.format(
+ "Cannot redo the snapshot branch cleanup of the
overwrite of partitions "
+ + "%s in table %s (snapshot %s): the
overwrite did not record "
+ + "where the snapshot branch stood when it
was published, so "
+ + "what it superseded cannot be told from
what was written "
+ + "after it. Clear the superseded rows of
these partitions "
+ + "manually.",
+ partitions, table.name(), overwrite.id()));
+ }
+ long asOfId = Long.parseLong(position);
+ if (asOfId == 0) {
+ // The snapshot branch had no snapshot when the overwrite was
published.
+ return;
+ }
+ FileStoreTable snapshotTable = snapshotTable();
+ SnapshotManager snapshotManager = snapshotTable.snapshotManager();
+ if (!snapshotManager.snapshotExists(asOfId)) {
+ throw new IllegalStateException(
+ String.format(
+ "Cannot redo the snapshot branch cleanup of the
overwrite of partitions "
+ + "%s in table %s (snapshot %s): the
snapshot branch no longer "
+ + "retains snapshot %s, where it stood
when that overwrite was "
+ + "published, so what the overwrite
superseded cannot be told "
+ + "from what was written after it. Retain
snapshots of the "
+ + "snapshot branch for longer than a job
may take to recover, "
+ + "and clear the superseded rows of these
partitions manually.",
+ partitions, table.name(), overwrite.id(), asOfId));
+ }
+ Snapshot asOf = snapshotManager.snapshot(asOfId);
Review Comment:
Reproduced with your steps
(`testChainOverwriteReplayAfterSnapshotBranchRollbackReusedTheRecordedId`:
after rolling the snapshot branch back to 1 and committing a new snapshot 2,
the replay deleted the later row), then fixed in e9a4f9e. Besides the id, the
overwrite now records the delta manifest list of that snapshot, which no other
snapshot shares, and `retry` loads the snapshot under that id and refuses to
clean up when its manifest list differs or none was recorded, so a replacement
lineage is never interpreted. A rollback that keeps the recorded snapshot still
cleans up as before
(`testChainOverwriteReplayAfterSnapshotBranchRollbackKeepingTheRecordedSnapshot`).
Both are mutation-checked.
--
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]
