JingsongLi commented on code in PR #10105:
URL: https://github.com/apache/paimon/pull/10105#discussion_r4253961103
##########
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:
[P1] Verify the snapshot-branch identity before replay cleanup
The recorded position is only a numeric snapshot ID, which rollback can
reuse. I reproduced this with sequential public operations: write
snapshot-branch snapshots 1 and 2; publish a delta overwrite while its
snapshot-branch cleanup fails; rollback the snapshot branch to 1; append a
genuinely later row (creating a replacement snapshot 2); then replay the
overwrite. snapshotExists(2) succeeds, and this code loads the replacement
snapshot as the original boundary, classifies its later row as superseded and
removes it. The added ChainTableFileStoreTableTest fails: after retry the
result contains delta-new but loses later. No concurrent rollback race is
required. Record the original snapshot UUID/content identity together with its
ID before publication and validate it from storage before interpreting these
files; refuse cleanup when lineage differs or cannot be established, rather
than deleting replacement-lineage data.
--
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]
