diff --git a/crates/dbx-core/src/agent_manager.rs b/crates/dbx-core/src/agent_manager.rs index 463b5a5f3..fa942b916 100644 --- a/crates/dbx-core/src/agent_manager.rs +++ b/crates/dbx-core/src/agent_manager.rs @@ -3,6 +3,7 @@ use std::path::{Path, PathBuf}; use serde::{Deserialize, Serialize}; use tokio::sync::Mutex; +use crate::database_capabilities; use crate::db::agent_driver::AgentDriverClient; use crate::models::connection::DatabaseType; @@ -166,33 +167,11 @@ impl AgentManager { } pub fn db_type_to_agent_key(db_type: &DatabaseType, driver_profile: Option<&str>) -> Option<&'static str> { - match db_type { - DatabaseType::Dameng => Some("dameng"), - DatabaseType::Kingbase => Some("kingbase"), - DatabaseType::Vastbase => Some("vastbase"), - DatabaseType::Goldendb => Some("goldendb"), - DatabaseType::Oracle => match driver_profile { - Some("oracle-10g") => Some("oracle-10g"), - _ => Some("oracle"), - }, - DatabaseType::H2 => Some("h2"), - DatabaseType::Snowflake => Some("snowflake"), - DatabaseType::Trino => Some("trino"), - DatabaseType::Hive => Some("hive"), - DatabaseType::Db2 => Some("db2"), - DatabaseType::Informix => Some("informix"), - DatabaseType::Neo4j => Some("neo4j"), - DatabaseType::Cassandra => Some("cassandra"), - DatabaseType::Bigquery => Some("bigquery"), - DatabaseType::Kylin => Some("kylin"), - DatabaseType::Sundb => Some("sundb"), - DatabaseType::Gaussdb => Some("gaussdb"), - _ => None, - } + database_capabilities::agent_key(db_type, driver_profile) } pub fn is_agent_type(db_type: &DatabaseType) -> bool { - Self::db_type_to_agent_key(db_type, None).is_some() + database_capabilities::is_agent_type(db_type) } pub async fn spawn( diff --git a/crates/dbx-core/src/connection.rs b/crates/dbx-core/src/connection.rs index c00029bca..c81fbd84e 100644 --- a/crates/dbx-core/src/connection.rs +++ b/crates/dbx-core/src/connection.rs @@ -3,6 +3,7 @@ use std::path::PathBuf; use std::sync::Arc; use tokio::sync::RwLock; +use crate::database_capabilities; use crate::db; use crate::db::proxy_tunnel::ProxyTunnelManager; use crate::db::ssh_tunnel::TunnelManager; @@ -61,7 +62,7 @@ pub struct AppState { pub fn metadata_connection_config(config: &ConnectionConfig) -> ConnectionConfig { let mut db_config = config.clone(); - if matches!(db_config.db_type, DatabaseType::Mysql | DatabaseType::Doris | DatabaseType::StarRocks) { + if database_capabilities::is_metadata_connection_scoped(&db_config.db_type) { db_config.database = None; } db_config @@ -122,17 +123,7 @@ impl AppState { configs.get(connection_id).map(|c| c.db_type.clone()) }; - let is_single_conn = matches!( - db_type, - Some(DatabaseType::Sqlite) - | Some(DatabaseType::DuckDb) - | Some(DatabaseType::Oracle) - | Some(DatabaseType::Dameng) - | Some(DatabaseType::Kingbase) - | Some(DatabaseType::Vastbase) - | Some(DatabaseType::Goldendb) - | Some(DatabaseType::Jdbc) - ); + let is_single_conn = db_type.as_ref().is_some_and(database_capabilities::is_single_connection_pool); let pool_key = if is_single_conn { connection_id.to_string() } else { @@ -357,15 +348,8 @@ impl AppState { configs .get(connection_id) .map(|c| { - matches!( - c.db_type, - DatabaseType::Oracle - | DatabaseType::Elasticsearch - | DatabaseType::Dameng - | DatabaseType::Kingbase - | DatabaseType::Vastbase - | DatabaseType::Goldendb - ) + database_capabilities::is_single_connection_pool(&c.db_type) + || c.db_type == DatabaseType::Elasticsearch }) .unwrap_or(false) }; @@ -404,29 +388,15 @@ pub fn redacted_connection_url_for_endpoint(config: &ConnectionConfig, host: &st } pub async fn probe_connection_endpoint(config: &ConnectionConfig, host: &str, port: u16) -> Result<(), String> { - match config.db_type { - DatabaseType::Sqlite | DatabaseType::DuckDb => Ok(()), - DatabaseType::MongoDb if config.connection_string.as_deref().is_some_and(|value| !value.is_empty()) => Ok(()), - DatabaseType::Jdbc => Ok(()), - DatabaseType::Dameng - | DatabaseType::Kingbase - | DatabaseType::Vastbase - | DatabaseType::Goldendb - | DatabaseType::Oracle - | DatabaseType::H2 - | DatabaseType::Snowflake - | DatabaseType::Trino - | DatabaseType::Hive - | DatabaseType::Db2 - | DatabaseType::Informix - | DatabaseType::Neo4j - | DatabaseType::Cassandra - | DatabaseType::Bigquery - | DatabaseType::Kylin - | DatabaseType::Sundb - | DatabaseType::Gaussdb => Ok(()), - _ => db::probe_tcp_endpoint(&format!("{:?}", config.db_type), host, port).await, + if config.db_type == DatabaseType::MongoDb + && config.connection_string.as_deref().is_some_and(|value| !value.is_empty()) + { + return Ok(()); } + if database_capabilities::skips_tcp_probe(&config.db_type) { + return Ok(()); + } + db::probe_tcp_endpoint(&format!("{:?}", config.db_type), host, port).await } async fn detect_ob_oracle_mode(config: &ConnectionConfig, pool: &sqlx::mysql::MySqlPool) -> MysqlMode { diff --git a/crates/dbx-core/src/db/agent_driver.rs b/crates/dbx-core/src/db/agent_driver.rs index 13038d5e3..b303ae50a 100644 --- a/crates/dbx-core/src/db/agent_driver.rs +++ b/crates/dbx-core/src/db/agent_driver.rs @@ -1,5 +1,7 @@ +use std::collections::VecDeque; use std::io::{BufRead, BufReader, BufWriter, Write}; -use std::process::{Child, ChildStdin, ChildStdout, Command, Stdio}; +use std::process::{Child, ChildStderr, ChildStdin, ChildStdout, Command, Stdio}; +use std::sync::{Arc, Mutex}; use std::time::Duration; use serde::de::DeserializeOwned; @@ -7,6 +9,7 @@ use serde_json::Value; const RPC_TIMEOUT_SECS: u64 = 30; const STARTUP_TIMEOUT_SECS: u64 = 15; +const STDERR_TAIL_LINES: usize = 20; #[cfg(windows)] const CREATE_NO_WINDOW: u32 = 0x08000000; @@ -14,9 +17,41 @@ pub struct AgentDriverClient { child: Child, stdin: Option>, stdout: Option>, + stderr_tail: Arc>, next_id: u64, } +struct StderrTail { + lines: VecDeque, + capacity: usize, +} + +impl Default for StderrTail { + fn default() -> Self { + Self::with_capacity(STDERR_TAIL_LINES) + } +} + +impl StderrTail { + fn with_capacity(capacity: usize) -> Self { + Self { lines: VecDeque::with_capacity(capacity), capacity } + } + + fn push_line(&mut self, line: String) { + if self.capacity == 0 { + return; + } + while self.lines.len() >= self.capacity { + self.lines.pop_front(); + } + self.lines.push_back(line.trim_end().to_string()); + } + + fn snapshot(&self) -> String { + self.lines.iter().filter(|line| !line.trim().is_empty()).cloned().collect::>().join("\n") + } +} + impl AgentDriverClient { /// Spawn a Java agent process and wait for it to signal readiness. /// @@ -36,7 +71,7 @@ impl AgentDriverClient { ]) .stdin(Stdio::piped()) .stdout(Stdio::piped()) - .stderr(Stdio::inherit()); + .stderr(Stdio::piped()); #[cfg(windows)] { @@ -48,12 +83,15 @@ impl AgentDriverClient { let child_stdin = child.stdin.take().ok_or("Failed to capture agent stdin")?; let child_stdout = child.stdout.take().ok_or("Failed to capture agent stdout")?; + let child_stderr = child.stderr.take().ok_or("Failed to capture agent stderr")?; let stdin = BufWriter::new(child_stdin); let mut stdout = BufReader::new(child_stdout); + let stderr_tail = Arc::new(Mutex::new(StderrTail::default())); + start_stderr_collector(child_stderr, stderr_tail.clone()); // Wait for the agent to signal readiness with {"ready":true} - let ready_stdout = tokio::time::timeout( + let startup_result = tokio::time::timeout( Duration::from_secs(STARTUP_TIMEOUT_SECS), tokio::task::spawn_blocking(move || { let line = read_agent_line(&mut stdout, "startup line")?; @@ -65,11 +103,34 @@ impl AgentDriverClient { Ok(stdout) }), ) - .await - .map_err(|_| format!("Agent startup timed out ({STARTUP_TIMEOUT_SECS}s)"))? - .map_err(|e| format!("Agent startup task failed: {e}"))??; + .await; - Ok(Self { child, stdin: Some(stdin), stdout: Some(ready_stdout), next_id: 0 }) + let ready_stdout = match startup_result { + Ok(Ok(Ok(stdout))) => stdout, + Ok(Ok(Err(e))) => { + return Err(format_agent_process_error( + &e, + child_exit_status(&mut child), + &stderr_tail_snapshot(&stderr_tail), + )); + } + Ok(Err(e)) => { + return Err(format_agent_process_error( + &format!("Agent startup task failed: {e}"), + child_exit_status(&mut child), + &stderr_tail_snapshot(&stderr_tail), + )); + } + Err(_) => { + return Err(format_agent_process_error( + &format!("Agent startup timed out ({STARTUP_TIMEOUT_SECS}s)"), + child_exit_status(&mut child), + &stderr_tail_snapshot(&stderr_tail), + )); + } + }; + + Ok(Self { child, stdin: Some(stdin), stdout: Some(ready_stdout), stderr_tail, next_id: 0 }) } /// Send a JSON-RPC 2.0 request and wait for the response. @@ -91,11 +152,18 @@ impl AgentDriverClient { serde_json::to_string(&request).map_err(|e| format!("Failed to serialize JSON-RPC request: {e}"))?; // Write request to stdin - { + let write_result = { let writer = self.stdin.as_mut().ok_or("Agent stdin not available")?; - writer.write_all(request_line.as_bytes()).map_err(|e| format!("Failed to write to agent stdin: {e}"))?; - writer.write_all(b"\n").map_err(|e| format!("Failed to write newline to agent stdin: {e}"))?; - writer.flush().map_err(|e| format!("Failed to flush agent stdin: {e}"))?; + writer + .write_all(request_line.as_bytes()) + .map_err(|e| format!("Failed to write to agent stdin: {e}")) + .and_then(|_| { + writer.write_all(b"\n").map_err(|e| format!("Failed to write newline to agent stdin: {e}")) + }) + .and_then(|_| writer.flush().map_err(|e| format!("Failed to flush agent stdin: {e}"))) + }; + if let Err(e) = write_result { + return Err(self.format_agent_process_error(&e)); } // Read response from stdout (blocking, with timeout) @@ -135,7 +203,7 @@ impl AgentDriverClient { .map_err(|e| format!("Agent RPC task failed: {e}"))?; let _ = self.stdout.insert(returned_reader); - result + result.map_err(|e| self.format_agent_process_error(&e)) } /// Send a shutdown message to the agent and wait for the process to exit. @@ -177,6 +245,64 @@ fn read_agent_line(reader: &mut R, context: &str) -> Result>) { + std::thread::spawn(move || { + let mut reader = BufReader::new(stderr); + let mut line = String::new(); + loop { + line.clear(); + match reader.read_line(&mut line) { + Ok(0) => break, + Ok(_) => { + log::warn!("[agent:stderr] {}", line.trim_end()); + if let Ok(mut tail) = stderr_tail.lock() { + tail.push_line(line.clone()); + } + } + Err(err) => { + log::warn!("[agent:stderr] failed to read stderr: {err}"); + break; + } + } + } + }); +} + +fn child_exit_status(child: &mut Child) -> Option { + match child.try_wait() { + Ok(Some(status)) => Some(status.to_string()), + Ok(None) => None, + Err(err) => Some(format!("status unavailable: {err}")), + } +} + +fn stderr_tail_snapshot(stderr_tail: &Arc>) -> StderrTail { + let snapshot = stderr_tail.lock().map(|tail| tail.snapshot()).unwrap_or_default(); + let mut tail = StderrTail::with_capacity(STDERR_TAIL_LINES); + for line in snapshot.lines() { + tail.push_line(line.to_string()); + } + tail +} + +fn format_agent_process_error(base: &str, exit_status: Option, stderr_tail: &StderrTail) -> String { + let mut parts = vec![base.to_string()]; + if let Some(status) = exit_status { + parts.push(format!("agent process exited with {status}")); + } + let stderr = stderr_tail.snapshot(); + if !stderr.is_empty() { + parts.push(format!("recent stderr:\n{stderr}")); + } + parts.join(". ") +} + +impl AgentDriverClient { + fn format_agent_process_error(&mut self, base: &str) -> String { + format_agent_process_error(base, child_exit_status(&mut self.child), &stderr_tail_snapshot(&self.stderr_tail)) + } +} + impl Drop for AgentDriverClient { fn drop(&mut self) { self.kill(); @@ -185,7 +311,7 @@ impl Drop for AgentDriverClient { #[cfg(test)] mod tests { - use super::read_agent_line; + use super::{format_agent_process_error, read_agent_line, StderrTail}; use std::io::Cursor; #[test] @@ -197,4 +323,34 @@ mod tests { assert_eq!(line, format!("{{\"error\":{}}}\n", "\u{fffd}\u{fffd}")); } + + #[test] + fn formats_agent_process_error_with_exit_status_and_stderr_tail() { + let mut stderr_tail = StderrTail::default(); + stderr_tail.push_line("java.lang.NoClassDefFoundError: org/apache/hive/jdbc/HiveDriver".to_string()); + stderr_tail.push_line("\tat com.dbx.agent.hive.HiveAgent.connect(HiveAgent.kt:21)".to_string()); + + let message = format_agent_process_error( + "Failed to read response from agent: end of stream", + Some("exit status: 1".to_string()), + &stderr_tail, + ); + + assert!(message.contains("Failed to read response from agent: end of stream")); + assert!(message.contains("agent process exited with exit status: 1")); + assert!(message.contains("recent stderr:")); + assert!(message.contains("NoClassDefFoundError")); + assert!(message.contains("HiveAgent.connect")); + } + + #[test] + fn stderr_tail_keeps_recent_lines_only() { + let mut stderr_tail = StderrTail::with_capacity(3); + stderr_tail.push_line("line 1".to_string()); + stderr_tail.push_line("line 2".to_string()); + stderr_tail.push_line("line 3".to_string()); + stderr_tail.push_line("line 4".to_string()); + + assert_eq!(stderr_tail.snapshot(), "line 2\nline 3\nline 4"); + } } diff --git a/crates/dbx-core/src/lib.rs b/crates/dbx-core/src/lib.rs index 277ce374a..859f270e8 100644 --- a/crates/dbx-core/src/lib.rs +++ b/crates/dbx-core/src/lib.rs @@ -2,6 +2,7 @@ pub mod agent_manager; pub mod ai; pub mod connection; pub mod connection_secrets; +pub mod database_capabilities; pub mod database_export; pub mod db; pub mod external; diff --git a/src-tauri/src/commands/connection.rs b/src-tauri/src/commands/connection.rs index 8ec328423..1dc79816f 100644 --- a/src-tauri/src/commands/connection.rs +++ b/src-tauri/src/commands/connection.rs @@ -5,6 +5,7 @@ pub use dbx_core::connection::{ connection_url_for_endpoint, expand_tilde, metadata_connection_config, probe_connection_endpoint, redacted_connection_url_for_endpoint, AppState, MysqlMode, PoolKind, }; +use dbx_core::database_capabilities; use dbx_core::db; use dbx_core::models::connection::{rewrite_jdbc_url_host, ConnectionConfig, DatabaseType}; @@ -106,23 +107,7 @@ pub async fn test_connection(state: State<'_, Arc>, config: Connection ); db::elasticsearch_driver::test_connection(&client).await.map(|_| "Connection successful".to_string()) } - DatabaseType::Dameng - | DatabaseType::Kingbase - | DatabaseType::Vastbase - | DatabaseType::Goldendb - | DatabaseType::Oracle - | DatabaseType::H2 - | DatabaseType::Snowflake - | DatabaseType::Trino - | DatabaseType::Hive - | DatabaseType::Db2 - | DatabaseType::Informix - | DatabaseType::Neo4j - | DatabaseType::Cassandra - | DatabaseType::Bigquery - | DatabaseType::Kylin - | DatabaseType::Sundb - | DatabaseType::Gaussdb => { + db_type if database_capabilities::is_agent_type(&db_type) => { state .agent_manager .call_daemon::( @@ -149,6 +134,7 @@ pub async fn test_connection(state: State<'_, Arc>, config: Connection } state.test_external_driver("jdbc", &jdbc_config).await } + db_type => Err(format!("Unsupported database type: {db_type:?}")), }, }; @@ -222,23 +208,7 @@ pub async fn connect_db(state: State<'_, Arc>, config: ConnectionConfi db::elasticsearch_driver::test_connection(&client).await?; PoolKind::Elasticsearch(client) } - DatabaseType::Dameng - | DatabaseType::Kingbase - | DatabaseType::Vastbase - | DatabaseType::Goldendb - | DatabaseType::Oracle - | DatabaseType::H2 - | DatabaseType::Snowflake - | DatabaseType::Trino - | DatabaseType::Hive - | DatabaseType::Db2 - | DatabaseType::Informix - | DatabaseType::Neo4j - | DatabaseType::Cassandra - | DatabaseType::Bigquery - | DatabaseType::Kylin - | DatabaseType::Sundb - | DatabaseType::Gaussdb => { + db_type if database_capabilities::is_agent_type(&db_type) => { let mut client = state.agent_manager.spawn(&db_config.db_type, db_config.driver_profile.as_deref()).await?; client .call::( @@ -255,6 +225,7 @@ pub async fn connect_db(state: State<'_, Arc>, config: ConnectionConfi PoolKind::Agent(std::sync::Arc::new(tokio::sync::Mutex::new(client))) } DatabaseType::Jdbc => state.external_driver_pool("jdbc", &db_config).await?, + db_type => return Err(format!("Unsupported database type: {db_type:?}")), }; state.connections.write().await.insert(id.clone(), pool); diff --git a/src/components/grid/DataGrid.vue b/src/components/grid/DataGrid.vue index 3bd534bf6..3de0e085a 100644 --- a/src/components/grid/DataGrid.vue +++ b/src/components/grid/DataGrid.vue @@ -77,7 +77,7 @@ import { import { formatGridSqlLiteral } from "@/lib/dataGridSql"; import { matchesRowStatusFilter, type RowStatus, type RowStatusFilter } from "@/lib/gridRowStatus"; import { displayCellValue, type CellValue } from "@/lib/cellValue"; -import { isCancelSearchShortcut } from "@/lib/keyboardShortcuts"; +import { isCancelSearchShortcut, isFocusSearchShortcut } from "@/lib/keyboardShortcuts"; import { useToast } from "@/composables/useToast"; import { useDataGridExport } from "@/composables/useDataGridExport"; @@ -224,6 +224,8 @@ const sortColIndex = ref(null); const sortDir = ref<"asc" | "desc">("asc"); const searchText = ref(""); const deferredClientSearchText = ref(""); +const searchOverlayVisible = ref(false); +const currentMatchIndex = ref(-1); let _searchTimer: ReturnType | undefined; const searchSuggestions = ref([]); @@ -467,14 +469,23 @@ function navigateSuggestion(delta: number) { } function focusSearch(): boolean { - const input = searchInputRef.value; - if (!input) return false; - input.focus(); - input.select(); - updateSuggestionPosition(); + searchOverlayVisible.value = true; + nextTick(() => { + const input = searchInputRef.value; + if (!input) return; + input.focus(); + input.select(); + updateSuggestionPosition(); + }); return true; } +function closeSearch() { + searchOverlayVisible.value = false; + searchText.value = ""; + searchSuggestions.value = []; +} + const PAIRS: Record = { "'": "'", '"': '"', "(": ")" }; function onSearchKeydown(e: KeyboardEvent) { @@ -536,7 +547,12 @@ function onSearchKeydown(e: KeyboardEvent) { } if (isCancelSearchShortcut(e)) { e.preventDefault(); - searchText.value = ""; + closeSearch(); + return; + } + if (e.key === "Enter") { + e.preventDefault(); + navigateMatch(e.shiftKey ? -1 : 1); } } @@ -1068,6 +1084,67 @@ const displayItems = computed(() => { return items.filter((item) => matchesRowStatusFilter(item.status, rowStatusFilter.value)); }); +interface SearchMatch { + displayRow: number; + col: number; +} + +const searchMatches = computed(() => { + const q = deferredClientSearchText.value; + if (!q) return []; + const items = displayItems.value; + const matches: SearchMatch[] = []; + for (let r = 0; r < items.length; r++) { + const data = items[r].data; + for (let c = 0; c < data.length; c++) { + if (data[c] !== null && String(data[c]).toLowerCase().includes(q)) { + matches.push({ displayRow: r, col: c }); + } + } + } + return matches; +}); + +const searchMatchSet = computed(() => { + const set = new Set(); + for (const m of searchMatches.value) { + set.add(`${m.displayRow}:${m.col}`); + } + return set; +}); + +watch(searchMatches, (matches) => { + currentMatchIndex.value = matches.length > 0 ? 0 : -1; +}); + +function cellIsSearchMatch(displayRow: number, col: number): boolean { + return searchMatchSet.value.has(`${displayRow}:${col}`); +} + +function cellIsCurrentMatch(displayRow: number, col: number): boolean { + const idx = currentMatchIndex.value; + if (idx < 0 || idx >= searchMatches.value.length) return false; + const m = searchMatches.value[idx]; + return m.displayRow === displayRow && m.col === col; +} + +function navigateMatch(delta: number) { + const total = searchMatches.value.length; + if (total === 0) return; + currentMatchIndex.value = (currentMatchIndex.value + delta + total) % total; + scrollToCurrentMatch(); +} + +function scrollToCurrentMatch() { + const idx = currentMatchIndex.value; + if (idx < 0 || idx >= searchMatches.value.length) return; + const match = searchMatches.value[idx]; + const scrollEl = gridRef.value; + if (!scrollEl) return; + const rowEl = scrollEl.querySelector(`[data-row-index="${match.displayRow}"]`) as HTMLElement | null; + if (rowEl) rowEl.scrollIntoView({ block: "center" }); +} + function getRowItem(rowId: number): RowItem | undefined { return displayItems.value.find((item) => item.id === rowId); } @@ -1575,6 +1652,11 @@ function cutSelection() { } async function onGridKeydown(event: KeyboardEvent) { + if (isFocusSearchShortcut(event)) { + event.preventDefault(); + focusSearch(); + return; + } if (eventTargetAllowsNativeClipboard(event)) return; if (clipboardShortcut(event, "c")) { if (!hasCellSelection.value) return; @@ -1862,64 +1944,24 @@ defineExpose({ -
- - -