feat(query): task support ownership

Author: TCeason

src/common/exception/src/exception_code.rs

@@ -370,6 +370,9 @@ build_exceptions! {
 
     // sequence
     SequenceError(3101),
+
+    // Task
+    UnknownTask(3201),
 }
 
 // Storage errors [3001, 4000].

src/meta/app/src/principal/ownership_object.rs

@@ -50,6 +50,10 @@ pub enum OwnershipObject {
     UDF {
         name: String,
     },
+
+    Task {
+        name: String,
+    },
 }
 
 impl OwnershipObject {
@@ -97,6 +101,7 @@ impl KeyCodec for OwnershipObject {
             }
             OwnershipObject::Stage { name } => b.push_raw("stage-by-name").push_str(name),
             OwnershipObject::UDF { name } => b.push_raw("udf-by-name").push_str(name),
+            OwnershipObject::Task { name } => b.push_raw("task-by-name").push_str(name),
         }
     }
 
@@ -143,9 +148,13 @@ impl KeyCodec for OwnershipObject {
                 let name = p.next_str()?;
                 Ok(OwnershipObject::UDF { name })
             }
+            "task-by-name" => {
+                let name = p.next_str()?;
+                Ok(OwnershipObject::Task { name })
+            }
             _ => Err(kvapi::KeyError::InvalidSegment {
                 i: p.index(),
-                expect: "database-by-id|database-by-catalog-id|table-by-id|table-by-catalog-id|stage-by-name|udf-by-name"
+                expect: "database-by-id|database-by-catalog-id|table-by-id|table-by-catalog-id|stage-by-name|udf-by-name|task-by-name"
                     .to_string(),
                 got: q.to_string(),
             }),

src/meta/app/src/principal/tenant_ownership_object_ident.rs

@@ -252,6 +252,22 @@ mod tests {
             let parsed = TenantOwnershipObjectIdent::from_str_key(&key).unwrap();
             assert_eq!(role_grantee, parsed);
         }
+
+        // udf
+        {
+            let role_grantee = TenantOwnershipObjectIdent::new_unchecked(
+                Tenant::new_literal("test"),
+                OwnershipObject::Task {
+                    name: "t1".to_string(),
+                },
+            );
+
+            let key = role_grantee.to_string_key();
+            assert_eq!("__fd_object_owners/test/task-by-name/t1", key);
+
+            let parsed = TenantOwnershipObjectIdent::from_str_key(&key).unwrap();
+            assert_eq!(role_grantee, parsed);
+        }
     }
 
     #[test]

src/meta/app/src/principal/user_grant.rs

@@ -30,6 +30,7 @@ pub enum GrantObject {
     TableById(String, u64, u64),
     UDF(String),
     Stage(String),
+    Task(String),
 }
 
 impl GrantObject {
@@ -62,6 +63,7 @@ impl GrantObject {
             (GrantObject::Table(_, _, _), _) => false,
             (GrantObject::Stage(lstage), GrantObject::Stage(rstage)) => lstage == rstage,
             (GrantObject::UDF(udf), GrantObject::UDF(rudf)) => udf == rudf,
+            (GrantObject::Task(task), GrantObject::Task(rtask)) => task == rtask,
             _ => false,
         }
     }
@@ -82,12 +84,18 @@ impl GrantObject {
             GrantObject::Stage(_) => {
                 UserPrivilegeSet::available_privileges_on_stage(available_ownership)
             }
+            GrantObject::Task(_) => {
+                UserPrivilegeSet::available_privileges_on_task(available_ownership)
+            }
         }
     }
 
     pub fn catalog(&self) -> Option<String> {
         match self {
-            GrantObject::Global | GrantObject::Stage(_) | GrantObject::UDF(_) => None,
+            GrantObject::Global
+            | GrantObject::Stage(_)
+            | GrantObject::UDF(_)
+            | GrantObject::Task(_) => None,
             GrantObject::Database(cat, _) | GrantObject::DatabaseById(cat, _) => Some(cat.clone()),
             GrantObject::Table(cat, _, _) | GrantObject::TableById(cat, _, _) => Some(cat.clone()),
         }
@@ -108,6 +116,7 @@ impl fmt::Display for GrantObject {
             }
             GrantObject::UDF(udf) => write!(f, "UDF {udf}"),
             GrantObject::Stage(stage) => write!(f, "STAGE {stage}"),
+            GrantObject::Task(task) => write!(f, "task {task}"),
         }
     }
 }

src/meta/app/src/principal/user_privilege.rs

@@ -75,6 +75,8 @@ pub enum UserPrivilegeType {
     Write = 1 << 19,
     // Privilege to Create database
     CreateDatabase = 1 << 20,
+    // Privilege to Create task
+    CreateTask = 1 << 21,
     // Discard Privilege Type
     Set = 1 << 4,
 }
@@ -101,6 +103,7 @@ const ALL_PRIVILEGES: BitFlags<UserPrivilegeType> = make_bitflags!(
         | Read
         | Write
         | CreateDatabase
+        | CreateTask
     }
 );
 
@@ -128,6 +131,7 @@ impl std::fmt::Display for UserPrivilegeType {
             UserPrivilegeType::Read => "Read",
             UserPrivilegeType::Write => "Write",
             UserPrivilegeType::CreateDatabase => "CREATE DATABASE",
+            UserPrivilegeType::CreateTask => "CREATE TASK",
         })
     }
 }
