feat: create DB and Tables via REST and CLI

This commit does a few things:

1. It brings the database command naming scheme for types inline with
   the rest of the CLI types
2. It brings the table command naming scheme for types inline with
   the rest of the CLI types
3. Adds tests to check that the num of dbs is not exceeded and that you
   cannot create more than one database with a given name.
4. Adds tests to check that you can create a table and put data into it
   and querying it
5. Adds tests for the CLI for both the database and table commands
6. It creates an endpoint to create databases given a JSON blob
7. It creates an endpoint to create tables given a JSON blob

With this users can now create a database or table without first needing
to write to the database via the line protocol!

Closes #25640
Closes #25641

influxdb3/src/commands/manage/database.rs

@@ -5,13 +5,14 @@ use secrecy::ExposeSecret;
 use crate::commands::common::InfluxDb3Config;
 
 #[derive(Debug, clap::Parser)]
-pub(crate) struct ManageDatabaseConfig {
+pub(crate) struct Config {
     #[clap(subcommand)]
     command: Command,
 }
 
 #[derive(Debug, clap::Parser)]
 enum Command {
+    Create(DatabaseConfig),
     Delete(DatabaseConfig),
 }
 
@@ -21,8 +22,25 @@ pub struct DatabaseConfig {
     influxdb3_config: InfluxDb3Config,
 }
 
-pub async fn delete_database(config: ManageDatabaseConfig) -> Result<(), Box<dyn Error>> {
+pub async fn command(config: Config) -> Result<(), Box<dyn Error>> {
     match config.command {
+        Command::Create(config) => {
+            let InfluxDb3Config {
+                host_url,
+                database_name,
+                auth_token,
+            } = config.influxdb3_config;
+
+            let mut client = influxdb3_client::Client::new(host_url)?;
+
+            if let Some(t) = auth_token {
+                client = client.with_auth_token(t.expose_secret());
+            }
+
+            client.api_v3_configure_db_create(&database_name).await?;
+
+            println!("Database {:?} created successfully", &database_name);
+        }
         Command::Delete(config) => {
             let InfluxDb3Config {
                 host_url,

influxdb3/src/commands/manage/table.rs

@@ -1,31 +1,71 @@
-use std::{error::Error, io};
+use std::{error::Error, fmt::Display, io, str::FromStr};
 
 use secrecy::ExposeSecret;
 
 use crate::commands::common::InfluxDb3Config;
 
 #[derive(Debug, clap::Parser)]
-pub(crate) struct ManageTableConfig {
+pub(crate) struct Config {
     #[clap(subcommand)]
     command: Command,
 }
 
 #[derive(Debug, clap::Parser)]
 enum Command {
-    Delete(TableConfig),
+    Create(CreateTableConfig),
+    Delete(DeleteTableConfig),
 }
 
 #[derive(Debug, clap::Parser)]
-pub struct TableConfig {
-    #[clap(short = 't', long = "table")]
-    table: String,
+pub struct DeleteTableConfig {
+    #[clap(short = 't', long = "table", required = true)]
+    table_name: String,
 
     #[clap(flatten)]
     influxdb3_config: InfluxDb3Config,
 }
 
-pub async fn delete_table(config: ManageTableConfig) -> Result<(), Box<dyn Error>> {
+#[derive(Debug, clap::Parser)]
+pub struct CreateTableConfig {
+    #[clap(short = 't', long = "table", required = true)]
+    table_name: String,
+
+    #[clap(long = "tags", required = true, num_args=0..)]
+    tags: Vec<String>,
+
+    #[clap(short = 'f', long = "fields", value_parser = parse_key_val::<String, DataType>, required = true, num_args=0..)]
+    fields: Vec<(String, DataType)>,
+
+    #[clap(flatten)]
+    influxdb3_config: InfluxDb3Config,
+}
+
+pub async fn command(config: Config) -> Result<(), Box<dyn Error>> {
     match config.command {
+        Command::Create(CreateTableConfig {
+            table_name,
+            tags,
+            fields,
+            influxdb3_config:
+                InfluxDb3Config {
+                    host_url,
+                    database_name,
+                    auth_token,
+                },
+        }) => {
+            let mut client = influxdb3_client::Client::new(host_url)?;
+            if let Some(t) = auth_token {
+                client = client.with_auth_token(t.expose_secret());
+            }
+            client
+                .api_v3_configure_table_create(&database_name, &table_name, tags, fields)
+                .await?;
+
+            println!(
+                "Table {:?}.{:?} created successfully",
+                &database_name, &table_name
+            );
+        }
         Command::Delete(config) => {
             let InfluxDb3Config {
                 host_url,
@@ -34,7 +74,7 @@ pub async fn delete_table(config: ManageTableConfig) -> Result<(), Box<dyn Error
             } = config.influxdb3_config;
             println!(
                 "Are you sure you want to delete {:?}.{:?}? Enter 'yes' to confirm",
-                database_name, &config.table,
+                database_name, &config.table_name,
             );
             let mut confirmation = String::new();
             let _ = io::stdin().read_line(&mut confirmation);
@@ -46,15 +86,74 @@ pub async fn delete_table(config: ManageTableConfig) -> Result<(), Box<dyn Error
                     client = client.with_auth_token(t.expose_secret());
                 }
                 client
-                    .api_v3_configure_table_delete(&database_name, &config.table)
+                    .api_v3_configure_table_delete(&database_name, &config.table_name)
                     .await?;
 
                 println!(
                     "Table {:?}.{:?} deleted successfully",
-                    &database_name, &config.table
+                    &database_name, &config.table_name
                 );
             }
         }
     }
     Ok(())
 }
+
+#[derive(Debug, Copy, Clone, PartialEq, Eq)]
+pub enum DataType {
+    Int64,
+    Uint64,
+    Float64,
+    Utf8,
+    Bool,
+}
+
+#[derive(Debug, PartialEq, Eq, thiserror::Error)]
+#[error("{0} is not a valid data type, values are int64, uint64, float64, utf8, and bool")]
+pub struct ParseDataTypeError(String);
+
+impl FromStr for DataType {
+    type Err = ParseDataTypeError;
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        match s {
+            "int64" => Ok(Self::Int64),
+            "uint64" => Ok(Self::Uint64),
+            "float64" => Ok(Self::Float64),
+            "utf8" => Ok(Self::Utf8),
+            "bool" => Ok(Self::Bool),
+            _ => Err(ParseDataTypeError(s.into())),
+        }
+    }
+}
+
+impl Display for DataType {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::Int64 => write!(f, "int64"),
+            Self::Uint64 => write!(f, "uint64"),
+            Self::Float64 => write!(f, "float64"),
+            Self::Utf8 => write!(f, "utf8"),
+            Self::Bool => write!(f, "bool"),
+        }
+    }
+}
+
+impl From<DataType> for String {
+    fn from(data: DataType) -> Self {
+        data.to_string()
+    }
+}
+
+/// Parse a single key-value pair
+fn parse_key_val<T, U>(s: &str) -> Result<(T, U), Box<dyn Error + Send + Sync + 'static>>
+where
+    T: std::str::FromStr,
+    T::Err: Error + Send + Sync + 'static,
+    U: std::str::FromStr,
+    U::Err: Error + Send + Sync + 'static,
+{
+    let pos = s
+        .find(':')
+        .ok_or_else(|| format!("invalid FIELD:VALUE. No `:` found in `{s}`"))?;
+    Ok((s[..pos].parse()?, s[pos + 1..].parse()?))
+}

influxdb3/src/main.rs

@@ -97,10 +97,10 @@ enum Command {
     MetaCache(commands::meta_cache::Config),
 
     /// Manage database (delete only for the moment)
-    Database(commands::manage::database::ManageDatabaseConfig),
+    Database(commands::manage::database::Config),
 
     /// Manage table (delete only for the moment)
-    Table(commands::manage::table::ManageTableConfig),
+    Table(commands::manage::table::Config),
 }
 
 fn main() -> Result<(), std::io::Error> {
@@ -166,14 +166,14 @@ fn main() -> Result<(), std::io::Error> {
                 }
             }
             Some(Command::Database(config)) => {
-                if let Err(e) = commands::manage::database::delete_database(config).await {
-                    eprintln!("Database delete command failed: {e}");
+                if let Err(e) = commands::manage::database::command(config).await {
+                    eprintln!("Database command failed: {e}");
                     std::process::exit(ReturnCode::Failure as _)
                 }
             }
             Some(Command::Table(config)) => {
-                if let Err(e) = commands::manage::table::delete_table(config).await {
-                    eprintln!("Table delete command failed: {e}");
+                if let Err(e) = commands::manage::table::command(config).await {
+                    eprintln!("Table command failed: {e}");
                     std::process::exit(ReturnCode::Failure as _)
                 }
             }

influxdb3/tests/server/cli.rs

@@ -6,6 +6,8 @@ use std::{
 
 use assert_cmd::cargo::CommandCargoExt;
 use observability_deps::tracing::debug;
+use pretty_assertions::assert_eq;
+use serde_json::{json, Value};
 use test_helpers::assert_contains;
 
 use crate::TestServer;
@@ -77,6 +79,57 @@ pub fn run_with_confirmation_and_err(args: &[&str]) -> String {
         .into()
 }
 
+#[test_log::test(tokio::test)]
+async fn test_create_database() {
+    let server = TestServer::spawn().await;
+    let server_addr = server.client_addr();
+    let db_name = "foo";
+    let result = run_with_confirmation(&[
+        "database",
+        "create",
+        "--dbname",
+        db_name,
+        "--host",
+        &server_addr,
+    ]);
+    debug!(result = ?result, "create database");
+    assert_contains!(&result, "Database \"foo\" created successfully");
+}
+
+#[test_log::test(tokio::test)]
+async fn test_create_database_limit() {
+    let server = TestServer::spawn().await;
+    let server_addr = server.client_addr();
+    let db_name = "foo";
+    for i in 0..5 {
+        let name = format!("{db_name}{i}");
+        let result = run_with_confirmation(&[
+            "database",
+            "create",
+            "--dbname",
+            &name,
+            "--host",
+            &server_addr,
+        ]);
+        debug!(result = ?result, "create database");
+        assert_contains!(&result, format!("Database \"{name}\" created successfully"));
+    }
+
+    let result = run_with_confirmation_and_err(&[
+        "database",
+        "create",
+        "--dbname",
+        "foo5",
+        "--host",
+        &server_addr,
+    ]);
+    debug!(result = ?result, "create database");
+    assert_contains!(
+        &result,
+        "Adding a new database would exceed limit of 5 databases"
+    );
+}
+
 #[test_log::test(tokio::test)]
 async fn test_delete_database() {
     let server = TestServer::spawn().await;
@@ -119,6 +172,91 @@ async fn test_delete_missing_database() {
     assert_contains!(&result, "404");
 }
 
+#[test_log::test(tokio::test)]
+async fn test_create_table() {
+    let server = TestServer::spawn().await;
+    let server_addr = server.client_addr();
+    let db_name = "foo";
+    let table_name = "bar";
+    let result = run_with_confirmation(&[
+        "database",
+        "create",
+        "--dbname",
+        db_name,
+        "--host",
+        &server_addr,
+    ]);
+    debug!(result = ?result, "create database");
+    assert_contains!(&result, "Database \"foo\" created successfully");
+    let result = run_with_confirmation(&[
+        "table",
+        "create",
+        "--dbname",
+        db_name,
+        "--table",
+        table_name,
+        "--host",
+        &server_addr,
+        "--tags",
+        "one",
+        "two",
+        "three",
+        "--fields",
+        "four:utf8",
+        "five:uint64",
+        "six:float64",
+        "seven:int64",
+        "eight:bool",
+    ]);
+    debug!(result = ?result, "create table");
+    assert_contains!(&result, "Table \"foo\".\"bar\" created successfully");
+    // Check that we can query the table and that it has no values
+    let result = server
+        .api_v3_query_sql(&[
+            ("db", "foo"),
+            ("q", "SELECT * FROM bar"),
+            ("format", "json"),
+        ])
+        .await
+        .json::<Value>()
+        .await
+        .unwrap();
+    assert_eq!(result, json!([]));
+    server
+        .write_lp_to_db(
+            db_name,
+            format!("{table_name},one=1,two=2,three=3 four=\"4\",five=5u,six=6,seven=7i,eight=true 1000"),
+            influxdb3_client::Precision::Second,
+        )
+        .await
+        .expect("write to db");
+    // Check that we can get data from the table
+    let result = server
+        .api_v3_query_sql(&[
+            ("db", "foo"),
+            ("q", "SELECT * FROM bar"),
+            ("format", "json"),
+        ])
+        .await
+        .json::<Value>()
+        .await
+        .unwrap();
+    assert_eq!(
+        result,
+        json!([{
+            "one": "1",
+            "two": "2",
+            "three": "3",
+            "four": "4",
+            "five": 5,
+            "six": 6.0,
+            "seven": 7,
+            "eight": true,
+            "time": "1970-01-01T00:16:40"
+        }])
+    );
+}
+
 #[test_log::test(tokio::test)]
 async fn test_delete_table() {
     let server = TestServer::spawn().await;