fix: remove incorrect table_idents_to_full_name (#967)

src/frontend/src/instance.rs

@@ -37,6 +37,7 @@ use common_recordbatch::RecordBatches;
 use common_telemetry::logging::{debug, info};
 use datafusion::sql::sqlparser::ast::ObjectName;
 use datafusion_common::TableReference;
+use datanode::instance::sql::table_idents_to_full_name;
 use datanode::instance::InstanceRef as DnInstanceRef;
 use datatypes::schema::Schema;
 use distributed::DistInstance;
@@ -63,8 +64,8 @@ use sql::statements::statement::Statement;
 use crate::catalog::FrontendCatalogManager;
 use crate::datanode::DatanodeClients;
 use crate::error::{
-    self, Error, ExecutePromqlSnafu, MissingMetasrvOptsSnafu, NotSupportedSnafu, ParseSqlSnafu,
-    Result, SqlExecInterceptedSnafu,
+    self, Error, ExecutePromqlSnafu, ExternalSnafu, MissingMetasrvOptsSnafu, NotSupportedSnafu,
+    ParseSqlSnafu, Result, SqlExecInterceptedSnafu,
 };
 use crate::expr_factory::{CreateExprFactoryRef, DefaultCreateExprFactory};
 use crate::frontend::FrontendOptions;
@@ -557,7 +558,11 @@ pub fn check_permission(
         Statement::ShowCreateTable(_) | Statement::Alter(_) => {}
 
         Statement::Insert(insert) => {
-            let (catalog, schema, _) = insert.full_table_name().context(ParseSqlSnafu)?;
+            let (catalog, schema, _) =
+                table_idents_to_full_name(insert.table_name(), query_ctx.clone())
+                    .map_err(BoxedError::new)
+                    .context(ExternalSnafu)?;
+
             validate_param(&catalog, &schema, query_ctx)?;
         }
         Statement::CreateTable(stmt) => {

src/frontend/src/instance/distributed.rs

@@ -295,7 +295,7 @@ impl DistInstance {
             }
             Statement::Insert(insert) => {
                 let (catalog, schema, table) =
-                    table_idents_to_full_name(insert.table_name(), query_ctx)
+                    table_idents_to_full_name(insert.table_name(), query_ctx.clone())
                         .map_err(BoxedError::new)
                         .context(error::ExternalSnafu)?;
 
@@ -305,7 +305,7 @@ impl DistInstance {
                     .context(CatalogSnafu)?
                     .context(TableNotFoundSnafu { table_name: table })?;
 
-                let insert_request = insert_to_request(&table, *insert)?;
+                let insert_request = insert_to_request(&table, *insert, query_ctx)?;
 
                 return Ok(Output::AffectedRows(
                     table.insert(insert_request).await.context(TableSnafu)?,

src/frontend/src/sql.rs

@@ -12,10 +12,13 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use common_error::ext::BoxedError;
 use common_error::snafu::ensure;
+use datanode::instance::sql::table_idents_to_full_name;
 use datatypes::data_type::DataType;
 use datatypes::prelude::MutableVector;
 use datatypes::schema::ColumnSchema;
+use session::context::QueryContextRef;
 use snafu::{OptionExt, ResultExt};
 use sql::ast::Value as SqlValue;
 use sql::statements;
@@ -23,17 +26,24 @@ use sql::statements::insert::Insert;
 use table::requests::InsertRequest;
 use table::TableRef;
 
-use crate::error::{self, Result};
+use crate::error::{self, ExternalSnafu, Result};
 
 const DEFAULT_PLACEHOLDER_VALUE: &str = "default";
 
 // TODO(fys): Extract the common logic in datanode and frontend in the future.
 // This function convert insert statement to an `InsertRequest` to region 0.
-pub(crate) fn insert_to_request(table: &TableRef, stmt: Insert) -> Result<InsertRequest> {
+pub(crate) fn insert_to_request(
+    table: &TableRef,
+    stmt: Insert,
+    query_ctx: QueryContextRef,
+) -> Result<InsertRequest> {
     let columns = stmt.columns();
     let values = stmt.values().context(error::ParseSqlSnafu)?;
+
     let (catalog_name, schema_name, table_name) =
-        stmt.full_table_name().context(error::ParseSqlSnafu)?;
+        table_idents_to_full_name(stmt.table_name(), query_ctx)
+            .map_err(BoxedError::new)
+            .context(ExternalSnafu)?;
 
     let schema = table.schema();
     let columns_num = if columns.is_empty() {

src/query/src/query_engine/options.rs

@@ -18,7 +18,7 @@ use snafu::ensure;
 
 use crate::error::{QueryAccessDeniedSnafu, Result};
 
-#[derive(Default)]
+#[derive(Default, Clone)]
 pub struct QueryOptions {
     pub disallow_cross_schema_query: bool,
 }

src/sql/src/statements.rs

@@ -25,7 +25,6 @@ use std::str::FromStr;
 
 use api::helper::ColumnDataTypeWrapper;
 use common_base::bytes::Bytes;
-use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
 use common_time::Timestamp;
 use datatypes::data_type::DataType;
 use datatypes::prelude::ConcreteDataType;
@@ -35,8 +34,7 @@ use datatypes::value::Value;
 use snafu::{ensure, OptionExt, ResultExt};
 
 use crate::ast::{
-    ColumnDef, ColumnOption, ColumnOptionDef, DataType as SqlDataType, Expr, ObjectName,
-    Value as SqlValue,
+    ColumnDef, ColumnOption, ColumnOptionDef, DataType as SqlDataType, Expr, Value as SqlValue,
 };
 use crate::error::{
     self, ColumnTypeMismatchSnafu, ConvertToGrpcDataTypeSnafu, InvalidSqlValueSnafu,
@@ -44,36 +42,6 @@ use crate::error::{
     UnsupportedDefaultValueSnafu,
 };
 
-// TODO(LFC): Get rid of this function, use session context aware version of "table_idents_to_full_name" instead.
-// Current obstacles remain in some usage in Frontend, and other SQLs like "describe", "drop" etc.
-/// Converts maybe fully-qualified table name (`<catalog>.<schema>.<table>` or `<table>` when
-/// catalog and schema are default) to tuple.
-pub fn table_idents_to_full_name(obj_name: &ObjectName) -> Result<(String, String, String)> {
-    match &obj_name.0[..] {
-        [table] => Ok((
-            DEFAULT_CATALOG_NAME.to_string(),
-            DEFAULT_SCHEMA_NAME.to_string(),
-            table.value.clone(),
-        )),
-        [schema, table] => Ok((
-            DEFAULT_CATALOG_NAME.to_string(),
-            schema.value.clone(),
-            table.value.clone(),
-        )),
-        [catalog, schema, table] => Ok((
-            catalog.value.clone(),
-            schema.value.clone(),
-            table.value.clone(),
-        )),
-        _ => error::InvalidSqlSnafu {
-            msg: format!(
-                "expect table name to be <catalog>.<schema>.<table>, <schema>.<table> or <table>, actual: {obj_name}",
-            ),
-        }
-        .fail(),
-    }
-}
-
 fn parse_string_to_value(
     column_name: &str,
     s: String,
@@ -368,6 +336,7 @@ mod tests {
     use common_time::timestamp::TimeUnit;
     use datatypes::types::BooleanType;
     use datatypes::value::OrderedFloat;
+    use sqlparser::ast::ObjectName;
 
     use super::*;
     use crate::ast::{Ident, TimezoneInfo};

src/sql/src/statements/alter.rs

@@ -12,12 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use api::v1::{alter_expr, AddColumn, AlterExpr, DropColumn};
 use sqlparser::ast::{ColumnDef, Ident, ObjectName, TableConstraint};
 
-use crate::error::UnsupportedAlterTableStatementSnafu;
-use crate::statements::{sql_column_def_to_grpc_column_def, table_idents_to_full_name};
-
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct AlterTable {
     table_name: ObjectName,
@@ -52,45 +48,3 @@ pub enum AlterTableOperation {
     /// `RENAME <new_table_name>`
     RenameTable { new_table_name: String },
 }
-
-/// Convert `AlterTable` statement to `AlterExpr` for gRPC
-impl TryFrom<AlterTable> for AlterExpr {
-    type Error = crate::error::Error;
-
-    fn try_from(value: AlterTable) -> Result<Self, Self::Error> {
-        let (catalog_name, schema_name, table_name) = table_idents_to_full_name(&value.table_name)?;
-
-        let kind = match value.alter_operation {
-            AlterTableOperation::AddConstraint(_) => {
-                return UnsupportedAlterTableStatementSnafu {
-                    msg: "ADD CONSTRAINT not supported yet.",
-                }
-                .fail();
-            }
-            AlterTableOperation::AddColumn { column_def } => {
-                alter_expr::Kind::AddColumns(api::v1::AddColumns {
-                    add_columns: vec![AddColumn {
-                        column_def: Some(sql_column_def_to_grpc_column_def(column_def)?),
-                        is_key: false,
-                    }],
-                })
-            }
-            AlterTableOperation::DropColumn { name } => {
-                alter_expr::Kind::DropColumns(api::v1::DropColumns {
-                    drop_columns: vec![DropColumn { name: name.value }],
-                })
-            }
-            AlterTableOperation::RenameTable { new_table_name } => {
-                alter_expr::Kind::RenameTable(api::v1::RenameTable { new_table_name })
-            }
-        };
-        let expr = AlterExpr {
-            catalog_name,
-            schema_name,
-            table_name,
-            kind: Some(kind),
-        };
-
-        Ok(expr)
-    }
-}

src/sql/src/statements/insert.rs

@@ -17,7 +17,6 @@ use sqlparser::parser::ParserError;
 
 use crate::ast::{Expr, Value};
 use crate::error::{self, Result};
-use crate::statements::table_idents_to_full_name;
 
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct Insert {
@@ -26,13 +25,6 @@ pub struct Insert {
 }
 
 impl Insert {
-    pub fn full_table_name(&self) -> Result<(String, String, String)> {
-        match &self.inner {
-            Statement::Insert { table_name, .. } => table_idents_to_full_name(table_name),
-            _ => unreachable!(),
-        }
-    }
-
     pub fn table_name(&self) -> &ObjectName {
         match &self.inner {
             Statement::Insert { table_name, .. } => table_name,