Add table comments interface in base JDBC and implement in MySQL

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

@@ -205,7 +205,7 @@ public Optional<JdbcTableHandle> getTableHandle(ConnectorSession session, Schema
             try (ResultSet resultSet = getTables(connection, Optional.of(remoteSchema), Optional.of(remoteTable))) {
                 List<JdbcTableHandle> tableHandles = new ArrayList<>();
                 while (resultSet.next()) {
-                    tableHandles.add(new JdbcTableHandle(schemaTableName, getRemoteTable(resultSet)));
+                    tableHandles.add(new JdbcTableHandle(schemaTableName, getRemoteTable(resultSet), getTableComment(resultSet)));
                 }
                 if (tableHandles.isEmpty()) {
                     return Optional.empty();

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

@@ -377,6 +377,20 @@ public void renameSchema(ConnectorSession session, String schemaName, String new
         invalidateSchemasCache();
     }
 
+    @Override
+    public Optional<String> getTableComment(ResultSet resultSet)
+            throws SQLException
+    {
+        return delegate.getTableComment(resultSet);
+    }
+
+    @Override
+    public void setTableComment(ConnectorSession session, JdbcTableHandle handle, Optional<String> comment)
+    {
+        delegate.setTableComment(session, handle, comment);
+        invalidateTableCaches(handle.asPlainTable().getSchemaTableName());
+    }
+
     @Override
     public void setColumnComment(ConnectorSession session, JdbcTableHandle handle, JdbcColumnHandle column, Optional<String> comment)
     {

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

@@ -592,7 +592,8 @@ public ConnectorTableMetadata getTableMetadata(ConnectorSession session, Connect
                 jdbcClient.getColumns(session, handle).stream()
                         .map(JdbcColumnHandle::getColumnMetadata)
                         .collect(toImmutableList()),
-                jdbcClient.getTableProperties(session, handle));
+                jdbcClient.getTableProperties(session, handle),
+                getTableComment(handle));
     }
 
     public static SchemaTableName getSchemaTableName(JdbcTableHandle handle)
@@ -603,6 +604,11 @@ public static SchemaTableName getSchemaTableName(JdbcTableHandle handle)
                 : new SchemaTableName("_generated", "_generated_query");
     }
 
+    public static Optional<String> getTableComment(JdbcTableHandle handle)
+    {
+        return handle.isNamedRelation() ? handle.getRequiredNamedRelation().getComment() : Optional.empty();
+    }
+
     @Override
     public List<SchemaTableName> listTables(ConnectorSession session, Optional<String> schemaName)
     {
@@ -743,6 +749,14 @@ public void truncateTable(ConnectorSession session, ConnectorTableHandle tableHa
         jdbcClient.truncateTable(session, (JdbcTableHandle) tableHandle);
     }
 
+    @Override
+    public void setTableComment(ConnectorSession session, ConnectorTableHandle table, Optional<String> comment)
+    {
+        JdbcTableHandle tableHandle = (JdbcTableHandle) table;
+        verify(!tableHandle.isSynthetic(), "Not a table reference: %s", tableHandle);
+        jdbcClient.setTableComment(session, tableHandle, comment);
+    }
+
     @Override
     public void setColumnComment(ConnectorSession session, ConnectorTableHandle table, ColumnHandle column, Optional<String> comment)
     {

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

@@ -264,6 +264,19 @@ public boolean isLimitGuaranteed(ConnectorSession session)
         return delegate().isLimitGuaranteed(session);
     }
 
+    @Override
+    public Optional<String> getTableComment(ResultSet resultSet)
+            throws SQLException
+    {
+        return delegate().getTableComment(resultSet);
+    }
+
+    @Override
+    public void setTableComment(ConnectorSession session, JdbcTableHandle handle, Optional<String> comment)
+    {
+        delegate().setTableComment(session, handle, comment);
+    }
+
     @Override
     public void setColumnComment(ConnectorSession session, JdbcTableHandle handle, JdbcColumnHandle column, Optional<String> comment)
     {

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

@@ -123,6 +123,17 @@ Optional<PreparedQuery> implementJoin(
 
     boolean isLimitGuaranteed(ConnectorSession session);
 
+    default Optional<String> getTableComment(ResultSet resultSet)
+            throws SQLException
+    {
+        return Optional.ofNullable(resultSet.getString("REMARKS"));
+    }
+
+    default void setTableComment(ConnectorSession session, JdbcTableHandle handle, Optional<String> comment)
+    {
+        throw new TrinoException(NOT_SUPPORTED, "This connector does not support setting table comments");
+    }
+
     default void setColumnComment(ConnectorSession session, JdbcTableHandle handle, JdbcColumnHandle column, Optional<String> comment)
     {
         throw new TrinoException(NOT_SUPPORTED, "This connector does not support setting column comments");

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

@@ -18,6 +18,7 @@
 import io.trino.spi.connector.SchemaTableName;
 
 import java.util.Objects;
+import java.util.Optional;
 
 import static java.lang.String.format;
 import static java.util.Objects.requireNonNull;
@@ -27,14 +28,17 @@ public class JdbcNamedRelationHandle
 {
     private final SchemaTableName schemaTableName;
     private final RemoteTableName remoteTableName;
+    private final Optional<String> comment;
 
     @JsonCreator
     public JdbcNamedRelationHandle(
             @JsonProperty("schemaTableName") SchemaTableName schemaTableName,
-            @JsonProperty("remoteTableName") RemoteTableName remoteTableName)
+            @JsonProperty("remoteTableName") RemoteTableName remoteTableName,
+            @JsonProperty("comment") Optional<String> comment)
     {
         this.schemaTableName = requireNonNull(schemaTableName, "schemaTableName is null");
         this.remoteTableName = requireNonNull(remoteTableName, "remoteTableName is null");
+        this.comment = requireNonNull(comment, "comment is null");
     }
 
     @JsonProperty
@@ -49,6 +53,12 @@ public RemoteTableName getRemoteTableName()
         return remoteTableName;
     }
 
+    @JsonProperty
+    public Optional<String> getComment()
+    {
+        return comment;
+    }
+
     @Override
     public boolean equals(Object o)
     {

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

@@ -63,13 +63,13 @@ public final class JdbcTableHandle
     @Deprecated
     public JdbcTableHandle(SchemaTableName schemaTableName, @Nullable String catalogName, @Nullable String schemaName, String tableName)
     {
-        this(schemaTableName, new RemoteTableName(Optional.ofNullable(catalogName), Optional.ofNullable(schemaName), tableName));
+        this(schemaTableName, new RemoteTableName(Optional.ofNullable(catalogName), Optional.ofNullable(schemaName), tableName), Optional.empty());
     }
 
-    public JdbcTableHandle(SchemaTableName schemaTableName, RemoteTableName remoteTableName)
+    public JdbcTableHandle(SchemaTableName schemaTableName, RemoteTableName remoteTableName, Optional<String> comment)
     {
         this(
-                new JdbcNamedRelationHandle(schemaTableName, remoteTableName),
+                new JdbcNamedRelationHandle(schemaTableName, remoteTableName, comment),
                 TupleDomain.all(),
                 ImmutableList.of(),
                 Optional.empty(),

plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/jmx/JdbcClientStats.java

@@ -29,6 +29,8 @@ public final class JdbcClientStats
     private final JdbcApiStats commitCreateTable = new JdbcApiStats();
     private final JdbcApiStats createSchema = new JdbcApiStats();
     private final JdbcApiStats createTable = new JdbcApiStats();
+    private final JdbcApiStats getTableComment = new JdbcApiStats();
+    private final JdbcApiStats setTableComment = new JdbcApiStats();
     private final JdbcApiStats setColumnComment = new JdbcApiStats();
     private final JdbcApiStats dropColumn = new JdbcApiStats();
     private final JdbcApiStats dropSchema = new JdbcApiStats();
@@ -135,6 +137,20 @@ public JdbcApiStats getCreateTable()
         return createTable;
     }
 
+    @Managed
+    @Nested
+    public JdbcApiStats getGetTableComment()
+    {
+        return getTableComment;
+    }
+
+    @Managed
+    @Nested
+    public JdbcApiStats getSetTableComment()
+    {
+        return setTableComment;
+    }
+
     @Managed
     @Nested
     public JdbcApiStats getSetColumnComment()

plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/jmx/StatisticsAwareJdbcClient.java

@@ -197,6 +197,19 @@ public Optional<PreparedQuery> implementJoin(ConnectorSession session,
         return stats.getImplementJoin().wrap(() -> delegate().implementJoin(session, joinType, leftSource, rightSource, joinConditions, rightAssignments, leftAssignments, statistics));
     }
 
+    @Override
+    public Optional<String> getTableComment(ResultSet resultSet)
+            throws SQLException
+    {
+        return stats.getGetTableComment().wrap(() -> delegate().getTableComment(resultSet));
+    }
+
+    @Override
+    public void setTableComment(ConnectorSession session, JdbcTableHandle handle, Optional<String> comment)
+    {
+        stats.getSetTableComment().wrap(() -> delegate().setTableComment(session, handle, comment));
+    }
+
     @Override
     public void setColumnComment(ConnectorSession session, JdbcTableHandle handle, JdbcColumnHandle column, Optional<String> comment)
     {

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

@@ -93,7 +93,8 @@ public class TestDefaultJdbcQueryBuilder
 {
     private static final JdbcNamedRelationHandle TEST_TABLE = new JdbcNamedRelationHandle(new SchemaTableName(
             "some_test_schema", "test_table"),
-            new RemoteTableName(Optional.empty(), Optional.empty(), "test_table"));
+            new RemoteTableName(Optional.empty(), Optional.empty(), "test_table"),
+            Optional.empty());
     private static final ConnectorSession SESSION = TestingConnectorSession.builder()
             .setPropertyMetadata(new JdbcMetadataSessionProperties(new JdbcMetadataConfig(), Optional.empty()).getSessionProperties())
             .build();

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

@@ -81,7 +81,8 @@ private JdbcTableHandle createNamedHandle()
         return new JdbcTableHandle(
                 new JdbcNamedRelationHandle(
                         new SchemaTableName("schema", "table"),
-                        new RemoteTableName(Optional.of("catalog"), Optional.of("schema"), "table")),
+                        new RemoteTableName(Optional.of("catalog"), Optional.of("schema"), "table"),
+                        Optional.empty()),
                 TupleDomain.all(),
                 ImmutableList.of(),
                 Optional.empty(),

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

@@ -34,6 +34,7 @@
 import net.jodah.failsafe.RetryPolicy;
 
 import java.sql.Connection;
+import java.sql.ResultSet;
 import java.sql.Types;
 import java.util.Collection;
 import java.util.List;
@@ -101,6 +102,13 @@ public Collection<String> listSchemas(Connection connection)
                 .get(() -> super.listSchemas(connection));
     }
 
+    @Override
+    public Optional<String> getTableComment(ResultSet resultSet)
+    {
+        // Don't return a comment until the connector supports creating tables with comment
+        return Optional.empty();
+    }
+
     @Override
     public boolean supportsAggregationPushdown(ConnectorSession session, JdbcTableHandle table, List<AggregateFunction> aggregates, Map<String, ColumnHandle> assignments, List<List<ColumnHandle>> groupingSets)
     {

plugin/trino-clickhouse/src/main/java/io/trino/plugin/clickhouse/ClickHouseClient.java

@@ -264,6 +264,13 @@ protected void copyTableSchema(Connection connection, String catalogName, String
         execute(connection, sql);
     }
 
+    @Override
+    public Optional<String> getTableComment(ResultSet resultSet)
+    {
+        // Don't return a comment until the connector supports creating tables with comment
+        return Optional.empty();
+    }
+
     @Override
     protected String createTableSql(RemoteTableName remoteTableName, List<String> columns, ConnectorTableMetadata tableMetadata)
     {

plugin/trino-druid/src/main/java/io/trino/plugin/druid/DruidJdbcClient.java

@@ -178,6 +178,13 @@ public ResultSet getTables(Connection connection, Optional<String> schemaName, O
                 null);
     }
 
+    @Override
+    public Optional<String> getTableComment(ResultSet resultSet)
+    {
+        // Don't return a comment until the connector supports creating tables with comment
+        return Optional.empty();
+    }
+
     @Override
     public Optional<ColumnMapping> toColumnMapping(ConnectorSession session, Connection connection, JdbcTypeHandle typeHandle)
     {
@@ -284,7 +291,8 @@ private JdbcTableHandle prepareTableHandleForQuery(JdbcTableHandle table)
                             new RemoteTableName(
                                     Optional.empty(),
                                     table.getRequiredNamedRelation().getRemoteTableName().getSchemaName(),
-                                    table.getRequiredNamedRelation().getRemoteTableName().getTableName())),
+                                    table.getRequiredNamedRelation().getRemoteTableName().getTableName()),
+                            Optional.empty()),
                     table.getConstraint(),
                     table.getConstraintExpressions(),
                     table.getSortOrder(),

plugin/trino-mysql/src/main/java/io/trino/plugin/mysql/MySqlClient.java

@@ -30,6 +30,7 @@
 import io.trino.plugin.jdbc.JdbcTypeHandle;
 import io.trino.plugin.jdbc.PreparedQuery;
 import io.trino.plugin.jdbc.QueryBuilder;
+import io.trino.plugin.jdbc.RemoteTableName;
 import io.trino.plugin.jdbc.WriteMapping;
 import io.trino.plugin.jdbc.aggregation.ImplementAvgDecimal;
 import io.trino.plugin.jdbc.aggregation.ImplementAvgFloatingPoint;
@@ -78,6 +79,8 @@
 import java.util.function.BiFunction;
 import java.util.stream.Stream;
 
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Strings.emptyToNull;
 import static com.google.common.base.Verify.verify;
 import static com.google.common.util.concurrent.MoreExecutors.directExecutor;
 import static com.mysql.cj.exceptions.MysqlErrorNumbers.SQL_STATE_ER_TABLE_EXISTS_ERROR;
@@ -142,6 +145,7 @@
 import static java.lang.Math.max;
 import static java.lang.Math.min;
 import static java.lang.String.format;
+import static java.lang.String.join;
 import static java.util.Locale.ENGLISH;
 import static java.util.stream.Collectors.joining;
 
@@ -157,6 +161,9 @@ public class MySqlClient
     // MySQL driver returns width of time types instead of precision, same as the above timestamp type.
     private static final int ZERO_PRECISION_TIME_COLUMN_SIZE = 8;
 
+    // An empty character means that the table doesn't have a comment in MySQL
+    private static final String NO_COMMENT = "";
+
     private final Type jsonType;
     private final ConnectorExpressionRewriter<String> connectorExpressionRewriter;
     private final AggregateFunctionRewriter<JdbcExpression, String> aggregateFunctionRewriter;
@@ -273,6 +280,24 @@ public ResultSet getTables(Connection connection, Optional<String> schemaName, O
                 getTableTypes().map(types -> types.toArray(String[]::new)).orElse(null));
     }
 
+    @Override
+    public Optional<String> getTableComment(ResultSet resultSet)
+            throws SQLException
+    {
+        // Empty remarks means that the table doesn't have a comment in MySQL
+        return Optional.ofNullable(emptyToNull(resultSet.getString("REMARKS")));
+    }
+
+    @Override
+    public void setTableComment(ConnectorSession session, JdbcTableHandle handle, Optional<String> comment)
+    {
+        String sql = format(
+                "ALTER TABLE %s COMMENT = '%s'",
+                quoted(handle.asPlainTable().getRemoteTableName()),
+                comment.orElse(NO_COMMENT)); // An empty character removes the existing comment in MySQL
+        execute(session, sql);
+    }
+
     @Override
     protected String getTableSchemaName(ResultSet resultSet)
             throws SQLException
@@ -281,6 +306,13 @@ protected String getTableSchemaName(ResultSet resultSet)
         return resultSet.getString("TABLE_CAT");
     }
 
+    @Override
+    protected String createTableSql(RemoteTableName remoteTableName, List<String> columns, ConnectorTableMetadata tableMetadata)
+    {
+        checkArgument(tableMetadata.getProperties().isEmpty(), "Unsupported table properties: %s", tableMetadata.getProperties());
+        return format("CREATE TABLE %s (%s) COMMENT '%s'", quoted(remoteTableName), join(", ", columns), tableMetadata.getComment().orElse(NO_COMMENT));
+    }
+
     @Override
     public Optional<ColumnMapping> toColumnMapping(ConnectorSession session, Connection connection, JdbcTypeHandle typeHandle)
     {

plugin/trino-mysql/src/test/java/io/trino/plugin/mysql/BaseMySqlConnectorTest.java

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

plugin/trino-oracle/src/main/java/io/trino/plugin/oracle/OracleClient.java

@@ -264,6 +264,13 @@ public void renameSchema(ConnectorSession session, String schemaName, String new
         throw new TrinoException(NOT_SUPPORTED, "This connector does not support renaming schemas");
     }
 
+    @Override
+    public Optional<String> getTableComment(ResultSet resultSet)
+    {
+        // Don't return a comment until the connector supports creating tables with comment
+        return Optional.empty();
+    }
+
     @Override
     public Optional<ColumnMapping> toColumnMapping(ConnectorSession session, Connection connection, JdbcTypeHandle typeHandle)
     {