From e705996aa6a2beb35ae8d5fcafc428a8e3cd3a34 Mon Sep 17 00:00:00 2001 From: James Greenhill Date: Thu, 3 Sep 2026 23:15:05 +0000 Subject: [PATCH 1/2] Detect DuckLake commit conflicts, retry when safe A DuckLake catalog orders every change on one chain of snapshots, so a commit has to claim the snapshot after the newest one. With a DuckDB ingest pipeline writing to the same catalog, Trino loses that race often. The connector already replayed the action when the database reported the race, but it never asked whether replaying was allowed: it re-based onto the newer snapshot whatever the other writer had done. An insert whose target table was altered underneath it registered data files that no longer described the table's columns. Apply the DuckLake conflict rules before the snapshot is written. The commit compares the changes it is about to record against the changes of every snapshot committed since the one the statement read, and refuses to land when the other writer touched an object the result depends on: inserted into T vs T dropped, altered, or deleted from deleted from T vs T dropped, altered, inserted into, or compacted altered T vs T dropped or altered dropped T vs T dropped These are DuckDB's rules for the same catalog, so a statement is accepted or rejected here exactly as it would be there. Two writers inserting into one table is not a conflict, which is the case that matters for an ingest pipeline running beside a transform. Changes recorded by name -- creating a schema, table, or view -- carry no rule, because the replay re-resolves the name against the newer catalog and reports the ordinary "already exists" or "not found" error there. A refusal now carries its own error code, DUCKLAKE_COMMIT_CONFLICT, so that a client such as SQLMesh can tell a lost race apart from a broken catalog. The retry bound and backoff become catalog properties, ducklake.commit.max-retries and ducklake.commit.retry-backoff. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01WCY5Jf2BQPCVKJTZU1TpEe --- docs/src/main/sphinx/connector/ducklake.md | 51 +++ .../trino/plugin/ducklake/DuckLakeConfig.java | 32 ++ .../plugin/ducklake/DuckLakeErrorCode.java | 1 + .../plugin/ducklake/DuckLakeMetadata.java | 8 +- .../ducklake/metastore/DuckLakeCommit.java | 42 ++- .../metastore/DuckLakeSnapshotChanges.java | 195 +++++++++++ .../metastore/JdbcDuckLakeMetastore.java | 53 ++- .../plugin/ducklake/TestDuckLakeConfig.java | 9 +- .../TestDuckLakeConcurrentCommits.java | 322 ++++++++++++++++++ .../TestDuckLakeSnapshotChanges.java | 138 ++++++++ 10 files changed, 827 insertions(+), 24 deletions(-) create mode 100644 plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/metastore/DuckLakeSnapshotChanges.java create mode 100644 plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/metastore/TestDuckLakeConcurrentCommits.java create mode 100644 plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/metastore/TestDuckLakeSnapshotChanges.java diff --git a/docs/src/main/sphinx/connector/ducklake.md b/docs/src/main/sphinx/connector/ducklake.md index fdd2cf3d1cc4..db6b08a0d9c0 100644 --- a/docs/src/main/sphinx/connector/ducklake.md +++ b/docs/src/main/sphinx/connector/ducklake.md @@ -98,6 +98,15 @@ 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, @@ -105,6 +114,48 @@ 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. + ## Type mapping The connector maps DuckLake column types to Trino types as follows: diff --git a/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/DuckLakeConfig.java b/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/DuckLakeConfig.java index 2a186cada3b7..d16831b98512 100644 --- a/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/DuckLakeConfig.java +++ b/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; + } } diff --git a/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/DuckLakeErrorCode.java b/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/DuckLakeErrorCode.java index 063305b88452..beb4e6fcb3b8 100644 --- a/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/DuckLakeErrorCode.java +++ b/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/DuckLakeErrorCode.java @@ -33,6 +33,7 @@ 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), /**/; private final ErrorCode errorCode; diff --git a/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/DuckLakeMetadata.java b/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/DuckLakeMetadata.java index ae126c933fc5..52c3462e1de9 100644 --- a/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/DuckLakeMetadata.java +++ b/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/DuckLakeMetadata.java @@ -197,8 +197,10 @@ public synchronized long snapshotId() */ private T commit(DuckLakeCommitAction 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 { @@ -206,7 +208,7 @@ class Result 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(); diff --git a/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/metastore/DuckLakeCommit.java b/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/metastore/DuckLakeCommit.java index b4913125f614..0967eda3f86b 100644 --- a/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/metastore/DuckLakeCommit.java +++ b/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; @@ -807,6 +807,40 @@ public long effectiveSnapshotId() 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. + *

