add WAREHOUSE as ident

note: only support_forward_warehouse_request will apply warehouse rbac. Now only support SystemResourcesManagement

src/query/ast/src/parser/statement.rs

@@ -3459,7 +3459,7 @@ pub fn grant_ownership_level(i: Input) -> IResult<AccountMgrLevel> {
     let object = alt((
         value(Object::Udf, rule! { UDF }),
         value(Object::Stage, rule! { STAGE }),
-        value(Object::Warehouse, rule! { STAGE }),
+        value(Object::Warehouse, rule! { WAREHOUSE }),
     ));
 
     // Object object_name

src/query/ast/src/parser/token.rs

@@ -1753,6 +1753,7 @@ impl TokenKind {
             // | TokenKind::RETURNING
             | TokenKind::STAGE
             | TokenKind::UDF
+            | TokenKind::WAREHOUSE
             | TokenKind::SHARE
             | TokenKind::SHARES
             | TokenKind::TO

src/query/ast/tests/it/parser.rs

@@ -774,6 +774,7 @@ fn test_statement() {
         r#"GRANT OWNERSHIP ON d20_0014.* TO ROLE 'd20_0015_owner';"#,
         r#"GRANT OWNERSHIP ON d20_0014.t TO ROLE 'd20_0015_owner';"#,
         r#"GRANT OWNERSHIP ON STAGE s1 TO ROLE 'd20_0015_owner';"#,
+        r#"GRANT OWNERSHIP ON WAREHOUSE w1 TO ROLE 'd20_0015_owner';"#,
         r#"GRANT OWNERSHIP ON UDF f1 TO ROLE 'd20_0015_owner';"#,
         r#"attach table t 's3://a' connection=(access_key_id ='x' secret_access_key ='y' endpoint_url='http://127.0.0.1:9900')"#,
         r#"CREATE FUNCTION IF NOT EXISTS isnotempty AS(p) -> not(is_null(p));"#,

src/query/ast/tests/it/testdata/stmt.txt

@@ -23666,6 +23666,28 @@ Grant(
 )
 
 
+---------- Input ----------
+GRANT OWNERSHIP ON WAREHOUSE w1 TO ROLE 'd20_0015_owner';
+---------- Output ---------
+GRANT OWNERSHIP ON  WAREHOUSE w1 TO ROLE 'd20_0015_owner'
+---------- AST ------------
+Grant(
+    GrantStmt {
+        source: Privs {
+            privileges: [
+                Ownership,
+            ],
+            level: Warehouse(
+                "w1",
+            ),
+        },
+        principal: Role(
+            "d20_0015_owner",
+        ),
+    },
+)
+
+
 ---------- Input ----------
 GRANT OWNERSHIP ON UDF f1 TO ROLE 'd20_0015_owner';
 ---------- Output ---------

src/query/service/src/interpreters/access/privilege_access.rs

@@ -1361,6 +1361,11 @@ impl AccessChecker for PrivilegeAccess {
                 self.validate_warehouse_ownership(plan.warehouse.clone(), identity).await.transpose()?;
             }
             Plan::CreateWarehouse(_) => {
+                let warehouse_mgr = GlobalInstance::get::<Arc<dyn ResourcesManagement>>();
+                // Only check support_forward_warehouse_request privileges
+                if !warehouse_mgr.support_forward_warehouse_request() {
+                    return Ok(());
+                }
                 // only current role has global level create warehouse privilege, it will pass
                 self.validate_access(&GrantObject::Global, UserPrivilegeType::CreateWarehouse, true, false)
                     .await?;

src/query/service/src/interpreters/common/grant.rs

@@ -14,10 +14,13 @@
 
 use std::sync::Arc;
 
+use databend_common_base::base::GlobalInstance;
 use databend_common_catalog::table_context::TableContext;
 use databend_common_exception::Result;
+use databend_common_management::WarehouseInfo;
 use databend_common_meta_app::principal::GrantObject;
 use databend_common_users::UserApiProvider;
+use databend_enterprise_resources_management::ResourcesManagement;
 
 use crate::sessions::QueryContext;
 
@@ -93,9 +96,26 @@ pub async fn validate_grant_object_exists(
                 )));
             }
         }
-        GrantObject::Warehouse(_w) => {
-            // TODO
-            return Ok(());
+        GrantObject::Warehouse(w) => {
+            let warehouse_mgr = GlobalInstance::get::<Arc<dyn ResourcesManagement>>();
+            // Only check support_forward_warehouse_request
+            if !warehouse_mgr.support_forward_warehouse_request() {
+                return Ok(());
+            }
+            let ws = warehouse_mgr.list_warehouses().await?;
+            return if ws.iter().any(|warehouse| {
+                if let WarehouseInfo::SystemManaged(sw) = warehouse {
+                    &sw.id == w
+                } else {
+                    false
+                }
+            }) {
+                Ok(())
+            } else {
+                Err(databend_common_exception::ErrorCode::UnknownWarehouse(
+                    format!("warehouse {w} not exists"),
+                ))
+            };
         }
         GrantObject::Global => (),
     }

src/query/service/src/interpreters/interpreter_privilege_grant.rs

@@ -131,7 +131,7 @@ impl GrantPrivilegeInterpreter {
         }
 
         let mut log_msg = format!(
-            "{}: grant ownership on {:?}  to {}",
+            "{}: grant ownership on {:?} to {}",
             ctx.get_id(),
             owner_object,
             new_role