feat: create DB and Tables via REST and CLI

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;