diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/RenamingSnapshotCommit.java b/paimon-core/src/main/java/org/apache/paimon/catalog/RenamingSnapshotCommit.java index 16d577ecc7e4..77988139faf4 100644 --- a/paimon-core/src/main/java/org/apache/paimon/catalog/RenamingSnapshotCommit.java +++ b/paimon-core/src/main/java/org/apache/paimon/catalog/RenamingSnapshotCommit.java @@ -94,6 +94,11 @@ protected boolean commitImpl( }; // fs.rename may not return false if the target file already exists, or even be atomic. // The bound lock lets us check existence before publishing. - return !fileIO.exists(newSnapshotPath) && callable.call(); + if (fileIO.exists(newSnapshotPath)) { + return false; + } + // Checking existence may outlast the lease, so check it again before writing. + ensureValid(); + return callable.call(); } } diff --git a/paimon-core/src/test/java/org/apache/paimon/catalog/RenamingSnapshotCommitTest.java b/paimon-core/src/test/java/org/apache/paimon/catalog/RenamingSnapshotCommitTest.java index d7213ff08811..42919ac0a3a7 100644 --- a/paimon-core/src/test/java/org/apache/paimon/catalog/RenamingSnapshotCommitTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/catalog/RenamingSnapshotCommitTest.java @@ -19,21 +19,43 @@ package org.apache.paimon.catalog; import org.apache.paimon.Snapshot; +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.data.InternalRow; import org.apache.paimon.fs.FileIO; import org.apache.paimon.fs.Path; import org.apache.paimon.fs.local.LocalFileIO; import org.apache.paimon.operation.Lock; +import org.apache.paimon.options.Options; +import org.apache.paimon.reader.RecordReader; +import org.apache.paimon.schema.FileSystemSchemaManager; +import org.apache.paimon.schema.Schema; +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; +import org.apache.paimon.table.sink.BatchTableWrite; +import org.apache.paimon.table.sink.BatchWriteBuilder; +import org.apache.paimon.table.source.ReadBuilder; +import org.apache.paimon.types.DataTypes; import org.apache.paimon.utils.HintFileUtils; import org.apache.paimon.utils.SnapshotManager; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; +import javax.annotation.Nullable; + import java.io.IOException; import java.nio.file.Files; +import java.nio.file.StandardCopyOption; +import java.util.ArrayList; import java.util.Collections; +import java.util.List; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.junit.jupiter.api.Assertions.assertThrows; /** @@ -146,6 +168,129 @@ public void testCommitRenameThrowsAndTargetMissing(@TempDir java.nio.file.Path t assertThat(fileIO.exists(snapshotManager.snapshotPath(snapshot.id()))).isFalse(); } + /** + * The lease of the commit lock expires while the commit checks whether the snapshot file + * already exists. The commit must not publish the snapshot after that. + */ + @Test + public void testCommitRechecksLeaseAfterCheckingSnapshot(@TempDir java.nio.file.Path tmp) + throws Exception { + AtomicBoolean expired = new AtomicBoolean(); + Path tablePath = new Path(tmp.toUri()); + SnapshotManager snapshotManager = + new SnapshotManager(new LocalFileIO(), tablePath, null, null, null); + Path snapshotPath = snapshotManager.snapshotPath(5L); + FileIO fileIO = + new LocalFileIO() { + @Override + public boolean exists(Path path) throws IOException { + boolean exists = super.exists(path); + if (path.equals(snapshotPath)) { + expired.set(true); + } + return exists; + } + }; + snapshotManager = new SnapshotManager(fileIO, tablePath, null, null, null); + Lock lock = + new Lock() { + @Override + public AutoCloseable acquire() { + return () -> {}; + } + + @Override + public void ensureValid() { + if (expired.get()) { + throw new IllegalStateException("Lease expired"); + } + } + + @Override + public void close() {} + }; + + RenamingSnapshotCommit commit = new RenamingSnapshotCommit(snapshotManager, lock); + Snapshot snapshot = createSnapshot(5L); + // thrown by the check in commitImpl, not wrapped like a failure to acquire the lock + assertThatThrownBy(() -> commit.commit(null, snapshot, "main", Collections.emptyList())) + .isInstanceOf(IllegalStateException.class) + .hasMessage("Lease expired"); + + assertThat(new LocalFileIO().exists(snapshotPath)).isFalse(); + } + + /** + * Two writers share a lease lock on a file system whose rename overwrites the target, as some + * object stores do. Writer A's lease expires while it checks whether snapshot 2 exists, and + * writer B, which gets the lock then, commits snapshot 2. Writer A must not overwrite it. + */ + @Test + public void testExpiredLeaseDoesNotOverwriteAnotherWritersSnapshot( + @TempDir java.nio.file.Path tmp) throws Exception { + Path tablePath = new Path(tmp.toUri()); + OverwritingRenameLocalFileIO fileIO = new OverwritingRenameLocalFileIO(); + new FileSystemSchemaManager(fileIO, tablePath) + .createTable(Schema.newBuilder().column("id", DataTypes.INT()).build()); + LeaseLockFactory locks = new LeaseLockFactory(); + FileStoreTable writerA = createTable(fileIO, tablePath, locks); + FileStoreTable writerB = createTable(fileIO, tablePath, locks); + commitRow(writerA, 1); + + fileIO.onCommitExists( + writerA.snapshotManager().snapshotPath(2), + () -> { + locks.expireHolder(); + Thread writer = new Thread(() -> commitRow(writerB, 2)); + writer.start(); + writer.join(); + }); + commitRow(writerA, 3); + + assertThat(readIds(FileStoreTableFactory.create(fileIO, tablePath))) + .containsExactlyInAnyOrder(1, 2, 3); + } + + private static FileStoreTable createTable( + FileIO fileIO, Path tablePath, CatalogLockFactory lockFactory) { + CatalogEnvironment env = + new CatalogEnvironment( + Identifier.create("db", "t"), + null, + null, + lockFactory, + null, + null, + false, + false); + return FileStoreTableFactory.create( + fileIO, + tablePath, + new FileSystemSchemaManager(fileIO, tablePath).latest().get(), + env); + } + + private static void commitRow(FileStoreTable table, int id) { + BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder(); + try (BatchTableWrite write = writeBuilder.newWrite(); + BatchTableCommit commit = writeBuilder.newCommit()) { + write.write(GenericRow.of(id)); + commit.commit(write.prepareCommit()); + } catch (Exception e) { + throw new RuntimeException(e); + } + } + + private static List readIds(FileStoreTable table) throws IOException { + ReadBuilder readBuilder = table.newReadBuilder(); + List ids = new ArrayList<>(); + try (RecordReader reader = + readBuilder.newRead().createReader(readBuilder.newScan().plan())) { + reader.forEachRemaining(row -> ids.add(row.getInt(0))); + } + return ids; + } + private static Snapshot createSnapshot(long id) throws IOException { long schemaId = 1L; String baseManifestList = "manifest-list-base"; @@ -217,6 +362,106 @@ public boolean rename(Path src, Path dst) throws IOException { } } + /** + * A LocalFileIO whose rename replaces an existing target, and that can run an action once when + * a snapshot commit checks whether a file exists. + */ + private static class OverwritingRenameLocalFileIO extends LocalFileIO { + + private volatile Path trigger; + private volatile ThrowingRunnable action; + + void onCommitExists(Path path, ThrowingRunnable action) { + this.action = action; + this.trigger = path; + } + + @Override + public boolean exists(Path path) throws IOException { + boolean exists = super.exists(path); + if (path.equals(trigger) && calledFromCommitImpl()) { + trigger = null; + try { + action.run(); + } catch (Exception e) { + throw new IOException(e); + } + } + return exists; + } + + @Override + public boolean rename(Path src, Path dst) throws IOException { + java.nio.file.Path target = java.nio.file.Paths.get(dst.toUri()); + Files.createDirectories(target.getParent()); + Files.move( + java.nio.file.Paths.get(src.toUri()), + target, + StandardCopyOption.REPLACE_EXISTING); + return true; + } + + private static boolean calledFromCommitImpl() { + for (StackTraceElement e : Thread.currentThread().getStackTrace()) { + if (e.getClassName().startsWith(RenamingSnapshotCommit.class.getName()) + && e.getMethodName().equals("commitImpl")) { + return true; + } + } + return false; + } + } + + private interface ThrowingRunnable { + void run() throws Exception; + } + + /** + * Hands out lease locks that exclude each other. {@link #expireHolder} ends the current + * holder's lease, so another lock can be acquired and the holder fails {@code ensureValid}. + */ + private static class LeaseLockFactory implements CatalogLockFactory { + + private final AtomicReference holder = new AtomicReference<>(); + + void expireHolder() { + holder.set(null); + } + + @Override + public String identifier() { + return "lease"; + } + + @Override + public Lock createLock( + CatalogLockContext context, + Identifier identifier, + @Nullable String tableUuid, + @Nullable String commitUser, + Options tableOptions) { + return new Lock() { + @Override + public AutoCloseable acquire() { + if (!holder.compareAndSet(null, this)) { + throw new IllegalStateException("The lock is held by another writer"); + } + return () -> holder.compareAndSet(this, null); + } + + @Override + public void ensureValid() { + if (holder.get() != this) { + throw new IllegalStateException("Lease expired"); + } + } + + @Override + public void close() {} + }; + } + } + /** * A FileIO based on LocalFileIO that always throws from rename(). *