DuckLake: detect commit conflicts against concurrent writers, retry only when safe

docs/src/main/sphinx/connector/ducklake.md

@@ -98,13 +98,70 @@ The following configuration properties are available:
     split. Also configurable per query with the
     `max_split_size` [catalog session property](/sql/set-session).
   - `64MB`
+* - `ducklake.commit.max-retries`
+  - How often a commit that lost the race for the next snapshot is attempted again against the
+    newer state. Raise this on a catalog with many concurrent writers. See
+    [](ducklake-concurrent-writers).
+  - `10`
+* - `ducklake.commit.retry-backoff`
+  - How long to wait before attempting a commit again. The wait is doubled after each attempt,
+    up to 32 times this value.
+  - `20ms`
 :::
 
 The connector supports reading from S3, Azure Storage, Google Cloud Storage,
 and HDFS using the same [file system configuration](/object-storage) as other
 object storage connectors, such as `fs.native-s3.enabled=true` and the `s3.*`
 properties.
 
+(ducklake-concurrent-writers)=
+### Concurrent writers
+
+A DuckLake catalog orders every change on one chain of snapshots, so a commit
+has to claim the snapshot following the newest one. When another writer, such
+as DuckDB or a second Trino cluster, claims it first, the connector applies the
+DuckLake conflict rules and either lands the commit on the newer snapshot or
+fails the query. It never rewrites the data files it already wrote, and a
+failed attempt leaves nothing behind in the catalog.
+
+The commit lands on the newer snapshot when the other writer changed something
+this statement does not depend on. Two writers inserting into the same table is
+the common case and always succeeds, as does any pair of statements writing to
+different tables. The connector attempts the commit again up to
+`ducklake.commit.max-retries` times, waiting `ducklake.commit.retry-backoff`
+before the first attempt and doubling the wait after each one.
+
+The query fails with the `DUCKLAKE_COMMIT_CONFLICT` error code when the other
+writer changed the table this statement writes to, in a way that invalidates
+the result:
+
+* Inserting into a table another writer altered, dropped, or deleted from.
+* Deleting from a table another writer altered, dropped, inserted into, or
+  compacted.
+* Altering a table another writer altered or dropped.
+* Dropping a table another writer dropped.
+* Rewriting a data file or a delete file another writer replaced.
+
+Statements naming what they create, such as `CREATE TABLE` and `CREATE SCHEMA`,
+resolve the name against the newer catalog instead, and report the ordinary
+"already exists" or "not found" error if the other writer took it.
+
+These are the rules DuckDB applies to the same catalog, so a statement is
+accepted or rejected here exactly as it would be there. A client driving the
+statement can match `DUCKLAKE_COMMIT_CONFLICT` to tell a lost race apart from a
+broken catalog, and run the statement again once it has been replanned.
+
+A statement reads one snapshot throughout, so a table it only reads from can
+change under it without failing the commit. The result is then computed from
+the snapshot the statement started at, which is what reading a snapshot means,
+rather than from the newest one.
+
+A query fails with `DUCKLAKE_UNSUPPORTED_CHANGE_TYPE` when another writer
+recorded a kind of change this connector does not know. The connector cannot
+decide whether committing on top of that snapshot is safe, so it refuses rather
+than risk dropping the other writer's work. Upgrade the connector to a version
+that understands the DuckLake version the other writer uses.
+
 ## Type mapping
 
 The connector maps DuckLake column types to Trino types as follows:

plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/DuckLakeConfig.java

@@ -30,6 +30,7 @@
 import java.util.Optional;
 
 import static io.airlift.units.DataSize.Unit.MEGABYTE;
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
 import static java.util.concurrent.TimeUnit.MINUTES;
 import static java.util.concurrent.TimeUnit.SECONDS;
 
@@ -48,6 +49,8 @@ public class DuckLakeConfig
     private DataSize maxSplitSize = DataSize.of(64, MEGABYTE);
     private DataSize targetMaxFileSize = DataSize.of(128, MEGABYTE);
     private int maxOpenPartitions = 100;
