Verify database name explicitly in File metastore

plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/file/FileHiveMetastore.java

@@ -141,6 +141,9 @@ public class FileHiveMetastore
     // todo there should be a way to manage the admins list
     private static final Set<String> ADMIN_USERS = ImmutableSet.of("admin", "hive", "hdfs");
 
+    // 128 is equals to the max database name length of Thrift Hive metastore
+    private static final int MAX_DATABASE_NAME_LENGTH = 128;
+
     private final String currentVersion;
     private final VersionCompatibility versionCompatibility;
     private final HdfsEnvironment hdfsEnvironment;
@@ -197,6 +200,7 @@ public synchronized void createDatabase(Database database)
             throw new TrinoException(HIVE_METASTORE_ERROR, "Database cannot be created with a location set");
         }
 
+        verifyDatabaseNameLength(database.getDatabaseName());
         verifyDatabaseNotExists(database.getDatabaseName());
 
         Path databaseMetadataDirectory = getDatabaseMetadataDirectory(database.getDatabaseName());
@@ -234,6 +238,7 @@ public synchronized void renameDatabase(String databaseName, String newDatabaseN
         requireNonNull(databaseName, "databaseName is null");
         requireNonNull(newDatabaseName, "newDatabaseName is null");
 
+        verifyDatabaseNameLength(newDatabaseName);
         getRequiredDatabase(databaseName);
         verifyDatabaseNotExists(newDatabaseName);
 
@@ -283,6 +288,13 @@ private Database getRequiredDatabase(String databaseName)
                 .orElseThrow(() -> new SchemaNotFoundException(databaseName));
     }
 
+    private void verifyDatabaseNameLength(String databaseName)
+    {
+        if (databaseName.length() > MAX_DATABASE_NAME_LENGTH) {
+            throw new TrinoException(NOT_SUPPORTED, format("Schema name must be shorter than or equal to '%s' characters but got '%s'", MAX_DATABASE_NAME_LENGTH, databaseName.length()));
+        }
+    }
+
     private void verifyDatabaseNotExists(String databaseName)
     {
         if (getDatabase(databaseName).isPresent()) {

plugin/trino-hive/src/test/java/io/trino/plugin/hive/BaseHiveConnectorTest.java

@@ -8486,13 +8486,13 @@ protected TestTable createTableWithDefaultColumns()
     protected OptionalInt maxSchemaNameLength()
     {
         // This value depends on metastore type
-        return OptionalInt.of(255 - "..".length() - ".trinoSchema.crc".length());
+        return OptionalInt.of(128);
     }
 
     @Override
     protected void verifySchemaNameLengthFailurePermissible(Throwable e)
     {
-        assertThat(e).hasMessageMatching("Could not (write|rename) database schema");
+        assertThat(e).hasMessageMatching("Schema name must be shorter than or equal to '128' characters but got '129'");
     }
 
     @Override

plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/BaseIcebergConnectorTest.java

@@ -5611,13 +5611,13 @@ public Object[][] targetAndSourceWithDifferentPartitioning()
     protected OptionalInt maxSchemaNameLength()
     {
         // This value depends on metastore type
-        return OptionalInt.of(255 - "..".length() - ".trinoSchema.crc".length());
+        return OptionalInt.of(128);
     }
 
     @Override
     protected void verifySchemaNameLengthFailurePermissible(Throwable e)
     {
-        assertThat(e).hasMessageMatching("Could not (write|rename) database schema");
+        assertThat(e).hasMessageMatching("Schema name must be shorter than or equal to '128' characters but got '129'");
     }
 
     @Override