diff --git a/apps/desktop/src/components/connection/ConnectionDialog.vue b/apps/desktop/src/components/connection/ConnectionDialog.vue index 8b03a1c930..d31679b450 100644 --- a/apps/desktop/src/components/connection/ConnectionDialog.vue +++ b/apps/desktop/src/components/connection/ConnectionDialog.vue @@ -1409,22 +1409,26 @@ function buildMqttExternalConfig(): MqttConnectionConfig { }; } +const INFLUXDB_V1V2_DEFAULT_PORT = 8086; +const INFLUXDB_V3_DEFAULT_PORT = 8181; + const influxDbVersion = ref("1"); const influxDbOrg = ref(""); const victoriaMetricsApiPath = ref("/prometheus"); const victoriaMetricsLookback = ref("1h"); -function resetInfluxDbFields(config?: Partial) { - influxDbVersion.value = config?.version === "2" ? "2" : "1"; +function resetInfluxDbFields(config?: Partial, versionHint?: InfluxDbVersion) { + const version = versionHint ?? (config?.version === "2" ? "2" : config?.version === "3" ? "3" : "1"); + influxDbVersion.value = version; influxDbOrg.value = config?.org?.trim() || ""; } -function hydrateInfluxDbFields(value: unknown) { +function hydrateInfluxDbFields(value: unknown, versionHint?: InfluxDbVersion) { if (!value || typeof value !== "object") { - resetInfluxDbFields(); + resetInfluxDbFields(undefined, versionHint); return; } - resetInfluxDbFields(value as Partial); + resetInfluxDbFields(value as Partial, versionHint); } function resetHiveKerberosFields(config?: Pick) { @@ -1442,6 +1446,9 @@ function resetDamengJvmOptions(config?: Pick { +watch(influxDbVersion, (version, previousVersion) => { if (form.value.db_type !== "influxdb") return; - if (version === "2") { + if (version === "2" || version === "3") { form.value.username = ""; } + const port = form.value.port; + if (version === "3" && (!port || port === INFLUXDB_V1V2_DEFAULT_PORT)) { + form.value.port = INFLUXDB_V3_DEFAULT_PORT; + } else if (previousVersion === "3" && port === INFLUXDB_V3_DEFAULT_PORT) { + form.value.port = INFLUXDB_V1V2_DEFAULT_PORT; + } }); function requireMqField(value: string, message: string): string { @@ -2556,8 +2569,15 @@ watch( } else { resetMqttFields(); } - if (config.db_type === "influxdb") { - hydrateInfluxDbFields(config.external_config); + if (config.db_type === "influxdb" || config.db_type === "influxdb3") { + // The influxdb3 engine is presented as the InfluxDB card with + // version = 3. Save-side (`applyConnectionFormToConfig`) swaps + // db_type back to `influxdb3` when saving. + const versionHint: InfluxDbVersion | undefined = config.db_type === "influxdb3" ? "3" : undefined; + hydrateInfluxDbFields(config.external_config, versionHint); + if (config.db_type === "influxdb3") { + form.value.db_type = "influxdb"; + } } else { resetInfluxDbFields(); } @@ -2831,7 +2851,9 @@ function jdbcProductCategory(profileId: string): DbCategoryKey { return category; } -const dbOptions: DbOption[] = [...CONNECTION_PICKER_OPTIONS, ...jdbcProductPickerOptions().map((option) => ({ ...option, category: jdbcProductCategory(option.value) }))]; +// `influxdb3` is presented as a version option inside the InfluxDB card +// (see the version @@ -7247,6 +7276,16 @@ function openExternalUrl(url: string) { +
- +
diff --git a/apps/desktop/src/components/icons/DatabaseIcon.vue b/apps/desktop/src/components/icons/DatabaseIcon.vue index 90d8dc3daf..729e259e0f 100644 --- a/apps/desktop/src/components/icons/DatabaseIcon.vue +++ b/apps/desktop/src/components/icons/DatabaseIcon.vue @@ -113,6 +113,7 @@ const assetIcons: Record = { consul: "consul", iris: "iris", influxdb: "influxdb", + influxdb3: "influxdb", victoriametrics: "victoriametrics.png", zookeeper: "zookeeper", oscar: "oscar.png", diff --git a/apps/desktop/src/lib/connection/connectionPresentation.ts b/apps/desktop/src/lib/connection/connectionPresentation.ts index b1a0badcbb..d3b10d4f09 100644 --- a/apps/desktop/src/lib/connection/connectionPresentation.ts +++ b/apps/desktop/src/lib/connection/connectionPresentation.ts @@ -263,6 +263,9 @@ export function connectionUrlPlaceholder(dbType: DatabaseType): string { case "influxdb": return "influxdb://user:password@host:port/database"; + case "influxdb3": + return "influxdb3://token@host:8181/database"; + case "victoriametrics": return "http://user:password@host:port/prometheus"; diff --git a/apps/desktop/src/lib/database/databaseFeatureSupport.ts b/apps/desktop/src/lib/database/databaseFeatureSupport.ts index 5218ce16b8..0a9ca61029 100644 --- a/apps/desktop/src/lib/database/databaseFeatureSupport.ts +++ b/apps/desktop/src/lib/database/databaseFeatureSupport.ts @@ -229,7 +229,7 @@ export function supportsObjectBrowserTreeNode(dbType: DatabaseType | undefined, } export function supportsTableTruncate(dbType?: DatabaseType): boolean { - return !!dbType && dbType !== "impala" && dbType !== "sqlite" && dbType !== "rqlite" && dbType !== "turso" && dbType !== "cloudflare-d1" && dbType !== "duckdb" && dbType !== "influxdb" && dbType !== "victoriametrics" && dbType !== "manticoresearch"; + return !!dbType && dbType !== "impala" && dbType !== "sqlite" && dbType !== "rqlite" && dbType !== "turso" && dbType !== "cloudflare-d1" && dbType !== "duckdb" && dbType !== "influxdb" && dbType !== "influxdb3" && dbType !== "victoriametrics" && dbType !== "manticoresearch"; } export function supportsTableVacuum(dbType?: DatabaseType): boolean { diff --git a/apps/desktop/src/lib/database/databaseNamespaceCreation.ts b/apps/desktop/src/lib/database/databaseNamespaceCreation.ts index bfc5f3acad..8b68f5a33a 100644 --- a/apps/desktop/src/lib/database/databaseNamespaceCreation.ts +++ b/apps/desktop/src/lib/database/databaseNamespaceCreation.ts @@ -90,6 +90,7 @@ export const DATABASE_NAMESPACE_CREATION_MATRIX = { zookeeper: { deferred: "key-value namespaces are not databases" }, iris: { database: "schema" }, influxdb: { connection: "database" }, + influxdb3: { deferred: "database creation on InfluxDB 3.x goes through the /api/v3/configure/database admin endpoint, which is not part of the SQL execution path" }, victoriametrics: { deferred: "metric namespaces are managed by VictoriaMetrics deployment configuration" }, jdbc: { deferred: "generic JDBC does not expose a reliable dialect-specific create target" }, mq: { deferred: "message queue namespaces are handled by MQ admin panels" }, diff --git a/apps/desktop/src/lib/database/databaseObjectCapabilities.ts b/apps/desktop/src/lib/database/databaseObjectCapabilities.ts index 50f9e262ff..850e38c4ed 100644 --- a/apps/desktop/src/lib/database/databaseObjectCapabilities.ts +++ b/apps/desktop/src/lib/database/databaseObjectCapabilities.ts @@ -92,6 +92,7 @@ const DATABASE_TYPE_OBJECTS = new Map([ ["neo4j", TABLE_VIEW_OBJECTS], // others ["influxdb", ["TABLE"]], + ["influxdb3", ["TABLE"]], ["victoriametrics", ["TABLE"]], ["hbase", ["TABLE"]], ["questdb", ["TABLE", "VIEW", "MATERIALIZED_VIEW"]], diff --git a/apps/desktop/src/lib/database/databasePropertyEditing.ts b/apps/desktop/src/lib/database/databasePropertyEditing.ts index 60b9b749fd..631ffa7197 100644 --- a/apps/desktop/src/lib/database/databasePropertyEditing.ts +++ b/apps/desktop/src/lib/database/databasePropertyEditing.ts @@ -90,6 +90,7 @@ export const DATABASE_PROPERTY_EDITING_MATRIX = { zookeeper: { deferred: "key-value namespaces are not databases" }, iris: { deferred: "schema properties need product-specific handling" }, influxdb: { deferred: "database retention policies need a dedicated workflow" }, + influxdb3: { deferred: "database retention policies need a dedicated workflow" }, victoriametrics: { deferred: "metric and retention settings are managed by VictoriaMetrics deployment configuration" }, jdbc: { deferred: "generic JDBC does not expose reliable dialect-specific properties" }, mq: { deferred: "message queue namespaces are handled by MQ admin panels" }, diff --git a/apps/desktop/src/lib/table/tableMetadataCapabilities.ts b/apps/desktop/src/lib/table/tableMetadataCapabilities.ts index be1da7af8a..0b1733edc2 100644 --- a/apps/desktop/src/lib/table/tableMetadataCapabilities.ts +++ b/apps/desktop/src/lib/table/tableMetadataCapabilities.ts @@ -102,6 +102,12 @@ const capabilityByType: Partial), ExternalDriver { @@ -143,6 +144,7 @@ impl PoolKind { Self::HBase(client) => Some(Self::HBase(client.clone())), Self::VectorDb(client) => Some(Self::VectorDb(client.clone())), Self::InfluxDb(client) => Some(Self::InfluxDb(client.clone())), + Self::InfluxDb3(client) => Some(Self::InfluxDb3(client.clone())), Self::VictoriaMetrics(client) => Some(Self::VictoriaMetrics(client.clone())), Self::Agent(client) => Some(Self::Agent(client.clone())), Self::ExternalDriver { driver_id, config, session } => Some(Self::ExternalDriver { @@ -2403,6 +2405,11 @@ impl AppState { db::influxdb_driver::test_connection(&client, connect_timeout).await?; PoolKind::InfluxDb(client) } + DatabaseType::InfluxDb3 => { + let client = db::influxdb3_driver::Influxdb3Client::new_for_config(&url, &db_config, connect_timeout)?; + db::influxdb3_driver::test_connection(&client, connect_timeout).await?; + PoolKind::InfluxDb3(client) + } DatabaseType::VictoriaMetrics => { let client = db::victoriametrics_driver::VictoriaMetricsClient::new_for_config( &url, @@ -3613,6 +3620,18 @@ impl AppState { } } } + PoolKind::InfluxDb3(client) => { + let client = client.clone(); + drop(connections); + let timeout = crate::db::connection_timeout(); + match db::influxdb3_driver::test_connection(&client, timeout).await { + Ok(()) => false, + Err(err) => { + log::warn!("InfluxDB 3 connection pool '{pool_key}' is stale: {err}"); + true + } + } + } PoolKind::VictoriaMetrics(client) => { let client = client.clone(); drop(connections); @@ -4689,6 +4708,13 @@ impl AppState { false } }, + PoolKind::InfluxDb3(client) => match db::influxdb3_driver::test_connection(client, timeout).await { + Ok(()) => true, + Err(e) => { + log::warn!("InfluxDB 3 connection pool '{key}' is unhealthy: {e}"); + false + } + }, PoolKind::VictoriaMetrics(client) => { match db::victoriametrics_driver::test_connection(client, timeout).await { Ok(()) => true, @@ -4989,6 +5015,7 @@ enum KeepaliveTarget { HBase(db::hbase_driver::HBaseClient), VectorDb(db::vector_driver::VectorClient), InfluxDb(db::influxdb_driver::InfluxdbClient), + InfluxDb3(db::influxdb3_driver::Influxdb3Client), VictoriaMetrics(db::victoriametrics_driver::VictoriaMetricsClient), Agent(Arc), #[cfg(feature = "mq-admin")] @@ -5089,6 +5116,7 @@ fn keepalive_target_from_pool(pool: &PoolKind, config: &ConnectionConfig) -> Opt PoolKind::HBase(client) => Some(KeepaliveTarget::HBase(client.clone())), PoolKind::VectorDb(client) => Some(KeepaliveTarget::VectorDb(client.clone())), PoolKind::InfluxDb(client) => Some(KeepaliveTarget::InfluxDb(client.clone())), + PoolKind::InfluxDb3(client) => Some(KeepaliveTarget::InfluxDb3(client.clone())), PoolKind::VictoriaMetrics(client) => Some(KeepaliveTarget::VictoriaMetrics(client.clone())), PoolKind::Agent(client) => Some(KeepaliveTarget::Agent(client.clone())), _ => None, @@ -5136,6 +5164,9 @@ async fn ping_keepalive_target(target: &mut KeepaliveTarget, timeout: Duration) KeepaliveTarget::InfluxDb(client) => { db::influxdb_driver::test_connection(client, timeout).await.map_err(Into::into) } + KeepaliveTarget::InfluxDb3(client) => { + db::influxdb3_driver::test_connection(client, timeout).await.map_err(Into::into) + } KeepaliveTarget::VictoriaMetrics(client) => { db::victoriametrics_driver::test_connection(client, timeout).await.map_err(Into::into) } @@ -5494,6 +5525,7 @@ fn clone_pool_kind(pool: &PoolKind) -> PoolKind { PoolKind::HBase(client) => PoolKind::HBase(client.clone()), PoolKind::VectorDb(client) => PoolKind::VectorDb(client.clone()), PoolKind::InfluxDb(client) => PoolKind::InfluxDb(client.clone()), + PoolKind::InfluxDb3(client) => PoolKind::InfluxDb3(client.clone()), PoolKind::VictoriaMetrics(client) => PoolKind::VictoriaMetrics(client.clone()), PoolKind::Agent(client) => PoolKind::Agent(client.clone()), PoolKind::ExternalDriver { driver_id, config, session } => { @@ -5571,6 +5603,9 @@ async fn close_pool_kind(pool: PoolKind) -> Result<(), String> { PoolKind::InfluxDb(client) => { drop(client); } + PoolKind::InfluxDb3(client) => { + drop(client); + } PoolKind::VictoriaMetrics(client) => { drop(client); } diff --git a/crates/dbx-core/src/db/influxdb3_driver.rs b/crates/dbx-core/src/db/influxdb3_driver.rs new file mode 100644 index 0000000000..1ddeb3c5b0 --- /dev/null +++ b/crates/dbx-core/src/db/influxdb3_driver.rs @@ -0,0 +1,510 @@ +//! InfluxDB 3.x native driver. +//! +//! Transport is HTTP JSON — `POST /api/v3/query_sql` for queries, +//! `GET /api/v3/configure/database?format=json` for the database list. +//! This is intentionally the same wire path Core's own tooling uses, +//! and every metadata / query / test call funnels through +//! [`post_query_json`] so the connection test proves the exact code +//! path a real query will take. +//! +//! Row bounding is implemented by rewriting eligible `SELECT` statements +//! with `LIMIT N + 1` when [`execute_query`] is given a `max_rows` cap. +//! That keeps the server from serializing a full table into JSON when +//! the caller only ever wanted the first few thousand rows. +//! +//! Database creation is deliberately **not** implemented here — Core's +//! `POST /api/v3/configure/database` admin endpoint exists but is not +//! part of the generic SQL execution path, and the driver manifest +//! reflects this by advertising `databaseCreate: false`. + +use regex::Regex; +use reqwest::{Certificate, Client as HttpClient, Response, StatusCode}; +use serde::{Deserialize, Serialize}; +use std::fs; +use std::sync::OnceLock; +use std::time::{Duration, Instant}; + +use super::with_connection_timeout; +use crate::models::connection::ConnectionConfig; +use crate::types::{ColumnInfo, DatabaseInfo, QueryResult, TableInfo}; + +/// InfluxDB 3 Core's internal database — present on every install. We +/// use it as the bootstrap target for `test_connection` when the user +/// hasn't set a default database, so the test still exercises the real +/// query endpoint. +const BOOTSTRAP_DATABASE: &str = "_internal"; + +/// Only user tables live in the `iox` schema. `system` and +/// `information_schema` are engine internals and get filtered out of +/// the sidebar. +const USER_SCHEMA: &str = "iox"; + +#[derive(Clone)] +pub struct Influxdb3Client { + http: HttpClient, + /// Normalized `http[s]://host:port` — no trailing slash. + base_url: String, + /// Bearer token from `config.password`. `None` when the server runs + /// with `--without-auth`. + token: Option, + /// Optional custom query-string appended to `/api/v3/query_sql`. + url_params: Option, + /// Default target database from `config.database`. Used when no + /// caller-supplied database is given (list_databases, + /// test_connection). + default_database: Option, +} + +impl Influxdb3Client { + pub fn new_for_config(url: &str, config: &ConnectionConfig, timeout: Duration) -> Result { + let http = build_http_client(Some(&config.ca_cert_path), timeout)?; + let default_database = + config.database.as_deref().map(str::trim).filter(|value| !value.is_empty()).map(str::to_string); + let token = (!config.password.is_empty()).then_some(config.password.clone()); + Ok(Self { + http, + base_url: url.trim_end_matches('/').to_string(), + token, + url_params: config.url_params.clone(), + default_database, + }) + } +} + +fn build_http_client(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 3 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 3 CA certificate at {path}: {e}"))?; + builder = builder.add_root_certificate(cert); + } + builder.build().map_err(|e| format!("Failed to configure InfluxDB 3 HTTP client: {e}")) +} + +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() +} + +fn auth(client: &Influxdb3Client, req: reqwest::RequestBuilder) -> reqwest::RequestBuilder { + match client.token.as_deref().map(str::trim).filter(|token| !token.is_empty()) { + Some(token) => req.header("Authorization", format!("Bearer {token}")), + None => req, + } +} + +fn query_url(client: &Influxdb3Client) -> String { + match client.url_params.as_deref().map(str::trim).filter(|value| !value.is_empty()) { + Some(params) => format!("{}/api/v3/query_sql?{params}", client.base_url), + None => format!("{}/api/v3/query_sql", client.base_url), + } +} + +#[derive(Serialize)] +struct QueryBody<'a> { + db: &'a str, + q: &'a str, + format: &'static str, +} + +#[derive(Deserialize, Default)] +struct ErrorBody { + #[serde(default)] + error: Option, + #[serde(default)] + message: Option, +} + +/// The single HTTP round-trip every code path (test_connection, +/// list_databases, list_tables, get_columns, execute_query) ends up +/// executing. Keeping this as the sole "does the wire actually work" +/// primitive is why test_connection can prove the real query path. +async fn post_query_json( + client: &Influxdb3Client, + database: &str, + sql: &str, +) -> Result, String> { + let url = query_url(client); + let body = QueryBody { db: database, q: sql, format: "json" }; + let resp = auth(client, client.http.post(&url).json(&body)) + .send() + .await + .map_err(|e| format!("InfluxDB 3 request failed: {e}"))?; + if !resp.status().is_success() { + return Err(error_message(resp).await); + } + let text = resp.text().await.unwrap_or_default(); + if text.trim().is_empty() { + return Ok(Vec::new()); + } + serde_json::from_str::>(&text) + .map_err(|e| format!("InfluxDB 3 JSON parse error: {e}; response: {text}")) +} + +async fn error_message(resp: Response) -> String { + let status = resp.status(); + let hint = match status { + StatusCode::UNAUTHORIZED => Some("Unauthorized — check the token."), + StatusCode::FORBIDDEN => Some("Access denied."), + StatusCode::NOT_FOUND => Some("Endpoint not found — verify the InfluxDB 3 base URL."), + _ => None, + }; + let body = resp.text().await.unwrap_or_default(); + let extracted = serde_json::from_str::(&body).ok().and_then(|value| { + value.error.filter(|msg| !msg.trim().is_empty()).or_else(|| value.message.filter(|msg| !msg.trim().is_empty())) + }); + let message = extracted.or_else(|| hint.map(str::to_string)).unwrap_or_else(|| { + if body.trim().is_empty() { + format!("Unknown error (HTTP {})", status.as_str()) + } else { + body + } + }); + format!("InfluxDB 3 error: status = {}, message = {message}", status.as_str()) +} + +// ----------------------------------------------------------------------------- +// Public API +// ----------------------------------------------------------------------------- + +/// Validate the connection by running the same `POST /api/v3/query_sql` +/// call every real query uses — proves the base URL, TLS, auth token, +/// **and** the query endpoint all work end-to-end. With no configured +/// default database the first database visible to the token is probed; +/// `_internal` exists on every Core install but is admin-only on +/// Enterprise, so it is only the last resort. +pub async fn test_connection(client: &Influxdb3Client, timeout: Duration) -> Result<(), String> { + let fallback_database = match client.default_database.as_deref() { + Some(database) => Some(database.to_string()), + None => list_databases(client) + .await + .ok() + .and_then(|databases| databases.first().map(|database| database.name.clone())), + }; + let database = fallback_database.as_deref().unwrap_or(BOOTSTRAP_DATABASE); + with_connection_timeout("InfluxDB 3", timeout, async { + post_query_json(client, database, "SELECT 1").await.map(|_| ()) + }) + .await +} + +pub async fn list_databases(client: &Influxdb3Client) -> Result, String> { + // The admin endpoint is the canonical way to enumerate databases on + // Core. `?format=json` is required — omitting it returns 400. + let url = format!("{}/api/v3/configure/database?format=json", client.base_url); + let resp = + auth(client, client.http.get(&url)).send().await.map_err(|e| format!("InfluxDB 3 request failed: {e}"))?; + if !resp.status().is_success() { + return Err(error_message(resp).await); + } + let text = resp.text().await.unwrap_or_default(); + if text.trim().is_empty() { + return Ok(Vec::new()); + } + let rows: Vec = + serde_json::from_str(&text).map_err(|e| format!("InfluxDB 3 JSON parse error: {e}; response: {text}"))?; + Ok(rows + .into_iter() + .filter_map(|row| { + let obj = row.as_object()?; + obj.get("iox::database") + .or_else(|| obj.get("db_name")) + .or_else(|| obj.get("name")) + .and_then(serde_json::Value::as_str) + .filter(|value| !value.is_empty()) + .map(|name| DatabaseInfo { + name: name.to_string(), + size_bytes: None, + created_at: None, + updated_at: None, + comment: None, + default_charset: None, + default_collation: None, + }) + }) + .collect()) +} + +pub async fn list_tables(client: &Influxdb3Client, database: &str) -> Result, String> { + // `SHOW TABLES` returns rows across every catalog schema. User + // measurements live under `iox`; `system` / `information_schema` + // are engine internals and would just clutter the sidebar. + let rows = post_query_json(client, database, "SHOW TABLES").await?; + Ok(rows + .into_iter() + .filter_map(|row| { + let obj = row.as_object()?; + let schema = obj.get("table_schema").and_then(serde_json::Value::as_str).unwrap_or(""); + if schema != USER_SCHEMA { + return None; + } + let name = + obj.get("table_name").or_else(|| obj.get("name")).and_then(serde_json::Value::as_str)?.to_string(); + Some(TableInfo { + name, + table_type: "TABLE".to_string(), + comment: None, + parent_schema: None, + parent_name: None, + }) + }) + .collect()) +} + +pub async fn get_columns(client: &Influxdb3Client, database: &str, table: &str) -> Result, String> { + let sql = format!( + "SELECT column_name, data_type, is_nullable \ + FROM information_schema.columns \ + WHERE table_name = '{}' AND table_schema = '{}'", + escape_sql_literal(table), + USER_SCHEMA + ); + let rows = post_query_json(client, database, &sql).await?; + Ok(rows + .into_iter() + .filter_map(|row| { + let obj = row.as_object()?; + let name = obj.get("column_name").and_then(serde_json::Value::as_str).unwrap_or("").to_string(); + if name.is_empty() { + return None; + } + let data_type = obj.get("data_type").and_then(serde_json::Value::as_str).unwrap_or("").to_string(); + let is_nullable = obj + .get("is_nullable") + .and_then(serde_json::Value::as_str) + .map(|value| !value.eq_ignore_ascii_case("NO")) + .unwrap_or(true); + Some(ColumnInfo { + name: name.clone(), + data_type, + is_nullable, + is_primary_key: name == "time", + ..Default::default() + }) + }) + .collect()) +} + +pub async fn execute_query( + client: &Influxdb3Client, + database: &str, + sql: &str, + max_rows: Option, +) -> Result { + // When the caller only wants N rows, push `LIMIT N + 1` down into + // the SQL so the server never serializes more than that. The `+1` + // lets the caller detect truncation. Only rewrite pure top-level + // SELECT statements that don't already carry a LIMIT — anything + // else (SHOW, INSERT, CTEs with LIMIT, etc.) runs verbatim. + let start = Instant::now(); + let effective_sql = max_rows.and_then(|limit| inject_limit(sql, limit)).unwrap_or_else(|| sql.to_string()); + let rows = post_query_json(client, database, &effective_sql).await?; + Ok(build_query_result(rows, start)) +} + +fn build_query_result(rows: Vec, start: Instant) -> QueryResult { + // Preserve the first row's key order — /api/v3/query_sql already + // emits columns in a stable order, and the frontend grid renders + // headers from `columns`, so we mirror that here. + let mut columns: Vec = Vec::new(); + let mut column_index = std::collections::BTreeMap::::new(); + for row in &rows { + if let Some(obj) = row.as_object() { + for (key, _) in obj { + if !column_index.contains_key(key) { + column_index.insert(key.clone(), columns.len()); + columns.push(key.clone()); + } + } + } + } + let mut out_rows = Vec::with_capacity(rows.len()); + for row in rows { + let mut cells = vec![serde_json::Value::Null; columns.len()]; + if let Some(obj) = row.as_object() { + for (key, value) in obj { + if let Some(&idx) = column_index.get(key) { + cells[idx] = value.clone(); + } + } + } + out_rows.push(cells); + } + let affected = out_rows.len() as u64; + QueryResult { + column_sortables: columns.iter().map(|_| false).collect(), + spatial_columns: vec![], + spatial_values: vec![], + columns, + column_types: vec![], + rows: out_rows, + affected_rows: affected, + execution_time_ms: start.elapsed().as_millis(), + truncated: false, + session_id: None, + has_more: false, + elasticsearch_raw_body: None, + messages: vec![], + } +} + +fn escape_sql_literal(value: &str) -> String { + value.replace('\'', "''") +} + +fn select_head_re() -> &'static Regex { + static RE: OnceLock = OnceLock::new(); + RE.get_or_init(|| Regex::new(r"(?is)^\s*(--[^\n]*\n|/\*.*?\*/|\s)*select\b").unwrap()) +} + +fn limit_clause_re() -> &'static Regex { + static RE: OnceLock = OnceLock::new(); + // Word-boundary LIMIT that isn't inside a string literal. The + // (?i) makes it case-insensitive; a real SQL parser would be + // safer, but this is only used to *avoid over-applying* the + // rewrite — false positives just leave the SQL unchanged. + RE.get_or_init(|| Regex::new(r"(?i)\blimit\b").unwrap()) +} + +fn inject_limit(sql: &str, max_rows: usize) -> Option { + let trimmed = sql.trim_end_matches(|c: char| c.is_whitespace() || c == ';'); + if !select_head_re().is_match(trimmed) { + return None; + } + if limit_clause_re().is_match(trimmed) { + return None; + } + let cap = max_rows.saturating_add(1); + Some(format!("{trimmed} LIMIT {cap}")) +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + fn client() -> Influxdb3Client { + Influxdb3Client { + http: HttpClient::new(), + base_url: "http://localhost:8181".to_string(), + token: Some("t".to_string()), + url_params: None, + default_database: Some("mydb".to_string()), + } + } + + #[test] + fn config_reads_bearer_token_and_default_database() { + let config: ConnectionConfig = serde_json::from_value(json!({ + "id": "influx3", + "name": "InfluxDB 3", + "db_type": "influxdb3", + "host": "127.0.0.1", + "port": 8181, + "username": "", + "password": "my-token", + "database": "metrics" + })) + .unwrap(); + let c = Influxdb3Client::new_for_config("http://localhost:8181/", &config, Duration::from_secs(1)).unwrap(); + assert_eq!(c.token.as_deref(), Some("my-token")); + assert_eq!(c.default_database.as_deref(), Some("metrics")); + assert_eq!(c.base_url, "http://localhost:8181"); + } + + #[test] + fn empty_password_skips_authorization_header() { + // Servers running with `--without-auth` accept the driver + // when the user leaves the token blank. + let config: ConnectionConfig = serde_json::from_value(json!({ + "id": "influx3", + "name": "InfluxDB 3", + "db_type": "influxdb3", + "host": "127.0.0.1", + "port": 8181, + "username": "", + "password": "" + })) + .unwrap(); + let c = Influxdb3Client::new_for_config("http://localhost:8181", &config, Duration::from_secs(1)).unwrap(); + assert!(c.token.is_none()); + } + + #[test] + fn query_url_appends_url_params() { + let mut c = client(); + c.url_params = Some("trace=1".to_string()); + assert_eq!(query_url(&c), "http://localhost:8181/api/v3/query_sql?trace=1"); + } + + #[test] + fn inject_limit_wraps_bare_select() { + let sql = "SELECT * FROM cpu ORDER BY time DESC"; + assert_eq!(inject_limit(sql, 100).as_deref(), Some("SELECT * FROM cpu ORDER BY time DESC LIMIT 101")); + } + + #[test] + fn inject_limit_respects_existing_limit() { + let sql = "SELECT * FROM cpu LIMIT 50"; + assert!(inject_limit(sql, 100).is_none()); + } + + #[test] + fn inject_limit_ignores_non_select_statements() { + assert!(inject_limit("SHOW TABLES", 100).is_none()); + assert!(inject_limit("INSERT INTO t VALUES (1)", 100).is_none()); + } + + #[test] + fn inject_limit_handles_leading_comments() { + let sql = "-- keep only recent rows\nSELECT * FROM cpu"; + assert_eq!(inject_limit(sql, 5).as_deref(), Some("-- keep only recent rows\nSELECT * FROM cpu LIMIT 6")); + } + + #[test] + fn inject_limit_trims_trailing_semicolon() { + let sql = "SELECT 1;"; + assert_eq!(inject_limit(sql, 10).as_deref(), Some("SELECT 1 LIMIT 11")); + } + + #[test] + fn build_query_result_flattens_json_rows_preserving_first_seen_column_order() { + let rows = vec![ + json!({ "time": "2026-01-01T00:00:00Z", "host": "a", "value": 1 }), + json!({ "time": "2026-01-01T00:01:00Z", "host": "b", "value": 2 }), + ]; + let result = build_query_result(rows, Instant::now()); + assert_eq!(result.columns, vec!["time", "host", "value"]); + assert_eq!(result.rows.len(), 2); + assert_eq!(result.affected_rows, 2); + } + + #[test] + fn escape_sql_literal_doubles_single_quotes() { + assert_eq!(escape_sql_literal("a'b"), "a''b"); + } +} diff --git a/crates/dbx-core/src/db/mod.rs b/crates/dbx-core/src/db/mod.rs index fa78f0e754..9f3b2ca669 100644 --- a/crates/dbx-core/src/db/mod.rs +++ b/crates/dbx-core/src/db/mod.rs @@ -24,6 +24,7 @@ pub mod elasticsearch_sql; pub mod file_validator; pub mod hbase_driver; pub mod http_tunnel; +pub mod influxdb3_driver; pub mod influxdb_driver; pub mod manticoresearch; pub mod meilisearch_driver; diff --git a/crates/dbx-core/src/models/connection.rs b/crates/dbx-core/src/models/connection.rs index fd9fe1ba76..c47ee49267 100644 --- a/crates/dbx-core/src/models/connection.rs +++ b/crates/dbx-core/src/models/connection.rs @@ -1119,7 +1119,7 @@ impl ConnectionConfig { format!("zookeeper://{host}:{port}") } DatabaseType::Iris => format!("iris://{host}:{port}{db_part}"), - DatabaseType::InfluxDb | DatabaseType::VictoriaMetrics => { + DatabaseType::InfluxDb | DatabaseType::InfluxDb3 | DatabaseType::VictoriaMetrics => { let scheme = if self.ssl { "https" } else { "http" }; format!("{scheme}://{host}:{port}") } @@ -1395,7 +1395,7 @@ impl ConnectionConfig { DatabaseType::Iris => { format!("iris://{}:{}@{host}:{port}{db_part}", username, password) } - DatabaseType::InfluxDb | DatabaseType::VictoriaMetrics => { + DatabaseType::InfluxDb | DatabaseType::InfluxDb3 | DatabaseType::VictoriaMetrics => { let scheme = if self.ssl { "https" } else { "http" }; format!("{scheme}://{host}:{port}") } diff --git a/crates/dbx-core/src/query.rs b/crates/dbx-core/src/query.rs index c351e8d958..4d1d960cb7 100644 --- a/crates/dbx-core/src/query.rs +++ b/crates/dbx-core/src/query.rs @@ -1319,6 +1319,7 @@ fn should_discard_pool_after_query_timeout(db_type: Option) -> boo | DatabaseType::Weaviate | DatabaseType::ChromaDb | DatabaseType::InfluxDb + | DatabaseType::InfluxDb3 | DatabaseType::VictoriaMetrics ) } @@ -2085,6 +2086,23 @@ async fn do_execute_typed( } result } + PoolKind::InfluxDb3(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); + let result = wait_for_query_opt( + cancel_token, + query_timeout, + db::influxdb3_driver::execute_query(&client, &database, sql, max_rows), + ) + .await + .map(|result| truncate_result_with_max_rows(result, max_rows)); + if matches!(result.as_ref(), Err(err) if should_discard_pool_after_error(pool_db_type, err)) { + state.remove_pool_by_key(pool_key).await; + } + result + } PoolKind::VictoriaMetrics(client) => { let client = client.clone(); let max_rows = options.max_rows; @@ -3909,6 +3927,7 @@ fn pool_kind_has_transactional_path(pool: &PoolKind) -> bool { | PoolKind::Meilisearch(_) | PoolKind::VectorDb(_) | PoolKind::InfluxDb(_) + | PoolKind::InfluxDb3(_) | PoolKind::VictoriaMetrics(_) | PoolKind::ExternalDriver { .. } => false, #[cfg(feature = "mq-admin")] @@ -4178,6 +4197,7 @@ pub async fn execute_statements_in_transaction_on_pool_typed( | PoolKind::Meilisearch(_) | PoolKind::VectorDb(_) | PoolKind::InfluxDb(_) + | PoolKind::InfluxDb3(_) | PoolKind::VictoriaMetrics(_) | PoolKind::ExternalDriver { .. } => TxPath::None, }) diff --git a/crates/dbx-core/src/query_execution_sql.rs b/crates/dbx-core/src/query_execution_sql.rs index 2fb79dc545..b61cd5da94 100644 --- a/crates/dbx-core/src/query_execution_sql.rs +++ b/crates/dbx-core/src/query_execution_sql.rs @@ -161,6 +161,7 @@ pub fn supports_sql_query(database_type: DatabaseType) -> bool { | DatabaseType::Weaviate | DatabaseType::ChromaDb | DatabaseType::InfluxDb + | DatabaseType::InfluxDb3 | DatabaseType::VictoriaMetrics | DatabaseType::Neo4j | DatabaseType::Etcd diff --git a/crates/dbx-core/src/schema.rs b/crates/dbx-core/src/schema.rs index f3e9e5e73d..5797b993f5 100644 --- a/crates/dbx-core/src/schema.rs +++ b/crates/dbx-core/src/schema.rs @@ -741,6 +741,10 @@ async fn list_databases_once(state: &AppState, connection_id: &str) -> Result DdlDialectProfile { // Non-tabular / not applicable for relational CREATE TABLE Redis | MongoDb | DynamoDb | Elasticsearch | Easysearch | Meilisearch | Qdrant | Milvus | Weaviate - | ChromaDb | Neo4j | Cassandra | Etcd | ZooKeeper | Nacos | Consul | InfluxDb | VictoriaMetrics + | ChromaDb | Neo4j | Cassandra | Etcd | ZooKeeper | Nacos | Consul | InfluxDb | InfluxDb3 | VictoriaMetrics | MessageQueue | Mqtt | Hbase => conservative_ansi(db_type), } } diff --git a/crates/dbx-core/src/sql_dialect/table_select.rs b/crates/dbx-core/src/sql_dialect/table_select.rs index d54d71e6dc..c2343dda20 100644 --- a/crates/dbx-core/src/sql_dialect/table_select.rs +++ b/crates/dbx-core/src/sql_dialect/table_select.rs @@ -238,8 +238,9 @@ pub fn build_table_data_select_sql_with_database( }; let predicate = normalize_where_input(options.where_input.as_deref()); let where_clause = if predicate.is_empty() { String::new() } else { format!(" WHERE ({predicate})") }; - let default_order_by = if database_type == Some(DatabaseType::InfluxDb) { - // InfluxQL only allows sorting of timestamp column + let default_order_by = if matches!(database_type, Some(DatabaseType::InfluxDb) | Some(DatabaseType::InfluxDb3)) { + // InfluxQL only allows sorting of the timestamp column; SQL-mode + // InfluxDB 3 tables also key naturally on `time`. Some("time DESC".to_string()) } else if database_type == Some(DatabaseType::Impala) { // Impala requires ORDER BY when OFFSET is present. Keeping the same diff --git a/crates/dbx-core/src/transfer.rs b/crates/dbx-core/src/transfer.rs index b6372e668b..016fe9b8bc 100644 --- a/crates/dbx-core/src/transfer.rs +++ b/crates/dbx-core/src/transfer.rs @@ -5156,6 +5156,13 @@ pub async fn get_columns_for_transfer( drop(connections); return db::influxdb_driver::get_columns(&client, &database, &table).await; } + if let Some(PoolKind::InfluxDb3(client)) = connections.get(pool_key) { + let client = client.clone(); + let database = database.to_string(); + let table = table.to_string(); + drop(connections); + return db::influxdb3_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/plugins/connection-types/influxdb3.yaml b/plugins/connection-types/influxdb3.yaml new file mode 100644 index 0000000000..c502c03806 --- /dev/null +++ b/plugins/connection-types/influxdb3.yaml @@ -0,0 +1,29 @@ +schemaVersion: 1 +order: 755 +dbType: influxdb3 +rustVariant: InfluxDb3 +label: InfluxDB 3.x +runtimeMode: native +mcpMode: bridge +singleConnectionPool: false +metadataConnectionScoped: false +skipTcpProbe: true +defaultPort: 8181 +supportLevel: connect +capabilities: + queryExecution: true + metadataBrowse: true + objectBrowser: true + objectSource: false + schemaSearch: false + diagram: false + tableDataEdit: false + tableStructureEdit: false + tableImport: false + dataTransfer: false + sqlFileExecution: true + databaseCreate: false + fieldLineage: false + sqlExplain: false + userAdmin: false + driverManagement: false diff --git a/plugins/connection-types/profiles/catalog.yaml b/plugins/connection-types/profiles/catalog.yaml index ff4dbe6d37..4e239a2746 100644 --- a/plugins/connection-types/profiles/catalog.yaml +++ b/plugins/connection-types/profiles/catalog.yaml @@ -700,6 +700,14 @@ profiles: port: 8086 user: "" category: timeseries + - id: influxdb3 + dbType: influxdb3 + label: InfluxDB 3.x + icon: InfluxDB + pickerIcon: influxdb + port: 8181 + user: "" + category: timeseries - id: victoriametrics dbType: victoriametrics label: VictoriaMetrics diff --git a/src-tauri/src/commands/connection.rs b/src-tauri/src/commands/connection.rs index 98089a094a..cf54aa343b 100644 --- a/src-tauri/src/commands/connection.rs +++ b/src-tauri/src/commands/connection.rs @@ -1493,6 +1493,12 @@ async fn test_connection_with_info_inner( .await .map(|_| "Connection successful".to_string()) } + DatabaseType::InfluxDb3 => { + let client = db::influxdb3_driver::Influxdb3Client::new_for_config(&url, &config, connect_timeout)?; + db::influxdb3_driver::test_connection(&client, connect_timeout) + .await + .map(|_| "Connection successful".to_string()) + } DatabaseType::VictoriaMetrics => { let client = db::victoriametrics_driver::VictoriaMetricsClient::new_for_config(&url, &config, connect_timeout)?; @@ -1924,6 +1930,11 @@ pub async fn connect_db( db::influxdb_driver::test_connection(&client, connect_timeout).await?; PoolKind::InfluxDb(client) } + DatabaseType::InfluxDb3 => { + let client = db::influxdb3_driver::Influxdb3Client::new_for_config(&url, &db_config, connect_timeout)?; + db::influxdb3_driver::test_connection(&client, connect_timeout).await?; + PoolKind::InfluxDb3(client) + } DatabaseType::VictoriaMetrics => { let client = db::victoriametrics_driver::VictoriaMetricsClient::new_for_config(&url, &db_config, connect_timeout)?;