test: Sqlness tests for distribute mode (#979)

* test: Sqlness tests for distribute mode

* ci

* fix: resolve PR comments

* fix: resolve PR comments

Author: MichaelScofield

Cargo.lock

@@ -335,7 +335,7 @@ impl ErrorExt for Error {
                 source.status_code()
             }
             Error::DecodeLogicalPlan { source } => source.status_code(),
-            Error::NewCatalog { source } => source.status_code(),
+            Error::NewCatalog { source } | Error::RegisterSchema { source } => source.status_code(),
             Error::FindTable { source, .. } => source.status_code(),
             Error::CreateTable { source, .. }
             | Error::GetTable { source, .. }
@@ -379,7 +379,6 @@ impl ErrorExt for Error {
             | Error::CreateDir { .. }
             | Error::InsertSystemCatalog { .. }
             | Error::RenameTable { .. }
-            | Error::RegisterSchema { .. }
             | Error::Catalog { .. }
             | Error::MissingRequiredField { .. }
             | Error::IncorrectInternalState { .. } => StatusCode::Internal,

src/datanode/src/error.rs

@@ -72,7 +72,7 @@ impl SqlHandler {
         alter_table: AlterTable,
         table_ref: TableReference,
     ) -> Result<AlterTableRequest> {
-        let alter_kind = match alter_table.alter_operation() {
+        let alter_kind = match &alter_table.alter_operation() {
             AlterTableOperation::AddConstraint(table_constraint) => {
                 return error::InvalidSqlSnafu {
                     msg: format!("unsupported table constraint {table_constraint}"),

src/datanode/src/sql/alter.rs

@@ -82,7 +82,12 @@ impl CatalogManager for FrontendCatalogManager {
         Ok(true)
     }
 
-    async fn deregister_table(&self, _request: DeregisterTableRequest) -> CatalogResult<bool> {
+    async fn deregister_table(&self, request: DeregisterTableRequest) -> CatalogResult<bool> {
+        let table_name = TableName::new(request.catalog, request.schema, request.table_name);
+        self.partition_manager
+            .table_routes()
+            .invalidate_table_route(&table_name)
+            .await;
         Ok(true)
     }
 

src/frontend/src/catalog.rs

@@ -114,7 +114,7 @@ pub enum Error {
         backtrace: Backtrace,
     },
 
-    #[snafu(display("Table not found: {}", table_name))]
+    #[snafu(display("Table `{}` not exist", table_name))]
     TableNotFound {
         table_name: String,
         backtrace: Backtrace,
@@ -214,6 +214,9 @@ pub enum Error {
         backtrace: Backtrace,
     },
 
+    #[snafu(display("Schema {} already exists", name))]
+    SchemaExists { name: String, backtrace: Backtrace },
+
     #[snafu(display("Table occurs error, source: {}", source))]
     Table {
         #[snafu(backtrace)]
@@ -392,8 +395,9 @@ impl ErrorExt for Error {
             Error::StartMetaClient { source } | Error::RequestMeta { source } => {
                 source.status_code()
             }
-            Error::SchemaNotFound { .. } => StatusCode::InvalidArguments,
-            Error::CatalogNotFound { .. } => StatusCode::InvalidArguments,
+            Error::CatalogNotFound { .. }
+            | Error::SchemaNotFound { .. }
+            | Error::SchemaExists { .. } => StatusCode::InvalidArguments,
 
             Error::BuildCreateExprOnInsertion { source }
             | Error::ToTableInsertRequest { source }

src/frontend/src/error.rs

@@ -396,7 +396,7 @@ impl Instance {
             | Statement::Alter(_)
             | Statement::DropTable(_) => self.sql_handler.do_statement_query(stmt, query_ctx).await,
             Statement::Use(db) => self.handle_use(db, query_ctx),
-            _ => NotSupportedSnafu {
+            Statement::ShowCreateTable(_) => NotSupportedSnafu {
                 feat: format!("{stmt:?}"),
             }
             .fail(),

src/frontend/src/instance.rs

@@ -38,8 +38,8 @@ use datatypes::schema::{RawSchema, Schema};
 use meta_client::client::MetaClient;
 use meta_client::rpc::router::DeleteRequest as MetaDeleteRequest;
 use meta_client::rpc::{
-    CreateRequest as MetaCreateRequest, Partition as MetaPartition, PutRequest, RouteResponse,
-    TableName,
+    CompareAndPutRequest, CreateRequest as MetaCreateRequest, Partition as MetaPartition,
+    RouteResponse, TableName,
 };
 use partition::partition::{PartitionBound, PartitionDef};
 use query::parser::QueryStatement;
@@ -60,8 +60,9 @@ use crate::datanode::DatanodeClients;
 use crate::error::{
     self, AlterExprToRequestSnafu, CatalogEntrySerdeSnafu, CatalogNotFoundSnafu, CatalogSnafu,
     ColumnDataTypeSnafu, DeserializePartitionSnafu, ParseSqlSnafu, PrimaryKeyNotFoundSnafu,
-    RequestDatanodeSnafu, RequestMetaSnafu, Result, SchemaNotFoundSnafu, StartMetaClientSnafu,
-    TableNotFoundSnafu, TableSnafu, ToTableInsertRequestSnafu,
+    RequestDatanodeSnafu, RequestMetaSnafu, Result, SchemaExistsSnafu, SchemaNotFoundSnafu,
+    StartMetaClientSnafu, TableAlreadyExistSnafu, TableNotFoundSnafu, TableSnafu,
+    ToTableInsertRequestSnafu,
 };
 use crate::expr_factory;
 use crate::instance::parse_stmt;
@@ -105,6 +106,26 @@ impl DistInstance {
             &create_table.table_name,
         );
 
+        if self
+            .catalog_manager
+            .table(
+                &table_name.catalog_name,
+                &table_name.schema_name,
+                &table_name.table_name,
+            )
+            .context(CatalogSnafu)?
+            .is_some()
+        {
+            return if create_table.create_if_not_exists {
+                Ok(Output::AffectedRows(0))
+            } else {
+                TableAlreadyExistSnafu {
+                    table: table_name.to_string(),
+                }
+                .fail()
+            };
+        }
+
         let mut table_info = create_table_info(create_table)?;
 
         let response = self
@@ -157,7 +178,7 @@ impl DistInstance {
                 .register_table(request)
                 .await
                 .context(CatalogSnafu)?,
-            error::TableAlreadyExistSnafu {
+            TableAlreadyExistSnafu {
                 table: table_name.to_string()
             }
         );
@@ -266,6 +287,10 @@ impl DistInstance {
                 let create_expr = &mut expr_factory::create_to_expr(&stmt, query_ctx)?;
                 Ok(self.create_table(create_expr, stmt.partitions).await?)
             }
+            Statement::Alter(alter_table) => {
+                let expr = grpc::to_alter_expr(alter_table, query_ctx)?;
+                return self.handle_alter_table(expr).await;
+            }
             Statement::DropTable(stmt) => {
                 let (catalog, schema, table) =
                     table_idents_to_full_name(stmt.table_name(), query_ctx)
@@ -358,10 +383,19 @@ impl DistInstance {
             .store_client()
             .context(StartMetaClientSnafu)?;
 
-        let request = PutRequest::default()
+        let request = CompareAndPutRequest::new()
             .with_key(key.to_string())
             .with_value(value.as_bytes().context(CatalogEntrySerdeSnafu)?);
-        client.put(request.into()).await.context(RequestMetaSnafu)?;
+        let response = client
+            .compare_and_put(request.into())
+            .await
+            .context(RequestMetaSnafu)?;
+        ensure!(
+            response.success,
+            SchemaExistsSnafu {
+                name: key.schema_name
+            }
+        );
 
         Ok(Output::AffectedRows(1))
     }

src/frontend/src/instance/distributed.rs

@@ -12,16 +12,22 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use alter_expr::Kind;
 use api::v1::ddl_request::Expr as DdlExpr;
 use api::v1::greptime_request::Request;
+use api::v1::{alter_expr, AddColumn, AddColumns, AlterExpr, DropColumn, DropColumns, RenameTable};
 use async_trait::async_trait;
+use common_error::prelude::BoxedError;
 use common_query::Output;
+use datanode::instance::sql::table_idents_to_full_name;
 use meta_client::rpc::TableName;
 use servers::query_handler::grpc::GrpcQueryHandler;
 use session::context::QueryContextRef;
-use snafu::OptionExt;
+use snafu::{OptionExt, ResultExt};
+use sql::statements::alter::{AlterTable, AlterTableOperation};
+use sql::statements::sql_column_def_to_grpc_column_def;
 
-use crate::error::{self, Result};
+use crate::error::{self, ExternalSnafu, Result};
 use crate::instance::distributed::DistInstance;
 
 #[async_trait]
@@ -56,3 +62,47 @@ impl GrpcQueryHandler for DistInstance {
         }
     }
 }
+
+pub(crate) fn to_alter_expr(
+    alter_table: AlterTable,
+    query_ctx: QueryContextRef,
+) -> Result<AlterExpr> {
+    let (catalog_name, schema_name, table_name) =
+        table_idents_to_full_name(alter_table.table_name(), query_ctx)
+            .map_err(BoxedError::new)
+            .context(ExternalSnafu)?;
+
+    let kind = match alter_table.alter_operation() {
+        AlterTableOperation::AddConstraint(_) => {
+            return error::NotSupportedSnafu {
+                feat: "ADD CONSTRAINT",
+            }
+            .fail();
+        }
+        AlterTableOperation::AddColumn { column_def } => Kind::AddColumns(AddColumns {
+            add_columns: vec![AddColumn {
+                column_def: Some(
+                    sql_column_def_to_grpc_column_def(column_def)
+                        .map_err(BoxedError::new)
+                        .context(ExternalSnafu)?,
+                ),
+                is_key: false,
+            }],
+        }),
+        AlterTableOperation::DropColumn { name } => Kind::DropColumns(DropColumns {
+            drop_columns: vec![DropColumn {
+                name: name.value.to_string(),
+            }],
+        }),
+        AlterTableOperation::RenameTable { new_table_name } => Kind::RenameTable(RenameTable {
+            new_table_name: new_table_name.to_string(),
+        }),
+    };
+
+    Ok(AlterExpr {
+        catalog_name,
+        schema_name,
+        table_name,
+        kind: Some(kind),
+    })
+}

src/frontend/src/instance/distributed/grpc.rs

@@ -46,6 +46,10 @@ impl PartitionRuleManager {
         Self { table_routes }
     }
 
+    pub fn table_routes(&self) -> &TableRoutes {
+        self.table_routes.as_ref()
+    }
+
     /// Find table route of given table name.
     pub async fn find_table_route(&self, table: &TableName) -> Result<Arc<TableRoute>> {
         self.table_routes.get_route(table).await

src/partition/src/manager.rs

@@ -24,6 +24,7 @@ use crate::error::{self, Result};
 
 pub struct TableRoutes {
     meta_client: Arc<MetaClient>,
+    // TODO(LFC): Use table id as cache key, then remove all the manually invoked cache invalidations.
     cache: Cache<TableName, Arc<TableRoute>>,
 }
 
@@ -72,4 +73,8 @@ impl TableRoutes {
     pub async fn insert_table_route(&self, table_name: TableName, table_route: Arc<TableRoute>) {
         self.cache.insert(table_name, table_route).await
     }
+
+    pub async fn invalidate_table_route(&self, table_name: &TableName) {
+        self.cache.invalidate(table_name).await
+    }
 }