Add test to verify failure for renaming to long table name

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

@@ -700,17 +700,22 @@ protected void renameTable(ConnectorSession session, String catalogName, String
             ConnectorIdentity identity = session.getIdentity();
             String newRemoteSchemaName = identifierMapping.toRemoteSchemaName(identity, connection, newSchemaName);
             String newRemoteTableName = identifierMapping.toRemoteTableName(identity, connection, newRemoteSchemaName, newTableName);
-            String sql = format(
-                    "ALTER TABLE %s RENAME TO %s",
-                    quoted(catalogName, remoteSchemaName, remoteTableName),
-                    quoted(catalogName, newRemoteSchemaName, newRemoteTableName));
+            String sql = renameTableSql(catalogName, remoteSchemaName, remoteTableName, newRemoteSchemaName, newRemoteTableName);
             execute(connection, sql);
         }
         catch (SQLException e) {
             throw new TrinoException(JDBC_ERROR, e);
         }
     }
 
+    protected String renameTableSql(String catalogName, String remoteSchemaName, String remoteTableName, String newRemoteSchemaName, String newRemoteTableName)
+    {
+        return format(
+                "ALTER TABLE %s RENAME TO %s",
+                quoted(catalogName, remoteSchemaName, remoteTableName),
+                quoted(catalogName, newRemoteSchemaName, newRemoteTableName));
+    }
+
     @Override
     public void finishInsertTable(ConnectorSession session, JdbcOutputTableHandle handle)
     {

plugin/trino-clickhouse/src/main/java/io/trino/plugin/clickhouse/ClickHouseClient.java

@@ -53,7 +53,6 @@
 import io.trino.spi.connector.ColumnMetadata;
 import io.trino.spi.connector.ConnectorSession;
 import io.trino.spi.connector.ConnectorTableMetadata;
-import io.trino.spi.connector.SchemaTableName;
 import io.trino.spi.type.CharType;
 import io.trino.spi.type.DecimalType;
 import io.trino.spi.type.Decimals;
@@ -474,14 +473,13 @@ public void dropTable(ConnectorSession session, JdbcTableHandle handle)
     }
 
     @Override
-    protected void renameTable(ConnectorSession session, String catalogName, String schemaName, String tableName, SchemaTableName newTable)
+    protected String renameTableSql(String catalogName, String remoteSchemaName, String remoteTableName, String newRemoteSchemaName, String newRemoteTableName)
     {
-        String sql = format("RENAME TABLE %s.%s TO %s.%s",
-                quoted(schemaName),
-                quoted(tableName),
-                quoted(newTable.getSchemaName()),
-                quoted(newTable.getTableName()));
-        execute(session, sql);
+        return format("RENAME TABLE %s.%s TO %s.%s",
+                quoted(remoteSchemaName),
+                quoted(remoteTableName),
+                quoted(newRemoteSchemaName),
+                quoted(newRemoteTableName));
     }
 
     @Override

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

@@ -731,6 +731,30 @@ protected void verifySchemaNameLengthFailurePermissible(Throwable e)
         assertThat(e).hasMessageContaining("File name too long");
     }
 
+    @Override
+    public void testRenameTableToLongTableName()
+    {
+        // Override because ClickHouse connector can rename to a table which can't be dropped
+        String sourceTableName = "test_source_long_table_name_" + randomTableSuffix();
+        assertUpdate("CREATE TABLE " + sourceTableName + " AS SELECT 123 x", 1);
+
+        String baseTableName = "test_target_long_table_name_" + randomTableSuffix();
+        // The max length is different from CREATE TABLE case
+        String validTargetTableName = baseTableName + "z".repeat(255 - ".sql".length() - baseTableName.length());
+
+        assertUpdate("ALTER TABLE " + sourceTableName + " RENAME TO " + validTargetTableName);
+        assertTrue(getQueryRunner().tableExists(getSession(), validTargetTableName));
+        assertQuery("SELECT x FROM " + validTargetTableName, "VALUES 123");
+        assertThatThrownBy(() -> assertUpdate("DROP TABLE " + validTargetTableName))
+                .hasMessageMatching("(?s).*(Bad path syntax|File name too long).*");
+
+        assertUpdate("CREATE TABLE " + sourceTableName + " AS SELECT 123 x", 1);
+        String invalidTargetTableName = validTargetTableName + "z";
+        assertThatThrownBy(() -> assertUpdate("ALTER TABLE " + sourceTableName + " RENAME TO " + invalidTargetTableName))
+                .hasMessageMatching("(?s).*(Cannot rename|File name too long).*");
+        assertFalse(getQueryRunner().tableExists(getSession(), invalidTargetTableName));
+    }
+
     @Override
     protected SqlExecutor onRemoteDatabase()
     {

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

@@ -335,6 +335,14 @@ public void testRenameTableToUnqualifiedPreservesSchema()
                 .hasStackTraceContaining("SQL: ALTER TABLE test_source_schema_");
     }
 
