diff --git a/docs/src/main/sphinx/connector/ducklake.md b/docs/src/main/sphinx/connector/ducklake.md index fdd2cf3d1cc4..008d01927850 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,54 @@ 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: 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..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 @@ -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; 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..8efb29e0c055 --- /dev/null +++ b/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/metastore/DuckLakeSnapshotChanges.java @@ -0,0 +1,208 @@ +/* + * 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 io.trino.spi.TrinoException; + +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 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; + +/** + * 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. + *

+ * 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) + { + Builder builder = new Builder(); + for (String change : splitChanges(changesMade)) { + int separator = change.indexOf(':'); + if (separator < 0) { + throw new TrinoException(DUCKLAKE_INVALID_METADATA, "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 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)); + } + } + + private static long objectId(String value) + { + try { + return Long.parseLong(value); + } + catch (NumberFormatException e) { + throw new TrinoException(DUCKLAKE_INVALID_METADATA, "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..9dfdc72893d1 --- /dev/null +++ b/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/metastore/TestDuckLakeConcurrentCommits.java @@ -0,0 +1,370 @@ +/* + * 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.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; +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)); + } + + /** + * 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 + * 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 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 { + 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..e06c8bbced20 --- /dev/null +++ b/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/metastore/TestDuckLakeSnapshotChanges.java @@ -0,0 +1,155 @@ +/* + * 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.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; + +/** + * 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, and named so + * that the remedy is obvious. + */ + @Test + void testUnknownChangeTypeIsRejected() + { + assertThatThrownBy(() -> parse("teleported_table:1")) + .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(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(TrinoException.class) + .matches(failure -> ((TrinoException) failure).getErrorCode().equals(DUCKLAKE_INVALID_METADATA.toErrorCode())) + .hasMessageContaining("Malformed DuckLake change entry: dropped_table"); + } +}