feat: support authorization syntax for CREATE SCHEMA and omitted sche…

…ma name

e2e_test/ddl/schema.slt

@@ -67,3 +67,35 @@ drop schema Myschema;
 
 statement ok
 drop schema "Myschema";
+
+# test derived schema name and authorization with user
+statement ok
+create user user_for_authorization;
+
+statement ok
+create schema if not exists authorization user_for_authorization;
+
+statement ok
+create schema myschema authorization user_for_authorization;
+
+query T rowsort
+show schemas;
+----
+information_schema
+myschema
+pg_catalog
+public
+rw_catalog
+user_for_authorization
+
+statement error Permission denied
+drop user user_for_authorization;
+
+statement ok
+drop schema myschema;
+
+statement ok
+drop schema user_for_authorization;
+
+statement ok
+drop user user_for_authorization;

src/frontend/planner_test/src/lib.rs

@@ -568,9 +568,15 @@ impl TestCase {
                 Statement::CreateSchema {
                     schema_name,
                     if_not_exists,
+                    user_specified,
                 } => {
-                    create_schema::handle_create_schema(handler_args, schema_name, if_not_exists)
-                        .await?;
+                    create_schema::handle_create_schema(
+                        handler_args,
+                        schema_name,
+                        if_not_exists,
+                        user_specified,
+                    )
+                    .await?;
                 }
                 _ => return Err(anyhow!("Unsupported statement type")),
             }

src/frontend/src/handler/create_schema.rs

@@ -29,6 +29,7 @@ pub async fn handle_create_schema(
     handler_args: HandlerArgs,
     schema_name: ObjectName,
     if_not_exist: bool,
+    user_specified: Option<ObjectName>,
 ) -> Result<RwPgResponse> {
     let session = handler_args.session;
     let database_name = session.database();
@@ -62,6 +63,19 @@ pub async fn handle_create_schema(
         (db.id(), db.owner())
     };
 
+    let schema_owner = if let Some(user_specified) = user_specified {
+        let user_specified = Binder::resolve_user_name(user_specified)?;
+        session
+            .env()
+            .user_info_reader()
+            .read_guard()
+            .get_user_by_name(&user_specified)
+            .map(|u| u.id)
+            .ok_or_else(|| CatalogError::NotFound("user", user_specified.to_string()))?
+    } else {
+        session.user_id()
+    };
+
     session.check_privileges(&[ObjectCheckItem::new(
         db_owner,
         AclMode::Create,
@@ -70,7 +84,7 @@ pub async fn handle_create_schema(
 
     let catalog_writer = session.catalog_writer()?;
     catalog_writer
-        .create_schema(db_id, &schema_name, session.user_id())
+        .create_schema(db_id, &schema_name, schema_owner)
         .await?;
     Ok(PgResponse::empty_result(StatementType::CREATE_SCHEMA))
 }

src/frontend/src/handler/mod.rs

@@ -363,7 +363,16 @@ pub async fn handle(
         Statement::CreateSchema {
             schema_name,
             if_not_exists,
-        } => create_schema::handle_create_schema(handler_args, schema_name, if_not_exists).await,
+            user_specified,
+        } => {
+            create_schema::handle_create_schema(
+                handler_args,
+                schema_name,
+                if_not_exists,
+                user_specified,
+            )
+            .await
+        }
         Statement::CreateUser(stmt) => create_user::handle_create_user(handler_args, stmt).await,
         Statement::DeclareCursor { stmt } => {
             declare_cursor::handle_declare_cursor(handler_args, stmt).await

src/sqlparser/src/ast/mod.rs

@@ -1452,6 +1452,7 @@ pub enum Statement {
     CreateSchema {
         schema_name: ObjectName,
         if_not_exists: bool,
+        user_specified: Option<ObjectName>,
     },
     /// CREATE DATABASE
     CreateDatabase {
@@ -1980,12 +1981,19 @@ impl fmt::Display for Statement {
             Statement::CreateSchema {
                 schema_name,
                 if_not_exists,
-            } => write!(
-                f,
-                "CREATE SCHEMA {if_not_exists}{name}",
-                if_not_exists = if *if_not_exists { "IF NOT EXISTS " } else { "" },
-                name = schema_name
-            ),
+                user_specified,
+            } => {
+                write!(
+                    f,
+                    "CREATE SCHEMA {if_not_exists}{name}",
+                    if_not_exists = if *if_not_exists { "IF NOT EXISTS " } else { "" },
+                    name = schema_name
+                )?;
+                if let Some(user) = user_specified {
+                    write!(f, " AUTHORIZATION {}", user)?;
+                }
+                Ok(())
+            },
             Statement::Grant {
                 privileges,
                 objects,

src/sqlparser/src/parser.rs

@@ -2152,10 +2152,22 @@ impl Parser {
 
     pub fn parse_create_schema(&mut self) -> Result<Statement, ParserError> {
         let if_not_exists = self.parse_keywords(&[Keyword::IF, Keyword::NOT, Keyword::EXISTS]);
-        let schema_name = self.parse_object_name()?;
+        let (schema_name, user_specified) = if self.parse_keyword(Keyword::AUTHORIZATION) {
+            let user_specified = self.parse_object_name()?;
+            (user_specified.clone(), Some(user_specified))
+        } else {
+            let schema_name = self.parse_object_name()?;
+            let user_specified = if self.parse_keyword(Keyword::AUTHORIZATION) {
+                Some(self.parse_object_name()?)
+            } else {
+                None
+            };
+            (schema_name, user_specified)
+        };
         Ok(Statement::CreateSchema {
             schema_name,
             if_not_exists,
+            user_specified,
         })
     }
 

src/sqlparser/tests/sqlparser_postgres.rs

@@ -342,6 +342,7 @@ fn parse_create_schema_if_not_exists() {
         Statement::CreateSchema {
             if_not_exists: true,
             schema_name,
+            ..
         } => assert_eq!("schema_name", schema_name.to_string()),
         _ => unreachable!(),
     }