Pass properties when checking access for CREATE SCHEMA

core/trino-main/src/main/java/io/trino/execution/CreateSchemaTask.java

@@ -88,17 +88,6 @@ static ListenableFuture<Void> internalExecute(
     {
         CatalogSchemaName schema = createCatalogSchemaName(session, statement, Optional.of(statement.getSchemaName()));
 
-        // TODO: validate that catalog exists
-
-        accessControl.checkCanCreateSchema(session.toSecurityContext(), schema);
-
-        if (plannerContext.getMetadata().schemaExists(session, schema)) {
-            if (!statement.isNotExists()) {
-                throw semanticException(SCHEMA_ALREADY_EXISTS, statement, "Schema '%s' already exists", schema);
-            }
-            return immediateVoidFuture();
-        }
-
         String catalogName = schema.getCatalogName();
         CatalogHandle catalogHandle = getRequiredCatalogHandle(plannerContext.getMetadata(), session, statement, catalogName);
 
@@ -112,6 +101,15 @@ static ListenableFuture<Void> internalExecute(
                 bindParameters(statement, parameters),
                 true);
 
+        accessControl.checkCanCreateSchema(session.toSecurityContext(), schema, properties);
+
+        if (plannerContext.getMetadata().schemaExists(session, schema)) {
+            if (!statement.isNotExists()) {
+                throw semanticException(SCHEMA_ALREADY_EXISTS, statement, "Schema '%s' already exists", schema);
+            }
+            return immediateVoidFuture();
+        }
+
         TrinoPrincipal principal = getCreatePrincipal(statement, session, plannerContext.getMetadata(), catalogName);
         try {
             plannerContext.getMetadata().createSchema(session, schema, properties, principal);

core/trino-main/src/main/java/io/trino/security/AccessControl.java

@@ -109,7 +109,7 @@ public interface AccessControl
      *
      * @throws AccessDeniedException if not allowed
      */
-    void checkCanCreateSchema(SecurityContext context, CatalogSchemaName schemaName);
+    void checkCanCreateSchema(SecurityContext context, CatalogSchemaName schemaName, Map<String, Object> properties);
 
     /**
      * Check if identity is allowed to drop the specified schema.

core/trino-main/src/main/java/io/trino/security/AccessControlManager.java

@@ -303,16 +303,16 @@ public Set<String> filterCatalogs(SecurityContext securityContext, Set<String> c
     }
 
     @Override
-    public void checkCanCreateSchema(SecurityContext securityContext, CatalogSchemaName schemaName)
+    public void checkCanCreateSchema(SecurityContext securityContext, CatalogSchemaName schemaName, Map<String, Object> properties)
     {
         requireNonNull(securityContext, "securityContext is null");
         requireNonNull(schemaName, "schemaName is null");
 
         checkCanAccessCatalog(securityContext, schemaName.getCatalogName());
 
-        systemAuthorizationCheck(control -> control.checkCanCreateSchema(securityContext.toSystemSecurityContext(), schemaName));
+        systemAuthorizationCheck(control -> control.checkCanCreateSchema(securityContext.toSystemSecurityContext(), schemaName, properties));
 
-        catalogAuthorizationCheck(schemaName.getCatalogName(), securityContext, (control, context) -> control.checkCanCreateSchema(context, schemaName.getSchemaName()));
+        catalogAuthorizationCheck(schemaName.getCatalogName(), securityContext, (control, context) -> control.checkCanCreateSchema(context, schemaName.getSchemaName(), properties));
     }
 
     @Override

core/trino-main/src/main/java/io/trino/security/AllowAllAccessControl.java

@@ -79,7 +79,7 @@ public Set<String> filterCatalogs(SecurityContext context, Set<String> catalogs)
     }
 
     @Override
-    public void checkCanCreateSchema(SecurityContext context, CatalogSchemaName schemaName)
+    public void checkCanCreateSchema(SecurityContext context, CatalogSchemaName schemaName, Map<String, Object> properties)
     {
     }
 

core/trino-main/src/main/java/io/trino/security/DenyAllAccessControl.java

@@ -152,7 +152,7 @@ public Set<String> filterCatalogs(SecurityContext context, Set<String> catalogs)
     }
 
     @Override
-    public void checkCanCreateSchema(SecurityContext context, CatalogSchemaName schemaName)
+    public void checkCanCreateSchema(SecurityContext context, CatalogSchemaName schemaName, Map<String, Object> properties)
     {
         denyCreateSchema(schemaName.toString());
     }

core/trino-main/src/main/java/io/trino/security/ForwardingAccessControl.java

@@ -108,9 +108,9 @@ public Set<String> filterCatalogs(SecurityContext context, Set<String> catalogs)
     }
 
     @Override
-    public void checkCanCreateSchema(SecurityContext context, CatalogSchemaName schemaName)
+    public void checkCanCreateSchema(SecurityContext context, CatalogSchemaName schemaName, Map<String, Object> properties)
     {
-        delegate().checkCanCreateSchema(context, schemaName);
+        delegate().checkCanCreateSchema(context, schemaName, properties);
     }
 
     @Override

core/trino-main/src/main/java/io/trino/security/InjectedConnectorAccessControl.java

@@ -14,6 +14,7 @@
 package io.trino.security;
 
 import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
 import io.trino.metadata.QualifiedObjectName;
 import io.trino.spi.TrinoException;
 import io.trino.spi.connector.CatalogSchemaName;
@@ -52,11 +53,18 @@ public InjectedConnectorAccessControl(AccessControl accessControl, SecurityConte
         this.catalogName = requireNonNull(catalogName, "catalogName is null");
     }
 
+    @Override
+    public void checkCanCreateSchema(ConnectorSecurityContext context, String schemaName, Map<String, Object> properties)
+    {
+        checkArgument(context == null, "context must be null");
+        accessControl.checkCanCreateSchema(securityContext, getCatalogSchemaName(schemaName), properties);
+    }
+
     @Override
     public void checkCanCreateSchema(ConnectorSecurityContext context, String schemaName)
     {
         checkArgument(context == null, "context must be null");
-        accessControl.checkCanCreateSchema(securityContext, getCatalogSchemaName(schemaName));
+        accessControl.checkCanCreateSchema(securityContext, getCatalogSchemaName(schemaName), ImmutableMap.of());
     }
 
     @Override

core/trino-main/src/main/java/io/trino/testing/AllowAllAccessControlManager.java

@@ -67,7 +67,7 @@ public Set<String> filterCatalogs(SecurityContext context, Set<String> catalogs)
     }
 
     @Override
-    public void checkCanCreateSchema(SecurityContext context, CatalogSchemaName schemaName) {}
+    public void checkCanCreateSchema(SecurityContext context, CatalogSchemaName schemaName, Map<String, Object> properties) {}
 
     @Override
     public void checkCanDropSchema(SecurityContext context, CatalogSchemaName schemaName) {}

core/trino-main/src/main/java/io/trino/testing/TestingAccessControlManager.java

@@ -309,13 +309,13 @@ public void checkCanKillQueryOwnedBy(Identity identity, Identity queryOwner)
     }
 
     @Override
-    public void checkCanCreateSchema(SecurityContext context, CatalogSchemaName schemaName)
+    public void checkCanCreateSchema(SecurityContext context, CatalogSchemaName schemaName, Map<String, Object> properties)
     {
         if (shouldDenyPrivilege(context.getIdentity().getUser(), schemaName.getSchemaName(), CREATE_SCHEMA)) {
             denyCreateSchema(schemaName.toString());
         }
         if (denyPrivileges.isEmpty()) {
-            super.checkCanCreateSchema(context, schemaName);
+            super.checkCanCreateSchema(context, schemaName, properties);
         }
     }
 

core/trino-main/src/test/java/io/trino/security/TestFileBasedSystemAccessControl.java

@@ -250,13 +250,13 @@ public void testSchemaOperations()
                     assertEquals(accessControlManager.filterSchemas(new SecurityContext(transactionId, alice, queryId), "alice-catalog", aliceSchemas), aliceSchemas);
                     assertEquals(accessControlManager.filterSchemas(new SecurityContext(transactionId, bob, queryId), "alice-catalog", aliceSchemas), ImmutableSet.of());
 
-                    accessControlManager.checkCanCreateSchema(new SecurityContext(transactionId, alice, queryId), aliceSchema);
+                    accessControlManager.checkCanCreateSchema(new SecurityContext(transactionId, alice, queryId), aliceSchema, ImmutableMap.of());
                     accessControlManager.checkCanDropSchema(new SecurityContext(transactionId, alice, queryId), aliceSchema);
                     accessControlManager.checkCanRenameSchema(new SecurityContext(transactionId, alice, queryId), aliceSchema, "new-schema");
                     accessControlManager.checkCanShowSchemas(new SecurityContext(transactionId, alice, queryId), "alice-catalog");
                 });
         assertThatThrownBy(() -> transaction(transactionManager, accessControlManager).execute(transactionId -> {
-            accessControlManager.checkCanCreateSchema(new SecurityContext(transactionId, bob, queryId), aliceSchema);
+            accessControlManager.checkCanCreateSchema(new SecurityContext(transactionId, bob, queryId), aliceSchema, ImmutableMap.of());
         })).isInstanceOf(AccessDeniedException.class)
                 .hasMessage("Access Denied: Cannot access catalog alice-catalog");
     }
@@ -277,7 +277,7 @@ public void testSchemaOperationsReadOnly()
                 });
 
         assertThatThrownBy(() -> transaction(transactionManager, accessControlManager).execute(transactionId -> {
-            accessControlManager.checkCanCreateSchema(new SecurityContext(transactionId, alice, queryId), aliceSchema);
+            accessControlManager.checkCanCreateSchema(new SecurityContext(transactionId, alice, queryId), aliceSchema, ImmutableMap.of());
         })).isInstanceOf(AccessDeniedException.class)
                 .hasMessage("Access Denied: Cannot create schema alice-catalog.schema");
 
@@ -292,7 +292,7 @@ public void testSchemaOperationsReadOnly()
                 .hasMessage("Access Denied: Cannot rename schema from alice-catalog.schema to new-schema");
 
         assertThatThrownBy(() -> transaction(transactionManager, accessControlManager).execute(transactionId -> {
-            accessControlManager.checkCanCreateSchema(new SecurityContext(transactionId, bob, queryId), aliceSchema);
+            accessControlManager.checkCanCreateSchema(new SecurityContext(transactionId, bob, queryId), aliceSchema, ImmutableMap.of());
         })).isInstanceOf(AccessDeniedException.class)
                 .hasMessage("Access Denied: Cannot access catalog alice-catalog");
     }

core/trino-spi/src/main/java/io/trino/spi/connector/ConnectorAccessControl.java

@@ -88,11 +88,22 @@
 
 public interface ConnectorAccessControl
 {
+    /**
+     * Check if identity is allowed to create the specified schema with properties.
+     *
+     * @throws io.trino.spi.security.AccessDeniedException if not allowed
+     */
+    default void checkCanCreateSchema(ConnectorSecurityContext context, String schemaName, Map<String, Object> properties)
+    {
+        denyCreateSchema(schemaName);
+    }
+
     /**
      * Check if identity is allowed to create the specified schema.
      *
      * @throws io.trino.spi.security.AccessDeniedException if not allowed
      */
+    @Deprecated
     default void checkCanCreateSchema(ConnectorSecurityContext context, String schemaName)
     {
         denyCreateSchema(schemaName);

core/trino-spi/src/main/java/io/trino/spi/security/SystemAccessControl.java

@@ -253,11 +253,22 @@ default Set<String> filterCatalogs(SystemSecurityContext context, Set<String> ca
         return emptySet();
     }
 
+    /**
+     * Check if identity is allowed to create the specified schema with properties in a catalog.
+     *
+     * @throws AccessDeniedException if not allowed
+     */
+    default void checkCanCreateSchema(SystemSecurityContext context, CatalogSchemaName schema, Map<String, Object> properties)
+    {
+        denyCreateSchema(schema.toString());
+    }
+
     /**
      * Check if identity is allowed to create the specified schema in a catalog.
      *
      * @throws AccessDeniedException if not allowed
      */
+    @Deprecated
     default void checkCanCreateSchema(SystemSecurityContext context, CatalogSchemaName schema)
     {
         denyCreateSchema(schema.toString());

lib/trino-plugin-toolkit/src/main/java/io/trino/plugin/base/classloader/ClassLoaderSafeConnectorAccessControl.java

@@ -46,6 +46,14 @@ public ClassLoaderSafeConnectorAccessControl(@ForClassLoaderSafe ConnectorAccess
         this.classLoader = requireNonNull(classLoader, "classLoader is null");
     }
 
+    @Override
+    public void checkCanCreateSchema(ConnectorSecurityContext context, String schemaName, Map<String, Object> properties)
+    {
+        try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(classLoader)) {
+            delegate.checkCanCreateSchema(context, schemaName, properties);
+        }
+    }
+
     @Override
     public void checkCanCreateSchema(ConnectorSecurityContext context, String schemaName)
     {

lib/trino-plugin-toolkit/src/main/java/io/trino/plugin/base/security/AllowAllAccessControl.java

@@ -32,6 +32,11 @@
 public class AllowAllAccessControl
         implements ConnectorAccessControl
 {
+    @Override
+    public void checkCanCreateSchema(ConnectorSecurityContext context, String schemaName, Map<String, Object> properties)
+    {
+    }
+
     @Override
     public void checkCanCreateSchema(ConnectorSecurityContext context, String schemaName)
     {

lib/trino-plugin-toolkit/src/main/java/io/trino/plugin/base/security/AllowAllSystemAccessControl.java

@@ -136,6 +136,11 @@ public Set<String> filterCatalogs(SystemSecurityContext context, Set<String> cat
         return catalogs;
     }
 
+    @Override
+    public void checkCanCreateSchema(SystemSecurityContext context, CatalogSchemaName schema, Map<String, Object> properties)
+    {
+    }
+
     @Override
     public void checkCanCreateSchema(SystemSecurityContext context, CatalogSchemaName schema)
     {

lib/trino-plugin-toolkit/src/main/java/io/trino/plugin/base/security/FileBasedAccessControl.java

@@ -139,6 +139,12 @@ public FileBasedAccessControl(CatalogName catalogName, File configFile)
         this.anySchemaPermissionsRules = anySchemaPermissionsRules.build();
     }
 
+    @Override
+    public void checkCanCreateSchema(ConnectorSecurityContext context, String schemaName, Map<String, Object> properties)
+    {
+        checkCanCreateSchema(context, schemaName);
+    }
+
     @Override
     public void checkCanCreateSchema(ConnectorSecurityContext context, String schemaName)
     {

lib/trino-plugin-toolkit/src/main/java/io/trino/plugin/base/security/FileBasedSystemAccessControl.java

@@ -456,6 +456,12 @@ public Set<String> filterCatalogs(SystemSecurityContext context, Set<String> cat
         return filteredCatalogs.build();
     }
 
+    @Override
+    public void checkCanCreateSchema(SystemSecurityContext context, CatalogSchemaName schema, Map<String, Object> properties)
+    {
+        checkCanCreateSchema(context, schema);
+    }
+
     @Override
     public void checkCanCreateSchema(SystemSecurityContext context, CatalogSchemaName schema)
     {

lib/trino-plugin-toolkit/src/main/java/io/trino/plugin/base/security/ForwardingConnectorAccessControl.java

@@ -49,6 +49,12 @@ protected ConnectorAccessControl delegate()
 
     protected abstract ConnectorAccessControl delegate();
 
+    @Override
+    public void checkCanCreateSchema(ConnectorSecurityContext context, String schemaName, Map<String, Object> properties)
+    {
+        delegate().checkCanCreateSchema(context, schemaName, properties);
+    }
+
     @Override
     public void checkCanCreateSchema(ConnectorSecurityContext context, String schemaName)
     {

lib/trino-plugin-toolkit/src/main/java/io/trino/plugin/base/security/ForwardingSystemAccessControl.java

@@ -139,6 +139,12 @@ public Set<String> filterCatalogs(SystemSecurityContext context, Set<String> cat
         return delegate().filterCatalogs(context, catalogs);
     }
 
+    @Override
+    public void checkCanCreateSchema(SystemSecurityContext context, CatalogSchemaName schema, Map<String, Object> properties)
+    {
+        delegate().checkCanCreateSchema(context, schema, properties);
+    }
+
     @Override
     public void checkCanCreateSchema(SystemSecurityContext context, CatalogSchemaName schema)
     {

lib/trino-plugin-toolkit/src/test/java/io/trino/plugin/base/security/TestFileBasedAccessControl.java

@@ -67,7 +67,7 @@ public void testEmptyFile()
     {
         ConnectorAccessControl accessControl = createAccessControl("empty.json");
 
-        accessControl.checkCanCreateSchema(UNKNOWN, "unknown");
+        accessControl.checkCanCreateSchema(UNKNOWN, "unknown", ImmutableMap.of());
         accessControl.checkCanDropSchema(UNKNOWN, "unknown");
         accessControl.checkCanRenameSchema(UNKNOWN, "unknown", "new_unknown");
         accessControl.checkCanSetSchemaAuthorization(UNKNOWN, "unknown", new TrinoPrincipal(PrincipalType.ROLE, "some_role"));
@@ -145,20 +145,21 @@ public void testSchemaRules()
     {
         ConnectorAccessControl accessControl = createAccessControl("schema.json");
 
-        accessControl.checkCanCreateSchema(ADMIN, "bob");
-        accessControl.checkCanCreateSchema(ADMIN, "staff");
-        accessControl.checkCanCreateSchema(ADMIN, "authenticated");
-        accessControl.checkCanCreateSchema(ADMIN, "test");
-
-        accessControl.checkCanCreateSchema(BOB, "bob");
-        accessControl.checkCanCreateSchema(BOB, "staff");
-        accessControl.checkCanCreateSchema(BOB, "authenticated");
-        assertDenied(() -> accessControl.checkCanCreateSchema(BOB, "test"));
-
-        assertDenied(() -> accessControl.checkCanCreateSchema(CHARLIE, "bob"));
-        assertDenied(() -> accessControl.checkCanCreateSchema(CHARLIE, "staff"));
-        accessControl.checkCanCreateSchema(CHARLIE, "authenticated");
-        assertDenied(() -> accessControl.checkCanCreateSchema(CHARLIE, "test"));
+        Map<String, Object> properties = ImmutableMap.of();
+        accessControl.checkCanCreateSchema(ADMIN, "bob", properties);
+        accessControl.checkCanCreateSchema(ADMIN, "staff", properties);
+        accessControl.checkCanCreateSchema(ADMIN, "authenticated", properties);
+        accessControl.checkCanCreateSchema(ADMIN, "test", properties);
+
+        accessControl.checkCanCreateSchema(BOB, "bob", properties);
+        accessControl.checkCanCreateSchema(BOB, "staff", properties);
+        accessControl.checkCanCreateSchema(BOB, "authenticated", properties);
+        assertDenied(() -> accessControl.checkCanCreateSchema(BOB, "test", properties));
+
+        assertDenied(() -> accessControl.checkCanCreateSchema(CHARLIE, "bob", properties));
+        assertDenied(() -> accessControl.checkCanCreateSchema(CHARLIE, "staff", properties));
+        accessControl.checkCanCreateSchema(CHARLIE, "authenticated", properties);
+        assertDenied(() -> accessControl.checkCanCreateSchema(CHARLIE, "test", properties));
 
         accessControl.checkCanDropSchema(ADMIN, "bob");
         accessControl.checkCanDropSchema(ADMIN, "staff");