Support creating tables with table comment in Delta Lake

plugin/trino-accumulo/src/main/java/io/trino/plugin/accumulo/AccumuloMetadata.java

@@ -102,6 +102,9 @@ public ConnectorOutputTableHandle beginCreateTable(ConnectorSession session, Con
         if (retryMode != NO_RETRIES) {
             throw new TrinoException(NOT_SUPPORTED, "This connector does not support query retries");
         }
+        if (tableMetadata.getComment().isPresent()) {
+            throw new TrinoException(NOT_SUPPORTED, "This connector does not support creating tables with table comment");
+        }
 
         checkNoRollback();
 
@@ -136,6 +139,9 @@ private void rollbackCreateTable(AccumuloTable table)
     @Override
     public void createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, boolean ignoreExisting)
     {
+        if (tableMetadata.getComment().isPresent()) {
+            throw new TrinoException(NOT_SUPPORTED, "This connector does not support creating tables with table comment");
+        }
         client.createTable(tableMetadata);
     }
 

plugin/trino-accumulo/src/test/java/io/trino/plugin/accumulo/TestAccumuloConnectorTest.java

@@ -68,6 +68,9 @@ protected boolean hasBehavior(TestingConnectorBehavior connectorBehavior)
             case SUPPORTS_DROP_COLUMN:
                 return false;
 
+            case SUPPORTS_CREATE_TABLE_WITH_TABLE_COMMENT:
+                return false;
+
             case SUPPORTS_COMMENT_ON_TABLE:
             case SUPPORTS_COMMENT_ON_COLUMN:
                 return false;

plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/BaseJdbcClient.java

