feat: Cleanup CLI flags for InfluxDB 3 Core

Cargo.toml

@@ -3,7 +3,8 @@
 members = [
     "influxdb3",
     "influxdb3_cache",
-    "influxdb3_catalog", "influxdb3_clap_blocks",
+    "influxdb3_catalog",
+    "influxdb3_clap_blocks",
     "influxdb3_client",
     "influxdb3_id",
     "influxdb3_load_generator",
@@ -77,10 +78,12 @@ humantime = "2.1.0"
 hyper = "0.14"
 insta = { version = "1.39", features = ["json", "redactions", "yaml"] }
 indexmap = { version = "2.2.6" }
+itertools = "0.13.0"
 libc = { version = "0.2" }
 mime = "0.3.17"
 mockito = { version = "1.4.0", default-features = false }
 mockall = { version = "0.13.0" }
+non-empty-string = "0.2.5"
 num_cpus = "1.16.0"
 object_store = "0.11.1"
 parking_lot = "0.12.1"
@@ -105,9 +108,11 @@ serde_json = "1.0.127"
 serde_urlencoded = "0.7.0"
 serde_with = "3.8.1"
 sha2 = "0.10.8"
+snafu = "0.8"
 snap = "1.0.0"
 sqlparser = "0.48.0"
 sysinfo = "0.30.8"
+tempfile = "3.14.0"
 test-log = { version = "0.2.16", features = ["trace"] }
 thiserror = "1.0"
 tokio = { version = "1.42", features = ["full"] }
@@ -126,7 +131,6 @@ num = { version = "0.4.3" }
 # Core.git crates we depend on
 arrow_util = { git = "https://github.com/influxdata/influxdb3_core", rev = "a5f6076c966f4940a67998e0b85d12c3e8596715" }
 authz = { git = "https://github.com/influxdata/influxdb3_core", rev = "a5f6076c966f4940a67998e0b85d12c3e8596715" }
-clap_blocks = { git = "https://github.com/influxdata/influxdb3_core", rev = "a5f6076c966f4940a67998e0b85d12c3e8596715" }
 data_types = { git = "https://github.com/influxdata/influxdb3_core", rev = "a5f6076c966f4940a67998e0b85d12c3e8596715" }
 datafusion_util = { git = "https://github.com/influxdata/influxdb3_core", rev = "a5f6076c966f4940a67998e0b85d12c3e8596715" }
 executor = { git = "https://github.com/influxdata/influxdb3_core", rev = "a5f6076c966f4940a67998e0b85d12c3e8596715" }
@@ -154,7 +158,7 @@ trace = { git = "https://github.com/influxdata/influxdb3_core", rev = "a5f6076c9
 trace_exporters = { git = "https://github.com/influxdata/influxdb3_core", rev = "a5f6076c966f4940a67998e0b85d12c3e8596715" }
 trace_http = { git = "https://github.com/influxdata/influxdb3_core", rev = "a5f6076c966f4940a67998e0b85d12c3e8596715" }
 tracker = { git = "https://github.com/influxdata/influxdb3_core", rev = "a5f6076c966f4940a67998e0b85d12c3e8596715" }
-trogging = { git = "https://github.com/influxdata/influxdb3_core", rev = "a5f6076c966f4940a67998e0b85d12c3e8596715" }
+trogging = { git = "https://github.com/influxdata/influxdb3_core", rev = "a5f6076c966f4940a67998e0b85d12c3e8596715", features = ["clap"] }
 
 [workspace.lints.rust]
 missing_copy_implementations = "deny"

influxdb3/Cargo.toml

@@ -8,7 +8,6 @@ license.workspace = true
 [dependencies]
 # Core Crates
 authz.workspace = true
-clap_blocks.workspace = true
 datafusion_util.workspace = true
 iox_query.workspace = true
 iox_time.workspace = true
@@ -63,9 +62,9 @@ console-subscriber = { version = "0.1.10", optional = true, features = ["parking
 [features]
 default = ["jemalloc_replacing_malloc", "azure", "gcp", "aws"]
 
-azure = ["clap_blocks/azure"] # Optional Azure Object store support
-gcp = ["clap_blocks/gcp"] # Optional GCP object store support
-aws = ["clap_blocks/aws"] # Optional AWS / S3 object store support
+azure = ["influxdb3_clap_blocks/azure"] # Optional Azure Object store support
+gcp = ["influxdb3_clap_blocks/gcp"] # Optional GCP object store support
+aws = ["influxdb3_clap_blocks/aws"] # Optional AWS / S3 object store support
 
 # Enable tokio_console support (https://github.com/tokio-rs/console)
 #

influxdb3/src/commands/activate.rs

@@ -0,0 +1,63 @@
+use crate::commands::common::InfluxDb3Config;
+use influxdb3_client::Client;
+use secrecy::ExposeSecret;
+use std::error::Error;
+
+#[derive(Debug, clap::Parser)]
+pub struct Config {
+    #[clap(subcommand)]
+    cmd: SubCommand,
+}
+
+impl Config {
+    fn get_client(&self) -> Result<Client, Box<dyn Error>> {
+        let (host_url, auth_token) = match &self.cmd {
+            SubCommand::Trigger(TriggerConfig {
+                influxdb3_config:
+                    InfluxDb3Config {
+                        host_url,
+                        auth_token,
+                        ..
+                    },
+                ..
+            }) => (host_url, auth_token),
+        };
+        let mut client = Client::new(host_url.clone())?;
+        if let Some(token) = &auth_token {
+            client = client.with_auth_token(token.expose_secret());
+        }
+        Ok(client)
+    }
+}
+
+#[derive(Debug, clap::Subcommand)]
+enum SubCommand {
+    /// Activate a trigger to enable plugin execution
+    Trigger(TriggerConfig),
+}
+
+#[derive(Debug, clap::Parser)]
+struct TriggerConfig {
+    #[clap(flatten)]
+    influxdb3_config: InfluxDb3Config,
+
+    /// Name of trigger to manage
+    #[clap(required = true)]
+    trigger_name: String,
+}
+
+pub async fn command(config: Config) -> Result<(), Box<dyn Error>> {
+    let client = config.get_client()?;
+    match config.cmd {
+        SubCommand::Trigger(TriggerConfig {
+            influxdb3_config: InfluxDb3Config { database_name, .. },
+            trigger_name,
+        }) => {
+            client
+                .api_v3_configure_processing_engine_trigger_activate(database_name, &trigger_name)
+                .await?;
+            println!("Trigger {} activated successfully", trigger_name);
+        }
+    }
+    Ok(())
+}

influxdb3/src/commands/common.rs

@@ -1,25 +1,26 @@
-use std::str::FromStr;
-
 use clap::Parser;
 use secrecy::Secret;
+use std::error::Error;
+use std::fmt::Display;
+use std::str::FromStr;
 use url::Url;
 
 #[derive(Debug, Parser)]
 pub struct InfluxDb3Config {
-    /// The host URL of the running InfluxDB 3.0 server
+    /// The host URL of the running InfluxDB 3 Core server
     #[clap(
-        short = 'h',
+        short = 'H',
         long = "host",
         env = "INFLUXDB3_HOST_URL",
         default_value = "http://127.0.0.1:8181"
     )]
     pub host_url: Url,
 
-    /// The database name to run the query against
-    #[clap(short = 'd', long = "dbname", env = "INFLUXDB3_DATABASE_NAME")]
+    /// The name of the database to operate on
+    #[clap(short = 'd', long = "database", env = "INFLUXDB3_DATABASE_NAME")]
     pub database_name: String,
 
-    /// The token for authentication with the InfluxDB 3.0 server
+    /// The token for authentication with the InfluxDB 3 Core server
     #[clap(long = "token", env = "INFLUXDB3_AUTH_TOKEN")]
     pub auth_token: Option<Secret<String>>,
 }
@@ -77,3 +78,61 @@ impl<T, const SEPARATOR: char> IntoIterator for SeparatedList<T, SEPARATOR> {
         self.0.into_iter()
     }
 }
+#[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
+pub 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()?))
+}