[apache#1352] fix(trino-connector): Support the NOT NULL constraint i…

…n create table syntax (apache#1364)

### What changes were proposed in this pull request?

Add `NOT_NULL_COLUMN_CONSTRAINT` capability to `GravitinoConnector`	

### Why are the changes needed?

We can't create a table with not null column by trino

Fix: apache#1352 

### Does this PR introduce _any_ user-facing change?

N/A

### How was this patch tested?

Add IT test case `testColumnTypeNotNullByTrino` was added.

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

@@ -28,6 +28,7 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Set;
 import org.apache.hadoop.hive.conf.HiveConf;
@@ -564,6 +565,99 @@ private ColumnDTO[] createFullTypeColumns() {
     };
   }
 
+  @Test
+  void testColumnTypeNotNullByTrino() throws InterruptedException {
+    String catalogName = GravitinoITUtils.genRandomName("mysql_catalog").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());
+
+    String sql = String.format("show catalogs like '%s.%s'", metalakeName, catalogName);
+    Assertions.assertTrue(checkTrinoHasLoaded(sql, 30));
+
+    String schemaName = GravitinoITUtils.genRandomName("schema").toLowerCase();
+    String tableName = GravitinoITUtils.genRandomName("table").toLowerCase();
+    String createSchemaSql =
+        String.format("CREATE SCHEMA \"%s.%s\".%s", metalakeName, catalogName, schemaName);
+    containerSuite.getTrinoContainer().executeUpdateSQL(createSchemaSql);
+
+    sql = String.format("show create schema \"%s.%s\".%s", metalakeName, catalogName, schemaName);
+    Assertions.assertTrue(checkTrinoHasLoaded(sql, 30));
+
+    String createTableSql =
+        String.format(
+            "CREATE TABLE \"%s.%s\".%s.%s (id int not null, name varchar not null)",
+            metalakeName, catalogName, schemaName, tableName);
+    containerSuite.getTrinoContainer().executeUpdateSQL(createTableSql);
+
+    String showCreateTableSql =
+        String.format(
+            "show create table \"%s.%s\".%s.%s", metalakeName, catalogName, schemaName, tableName);
+    ArrayList<ArrayList<String>> rs =
+        containerSuite.getTrinoContainer().executeQuerySQL(showCreateTableSql);
+    Assertions.assertTrue(rs.get(0).get(0).toLowerCase(Locale.ENGLISH).contains("not null"));
+
+    containerSuite
+        .getTrinoContainer()
+        .executeUpdateSQL(
+            String.format(
+                "insert into \"%s.%s\".%s.%s values(1, 'a')",
+                metalakeName, catalogName, schemaName, tableName));
+    Assertions.assertThrows(
+        RuntimeException.class,
+        () ->
+            containerSuite
+                .getTrinoContainer()
+                .executeUpdateSQL(
+                    String.format(
+                        "insert into \"%s.%s\".%s.%s values(null, 'a')",
+                        metalakeName, catalogName, schemaName, tableName)));
+    Assertions.assertThrows(
+        RuntimeException.class,
+        () ->
+            containerSuite
+                .getTrinoContainer()
+                .executeUpdateSQL(
+                    String.format(
+                        "insert into \"%s.%s\".%s.%s values(1, null)",
+                        metalakeName, catalogName, schemaName, tableName)));
+    Assertions.assertThrows(
+        RuntimeException.class,
+        () ->
+            containerSuite
+                .getTrinoContainer()
+                .executeUpdateSQL(
+                    String.format(
+                        "insert into \"%s.%s\".%s.%s values(null, null)",
+                        metalakeName, catalogName, schemaName, tableName)));
+
+    catalog
+        .asTableCatalog()
+        .loadTable(NameIdentifier.of(metalakeName, catalogName, schemaName, tableName));
+  }
+
   @Test
   void testHiveTableCreatedByGravitino() throws InterruptedException {
     String catalogName = GravitinoITUtils.genRandomName("catalog").toLowerCase();

trino-connector/src/main/java/com/datastrato/gravitino/trino/connector/GravitinoConnector.java

@@ -4,6 +4,9 @@
  */
 package com.datastrato.gravitino.trino.connector;
 
+import static com.google.common.collect.Sets.immutableEnumSet;
+import static io.trino.spi.connector.ConnectorCapabilities.NOT_NULL_COLUMN_CONSTRAINT;
+
 import com.datastrato.gravitino.NameIdentifier;
 import com.datastrato.gravitino.client.GravitinoMetaLake;
 import com.datastrato.gravitino.trino.connector.catalog.CatalogConnectorContext;
@@ -12,6 +15,7 @@
 import io.trino.plugin.base.security.AllowAllAccessControl;
 import io.trino.spi.connector.Connector;
 import io.trino.spi.connector.ConnectorAccessControl;
+import io.trino.spi.connector.ConnectorCapabilities;
 import io.trino.spi.connector.ConnectorMetadata;
 import io.trino.spi.connector.ConnectorPageSinkProvider;
 import io.trino.spi.connector.ConnectorPageSourceProvider;
@@ -22,6 +26,7 @@
 import io.trino.spi.session.PropertyMetadata;
 import io.trino.spi.transaction.IsolationLevel;
 import java.util.List;
+import java.util.Set;
 
 /**
  * GravitinoConnector serves as the entry point for operations on the connector managed by Trino and
@@ -134,4 +139,9 @@ public void commit(ConnectorTransactionHandle transactionHandle) {
   public ConnectorAccessControl getAccessControl() {
     return new AllowAllAccessControl();
   }
+
+  @Override
+  public Set<ConnectorCapabilities> getCapabilities() {
+    return immutableEnumSet(NOT_NULL_COLUMN_CONSTRAINT);
+  }
 }

trino-connector/src/main/java/com/datastrato/gravitino/trino/connector/metadata/GravitinoTable.java

@@ -88,6 +88,7 @@ public ColumnDTO[] getColumnDTOs() {
               .withName(columns.get(i).getName())
               .withDataType(columns.get(i).getType())
               .withComment(columns.get(i).getComment())
+              .withNullable(columns.get(i).isNullable())
               .build();
     }
     return gravitinoColumns;