+    @Override
+    public void testRenameTableToLongTableName()
+    {
+        assertThatThrownBy(super::testRenameTableToLongTableName)
+                .hasMessage("Renaming managed tables is not supported")
+                .hasStackTraceContaining("SQL: ALTER TABLE test_rename_");
+    }
+
     @Override
     public void testDropNonEmptySchemaWithTable()
     {

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

@@ -8366,7 +8366,7 @@ protected OptionalInt maxTableNameLength()
     @Override
     protected void verifyTableNameLengthFailurePermissible(Throwable e)
     {
-        assertThat(e).hasMessageContaining("Failed to create directory");
+        assertThat(e).hasMessageMatching("Failed to create directory.*|Could not rename table directory");
     }
 
     private Session withTimestampPrecision(Session session, HiveTimestampPrecision precision)

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

@@ -70,7 +70,6 @@
 import java.util.Optional;
 import java.util.OptionalInt;
 import java.util.Set;
-import java.util.UUID;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 import java.util.regex.Matcher;
@@ -130,6 +129,7 @@
 import static org.testng.Assert.assertFalse;
 import static org.testng.Assert.assertNotEquals;
 import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertTrue;
 
 public abstract class BaseIcebergConnectorTest
         extends BaseConnectorTest
@@ -5480,18 +5480,43 @@ protected void verifySchemaNameLengthFailurePermissible(Throwable e)
         assertThat(e).hasMessageMatching("Could not (write|rename) database schema");
     }
 
+    @Override
+    public void testRenameTableToLongTableName()
+    {
+        // Override because the max name length is different from CREATE TABLE case
+        String sourceTableName = "test_rename_source_" + randomTableSuffix();
+        assertUpdate("CREATE TABLE " + sourceTableName + " AS SELECT 123 x", 1);
+
+        String baseTableName = "test_rename_target_" + randomTableSuffix();
+
+        int maxLength = 255;
+
+        String validTargetTableName = baseTableName + "z".repeat(maxLength - baseTableName.length());
+        assertUpdate("ALTER TABLE " + sourceTableName + " RENAME TO " + validTargetTableName);
+        assertTrue(getQueryRunner().tableExists(getSession(), validTargetTableName));
+        assertQuery("SELECT x FROM " + validTargetTableName, "VALUES 123");
+        assertUpdate("DROP TABLE " + validTargetTableName);
+
+        assertUpdate("CREATE TABLE " + sourceTableName + " AS SELECT 123 x", 1);
+        String invalidTargetTableName = validTargetTableName + "z";
+        assertThatThrownBy(() -> assertUpdate("ALTER TABLE " + sourceTableName + " RENAME TO " + invalidTargetTableName))
+                .satisfies(this::verifyTableNameLengthFailurePermissible);
+        assertFalse(getQueryRunner().tableExists(getSession(), invalidTargetTableName));
+    }
+
     @Override
     protected OptionalInt maxTableNameLength()
     {
         // This value depends on metastore type
         // The connector appends uuids to the end of all table names
-        return OptionalInt.of(255 - UUID.randomUUID().toString().length());
+        // 33 is the length of random suffix. e.g. {table name}-142763c594d54e4b9329a98f90528caf
+        return OptionalInt.of(255 - 33);
     }
 
     @Override
     protected void verifyTableNameLengthFailurePermissible(Throwable e)
     {
-        assertThat(e).hasMessageContaining("Failed to create file");
+        assertThat(e).hasMessageMatching("Failed to create file.*|Could not create new table directory");
     }
 
     private Session prepareCleanUpSession()

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

@@ -24,6 +24,7 @@
 import org.testng.annotations.Test;
 
 import java.util.Optional;
+import java.util.OptionalInt;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
@@ -678,6 +679,18 @@ protected String tableDefinitionForQueryLoggingCount()
                 ")";
     }
 
