operator/statement/
comment.rs1use api::v1::CommentOnExpr;
16use common_error::ext::BoxedError;
17use common_meta::cache_invalidator::Context;
18use common_meta::procedure_executor::ExecutorContext;
19use common_meta::rpc::ddl::{
20 CommentObjectType, CommentOnTask, DdlTask, SubmitDdlTaskRequest, TriggerReason,
21};
22use common_query::Output;
23use session::context::QueryContextRef;
24use session::table_name::table_idents_to_full_name;
25use snafu::ResultExt;
26use sql::ast::ObjectNamePartExt;
27use sql::statements::comment::{Comment, CommentObject};
28
29use crate::error::{
30 self, ExecuteDdlSnafu, ExternalSnafu, InvalidSqlSnafu, Result, TableMetadataManagerSnafu,
31};
32use crate::statement::StatementExecutor;
33use crate::utils::to_meta_query_context_with_trigger_reason;
34
35impl StatementExecutor {
36 pub async fn comment(&self, stmt: Comment, query_ctx: QueryContextRef) -> Result<Output> {
47 let mut comment_on_task = self.create_comment_on_task_from_stmt(stmt, &query_ctx)?;
48 comment_on_task
49 .enrich_object_id(
50 self.table_metadata_manager.table_name_manager(),
51 self.flow_metadata_manager.flow_name_manager(),
52 )
53 .await
54 .context(TableMetadataManagerSnafu)?;
55 let cache_idents = comment_on_task.cache_idents();
56
57 let meta_query_context =
58 to_meta_query_context_with_trigger_reason(query_ctx, TriggerReason::Manual);
59 let request =
60 SubmitDdlTaskRequest::new(meta_query_context, DdlTask::new_comment_on(comment_on_task));
61
62 self.procedure_executor
63 .submit_ddl_task(&ExecutorContext::default(), request)
64 .await
65 .context(ExecuteDdlSnafu)?;
66
67 self.cache_invalidator
69 .invalidate(&Context::default(), &cache_idents)
70 .await
71 .context(error::InvalidateTableCacheSnafu)?;
72
73 Ok(Output::new_with_affected_rows(0))
74 }
75
76 pub async fn comment_by_expr(
77 &self,
78 expr: CommentOnExpr,
79 query_ctx: QueryContextRef,
80 ) -> Result<Output> {
81 let mut comment_on_task = self.create_comment_on_task_from_expr(expr)?;
82 comment_on_task
83 .enrich_object_id(
84 self.table_metadata_manager.table_name_manager(),
85 self.flow_metadata_manager.flow_name_manager(),
86 )
87 .await
88 .context(TableMetadataManagerSnafu)?;
89 let cache_idents = comment_on_task.cache_idents();
90
91 let meta_query_context =
92 to_meta_query_context_with_trigger_reason(query_ctx, TriggerReason::Manual);
93 let request =
94 SubmitDdlTaskRequest::new(meta_query_context, DdlTask::new_comment_on(comment_on_task));
95
96 self.procedure_executor
97 .submit_ddl_task(&ExecutorContext::default(), request)
98 .await
99 .context(ExecuteDdlSnafu)?;
100
101 self.cache_invalidator
103 .invalidate(&Context::default(), &cache_idents)
104 .await
105 .context(error::InvalidateTableCacheSnafu)?;
106
107 Ok(Output::new_with_affected_rows(0))
108 }
109
110 fn create_comment_on_task_from_expr(&self, expr: CommentOnExpr) -> Result<CommentOnTask> {
111 let object_type = match expr.object_type {
112 0 => CommentObjectType::Table,
113 1 => CommentObjectType::Column,
114 2 => CommentObjectType::Flow,
115 _ => {
116 return InvalidSqlSnafu {
117 err_msg: format!(
118 "Invalid CommentObjectType value: {}. Valid values are: 0 (Table), 1 (Column), 2 (Flow)",
119 expr.object_type
120 ),
121 }
122 .fail();
123 }
124 };
125
126 Ok(CommentOnTask {
127 catalog_name: expr.catalog_name,
128 schema_name: expr.schema_name,
129 object_type,
130 object_name: expr.object_name,
131 column_name: if expr.column_name.is_empty() {
132 None
133 } else {
134 Some(expr.column_name)
135 },
136 object_id: None,
137 comment: if expr.comment.is_empty() {
138 None
139 } else {
140 Some(expr.comment)
141 },
142 })
143 }
144
145 fn create_comment_on_task_from_stmt(
146 &self,
147 stmt: Comment,
148 query_ctx: &QueryContextRef,
149 ) -> Result<CommentOnTask> {
150 match stmt.object {
151 CommentObject::Table(table) => {
152 let (catalog_name, schema_name, table_name) =
153 table_idents_to_full_name(&table, query_ctx)
154 .map_err(BoxedError::new)
155 .context(ExternalSnafu)?;
156
157 Ok(CommentOnTask {
158 catalog_name,
159 schema_name,
160 object_type: CommentObjectType::Table,
161 object_name: table_name,
162 column_name: None,
163 object_id: None,
164 comment: stmt.comment,
165 })
166 }
167 CommentObject::Column { table, column } => {
168 let (catalog_name, schema_name, table_name) =
169 table_idents_to_full_name(&table, query_ctx)
170 .map_err(BoxedError::new)
171 .context(ExternalSnafu)?;
172
173 Ok(CommentOnTask {
174 catalog_name,
175 schema_name,
176 object_type: CommentObjectType::Column,
177 object_name: table_name,
178 column_name: Some(column.value),
179 object_id: None,
180 comment: stmt.comment,
181 })
182 }
183 CommentObject::Flow(flow_name) => {
184 let (catalog_name, flow_name_str) = match &flow_name.0[..] {
185 [flow] => (
186 query_ctx.current_catalog().to_string(),
187 flow.to_string_unquoted(),
188 ),
189 [catalog, flow] => (catalog.to_string_unquoted(), flow.to_string_unquoted()),
190 _ => {
191 return InvalidSqlSnafu {
192 err_msg: format!(
193 "expect flow name to be <catalog>.<flow_name> or <flow_name>, actual: {flow_name}"
194 ),
195 }
196 .fail();
197 }
198 };
199
200 Ok(CommentOnTask {
201 catalog_name,
202 schema_name: String::new(), object_type: CommentObjectType::Flow,
204 object_name: flow_name_str,
205 column_name: None,
206 object_id: None,
207 comment: stmt.comment,
208 })
209 }
210 }
211 }
212}