From ad094a9e2f7c081f1a0b7dd081b4c7a4248cb9db Mon Sep 17 00:00:00 2001 From: yangjie01 Date: Tue, 27 Oct 2026 19:54:27 +0800 Subject: [PATCH 1/2] [core] Keep main's snapshots when fast_forward fails and hold the commit lock --- .../paimon/table/AbstractFileStoreTable.java | 3 +- .../paimon/table/CatalogEnvironment.java | 8 +- .../paimon/utils/FileSystemBranchManager.java | 308 ++++++++-- .../utils/FileSystemBranchManagerTest.java | 575 ++++++++++++++++++ 4 files changed, 856 insertions(+), 38 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java b/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java index 459713687e88..bdf48f318a74 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java @@ -850,7 +850,8 @@ public BranchManager branchManager() { snapshotManager(), tagManager(), schemaManager(), - new BranchMergeHandler(this::switchToBranch)); + new BranchMergeHandler(this::switchToBranch), + () -> catalogEnvironment.snapshotCommit(snapshotManager(), coreOptions(), null)); } @Override diff --git a/paimon-core/src/main/java/org/apache/paimon/table/CatalogEnvironment.java b/paimon-core/src/main/java/org/apache/paimon/table/CatalogEnvironment.java index 5e85a884ec64..b4eb8fd60aa7 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/CatalogEnvironment.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/CatalogEnvironment.java @@ -131,9 +131,13 @@ public SchemaModification schemaModification() { return SchemaModification.create(catalog, identifier); } - /** Create a publisher owning the writer's lock and its catalog resources. */ + /** + * Create a publisher owning the writer's lock and its catalog resources. Callers that only need + * the lock pass a null commit user; locks that require a writer identity, such as REST commit + * leases, reject it. + */ public SnapshotCommit snapshotCommit( - SnapshotManager snapshotManager, CoreOptions options, String commitUser) { + SnapshotManager snapshotManager, CoreOptions options, @Nullable String commitUser) { Identifier lockIdentifier = identifier == null ? null diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/FileSystemBranchManager.java b/paimon-core/src/main/java/org/apache/paimon/utils/FileSystemBranchManager.java index c84b2559cabe..9e7b94518d4e 100644 --- a/paimon-core/src/main/java/org/apache/paimon/utils/FileSystemBranchManager.java +++ b/paimon-core/src/main/java/org/apache/paimon/utils/FileSystemBranchManager.java @@ -21,10 +21,12 @@ import org.apache.paimon.CoreOptions; import org.apache.paimon.Snapshot; import org.apache.paimon.fs.FileIO; +import org.apache.paimon.fs.FileStatus; import org.apache.paimon.fs.Path; import org.apache.paimon.manifest.FileEntry; import org.apache.paimon.manifest.FileKind; import org.apache.paimon.manifest.ManifestEntry; +import org.apache.paimon.operation.Lock; import org.apache.paimon.schema.FileSystemSchemaManager; import org.apache.paimon.schema.SchemaManager; import org.apache.paimon.schema.TableSchema; @@ -35,15 +37,21 @@ import java.io.IOException; import java.util.ArrayList; -import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Set; +import java.util.function.Function; +import java.util.function.Supplier; import java.util.stream.Collectors; -import java.util.stream.Stream; +import static java.nio.charset.StandardCharsets.UTF_8; import static org.apache.paimon.utils.FileUtils.listVersionedDirectories; +import static org.apache.paimon.utils.FileUtils.listVersionedFileStatus; import static org.apache.paimon.utils.Preconditions.checkArgument; +import static org.apache.paimon.utils.SnapshotManager.SNAPSHOT_PREFIX; /** A {@link BranchManager} implementation to manage branches via file system. */ public class FileSystemBranchManager implements BranchManager { @@ -56,6 +64,7 @@ public class FileSystemBranchManager implements BranchManager { private final TagManager tagManager; private final SchemaManager schemaManager; private final BranchMergeHandler mergeHandler; + private final Supplier lockSupplier; public FileSystemBranchManager( FileIO fileIO, @@ -64,12 +73,28 @@ public FileSystemBranchManager( TagManager tagManager, SchemaManager schemaManager, BranchMergeHandler mergeHandler) { + this(fileIO, path, snapshotManager, tagManager, schemaManager, mergeHandler, Lock::empty); + } + + /** + * Create a branch manager whose fast-forward holds the lock from {@code lockSupplier}, which + * must be the lock that commits to the current branch use. Fast-forward closes it after use. + */ + public FileSystemBranchManager( + FileIO fileIO, + Path path, + SnapshotManager snapshotManager, + TagManager tagManager, + SchemaManager schemaManager, + BranchMergeHandler mergeHandler, + Supplier lockSupplier) { this.fileIO = fileIO; this.tablePath = path; this.snapshotManager = snapshotManager; this.tagManager = tagManager; this.schemaManager = schemaManager; this.mergeHandler = mergeHandler; + this.lockSupplier = lockSupplier; } /** Return the path of a branch. */ @@ -171,6 +196,28 @@ public void fastForward(String branchName) { BranchManager.fastForwardValidate(branchName, snapshotManager.branch()); checkArgument(branchExists(branchName), "Branch name '%s' doesn't exist.", branchName); + // Hold the lock that commits to the current branch use, so a fast-forward and a commit + // never interleave their snapshot writes. + Lock lock = lockSupplier.get(); + try { + lock.runWithLock( + () -> { + fastForwardWithLock(branchName, lock); + return null; + }); + } catch (RuntimeException e) { + throw e; + } catch (Exception e) { + throw new RuntimeException( + String.format("Failed to fast forward branch '%s'.", branchName), e); + } finally { + // Like a commit, do not report a finished fast-forward as failed because the lock + // could not be closed. + IOUtils.closeQuietly(lock); + } + } + + private void fastForwardWithLock(String branchName, Lock lock) { Long earliestSnapshotId = snapshotManager.copyWithBranch(branchName).earliestSnapshotId(); if (earliestSnapshotId == null) { throw new RuntimeException( @@ -178,41 +225,10 @@ public void fastForward(String branchName) { + branchName + ", because it does not have snapshot."); } - Snapshot earliestSnapshot = - snapshotManager.copyWithBranch(branchName).snapshot(earliestSnapshotId); - long earliestSchemaId = earliestSnapshot.schemaId(); try { - // Delete snapshot, schema, and tag from the main branch which occurs after - // earliestSnapshotId - List deleteSnapshotPaths = - snapshotManager.snapshotPaths(id -> id >= earliestSnapshotId); - List deleteSchemaPaths = schemaManager.schemaPaths(id -> id >= earliestSchemaId); - List deleteTagPaths = - tagManager.tagPaths( - path -> Tag.fromPath(fileIO, path).id() >= earliestSnapshotId); - - List deletePaths = - Stream.of(deleteSnapshotPaths, deleteSchemaPaths, deleteTagPaths) - .flatMap(Collection::stream) - .collect(Collectors.toList()); - - // Delete latest snapshot hint - snapshotManager.deleteLatestHint(); - - fileIO.deleteFilesQuietly(deletePaths); - fileIO.copyFiles( - snapshotManager.copyWithBranch(branchName).snapshotDirectory(), - snapshotManager.snapshotDirectory(), - true); - fileIO.copyFiles( - schemaManager.copyWithBranch(branchName).schemaDirectory(), - schemaManager.schemaDirectory(), - true); - // Continue fast-forward even without tags. - Path branchTagDirectory = tagManager.copyWithBranch(branchName).tagDirectory(); - if (fileIO.exists(branchTagDirectory)) { - fileIO.copyFiles(branchTagDirectory, tagManager.tagDirectory(), true); + if (!tryAppendFromBranch(branchName, earliestSnapshotId, lock)) { + replaceFromBranch(branchName, earliestSnapshotId, lock); } snapshotManager.invalidateCache(); } catch (IOException e) { @@ -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 sizes = snapshotSizes(snapshotManager); + Map 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 snapshotSizes(SnapshotManager manager) throws IOException { + Map 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 sizes, Map 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 ids = new HashSet<>(schemaManager.listAllIds()); + if (ids.isEmpty()) { + return false; + } + long latestId = Collections.max(ids); + Set 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 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 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)) + && !content.equals(readExactly(target))) { + throw new IOException( + String.format( + "Cannot copy %s to %s, because the target holds other content.", + source, target)); + } + } + + /** Like {@link #copyAtomically}, but first deletes a target that holds other content. */ + private void replaceAtomically(Lock lock, Path source, Path target) throws IOException { + if (fileIO.exists(target) && !readExactly(source).equals(readExactly(target))) { + lock.ensureValid(); + fileIO.delete(target, false); + } + copyAtomically(lock, source, target); + } + + /** Whether two files hold the same content, comparing parsed values when the bytes differ. */ + private boolean sameContent(Path path, Path branchPath, Function parser) + throws IOException { + String content = readExactly(path); + String branchContent = readExactly(branchPath); + if (content.equals(branchContent)) { + return true; + } + try { + return parser.apply(content).equals(parser.apply(branchContent)); + } catch (RuntimeException e) { + // A file cut short by an interrupted copy does not parse. + return false; + } + } + + /** Reads a file with its line breaks, unlike {@link FileIO#readFileUtf8}. */ + private String readExactly(Path path) throws IOException { + return new String(IOUtils.readFully(fileIO.newInputStream(path), true), UTF_8); + } + + private static List sorted(List ids) { + return ids.stream().sorted().collect(Collectors.toList()); + } + @Override public void mergeBranch(String sourceBranch, String targetBranch) { BranchManager.mergeValidate(sourceBranch, targetBranch); diff --git a/paimon-core/src/test/java/org/apache/paimon/utils/FileSystemBranchManagerTest.java b/paimon-core/src/test/java/org/apache/paimon/utils/FileSystemBranchManagerTest.java index ed4cf8d178a9..77916a3cad4b 100644 --- a/paimon-core/src/test/java/org/apache/paimon/utils/FileSystemBranchManagerTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/utils/FileSystemBranchManagerTest.java @@ -18,14 +18,25 @@ package org.apache.paimon.utils; +import org.apache.paimon.Snapshot; +import org.apache.paimon.catalog.CatalogLockContext; +import org.apache.paimon.catalog.CatalogLockFactory; +import org.apache.paimon.catalog.Identifier; import org.apache.paimon.data.GenericRow; import org.apache.paimon.fs.FileIO; import org.apache.paimon.fs.FileIOFinder; +import org.apache.paimon.fs.FileStatus; import org.apache.paimon.fs.Path; +import org.apache.paimon.fs.PositionOutputStream; +import org.apache.paimon.fs.local.LocalFileIO; +import org.apache.paimon.operation.Lock; +import org.apache.paimon.options.ExpireConfig; +import org.apache.paimon.options.Options; import org.apache.paimon.schema.FileSystemSchemaManager; import org.apache.paimon.schema.Schema; import org.apache.paimon.schema.SchemaChange; import org.apache.paimon.schema.SchemaManager; +import org.apache.paimon.table.CatalogEnvironment; import org.apache.paimon.table.FileStoreTable; import org.apache.paimon.table.FileStoreTableFactory; import org.apache.paimon.table.sink.BatchTableCommit; @@ -39,9 +50,21 @@ import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.ValueSource; +import javax.annotation.Nullable; + +import java.io.File; +import java.io.IOException; import java.nio.file.Files; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Comparator; import java.util.List; import java.util.Set; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.BooleanSupplier; import java.util.stream.Collectors; import java.util.stream.Stream; @@ -255,4 +278,556 @@ void testRenameBranchToSameName() { .isInstanceOf(RuntimeException.class) .hasMessageContaining("already exists"); } + + @Test + void testFailedFastForwardKeepsMainSnapshots() throws Exception { + InterceptingFileIO io = new InterceptingFileIO(); + FileStoreTable table = createTableWithBranchAhead(io); + + io.failSnapshotWrite(table.snapshotManager().snapshotDirectory(), 0); + assertThatThrownBy(() -> table.fastForward("b1")) + .hasStackTraceContaining("Injected write failure"); + + // Snapshot 2 was committed to main before the branch existed; a failed fast-forward must + // not take it away. + FileStoreTable reloaded = FileStoreTableFactory.create(io, tablePath); + assertThat(reloaded.snapshotManager().latestSnapshotId()).isEqualTo(2L); + assertThat(readRows(reloaded)).containsExactlyInAnyOrder("1,10,100", "2,20,200"); + + io.watch(table.snapshotManager().snapshotDirectory(), "snapshot-"); + table.fastForward("b1"); + assertThat(io.changes).containsExactly("write snapshot-3", "write snapshot-4"); + assertThat(readRows(FileStoreTableFactory.create(io, tablePath))) + .containsExactlyInAnyOrder("1,10,100", "2,20,200", "3,30,300", "4,40,400"); + } + + @Test + void testFastForwardResumesAfterPartialFailure() throws Exception { + InterceptingFileIO io = new InterceptingFileIO(); + FileStoreTable table = createTableWithBranchAhead(io); + + io.failSnapshotWrite(table.snapshotManager().snapshotDirectory(), 1); + assertThatThrownBy(() -> table.fastForward("b1")) + .hasStackTraceContaining("Injected write failure"); + + // Only the first of the branch's snapshots reached main, and main reads it consistently. + FileStoreTable reloaded = FileStoreTableFactory.create(io, tablePath); + assertThat(reloaded.snapshotManager().latestSnapshotId()).isEqualTo(3L); + assertThat(readRows(reloaded)) + .containsExactlyInAnyOrder("1,10,100", "2,20,200", "3,30,300"); + + // The retry continues where the failed run stopped instead of replacing main. + io.watch(table.snapshotManager().snapshotDirectory(), "snapshot-"); + table.fastForward("b1"); + assertThat(io.changes).containsExactly("write snapshot-4"); + reloaded = FileStoreTableFactory.create(io, tablePath); + assertThat(reloaded.snapshotManager().latestSnapshotId()).isEqualTo(4L); + assertThat(readRows(reloaded)) + .containsExactlyInAnyOrder("1,10,100", "2,20,200", "3,30,300", "4,40,400"); + } + + @Test + void testFastForwardAfterMainDivergedReplacesMain() throws Exception { + InterceptingFileIO io = new InterceptingFileIO(); + FileStoreTable table = createTableWithBranchAhead(io); + commitRow(table, 5); + + table.fastForward("b1"); + + // Main moved on after the branch was created, so fast-forward resets main to the branch + // as it always has, including main's own snapshot 3. + FileStoreTable reloaded = FileStoreTableFactory.create(io, tablePath); + assertThat(reloaded.snapshotManager().snapshot(3)) + .isEqualTo(table.switchToBranch("b1").snapshotManager().snapshot(3)); + assertThat(readRows(reloaded)) + .containsExactlyInAnyOrder("1,10,100", "2,20,200", "3,30,300", "4,40,400"); + } + + @Test + void testFailedReplaceKeepsMainSchemas() throws Exception { + InterceptingFileIO io = new InterceptingFileIO(); + FileStoreTable table = createTableWithBranchAhead(io); + commitRow(table, 5); + + io.failSnapshotWrite(table.snapshotManager().snapshotDirectory(), 0); + assertThatThrownBy(() -> table.fastForward("b1")) + .hasStackTraceContaining("Injected write failure"); + + // Replacing main dropped its snapshots from 2 on, but snapshot 1 keeps its schema. + FileStoreTable reloaded = FileStoreTableFactory.create(io, tablePath); + assertThat(reloaded.schemaManager().listAllIds()).containsExactlyInAnyOrder(0L, 1L); + assertThat(readRows(reloaded)).containsExactly("1,10,100"); + + table.fastForward("b1"); + assertThat(readRows(FileStoreTableFactory.create(io, tablePath))) + .containsExactlyInAnyOrder("1,10,100", "2,20,200", "3,30,300", "4,40,400"); + } + + @Test + void testFailedSchemaReplaceKeepsMainReadable() throws Exception { + InterceptingFileIO io = new InterceptingFileIO(); + FileStoreTable table = createTableWithBranchAhead(io); + // Both sides create a schema 2, so replacing main rewrites main's schema 2. + table.schemaManager().commitChanges(SchemaChange.addColumn("m_col", DataTypes.INT())); + table.switchToBranch("b1") + .schemaManager() + .commitChanges(SchemaChange.addColumn("b_col", DataTypes.INT())); + commitWideRow(table.switchToBranch("b1"), 5); + + io.failWrite(table.schemaManager().schemaDirectory(), "schema-", 0); + assertThatThrownBy(() -> table.fastForward("b1")) + .hasStackTraceContaining("Injected write failure"); + + // Snapshot 1, which main keeps, still has its schema. + FileStoreTable reloaded = FileStoreTableFactory.create(io, tablePath); + assertThat(reloaded.schemaManager().listAllIds()).contains(0L, 1L); + assertThat(readRows(reloaded)).containsExactly("1,10,100"); + + table.fastForward("b1"); + reloaded = FileStoreTableFactory.create(io, tablePath); + assertThat(reloaded.schema().fieldNames()).contains("b_col").doesNotContain("m_col"); + assertThat(readRows(reloaded)) + .containsExactlyInAnyOrder( + "1,10,100", "2,20,200", "3,30,300", "4,40,400", "5,50,500"); + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void testReplaceMovesEarliestHintBack(boolean failFirstCopy) throws Exception { + InterceptingFileIO io = new InterceptingFileIO(); + FileStoreTable table = createTableWithBranchAhead(io); + commitRow(table, 5); + expireAllButLatest(table); + assertThat(table.snapshotManager().earliestSnapshotId()).isEqualTo(3L); + + if (failFirstCopy) { + // Replacing main copies the branch's snapshot 2, then fails on snapshot 3. The retry + // sees main as a prefix of the branch and only appends. + io.failSnapshotWrite(table.snapshotManager().snapshotDirectory(), 1); + assertThatThrownBy(() -> table.fastForward("b1")) + .hasStackTraceContaining("Injected write failure"); + io.watch(table.snapshotManager().snapshotDirectory(), "snapshot-"); + } + table.fastForward("b1"); + if (failFirstCopy) { + assertThat(io.changes).containsExactly("write snapshot-3", "write snapshot-4"); + } + + // Main is the branch from snapshot 2 on, so 2 is its earliest snapshot again; otherwise + // expiration would never reach snapshot 2. + assertThat( + FileStoreTableFactory.create(io, tablePath) + .snapshotManager() + .earliestSnapshotId()) + .isEqualTo(2L); + } + + @ParameterizedTest + @ValueSource(strings = {"latest-missing", "gap", "truncated-latest", "truncated-middle"}) + void testFastForwardRepairsInterruptedReplace(String state) throws Exception { + InterceptingFileIO io = new InterceptingFileIO(); + FileStoreTable table = createTableWithBranchAhead(io); + SnapshotManager main = table.snapshotManager(); + SnapshotManager branch = table.switchToBranch("b1").snapshotManager(); + // What an older version's replace leaves in main's snapshot directory when it fails: the + // branch's LATEST hint copied first, then some of the branch's snapshot files in listing + // order, which is not id order once ids reach 10. Only the snapshot directory is set up. + main.commitLatestHint(4); + copySnapshot(io, branch.snapshotPath(3), main.snapshotPath(3), state, "middle", "gap"); + copySnapshot(io, branch.snapshotPath(4), main.snapshotPath(4), state, "latest", "missing"); + + table.fastForward("b1"); + + for (long id = 2; id <= 4; id++) { + assertThat(readBytes(io, main.snapshotPath(id))) + .isEqualTo(readBytes(io, branch.snapshotPath(id))); + } + assertThat(readRows(FileStoreTableFactory.create(io, tablePath))) + .containsExactlyInAnyOrder("1,10,100", "2,20,200", "3,30,300", "4,40,400"); + } + + /** + * Copy a snapshot file byte for byte, as an older replace did, unless {@code state} names it + * truncated (then write its first half) or absent. + */ + private static void copySnapshot( + FileIO io, Path source, Path target, String state, String truncatedAs, String absentAs) + throws IOException { + if (state.endsWith(absentAs)) { + return; + } + byte[] bytes = readBytes(io, source); + int length = state.equals("truncated-" + truncatedAs) ? bytes.length / 2 : bytes.length; + try (PositionOutputStream out = io.newOutputStream(target, false)) { + out.write(bytes, 0, length); + } + } + + private static byte[] readBytes(FileIO io, Path path) throws IOException { + return IOUtils.readFully(io.newInputStream(path), true); + } + + @Test + void testFastForwardCopiesBranchSchemas() throws Exception { + InterceptingFileIO io = new InterceptingFileIO(); + FileStoreTable table = createTableWithBranchAhead(io); + table.switchToBranch("b1") + .schemaManager() + .commitChanges(SchemaChange.addColumn("b_col", DataTypes.INT())); + commitWideRow(table.switchToBranch("b1"), 5); + + table.fastForward("b1"); + + FileStoreTable reloaded = FileStoreTableFactory.create(io, tablePath); + assertThat(reloaded.schema().fieldNames()).contains("b_col"); + assertThat(readRows(reloaded)) + .containsExactlyInAnyOrder( + "1,10,100", "2,20,200", "3,30,300", "4,40,400", "5,50,500"); + } + + @Test + void testFastForwardAfterMainSchemaChangeReplacesMain() throws Exception { + InterceptingFileIO io = new InterceptingFileIO(); + FileStoreTable table = createTableWithBranchAhead(io); + // Both sides create a schema with the same id after the branch was created. + table.schemaManager().commitChanges(SchemaChange.addColumn("m_col", DataTypes.INT())); + table.switchToBranch("b1") + .schemaManager() + .commitChanges(SchemaChange.addColumn("b_col", DataTypes.INT())); + commitWideRow(table.switchToBranch("b1"), 5); + + table.fastForward("b1"); + + // Main's own schema must not be kept under the id the branch's snapshots refer to. + FileStoreTable reloaded = FileStoreTableFactory.create(io, tablePath); + assertThat(reloaded.schema().fieldNames()).contains("b_col").doesNotContain("m_col"); + assertThat(readRows(reloaded)) + .containsExactlyInAnyOrder( + "1,10,100", "2,20,200", "3,30,300", "4,40,400", "5,50,500"); + } + + @Test + void testFastForwardStopsWhenLockLeaseExpires() throws Exception { + schemaManager.commitChanges(SchemaChange.setOption(BUCKET.key(), "1")); + InterceptingFileIO io = new InterceptingFileIO(); + HeldFlagLockFactory lockFactory = new HeldFlagLockFactory(); + FileStoreTable table = createTableWithLock(io, lockFactory); + commitRow(table, 1); + table.createTag("tag1", 1); + table.createBranch("b1", "tag1"); + FileStoreTable branch = table.switchToBranch("b1"); + commitRow(branch, 2); + commitRow(branch, 3); + + // The lease expires once the first of the branch's snapshots reaches main. + io.watch(table.snapshotManager().snapshotDirectory(), "snapshot-"); + io.afterWatchedWrite = () -> lockFactory.expired.set(true); + assertThatThrownBy(() -> table.fastForward("b1")).hasStackTraceContaining("Lease expired"); + + FileStoreTable reloaded = FileStoreTableFactory.create(io, tablePath); + assertThat(reloaded.snapshotManager().latestSnapshotId()).isEqualTo(2L); + assertThat(reloaded.snapshotManager().snapshotExists(3)).isFalse(); + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void testFastForwardRunsUnderCommitLock(boolean mainDiverged) throws Exception { + schemaManager.commitChanges(SchemaChange.setOption(BUCKET.key(), "1")); + InterceptingFileIO io = new InterceptingFileIO(); + HeldFlagLockFactory lockFactory = new HeldFlagLockFactory(); + FileStoreTable table = createTableWithLock(io, lockFactory); + Path snapshotDirectory = table.snapshotManager().snapshotDirectory(); + io.watch(snapshotDirectory, "snapshot-"); + io.lockHeld = lockFactory::isHeld; + + // A commit publishes its main snapshot under the lock, which shows the recording works. + commitRow(table, 1); + assertThat(io.lockHeldOnChanges).isNotEmpty().containsOnly(true); + Identifier mainCommitLock = lockFactory.lockedIdentifiers.get(0); + + table.createTag("tag1", 1); + table.createBranch("b1", "tag1"); + commitRow(table.switchToBranch("b1"), 2); + if (mainDiverged) { + commitRow(table, 3); + } + io.watch(snapshotDirectory, "snapshot-"); + lockFactory.lockedIdentifiers.clear(); + + table.fastForward("b1"); + assertThat(io.lockHeldOnChanges).isNotEmpty().containsOnly(true); + // Replacing a diverged main deletes its snapshots, which must also happen under the lock. + assertThat(io.changes.stream().anyMatch(change -> change.startsWith("delete"))) + .isEqualTo(mainDiverged); + assertThat(lockFactory.lockedIdentifiers).containsOnly(mainCommitLock); + } + + @Test + void testFastForwardCopiesSchemasInIdOrder() throws Exception { + InterceptingFileIO io = new InterceptingFileIO(); + FileStoreTable table = createTableWithBranchAhead(io); + // Schemas 2 to 10, which a name-ordered listing returns as 10, 2, 3, ... + for (int i = 2; i <= 10; i++) { + table.switchToBranch("b1") + .schemaManager() + .commitChanges(SchemaChange.updateComment("comment " + i)); + } + commitRow(table.switchToBranch("b1"), 5); + + io.watch(table.schemaManager().schemaDirectory(), "schema-"); + table.fastForward("b1"); + + // In id order, a failure part way leaves main with no missing schema id below its latest. + List expected = new ArrayList<>(); + for (int i = 2; i <= 10; i++) { + expected.add("write schema-" + i); + } + assertThat(io.changes).containsExactlyElementsOf(expected); + } + + @Test + void testFastForwardKeepsMainTagsUnchanged() throws Exception { + InterceptingFileIO io = new InterceptingFileIO(); + FileStoreTable table = createTableWithBranchAhead(io); + table.switchToBranch("b1").createTag("btag", 3); + + io.watch(table.tagManager().tagDirectory(), "tag-"); + table.fastForward("b1"); + + // tag2, which main and the branch share, is not rewritten. + assertThat(io.changes).containsExactly("write tag-btag"); + } + + @Test + void testFastForwardAppendsAfterMainExpiredSharedSnapshots() throws Exception { + InterceptingFileIO io = new InterceptingFileIO(); + FileStoreTable table = createTableWithBranchAhead(io); + table.fastForward("b1"); + expireAllButLatest(table); + assertThat(table.snapshotManager().earliestSnapshotId()).isEqualTo(4L); + commitRow(table.switchToBranch("b1"), 5); + assertThat(table.switchToBranch("b1").snapshotManager().latestSnapshot().commitKind()) + .isEqualTo(Snapshot.CommitKind.COMPACT); + + // Main no longer has the branch's snapshots 2 and 3, but it still ends at the branch's + // snapshot 4, so the next publication of the branch only appends. The fifth sorted run + // also triggers a compaction on the branch, so it has snapshots 5 and 6. + io.watch(table.snapshotManager().snapshotDirectory(), "snapshot-"); + table.fastForward("b1"); + assertThat(io.changes).containsExactly("write snapshot-5", "write snapshot-6"); + } + + /** Main has snapshots 1 and 2; branch b1 starts from snapshot 2 and adds snapshots 3 and 4. */ + private FileStoreTable createTableWithBranchAhead(FileIO io) throws Exception { + schemaManager.commitChanges(SchemaChange.setOption(BUCKET.key(), "1")); + FileStoreTable table = FileStoreTableFactory.create(io, tablePath); + commitRow(table, 1); + commitRow(table, 2); + table.createTag("tag2", 2); + table.createBranch("b1", "tag2"); + FileStoreTable branch = table.switchToBranch("b1"); + commitRow(branch, 3); + commitRow(branch, 4); + return table; + } + + private FileStoreTable createTableWithLock(FileIO io, CatalogLockFactory lockFactory) { + CatalogEnvironment env = + new CatalogEnvironment( + Identifier.create("db", "t"), + null, + null, + lockFactory, + null, + null, + false, + false); + return FileStoreTableFactory.create(io, tablePath, schemaManager.latest().get(), env); + } + + private static void expireAllButLatest(FileStoreTable table) { + table.newExpireSnapshots() + .config( + ExpireConfig.builder() + .snapshotRetainMax(1) + .snapshotRetainMin(1) + .snapshotTimeRetain(Duration.ofMillis(Long.MAX_VALUE)) + .build()) + .expire(); + } + + private static void commitRow(FileStoreTable table, int id) throws Exception { + commit(table, GenericRow.of(id, id * 10, id * 100)); + } + + /** Commit a row to a table that has one more INT column than the initial schema. */ + private static void commitWideRow(FileStoreTable table, int id) throws Exception { + commit(table, GenericRow.of(id, id * 10, id * 100, id)); + } + + private static void commit(FileStoreTable table, GenericRow row) throws Exception { + BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder(); + try (BatchTableWrite write = writeBuilder.newWrite(); + BatchTableCommit commit = writeBuilder.newCommit()) { + write.write(row); + commit.commit(write.prepareCommit()); + } + } + + private static List readRows(FileStoreTable table) throws Exception { + return getResult( + table.newRead(), + table.newReadBuilder().newScan().plan().splits(), + row -> row.getInt(0) + "," + row.getInt(1) + "," + row.getInt(2)); + } + + /** + * A {@link LocalFileIO} that lists directories in name order, as HDFS and object stores do. It + * can fail one write of a prefixed file into a directory, and records each change to the + * prefixed files of a watched directory together with whether the table lock was held. Files + * are written by copying, by renaming a temporary file, or directly; all three count as writes. + */ + private static class InterceptingFileIO extends LocalFileIO { + + @Nullable private volatile Path failingDirectory; + private volatile String failingPrefix = ""; + private final AtomicInteger writesBeforeFailure = new AtomicInteger(); + @Nullable private volatile Path watchedDirectory; + private volatile String watchedPrefix = ""; + private volatile BooleanSupplier lockHeld = () -> false; + private volatile Runnable afterWatchedWrite = () -> {}; + private final List changes = new CopyOnWriteArrayList<>(); + private final List lockHeldOnChanges = new CopyOnWriteArrayList<>(); + + void failSnapshotWrite(Path directory, int successfulWritesFirst) { + failWrite(directory, "snapshot-", successfulWritesFirst); + } + + /** + * Let the given number of writes of prefixed files into the directory succeed, then fail. + */ + void failWrite(Path directory, String prefix, int successfulWritesFirst) { + writesBeforeFailure.set(successfulWritesFirst); + failingPrefix = prefix; + failingDirectory = directory; + } + + /** Record changes to the prefixed files of the directory from now on. */ + void watch(Path directory, String prefix) { + changes.clear(); + lockHeldOnChanges.clear(); + watchedPrefix = prefix; + watchedDirectory = directory; + } + + @Override + public FileStatus[] listStatus(Path path) throws IOException { + FileStatus[] statuses = super.listStatus(path); + Arrays.sort(statuses, Comparator.comparing((FileStatus s) -> s.getPath().getName())); + return statuses; + } + + @Override + public PositionOutputStream newOutputStream(Path path, boolean overwrite) + throws IOException { + beforeWrite(path); + return super.newOutputStream(path, overwrite); + } + + @Override + public void copyFile(Path sourcePath, Path targetPath, boolean overwrite) + throws IOException { + beforeWrite(targetPath); + super.copyFile(sourcePath, targetPath, overwrite); + } + + @Override + public boolean rename(Path src, Path dst) throws IOException { + beforeWrite(dst); + boolean renamed = super.rename(src, dst); + if (renamed && isIn(dst, watchedDirectory, watchedPrefix)) { + afterWatchedWrite.run(); + } + return renamed; + } + + @Override + public boolean delete(Path path, boolean recursive) throws IOException { + if (isIn(path, watchedDirectory, watchedPrefix)) { + record("delete", path); + } + return super.delete(path, recursive); + } + + private void beforeWrite(Path path) throws IOException { + if (isIn(path, failingDirectory, failingPrefix) + && writesBeforeFailure.getAndDecrement() <= 0) { + failingDirectory = null; + throw new IOException("Injected write failure"); + } + if (isIn(path, watchedDirectory, watchedPrefix)) { + record("write", path); + } + } + + private void record(String change, Path path) { + changes.add(change + " " + path.getName()); + lockHeldOnChanges.add(lockHeld.getAsBoolean()); + } + + private static boolean isIn(Path path, @Nullable Path directory, String prefix) { + return directory != null + && path.getName().startsWith(prefix) + && new File(path.toUri().getPath()) + .getParentFile() + .equals(new File(directory.toUri().getPath())); + } + } + + /** + * Hands out locks that record which identifiers were locked and whether one of them is held. + * Once {@code expired} is set, a held lock fails its validity check like an expired lease. + */ + private static class HeldFlagLockFactory implements CatalogLockFactory { + + private final AtomicBoolean held = new AtomicBoolean(); + private final AtomicBoolean expired = new AtomicBoolean(); + private final List lockedIdentifiers = new CopyOnWriteArrayList<>(); + + @Override + public String identifier() { + return "held-flag"; + } + + @Override + public Lock createLock( + CatalogLockContext context, + Identifier identifier, + @Nullable String tableUuid, + @Nullable String commitUser, + Options tableOptions) { + lockedIdentifiers.add(identifier); + return new Lock() { + @Override + public AutoCloseable acquire() { + held.set(true); + return () -> held.set(false); + } + + @Override + public void ensureValid() { + if (expired.get()) { + throw new IllegalStateException("Lease expired"); + } + } + + @Override + public void close() {} + }; + } + + boolean isHeld() { + return held.get(); + } + } } From 02d7d676dc83dd477a957afff231b19ea5fb4517 Mon Sep 17 00:00:00 2001 From: yangjie01 Date: Wed, 28 Oct 2026 20:05:29 +0800 Subject: [PATCH 2/2] [core] Check the lease after reading, fail on undeleted files and compare the whole prefix --- .../paimon/utils/FileSystemBranchManager.java | 65 ++++--- .../utils/FileSystemBranchManagerTest.java | 161 +++++++++++++++++- 2 files changed, 204 insertions(+), 22 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/FileSystemBranchManager.java b/paimon-core/src/main/java/org/apache/paimon/utils/FileSystemBranchManager.java index 9e7b94518d4e..3a038cf475d1 100644 --- a/paimon-core/src/main/java/org/apache/paimon/utils/FileSystemBranchManager.java +++ b/paimon-core/src/main/java/org/apache/paimon/utils/FileSystemBranchManager.java @@ -264,11 +264,8 @@ private boolean tryAppendFromBranch(String branchName, long branchEarliestId, Lo 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) + || !samePrefix( + sizes, branchSizes, branchSnapshotManager, branchEarliestId, latestId) || !sameSchemas(branchSchemaManager)) { return false; } @@ -284,6 +281,7 @@ private boolean tryAppendFromBranch(String branchName, long branchEarliestId, Lo } copySnapshots(branchSnapshotManager, latestId + 1, branchLatestId, lock); copyBranchTags(branchName, lock); + lock.ensureValid(); snapshotManager.commitLatestHint(branchLatestId); return true; } @@ -305,17 +303,31 @@ private Map snapshotSizes(SnapshotManager manager) throws IOExceptio /** * 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. + * 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 static boolean sameSizes( - Map sizes, Map branchSizes, long from, long to) { - for (long id = Math.max(from, Collections.min(sizes.keySet())); id <= to; id++) { + private boolean samePrefix( + Map sizes, + Map 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; } @@ -355,13 +367,18 @@ private void replaceFromBranch(String branchName, long earliestSnapshotId, Lock SchemaManager branchSchemaManager = schemaManager.copyWithBranch(branchName); long earliestSchemaId = branchSnapshotManager.snapshot(earliestSnapshotId).schemaId(); + // 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 deletePaths = - new ArrayList<>(snapshotManager.snapshotPaths(id -> id >= earliestSnapshotId)); - deletePaths.addAll( - tagManager.tagPaths(path -> Tag.fromPath(fileIO, path).id() >= earliestSnapshotId)); + new ArrayList<>( + tagManager.tagPaths( + path -> Tag.fromPath(fileIO, path).id() >= earliestSnapshotId)); + deletePaths.addAll(snapshotManager.snapshotPaths(id -> id >= earliestSnapshotId)); lock.ensureValid(); snapshotManager.deleteLatestHint(); - fileIO.deleteFilesQuietly(deletePaths); + 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. @@ -369,6 +386,7 @@ private void replaceFromBranch(String branchName, long earliestSnapshotId, Lock HintFileUtils.readHint( fileIO, HintFileUtils.EARLIEST, snapshotManager.snapshotDirectory()); if (earliestHint != null && earliestHint > earliestSnapshotId) { + lock.ensureValid(); snapshotManager.commitEarliestHint(earliestSnapshotId); } @@ -379,8 +397,7 @@ private void replaceFromBranch(String branchName, long earliestSnapshotId, Lock } for (long id : schemaManager.listAllIds()) { if (id >= earliestSchemaId && !branchSchemaIds.contains(id)) { - lock.ensureValid(); - fileIO.deleteQuietly(schemaManager.toSchemaPath(id)); + delete(lock, schemaManager.toSchemaPath(id)); } } @@ -388,6 +405,7 @@ private void replaceFromBranch(String branchName, long earliestSnapshotId, Lock long branchLatestId = Collections.max(snapshotSizes(branchSnapshotManager).keySet()); copySnapshots(branchSnapshotManager, earliestSnapshotId, branchLatestId, lock); copyBranchTags(branchName, lock); + lock.ensureValid(); snapshotManager.commitLatestHint(branchLatestId); } @@ -416,9 +434,9 @@ private void copyBranchTags(String branchName, Lock lock) throws IOException { /** 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); + // Reading the source may outlast the lease, so check it right before writing. + lock.ensureValid(); if ((fileIO.exists(target) || !fileIO.tryToWriteAtomic(target, content)) && !content.equals(readExactly(target))) { throw new IOException( @@ -431,12 +449,19 @@ private void copyAtomically(Lock lock, Path source, Path target) throws IOExcept /** Like {@link #copyAtomically}, but first deletes a target that holds other content. */ private void replaceAtomically(Lock lock, Path source, Path target) throws IOException { if (fileIO.exists(target) && !readExactly(source).equals(readExactly(target))) { - lock.ensureValid(); - fileIO.delete(target, false); + delete(lock, target); } copyAtomically(lock, source, target); } + /** Deletes a file while the lease is valid, failing if it remains. */ + private void delete(Lock lock, Path path) throws IOException { + lock.ensureValid(); + if (!fileIO.delete(path, false) && fileIO.exists(path)) { + throw new IOException("Failed to delete " + path); + } + } + /** Whether two files hold the same content, comparing parsed values when the bytes differ. */ private boolean sameContent(Path path, Path branchPath, Function parser) throws IOException { diff --git a/paimon-core/src/test/java/org/apache/paimon/utils/FileSystemBranchManagerTest.java b/paimon-core/src/test/java/org/apache/paimon/utils/FileSystemBranchManagerTest.java index 77916a3cad4b..45872f20e1ca 100644 --- a/paimon-core/src/test/java/org/apache/paimon/utils/FileSystemBranchManagerTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/utils/FileSystemBranchManagerTest.java @@ -18,6 +18,7 @@ package org.apache.paimon.utils; +import org.apache.paimon.CoreOptions; import org.apache.paimon.Snapshot; import org.apache.paimon.catalog.CatalogLockContext; import org.apache.paimon.catalog.CatalogLockFactory; @@ -28,6 +29,7 @@ import org.apache.paimon.fs.FileStatus; import org.apache.paimon.fs.Path; import org.apache.paimon.fs.PositionOutputStream; +import org.apache.paimon.fs.SeekableInputStream; import org.apache.paimon.fs.local.LocalFileIO; import org.apache.paimon.operation.Lock; import org.apache.paimon.options.ExpireConfig; @@ -529,6 +531,123 @@ void testFastForwardStopsWhenLockLeaseExpires() throws Exception { assertThat(reloaded.snapshotManager().snapshotExists(3)).isFalse(); } + @Test + void testFastForwardRechecksLeaseAfterReadingSource() throws Exception { + schemaManager.commitChanges(SchemaChange.setOption(BUCKET.key(), "1")); + InterceptingFileIO io = new InterceptingFileIO(); + HeldFlagLockFactory lockFactory = new HeldFlagLockFactory(); + FileStoreTable table = createTableWithLock(io, lockFactory); + commitRow(table, 1); + table.createTag("tag1", 1); + table.createBranch("b1", "tag1"); + FileStoreTable branch = table.switchToBranch("b1"); + commitRow(branch, 2); + + // The lease expires while the branch's snapshot 2 is read, before main gets a copy. + io.onOpen(branch.snapshotManager().snapshotPath(2), () -> lockFactory.expired.set(true)); + assertThatThrownBy(() -> table.fastForward("b1")).hasStackTraceContaining("Lease expired"); + + assertThat(io.exists(table.snapshotManager().snapshotPath(2))).isFalse(); + } + + @Test + void testReplaceStopsDeletingWhenLeaseExpires() throws Exception { + schemaManager.commitChanges(SchemaChange.setOption(BUCKET.key(), "1")); + InterceptingFileIO io = new InterceptingFileIO(); + HeldFlagLockFactory lockFactory = new HeldFlagLockFactory(); + FileStoreTable table = createTableWithLock(io, lockFactory); + commitRow(table, 1); + table.createTag("tag1", 1); + table.createBranch("b1", "tag1"); + commitRow(table.switchToBranch("b1"), 2); + commitRow(table, 2); + commitRow(table, 3); + + // Replacing the diverged main deletes its snapshots 1 to 3; the lease expires after the + // first delete. + Path snapshotDirectory = table.snapshotManager().snapshotDirectory(); + io.watch(snapshotDirectory, "snapshot-"); + io.afterWatchedDelete = () -> lockFactory.expired.set(true); + assertThatThrownBy(() -> table.fastForward("b1")).hasStackTraceContaining("Lease expired"); + + assertThat(io.changes).containsExactly("delete snapshot-1"); + } + + @Test + void testReplaceDeletesTagsBeforeSnapshots() throws Exception { + schemaManager.commitChanges( + SchemaChange.setOption(BUCKET.key(), "1"), + SchemaChange.setOption(CoreOptions.WRITE_ONLY.key(), "true")); + InterceptingFileIO io = new InterceptingFileIO(); + FileStoreTable table = FileStoreTableFactory.create(io, tablePath); + for (int i = 1; i <= 9; i++) { + commitRow(table, i); + } + table.createTag("tag9", 9); + table.createBranch("b1", "tag9"); + commitRow(table.switchToBranch("b1"), 10); + // Main's own snapshots 10 and 11, with a tag on 11. + commitRow(table, 20); + commitRow(table, 21); + table.createTag("mtag", 11); + + // Replacing main deletes snapshot-10 and snapshot-11 before snapshot-9 in name order, and + // fails on snapshot-9. Main is then a prefix of the branch, so the retry only appends. + io.failDeleteOf(table.snapshotManager().snapshotPath(9)); + assertThatThrownBy(() -> table.fastForward("b1")) + .hasStackTraceContaining("Failed to delete"); + assertThat(table.snapshotManager().snapshotExists(9)).isTrue(); + assertThat(table.snapshotManager().snapshotExists(10)).isFalse(); + + table.fastForward("b1"); + assertThat(table.tagManager().tagExists("mtag")).isFalse(); + assertThat(table.tagManager().tagExists("tag9")).isTrue(); + } + + @Test + void testReplaceFailsWhenSnapshotCannotBeDeleted() throws Exception { + InterceptingFileIO io = new InterceptingFileIO(); + FileStoreTable table = createTableWithBranchAhead(io); + for (int i = 5; i <= 7; i++) { + commitRow(table, i); + } + SnapshotManager main = table.snapshotManager(); + long mainLatest = main.latestSnapshotId(); + assertThat(mainLatest).isGreaterThan(4L); + + // Main's own latest snapshot is above all of the branch's. If it stayed, main would go back + // to it once commits fill the gap below it. + io.failDeleteOf(main.snapshotPath(mainLatest)); + assertThatThrownBy(() -> table.fastForward("b1")) + .hasStackTraceContaining("Failed to delete"); + + table.fastForward("b1"); + FileStoreTable reloaded = FileStoreTableFactory.create(io, tablePath); + assertThat(reloaded.snapshotManager().latestSnapshotId()).isEqualTo(4L); + assertThat(readRows(reloaded)) + .containsExactlyInAnyOrder("1,10,100", "2,20,200", "3,30,300", "4,40,400"); + } + + @Test + void testFastForwardReplacesDivergentSnapshotOfSameLength() throws Exception { + InterceptingFileIO io = new InterceptingFileIO(); + FileStoreTable table = createTableWithBranchAhead(io); + SnapshotManager main = table.snapshotManager(); + SnapshotManager branch = table.switchToBranch("b1").snapshotManager(); + // What an older version's replace leaves when it failed to delete main's own snapshot 3, + // copied the branch's snapshot 4 and then stopped before snapshot 3. + commitRow(table, 5); + io.copyFile(branch.snapshotPath(4), main.snapshotPath(4), false); + main.commitLatestHint(4); + assertThat(io.getFileSize(main.snapshotPath(3))) + .isEqualTo(io.getFileSize(branch.snapshotPath(3))); + + table.fastForward("b1"); + + assertThat(readBytes(io, main.snapshotPath(3))) + .isEqualTo(readBytes(io, branch.snapshotPath(3))); + } + @ParameterizedTest @ValueSource(booleans = {false, true}) void testFastForwardRunsUnderCommitLock(boolean mainDiverged) throws Exception { @@ -696,9 +815,24 @@ private static class InterceptingFileIO extends LocalFileIO { private volatile String watchedPrefix = ""; private volatile BooleanSupplier lockHeld = () -> false; private volatile Runnable afterWatchedWrite = () -> {}; + private volatile Runnable afterWatchedDelete = () -> {}; + @Nullable private volatile Path openTrigger; + private volatile Runnable onOpenAction = () -> {}; + @Nullable private volatile Path undeletable; private final List changes = new CopyOnWriteArrayList<>(); private final List lockHeldOnChanges = new CopyOnWriteArrayList<>(); + /** Run the action once, when the file is next opened for reading. */ + void onOpen(Path path, Runnable action) { + onOpenAction = action; + openTrigger = path; + } + + /** Make the next delete of the file report failure and leave it in place. */ + void failDeleteOf(Path path) { + undeletable = path; + } + void failSnapshotWrite(Path directory, int successfulWritesFirst) { failWrite(directory, "snapshot-", successfulWritesFirst); } @@ -751,12 +885,35 @@ public boolean rename(Path src, Path dst) throws IOException { return renamed; } + @Override + public SeekableInputStream newInputStream(Path path) throws IOException { + if (sameFile(path, openTrigger)) { + openTrigger = null; + onOpenAction.run(); + } + return super.newInputStream(path); + } + @Override public boolean delete(Path path, boolean recursive) throws IOException { - if (isIn(path, watchedDirectory, watchedPrefix)) { + if (sameFile(path, undeletable)) { + undeletable = null; + return false; + } + boolean watched = isIn(path, watchedDirectory, watchedPrefix); + if (watched) { record("delete", path); } - return super.delete(path, recursive); + boolean deleted = super.delete(path, recursive); + if (watched) { + afterWatchedDelete.run(); + } + return deleted; + } + + private static boolean sameFile(Path path, @Nullable Path other) { + return other != null + && new File(path.toUri().getPath()).equals(new File(other.toUri().getPath())); } private void beforeWrite(Path path) throws IOException {