From 5999f1cba151460d90a60e2237f31009aa8b1355 Mon Sep 17 00:00:00 2001 From: Antigravity Agent Date: Thu, 17 Sep 2026 19:04:42 +0700 Subject: [PATCH 01/32] fix(query): route results to origin tab, report affected rows, cancel on server - Give QueryTab a stable id and carry it through QueryJob/QueryResultMessage so a result that finishes after a tab switch lands in the tab that ran it. - Run INSERT/UPDATE/DELETE/DDL via execute() and report the driver's rows_affected instead of the result-set length (always 0 before). - Split statements with the quote/dollar-quote/comment aware splitter in the executors, auto-pagination check and user manager (was split(';')). - Execute statements that start with a comment instead of skipping them. - Decode PostgreSQL values by native type (INT4, NUMERIC, TIMESTAMP, UUID, JSON, arrays, ...) instead of showing [unsupported]/Error. - MySQL: stop retrying the whole job after a statement error or timeout, which could execute DML up to three times; retry only on connect failure. - Replace hard-coded 10s/15s/60s timeouts with a configurable query timeout (default: none) and add a max-rows-per-result limit with a notice. - Record backend pid per job; cancel and timeout send pg_cancel_backend / KILL QUERY so the statement stops on the server too. - Surface query start failures as toasts and reset the running state. - Collapse the duplicated synchronous executor into a wrapper over the async executor (~780 lines removed). - Fix pre-existing clippy errors that broke CI. Co-Authored-By: Claude Opus 5 (1M context) --- src/config.rs | 37 +- src/connection/execute.rs | 1315 ++++++++++------------------ src/connection/session.rs | 96 +- src/connection/sql.rs | 83 +- src/connection/types.rs | 49 ++ src/data_table/export_clipboard.rs | 4 +- src/driver_postgres.rs | 97 ++ src/editor.rs | 40 +- src/models/structs.rs | 5 + src/query_tools/text_actions.rs | 11 +- src/sidebar_query.rs | 3 + src/user_manager.rs | 5 +- src/window_egui/app_impl.rs | 37 + src/window_egui/init.rs | 7 + src/window_egui/mod.rs | 8 + src/window_egui/query_jobs.rs | 174 +++- 16 files changed, 1034 insertions(+), 937 deletions(-) diff --git a/src/config.rs b/src/config.rs index fcf31d6a..1311f6d8 100644 --- a/src/config.rs +++ b/src/config.rs @@ -228,12 +228,31 @@ pub struct AppPreferences { pub redis_browser_auto_refresh_seconds: u32, #[serde(default)] pub sync_server_url: Option, + /// Timeout query per statement dalam detik; 0 berarti tanpa batas. + #[serde(default)] + pub query_timeout_secs: u32, + /// Jumlah baris maksimum yang disimpan dari satu result set tanpa paginasi. + #[serde(default = "default_max_result_rows")] + pub max_result_rows: u32, + /// Buka kembali tab query dari sesi sebelumnya (termasuk draft yang belum disimpan). + #[serde(default = "default_true")] + pub restore_session: bool, } fn default_redis_browser_auto_refresh_seconds() -> u32 { 5 } +pub const DEFAULT_MAX_RESULT_ROWS: u32 = 50_000; + +fn default_max_result_rows() -> u32 { + DEFAULT_MAX_RESULT_ROWS +} + +fn default_true() -> bool { + true +} + impl Default for AppPreferences { fn default() -> Self { Self { @@ -254,6 +273,9 @@ impl Default for AppPreferences { ai_base_url: String::new(), redis_browser_auto_refresh_seconds: default_redis_browser_auto_refresh_seconds(), sync_server_url: Some("https://api.tabular.id".to_string()), + query_timeout_secs: 0, + max_result_rows: DEFAULT_MAX_RESULT_ROWS, + restore_session: true, } } } @@ -366,6 +388,9 @@ impl ConfigStore { redis_browser_auto_refresh_seconds: default_redis_browser_auto_refresh_seconds(), sync_server_url: Some("https://api.tabular.id".to_string()), ui_mode: UiModePreference::Auto, + query_timeout_secs: 0, + max_result_rows: DEFAULT_MAX_RESULT_ROWS, + restore_session: true, }; // Set when a legacy plaintext AI key was migrated to the secret @@ -413,6 +438,11 @@ impl ConfigStore { "sync_server_url" => { prefs.sync_server_url = if v.is_empty() { None } else { Some(v) } } + "query_timeout_secs" => prefs.query_timeout_secs = v.parse().unwrap_or(0), + "max_result_rows" => { + prefs.max_result_rows = v.parse().unwrap_or(DEFAULT_MAX_RESULT_ROWS) + } + "restore_session" => prefs.restore_session = v == "1", _ => {} } } @@ -468,7 +498,9 @@ impl ConfigStore { // The key goes to the OS keychain; the row keeps only a sentinel. let ai_api_key_stored = crate::secrets::store_or_keep("pref:ai_api_key", &prefs.ai_api_key); - let entries: [(&str, &str); 16] = [ + let query_timeout_secs = prefs.query_timeout_secs.to_string(); + let max_result_rows = prefs.max_result_rows.to_string(); + let entries: [(&str, &str); 19] = [ ("theme", prefs.theme.as_str()), ("ui_mode", prefs.ui_mode.as_str()), ( @@ -507,6 +539,9 @@ impl ConfigStore { "sync_server_url", prefs.sync_server_url.as_deref().unwrap_or(""), ), + ("query_timeout_secs", &query_timeout_secs), + ("max_result_rows", &max_result_rows), + ("restore_session", if prefs.restore_session { "1" } else { "0" }), ]; for (k, v) in entries.iter() { diff --git a/src/connection/execute.rs b/src/connection/execute.rs index 5fb97ecc..fcd7060c 100644 --- a/src/connection/execute.rs +++ b/src/connection/execute.rs @@ -11,13 +11,105 @@ use std::time::Instant; use super::pool::{resolve_connection_target_async, try_get_connection_pool}; use super::sql::{ - infer_column_origins, infer_select_headers, is_simple_select_statement, - query_contains_pagination, should_enable_auto_pagination, + infer_column_origins, infer_select_headers, is_comment_only_statement, + is_simple_select_statement, query_contains_pagination, should_enable_auto_pagination, + split_sql_statements, statement_returns_rows, strip_leading_sql_comments, }; use super::types::{ - QueryExecutionError, QueryExecutionOptions, QueryJob, QueryJobOutput, QueryPreparationError, - QueryResultMessage, + BackendPidGuard, QueryExecutionError, QueryExecutionOptions, QueryJob, QueryJobOutput, + QueryPreparationError, QueryResultMessage, }; +use futures_util::TryStreamExt; + +/// Membaca result set lewat stream dan berhenti setelah `$max` baris. +/// Menghasilkan `Result<(Vec, bool /* terpotong */), sqlx::Error>`. +macro_rules! fetch_rows_limited { + ($query:expr, $executor:expr, $max:expr) => { + async { + let mut stream = $query.fetch($executor); + let mut rows = Vec::new(); + let mut truncated = false; + while let Some(row) = stream.try_next().await? { + if rows.len() >= $max { + truncated = true; + break; + } + rows.push(row); + } + Ok::<_, sqlx::Error>((rows, truncated)) + } + }; +} + +/// Menjalankan future dengan batas waktu opsional. `Err(())` berarti timeout. +async fn run_with_timeout( + timeout: Option, + fut: F, +) -> Result { + match timeout { + Some(limit) => tokio::time::timeout(limit, fut).await.map_err(|_| ()), + None => Ok(fut.await), + } +} + +/// Pesan error timeout yang konsisten untuk semua driver. +fn timeout_message(options: &QueryExecutionOptions) -> String { + match options.query_timeout { + Some(limit) => format!( + "Query timed out after {}s and was cancelled. Adjust the limit in Settings → Performance → Query timeout.", + limit.as_secs() + ), + None => "Query timed out".to_string(), + } +} + +/// Memecah query job menjadi statement dengan splitter yang paham quote, +/// dollar-quote, dan komentar; statement yang hanya berisi komentar dibuang. +fn job_statements(options: &QueryExecutionOptions) -> Vec { + let hash_is_comment = matches!( + options.connection.connection_type, + models::enums::DatabaseType::MySQL + ); + split_sql_statements(&options.query, hash_is_comment) + .into_iter() + .filter(|s| !is_comment_only_statement(s)) + .collect() +} + +/// Potong teks untuk pratinjau tanpa memotong di tengah karakter multibyte. +fn preview_text(text: &str, max_chars: usize) -> String { + if text.chars().count() > max_chars { + format!("{}...", text.chars().take(max_chars).collect::()) + } else { + text.to_string() + } +} + +/// Minta server menghentikan statement yang sedang berjalan pada sesi `pid`. +/// Dipakai saat user menekan cancel atau saat timeout tercapai. +pub(crate) async fn cancel_backend_query(pool: models::enums::DatabasePool, pid: i64) { + match pool { + models::enums::DatabasePool::PostgreSQL(pg) => { + let result = sqlx::query("SELECT pg_cancel_backend($1)") + .bind(pid as i32) + .execute(pg.as_ref()) + .await; + if let Err(e) = result { + log::warn!("[CANCEL] pg_cancel_backend({}) failed: {}", pid, e); + } + } + models::enums::DatabasePool::MySQL(my) => { + let kill = format!("KILL QUERY {}", pid); + if let Err(e) = sqlx::query(sqlx::AssertSqlSafe(kill.as_str())) + .execute(my.as_ref()) + .await + { + log::warn!("[CANCEL] KILL QUERY {} failed: {}", pid, e); + } + } + _ => {} + } +} pub(crate) fn prepare_query_job( tabular: &mut Tabular, @@ -72,10 +164,18 @@ pub(crate) fn prepare_query_job( dba_special_mode, save_to_history: true, ast_enabled: cfg!(feature = "query_ast"), + job_id, + query_timeout: (tabular.query_timeout_secs > 0) + .then(|| std::time::Duration::from_secs(tabular.query_timeout_secs as u64)), + max_rows: tabular.max_result_rows.max(1) as usize, + backend_pids: tabular.query_backend_pids.clone(), }; + let tab_id = tabular.query_tabs.get(tabular.active_tab_index).map(|t| t.id); + Ok(QueryJob { job_id, + tab_id, options, connection_pool, started_at: Instant::now(), @@ -138,6 +238,7 @@ fn skipped_statement_message(job: &QueryJob) -> QueryResultMessage { let message = "Skipped: a previous statement in this batch failed".to_string(); QueryResultMessage { job_id: job.job_id, + tab_id: job.tab_id, connection_id: job.options.connection_id, success: false, headers: vec!["Error".to_string()], @@ -150,11 +251,13 @@ fn skipped_statement_message(job: &QueryJob) -> QueryResultMessage { ast_headers: None, affected_rows: None, column_metadata: None, + truncated: false, } } async fn execute_query_job(job: QueryJob) -> QueryResultMessage { let start = job.started_at; + let tab_id = job.tab_id; let connection_id = job.options.connection_id; let query = job.options.query.clone(); let dba_special_mode = job.options.dba_special_mode.clone(); @@ -186,6 +289,7 @@ async fn execute_query_job(job: QueryJob) -> QueryResultMessage { match outcome { Ok(output) => QueryResultMessage { job_id: job.job_id, + tab_id, connection_id, success: true, headers: output.headers.clone(), @@ -196,13 +300,15 @@ async fn execute_query_job(job: QueryJob) -> QueryResultMessage { dba_special_mode, ast_debug_sql: output.ast_debug_sql, ast_headers: output.ast_headers, - affected_rows: Some(output.rows.len()), + affected_rows: output.affected_rows.map(|n| n as usize), column_metadata: output.column_metadata, + truncated: output.truncated, }, Err(err) => { let message = describe_execution_error(err); QueryResultMessage { job_id: job.job_id, + tab_id, connection_id, success: false, headers: vec!["Error".to_string()], @@ -215,6 +321,7 @@ async fn execute_query_job(job: QueryJob) -> QueryResultMessage { ast_headers: None, affected_rows: None, column_metadata: None, + truncated: false, } } } @@ -232,7 +339,7 @@ fn describe_execution_error(err: QueryExecutionError) -> String { async fn execute_mysql_query_job( options: &QueryExecutionOptions, - _pool: models::enums::DatabasePool, + pool: models::enums::DatabasePool, ) -> Result { debug!( "[async] Executing MySQL query (conn_id={})", @@ -243,12 +350,8 @@ async fn execute_mysql_query_job( .await .map_err(QueryExecutionError::Message)?; - let statements_raw: Vec<&str> = options - .query - .split(';') - .map(|s| s.trim()) - .filter(|s| !s.is_empty()) - .collect(); + let statements_owned = job_statements(options); + let statements_raw: Vec<&str> = statements_owned.iter().map(|s| s.as_str()).collect(); #[cfg(feature = "query_ast")] let mut inferred_headers_from_ast: Option> = None; @@ -380,24 +483,29 @@ async fn execute_mysql_query_job( .execute(&mut conn) .await; + // Catat connection id supaya cancel/timeout bisa mengirim KILL QUERY. + let mut _pid_guard = sqlx::query_scalar::<_, u64>("SELECT CONNECTION_ID()") + .fetch_one(&mut conn) + .await + .ok() + .map(|pid| BackendPidGuard::register(&options.backend_pids, options.job_id, pid as i64)); + let mut final_headers: Vec = Vec::new(); let mut final_data: Vec> = Vec::new(); let mut final_column_metadata: Option> = None; + let mut final_affected: Option = None; + let mut final_truncated = false; let mut execution_success = true; for (idx, statement) in statements_ref.iter().enumerate() { let trimmed = statement.trim(); - if trimmed.is_empty() - || trimmed.starts_with("--") - || trimmed.starts_with('#') - || trimmed.starts_with("/*") - { + if is_comment_only_statement(trimmed) { continue; } debug!("[mysql] about to run statement[{}]: {:?}", idx + 1, trimmed); - let upper = trimmed.to_uppercase(); + let upper = strip_leading_sql_comments(trimmed).to_uppercase(); let is_admin_command = { upper.starts_with("PURGE BINARY LOGS") @@ -411,7 +519,7 @@ async fn execute_mysql_query_job( }; if upper.starts_with("USE ") { - let db_part = trimmed[3..].trim(); + let db_part = strip_leading_sql_comments(trimmed)[3..].trim(); let db_name = db_part .trim_matches('`') .trim_matches('"') @@ -447,6 +555,17 @@ async fn execute_mysql_query_job( .execute(&mut new_conn) .await; conn = new_conn; + _pid_guard = sqlx::query_scalar::<_, u64>("SELECT CONNECTION_ID()") + .fetch_one(&mut conn) + .await + .ok() + .map(|pid| { + BackendPidGuard::register( + &options.backend_pids, + options.job_id, + pid as i64, + ) + }); } Err(e) => { last_error = Some(format!("USE failed (reconnect): {}", e)); @@ -458,15 +577,30 @@ async fn execute_mysql_query_job( continue; } - let query_result = tokio::time::timeout( - std::time::Duration::from_secs(60), - sqlx::query(sqlx::AssertSqlSafe(trimmed)).fetch_all(&mut conn), - ) + let returns_rows = statement_returns_rows(trimmed); + let query_result = run_with_timeout(options.query_timeout, async { + if returns_rows { + fetch_rows_limited!( + sqlx::query(sqlx::AssertSqlSafe(trimmed)), + &mut conn, + options.max_rows + ) + .await + .map(|(rows, truncated)| (rows, truncated, None)) + } else { + sqlx::query(sqlx::AssertSqlSafe(trimmed)) + .execute(&mut conn) + .await + .map(|r| (Vec::new(), false, Some(r.rows_affected()))) + } + }) .await; match query_result { - Ok(Ok(rows)) => { + Ok(Ok((rows, truncated, affected))) => { if idx == statements_ref.len() - 1 { + final_affected = affected; + final_truncated = truncated; if !rows.is_empty() { final_headers = rows[0] .columns() @@ -828,12 +962,7 @@ async fn execute_mysql_query_job( } } else { if failing_stmt_preview.is_none() { - let prev = if trimmed.len() > 200 { - format!("{}...", &trimmed[..200]) - } else { - trimmed.to_string() - }; - failing_stmt_preview = Some(prev); + failing_stmt_preview = Some(preview_text(trimmed, 200)); } if err_str.contains("1146") || err_str.to_lowercase().contains("doesn't exist") @@ -855,7 +984,18 @@ async fn execute_mysql_query_job( } } Err(_) => { - last_error = Some("Query timeout after 60s".to_string()); + // Future yang di-drop tidak menghentikan query di server, + // jadi kirim KILL QUERY lewat koneksi lain dari pool. + let pid = options + .backend_pids + .lock() + .ok() + .and_then(|m| m.get(&options.job_id).copied()); + if let Some(pid) = pid { + cancel_backend_query(pool.clone(), pid).await; + } + last_error = Some(timeout_message(options)); + failing_stmt_preview.get_or_insert_with(|| preview_text(trimmed, 200)); execution_success = false; break; } @@ -881,8 +1021,17 @@ async fn execute_mysql_query_job( ast_debug_sql, ast_headers, column_metadata: final_column_metadata, + affected_rows: final_affected, + truncated: final_truncated, }); } + + // Koneksi sudah terbentuk tetapi statement gagal atau timeout. Jangan + // diulang: statement sebelumnya (atau statement yang timeout itu + // sendiri) mungkin sudah berefek, sehingga retry bisa menjalankan DML + // dua kali. Retry hanya untuk kegagalan membuka koneksi (lihat `continue` + // di atas). + break; } let mut final_err = last_error.unwrap_or_else(|| "Unknown MySQL error".to_string()); @@ -905,12 +1054,8 @@ async fn execute_postgres_query_job( } }; - let statements_raw: Vec<&str> = options - .query - .split(';') - .map(|s| s.trim()) - .filter(|s| !s.is_empty()) - .collect(); + let statements_owned = job_statements(options); + let statements_raw: Vec<&str> = statements_owned.iter().map(|s| s.as_str()).collect(); #[cfg(feature = "query_ast")] let mut inferred_headers_from_ast: Option> = None; @@ -963,52 +1108,60 @@ async fn execute_postgres_query_job( #[cfg(not(feature = "query_ast"))] let statements_ref: Vec<&str> = statements.iter().map(|s| s.as_str()).collect(); + // Semua statement dalam job memakai satu koneksi yang sama, sehingga SET / + // search_path dan statement berikutnya konsisten, dan backend pid-nya + // diketahui untuk keperluan cancel. + let mut conn = pg_pool.acquire().await.map_err(|e| { + QueryExecutionError::Message(format!("PostgreSQL connection error: {}", e)) + })?; + let _pid_guard = sqlx::query_scalar::<_, i32>("SELECT pg_backend_pid()") + .fetch_one(&mut *conn) + .await + .ok() + .map(|pid| BackendPidGuard::register(&options.backend_pids, options.job_id, pid as i64)); + let mut final_headers = Vec::new(); let mut final_data = Vec::new(); + let mut final_affected: Option = None; + let mut final_truncated = false; for (i, statement) in statements_ref.iter().enumerate() { let trimmed = statement.trim(); - if trimmed.is_empty() || trimmed.starts_with("--") || trimmed.starts_with("/*") { + if is_comment_only_statement(trimmed) { continue; } - let result = tokio::time::timeout( - std::time::Duration::from_secs(15), - sqlx::query(sqlx::AssertSqlSafe(trimmed)).fetch_all(pg_pool.as_ref()), - ) + let returns_rows = statement_returns_rows(trimmed); + let result = run_with_timeout(options.query_timeout, async { + if returns_rows { + fetch_rows_limited!( + sqlx::query(sqlx::AssertSqlSafe(trimmed)), + &mut *conn, + options.max_rows + ) + .await + .map(|(rows, truncated)| (rows, truncated, None)) + } else { + sqlx::query(sqlx::AssertSqlSafe(trimmed)) + .execute(&mut *conn) + .await + .map(|r| (Vec::new(), false, Some(r.rows_affected()))) + } + }) .await; match result { - Ok(Ok(rows)) => { + Ok(Ok((rows, truncated, affected))) => { if i == statements_ref.len() - 1 { + final_affected = affected; + final_truncated = truncated; if !rows.is_empty() { final_headers = rows[0] .columns() .iter() .map(|c| c.name().to_string()) .collect(); - final_data = rows - .into_iter() - .map(|row| { - (0..row.len()) - .map(|idx| match row.try_get::, _>(idx) { - Ok(Some(v)) => v, - Ok(None) => "NULL".to_string(), - Err(_) => { - if let Ok(val) = row.try_get::(idx) { - val.to_string() - } else if let Ok(val) = row.try_get::(idx) { - val.to_string() - } else if let Ok(val) = row.try_get::(idx) { - val.to_string() - } else { - "[unsupported]".to_string() - } - } - }) - .collect() - }) - .collect(); + final_data = crate::driver_postgres::convert_postgres_rows_to_table_data(rows); } else { #[cfg(feature = "query_ast")] if final_headers.is_empty() @@ -1018,7 +1171,9 @@ async fn execute_postgres_query_job( final_headers = hh; } if final_headers.is_empty() - && trimmed.to_uppercase().starts_with("SELECT") + && strip_leading_sql_comments(trimmed) + .to_uppercase() + .starts_with("SELECT") { let inferred = infer_select_headers(trimmed); if !inferred.is_empty() { @@ -1036,9 +1191,24 @@ async fn execute_postgres_query_job( ))); } Err(_) => { - return Err(QueryExecutionError::Message( - "PostgreSQL query timed out".to_string(), - )); + // Drop future tidak menghentikan query di server; kirim + // pg_cancel_backend lewat koneksi lain dari pool. + let pid = options + .backend_pids + .lock() + .ok() + .and_then(|m| m.get(&options.job_id).copied()); + // Koneksi ini masih menunggu hasil query yang dibatalkan; + // lepaskan (tutup) supaya tidak dikembalikan ke pool. + drop(conn.detach()); + if let Some(pid) = pid { + cancel_backend_query( + models::enums::DatabasePool::PostgreSQL(pg_pool.clone()), + pid, + ) + .await; + } + return Err(QueryExecutionError::Message(timeout_message(options))); } } } @@ -1049,6 +1219,8 @@ async fn execute_postgres_query_job( ast_debug_sql, ast_headers, column_metadata: None, + affected_rows: final_affected, + truncated: final_truncated, }) } @@ -1065,12 +1237,8 @@ async fn execute_sqlite_query_job( } }; - let statements_raw: Vec<&str> = options - .query - .split(';') - .map(|s| s.trim()) - .filter(|s| !s.is_empty()) - .collect(); + let statements_owned = job_statements(options); + let statements_raw: Vec<&str> = statements_owned.iter().map(|s| s.as_str()).collect(); #[cfg(feature = "query_ast")] let mut inferred_headers_from_ast: Option> = None; @@ -1125,50 +1293,46 @@ async fn execute_sqlite_query_job( let mut final_headers = Vec::new(); let mut final_data = Vec::new(); + let mut final_affected: Option = None; + let mut final_truncated = false; for (i, statement) in statements_ref.iter().enumerate() { let trimmed = statement.trim(); - if trimmed.is_empty() || trimmed.starts_with("--") || trimmed.starts_with("/*") { + if is_comment_only_statement(trimmed) { continue; } - let result = tokio::time::timeout( - std::time::Duration::from_secs(10), - sqlx::query(sqlx::AssertSqlSafe(trimmed)).fetch_all(sqlite_pool.as_ref()), - ) + let returns_rows = statement_returns_rows(trimmed); + let result = run_with_timeout(options.query_timeout, async { + if returns_rows { + fetch_rows_limited!( + sqlx::query(sqlx::AssertSqlSafe(trimmed)), + sqlite_pool.as_ref(), + options.max_rows + ) + .await + .map(|(rows, truncated)| (rows, truncated, None)) + } else { + sqlx::query(sqlx::AssertSqlSafe(trimmed)) + .execute(sqlite_pool.as_ref()) + .await + .map(|r| (Vec::new(), false, Some(r.rows_affected()))) + } + }) .await; match result { - Ok(Ok(rows)) => { + Ok(Ok((rows, truncated, affected))) => { if i == statements_ref.len() - 1 { + final_affected = affected; + final_truncated = truncated; if !rows.is_empty() { final_headers = rows[0] .columns() .iter() .map(|c| c.name().to_string()) .collect(); - final_data = rows - .into_iter() - .map(|row| { - (0..row.len()) - .map(|idx| match row.try_get::, _>(idx) { - Ok(Some(v)) => v, - Ok(None) => "NULL".to_string(), - Err(_) => { - if let Ok(val) = row.try_get::(idx) { - val.to_string() - } else if let Ok(val) = row.try_get::(idx) { - val.to_string() - } else if let Ok(val) = row.try_get::(idx) { - val.to_string() - } else { - "[unsupported]".to_string() - } - } - }) - .collect() - }) - .collect(); + final_data = driver_sqlite::convert_sqlite_rows_to_table_data(rows); } else { #[cfg(feature = "query_ast")] if final_headers.is_empty() @@ -1178,7 +1342,9 @@ async fn execute_sqlite_query_job( final_headers = hh; } if final_headers.is_empty() - && trimmed.to_uppercase().starts_with("SELECT") + && strip_leading_sql_comments(trimmed) + .to_uppercase() + .starts_with("SELECT") { let inferred = infer_select_headers(trimmed); if !inferred.is_empty() { @@ -1193,9 +1359,7 @@ async fn execute_sqlite_query_job( return Err(QueryExecutionError::Message(format!("SQLite error: {}", e))); } Err(_) => { - return Err(QueryExecutionError::Message( - "SQLite query timed out".to_string(), - )); + return Err(QueryExecutionError::Message(timeout_message(options))); } } } @@ -1206,6 +1370,8 @@ async fn execute_sqlite_query_job( ast_debug_sql, ast_headers, column_metadata: None, + affected_rows: final_affected, + truncated: final_truncated, }) } @@ -1282,6 +1448,8 @@ async fn execute_redis_query_job( ast_debug_sql: None, ast_headers: None, column_metadata: None, + affected_rows: None, + truncated: false, }), Ok(Ok(None)) => Ok(QueryJobOutput { headers: vec!["Key".to_string(), "Value".to_string()], @@ -1289,6 +1457,8 @@ async fn execute_redis_query_job( ast_debug_sql: None, ast_headers: None, column_metadata: None, + affected_rows: None, + truncated: false, }), _ => Err(QueryExecutionError::Message( "Redis GET timed out or failed".to_string(), @@ -1316,6 +1486,8 @@ async fn execute_redis_query_job( ast_debug_sql: None, ast_headers: None, column_metadata: None, + affected_rows: None, + truncated: false, }) } _ => Err(QueryExecutionError::Message( @@ -1423,6 +1595,8 @@ async fn execute_redis_query_job( ast_debug_sql: None, ast_headers: None, column_metadata: None, + affected_rows: None, + truncated: false, }) } _ => Err(QueryExecutionError::Message( @@ -1456,6 +1630,8 @@ async fn execute_redis_query_job( ast_debug_sql: None, ast_headers: None, column_metadata: None, + affected_rows: None, + truncated: false, }) } _ => Err(QueryExecutionError::Message( @@ -1496,6 +1672,8 @@ async fn execute_redis_query_job( ast_debug_sql: None, ast_headers: None, column_metadata: None, + affected_rows: None, + truncated: false, }) } _ => Err(QueryExecutionError::Message( @@ -1541,6 +1719,8 @@ async fn execute_mssql_query_job( ast_debug_sql: None, ast_headers: None, column_metadata: None, + affected_rows: None, + truncated: false, }), Err(e) => Err(QueryExecutionError::Message(format!("Query error: {}", e))), } @@ -1560,6 +1740,8 @@ async fn execute_mongodb_query_job( ast_debug_sql: None, ast_headers: None, column_metadata: None, + affected_rows: None, + truncated: false, }), _ => Err(QueryExecutionError::Message( "Invalid pool type for MongoDB".to_string(), @@ -1651,6 +1833,15 @@ pub(crate) fn execute_query_with_connection( } } +/// Jalankan query secara sinkron (memblokir thread pemanggil) dan kembalikan +/// `(headers, rows)`. Jika gagal, hasilnya berupa satu kolom `"Error"` berisi +/// pesan error, sesuai kontrak lama yang dipakai pemanggil. +/// +/// Fungsi ini sekarang hanya pembungkus tipis di atas executor async +/// (`execute_query_job`), sehingga pemisahan statement, decoding tipe, +/// timeout, batas baris, dan cancel di server sama persis dengan eksekusi +/// dari editor. Pemanggil baru sebaiknya memakai `prepare_query_job` + +/// `spawn_query_job` supaya UI tidak freeze. pub(crate) fn execute_table_query_sync( tabular: &mut Tabular, connection_id: i64, @@ -1661,777 +1852,71 @@ pub(crate) fn execute_table_query_sync( let runtime = match &tabular.runtime { Some(rt) => rt.clone(), - None => { - debug!("No runtime available, creating temporary one"); - match tokio::runtime::Runtime::new() { - Ok(rt) => Arc::new(rt), - Err(e) => { - debug!("Failed to create runtime: {}", e); - return None; - } + None => match tokio::runtime::Runtime::new() { + Ok(rt) => Arc::new(rt), + Err(e) => { + log::error!("Failed to create runtime for synchronous query: {}", e); + return None; } - } + }, }; - runtime.block_on(async { - match try_get_connection_pool(tabular, connection_id).await { - Some(pool) => { - match pool { - models::enums::DatabasePool::MySQL(_mysql_pool) => { - debug!("Executing MySQL query: {}", query); - - let (target_host, target_port) = match resolve_connection_target_async( - connection, - ) - .await - { - Ok(tuple) => tuple, - Err(err) => { - return Some(( - vec!["Error".to_string()], - vec![vec![format!( - "Failed to resolve MySQL connection target: {}", - err - )]], - )); - } - }; - - let statements: Vec<&str> = query - .split(';') - .map(|s| s.trim()) - .filter(|s| !s.is_empty()) - .collect(); - #[cfg(feature = "query_ast")] - let mut _inferred_headers_from_ast: Option> = None; - #[cfg(feature = "query_ast")] - let statements: Vec = { - let allow_ast_rewrite = statements.len() == 1 - && statements[0].to_uppercase().starts_with("SELECT") - && is_simple_select_statement(statements[0]); - - if allow_ast_rewrite { - let should_paginate = tabular.use_server_pagination - && !query_contains_pagination(statements[0]); - let pagination_opt = if should_paginate { - Some((tabular.current_page as u64, tabular.page_size as u64)) - } else { - None - }; - let inject_auto_limit = should_paginate; - match crate::query_ast::compile_single_select( - statements[0], - &connection.connection_type, - pagination_opt, - inject_auto_limit, - ) { - Ok((new_sql, hdrs)) => { - if !hdrs.is_empty() { - _inferred_headers_from_ast = Some(hdrs.clone()); - } - tabular.last_compiled_sql = Some(new_sql.clone()); - tabular.last_compiled_headers = hdrs.clone(); - if let Ok(plan_txt) = - crate::query_ast::debug_plan(statements[0], &connection.connection_type) - { - tabular.last_debug_plan = Some(plan_txt); - } - let (h, m) = crate::query_ast::cache_stats(); - tabular.last_cache_hits = h; - tabular.last_cache_misses = m; - vec![new_sql] - } - Err(_e) => statements.iter().map(|s| s.to_string()).collect(), - } - } else { - statements.iter().map(|s| s.to_string()).collect() - } - }; - #[cfg(not(feature = "query_ast"))] - let statements: Vec = statements.iter().map(|s| s.to_string()).collect(); - #[cfg(feature = "query_ast")] - let statements: Vec<&str> = statements.iter().map(|s| s.as_str()).collect(); - #[cfg(not(feature = "query_ast"))] - let statements: Vec<&str> = statements.iter().map(|s| s.as_str()).collect(); - debug!("Found {} SQL statements to execute", statements.len()); - - let mut final_headers = Vec::new(); - let mut final_data = Vec::new(); - let (replication_status_mode, master_status_mode) = { - if let Some(active_tab) = tabular.query_tabs.get(tabular.active_tab_index) { - match active_tab.dba_special_mode { - Some(models::enums::DBASpecialMode::ReplicationStatus) => (true, false), - Some(models::enums::DBASpecialMode::MasterStatus) => (false, true), - _ => (false, false), - } - } else { - (false, false) - } - }; - - let mut attempts = 0; - let max_attempts = 3; - while attempts < max_attempts { - attempts += 1; - let mut execution_success = true; - let mut error_message = String::new(); - let encoded_username = modules::url_encode(&connection.username); - let encoded_password = modules::url_encode(&connection.password); - let dsn = format!( - "mysql://{}:{}@{}:{}/{}", - encoded_username, - encoded_password, - target_host, - target_port, - connection.database - ); - let mut conn = match MySqlConnection::connect(&dsn).await { - Ok(c) => c, - Err(e) => { - error_message = e.to_string(); - debug!("Failed to open MySQL connection: {}", error_message); - if attempts >= max_attempts { - break; - } else { - continue; - } - } - }; - let _ = sqlx::query("SET SESSION wait_timeout = 600").execute(&mut conn).await; - let _ = sqlx::query("SET SESSION interactive_timeout = 600").execute(&mut conn).await; - let _ = sqlx::query("SET SESSION net_read_timeout = 120").execute(&mut conn).await; - let _ = sqlx::query("SET SESSION net_write_timeout = 120").execute(&mut conn).await; - let _ = sqlx::query("SET SESSION max_allowed_packet = 1073741824").execute(&mut conn).await; - let _ = sqlx::query("SET SESSION sql_mode = 'TRADITIONAL'").execute(&mut conn).await; - - for (i, statement) in statements.iter().enumerate() { - let trimmed = statement.trim(); - if trimmed.is_empty() - || trimmed.starts_with("--") - || trimmed.starts_with('#') - || trimmed.starts_with("/*") - { - debug!("Skipping statement {}: '{}'", i + 1, trimmed); - continue; - } - debug!("Executing statement {}: '{}'", i + 1, trimmed); - let upper = trimmed.to_uppercase(); - - if upper.starts_with("USE ") { - let db_part = trimmed[3..].trim(); - let db_name = db_part - .trim_matches('`') - .trim_matches('"') - .trim_matches('[') - .trim_matches(']') - .trim(); - - match sqlx::query(sqlx::AssertSqlSafe(format!("USE `{}`", db_name))) - .execute(&mut conn) - .await - { - Ok(_) => { - debug!("✅ Switched MySQL database using USE to '{}'.", db_name); - } - Err(_) => { - debug!("⚠️ USE statement failed, falling back to reconnection..."); - let new_dsn = format!( - "mysql://{}:{}@{}:{}/{}", - encoded_username, - encoded_password, - target_host, - target_port, - db_name - ); - match MySqlConnection::connect(&new_dsn).await { - Ok(new_conn) => { - let mut new_conn = new_conn; - let _ = sqlx::query("SET SESSION wait_timeout = 600").execute(&mut new_conn).await; - let _ = sqlx::query("SET SESSION interactive_timeout = 600").execute(&mut new_conn).await; - let _ = sqlx::query("SET SESSION net_read_timeout = 120").execute(&mut new_conn).await; - let _ = sqlx::query("SET SESSION net_write_timeout = 120").execute(&mut new_conn).await; - let _ = sqlx::query("SET SESSION max_allowed_packet = 1073741824").execute(&mut new_conn).await; - let _ = sqlx::query("SET SESSION sql_mode = 'TRADITIONAL'").execute(&mut new_conn).await; - conn = new_conn; - } - Err(e) => { - error_message = - format!("USE failed (reconnect): {}", e); - break; - } - } - } - } - continue; - } - - let is_admin_command = { - let cmd_upper = trimmed.to_uppercase(); - cmd_upper.starts_with("PURGE BINARY LOGS") - || cmd_upper.starts_with("PURGE MASTER LOGS") - || cmd_upper.starts_with("RESET MASTER") - || cmd_upper.starts_with("RESET SLAVE") - || cmd_upper.starts_with("RESET REPLICA") - || cmd_upper.starts_with("CHANGE MASTER") - || cmd_upper.starts_with("CHANGE REPLICATION SOURCE") - || cmd_upper.starts_with("FLUSH") - }; - - let query_result = tokio::time::timeout( - std::time::Duration::from_secs(60), - sqlx::query(sqlx::AssertSqlSafe(trimmed)).fetch_all(&mut conn), - ) - .await; - - let handle_admin_error = - |e: sqlx::Error| -> Result, sqlx::Error> { - let err_str = e.to_string(); - if err_str.contains("1295") - || err_str.contains("prepared statement protocol") - { - debug!("Admin command executed via sqlx (1295 expected)"); - Ok(vec![]) - } else { - Err(e) - } - }; - - let query_result = query_result.map(|result| { - result.or_else(|e| { - if is_admin_command { - handle_admin_error(e) - } else { - Err(e) - } - }) - }); - - match query_result { - Ok(Ok(rows)) => { - debug!("Query executed successfully: {} rows", rows.len()); - if i == statements.len() - 1 { - if !rows.is_empty() { - final_headers = rows[0] - .columns() - .iter() - .map(|c| c.name().to_string()) - .collect(); - final_data = driver_mysql::convert_mysql_rows_to_table_data(rows); - if replication_status_mode || master_status_mode { - let version_str = match sqlx::query("SELECT VERSION() AS v").fetch_one(&mut conn).await { - Ok(vrow) => vrow.try_get::("v").unwrap_or_default(), - Err(_) => String::new(), - }; - let is_mariadb = version_str.to_lowercase().contains("mariadb"); - if replication_status_mode - && final_data.is_empty() - && let Ok(fallback_rows) = sqlx::query("SHOW SLAVE STATUS").fetch_all(&mut conn).await - && !fallback_rows.is_empty() - { - final_headers = fallback_rows[0].columns().iter().map(|c| c.name().to_string()).collect(); - final_data = driver_mysql::convert_mysql_rows_to_table_data(fallback_rows); - } - if !final_headers.is_empty() && !final_data.is_empty() { - let header_index = |name: &str| final_headers.iter().position(|h| h.eq_ignore_ascii_case(name)); - let mut summary: Vec<(String, String)> = Vec::new(); - if replication_status_mode { - let first = &final_data[0]; - if let Some(idx) = header_index("Replica_IO_Running").or_else(|| header_index("Slave_IO_Running")) { summary.push(("IO Thread".into(), first[idx].clone())); } - if let Some(idx) = header_index("Replica_SQL_Running").or_else(|| header_index("Slave_SQL_Running")) { summary.push(("SQL Thread".into(), first[idx].clone())); } - if let Some(idx) = header_index("Seconds_Behind_Source").or_else(|| header_index("Seconds_Behind_Master")) { summary.push(("Seconds Behind".into(), first[idx].clone())); } - if let Some(idx) = header_index("Channel_Name") { summary.push(("Channel".into(), first[idx].clone())); } - if let Some(idx) = header_index("Retrieved_Gtid_Set") { summary.push(("Retrieved GTID".into(), first[idx].clone())); } - if let Some(idx) = header_index("Executed_Gtid_Set") { summary.push(("Executed GTID".into(), first[idx].clone())); } - } - if master_status_mode { - let first = &final_data[0]; - if let Some(idx) = header_index("File") { summary.push(("Binary Log File".into(), first[idx].clone())); } - if let Some(idx) = header_index("Position") { summary.push(("Position".into(), first[idx].clone())); } - if let Some(idx) = header_index("Binlog_Do_DB") { summary.push(("Binlog Do DB".into(), first[idx].clone())); } - if let Some(idx) = header_index("Binlog_Ignore_DB") { summary.push(("Binlog Ignore DB".into(), first[idx].clone())); } - } - if !summary.is_empty() { - let mut summary_table: Vec> = summary.into_iter().map(|(m, v)| vec![m, v]).collect(); - summary_table.push(vec!["Server Version".into(), version_str.clone()]); - summary_table.push(vec!["Engine".into(), if is_mariadb { "MariaDB".into() } else { "MySQL".into() }]); - final_headers = vec!["Metric".into(), "Value".into()]; - final_data = summary_table; - } - } - } - } else if is_admin_command { - debug!("Admin command executed successfully"); - final_headers = vec!["Status".to_string()]; - final_data = vec![vec!["Command executed successfully".to_string()]]; - } else { - #[cfg(feature = "query_ast")] - if final_headers.is_empty() - && let Some(hh) = _inferred_headers_from_ast.clone() - && !hh.is_empty() - { - final_headers = hh; - } - if trimmed.to_uppercase().starts_with("SELECT") { - let inferred = infer_select_headers(trimmed); - if !inferred.is_empty() { - final_headers = inferred; - } - } - if trimmed.to_uppercase().contains("FROM") { - let words: Vec<&str> = trimmed.split_whitespace().collect(); - if let Some(from_idx) = words.iter().position(|&w| w.to_uppercase() == "FROM") - && let Some(table_name) = words.get(from_idx + 1) - { - let describe_query = format!("DESCRIBE {}", table_name); - match tokio::time::timeout( - std::time::Duration::from_secs(30), - sqlx::query(sqlx::AssertSqlSafe(describe_query.as_str())).fetch_all(&mut conn), - ).await { - Ok(Ok(desc_rows)) => { - if !desc_rows.is_empty() { - final_headers = desc_rows.iter().map(|row| { - row.try_get::(0).unwrap_or_else(|_| "Field".to_string()) - }).collect(); - } - } - _ => { - let info_query = format!("{} LIMIT 0", trimmed); - match tokio::time::timeout( - std::time::Duration::from_secs(30), - sqlx::query(sqlx::AssertSqlSafe(info_query.as_str())).fetch_all(&mut conn), - ).await { - Ok(Ok(info_rows)) => { - if !info_rows.is_empty() { - final_headers = info_rows[0].columns().iter().map(|c| c.name().to_string()).collect(); - } - } - _ => { final_headers = Vec::new(); } - } - } - } - } - } else { - final_headers = Vec::new(); - } - final_data = Vec::new(); - } - } - } - Ok(Err(e)) => { - error_message = e.to_string(); - execution_success = false; - break; - } - Err(_) => { - error_message = "Query timeout after 60s".to_string(); - execution_success = false; - break; - } - } - } + let Some(pool) = runtime.block_on(try_get_connection_pool(tabular, connection_id)) else { + debug!( + "Failed to get connection pool for connection_id: {}", + connection_id + ); + return Some(( + vec!["Error".to_string()], + vec![vec!["Failed to connect to database".to_string()]], + )); + }; - if execution_success { - return Some((final_headers, final_data)); - } else { - debug!("MySQL query failed on attempt {}: {}", attempts, error_message); - if (error_message.contains("timeout") || error_message.contains("pool")) && attempts < max_attempts { - tabular.connection_pools.remove(&connection_id); - continue; - } - if attempts >= max_attempts { - return Some(( - vec!["Error".to_string()], - vec![vec![format!("Query error: {}", error_message)]], - )); - } - } - } + let active_tab = tabular.query_tabs.get(tabular.active_tab_index); + let job_id = tabular.next_query_job_id; + tabular.next_query_job_id = tabular.next_query_job_id.wrapping_add(1); - Some(( - vec!["Error".to_string()], - vec![vec!["Failed to execute query after multiple attempts".to_string()]], - )) - } - models::enums::DatabasePool::PostgreSQL(pg_pool) => { - debug!("Executing PostgreSQL query: {}", query); - let statements: Vec<&str> = query - .split(';') - .map(|s| s.trim()) - .filter(|s| !s.is_empty()) - .collect(); - #[cfg(feature = "query_ast")] - let mut _inferred_headers_from_ast: Option> = None; - #[cfg(feature = "query_ast")] - let statements: Vec = { - let allow_ast_rewrite = statements.len() == 1 - && statements[0].to_uppercase().starts_with("SELECT") - && is_simple_select_statement(statements[0]); - - if allow_ast_rewrite { - let should_paginate = tabular.use_server_pagination - && !query_contains_pagination(statements[0]); - let pagination_opt = if should_paginate { - Some((tabular.current_page as u64, tabular.page_size as u64)) - } else { - None - }; - let inject_auto_limit = should_paginate; - match crate::query_ast::compile_single_select( - statements[0], - &connection.connection_type, - pagination_opt, - inject_auto_limit, - ) { - Ok((new_sql, hdrs)) => { - if !hdrs.is_empty() { - _inferred_headers_from_ast = Some(hdrs.clone()); - } - tabular.last_compiled_sql = Some(new_sql.clone()); - tabular.last_compiled_headers = hdrs.clone(); - if let Ok(plan_txt) = crate::query_ast::debug_plan(statements[0], &connection.connection_type) { - tabular.last_debug_plan = Some(plan_txt); - } - let (h, m) = crate::query_ast::cache_stats(); - tabular.last_cache_hits = h; - tabular.last_cache_misses = m; - vec![new_sql] - } - Err(_) => statements.iter().map(|s| s.to_string()).collect(), - } - } else { - statements.iter().map(|s| s.to_string()).collect() - } - }; - #[cfg(not(feature = "query_ast"))] - let statements: Vec = statements.iter().map(|s| s.to_string()).collect(); - #[cfg(feature = "query_ast")] - let statements: Vec<&str> = statements.iter().map(|s| s.as_str()).collect(); - #[cfg(not(feature = "query_ast"))] - let statements: Vec<&str> = statements.iter().map(|s| s.as_str()).collect(); - debug!("Found {} SQL statements to execute", statements.len()); - - let mut final_headers = Vec::new(); - let mut final_data = Vec::new(); - - for (i, statement) in statements.iter().enumerate() { - let trimmed = statement.trim(); - if trimmed.is_empty() || trimmed.starts_with("--") || trimmed.starts_with("/*") { - continue; - } - match tokio::time::timeout( - std::time::Duration::from_secs(10), - sqlx::query(sqlx::AssertSqlSafe(trimmed)).fetch_all(pg_pool.as_ref()), - ) - .await - { - Ok(Ok(rows)) => { - if i == statements.len() - 1 { - if !rows.is_empty() { - final_headers = rows[0].columns().iter().map(|c| c.name().to_string()).collect(); - final_data = rows.iter().map(|row| { - (0..row.len()).map(|j| match row.try_get::, _>(j) { - Ok(Some(v)) => v, - Ok(None) => "NULL".to_string(), - Err(_) => "Error".to_string(), - }).collect() - }).collect(); - } else { - #[cfg(feature = "query_ast")] - if final_headers.is_empty() - && let Some(hh) = _inferred_headers_from_ast.clone() - && !hh.is_empty() - { - final_headers = hh; - } - if statement.to_uppercase().starts_with("SELECT") { - let inferred = infer_select_headers(statement); - if !inferred.is_empty() { final_headers = inferred; } - } - if statement.to_uppercase().contains("FROM") { - let words: Vec<&str> = statement.split_whitespace().collect(); - if let Some(from_idx) = words.iter().position(|&w| w.to_uppercase() == "FROM") - && let Some(table_name) = words.get(from_idx + 1) - { - let clean_table = table_name.trim_matches('"').trim_matches('`'); - let info_query = format!( - "SELECT column_name FROM information_schema.columns WHERE table_name = '{}' ORDER BY ordinal_position", - clean_table - ); - match tokio::time::timeout( - std::time::Duration::from_secs(10), - sqlx::query(sqlx::AssertSqlSafe(info_query.as_str())).fetch_all(pg_pool.as_ref()), - ).await { - Ok(Ok(info_rows)) => { - final_headers = info_rows.iter().map(|row| { - match row.try_get::(0) { - Ok(col_name) => col_name, - Err(_) => "Column".to_string(), - } - }).collect(); - } - _ => { - let limit_query = format!("{} LIMIT 0", statement); - match tokio::time::timeout( - std::time::Duration::from_secs(10), - sqlx::query(sqlx::AssertSqlSafe(limit_query.as_str())).fetch_all(pg_pool.as_ref()), - ).await { - Ok(Ok(limit_rows)) => { - if !limit_rows.is_empty() { - final_headers = limit_rows[0].columns().iter().map(|c| c.name().to_string()).collect(); - } - } - _ => { - if final_headers.is_empty() { final_headers = infer_select_headers(statement); } - if final_headers.is_empty() { final_headers = Vec::new(); } - } - } - } - } - } - } else { - final_headers = Vec::new(); - } - final_data = Vec::new(); - } - } - } - _ => { - return Some(( - vec!["Error".to_string()], - vec![vec!["Query timed out or failed".to_string()]], - )); - } - } - } - Some((final_headers, final_data)) - } - models::enums::DatabasePool::SQLite(sqlite_pool) => { - debug!("Executing SQLite query: {}", query); - let statements: Vec<&str> = query - .split(';') - .map(|s| s.trim()) - .filter(|s| !s.is_empty()) - .collect(); - #[cfg(feature = "query_ast")] - let mut _inferred_headers_from_ast: Option> = None; - #[cfg(feature = "query_ast")] - let statements: Vec = { - let allow_ast_rewrite = statements.len() == 1 - && statements[0].to_uppercase().starts_with("SELECT") - && is_simple_select_statement(statements[0]); - - if allow_ast_rewrite { - let should_paginate = tabular.use_server_pagination - && !query_contains_pagination(statements[0]); - let pagination_opt = if should_paginate { - Some((tabular.current_page as u64, tabular.page_size as u64)) - } else { - None - }; - let inject_auto_limit = should_paginate; - match crate::query_ast::compile_single_select( - statements[0], - &connection.connection_type, - pagination_opt, - inject_auto_limit, - ) { - Ok((new_sql, hdrs)) => { - if !hdrs.is_empty() { - _inferred_headers_from_ast = Some(hdrs.clone()); - } - vec![new_sql] - } - Err(_) => statements.iter().map(|s| s.to_string()).collect(), - } - } else { - statements.iter().map(|s| s.to_string()).collect() - } - }; - #[cfg(not(feature = "query_ast"))] - let statements: Vec = statements.iter().map(|s| s.to_string()).collect(); - #[cfg(feature = "query_ast")] - let statements: Vec<&str> = statements.iter().map(|s| s.as_str()).collect(); - #[cfg(not(feature = "query_ast"))] - let statements: Vec<&str> = statements.iter().map(|s| s.as_str()).collect(); - debug!("Found {} SQL statements to execute", statements.len()); - - let mut final_headers = Vec::new(); - let mut final_data = Vec::new(); - - for (i, statement) in statements.iter().enumerate() { - let trimmed = statement.trim(); - if trimmed.is_empty() || trimmed.starts_with("--") || trimmed.starts_with("/*") { - continue; - } - match tokio::time::timeout( - std::time::Duration::from_secs(10), - sqlx::query(sqlx::AssertSqlSafe(trimmed)).fetch_all(sqlite_pool.as_ref()), - ) - .await - { - Ok(Ok(rows)) => { - if i == statements.len() - 1 { - if !rows.is_empty() { - final_headers = rows[0].columns().iter().map(|c| c.name().to_string()).collect(); - final_data = driver_sqlite::convert_sqlite_rows_to_table_data(rows); - } else { - #[cfg(feature = "query_ast")] - if final_headers.is_empty() - && let Some(hh) = _inferred_headers_from_ast.clone() - && !hh.is_empty() - { - final_headers = hh; - } - if statement.to_uppercase().starts_with("SELECT") { - let inferred = infer_select_headers(statement); - if !inferred.is_empty() { final_headers = inferred; } - } - if statement.to_uppercase().contains("FROM") { - let words: Vec<&str> = statement.split_whitespace().collect(); - if let Some(from_idx) = words.iter().position(|&w| w.to_uppercase() == "FROM") - && let Some(table_name) = words.get(from_idx + 1) - { - let clean_table = table_name.trim_matches('"').trim_matches('`').trim_matches('[').trim_matches(']'); - let pragma_query = format!("PRAGMA table_info(\"{}\")", clean_table.replace('\"', "\"\"")); - match tokio::time::timeout( - std::time::Duration::from_secs(10), - sqlx::query(sqlx::AssertSqlSafe(pragma_query.as_str())).fetch_all(sqlite_pool.as_ref()), - ).await { - Ok(Ok(pragma_rows)) => { - final_headers = pragma_rows.iter().map(|row| { - match row.try_get::(1) { - Ok(col_name) => col_name, - Err(_) => "Column".to_string(), - } - }).collect(); - } - _ => { - let limit_query = format!("{} LIMIT 0", statement); - match tokio::time::timeout( - std::time::Duration::from_secs(10), - sqlx::query(sqlx::AssertSqlSafe(limit_query.as_str())).fetch_all(sqlite_pool.as_ref()), - ).await { - Ok(Ok(limit_rows)) => { - if !limit_rows.is_empty() { - final_headers = limit_rows[0].columns().iter().map(|c| c.name().to_string()).collect(); - } - } - _ => { final_headers = Vec::new(); } - } - } - } - } - } else { - final_headers = Vec::new(); - } - final_data = Vec::new(); - } - } - } - _ => { - return Some(( - vec!["Error".to_string()], - vec![vec!["Query timed out or failed".to_string()]], - )); - } - } - } - Some((final_headers, final_data)) - } - models::enums::DatabasePool::Redis(redis_manager) => { - debug!("Executing Redis command: {}", query); - let mut conn = redis_manager.as_ref().clone(); - use redis::AsyncCommands; - - let parts: Vec<&str> = query.split_whitespace().collect(); - if parts.is_empty() { - return Some(( - vec!["Error".to_string()], - vec![vec!["Empty command".to_string()]], - )); - } + let job = QueryJob { + job_id, + tab_id: active_tab.map(|t| t.id), + options: QueryExecutionOptions { + connection_id, + connection: connection.clone(), + query: query.to_string(), + selected_database: active_tab + .and_then(|t| t.database_name.clone()) + .filter(|s| !s.trim().is_empty()), + use_server_pagination: tabular.use_server_pagination, + current_page: tabular.current_page, + page_size: tabular.page_size, + base_query: None, + dba_special_mode: active_tab.and_then(|t| t.dba_special_mode.clone()), + save_to_history: false, + ast_enabled: cfg!(feature = "query_ast"), + job_id, + query_timeout: (tabular.query_timeout_secs > 0) + .then(|| std::time::Duration::from_secs(tabular.query_timeout_secs as u64)), + max_rows: tabular.max_result_rows.max(1) as usize, + backend_pids: tabular.query_backend_pids.clone(), + }, + connection_pool: pool, + started_at: Instant::now(), + }; - match parts[0].to_uppercase().as_str() { - "GET" => { - if parts.len() != 2 { - return Some((vec!["Error".to_string()], vec![vec!["GET requires exactly one key".to_string()]])); - } - match tokio::time::timeout(std::time::Duration::from_secs(10), conn.get::<&str, Option>(parts[1])).await { - Ok(Ok(Some(value))) => Some((vec!["Key".to_string(), "Value".to_string()], vec![vec![parts[1].to_string(), value]])), - Ok(Ok(None)) => Some((vec!["Key".to_string(), "Value".to_string()], vec![vec![parts[1].to_string(), "NULL".to_string()]])), - _ => Some((vec!["Error".to_string()], vec![vec!["Redis GET timed out or failed".to_string()]])), - } - } - "KEYS" => { - if parts.len() != 2 { - return Some((vec!["Error".to_string()], vec![vec!["KEYS requires exactly one pattern".to_string()]])); - } - match tokio::time::timeout(std::time::Duration::from_secs(10), conn.keys::<&str, Vec>(parts[1])).await { - Ok(Ok(keys)) => Some((vec!["Key".to_string()], keys.into_iter().map(|k| vec![k]).collect())), - _ => Some((vec!["Error".to_string()], vec![vec!["Redis KEYS timed out or failed".to_string()]])), - } - } - "INFO" => { - let section = if parts.len() > 1 { parts[1] } else { "default" }; - match tokio::time::timeout(std::time::Duration::from_secs(10), redis::cmd("INFO").arg(section).query_async::(&mut conn)).await { - Ok(Ok(info_result)) => { - let mut table_data = Vec::new(); - for line in info_result.lines() { - if line.trim().is_empty() || line.starts_with('#') { continue; } - if let Some((key, value)) = line.split_once(':') { table_data.push(vec![key.to_string(), value.to_string()]); } - } - Some((vec!["Property".to_string(), "Value".to_string()], table_data)) - } - _ => Some((vec!["Error".to_string()], vec![vec!["Redis INFO timed out or failed".to_string()]])), - } - } - "HGETALL" => { - if parts.len() != 2 { return Some((vec!["Error".to_string()], vec![vec!["HGETALL requires exactly one key".to_string()]])); } - match tokio::time::timeout(std::time::Duration::from_secs(10), redis::cmd("HGETALL").arg(parts[1]).query_async::>(&mut conn)).await { - Ok(Ok(hash_data)) => { - let mut table_data = Vec::new(); - for chunk in hash_data.chunks(2) { if chunk.len() == 2 { table_data.push(vec![chunk[0].clone(), chunk[1].clone()]); } } - if table_data.is_empty() { table_data.push(vec!["No data".to_string(), "Hash is empty or key does not exist".to_string()]); } - Some((vec!["Field".to_string(), "Value".to_string()], table_data)) - } - _ => Some((vec!["Error".to_string()], vec![vec!["Redis HGETALL timed out or failed".to_string()]])), - } - } - _ => Some((vec!["Error".to_string()], vec![vec![format!("Unsupported Redis command: {}", parts[0])]])), - } - } - models::enums::DatabasePool::MsSQL(mssql_cfg) => { - debug!("Executing MsSQL query: {}", query); - let mut query_str = query.to_string(); - if query_str.contains("TOP") && query_str.contains("ROWS FETCH NEXT") { - query_str = query_str.replace("TOP 10000", ""); - } - match driver_mssql::execute_query(mssql_cfg.clone(), &query_str).await { - Ok((h, d)) => Some((h, d)), - Err(e) => Some(( - vec!["Error".to_string()], - vec![vec![format!("Query error: {}", e)]], - )), - } - } - models::enums::DatabasePool::MongoDB(_client) => Some(( - vec!["Info".to_string()], - vec![vec!["MongoDB query execution is not supported. Use tree to browse collections.".to_string()]], - )), - } - } - None => { - debug!( - "Failed to get connection pool for connection_id: {}", - connection_id - ); - Some(( - vec!["Error".to_string()], - vec![vec!["Failed to connect to database".to_string()]], - )) - } - } - }) + let message = runtime.block_on(execute_query_job(job)); + if let Some(sql) = message.ast_debug_sql.clone() { + tabular.last_compiled_sql = Some(sql); + } + if let Some(headers) = message.ast_headers.clone() { + tabular.last_compiled_headers = headers; + } + if message.truncated { + tabular.toasts.warning(format!( + "Result truncated to the first {} rows.", + message.rows.len() + )); + } + Some((message.headers, message.rows)) } /// Execute multiple queries concurrently (non-blocking for slow connections). @@ -2470,3 +1955,89 @@ pub(crate) async fn execute_multiple_queries_concurrently( results } + +#[cfg(test)] +mod tests { + use super::*; + + async fn sqlite_job(query: &str, max_rows: usize) -> QueryResultMessage { + let pool = sqlx::sqlite::SqlitePoolOptions::new() + .max_connections(1) + .connect("sqlite::memory:") + .await + .expect("in-memory sqlite"); + for stmt in [ + "CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT)", + "INSERT INTO t (name) VALUES ('a'), ('b;c'), ('d')", + ] { + sqlx::query(stmt).execute(&pool).await.expect("seed"); + } + let connection = models::structs::ConnectionConfig { + connection_type: models::enums::DatabaseType::SQLite, + ..Default::default() + }; + let job = QueryJob { + job_id: 7, + tab_id: Some(3), + options: QueryExecutionOptions { + connection_id: 1, + connection, + query: query.to_string(), + selected_database: None, + use_server_pagination: false, + current_page: 0, + page_size: 100, + base_query: None, + dba_special_mode: None, + save_to_history: false, + ast_enabled: false, + job_id: 7, + query_timeout: None, + max_rows, + backend_pids: Default::default(), + }, + connection_pool: models::enums::DatabasePool::SQLite(Arc::new(pool)), + started_at: Instant::now(), + }; + execute_query_job(job).await + } + + #[tokio::test] + async fn statement_with_leading_comment_is_executed() { + let msg = sqlite_job("-- ambil semua\nSELECT name FROM t ORDER BY id", 100).await; + assert!(msg.success, "{:?}", msg.error); + assert_eq!(msg.tab_id, Some(3)); + assert_eq!(msg.rows.len(), 3); + assert_eq!(msg.affected_rows, None); + } + + #[tokio::test] + async fn semicolon_inside_string_is_not_split() { + let msg = sqlite_job("SELECT id FROM t WHERE name = 'b;c'", 100).await; + assert!(msg.success, "{:?}", msg.error); + assert_eq!(msg.rows, vec![vec!["2".to_string()]]); + } + + #[tokio::test] + async fn update_reports_driver_affected_rows() { + let msg = sqlite_job("UPDATE t SET name = 'z' WHERE id >= 2", 100).await; + assert!(msg.success, "{:?}", msg.error); + assert_eq!(msg.affected_rows, Some(2)); + assert!(msg.rows.is_empty()); + } + + #[tokio::test] + async fn result_set_is_truncated_at_row_limit() { + let msg = sqlite_job("SELECT * FROM t", 2).await; + assert!(msg.success, "{:?}", msg.error); + assert_eq!(msg.rows.len(), 2); + assert!(msg.truncated); + } + + #[tokio::test] + async fn sql_error_is_reported_as_failure() { + let msg = sqlite_job("SELECT * FROM missing_table", 100).await; + assert!(!msg.success); + assert!(msg.error.unwrap_or_default().contains("missing_table")); + } +} diff --git a/src/connection/session.rs b/src/connection/session.rs index 41c3fffc..eb34b608 100644 --- a/src/connection/session.rs +++ b/src/connection/session.rs @@ -86,6 +86,7 @@ pub fn spawn_session( return None; }; + let tab_id = tabular.query_tabs.get(tabular.active_tab_index).map(|t| t.id); let runtime = tabular.runtime.clone()?; let result_sender = tabular.query_result_sender.clone(); let (tx, rx) = tokio::sync::mpsc::unbounded_channel(); @@ -93,6 +94,7 @@ pub fn spawn_session( let handle = runtime.spawn(run_session( pool, connection_type, + tab_id, connection_id, database_name, rx, @@ -109,6 +111,7 @@ pub fn spawn_session( async fn run_session( pool: models::enums::DatabasePool, connection_type: models::enums::DatabaseType, + tab_id: Option, connection_id: i64, database_name: Option, mut rx: tokio::sync::mpsc::UnboundedReceiver, @@ -127,6 +130,7 @@ async fn run_session( Err(e) => { let _ = result_sender.send(session_message( job_id, + tab_id, connection_id, &sql, Err(format!("Cannot open session connection: {}", e)), @@ -148,6 +152,7 @@ async fn run_session( if let Err(e) = run_simple(c, begin).await { let _ = result_sender.send(session_message( job_id, + tab_id, connection_id, &sql, Err(format!("BEGIN failed: {}", e)), @@ -158,9 +163,10 @@ async fn run_session( tx_open = true; } - let outcome = run_query(c, &sql).await; + let outcome = run_statement(c, &sql).await; let _ = result_sender.send(session_message( job_id, + tab_id, connection_id, &sql, outcome, @@ -169,9 +175,12 @@ async fn run_session( } SessionCommand::Commit { job_id } => { let started = Instant::now(); - let outcome = finish_tx(conn.as_mut(), &mut tx_open, "COMMIT").await; + let outcome = finish_tx(conn.as_mut(), &mut tx_open, "COMMIT") + .await + .map(|(h, r)| (h, r, None)); let _ = result_sender.send(session_message( job_id, + tab_id, connection_id, "COMMIT", outcome, @@ -180,9 +189,12 @@ async fn run_session( } SessionCommand::Rollback { job_id } => { let started = Instant::now(); - let outcome = finish_tx(conn.as_mut(), &mut tx_open, "ROLLBACK").await; + let outcome = finish_tx(conn.as_mut(), &mut tx_open, "ROLLBACK") + .await + .map(|(h, r)| (h, r, None)); let _ = result_sender.send(session_message( job_id, + tab_id, connection_id, "ROLLBACK", outcome, @@ -291,6 +303,46 @@ async fn run_simple(conn: &mut SessionConn, sql: &str) -> Result<(), String> { } } +/// Hasil satu statement di sesi: header, baris, dan jumlah baris terdampak +/// (Some hanya untuk statement pengubah data). +type StatementOutput = (Vec, Vec>, Option); + +/// Jalankan satu statement di koneksi sesi. Statement pengubah data dijalankan +/// lewat `execute()` supaya jumlah baris terdampak dari driver bisa dilaporkan. +async fn run_statement(conn: &mut SessionConn, sql: &str) -> Result { + if !crate::connection::sql::statement_returns_rows(sql) { + let affected = match conn { + SessionConn::MySql(c) => Some( + sqlx::query(sqlx::AssertSqlSafe(sql)) + .execute(&mut **c) + .await + .map_err(|e| e.to_string())? + .rows_affected(), + ), + SessionConn::Postgres(c) => Some( + sqlx::query(sqlx::AssertSqlSafe(sql)) + .execute(&mut **c) + .await + .map_err(|e| e.to_string())? + .rows_affected(), + ), + SessionConn::Sqlite(c) => Some( + sqlx::query(sqlx::AssertSqlSafe(sql)) + .execute(&mut **c) + .await + .map_err(|e| e.to_string())? + .rows_affected(), + ), + // Driver MsSQL mengembalikan hasil lewat jalur query biasa. + SessionConn::MsSQL(_) => None, + }; + if let Some(n) = affected { + return Ok((Vec::new(), Vec::new(), Some(n))); + } + } + run_query(conn, sql).await.map(|(h, r)| (h, r, None)) +} + async fn run_query( conn: &mut SessionConn, sql: &str, @@ -319,29 +371,10 @@ async fn run_query( .first() .map(|r| r.columns().iter().map(|c| c.name().to_string()).collect()) .unwrap_or_default(); - let data = rows - .into_iter() - .map(|row| { - (0..row.len()) - .map(|idx| match row.try_get::, _>(idx) { - Ok(Some(v)) => v, - Ok(None) => "NULL".to_string(), - Err(_) => { - if let Ok(val) = row.try_get::(idx) { - val.to_string() - } else if let Ok(val) = row.try_get::(idx) { - val.to_string() - } else if let Ok(val) = row.try_get::(idx) { - val.to_string() - } else { - "[unsupported]".to_string() - } - } - }) - .collect() - }) - .collect(); - Ok((headers, data)) + Ok(( + headers, + crate::driver_postgres::convert_postgres_rows_to_table_data(rows), + )) } SessionConn::Sqlite(c) => { let rows = sqlx::query(sqlx::AssertSqlSafe(sql)) @@ -368,17 +401,20 @@ async fn run_query( fn session_message( job_id: u64, + tab_id: Option, connection_id: i64, query: &str, - outcome: Result<(Vec, Vec>), String>, + outcome: Result, started: Instant, ) -> QueryResultMessage { match outcome { - Ok((headers, rows)) => QueryResultMessage { + Ok((headers, rows, affected)) => QueryResultMessage { job_id, + tab_id, connection_id, success: true, - affected_rows: Some(rows.len()), + affected_rows: affected.map(|n| n as usize), + truncated: false, headers, rows, error: None, @@ -391,6 +427,7 @@ fn session_message( }, Err(message) => QueryResultMessage { job_id, + tab_id, connection_id, success: false, headers: vec!["Error".to_string()], @@ -403,6 +440,7 @@ fn session_message( ast_headers: None, affected_rows: None, column_metadata: None, + truncated: false, }, } } diff --git a/src/connection/sql.rs b/src/connection/sql.rs index 566295a2..c125e75a 100644 --- a/src/connection/sql.rs +++ b/src/connection/sql.rs @@ -339,15 +339,11 @@ pub fn should_enable_auto_pagination(sql: &str) -> bool { } let mut simple_select_count = 0; - for stmt in sql.split(';') { - let trimmed = stmt.trim(); - if trimmed.is_empty() { - continue; - } - - if trimmed.trim_start().starts_with(['-', '#']) { + for stmt in split_sql_statements(sql, false) { + if is_comment_only_statement(&stmt) { continue; } + let trimmed = strip_leading_sql_comments(&stmt); if trimmed.to_uppercase().starts_with("SELECT") { let is_simple = is_simple_select_statement(trimmed); @@ -633,10 +629,83 @@ pub fn split_sql_statements(sql: &str, hash_is_comment: bool) -> Vec { statements } +/// Lewati spasi, komentar baris `-- …` / `# …`, dan komentar blok `/* … */` di +/// awal statement supaya statement bisa diklasifikasi dari keyword pertamanya. +pub fn strip_leading_sql_comments(sql: &str) -> &str { + let mut rest = sql.trim_start(); + loop { + if let Some(after) = rest.strip_prefix("--").or_else(|| rest.strip_prefix('#')) { + rest = match after.find('\n') { + Some(pos) => after[pos + 1..].trim_start(), + None => "", + }; + } else if let Some(after) = rest.strip_prefix("/*") { + rest = match after.find("*/") { + Some(pos) => after[pos + 2..].trim_start(), + None => "", + }; + } else { + return rest; + } + } +} + +/// True jika statement hanya berisi komentar dan spasi. +pub fn is_comment_only_statement(sql: &str) -> bool { + strip_leading_sql_comments(sql).trim_end_matches(';').trim().is_empty() +} + +/// Menentukan apakah statement diharapkan menghasilkan result set. +/// +/// Perubahan data/skema tanpa `RETURNING`/`OUTPUT` dijalankan lewat `execute()` +/// agar jumlah baris terdampak dari driver bisa dilaporkan. Selain itu (termasuk +/// statement yang tidak bisa diklasifikasi) diambil sebagai baris, yang selalu aman. +pub fn statement_returns_rows(sql: &str) -> bool { + const MODIFYING: &[&str] = &[ + "INSERT", "UPDATE", "DELETE", "REPLACE", "MERGE", "UPSERT", "TRUNCATE", "CREATE", + "ALTER", "DROP", "GRANT", "REVOKE", "RENAME", "COMMENT", + ]; + let body = strip_leading_sql_comments(sql); + let first = body + .split(|c: char| !c.is_ascii_alphabetic()) + .next() + .unwrap_or("") + .to_ascii_uppercase(); + if !MODIFYING.contains(&first.as_str()) { + return true; + } + body.split(|c: char| !(c.is_ascii_alphanumeric() || c == '_')) + .any(|word| word.eq_ignore_ascii_case("RETURNING") || word.eq_ignore_ascii_case("OUTPUT")) +} + #[cfg(test)] mod tests { use super::*; + #[test] + fn leading_comments_are_stripped() { + assert_eq!(strip_leading_sql_comments("-- note\nSELECT 1"), "SELECT 1"); + assert_eq!(strip_leading_sql_comments("/* a */ /* b */\n UPDATE t"), "UPDATE t"); + assert_eq!(strip_leading_sql_comments("# mysql\nDELETE FROM t"), "DELETE FROM t"); + assert!(is_comment_only_statement("-- just a note")); + assert!(is_comment_only_statement("/* unterminated")); + assert!(!is_comment_only_statement("-- note\nSELECT 1")); + } + + #[test] + fn classifies_row_returning_statements() { + assert!(statement_returns_rows("SELECT * FROM t")); + assert!(statement_returns_rows("-- c\nWITH x AS (SELECT 1) SELECT * FROM x")); + assert!(statement_returns_rows("SHOW TABLES")); + assert!(statement_returns_rows("EXPLAIN UPDATE t SET a = 1")); + assert!(!statement_returns_rows("update t set a = 1 where id = 2")); + assert!(!statement_returns_rows("/* bulk */ INSERT INTO t VALUES (1)")); + assert!(!statement_returns_rows("CREATE TABLE t (id int)")); + assert!(statement_returns_rows("INSERT INTO t VALUES (1) RETURNING id")); + assert!(statement_returns_rows("DELETE FROM t OUTPUT deleted.id")); + assert!(!statement_returns_rows("UPDATE t SET returning_customer = 1")); + } + #[test] fn simple_select_allows_auto_pagination() { assert!(should_enable_auto_pagination("SELECT * FROM users")); diff --git a/src/connection/types.rs b/src/connection/types.rs index 989eb80e..2903bc15 100644 --- a/src/connection/types.rs +++ b/src/connection/types.rs @@ -1,6 +1,40 @@ use crate::models; use std::time::Instant; +/// Id sesi di sisi server (backend pid PostgreSQL / connection id MySQL) untuk +/// setiap job query yang sedang berjalan, dengan key job id. Dipakai supaya +/// permintaan cancel benar-benar menghentikan statement di server, bukan hanya +/// meninggalkannya di sisi klien. +pub type BackendPidRegistry = + std::sync::Arc>>; + +/// Menghapus backend pid sebuah job dari registry saat job selesai atau +/// task-nya di-abort. +pub struct BackendPidGuard { + registry: BackendPidRegistry, + job_id: u64, +} + +impl BackendPidGuard { + pub fn register(registry: &BackendPidRegistry, job_id: u64, pid: i64) -> Self { + if let Ok(mut map) = registry.lock() { + map.insert(job_id, pid); + } + Self { + registry: registry.clone(), + job_id, + } + } +} + +impl Drop for BackendPidGuard { + fn drop(&mut self) { + if let Ok(mut map) = self.registry.lock() { + map.remove(&self.job_id); + } + } +} + #[derive(Clone, Debug)] pub struct QueryExecutionOptions { pub connection_id: i64, @@ -14,11 +48,19 @@ pub struct QueryExecutionOptions { pub dba_special_mode: Option, pub save_to_history: bool, pub ast_enabled: bool, + pub job_id: u64, + /// Batalkan statement setelah durasi ini (None = tanpa batas). + pub query_timeout: Option, + /// Berhenti membaca result set setelah jumlah baris ini. + pub max_rows: usize, + pub backend_pids: BackendPidRegistry, } #[derive(Clone)] pub struct QueryJob { pub job_id: u64, + /// `QueryTab::id` milik tab yang menjalankan job ini (None jika tidak ada tab aktif). + pub tab_id: Option, pub options: QueryExecutionOptions, pub connection_pool: models::enums::DatabasePool, pub started_at: Instant, @@ -36,6 +78,8 @@ pub struct QueryJobStatus { #[derive(Debug, Clone)] pub struct QueryResultMessage { pub job_id: u64, + /// `QueryTab::id` milik tab yang menjalankan job; hasil dikirim ke tab ini. + pub tab_id: Option, pub connection_id: i64, pub success: bool, pub headers: Vec, @@ -48,6 +92,8 @@ pub struct QueryResultMessage { pub ast_headers: Option>, pub affected_rows: Option, // Number of affected rows for INSERT/UPDATE/DELETE pub column_metadata: Option>, + /// True jika result set dipotong karena mencapai batas baris. + pub truncated: bool, } #[derive(Debug, Clone)] @@ -57,6 +103,9 @@ pub struct QueryJobOutput { pub ast_debug_sql: Option, pub ast_headers: Option>, pub column_metadata: Option>, + /// Jumlah baris terdampak dari driver jika statement terakhir mengubah data. + pub affected_rows: Option, + pub truncated: bool, } #[derive(Debug)] diff --git a/src/data_table/export_clipboard.rs b/src/data_table/export_clipboard.rs index bb174dba..25a4b798 100644 --- a/src/data_table/export_clipboard.rs +++ b/src/data_table/export_clipboard.rs @@ -1,5 +1,5 @@ -/// Utilities for exporting query result sets to clipboard in various formats: -/// Markdown Table, JSON, CSV, and SQL INSERT statements. +//! Utilities for exporting query result sets to clipboard in various formats: +//! Markdown Table, JSON, CSV, and SQL INSERT statements. pub fn format_as_markdown_table(headers: &[String], rows: &[Vec]) -> String { if headers.is_empty() { diff --git a/src/driver_postgres.rs b/src/driver_postgres.rs index e31d033c..9985d187 100644 --- a/src/driver_postgres.rs +++ b/src/driver_postgres.rs @@ -300,3 +300,100 @@ pub(crate) fn fetch_tables_from_postgres_connection( } }) } + +/// Mengubah satu nilai PostgreSQL menjadi teks tampilan. +/// +/// sqlx mengecek kompatibilitas tipe secara ketat (kolom `INT4` tidak bisa dibaca +/// sebagai `i64`, `NUMERIC` tidak bisa sebagai `String`), jadi setiap keluarga tipe +/// di-decode dengan tipe Rust yang sesuai. Tipe yang tidak dikenal memakai byte +/// mentah dari protokol. +fn pg_value_to_string(row: &sqlx::postgres::PgRow, idx: usize) -> String { + use sqlx::{Column, TypeInfo, ValueRef}; + + fn show(v: Result, sqlx::Error>) -> Option { + v.ok().map(|o| o.map(|x| x.to_string()).unwrap_or_else(|| "NULL".to_string())) + } + fn show_array(v: Result>>, sqlx::Error>) -> Option { + v.ok().map(|o| match o { + None => "NULL".to_string(), + Some(items) => format!( + "{{{}}}", + items + .iter() + .map(|i| i.as_ref().map(|x| x.to_string()).unwrap_or_else(|| "NULL".to_string())) + .collect::>() + .join(",") + ), + }) + } + + match row.try_get_raw(idx) { + Ok(raw) if raw.is_null() => return "NULL".to_string(), + Err(e) => return format!("[error: {}]", e), + Ok(_) => {} + } + + let type_name = row.columns()[idx].type_info().name().to_ascii_uppercase(); + let decoded = match type_name.as_str() { + "BOOL" => show(row.try_get::, _>(idx)), + "INT2" | "SMALLINT" | "SMALLSERIAL" => show(row.try_get::, _>(idx)), + "INT4" | "INT" | "SERIAL" => show(row.try_get::, _>(idx)), + "INT8" | "BIGINT" | "BIGSERIAL" => show(row.try_get::, _>(idx)), + "OID" => show(row.try_get::, _>(idx).map(|o| o.map(|v| v.0))), + "FLOAT4" | "REAL" => show(row.try_get::, _>(idx)), + "FLOAT8" | "DOUBLE PRECISION" => show(row.try_get::, _>(idx)), + "NUMERIC" => show(row.try_get::, _>(idx)), + "TIMESTAMP" => show(row.try_get::, _>(idx)), + "TIMESTAMPTZ" => show(row.try_get::>, _>(idx)), + "DATE" => show(row.try_get::, _>(idx)), + "TIME" => show(row.try_get::, _>(idx)), + "JSON" | "JSONB" => show(row.try_get::, _>(idx)), + "BYTEA" => row.try_get::>, _>(idx).ok().map(|o| match o { + None => "NULL".to_string(), + Some(b) => format!("\\x{}", hex::encode(b)), + }), + "UUID" => row.try_get_raw(idx).ok().and_then(|raw| { + let bytes = raw.as_bytes().ok()?; + (bytes.len() == 16).then(|| { + let h = hex::encode(bytes); + format!("{}-{}-{}-{}-{}", &h[0..8], &h[8..12], &h[12..16], &h[16..20], &h[20..32]) + }) + }), + "INT2[]" => show_array(row.try_get::>>, _>(idx)), + "INT4[]" => show_array(row.try_get::>>, _>(idx)), + "INT8[]" => show_array(row.try_get::>>, _>(idx)), + "FLOAT8[]" => show_array(row.try_get::>>, _>(idx)), + "BOOL[]" => show_array(row.try_get::>>, _>(idx)), + "TEXT[]" | "VARCHAR[]" | "NAME[]" | "BPCHAR[]" => { + show_array(row.try_get::>>, _>(idx)) + } + _ => None, + }; + if let Some(text) = decoded { + return text; + } + + // Tipe mirip teks (TEXT, VARCHAR, NAME, CITEXT, enum, …) di-decode sebagai String. + if let Ok(v) = row.try_get_unchecked::, _>(idx) + && let Some(s) = v + { + return s; + } + match row.try_get_raw(idx).ok().and_then(|raw| raw.as_bytes().ok()) { + Some(bytes) => match std::str::from_utf8(bytes) { + Ok(s) if s.chars().all(|c| !c.is_control() || c.is_whitespace()) => s.to_string(), + _ => format!("\\x{}", hex::encode(bytes)), + }, + None => format!("[unsupported {}]", type_name), + } +} + +/// Mengubah baris PostgreSQL menjadi string tampilan, dengan men-decode setiap +/// kolom memakai tipe aslinya (lihat [`pg_value_to_string`]). +pub(crate) fn convert_postgres_rows_to_table_data( + rows: Vec, +) -> Vec> { + rows.iter() + .map(|row| (0..row.len()).map(|idx| pg_value_to_string(row, idx)).collect()) + .collect() +} diff --git a/src/editor.rs b/src/editor.rs index 29ec09e0..62336d65 100644 --- a/src/editor.rs +++ b/src/editor.rs @@ -24,6 +24,7 @@ pub(crate) fn create_new_tab( tabular.next_tab_id += 1; let new_tab = models::structs::QueryTab { + id: tab_id, title, content: content.clone(), file_path: None, @@ -7189,12 +7190,12 @@ pub(crate) fn execute_query_bypass_checks(tabular: &mut window_egui::Tabular, qu } Err(err) => { tabular.active_query_jobs.remove(&job_id); - debug!("Failed to spawn async job: {:?}", err); + report_query_start_failure(tabular, &err); } } } Err(err) => { - debug!("Failed to prepare async job: {:?}", err); + report_query_start_failure(tabular, &err); } } } else { @@ -7226,7 +7227,14 @@ pub(crate) fn execute_query_bypass_checks(tabular: &mut window_egui::Tabular, qu jobs.push(job); } Err(err) => { - debug!("Failed to prepare statement {}/{}: {:?}", idx + 1, total, err); + // Tanpa statement ini urutan script jadi tidak utuh, + // jadi batalkan seluruh batch daripada menjalankan sebagian. + for job_id in &job_ids { + tabular.active_query_jobs.remove(job_id); + } + log::warn!("Failed to prepare statement {}/{}: {:?}", idx + 1, total, err); + report_query_start_failure(tabular, &err); + return; } } } @@ -7252,14 +7260,36 @@ pub(crate) fn execute_query_bypass_checks(tabular: &mut window_egui::Tabular, qu for job_id in &job_ids { tabular.active_query_jobs.remove(job_id); } - tabular.query_execution_in_progress = false; - debug!("Failed to spawn batch job: {:?}", err); + report_query_start_failure(tabular, &err); } } } } } +/// Tampilkan alasan query gagal dimulai dan kembalikan status eksekusi ke idle. +/// Sebelumnya kegagalan ini hanya masuk ke log debug, sehingga tombol Run +/// terlihat tidak melakukan apa-apa dan spinner bisa terus berputar. +fn report_query_start_failure( + tabular: &mut window_egui::Tabular, + err: &connection::types::QueryPreparationError, +) { + use connection::types::QueryPreparationError as E; + let reason = match err { + E::ConnectionNotFound => "the connection for this tab no longer exists", + E::PoolUnavailable => "the database connection is not open yet — try again in a moment", + E::RuntimeUnavailable => "the background runtime is not available", + E::UnsupportedDatabase => "this database type does not support running queries here", + }; + log::warn!("Query could not be started: {:?}", err); + tabular.toasts.error(format!("Query could not be started: {}", reason)); + if tabular.active_query_jobs.is_empty() { + tabular.query_execution_in_progress = false; + tabular.current_table_name.clear(); + tabular.extend_query_icon_hold(); + } +} + /// Send statements to the active tab's dedicated session connection /// (manual-commit mode), creating or replacing the session as needed. fn execute_statements_in_session( diff --git a/src/models/structs.rs b/src/models/structs.rs index 374984c8..2914adba 100644 --- a/src/models/structs.rs +++ b/src/models/structs.rs @@ -705,6 +705,10 @@ pub struct QueryResult { #[derive(Clone, Debug)] pub struct QueryTab { + /// Identitas tab yang stabil. Berbeda dengan index di `query_tabs`, id ini + /// tidak berubah saat tab diurutkan ulang atau tab lain ditutup, sehingga + /// hasil query async bisa dikembalikan ke tab yang menjalankannya. + pub id: usize, pub title: String, pub content: String, pub file_path: Option, @@ -1786,6 +1790,7 @@ mod tests { #[test] fn test_query_tab_pinning() { let mut tab = QueryTab { + id: 1, title: "Test Tab".to_string(), content: "SELECT 1;".to_string(), file_path: None, diff --git a/src/query_tools/text_actions.rs b/src/query_tools/text_actions.rs index c9f439ec..45c109c5 100644 --- a/src/query_tools/text_actions.rs +++ b/src/query_tools/text_actions.rs @@ -1,4 +1,4 @@ -/// Pure functions for ergonomic text manipulation in SQL editor. +//! Pure functions for ergonomic text manipulation in SQL editor. /// Toggle SQL line comments (`-- `) on lines spanned by selection. /// If all non-empty selected lines already start with `--`, comments are removed. @@ -67,15 +67,10 @@ pub fn toggle_line_comments(text: &str, selection_start: usize, selection_end: u if all_commented { // Uncomment: remove leading `-- ` or `--` let trimmed = line.trim_start(); - if trimmed.starts_with("--") { + if let Some(after_dashes) = trimmed.strip_prefix("--") { let indent_len = line.len() - trimmed.len(); let indent = &line[..indent_len]; - let after_dashes = &trimmed[2..]; - let rest = if after_dashes.starts_with(' ') { - &after_dashes[1..] - } else { - after_dashes - }; + let rest = after_dashes.strip_prefix(' ').unwrap_or(after_dashes); let modified = format!("{}{}", indent, rest); let diff = modified.len() as isize - line.len() as isize; if l_start < sel_min { diff --git a/src/sidebar_query.rs b/src/sidebar_query.rs index 60c2cb25..c648898e 100644 --- a/src/sidebar_query.rs +++ b/src/sidebar_query.rs @@ -526,7 +526,10 @@ pub(crate) fn open_query_file( }; let effective_connection_id = resolved_connection_id.or(auto_single_connection); + let tab_id = tabular.next_tab_id; + tabular.next_tab_id += 1; let new_tab = models::structs::QueryTab { + id: tab_id, title: filename, content: content.clone(), file_path: Some(file_path.to_string()), diff --git a/src/user_manager.rs b/src/user_manager.rs index 3f292df9..60f1de67 100644 --- a/src/user_manager.rs +++ b/src/user_manager.rs @@ -301,9 +301,10 @@ pub async fn execute_user_manager_command( Ok(()) } DatabasePool::MySQL(my_pool) => { - for stmt in query_owned.split(';') { + // Splitter yang paham quote: password seperti 'a;b' tidak ikut terpecah. + for stmt in crate::connection::split_sql_statements(&query_owned, true) { let trimmed = stmt.trim(); - if !trimmed.is_empty() { + if !crate::connection::sql::is_comment_only_statement(trimmed) { sqlx::query(sqlx::AssertSqlSafe(trimmed)) .execute(&**my_pool) .await diff --git a/src/window_egui/app_impl.rs b/src/window_egui/app_impl.rs index 11918ba7..4b44399d 100644 --- a/src/window_egui/app_impl.rs +++ b/src/window_egui/app_impl.rs @@ -485,6 +485,37 @@ impl Tabular { } }); ui.label(egui::RichText::new("Default interval used when Redis browser auto-refresh is enabled.").size(11.0).color(egui::Color32::from_gray(120))); + ui.add_space(8.0); + ui.heading("Query Execution"); + ui.horizontal(|ui| { + ui.label("Query timeout (seconds):"); + let mut secs = self.query_timeout_secs as i64; + if ui.add(egui::DragValue::new(&mut secs).range(0..=86_400)).changed() { + self.query_timeout_secs = secs.max(0) as u32; + self.prefs_dirty = true; + self.try_save_prefs(); + } + if self.query_timeout_secs == 0 { + ui.label(egui::RichText::new("no limit").size(11.0).color(egui::Color32::from_gray(120))); + } + }); + ui.label(egui::RichText::new("A statement running longer than this is cancelled on the server. 0 = never time out.").size(11.0).color(egui::Color32::from_gray(120))); + ui.add_space(4.0); + ui.horizontal(|ui| { + ui.label("Max rows per result:"); + let mut rows = self.max_result_rows as i64; + if ui.add(egui::DragValue::new(&mut rows).range(100..=5_000_000).speed(100)).changed() { + self.max_result_rows = rows.max(100) as u32; + self.prefs_dirty = true; + self.try_save_prefs(); + } + }); + ui.label(egui::RichText::new("Larger result sets are truncated (with a notice) so a stray SELECT * cannot exhaust memory.").size(11.0).color(egui::Color32::from_gray(120))); + ui.add_space(4.0); + if ui.checkbox(&mut self.restore_session, "Restore open tabs and unsaved drafts on startup").changed() { + self.prefs_dirty = true; + self.try_save_prefs(); + } } PrefTab::DataDirectory => { ui.heading("Data Directory"); @@ -4573,6 +4604,9 @@ impl Tabular { redis_browser_auto_refresh_seconds: self.redis_browser_auto_refresh_default_seconds.max(1), sync_server_url: Some(self.sync_server_url.clone()), ui_mode: self.ui_mode, + query_timeout_secs: self.query_timeout_secs, + max_result_rows: self.max_result_rows.max(1), + restore_session: self.restore_session, }; rt.block_on(store.save(&prefs)); log::debug!( @@ -4933,6 +4967,9 @@ impl App for Tabular { // Load server pagination preference self.use_server_pagination = prefs.use_server_pagination; + self.query_timeout_secs = prefs.query_timeout_secs; + self.max_result_rows = prefs.max_result_rows.max(1); + self.restore_session = prefs.restore_session; self.config_store = Some(store); self.last_saved_prefs = Some(prefs.clone()); diff --git a/src/window_egui/init.rs b/src/window_egui/init.rs index 44df25f7..247596a5 100644 --- a/src/window_egui/init.rs +++ b/src/window_egui/init.rs @@ -58,6 +58,9 @@ impl super::Tabular { self.auto_check_updates = prefs.auto_check_updates; self.use_server_pagination = prefs.use_server_pagination; self.enable_debug_logging = prefs.enable_debug_logging; + self.query_timeout_secs = prefs.query_timeout_secs; + self.max_result_rows = prefs.max_result_rows.max(1); + self.restore_session = prefs.restore_session; self.redis_browser_auto_refresh_default_seconds = prefs.redis_browser_auto_refresh_seconds.max(1); // Mirror AI settings self.ai_api_key = prefs.ai_api_key.clone(); @@ -263,6 +266,7 @@ impl super::Tabular { user_manager_result_receiver, active_query_jobs: std::collections::HashMap::new(), active_query_handles: std::collections::HashMap::new(), + query_backend_pids: Default::default(), cancelled_query_jobs: std::collections::HashMap::new(), query_job_batches: Vec::new(), pending_paginated_jobs: std::collections::HashSet::new(), @@ -482,6 +486,9 @@ impl super::Tabular { update_stage_receiver: None, staged_update_script: None, enable_debug_logging: false, // Default to false + query_timeout_secs: 0, + max_result_rows: crate::config::DEFAULT_MAX_RESULT_ROWS, + restore_session: true, auto_updater: crate::auto_updater::AutoUpdater::new().ok(), settings_active_pref_tab: PrefTab::ApplicationTheme, show_settings_menu: false, diff --git a/src/window_egui/mod.rs b/src/window_egui/mod.rs index aa34eb94..b688a326 100644 --- a/src/window_egui/mod.rs +++ b/src/window_egui/mod.rs @@ -136,6 +136,8 @@ pub struct Tabular { pub user_manager_result_receiver: Receiver<(usize, crate::user_manager::UserManagerResult)>, pub active_query_jobs: std::collections::HashMap, pub active_query_handles: std::collections::HashMap>, + /// Backend pid per job yang sedang berjalan, untuk cancel di sisi server. + pub query_backend_pids: crate::connection::types::BackendPidRegistry, pub cancelled_query_jobs: std::collections::HashMap, /// Sequential statement batches: member job ids + one abort handle for /// the whole batch (cancelling any member cancels the entire batch). @@ -446,6 +448,12 @@ pub struct Tabular { /// passed to `restart_app()` on "Restart Now". pub staged_update_script: Option, pub enable_debug_logging: bool, // New field for debug logging + /// Timeout query per statement dalam detik (0 = tanpa batas). + pub query_timeout_secs: u32, + /// Jumlah baris maksimum yang disimpan dari satu result set tanpa paginasi. + pub max_result_rows: u32, + /// Buka kembali tab dari sesi sebelumnya saat startup. + pub restore_session: bool, // Auto updater instance pub auto_updater: Option, // Preferences window active tab diff --git a/src/window_egui/query_jobs.rs b/src/window_egui/query_jobs.rs index 0d2fd41a..31397c74 100644 --- a/src/window_egui/query_jobs.rs +++ b/src/window_egui/query_jobs.rs @@ -57,6 +57,30 @@ impl super::Tabular { let was_paginated = self.pending_paginated_jobs.remove(&message.job_id); + if message.truncated { + self.toasts.warning(format!( + "Result truncated to the first {} rows. Add a LIMIT or raise “Max rows per result” in Settings → Performance.", + message.rows.len() + )); + } + + // User bisa saja pindah tab selama query berjalan. Hasil tab aktif + // disimpan di state tampilan global, sedangkan tab lain di field + // miliknya sendiri. Jadi hasil untuk tab di latar belakang ditulis + // langsung ke tab tersebut, tanpa menimpa data yang sedang tampil. + if let Some(origin_idx) = message + .tab_id + .and_then(|id| self.query_tabs.iter().position(|t| t.id == id)) + && origin_idx != self.active_tab_index + { + self.apply_result_to_background_tab(origin_idx, &message, was_paginated); + if self.active_query_jobs.is_empty() { + self.query_execution_in_progress = false; + self.extend_query_icon_hold(); + } + return; + } + if let Some(ast_sql) = message.ast_debug_sql.clone() { self.last_compiled_sql = Some(ast_sql); } @@ -66,14 +90,7 @@ impl super::Tabular { // Update query message panel if message.success { - let duration_ms = message.duration.as_millis(); - let row_count = message.affected_rows.unwrap_or(message.rows.len()); - self.query_message = format!( - "Query executed successfully in {}.{:03}s • {} row(s) affected", - duration_ms / 1000, - duration_ms % 1000, - row_count - ); + self.query_message = describe_query_outcome(&message); self.query_message_is_error = false; // Auto-switch to Data tab to show results self.table_bottom_view = models::structs::TableBottomView::Data; @@ -187,6 +204,85 @@ impl super::Tabular { crate::connection::ensure_background_pool_creation(self, cid); } } + /// Menyimpan hasil query yang selesai ke tab yang sedang tidak ditampilkan. + /// Data masuk ke field hasil milik tab tersebut, lalu `switch_to_tab` + /// menukarnya ke tampilan saat user kembali ke tab itu. + fn apply_result_to_background_tab( + &mut self, + tab_index: usize, + message: &connection::QueryResultMessage, + was_paginated: bool, + ) { + let query_message = describe_query_outcome(message); + let tab_title; + { + let Some(tab) = self.query_tabs.get_mut(tab_index) else { + return; + }; + tab_title = tab.title.clone(); + tab.has_executed_query = true; + tab.query_message = query_message.clone(); + tab.query_message_is_error = !message.success; + + if !(was_paginated && message.success) { + let new_index = tab.results.len(); + tab.results.push(models::structs::QueryResult { + headers: message.headers.clone(), + rows: message.rows.clone(), + all_rows: message.rows.clone(), + table_name: if message.success { + format!("Result {}", new_index + 1) + } else { + "Error".to_string() + }, + current_page: 0, + page_size: tab.page_size.max(1), + total_rows: message.rows.len(), + query_message: query_message.clone(), + query_message_is_error: !message.success, + execution_time_ms: message.duration.as_millis(), + column_metadata: message.column_metadata.clone(), + explain_plan_json: None, + pinned_columns: std::collections::HashSet::new(), + }); + if new_index > 0 { + // Statement berikutnya dalam batch hanya menambah tab hasil. + tab.active_result_index = tab.active_result_index.min(new_index); + } + } + + let is_primary = was_paginated || tab.results.len() <= 1; + if is_primary { + tab.active_result_index = 0; + tab.result_headers = message.headers.clone(); + tab.result_all_rows = message.rows.clone(); + tab.result_rows = message.rows.clone(); + tab.result_column_metadata = message.column_metadata.clone(); + tab.total_rows = message.rows.len(); + if !was_paginated { + tab.current_page = 0; + } + tab.result_table_name = if !message.success { + "Error".to_string() + } else if message.rows.is_empty() { + "Query executed successfully (no results)".to_string() + } else { + format!("Query Results ({} rows)", message.rows.len()) + }; + } + } + + if message.success && !was_paginated { + sidebar_history::save_query_to_history(self, &message.query, message.connection_id); + } + + let summary = format!("“{}” finished: {}", tab_title, query_message); + if message.success { + self.toasts.info(summary); + } else { + self.toasts.error(summary); + } + } pub fn apply_paginated_query_result(&mut self, message: &connection::QueryResultMessage) { self.current_table_headers = message.headers.clone(); self.current_table_data = message.rows.clone(); @@ -213,9 +309,42 @@ impl super::Tabular { active_tab.total_rows = self.actual_total_rows.unwrap_or(self.total_rows); } } + /// Kirim perintah cancel ke server (pg_cancel_backend / KILL QUERY) untuk + /// job yang backend pid-nya sudah tercatat. `abort()` pada task saja hanya + /// menghentikan penantian di klien, query tetap berjalan di server. + fn cancel_queries_on_server(&self, job_ids: &[u64]) { + let Some(runtime) = self.runtime.clone() else { + return; + }; + for job_id in job_ids { + let pid = self + .query_backend_pids + .lock() + .ok() + .and_then(|m| m.get(job_id).copied()); + let pool = self + .active_query_jobs + .get(job_id) + .and_then(|status| self.connection_pools.get(&status.connection_id).cloned()); + if let (Some(pid), Some(pool)) = (pid, pool) { + runtime.spawn(connection::execute::cancel_backend_query(pool, pid)); + } + } + } + pub fn cancel_active_query_job(&mut self, job_id: u64) -> bool { self.prune_cancelled_jobs(); + let mut server_side_ids = vec![job_id]; + if let Some((ids, _)) = self + .query_job_batches + .iter() + .find(|(ids, _)| ids.contains(&job_id)) + { + server_side_ids.extend(ids.iter().copied().filter(|id| *id != job_id)); + } + self.cancel_queries_on_server(&server_side_ids); + let preview_text = self .active_query_jobs .get(&job_id) @@ -266,11 +395,10 @@ impl super::Tabular { } else { preview }; - self.error_message = format!("Query cancelled: {}", truncated.trim()); + self.toasts.info(format!("Query cancelled: {}", truncated.trim())); } else { - self.error_message = "Query cancelled.".to_string(); + self.toasts.info("Query cancelled."); } - self.show_error_message = true; self.current_table_name = "Query cancelled".to_string(); } @@ -302,3 +430,27 @@ impl super::Tabular { Some(std::time::Instant::now() + std::time::Duration::from_millis(900)); } } + +/// Baris status untuk query yang selesai: jumlah baris yang dikembalikan untuk +/// result set, jumlah baris terdampak untuk perubahan data (dari driver), atau +/// teks error. +pub(crate) fn describe_query_outcome(message: &connection::QueryResultMessage) -> String { + if !message.success { + return format!( + "Error: {}", + message.error.as_deref().unwrap_or("Unknown error") + ); + } + let duration_ms = message.duration.as_millis(); + let count = match message.affected_rows { + Some(n) => format!("{} row(s) affected", n), + None if message.truncated => format!("first {} row(s) returned (truncated)", message.rows.len()), + None => format!("{} row(s) returned", message.rows.len()), + }; + format!( + "Query executed successfully in {}.{:03}s • {}", + duration_ms / 1000, + duration_ms % 1000, + count + ) +} From a023271fdbf29c04b6070eca9a7859923b9722aa Mon Sep 17 00:00:00 2001 From: Antigravity Agent Date: Thu, 17 Sep 2026 19:14:56 +0700 Subject: [PATCH 02/32] feat(stability): file logging, crash reports, session restore, unsaved guards MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Compile logging in (was max_level_off) and write it to /logs/tabular.log with size-based rotation; "Enable Debug Logging" now applies immediately. - Install a panic hook that saves crash-