DuckLake: make CREATE OR REPLACE TABLE atomic; end sort orders on drop

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 <sql-globally-available>` and {ref}`read
+operation <sql-read-operations>` 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.

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<ConnectorTableCredentials> 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<DuckLakeWriteColumn> columns,
+            List<DuckLakePartitioning.Field> 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);

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.
+ * <p>
+ * 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.
+ * <p>
+ * 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<DuckLakeWriteColumn> columns,
+        @JsonProperty List<DuckLakePartitioning.Field> partitionFields,
+        @JsonProperty Map<Long, String> columnComments,
+        @JsonProperty Optional<String> 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();
+    }
+}

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

@@ -63,9 +63,9 @@ public final class DuckLakeWritePartitioner
     private final List<Type> sourceTypes;
     private final List<Type> partitionTypes;
 
-    public DuckLakeWritePartitioner(DuckLakePartitioning partitioning, List<DuckLakeWriteColumn> columns)
+    public DuckLakeWritePartitioner(List<DuckLakePartitioning.Field> fields, List<DuckLakeWriteColumn> 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());

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

@@ -31,7 +31,9 @@
  * <p>
  * 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,

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

@@ -50,14 +50,15 @@
     private final long baseSnapshotId;
     private final long snapshotId;
     private final long baseSchemaVersion;
+    private final boolean sortInfoSupported;
 
     private long nextCatalogId;
     private long nextFileId;
     private long schemaVersion;
     private final List<String> changes = new ArrayList<>();
     private final Set<Long> 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 @@
         this.schemaVersion = state.schemaVersion();
         this.nextCatalogId = state.nextCatalogId();
         this.nextFileId = state.nextFileId();
+        this.sortInfoSupported = sortInfoSupported;
     }
 
     /**
@@ -184,20 +186,21 @@
     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();
     }
 
@@ -212,7 +215,7 @@
 
     public void insertTableRow(long tableId, long schemaId, String tableName, String path, Optional<String> tableUuid)
     {
         String uuidExpression = tableUuid.map(_ -> ":tableUuid" + uuidCast()).orElseGet(DuckLakeCommit::randomUuid);
         var update = handle.createUpdate(
                         """
                         INSERT INTO %s (table_id, table_uuid, begin_snapshot, end_snapshot, schema_id, table_name, path, path_is_relative)
@@ -233,8 +236,12 @@
 
     /**
      * 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.
+     * <p>
+     * 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)
     {
@@ -244,6 +251,9 @@
         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<TableIdentity> findTable(String schemaName, String tableName)
@@ -274,15 +284,16 @@
     {
         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)
                 .bind("schemaName", schemaName.toLowerCase(ENGLISH))
                 .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 +895,7 @@
 
     public record TableIdentity(long tableId, long schemaId, String tableName, String path, Optional<String> 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) {}
 

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> T commit(DuckLakeCommitAction<T> 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;