Support altering table comments in Hive Glue catalogs

Additionally, move the comment storage from a parameter to the
description field.

plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/glue/GlueHiveMetastore.java

@@ -145,6 +145,7 @@
 import static io.trino.plugin.hive.HiveErrorCode.HIVE_FILESYSTEM_ERROR;
 import static io.trino.plugin.hive.HiveErrorCode.HIVE_INVALID_METADATA;
 import static io.trino.plugin.hive.HiveErrorCode.HIVE_METASTORE_ERROR;
+import static io.trino.plugin.hive.HiveMetadata.TABLE_COMMENT;
 import static io.trino.plugin.hive.TableType.MANAGED_TABLE;
 import static io.trino.plugin.hive.TableType.VIRTUAL_VIEW;
 import static io.trino.plugin.hive.metastore.MetastoreUtil.makePartitionName;
@@ -684,7 +685,11 @@ private TableInput convertGlueTableToTableInput(com.amazonaws.services.glue.mode
     @Override
     public void commentTable(String databaseName, String tableName, Optional<String> comment)
     {
-        throw new TrinoException(NOT_SUPPORTED, "Table comment is not yet supported by Glue service");
+        Table oldTable = getExistingTable(databaseName, tableName);
+        Table newTable = Table.builder(oldTable)
+                .setParameter(TABLE_COMMENT, comment)
+                .build();
+        replaceTable(databaseName, tableName, newTable, null);
     }
 
     @Override

plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/glue/converter/GlueInputConverter.java

@@ -36,9 +36,12 @@
 import io.trino.spi.function.LanguageFunction;
 
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 
 import static com.google.common.collect.ImmutableList.toImmutableList;
+import static com.google.common.collect.ImmutableMap.toImmutableMap;
+import static io.trino.plugin.hive.HiveMetadata.TABLE_COMMENT;
 import static io.trino.plugin.hive.metastore.thrift.ThriftMetastoreUtil.metastoreFunctionName;
 import static io.trino.plugin.hive.metastore.thrift.ThriftMetastoreUtil.toResourceUris;
 import static io.trino.plugin.hive.metastore.thrift.ThriftMetastoreUtil.updateStatisticsParameters;
@@ -61,15 +64,21 @@ public static DatabaseInput convertDatabase(Database database)
 
     public static TableInput convertTable(Table table)
     {
+        Optional<String> comment = Optional.ofNullable(table.getParameters().get(TABLE_COMMENT));
+        Map<String, String> tableParameters = table.getParameters().entrySet().stream()
+                .filter(entry -> !entry.getKey().equals(TABLE_COMMENT))
+                .collect(toImmutableMap(Map.Entry::getKey, Map.Entry::getValue));
+
         TableInput input = new TableInput();
         input.setName(table.getTableName());
         input.setOwner(table.getOwner().orElse(null));
         input.setTableType(table.getTableType());
         input.setStorageDescriptor(convertStorage(table.getStorage(), table.getDataColumns()));
         input.setPartitionKeys(table.getPartitionColumns().stream().map(GlueInputConverter::convertColumn).collect(toImmutableList()));
-        input.setParameters(table.getParameters());
+        input.setParameters(tableParameters);
         table.getViewOriginalText().ifPresent(input::setViewOriginalText);
         table.getViewExpandedText().ifPresent(input::setViewExpandedText);
+        comment.ifPresent(input::setDescription);
         return input;
     }
 

plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/glue/converter/GlueToTrinoConverter.java

@@ -53,6 +53,7 @@
 import static com.google.common.base.Strings.nullToEmpty;
 import static io.trino.plugin.hive.HiveErrorCode.HIVE_INVALID_METADATA;
 import static io.trino.plugin.hive.HiveErrorCode.HIVE_UNSUPPORTED_FORMAT;
+import static io.trino.plugin.hive.HiveMetadata.TABLE_COMMENT;
 import static io.trino.plugin.hive.HiveType.HIVE_INT;
 import static io.trino.plugin.hive.TableType.EXTERNAL_TABLE;
 import static io.trino.plugin.hive.ViewReaderUtil.isTrinoMaterializedView;
@@ -126,7 +127,16 @@ public static Table convertTable(com.amazonaws.services.glue.model.Table glueTab
         SchemaTableName table = new SchemaTableName(dbName, glueTable.getName());
 
         String tableType = getTableType(glueTable);
-        Map<String, String> tableParameters = ImmutableMap.copyOf(getTableParameters(glueTable));
+
+        ImmutableMap.Builder<String, String> parameters = ImmutableMap.builder();
+        Optional<String> description = Optional.ofNullable(glueTable.getDescription());
+        description.ifPresent(comment -> parameters.put(TABLE_COMMENT, comment));
+        getTableParameters(glueTable).entrySet().stream()
+                // If the description was set we may have two "comment"s, prefer the description field
+                .filter(entry -> description.isEmpty() || !entry.getKey().equals(TABLE_COMMENT))
+                .forEach(parameters::put);
+        Map<String, String> tableParameters = parameters.buildOrThrow();
+
         Table.Builder tableBuilder = Table.builder()
                 .setDatabaseName(table.getSchemaName())
                 .setTableName(table.getTableName())

plugin/trino-hive/src/test/java/io/trino/plugin/hive/metastore/glue/TestHiveGlueMetastore.java

@@ -1419,6 +1419,29 @@ public void testGlueObjectsWithoutStorageDescriptor()
         }
     }
 
+    @Test
+    public void testAlterTableComment()
+            throws Exception
+    {
+        SchemaTableName tableName = temporaryTable("test_alter_table_comment");
+        doCreateEmptyTable(tableName, ORC, ImmutableList.of(new ColumnMetadata("name", BIGINT)), ImmutableList.of());
+        try {
+            assertThat(metastore.getTable(tableName.getSchemaName(), tableName.getTableName()).orElseThrow().getParameters()).doesNotContainKey(TABLE_COMMENT);
+            metastore.commentTable(tableName.getSchemaName(), tableName.getTableName(), Optional.of("a table comment"));
+            Map<String, String> tableParameters = metastore.getTable(tableName.getSchemaName(), tableName.getTableName()).orElseThrow().getParameters();
+            assertThat(tableParameters.get(TABLE_COMMENT)).isEqualTo("a table comment");
+
+            metastore.commentTable(tableName.getSchemaName(), tableName.getTableName(), Optional.empty());
+            tableParameters = metastore.getTable(tableName.getSchemaName(), tableName.getTableName()).orElseThrow().getParameters();
+            assertThat(tableParameters.get(TABLE_COMMENT)).isNull();
+        }
+        finally {
+            glueClient.deleteTable(new DeleteTableRequest()
+                    .withDatabaseName(tableName.getSchemaName())
+                    .withName(tableName.getTableName()));
+        }
+    }
+
     @Test
     public void testAlterColumnComment()
             throws Exception