Verify schema name length when creating or renaming in Postgres

plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/BaseJdbcClient.java

@@ -884,6 +884,7 @@ public void createSchema(ConnectorSession session, String schemaName)
         ConnectorIdentity identity = session.getIdentity();
         try (Connection connection = connectionFactory.openConnection(session)) {
             schemaName = identifierMapping.toRemoteSchemaName(identity, connection, schemaName);
+            verifySchemaName(connection.getMetaData(), schemaName);
             execute(connection, createSchemaSql(schemaName));
         }
         catch (SQLException e) {
@@ -921,6 +922,7 @@ public void renameSchema(ConnectorSession session, String schemaName, String new
         try (Connection connection = connectionFactory.openConnection(session)) {
             String remoteSchemaName = identifierMapping.toRemoteSchemaName(identity, connection, schemaName);
             String newRemoteSchemaName = identifierMapping.toRemoteSchemaName(identity, connection, newSchemaName);
+            verifySchemaName(connection.getMetaData(), newRemoteSchemaName);
             execute(connection, renameSchemaSql(remoteSchemaName, newRemoteSchemaName));
         }
         catch (SQLException e) {
@@ -1079,6 +1081,12 @@ public void truncateTable(ConnectorSession session, JdbcTableHandle handle)
         execute(session, sql);
     }
 
+    protected void verifySchemaName(DatabaseMetaData databaseMetadata, String schemaName)
+            throws SQLException
+    {
+        // expect remote databases throw an exception for unsupported schema names
+    }
+
     protected void verifyTableName(DatabaseMetaData databaseMetadata, String tableName)
             throws SQLException
     {

plugin/trino-bigquery/src/test/java/io/trino/plugin/bigquery/TestBigQueryConnectorTest.java

@@ -729,6 +729,18 @@ public void testMissingWildcardTable()
                 .hasMessageEndingWith("does not match any table.");
     }
 
+    @Override
+    protected OptionalInt maxSchemaNameLength()
+    {
+        return OptionalInt.of(1024);
+    }
+
+    @Override
+    protected void verifySchemaNameLengthFailurePermissible(Throwable e)
+    {
+        assertThat(e).hasMessageContaining("Invalid dataset ID");
+    }
+
     @Override
     protected OptionalInt maxTableNameLength()
     {

plugin/trino-clickhouse/src/test/java/io/trino/plugin/clickhouse/BaseClickHouseConnectorTest.java

@@ -33,6 +33,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.OptionalInt;
 
 import static io.trino.plugin.clickhouse.ClickHouseTableProperties.ENGINE_PROPERTY;
 import static io.trino.plugin.clickhouse.ClickHouseTableProperties.ORDER_BY_PROPERTY;
@@ -693,6 +694,43 @@ public void testCreateTableWithLongTableName()
         assertTrue(getQueryRunner().tableExists(getSession(), validTableName));
     }
 
+    @Override
+    public void testRenameSchemaToLongName()
+    {
+        // Override because the max length is different from CREATE SCHEMA case
+        String sourceTableName = "test_rename_source_" + randomTableSuffix();
+        assertUpdate("CREATE SCHEMA " + sourceTableName);
+
+        String baseSchemaName = "test_rename_target_" + randomTableSuffix();
+
+        // The numeric value depends on file system
+        int maxLength = 255 - ".sql".length();
+
+        String validTargetSchemaName = baseSchemaName + "z".repeat(maxLength - baseSchemaName.length());
+        assertUpdate("ALTER SCHEMA " + sourceTableName + " RENAME TO " + validTargetSchemaName);
+        assertThat(computeActual("SHOW SCHEMAS").getOnlyColumnAsSet()).contains(validTargetSchemaName);
+        assertUpdate("DROP SCHEMA " + validTargetSchemaName);
+
+        assertUpdate("CREATE SCHEMA " + sourceTableName);
+        String invalidTargetSchemaName = validTargetSchemaName + "z";
+        assertThatThrownBy(() -> assertUpdate("ALTER SCHEMA " + sourceTableName + " RENAME TO " + invalidTargetSchemaName))
+                .satisfies(this::verifySchemaNameLengthFailurePermissible);
+        assertThat(computeActual("SHOW SCHEMAS").getOnlyColumnAsSet()).doesNotContain(invalidTargetSchemaName);
+    }
+
+    @Override
+    protected OptionalInt maxSchemaNameLength()
+    {
+        // The numeric value depends on file system
+        return OptionalInt.of(255 - ".sql.tmp".length());
+    }
+
+    @Override
+    protected void verifySchemaNameLengthFailurePermissible(Throwable e)
+    {
+        assertThat(e).hasMessageContaining("File name too long");
+    }
+
     @Override
     protected SqlExecutor onRemoteDatabase()
     {

plugin/trino-delta-lake/src/test/java/io/trino/plugin/deltalake/BaseDeltaLakeMinioConnectorTest.java

@@ -518,6 +518,18 @@ protected String createSchemaSql(String schemaName)
         return "CREATE SCHEMA " + schemaName + " WITH (location = 's3://" + bucketName + "/" + schemaName + "')";
     }
 
+    @Override
+    protected OptionalInt maxSchemaNameLength()
+    {
+        return OptionalInt.of(128);
+    }
+
+    @Override
+    protected void verifySchemaNameLengthFailurePermissible(Throwable e)
+    {
+        assertThat(e).hasMessageMatching("(?s)(.*Read timed out)|(.*\"`NAME`\" that has maximum length of 128.*)");
+    }
+
     @Override
     protected OptionalInt maxTableNameLength()
     {

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

@@ -8333,6 +8333,19 @@ protected TestTable createTableWithDefaultColumns()
         throw new SkipException("Hive connector does not support column default values");
     }
 
+    @Override
+    protected OptionalInt maxSchemaNameLength()
+    {
+        // This value depends on metastore type
+        return OptionalInt.of(255 - "..".length() - ".trinoSchema.crc".length());
+    }
+
+    @Override
+    protected void verifySchemaNameLengthFailurePermissible(Throwable e)
+    {
+        assertThat(e).hasMessageMatching("Could not (write|rename) database schema");
+    }
+
     @Override
     protected OptionalInt maxTableNameLength()
     {

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

@@ -5221,6 +5221,19 @@ public void testReadFromVersionedTableWithExpiredHistory()
         assertQueryFails("SELECT * FROM " + tableName + " FOR TIMESTAMP AS OF " + timestampLiteral(v1EpochMillis, 9), "No version history table .* at or before .*");
     }
 
+    @Override
+    protected OptionalInt maxSchemaNameLength()
+    {
+        // This value depends on metastore type
+        return OptionalInt.of(255 - "..".length() - ".trinoSchema.crc".length());
+    }
+
+    @Override
+    protected void verifySchemaNameLengthFailurePermissible(Throwable e)
+    {
+        assertThat(e).hasMessageMatching("Could not (write|rename) database schema");
+    }
+
     @Override
     protected OptionalInt maxTableNameLength()
     {

plugin/trino-kudu/src/test/java/io/trino/plugin/kudu/TestKuduConnectorTest.java

@@ -93,6 +93,14 @@ public void testCreateSchema()
                 .hasMessage("Creating schema in Kudu connector not allowed if schema emulation is disabled.");
     }
 
+    @Override
+    public void testCreateSchemaWithLongName()
+    {
+        // TODO: Add a test to BaseKuduConnectorSmokeTest
+        assertThatThrownBy(super::testCreateSchemaWithLongName)
+                .hasMessage("Creating schema in Kudu connector not allowed if schema emulation is disabled.");
+    }
+
     @Test
     @Override
     public void testDropNonEmptySchemaWithTable()

plugin/trino-mariadb/src/test/java/io/trino/plugin/mariadb/BaseMariaDbConnectorTest.java

@@ -304,6 +304,18 @@ protected String errorMessageForInsertIntoNotNullColumn(String columnName)
         return format("Failed to insert data: .* \\(conn=.*\\) Field '%s' doesn't have a default value", columnName);
     }
 
+    @Override
+    protected OptionalInt maxSchemaNameLength()
+    {
+        return OptionalInt.of(64);
+    }
+
+    @Override
+    protected void verifySchemaNameLengthFailurePermissible(Throwable e)
+    {
+        assertThat(e).hasMessageContaining("Incorrect database name");
+    }
+
     @Override
     protected OptionalInt maxTableNameLength()
     {

plugin/trino-mongodb/src/test/java/io/trino/plugin/mongodb/BaseMongoConnectorTest.java

@@ -577,6 +577,18 @@ public void testAddColumnConcurrently()
         throw new SkipException("TODO");
     }
 
+    @Override
+    protected OptionalInt maxSchemaNameLength()
+    {
+        return OptionalInt.of(63);
+    }
+
+    @Override
+    protected void verifySchemaNameLengthFailurePermissible(Throwable e)
+    {
+        assertThat(e).hasMessageContaining("Invalid database name");
+    }
+
     @Override
     protected OptionalInt maxTableNameLength()
     {

plugin/trino-mysql/src/test/java/io/trino/plugin/mysql/BaseMySqlConnectorTest.java

@@ -383,6 +383,18 @@ private String getLongInClause(int start, int length)
         return "orderkey IN (" + longValues + ")";
     }
 
+    @Override
+    protected OptionalInt maxSchemaNameLength()
+    {
+        return OptionalInt.of(64);
+    }
+
+    @Override
+    protected void verifySchemaNameLengthFailurePermissible(Throwable e)
+    {
+        assertThat(e).hasMessageMatching("Identifier name .* is too long");
+    }
+
     @Override
     protected OptionalInt maxTableNameLength()
     {

plugin/trino-oracle/src/test/java/io/trino/plugin/oracle/BaseOracleConnectorTest.java

@@ -476,6 +476,18 @@ protected void verifyConcurrentAddColumnFailurePermissible(Exception e)
                 .hasMessage("ORA-14411: The DDL cannot be run concurrently with other DDLs\n");
     }
 
+    @Override
+    protected OptionalInt maxSchemaNameLength()
+    {
+        return OptionalInt.of(30);
+    }
+
+    @Override
+    protected void verifySchemaNameLengthFailurePermissible(Throwable e)
+    {
+        assertThat(e).hasMessage("ORA-00972: identifier is too long\n");
+    }
+
     @Override
     protected OptionalInt maxTableNameLength()
     {

plugin/trino-phoenix5/src/test/java/io/trino/plugin/phoenix5/TestPhoenixConnectorTest.java

@@ -664,6 +664,13 @@ protected void verifyConcurrentAddColumnFailurePermissible(Exception e)
                 .hasMessageContaining("Concurrent modification to table");
     }
 
+    @Override
+    public void testCreateSchemaWithLongName()
+    {
+        // TODO: Find the maximum table schema length in Phoenix and enable this test.
+        throw new SkipException("TODO");
+    }
+
     @Override
     public void testCreateTableWithLongTableName()
     {

plugin/trino-postgresql/src/main/java/io/trino/plugin/postgresql/PostgreSqlClient.java

@@ -1005,6 +1005,16 @@ protected boolean isSupportedJoinCondition(ConnectorSession session, JdbcJoinCon
         return true;
     }
 
+    @Override
+    protected void verifySchemaName(DatabaseMetaData databaseMetadata, String schemaName)
+            throws SQLException
+    {
+        // PostgreSQL truncates schema name to 63 chars silently
+        if (schemaName.length() > databaseMetadata.getMaxSchemaNameLength()) {
+            throw new TrinoException(NOT_SUPPORTED, format("Schema name must be shorter than or equal to '%s' characters but got '%s'", databaseMetadata.getMaxSchemaNameLength(), schemaName.length()));
+        }
+    }
+
     @Override
     protected void verifyTableName(DatabaseMetaData databaseMetadata, String tableName)
             throws SQLException

plugin/trino-postgresql/src/test/java/io/trino/plugin/postgresql/TestPostgreSqlConnectorTest.java

@@ -987,6 +987,18 @@ protected Session joinPushdownEnabled(Session session)
                 .build();
     }
 
+    @Override
+    protected OptionalInt maxSchemaNameLength()
+    {
+        return OptionalInt.of(63);
+    }
+
+    @Override
+    protected void verifySchemaNameLengthFailurePermissible(Throwable e)
+    {
+        assertThat(e).hasMessage("Schema name must be shorter than or equal to '63' characters but got '64'");
+    }
+
     @Override
     protected OptionalInt maxTableNameLength()
     {

plugin/trino-singlestore/src/test/java/io/trino/plugin/singlestore/TestSingleStoreConnectorTest.java

@@ -360,6 +360,19 @@ private String getLongInClause(int start, int length)
         return "orderkey IN (" + longValues + ")";
     }
 
+    @Override
+    protected OptionalInt maxSchemaNameLength()
+    {
+        return OptionalInt.of(62);
+    }
+
+    @Override
+    protected void verifySchemaNameLengthFailurePermissible(Throwable e)
+    {
+        // The error message says 60 char, but the actual limitation is 62
+        assertThat(e).hasMessageContaining("Distributed MemSQL requires the length of the database name to be at most 60 characters");
+    }
+
     @Override
     protected OptionalInt maxTableNameLength()
     {

plugin/trino-sqlserver/src/test/java/io/trino/plugin/sqlserver/BaseSqlServerConnectorTest.java

@@ -551,6 +551,18 @@ protected String errorMessageForInsertIntoNotNullColumn(String columnName)
         return format("Cannot insert the value NULL into column '%s'.*", columnName);
     }
 
+    @Override
+    protected OptionalInt maxSchemaNameLength()
+    {
+        return OptionalInt.of(128);
+    }
+
+    @Override
+    protected void verifySchemaNameLengthFailurePermissible(Throwable e)
+    {
+        assertThat(e).hasMessageMatching("The identifier that starts with '.*' is too long. Maximum length is 128.");
+    }
+
     @Override
     protected OptionalInt maxTableNameLength()
     {