Fix failure when creating a schema with non-lowercase username on Iceberg Glue catalog

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

@@ -698,6 +698,15 @@ public void testSchemaAuthorizationForRole()
         assertUpdate(admin, "DROP ROLE authorized_users IN hive");
     }
 
+    @Override
+    public void testCreateSchemaWithNonLowercaseOwnerName()
+    {
+        // Override because HivePrincipal's username is case-sensitive unlike TrinoPrincipal
+        assertThatThrownBy(super::testCreateSchemaWithNonLowercaseOwnerName)
+                .hasMessageContaining("Access Denied: Cannot create schema")
+                .hasStackTraceContaining("CREATE SCHEMA");
+    }
+
     @Test
     public void testCreateSchemaWithAuthorizationForUser()
     {

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

@@ -96,4 +96,13 @@ public void testShowCreateTable()
                         "   format = 'ORC'\n" +
                         ")");
     }
+
+    @Override
+    public void testCreateSchemaWithNonLowercaseOwnerName()
+    {
+        // Override because HivePrincipal's username is case-sensitive unlike TrinoPrincipal
+        assertThatThrownBy(super::testCreateSchemaWithNonLowercaseOwnerName)
+                .hasMessageContaining("Access Denied: Cannot create schema")
+                .hasStackTraceContaining("CREATE SCHEMA");
+    }
 }

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/catalog/glue/TrinoGlueCatalog.java

@@ -253,7 +253,7 @@ public Optional<TrinoPrincipal> getNamespacePrincipal(ConnectorSession session,
     public void createNamespace(ConnectorSession session, String namespace, Map<String, Object> properties, TrinoPrincipal owner)
     {
         checkArgument(owner.getType() == PrincipalType.USER, "Owner type must be USER");
-        checkArgument(owner.getName().equals(session.getUser()), "Explicit schema owner is not supported");
+        checkArgument(owner.getName().equals(session.getUser().toLowerCase(ENGLISH)), "Explicit schema owner is not supported");
 
         try {
             stats.getCreateDatabase().call(() ->

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

@@ -57,6 +57,13 @@ public void testCreateSchema()
                 .hasMessage("Creating schema in Kudu connector not allowed if schema emulation is disabled.");
     }
 
+    @Override
+    public void testCreateSchemaWithNonLowercaseOwnerName()
+    {
+        assertThatThrownBy(super::testCreateSchemaWithNonLowercaseOwnerName)
+                .hasMessage("Creating schema in Kudu connector not allowed if schema emulation is disabled.");
+    }
+
     @Test
     @Override
     public void testRenameTableAcrossSchemas()

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

@@ -139,6 +139,13 @@ public void testCreateSchema()
                 .hasMessage("Creating schema in Kudu connector not allowed if schema emulation is disabled.");
     }
 
+    @Override
+    public void testCreateSchemaWithNonLowercaseOwnerName()
+    {
+        assertThatThrownBy(super::testCreateSchemaWithNonLowercaseOwnerName)
+                .hasMessage("Creating schema in Kudu connector not allowed if schema emulation is disabled.");
+    }
+
     @Override
     public void testCreateSchemaWithLongName()
     {

testing/trino-testing/src/main/java/io/trino/testing/BaseConnectorSmokeTest.java

@@ -14,6 +14,8 @@
 package io.trino.testing;
 
 import com.google.common.collect.ImmutableList;
+import io.trino.Session;
+import io.trino.spi.security.Identity;
 import io.trino.testing.sql.TestTable;
 import io.trino.tpch.TpchTable;
 import org.testng.SkipException;
@@ -315,6 +317,22 @@ public void testCreateSchema()
         assertUpdate("DROP SCHEMA " + schemaName);
     }
 
+    @Test
+    public void testCreateSchemaWithNonLowercaseOwnerName()
+    {
+        skipTestUnless(hasBehavior(SUPPORTS_CREATE_SCHEMA));
+
+        Session newSession = Session.builder(getSession())
+                .setIdentity(Identity.ofUser("ADMIN"))
+                .build();
+        String schemaName = "test_schema_create_uppercase_owner_name_" + randomNameSuffix();
+        assertUpdate(newSession, createSchemaSql(schemaName));
+        assertThat(query(newSession, "SHOW SCHEMAS"))
+                .skippingTypesCheck()
+                .containsAll(format("VALUES '%s'", schemaName));
+        assertUpdate(newSession, "DROP SCHEMA " + schemaName);
+    }
+
     @Test
     public void testRenameSchema()
     {

testing/trino-testing/src/main/java/io/trino/testing/BaseConnectorTest.java

@@ -34,6 +34,7 @@
 import io.trino.server.BasicQueryInfo;
 import io.trino.spi.connector.ConnectorSession;
 import io.trino.spi.connector.MaterializedViewFreshness;
+import io.trino.spi.security.Identity;
 import io.trino.sql.planner.OptimizerConfig.JoinDistributionType;
 import io.trino.sql.planner.Plan;
 import io.trino.sql.planner.assertions.PlanMatchPattern;
@@ -2881,6 +2882,22 @@ public void testCreateTable()
         assertFalse(getQueryRunner().tableExists(getSession(), tableNameLike));
     }
 
+    @Test
+    public void testCreateSchemaWithNonLowercaseOwnerName()
+    {
+        skipTestUnless(hasBehavior(SUPPORTS_CREATE_SCHEMA));
+
+        Session newSession = Session.builder(getSession())
+                .setIdentity(Identity.ofUser("ADMIN"))
+                .build();
+        String schemaName = "test_schema_create_uppercase_owner_name_" + randomNameSuffix();
+        assertUpdate(newSession, createSchemaSql(schemaName));
+        assertThat(query(newSession, "SHOW SCHEMAS"))
+                .skippingTypesCheck()
+                .containsAll(format("VALUES '%s'", schemaName));
+        assertUpdate(newSession, "DROP SCHEMA " + schemaName);
+    }
+
     @Test
     public void testCreateSchemaWithLongName()
     {