From a6c43c13ea869bb2aa4c7fa15800215e58c11be7 Mon Sep 17 00:00:00 2001 From: James Greenhill Date: Thu, 3 Sep 2026 23:13:43 +0000 Subject: [PATCH 1/3] Make CREATE OR REPLACE TABLE atomic in DuckLake Both forms of the statement, the DDL one and the CTAS one, used to fail with "This connector does not support replacing tables". A writer that wanted to rebuild a table had to run DELETE and then INSERT, which is two DuckLake snapshots with an empty table visible in between. Anything reading the table at that moment reads nothing. A replace is now one snapshot, as it is in DuckDB. Within it the table that holds the name is ended -- its row, its columns, its partitioning, its data and delete files, its tags and its statistics, exactly what DROP TABLE ends -- and the table that takes the name is created. A reader at the snapshot before sees the old rows and a reader at the snapshot after sees the new ones. DuckDB does the same, in DuckLakeSchemaEntry::HandleCreateConflict: it drops the existing entry and creates a new one with a fresh catalog id, both staged in one transaction and committed as one snapshot. So the replacement is a new table, with a new identifier, and nothing of the old one carries over. This connector matches that: a comment or a partitioning the new definition does not state is gone with the old table. It differs in one place, which is the data directory. DuckDB derives it from a fresh table UUID, while this connector names it after the table, so a replaced table keeps writing into the directory the old one used. Files there are named by UUID and never collide, and a drop followed by a create of the same name already behaves this way. The CTAS form cannot create its table up front the way the plain form does, because the name still has to resolve to the table being replaced while the rows are written. It carries a DuckLakeReplaceTarget through the write instead, which holds only what the workers need, and finishCreateTable makes the one commit that ends the old table, creates the new one and registers the files. A commit that loses a race is still discarded and replayed against the newer state, as every other commit here is. Views keep working across a replace, because a view refers to the table by name. Claude-Session: https://claude.ai/code/session_01WCY5Jf2BQPCVKJTZU1TpEe --- docs/src/main/sphinx/connector/ducklake.md | 56 ++++- .../plugin/ducklake/DuckLakeMetadata.java | 225 ++++++++++++++---- .../ducklake/DuckLakePageSinkProvider.java | 24 +- .../ducklake/DuckLakeReplaceTarget.java | 68 ++++++ .../ducklake/DuckLakeWritePartitioner.java | 4 +- .../plugin/ducklake/DuckLakeWriteTarget.java | 4 +- .../ducklake/metastore/DuckLakeCommit.java | 14 +- .../plugin/ducklake/TestDuckLakeWrites.java | 160 +++++++++++++ 8 files changed, 489 insertions(+), 66 deletions(-) create mode 100644 plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/DuckLakeReplaceTarget.java diff --git a/docs/src/main/sphinx/connector/ducklake.md b/docs/src/main/sphinx/connector/ducklake.md index fdd2cf3d1cc4..e9bf2f679fbf 100644 --- a/docs/src/main/sphinx/connector/ducklake.md +++ b/docs/src/main/sphinx/connector/ducklake.md @@ -11,9 +11,11 @@ reads the metadata directly from the catalog database (PostgreSQL) and scans the Parquet data files with Trino's native Parquet reader, so queries run distributed across the cluster without going through DuckDB. -The connector is read-only. Tables are written by DuckDB (or other DuckLake -writers) and queried from Trino. Rows removed with `DELETE` in DuckDB are -recorded in positional delete files, which the connector applies when reading. +The connector reads and writes the same catalog DuckDB does. A table written +from Trino is a DuckLake table like any other, so DuckDB (or another DuckLake +writer) reads it, and Trino reads what those engines write. Rows removed with +`DELETE` are recorded in positional delete files, which every engine applies +when reading. ## Requirements @@ -208,12 +210,54 @@ delete files. Each split accounts for its own reads, including repeated reads of the same delete file by different splits. These statistics do not include coordinator-side reads performed while committing changes. +## SQL support + +Beyond the {ref}`globally available ` and {ref}`read +operation ` statements, the connector supports: + +- {doc}`/sql/insert` +- {doc}`/sql/update` +- {doc}`/sql/delete` +- {doc}`/sql/merge` +- {doc}`/sql/truncate` +- {doc}`/sql/create-table`, including `CREATE OR REPLACE TABLE` +- {doc}`/sql/create-table-as`, including `CREATE OR REPLACE TABLE ... AS` +- {doc}`/sql/drop-table` +- {doc}`/sql/alter-table` +- {doc}`/sql/comment` +- {doc}`/sql/create-schema`, {doc}`/sql/drop-schema`, and {doc}`/sql/alter-schema` +- {ref}`sql-view-management` + +Every statement writes one DuckLake snapshot, whatever it changes. A +`CREATE OR REPLACE TABLE` therefore ends the table that held the name and +creates the one that takes it in a single snapshot: a reader sees the rows the +table held before, or the rows it holds now, and never a table with no rows in +it. As in DuckDB, the table that takes the name is a new table. Nothing of the +old one carries over, so a comment or a `partitioning` property that the new +definition does not state is gone. + ## Limitations -- The connector is read-only; `INSERT`, `UPDATE`, `DELETE`, `MERGE`, and DDL - statements are not supported. - Each query reads at the latest catalog snapshot committed when the query - starts; time travel with `FOR VERSION AS OF` is not yet supported. + starts; time travel with `FOR VERSION AS OF` is not supported. +- `INSERT` writes every column of the table. A statement that names a subset of + the columns fails, because each data file holds all of them. +- `ALTER TABLE ... ADD COLUMN` adds the column at the end of the table. A + position given with `FIRST` or `AFTER` fails, because the position of a + column is what identifies it in the data files already written. +- `ALTER TABLE ... ALTER COLUMN ... SET DATA TYPE` only widens a type, to one + that every value already written reads back as: `TINYINT` to `SMALLINT`, + `INTEGER` or `BIGINT`, `SMALLINT` to `INTEGER` or `BIGINT`, `INTEGER` to + `BIGINT`, and `REAL` to `DOUBLE`. DuckLake does not rewrite data files for a + type change. +- `ALTER TABLE ... DROP COLUMN` cannot drop the only column of a table. +- `DROP SCHEMA ... CASCADE` is not supported; drop the tables of the schema + first. +- `partitioning` is the only supported table property, and views take no + properties. +- A view written in another engine's dialect is listed but cannot be queried + from Trino, because its query text does not parse here. +- A column with a non-`NULL` default value is not supported. - Queries on a column whose name mapping reads the values from a Hive partition in the file path (`ducklake_name_mapping.is_partition`), or maps the fields nested inside the column, fail. Other columns of such a table can be read. 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..22d59bf955be 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 @@ -260,7 +260,7 @@ public void renameSchema(ConnectorSession session, String source, String target) } commit.endSchema(schema.schemaId()); // the location keeps the original name, matching how DuckDB renames a schema - commit.insertSchemaRow(schema.schemaId(), target, schema.path()); + commit.insertSchemaRow(schema.schemaId(), target, schema.path(), schema.pathIsRelative()); commit.recordCreatedSchema(target); return null; }); @@ -399,22 +399,18 @@ public ColumnMetadata getColumnMetadata(ConnectorSession session, ConnectorTable @Override public void createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, SaveMode saveMode) { - SchemaTableName tableName = tableMetadata.getTable(); - commit(commit -> { - NewTable table = createTable(commit, tableName, tableMetadata.getColumns(), DuckLakeTableProperties.getPartitioning(tableMetadata.getProperties()), saveMode); - if (table != null) { - tableMetadata.getComment().ifPresent(comment -> commit.setTableTag(table.tableId(), COMMENT_TAG_KEY, Optional.of(comment))); - } - return null; - }); + commit(commit -> createTable(commit, tableMetadata, saveMode)); } /** - * Creates the catalog rows of a new table and returns its identifier and location. Shared by - * {@code CREATE TABLE} and {@code CREATE TABLE AS}, which then writes data into it. + * Creates the catalog rows of a new table and returns its identifier and location, or null when + * the table already exists and {@code IGNORE} leaves it alone. Shared by {@code CREATE TABLE} + * and {@code CREATE TABLE AS}, which then writes data into it. */ - private NewTable createTable(DuckLakeCommit commit, SchemaTableName tableName, List columns, List partitionKeys, SaveMode saveMode) + @Nullable + private NewTable createTable(DuckLakeCommit commit, ConnectorTableMetadata tableMetadata, SaveMode saveMode) { + SchemaTableName tableName = tableMetadata.getTable(); DuckLakeCommit.SchemaIdentity schema = commit.findSchema(tableName.getSchemaName()) .orElseThrow(() -> new SchemaNotFoundException(tableName.getSchemaName())); if (commit.findView(tableName.getSchemaName(), tableName.getTableName()).isPresent()) { @@ -427,29 +423,64 @@ private NewTable createTable(DuckLakeCommit commit, SchemaTableName tableName, L case IGNORE -> { return null; } - case REPLACE -> throw new TrinoException(NOT_SUPPORTED, "This connector does not support replacing tables"); + // the table being replaced ends in the snapshot the new one begins in, so the name + // resolves to the old table or to the new one, never to nothing + case REPLACE -> endTable(commit, existing.get().tableId()); } } + List columns = DuckLakeColumns.assignColumnIds(tableMetadata.getColumns()); + return insertTable( + commit, + schema, + tableName, + columns, + columnComments(tableMetadata.getColumns(), columns), + tableMetadata.getComment(), + resolvePartitioning(columns, DuckLakeTableProperties.getPartitioning(tableMetadata.getProperties()))); + } + /** + * Writes the catalog rows that describe a table: the table itself, its columns, the comments on + * both, and its partitioning scheme. + */ + private NewTable insertTable( + DuckLakeCommit commit, + DuckLakeCommit.SchemaIdentity schema, + SchemaTableName tableName, + List columns, + Map columnComments, + Optional comment, + List partitionFields) + { long tableId = commit.allocateCatalogId(); String tablePath = directoryName(tableName.getTableName()); commit.insertTableRow(tableId, schema.schemaId(), tableName.getTableName(), tablePath, Optional.empty()); - List writeColumns = DuckLakeColumns.assignColumnIds(columns); - for (DuckLakeColumnRow row : DuckLakeColumns.toColumnRows(writeColumns)) { + for (DuckLakeColumnRow row : DuckLakeColumns.toColumnRows(columns)) { commit.insertColumn(tableId, row); } - for (int i = 0; i < columns.size(); i++) { - Optional comment = columns.get(i).getComment(); + columnComments.forEach((columnId, columnComment) -> commit.setColumnTag(tableId, columnId, COMMENT_TAG_KEY, Optional.of(columnComment))); + comment.ifPresent(value -> commit.setTableTag(tableId, COMMENT_TAG_KEY, Optional.of(value))); + Optional partitioning = insertPartitioning(commit, tableId, partitionFields); + commit.recordCreatedTable(tableName.getSchemaName(), tableName.getTableName(), tableId); + + String schemaLocation = PathResolver.resolve(dataPath, schema.path(), schema.pathIsRelative()); + return new NewTable(tableId, columns, PathResolver.resolve(schemaLocation, tablePath, true), partitioning); + } + + /** + * The comment of each column that has one, by column identifier. + */ + private static Map columnComments(List columns, List writeColumns) + { + ImmutableMap.Builder comments = ImmutableMap.builder(); + for (int index = 0; index < columns.size(); index++) { + Optional comment = columns.get(index).getComment(); if (comment.isPresent()) { - commit.setColumnTag(tableId, writeColumns.get(i).columnId(), COMMENT_TAG_KEY, comment); + comments.put(writeColumns.get(index).columnId(), comment.get()); } } - Optional partitioning = createPartitioning(commit, tableId, writeColumns, partitionKeys); - commit.recordCreatedTable(tableName.getSchemaName(), tableName.getTableName(), tableId); - - String schemaLocation = PathResolver.resolve(dataPath, schema.path(), true); - return new NewTable(tableId, writeColumns, PathResolver.resolve(schemaLocation, tablePath, true), partitioning); + return comments.buildOrThrow(); } @Override @@ -457,15 +488,23 @@ public void dropTable(ConnectorSession session, ConnectorTableHandle tableHandle { DuckLakeTableHandle handle = (DuckLakeTableHandle) tableHandle; commit(commit -> { - // the data files stay in place, still visible to readers of the snapshots they belonged to - commit.endTable(handle.tableId()); - commit.endTableContents(handle.tableId()); - commit.deleteTableStats(handle.tableId()); - commit.recordDroppedTable(handle.tableId()); + endTable(commit, handle.tableId()); return null; }); } + /** + * Ends every row describing a table, so that it is gone from this snapshot on. The data files + * stay in place, still visible to readers of the snapshots they belonged to. + */ + private static void endTable(DuckLakeCommit commit, long tableId) + { + commit.endTable(tableId); + commit.endTableContents(tableId); + commit.deleteTableStats(tableId); + commit.recordDroppedTable(tableId); + } + @Override public void renameTable(ConnectorSession session, ConnectorTableHandle tableHandle, SchemaTableName newTableName) { @@ -702,19 +741,45 @@ private static Optional toWriteLayout(List partiti @Override public ConnectorOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Optional layout, RetryMode retryMode, boolean replace) { + SchemaTableName tableName = tableMetadata.getTable(); if (replace) { - throw new TrinoException(NOT_SUPPORTED, "This connector does not support replacing tables"); + // The name still resolves to the table being replaced, which keeps its rows until the + // statement finishes. Creating the new table now would make the name resolve to an + // empty table in the meantime, so the whole statement is left to a single commit in + // finishCreateTable. + List columns = DuckLakeColumns.assignColumnIds(tableMetadata.getColumns()); + return new DuckLakeReplaceTarget( + tableName, + newTableLocation(tableName), + columns, + resolvePartitioning(columns, DuckLakeTableProperties.getPartitioning(tableMetadata.getProperties())), + columnComments(tableMetadata.getColumns(), columns), + tableMetadata.getComment()); } - SchemaTableName tableName = tableMetadata.getTable(); // the table is created in its own snapshot, so that the rows written afterwards land in a // table that already exists, exactly as an insert into an existing table would return commit(commit -> { - NewTable table = createTable(commit, tableName, tableMetadata.getColumns(), DuckLakeTableProperties.getPartitioning(tableMetadata.getProperties()), SaveMode.FAIL); - tableMetadata.getComment().ifPresent(comment -> commit.setTableTag(table.tableId(), COMMENT_TAG_KEY, Optional.of(comment))); + NewTable table = createTable(commit, tableMetadata, SaveMode.FAIL); return new DuckLakeWriteTarget(tableName, table.tableId(), table.location(), table.columns(), table.partitioning()); }); } + /** + * Where a table of the given name keeps its data files. The directory is named after the table + * itself, so it can be resolved before the table exists — which the replacing form of + * {@code CREATE TABLE AS} needs, because it writes its rows before the commit that creates the + * table it writes them into. + */ + private String newTableLocation(SchemaTableName tableName) + { + DuckLakeSchemaEntry schema = metastore.listSchemas(snapshotId()).stream() + .filter(entry -> entry.schemaName().equalsIgnoreCase(tableName.getSchemaName())) + .findFirst() + .orElseThrow(() -> new SchemaNotFoundException(tableName.getSchemaName())); + String schemaLocation = PathResolver.resolve(dataPath, schema.path(), schema.pathIsRelative()); + return PathResolver.resolve(schemaLocation, directoryName(tableName.getTableName()), true); + } + @Override public Optional finishCreateTable( ConnectorSession session, @@ -722,9 +787,51 @@ public Optional finishCreateTable( Collection fragments, Collection computedStatistics) { + if (tableHandle instanceof DuckLakeReplaceTarget replaceTarget) { + return finishReplaceTable(replaceTarget, fragments); + } return finishWrite((DuckLakeWriteTarget) tableHandle, fragments); } + /** + * Ends the table being replaced, creates the one that takes its name, and registers the rows + * written for it — all in one commit, so a reader sees the rows the table held before or the + * rows it holds now, and never a table that exists but holds nothing. + *

