fix: clarify table creation conflict error message (#25936)

Also include a basic CLI integration test to exemplify the new error message.

influxdb3/tests/server/cli.rs

@@ -473,6 +473,48 @@ async fn test_create_table() {
     );
 }
 
+#[test_log::test(tokio::test)]
+async fn test_create_table_fail_existing() {
+    let server = TestServer::spawn().await;
+    let server_addr = server.client_addr();
+    let db_name = "foo";
+    let table_name = "bar";
+    let result = run_with_confirmation(&["create", "database", db_name, "--host", &server_addr]);
+    debug!(result = ?result, "create database");
+    assert_contains!(&result, "Database \"foo\" created successfully");
+    let result = run_with_confirmation(&[
+        "create",
+        "table",
+        table_name,
+        "--database",
+        db_name,
+        "--host",
+        &server_addr,
+        "--tags",
+        "one",
+        "--fields",
+        "four:utf8",
+    ]);
+    debug!(result = ?result, "create table");
+    assert_contains!(&result, "Table \"foo\".\"bar\" created successfully");
+
+    let result = run_with_confirmation_and_err(&[
+        "create",
+        "table",
+        table_name,
+        "--database",
+        db_name,
+        "--host",
+        &server_addr,
+        "--tags",
+        "one",
+        "--fields",
+        "four:utf8",
+    ]);
+
+    insta::assert_snapshot!("test_create_table_fail_existing", result);
+}
+
 #[test_log::test(tokio::test)]
 async fn test_delete_table() {
     let server = TestServer::spawn().await;

influxdb3/tests/server/snapshots/server__cli__test_create_table_fail_existing.snap

@@ -0,0 +1,6 @@
+---
+source: influxdb3/tests/server/cli.rs
+expression: result
+snapshot_kind: text
+---
+Create command failed: server responded with error [409 Conflict]: table 'foo.bar' already exists

influxdb3_catalog/src/catalog.rs

@@ -30,8 +30,8 @@ const SOFT_DELETION_TIME_FORMAT: &str = "%Y%m%dT%H%M%S";
 
 #[derive(Debug, Error, Clone)]
 pub enum Error {
-    #[error("catalog updated elsewhere")]
-    CatalogUpdatedElsewhere,
+    #[error("table '{table_name}' already exists")]
+    CatalogUpdatedElsewhere { table_name: Arc<str> },
 
     #[error(
         "Update to schema would exceed number of columns per table limit of {} columns",
@@ -609,7 +609,9 @@ impl DatabaseSchema {
                 // for existence and insertion.
                 // We'd like this to be automatically handled by the system,
                 // but for now it is better to error than get into an inconsistent state.
-                return Err(CatalogUpdatedElsewhere);
+                return Err(CatalogUpdatedElsewhere {
+                    table_name: Arc::clone(&table_def.table_name),
+                });
             }
             Overwritten::Neither | Overwritten::Pair(_, _) => {}
         }

influxdb3_server/src/http.rs

@@ -353,6 +353,15 @@ impl Error {
                     .body(Body::from(mc_err.to_string()))
                     .unwrap(),
             },
+            Self::WriteBuffer(
+                err @ WriteBufferError::CatalogUpdateError(CatalogError::CatalogUpdatedElsewhere {
+                    ..
+                })
+                | err @ WriteBufferError::TableAlreadyExists { .. },
+            ) => Response::builder()
+                .status(StatusCode::CONFLICT)
+                .body(Body::from(err.to_string()))
+                .unwrap(),
             Self::DbName(e) => {
                 let err: ErrorMessage<()> = ErrorMessage {
                     error: e.to_string(),

influxdb3_write/src/write_buffer/mod.rs

@@ -99,6 +99,12 @@ pub enum Error {
     #[error("tried accessing table that do not exist")]
     TableDoesNotExist,
 
+    #[error("table '{db_name}.{table_name}' already exists")]
+    TableAlreadyExists {
+        db_name: Arc<str>,
+        table_name: Arc<str>,
+    },
+
     #[error("tried accessing column with name ({0}) that does not exist")]
     ColumnDoesNotExist(String),
 
@@ -730,7 +736,15 @@ impl DatabaseManager for WriteBufferImpl {
         };
 
         let table_id = TableId::new();
-        let table_name = table.into();
+        let table_name: Arc<str> = table.into();
+
+        if db_schema.table_map.contains_right(&table_name) {
+            return Err(Error::TableAlreadyExists {
+                db_name: Arc::clone(&db_schema.name),
+                table_name: Arc::clone(&table_name),
+            });
+        }
+
         let mut key = Vec::new();
         let field_definitions = {
             let mut field_definitions = Vec::new();