diff --git a/apps/desktop/public/icons/database/influxdb.svg b/apps/desktop/public/icons/database/influxdb.svg new file mode 100644 index 000000000..18239a1ef --- /dev/null +++ b/apps/desktop/public/icons/database/influxdb.svg @@ -0,0 +1,10 @@ + + + + + + + + + + \ No newline at end of file diff --git a/apps/desktop/src/components/connection/ConnectionDialog.vue b/apps/desktop/src/components/connection/ConnectionDialog.vue index b2bf1b0f1..30daf2e9f 100644 --- a/apps/desktop/src/components/connection/ConnectionDialog.vue +++ b/apps/desktop/src/components/connection/ConnectionDialog.vue @@ -416,6 +416,7 @@ const driverProfiles: Record< iotdb: { type: "iotdb", port: 6667, user: "root", label: "Apache IoTDB", icon: "iotdb" }, etcd: { type: "etcd", port: 2379, user: "", label: "etcd", icon: "etcd" }, iris: { type: "iris", port: 1972, user: "_SYSTEM", label: "IRIS", icon: "iris" }, + influxdb: { type: "influxdb", port: 8086, user: "", label: "InfluxDB", icon: "InfluxDB" }, custom_mysql: { type: "mysql", port: 3306, @@ -715,6 +716,7 @@ const iconTypeMap: Record = { bigquery: "bigquery", kylin: "kylin", sundb: "sundb", + influxdb: "influxdb", jdbc: "jdbc", custom_mysql: "mysql", custom_postgres: "postgres", @@ -777,6 +779,7 @@ const dbOptions = [ { value: "xugu", label: "虚谷 XuguDB" }, { value: "iotdb", label: "Apache IoTDB" }, { value: "etcd", label: "etcd" }, + { value: "influxdb", label: "InfluxDB" }, { value: "iris", label: "IRIS" }, { value: "jdbc", label: "JDBC" }, { value: "custom_mysql", label: "Custom (MySQL)" }, @@ -824,7 +827,7 @@ const sqliteExtensionPaths = computed({ form.value.url_params = setSqliteExtensionPaths(form.value.url_params, value); }, }); -const tlsCapableDatabaseTypes = new Set(["mysql", "postgres", "redshift", "gaussdb", "kwdb", "opengauss", "redis", "etcd", "clickhouse", "elasticsearch"]); +const tlsCapableDatabaseTypes = new Set(["mysql", "postgres", "redshift", "gaussdb", "kwdb", "opengauss", "redis", "etcd", "clickhouse", "elasticsearch", "influxdb"]); const supportsTlsToggle = computed(() => tlsCapableDatabaseTypes.has(form.value.db_type)); const supportsCaCertificatePath = computed(() => form.value.db_type === "clickhouse"); const bareMysqlProfiles = new Set(["doris", "starrocks", "selectdb", "oceanbase"]); diff --git a/apps/desktop/src/components/icons/DatabaseIcon.vue b/apps/desktop/src/components/icons/DatabaseIcon.vue index bb83db01c..7249727a6 100644 --- a/apps/desktop/src/components/icons/DatabaseIcon.vue +++ b/apps/desktop/src/components/icons/DatabaseIcon.vue @@ -73,6 +73,7 @@ const assetIcons: Record = { iotdb: "iotdb", etcd: "etcd", iris: "iris.png", + influxdb: "influxdb", }; const letterIcons: Record = {}; diff --git a/apps/desktop/src/components/sidebar/TreeItem.vue b/apps/desktop/src/components/sidebar/TreeItem.vue index 05388f25a..f950cbc11 100644 --- a/apps/desktop/src/components/sidebar/TreeItem.vue +++ b/apps/desktop/src/components/sidebar/TreeItem.vue @@ -184,7 +184,11 @@ function getIconInfo(node: TreeNode): { icon: any; colorClass: string } | null { case "view": return { icon: Eye, colorClass: "text-purple-500" }; case "column": - return { icon: Columns3, colorClass: "text-muted-foreground" }; + if ((node.meta as ColumnInfo).is_primary_key) { + return { icon: Columns3, colorClass: "text-orange-400" }; + } else { + return { icon: Columns3, colorClass: "text-muted-foreground" }; + } case "group-columns": return { icon: ListTree, colorClass: "text-green-400" }; case "group-indexes": diff --git a/apps/desktop/src/lib/connectionPresentation.ts b/apps/desktop/src/lib/connectionPresentation.ts index b3a1bf765..6bb3e0149 100644 --- a/apps/desktop/src/lib/connectionPresentation.ts +++ b/apps/desktop/src/lib/connectionPresentation.ts @@ -145,6 +145,9 @@ export function connectionUrlPlaceholder(dbType: DatabaseType): string { case "iris": return "iris://user:password@host:port/namespace"; + case "influxdb": + return "influxdb://user:password@host:port/database"; + case "jdbc": return "jdbc:mysql://host:3306/database"; diff --git a/apps/desktop/src/lib/databaseCapabilitySets.ts b/apps/desktop/src/lib/databaseCapabilitySets.ts index 245487835..15653c334 100644 --- a/apps/desktop/src/lib/databaseCapabilitySets.ts +++ b/apps/desktop/src/lib/databaseCapabilitySets.ts @@ -145,7 +145,7 @@ export const TABLE_STRUCTURE_SUPPORTED_TYPES = new Set([ "access", ]); -export const CREATE_DATABASE_SUPPORTED_TYPES = new Set(["mysql", "postgres", "sqlserver", "clickhouse", "oracle", "gaussdb", "kwdb", "opengauss", "oceanbase-oracle", "doris", "starrocks", "redshift"]); +export const CREATE_DATABASE_SUPPORTED_TYPES = new Set(["mysql", "postgres", "sqlserver", "clickhouse", "oracle", "gaussdb", "kwdb", "opengauss", "oceanbase-oracle", "doris", "starrocks", "redshift", "influxdb"]); export const FIELD_LINEAGE_SUPPORTED_TYPES = new Set(["mysql", "postgres", "sqlite", "rqlite", "turso", "sqlserver", "oracle", "redshift", "dameng", "gaussdb", "kwdb", "opengauss", "oceanbase-oracle"]); diff --git a/apps/desktop/src/lib/databaseFeatureSupport.ts b/apps/desktop/src/lib/databaseFeatureSupport.ts index 23d6233c2..bb19599ec 100644 --- a/apps/desktop/src/lib/databaseFeatureSupport.ts +++ b/apps/desktop/src/lib/databaseFeatureSupport.ts @@ -96,7 +96,7 @@ export function supportsObjectBrowserTreeNode(dbType: DatabaseType | undefined, } export function supportsTableTruncate(dbType?: DatabaseType): boolean { - return !!dbType && dbType !== "sqlite" && dbType !== "rqlite" && dbType !== "turso" && dbType !== "duckdb"; + return !!dbType && dbType !== "sqlite" && dbType !== "rqlite" && dbType !== "turso" && dbType !== "duckdb" && dbType !== "influxdb"; } export function usesPostgresLikeStructureCopy(dbType?: DatabaseType): boolean { diff --git a/apps/desktop/src/lib/databaseTableDataCapabilities.ts b/apps/desktop/src/lib/databaseTableDataCapabilities.ts index e45fd3f1e..3eda9b8ed 100644 --- a/apps/desktop/src/lib/databaseTableDataCapabilities.ts +++ b/apps/desktop/src/lib/databaseTableDataCapabilities.ts @@ -137,6 +137,18 @@ const DATABASE_CAPABILITY_OVERRIDES: Partial = { gaussdb: { dbType: "gaussdb", profile: "gaussdb", label: "GaussDB", port: 5432, user: "gaussdb" }, kwdb: { dbType: "kwdb", profile: "kwdb", label: "KWDB", port: 26257, user: "root" }, opengauss: { dbType: "gaussdb", profile: "opengauss", label: "openGauss", port: 5432, user: "gaussdb" }, + influxdb: { dbType: "influxdb", profile: "influxdb", label: "InfluxDB", port: 8086, user: "" }, }; function normalizeKey(value: unknown) { diff --git a/apps/desktop/src/lib/tableStructureCapabilities.ts b/apps/desktop/src/lib/tableStructureCapabilities.ts index 34199a0af..ce0d792f3 100644 --- a/apps/desktop/src/lib/tableStructureCapabilities.ts +++ b/apps/desktop/src/lib/tableStructureCapabilities.ts @@ -1,6 +1,6 @@ import type { DatabaseType } from "@/types/database"; -export type TableStructureDialect = "mysql" | "postgres" | "sqlite" | "duckdb" | "sqlserver" | "oracle" | "h2" | "clickhouse" | "unsupported"; +export type TableStructureDialect = "mysql" | "postgres" | "sqlite" | "duckdb" | "sqlserver" | "oracle" | "h2" | "clickhouse" | "influxdb" | "unsupported"; export interface TableStructureCapabilities { dialect: TableStructureDialect; @@ -199,6 +199,20 @@ const accessCapabilities = capabilities({ createIndex: true, }); +const influxdbCapabilities = capabilities({ + dialect: "influxdb", + createTable: false, + addColumn: false, + dropColumn: false, + renameColumn: false, + alterExistingColumn: false, + alterType: false, + alterNullability: false, + alterDefault: false, + reorderColumn: false, + comment: false, +}); + const capabilityByType: Partial> = { mysql: mysqlCapabilities, doris: mysqlCapabilities, @@ -225,6 +239,7 @@ const capabilityByType: Partial h2: h2Capabilities, access: accessCapabilities, clickhouse: clickhouseCapabilities, + influxdb: influxdbCapabilities, }; export function getTableStructureCapabilities(dbType?: DatabaseType): TableStructureCapabilities { diff --git a/apps/desktop/src/stores/connectionStore.ts b/apps/desktop/src/stores/connectionStore.ts index b8fcf40d2..82502973f 100644 --- a/apps/desktop/src/stores/connectionStore.ts +++ b/apps/desktop/src/stores/connectionStore.ts @@ -257,6 +257,7 @@ export const useConnectionStore = defineStore("connection", () => { bigquery: "BigQuery", kylin: "Kylin", sundb: "SunDB", + influxdb: "InfluxDB", }; const profile = config.driver_profile || config.db_type; @@ -1201,7 +1202,8 @@ export const useConnectionStore = defineStore("connection", () => { }, ]; - if (node.type === "table") { + const config = getConfig(connectionId); + if (node.type === "table" && config?.db_type !== "influxdb") { children.push( { id: `${parentId}:__indexes`, diff --git a/apps/desktop/src/types/database.ts b/apps/desktop/src/types/database.ts index d72a066bc..4b4b8ac42 100644 --- a/apps/desktop/src/types/database.ts +++ b/apps/desktop/src/types/database.ts @@ -49,6 +49,7 @@ export type DatabaseType = | "iotdb" | "etcd" | "iris" + | "influxdb" | "jdbc"; export interface SqlSnippet { diff --git a/crates/dbx-core/assets/database-drivers.manifest.json b/crates/dbx-core/assets/database-drivers.manifest.json index f3a47e11d..3e7e664a4 100644 --- a/crates/dbx-core/assets/database-drivers.manifest.json +++ b/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", diff --git a/crates/dbx-core/src/connection.rs b/crates/dbx-core/src/connection.rs index 4f22fa657..8037a4dd9 100644 --- a/crates/dbx-core/src/connection.rs +++ b/crates/dbx-core/src/connection.rs @@ -48,6 +48,7 @@ pub enum PoolKind { ClickHouse(db::clickhouse_driver::ChClient), SqlServer(Arc>), Elasticsearch(db::elasticsearch_driver::EsClient), + InfluxDb(db::influxdb_driver::InfluxdbClient), Agent(Arc>), ExternalTabular(Arc), ExternalDriver { driver_id: String, config: Arc, session: Arc }, @@ -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; diff --git a/crates/dbx-core/src/db/influxdb_driver.rs b/crates/dbx-core/src/db/influxdb_driver.rs new file mode 100644 index 000000000..755ee58ef --- /dev/null +++ b/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, + password: Option, +} + +impl InfluxdbClient { + pub fn new(url: &str, username: Option, password: Option, 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, + password: Option, + ca_cert_path: Option<&str>, + timeout: Duration, + ) -> Result { + 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, +} + +#[derive(Deserialize)] +#[allow(dead_code)] +struct InfluxQueryResult { + statement_id: usize, + #[serde(default)] + #[allow(dead_code)] + series: Vec, +} + +#[derive(Deserialize)] +#[allow(dead_code)] +struct InfluxSeries { + #[serde(default)] + #[allow(dead_code)] + name: String, + columns: Vec, + values: Vec>, +} + +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 { + 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::().await.unwrap_or_default(); + let msg = error_json.error; + log::error!("[influxdb] error: {msg}"); + return Err(format!("InfluxDB error: {msg}")); + } + resp.json::().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, 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, 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, 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 = 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 { + 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::().await.unwrap_or_default(); + let msg = error_json.error; + return Err(format!("InfluxDB error: {msg}")); + } + let json = resp.json::().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"); + } +} diff --git a/crates/dbx-core/src/db/mod.rs b/crates/dbx-core/src/db/mod.rs index e39380560..88b982eac 100644 --- a/crates/dbx-core/src/db/mod.rs +++ b/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; diff --git a/crates/dbx-core/src/db_admin_sql.rs b/crates/dbx-core/src/db_admin_sql.rs index 3e5a31290..53bdca888 100644 --- a/crates/dbx-core/src/db_admin_sql.rs +++ b/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};") } diff --git a/crates/dbx-core/src/models/connection.rs b/crates/dbx-core/src/models/connection.rs index e64b562f1..37bd4f726 100644 --- a/crates/dbx-core/src/models/connection.rs +++ b/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:".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() } diff --git a/crates/dbx-core/src/query.rs b/crates/dbx-core/src/query.rs index f9f662789..bed6b9264 100644 --- a/crates/dbx-core/src/query.rs +++ b/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, }) diff --git a/crates/dbx-core/src/schema.rs b/crates/dbx-core/src/schema.rs index b3ce9fafb..d5a670b91 100644 --- a/crates/dbx-core/src/schema.rs +++ b/crates/dbx-core/src/schema.rs @@ -275,6 +275,10 @@ async fn list_databases_once(state: &AppState, connection_id: &str) -> Result 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 diff --git a/crates/dbx-core/src/transfer.rs b/crates/dbx-core/src/transfer.rs index ca6c3ea37..37722708a 100644 --- a/crates/dbx-core/src/transfer.rs +++ b/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(); diff --git a/src-tauri/src/commands/connection.rs b/src-tauri/src/commands/connection.rs index 4467102da..5dbcb4649 100644 --- a/src-tauri/src/commands/connection.rs +++ b/src-tauri/src/commands/connection.rs @@ -426,6 +426,20 @@ pub async fn test_connection(state: State<'_, Arc>, config: Connection .await .map(|_| "Connection successful".to_string()) } + DatabaseType::InfluxDb => { + let username = if config.username.is_empty() { None } else { Some(config.username.clone()) }; + let password = if config.password.is_empty() { None } else { Some(config.password.clone()) }; + let client = db::influxdb_driver::InfluxdbClient::new_with_ca_cert( + &url, + username, + password, + Some(&config.ca_cert_path), + connect_timeout, + )?; + db::influxdb_driver::test_connection(&client, connect_timeout) + .await + .map(|_| "Connection successful".to_string()) + } db_type if database_capabilities::is_agent_type(&db_type) => { test_agent_connection(state.inner(), &config, &host, port).await } @@ -617,6 +631,19 @@ pub async fn connect_db(state: State<'_, Arc>, config: ConnectionConfi db::turso_driver::test_connection(&client, connect_timeout).await?; PoolKind::Turso(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) + } db_type if database_capabilities::is_agent_type(&db_type) => { connect_agent_pool(state.inner(), &db_config, &host, port).await? }