@@ -540,6 +540,9 @@ protected JdbcOutputTableHandle createTable(ConnectorSession session, ConnectorT
 
     protected String createTableSql(RemoteTableName remoteTableName, List<String> columns, ConnectorTableMetadata tableMetadata)
     {
+        if (tableMetadata.getComment().isPresent()) {
+            throw new TrinoException(NOT_SUPPORTED, "This connector does not support creating tables with table comment");
+        }
         checkArgument(tableMetadata.getProperties().isEmpty(), "Unsupported table properties: %s", tableMetadata.getProperties());
         return format("CREATE TABLE %s (%s)", quoted(remoteTableName), join(", ", columns));
     }

plugin/trino-base-jdbc/src/test/java/io/trino/plugin/jdbc/TestJdbcConnectorTest.java

@@ -66,6 +66,7 @@ protected boolean hasBehavior(TestingConnectorBehavior connectorBehavior)
             case SUPPORTS_RENAME_TABLE_ACROSS_SCHEMAS:
                 return false;
 
+            case SUPPORTS_CREATE_TABLE_WITH_TABLE_COMMENT:
             case SUPPORTS_COMMENT_ON_TABLE:
             case SUPPORTS_COMMENT_ON_COLUMN:
             case SUPPORTS_ADD_COLUMN_WITH_COMMENT:

plugin/trino-bigquery/src/main/java/io/trino/plugin/bigquery/BigQueryMetadata.java

@@ -79,6 +79,7 @@
 import static io.trino.plugin.bigquery.BigQueryTableHandle.BigQueryPartitionType.INGESTION;
 import static io.trino.plugin.bigquery.BigQueryType.toField;
 import static io.trino.plugin.bigquery.BigQueryUtil.isWildcardTable;
+import static io.trino.spi.StandardErrorCode.NOT_SUPPORTED;
 import static java.util.Locale.ENGLISH;
 import static java.util.Objects.requireNonNull;
 import static java.util.function.Function.identity;
@@ -371,6 +372,9 @@ public void dropSchema(ConnectorSession session, String schemaName)
     @Override
     public void createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, boolean ignoreExisting)
     {
+        if (tableMetadata.getComment().isPresent()) {
+            throw new TrinoException(NOT_SUPPORTED, "This connector does not support creating tables with table comment");
+        }
         try {
             createTable(session, tableMetadata);
         }

plugin/trino-bigquery/src/test/java/io/trino/plugin/bigquery/TestBigQueryConnectorTest.java

@@ -69,6 +69,7 @@ protected boolean hasBehavior(TestingConnectorBehavior connectorBehavior)
             case SUPPORTS_RENAME_TABLE:
             case SUPPORTS_NOT_NULL_CONSTRAINT:
             case SUPPORTS_CREATE_TABLE_WITH_DATA:
+            case SUPPORTS_CREATE_TABLE_WITH_TABLE_COMMENT:
             case SUPPORTS_DELETE:
             case SUPPORTS_INSERT:
             case SUPPORTS_ADD_COLUMN:

plugin/trino-cassandra/src/main/java/io/trino/plugin/cassandra/CassandraMetadata.java

@@ -288,6 +288,9 @@ public ConnectorOutputTableHandle beginCreateTable(ConnectorSession session, Con
 
     private CassandraOutputTableHandle createTable(ConnectorTableMetadata tableMetadata)
     {
+        if (tableMetadata.getComment().isPresent()) {
+            throw new TrinoException(NOT_SUPPORTED, "This connector does not support creating tables with table comment");
+        }
         ImmutableList.Builder<String> columnNames = ImmutableList.builder();
         ImmutableList.Builder<Type> columnTypes = ImmutableList.builder();
         ImmutableList.Builder<ExtraColumnMetadata> columnExtra = ImmutableList.builder();

plugin/trino-cassandra/src/test/java/io/trino/plugin/cassandra/TestCassandraConnectorTest.java

@@ -95,6 +95,9 @@ protected boolean hasBehavior(TestingConnectorBehavior connectorBehavior)
             case SUPPORTS_CREATE_VIEW:
                 return false;
 
+            case SUPPORTS_CREATE_TABLE_WITH_TABLE_COMMENT:
+                return false;
+
             case SUPPORTS_RENAME_TABLE:
                 return false;
 

plugin/trino-clickhouse/src/test/java/io/trino/plugin/clickhouse/TestClickHouseConnectorTest.java

@@ -42,4 +42,20 @@ public void testCommentTable()
         assertThatThrownBy(super::testCommentTable)
                 .hasMessageContaining("Code: 62, e.displayText() = DB::Exception: Syntax error");
     }
+
+    @Override
+    public void testCreateTableWithTableComment()
+    {
+        // Table comment is unsupported in old ClickHouse version
+        assertThatThrownBy(super::testCreateTableWithTableComment)
+                .hasMessageMatching("(?s).* Syntax error: .* COMMENT 'test comment'.*");
+    }
+
+    @Override
+    public void testCreateTableAsSelectWithTableComment()
+    {
+        // Table comment is unsupported in old ClickHouse version
+        assertThatThrownBy(super::testCreateTableAsSelectWithTableComment)
+                .hasMessageMatching("(?s).* Syntax error: .* COMMENT 'test comment'.*");
+    }
 }

plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeMetadata.java

@@ -403,7 +403,7 @@ public ConnectorTableMetadata getTableMetadata(ConnectorSession session, Connect
                 tableHandle.getSchemaTableName(),
                 columns,
                 properties.buildOrThrow(),
-                Optional.empty());
+                Optional.ofNullable(tableHandle.getMetadataEntry().getDescription()));
     }
 
     @Override
@@ -626,7 +626,8 @@ public void createTable(ConnectorSession session, ConnectorTableMetadata tableMe
                         CREATE_TABLE_OPERATION,
                         session,
                         nodeVersion,
-                        nodeId);
+                        nodeId,
+                        tableMetadata.getComment());
 
                 setRollback(() -> deleteRecursivelyIfExists(new HdfsContext(session), hdfsEnvironment, deltaLogDirectory));
                 transactionLogWriter.flush();
@@ -735,7 +736,8 @@ public DeltaLakeOutputTableHandle beginCreateTable(ConnectorSession session, Con
                 tableMetadata.getColumns().stream().map(column -> toColumnHandle(column, partitionedBy)).collect(toImmutableList()),
                 location,
                 DeltaLakeTableProperties.getCheckpointInterval(tableMetadata.getProperties()),
-                external);
+                external,
+                tableMetadata.getComment());
     }
 
     private Optional<String> getSchemaLocation(Database database)
@@ -885,7 +887,8 @@ public Optional<ConnectorOutputMetadata> finishCreateTable(
                     CREATE_TABLE_AS_OPERATION,
                     session,
                     nodeVersion,
-                    nodeId);
+                    nodeId,
+                    handle.getComment());
             appendAddFileEntries(transactionLogWriter, dataFileInfos, handle.getPartitionedBy(), true);
             transactionLogWriter.flush();
             PrincipalPrivileges principalPrivileges = buildInitialPrivilegeSet(table.getOwner().orElseThrow());