+ * The new table is a table of its own, with a new identifier, as it is in DuckDB. Nothing of + * the table it replaces carries over: a comment or a partitioning scheme the new definition + * does not state is gone with the old table. + */ + private Optional finishReplaceTable(DuckLakeReplaceTarget replaceTarget, Collection fragments) + { + List dataFiles = parseDataFiles(fragments); + SchemaTableName tableName = replaceTarget.tableName(); + commit(commit -> { + DuckLakeCommit.SchemaIdentity schema = commit.findSchema(tableName.getSchemaName()) + .orElseThrow(() -> new SchemaNotFoundException(tableName.getSchemaName())); + if (commit.findView(tableName.getSchemaName(), tableName.getTableName()).isPresent()) { + throw new TrinoException(ALREADY_EXISTS, "View already exists: " + tableName); + } + commit.findTable(tableName.getSchemaName(), tableName.getTableName()) + .ifPresent(existing -> endTable(commit, existing.tableId())); + NewTable table = insertTable( + commit, + schema, + tableName, + replaceTarget.columns(), + replaceTarget.columnComments(), + replaceTarget.comment(), + replaceTarget.partitionFields()); + if (!dataFiles.isEmpty()) { + DuckLakeWriteTarget target = new DuckLakeWriteTarget(tableName, table.tableId(), table.location(), table.columns(), table.partitioning()); + addDataFiles(commit, target, dataFiles); + commit.recordInsert(table.tableId()); + } + return null; + }); + return Optional.empty(); + } + @Override public ConnectorInsertTableHandle beginInsert(ConnectorSession session, ConnectorTableHandle tableHandle, List columns, RetryMode retryMode) { @@ -767,9 +874,7 @@ public Optional finishInsert( */ private Optional finishWrite(DuckLakeWriteTarget target, Collection fragments) { - List dataFiles = fragments.stream() - .map(fragment -> dataFileCodec.fromJson(fragment.getBytes())) - .collect(toImmutableList()); + List dataFiles = parseDataFiles(fragments); if (dataFiles.isEmpty()) { return Optional.empty(); } @@ -781,6 +886,16 @@ private Optional finishWrite(DuckLakeWriteTarget target return Optional.empty(); } + /** + * The data files the workers report having written, one per fragment. + */ + private List parseDataFiles(Collection fragments) + { + return fragments.stream() + .map(fragment -> dataFileCodec.fromJson(fragment.getBytes())) + .collect(toImmutableList()); + } + /** * Adds data files to a table, numbering their rows from the table's next row identifier and * updating the statistics the catalog keeps for the whole table. @@ -1459,18 +1574,17 @@ private static String directoryName(String name) private record NewTable(long tableId, List columns, String location, Optional partitioning) {} /** - * Records the partitioning of a table being created or altered, resolving each key to the - * column it reads and checking that the transform can be applied to it. + * Resolves the partition keys of a table against its columns, checking that each transform can + * be applied to the column it reads. + *

+ * This writes nothing, so a statement that has to settle how it lays its rows out before the + * commit that records the layout can still do so. */ - private static Optional createPartitioning(DuckLakeCommit commit, long tableId, List columns, List partitionKeys) + private static List resolvePartitioning(List columns, List partitionKeys) { - if (partitionKeys.isEmpty()) { - return Optional.empty(); - } - ImmutableList.Builder catalogColumns = ImmutableList.builder(); ImmutableList.Builder fields = ImmutableList.builder(); - for (int index = 0; index < partitionKeys.size(); index++) { - DuckLakeTableProperties.PartitionKey key = DuckLakeTableProperties.parsePartitionKey(partitionKeys.get(index)); + for (String partitionKey : partitionKeys) { + DuckLakeTableProperties.PartitionKey key = DuckLakeTableProperties.parsePartitionKey(partitionKey); int channel = -1; for (int column = 0; column < columns.size(); column++) { if (columns.get(column).name().equalsIgnoreCase(key.columnName())) { @@ -1483,11 +1597,26 @@ private static Optional createPartitioning(DuckLakeCommit } DuckLakeWriteColumn column = columns.get(channel); DuckLakeWritePartitioner.validateTransform(key.transform(), column.name(), column.type()); - catalogColumns.add(new DuckLakePartitionColumn(index, column.columnId(), key.transform())); fields.add(new DuckLakePartitioning.Field(channel, column.columnId(), column.name(), key.transform())); } - long partitionId = commit.insertPartitioning(tableId, catalogColumns.build()); - return Optional.of(new DuckLakePartitioning(partitionId, fields.build())); + return fields.build(); + } + + /** + * Records the partitioning of a table being created or altered, which the data files written + * from now on are filed under. + */ + private static Optional insertPartitioning(DuckLakeCommit commit, long tableId, List fields) + { + if (fields.isEmpty()) { + return Optional.empty(); + } + ImmutableList.Builder catalogColumns = ImmutableList.builder(); + for (int index = 0; index < fields.size(); index++) { + DuckLakePartitioning.Field field = fields.get(index); + catalogColumns.add(new DuckLakePartitionColumn(index, field.columnId(), field.transform())); + } + return Optional.of(new DuckLakePartitioning(commit.insertPartitioning(tableId, catalogColumns.build()), fields)); } @Override @@ -1508,7 +1637,7 @@ public void setTableProperties(ConnectorSession session, ConnectorTableHandle ta // data files already written keep the partitioning they were written with, which the // read path notices and stops using partition values to prune with commit.endPartitioning(handle.tableId()); - createPartitioning(commit, handle.tableId(), columns, partitionKeys); + insertPartitioning(commit, handle.tableId(), resolvePartitioning(columns, partitionKeys)); commit.recordAlteredTable(handle.tableId()); return null; }); diff --git a/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/DuckLakePageSinkProvider.java b/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/DuckLakePageSinkProvider.java index 98a0f7ef356b..ec623d1ad0ab 100644 --- a/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/DuckLakePageSinkProvider.java +++ b/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/DuckLakePageSinkProvider.java @@ -13,6 +13,7 @@ */ package io.trino.plugin.ducklake; +import com.google.common.collect.ImmutableList; import com.google.inject.Inject; import io.airlift.json.JsonCodec; import io.trino.filesystem.TrinoFileSystemFactory; @@ -30,6 +31,7 @@ import io.trino.spi.connector.ConnectorTransactionHandle; import io.trino.spi.connector.MemoryContext; +import java.util.List; import java.util.Optional; import static io.trino.plugin.ducklake.DuckLakeSessionProperties.getTargetMaxFileSize; @@ -70,6 +72,9 @@ public ConnectorPageSink createPageSink( Optional tableCredentials, ConnectorPageSinkId pageSinkId) { + if (tableHandle instanceof DuckLakeReplaceTarget target) { + return createPageSink(session, target.tableLocation(), target.columns(), target.partitionFields()); + } return createPageSink(session, (DuckLakeWriteTarget) tableHandle); } @@ -102,15 +107,28 @@ public ConnectorMergeSink createMergeSink( } private ConnectorPageSink createPageSink(ConnectorSession session, DuckLakeWriteTarget target) + { + return createPageSink( + session, + target.tableLocation(), + target.columns(), + target.partitioning().map(DuckLakePartitioning::fields).orElseGet(ImmutableList::of)); + } + + private ConnectorPageSink createPageSink( + ConnectorSession session, + String tableLocation, + List columns, + List partitionFields) { return new DuckLakePageSink( session, fileSystemFactory.create(session), writerFactory, pageIndexerFactory, - DuckLakeParquetSchema.create(target.columns()), - target.partitioning().map(partitioning -> new DuckLakeWritePartitioner(partitioning, target.columns())), - target.tableLocation(), + DuckLakeParquetSchema.create(columns), + partitionFields.isEmpty() ? Optional.empty() : Optional.of(new DuckLakeWritePartitioner(partitionFields, columns)), + tableLocation, getTargetMaxFileSize(session).toBytes(), maxOpenPartitions, dataFileCodec); diff --git a/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/DuckLakeReplaceTarget.java b/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/DuckLakeReplaceTarget.java new file mode 100644 index 000000000000..476c470828ba --- /dev/null +++ b/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/DuckLakeReplaceTarget.java @@ -0,0 +1,68 @@ +/* + * 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; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonProperty; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; +import io.trino.spi.connector.ConnectorOutputTableHandle; +import io.trino.spi.connector.SchemaTableName; + +import java.util.List; +import java.util.Map; +import java.util.Optional; + +import static java.util.Objects.requireNonNull; + +/** + * The table a {@code CREATE OR REPLACE TABLE ... AS SELECT} is about to define, which does not + * exist in the catalog yet. + *

+ * The name it takes still resolves to the table being replaced, holding the rows it has always + * held. Creating the new table in its own snapshot first, the way {@link DuckLakeWriteTarget} does, + * would leave the name resolving to an empty table until the rows landed. So the definition travels + * with the write instead, and the commit that registers the data files is the one that ends the old + * table and creates this one. + *

+ * Nothing here identifies a catalog object: the table identifier and the identifier of its + * partitioning scheme are drawn by that commit. Only what the workers need to write the files is + * settled in advance, which is the column layout and the location the files go to. + */ +public record DuckLakeReplaceTarget( + @JsonProperty SchemaTableName tableName, + @JsonProperty String tableLocation, + @JsonProperty List columns, + @JsonProperty List partitionFields, + @JsonProperty Map columnComments, + @JsonProperty Optional comment) + implements ConnectorOutputTableHandle +{ + @JsonCreator + public DuckLakeReplaceTarget + { + requireNonNull(tableName, "tableName is null"); + requireNonNull(tableLocation, "tableLocation is null"); + columns = ImmutableList.copyOf(columns); + partitionFields = ImmutableList.copyOf(partitionFields); + columnComments = ImmutableMap.copyOf(columnComments); + requireNonNull(comment, "comment is null"); + } + + @Override + public String toString() + { + return tableName.toString(); + } +} diff --git a/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/DuckLakeWritePartitioner.java b/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/DuckLakeWritePartitioner.java index a9242fb9258c..7b944ad15713 100644 --- a/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/DuckLakeWritePartitioner.java +++ b/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/DuckLakeWritePartitioner.java @@ -63,9 +63,9 @@ public final class DuckLakeWritePartitioner private final List sourceTypes; private final List partitionTypes; - public DuckLakeWritePartitioner(DuckLakePartitioning partitioning, List columns) + public DuckLakeWritePartitioner(List fields, List columns) { - this.fields = ImmutableList.copyOf(requireNonNull(partitioning, "partitioning is null").fields()); + this.fields = ImmutableList.copyOf(requireNonNull(fields, "fields is null")); this.sourceTypes = fields.stream() .map(field -> columns.get(field.sourceChannel()).type()) .collect(toImmutableList()); diff --git a/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/DuckLakeWriteTarget.java b/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/DuckLakeWriteTarget.java index 3302082b9680..198de1763290 100644 --- a/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/DuckLakeWriteTarget.java +++ b/plugin/trino-ducklake/src/main/java/io/trino/plugin/ducklake/DuckLakeWriteTarget.java @@ -31,7 +31,9 @@ *

* The same description serves {@code INSERT} and {@code CREATE TABLE AS}. For the latter the table * already exists in the catalog by the time rows are written — the statement creates it in its own - * snapshot first — so nothing distinguishes the two on the write side. + * snapshot first — so nothing distinguishes the two on the write side. The replacing form of + * {@code CREATE TABLE AS} cannot create its table in advance, and uses {@link DuckLakeReplaceTarget} + * instead. */ public record DuckLakeWriteTarget( @JsonProperty SchemaTableName tableName, 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..9b901790dad8 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 @@ -184,20 +184,21 @@ private void bumpSchemaVersion() public long insertSchema(String schemaName, String path) { long schemaId = allocateCatalogId(); - insertSchemaRow(schemaId, schemaName, path); + insertSchemaRow(schemaId, schemaName, path, true); return schemaId; } - public void insertSchemaRow(long schemaId, String schemaName, String path) + public void insertSchemaRow(long schemaId, String schemaName, String path, boolean pathIsRelative) { handle.createUpdate( """ INSERT INTO %s (schema_id, schema_uuid, begin_snapshot, end_snapshot, schema_name, path, path_is_relative) - VALUES (:schemaId, %s, :snapshot, NULL, :schemaName, :path, true)""".formatted(table("ducklake_schema"), randomUuid())) + VALUES (:schemaId, %s, :snapshot, NULL, :schemaName, :path, :pathIsRelative)""".formatted(table("ducklake_schema"), randomUuid())) .bind("schemaId", schemaId) .bind("snapshot", snapshotId) .bind("schemaName", schemaName) .bind("path", path) + .bind("pathIsRelative", pathIsRelative) .execute(); } @@ -274,7 +275,7 @@ public Optional findSchema(String schemaName) { return handle.createQuery( """ - SELECT schema_id, schema_name, path + SELECT schema_id, schema_name, path, path_is_relative FROM %s WHERE %s AND lower(schema_name) = :schemaName""".formatted(table("ducklake_schema"), visibleUnaliased())) .bind("snapshot", baseSnapshotId) @@ -282,7 +283,8 @@ public Optional findSchema(String schemaName) .map((rs, _) -> new SchemaIdentity( rs.getLong("schema_id"), rs.getString("schema_name"), - Optional.ofNullable(rs.getString("path")).orElse(""))) + Optional.ofNullable(rs.getString("path")).orElse(""), + rs.getBoolean("path_is_relative"))) .findFirst(); } @@ -884,7 +886,7 @@ private static String quoteName(String name) public record TableIdentity(long tableId, long schemaId, String tableName, String path, Optional tableUuid) {} - public record SchemaIdentity(long schemaId, String schemaName, String path) {} + public record SchemaIdentity(long schemaId, String schemaName, String path, boolean pathIsRelative) {} public record ViewIdentity(long viewId, String dialect) {} diff --git a/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/TestDuckLakeWrites.java b/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/TestDuckLakeWrites.java index 5b50de387a52..ad7953d83d53 100644 --- a/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/TestDuckLakeWrites.java +++ b/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/TestDuckLakeWrites.java @@ -187,6 +187,135 @@ void testCreateTableAsSelect() } } + @Test + void testReplaceTableDefinitionKeepsTheTableReadableThroughout() + { + String table = "replace_ddl_" + randomNameSuffix(); + assertUpdate("CREATE TABLE " + table + " AS SELECT id, 'old' AS tag FROM UNNEST(sequence(1, 3)) AS t(id)", 3); + try { + String replacedTableId = tableId(table); + assertUpdate("CREATE OR REPLACE TABLE " + table + " (n INTEGER COMMENT 'the only column') COMMENT 'replaced'"); + + assertThat(replacedTableAtSnapshot(table, replacedTableId)) + .isEqualTo("dropped_table:%s,created_table:\"main\".\"%s\"".formatted(replacedTableId, table)); + assertQuery("SELECT count(*) FROM " + table, "VALUES 0"); + assertQuery("SELECT column_name FROM information_schema.columns WHERE table_name = '" + table + "'", "VALUES 'n'"); + assertThat((String) computeScalar("SHOW CREATE TABLE " + table)).contains("COMMENT 'replaced'", "COMMENT 'the only column'"); + assertThat(duckDbRows("SELECT column_name FROM duckdb_columns() WHERE table_name = '" + table + "'")).isEqualTo(List.of("n")); + + // the table that took the name is an ordinary table, which rows can be added to + assertUpdate("INSERT INTO " + table + " VALUES 7", 1); + assertQuery("SELECT n FROM " + table, "VALUES 7"); + assertThat(duckDbScalar("SELECT n::VARCHAR FROM " + table)).isEqualTo("7"); + } + finally { + assertUpdate("DROP TABLE " + table); + } + } + + @Test + void testReplaceTableAsSelectSwapsTheRowsInOneSnapshot() + { + String table = "replace_ctas_" + randomNameSuffix(); + assertUpdate("CREATE TABLE " + table + " AS SELECT id FROM UNNEST(sequence(1, 3)) AS t(id)", 3); + try { + String replacedTableId = tableId(table); + String replacedFile = duckDbScalar("SELECT path FROM __ducklake_metadata_lake.ducklake_data_file WHERE table_id = " + replacedTableId); + + assertUpdate("CREATE OR REPLACE TABLE " + table + " AS SELECT id, id * 10 AS scaled FROM UNNEST(sequence(1, 5)) AS t(id)", 5); + + // the one snapshot ends the old table, creates the new one and fills it, in the order + // DuckDB records the same three changes in + assertThat(replacedTableAtSnapshot(table, replacedTableId)) + .isEqualTo("dropped_table:%s,created_table:\"main\".\"%s\",inserted_into_table:%s".formatted(replacedTableId, table, tableId(table))); + assertQuery("SELECT count(*), sum(scaled) FROM " + table, "VALUES (5, 150)"); + assertThat(duckDbScalar("SELECT sum(scaled) FROM " + table)).isEqualTo("150"); + + // the files of the table that was replaced are ended rather than removed, so a reader + // of the snapshots they belonged to still finds them where they always were + assertThat(duckDbScalar( + "SELECT count(*) FROM __ducklake_metadata_lake.ducklake_data_file WHERE table_id = %s AND end_snapshot IS NOT NULL".formatted(replacedTableId))) + .isEqualTo("1"); + assertThat(catalog.dataPath().resolve("main").resolve(table).resolve(replacedFile)).exists(); + } + finally { + assertUpdate("DROP TABLE " + table); + } + } + + @Test + void testReplacingATableKeepsNothingOfTheOldDefinition() + { + String table = "replace_reset_" + randomNameSuffix(); + assertUpdate("CREATE TABLE " + table + " (id BIGINT, region VARCHAR) COMMENT 'first' WITH (partitioning = ARRAY['region'])"); + try { + assertUpdate("INSERT INTO %s VALUES (1, 'us'), (2, 'eu')".formatted(table), 2); + + // neither the comment nor the partitioning of the old table carries over + assertUpdate("CREATE OR REPLACE TABLE " + table + " AS SELECT 'x' AS region", 1); + assertThat((String) computeScalar("SHOW CREATE TABLE " + table)).doesNotContain("partitioning", "COMMENT 'first'"); + assertQuery("SELECT region FROM " + table, "VALUES 'x'"); + + // a partitioning the new definition does state files the rows it writes + assertUpdate( + "CREATE OR REPLACE TABLE %s WITH (partitioning = ARRAY['region']) AS SELECT * FROM (VALUES 'us', 'eu', 'us') AS t(region)".formatted(table), + 3); + assertThat((String) computeScalar("SHOW CREATE TABLE " + table)).contains("partitioning = ARRAY['region']"); + assertQuery("SELECT count(*) FROM " + table + " WHERE region = 'us'", "VALUES 2"); + assertThat(duckDbScalar( + "SELECT count(DISTINCT data_file_id) FROM __ducklake_metadata_lake.ducklake_file_partition_value WHERE table_id = " + tableId(table))) + .isEqualTo("2"); + assertThat(duckDbScalar("SELECT count(*) FROM " + table)).isEqualTo("3"); + } + finally { + assertUpdate("DROP TABLE " + table); + } + } + + @Test + void testReplacingATableThatDoesNotExistCreatesIt() + { + String ddl = "replace_new_ddl_" + randomNameSuffix(); + String ctas = "replace_new_ctas_" + randomNameSuffix(); + String empty = "replace_new_empty_" + randomNameSuffix(); + assertUpdate("CREATE OR REPLACE TABLE " + ddl + " (a INTEGER)"); + assertUpdate("CREATE OR REPLACE TABLE " + ctas + " AS SELECT 1 AS a", 1); + // a query that selects nothing still leaves the table behind, with no rows in it + assertUpdate("CREATE OR REPLACE TABLE " + empty + " AS SELECT 1 AS a WHERE false", 0); + try { + assertQuery("SELECT count(*) FROM " + ddl, "VALUES 0"); + assertQuery("SELECT a FROM " + ctas, "VALUES 1"); + assertQuery("SELECT count(*) FROM " + empty, "VALUES 0"); + assertThat(duckDbScalar("SELECT a::VARCHAR FROM " + ctas)).isEqualTo("1"); + assertThat(duckDbScalar("SELECT count(*) FROM " + empty)).isEqualTo("0"); + } + finally { + assertUpdate("DROP TABLE " + ddl); + assertUpdate("DROP TABLE " + ctas); + assertUpdate("DROP TABLE " + empty); + } + } + + @Test + void testAViewOverAReplacedTableReadsTheNewRows() + { + String table = "replaced_under_view_" + randomNameSuffix(); + String view = "over_replaced_" + randomNameSuffix(); + assertUpdate("CREATE TABLE " + table + " AS SELECT id FROM UNNEST(sequence(1, 3)) AS t(id)", 3); + try { + assertUpdate("CREATE VIEW %s AS SELECT sum(id) AS total FROM %s".formatted(view, table)); + assertQuery("SELECT total FROM " + view, "VALUES 6"); + + // the view names the table, not the identifier a replacement gives it + assertUpdate("CREATE OR REPLACE TABLE %s AS SELECT id FROM UNNEST(sequence(1, 10)) AS t(id)".formatted(table), 10); + assertQuery("SELECT total FROM " + view, "VALUES 55"); + } + finally { + assertUpdate("DROP VIEW " + view); + assertUpdate("DROP TABLE " + table); + } + } + @Test void testInsertIntoTableCreatedByDuckDb() throws SQLException @@ -737,6 +866,37 @@ void testTargetFileSizeRollsWritesOverIntoSeveralFiles() } } + /** + * The changes recorded for the snapshot that replaced the table, after checking that it was one + * snapshot: the row of the table that was there ends where the row of the table that took its + * name begins, and the two are different tables. + */ + private String replacedTableAtSnapshot(String tableName, String replacedTableId) + { + List rows = duckDbRows( + """ + SELECT table_id::VARCHAR, begin_snapshot::VARCHAR, coalesce(end_snapshot::VARCHAR, 'open') + FROM __ducklake_metadata_lake.ducklake_table + WHERE table_name = '%s' ORDER BY begin_snapshot""".formatted(tableName)); + assertThat(rows).hasSize(6); + assertThat(rows.getFirst()).isEqualTo(replacedTableId); + // the replacement is a table of its own, with an identifier of its own, as it is in DuckDB + assertThat(rows.get(3)).isNotEqualTo(replacedTableId); + String snapshot = rows.get(2); + assertThat(snapshot).isEqualTo(rows.get(4)); + assertThat(rows.get(5)).isEqualTo("open"); + return duckDbScalar("SELECT changes_made FROM __ducklake_metadata_lake.ducklake_snapshot_changes WHERE snapshot_id = " + snapshot); + } + + /** + * The identifier of the table currently going by the given name. + */ + private String tableId(String tableName) + { + return duckDbScalar("SELECT table_id::VARCHAR FROM __ducklake_metadata_lake.ducklake_table " + + "WHERE table_name = '" + tableName + "' AND end_snapshot IS NULL"); + } + private String duckDbScalar(@Language("SQL") String sql) { List row = duckDbRows(sql); From f073a5439747bee2224806b4f04eeb0dac669674 Mon Sep 17 00:00:00 2001 From: James Greenhill Date: Thu, 3 Sep 2026 23:37:19 +0000 Subject: [PATCH 2/3] Retry a DuckDB statement whose commit lost a race Test methods of a class run in parallel against one catalog, and several of them write to it from DuckDB while others write to it from Trino. Every change to a DuckLake catalog claims the next snapshot, so two writers that start from the same one conflict. The connector retries such a commit. DuckDB reports it, and the test that ran the statement fails. That was already possible, and it got likelier with five more tests writing to the same catalog: TestDuckLakeWrites failed twice in seven runs, each time in a different test, always on a DuckDB statement. So the fixture now runs such a statement again, which is what a writer facing the conflict would do. A failed commit leaves the connection unusable, so each attempt after the first opens a new one. Options an earlier statement set survive that, because DuckLake keeps them in the catalog rather than on the connection. TestDuckLakeWrites and TestDuckLakeReads ran eight times each without a failure afterwards. Claude-Session: https://claude.ai/code/session_01WCY5Jf2BQPCVKJTZU1TpEe --- .../plugin/ducklake/TestDuckLakeWrites.java | 2 +- .../ducklake/TestingDuckLakeCatalog.java | 50 +++++++++++++++++-- 2 files changed, 48 insertions(+), 4 deletions(-) diff --git a/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/TestDuckLakeWrites.java b/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/TestDuckLakeWrites.java index ad7953d83d53..57183798e775 100644 --- a/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/TestDuckLakeWrites.java +++ b/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/TestDuckLakeWrites.java @@ -188,7 +188,7 @@ void testCreateTableAsSelect() } @Test - void testReplaceTableDefinitionKeepsTheTableReadableThroughout() + void testReplaceTableDefinitionInOneSnapshot() { String table = "replace_ddl_" + randomNameSuffix(); assertUpdate("CREATE TABLE " + table + " AS SELECT id, 'old' AS tag FROM UNNEST(sequence(1, 3)) AS t(id)", 3); diff --git a/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/TestingDuckLakeCatalog.java b/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/TestingDuckLakeCatalog.java index c71de2c3c226..c18dcae27db3 100644 --- a/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/TestingDuckLakeCatalog.java +++ b/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/TestingDuckLakeCatalog.java @@ -26,7 +26,9 @@ import java.sql.ResultSet; import java.sql.SQLException; import java.sql.Statement; +import java.time.Duration; +import static com.google.common.base.Strings.nullToEmpty; import static com.google.common.base.Verify.verify; import static io.trino.plugin.ducklake.metastore.DuckLakeMetastoreConnectionFactory.APPLICATION_NAME; import static java.lang.String.format; @@ -42,6 +44,9 @@ public final class TestingDuckLakeCatalog public static final String USER = "test"; public static final String PASSWORD = "test"; private static final String DATABASE = "lakedb"; + private static final String COMMIT_CONFLICT_MESSAGE = "Failed to commit DuckLake transaction"; + private static final int COMMIT_ATTEMPTS = 5; + private static final Duration COMMIT_RETRY_DELAY = Duration.ofMillis(100); private final PostgreSQLContainer dockerContainer; private final Path dataPath; @@ -146,15 +151,54 @@ public long connectorConnectionCount() } } + /** + * Runs the statements in DuckDB, in order, running one again when its DuckLake commit lost a + * race against another writer. + *

+ * Every change to a DuckLake catalog claims the next snapshot, so a commit conflicts with any + * other commit that started from the same one. The connector retries such a commit; DuckDB + * reports it and leaves the connection it came from unusable. Tests write to one catalog from + * both engines at the same time, so a statement that loses the race runs again on a new + * connection, which is what a writer facing the conflict would do. Options an earlier + * statement set survive the new connection, because DuckLake keeps them in the catalog rather + * than on the connection. + */ public void executeInDuckDb(@Language("SQL") String... statements) throws SQLException { - try (Connection connection = openDuckDbConnection(); - Statement statement = connection.createStatement()) { + Connection connection = openDuckDbConnection(); + try { for (String sql : statements) { - statement.execute(sql); + for (int attempt = 1; ; attempt++) { + try (Statement statement = connection.createStatement()) { + statement.execute(sql); + break; + } + catch (SQLException e) { + if (attempt == COMMIT_ATTEMPTS || !nullToEmpty(e.getMessage()).contains(COMMIT_CONFLICT_MESSAGE)) { + throw e; + } + } + connection.close(); + sleepBeforeRetry(attempt); + connection = openDuckDbConnection(); + } } } + finally { + connection.close(); + } + } + + private static void sleepBeforeRetry(int attempt) + { + try { + Thread.sleep(COMMIT_RETRY_DELAY.toMillis() * attempt); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException(e); + } } @Override From 8a905f9eb28295db6e1906d871f3ca5521719112 Mon Sep 17 00:00:00 2001 From: James Greenhill Date: Thu, 3 Sep 2026 23:58:29 +0000 Subject: [PATCH 3/3] End the sort order of a dropped or replaced table DuckDB ends ducklake_sort_info along with the rest of a table it drops, in DuckLakeMetadataManager::DropTables. This connector ended everything else that method ends and left the sort order behind, so a table DuckDB had sorted kept a live sort_info row pointing at a table id that no longer resolved. That is worse than a stale row. DuckDB validates sort entries when it attaches a catalog, so the leftover row made every later DuckDB connection fail with "Invalid Input Error: Could not find matching table for sort entry" -- the catalog could no longer be opened at all. Both tests below reproduce exactly that without the fix. ducklake_sort_info arrived in format version 0.4, and the connector still reads 0.1 catalogs, so the metastore checks for the table the way it already checks for ducklake_view and ducklake_name_mapping, and passes the answer to the commit. The check is cached, and runs outside the commit transaction so a commit still needs one connection. Rows keyed by a sort order rather than by the table, such as ducklake_sort_expression, carry no snapshot range and are left alone, as DuckDB leaves them. Two further things, both about work done for nothing: beginCreateTable now refuses to replace a name a view holds, instead of leaving that to the commit. The commit keeps the authoritative check, which is what catches a view created after this read. But every row the statement selects is written before that commit runs, and none of it could be used, so the cheap read here saves the whole write. TestDuckLakeAbsoluteSchemaPath covers reading a schema whose catalog row holds an absolute path. It needs a catalog of its own, because that path names a Trino file system DuckDB cannot resolve. Claude-Session: https://claude.ai/code/session_01WCY5Jf2BQPCVKJTZU1TpEe --- .../plugin/ducklake/DuckLakeMetadata.java | 7 ++ .../ducklake/metastore/DuckLakeCommit.java | 15 +++- .../metastore/JdbcDuckLakeMetastore.java | 19 ++++- .../TestDuckLakeAbsoluteSchemaPath.java | 70 +++++++++++++++++++ .../ducklake/TestDuckLakeLegacyCatalog.java | 11 +-- .../plugin/ducklake/TestDuckLakeWrites.java | 67 ++++++++++++++++++ .../ducklake/TestingDuckLakeCatalog.java | 16 +++++ 7 files changed, 193 insertions(+), 12 deletions(-) create mode 100644 plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/TestDuckLakeAbsoluteSchemaPath.java 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 22d59bf955be..b90339d7d5af 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 @@ -747,6 +747,13 @@ public ConnectorOutputTableHandle beginCreateTable(ConnectorSession session, Con // statement finishes. Creating the new table now would make the name resolve to an // empty table in the meantime, so the whole statement is left to a single commit in // finishCreateTable. + // + // A name a view holds is one no table can take, and the commit will say so. It is + // worth saying here too, because everything the statement selects is written before + // that commit runs, and none of it could be used. + if (metastore.findView(snapshotId(), tableName.getSchemaName(), tableName.getTableName()).isPresent()) { + throw new TrinoException(ALREADY_EXISTS, "View already exists: " + tableName); + } List columns = DuckLakeColumns.assignColumnIds(tableMetadata.getColumns()); return new DuckLakeReplaceTarget( tableName, 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 9b901790dad8..62a6b2616672 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 @@ -50,6 +50,7 @@ public final class DuckLakeCommit private final long baseSnapshotId; private final long snapshotId; private final long baseSchemaVersion; + private final boolean sortInfoSupported; private long nextCatalogId; private long nextFileId; @@ -57,7 +58,7 @@ public final class DuckLakeCommit private final List changes = new ArrayList<>(); private final Set tablesWithSchemaChange = new LinkedHashSet<>(); - DuckLakeCommit(Handle handle, String metadataSchema, SnapshotState state) + DuckLakeCommit(Handle handle, String metadataSchema, SnapshotState state, boolean sortInfoSupported) { this.handle = requireNonNull(handle, "handle is null"); this.metadataSchema = requireNonNull(metadataSchema, "metadataSchema is null"); @@ -67,6 +68,7 @@ public final class DuckLakeCommit this.schemaVersion = state.schemaVersion(); this.nextCatalogId = state.nextCatalogId(); this.nextFileId = state.nextFileId(); + this.sortInfoSupported = sortInfoSupported; } /** @@ -234,8 +236,12 @@ public void endTable(long tableId) /** * Ends every row describing the table other than the table row itself: its columns, its - * partitioning scheme, its data and delete files and its comments. Used when the table is - * dropped, so that no part of it stays visible at later snapshots. + * partitioning scheme, its sort order, its data and delete files and its comments. Used when + * the table is dropped, so that no part of it stays visible at later snapshots. + *

+ * This is the set DuckDB ends in {@code DuckLakeMetadataManager::DropTables}. The rows keyed by + * one of those, such as {@code ducklake_partition_column} and {@code ducklake_sort_expression}, + * carry no snapshot range of their own and are left alone, as DuckDB leaves them. */ public void endTableContents(long tableId) { @@ -245,6 +251,9 @@ public void endTableContents(long tableId) endRows("ducklake_delete_file", "table_id = :objectId", tableId); endRows("ducklake_column_tag", "table_id = :objectId", tableId); endRows("ducklake_tag", "object_id = :objectId", tableId); + if (sortInfoSupported) { + endRows("ducklake_sort_info", "table_id = :objectId", tableId); + } } public Optional findTable(String schemaName, String tableName) 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..c510a9b0a0ba 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 @@ -64,6 +64,7 @@ public class JdbcDuckLakeMetastore private volatile Boolean dataFileHasPartialMax; private volatile Boolean inlinedDataTablesRegistryExists; private volatile Boolean nameMappingTableExists; + private volatile Boolean sortInfoTableExists; private volatile Boolean viewTableExists; private volatile Boolean nameMappingHasIsPartition; @@ -100,11 +101,13 @@ public long currentSnapshotId() */ public T commit(DuckLakeCommitAction action) { + // read outside the transaction, so that the commit itself needs one connection only + boolean sortInfoSupported = sortInfoTableExists(); RuntimeException conflict = null; for (int attempt = 0; attempt < MAX_COMMIT_ATTEMPTS; attempt++) { try { return jdbi.inTransaction(TransactionIsolationLevel.SERIALIZABLE, handle -> { - DuckLakeCommit commit = new DuckLakeCommit(handle, metadataSchema, snapshotState(handle)); + DuckLakeCommit commit = new DuckLakeCommit(handle, metadataSchema, snapshotState(handle), sortInfoSupported); T result = action.run(commit); commit.writeSnapshot(); return result; @@ -880,6 +883,20 @@ private boolean viewTableExists() return tableExists; } + /** + * Whether the catalog records sort orders. {@code ducklake_sort_info} was added to the format + * in version 0.4, so a catalog written against an older one does not have the table at all. + */ + private boolean sortInfoTableExists() + { + Boolean tableExists = sortInfoTableExists; + if (tableExists == null) { + tableExists = tableExists("ducklake_sort_info"); + sortInfoTableExists = tableExists; + } + return tableExists; + } + private boolean nameMappingTableExists() { Boolean tableExists = nameMappingTableExists; diff --git a/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/TestDuckLakeAbsoluteSchemaPath.java b/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/TestDuckLakeAbsoluteSchemaPath.java new file mode 100644 index 000000000000..c674182ea941 --- /dev/null +++ b/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/TestDuckLakeAbsoluteSchemaPath.java @@ -0,0 +1,70 @@ +/* + * 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; + +import io.trino.testing.AbstractTestQueryFramework; +import io.trino.testing.QueryRunner; +import org.junit.jupiter.api.Test; + +import java.sql.SQLException; + +import static io.trino.testing.TestingNames.randomNameSuffix; +import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.jupiter.api.Assumptions.abort; + +/** + * Verifies that a schema whose catalog row holds an absolute path keeps its tables there. + *

+ * DuckLake stores each path either relative to the one above it or as an absolute path of its own. + * This connector only ever writes relative ones, so the fixture writes an absolute one by hand. + * That needs a catalog of its own, because the absolute path names a Trino file system that DuckDB + * cannot resolve, and DuckDB checks what it reads when it attaches. + */ +final class TestDuckLakeAbsoluteSchemaPath + extends AbstractTestQueryFramework +{ + private TestingDuckLakeCatalog catalog; + + @Override + protected QueryRunner createQueryRunner() + throws Exception + { + catalog = closeAfterClass(new TestingDuckLakeCatalog()); + try { + // the catalog has to exist before Trino can attach to it, and only DuckDB creates one + 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); + } + return DuckLakeQueryRunner.builder(catalog).build(); + } + + @Test + void testTablesOfSuchASchemaAreWrittenAndReadThere() + throws SQLException + { + String schema = "abs_schema_" + randomNameSuffix(); + String directory = "elsewhere_" + randomNameSuffix(); + assertUpdate("CREATE SCHEMA " + schema); + catalog.executeInMetastore("UPDATE public.ducklake_schema SET path = 'local:///%s/', path_is_relative = false WHERE schema_name = '%s'" + .formatted(directory, schema)); + + assertUpdate("CREATE TABLE %s.t AS SELECT 1 AS a".formatted(schema), 1); + assertThat(catalog.dataPath().resolve(directory).resolve("t")).isDirectory(); + assertQuery("SELECT a FROM %s.t".formatted(schema), "VALUES 1"); + assertUpdate("INSERT INTO %s.t VALUES 2".formatted(schema), 1); + assertQuery("SELECT a FROM %s.t ORDER BY a".formatted(schema), "VALUES 1, 2"); + } +} diff --git a/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/TestDuckLakeLegacyCatalog.java b/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/TestDuckLakeLegacyCatalog.java index f2444456bfd9..b47c6b86bd52 100644 --- a/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/TestDuckLakeLegacyCatalog.java +++ b/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/TestDuckLakeLegacyCatalog.java @@ -17,10 +17,7 @@ import io.trino.testing.QueryRunner; import org.junit.jupiter.api.Test; -import java.sql.Connection; -import java.sql.DriverManager; import java.sql.SQLException; -import java.sql.Statement; import static org.assertj.core.api.Assertions.assertThat; import static org.junit.jupiter.api.Assumptions.abort; @@ -57,11 +54,9 @@ protected QueryRunner createQueryRunner() private static void makeCatalogLegacy(TestingDuckLakeCatalog catalog) throws SQLException { - try (Connection connection = DriverManager.getConnection(catalog.jdbcUrl(), TestingDuckLakeCatalog.USER, TestingDuckLakeCatalog.PASSWORD); - Statement statement = connection.createStatement()) { - statement.execute("DROP TABLE IF EXISTS public.ducklake_inlined_data_tables"); - statement.execute("ALTER TABLE public.ducklake_name_mapping DROP COLUMN IF EXISTS is_partition"); - } + catalog.executeInMetastore( + "DROP TABLE IF EXISTS public.ducklake_inlined_data_tables", + "ALTER TABLE public.ducklake_name_mapping DROP COLUMN IF EXISTS is_partition"); } @Test diff --git a/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/TestDuckLakeWrites.java b/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/TestDuckLakeWrites.java index 57183798e775..ac65629e9b5e 100644 --- a/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/TestDuckLakeWrites.java +++ b/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/TestDuckLakeWrites.java @@ -316,6 +316,49 @@ void testAViewOverAReplacedTableReadsTheNewRows() } } + @Test + void testReplacingATableEndsTheSortOrderItHad() + throws SQLException + { + String table = "replace_sorted_" + randomNameSuffix(); + catalog.executeInDuckDb( + "CREATE TABLE %s (id INTEGER, v VARCHAR)".formatted(table), + "ALTER TABLE %s SET SORTED BY (id ASC)".formatted(table), + "INSERT INTO %s VALUES (2, 'b'), (1, 'a')".formatted(table)); + try { + String replacedTableId = tableId(table); + assertThat(sortOrderEndSnapshots(replacedTableId)).isEqualTo(List.of("open")); + + assertUpdate("CREATE OR REPLACE TABLE %s AS SELECT 9 AS id".formatted(table), 1); + + // the sort order ends with the table it described, in the same snapshot + assertThat(sortOrderEndSnapshots(replacedTableId)).isEqualTo(List.of(tableEndSnapshot(replacedTableId))); + // and the table that took the name has none, because its definition states none + assertThat(sortOrderEndSnapshots(tableId(table))).isEmpty(); + assertQuery("SELECT id FROM " + table, "VALUES 9"); + } + finally { + assertUpdate("DROP TABLE " + table); + } + } + + @Test + void testDroppingATableEndsTheSortOrderItHad() + throws SQLException + { + String table = "drop_sorted_" + randomNameSuffix(); + catalog.executeInDuckDb( + "CREATE TABLE %s (id INTEGER)".formatted(table), + "ALTER TABLE %s SET SORTED BY (id ASC)".formatted(table)); + String droppedTableId = tableId(table); + assertThat(sortOrderEndSnapshots(droppedTableId)).isEqualTo(List.of("open")); + + assertUpdate("DROP TABLE " + table); + + assertThat(sortOrderEndSnapshots(droppedTableId)).isEqualTo(List.of(tableEndSnapshot(droppedTableId))); + assertThat(duckDbScalar("SELECT count(*) FROM duckdb_tables() WHERE table_name = '" + table + "'")).isEqualTo("0"); + } + @Test void testInsertIntoTableCreatedByDuckDb() throws SQLException @@ -757,6 +800,12 @@ void testAViewAndATableCannotShareAName() try { assertThatThrownBy(() -> assertUpdate("CREATE TABLE %s (a INTEGER)".formatted(viewName))) .hasMessageContaining("already exists"); + + // replacing is refused before the rows are selected, so nothing is written for a + // statement that cannot finish + assertThatThrownBy(() -> assertUpdate("CREATE OR REPLACE TABLE %s AS SELECT 1 AS a".formatted(viewName), 1)) + .hasMessageContaining("already exists"); + assertThat(catalog.dataPath().resolve("main").resolve(viewName)).doesNotExist(); } finally { assertUpdate("DROP VIEW " + viewName); @@ -897,6 +946,24 @@ private String tableId(String tableName) + "WHERE table_name = '" + tableName + "' AND end_snapshot IS NULL"); } + /** + * The snapshot the row of the given table ended in, or {@code open} while it is still current. + */ + private String tableEndSnapshot(String tableId) + { + return duckDbScalar("SELECT coalesce(end_snapshot::VARCHAR, 'open') FROM __ducklake_metadata_lake.ducklake_table " + + "WHERE table_id = " + tableId); + } + + /** + * One entry per sort order the table has ever had, saying which snapshot ended it. + */ + private List sortOrderEndSnapshots(String tableId) + { + return duckDbRows("SELECT coalesce(end_snapshot::VARCHAR, 'open') FROM __ducklake_metadata_lake.ducklake_sort_info " + + "WHERE table_id = " + tableId + " ORDER BY sort_id"); + } + private String duckDbScalar(@Language("SQL") String sql) { List row = duckDbRows(sql); diff --git a/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/TestingDuckLakeCatalog.java b/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/TestingDuckLakeCatalog.java index c18dcae27db3..1c4c8cecceb4 100644 --- a/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/TestingDuckLakeCatalog.java +++ b/plugin/trino-ducklake/src/test/java/io/trino/plugin/ducklake/TestingDuckLakeCatalog.java @@ -190,6 +190,22 @@ public void executeInDuckDb(@Language("SQL") String... statements) } } + /** + * Runs the statements against the catalog database itself, rather than through DuckDB. Used to + * put the catalog into a state no engine writes, such as the one an older DuckLake version + * would have left. + */ + public void executeInMetastore(@Language("SQL") String... statements) + throws SQLException + { + try (Connection connection = DriverManager.getConnection(jdbcUrl(), USER, PASSWORD); + Statement statement = connection.createStatement()) { + for (String sql : statements) { + statement.execute(sql); + } + } + } + private static void sleepBeforeRetry(int attempt) { try {