operator/statement/
dml.rs1use common_catalog::consts::is_ddl_reserved_table;
16use common_error::ext::BoxedError;
17use common_query::Output;
18use common_telemetry::tracing;
19use query::parser::QueryStatement;
20use session::context::QueryContextRef;
21use session::table_name::table_idents_to_full_name;
22use snafu::ResultExt;
23use sql::statements::insert::Insert;
24use sql::statements::statement::Statement;
25
26use crate::error::{ExternalSnafu, ParseSqlSnafu, Result};
27use crate::statement::StatementExecutor;
28
29impl StatementExecutor {
30 #[tracing::instrument(skip_all)]
31 pub async fn insert(&self, insert: Box<Insert>, query_ctx: QueryContextRef) -> Result<Output> {
32 self.create_ddl_reserved_target_on_demand(&insert, &query_ctx)
33 .await?;
34 if insert.can_extract_values() {
35 self.inserter
37 .handle_statement_insert(insert.as_ref(), &query_ctx)
38 .await
39 } else {
40 let statement = QueryStatement::Sql(Statement::Insert(insert));
42 self.plan_exec(statement, query_ctx).await
43 }
44 }
45
46 async fn create_ddl_reserved_target_on_demand(
49 &self,
50 insert: &Insert,
51 query_ctx: &QueryContextRef,
52 ) -> Result<()> {
53 let table_name = insert.table_name().context(ParseSqlSnafu)?;
54 let (catalog, schema, table) = table_idents_to_full_name(table_name, query_ctx)
55 .map_err(BoxedError::new)
56 .context(ExternalSnafu)?;
57 if !is_ddl_reserved_table(&schema, &table) {
58 return Ok(());
59 }
60 let exists = self
61 .catalog_manager
62 .table_exists(&catalog, &schema, &table, Some(query_ctx))
63 .await
64 .map_err(BoxedError::new)
65 .context(ExternalSnafu)?;
66 if !exists {
67 self.create_declared_relationships_table(&catalog, query_ctx.clone())
68 .await?;
69 }
70 Ok(())
71 }
72}