@@ -960,7 +963,8 @@ public void addColumn(ConnectorSession session, ConnectorTableHandle tableHandle
                     ADD_COLUMN_OPERATION,
                     session,
                     nodeVersion,
-                    nodeId);
+                    nodeId,
+                    Optional.ofNullable(handle.getMetadataEntry().getDescription()));
             transactionLogWriter.flush();
         }
         catch (Exception e) {
@@ -978,7 +982,8 @@ private static void appendTableEntries(
             String operation,
             ConnectorSession session,
             String nodeVersion,
-            String nodeId)
+            String nodeId,
+            Optional<String> comment)
     {
         long createdTime = System.currentTimeMillis();
         transactionLogWriter.appendCommitInfoEntry(
@@ -1002,7 +1007,7 @@ private static void appendTableEntries(
                 new MetadataEntry(
                         tableId,
                         null,
-                        null,
+                        comment.orElse(null),
                         new Format("parquet", ImmutableMap.of()),
                         serializeSchemaAsJson(columns),
                         partitionColumnNames,

plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeOutputTableHandle.java

@@ -35,6 +35,7 @@ public class DeltaLakeOutputTableHandle
     private final String location;
     private final Optional<Long> checkpointInterval;
     private final boolean external;
+    private final Optional<String> comment;
 
     @JsonCreator
     public DeltaLakeOutputTableHandle(
@@ -43,14 +44,16 @@ public DeltaLakeOutputTableHandle(
             @JsonProperty("inputColumns") List<DeltaLakeColumnHandle> inputColumns,
             @JsonProperty("location") String location,
             @JsonProperty("checkpointInterval") Optional<Long> checkpointInterval,
-            @JsonProperty("external") boolean external)
+            @JsonProperty("external") boolean external,
+            @JsonProperty("comment") Optional<String> comment)
     {
         this.schemaName = requireNonNull(schemaName, "schemaName is null");
         this.tableName = requireNonNull(tableName, "tableName is null");
         this.inputColumns = ImmutableList.copyOf(inputColumns);
         this.location = requireNonNull(location, "location is null");
         this.checkpointInterval = checkpointInterval;
         this.external = external;
+        this.comment = requireNonNull(comment, "comment is null");
     }
 
     @JsonProperty
@@ -97,4 +100,10 @@ public boolean isExternal()
     {
         return external;
     }
+
+    @JsonProperty
+    public Optional<String> getComment()
+    {
+        return comment;
+    }
 }

plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/transactionlog/checkpoint/CheckpointEntryIterator.java

@@ -521,23 +521,30 @@ private DeltaLakeTransactionLogEntry buildTxnEntry(ConnectorSession session, Blo
         return DeltaLakeTransactionLogEntry.transactionEntry(result);
     }
 
+    @Nullable
     private String getString(Block block, int position)
     {
+        if (block.isNull(position)) {
+            return null;
+        }
         return block.getSlice(position, 0, block.getSliceLength(position)).toString(UTF_8);
     }
 
     private long getLong(Block block, int position)
     {
+        checkArgument(!block.isNull(position));
         return block.getLong(position, 0);
     }
 
     private int getInt(Block block, int position)
     {
+        checkArgument(!block.isNull(position));
         return block.getInt(position, 0);
     }
 
     private byte getByte(Block block, int position)
     {
+        checkArgument(!block.isNull(position));
         return block.getByte(position, 0);
     }
 

plugin/trino-delta-lake/src/test/java/io/trino/plugin/deltalake/TestDeltaLakePageSink.java

@@ -164,7 +164,8 @@ private static ConnectorPageSink createPageSink(Path outputPath, DeltaLakeWriter
                 getColumnHandles(),
                 outputPath.toString(),
                 Optional.of(deltaLakeConfig.getDefaultCheckpointWritingInterval()),
-                true);
+                true,
+                Optional.empty());
 
         DeltaLakePageSinkProvider provider = new DeltaLakePageSinkProvider(
                 new GroupByHashPageIndexerFactory(new JoinCompiler(new TypeOperators()), new BlockTypeOperators()),

plugin/trino-delta-lake/src/test/java/io/trino/plugin/deltalake/transactionlog/checkpoint/TestCheckpointEntryIterator.java

@@ -87,8 +87,8 @@ public void testReadMetadataEntry()
                 .isEqualTo(
                         new MetadataEntry(
                                 "b6aeffad-da73-4dde-b68e-937e468b1fde",
-                                "",
-                                "",
+                                null,
+                                null,
                                 new MetadataEntry.Format("parquet", Map.of()),
                                 "{\"type\":\"struct\",\"fields\":[" +
                                         "{\"name\":\"name\",\"type\":\"string\",\"nullable\":true,\"metadata\":{}}," +

plugin/trino-kudu/src/main/java/io/trino/plugin/kudu/KuduMetadata.java

@@ -247,6 +247,9 @@ public void dropSchema(ConnectorSession session, String schemaName)
     @Override
     public void createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, boolean ignoreExisting)
     {
+        if (tableMetadata.getComment().isPresent()) {
+            throw new TrinoException(NOT_SUPPORTED, "This connector does not support creating tables with table comment");
+        }
         clientSession.createTable(tableMetadata, ignoreExisting);
     }
 
@@ -329,6 +332,9 @@ public ConnectorOutputTableHandle beginCreateTable(
         if (retryMode != NO_RETRIES) {
             throw new TrinoException(NOT_SUPPORTED, "This connector does not support query retries");
         }
+        if (tableMetadata.getComment().isPresent()) {
+            throw new TrinoException(NOT_SUPPORTED, "This connector does not support creating tables with table comment");
+        }
         PartitionDesign design = KuduTableProperties.getPartitionDesign(tableMetadata.getProperties());
         boolean generateUUID = !design.hasPartitions();
         ConnectorTableMetadata finalTableMetadata = tableMetadata;

plugin/trino-kudu/src/test/java/io/trino/plugin/kudu/AbstractKuduConnectorTest.java

@@ -70,6 +70,7 @@ protected boolean hasBehavior(TestingConnectorBehavior connectorBehavior)
             case SUPPORTS_DELETE:
                 return true;
             case SUPPORTS_RENAME_SCHEMA:
+            case SUPPORTS_CREATE_TABLE_WITH_TABLE_COMMENT:
             case SUPPORTS_COMMENT_ON_TABLE:
             case SUPPORTS_COMMENT_ON_COLUMN:
             case SUPPORTS_ARRAY:

plugin/trino-memory/src/main/java/io/trino/plugin/memory/MemoryMetadata.java

@@ -67,6 +67,7 @@
 import static com.google.common.collect.ImmutableMap.toImmutableMap;
 import static io.trino.spi.StandardErrorCode.ALREADY_EXISTS;
 import static io.trino.spi.StandardErrorCode.NOT_FOUND;
+import static io.trino.spi.StandardErrorCode.NOT_SUPPORTED;
 import static io.trino.spi.StandardErrorCode.SCHEMA_NOT_EMPTY;
 import static io.trino.spi.connector.RetryMode.NO_RETRIES;
 import static io.trino.spi.connector.SampleType.SYSTEM;
@@ -235,6 +236,9 @@ public synchronized void createTable(ConnectorSession session, ConnectorTableMet
     @Override
     public synchronized MemoryOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, Optional<ConnectorTableLayout> layout, RetryMode retryMode)
     {
+        if (tableMetadata.getComment().isPresent()) {
+            throw new TrinoException(NOT_SUPPORTED, "This connector does not support creating tables with table comment");
+        }
         checkSchemaExists(tableMetadata.getTable().getSchemaName());
         checkTableNotExists(tableMetadata.getTable());
         long tableId = nextTableId.getAndIncrement();

plugin/trino-memory/src/test/java/io/trino/plugin/memory/TestMemoryConnectorTest.java

@@ -97,6 +97,7 @@ protected boolean hasBehavior(TestingConnectorBehavior connectorBehavior)
             case SUPPORTS_RENAME_COLUMN:
                 return false;
 
+            case SUPPORTS_CREATE_TABLE_WITH_TABLE_COMMENT:
             case SUPPORTS_COMMENT_ON_TABLE:
             case SUPPORTS_COMMENT_ON_COLUMN:
                 return false;

plugin/trino-oracle/src/test/java/io/trino/plugin/oracle/BaseOracleConnectorTest.java

@@ -61,6 +61,9 @@ protected boolean hasBehavior(TestingConnectorBehavior connectorBehavior)
             case SUPPORTS_JOIN_PUSHDOWN_WITH_DISTINCT_FROM:
                 return false;
 
+            case SUPPORTS_CREATE_TABLE_WITH_TABLE_COMMENT:
+                return false;
+
             case SUPPORTS_COMMENT_ON_TABLE:
                 return false;
 

plugin/trino-phoenix/src/main/java/io/trino/plugin/phoenix/PhoenixClient.java

@@ -572,6 +572,9 @@ public Optional<String> getTableComment(ResultSet resultSet)
     @Override
     public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)
     {
+        if (tableMetadata.getComment().isPresent()) {
+            throw new TrinoException(NOT_SUPPORTED, "This connector does not support creating tables with table comment");
+        }
         SchemaTableName schemaTableName = tableMetadata.getTable();
         String schema = schemaTableName.getSchemaName();
         String table = schemaTableName.getTableName();

plugin/trino-phoenix/src/test/java/io/trino/plugin/phoenix/TestPhoenixConnectorTest.java

@@ -71,6 +71,7 @@ protected boolean hasBehavior(TestingConnectorBehavior connectorBehavior)
             case SUPPORTS_AGGREGATION_PUSHDOWN:
                 return false;
 
+            case SUPPORTS_CREATE_TABLE_WITH_TABLE_COMMENT:
             case SUPPORTS_COMMENT_ON_TABLE:
             case SUPPORTS_COMMENT_ON_COLUMN:
             case SUPPORTS_ADD_COLUMN_WITH_COMMENT:

plugin/trino-phoenix5/src/main/java/io/trino/plugin/phoenix5/PhoenixClient.java

@@ -564,6 +564,9 @@ public Optional<String> getTableComment(ResultSet resultSet)
     @Override
     public JdbcOutputTableHandle beginCreateTable(ConnectorSession session, ConnectorTableMetadata tableMetadata)
     {
+        if (tableMetadata.getComment().isPresent()) {
+            throw new TrinoException(NOT_SUPPORTED, "This connector does not support creating tables with table comment");
+        }
         SchemaTableName schemaTableName = tableMetadata.getTable();
         String schema = schemaTableName.getSchemaName();
         String table = schemaTableName.getTableName();

plugin/trino-phoenix5/src/test/java/io/trino/plugin/phoenix5/TestPhoenixConnectorTest.java

@@ -89,6 +89,7 @@ protected boolean hasBehavior(TestingConnectorBehavior connectorBehavior)
             case SUPPORTS_AGGREGATION_PUSHDOWN:
                 return false;
 
+            case SUPPORTS_CREATE_TABLE_WITH_TABLE_COMMENT:
             case SUPPORTS_COMMENT_ON_TABLE:
             case SUPPORTS_COMMENT_ON_COLUMN:
             case SUPPORTS_ADD_COLUMN_WITH_COMMENT:

plugin/trino-postgresql/src/test/java/io/trino/plugin/postgresql/TestPostgreSqlConnectorTest.java

@@ -118,6 +118,7 @@ protected boolean hasBehavior(TestingConnectorBehavior connectorBehavior)
             case SUPPORTS_JOIN_PUSHDOWN:
                 return true;
 
+            case SUPPORTS_CREATE_TABLE_WITH_TABLE_COMMENT:
             case SUPPORTS_COMMENT_ON_TABLE:
             case SUPPORTS_ADD_COLUMN_WITH_COMMENT:
                 return false;

plugin/trino-raptor-legacy/src/main/java/io/trino/plugin/raptor/legacy/RaptorMetadata.java

@@ -550,6 +550,9 @@ public ConnectorOutputTableHandle beginCreateTable(ConnectorSession session, Con
         if (retryMode != NO_RETRIES) {
             throw new TrinoException(NOT_SUPPORTED, "This connector does not support query retries");
         }
+        if (tableMetadata.getComment().isPresent()) {
+            throw new TrinoException(NOT_SUPPORTED, "This connector does not support creating tables with table comment");
+        }
 
         if (viewExists(session, tableMetadata.getTable())) {
             throw new TrinoException(ALREADY_EXISTS, "View already exists: " + tableMetadata.getTable());

plugin/trino-raptor-legacy/src/test/java/io/trino/plugin/raptor/legacy/BaseRaptorConnectorTest.java

@@ -74,6 +74,7 @@ protected boolean hasBehavior(TestingConnectorBehavior connectorBehavior)
                 return true;
             case SUPPORTS_CREATE_SCHEMA:
             case SUPPORTS_RENAME_SCHEMA:
+            case SUPPORTS_CREATE_TABLE_WITH_TABLE_COMMENT:
             case SUPPORTS_COMMENT_ON_TABLE:
             case SUPPORTS_COMMENT_ON_COLUMN:
             case SUPPORTS_ADD_COLUMN_WITH_COMMENT: