JingsongLi commented on code in PR #10845:
URL: https://github.com/apache/paimon/pull/10845#discussion_r4399558648
##########
paimon-core/src/main/java/org/apache/paimon/utils/FileSystemBranchManager.java:
##########
@@ -224,6 +240,228 @@ 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.
+ Map<Long, Long> sizes = snapshotSizes(snapshotManager);
+ Map<Long, Long> branchSizes = snapshotSizes(branchSnapshotManager);
+ if (sizes.isEmpty() || branchSizes.isEmpty()) {
+ return false;
+ }
+ long latestId = Collections.max(sizes.keySet());
+ long branchLatestId = Collections.max(branchSizes.keySet());
+ if (latestId < branchEarliestId
+ || latestId > branchLatestId
+ || !sameSizes(sizes, branchSizes, branchEarliestId, latestId)
+ || !sameContent(
+ snapshotManager.snapshotPath(latestId),
+ branchSnapshotManager.snapshotPath(latestId),
+ Snapshot::fromJson)
+ || !sameSchemas(branchSchemaManager)) {
+ return false;
+ }
+
+ long latestSchemaId = Collections.max(schemaManager.listAllIds());
+ for (long schemaId : sorted(branchSchemaManager.listAllIds())) {
+ if (schemaId > latestSchemaId) {
+ copyAtomically(
+ lock,
+ branchSchemaManager.toSchemaPath(schemaId),
+ schemaManager.toSchemaPath(schemaId));
+ }
+ }
+ copySnapshots(branchSnapshotManager, latestId + 1, branchLatestId,
lock);
+ copyBranchTags(branchName, lock);
+ snapshotManager.commitLatestHint(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 of the branch's size. A copy cut short by an older
version's non-atomic
+ * replace has another size.
+ */
+ private static boolean sameSizes(
+ Map<Long, Long> sizes, Map<Long, Long> branchSizes, long from,
long to) {
+ for (long id = Math.max(from, Collections.min(sizes.keySet())); id <=
to; id++) {
+ Long size = sizes.get(id);
+ if (size == null || !size.equals(branchSizes.get(id))) {
Review Comment:
[P2] Verify intermediate snapshot content before treating main as a prefix
Equal size does not establish snapshot identity. With real writes, divergent
main and branch snapshot 3 are both 639 bytes. I executed the exact baseline
fastForward with failed quiet metadata deletions, listing snapshot 4 before
snapshot 3, and an injected failure before overwriting snapshot 3. This leaves
branch snapshot 4 on main, the divergent full-length main snapshot 3, and
matching schema files. After disabling all faults, this head's fastForward
returns success because only the latest snapshot is compared by content. A
time-travel read of main snapshot 3 still returns rows [1,2,5], while branch
snapshot 3 returns [1,2,3]. Removing main snapshot 3 to force replacement
repairs the same table. Please establish content identity for every retained
snapshot in the claimed prefix, or use another reliable lineage/integrity
check; length is insufficient for the interrupted-old-replace recovery this PR
supports.
##########
paimon-core/src/main/java/org/apache/paimon/utils/FileSystemBranchManager.java:
##########
@@ -224,6 +240,228 @@ 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.
+ Map<Long, Long> sizes = snapshotSizes(snapshotManager);
+ Map<Long, Long> branchSizes = snapshotSizes(branchSnapshotManager);
+ if (sizes.isEmpty() || branchSizes.isEmpty()) {
+ return false;
+ }
+ long latestId = Collections.max(sizes.keySet());
+ long branchLatestId = Collections.max(branchSizes.keySet());
+ if (latestId < branchEarliestId
+ || latestId > branchLatestId
+ || !sameSizes(sizes, branchSizes, branchEarliestId, latestId)
+ || !sameContent(
+ snapshotManager.snapshotPath(latestId),
+ branchSnapshotManager.snapshotPath(latestId),
+ Snapshot::fromJson)
+ || !sameSchemas(branchSchemaManager)) {
+ return false;
+ }
+
+ long latestSchemaId = Collections.max(schemaManager.listAllIds());
+ for (long schemaId : sorted(branchSchemaManager.listAllIds())) {
+ if (schemaId > latestSchemaId) {
+ copyAtomically(
+ lock,
+ branchSchemaManager.toSchemaPath(schemaId),
+ schemaManager.toSchemaPath(schemaId));
+ }
+ }
+ copySnapshots(branchSnapshotManager, latestId + 1, branchLatestId,
lock);
+ copyBranchTags(branchName, lock);
+ snapshotManager.commitLatestHint(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 of the branch's size. A copy cut short by an older
version's non-atomic
+ * replace has another size.
+ */
+ private static boolean sameSizes(
+ Map<Long, Long> sizes, Map<Long, Long> branchSizes, long from,
long to) {
+ for (long id = Math.max(from, Collections.min(sizes.keySet())); id <=
to; id++) {
+ Long size = sizes.get(id);
+ if (size == null || !size.equals(branchSizes.get(id))) {
+ 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) throws
IOException {
+ Set<Long> ids = new HashSet<>(schemaManager.listAllIds());
+ if (ids.isEmpty()) {
+ return false;
+ }
+ long latestId = Collections.max(ids);
+ Set<Long> branchIds =
+ branchSchemaManager.listAllIds().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();
+
+ List<Path> deletePaths =
+ new ArrayList<>(snapshotManager.snapshotPaths(id -> id >=
earliestSnapshotId));
+ deletePaths.addAll(
+ tagManager.tagPaths(path -> Tag.fromPath(fileIO, path).id() >=
earliestSnapshotId));
+ lock.ensureValid();
+ snapshotManager.deleteLatestHint();
+ fileIO.deleteFilesQuietly(deletePaths);
+ // 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) {
+ snapshotManager.commitEarliestHint(earliestSnapshotId);
+ }
+
+ List<Long> branchSchemaIds = sorted(branchSchemaManager.listAllIds());
+ for (long id : branchSchemaIds) {
+ replaceAtomically(
+ lock, branchSchemaManager.toSchemaPath(id),
schemaManager.toSchemaPath(id));
+ }
+ for (long id : schemaManager.listAllIds()) {
+ if (id >= earliestSchemaId && !branchSchemaIds.contains(id)) {
+ lock.ensureValid();
+ fileIO.deleteQuietly(schemaManager.toSchemaPath(id));
+ }
+ }
+
+ // List rather than trust the branch's LATEST hint, as the append path
does.
+ long branchLatestId =
Collections.max(snapshotSizes(branchSnapshotManager).keySet());
+ copySnapshots(branchSnapshotManager, earliestSnapshotId,
branchLatestId, lock);
+ copyBranchTags(branchName, lock);
+ snapshotManager.commitLatestHint(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));
+ }
+ }
+
+ private void copyBranchTags(String branchName, Lock lock) throws
IOException {
+ // Continue fast-forward even without tags.
+ Path branchTagDirectory =
tagManager.copyWithBranch(branchName).tagDirectory();
+ if (!fileIO.exists(branchTagDirectory)) {
+ return;
+ }
+ for (FileStatus status : fileIO.listStatus(branchTagDirectory)) {
+ if (!status.isDir()) {
+ Path tag = status.getPath();
+ replaceAtomically(lock, tag, new
Path(tagManager.tagDirectory(), tag.getName()));
+ }
+ }
+ }
+
+ /** Copies a file without exposing partial content, accepting a target
already copied. */
+ private void copyAtomically(Lock lock, Path source, Path target) throws
IOException {
+ // A lease may expire while many files are copied; stop before writing
past it.
+ lock.ensureValid();
+ String content = readExactly(source);
+ if ((fileIO.exists(target) || !fileIO.tryToWriteAtomic(target,
content))
Review Comment:
[P2] Revalidate the lease after source IO and before publication
The lease check occurs before readExactly(source), so a remote read can
outlast the lease or a failed heartbeat. I reproduced this through actual
table.fastForward: main has snapshot 1, a tag-derived branch has snapshot 2,
and FileIO expires the lease while opening branch snapshot 2. tryToWriteAtomic
still publishes main snapshot 2 with expired=true; fastForward only throws at a
later check. An identical control with ensureValid() immediately after the
source read leaves main snapshot 2 absent. Please recheck after potentially
blocking preparation and before the actual mutation, and cover this timing
rather than only expiry between completed writes. This is a gap in the newly
promised lease protection (the baseline did not provide locking). Apply the
same boundary discipline to hint writes and multi-file deletes.
--
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]