+ * 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 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 @@ public long visibleRecordCount() /** * 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); } } diff --git a/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/metastore/DuckLakeSnapshotChanges.java b/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/metastore/DuckLakeSnapshotChanges.java new file mode 100644 index 000000000000..3f0f47892896 --- /dev/null +++ b/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/metastore/DuckLakeSnapshotChanges.java @@ -0,0 +1,195 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.trino.plugin.ducklake.metastore; + +import com.google.common.collect.ImmutableSet; + +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; +import java.util.Set; + +import static com.google.common.collect.Sets.intersection; +import static java.util.Objects.requireNonNull; + +/** + * What one or more snapshots changed, read from {@code ducklake_snapshot_changes}. + *

+ * DuckLake decides whether two commits may coexist from these records alone: a commit is allowed + * to land on top of another one unless the two touched the same object. The vocabulary is shared + * with DuckDB, which writes the same strings and applies the same rules, so a statement is + * accepted or rejected here exactly as it would be there. + *

+ * Only the changes keyed by object identifier are kept. The ones keyed by name — creating a + * schema, table or view — need no record, because a commit that creates something by name + * re-resolves that name against the newer catalog when it is replayed, and fails there if the + * name has been taken or its schema has gone. + */ +public final class DuckLakeSnapshotChanges +{ + private final Set droppedRelations; + private final Set alteredRelations; + private final Set insertedIntoTables; + private final Set deletedFromTables; + private final Set compactedTables; + + /** + * Reads the changes from the comma-separated form DuckLake stores them in. Entries whose type + * is not recognized are rejected rather than ignored: an unknown change was written by a newer + * DuckLake version, and treating it as harmless could drop the writer's work. + */ + public static DuckLakeSnapshotChanges parse(String changesMade) + { + Builder builder = new Builder(); + for (String change : splitChanges(changesMade)) { + int separator = change.indexOf(':'); + if (separator < 0) { + throw new IllegalArgumentException("Malformed DuckLake change entry: " + change); + } + builder.add(change.substring(0, separator), change.substring(separator + 1)); + } + return builder.build(); + } + + /** + * Splits on the commas separating entries, which are the ones outside a quoted name. A created + * schema, table or view is recorded under its quoted name, and a name may hold a comma. + */ + private static List splitChanges(String changesMade) + { + List changes = new ArrayList<>(); + boolean quoted = false; + int start = 0; + for (int i = 0; i < changesMade.length(); i++) { + char character = changesMade.charAt(i); + if (character == '"') { + quoted = !quoted; + } + else if (character == ',' && !quoted) { + changes.add(changesMade.substring(start, i)); + start = i + 1; + } + } + if (start < changesMade.length()) { + changes.add(changesMade.substring(start)); + } + return changes; + } + + private DuckLakeSnapshotChanges( + Set droppedRelations, + Set alteredRelations, + Set insertedIntoTables, + Set deletedFromTables, + Set compactedTables) + { + this.droppedRelations = requireNonNull(droppedRelations, "droppedRelations is null"); + this.alteredRelations = requireNonNull(alteredRelations, "alteredRelations is null"); + this.insertedIntoTables = requireNonNull(insertedIntoTables, "insertedIntoTables is null"); + this.deletedFromTables = requireNonNull(deletedFromTables, "deletedFromTables is null"); + this.compactedTables = requireNonNull(compactedTables, "compactedTables is null"); + } + + public boolean isEmpty() + { + return droppedRelations.isEmpty() + && alteredRelations.isEmpty() + && insertedIntoTables.isEmpty() + && deletedFromTables.isEmpty() + && compactedTables.isEmpty(); + } + + /** + * Describes the first reason these changes cannot land on top of {@code other}, or an empty + * value when they can. + *

+ * The rules are DuckLake's. Two commits that touched different tables never conflict, and + * neither do two inserts into the same table: rows added by one are simply visible beside the + * rows added by the other. What conflicts is a commit whose result depends on the state + * another commit changed — inserting into a table whose columns were altered, deleting rows + * another writer already removed or rewrote, altering a table that was dropped. + */ + public Optional conflictWith(DuckLakeSnapshotChanges other) + { + return firstConflict(droppedRelations, other.droppedRelations, "dropped", "dropped it") + .or(() -> firstConflict(insertedIntoTables, other.droppedRelations, "inserted into", "dropped it")) + .or(() -> firstConflict(insertedIntoTables, other.alteredRelations, "inserted into", "altered it")) + .or(() -> firstConflict(insertedIntoTables, other.deletedFromTables, "inserted into", "deleted from it")) + .or(() -> firstConflict(deletedFromTables, other.droppedRelations, "deleted from", "dropped it")) + .or(() -> firstConflict(deletedFromTables, other.alteredRelations, "deleted from", "altered it")) + .or(() -> firstConflict(deletedFromTables, other.insertedIntoTables, "deleted from", "inserted into it")) + .or(() -> firstConflict(deletedFromTables, other.compactedTables, "deleted from", "compacted it")) + .or(() -> firstConflict(alteredRelations, other.droppedRelations, "altered", "dropped it")) + .or(() -> firstConflict(alteredRelations, other.alteredRelations, "altered", "altered it")); + } + + private static Optional firstConflict(Set mine, Set theirs, String action, String otherAction) + { + return intersection(mine, theirs).stream() + .min(Long::compare) + .map(objectId -> "this statement %s object %s, but another transaction %s".formatted(action, objectId, otherAction)); + } + + private static final class Builder + { + private final ImmutableSet.Builder droppedRelations = ImmutableSet.builder(); + private final ImmutableSet.Builder alteredRelations = ImmutableSet.builder(); + private final ImmutableSet.Builder insertedIntoTables = ImmutableSet.builder(); + private final ImmutableSet.Builder deletedFromTables = ImmutableSet.builder(); + private final ImmutableSet.Builder compactedTables = ImmutableSet.builder(); + + void add(String type, String value) + { + switch (type) { + // Tables and views are numbered from one counter, so an identifier names exactly + // one of them. Keeping them in one set makes a rule hold however the writer + // spelled the change, which DuckDB and this connector do not always do alike. + case "dropped_table", "dropped_view" -> droppedRelations.add(objectId(value)); + case "altered_table", "altered_view" -> alteredRelations.add(objectId(value)); + // Inlined rows live in the catalog database rather than a data file, but they are + // rows of the table either way and conflict with the same statements. + case "inserted_into_table", "inlined_insert" -> insertedIntoTables.add(objectId(value)); + case "deleted_from_table", "inlined_delete" -> deletedFromTables.add(objectId(value)); + case "compacted_table", "merge_adjacent", "rewrite_delete" -> compactedTables.add(objectId(value)); + // Creating something by name, dropping a schema, and flushing inlined rows into a + // data file are recorded but need no rule here; see the class comment for creates, + // and a flush adds a file without touching the ones a statement already read. + case "created_schema", "created_table", "created_view", "created_scalar_macro", "created_table_macro", + "dropped_schema", "dropped_scalar_macro", "dropped_table_macro", + "flushed_inlined", "inline_flush" -> {} + default -> throw new IllegalArgumentException("Unknown DuckLake change type: " + type); + } + } + + private static long objectId(String value) + { + try { + return Long.parseLong(value); + } + catch (NumberFormatException e) { + throw new IllegalArgumentException("Malformed DuckLake object identifier: " + value, e); + } + } + + DuckLakeSnapshotChanges build() + { + return new DuckLakeSnapshotChanges( + droppedRelations.build(), + alteredRelations.build(), + insertedIntoTables.build(), + deletedFromTables.build(), + compactedTables.build()); + } + } +} diff --git a/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/metastore/JdbcDuckLakeMetastore.java b/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/metastore/JdbcDuckLakeMetastore.java index 545cd9661480..09045f82770d 100644 --- a/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/metastore/JdbcDuckLakeMetastore.java +++ b/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/metastore/JdbcDuckLakeMetastore.java @@ -54,12 +54,16 @@ public class JdbcDuckLakeMetastore private static final String SERIALIZATION_FAILURE_SQL_STATE = "40001"; private static final String DEADLOCK_DETECTED_SQL_STATE = "40P01"; private static final String UNIQUE_VIOLATION_SQL_STATE = "23505"; - private static final int MAX_COMMIT_ATTEMPTS = 10; - private static final long COMMIT_RETRY_BASE_DELAY_MILLIS = 20; - private static final long MAX_COMMIT_RETRY_DELAY_MILLIS = 1000; + /** + * How often the wait between commit attempts is doubled. Waiting longer than this does not help + * a writer that keeps losing the race, and the bound holds whatever backoff is configured. + */ + private static final int MAX_COMMIT_RETRY_DOUBLINGS = 5; private final Jdbi jdbi; private final String metadataSchema; + private final int maxCommitRetries; + private final long commitRetryBackoffMillis; private volatile Boolean dataFileHasPartialMax; private volatile Boolean inlinedDataTablesRegistryExists; @@ -72,6 +76,8 @@ public JdbcDuckLakeMetastore(ConnectionFactory connectionFactory, DuckLakeConfig { this.jdbi = Jdbi.create(requireNonNull(connectionFactory, "connectionFactory is null")); this.metadataSchema = config.getMetadataSchema(); + this.maxCommitRetries = config.getCommitMaxRetries(); + this.commitRetryBackoffMillis = config.getCommitRetryBackoff().toMillis(); } public long currentSnapshotId() @@ -88,24 +94,32 @@ public long currentSnapshotId() } /** - * Runs the action against a new snapshot and commits it atomically. + * Runs the action against a new snapshot and commits it atomically, re-basing it onto whatever + * another writer committed in the meantime. *

- * DuckLake orders all changes to a catalog on a single snapshot chain, so a commit conflicts - * with any other commit that started from the same snapshot. Conflicts are detected by the - * database rather than avoided by locking: the transaction runs at {@code SERIALIZABLE}, and - * the snapshot table's primary key rejects a second commit claiming the same snapshot - * identifier. Either way the action is discarded and replayed against the newer state, which - * is safe because it only reads catalog rows and writes them through this commit — the data - * files it registers were written before the commit began and are unaffected by a replay. + * DuckLake orders all changes to a catalog on a single snapshot chain, so every commit has to + * claim the snapshot following the newest one. Losing that race is ordinary rather than + * exceptional here, because a DuckDB writer commits to the same catalog: the database reports + * it — as a serialization failure, or as a unique violation on the snapshot table's primary + * key — and the action is discarded and replayed against the newer state. A replay is safe + * because the action only reads catalog rows and writes them through this commit; the data + * files it registers were written to object storage before the commit began, and the discarded + * attempt left no rows behind. + *

+ * Re-basing is only allowed where DuckLake allows it. {@code readSnapshotId} is the snapshot the + * statement was planned and read against, and the commit refuses to land if anything committed + * after it changed an object the statement's own result depends on. The refusal carries + * {@code DUCKLAKE_COMMIT_CONFLICT} and is not retried: only replanning the statement can help. */ - public T commit(DuckLakeCommitAction action) + public T commit(long readSnapshotId, DuckLakeCommitAction action) { RuntimeException conflict = null; - for (int attempt = 0; attempt < MAX_COMMIT_ATTEMPTS; attempt++) { + for (int attempt = 0; attempt <= maxCommitRetries; attempt++) { try { return jdbi.inTransaction(TransactionIsolationLevel.SERIALIZABLE, handle -> { DuckLakeCommit commit = new DuckLakeCommit(handle, metadataSchema, snapshotState(handle)); T result = action.run(commit); + commit.verifyNoConflictSince(readSnapshotId); commit.writeSnapshot(); return result; }); @@ -119,9 +133,14 @@ public T commit(DuckLakeCommitAction action) catch (DuckLakeCommit.ConcurrentModificationFailure e) { throw e; } - sleepBeforeRetry(attempt); + if (attempt < maxCommitRetries) { + sleepBeforeRetry(attempt); + } } - throw new TrinoException(DUCKLAKE_COMMIT_FAILED, "Failed to commit to the DuckLake catalog after %s attempts because of concurrent updates".formatted(MAX_COMMIT_ATTEMPTS), conflict); + throw new TrinoException( + DUCKLAKE_COMMIT_FAILED, + "Failed to commit to the DuckLake catalog after %s retries because of concurrent updates; raise ducklake.commit.max-retries if this is common".formatted(maxCommitRetries), + conflict); } private DuckLakeCommit.SnapshotState snapshotState(Handle handle) @@ -156,10 +175,10 @@ private static boolean isRetriableConflict(Throwable throwable) return false; } - private static void sleepBeforeRetry(int attempt) + private void sleepBeforeRetry(int attempt) { try { - Thread.sleep(Math.min(COMMIT_RETRY_BASE_DELAY_MILLIS << attempt, MAX_COMMIT_RETRY_DELAY_MILLIS)); + Thread.sleep(commitRetryBackoffMillis << Math.min(attempt, MAX_COMMIT_RETRY_DOUBLINGS)); } catch (InterruptedException e) { Thread.currentThread().interrupt(); diff --git a/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/TestDuckLakeConfig.java b/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/TestDuckLakeConfig.java index dc44d8193752..00e777c379c4 100644 --- a/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/TestDuckLakeConfig.java +++ b/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/TestDuckLakeConfig.java @@ -34,6 +34,7 @@ import static io.airlift.testing.ValidationAssertions.assertFailsValidation; import static io.airlift.testing.ValidationAssertions.assertValidates; 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; import static org.assertj.core.api.Assertions.assertThat; @@ -59,7 +60,9 @@ void testDefaults() .setFileStatisticsPruningEnabled(true) .setMaxSplitSize(DataSize.of(64, MEGABYTE)) .setTargetMaxFileSize(DataSize.of(128, MEGABYTE)) - .setMaxOpenPartitions(100)); + .setMaxOpenPartitions(100) + .setCommitMaxRetries(10) + .setCommitRetryBackoff(new Duration(20, MILLISECONDS))); } /** @@ -113,6 +116,8 @@ private static Map commonProperties() .put("ducklake.max-split-size", "32MB") .put("ducklake.target-max-file-size", "48MB") .put("ducklake.max-open-partitions", "7") + .put("ducklake.commit.max-retries", "4") + .put("ducklake.commit.retry-backoff", "250ms") .buildOrThrow(); } @@ -129,6 +134,8 @@ private static void assertCommonProperties(DuckLakeConfig config) assertThat(config.getMaxSplitSize()).isEqualTo(DataSize.of(32, MEGABYTE)); assertThat(config.getTargetMaxFileSize()).isEqualTo(DataSize.of(48, MEGABYTE)); assertThat(config.getMaxOpenPartitions()).isEqualTo(7); + assertThat(config.getCommitMaxRetries()).isEqualTo(4); + assertThat(config.getCommitRetryBackoff()).isEqualTo(new Duration(250, MILLISECONDS)); } @Test diff --git a/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/metastore/TestDuckLakeConcurrentCommits.java b/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/metastore/TestDuckLakeConcurrentCommits.java new file mode 100644 index 000000000000..cf14a13960fe --- /dev/null +++ b/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/metastore/TestDuckLakeConcurrentCommits.java @@ -0,0 +1,322 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.trino.plugin.ducklake.metastore; + +import io.trino.plugin.ducklake.DuckLakeConfig; +import io.trino.plugin.ducklake.TestingDuckLakeCatalog; +import io.trino.spi.TrinoException; +import org.intellij.lang.annotations.Language; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.TestInstance; + +import java.io.IOException; +import java.io.UncheckedIOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Statement; +import java.util.ArrayList; +import java.util.List; +import java.util.OptionalLong; +import java.util.concurrent.atomic.AtomicInteger; + +import static com.google.common.base.Verify.verify; +import static io.trino.plugin.ducklake.DuckLakeErrorCode.DUCKLAKE_COMMIT_CONFLICT; +import static io.trino.testing.TestingNames.randomNameSuffix; +import static java.nio.file.StandardCopyOption.REPLACE_EXISTING; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.junit.jupiter.api.Assumptions.abort; +import static org.junit.jupiter.api.TestInstance.Lifecycle.PER_CLASS; + +/** + * An ingest pipeline on DuckDB and a transform pipeline on Trino commit to one DuckLake catalog, so + * a Trino commit regularly finds that another writer claimed the next snapshot first. + *

+ * The race is made deterministic by committing the foreign snapshot from DuckDB inside the + * connector's own commit action, which is the moment between reading the newest snapshot and + * writing the new one. The first attempt therefore always loses it. + */ +@TestInstance(PER_CLASS) +final class TestDuckLakeConcurrentCommits +{ + private TestingDuckLakeCatalog catalog; + private JdbcDuckLakeMetastore metastore; + + @BeforeAll + void setUp() + { + catalog = new TestingDuckLakeCatalog(); + try { + catalog.executeInDuckDb("CREATE TABLE bootstrap (x INTEGER)", "DROP TABLE bootstrap"); + } + catch (SQLException e) { + abort("Failed to create a DuckLake catalog with DuckDB (extension download requires network access): " + e); + } + metastore = new JdbcDuckLakeMetastore( + () -> DriverManager.getConnection(catalog.jdbcUrl(), TestingDuckLakeCatalog.USER, TestingDuckLakeCatalog.PASSWORD), + new DuckLakeConfig()); + } + + @AfterAll + void tearDown() + { + metastore = null; + catalog.close(); + catalog = null; + } + + /** + * The foreign snapshot changes a table this statement did not touch, so the commit is re-based + * onto it and lands with the data file it already wrote. + */ + @Test + void testCommitIsRebasedOntoAForeignSnapshotOnAnotherTable() + throws Exception + { + String table = "rebase_" + randomNameSuffix(); + String other = "unrelated_" + randomNameSuffix(); + createTable(table); + createTable(other); + DataFile dataFile = stageDataFile(table, 42); + + long readSnapshotId = metastore.currentSnapshotId(); + AtomicInteger attempts = new AtomicInteger(); + long committed = registerDataFile(readSnapshotId, table, dataFile, attempts, () -> executeInDuckDb("INSERT INTO " + other + " VALUES (7)")); + + assertThat(attempts).hasValue(2); + // the foreign snapshot took the identifier the first attempt wanted, and this commit took the next one + assertThat(committed).isEqualTo(readSnapshotId + 2); + assertThat(committed).isEqualTo(metastore.currentSnapshotId()); + + // the file was registered once, and DuckDB reads its row through the re-based commit + assertThat(dataFileCount(table)).isEqualTo(1); + assertThat(duckDbRows("SELECT x FROM " + table)).isEqualTo(List.of(42L)); + assertThat(duckDbRows("SELECT x FROM " + other)).isEqualTo(List.of(7L)); + } + + /** + * The foreign snapshot alters the very table this statement writes to, so the data file it wrote + * no longer describes the table's columns. Re-basing would register it anyway, so the commit + * fails and says so. + */ + @Test + void testCommitFailsWhenAForeignSnapshotAltersTheTargetTable() + throws Exception + { + String table = "conflict_" + randomNameSuffix(); + createTable(table); + DataFile dataFile = stageDataFile(table, 42); + + long readSnapshotId = metastore.currentSnapshotId(); + AtomicInteger attempts = new AtomicInteger(); + + assertThatThrownBy(() -> registerDataFile(readSnapshotId, table, dataFile, attempts, () -> executeInDuckDb("ALTER TABLE " + table + " ADD COLUMN z INTEGER"))) + .isInstanceOf(TrinoException.class) + .matches(failure -> ((TrinoException) failure).getErrorCode().equals(DUCKLAKE_COMMIT_CONFLICT.toErrorCode())) + .hasMessageContaining("Conflicting concurrent commit to the DuckLake catalog") + .hasMessageContaining("but another transaction altered it"); + + // every attempt was rolled back, so no row of the statement reached the catalog + assertThat(attempts).hasValue(2); + assertThat(dataFileCount(table)).isEqualTo(0); + assertThat(duckDbRows("SELECT x FROM " + table)).isEmpty(); + } + + /** + * Two writers adding rows to one table is the ordinary case of an ingest pipeline running beside + * a transform, and DuckLake lets both land. + */ + @Test + void testCommitIsRebasedOntoAForeignInsertIntoTheSameTable() + throws Exception + { + String table = "shared_" + randomNameSuffix(); + createTable(table); + DataFile dataFile = stageDataFile(table, 42); + + long readSnapshotId = metastore.currentSnapshotId(); + AtomicInteger attempts = new AtomicInteger(); + registerDataFile(readSnapshotId, table, dataFile, attempts, () -> executeInDuckDb("INSERT INTO " + table + " VALUES (7)")); + + assertThat(attempts).hasValue(2); + assertThat(dataFileCount(table)).isEqualTo(2); + assertThat(duckDbRows("SELECT x FROM " + table + " ORDER BY x")).isEqualTo(List.of(7L, 42L)); + } + + /** + * Registers an already written data file into a table the way {@code finishInsert} does, while + * {@code foreignCommit} lands a snapshot from DuckDB during the first attempt. Returns the + * snapshot the commit ended up on. + */ + private long registerDataFile(long readSnapshotId, String tableName, DataFile dataFile, AtomicInteger attempts, Runnable foreignCommit) + { + return metastore.commit(readSnapshotId, commit -> { + if (attempts.incrementAndGet() == 1) { + foreignCommit.run(); + } + DuckLakeCommit.TableIdentity table = commit.findTable("main", tableName).orElseThrow(); + DuckLakeCommit.TableStatsRow stats = commit.tableStats(table.tableId()) + .orElseGet(() -> new DuckLakeCommit.TableStatsRow(0, 0, 0)); + commit.insertDataFile(table.tableId(), commit.allocateFileId(), new DuckLakeCommit.DataFileRow( + dataFile.path(), + dataFile.recordCount(), + dataFile.fileSizeBytes(), + dataFile.footerSize(), + stats.nextRowId(), + OptionalLong.empty())); + commit.writeTableStats(table.tableId(), new DuckLakeCommit.TableStatsRow( + stats.recordCount() + dataFile.recordCount(), + stats.nextRowId() + dataFile.recordCount(), + stats.fileSizeBytes() + dataFile.fileSizeBytes())); + commit.recordInsert(table.tableId()); + return commit.effectiveSnapshotId(); + }); + } + + private void createTable(String tableName) + throws SQLException + { + catalog.executeInDuckDb("CREATE TABLE %s (x INTEGER)".formatted(tableName)); + } + + /** + * Produces a data file lying in the table's directory but registered nowhere, standing in for + * the file a Trino worker writes before the commit that registers it. DuckDB writes it into a + * throwaway table of the same shape, so that it carries the DuckLake field ids a reader needs, + * and the copy is made under the target table where the registered path resolves to it. + */ + private DataFile stageDataFile(String tableName, int value) + throws SQLException + { + String sourceTable = tableName + "_source"; + catalog.executeInDuckDb( + "CREATE TABLE %s (x INTEGER)".formatted(sourceTable), + "INSERT INTO %s VALUES (%s)".formatted(sourceTable, value)); + + DataFile source = onlyDataFile(sourceTable); + Path staged = tableDirectory(tableName).resolve("staged.parquet"); + try { + Files.createDirectories(staged.getParent()); + Files.copy(tableDirectory(sourceTable).resolve(source.path()), staged, REPLACE_EXISTING); + } + catch (IOException e) { + throw new UncheckedIOException(e); + } + // the source table is dropped so that its own file is not counted with the target's rows + catalog.executeInDuckDb("DROP TABLE " + sourceTable); + return new DataFile(staged.getFileName().toString(), source.recordCount(), source.fileSizeBytes(), source.footerSize()); + } + + /** + * Where the data files of a table live, which is the table's path resolved under its schema's. + * A registered path is relative to it. + */ + private Path tableDirectory(String tableName) + { + @Language("SQL") String sql = + """ + SELECT s.path AS schema_path, t.path AS table_path + FROM ducklake_table t + JOIN ducklake_schema s ON t.schema_id = s.schema_id + WHERE t.table_name = '%s' AND t.end_snapshot IS NULL AND s.end_snapshot IS NULL""".formatted(tableName); + return inCatalogDatabase(sql, resultSet -> { + verify(resultSet.next(), "no table named %s", tableName); + return catalog.dataPath().resolve(resultSet.getString("schema_path")).resolve(resultSet.getString("table_path")); + }); + } + + private DataFile onlyDataFile(String tableName) + { + @Language("SQL") String sql = + """ + SELECT f.path, f.record_count, f.file_size_bytes, f.footer_size + FROM ducklake_data_file f + JOIN ducklake_table t ON f.table_id = t.table_id + WHERE t.table_name = '%s' AND f.end_snapshot IS NULL""".formatted(tableName); + return inCatalogDatabase(sql, resultSet -> { + verify(resultSet.next(), "no data file for table %s", tableName); + DataFile file = new DataFile( + resultSet.getString("path"), + resultSet.getLong("record_count"), + resultSet.getLong("file_size_bytes"), + resultSet.getLong("footer_size")); + verify(!resultSet.next(), "more than one data file for table %s", tableName); + return file; + }); + } + + private long dataFileCount(String tableName) + { + @Language("SQL") String sql = + """ + SELECT count(*) FROM ducklake_data_file f + JOIN ducklake_table t ON f.table_id = t.table_id + WHERE t.table_name = '%s' AND f.end_snapshot IS NULL AND t.end_snapshot IS NULL""".formatted(tableName); + return inCatalogDatabase(sql, resultSet -> { + verify(resultSet.next(), "no result returned by %s", sql); + return resultSet.getLong(1); + }); + } + + private T inCatalogDatabase(@Language("SQL") String sql, ResultSetMapper mapper) + { + try (Connection connection = DriverManager.getConnection(catalog.jdbcUrl(), TestingDuckLakeCatalog.USER, TestingDuckLakeCatalog.PASSWORD); + Statement statement = connection.createStatement(); + ResultSet resultSet = statement.executeQuery(sql)) { + return mapper.map(resultSet); + } + catch (SQLException e) { + throw new RuntimeException(e); + } + } + + private void executeInDuckDb(@Language("SQL") String sql) + { + try { + catalog.executeInDuckDb(sql); + } + catch (SQLException e) { + throw new RuntimeException(e); + } + } + + private List duckDbRows(@Language("SQL") String sql) + throws SQLException + { + try (Connection connection = catalog.openDuckDbConnection(); + Statement statement = connection.createStatement(); + ResultSet resultSet = statement.executeQuery(sql)) { + List rows = new ArrayList<>(); + while (resultSet.next()) { + rows.add(resultSet.getLong(1)); + } + return rows; + } + } + + private record DataFile(String path, long recordCount, long fileSizeBytes, long footerSize) {} + + @FunctionalInterface + private interface ResultSetMapper + { + T map(ResultSet resultSet) + throws SQLException; + } +} diff --git a/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/metastore/TestDuckLakeSnapshotChanges.java b/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/metastore/TestDuckLakeSnapshotChanges.java new file mode 100644 index 000000000000..4fd2737af37e --- /dev/null +++ b/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/metastore/TestDuckLakeSnapshotChanges.java @@ -0,0 +1,138 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.trino.plugin.ducklake.metastore; + +import org.junit.jupiter.api.Test; + +import static io.trino.plugin.ducklake.metastore.DuckLakeSnapshotChanges.parse; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * The rules deciding whether a commit may land on top of another one. They are DuckLake's, so each + * case here is one a DuckDB writer against the same catalog would decide the same way. + */ +final class TestDuckLakeSnapshotChanges +{ + @Test + void testUnrelatedTablesNeverConflict() + { + assertThat(parse("inserted_into_table:1").conflictWith(parse("altered_table:2,dropped_table:3,deleted_from_table:4"))).isEmpty(); + assertThat(parse("deleted_from_table:1").conflictWith(parse("inserted_into_table:2,merge_adjacent:3"))).isEmpty(); + assertThat(parse("altered_table:1").conflictWith(parse("altered_table:2"))).isEmpty(); + } + + @Test + void testConcurrentInsertsIntoOneTableAreAllowed() + { + // both writers only add rows, and the rows of one are visible beside the rows of the other + assertThat(parse("inserted_into_table:1").conflictWith(parse("inserted_into_table:1"))).isEmpty(); + assertThat(parse("inserted_into_table:1").conflictWith(parse("inlined_insert:1"))).isEmpty(); + } + + @Test + void testInsertConflictsWithChangesToTheTableItWrites() + { + // the data files were written against the columns the table had, so an alter invalidates them + assertThat(parse("inserted_into_table:1").conflictWith(parse("altered_table:1"))) + .contains("this statement inserted into object 1, but another transaction altered it"); + assertThat(parse("inserted_into_table:1").conflictWith(parse("dropped_table:1"))) + .contains("this statement inserted into object 1, but another transaction dropped it"); + assertThat(parse("inserted_into_table:1").conflictWith(parse("deleted_from_table:1"))) + .contains("this statement inserted into object 1, but another transaction deleted from it"); + assertThat(parse("inserted_into_table:1").conflictWith(parse("inlined_delete:1"))) + .contains("this statement inserted into object 1, but another transaction deleted from it"); + } + + @Test + void testDeleteConflictsWithChangesToTheTableItRewrites() + { + assertThat(parse("deleted_from_table:1").conflictWith(parse("dropped_table:1"))).isPresent(); + assertThat(parse("deleted_from_table:1").conflictWith(parse("altered_table:1"))).isPresent(); + assertThat(parse("deleted_from_table:1").conflictWith(parse("inserted_into_table:1"))) + .contains("this statement deleted from object 1, but another transaction inserted into it"); + assertThat(parse("deleted_from_table:1").conflictWith(parse("merge_adjacent:1"))) + .contains("this statement deleted from object 1, but another transaction compacted it"); + assertThat(parse("deleted_from_table:1").conflictWith(parse("rewrite_delete:1"))).isPresent(); + assertThat(parse("deleted_from_table:1").conflictWith(parse("compacted_table:1"))).isPresent(); + } + + @Test + void testAlterAndDropConflictWithThemselves() + { + assertThat(parse("altered_table:1").conflictWith(parse("altered_table:1"))).isPresent(); + assertThat(parse("altered_table:1").conflictWith(parse("dropped_table:1"))).isPresent(); + assertThat(parse("dropped_table:1").conflictWith(parse("dropped_table:1"))) + .contains("this statement dropped object 1, but another transaction dropped it"); + // a view carries an identifier from the same counter, whichever word the writer used for it + assertThat(parse("altered_table:1").conflictWith(parse("dropped_view:1"))).isPresent(); + assertThat(parse("dropped_view:1").conflictWith(parse("dropped_table:1"))).isPresent(); + assertThat(parse("altered_view:1").conflictWith(parse("altered_table:1"))).isPresent(); + + // dropping something another writer altered is not a conflict: the drop removes it either way + assertThat(parse("dropped_table:1").conflictWith(parse("altered_table:1"))).isEmpty(); + } + + /** + * Creating a schema, table or view is recorded by name. A commit that creates one re-resolves + * the name when it is replayed against the newer catalog and fails there, so these changes + * carry no rule of their own. + */ + @Test + void testChangesRecordedByNameCarryNoRule() + { + assertThat(parse("created_table:\"main\".\"orders\"").isEmpty()).isTrue(); + assertThat(parse("created_schema:\"main\",dropped_schema:7").isEmpty()).isTrue(); + assertThat(parse("created_view:\"main\".\"v\"").conflictWith(parse("created_view:\"main\".\"v\""))).isEmpty(); + } + + @Test + void testFlushingInlinedRowsCarriesNoRule() + { + // the flush writes a new data file; it changes none of the files a statement read + assertThat(parse("inserted_into_table:1").conflictWith(parse("flushed_inlined:1"))).isEmpty(); + assertThat(parse("deleted_from_table:1").conflictWith(parse("inline_flush:1"))).isEmpty(); + } + + @Test + void testNamesHoldingACommaAreOneEntry() + { + assertThat(parse("created_table:\"main\".\"a,b\",inserted_into_table:1").conflictWith(parse("altered_table:1"))).isPresent(); + } + + @Test + void testEmptyChanges() + { + assertThat(parse("").isEmpty()).isTrue(); + assertThat(parse("inserted_into_table:1").conflictWith(parse(""))).isEmpty(); + } + + /** + * A change type this connector does not know was written by a newer DuckLake version. Ignoring + * it could drop the work of the writer that made it, so it is rejected instead. + */ + @Test + void testUnknownChangeTypeIsRejected() + { + assertThatThrownBy(() -> parse("teleported_table:1")) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Unknown DuckLake change type: teleported_table"); + assertThatThrownBy(() -> parse("dropped_table:not_a_number")) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Malformed DuckLake object identifier: not_a_number"); + assertThatThrownBy(() -> parse("dropped_table")) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Malformed DuckLake change entry: dropped_table"); + } +} From 8713e0c376e1cb004a3c24bcd8c85def55c35e73 Mon Sep 17 00:00:00 2001 From: James Greenhill Date: Thu, 3 Sep 2026 23:28:40 +0000 Subject: [PATCH 2/2] Type the error for an unknown snapshot change Parsing a foreign snapshot's changes threw IllegalArgumentException. That is neither JdbiException nor ConcurrentModificationFailure, so it escaped the commit retry loop and reached Trino as an unclassified internal error. It is reachable without corruption: a newer DuckLake writer commits a change type this connector does not know, and every commit whose read snapshot precedes that row hits it. Throw a TrinoException instead, and separate the two cases, which mean different things to whoever reads the error: DUCKLAKE_UNSUPPORTED_CHANGE_TYPE a newer DuckLake wrote a change type this connector cannot interpret, so it cannot tell whether committing on top is safe. The message names the token and says to upgrade the connector. DUCKLAKE_INVALID_METADATA the row is not shaped like a change at all, which is corruption. Failing stays the right direction: ignoring an unknown change could drop the other writer's work. Neither failure is retried, because reading the row again reaches the same conclusion. The malformed-entry path had the same defect and is fixed with it. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01WCY5Jf2BQPCVKJTZU1TpEe --- docs/src/main/sphinx/connector/ducklake.md | 6 +++ .../plugin/ducklake/DuckLakeErrorCode.java | 1 + .../metastore/DuckLakeSnapshotChanges.java | 25 +++++++--- .../TestDuckLakeConcurrentCommits.java | 48 +++++++++++++++++++ .../TestDuckLakeSnapshotChanges.java | 31 +++++++++--- 5 files changed, 98 insertions(+), 13 deletions(-) diff --git a/docs/src/main/sphinx/connector/ducklake.md b/docs/src/main/sphinx/connector/ducklake.md index db6b08a0d9c0..008d01927850 100644 --- a/docs/src/main/sphinx/connector/ducklake.md +++ b/docs/src/main/sphinx/connector/ducklake.md @@ -156,6 +156,12 @@ 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: diff --git a/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/DuckLakeErrorCode.java b/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/DuckLakeErrorCode.java index beb4e6fcb3b8..64142b21015b 100644 --- a/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/DuckLakeErrorCode.java +++ b/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/DuckLakeErrorCode.java @@ -34,6 +34,7 @@ public enum DuckLakeErrorCode 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; diff --git a/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/metastore/DuckLakeSnapshotChanges.java b/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/metastore/DuckLakeSnapshotChanges.java index 3f0f47892896..8efb29e0c055 100644 --- a/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/metastore/DuckLakeSnapshotChanges.java +++ b/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/metastore/DuckLakeSnapshotChanges.java @@ -14,6 +14,7 @@ package io.trino.plugin.ducklake.metastore; import com.google.common.collect.ImmutableSet; +import io.trino.spi.TrinoException; import java.util.ArrayList; import java.util.List; @@ -21,6 +22,8 @@ import java.util.Set; import static com.google.common.collect.Sets.intersection; +import static io.trino.plugin.ducklake.DuckLakeErrorCode.DUCKLAKE_INVALID_METADATA; +import static io.trino.plugin.ducklake.DuckLakeErrorCode.DUCKLAKE_UNSUPPORTED_CHANGE_TYPE; import static java.util.Objects.requireNonNull; /** @@ -45,9 +48,15 @@ public final class DuckLakeSnapshotChanges private final Set compactedTables; /** - * Reads the changes from the comma-separated form DuckLake stores them in. Entries whose type - * is not recognized are rejected rather than ignored: an unknown change was written by a newer - * DuckLake version, and treating it as harmless could drop the writer's work. + * Reads the changes from the comma-separated form DuckLake stores them in. + *

+ * Two things can go wrong, and they mean different things to whoever reads the error. A change + * type this connector does not know is a catalog written by a newer DuckLake than this + * connector understands, and it fails with {@code DUCKLAKE_UNSUPPORTED_CHANGE_TYPE}: the + * connector cannot tell whether committing on top of it is safe, and ignoring it could drop the + * other writer's work. An entry that is not shaped like a change at all is a corrupt row, and + * fails as invalid metadata. Neither is retried, because attempting the commit again reads the + * same row and reaches the same conclusion. */ public static DuckLakeSnapshotChanges parse(String changesMade) { @@ -55,7 +64,7 @@ public static DuckLakeSnapshotChanges parse(String changesMade) for (String change : splitChanges(changesMade)) { int separator = change.indexOf(':'); if (separator < 0) { - throw new IllegalArgumentException("Malformed DuckLake change entry: " + change); + throw new TrinoException(DUCKLAKE_INVALID_METADATA, "Malformed DuckLake change entry: " + change); } builder.add(change.substring(0, separator), change.substring(separator + 1)); } @@ -168,7 +177,11 @@ void add(String type, String value) case "created_schema", "created_table", "created_view", "created_scalar_macro", "created_table_macro", "dropped_schema", "dropped_scalar_macro", "dropped_table_macro", "flushed_inlined", "inline_flush" -> {} - default -> throw new IllegalArgumentException("Unknown DuckLake change type: " + type); + default -> throw new TrinoException( + DUCKLAKE_UNSUPPORTED_CHANGE_TYPE, + ("Another writer recorded the DuckLake change type '%s', which this connector does not understand, " + + "so it cannot tell whether committing on top of that snapshot is safe. " + + "Upgrade the DuckLake connector to a version that knows this change type.").formatted(type)); } } @@ -178,7 +191,7 @@ private static long objectId(String value) return Long.parseLong(value); } catch (NumberFormatException e) { - throw new IllegalArgumentException("Malformed DuckLake object identifier: " + value, e); + throw new TrinoException(DUCKLAKE_INVALID_METADATA, "Malformed DuckLake object identifier: " + value, e); } } diff --git a/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/metastore/TestDuckLakeConcurrentCommits.java b/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/metastore/TestDuckLakeConcurrentCommits.java index cf14a13960fe..9dfdc72893d1 100644 --- a/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/metastore/TestDuckLakeConcurrentCommits.java +++ b/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/metastore/TestDuckLakeConcurrentCommits.java @@ -38,6 +38,7 @@ import static com.google.common.base.Verify.verify; import static io.trino.plugin.ducklake.DuckLakeErrorCode.DUCKLAKE_COMMIT_CONFLICT; +import static io.trino.plugin.ducklake.DuckLakeErrorCode.DUCKLAKE_UNSUPPORTED_CHANGE_TYPE; import static io.trino.testing.TestingNames.randomNameSuffix; import static java.nio.file.StandardCopyOption.REPLACE_EXISTING; import static org.assertj.core.api.Assertions.assertThat; @@ -160,6 +161,42 @@ void testCommitIsRebasedOntoAForeignInsertIntoTheSameTable() assertThat(duckDbRows("SELECT x FROM " + table + " ORDER BY x")).isEqualTo(List.of(7L, 42L)); } + /** + * A newer DuckLake records a change type this connector has never seen. It cannot tell whether + * its own commit may land on top of that snapshot, so it fails saying what it read and what to + * do about it, rather than committing and risking the other writer's work. + */ + @Test + void testCommitFailsOnAChangeTypeTheConnectorDoesNotUnderstand() + throws Exception + { + String table = "unknown_" + randomNameSuffix(); + String other = "elsewhere_" + randomNameSuffix(); + createTable(table); + createTable(other); + DataFile dataFile = stageDataFile(table, 42); + + long readSnapshotId = metastore.currentSnapshotId(); + AtomicInteger attempts = new AtomicInteger(); + + assertThatThrownBy(() -> registerDataFile(readSnapshotId, table, dataFile, attempts, () -> { + // a real foreign snapshot, rewritten to record a change this connector cannot interpret + executeInDuckDb("INSERT INTO " + other + " VALUES (7)"); + executeInCatalogDatabase( + """ + UPDATE ducklake_snapshot_changes SET changes_made = 'teleported_table:99' + WHERE snapshot_id = (SELECT max(snapshot_id) FROM ducklake_snapshot_changes)"""); + })) + .isInstanceOf(TrinoException.class) + .matches(failure -> ((TrinoException) failure).getErrorCode().equals(DUCKLAKE_UNSUPPORTED_CHANGE_TYPE.toErrorCode())) + .hasMessageContaining("recorded the DuckLake change type 'teleported_table'") + .hasMessageContaining("Upgrade the DuckLake connector"); + + // reading the row again reaches the same conclusion, so the failure is not retried + assertThat(attempts).hasValue(2); + assertThat(dataFileCount(table)).isEqualTo(0); + } + /** * Registers an already written data file into a table the way {@code finishInsert} does, while * {@code foreignCommit} lands a snapshot from DuckDB during the first attempt. Returns the @@ -287,6 +324,17 @@ private T inCatalogDatabase(@Language("SQL") String sql, ResultSetMapper } } + private void executeInCatalogDatabase(@Language("SQL") String sql) + { + try (Connection connection = DriverManager.getConnection(catalog.jdbcUrl(), TestingDuckLakeCatalog.USER, TestingDuckLakeCatalog.PASSWORD); + Statement statement = connection.createStatement()) { + statement.executeUpdate(sql); + } + catch (SQLException e) { + throw new RuntimeException(e); + } + } + private void executeInDuckDb(@Language("SQL") String sql) { try { diff --git a/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/metastore/TestDuckLakeSnapshotChanges.java b/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/metastore/TestDuckLakeSnapshotChanges.java index 4fd2737af37e..e06c8bbced20 100644 --- a/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/metastore/TestDuckLakeSnapshotChanges.java +++ b/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/metastore/TestDuckLakeSnapshotChanges.java @@ -13,8 +13,11 @@ */ package io.trino.plugin.ducklake.metastore; +import io.trino.spi.TrinoException; import org.junit.jupiter.api.Test; +import static io.trino.plugin.ducklake.DuckLakeErrorCode.DUCKLAKE_INVALID_METADATA; +import static io.trino.plugin.ducklake.DuckLakeErrorCode.DUCKLAKE_UNSUPPORTED_CHANGE_TYPE; import static io.trino.plugin.ducklake.metastore.DuckLakeSnapshotChanges.parse; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; @@ -120,19 +123,33 @@ void testEmptyChanges() /** * A change type this connector does not know was written by a newer DuckLake version. Ignoring - * it could drop the work of the writer that made it, so it is rejected instead. + * it could drop the work of the writer that made it, so it is rejected instead, and named so + * that the remedy is obvious. */ @Test void testUnknownChangeTypeIsRejected() { assertThatThrownBy(() -> parse("teleported_table:1")) - .isInstanceOf(IllegalArgumentException.class) - .hasMessage("Unknown DuckLake change type: teleported_table"); + .isInstanceOf(TrinoException.class) + .matches(failure -> ((TrinoException) failure).getErrorCode().equals(DUCKLAKE_UNSUPPORTED_CHANGE_TYPE.toErrorCode())) + .hasMessageContaining("recorded the DuckLake change type 'teleported_table'") + .hasMessageContaining("Upgrade the DuckLake connector"); + } + + /** + * A row that is not shaped like a change at all is corruption rather than a newer writer, and + * says so with a different error code. + */ + @Test + void testCorruptChangeEntryIsInvalidMetadata() + { assertThatThrownBy(() -> parse("dropped_table:not_a_number")) - .isInstanceOf(IllegalArgumentException.class) - .hasMessage("Malformed DuckLake object identifier: not_a_number"); + .isInstanceOf(TrinoException.class) + .matches(failure -> ((TrinoException) failure).getErrorCode().equals(DUCKLAKE_INVALID_METADATA.toErrorCode())) + .hasMessageContaining("Malformed DuckLake object identifier: not_a_number"); assertThatThrownBy(() -> parse("dropped_table")) - .isInstanceOf(IllegalArgumentException.class) - .hasMessage("Malformed DuckLake change entry: dropped_table"); + .isInstanceOf(TrinoException.class) + .matches(failure -> ((TrinoException) failure).getErrorCode().equals(DUCKLAKE_INVALID_METADATA.toErrorCode())) + .hasMessageContaining("Malformed DuckLake change entry: dropped_table"); } }