fix: check table exists before allocating table id (#2546)

* fix: check table exists before allocating table_id

* chore: apply suggestions from CR

src/frontend/src/error.rs

@@ -182,9 +182,6 @@ pub enum Error {
     #[snafu(display("Failed to find leaders when altering table, table: {}", table))]
     LeaderNotFound { table: String, location: Location },
 
-    #[snafu(display("Table already exists: `{}`", table))]
-    TableAlreadyExist { table: String, location: Location },
-
     #[snafu(display("Failed to found context value: {}", key))]
     ContextValueNotFound { key: String, location: Location },
 
@@ -345,7 +342,6 @@ impl ErrorExt for Error {
             | Error::ExecLogicalPlan { source, .. } => source.status_code(),
 
             Error::LeaderNotFound { .. } => StatusCode::StorageUnavailable,
-            Error::TableAlreadyExist { .. } => StatusCode::TableAlreadyExists,
             Error::InvokeRegionServer { source, .. } => source.status_code(),
 
             Error::External { source, .. } => source.status_code(),

src/operator/src/error.rs

@@ -28,6 +28,9 @@ use snafu::{Location, Snafu};
 #[snafu(visibility(pub))]
 #[stack_trace_debug]
 pub enum Error {
+    #[snafu(display("Table already exists: `{}`", table))]
+    TableAlreadyExists { table: String, location: Location },
+
     #[snafu(display("Failed to invalidate table cache"))]
     InvalidateTableCache {
         location: Location,
@@ -512,6 +515,8 @@ impl ErrorExt for Error {
             | Error::InferFileTableSchema { .. }
             | Error::SchemaIncompatible { .. } => StatusCode::InvalidArguments,
 
+            Error::TableAlreadyExists { .. } => StatusCode::TableAlreadyExists,
+
             Error::NotSupported { .. } => StatusCode::Unsupported,
 
             Error::TableMetadataManager { source, .. } => source.status_code(),

src/operator/src/statement/ddl.rs

@@ -95,6 +95,31 @@ impl StatementExecutor {
             .fail();
         };
 
+        // if table exists.
+        if let Some(table) = self
+            .catalog_manager
+            .table(
+                &create_table.catalog_name,
+                &create_table.schema_name,
+                &create_table.table_name,
+            )
+            .await
+            .context(error::CatalogSnafu)?
+        {
+            return if create_table.create_if_not_exists {
+                Ok(table)
+            } else {
+                error::TableAlreadyExistsSnafu {
+                    table: format_full_table_name(
+                        &create_table.catalog_name,
+                        &create_table.schema_name,
+                        &create_table.table_name,
+                    ),
+                }
+                .fail()
+            };
+        }
+
         let table_name = TableName::new(
             &create_table.catalog_name,
             &create_table.schema_name,

tests/cases/standalone/common/create/create.result

@@ -46,6 +46,10 @@ CREATE TABLE test2 (i INTEGER, j TIMESTAMP TIME INDEX);
 
 Affected Rows: 0
 
+CREATE TABLE test2 (i INTEGER, j TIMESTAMP TIME INDEX);
+
+Error: 4000(TableAlreadyExists), Table already exists: `greptime.public.test2`
+
 DESC TABLE integers;
 
 +--------+----------------------+-----+------+---------+---------------+

tests/cases/standalone/common/create/create.sql

@@ -22,6 +22,8 @@ CREATE TABLE test2 (i INTEGER, j TIMESTAMP TIME INDEX NULL);
 
 CREATE TABLE test2 (i INTEGER, j TIMESTAMP TIME INDEX);
 
+CREATE TABLE test2 (i INTEGER, j TIMESTAMP TIME INDEX);
+
 DESC TABLE integers;
 
 DESC TABLE test1;