diff --git a/catalogs/catalog-jdbc-mysql/src/main/java/org/apache/gravitino/catalog/mysql/operation/MysqlTableOperations.java b/catalogs/catalog-jdbc-mysql/src/main/java/org/apache/gravitino/catalog/mysql/operation/MysqlTableOperations.java
index b8cc2f87233..36b4daebf9b 100644
--- a/catalogs/catalog-jdbc-mysql/src/main/java/org/apache/gravitino/catalog/mysql/operation/MysqlTableOperations.java
+++ b/catalogs/catalog-jdbc-mysql/src/main/java/org/apache/gravitino/catalog/mysql/operation/MysqlTableOperations.java
@@ -106,6 +106,7 @@ protected String generateCreateTableSql(
       }
     }
 
+    validateIndexes(indexes, columns);
     appendIndexesSql(indexes, sqlBuilder);
 
     sqlBuilder.append("\n)");
@@ -642,4 +643,33 @@ private StringBuilder appendColumnDefinition(JdbcColumn column, StringBuilder sq
   private static String quote(String name) {
     return BACK_QUOTE + name + BACK_QUOTE;
   }
+
+  /**
+   * Verify the columns in the index.
+   *
+   * @param columns jdbc column
+   * @param indexes table indexes
+   */
+  private static void validateIndexes(Index[] indexes, JdbcColumn[] columns) {
+    Map<String, JdbcColumn> columnMap =
+        Arrays.stream(columns).collect(Collectors.toMap(JdbcColumn::name, c -> c));
+    for (Index index : indexes) {
+      if (index.type() == Index.IndexType.UNIQUE_KEY) {
+        // the column in the unique index must be not null
+        for (String[] colNames : index.fieldNames()) {
+          JdbcColumn column = columnMap.get(colNames[0]);
+          Preconditions.checkArgument(
+              column != null,
+              "Column %s in the unique index %s does not exist in the table",
+              colNames[0],
+              index.name());
+          Preconditions.checkArgument(
+              !column.nullable(),
+              "Column %s in the unique index %s must be a not null column",
+              colNames[0],
+              index.name());
+        }
+      }
+    }
+  }
 }
diff --git a/catalogs/catalog-jdbc-mysql/src/test/java/org/apache/gravitino/catalog/mysql/integration/test/CatalogMysqlIT.java b/catalogs/catalog-jdbc-mysql/src/test/java/org/apache/gravitino/catalog/mysql/integration/test/CatalogMysqlIT.java
index a80da4795a0..9bd949b7b31 100644
--- a/catalogs/catalog-jdbc-mysql/src/test/java/org/apache/gravitino/catalog/mysql/integration/test/CatalogMysqlIT.java
+++ b/catalogs/catalog-jdbc-mysql/src/test/java/org/apache/gravitino/catalog/mysql/integration/test/CatalogMysqlIT.java
@@ -1037,6 +1037,27 @@ void testCreateTableIndex() {
     Assertions.assertEquals(2, table.index().length);
     Assertions.assertNotNull(table.index()[0].name());
     Assertions.assertNotNull(table.index()[1].name());
+
+    Column notNullCol = Column.of("col_6", Types.LongType.get(), "id", true, false, null);
+    Exception exception =
+        assertThrows(
+            IllegalArgumentException.class,
+            () ->
+                tableCatalog.createTable(
+                    tableIdent,
+                    new Column[] {notNullCol},
+                    table_comment,
+                    properties,
+                    Transforms.EMPTY_TRANSFORM,
+                    Distributions.NONE,
+                    new SortOrder[0],
+                    new Index[] {
+                      Indexes.of(Index.IndexType.UNIQUE_KEY, null, new String[][] {{"col_6"}}),
+                    }));
+    Assertions.assertTrue(
+        exception
+            .getMessage()
+            .contains("Column col_6 in the unique index null must be a not null column"));
   }
 
   @Test
diff --git a/catalogs/catalog-jdbc-mysql/src/test/java/org/apache/gravitino/catalog/mysql/operation/TestMysqlTableOperations.java b/catalogs/catalog-jdbc-mysql/src/test/java/org/apache/gravitino/catalog/mysql/operation/TestMysqlTableOperations.java
index 9eac348cd91..923e20fa0c0 100644
--- a/catalogs/catalog-jdbc-mysql/src/test/java/org/apache/gravitino/catalog/mysql/operation/TestMysqlTableOperations.java
+++ b/catalogs/catalog-jdbc-mysql/src/test/java/org/apache/gravitino/catalog/mysql/operation/TestMysqlTableOperations.java
@@ -64,7 +64,7 @@ public void testOperationTable() {
             .withName("col_1")
             .withType(VARCHAR)
             .withComment("test_comment")
-            .withNullable(true)
+            .withNullable(false)
             .build());
     columns.add(
         JdbcColumn.builder()
@@ -573,7 +573,7 @@ public void testCreateAndLoadTable() {
         JdbcColumn.builder()
             .withName("col_4")
             .withType(Types.DateType.get())
-            .withNullable(true)
+            .withNullable(false)
             .withComment("date")
             .withDefaultValue(Column.DEFAULT_VALUE_NOT_SET)
             .build());