@@ -160,10 +164,12 @@ impl UserPrivilegeSet {
         let database_privs = Self::available_privileges_on_database(false);
         let stage_privs_without_ownership = Self::available_privileges_on_stage(false);
         let udf_privs_without_ownership = Self::available_privileges_on_udf(false);
-        let privs = make_bitflags!(UserPrivilegeType::{ Usage | Super | CreateUser | DropUser | CreateRole | DropRole | CreateDatabase | Grant | CreateDataMask });
+        let task_privs_without_ownership = Self::available_privileges_on_task(false);
+        let privs = make_bitflags!(UserPrivilegeType::{ Usage | Super | CreateUser | DropUser | CreateRole | DropRole | CreateDatabase | Grant | CreateDataMask | CreateTask });
         (database_privs.privileges
             | privs
             | stage_privs_without_ownership.privileges
+            | task_privs_without_ownership.privileges
             | udf_privs_without_ownership.privileges)
             .into()
     }
@@ -201,6 +207,14 @@ impl UserPrivilegeSet {
         }
     }
 
+    pub fn available_privileges_on_task(available_ownership: bool) -> Self {
+        if available_ownership {
+            make_bitflags!(UserPrivilegeType::{ Drop | Alter | Ownership }).into()
+        } else {
+            make_bitflags!(UserPrivilegeType::{ Drop | Alter }).into()
+        }
+    }
+
     // TODO: remove this, as ALL has different meanings on different objects
     pub fn all_privileges() -> Self {
         ALL_PRIVILEGES.into()

src/meta/proto-conv/src/ownership_from_to_protobuf_impl.rs

@@ -82,6 +82,9 @@ impl FromToProto for mt::principal::OwnershipObject {
             Some(pb::ownership_object::Object::Stage(
                 pb::ownership_object::OwnershipStageObject { stage },
             )) => Ok(mt::principal::OwnershipObject::Stage { name: stage }),
+            Some(pb::ownership_object::Object::Task(
+                pb::ownership_object::OwnershipTaskObject { task },
+            )) => Ok(mt::principal::OwnershipObject::Task { name: task }),
             _ => Err(Incompatible {
                 reason: "OwnershipObject cannot be None".to_string(),
             }),
@@ -115,6 +118,11 @@ impl FromToProto for mt::principal::OwnershipObject {
                     pb::ownership_object::OwnershipUdfObject { udf: name.clone() },
                 ))
             }
+            mt::principal::OwnershipObject::Task { name } => {
+                Some(pb::ownership_object::Object::Task(
+                    pb::ownership_object::OwnershipTaskObject { task: name.clone() },
+                ))
+            }
             mt::principal::OwnershipObject::Stage { name } => Some(
                 pb::ownership_object::Object::Stage(pb::ownership_object::OwnershipStageObject {
                     stage: name.clone(),

src/meta/proto-conv/src/user_from_to_protobuf_impl.rs

@@ -180,6 +180,9 @@ impl FromToProto for mt::principal::GrantObject {
             Some(pb::grant_object::Object::Stage(pb::grant_object::GrantStageObject { stage })) => {
                 Ok(mt::principal::GrantObject::Stage(stage))
             }
+            Some(pb::grant_object::Object::Task(pb::grant_object::GrantTaskObject { task })) => {
+                Ok(mt::principal::GrantObject::Task(task))
+            }
             _ => Err(Incompatible {
                 reason: "GrantObject cannot be None".to_string(),
             }),
@@ -225,6 +228,9 @@ impl FromToProto for mt::principal::GrantObject {
                     stage: stage.clone(),
                 },
             )),
+            mt::principal::GrantObject::Task(task) => Some(pb::grant_object::Object::Task(
+                pb::grant_object::GrantTaskObject { task: task.clone() },
+            )),
         };
         Ok(pb::GrantObject {
             ver: VER,

src/meta/proto-conv/src/util.rs

@@ -119,6 +119,7 @@ const META_CHANGE_LOG: &[(u64, &str)] = &[
     (87, "2024-04-17: Add: UserOption::disabled"),
     (88, "2024-04-17: Add: SequenceMeta"),
     (89, "2024-04-19: Add: geometry_output_format settings"),
+    (90, "2024-05-09: Add: GrantTaskObject"),
     // Dear developer:
     //      If you're gonna add a new metadata version, you'll have to add a test for it.
     //      You could just copy an existing test file(e.g., `../tests/it/v024_table_meta.rs`)

src/meta/protos/proto/ownership.proto

@@ -46,10 +46,15 @@ message OwnershipObject {
     string stage = 1;
   }
 
+  message OwnershipTaskObject {
+    string task = 1;
+  }
+
   oneof object {
     OwnershipDatabaseObject database = 1;
     OwnershipTableObject table = 2;
     OwnershipUdfObject udf = 3;
     OwnershipStageObject stage = 4;
+    OwnershipTaskObject task = 5;
   }
 }

src/meta/protos/proto/user.proto

@@ -72,6 +72,10 @@ message GrantObject {
     string udf = 1;
   }
 
+  message GrantTaskObject {
+    string task = 1;
+  }
+
   message GrantStageObject {
     string stage = 1;
   }
@@ -84,6 +88,7 @@ message GrantObject {
     GrantStageObject stage = 5;
     GrantDatabaseIdObject databasebyid = 6;
     GrantTableIdObject tablebyid = 7;
+    GrantTaskObject task = 8;
   }
 }