+    @Override
+    protected OptionalInt maxTableNameLength()
+    {
+        return OptionalInt.of(256);
+    }
+
+    @Override
+    protected void verifyTableNameLengthFailurePermissible(Throwable e)
+    {
+        assertThat(e).hasMessageContaining("invalid table name: identifier");
+    }
+
     private void assertTableProperty(String tableProperties, String key, String regexValue)
     {
         assertTrue(Pattern.compile(key + "\\s*=\\s*" + regexValue + ",?\\s+").matcher(tableProperties).find(),

plugin/trino-mongodb/src/main/java/io/trino/plugin/mongodb/MongoMetadata.java

@@ -58,6 +58,8 @@
 import static io.trino.spi.StandardErrorCode.NOT_SUPPORTED;
 import static io.trino.spi.connector.RetryMode.NO_RETRIES;
 import static java.lang.Math.toIntExact;
+import static java.lang.String.format;
+import static java.nio.charset.StandardCharsets.UTF_8;
 import static java.util.Locale.ENGLISH;
 import static java.util.Objects.requireNonNull;
 import static java.util.stream.Collectors.toList;
@@ -67,6 +69,8 @@ public class MongoMetadata
 {
     private static final Logger log = Logger.get(MongoMetadata.class);
 
+    private static final int MAX_QUALIFIED_IDENTIFIER_BYTE_LENGTH = 120;
+
     private final MongoSession mongoSession;
 
     private final AtomicReference<Runnable> rollbackAction = new AtomicReference<>();
@@ -205,6 +209,9 @@ public void setColumnComment(ConnectorSession session, ConnectorTableHandle tabl
     @Override
     public void renameTable(ConnectorSession session, ConnectorTableHandle tableHandle, SchemaTableName newTableName)
     {
+        if (newTableName.toString().getBytes(UTF_8).length > MAX_QUALIFIED_IDENTIFIER_BYTE_LENGTH) {
+            throw new TrinoException(NOT_SUPPORTED, format("Qualified identifier name must be shorter than or equal to '%s' bytes: '%s'", MAX_QUALIFIED_IDENTIFIER_BYTE_LENGTH, newTableName));
+        }
         MongoTableHandle table = (MongoTableHandle) tableHandle;
         mongoSession.renameTable(table.getSchemaTableName(), newTableName);
     }

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

@@ -39,6 +39,7 @@
 import java.util.Optional;
 import java.util.OptionalInt;
 
+import static io.trino.testing.sql.TestTable.randomTableSuffix;
 import static java.lang.String.format;
 import static java.nio.charset.StandardCharsets.UTF_8;
 import static org.assertj.core.api.Assertions.assertThat;
@@ -577,6 +578,27 @@ public void testAddColumnConcurrently()
         throw new SkipException("TODO");
     }
 
+    @Test
+    public void testRenameTableTo120bytesTableName()
+    {
+        String sourceTableName = "test_rename_source_" + randomTableSuffix();
+        assertUpdate("CREATE TABLE " + sourceTableName + " AS SELECT 123 x", 1);
+
+        // The new table has 120 bytes as fully qualified identifier (あ is 3 bytes char)
+        String targetTableName = "a".repeat(120 - "tpch.".length() - 3) + "あ";
+        assertThat(targetTableName.length()).isLessThan(120);
+        assertUpdate("ALTER TABLE " + sourceTableName + " RENAME TO \"" + targetTableName + "\"");
+        assertQuery("SELECT x FROM \"" + targetTableName + "\"", "VALUES 123");
+        assertUpdate("DROP TABLE \"" + targetTableName + "\"");
+
+        targetTableName = targetTableName + "z";
+        assertUpdate("CREATE TABLE " + sourceTableName + " AS SELECT 123 x", 1);
+        assertQueryFails(
+                "ALTER TABLE " + sourceTableName + " RENAME TO \"" + targetTableName + "\"",
+                "Qualified identifier name must be shorter than or equal to '120' bytes: .*");
+        assertUpdate("DROP TABLE \"" + sourceTableName + "\"");
+    }
+
     @Override
     protected OptionalInt maxSchemaNameLength()
     {
@@ -598,7 +620,7 @@ protected OptionalInt maxTableNameLength()
     @Override
     protected void verifyTableNameLengthFailurePermissible(Throwable e)
     {
-        assertThat(e).hasMessageMatching(".*fully qualified namespace .* is too long.*");
+        assertThat(e).hasMessageMatching(".*fully qualified namespace .* is too long.*|Qualified identifier name must be shorter than or equal to '120'.*");
     }
 
     private void assertOneNotNullResult(String query)

plugin/trino-oracle/src/main/java/io/trino/plugin/oracle/OracleClient.java

@@ -280,18 +280,17 @@ protected void renameTable(ConnectorSession session, String catalogName, String
             throw new TrinoException(NOT_SUPPORTED, "This connector does not support renaming tables across schemas");
         }
 
-        String newTableName = newTable.getTableName().toUpperCase(ENGLISH);
-        String sql = format(
+        super.renameTable(session, catalogName, schemaName, tableName, newTable);
+    }
+
+    @Override
+    protected String renameTableSql(String catalogName, String remoteSchemaName, String remoteTableName, String newRemoteSchemaName, String newRemoteTableName)
+    {
+        String newTableName = newRemoteTableName.toUpperCase(ENGLISH);
+        return format(
                 "ALTER TABLE %s RENAME TO %s",
-                quoted(catalogName, schemaName, tableName),
+                quoted(catalogName, remoteSchemaName, remoteTableName),
                 quoted(newTableName));
-
-        try (Connection connection = connectionFactory.openConnection(session)) {
-            execute(connection, sql);
-        }
-        catch (SQLException e) {
-            throw new TrinoException(JDBC_ERROR, e);
-        }
     }
 
     @Override

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

@@ -367,11 +367,16 @@ protected void renameTable(ConnectorSession session, String catalogName, String
             throw new TrinoException(NOT_SUPPORTED, "This connector does not support renaming tables across schemas");
         }
 
-        String sql = format(
+        super.renameTable(session, catalogName, schemaName, tableName, newTable);
+    }
+
+    @Override
+    protected String renameTableSql(String catalogName, String remoteSchemaName, String remoteTableName, String newRemoteSchemaName, String newRemoteTableName)
+    {
+        return format(
                 "ALTER TABLE %s RENAME TO %s",
-                quoted(catalogName, schemaName, tableName),
-                quoted(newTable.getTableName()));
-        execute(session, sql);
+                quoted(catalogName, remoteSchemaName, remoteTableName),
+                quoted(newRemoteTableName));
     }
 
     @Override

plugin/trino-redshift/src/main/java/io/trino/plugin/redshift/RedshiftClient.java

@@ -109,11 +109,16 @@ protected void renameTable(ConnectorSession session, String catalogName, String
             throw new TrinoException(NOT_SUPPORTED, "This connector does not support renaming tables across schemas");
         }
 
-        String sql = format(
+        super.renameTable(session, catalogName, schemaName, tableName, newTable);
+    }
+
+    @Override
+    protected String renameTableSql(String catalogName, String remoteSchemaName, String remoteTableName, String newRemoteSchemaName, String newRemoteTableName)
+    {
+        return format(
                 "ALTER TABLE %s RENAME TO %s",
-                quoted(catalogName, schemaName, tableName),
-                quoted(newTable.getTableName()));
-        execute(session, sql);
+                quoted(catalogName, remoteSchemaName, remoteTableName),
+                quoted(newRemoteTableName));
     }
 
     @Override

plugin/trino-sqlserver/src/main/java/io/trino/plugin/sqlserver/SqlServerClient.java

@@ -84,6 +84,7 @@
 
 import java.sql.CallableStatement;
 import java.sql.Connection;
+import java.sql.DatabaseMetaData;
 import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.SQLException;
@@ -277,18 +278,33 @@ protected boolean isTableLockNeeded(ConnectorSession session)
         return isBulkCopyForWrite(session) && isBulkCopyForWriteLockDestinationTable(session);
     }
 
+    @Override
+    protected void verifyTableName(DatabaseMetaData databaseMetadata, String tableName)
+            throws SQLException
+    {
+        // SQL Server truncates table name to the max length silently when renaming a table
+        if (tableName.length() > databaseMetadata.getMaxTableNameLength()) {
+            throw new TrinoException(NOT_SUPPORTED, format("Table name must be shorter than or equal to '%s' characters but got '%s'", databaseMetadata.getMaxTableNameLength(), tableName.length()));
+        }
+    }
+
     @Override
     protected void renameTable(ConnectorSession session, String catalogName, String schemaName, String tableName, SchemaTableName newTable)
     {
         if (!schemaName.equals(newTable.getSchemaName())) {
             throw new TrinoException(NOT_SUPPORTED, "This connector does not support renaming tables across schemas");
         }
 
-        String sql = format(
+        super.renameTable(session, catalogName, schemaName, tableName, newTable);
+    }
+
+    @Override
+    protected String renameTableSql(String catalogName, String remoteSchemaName, String remoteTableName, String newRemoteSchemaName, String newRemoteTableName)
+    {
+        return format(
                 "sp_rename %s, %s",
-                singleQuote(catalogName, schemaName, tableName),
-                singleQuote(newTable.getTableName()));
-        execute(session, sql);
+                singleQuote(catalogName, remoteSchemaName, remoteTableName),
+                singleQuote(newRemoteTableName));
     }
 
     @Override

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

@@ -572,7 +572,7 @@ protected OptionalInt maxTableNameLength()
     @Override
     protected void verifyTableNameLengthFailurePermissible(Throwable e)
     {
-        assertThat(e).hasMessageMatching("The identifier that starts with '.*' is too long. Maximum length is 128.");
+        assertThat(e).hasMessageMatching("(The identifier that starts with '.*' is too long. Maximum length is 128.|Table name must be shorter than or equal to '128' characters but got '129')");
     }
 
     private String getLongInClause(int start, int length)