+    private int commitMaxRetries = 10;
+    private Duration commitRetryBackoff = new Duration(20, MILLISECONDS);
 
     @NotNull
     public String getConnectionUrl()
@@ -240,4 +243,33 @@ public DuckLakeConfig setMaxOpenPartitions(int maxOpenPartitions)
         this.maxOpenPartitions = maxOpenPartitions;
         return this;
     }
+
+    @Min(0)
+    public int getCommitMaxRetries()
+    {
+        return commitMaxRetries;
+    }
+
+    @Config("ducklake.commit.max-retries")
+    @ConfigDescription("Re-attempt a commit that lost the race for the next snapshot this many times before failing")
+    public DuckLakeConfig setCommitMaxRetries(int commitMaxRetries)
+    {
+        this.commitMaxRetries = commitMaxRetries;
+        return this;
+    }
+
+    @NotNull
+    @MinDuration("0ms")
+    public Duration getCommitRetryBackoff()
+    {
+        return commitRetryBackoff;
+    }
+
+    @Config("ducklake.commit.retry-backoff")
+    @ConfigDescription("Wait this long before re-attempting a commit, doubling after each attempt up to 32 times this value")
+    public DuckLakeConfig setCommitRetryBackoff(Duration commitRetryBackoff)
+    {
+        this.commitRetryBackoff = commitRetryBackoff;
+        return this;
+    }
 }

plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/DuckLakeErrorCode.java

@@ -33,6 +33,8 @@ public enum DuckLakeErrorCode
     DUCKLAKE_COMMIT_FAILED(7, EXTERNAL),
     DUCKLAKE_WRITER_ERROR(8, EXTERNAL),
     DUCKLAKE_TOO_MANY_OPEN_PARTITIONS(9, USER_ERROR),
+    DUCKLAKE_COMMIT_CONFLICT(10, EXTERNAL),
+    DUCKLAKE_UNSUPPORTED_CHANGE_TYPE(11, EXTERNAL),
     /**/;
 
     private final ErrorCode errorCode;

plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/DuckLakeMetadata.java

