Task privilege Part2

1. add function: list_task_ownerships, only list task prefix.

2. add visibility in task/task_history system table.

3. refactor showtasks, directly use select * from system.task;

Author: TCeason

src/common/cloud_control/proto/task.proto

@@ -119,6 +119,7 @@ message ShowTasksRequest {// every owner has a roles list like ["role1", "role2"
   int32 result_limit = 4;
   repeated string owners = 5;   // all available roles under current client
   repeated string task_ids = 6; // all task ids which permit to access for given user
+  repeated string task_names = 7; // all task names which permit to access for given user
 }
 
 message ShowTasksResponse {
@@ -170,6 +171,7 @@ message ShowTaskRunsRequest {
   repeated string owners = 6;
   repeated string task_ids = 7;
   string task_name = 8;
+  repeated string task_names = 9;
 
   optional int32 page_size = 90; // 100 by default
   optional int64 next_page_token = 91;

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

@@ -253,7 +253,7 @@ mod tests {
             assert_eq!(role_grantee, parsed);
         }
 
-        // udf
+        // task
         {
             let role_grantee = TenantOwnershipObjectIdent::new_unchecked(
                 Tenant::new_literal("test"),

src/query/management/src/role/role_api.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
 use databend_common_meta_app::principal::OwnershipInfo;
 use databend_common_meta_app::principal::OwnershipObject;
@@ -30,6 +31,10 @@ pub trait RoleApi: Sync + Send {
 
     async fn get_ownerships(&self) -> Result<Vec<SeqV<OwnershipInfo>>>;
 
+    async fn list_tasks_ownerships(
+        &self,
+    ) -> std::result::Result<Vec<SeqV<OwnershipInfo>>, ErrorCode>;
+
     /// General role update.
     ///
     /// It fetches the role that matches the specified seq number, update it in place, then write it back with the seq it sees.

src/query/management/src/role/role_mgr.rs

@@ -217,6 +217,28 @@ impl RoleApi for RoleMgr {
         Ok(r)
     }
 
+    #[async_backtrace::framed]
+    #[minitrace::trace]
+    async fn list_tasks_ownerships(&self) -> Result<Vec<SeqV<OwnershipInfo>>, ErrorCode> {
+        let mut task_object_owner_prefix = self.ownership_object_prefix();
+        task_object_owner_prefix.push_str("task-by-name/");
+        let values = self
+            .kv_api
+            .prefix_list_kv(task_object_owner_prefix.as_str())
+            .await?;
+
+        let mut r = vec![];
+
+        let mut quota = Quota::new(func_name!());
+
+        for (key, val) in values {
+            let u = check_and_upgrade_to_pb(&mut quota, key, &val, self.kv_api.as_ref()).await?;
+            r.push(u);
+        }
+
+        Ok(r)
+    }
+
     /// General role update.
     ///
     /// It fetch the role that matches the specified seq number, update it in place, then write it back with the seq it sees.

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

@@ -56,7 +56,6 @@ use crate::interpreters::interpreter_task_create::CreateTaskInterpreter;
 use crate::interpreters::interpreter_task_describe::DescribeTaskInterpreter;
 use crate::interpreters::interpreter_task_drop::DropTaskInterpreter;
 use crate::interpreters::interpreter_task_execute::ExecuteTaskInterpreter;
-use crate::interpreters::interpreter_tasks_show::ShowTasksInterpreter;
 use crate::interpreters::interpreter_txn_abort::AbortInterpreter;
 use crate::interpreters::interpreter_txn_begin::BeginInterpreter;
 use crate::interpreters::interpreter_txn_commit::CommitInterpreter;
@@ -546,7 +545,6 @@ impl InterpreterFactory {
                 ctx,
                 *p.clone(),
             )?)),
-            Plan::ShowTasks(p) => Ok(Arc::new(ShowTasksInterpreter::try_create(ctx, *p.clone())?)),
 
             Plan::CreateConnection(p) => Ok(Arc::new(CreateConnectionInterpreter::try_create(
                 ctx,

src/query/service/src/interpreters/interpreter_factory.rs

@@ -121,7 +121,6 @@ mod interpreter_task_create;
 mod interpreter_task_describe;
 mod interpreter_task_drop;
 mod interpreter_task_execute;
-mod interpreter_tasks_show;
 mod interpreter_txn_abort;
 mod interpreter_txn_begin;
 mod interpreter_txn_commit;

src/query/service/src/interpreters/interpreter_tasks_show.rs

@@ -609,7 +609,7 @@ impl<'a> Binder {
                 self.bind_execute_task(stmt).await?
             }
             Statement::ShowTasks(stmt) => {
-                self.bind_show_tasks(stmt).await?
+                self.bind_show_tasks(bind_context, stmt).await?
             }
 
             // Streams

src/query/service/src/interpreters/mod.rs

@@ -23,22 +23,26 @@ use databend_common_ast::ast::DescribeTaskStmt;
 use databend_common_ast::ast::DropTaskStmt;
 use databend_common_ast::ast::ExecuteTaskStmt;
 use databend_common_ast::ast::ScheduleOptions;
+use databend_common_ast::ast::ShowLimit;
 use databend_common_ast::ast::ShowTasksStmt;
 use databend_common_ast::ast::TaskSql;
 use databend_common_ast::parser::parse_sql;
 use databend_common_ast::parser::tokenize_sql;
 use databend_common_ast::parser::Dialect;
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
+use log::debug;
 
 use crate::plans::AlterTaskPlan;
 use crate::plans::CreateTaskPlan;
 use crate::plans::DescribeTaskPlan;
 use crate::plans::DropTaskPlan;
 use crate::plans::ExecuteTaskPlan;
 use crate::plans::Plan;
-use crate::plans::ShowTasksPlan;
+use crate::plans::RewriteKind;
+use crate::BindContext;
 use crate::Binder;
+use crate::SelectBuilder;
 
 fn verify_single_statement(sql: &String) -> Result<()> {
     let tokens = tokenize_sql(sql.as_str()).map_err(|e| {
@@ -243,16 +247,27 @@ impl Binder {
     #[async_backtrace::framed]
     pub(in crate::planner::binder) async fn bind_show_tasks(
         &mut self,
+        bind_context: &mut BindContext,
         stmt: &ShowTasksStmt,
     ) -> Result<Plan> {
         let ShowTasksStmt { limit } = stmt;
 
-        let tenant = self.ctx.get_tenant();
+        let mut select_builder = SelectBuilder::from("system.tasks");
 
-        let plan = ShowTasksPlan {
-            tenant,
-            limit: limit.clone(),
+        let query = match limit {
+            None => select_builder.build(),
+            Some(ShowLimit::Like { pattern }) => {
+                select_builder.with_filter(format!("name LIKE '{pattern}'"));
+                select_builder.build()
+            }
+            Some(ShowLimit::Where { selection }) => {
+                select_builder.with_filter(format!("({selection})"));
+                select_builder.build()
+            }
         };
-        Ok(Plan::ShowTasks(Box::new(plan)))
+
+        debug!("show tasks rewrite to: {:?}", query);
+        self.bind_rewrite_to_query(bind_context, query.as_str(), RewriteKind::ShowTasks)
+            .await
     }
 }