[#1632] fix(jdbc-mysql): Supporting create not null column for MySQL table.

catalogs/catalog-jdbc-mysql/src/main/java/com/datastrato/gravitino/catalog/mysql/operation/MysqlTableOperations.java

@@ -308,8 +308,7 @@ protected String generateAlterTableSql(
         throw new IllegalArgumentException("Remove property is not supported yet");
       } else if (change instanceof TableChange.AddColumn) {
         TableChange.AddColumn addColumn = (TableChange.AddColumn) change;
-        lazyLoadCreateTable = getOrCreateTable(databaseName, tableName, lazyLoadCreateTable);
-        alterSql.add(addColumnFieldDefinition(addColumn, lazyLoadCreateTable));
+        alterSql.add(addColumnFieldDefinition(addColumn));
       } else if (change instanceof TableChange.RenameColumn) {
         lazyLoadCreateTable = getOrCreateTable(databaseName, tableName, lazyLoadCreateTable);
         TableChange.RenameColumn renameColumn = (TableChange.RenameColumn) change;
@@ -435,8 +434,7 @@ private String updateColumnCommentFieldDefinition(
     return "MODIFY COLUMN " + col + appendColumnDefinition(updateColumn, new StringBuilder());
   }
 
-  private String addColumnFieldDefinition(
-      TableChange.AddColumn addColumn, CreateTable createTable) {
+  private String addColumnFieldDefinition(TableChange.AddColumn addColumn) {
     String dataType = (String) typeConverter.fromGravitinoType(addColumn.getDataType());
     if (addColumn.fieldName().length > 1) {
       throw new UnsupportedOperationException("Mysql does not support nested column names.");
@@ -445,6 +443,11 @@ private String addColumnFieldDefinition(
 
     StringBuilder columnDefinition = new StringBuilder();
     columnDefinition.append("ADD COLUMN ").append(col).append(SPACE).append(dataType).append(SPACE);
+
+    if (!addColumn.isNullable()) {
+      columnDefinition.append("NOT NULL ");
+    }
+
     // Append comment if available
     if (StringUtils.isNotEmpty(addColumn.getComment())) {
       columnDefinition.append("COMMENT '").append(addColumn.getComment()).append("' ");

integration-test/src/test/java/com/datastrato/gravitino/integration/test/trino/TrinoConnectorIT.java

@@ -1105,6 +1105,85 @@ void testMySQLTableCreatedByGravitino() throws InterruptedException {
     }
   }
 
+  @Test
+  void testMySQLTableCreatedByTrino() throws InterruptedException {
+    String catalogName = GravitinoITUtils.genRandomName("mysql_catalog").toLowerCase();
+    String schemaName = GravitinoITUtils.genRandomName("mysql_schema").toLowerCase();
+    String tableName = GravitinoITUtils.genRandomName("mysql_table").toLowerCase();
+    GravitinoMetaLake createdMetalake = client.loadMetalake(NameIdentifier.of(metalakeName));
+    String[] command = {
+      "mysql",
+      "-h127.0.0.1",
+      "-uroot",
+      "-pds123", // username and password are referred from Hive dockerfile.
+      "-e",
+      "grant all privileges on *.* to root@'%' identified by 'ds123'"
+    };
+
+    // There exists a mysql instance in Hive the container.
+    containerSuite.getHiveContainer().executeInContainer(command);
+    String hiveHost = containerSuite.getHiveContainer().getContainerIpAddress();
+
+    createdMetalake.createCatalog(
+        NameIdentifier.of(metalakeName, catalogName),
+        Catalog.Type.RELATIONAL,
+        "jdbc-mysql",
+        "comment",
+        ImmutableMap.<String, String>builder()
+            .put("jdbc-driver", "com.mysql.cj.jdbc.Driver")
+            .put("jdbc-user", "root")
+            .put("jdbc-password", "ds123")
+            .put("jdbc-url", String.format("jdbc:mysql://%s:3306?useSSL=false", hiveHost))
+            .build());
+    Catalog catalog = createdMetalake.loadCatalog(NameIdentifier.of(metalakeName, catalogName));
+    Assertions.assertEquals("root", catalog.properties().get("jdbc-user"));
+
+    String sql = String.format("show catalogs like '%s.%s'", metalakeName, catalogName);
+    boolean success = checkTrinoHasLoaded(sql, 30);
+    if (!success) {
+      Assertions.fail("Trino fail to load catalogs created by gravitino: " + sql);
+    }
+
+    String data = containerSuite.getTrinoContainer().executeQuerySQL(sql).get(0).get(0);
+    Assertions.assertEquals(metalakeName + "." + catalogName, data);
+
+    // Create schema
+    sql = String.format("create schema \"%s.%s\".%s", metalakeName, catalogName, schemaName);
+    containerSuite.getTrinoContainer().executeUpdateSQL(sql);
+
+    // create table
+    sql =
+        String.format(
+            "create table \"%s.%s\".%s.%s (id int, name varchar)",
+            metalakeName, catalogName, schemaName, tableName);
+    containerSuite.getTrinoContainer().executeUpdateSQL(sql);
+
+    // Add a not null column
+    sql =
+        String.format(
+            "alter table \"%s.%s\".%s.%s add column age int not null comment 'age of users'",
+            metalakeName, catalogName, schemaName, tableName);
+    containerSuite.getTrinoContainer().executeUpdateSQL(sql);
+
+    sql =
+        String.format(
+            "alter table \"%s.%s\".%s.%s add column address varchar(20) not null comment 'address of users'",
+            metalakeName, catalogName, schemaName, tableName);
+    containerSuite.getTrinoContainer().executeUpdateSQL(sql);
+
+    catalog
+        .asTableCatalog()
+        .loadTable(NameIdentifier.of(metalakeName, catalogName, schemaName, tableName));
+
+    sql =
+        String.format(
+            "show create table \"%s.%s\".%s.%s", metalakeName, catalogName, schemaName, tableName);
+
+    data = containerSuite.getTrinoContainer().executeQuerySQL(sql).get(0).get(0);
+    Assertions.assertTrue(data.contains("age integer NOT NULL"));
+    Assertions.assertTrue(data.contains("address varchar(20) NOT NULL"));
+  }
+
   @Test
   void testDropCatalogAndCreateAgain() throws InterruptedException {
     String catalogName = GravitinoITUtils.genRandomName("mysql_catalog").toLowerCase();

trino-connector/src/main/java/com/datastrato/gravitino/trino/connector/catalog/CatalogConnectorMetadata.java

@@ -184,7 +184,7 @@ public void renameSchema(String source, String target) {
     throw new NotImplementedException();
   }
 
-  private void applyAlter(SchemaTableName tableName, TableChange change) {
+  private void applyAlter(SchemaTableName tableName, TableChange... change) {
     try {
       tableCatalog.alterTable(
           NameIdentifier.ofTable(
@@ -230,11 +230,14 @@ public void setTableProperties(SchemaTableName schemaTableName, Map<String, Stri
   public void addColumn(SchemaTableName schemaTableName, GravitinoColumn column) {
     String[] columnNames = {column.getName()};
     if (Strings.isNullOrEmpty(column.getComment()))
-      applyAlter(schemaTableName, TableChange.addColumn(columnNames, column.getType()));
+      applyAlter(
+          schemaTableName,
+          TableChange.addColumn(columnNames, column.getType(), column.isNullable()));
     else {
       applyAlter(
           schemaTableName,
-          TableChange.addColumn(columnNames, column.getType(), column.getComment()));
+          TableChange.addColumn(
+              columnNames, column.getType(), column.getComment(), column.isNullable()));
     }
   }