JingsongLi commented on code in PR #10845:
URL: https://github.com/apache/paimon/pull/10845#discussion_r4491207851
##########
paimon-core/src/main/java/org/apache/paimon/utils/FileSystemBranchManager.java:
##########
@@ -224,6 +240,261 @@ public void fastForward(String branchName) {
}
}
+ /**
+ * Fast-forwards by appending when the current branch is a prefix of the
branch: its snapshots
+ * from the branch's earliest one (or its own earliest, if later) to its
latest are the
+ * branch's, and so are its schemas. Nothing is deleted and files are
written in ascending id
+ * order, so a failure leaves the current branch at one of the branch's
snapshots and a retry
+ * continues from there.
+ *
+ * @return false if the current branch is not a prefix of the branch
+ */
+ private boolean tryAppendFromBranch(String branchName, long
branchEarliestId, Lock lock)
+ throws IOException {
+ SnapshotManager branchSnapshotManager =
snapshotManager.copyWithBranch(branchName);
+ SchemaManager branchSchemaManager =
schemaManager.copyWithBranch(branchName);
+ // List rather than trust the LATEST hint: an interrupted replace by
an older version copies
+ // the branch's hint before the snapshots it names. The branch's
snapshots are listed
+ // before its schemas, and only listed snapshots and schemas are
published, so commits to
+ // the branch after this point are left for the next fast-forward.
+ Map<Long, Long> sizes = snapshotSizes(snapshotManager);
+ Map<Long, Long> branchSizes = snapshotSizes(branchSnapshotManager);
+ List<Long> branchSchemaIds = sorted(branchSchemaManager.listAllIds());
+ if (sizes.isEmpty() || branchSizes.isEmpty()) {
+ return false;
+ }
+ long latestId = Collections.max(sizes.keySet());
+ long branchLatestId = Collections.max(branchSizes.keySet());
+ if (latestId < branchEarliestId
+ || latestId > branchLatestId
+ || !samePrefix(
+ sizes, branchSizes, branchSnapshotManager,
branchEarliestId, latestId)
+ || !sameSchemas(branchSchemaManager, branchSchemaIds)) {
+ return false;
+ }
+
+ long latestSchemaId = Collections.max(schemaManager.listAllIds());
+ for (long schemaId : branchSchemaIds) {
+ if (schemaId > latestSchemaId) {
+ copyAtomically(
+ lock,
+ branchSchemaManager.toSchemaPath(schemaId),
+ schemaManager.toSchemaPath(schemaId));
+ }
+ }
+ copySnapshots(branchSnapshotManager, latestId + 1, branchLatestId,
lock);
+ copyBranchTags(branchName, branchLatestId, lock);
+ commitHint(lock, HintFileUtils.LATEST, branchLatestId);
+ return true;
+ }
+
+ private Map<Long, Long> snapshotSizes(SnapshotManager manager) throws
IOException {
+ Map<Long, Long> sizes = new HashMap<>();
+ for (FileStatus status :
+ listVersionedFileStatus(fileIO, manager.snapshotDirectory(),
SNAPSHOT_PREFIX)
+ .collect(Collectors.toList())) {
+ String version =
status.getPath().getName().substring(SNAPSHOT_PREFIX.length());
+ try {
+ sizes.put(Long.parseLong(version), status.getLen());
+ } catch (NumberFormatException ignored) {
+ // not a snapshot file, for example a temporary file of a
Python writer
+ }
+ }
+ return sizes;
+ }
+
+ /**
+ * Whether the current branch has every snapshot from {@code from} (or its
earliest, if later)
+ * to {@code to}, each with the branch's content. Sizes are compared first
because they come
+ * with the listing; equal sizes do not mean equal snapshots.
+ */
+ private boolean samePrefix(
+ Map<Long, Long> sizes,
+ Map<Long, Long> branchSizes,
+ SnapshotManager branchSnapshotManager,
+ long from,
+ long to)
+ throws IOException {
+ long start = Math.max(from, Collections.min(sizes.keySet()));
+ for (long id = start; id <= to; id++) {
+ Long size = sizes.get(id);
+ if (size == null || !size.equals(branchSizes.get(id))) {
+ return false;
+ }
+ }
+ for (long id = start; id <= to; id++) {
+ if (!sameContent(
+ snapshotManager.snapshotPath(id),
+ branchSnapshotManager.snapshotPath(id),
+ Snapshot::fromJson)) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ /** Whether the current branch's schemas are the branch's schemas up to
the current latest. */
+ private boolean sameSchemas(SchemaManager branchSchemaManager, List<Long>
branchSchemaIds)
+ throws IOException {
+ Set<Long> ids = new HashSet<>(schemaManager.listAllIds());
+ if (ids.isEmpty()) {
+ return false;
+ }
+ long latestId = Collections.max(ids);
+ Set<Long> branchIds =
+ branchSchemaIds.stream().filter(id -> id <=
latestId).collect(Collectors.toSet());
+ if (!ids.equals(branchIds)) {
+ return false;
+ }
+ for (long id : ids) {
+ if (!sameContent(
+ schemaManager.toSchemaPath(id),
+ branchSchemaManager.toSchemaPath(id),
+ TableSchema::fromJson)) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ /**
+ * Replaces the current branch's snapshots and tags from the branch's
earliest snapshot onwards,
+ * and its schemas, with the branch's. Schemas are replaced file by file
instead of deleted up
+ * front, so a failure never leaves the remaining snapshots without their
schemas.
+ */
+ private void replaceFromBranch(String branchName, long earliestSnapshotId,
Lock lock)
+ throws IOException {
+ SnapshotManager branchSnapshotManager =
snapshotManager.copyWithBranch(branchName);
+ SchemaManager branchSchemaManager =
schemaManager.copyWithBranch(branchName);
+ long earliestSchemaId =
branchSnapshotManager.snapshot(earliestSnapshotId).schemaId();
+ // Fix what is published before copying anything, as the append path
does: the branch's
+ // snapshots are listed before its schemas, so every listed snapshot
has its schema.
+ long branchLatestId =
Collections.max(snapshotSizes(branchSnapshotManager).keySet());
+ List<Long> branchSchemaIds = sorted(branchSchemaManager.listAllIds());
+
+ // Delete tags first: a retry decides from snapshots and schemas alone
whether to append,
+ // which keeps main's tags, so the tags must be gone before either
changes.
+ List<Path> deletePaths =
+ new ArrayList<>(
+ tagManager.tagPaths(
+ path -> Tag.fromPath(fileIO, path).id() >=
earliestSnapshotId));
+ deletePaths.addAll(snapshotManager.snapshotPaths(id -> id >=
earliestSnapshotId));
+ lock.ensureValid();
+ snapshotManager.deleteLatestHint();
+ for (Path path : deletePaths) {
+ delete(lock, path);
+ }
+ // Main expired its own snapshots up to its EARLIEST hint; the
branch's snapshots below
+ // that hint become main's. Move the hint before copying them, so that
a retry after a
+ // failed copy, which appends the rest, finds it already moved.
+ Long earliestHint =
+ HintFileUtils.readHint(
+ fileIO, HintFileUtils.EARLIEST,
snapshotManager.snapshotDirectory());
+ if (earliestHint != null && earliestHint > earliestSnapshotId) {
+ commitHint(lock, HintFileUtils.EARLIEST, earliestSnapshotId);
+ }
+
+ for (long id : branchSchemaIds) {
+ replaceAtomically(
+ lock, branchSchemaManager.toSchemaPath(id),
schemaManager.toSchemaPath(id));
+ }
+ for (long id : schemaManager.listAllIds()) {
+ if (id >= earliestSchemaId && !branchSchemaIds.contains(id)) {
+ delete(lock, schemaManager.toSchemaPath(id));
+ }
+ }
+
+ copySnapshots(branchSnapshotManager, earliestSnapshotId,
branchLatestId, lock);
+ copyBranchTags(branchName, branchLatestId, lock);
+ commitHint(lock, HintFileUtils.LATEST, branchLatestId);
+ }
+
+ private void copySnapshots(
+ SnapshotManager branchSnapshotManager, long fromId, long toId,
Lock lock)
+ throws IOException {
+ for (long id = fromId; id <= toId; id++) {
+ copyAtomically(
+ lock, branchSnapshotManager.snapshotPath(id),
snapshotManager.snapshotPath(id));
+ }
+ }
+
+ /** Copies the branch's tags on snapshots up to {@code latestId}. */
+ private void copyBranchTags(String branchName, long latestId, Lock lock)
throws IOException {
+ // Only tag files: a writer killed before renaming leaves a temporary
file in the directory.
+ for (Path tag :
+ tagManager
+ .copyWithBranch(branchName)
+ .tagPaths(path -> Tag.fromPath(fileIO, path).id() <=
latestId)) {
+ replaceAtomically(lock, tag, new Path(tagManager.tagDirectory(),
tag.getName()));
Review Comment:
[P2] Publish the tag content that passed the snapshot bound
The predicate validates one read of a mutable tag path, but
`replaceAtomically` reads that path again. With the real branch-ahead fixture,
I let a source writer add a column, commit, and call `replaceTag("tag2", null,
null)` after the predicate accepted snapshot 2 and while main's tag existence
was checked. On both append and replacement paths, fast-forward returns
successfully with main still at snapshot 4, but its copied tag points to the
later source snapshot; reading that tag fails with missing `schema/schema-2`.
The main commit lock does not freeze the source branch. This leaves the new
fixed tag upper-bound guarantee incomplete. Read the tag once, validate the
snapshot ID in those bytes, and publish that same content; that diagnostic
control passes both reproductions. Please cover replacement of an existing tag
between validation and publication.
--
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]
