diff --git a/docs/architecture/backend/data-transfer.md b/docs/architecture/backend/data-transfer.md
index d95a3fe0d..8a6bb5cfb 100644
--- a/docs/architecture/backend/data-transfer.md
+++ b/docs/architecture/backend/data-transfer.md
@@ -81,7 +81,7 @@ peek 不消耗计划；claim 原子将 available 转为 executing 并建立活
 
 ## 6. SQL 文件与取消
 
-`data_transfer/sql_file.rs` 负责 SQL 文件输出；SQL 文件目的地不建立目标数据库写会话，但仍需源读取与渲染能力。文件输出失败或取消不能标成完整产物；结果必须表达已输出部分和错误。
+`data_transfer/sql_file.rs` 负责 SQL 文件输出；SQL 文件目的地不建立目标数据库写会话，但仍需源读取与渲染能力。INSERT 字面量必须由目标 SQL 驱动声明的 `SqlLiteralDialect` 生成；无方言或值无法安全表示时停止生成，不能回退到默认转义。文件输出失败或取消不能标成完整产物；结果必须表达已输出部分和错误。
 
 旧执行与兼容路径仍由 `commands/data_transfer/exec.rs` 等入口管理；P5 新路径在 `commands/data_transfer/job_api/`。新路径的 cancel intent 持久写入本机 Job repository，运行时通过 `CancelWatch` 将它传给当前 stage。该 SQLite repository 是单机桌面 host，不是团队服务跨实例协调器。Job 只存 Artifact ID 引用，不存文件字节；当前引用 TTL 为 30 天，且不保证仍可下载对应内容。
 
diff --git a/docs/architecture/backend/drivers.md b/docs/architecture/backend/drivers.md
index e6c1144a0..3f8ede0ef 100644
--- a/docs/architecture/backend/drivers.md
+++ b/docs/architecture/backend/drivers.md
@@ -32,7 +32,7 @@ Driver 基础能力包括：
 - connect / test_connection / disconnect
 - get_databases / get_tables / get_table_schema
 - query / query_multi / query_stream
-- query_with_params / execute
+- query_with_params / execute / execute_with_params
 - transaction
 - EXPLAIN
 - Driver Commands
@@ -51,6 +51,10 @@ cleanup_query_execution
 
 只有实际声明精确取消能力的 Driver 才会被 Host 当作 cancellable；兼容默认实现不会自动获得取消能力。
 
+Host 生成的写语句必须把 SQL 模板与 `Vec<Value>` 分开，通过 `parameter_placeholder()` 和 `execute_with_params()` 传值；没有绑定写能力的驱动返回 `Unsupported`，Host 在开启事务前拒绝该批写入。`build_update_statement()` / `build_delete_statement()` 生成的预览只含占位符，不能把值插入 SQL 文本。
+
+SQL 筛选与 SQL 文件产物使用 `try_format_sql_literal()`。每个 SQL 驱动显式声明 `SqlLiteralDialect`，共享格式器按方言使用对会话转义模式稳定的表示（例如 MySQL 的 UTF-8 十六进制转换、PostgreSQL / DuckDB 的 dollar-quoted 文本）；未声明方言时返回 `Unsupported`，不会猜测反斜杠规则。无方言的键值 / 文档驱动不应调用 SQL 字面量接口。旧的无结果 `format_sql_literal()` 和内插式 `build_*_sql()` 仅保留兼容，产品写路径不使用它们。
+
 ### 2.1 Schema 元数据目标
 
 `get_tables`、`get_table_schema`、`get_columns` 和 `get_all_columns` 都接收显式 `database` 与可选 `schema`，驱动按传入目标读取，不依赖会话当前选中的数据库。Host 的 `list_catalog`、`read_relation_columns`、`read_relation_schema` 和 `refresh_schema_metadata` Driver Commands 以 `RelationRef { database, schema, name }` 表达目标；前端与扩展共用 Driver Command 网关，不再需要按数据库类型选择旧 Host schema IPC。
diff --git a/docs/architecture/backend/services.md b/docs/architecture/backend/services.md
index 9da5a70bf..d5c7d2536 100644
--- a/docs/architecture/backend/services.md
+++ b/docs/architecture/backend/services.md
@@ -129,6 +129,7 @@ DataTable 的行编辑与数据导出是两条不同的写路径，不要按同
 
 - **创建 / 复用**：都只在调用方已建的会话上执行。会话找不到时走同一条透明重建路径。
 - **事务**：预览待提交改动**不**开事务，只读；真正执行改动计划时，若该会话已有显式事务就复用，否则自己开一个。语义不明确的写入（例如无法判定影响行数）一律拒绝，不做「尽力而为」。