@@ -197,16 +197,18 @@
      */
     private <T> T commit(DuckLakeCommitAction<T> action)
     {
-        // pin the transaction to a snapshot first, which also verifies the catalog format version
-        snapshotId();
+        // pin the transaction to a snapshot first, which also verifies the catalog format version.
+        // Everything this statement read, it read there, so that is the snapshot the commit has to
+        // be checked for conflicts against.
+        long readSnapshotId = snapshotId();
 
         class Result
         {
             T value;
             long snapshot;
         }
 
-        Result result = metastore.commit(commit -> {
+        Result result = metastore.commit(readSnapshotId, commit -> {
             Result committed = new Result();
             committed.value = action.run(commit);
             committed.snapshot = commit.effectiveSnapshotId();
@@ -422,7 +424,7 @@
         }
         Optional<DuckLakeCommit.TableIdentity> existing = commit.findTable(tableName.getSchemaName(), tableName.getTableName());
         if (existing.isPresent()) {
             switch (saveMode) {
                 case FAIL -> throw new TrinoException(TABLE_ALREADY_EXISTS, "Table already exists: " + tableName);
                 case IGNORE -> {
                     return null;
@@ -523,7 +525,7 @@
             existing.stream()
                     .filter(row -> row.parentColumn().isEmpty() && row.columnName().equalsIgnoreCase(column.getName()))
                     .findAny()
                     .ifPresent(_ -> {
                         throw new TrinoException(COLUMN_ALREADY_EXISTS, "Column already exists: " + column.getName());
                     });
             // identifiers are never reused, because data files written while a dropped column
@@ -1626,7 +1628,7 @@
                 || type.equals(INTEGER)
                 || type.equals(BIGINT)
                 || type.equals(BOOLEAN)
                 || type instanceof VarcharType varcharType && varcharType.isUnbounded();
     }
 
     /**
@@ -1702,7 +1704,7 @@
                     .build();
         }
         TableStatistics.Builder tableStatistics = TableStatistics.builder()
                 .setRowCount(Estimate.of(rowCount(handle).rowCount()));
 
         Map<Long, DuckLakeTableColumnStats> columnStats = metastore.tableColumnStatistics(handle.tableId()).stream()
                 .collect(toImmutableMap(DuckLakeTableColumnStats::columnId, stats -> stats));

plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/metastore/DuckLakeCommit.java

@@ -24,7 +24,7 @@
 import java.util.OptionalLong;
 import java.util.Set;
 
-import static io.trino.plugin.ducklake.DuckLakeErrorCode.DUCKLAKE_INVALID_METADATA;
+import static io.trino.plugin.ducklake.DuckLakeErrorCode.DUCKLAKE_COMMIT_CONFLICT;
 import static java.lang.String.join;
 import static java.util.Locale.ENGLISH;
 import static java.util.Objects.requireNonNull;
@@ -212,7 +212,7 @@
 
     public void insertTableRow(long tableId, long schemaId, String tableName, String path, Optional<String> tableUuid)
     {
         String uuidExpression = tableUuid.map(_ -> ":tableUuid" + uuidCast()).orElseGet(DuckLakeCommit::randomUuid);
         var update = handle.createUpdate(
                         """
                         INSERT INTO %s (table_id, table_uuid, begin_snapshot, end_snapshot, schema_id, table_name, path, path_is_relative)
@@ -807,6 +807,40 @@
         return snapshotId;
     }
 
+    /**
+     * Fails unless this commit may land on top of everything committed since {@code readSnapshotId},
+     * which is the snapshot the statement was planned and read against.
+     * <p>
+     * A statement reads a single snapshot, so what another writer did afterwards cannot change what
+     * it read. What it can change is whether the result may be written down: the DuckLake rules,
+     * applied here, say when it may not. When they allow it the commit lands on the newer snapshot
+     * unchanged, keeping the data files it already wrote.
+     */
+    void verifyNoConflictSince(long readSnapshotId)
+    {
+        if (changes.isEmpty() || baseSnapshotId <= readSnapshotId) {
+            // nothing to commit, or nothing was committed by anyone else in the meantime
+            return;
+        }
+        DuckLakeSnapshotChanges ourChanges = DuckLakeSnapshotChanges.parse(join(",", changes));
+        if (ourChanges.isEmpty()) {
+            // the statement created something by name, which the replay re-resolves by itself
+            return;
+        }
+        List<String> otherChangesMade = handle.createQuery(
+                        """
+                        SELECT changes_made FROM %s
+                        WHERE snapshot_id > :snapshot AND changes_made IS NOT NULL AND changes_made <> ''
+                        ORDER BY snapshot_id""".formatted(table("ducklake_snapshot_changes")))
+                .bind("snapshot", readSnapshotId)
+                .mapTo(String.class)
+                .list();
+        DuckLakeSnapshotChanges otherChanges = DuckLakeSnapshotChanges.parse(join(",", otherChangesMade));
+        ourChanges.conflictWith(otherChanges).ifPresent(conflict -> {
+            throw new ConcurrentModificationFailure("Conflicting concurrent commit to the DuckLake catalog: " + conflict);
+        });
+    }
+
     void writeSnapshot()
     {
         if (changes.isEmpty()) {
@@ -922,14 +956,16 @@
 
     /**
      * Signals that the state a statement was planned against changed before it could commit.
-     * Retrying the commit cannot help, because the statement has to be replanned.
+     * Retrying the commit cannot help, because the statement has to be replanned. It carries its
+     * own error code so that a client driving the statement can tell this apart from a catalog
+     * that is broken, and decide for itself whether to run the statement again.
      */
     public static class ConcurrentModificationFailure
             extends TrinoException
     {
         public ConcurrentModificationFailure(String message)
         {
-            super(DUCKLAKE_INVALID_METADATA, message);
+            super(DUCKLAKE_COMMIT_CONFLICT, message);
         }
     }