feat(dolt): add native profile, branch database discovery, and SQL completion

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

@@ -0,0 +1,123 @@
+use mysql_async::prelude::Queryable;
+use std::collections::HashSet;
+
+use crate::models::connection::{ConnectionConfig, DatabaseType};
+use crate::types::DatabaseInfo;
+
+use super::mysql::{get_conn_with_timeout, quote_identifier, MySqlPool};
+
+const ENABLE_BRANCH_DATABASES_SQL: &str = "SET @@dolt_show_branch_databases = 1";
+// 与设置对称地复位 session 变量：连接从池中取出后是单条会被复用的 session 连接，
+// 若不在归还前复位，后续无关的 SHOW DATABASES（其他元数据路径复用同一条连接时）
+// 会意外看到 branch 数据库。复位是 best-effort，失败不应让 list_databases 整体失败。
+const DISABLE_BRANCH_DATABASES_SQL: &str = "SET @@dolt_show_branch_databases = 0";
+const SHOW_DATABASES_SQL: &str = "SHOW DATABASES";
+
+pub fn is_config(config: &ConnectionConfig) -> bool {
+    is_profile(&config.db_type, config.driver_profile.as_deref())
+}
+
+fn is_profile(db_type: &DatabaseType, driver_profile: Option<&str>) -> bool {
+    *db_type == DatabaseType::Mysql && driver_profile.is_some_and(|profile| profile.eq_ignore_ascii_case("dolt"))
+}
+
+pub async fn list_databases(pool: &MySqlPool) -> Result<Vec<DatabaseInfo>, String> {
+    let mut conn = get_conn_with_timeout(pool, super::connection_timeout()).await?;
+    if let Err(error) = conn.query_drop(ENABLE_BRANCH_DATABASES_SQL).await {
+        log::debug!("Dolt branch database system variable is unavailable; falling back to dolt_branches: {error}");
+        return list_databases_from_branches(&mut conn).await;
+    }
+    let databases = query_database_names(&mut conn).await?;
+    // 主路径成功读完 SHOW DATABASES 后，归还连接前复位 session 变量，
+    // 避免该连接被复用于无关元数据查询时仍残留 branch 数据库可见性。
+    if let Err(error) = conn.query_drop(DISABLE_BRANCH_DATABASES_SQL).await {
+        log::warn!("Failed to reset Dolt branch database session variable before returning connection: {error}");
+    }
+    Ok(database_infos(databases))
+}
+
+async fn list_databases_from_branches(conn: &mut mysql_async::Conn) -> Result<Vec<DatabaseInfo>, String> {
+    let base_names = query_database_names(conn).await?;
+    let discovery_names: Vec<String> =
+        base_names.iter().filter(|name| is_branch_discovery_database(name)).cloned().collect();
+    let mut seen: HashSet<String> = base_names.iter().cloned().collect();
+    let mut names = base_names;
+
+    for database in discovery_names {
+        let branches: Vec<String> = match conn.query(list_branches_sql(&database)).await {
+            Ok(branches) => branches,
+            Err(error) => {
+                // fallback 进入这里说明实例不支持 dolt_show_branch_databases 系统变量。
+                // 首个非系统库的 dolt_branches 查询报错即可判定该实例并非 Dolt
+                // （可能是同一实例上混用的纯 MySQL 库），逐库试探只会对每个普通库
+                // 产生一次无效报错噪音。首次失败即整体降级为纯 SHOW DATABASES 结果，
+                // 不再继续试探剩余库。注意区分「查询报错（非 Dolt）」与「查询成功但为空
+                // （是 Dolt 但无 branch）」：只有前者触发降级，后者继续遍历后续库。
+                log::debug!(
+                    "Dolt branch discovery failed on `{database}`, treating instance as non-Dolt and falling back to plain SHOW DATABASES: {error}"
+                );
+                return Ok(database_infos(names));
+            }
+        };
+        for branch in branches {
+            let revision = format!("{database}/{}", branch.trim());
+            if !branch.trim().is_empty() && seen.insert(revision.clone()) {
+                names.push(revision);
+            }
+        }
+    }
+
+    Ok(database_infos(names))
+}
+
+async fn query_database_names(conn: &mut mysql_async::Conn) -> Result<Vec<String>, String> {
+    conn.query(SHOW_DATABASES_SQL).await.map_err(|error| error.to_string())
+}
+
+fn database_infos(names: Vec<String>) -> Vec<DatabaseInfo> {
+    names
+        .into_iter()
+        .map(|name| name.trim().to_string())
+        .filter(|name| !name.is_empty())
+        .map(|name| DatabaseInfo { name })
+        .collect()
+}
+
+fn is_branch_discovery_database(name: &str) -> bool {
+    let normalized = name.trim().to_ascii_lowercase();
+    !normalized.is_empty()
+        && !normalized.contains('/')
+        && !matches!(normalized.as_str(), "information_schema" | "mysql" | "performance_schema" | "sys")
+}
+
+fn list_branches_sql(database: &str) -> String {
+    format!("SELECT name FROM {}.dolt_branches ORDER BY name", quote_identifier(database))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn profile_is_isolated_from_standard_mysql() {
+        assert!(is_profile(&DatabaseType::Mysql, Some("Dolt")));
+        assert!(!is_profile(&DatabaseType::Mysql, None));
+        assert!(!is_profile(&DatabaseType::Postgres, Some("dolt")));
+    }
+
+    #[test]
+    fn fallback_only_discovers_base_user_databases() {
+        assert!(is_branch_discovery_database("inventory"));
+        assert!(!is_branch_discovery_database("inventory/main"));
+        assert!(!is_branch_discovery_database("information_schema"));
+        assert!(!is_branch_discovery_database("mysql"));
+    }
+
+    #[test]
+    fn fallback_quotes_database_identifier() {
+        assert_eq!(
+            list_branches_sql("inventory`archive"),
+            "SELECT name FROM `inventory``archive`.dolt_branches ORDER BY name"
+        );
+    }
+}

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

@@ -4,6 +4,7 @@ pub mod cloudberry;
 pub mod cloudflare_d1;
 pub use cloudflare_d1 as cloudflare_d1_driver;
 pub mod document_result;
+pub mod dolt;
 pub mod doris;
 pub mod duckdb_sql;
 #[cfg(feature = "duckdb-sidecar")]

crates/dbx-core/src/schema.rs

@@ -605,6 +605,11 @@ async fn list_databases_once(state: &AppState, connection_id: &str) -> Result<Ve
     let pool = connections.get(connection_id).ok_or("Connection not found")?;
 
     match pool {
+        PoolKind::Mysql(p, mode)
+            if *mode != MysqlMode::OceanBaseOracle && db_config.as_ref().is_some_and(db::dolt::is_config) =>
+        {
+            db::dolt::list_databases(p).await
+        }
         PoolKind::Mysql(p, _) if db_config.as_ref().is_some_and(is_doris_family_config) => {
             db::mysql::list_databases_show(p)
                 .await