+- **写入参数**：UPDATE / DELETE 的 SQL 只包含驱动占位符；列值与主键值通过驱动绑定 API 单独传递。驱动先构造整批语句，确认占位符与绑定执行均受支持后才开启事务。预览中的 SQL 模板不包含数据值。
 - **关闭**：不涉及。
 - **取消**：无独立的取消句柄——粒度是「提交前 / 提交后」，不是执行中途。
 - **配置目标**：提交前复核四件事：改动计划里的表必须属于这条会话的**所属连接**、驱动类型与 database/schema 必须与当前配置一致、连接不得只读、以及结构指纹是否仍然匹配。任一不符即中止，避免把过期界面上的改动写到已经变化的表上。表所属连接是从会话反查出来的（见 [2.1](#21-连接-id-约定)），该反查依赖 owner 映射在空闲回收后仍然保留。
@@ -136,6 +137,7 @@ DataTable 的行编辑与数据导出是两条不同的写路径，不要按同
 **数据导出**
 
 - **创建 / 复用**：只消费调用方的 `dbSessionId`，不创建；导出前先过 SQL 安全门闸，再以流式回调逐批输出。
+- **SQL 格式**：查询筛选和 SQL INSERT 导出使用驱动显式声明的字面量方言；不支持该方言或遇到无法表示的值时返回错误，不能退回共享反斜杠替换。CSV / JSON 导出不经过 SQL 字面量格式器。
 - **关闭 / 取消**：**导出没有中途取消机制**——没有取消令牌也没有取消标志位，与 Data Sync / Data Transfer 的 job 标志是两种模型。导出结果里的「已取消」只表示用户在保存对话框里放弃了选文件，也就是导出根本没开始；一旦开始流式写盘，就只能由查询本身失败而结束。
 - **配置目标**：database / schema 随导出请求传递，并从会话配置补齐缺省项，不读会话当前状态。
 
diff --git a/packages/data-transfer/src/job/sqlfile.rs b/packages/data-transfer/src/job/sqlfile.rs
index eda6f9f45..2a9a77bee 100644
--- a/packages/data-transfer/src/job/sqlfile.rs
+++ b/packages/data-transfer/src/job/sqlfile.rs
@@ -46,6 +46,33 @@ impl DataTransferHandler {
                 )
                 .await;
                 match result {
+                    Ok(res) if res.cancelled || res.partial => {
+                        let cancelled = res.cancelled;
+                        tracing::warn!(
+                            destination = destination.display().to_string(),
+                            cancelled,
+                            table_errors = ?res.tables.iter().filter_map(|table| table.error.as_deref()).collect::<Vec<_>>(),
+                            "SQL file was not published because generation was incomplete"
+                        );
+                        Ok(StageOutcome {
+                            stage_id: spec.stage_id.clone(),
+                            terminal: if cancelled {
+                                StageTerminal::Cancelled
+                            } else {
+                                StageTerminal::Failed
+                            },
+                            progress: JobProgress::default(),
+                            commit_boundaries: Vec::new(),
+                            execution_ids: Vec::new(),
+                            artifact_ids: Vec::new(),
+                            effect_outcome: EffectOutcome::RolledBack,
+                            error_code: Some(if cancelled {
+                                ExecutionErrorCode::Cancelled
+                            } else {
+                                ExecutionErrorCode::SqlError
+                            }),
+                        })
+                    }
                     Ok(res) => {
                         let digest = artifact_digest(destination)?;
                         let boundary = commit_boundary(
diff --git a/packages/data-transfer/src/job/tests.rs b/packages/data-transfer/src/job/tests.rs
index e70a93a1d..81b19d23b 100644
--- a/packages/data-transfer/src/job/tests.rs
+++ b/packages/data-transfer/src/job/tests.rs
@@ -88,6 +88,10 @@ fn unsupported<T>() -> Result<T, DriverError> {
 
 #[async_trait]
 impl DatabaseDriver for FakeDb {
+    fn sql_literal_dialect(&self) -> Option<SqlLiteralDialect> {
+        Some(SqlLiteralDialect::Postgres)
+    }
+
     async fn cancel_query(&self, _: &ConnectionHandle) -> Result<(), DriverError> {
         Ok(())
     }
diff --git a/packages/data-transfer/src/sql_file.rs b/packages/data-transfer/src/sql_file.rs
index 5522a1750..3ca8f65e2 100644
--- a/packages/data-transfer/src/sql_file.rs
+++ b/packages/data-transfer/src/sql_file.rs
@@ -724,13 +724,15 @@ fn insert_sql_batch(
                     mappings.len()
                 )));
             }
-            Ok(format!(
-                "({})",
-                row.iter()
-                    .map(|value| driver.format_sql_literal(value))
-                    .collect::<Vec<_>>()
-                    .join(", ")
-            ))
+            let values = row
+                .iter()
+                .map(|value| {
+                    driver
+                        .try_format_sql_literal(value)
+                        .map_err(|error| TransferError::unsupported(error.to_string()))
+                })
+                .collect::<Result<Vec<_>, _>>()?;
+            Ok(format!("({})", values.join(", ")))
         })
         .collect::<Result<Vec<_>, TransferError>>()?;
     render_sql_file_insert(
@@ -1513,7 +1515,7 @@ mod tests {
         )
         .unwrap();
         assert!(sql.contains("\"display_name\""));
-        assert!(sql.contains("'O''Reilly'"));
+        assert!(sql.contains("$datazen$O'Reilly$datazen$"));
     }
 
     #[test]
diff --git a/packages/driver-api/src/mock_driver.rs b/packages/driver-api/src/mock_driver.rs
index 8aeaa3dff..eb69e3e99 100644
--- a/packages/driver-api/src/mock_driver.rs
+++ b/packages/driver-api/src/mock_driver.rs
@@ -11,7 +11,7 @@ use crate::{
     is_schema_object_command, query_command_definition, query_stream_command_definition,
     schema_catalog_command_definitions, schema_object_command_definitions,
     try_execute_schema_catalog_command, validate_schema_target, CommandResult, DdlAtomicity,
-    DriverCommandDefinition, SchemaScope,
+    DriverCommandDefinition, SchemaScope, SqlLiteralDialect,
 };
 use crate::{
     ColumnInfo, ColumnSchema, ConnectionConfig, ConnectionHandle, DatabaseDriver, DatabaseType,
@@ -386,6 +386,18 @@ impl MockDriver {
 
 #[async_trait]
 impl DatabaseDriver for MockDriver {
+    fn sql_literal_dialect(&self) -> Option<SqlLiteralDialect> {
+        match self.db_type.to_ascii_lowercase().as_str() {
+            "clickhouse" => Some(SqlLiteralDialect::ClickHouse),
+            "duckdb" => Some(SqlLiteralDialect::DuckDb),
+            "mysql" | "mariadb" | "doris" | "starrocks" => Some(SqlLiteralDialect::MySql),
+            "postgres" | "postgresql" => Some(SqlLiteralDialect::Postgres),
+            "sqlserver" | "mssql" => Some(SqlLiteralDialect::SqlServer),
+            "sqlite" | "turso" | "rqlite" => Some(SqlLiteralDialect::Sqlite),
+            _ => None,
+        }
+    }
+
     fn driver_type(&self) -> DatabaseType {
         self.db_type.clone()
     }
@@ -394,6 +406,10 @@ impl DatabaseDriver for MockDriver {
         self.opts.category.clone()
     }
 
+    fn supports_bound_writes(&self) -> bool {
+        self.opts.parameterized_writes
+    }
+
     fn default_host(&self) -> Option<&'static str> {
         self.opts.default_host
     }
diff --git a/packages/driver-api/src/reuse.rs b/packages/driver-api/src/reuse.rs
index 07b1f98df..bae679e55 100644
--- a/packages/driver-api/src/reuse.rs
+++ b/packages/driver-api/src/reuse.rs
@@ -176,6 +176,18 @@ impl DatabaseDriver for ReuseDriver {
         self.inner.format_sql_literal(value)
     }
 
+    fn sql_literal_dialect(&self) -> Option<SqlLiteralDialect> {
+        self.inner.sql_literal_dialect()
+    }
+
+    fn supports_bound_writes(&self) -> bool {
+        self.inner.supports_bound_writes()
+    }
+
+    fn try_format_sql_literal(&self, value: &Option<Value>) -> Result<String, DriverError> {
+        self.inner.try_format_sql_literal(value)
+    }
+
     fn build_update_sql(
         &self,
         table: &str,
diff --git a/packages/driver-api/src/sql_dump/dump.rs b/packages/driver-api/src/sql_dump/dump.rs
index a69f5b0bf..0ee5cea23 100644
--- a/packages/driver-api/src/sql_dump/dump.rs
+++ b/packages/driver-api/src/sql_dump/dump.rs
@@ -251,12 +251,14 @@ where
             let tuples: Vec<String> = result
                 .rows
                 .iter()
-                .map(|row| {
-                    let vals: Vec<String> =
-                        row.iter().map(|v| driver.format_sql_literal(v)).collect();
-                    format!("({})", vals.join(", "))
+                .map(|row| -> Result<String, DriverError> {
+                    let vals: Vec<String> = row
+                        .iter()
+                        .map(|value| driver.try_format_sql_literal(value))
+                        .collect::<Result<_, _>>()?;
+                    Ok(format!("({})", vals.join(", ")))
                 })
-                .collect();
+                .collect::<Result<_, _>>()?;
             append_batched_inserts(
                 out,
                 &rel,
diff --git a/packages/driver-api/src/traits.rs b/packages/driver-api/src/traits.rs
index 438480f8c..39f3a4f9f 100644
--- a/packages/driver-api/src/traits.rs
+++ b/packages/driver-api/src/traits.rs
@@ -161,10 +161,26 @@ pub trait DatabaseDriver: Send + Sync {
         DdlAtomicity::Unknown
     }
 
+    /// Legacy best-effort SQL literal formatter. New code should use
+    /// [`Self::try_format_sql_literal`], which requires an explicit dialect.
     fn format_sql_literal(&self, value: &Option<Value>) -> String {
         sql_text::format_sql_literal(value)
     }
 
+    /// The literal grammar this driver can safely render for SQL artifacts.
+    /// Drivers with session-dependent or unsupported syntax must leave this
+    /// unset; their export/filter callers then fail closed.
+    fn sql_literal_dialect(&self) -> Option<SqlLiteralDialect> {
+        None
+    }
+
+    fn try_format_sql_literal(&self, value: &Option<Value>) -> Result<String, DriverError> {
+        let dialect = self.sql_literal_dialect().ok_or_else(|| {
+            DriverError::Unsupported("this driver has no declared SQL literal formatter".into())
+        })?;
+        sql_text::format_sql_literal_for_dialect(value, dialect)
+    }
+
     fn build_update_sql(
         &self,
         table: &str,
@@ -179,6 +195,25 @@ pub trait DatabaseDriver: Send + Sync {
         sql_text::build_delete_sql(self, table, pk_columns)
     }
 
+    /// Build an UPDATE with driver placeholders and values kept out of SQL.
+    fn build_update_statement(
+        &self,
+        table: &str,
+        set_columns: &[(&str, Option<Value>)],
+        pk_columns: &[(&str, Option<Value>)],
+    ) -> Result<BoundSqlStatement, DriverError> {
+        sql_text::build_update_statement(self, table, set_columns, pk_columns)
+    }
+
+    /// Build a DELETE with driver placeholders and values kept out of SQL.
+    fn build_delete_statement(
+        &self,
+        table: &str,
+        pk_columns: &[(&str, Option<Value>)],
+    ) -> Result<BoundSqlStatement, DriverError> {
+        sql_text::build_delete_statement(self, table, pk_columns)
+    }
+
     /// The host this driver dials when the connection config leaves `host` unset.
     ///
     /// `connect` resolves that default internally, so a config that omits
@@ -367,6 +402,13 @@ pub trait DatabaseDriver: Send + Sync {
         params: &[Value],
     ) -> Result<QueryResult, DriverError>;
 
+    /// Whether this driver implements both placeholder generation and bound
+    /// DML execution. Callers use this to reject writes before opening a
+    /// transaction.
+    fn supports_bound_writes(&self) -> bool {
+        false
+    }
+
     /// Render a parameter for this dialect. Unsupported drivers must fail before writes.
     /// `data_type` comes from the inspected target column metadata.
     fn parameter_placeholder(
diff --git a/packages/driver-api/src/traits/sql_text.rs b/packages/driver-api/src/traits/sql_text.rs
index 2f913bfb7..c8d95266f 100644
--- a/packages/driver-api/src/traits/sql_text.rs
+++ b/packages/driver-api/src/traits/sql_text.rs
@@ -11,7 +11,7 @@
 
 use super::DatabaseDriver;
 use crate::sql_target::SqlTarget;
-use crate::types::{PaginationSyntax, Value};
+use crate::types::{BoundSqlStatement, DriverError, PaginationSyntax, SqlLiteralDialect, Value};
 
 /// Lowercase hexadecimal encoding used by the default literal formatter.
 ///
@@ -22,6 +22,14 @@ fn bytes_to_hex(bytes: &[u8]) -> String {
     bytes.iter().map(|byte| format!("{byte:02x}")).collect()
 }
 
+fn utf16le_to_hex(text: &str) -> String {
+    let bytes = text
+        .encode_utf16()
+        .flat_map(u16::to_le_bytes)
+        .collect::<Vec<_>>();
+    bytes_to_hex(&bytes)
+}
+
 pub(crate) fn quote_ident<D: DatabaseDriver + ?Sized>(driver: &D, name: &str) -> String {
     let q = driver.quote_char();
     if q == '`' {
@@ -47,38 +55,101 @@ pub(crate) fn pagination_syntax<D: DatabaseDriver + ?Sized>(
     }
 }
 
+pub(crate) fn format_sql_literal_for_dialect(
+    value: &Option<Value>,
+    dialect: SqlLiteralDialect,
+) -> Result<String, DriverError> {
+    let string_literal = |value: &str| -> Result<String, DriverError> {
+        if value.contains('\0')
+            && matches!(
+                dialect,
+                SqlLiteralDialect::DuckDb | SqlLiteralDialect::Postgres
+            )
+        {
+            return Err(DriverError::Unsupported(
+                "this server does not accept NUL characters in text literals".into(),
+            ));
+        }
+        Ok(match dialect {
+            SqlLiteralDialect::ClickHouse => format!("unhex('{}')", bytes_to_hex(value.as_bytes())),
+            SqlLiteralDialect::MySql => {
+                format!(
+                    "CONVERT(X'{}' USING utf8mb4)",
+                    bytes_to_hex(value.as_bytes())
+                )
+            }
+            SqlLiteralDialect::DuckDb | SqlLiteralDialect::Postgres => {
+                let mut tag = "datazen".to_string();
+                let mut delimiter = format!("${tag}$");
+                let mut suffix = 0usize;
+                while value.contains(&delimiter) {
+                    suffix += 1;
+                    tag = format!("datazen_{suffix}");
+                    delimiter = format!("${tag}$");
+                }
+                format!("{delimiter}{value}{delimiter}")
+            }
+            SqlLiteralDialect::SqlServer => {
+                format!("CONVERT(nvarchar(max), 0x{})", utf16le_to_hex(value))
+            }
+            SqlLiteralDialect::Sqlite => {
+                format!("CAST(X'{}' AS TEXT)", bytes_to_hex(value.as_bytes()))
+            }
+        })
+    };
+
+    let bytes_literal = |bytes: &[u8]| match dialect {
+        SqlLiteralDialect::ClickHouse => format!("unhex('{}')", bytes_to_hex(bytes)),
+        SqlLiteralDialect::DuckDb => format!("from_hex('{}')", bytes_to_hex(bytes)),
+        SqlLiteralDialect::MySql | SqlLiteralDialect::Sqlite => {
+            format!("X'{}'", bytes_to_hex(bytes))
+        }
+        SqlLiteralDialect::Postgres => {
+            format!("decode('{}', 'hex')", bytes_to_hex(bytes))
+        }
+        SqlLiteralDialect::SqlServer => format!("0x{}", bytes_to_hex(bytes)),
+    };
+
+    match value {
+        None | Some(Value::Null) => Ok("NULL".to_string()),
+        Some(Value::Bool(b)) => Ok(match (dialect, b) {
+            (
+                SqlLiteralDialect::MySql | SqlLiteralDialect::Sqlite | SqlLiteralDialect::SqlServer,
+                true,
+            ) => "1",
+            (
+                SqlLiteralDialect::MySql | SqlLiteralDialect::Sqlite | SqlLiteralDialect::SqlServer,
+                false,
+            ) => "0",
+            (_, true) => "TRUE",
+            (_, false) => "FALSE",
+        }
+        .to_string()),
+        Some(Value::Integer(i)) => Ok(i.to_string()),
+        Some(Value::Float(f)) if f.is_finite() => Ok(f.to_string()),
+        Some(Value::Float(_)) => Err(DriverError::Unsupported(
+            "non-finite floating-point values cannot be rendered as SQL literals".into(),
+        )),
+        Some(Value::String(s) | Value::Timestamp(s)) => string_literal(s),
+        Some(Value::Bytes(bytes)) => Ok(bytes_literal(bytes)),
+        Some(Value::Json(json)) => string_literal(&json.to_string()),
+    }
+}
+
+/// Historical, infallible ANSI-style formatter kept for source compatibility.
+/// Product code must call `try_format_sql_literal` through the driver trait.
 pub(crate) fn format_sql_literal(value: &Option<Value>) -> String {
     match value {
         None | Some(Value::Null) => "NULL".to_string(),
-        Some(Value::Bool(b)) => {
-            if *b {
-                "TRUE".to_string()
-            } else {
-                "FALSE".to_string()
-            }
-        }
+        Some(Value::Bool(true)) => "TRUE".to_string(),
+        Some(Value::Bool(false)) => "FALSE".to_string(),
         Some(Value::Integer(i)) => i.to_string(),
         Some(Value::Float(f)) => f.to_string(),
-        Some(Value::String(s)) => {
-            let escaped = s.replace('\\', "\\\\");
-            format!("'{}'", escaped.replace('\'', "''"))
-        }
-        Some(Value::Bytes(b)) => {
-            // Keep the default dialect conservative and lossless. Drivers
-            // with a stricter binary-literal grammar should override this
-            // method (PostgreSQL uses bytea hex input; MySQL/SQLite use
-            // X'...'). Never turn arbitrary bytes into replacement UTF-8.
-            format!("X'{}'", bytes_to_hex(b))
-        }
-        Some(Value::Timestamp(s)) => {
-            let escaped = s.replace('\\', "\\\\");
-            format!("'{}'", escaped.replace('\'', "''"))
-        }
-        Some(Value::Json(j)) => {
-            let s = j.to_string();
-            let escaped = s.replace('\\', "\\\\");
-            format!("'{}'", escaped.replace('\'', "''"))
+        Some(Value::String(s) | Value::Timestamp(s)) => {
+            format!("'{}'", s.replace('\'', "''"))
         }
+        Some(Value::Bytes(b)) => format!("X'{}'", bytes_to_hex(b)),
+        Some(Value::Json(j)) => format!("'{}'", j.to_string().replace('\'', "''")),
     }
 }
 
@@ -117,6 +188,88 @@ pub(crate) fn build_update_sql<D: DatabaseDriver + ?Sized>(
     )
 }
 
+pub(crate) fn build_update_statement<D: DatabaseDriver + ?Sized>(
+    driver: &D,
+    table: &str,
+    set_columns: &[(&str, Option<Value>)],
+    pk_columns: &[(&str, Option<Value>)],
+) -> Result<BoundSqlStatement, DriverError> {
+    if !driver.supports_bound_writes() {
+        return Err(DriverError::Unsupported(
+            "parameterized row writes are not supported".into(),
+        ));
+    }
+    let mut parameters = Vec::new();
+    let mut bind = |value: &Option<Value>| -> Result<String, DriverError> {
+        let placeholder = driver.parameter_placeholder(parameters.len() + 1, None)?;
+        parameters.push(value.clone().unwrap_or(Value::Null));
+        Ok(placeholder)
+    };
+
+    let mut set_clauses = Vec::with_capacity(set_columns.len());
+    for (column, value) in set_columns {
+        set_clauses.push(format!("{} = {}", driver.quote_ident(column), bind(value)?));
+    }
+    let mut where_clauses = Vec::with_capacity(pk_columns.len());
+    for (column, value) in pk_columns {
+        let quoted = driver.quote_ident(column);
+        if value
+            .as_ref()
+            .map_or(true, |value| matches!(value, Value::Null))
+        {
+            where_clauses.push(format!("{quoted} IS NULL"));
+        } else {
+            where_clauses.push(format!("{quoted} = {}", bind(value)?));
+        }
+    }
+
+    Ok(BoundSqlStatement {
+        sql: format!(
+            "UPDATE {} SET {} WHERE {}",
+            driver.quote_ident(table),
+            set_clauses.join(", "),
+            where_clauses.join(" AND ")
+        ),
+        parameters,
+    })
+}
+
+pub(crate) fn build_delete_statement<D: DatabaseDriver + ?Sized>(
+    driver: &D,
+    table: &str,
+    pk_columns: &[(&str, Option<Value>)],
+) -> Result<BoundSqlStatement, DriverError> {
+    if !driver.supports_bound_writes() {
+        return Err(DriverError::Unsupported(
+            "parameterized row writes are not supported".into(),
+        ));
+    }
+    let mut parameters = Vec::new();
+    let mut where_clauses = Vec::with_capacity(pk_columns.len());
+    for (column, value) in pk_columns {
+        let quoted = driver.quote_ident(column);
+        if value
+            .as_ref()
+            .map_or(true, |value| matches!(value, Value::Null))
+        {
+            where_clauses.push(format!("{quoted} IS NULL"));
+        } else {
+            let placeholder = driver.parameter_placeholder(parameters.len() + 1, None)?;
+            parameters.push(value.clone().unwrap_or(Value::Null));
+            where_clauses.push(format!("{quoted} = {placeholder}"));
+        }
+    }
+
+    Ok(BoundSqlStatement {
+        sql: format!(
+            "DELETE FROM {} WHERE {}",
+            driver.quote_ident(table),
+            where_clauses.join(" AND ")
+        ),
+        parameters,
+    })
+}
+
 pub(crate) fn build_delete_sql<D: DatabaseDriver + ?Sized>(
     driver: &D,
     table: &str,
@@ -159,3 +312,109 @@ pub(crate) fn split_restore_sql<D: DatabaseDriver + ?Sized>(driver: &D, sql: &st
     out.extend(scanner.finish());
     out
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::mock_driver::{MockDriver, MockDriverOptions};
+
+    #[test]
+    fn literal_formatters_do_not_reinterpret_backslashes() {
+        let text = "quote' slash\\ trailing\\\n雪 $datazen$";
+        let value = Some(Value::String(text.to_string()));
+        let hex = bytes_to_hex(text.as_bytes());
+
+        assert_eq!(
+            format_sql_literal_for_dialect(&value, SqlLiteralDialect::MySql).unwrap(),
+            format!("CONVERT(X'{hex}' USING utf8mb4)")
+        );
+        assert_eq!(
+            format_sql_literal_for_dialect(&value, SqlLiteralDialect::ClickHouse).unwrap(),
+            format!("unhex('{hex}')")
+        );
+        assert_eq!(
+            format_sql_literal_for_dialect(&value, SqlLiteralDialect::Sqlite).unwrap(),
+            format!("CAST(X'{hex}' AS TEXT)")
+        );
+        assert_eq!(
+            format_sql_literal_for_dialect(&value, SqlLiteralDialect::Postgres).unwrap(),
+            format!("$datazen_1${text}$datazen_1$")
+        );
+        assert_eq!(
+            format_sql_literal_for_dialect(&value, SqlLiteralDialect::DuckDb).unwrap(),
+            format!("$datazen_1${text}$datazen_1$")
+        );
+        assert_eq!(
+            format_sql_literal_for_dialect(&value, SqlLiteralDialect::SqlServer).unwrap(),
+            format!("CONVERT(nvarchar(max), 0x{})", utf16le_to_hex(text))
+        );
+    }
+
+    #[test]
+    fn postgres_and_duckdb_text_literals_reject_nul_bytes() {
+        let value = Some(Value::String("before\0after".to_string()));
+        for dialect in [SqlLiteralDialect::Postgres, SqlLiteralDialect::DuckDb] {
+            assert!(format_sql_literal_for_dialect(&value, dialect).is_err());
+        }
+        assert!(format_sql_literal_for_dialect(&value, SqlLiteralDialect::Sqlite).is_ok());
+    }
+
+    #[test]
+    fn parameterized_update_and_delete_keep_values_out_of_sql() {
+        let mut options = MockDriverOptions::default();
+        options.parameterized_writes = true;
+        let driver = MockDriver::new("sqlite", options);
+        let update = driver
+            .build_update_statement(
+                "rows",
+                &[
+                    ("text", Some(Value::String("x'\\y".to_string()))),
+                    ("nullable", None),
+                ],
+                &[
+                    ("id", Some(Value::Integer(7))),
+                    ("tenant", Some(Value::Null)),
+                ],
+            )
+            .unwrap();
+
+        assert_eq!(
+            update.sql,
+            "UPDATE \"rows\" SET \"text\" = ?1, \"nullable\" = ?2 WHERE \"id\" = ?3 AND \"tenant\" IS NULL"
+        );
+        assert_eq!(update.parameters.len(), 3);
+        assert!(matches!(&update.parameters[0], Value::String(value) if value == "x'\\y"));
+        assert!(matches!(update.parameters[1], Value::Null));
+        assert!(matches!(update.parameters[2], Value::Integer(7)));
+
+        let delete = driver
+            .build_delete_statement(
+                "rows",
+                &[
+                    ("id", Some(Value::Integer(7))),
+                    ("tenant", Some(Value::Null)),
+                ],
+            )
+            .unwrap();
+        assert_eq!(
+            delete.sql,
+            "DELETE FROM \"rows\" WHERE \"id\" = ?1 AND \"tenant\" IS NULL"
+        );
+        assert!(matches!(delete.parameters.as_slice(), [Value::Integer(7)]));
+    }
+
+    #[test]
+    fn parameterized_statement_refuses_drivers_without_binding_support() {
+        let driver = MockDriver::new("sqlite", MockDriverOptions::default());
+        let result = driver.build_delete_statement("rows", &[("id", Some(Value::Integer(1)))]);
+        assert!(matches!(result, Err(DriverError::Unsupported(_))));
+    }
+
+    #[test]
+    fn legacy_generic_literal_formatter_does_not_double_backslashes() {
+        assert_eq!(
+            format_sql_literal(&Some(Value::String("a\\b".into()))),
+            "'a\\b'"
+        );
+    }
+}
diff --git a/packages/driver-api/src/types.rs b/packages/driver-api/src/types.rs
index ec984b986..eeeecb390 100644
--- a/packages/driver-api/src/types.rs
+++ b/packages/driver-api/src/types.rs
@@ -264,6 +264,25 @@ pub enum Value {
     Json(serde_json::Value),
 }
 
+/// SQL literal syntax supported by a driver for generated SQL artifacts and
+/// read-only filter statements. Runtime writes must use bound parameters.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum SqlLiteralDialect {
+    ClickHouse,
+    DuckDb,
+    MySql,
+    Postgres,
+    SqlServer,
+    Sqlite,
+}
+
+/// A SQL statement whose values remain separate from its SQL text.
+#[derive(Debug, Clone)]
+pub struct BoundSqlStatement {
+    pub sql: String,
+    pub parameters: Vec<Value>,
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct ColumnInfo {
diff --git a/packages/drivers/clickhouse/src/clickhouse.rs b/packages/drivers/clickhouse/src/clickhouse.rs
index ea3e6f503..3f6a4614b 100644
--- a/packages/drivers/clickhouse/src/clickhouse.rs
+++ b/packages/drivers/clickhouse/src/clickhouse.rs
@@ -186,6 +186,10 @@ impl ClickHouseDriver {
 
 #[async_trait]
 impl DatabaseDriver for ClickHouseDriver {
+    fn sql_literal_dialect(&self) -> Option<SqlLiteralDialect> {
+        Some(SqlLiteralDialect::ClickHouse)
+    }
+
     fn has_multi_database(&self) -> bool {
         true
     }
diff --git a/packages/drivers/duckdb/src/duckdb.rs b/packages/drivers/duckdb/src/duckdb.rs
index 819719e5d..c06dd7514 100644
--- a/packages/drivers/duckdb/src/duckdb.rs
+++ b/packages/drivers/duckdb/src/duckdb.rs
@@ -123,6 +123,10 @@ impl DuckDbDriver {
 
 #[async_trait]
 impl DatabaseDriver for DuckDbDriver {
+    fn sql_literal_dialect(&self) -> Option<SqlLiteralDialect> {
+        Some(SqlLiteralDialect::DuckDb)
+    }
+
     fn driver_type(&self) -> DatabaseType {
         "duckdb".to_string()
     }
diff --git a/packages/drivers/http-support/tests/support/real_driver_contract_plumbing.rs b/packages/drivers/http-support/tests/support/real_driver_contract_plumbing.rs
index ac286254e..551864925 100644
--- a/packages/drivers/http-support/tests/support/real_driver_contract_plumbing.rs
+++ b/packages/drivers/http-support/tests/support/real_driver_contract_plumbing.rs
@@ -10,8 +10,8 @@ use std::time::{SystemTime, UNIX_EPOCH};
 
 use datazen_driver_api::{
     async_trait, ConnectionConfig, ConnectionHandle, DatabaseDriver, DatabaseType, DdlAtomicity,
-    DriverError, QueryExecutionId, QueryResult, ServerInfo, SqlTarget, SslMode, StatementResult,
-    TableInfo, TableSchema, Value,
+    DriverError, QueryExecutionId, QueryResult, ServerInfo, SqlLiteralDialect, SqlTarget, SslMode,
+    StatementResult, TableInfo, TableSchema, Value,
 };
 
 use super::Dialect;
@@ -635,6 +635,18 @@ impl<D: DatabaseDriver> DatabaseDriver for WithheldPreciseCancel<D> {
         self.inner.format_sql_literal(value)
     }
 
+    fn sql_literal_dialect(&self) -> Option<SqlLiteralDialect> {
+        self.inner.sql_literal_dialect()
+    }
+
+    fn supports_bound_writes(&self) -> bool {
+        self.inner.supports_bound_writes()
+    }
+
+    fn try_format_sql_literal(&self, value: &Option<Value>) -> Result<String, DriverError> {
+        self.inner.try_format_sql_literal(value)
+    }
+
     fn supports_offset(&self) -> bool {
         self.inner.supports_offset()
     }
diff --git a/packages/drivers/http-support/tests/support/real_driver_contract_probe.rs b/packages/drivers/http-support/tests/support/real_driver_contract_probe.rs
index e3821cd75..17fe36212 100644
--- a/packages/drivers/http-support/tests/support/real_driver_contract_probe.rs
+++ b/packages/drivers/http-support/tests/support/real_driver_contract_probe.rs
@@ -31,7 +31,8 @@
 
 use datazen_driver_api::{
     async_trait, ConnectionConfig, ConnectionHandle, DatabaseDriver, DatabaseType, DdlAtomicity,
-    DriverCommandDefinition, DriverError, QueryResult, ServerInfo, TableInfo, TableSchema, Value,
+    DriverCommandDefinition, DriverError, QueryResult, ServerInfo, SqlLiteralDialect, TableInfo,
+    TableSchema, Value,
 };
 
 /// Suffix appended to a perturbed `String`. A suffix rather than a fixed
@@ -134,6 +135,24 @@ fn other_atomicity(hit: bool, base: DdlAtomicity) -> DdlAtomicity {
     }
 }
 
+/// A dialect guaranteed to differ from `base` when `hit`.
+fn other_sql_literal_dialect(
+    hit: bool,
+    base: Option<SqlLiteralDialect>,
+) -> Option<SqlLiteralDialect> {
+    if !hit {
+        return base;
+    }
+    Some(match base {
+        None | Some(SqlLiteralDialect::SqlServer) => SqlLiteralDialect::Sqlite,
+        Some(SqlLiteralDialect::Sqlite) => SqlLiteralDialect::Postgres,
+        Some(SqlLiteralDialect::Postgres) => SqlLiteralDialect::MySql,
+        Some(SqlLiteralDialect::MySql) => SqlLiteralDialect::ClickHouse,
+        Some(SqlLiteralDialect::ClickHouse) => SqlLiteralDialect::DuckDb,
+        Some(SqlLiteralDialect::DuckDb) => SqlLiteralDialect::SqlServer,
+    })
+}
+
 #[async_trait]
 impl<D: DatabaseDriver> DatabaseDriver for Misreports<D> {
     // --- the serverless surface: each method perturbs only when named
@@ -171,6 +190,32 @@ impl<D: DatabaseDriver> DatabaseDriver for Misreports<D> {
         )
     }
 
+    fn sql_literal_dialect(&self) -> Option<SqlLiteralDialect> {
+        other_sql_literal_dialect(
+            self.hit("sql_literal_dialect"),
+            self.inner.sql_literal_dialect(),
+        )
+    }
+
+    fn supports_bound_writes(&self) -> bool {
+        flip(
+            self.hit("supports_bound_writes"),
+            self.inner.supports_bound_writes(),
+        )
+    }
+
+    fn try_format_sql_literal(&self, value: &Option<Value>) -> Result<String, DriverError> {
+        let result = self.inner.try_format_sql_literal(value);
+        if self.hit("try_format_sql_literal") {
+            Ok(match result {
+                Ok(value) => format!("{value}{PROBE_MARK}"),
+                Err(_) => PROBE_MARK.to_string(),
+            })
+        } else {
+            result
+        }
+    }
+
     fn supports_offset(&self) -> bool {
         flip(self.hit("supports_offset"), self.inner.supports_offset())
     }
diff --git a/packages/drivers/http-support/tests/support/real_driver_contract_refusal.rs b/packages/drivers/http-support/tests/support/real_driver_contract_refusal.rs
index ac7ba0dc6..01fc63671 100644
--- a/packages/drivers/http-support/tests/support/real_driver_contract_refusal.rs
+++ b/packages/drivers/http-support/tests/support/real_driver_contract_refusal.rs
@@ -18,7 +18,7 @@
 
 use datazen_driver_api::{
     validate_schema_target, ConnectionHandle, DatabaseDriver, DdlAtomicity, DriverError,
-    QueryExecutionId, SchemaScope, Value,
+    QueryExecutionId, SchemaScope, SqlLiteralDialect, Value,
 };
 use std::fs;
 
@@ -58,6 +58,9 @@ pub const SERVERLESS_TRAIT_METHODS: &[&str] = &[
     "quote_ident",
     "ddl_atomicity",
     "format_sql_literal",
+    "sql_literal_dialect",
+    "supports_bound_writes",
+    "try_format_sql_literal",
     "supports_offset",
     "supports_explain",
     "command_definitions",
@@ -113,10 +116,10 @@ struct CommandRow {
 /// wrapper differs from the driver it wraps in exactly these two fields.
 ///
 /// The remaining fields exist so "identical everywhere else" is not a claim the
-/// type cannot back. This type once held six fields while the wrapper implements
-/// 25 trait methods, so 19 methods had no observation at all and a wrapper that
-/// appended `!` to every identifier, or inverted `supports_offset`, went green
-/// through this whole file. Every field here reads one of
+/// type cannot back. The original snapshot held six fields while the wrapper
+/// implemented 25 trait methods, so 19 methods had no observation at all and a
+/// wrapper that appended `!` to every identifier, or inverted `supports_offset`,
+/// went green through this whole file. Every field here reads one of
 /// [`SERVERLESS_TRAIT_METHODS`]; what is left over is named, with reasons, in
 /// [`SERVER_REQUIRED_TRAIT_METHODS`].
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -133,6 +136,10 @@ struct RefusalSnapshot {
     quoted_ident: String,
     literal_absent: String,
     literal_quote: String,
+    sql_literal_dialect: Option<SqlLiteralDialect>,
+    supports_bound_writes: bool,
+    try_literal_absent: Result<String, String>,
+    try_literal_quote: Result<String, String>,
     ddl_atomicity: DdlAtomicity,
     supports_offset: bool,
     supports_explain: bool,
@@ -173,6 +180,14 @@ impl RefusalSnapshot {
             literal_absent: driver.format_sql_literal(&None),
             literal_quote: driver
                 .format_sql_literal(&Some(Value::String(LITERAL_PROBE.to_string()))),
+            sql_literal_dialect: driver.sql_literal_dialect(),
+            supports_bound_writes: driver.supports_bound_writes(),
+            try_literal_absent: driver
+                .try_format_sql_literal(&None)
+                .map_err(|error| error.to_string()),
+            try_literal_quote: driver
+                .try_format_sql_literal(&Some(Value::String(LITERAL_PROBE.to_string())))
+                .map_err(|error| error.to_string()),
             ddl_atomicity: driver.ddl_atomicity(),
             supports_offset: driver.supports_offset(),
             supports_explain: driver.supports_explain(),
@@ -206,7 +221,8 @@ impl RefusalSnapshot {
     fn summary(&self) -> String {
         format!(
             "declared={:?} execution_cancel={} driver_type={:?} sync_family={:?} quote_char={:?} \
-             quoted_ident={:?} literal_absent={:?} literal_quote={:?} ddl_atomicity={:?} \
+             quoted_ident={:?} literal_absent={:?} literal_quote={:?} sql_literal_dialect={:?} \
+             supports_bound_writes={} try_literal_absent={:?} try_literal_quote={:?} ddl_atomicity={:?} \
              supports_offset={} supports_explain={} commands=[{}] original={:?} second={:?} \
              multi_database={} schema_level={}",
             self.declared,
@@ -217,6 +233,10 @@ impl RefusalSnapshot {
             self.quoted_ident,
             self.literal_absent,
             self.literal_quote,
+            self.sql_literal_dialect,
+            self.supports_bound_writes,
+            self.try_literal_absent,
+            self.try_literal_quote,
             self.ddl_atomicity,
             self.supports_offset,
             self.supports_explain,
@@ -450,8 +470,8 @@ async fn a_withheld_capability_is_refused_and_leaves_the_target_and_the_serverle
 /// lists below are read against the wrapper's actual source, so neither can drift
 /// away from it.
 ///
-/// Without this, "the snapshot has 6 fields" and "the wrapper implements 25
-/// methods" are two true statements that add up to 19 methods nothing observes.
+/// Without this, the original snapshot's 6 fields and the wrapper's 25 methods
+/// left 19 methods with no observation.
 /// That gap is invisible to a reader of the passing test, so it is made a
 /// failing one: add a delegation to the wrapper and this test fails until the new
 /// method is either observed or disclosed. Removing a method from the observed
diff --git a/packages/drivers/mysql/src/mysql.rs b/packages/drivers/mysql/src/mysql.rs
index 3c5a58117..c2c044ac5 100644
--- a/packages/drivers/mysql/src/mysql.rs
+++ b/packages/drivers/mysql/src/mysql.rs
@@ -726,6 +726,14 @@ impl MysqlDriver {
 
 #[async_trait]
 impl DatabaseDriver for MysqlDriver {
+    fn supports_bound_writes(&self) -> bool {
+        true
+    }
+
+    fn sql_literal_dialect(&self) -> Option<SqlLiteralDialect> {
+        Some(SqlLiteralDialect::MySql)
+    }
+
     fn default_host(&self) -> Option<&'static str> {
         Some(crate::mysql::connection::DEFAULT_HOST)
     }
diff --git a/packages/drivers/postgres/src/postgres.rs b/packages/drivers/postgres/src/postgres.rs
index 2bbfd039e..742c8cc90 100644
--- a/packages/drivers/postgres/src/postgres.rs
+++ b/packages/drivers/postgres/src/postgres.rs
@@ -57,6 +57,14 @@ impl PostgresDriver {
 
 #[async_trait]
 impl DatabaseDriver for PostgresDriver {
+    fn supports_bound_writes(&self) -> bool {
+        true
+    }
+
+    fn sql_literal_dialect(&self) -> Option<SqlLiteralDialect> {
+        Some(SqlLiteralDialect::Postgres)
+    }
+
     fn default_host(&self) -> Option<&'static str> {
         Some(crate::connection::DEFAULT_HOST)
     }
diff --git a/packages/drivers/rqlite/src/rqlite.rs b/packages/drivers/rqlite/src/rqlite.rs
index 4ceb3b4e5..4a4c71bf1 100644
--- a/packages/drivers/rqlite/src/rqlite.rs
+++ b/packages/drivers/rqlite/src/rqlite.rs
@@ -138,6 +138,10 @@ impl RqliteDriver {
 
 #[async_trait]
 impl DatabaseDriver for RqliteDriver {
+    fn sql_literal_dialect(&self) -> Option<SqlLiteralDialect> {
+        Some(SqlLiteralDialect::Sqlite)
+    }
+
     // This HTTP driver has no parameterized DML implementation. Zero makes
     // migration consumers reject an active projection before any target write.
     fn max_bound_parameters(&self) -> usize {
diff --git a/packages/drivers/sqlite/src/sqlite.rs b/packages/drivers/sqlite/src/sqlite.rs
index 09d44d687..5055c3592 100644
--- a/packages/drivers/sqlite/src/sqlite.rs
+++ b/packages/drivers/sqlite/src/sqlite.rs
@@ -183,6 +183,14 @@ impl std::hash::Hasher for FileIdentityHasher {
 
 #[async_trait]
 impl DatabaseDriver for SqliteDriver {
+    fn supports_bound_writes(&self) -> bool {
+        true
+    }
+
+    fn sql_literal_dialect(&self) -> Option<SqlLiteralDialect> {
+        Some(SqlLiteralDialect::Sqlite)
+    }
+
     fn max_bound_parameters(&self) -> usize {
         self.parameter_limit
     }
diff --git a/packages/drivers/sqlite/tests/migration_transfer_journey.rs b/packages/drivers/sqlite/tests/migration_transfer_journey.rs
index 506d0c454..0219a873b 100644
--- a/packages/drivers/sqlite/tests/migration_transfer_journey.rs
+++ b/packages/drivers/sqlite/tests/migration_transfer_journey.rs
@@ -57,6 +57,70 @@ async fn collect_rows(
     Arc::try_unwrap(rows).unwrap().into_inner().unwrap()
 }
 
+#[tokio::test]
+async fn sql_literal_and_bound_update_preserve_quotes_slashes_controls_and_unicode() {
+    let directory = std::env::temp_dir().join(format!("datazen-literal-{}", uuid::Uuid::new_v4()));
+    std::fs::create_dir_all(&directory).unwrap();
+    let path = directory.join("literal.db");
+    std::fs::File::create(&path).unwrap();
+
+    let driver = SqliteDriver::new();
+    let handle = driver
+        .connect(&config(path.to_str().unwrap()))
+        .await
+        .unwrap();
+    let expected = "quote' slash\\ trailing\\\n\t雪";
+    let literal = driver
+        .try_format_sql_literal(&Some(Value::String(expected.into())))
+        .unwrap();
+    let result = driver
+        .query(&handle, &format!("SELECT {literal}"))
+        .await
+        .unwrap();
+    assert!(matches!(
+        result.rows.first().and_then(|row| row.first()).and_then(Option::as_ref),
+        Some(Value::String(actual)) if actual == expected
+    ));
+
+    driver
+        .execute(
+            &handle,
+            "CREATE TABLE literal_rows (id INTEGER PRIMARY KEY, value TEXT NOT NULL)",
+        )
+        .await
+        .unwrap();
+    driver
+        .execute(&handle, "INSERT INTO literal_rows VALUES (1, 'initial')")
+        .await
+        .unwrap();
+    let statement = driver
+        .build_update_statement(
+            "literal_rows",
+            &[("value", Some(Value::String(expected.into())))],
+            &[("id", Some(Value::Integer(1)))],
+        )
+        .unwrap();
+    assert!(!statement.sql.contains(expected));
+    assert_eq!(
+        driver
+            .execute_with_params(&handle, &statement.sql, &statement.parameters)
+            .await
+            .unwrap(),
+        1
+    );
+    let result = driver
+        .query(&handle, "SELECT value FROM literal_rows WHERE id = 1")
+        .await
+        .unwrap();
+    assert!(matches!(
+        result.rows.first().and_then(|row| row.first()).and_then(Option::as_ref),
+        Some(Value::String(actual)) if actual == expected
+    ));
+
+    driver.disconnect(handle).await.unwrap();
+    std::fs::remove_dir_all(directory).unwrap();
+}
+
 #[tokio::test]
 async fn test_tester_projected_bound_transfer_preserves_values_and_rolls_back_failed_table() {
     let directory =
diff --git a/packages/drivers/sqlserver/src/sqlserver.rs b/packages/drivers/sqlserver/src/sqlserver.rs
index d3c50e68e..939ba41d3 100644
--- a/packages/drivers/sqlserver/src/sqlserver.rs
+++ b/packages/drivers/sqlserver/src/sqlserver.rs
@@ -55,6 +55,14 @@ impl SqlServerDriver {
 
 #[async_trait]
 impl DatabaseDriver for SqlServerDriver {
+    fn supports_bound_writes(&self) -> bool {
+        true
+    }
+
+    fn sql_literal_dialect(&self) -> Option<SqlLiteralDialect> {
+        Some(SqlLiteralDialect::SqlServer)
+    }
+
     fn default_port(&self) -> Option<u16> {
         Some(DEFAULT_PORT)
     }
diff --git a/packages/drivers/turso/src/turso.rs b/packages/drivers/turso/src/turso.rs
index a9c1cd77f..297bece17 100644
--- a/packages/drivers/turso/src/turso.rs
+++ b/packages/drivers/turso/src/turso.rs
@@ -218,6 +218,10 @@ fn turso_int(v: &serde_json::Value) -> Option<i64> {
 
 #[async_trait]
 impl DatabaseDriver for TursoDriver {
+    fn sql_literal_dialect(&self) -> Option<SqlLiteralDialect> {
+        Some(SqlLiteralDialect::Sqlite)
+    }
+
     // This HTTP driver has no parameterized DML implementation. Zero makes
     // migration consumers reject an active projection before any target write.
     fn max_bound_parameters(&self) -> usize {
diff --git a/src-tauri/src/commands/data.rs b/src-tauri/src/commands/data.rs
index 5914bafb2..540928a7c 100644
--- a/src-tauri/src/commands/data.rs
+++ b/src-tauri/src/commands/data.rs
@@ -1,6 +1,6 @@
 use super::error::{CmdExt, CommandError};
 use super::AppState;
-use crate::db::{DriverError, Value};
+use crate::db::{BoundSqlStatement, DriverError, Value};
 use serde::{Deserialize, Serialize};
 use sha2::{Digest, Sha256};
 use std::collections::{BTreeMap, HashMap, HashSet};
@@ -297,16 +297,23 @@ fn build_row_change_plan(
                     .map(|(column, value)| format!("PK {column}={}", value_summary(value))),
             )
             .collect();
+        let sql_template = if change.delete_marked {
+            driver
+                .build_delete_statement(&table.table, &pk_columns)
+                .map_err(CommandError::Driver)?
+                .sql
+        } else {
+            driver
+                .build_update_statement(&table.table, &set_columns, &pk_columns)
+                .map_err(CommandError::Driver)?
+                .sql
+        };
         let statement = PlannedStatement {
             row_identity: change.row_identity.clone(),
             original_values: change.original_values.clone(),
             current_values: change.current_values.clone(),
             changed_columns: change.changed_columns.clone(),
-            sql_template: if change.delete_marked {
-                driver.build_delete_sql(&table.table, &pk_columns)
-            } else {
-                driver.build_update_sql(&table.table, &set_columns, &pk_columns)
-            },
+            sql_template,
             parameter_summary,
         };
 
@@ -395,6 +402,44 @@ fn validate_immutable_plan(
     Ok(rebuilt)
 }
 
+fn bound_statement_for_plan(
+    driver: &dyn crate::db::DatabaseDriver,
+    table: &str,
+    planned: &PlannedStatement,
+    delete: bool,
+) -> Result<BoundSqlStatement, CommandError> {
+    let pk_columns: Vec<(&str, Option<Value>)> = planned
+        .row_identity
+        .iter()
+        .map(|(column, value)| (column.as_str(), value.clone()))
+        .collect();
+    if delete {
+        return driver
+            .build_delete_statement(table, &pk_columns)
+            .map_err(CommandError::Driver);
+    }
+
+    let set_columns: Vec<(&str, Option<Value>)> = planned
+        .changed_columns
+        .iter()
+        .map(|column| {
+            planned
+                .current_values
+                .get(column)
+                .cloned()
+                .map(|value| (column.as_str(), value))
+                .ok_or_else(|| {
+                    CommandError::Validation(format!(
+                        "Missing current value for changed column '{column}'"
+                    ))
+                })
+        })
+        .collect::<Result<_, _>>()?;
+    driver
+        .build_update_statement(table, &set_columns, &pk_columns)
+        .map_err(CommandError::Driver)
+}
+
 pub(crate) async fn preview_pending_changes_impl(
     state: &AppState,
     table: RowChangeTableContext,
@@ -437,6 +482,19 @@ async fn execute_row_change_plan_impl(
         });
     }
 
+    // Build every statement before opening a transaction. Drivers without
+    // bound-write support reject the plan here, before any database side effect.
+    let updates = plan
+        .updates
+        .iter()
+        .map(|planned| bound_statement_for_plan(driver.as_ref(), &plan.table.table, planned, false))
+        .collect::<Result<Vec<_>, _>>()?;
+    let deletes = plan
+        .deletes
+        .iter()
+        .map(|planned| bound_statement_for_plan(driver.as_ref(), &plan.table.table, planned, true))
+        .collect::<Result<Vec<_>, _>>()?;
+
     let session_open = state
         .session_transactions
         .lock()
@@ -462,9 +520,9 @@ async fn execute_row_change_plan_impl(
     let result: Result<(Vec<RowCommitStatementResult>, u64), CommandError> = async {
         let mut statements = Vec::with_capacity(plan.updates.len() + plan.deletes.len());
         let mut affected_rows = 0;
-        for planned in &plan.updates {
+        for (planned, statement) in plan.updates.iter().zip(&updates) {
             let affected = driver
-                .execute(handle, &planned.sql_template)
+                .execute_with_params(handle, &statement.sql, &statement.parameters)
                 .await
                 .cmd_err("commit_pending_changes")?;
             if affected != 1 {
@@ -479,9 +537,9 @@ async fn execute_row_change_plan_impl(
                 affected_rows: affected,
             });
         }
-        for planned in &plan.deletes {
+        for (planned, statement) in plan.deletes.iter().zip(&deletes) {
             let affected = driver
-                .execute(handle, &planned.sql_template)
+                .execute_with_params(handle, &statement.sql, &statement.parameters)
                 .await
                 .cmd_err("commit_pending_changes")?;
             if affected != 1 {
@@ -663,6 +721,25 @@ pub(crate) async fn commit_row_updates_impl(
         validate_legacy_pk_columns(&batch.pk_columns, "Row update")?;
     }
 
+    let statements = updates
+        .iter()
+        .map(|batch| {
+            let set_columns: Vec<(&str, Option<Value>)> = batch
+                .set_columns
+                .iter()
+                .map(|column| (column.column.as_str(), column.value.clone()))
+                .collect();
+            let pk_columns: Vec<(&str, Option<Value>)> = batch
+                .pk_columns
+                .iter()
+                .map(|column| (column.column.as_str(), column.value.clone()))
+                .collect();
+            driver
+                .build_update_statement(&table, &set_columns, &pk_columns)
+                .map_err(CommandError::Driver)
+        })
+        .collect::<Result<Vec<_>, _>>()?;
+
     let session_open = state
         .session_transactions
         .lock()
@@ -690,20 +767,9 @@ pub(crate) async fn commit_row_updates_impl(
     }
 
     let result: Result<(), CommandError> = async {
-        for batch in &updates {
-            let set_columns: Vec<(&str, Option<Value>)> = batch
-                .set_columns
-                .iter()
-                .map(|c| (c.column.as_str(), c.value.clone()))
-                .collect();
-            let pk_columns: Vec<(&str, Option<Value>)> = batch
-                .pk_columns
-                .iter()
-                .map(|c| (c.column.as_str(), c.value.clone()))
-                .collect();
-            let sql = driver.build_update_sql(&table, &set_columns, &pk_columns);
+        for statement in &statements {
             driver
-                .execute(&handle, &sql)
+                .execute_with_params(&handle, &statement.sql, &statement.parameters)
                 .await
                 .cmd_err("commit_row_updates")?;
         }
@@ -784,6 +850,20 @@ pub(crate) async fn commit_row_deletes_impl(
         validate_legacy_pk_columns(&batch.pk_columns, "Row delete")?;
     }
 
+    let statements = deletes
+        .iter()
+        .map(|batch| {
+            let pk_columns: Vec<(&str, Option<Value>)> = batch
+                .pk_columns
+                .iter()
+                .map(|column| (column.column.as_str(), column.value.clone()))
+                .collect();
+            driver
+                .build_delete_statement(&table, &pk_columns)
+                .map_err(CommandError::Driver)
+        })
+        .collect::<Result<Vec<_>, _>>()?;
+
     let session_open = state
         .session_transactions
         .lock()
@@ -808,15 +888,9 @@ pub(crate) async fn commit_row_deletes_impl(
     }
 
     let result: Result<(), CommandError> = async {
-        for batch in &deletes {
-            let pk_columns: Vec<(&str, Option<Value>)> = batch
-                .pk_columns
-                .iter()
-                .map(|c| (c.column.as_str(), c.value.clone()))
-                .collect();
-            let sql = driver.build_delete_sql(&table, &pk_columns);
+        for statement in &statements {
             driver
-                .execute(&handle, &sql)
+                .execute_with_params(&handle, &statement.sql, &statement.parameters)
                 .await
                 .cmd_err("commit_row_deletes")?;
         }
@@ -872,6 +946,12 @@ mod tests {
     use super::*;
     use crate::testing::app_state::TestAppState;
 
+    async fn with_parameterized_driver() -> TestAppState {
+        let mut options = crate::testing::app_state::rich_mock_options();
+        options.parameterized_writes = true;
+        TestAppState::with_options(options).await
+    }
+
     fn table_context(connection_id: &str, db_session_id: &str) -> RowChangeTableContext {
         RowChangeTableContext {
             connection_id: connection_id.into(),
@@ -928,7 +1008,7 @@ mod tests {
 
     #[tokio::test]
     async fn preview_builds_driver_sql_without_opening_a_transaction() {
-        let test = TestAppState::with_tables().await;
+        let test = with_parameterized_driver().await;
         let (_, conn_id) = test.save_and_connect("preview-only").await;
         let plan = preview_pending_changes_impl(
             &test.state,
@@ -941,7 +1021,8 @@ mod tests {
         assert_eq!(plan.table.db_session_id, conn_id);
         assert_eq!(plan.updates.len(), 1);
         assert!(plan.updates[0].sql_template.contains("UPDATE \"users\""));
-        assert!(plan.updates[0].sql_template.contains("\"id\" = 1"));
+        assert!(plan.updates[0].sql_template.contains("\"id\" = ?2"));
+        assert!(!plan.updates[0].sql_template.contains("Updated"));
         assert!(!crate::commands::session_transaction_status_impl(
             &test.state,
             plan.table.db_session_id.clone()
@@ -954,6 +1035,7 @@ mod tests {
     async fn commit_reuses_plan_fingerprint_and_returns_plan_id() {
         let mut options = crate::testing::app_state::rich_mock_options();
         options.execute_rows_affected = 1;
+        options.parameterized_writes = true;
         let test = TestAppState::with_options(options).await;
         let (_, conn_id) = test.save_and_connect("commit-plan").await;
         let plan = preview_pending_changes_impl(
@@ -982,7 +1064,7 @@ mod tests {
 
     #[tokio::test]
     async fn commit_rejects_stale_fingerprint_before_execution() {
-        let test = TestAppState::with_tables().await;
+        let test = with_parameterized_driver().await;
         let (_, conn_id) = test.save_and_connect("stale-plan").await;
         let plan = preview_pending_changes_impl(
             &test.state,
@@ -1012,7 +1094,7 @@ mod tests {
 
     #[tokio::test]
     async fn preview_rejects_changes_without_primary_key_identity() {
-        let test = TestAppState::with_tables().await;
+        let test = with_parameterized_driver().await;
         let (_, conn_id) = test.save_and_connect("preview-no-pk").await;
         let error = preview_pending_changes_impl(
             &test.state,
@@ -1088,6 +1170,7 @@ mod tests {
     async fn commit_rejects_database_context_change_without_switching_session() {
         let mut options = crate::testing::app_state::rich_mock_options();
         options.execute_rows_affected = 1;
+        options.parameterized_writes = true;
         let test = TestAppState::with_options(options).await;
         let (_, conn_id) = test.save_and_connect("context-database").await;
         let plan = preview_pending_changes_impl(
@@ -1119,6 +1202,7 @@ mod tests {
     async fn commit_allows_concrete_table_schema_when_connection_schema_is_unspecified() {
         let mut options = crate::testing::app_state::rich_mock_options();
         options.execute_rows_affected = 1;
+        options.parameterized_writes = true;
         let test = TestAppState::with_options(options).await;
         let (_, conn_id) = test.save_and_connect("context-schema-default").await;
         let plan = preview_pending_changes_impl(
@@ -1148,6 +1232,7 @@ mod tests {
     async fn commit_rejects_explicit_schema_context_change() {
         let mut options = crate::testing::app_state::rich_mock_options();
         options.execute_rows_affected = 1;
+        options.parameterized_writes = true;
         let test = TestAppState::with_options(options).await;
         let mut config =
             crate::testing::app_state::sample_postgres_config("context-schema-explicit");
@@ -1180,7 +1265,7 @@ mod tests {
 
     #[tokio::test]
     async fn commit_row_updates_success() {
-        let test = TestAppState::with_tables().await;
+        let test = with_parameterized_driver().await;
         let (_, conn_id) = test.save_and_connect("data-cfg").await;
         commit_row_updates_impl(
             &test.state,
@@ -1213,7 +1298,7 @@ mod tests {
 
     #[tokio::test]
     async fn commit_row_updates_rejects_read_only_connection() {
-        let test = TestAppState::with_tables().await;
+        let test = with_parameterized_driver().await;
         let mut config = crate::testing::app_state::sample_postgres_config("ro-edit");
         config.read_only = true;
         test.store.save_connection(config).await.unwrap();
@@ -1240,7 +1325,7 @@ mod tests {
 
     #[tokio::test]
     async fn commit_row_updates_joins_open_session_transaction() {
-        let test = TestAppState::with_tables().await;
+        let test = with_parameterized_driver().await;
         let (_, conn_id) = test.save_and_connect("tx-edit").await;
         crate::commands::begin_session_transaction_impl(&test.state, conn_id.clone())
             .await
@@ -1274,7 +1359,7 @@ mod tests {
 
     #[tokio::test]
     async fn commit_row_deletes_success() {
-        let test = TestAppState::with_tables().await;
+        let test = with_parameterized_driver().await;
         let (_, conn_id) = test.save_and_connect("data-del").await;
         commit_row_deletes_impl(
             &test.state,
@@ -1293,7 +1378,7 @@ mod tests {
 
     #[tokio::test]
     async fn commit_row_deletes_rejects_read_only_connection() {
-        let test = TestAppState::with_tables().await;
+        let test = with_parameterized_driver().await;
         let mut config = crate::testing::app_state::sample_postgres_config("ro-del");
         config.read_only = true;
         test.store.save_connection(config).await.unwrap();
@@ -1316,7 +1401,7 @@ mod tests {
 
     #[tokio::test]
     async fn commit_row_deletes_rejects_empty_pk() {
-        let test = TestAppState::with_tables().await;
+        let test = with_parameterized_driver().await;
         let (_, conn_id) = test.save_and_connect("data-del-empty-pk").await;
         let err = commit_row_deletes_impl(
             &test.state,
diff --git a/src-tauri/src/commands/export.rs b/src-tauri/src/commands/export.rs
index 85293ae46..c0644fde9 100644
--- a/src-tauri/src/commands/export.rs
+++ b/src-tauri/src/commands/export.rs
@@ -282,7 +282,11 @@ impl StreamFormatter {
     }
 
     /// Format a batch of rows into text.
-    fn rows(&mut self, rows: &[Vec<Option<Value>>], columns: &[String]) -> String {
+    fn rows(
+        &mut self,
+        rows: &[Vec<Option<Value>>],
+        columns: &[String],
+    ) -> Result<String, crate::db::DriverError> {
         match &mut self.state {
             FormatterState::Csv => {
                 let mut out = String::new();
@@ -295,7 +299,7 @@ impl StreamFormatter {
                     }
                     out.push('\n');
                 }
-                out
+                Ok(out)
             }
             FormatterState::Json { started } => {
                 let mut out = String::new();
@@ -313,7 +317,7 @@ impl StreamFormatter {
                     out.push('\n');
                     out.push_str(&obj);
                 }
-                out
+                Ok(out)
             }
             FormatterState::SqlInsert { pending } => {
                 pending.extend(rows.iter().cloned());
@@ -322,12 +326,12 @@ impl StreamFormatter {
         }
     }
 
-    fn flush_sql(&mut self, columns: &[String]) -> String {
+    fn flush_sql(&mut self, columns: &[String]) -> Result<String, crate::db::DriverError> {
         let FormatterState::SqlInsert { pending } = &mut self.state else {
-            return String::new();
+            return Ok(String::new());
         };
         if pending.is_empty() {
-            return String::new();
+            return Ok(String::new());
         }
         let mut out = String::new();
         let mut batch: Vec<Vec<Option<Value>>> = Vec::new();
@@ -340,7 +344,7 @@ impl StreamFormatter {
                     &self.database_type,
                     self.driver.as_ref(),
                     &batch,
-                ));
+                )?);
                 batch.clear();
             }
         }
@@ -351,29 +355,29 @@ impl StreamFormatter {
                 &self.database_type,
                 self.driver.as_ref(),
                 &batch,
-            ));
+            )?);
         }
-        out
+        Ok(out)
     }
 
     /// Text to write at the very end of a table dump (JSON `]`, SQL `COMMIT`).
-    fn tail(&mut self, columns: &[String]) -> String {
+    fn tail(&mut self, columns: &[String]) -> Result<String, crate::db::DriverError> {
         match self.format {
-            DataFormat::Json => {
+            DataFormat::Json => Ok({
                 let FormatterState::Json { started } = &mut self.state else {
-                    return String::new();
+                    return Ok(String::new());
                 };
                 if *started {
                     "\n]".into()
                 } else {
                     "]".into()
                 }
-            }
+            }),
             DataFormat::SqlInsert => {
                 let batched = self.flush_sql(columns);
-                format!("{batched}COMMIT;\n")
+                Ok(format!("{}COMMIT;\n", batched?))
             }
-            DataFormat::Csv => String::new(),
+            DataFormat::Csv => Ok(String::new()),
         }
     }
 }
@@ -384,7 +388,7 @@ fn emit_insert_batch(
     database_type: &Option<String>,
     driver: &dyn DatabaseDriver,
     rows: &[Vec<Option<Value>>],
-) -> String {
+) -> Result<String, crate::db::DriverError> {
     let db_type = database_type.as_deref();
     let col_list = columns
         .iter()
@@ -403,13 +407,13 @@ fn emit_insert_batch(
         }
         let values = row
             .iter()
-            .map(|v| driver.format_sql_literal(v))
-            .collect::<Vec<_>>()
+            .map(|value| driver.try_format_sql_literal(value))
+            .collect::<Result<Vec<_>, _>>()?
             .join(", ");
         out.push_str(&format!("({values})"));
     }
     out.push_str(";\n");
-    out
+    Ok(out)
 }
 
 // ---------------------------------------------------------------------------
@@ -655,6 +659,7 @@ async fn write_data_file(
     let rows_written = Arc::new(AtomicU64::new(0));
     let tail_stated = Arc::new(AtomicU64::new(0));
     let last_emit_rows = Arc::new(AtomicU64::new(0));
+    let format_error = Arc::new(Mutex::new(None::<String>));
 
     let cb_sink = Arc::clone(&sink);
     let cb_fmt = Arc::clone(&formatter);
@@ -663,6 +668,7 @@ async fn write_data_file(
     let cb_rows = Arc::clone(&rows_written);
     let cb_tail = Arc::clone(&tail_stated);
     let cb_last_emit = Arc::clone(&last_emit_rows);
+    let cb_format_error = Arc::clone(&format_error);
     let cb_app = app.clone();
     let cb_table = table.table_name.clone();
 
@@ -681,12 +687,15 @@ async fn write_data_file(
         QueryStreamEvent::Rows { rows, .. } => {
             ensure_header(&cb_cols, &cb_fmt, &cb_sink, &cb_header, false);
             let cols = lock_export_stream(&cb_cols).clone();
-            let text = lock_export_stream(&cb_fmt).rows(&rows, &cols);
-            if !text.is_empty() {
-                if let Err(e) = lock_export_stream(&cb_sink).append_str(&text) {
-                    // Cannot return an error through the Fn callback; the query
-                    // will surface via query_stream's error propagation instead.
-                    let _ = e;
+            match lock_export_stream(&cb_fmt).rows(&rows, &cols) {
+                Ok(text) if !text.is_empty() => {
+                    if let Err(e) = lock_export_stream(&cb_sink).append_str(&text) {
+                        let _ = e;
+                    }
+                }
+                Ok(_) => {}
+                Err(error) => {
+                    *lock_export_stream(&cb_format_error) = Some(error.to_string());
                 }
             }
             let total = cb_rows.fetch_add(rows.len() as u64, Ordering::Relaxed) + rows.len() as u64;
@@ -714,10 +723,15 @@ async fn write_data_file(
             );
             if cb_tail.fetch_add(1, Ordering::Relaxed) == 0 {
                 let cols = lock_export_stream(&cb_cols).clone();
-                let text = lock_export_stream(&cb_fmt).tail(&cols);
-                if !text.is_empty() {
-                    if let Err(e) = lock_export_stream(&cb_sink).append_str(&text) {
-                        let _ = e;
+                match lock_export_stream(&cb_fmt).tail(&cols) {
+                    Ok(text) if !text.is_empty() => {
+                        if let Err(e) = lock_export_stream(&cb_sink).append_str(&text) {
+                            let _ = e;
+                        }
+                    }
+                    Ok(_) => {}
+                    Err(error) => {
+                        *lock_export_stream(&cb_format_error) = Some(error.to_string());
                     }
                 }
             }
@@ -746,10 +760,16 @@ async fn write_data_file(
         .await
         .cmd_err("export")?;
 
+    if let Some(error) = lock_export(&format_error)?.take() {
+        return Err(CommandError::Validation(format!(
+            "SQL export could not safely render a value: {error}"
+        )));
+    }
+
     // Ensure tail is written even if no Done event fired.
     if tail_stated.load(Ordering::Relaxed) == 0 {
         let cols = lock_export(&columns)?.clone();
-        let text = lock_export(&formatter)?.tail(&cols);
+        let text = lock_export(&formatter)?.tail(&cols)?;
         if !text.is_empty() {
             lock_export(&sink)?.append_str(&text)?;
         }
@@ -1033,7 +1053,7 @@ mod tests {
 
     fn test_driver() -> Arc<dyn DatabaseDriver> {
         crate::testing::mock_driver::MockDriver::new(
-            "test",
+            "sqlite",
             crate::testing::mock_driver::MockDriverOptions::default(),
         )
     }
@@ -1055,9 +1075,9 @@ mod tests {
         let cols = vec!["id".to_string(), "name".to_string()];
         let header = f.header(&cols);
         assert_eq!(header, "id,name\n");
-        let text = f.rows(&rows2(), &cols);
+        let text = f.rows(&rows2(), &cols).unwrap();
         assert_eq!(text, "1,\"a,b\"\n2,\n");
-        assert_eq!(f.tail(&cols), "");
+        assert_eq!(f.tail(&cols).unwrap(), "");
     }
 
     #[test]
@@ -1066,7 +1086,7 @@ mod tests {
         let cols = vec!["text".to_string()];
         let _ = f.header(&cols);
         assert_eq!(
-            f.rows(&[vec![v_str("before\rafter")]], &cols),
+            f.rows(&[vec![v_str("before\rafter")]], &cols).unwrap(),
             "\"before\rafter\"\n"
         );
     }
@@ -1081,12 +1101,12 @@ mod tests {
             vec![Some(Value::Integer(2)), Some(Value::Null)],
         ];
         // First call writes object 1 with a leading newline, no comma yet.
-        let a = f.rows(&rows[..1], &cols);
+        let a = f.rows(&rows[..1], &cols).unwrap();
         assert_eq!(a, "\n{\"id\":1,\"name\":\"x\"}");
         // Second call adds a comma.
-        let b = f.rows(&rows[1..], &cols);
+        let b = f.rows(&rows[1..], &cols).unwrap();
         assert_eq!(b, ",\n{\"id\":2,\"name\":null}");
-        assert_eq!(f.tail(&cols), "\n]");
+        assert_eq!(f.tail(&cols).unwrap(), "\n]");
     }
 
     #[test]
@@ -1099,7 +1119,7 @@ mod tests {
             v_str("datazen:bytes:hex:00fffe"),
         ]];
         assert_eq!(
-            f.rows(&rows, &cols),
+            f.rows(&rows, &cols).unwrap(),
             "datazen:bytes:hex:00fffe,datazen:text:datazen:bytes:hex:00fffe\n"
         );
     }
@@ -1109,8 +1129,11 @@ mod tests {
         let mut f = StreamFormatter::new(DataFormat::Json, "t".into(), None, test_driver());
         let cols = vec!["payload".to_string()];
         let mut text = f.header(&cols);
-        text.push_str(&f.rows(&[vec![Some(Value::Bytes(vec![0x00, 0xff, 0xfe]))]], &cols));
-        text.push_str(&f.tail(&cols));
+        text.push_str(
+            &f.rows(&[vec![Some(Value::Bytes(vec![0x00, 0xff, 0xfe]))]], &cols)
+                .unwrap(),
+        );
+        text.push_str(&f.tail(&cols).unwrap());
         let parsed: serde_json::Value = serde_json::from_str(&text).expect("valid JSON export");
         assert_eq!(
             parsed[0]["payload"],
@@ -1152,19 +1175,19 @@ mod tests {
         let mut f = StreamFormatter::new(
             DataFormat::SqlInsert,
             "users".into(),
-            Some("postgres".into()),
+            Some("sqlite".into()),
             test_driver(),
         );
         let cols = vec!["id".to_string(), "email".to_string()];
         let header = f.header(&cols);
         assert_eq!(header, "BEGIN;\n");
         // Two rows -> single batched INSERT with two tuples.
-        let text = f.rows(&rows2(), &cols);
+        let text = f.rows(&rows2(), &cols).unwrap();
         assert_eq!(
             text,
-            "INSERT INTO \"users\" (\"id\", \"email\") VALUES\n  (1, 'a,b'),\n  (2, NULL);\n"
+            "INSERT INTO \"users\" (\"id\", \"email\") VALUES\n  (1, CAST(X'612c62' AS TEXT)),\n  (2, NULL);\n"
         );
-        let tail = f.tail(&cols);
+        let tail = f.tail(&cols).unwrap();
         assert_eq!(tail, "COMMIT;\n");
     }
 
@@ -1174,13 +1197,13 @@ mod tests {
         let cols = vec!["id".to_string()];
         let rows: Vec<Vec<Option<Value>>> =
             (0..600).map(|i| vec![Some(Value::Integer(i))]).collect();
-        let text = f.rows(&rows, &cols);
+        let text = f.rows(&rows, &cols).unwrap();
         // 600 rows -> two batched INSERT statements (500 + 100 rows), all in
         // this rows call since flush drains every pending row.
         assert_eq!(text.matches("INSERT INTO").count(), 2);
         // The last row (599) is in the second batch.
         assert!(text.contains("(599);"));
-        let tail = f.tail(&cols);
+        let tail = f.tail(&cols).unwrap();
         // No pending rows left; tail only adds COMMIT (single transaction).
         assert_eq!(tail, "COMMIT;\n");
         assert_eq!(tail.matches("INSERT INTO").count(), 0);
@@ -1191,7 +1214,9 @@ mod tests {
         let mut f = StreamFormatter::new(DataFormat::SqlInsert, "t".into(), None, test_driver());
         let cols = vec!["payload".to_string()];
         let _ = f.header(&cols);
-        let text = f.rows(&[vec![Some(Value::Bytes(vec![0x00, 0xff, 0xfe]))]], &cols);
+        let text = f
+            .rows(&[vec![Some(Value::Bytes(vec![0x00, 0xff, 0xfe]))]], &cols)
+            .unwrap();
         assert!(text.contains("X'00fffe'"), "{text}");
         assert!(!text.contains('\u{fffd}'), "{text}");
     }
diff --git a/src-tauri/src/services/query_executor.rs b/src-tauri/src/services/query_executor.rs
index 6b821b62d..98af4b128 100644
--- a/src-tauri/src/services/query_executor.rs
+++ b/src-tauri/src/services/query_executor.rs
@@ -68,7 +68,7 @@ impl QueryExecutor {
             .await?;
 
         let qi = |name: &str| driver.quote_ident(name);
-        let format_lit = |v: &Value| driver.format_sql_literal(&Some(v.clone()));
+        let format_lit = |v: &Value| driver.try_format_sql_literal(&Some(v.clone()));
 
         let data_sql = Self::build_select_sql(
             &cached.table_name,
@@ -82,7 +82,7 @@ impl QueryExecutor {
                 (page as u64).saturating_mul(page_size as u64),
             ),
             filter_logic,
-        );
+        )?;
 
         let target = SqlTarget::new(Some(database), schema);
 
@@ -105,7 +105,7 @@ impl QueryExecutor {
             &qi,
             &format_lit,
             filter_logic,
-        );
+        )?;
         tracing::info!(%table, "query_executor: count query");
         tracing::debug!(
             %table,
@@ -142,9 +142,9 @@ impl QueryExecutor {
         columns: &[ColumnSchema],
         filters: &Option<Vec<FilterCondition>>,
         qi: &dyn Fn(&str) -> String,
-        format_lit: &dyn Fn(&Value) -> String,
+        format_lit: &dyn Fn(&Value) -> Result<String, DriverError>,
         filter_logic: Option<&str>,
-    ) -> String {
+    ) -> Result<String, DriverError> {
         let _ = columns;
         let mut sql = format!("SELECT COUNT(*) FROM {}", qi(table_name));
 
@@ -152,15 +152,15 @@ impl QueryExecutor {
             let parts: Vec<String> = conditions
                 .iter()
                 .map(|c| Self::format_condition(c, qi, format_lit))
-                .filter(|s| !s.is_empty())
-                .collect();
+                .collect::<Result<_, _>>()?;
+            let parts: Vec<String> = parts.into_iter().filter(|s| !s.is_empty()).collect();
             if !parts.is_empty() {
                 sql.push_str(" WHERE ");
                 sql.push_str(&parts.join(filter_join(filter_logic)));
             }
         }
 
-        sql
+        Ok(sql)
     }
 
     fn build_select_sql(
@@ -169,10 +169,10 @@ impl QueryExecutor {
         filters: Option<Vec<FilterCondition>>,
         order_by: Option<OrderBy>,
         qi: &dyn Fn(&str) -> String,
-        format_lit: &dyn Fn(&Value) -> String,
+        format_lit: &dyn Fn(&Value) -> Result<String, DriverError>,
         pagination: &PaginationSyntax,
         filter_logic: Option<&str>,
-    ) -> String {
+    ) -> Result<String, DriverError> {
         let mut sql = String::new();
         sql.push_str("SELECT ");
         if columns.is_empty() {
@@ -193,7 +193,7 @@ impl QueryExecutor {
             let parts: Vec<String> = conditions
                 .iter()
                 .map(|c| Self::format_condition(c, qi, format_lit))
-                .collect();
+                .collect::<Result<_, _>>()?;
             let parts: Vec<String> = parts.into_iter().filter(|s| !s.is_empty()).collect();
             if !parts.is_empty() {
                 sql.push_str(" WHERE ");
@@ -249,35 +249,35 @@ impl QueryExecutor {
             sql.push(' ');
             sql.push_str(&pagination.clause);
         }
-        sql
+        Ok(sql)
     }
 
     fn format_condition(
         condition: &FilterCondition,
         qi: &dyn Fn(&str) -> String,
-        format_lit: &dyn Fn(&Value) -> String,
-    ) -> String {
+        format_lit: &dyn Fn(&Value) -> Result<String, DriverError>,
+    ) -> Result<String, DriverError> {
         // Incomplete filters (e.g. just-added eq with empty value) must not enter SQL.
         // Otherwise drivers cast '' onto integer PKs and the whole table load fails.
         if !Self::filter_is_complete(condition) {
-            return String::new();
+            return Ok(String::new());
         }
 
         let col = qi(&condition.column);
-        match condition.operator {
-            FilterOperator::Eq => format!("{col} = {}", format_lit(&condition.value)),
-            FilterOperator::Ne => format!("{col} != {}", format_lit(&condition.value)),
-            FilterOperator::Gt => format!("{col} > {}", format_lit(&condition.value)),
-            FilterOperator::Lt => format!("{col} < {}", format_lit(&condition.value)),
-            FilterOperator::Gte => format!("{col} >= {}", format_lit(&condition.value)),
-            FilterOperator::Lte => format!("{col} <= {}", format_lit(&condition.value)),
-            FilterOperator::Like => format!("{col} LIKE {}", format_lit(&condition.value)),
+        let sql = match condition.operator {
+            FilterOperator::Eq => format!("{col} = {}", format_lit(&condition.value)?),
+            FilterOperator::Ne => format!("{col} != {}", format_lit(&condition.value)?),
+            FilterOperator::Gt => format!("{col} > {}", format_lit(&condition.value)?),
+            FilterOperator::Lt => format!("{col} < {}", format_lit(&condition.value)?),
+            FilterOperator::Gte => format!("{col} >= {}", format_lit(&condition.value)?),
+            FilterOperator::Lte => format!("{col} <= {}", format_lit(&condition.value)?),
+            FilterOperator::Like => format!("{col} LIKE {}", format_lit(&condition.value)?),
             FilterOperator::In => match &condition.value {
                 Value::Json(serde_json::Value::Array(arr)) => {
                     let parts: Vec<String> = arr
                         .iter()
                         .map(|j| format_lit(&Value::Json(j.clone())))
-                        .collect();
+                        .collect::<Result<_, _>>()?;
                     format!("{col} IN ({})", parts.join(", "))
                 }
                 Value::String(s) => {
@@ -286,7 +286,7 @@ impl QueryExecutor {
                         .map(|p| p.trim())
                         .filter(|p| !p.is_empty())
                         .map(|p| format_lit(&Value::String(p.to_string())))
-                        .collect();
+                        .collect::<Result<_, _>>()?;
                     if parts.is_empty() {
                         String::new()
                     } else {
@@ -297,7 +297,8 @@ impl QueryExecutor {
             },
             FilterOperator::IsNull => format!("{col} IS NULL"),
             FilterOperator::IsNotNull => format!("{col} IS NOT NULL"),
-        }
+        };
+        Ok(sql)
     }
 
     /// Filters that still need a value (just added / cleared) are ignored until complete.
@@ -344,8 +345,8 @@ mod tests {
         format!("\"{}\"", name)
     }
 
-    fn simple_lit(value: &Value) -> String {
-        match value {
+    fn simple_lit(value: &Value) -> Result<String, DriverError> {
+        Ok(match value {
             Value::Null => "NULL".to_string(),
             Value::Bool(b) => {
                 if *b {
@@ -360,7 +361,7 @@ mod tests {
             Value::Bytes(_) => "NULL".to_string(),
             Value::Timestamp(s) => format!("'{}'", s.replace('\'', "''")),
             Value::Json(v) => format!("'{}'", v.to_string().replace('\'', "''")),
-        }
+        })
     }
 
     fn make_column(name: &str, is_pk: bool) -> ColumnSchema {
@@ -422,7 +423,8 @@ mod tests {
             &simple_lit,
             &tsql_syntax(3, 25),
             None,
-        );
+        )
+        .unwrap();
         assert!(
             sql.contains("OFFSET 75 ROWS FETCH NEXT 25 ROWS ONLY"),
             "expected T-SQL paging clause, got: {sql}"
@@ -447,7 +449,8 @@ mod tests {
             &simple_lit,
             &tsql_syntax(0, 50),
             None,
-        );
+        )
+        .unwrap();
         assert!(sql.contains("ORDER BY \"id\" ASC"), "got: {sql}");
         assert!(!sql.contains("(SELECT NULL)"), "got: {sql}");
         assert!(
@@ -473,7 +476,8 @@ mod tests {
             &simple_lit,
             &limit_syntax(0, 50),
             None,
-        );
+        )
+        .unwrap();
         assert!(
             sql.contains("ORDER BY \"id\" ASC"),
             "Expected default ORDER BY primary key, got: {sql}"
@@ -496,7 +500,8 @@ mod tests {
             &simple_lit,
             &limit_syntax(0, 50),
             None,
-        );
+        )
+        .unwrap();
         assert!(
             sql.contains("ORDER BY \"order_id\" ASC, \"product_id\" ASC"),
             "Expected composite PK ordering, got: {sql}"
@@ -519,7 +524,8 @@ mod tests {
             &simple_lit,
             &limit_syntax(0, 50),
             None,
-        );
+        )
+        .unwrap();
         assert!(
             sql.contains("ORDER BY \"name\" DESC"),
             "Expected explicit ORDER BY, got: {sql}"
@@ -542,7 +548,8 @@ mod tests {
             &simple_lit,
             &limit_syntax(0, 50),
             None,
-        );
+        )
+        .unwrap();
         assert!(
             sql.contains("ORDER BY \"col_a\" ASC"),
             "Expected fallback ORDER BY first column, got: {sql}"
@@ -566,7 +573,8 @@ mod tests {
             &simple_lit,
             &limit_syntax(0, 10),
             None,
-        );
+        )
+        .unwrap();
         assert!(
             sql.contains("WHERE \"flag\" = TRUE"),
             "Expected driver-style TRUE literal, got: {sql}"
@@ -588,7 +596,8 @@ mod tests {
             &simple_qi,
             &simple_lit,
             None,
-        );
+        )
+        .unwrap();
         assert!(sql.contains("WHERE \"status\" = 'active'"));
     }
 
@@ -616,7 +625,8 @@ mod tests {
             &simple_lit,
             &limit_syntax(0, 10),
             Some("or"),
-        );
+        )
+        .unwrap();
         assert!(
             sql.contains("WHERE \"status\" = 'active' OR \"role\" = 'admin'"),
             "got: {sql}"
@@ -645,7 +655,8 @@ mod tests {
             &simple_qi,
             &simple_lit,
             Some("OR"),
-        );
+        )
+        .unwrap();
         assert!(
             sql.contains("WHERE \"status\" = 'active' OR \"role\" = 'admin'"),
             "got: {sql}"
@@ -671,7 +682,8 @@ mod tests {
             &simple_lit,
             &limit_only_syntax(25),
             None,
-        );
+        )
+        .unwrap();
         assert!(sql.contains("LIMIT 25"));
         assert!(!sql.contains("OFFSET"));
     }
@@ -689,11 +701,11 @@ mod tests {
             value: Value::Null,
         };
         assert_eq!(
-            QueryExecutor::format_condition(&in_filter, &simple_qi, &simple_lit),
+            QueryExecutor::format_condition(&in_filter, &simple_qi, &simple_lit).unwrap(),
             "\"id\" IN ('1', '2', '3')"
         );
         assert_eq!(
-            QueryExecutor::format_condition(&null_filter, &simple_qi, &simple_lit),
+            QueryExecutor::format_condition(&null_filter, &simple_qi, &simple_lit).unwrap(),
             "\"deleted_at\" IS NULL"
         );
     }
@@ -715,7 +727,8 @@ mod tests {
             &simple_lit,
             &limit_syntax(0, 50),
             None,
-        );
+        )
+        .unwrap();
         assert!(
             !sql.contains("WHERE"),
             "empty eq value must not produce WHERE, got: {sql}"
@@ -746,7 +759,8 @@ mod tests {
             &simple_lit,
             &limit_syntax(0, 50),
             Some("and"),
-        );
+        )
+        .unwrap();
         assert!(
             sql.contains("WHERE \"name\" = 'alice'"),
             "expected only complete filter in WHERE, got: {sql}"