risingwave_frontend/catalog/system_catalog/rw_catalog/
rw_tables.rs1use risingwave_common::types::{Fields, Timestamptz};
16use risingwave_frontend_macro::system_catalog;
17use risingwave_pb::user::grant_privilege::Object as GrantObject;
18
19use crate::catalog::system_catalog::{SysCatalogReaderImpl, get_acl_items};
20use crate::error::Result;
21
22#[derive(Fields)]
23struct RwTable {
24 #[primary_key]
25 id: i32,
26 name: String,
27 schema_id: i32,
28 owner: i32,
29 definition: String,
30 append_only: bool,
31 acl: Vec<String>,
32 initialized_at: Option<Timestamptz>,
33 created_at: Option<Timestamptz>,
34 initialized_at_cluster_version: Option<String>,
35 created_at_cluster_version: Option<String>,
36}
37
38#[system_catalog(table, "rw_catalog.rw_tables")]
39fn read_rw_table_info(reader: &SysCatalogReaderImpl) -> Result<Vec<RwTable>> {
40 let catalog_reader = reader.catalog_reader.read_guard();
41 let schemas = catalog_reader.iter_schemas(&reader.auth_context.database)?;
42 let user_reader = reader.user_info_reader.read_guard();
43 let users = user_reader.get_all_users();
44 let current_user = user_reader
45 .get_user_by_name(&reader.auth_context.user_name)
46 .expect("user not found");
47 let username_map = user_reader.get_user_name_map();
48
49 Ok(schemas
50 .flat_map(|schema| {
51 schema
52 .iter_user_table_with_acl(current_user)
53 .map(|table| RwTable {
54 id: table.id.table_id as i32,
55 name: table.name().to_owned(),
56 schema_id: schema.id() as i32,
57 owner: table.owner as i32,
58 definition: table.create_sql_purified(),
59 append_only: table.append_only,
60 acl: get_acl_items(
61 &GrantObject::TableId(table.id.table_id),
62 true,
63 &users,
64 username_map,
65 ),
66 initialized_at: table.initialized_at_epoch.map(|e| e.as_timestamptz()),
67 created_at: table.created_at_epoch.map(|e| e.as_timestamptz()),
68 initialized_at_cluster_version: table.initialized_at_cluster_version.clone(),
69 created_at_cluster_version: table.created_at_cluster_version.clone(),
70 })
71 })
72 .collect())
73}