feat: add InfluxDB support (#980)

crates/dbx-core/assets/database-drivers.manifest.json

@@ -526,6 +526,17 @@
       "skipTcpProbe": true,
       "defaultPort": 1972
     },
+    {
+      "dbType": "influxdb",
+      "label": "InfluxDB",
+      "runtimeMode": "agent",
+      "mcpMode": "bridge",
+      "agentKey": "influxdb",
+      "singleConnectionPool": false,
+      "metadataConnectionScoped": false,
+      "skipTcpProbe": true,
+      "defaultPort": 8086
+    },
     {
       "dbType": "jdbc",
       "label": "JDBC",

crates/dbx-core/src/connection.rs

@@ -48,6 +48,7 @@ pub enum PoolKind {
     ClickHouse(db::clickhouse_driver::ChClient),
     SqlServer(Arc<tokio::sync::Mutex<db::sqlserver::SqlServerClient>>),
     Elasticsearch(db::elasticsearch_driver::EsClient),
+    InfluxDb(db::influxdb_driver::InfluxdbClient),
     Agent(Arc<tokio::sync::Mutex<db::agent_driver::AgentDriverClient>>),
     ExternalTabular(Arc<external::ExternalPool>),
     ExternalDriver { driver_id: String, config: Arc<ConnectionConfig>, session: Arc<PluginDriverSession> },
@@ -482,6 +483,19 @@ impl AppState {
                 db::elasticsearch_driver::test_connection(&mut client, connect_timeout).await?;
                 PoolKind::Elasticsearch(client)
             }
+            DatabaseType::InfluxDb => {
+                let username = if db_config.username.is_empty() { None } else { Some(db_config.username.clone()) };
+                let password = if db_config.password.is_empty() { None } else { Some(db_config.password.clone()) };
+                let client = db::influxdb_driver::InfluxdbClient::new_with_ca_cert(
+                    &url,
+                    username,
+                    password,
+                    Some(&db_config.ca_cert_path),
+                    connect_timeout,
+                )?;
+                db::influxdb_driver::test_connection(&client, connect_timeout).await?;
+                PoolKind::InfluxDb(client)
+            }
             DatabaseType::Dameng
             | DatabaseType::Kingbase
             | DatabaseType::Highgo
@@ -960,6 +974,7 @@ pub async fn close_pool_kind(pool: PoolKind) {
         PoolKind::ClickHouse(_) => {}
         PoolKind::SqlServer(_) => {}
         PoolKind::Elasticsearch(_) => {}
+        PoolKind::InfluxDb(_) => {}
         PoolKind::Agent(client) => {
             let mut client = client.lock().await;
             let _ = client.disconnect().await;

crates/dbx-core/src/db/influxdb_driver.rs

@@ -0,0 +1,313 @@
+use percent_encoding::{utf8_percent_encode, NON_ALPHANUMERIC};
+use reqwest::{Certificate, Client as HttpClient};
+use serde::Deserialize;
+use std::fs;
+use std::time::{Duration, Instant};
+
+use super::with_connection_timeout;
+use crate::sql::starts_with_executable_sql_keyword;
+use crate::types::{ColumnInfo, DatabaseInfo, QueryResult, TableInfo};
+
+pub struct InfluxdbClient {
+    http: HttpClient,
+    base_url: String,
+    username: Option<String>,
+    password: Option<String>,
+}
+
+impl InfluxdbClient {
+    pub fn new(url: &str, username: Option<String>, password: Option<String>, timeout: Duration) -> Self {
+        let http = HttpClient::builder().connect_timeout(timeout).build().unwrap_or_else(|_| HttpClient::new());
+        Self { http, base_url: url.trim_end_matches('/').to_string(), username, password }
+    }
+
+    pub fn new_with_ca_cert(
+        url: &str,
+        username: Option<String>,
+        password: Option<String>,
+        ca_cert_path: Option<&str>,
+        timeout: Duration,
+    ) -> Result<Self, String> {
+        let mut builder = HttpClient::builder().connect_timeout(timeout);
+        if let Some(path) = ca_cert_path.map(str::trim).filter(|path| !path.is_empty()) {
+            let path = expand_cert_path(path);
+            let cert_bytes =
+                fs::read(&path).map_err(|e| format!("Failed to read InfluxDB CA certificate at {path}: {e}"))?;
+            let cert = Certificate::from_pem(&cert_bytes)
+                .or_else(|_| Certificate::from_der(&cert_bytes))
+                .map_err(|e| format!("Failed to parse InfluxDB CA certificate at {path}: {e}"))?;
+            builder = builder.add_root_certificate(cert);
+        }
+        let http = builder.build().map_err(|e| format!("Failed to configure InfluxDB HTTP client: {e}"))?;
+        Ok(Self { http, base_url: url.trim_end_matches('/').to_string(), username, password })
+    }
+}
+
+fn expand_cert_path(path: &str) -> String {
+    let home = || std::env::var(if cfg!(windows) { "USERPROFILE" } else { "HOME" }).ok();
+    if path == "~" || path.starts_with("~/") || path.starts_with("~\\") {
+        if let Some(home) = home() {
+            return format!("{}{}", home, &path[1..]);
+        }
+    }
+    if let Some(rest) = path.strip_prefix("$HOME") {
+        if let Some(home) = home() {
+            return format!("{home}{rest}");
+        }
+    }
+    if let Some(rest) = path.strip_prefix("${HOME}") {
+        if let Some(home) = home() {
+            return format!("{home}{rest}");
+        }
+    }
+    if let Some(rest) = path.strip_prefix("%USERPROFILE%") {
+        if let Ok(home) = std::env::var("USERPROFILE") {
+            return format!("{home}{rest}");
+        }
+    }
+    path.to_string()
+}
+
+impl Clone for InfluxdbClient {
+    fn clone(&self) -> Self {
+        Self {
+            http: self.http.clone(),
+            base_url: self.base_url.clone(),
+            username: self.username.clone(),
+            password: self.password.clone(),
+        }
+    }
+}
+
+#[derive(Deserialize, Default)]
+struct InfluxErrorResult {
+    #[serde(default)]
+    error: String,
+}
+
+#[derive(Deserialize)]
+struct InfluxJsonResult {
+    results: Vec<InfluxQueryResult>,
+}
+
+#[derive(Deserialize)]
+#[allow(dead_code)]
+struct InfluxQueryResult {
+    statement_id: usize,
+    #[serde(default)]
+    #[allow(dead_code)]
+    series: Vec<InfluxSeries>,
+}
+
+#[derive(Deserialize)]
+#[allow(dead_code)]
+struct InfluxSeries {
+    #[serde(default)]
+    #[allow(dead_code)]
+    name: String,
+    columns: Vec<String>,
+    values: Vec<Vec<serde_json::Value>>,
+}
+
+fn build_query_url(base_url: &str, database: Option<&str>, sql: &str) -> String {
+    let mut url = format!("{}/query", base_url);
+    let mut has_param = false;
+    if let Some(db) = database {
+        url.push_str(&format!("?db={db}"));
+        has_param = true;
+    }
+    if has_param {
+        url.push('&');
+    } else {
+        url.push('?');
+    }
+    let encoded_sql = utf8_percent_encode(sql, NON_ALPHANUMERIC);
+    url.push_str(&format!("q={encoded_sql}"));
+    url
+}
+
+fn build_request(client: &InfluxdbClient, req: reqwest::RequestBuilder) -> reqwest::RequestBuilder {
+    match (&client.username, &client.password) {
+        (Some(u), Some(p)) if !u.is_empty() => req.basic_auth(u, Some(p)),
+        (Some(u), None) if !u.is_empty() => req.basic_auth(u, None::<&str>),
+        _ => req,
+    }
+}
+
+async fn influx_query(client: &InfluxdbClient, sql: &str, database: Option<&str>) -> Result<InfluxJsonResult, String> {
+    let url = build_query_url(&client.base_url, database, sql);
+    log::info!("[influxdb] query url={url} username={:?} password={}", client.username, client.password.is_some());
+
+    let req = if starts_with_executable_sql_keyword(sql, &["SELECT", "SHOW"]) {
+        build_request(client, client.http.get(&url))
+    } else {
+        build_request(client, client.http.post(&url))
+    };
+
+    let resp = req.send().await.map_err(|e| format!("InfluxDB request failed: {e}"))?;
+    log::info!("[influxdb] response status={}", resp.status());
+    if !resp.status().is_success() {
+        let error_json = resp.json::<InfluxErrorResult>().await.unwrap_or_default();
+        let msg = error_json.error;
+        log::error!("[influxdb] error: {msg}");
+        return Err(format!("InfluxDB error: {msg}"));
+    }
+    resp.json::<InfluxJsonResult>().await.map_err(|e| format!("InfluxDB parse error: {e}"))
+}
+
+pub async fn test_connection(client: &InfluxdbClient, timeout: Duration) -> Result<(), String> {
+    let url = format!("{}/query?q=SHOW DATABASES", client.base_url);
+    let req = build_request(client, client.http.get(&url));
+    let resp = with_connection_timeout("InfluxDB", timeout, async {
+        req.send().await.map_err(|e| format!("InfluxDB connection failed: {e}"))
+    })
+    .await?;
+    if !resp.status().is_success() {
+        let body = resp.text().await.unwrap_or_default();
+        return Err(format!("InfluxDB error: {body}"));
+    }
+    Ok(())
+}
+
+pub async fn list_databases(client: &InfluxdbClient) -> Result<Vec<DatabaseInfo>, String> {
+    let result = influx_query(client, "SHOW DATABASES", None).await?;
+    Ok(result
+        .results
+        .iter()
+        .flat_map(|r| &r.series)
+        .flat_map(|s| &s.values)
+        .map(|row| DatabaseInfo { name: row[0].as_str().unwrap_or("").to_string() })
+        .collect())
+}
+
+pub async fn list_tables(client: &InfluxdbClient, database: &str) -> Result<Vec<TableInfo>, String> {
+    let result = influx_query(client, "SHOW MEASUREMENTS", Some(database)).await?;
+    let empty = vec![];
+    let series = result.results.first().map(|r| &r.series).unwrap_or(&empty);
+    if series.is_empty() {
+        return Ok(vec![]);
+    }
+    let first_series = &series[0];
+    Ok(first_series
+        .values
+        .iter()
+        .map(|row| TableInfo {
+            name: row[0].as_str().unwrap_or("").to_string(),
+            table_type: "TABLE".to_string(),
+            comment: None,
+            parent_schema: None,
+            parent_name: None,
+        })
+        .collect())
+}
+
+pub async fn get_columns(client: &InfluxdbClient, database: &str, table: &str) -> Result<Vec<ColumnInfo>, String> {
+    let empty = vec![];
+
+    let tag_sql = format!("SHOW TAG KEYS FROM \"{}\"", table);
+    let tag_result = influx_query(client, &tag_sql, Some(database)).await?;
+    let tag_series = tag_result.results.first().map(|r| &r.series).unwrap_or(&empty);
+
+    let field_sql = format!("SHOW FIELD KEYS FROM \"{}\"", table);
+    let field_result = influx_query(client, &field_sql, Some(database)).await?;
+    let field_series = field_result.results.first().map(|r| &r.series).unwrap_or(&empty);
+
+    let time_col = ColumnInfo {
+        name: "time".to_string(),
+        data_type: "timestamp".to_string(),
+        is_nullable: false,
+        column_default: None,
+        is_primary_key: true,
+        extra: None,
+        comment: None,
+        numeric_precision: None,
+        numeric_scale: None,
+        character_maximum_length: None,
+    };
+
+    let cols: Vec<ColumnInfo> = std::iter::once(time_col)
+        .chain(tag_series.first().into_iter().flat_map(|s| s.values.iter()).map(|row| ColumnInfo {
+            name: row[0].as_str().unwrap_or("").to_string(),
+            data_type: "string".to_string(),
+            is_nullable: true,
+            column_default: None,
+            is_primary_key: true,
+            extra: None,
+            comment: None,
+            numeric_precision: None,
+            numeric_scale: None,
+            character_maximum_length: None,
+        }))
+        .chain(field_series.first().into_iter().flat_map(|s| s.values.iter()).map(|row| {
+            let data_type = row.get(1).and_then(|v| v.as_str()).unwrap_or("unknown").to_string();
+            ColumnInfo {
+                name: row[0].as_str().unwrap_or("").to_string(),
+                data_type,
+                is_nullable: true,
+                column_default: None,
+                is_primary_key: false,
+                extra: None,
+                comment: None,
+                numeric_precision: None,
+                numeric_scale: None,
+                character_maximum_length: None,
+            }
+        }))
+        .collect();
+
+    Ok(cols)
+}
+
+pub async fn execute_query(client: &InfluxdbClient, database: &str, sql: &str) -> Result<QueryResult, String> {
+    let start = Instant::now();
+    let url = build_query_url(&client.base_url, Some(database), sql);
+    let req = if starts_with_executable_sql_keyword(sql, &["SELECT", "SHOW"]) {
+        build_request(client, client.http.get(&url))
+    } else {
+        build_request(client, client.http.post(&url))
+    };
+    let resp = req.send().await.map_err(|e| format!("InfluxDB request failed: {e}"))?;
+    if !resp.status().is_success() {
+        let error_json = resp.json::<InfluxErrorResult>().await.unwrap_or_default();
+        let msg = error_json.error;
+        return Err(format!("InfluxDB error: {msg}"));
+    }
+    let json = resp.json::<InfluxJsonResult>().await.map_err(|e| format!("InfluxDB parse error: {e}"))?;
+    let series = json.results.iter().flat_map(|r| &r.series).next();
+    match series {
+        Some(s) => Ok(QueryResult {
+            columns: s.columns.clone(),
+            column_types: vec![],
+            column_sortables: s.columns.iter().map(|_| false).collect(),
+            rows: s.values.clone(),
+            affected_rows: s.values.len() as u64,
+            execution_time_ms: start.elapsed().as_millis(),
+            truncated: false,
+            session_id: None,
+            has_more: false,
+        }),
+        None => Ok(QueryResult {
+            columns: vec![],
+            column_types: vec![],
+            column_sortables: vec![],
+            rows: vec![],
+            affected_rows: 0,
+            execution_time_ms: start.elapsed().as_millis(),
+            truncated: false,
+            session_id: None,
+            has_more: false,
+        }),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn query_url() {
+        let url = build_query_url("http://localhost:8086", Some("sample"), "SHOW DATABASES");
+
+        assert_eq!(url, "http://localhost:8086/query?db=sample&q=SHOW%20DATABASES");
+    }
+}

crates/dbx-core/src/db/mod.rs

@@ -4,6 +4,7 @@ pub mod duckdb_driver;
 pub mod elasticsearch_driver;
 pub mod elasticsearch_sql;
 pub mod file_validator;
+pub mod influxdb_driver;
 pub mod mongo_driver;
 pub mod mysql;
 pub mod ob_oracle;

crates/dbx-core/src/db_admin_sql.rs

@@ -165,6 +165,8 @@ pub fn build_drop_table_sql(options: TableAdminSqlOptions) -> String {
     let table = qualified_name(options.database_type, options.schema.as_deref(), &options.table_name);
     if matches!(options.database_type, Some(DatabaseType::Iotdb)) {
         return format!("DELETE TIMESERIES {};", iotdb_timeseries_pattern(&table));
+    } else if matches!(options.database_type, Some(DatabaseType::InfluxDb)) {
+        return format!("DROP MEASUREMENT {};", table);
     }
     format!("DROP TABLE {table};")
 }

crates/dbx-core/src/models/connection.rs

@@ -274,6 +274,8 @@ pub enum DatabaseType {
     Iris,
     #[serde(rename = "turso")]
     Turso,
+    #[serde(rename = "influxdb")]
+    InfluxDb,
     Jdbc,
 }
 
@@ -728,6 +730,10 @@ impl ConnectionConfig {
                 format!("etcd://{host}:{port}")
             }
             DatabaseType::Iris => format!("iris://{host}:{port}{db_part}"),
+            DatabaseType::InfluxDb => {
+                let scheme = if self.ssl { "https" } else { "http" };
+                format!("{scheme}://{host}:{port}")
+            }
             DatabaseType::Jdbc => "jdbc:<redacted>".to_string(),
         }
     }
@@ -918,6 +924,10 @@ impl ConnectionConfig {
             DatabaseType::Iris => {
                 format!("iris://{}:{}@{host}:{port}{db_part}", username, password)
             }
+            DatabaseType::InfluxDb => {
+                let scheme = if self.ssl { "https" } else { "http" };
+                format!("{scheme}://{host}:{port}")
+            }
             DatabaseType::Jdbc => {
                 self.connection_string.as_deref().filter(|value| !value.is_empty()).unwrap_or("jdbc:").to_string()
             }

crates/dbx-core/src/query.rs

@@ -749,6 +749,15 @@ pub async fn do_execute(
         }
         PoolKind::Redis(_) => Err("Use Redis-specific commands".to_string()),
         PoolKind::MongoDb(_) => Err("Use MongoDB-specific commands".to_string()),
+        PoolKind::InfluxDb(client) => {
+            let client = client.clone();
+            let database = pool_key.split(':').nth(1).unwrap_or("default").to_string();
+            let max_rows = options.max_rows;
+            drop(connections);
+            wait_for_query_opt(cancel_token, query_timeout, db::influxdb_driver::execute_query(&client, &database, sql))
+                .await
+                .map(|result| truncate_result_with_max_rows(result, max_rows))
+        }
         PoolKind::Agent(client) => {
             let client = client.clone();
             let sql = sql.to_string();
@@ -1297,6 +1306,7 @@ pub async fn execute_statements_in_transaction(
             | PoolKind::Redis(_)
             | PoolKind::MongoDb(_)
             | PoolKind::Elasticsearch(_)
+            | PoolKind::InfluxDb(_)
             | PoolKind::ExternalTabular(_)
             | PoolKind::ExternalDriver { .. } => TxPath::None,
         })

crates/dbx-core/src/schema.rs

@@ -275,6 +275,10 @@ async fn list_databases_once(state: &AppState, connection_id: &str) -> Result<Ve
             drop(connections);
             return db::clickhouse_driver::list_databases(&client).await;
         }
+        if let Some(client) = extract_pool!(&connections, connection_id, InfluxDb) {
+            drop(connections);
+            return db::influxdb_driver::list_databases(&client).await;
+        }
         try_sqlserver!(connections, connection_id, list_databases);
         if let Some(client) = extract_pool!(&connections, connection_id, Agent) {
             let is_mongo =
@@ -448,6 +452,10 @@ async fn list_tables_once(
             drop(connections);
             return db::clickhouse_driver::list_tables(&client, clickhouse_metadata_database(database, schema)).await;
         }
+        if let Some(client) = extract_pool!(&connections, &pool_key, InfluxDb) {
+            drop(connections);
+            return db::influxdb_driver::list_tables(&client, database).await;
+        }
         try_sqlserver!(connections, &pool_key, list_tables, schema, filter, limit);
         if let Some(client) = extract_pool!(&connections, &pool_key, Agent) {
             let fallback_config = db_config.clone();
@@ -1236,6 +1244,10 @@ pub async fn get_columns_core(
                 .await
                 .map(deduplicate_column_infos);
         }
+        if let Some(client) = extract_pool!(&connections, &pool_key, InfluxDb) {
+            drop(connections);
+            return db::influxdb_driver::get_columns(&client, database, table).await.map(deduplicate_column_infos);
+        }
         try_sqlserver!(connections, &pool_key, get_columns, schema, table);
         if let Some(client) = extract_pool!(&connections, &pool_key, Agent) {
             let fallback_config = db_config.clone();

crates/dbx-core/src/sql_dialect.rs

@@ -58,7 +58,10 @@ pub fn build_table_data_select_sql(options: TableDataSelectSqlOptions) -> String
     let row_id_alias =
         if options.include_row_id && database_type == Some(DatabaseType::Oracle) { Some("t") } else { None };
     let default_order_alias = if database_type == Some(DatabaseType::Jdbc) { Some("dbx_t") } else { row_id_alias };
-    let default_order_by = if !options.primary_keys.is_empty() {
+    let default_order_by = if database_type == Some(DatabaseType::InfluxDb) {
+        // InfluxQL only allows sorting of timestamp column
+        Some("time DESC".to_string())
+    } else if !options.primary_keys.is_empty() {
         Some(
             options
                 .primary_keys

crates/dbx-core/src/transfer.rs

@@ -1822,6 +1822,13 @@ pub async fn get_columns_for_transfer(
         let mut client = client.lock().await;
         return db::sqlserver::get_columns(&mut client, &schema, &table).await;
     }
+    if let Some(PoolKind::InfluxDb(client)) = connections.get(pool_key) {
+        let client = client.clone();
+        let database = database.to_string();
+        let table = table.to_string();
+        drop(connections);
+        return db::influxdb_driver::get_columns(&client, &database, &table).await;
+    }
     if let Some(PoolKind::Agent(client)) = connections.get(pool_key) {
         let client = client.clone();
         let database = database.to_string();