diff --git a/src-tauri/src/commands/mod.rs b/src-tauri/src/commands/mod.rs
index 1fb11f1..3143f07 100644
--- a/src-tauri/src/commands/mod.rs
+++ b/src-tauri/src/commands/mod.rs
@@ -6,6 +6,7 @@ pub mod helpers;
 pub mod jdbc;
 pub mod query;
 pub mod server;
+pub mod sql_analysis;
 pub mod store;
 pub mod transfer;
 
diff --git a/src-tauri/src/commands/sql_analysis.rs b/src-tauri/src/commands/sql_analysis.rs
new file mode 100644
index 0000000..2316be7
--- /dev/null
+++ b/src-tauri/src/commands/sql_analysis.rs
@@ -0,0 +1,384 @@
+//! SQL editability analysis.
+//!
+//! Parses a SELECT statement with sqlparser and decides whether its result rows
+//! can be mapped back to a single base table (and therefore edited/deleted).
+//! A result is editable only when the query is a plain single-table SELECT
+//! (optionally with WHERE/ORDER BY/LIMIT) whose rows map 1:1 to the table.
+
+use serde::{Deserialize, Serialize};
+use sqlparser::ast::{
+    Expr, GroupByExpr, ObjectNamePart, Query, Select, SelectItem, SetExpr, Statement, TableFactor,
+};
+use sqlparser::dialect::{GenericDialect, MsSqlDialect, MySqlDialect, PostgreSqlDialect};
+use sqlparser::parser::Parser;
+
+/// Why a query result cannot be edited.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "kebab-case")]
+pub enum NonEditableReason {
+    /// Statement is not a SELECT (INSERT/UPDATE/DELETE/DDL...).
+    NotSelect,
+    /// Query starts with WITH (CTE) — row identity cannot be trusted.
+    Cte,
+    /// UNION/INTERSECT/EXCEPT — rows come from multiple statements.
+    SetOperation,
+    /// GROUP BY / HAVING / DISTINCT / aggregate functions — rows are aggregated.
+    Aggregation,
+    /// More than one table source (JOIN or comma-separated).
+    MultipleSources,
+    /// No FROM clause at all.
+    NoTable,
+    /// The FROM source is a subquery/table function/parenthesized join.
+    ComplexSource,
+}
+
+/// Result of analyzing whether a query's rows are editable.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct SqlEditability {
+    pub editable: bool,
+    /// Present only when editable — the single base table the query reads from.
+    pub table_name: Option<String>,
+    pub schema: Option<String>,
+    pub reason: Option<NonEditableReason>,
+}
+
+fn dialect_for(db_type: &str) -> Box<dyn sqlparser::dialect::Dialect> {
+    match db_type.to_ascii_lowercase().as_str() {
+        "postgres" | "postgresql" | "duckdb" | "cockroachdb" | "gbase8c" | "kingbasees" | "yashandb"
+        | "xugudb" | "timescaledb" | "redshift" | "yugabytedb" | "opengauss" | "highgo" | "uxdb"
+        | "gaussdb" => Box::new(PostgreSqlDialect {}),
+        "mysql" | "clickhouse" | "oceanbase" | "mariadb" | "gbase8a" => Box::new(MySqlDialect {}),
+        "sqlserver" | "mssql" => Box::new(MsSqlDialect {}),
+        _ => Box::new(GenericDialect {}),
+    }
+}
+
+/// Analyze a SQL statement and report whether its result rows are editable.
+///
+/// Safe by construction: only plain single-table SELECTs map result rows back
+/// to base rows. Aggregations, set operations, CTEs, and multi-source queries
+/// are reported as non-editable with a machine-readable reason.
+pub fn analyze_sql_editability(sql: &str, db_type: &str) -> SqlEditability {
+    let dialect = dialect_for(db_type);
+    let Ok(statements) = Parser::parse_sql(&*dialect, sql) else {
+        return SqlEditability {
+            editable: false,
+            table_name: None,
+            schema: None,
+            reason: Some(NonEditableReason::ComplexSource),
+        };
+    };
+
+    if statements.len() != 1 {
+        return SqlEditability {
+            editable: false,
+            table_name: None,
+            schema: None,
+            reason: Some(NonEditableReason::NotSelect),
+        };
+    }
+
+    let Some(query) = as_select_query(&statements[0]) else {
+        return SqlEditability {
+            editable: false,
+            table_name: None,
+            schema: None,
+            reason: Some(NonEditableReason::NotSelect),
+        };
+    };
+
+    // WITH (CTE) — the top-level FROM may reference a CTE instead of a table.
+    if query.with.is_some() {
+        return non_editable(NonEditableReason::Cte);
+    }
+
+    // UNION/INTERSECT/EXCEPT — rows come from multiple statements.
+    if !matches!(query.body.as_ref(), SetExpr::Select(_)) {
+        return non_editable(NonEditableReason::SetOperation);
+    }
+
+    let SetExpr::Select(select) = query.body.as_ref() else {
+        return non_editable(NonEditableReason::SetOperation);
+    };
+
+    if select_is_aggregated(select) {
+        return non_editable(NonEditableReason::Aggregation);
+    }
+
+    if select.from.is_empty() {
+        return non_editable(NonEditableReason::NoTable);
+    }
+
+    // Exactly one FROM source, and it must be a plain table (no subquery,
+    // no table function, no parenthesized join).
+    if select.from.len() > 1 {
+        return non_editable(NonEditableReason::MultipleSources);
+    }
+
+    let table_with_joins = &select.from[0];
+    if !table_with_joins.joins.is_empty() {
+        return non_editable(NonEditableReason::MultipleSources);
+    }
+
+    let TableFactor::Table { name, .. } = &table_with_joins.relation else {
+        return non_editable(NonEditableReason::ComplexSource);
+    };
+
+    // Extract schema (second-to-last part) and table (last part).
+    let parts: Vec<&String> = name
+        .0
+        .iter()
+        .filter_map(ObjectNamePart::as_ident)
+        .map(|ident| &ident.value)
+        .collect();
+
+    let Some(table_name) = parts.last().cloned().cloned() else {
+        return non_editable(NonEditableReason::NoTable);
+    };
+
+    let schema = if parts.len() >= 2 {
+        Some(parts[parts.len() - 2].clone())
+    } else {
+        None
+    };
+
+    SqlEditability {
+        editable: true,
+        table_name: Some(table_name),
+        schema,
+        reason: None,
+    }
+}
+
+fn non_editable(reason: NonEditableReason) -> SqlEditability {
+    SqlEditability {
+        editable: false,
+        table_name: None,
+        schema: None,
+        reason: Some(reason),
+    }
+}
+
+fn as_select_query(statement: &Statement) -> Option<&Query> {
+    match statement {
+        Statement::Query(query) => Some(query),
+        _ => None,
+    }
+}
+
+/// True when the SELECT is aggregated: GROUP BY/HAVING/DISTINCT or an
+/// aggregate function in the projection. Aggregated rows do not map 1:1 to
+/// base-table rows.
+fn select_is_aggregated(select: &Select) -> bool {
+    if select.distinct.is_some() {
+        return true;
+    }
+    if matches!(select.group_by, GroupByExpr::Expressions(ref exprs, _) if !exprs.is_empty()) {
+        return true;
+    }
+    if select.having.is_some() {
+        return true;
+    }
+    select.projection.iter().any(projection_has_aggregate)
+}
+
+fn projection_has_aggregate(item: &SelectItem) -> bool {
+    match item {
+        SelectItem::UnnamedExpr(expr) | SelectItem::ExprWithAlias { expr, .. } => {
+            expr_has_aggregate(expr)
+        }
+        _ => false,
+    }
+}
+
+fn expr_has_aggregate(expr: &Expr) -> bool {
+    match expr {
+        Expr::Function(function) => {
+            if is_aggregate_function(&function.name.to_string()) {
+                return true;
+            }
+            expr_function_has_aggregate_arg(&function.args)
+                || function.filter.as_deref().is_some_and(expr_has_aggregate)
+        }
+        Expr::BinaryOp { left, right, .. } => expr_has_aggregate(left) || expr_has_aggregate(right),
+        Expr::Nested(inner) | Expr::IsNull(inner) | Expr::IsNotNull(inner) => expr_has_aggregate(inner),
+        Expr::UnaryOp { expr: inner, .. } => expr_has_aggregate(inner),
+        Expr::Case { operand, conditions, else_result, .. } => {
+            operand.as_deref().is_some_and(expr_has_aggregate)
+                || conditions
+                    .iter()
+                    .any(|cond| expr_has_aggregate(&cond.condition) || expr_has_aggregate(&cond.result))
+                || else_result.as_deref().is_some_and(expr_has_aggregate)
+        }
+        Expr::Subquery(query) | Expr::Exists { subquery: query, .. } => {
+            let mut found = false;
+            if let SetExpr::Select(select) = query.body.as_ref() {
+                found = select.projection.iter().any(projection_has_aggregate);
+            }
+            found
+        }
+        _ => false,
+    }
+}
+
+fn expr_function_has_aggregate_arg(args: &sqlparser::ast::FunctionArguments) -> bool {
+    match args {
+        sqlparser::ast::FunctionArguments::List(list) => list
+            .args
+            .iter()
+            .any(|arg| match arg {
+                sqlparser::ast::FunctionArg::Unnamed(sqlparser::ast::FunctionArgExpr::Expr(e)) => {
+                    expr_has_aggregate(e)
+                }
+                sqlparser::ast::FunctionArg::Named { arg, .. }
+                | sqlparser::ast::FunctionArg::ExprNamed { arg, .. } => match arg {
+                    sqlparser::ast::FunctionArgExpr::Expr(e) => expr_has_aggregate(e),
+                    _ => false,
+                },
+                _ => false,
+            }),
+        sqlparser::ast::FunctionArguments::Subquery(query) => {
+            let mut found = false;
+            if let SetExpr::Select(select) = query.body.as_ref() {
+                found = select.projection.iter().any(projection_has_aggregate);
+            }
+            found
+        }
+        sqlparser::ast::FunctionArguments::None => false,
+    }
+}
+
+fn is_aggregate_function(name: &str) -> bool {
+    matches!(
+        name.to_ascii_uppercase().as_str(),
+        "COUNT" | "SUM" | "AVG" | "MIN" | "MAX" | "STDDEV" | "STDDEV_POP" | "STDDEV_SAMP" | "VARIANCE"
+            | "VAR_POP" | "VAR_SAMP" | "ARRAY_AGG" | "STRING_AGG" | "JSON_AGG" | "JSONB_AGG"
+            | "BOOL_AND" | "BOOL_OR" | "EVERY"
+    )
+}
+
+/// Tauri command: analyze whether a query's result rows are editable.
+///
+/// Returns the single base table (with optional schema) when the query is a
+/// plain single-table SELECT; otherwise a machine-readable non-editable reason
+/// the frontend can surface to the user.
+#[tauri::command]
+pub fn analyze_sql_editability_command(sql: String, database_type: Option<String>) -> SqlEditability {
+    analyze_sql_editability(&sql, database_type.as_deref().unwrap_or("generic"))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn editable(sql: &str) -> bool {
+        analyze_sql_editability(sql, "postgres").editable
+    }
+
+    fn reason(sql: &str) -> Option<NonEditableReason> {
+        analyze_sql_editability(sql, "postgres").reason
+    }
+
+    #[test]
+    fn plain_select_star_is_editable() {
+        let result = analyze_sql_editability("SELECT * FROM apps", "postgres");
+        assert!(result.editable);
+        assert_eq!(result.table_name.as_deref(), Some("apps"));
+        assert_eq!(result.schema, None);
+    }
+
+    #[test]
+    fn select_with_qualifier_returns_schema() {
+        let result = analyze_sql_editability("SELECT * FROM public.apps", "postgres");
+        assert!(result.editable);
+        assert_eq!(result.table_name.as_deref(), Some("apps"));
+        assert_eq!(result.schema.as_deref(), Some("public"));
+    }
+
+    #[test]
+    fn select_with_where_order_limit_is_editable() {
+        assert!(editable("SELECT id, name FROM customers WHERE id > 10 ORDER BY name LIMIT 100"));
+    }
+
+    #[test]
+    fn quoted_table_name_is_editable() {
+        let result = analyze_sql_editability("SELECT * FROM \"My Table\"", "postgres");
+        assert!(result.editable);
+        assert_eq!(result.table_name.as_deref(), Some("My Table"));
+    }
+
+    #[test]
+    fn join_is_not_editable() {
+        assert!(!editable("SELECT a.*, b.* FROM a JOIN b ON a.id = b.id"));
+        assert_eq!(reason("SELECT a.*, b.* FROM a JOIN b ON a.id = b.id"), Some(NonEditableReason::MultipleSources));
+    }
+
+    #[test]
+    fn comma_separated_sources_is_not_editable() {
+        assert_eq!(reason("SELECT * FROM a, b"), Some(NonEditableReason::MultipleSources));
+    }
+
+    #[test]
+    fn count_aggregation_is_not_editable() {
+        assert_eq!(reason("SELECT COUNT(*) FROM apps"), Some(NonEditableReason::Aggregation));
+    }
+
+    #[test]
+    fn group_by_is_not_editable() {
+        assert_eq!(reason("SELECT name, COUNT(*) FROM customers GROUP BY name"), Some(NonEditableReason::Aggregation));
+    }
+
+    #[test]
+    fn distinct_is_not_editable() {
+        assert_eq!(reason("SELECT DISTINCT name FROM customers"), Some(NonEditableReason::Aggregation));
+    }
+
+    #[test]
+    fn union_is_not_editable() {
+        assert_eq!(reason("SELECT * FROM a UNION SELECT * FROM b"), Some(NonEditableReason::SetOperation));
+    }
+
+    #[test]
+    fn cte_is_not_editable() {
+        assert_eq!(
+            reason("WITH t AS (SELECT * FROM apps) SELECT * FROM t"),
+            Some(NonEditableReason::Cte)
+        );
+    }
+
+    #[test]
+    fn subquery_source_is_not_editable() {
+        assert_eq!(
+            reason("SELECT * FROM (SELECT * FROM apps) t"),
+            Some(NonEditableReason::ComplexSource)
+        );
+    }
+
+    #[test]
+    fn non_select_statement_is_not_editable() {
+        assert_eq!(reason("UPDATE apps SET name = 'x'"), Some(NonEditableReason::NotSelect));
+        assert_eq!(reason("DELETE FROM apps"), Some(NonEditableReason::NotSelect));
+        assert_eq!(reason("INSERT INTO apps (name) VALUES ('x')"), Some(NonEditableReason::NotSelect));
+    }
+
+    #[test]
+    fn multiple_statements_is_not_editable() {
+        assert_eq!(
+            reason("SELECT * FROM apps; SELECT * FROM orders"),
+            Some(NonEditableReason::NotSelect)
+        );
+    }
+
+    #[test]
+    fn mysql_backtick_table_is_editable() {
+        let result = analyze_sql_editability("SELECT * FROM `customers`", "mysql");
+        assert!(result.editable);
+        assert_eq!(result.table_name.as_deref(), Some("customers"));
+    }
+
+    #[test]
+    fn no_from_is_not_editable() {
+        assert_eq!(reason("SELECT 1"), Some(NonEditableReason::NoTable));
+    }
+}
diff --git a/src-tauri/src/database/mysql.rs b/src-tauri/src/database/mysql.rs
index 7c5b824..1b43637 100644
--- a/src-tauri/src/database/mysql.rs
+++ b/src-tauri/src/database/mysql.rs
@@ -527,42 +527,42 @@ impl DatabaseAdapter for MySQLAdapter {
         let execution_time;
 
         if is_select {
-            // Execute query and get results
-            let result: Vec<Row> = if let Some(timeout_duration) = timeout {
-                tokio::time::timeout(timeout_duration, conn.query(query))
+            // Use query_iter so column metadata is available even when the
+            // result set is empty — an empty SELECT must still render its
+            // columns (empty table) rather than looking like a DML statement.
+            let mut result = if let Some(timeout_duration) = timeout {
+                tokio::time::timeout(timeout_duration, conn.query_iter(query))
                     .await
                     .map_err(|_| {
                         DbError::Timeout(format!("Query timed out after {:?}", timeout_duration))
                     })?
                     .map_err(mysql_error_to_db_error)?
             } else {
-                conn.query(query).await.map_err(mysql_error_to_db_error)?
+                conn.query_iter(query).await.map_err(mysql_error_to_db_error)?
             };
 
             execution_time = start.elapsed().as_millis() as u64;
 
-            if result.is_empty() {
-                Ok(QueryResult::new(Vec::new()).with_execution_time(execution_time))
-            } else {
-                let columns: Vec<String> = result[0]
-                    .columns_ref()
-                    .iter()
-                    .map(|col| col.name_str().to_string())
-                    .collect();
-                let column_types: Vec<String> = result[0]
-                    .columns_ref()
-                    .iter()
-                    .map(|col| format!("{:?}", col.column_type()))
-                    .collect();
-
-                let mut query_result = QueryResult::with_columns(columns, column_types);
-                for row in result {
-                    let query_row = Self::row_to_query_row(row)?;
-                    query_result.add_row(query_row);
-                }
+            let columns: Vec<String> = result
+                .columns()
+                .map(|cols| cols.iter().map(|col| col.name_str().to_string()).collect())
+                .unwrap_or_default();
+            let column_types: Vec<String> = result
+                .columns()
+                .map(|cols| {
+                    cols.iter()
+                        .map(|col| format!("{:?}", col.column_type()))
+                        .collect()
+                })
+                .unwrap_or_default();
 
-                Ok(query_result.with_execution_time(execution_time))
+            let mut query_result = QueryResult::with_columns(columns, column_types);
+            while let Some(row) = result.next().await.map_err(mysql_error_to_db_error)? {
+                let query_row = Self::row_to_query_row(row)?;
+                query_result.add_row(query_row);
             }
+
+            Ok(query_result.with_execution_time(execution_time))
         } else {
             // For INSERT, UPDATE, DELETE, etc.
             if let Some(timeout_duration) = timeout {
diff --git a/src-tauri/src/database/postgres.rs b/src-tauri/src/database/postgres.rs
index 26ce234..47524f2 100644
--- a/src-tauri/src/database/postgres.rs
+++ b/src-tauri/src/database/postgres.rs
@@ -1005,8 +1005,11 @@ impl DatabaseAdapter for PostgresAdapter {
         let execution_time;
 
         if is_select {
-            let result = if let Some(timeout_duration) = timeout {
-                tokio::time::timeout(timeout_duration, client.query(query, &[]))
+            // Prepare first so column metadata is available even when the
+            // result set is empty — an empty SELECT must still render its
+            // columns (empty table) rather than looking like a DML statement.
+            let statement = if let Some(timeout_duration) = timeout {
+                tokio::time::timeout(timeout_duration, client.prepare(query))
                     .await
                     .map_err(|_| {
                         DbError::Timeout(format!("Query timed out after {:?}", timeout_duration))
@@ -1014,35 +1017,45 @@ impl DatabaseAdapter for PostgresAdapter {
                     .map_err(postgres_error_to_db_error)?
             } else {
                 client
-                    .query(query, &[])
+                    .prepare(query)
                     .await
                     .map_err(postgres_error_to_db_error)?
             };
 
-            execution_time = start.elapsed().as_millis() as u64;
-
-            if result.is_empty() {
-                Ok(QueryResult::new(Vec::new()).with_execution_time(execution_time))
+            let result = if let Some(timeout_duration) = timeout {
+                tokio::time::timeout(timeout_duration, client.query(&statement, &[]))
+                    .await
+                    .map_err(|_| {
+                        DbError::Timeout(format!("Query timed out after {:?}", timeout_duration))
+                    })?
+                    .map_err(postgres_error_to_db_error)?
             } else {
-                let columns: Vec<String> = result[0]
-                    .columns()
-                    .iter()
-                    .map(|col| col.name().to_string())
-                    .collect();
-                let column_types: Vec<String> = result[0]
-                    .columns()
-                    .iter()
-                    .map(|col| col.type_().name().to_string())
-                    .collect();
+                client
+                    .query(&statement, &[])
+                    .await
+                    .map_err(postgres_error_to_db_error)?
+            };
 
-                let mut query_result = QueryResult::with_columns(columns, column_types);
-                for row in &result {
-                    let query_row = Self::row_to_query_row(row)?;
-                    query_result.add_row(query_row);
-                }
+            execution_time = start.elapsed().as_millis() as u64;
 
-                Ok(query_result.with_execution_time(execution_time))
+            let columns: Vec<String> = statement
+                .columns()
+                .iter()
+                .map(|col| col.name().to_string())
+                .collect();
+            let column_types: Vec<String> = statement
+                .columns()
+                .iter()
+                .map(|col| col.type_().name().to_string())
+                .collect();
+
+            let mut query_result = QueryResult::with_columns(columns, column_types);
+            for row in &result {
+                let query_row = Self::row_to_query_row(row)?;
+                query_result.add_row(query_row);
             }
+
+            Ok(query_result.with_execution_time(execution_time))
         } else {
             // For INSERT, UPDATE, DELETE, etc.
             let affected = if let Some(timeout_duration) = timeout {
diff --git a/src-tauri/src/database/sqlserver.rs b/src-tauri/src/database/sqlserver.rs
index 7412af7..5ca4d83 100644
--- a/src-tauri/src/database/sqlserver.rs
+++ b/src-tauri/src/database/sqlserver.rs
@@ -421,7 +421,7 @@ impl DatabaseAdapter for SqlServerAdapter {
             .map(Duration::from_millis);
 
         // Execute query with optional timeout
-        let stream = if let Some(timeout_duration) = timeout {
+        let mut stream = if let Some(timeout_duration) = timeout {
             tokio::time::timeout(timeout_duration, client.simple_query(query))
                 .await
                 .map_err(|_| {
@@ -436,6 +436,20 @@ impl DatabaseAdapter for SqlServerAdapter {
         };
 
         // Collect results
+        // QueryStream::columns() forwards to the first result-set metadata
+        // without consuming rows, so column names survive even when the result
+        // set is empty — an empty SELECT must still render its columns rather
+        // than looking like a DML statement.
+        let stream_columns = stream
+            .columns()
+            .await
+            .map_err(|e| DbError::QueryExecution(e.to_string()))?
+            .map(|cols| {
+                cols.iter()
+                    .map(|col| (col.name().to_string(), format!("{:?}", col.column_type())))
+                    .collect::<Vec<_>>()
+            });
+
         let results = stream
             .into_results()
             .await
@@ -452,28 +466,34 @@ impl DatabaseAdapter for SqlServerAdapter {
         } else {
             let result_set = &results[0];
 
-            if result_set.is_empty() {
-                Ok(QueryResult::new(Vec::new()).with_execution_time(execution_time))
-            } else {
-                let columns: Vec<String> = result_set[0]
-                    .columns()
-                    .iter()
-                    .map(|col| col.name().to_string())
-                    .collect();
-                let column_types: Vec<String> = result_set[0]
-                    .columns()
-                    .iter()
-                    .map(|col| format!("{:?}", col.column_type()))
-                    .collect();
-
-                let mut query_result = QueryResult::with_columns(columns, column_types);
-                for row in result_set {
-                    let query_row = Self::row_to_query_row(row)?;
-                    query_result.add_row(query_row);
-                }
+            // Fall back to the first row for column names when the stream
+            // metadata is unavailable (should not happen in practice).
+            let columns: Vec<String> = stream_columns
+                .as_ref()
+                .map(|cols| cols.iter().map(|(name, _)| name.clone()).collect())
+                .or_else(|| {
+                    result_set
+                        .first()
+                        .map(|row| row.columns().iter().map(|col| col.name().to_string()).collect())
+                })
+                .unwrap_or_default();
+            let column_types: Vec<String> = stream_columns
+                .as_ref()
+                .map(|cols| cols.iter().map(|(_, ty)| ty.clone()).collect())
+                .or_else(|| {
+                    result_set
+                        .first()
+                        .map(|row| row.columns().iter().map(|col| format!("{:?}", col.column_type())).collect())
+                })
+                .unwrap_or_default();
 
-                Ok(query_result.with_execution_time(execution_time))
+            let mut query_result = QueryResult::with_columns(columns, column_types);
+            for row in result_set {
+                let query_row = Self::row_to_query_row(row)?;
+                query_result.add_row(query_row);
             }
+
+            Ok(query_result.with_execution_time(execution_time))
         }
     }
 
diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs
index b2f2aaa..dbaac2e 100644
--- a/src-tauri/src/lib.rs
+++ b/src-tauri/src/lib.rs
@@ -286,6 +286,7 @@ pub fn run() {
             commands::execute_sorted_query,
             commands::cancel_query,
             commands::explain_query,
+            commands::sql_analysis::analyze_sql_editability_command,
             // Database browsing commands
             commands::list_databases,
             commands::list_schemas,
diff --git a/src/components/database-browser/QueryResultPanel.vue b/src/components/database-browser/QueryResultPanel.vue
index 70b69f9..d4af3e4 100644
--- a/src/components/database-browser/QueryResultPanel.vue
+++ b/src/components/database-browser/QueryResultPanel.vue
@@ -1,4 +1,5 @@
 <script setup lang="ts">
+import type { ColumnInfo } from '@/composables/sqlCompletion/metadata'
 import type { QueryResult } from '@/store/tabStore'
 import type { ApiError, ApiResponse } from '@/types/api'
 import type { ExplainResult } from '@/types/explainPlan'
@@ -12,6 +13,7 @@ import { Button } from '@/components/ui/button'
 import { Spinner } from '@/components/ui/spinner'
 import { Tooltip, TooltipContent, TooltipProvider, TooltipTrigger } from '@/components/ui/tooltip'
 import { useDataGridCopy } from '@/composables/useDataGridCopy'
+import { useConnectionStore } from '@/store'
 import { formatApiError, isApiError } from '@/types/api'
 
 type Props = {
@@ -88,6 +90,76 @@ const gridExecutionTimeMs = ref<number | undefined>(undefined)
 const activeSort = ref<SortColumn[]>([])
 const activeFilters = ref<ColumnFilter[]>([])
 
+// ── Result editability (single-table SELECTs only) ──
+const connectionStore = useConnectionStore()
+
+type SqlEditability = {
+  editable: boolean
+  tableName?: string | null
+  schema?: string | null
+  reason?: string | null
+}
+
+const editability = ref<SqlEditability>({ editable: false })
+const editableTableName = ref<string | undefined>(undefined)
+const editableSchema = ref<string | undefined>(undefined)
+const editablePrimaryKeys = ref<string[]>([])
+
+const connectionTypeForAnalysis = computed(() => {
+  if (!props.connectionId)
+    return undefined
+  const conn = connectionStore.getConnectionById(props.connectionId)
+  return conn?.type ? String(conn.type).toLowerCase() : undefined
+})
+
+async function refreshEditability() {
+  editability.value = { editable: false }
+  editableTableName.value = undefined
+  editableSchema.value = undefined
+  editablePrimaryKeys.value = []
+
+  if (!props.sql || !props.connectionId) {
+    return
+  }
+
+  try {
+    const result = await invoke<SqlEditability>('analyze_sql_editability_command', {
+      sql: props.sql,
+      databaseType: connectionTypeForAnalysis.value,
+    })
+    editability.value = result
+    if (!result.editable || !result.tableName) {
+      return
+    }
+
+    editableTableName.value = result.tableName
+    editableSchema.value = result.schema ?? undefined
+
+    // Fetch primary keys so edit/delete can locate rows.
+    const columns = await invoke<ColumnInfo[]>('list_columns', {
+      connectionId: props.connectionId,
+      database: props.database ?? null,
+      schema: result.schema ?? props.schema ?? null,
+      tableName: result.tableName,
+    })
+    editablePrimaryKeys.value = (columns ?? [])
+      .filter(col => col.is_primary_key)
+      .map(col => col.name)
+  }
+  catch {
+    // Analysis is best-effort; fall back to read-only grid on failure.
+    editability.value = { editable: false }
+  }
+}
+
+watch(
+  () => [props.sql, props.connectionId, props.database],
+  () => {
+    void refreshEditability()
+  },
+  { immediate: true },
+)
+
 // Sync from props when results change
 watch(() => props.results, (r) => {
   if (r) {
@@ -365,7 +437,10 @@ const displayExecutionTime = computed(() => gridExecutionTimeMs.value ?? props.e
           :execution-time-ms="displayExecutionTime"
           :connection-id="connectionId"
           :database="database"
-          :schema="schema"
+          :schema="editableSchema ?? schema"
+          :table-name="editableTableName"
+          :primary-keys="editablePrimaryKeys"
+          :editability-reason="editability.reason ?? undefined"
           :loading="displayExecuting"
           :hide-toolbar="true"
           :hide-batch-actions="true"
diff --git a/src/components/grid/CellContextMenu.vue b/src/components/grid/CellContextMenu.vue
index 6b2dc8b..eec1774 100644
--- a/src/components/grid/CellContextMenu.vue
+++ b/src/components/grid/CellContextMenu.vue
@@ -20,6 +20,7 @@ const props = defineProps<{
   column: string
   columnType?: string
   row: Record<string, unknown> | null
+  rowIndex?: number
   columns: string[]
   tableName?: string
 }>()
diff --git a/src/components/grid/DataGrid.vue b/src/components/grid/DataGrid.vue
index de5160e..b6edfc9 100644
--- a/src/components/grid/DataGrid.vue
+++ b/src/components/grid/DataGrid.vue
@@ -50,6 +50,8 @@ type Props = {
   connectionId?: string
   hideToolbar?: boolean
   hideBatchActions?: boolean
+  /** Why the result cannot be edited, when tableName is absent. */
+  editabilityReason?: string
 }
 
 const props = withDefaults(defineProps<Props>(), {
@@ -64,6 +66,7 @@ const props = withDefaults(defineProps<Props>(), {
   connectionId: undefined,
   hideToolbar: false,
   hideBatchActions: false,
+  editabilityReason: undefined,
 })
 
 const emit = defineEmits<DataGridEmits>()
@@ -222,13 +225,28 @@ function handleClearFilter(column: string) {
 }
 
 // ── Row Operations ──
+function editabilityMessage(): string {
+  if (props.editabilityReason) {
+    return t(`components.dataGrid.row.editUnsupported.${props.editabilityReason}`)
+  }
+  return t('components.dataGrid.row.noTableContext')
+}
+
 function openEditDialog(rowIndex: number) {
+  if (!props.tableName) {
+    toast.warning(editabilityMessage())
+    return
+  }
   editingRowIndex.value = rowIndex
   isDuplicateRow.value = false
   editDialogOpen.value = true
 }
 
 function openDuplicateDialog(rowIndex: number) {
+  if (!props.tableName) {
+    toast.warning(editabilityMessage())
+    return
+  }
   editingRowIndex.value = rowIndex
   isDuplicateRow.value = true
   editDialogOpen.value = true
@@ -246,6 +264,10 @@ function openJsonDialog(value: unknown, column: string) {
 }
 
 function openDeleteDialog(rowIndex: number) {
+  if (!props.tableName) {
+    toast.warning(editabilityMessage())
+    return
+  }
   deletingRowIndex.value = rowIndex
   deleteDialogOpen.value = true
 }
@@ -584,8 +606,8 @@ watch(() => [props.rows, props.columns], () => {
           v-if="connectionId"
           class="bg-muted flex-shrink-0 w-20 right-0 sticky z-10"
         >
-          <div class="flex h-8 items-center justify-center px-2">
-            <span class="text-xs font-medium text-muted-foreground truncate">{{ $t('components.dataGrid.row.actions') }}</span>
+          <div class="px-2 flex h-8 items-center justify-center">
+            <span class="text-xs text-muted-foreground font-medium truncate">{{ $t('components.dataGrid.row.actions') }}</span>
           </div>
         </div>
       </div>
diff --git a/src/lang/enUS.ts b/src/lang/enUS.ts
index 5ec90b5..b78545f 100644
--- a/src/lang/enUS.ts
+++ b/src/lang/enUS.ts
@@ -782,6 +782,13 @@ export const enUS = {
         acknowledge: 'Keep as Orphan',
         close: 'Close Tab',
       },
+      executeOrphanDialog: {
+        title: 'Connection Changed',
+        message: 'The original connection for this query tab has changed: {from} → {to}. Rebind to the current active connection and execute?',
+        cancel: 'Cancel',
+        close: 'Close Tab',
+        rebindAndExecute: 'Rebind and Execute',
+      },
       explain: {
         title: 'Explain Plan',
         treeView: 'Tree',
@@ -1392,6 +1399,16 @@ export const enUS = {
         copyAsJson: 'Copy Row as JSON',
         copyAsInsert: 'Copy Row as INSERT',
         actions: 'Actions',
+        noTableContext: 'Query result is not tied to a table, so rows cannot be edited or deleted',
+        editUnsupported: {
+          'not-select': 'This statement is not a SELECT query; result rows cannot be edited or deleted',
+          'cte': 'Query uses WITH (CTE); rows cannot be mapped back to a table, so editing is not supported',
+          'set-operation': 'Query uses UNION/INTERSECT/EXCEPT; rows come from multiple statements, so editing is not supported',
+          'aggregation': 'Query is aggregated (GROUP BY/DISTINCT/aggregate functions); rows cannot be mapped back to base rows, so editing is not supported',
+          'multiple-sources': 'Query joins multiple tables; rows cannot be edited or deleted',
+          'no-table': 'Query has no FROM table; rows cannot be edited or deleted',
+          'complex-source': 'Query source is a subquery or complex expression; rows cannot be edited or deleted',
+        },
       },
       cell: {
         copyValue: 'Copy Cell Value',
diff --git a/src/lang/zhCN.ts b/src/lang/zhCN.ts
index 263c42b..3c20cb3 100644
--- a/src/lang/zhCN.ts
+++ b/src/lang/zhCN.ts
@@ -782,6 +782,13 @@ export const zhCN = {
         acknowledge: '保持孤立',
         close: '关闭标签页',
       },
+      executeOrphanDialog: {
+        title: '连接已变更',
+        message: '此查询标签的原始连接已变更：{from} → {to}。是否绑定到当前活跃连接并执行？',
+        cancel: '取消',
+        close: '关闭标签页',
+        rebindAndExecute: '绑定并执行',
+      },
       explain: {
         title: '执行计划',
         treeView: '树形',
@@ -1330,6 +1337,16 @@ export const zhCN = {
         copyAsJson: '复制为 JSON',
         copyAsInsert: '复制为 INSERT',
         actions: '操作',
+        noTableContext: '查询结果未关联具体表，无法编辑或删除行',
+        editUnsupported: {
+          'not-select': '当前语句不是查询，无法编辑或删除结果行',
+          'cte': '包含 WITH 公共表表达式，无法确定行来源，不支持编辑',
+          'set-operation': '包含 UNION/INTERSECT/EXCEPT，行来自多条语句，不支持编辑',
+          'aggregation': '包含聚合、GROUP BY 或 DISTINCT，结果行无法映射回原始行，不支持编辑',
+          'multiple-sources': '查询关联了多张表（JOIN），不支持编辑或删除行',
+          'no-table': '查询没有 FROM 表，不支持编辑或删除行',
+          'complex-source': '查询来源为子查询或复杂表达式，不支持编辑或删除行',
+        },
       },
       cell: {
         copyValue: '复制单元格值',
diff --git a/src/pages/QueriesPage.vue b/src/pages/QueriesPage.vue
index a1be829..050f52c 100644
--- a/src/pages/QueriesPage.vue
+++ b/src/pages/QueriesPage.vue
@@ -53,6 +53,8 @@ const destructiveAction = ref<{ type: 'drop' | 'truncate', table: TableInfo, dat
 const isDestructiveActionExecuting = ref(false)
 const showOrphanTabDialog = ref(false)
 const orphanTabToHandle = ref<string | null>(null)
+const showExecuteOrphanDialog = ref(false)
+const pendingExecuteSql = ref<string | null>(null)
 
 // ── Database action dialog state ──
 const createDatabaseDialogOpen = ref(false)
@@ -219,7 +221,7 @@ const listingTabObjects = computed(() => {
 })
 
 async function executeQuery(details?: StatementToExecute) {
-  if (!activeTab.value || activeTab.value.orphanFromConnectionId) {
+  if (!activeTab.value) {
     return
   }
 
@@ -234,6 +236,15 @@ async function executeQuery(details?: StatementToExecute) {
     return
   }
 
+  // Orphaned tabs point at a connection that is no longer active. Instead of
+  // silently dropping the execution, surface a dialog so the user can rebind
+  // the tab to the current connection, close it, or cancel.
+  if (activeTab.value.orphanFromConnectionId) {
+    pendingExecuteSql.value = sqlToExecute
+    showExecuteOrphanDialog.value = true
+    return
+  }
+
   const connId = getConnectionId()
   if (!connId) {
     toast.warning(t('pages.queries.notifications.selectConnectionFirst'))
@@ -408,6 +419,54 @@ function handleOrphanTabAcknowledge() {
   showOrphanTabDialog.value = false
 }
 
+const activeConnectionForOrphan = computed(() => {
+  const connId = getConnectionId()
+  if (!connId)
+    return null
+  return connectionStore.getConnectionById(connId)
+})
+
+const orphanConnectionPair = computed<{ from: string, to: string }>(() => {
+  const fromId = activeTab.value?.orphanFromConnectionId
+  const fromName = fromId
+    ? (connectionStore.getConnectionById(fromId)?.name ?? fromId)
+    : ''
+  return {
+    from: fromName,
+    to: activeConnectionForOrphan.value?.name ?? '',
+  }
+})
+
+function handleExecuteOrphanRebind() {
+  const tab = activeTab.value
+  const connId = getConnectionId()
+  const sql = pendingExecuteSql.value
+  if (!tab || !connId || !sql) {
+    showExecuteOrphanDialog.value = false
+    pendingExecuteSql.value = null
+    return
+  }
+  tabStore.reattachTab(tab.id, connId, selectedDatabase.value || connectionStore.getCurrentDatabase(connId) || undefined)
+  showExecuteOrphanDialog.value = false
+  pendingExecuteSql.value = null
+  showResultPanel.value = true
+  void tabStore.executeQuery(tab.id, sql)
+}
+
+function handleExecuteOrphanClose() {
+  const tab = activeTab.value
+  if (tab) {
+    tabStore.closeTab(tab.id)
+  }
+  showExecuteOrphanDialog.value = false
+  pendingExecuteSql.value = null
+}
+
+function handleExecuteOrphanCancel() {
+  showExecuteOrphanDialog.value = false
+  pendingExecuteSql.value = null
+}
+
 function handleTabClose(tabId: string) {
   tabStore.closeTab(tabId)
 }
@@ -1179,7 +1238,7 @@ function closeResultPanel() {
                       variant="ghost"
                       size="icon"
                       class="text-emerald-600 h-7 w-7 dark:text-emerald-400 hover:bg-emerald-500/10 disabled:opacity-40 dark:hover:bg-emerald-500/20"
-                      :disabled="!activeTab || activeTab.orphanFromConnectionId"
+                      :disabled="!activeTab"
                       @click="executeQuery"
                     >
                       <span class="i-lucide-play shrink-0 h-3.5 w-3.5" />
@@ -1353,7 +1412,7 @@ function closeResultPanel() {
               :error="activeTab?.error"
               :is-executing="activeTab?.isExecuting"
               :execution-time="activeTab?.executionTime"
-              :sql="activeTab?.content"
+              :sql="activeTab?.lastExecutedSql ?? activeTab?.content"
               :connection-id="getConnectionId() ?? undefined"
               :database="activeTab?.database"
               :explain-plan="activeTab?.explainPlan ?? null"
@@ -1428,6 +1487,38 @@ function closeResultPanel() {
       </AlertDialogContent>
     </AlertDialog>
 
+    <!-- Execute Orphaned Tab Dialog -->
+    <AlertDialog v-model:open="showExecuteOrphanDialog">
+      <AlertDialogContent>
+        <AlertDialogHeader>
+          <AlertDialogTitle>{{ t('pages.queries.executeOrphanDialog.title') }}</AlertDialogTitle>
+          <AlertDialogDescription>
+            {{ t('pages.queries.executeOrphanDialog.message', {
+              from: orphanConnectionPair.from,
+              to: orphanConnectionPair.to,
+            }) }}
+          </AlertDialogDescription>
+        </AlertDialogHeader>
+        <AlertDialogFooter>
+          <AlertDialogCancel @click="handleExecuteOrphanCancel">
+            {{ t('pages.queries.executeOrphanDialog.cancel') }}
+          </AlertDialogCancel>
+          <Button
+            variant="outline"
+            @click="handleExecuteOrphanClose"
+          >
+            {{ t('pages.queries.executeOrphanDialog.close') }}
+          </Button>
+          <Button
+            :disabled="!activeConnectionForOrphan"
+            @click="handleExecuteOrphanRebind"
+          >
+            {{ t('pages.queries.executeOrphanDialog.rebindAndExecute') }}
+          </Button>
+        </AlertDialogFooter>
+      </AlertDialogContent>
+    </AlertDialog>
+
     <!-- Create Database dialog -->
     <CreateDatabaseDialog
       :open="createDatabaseDialogOpen"
diff --git a/src/store/tabStore.ts b/src/store/tabStore.ts
index 4b79623..1e9a940 100644
--- a/src/store/tabStore.ts
+++ b/src/store/tabStore.ts
@@ -48,6 +48,8 @@ export type QueryTab = {
   results?: QueryResult
   error?: ApiError | string
   executionTime?: number
+  /** The exact SQL statement that produced `results`. May differ from `content` (a multi-statement editor buffer). Used to map results back to a table. */
+  lastExecutedSql?: string
   tableView?: TableViewMeta
   erDiagram?: ErDiagramMeta
   listingTab?: ListingTabMeta
@@ -324,6 +326,7 @@ export const useTabStore = defineStore('tabs', {
         return
       }
 
+      tab.lastExecutedSql = sql
       const activeConnectionId = tab.connectionId
       if (!activeConnectionId) {
         return
@@ -472,6 +475,21 @@ export const useTabStore = defineStore('tabs', {
       }
     },
 
+    /**
+     * Rebind an orphaned tab to a connection so its SQL can execute again.
+     * Clears the orphan marker and points the tab at the given connection.
+     */
+    reattachTab(tabId: string, connectionId: string, database?: string) {
+      const tab = this.tabs.find(t => t.id === tabId)
+      if (!tab)
+        return
+      tab.connectionId = connectionId
+      tab.orphanFromConnectionId = undefined
+      if (database !== undefined) {
+        tab.database = database
+      }
+    },
+
     markTabSaved(tabId: string, filePath?: string) {
       const tab = this.tabs.find(t => t.id === tabId)
       if (tab) {