perf(import): bound bulk import memory and cancellation

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

@@ -25,6 +25,7 @@ use super::file_validator::validate_file_path;
 
 pub type MySqlPool = mysql_async::Pool;
 const MYSQL_TCP_KEEPALIVE_MS: u32 = 30_000;
+const MYSQL_SQL_PACKET_MARGIN_MAX_BYTES: usize = 64 * 1024;
 
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub(crate) enum MySqlCatalogDialect {
@@ -3655,6 +3656,23 @@ pub async fn execute_query(pool: &MySqlPool, sql: &str, bare: bool) -> Result<Qu
     execute_query_with_max_rows(pool, sql, bare, None, MySqlQueryDialect::default()).await
 }
 
+pub async fn max_allowed_packet(pool: &MySqlPool) -> Result<u64, String> {
+    let mut conn = get_conn_with_health_check(pool).await?;
+    conn.query_first::<u64, _>("SELECT @@max_allowed_packet")
+        .await
+        .map_err(|error| error.to_string())?
+        .ok_or_else(|| "MySQL did not return @@max_allowed_packet".to_string())
+}
+
+pub(crate) fn mysql_sql_statement_hard_limit(max_allowed_packet: u64) -> Option<usize> {
+    let packet_bytes = usize::try_from(max_allowed_packet).ok()?;
+    if packet_bytes == 0 {
+        return None;
+    }
+    let margin = (packet_bytes / 10).clamp(1024, MYSQL_SQL_PACKET_MARGIN_MAX_BYTES).min(packet_bytes / 2);
+    packet_bytes.checked_sub(margin).filter(|limit| *limit > 0)
+}
+
 pub async fn execute_query_with_max_rows(
     pool: &MySqlPool,
     sql: &str,
@@ -4158,6 +4176,17 @@ pub async fn list_triggers(pool: &MySqlPool, database: &str, table: &str) -> Res
 #[cfg(test)]
 mod tests {
     use super::*;
+
+    #[test]
+    fn mysql_sql_statement_limit_reserves_packet_headroom() {
+        let packet_bytes = 64 * 1024 * 1024;
+        let hard_limit = mysql_sql_statement_hard_limit(packet_bytes).unwrap();
+
+        assert!(hard_limit < packet_bytes as usize);
+        assert!(hard_limit >= packet_bytes as usize * 9 / 10);
+        assert_eq!(mysql_sql_statement_hard_limit(0), None);
+        assert_eq!(mysql_sql_statement_hard_limit(4096), Some(3072));
+    }
     use crate::db::connection_timeout;
     use mysql_async::consts::ColumnFlags;
     #[test]

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

@@ -8,11 +8,14 @@ use futures::{FutureExt, TryStreamExt};
 use sqlparser::ast::{Expr, Ident, SelectItem, SetExpr, Statement};
 use sqlparser::dialect::MsSqlDialect;
 use sqlparser::parser::Parser;
+use std::borrow::Cow;
 use std::future::Future;
 use std::panic::AssertUnwindSafe;
 use std::sync::{Arc as StdArc, Mutex as StdMutex};
 use std::time::{Duration, Instant};
-use tiberius::{AuthMethod, Client, ColumnData, ColumnType, Config, FromSql, QueryItem, QueryStream, Row, SqlBrowser};
+use tiberius::{
+    AuthMethod, Client, ColumnData, ColumnType, Config, FromSql, QueryItem, QueryStream, Row, SqlBrowser, TokenRow,
+};
 use tokio::net::TcpStream;
 use tokio_util::compat::{Compat, TokioAsyncWriteCompatExt};
 use tokio_util::sync::CancellationToken;
@@ -2084,6 +2087,57 @@ pub async fn execute_query(client: &mut SqlServerClient, sql: &str) -> Result<Qu
     execute_query_with_max_rows(client, sql, None).await
 }
 
+fn sqlserver_bulk_token_row(values: Vec<Option<String>>) -> TokenRow<'static> {
+    let mut row = TokenRow::with_capacity(values.len());
+    for value in values {
+        row.push(ColumnData::String(value.map(Cow::Owned)));
+    }
+    row
+}
+
+pub async fn bulk_insert_text_rows<T, F>(
+    client: &mut SqlServerClient,
+    staging_table: &str,
+    rows: &[T],
+    column_count: usize,
+    mut convert_row: F,
+) -> Result<u64, String>
+where
+    F: FnMut(usize, &T) -> Result<Vec<Option<String>>, String>,
+{
+    if rows.is_empty() {
+        return Ok(0);
+    }
+    if column_count == 0 {
+        return Err("SQL Server bulk load requires at least one mapped column".to_string());
+    }
+
+    let mut request = client
+        .bulk_insert(staging_table)
+        .await
+        .map_err(|error| format!("SQL Server bulk load initialization failed: {error}"))?;
+    for (row_index, source_row) in rows.iter().enumerate() {
+        let row = convert_row(row_index, source_row)?;
+        if row.len() != column_count {
+            return Err(format!(
+                "SQL Server bulk row {} has {} columns; expected {}",
+                row_index + 1,
+                row.len(),
+                column_count
+            ));
+        }
+        request
+            .send(sqlserver_bulk_token_row(row))
+            .await
+            .map_err(|error| format!("SQL Server bulk load send failed: {error}"))?;
+    }
+    request
+        .finalize()
+        .await
+        .map(|result| result.total())
+        .map_err(|error| format!("SQL Server bulk load finalize failed: {error}"))
+}
+
 pub async fn execute_query_with_max_rows(
     client: &mut SqlServerClient,
     sql: &str,
@@ -2336,6 +2390,10 @@ fn contains_transaction_control(sql: &str) -> bool {
 }
 
 fn requires_simple_query_batch(sql: &str) -> bool {
+    if creates_local_temp_table(sql) {
+        return true;
+    }
+
     let tokens = first_sql_tokens(sql, 4);
     if tokens.len() >= 2 && tokens[0].eq_ignore_ascii_case("SET") && tokens[1].eq_ignore_ascii_case("SHOWPLAN_XML") {
         return true;
@@ -2359,6 +2417,22 @@ fn requires_simple_query_batch(sql: &str) -> bool {
     false
 }
 
+fn creates_local_temp_table(sql: &str) -> bool {
+    if !sql.as_bytes().contains(&b'#') {
+        return false;
+    }
+
+    let Ok(statements) = Parser::parse_sql(&MsSqlDialect {}, sql) else {
+        return false;
+    };
+    statements.iter().any(|statement| {
+        let Statement::CreateTable(table) = statement else {
+            return false;
+        };
+        table.name.0.last().and_then(|part| part.as_ident()).is_some_and(|identifier| identifier.value.starts_with('#'))
+    })
+}
+
 fn first_sql_tokens(sql: &str, limit: usize) -> Vec<String> {
     let bytes = sql.as_bytes();
     let mut tokens = Vec::new();
@@ -2407,7 +2481,7 @@ mod tests {
         build_sqlserver_unsafe_type_query, capture_sqlserver_messages, format_sqlserver_numeric,
         is_blocking_sqlserver_unsafe_probe_error, is_sqlserver_spatial_column, is_sqlserver_variant_column,
         query_result_with_server_messages, requires_simple_query_batch, restore_sqlserver_legacy_probe_output_names,
-        sqlserver_batch_can_use_execute, sqlserver_cell_to_json, sqlserver_columns_sql,
+        sqlserver_batch_can_use_execute, sqlserver_bulk_token_row, sqlserver_cell_to_json, sqlserver_columns_sql,
         sqlserver_completion_assistant_sql, sqlserver_dml_output_returns_rows, sqlserver_filter_definition_error,
         sqlserver_hidden_schema_names, sqlserver_indexes_sql, sqlserver_legacy_indexes_sql, sqlserver_legacy_probe,
         sqlserver_legacy_probe_with_nonce, sqlserver_list_objects_sql, sqlserver_list_schemas_sql,
@@ -2423,6 +2497,15 @@ mod tests {
     use std::{borrow::Cow, time::Instant};
     use tiberius::{Column, ColumnData, ColumnType, IntoSql};
 
+    #[test]
+    fn sqlserver_bulk_token_row_owns_text_and_preserves_nulls() {
+        let row = sqlserver_bulk_token_row(vec![Some("Tieng Viet".to_string()), None]);
+        let values = row.iter().collect::<Vec<_>>();
+
+        assert!(matches!(&values[0], ColumnData::String(Some(value)) if value.as_ref() == "Tieng Viet"));
+        assert!(matches!(&values[1], ColumnData::String(None)));
+    }
+
     #[tokio::test]
     async fn sqlserver_ignores_non_info_tiberius_events() {
         let (_, messages) = capture_sqlserver_messages(async {
@@ -2588,6 +2671,16 @@ mod tests {
         assert!(!requires_simple_query_batch("UPDATE dbo.t SET id = 1;"));
     }
 
+    #[test]
+    fn sqlserver_local_temp_table_creation_keeps_session_scoped_query_path() {
+        assert!(requires_simple_query_batch("CREATE TABLE #stage (id INT);"));
+        assert!(requires_simple_query_batch("CREATE TABLE [#stage] ([id] INT);"));
+        assert!(requires_simple_query_batch(
+            "DECLARE @id INT = 1; CREATE TABLE #stage (id INT); INSERT INTO #stage VALUES (@id);"
+        ));
+        assert!(!requires_simple_query_batch("CREATE TABLE dbo.stage (id INT);"));
+    }
+
     #[test]
     fn sqlserver_cud_batches_use_execute_for_affected_rows() {
         assert!(sqlserver_batch_can_use_execute("UPDATE dbo.users SET active = 0 WHERE id = 1;"));

crates/dbx-core/src/transfer.rs

@@ -23,6 +23,33 @@ const MAX_ORACLE_INSERT_ALL_ROWS: usize = 500;
 const MAX_ORACLE_MERGE_ROWS: usize = 500;
 const TRANSFER_TARGET_TABLE_LOOKUP_LIMIT: usize = 1000;
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) struct SqlBatchLimits {
+    max_rows: usize,
+    target_sql_bytes: usize,
+    hard_sql_bytes: Option<usize>,
+}
+
+impl SqlBatchLimits {
+    pub(crate) fn for_database(db_type: &DatabaseType, requested_max_rows: usize) -> Self {
+        let max_rows = requested_max_rows.max(1).min(match db_type {
+            DatabaseType::SqlServer => MAX_SQLSERVER_INSERT_ROWS,
+            DatabaseType::Oracle => MAX_ORACLE_INSERT_ALL_ROWS,
+            _ => usize::MAX,
+        });
+        let target_sql_bytes = match db_type {
+            DatabaseType::CloudflareD1 => crate::db::cloudflare_d1::MAX_SQL_STATEMENT_BYTES,
+            _ => MAX_TRANSFER_WRITE_SQL_BYTES,
+        };
+        Self { max_rows, target_sql_bytes, hard_sql_bytes: None }
+    }
+
+    pub(crate) fn with_hard_sql_bytes(mut self, hard_sql_bytes: Option<usize>) -> Self {
+        self.hard_sql_bytes = hard_sql_bytes;
+        self
+    }
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize, Default, PartialEq)]
 #[serde(rename_all = "camelCase")]
 pub enum TransferMode {
@@ -308,12 +335,28 @@ fn is_postgres_integer_like_type(data_type: &str) -> bool {
     postgres_integer_bounds(data_type).is_some()
 }
 
-pub(crate) fn normalize_postgres_integer_literal(
+fn sqlserver_integer_bounds(data_type: &str) -> Option<(i128, i128)> {
+    let normalized = data_type.trim().to_ascii_lowercase();
+    match normalized.split(['(', ' ']).next().unwrap_or("") {
+        "bit" => Some((i128::MIN, i128::MAX)),
+        "tinyint" => Some((0, i128::from(u8::MAX))),
+        "smallint" => Some((i128::from(i16::MIN), i128::from(i16::MAX))),
+        "int" | "integer" => Some((i128::from(i32::MIN), i128::from(i32::MAX))),
+        "bigint" => Some((i128::from(i64::MIN), i128::from(i64::MAX))),
+        _ => None,
+    }
+}
+
+pub(crate) fn normalize_integer_literal(
     value: &str,
     db_type: &DatabaseType,
     column_type: Option<&str>,
 ) -> Option<String> {
-    let bounds = column_type.filter(|_| is_postgres_transfer_dialect(db_type)).and_then(postgres_integer_bounds)?;
+    let bounds = match db_type {
+        db_type if is_postgres_transfer_dialect(db_type) => column_type.and_then(postgres_integer_bounds),
+        DatabaseType::SqlServer => column_type.and_then(sqlserver_integer_bounds),
+        _ => None,
+    }?;
 
     // Excel numeric cells arrive as f64; normalize only an explicit zero fraction so real decimals,
     // scientific notation, and values outside the target integer range stay untouched.
@@ -1115,7 +1158,7 @@ pub fn escape_value_typed(val: &serde_json::Value, db_type: &DatabaseType, colum
             }
         },
         serde_json::Value::Number(n) => {
-            if let Some(integer_literal) = normalize_postgres_integer_literal(&n.to_string(), db_type, column_type) {
+            if let Some(integer_literal) = normalize_integer_literal(&n.to_string(), db_type, column_type) {
                 return integer_literal;
             }
             match db_type {
@@ -1130,7 +1173,7 @@ pub fn escape_value_typed(val: &serde_json::Value, db_type: &DatabaseType, colum
             }
         }
         serde_json::Value::String(s) => {
-            if let Some(integer_literal) = normalize_postgres_integer_literal(s, db_type, column_type) {
+            if let Some(integer_literal) = normalize_integer_literal(s, db_type, column_type) {
                 return integer_literal;
             }
             if let Some(binary_literal) = format_postgres_binary_sql_literal(s, db_type, column_type) {
@@ -1139,6 +1182,9 @@ pub fn escape_value_typed(val: &serde_json::Value, db_type: &DatabaseType, colum
             if let Some(binary_literal) = format_mysql_binary_sql_literal(s, db_type, column_type) {
                 return binary_literal;
             }
+            if let Some(binary_literal) = format_sqlserver_binary_sql_literal(s, db_type, column_type) {
+                return binary_literal;
+            }
             if let Some(numeric_literal) = format_mysql_numeric_string_literal(s, db_type, column_type) {
                 return numeric_literal;
             }
@@ -1250,6 +1296,29 @@ fn format_mysql_binary_sql_literal(value: &str, db_type: &DatabaseType, column_t
     }
 }
 
+fn format_sqlserver_binary_sql_literal(
+    value: &str,
+    db_type: &DatabaseType,
+    column_type: Option<&str>,
+) -> Option<String> {
+    if !matches!(db_type, DatabaseType::SqlServer) {
+        return None;
+    }
+    let column_type = column_type.filter(|column_type| is_binary_transfer_column_type(column_type))?;
+
+    let trimmed = value.trim();
+    if let Some(hex) = trimmed.strip_prefix("0x").or_else(|| trimmed.strip_prefix("0X")) {
+        if hex.len() % 2 == 0 && hex.as_bytes().iter().all(|byte| byte.is_ascii_hexdigit()) {
+            return Some(format!("0x{hex}"));
+        }
+    }
+    // SQL Server does not implicitly convert NVARCHAR literals to binary targets.
+    // Use the target type so text fallback preserves the same Unicode byte encoding
+    // that a direct typed conversion would produce.
+    let escaped = value.replace('\'', "''");
+    Some(format!("CONVERT({column_type}, N'{escaped}')"))
+}
+
 fn format_oracle_temporal_sql_literal(
     value: &str,
     db_type: &DatabaseType,
@@ -1914,21 +1983,71 @@ pub fn generate_insert_typed(
         return String::new();
     }
 
-    let full_table = qualified_table(table, schema, db_type);
-    let col_list = columns.iter().map(|c| quote_identifier(c, db_type)).collect::<Vec<_>>().join(", ");
-
+    let template = InsertSqlTemplate::new(columns, table, schema, db_type);
     let value_rows = value_rows_sql(rows, column_types, db_type);
-    if matches!(db_type, DatabaseType::Oracle) && rows.len() > 1 {
-        // Oracle 11g does not accept comma-separated multi-row VALUES lists.
-        let into_rows = value_rows
-            .iter()
-            .map(|values| format!("INTO {full_table} ({col_list}) VALUES {values}"))
-            .collect::<Vec<_>>()
-            .join("\n");
-        return format!("INSERT ALL\n{into_rows}\nSELECT 1 FROM dual");
+    template.build(&value_rows)
+}
+
+#[derive(Debug)]
+struct InsertSqlTemplate {
+    standard_prefix: String,
+    oracle_into_prefix: Option<String>,
+}
+
+impl InsertSqlTemplate {
+    fn new(columns: &[String], table: &str, schema: &str, db_type: &DatabaseType) -> Self {
+        let full_table = qualified_table(table, schema, db_type);
+        let col_list = columns.iter().map(|column| quote_identifier(column, db_type)).collect::<Vec<_>>().join(", ");
+        Self {
+            standard_prefix: format!("INSERT INTO {full_table} ({col_list}) VALUES\n"),
+            oracle_into_prefix: matches!(db_type, DatabaseType::Oracle)
+                .then(|| format!("INTO {full_table} ({col_list}) VALUES ")),
+        }
     }
 
-    format!("INSERT INTO {full_table} ({col_list}) VALUES\n{}", value_rows.join(",\n"))
+    fn build(&self, value_rows: &[String]) -> String {
+        if value_rows.is_empty() {
+            return String::new();
+        }
+        if let Some(into_prefix) = self.oracle_into_prefix.as_deref().filter(|_| value_rows.len() > 1) {
+            let mut sql = String::from("INSERT ALL\n");
+            for (index, values) in value_rows.iter().enumerate() {
+                if index > 0 {
+                    sql.push('\n');
+                }
+                sql.push_str(into_prefix);
+                sql.push_str(values);
+            }
+            sql.push_str("\nSELECT 1 FROM dual");
+            return sql;
+        }
+
+        let mut sql = self.standard_prefix.clone();
+        sql.push_str(&value_rows.join(",\n"));
+        sql
+    }
+
+    fn statement_bytes(&self, value_rows_bytes: usize, row_count: usize, db_type: &DatabaseType) -> usize {
+        if let Some(into_prefix) = self.oracle_into_prefix.as_deref().filter(|_| row_count > 1) {
+            return sql_text_bytes("INSERT ALL\n", db_type)
+                .saturating_add(sql_text_bytes(into_prefix, db_type).saturating_mul(row_count))
+                .saturating_add(value_rows_bytes)
+                .saturating_add(sql_text_bytes("\n", db_type).saturating_mul(row_count - 1))
+                .saturating_add(sql_text_bytes("\nSELECT 1 FROM dual", db_type));
+        }
+
+        sql_text_bytes(&self.standard_prefix, db_type)
+            .saturating_add(value_rows_bytes)
+            .saturating_add(sql_text_bytes(",\n", db_type).saturating_mul(row_count.saturating_sub(1)))
+    }
+}
+
+fn sql_text_bytes(sql: &str, db_type: &DatabaseType) -> usize {
+    if matches!(db_type, DatabaseType::SqlServer) {
+        sql.encode_utf16().count().saturating_mul(2)
+    } else {
+        sql.len()
+    }
 }
 
 fn value_rows_sql(
@@ -2167,53 +2286,55 @@ pub(crate) fn generate_insert_typed_sql_batches(
     table: &str,
     schema: &str,
     db_type: &DatabaseType,
-    requested_max_rows: usize,
-) -> Vec<(String, usize)> {
+    limits: SqlBatchLimits,
+) -> Result<Vec<(String, usize)>, String> {
     if rows.is_empty() {
-        return Vec::new();
+        return Ok(Vec::new());
     }
 
-    let max_rows = requested_max_rows.max(1);
-    let max_sql_bytes = match db_type {
-        DatabaseType::CloudflareD1 => crate::db::cloudflare_d1::MAX_SQL_STATEMENT_BYTES,
-        _ => MAX_TRANSFER_WRITE_SQL_BYTES,
-    };
+    let max_rows = limits.max_rows.max(1).min(match db_type {
+        DatabaseType::SqlServer => MAX_SQLSERVER_INSERT_ROWS,
+        DatabaseType::Oracle => MAX_ORACLE_INSERT_ALL_ROWS,
+        _ => usize::MAX,
+    });
+    let target_sql_bytes = limits.target_sql_bytes.max(1);
+    let batch_sql_bytes = limits.hard_sql_bytes.map_or(target_sql_bytes, |hard| target_sql_bytes.min(hard));
+    let template = InsertSqlTemplate::new(columns, table, schema, db_type);
+    let value_rows = value_rows_sql(rows, column_types, db_type);
+    let value_row_bytes = value_rows.iter().map(|row| sql_text_bytes(row, db_type)).collect::<Vec<_>>();
     let mut statements = Vec::new();
-    let mut start = 0;
+    let mut start = 0usize;
 
-    // First honor the row limit, then use binary search to find the largest statement that
-    // also fits the backend byte limit. This avoids generating every intermediate size.
-    while start < rows.len() {
-        let max_end = start.saturating_add(max_rows).min(rows.len());
-        let mut end = max_end;
-        let mut accepted = generate_insert_typed(columns, column_types, &rows[start..max_end], table, schema, db_type);
-
-        if accepted.len() > max_sql_bytes && max_end > start + 1 {
-            end = start + 1;
-            accepted = generate_insert_typed(columns, column_types, &rows[start..end], table, schema, db_type);
-            let mut low = start + 2;
-            let mut high = max_end;
-            while low <= high {
-                let candidate_end = low + (high - low) / 2;
-                let candidate =
-                    generate_insert_typed(columns, column_types, &rows[start..candidate_end], table, schema, db_type);
-                if candidate.len() <= max_sql_bytes {
-                    accepted = candidate;
-                    end = candidate_end;
-                    low = candidate_end + 1;
-                } else {
-                    high = candidate_end - 1;
+    while start < value_rows.len() {
+        let mut end = start;
+        let mut rows_bytes = 0usize;
+        while end < value_rows.len() && end - start < max_rows {
+            let single_row_bytes = template.statement_bytes(value_row_bytes[end], 1, db_type);
+            if let Some(hard_sql_bytes) = limits.hard_sql_bytes {
+                if single_row_bytes > hard_sql_bytes {
+                    return Err(format!(
+                        "SQL batch row {} requires {} bytes and exceeds the {} byte hard limit",
+                        end + 1,
+                        single_row_bytes,
+                        hard_sql_bytes
+                    ));
                 }
             }
+            let candidate_rows_bytes = rows_bytes.saturating_add(value_row_bytes[end]);
+            let candidate_row_count = end - start + 1;
+            let candidate_bytes = template.statement_bytes(candidate_rows_bytes, candidate_row_count, db_type);
+            if candidate_row_count > 1 && candidate_bytes > batch_sql_bytes {
+                break;
+            }
+            rows_bytes = candidate_rows_bytes;
+            end += 1;
         }
 
-        if !accepted.is_empty() {
-            statements.push((accepted, end - start));
-        }
+        statements.push((template.build(&value_rows[start..end]), end - start));
         start = end;
     }
 
-    statements
+    Ok(statements)
 }
 
 #[allow(clippy::too_many_arguments)]
@@ -2226,24 +2347,24 @@ fn generate_transfer_write_sql_batches(
     schema: &str,
     db_type: &DatabaseType,
     pk_columns: &[String],
-) -> Vec<String> {
+) -> Result<Vec<String>, String> {
     if rows.is_empty() {
-        return Vec::new();
+        return Ok(Vec::new());
     }
 
     if matches!(mode, TransferMode::Append | TransferMode::Overwrite) {
-        return generate_insert_typed_sql_batches(
+        return Ok(generate_insert_typed_sql_batches(
             columns,
             column_types,
             rows,
             table,
             schema,
             db_type,
-            max_transfer_write_rows(db_type, mode),
-        )
+            SqlBatchLimits::for_database(db_type, max_transfer_write_rows(db_type, mode)),
+        )?
         .into_iter()
         .map(|(sql, _)| sql)
-        .collect();
+        .collect());
     }
 
     let max_rows = max_transfer_write_rows(db_type, mode);
@@ -2291,7 +2412,7 @@ fn generate_transfer_write_sql_batches(
         start = end;
     }
 
-    statements
+    Ok(statements)
 }
 
 pub fn pagination_sql(
@@ -4107,7 +4228,7 @@ where
                 &request.target_schema,
                 target_db_type,
                 &[],
-            );
+            )?;
             for (statement_index, batch_sql) in write_statements.iter().enumerate() {
                 execute_on_pool(state, target_pool_key, batch_sql).await.map_err(|e| {
                     format!(
@@ -4452,7 +4573,7 @@ where
             &request.target_schema,
             target_db_type,
             &pk_columns,
-        );
+        )?;
         for (statement_index, batch_sql) in write_statements.iter().enumerate() {
             execute_transfer_write_statement(
                 state,
@@ -6313,6 +6434,34 @@ mod tests {
         assert_eq!(sql, "INSERT INTO [dbo].[flags] ([enabled], [deleted]) VALUES\n(1, 0)");
     }
 
+    #[test]
+    fn sqlserver_insert_formats_prefixed_hex_for_varbinary_columns() {
+        let sql = generate_insert_typed(
+            &[String::from("payload"), String::from("note")],
+            &[Some(String::from("varbinary(max)")), Some(String::from("nvarchar(64)"))],
+            &[vec![json!("0x0001ABff"), json!("0x0001ABff")]],
+            "files",
+            "dbo",
+            &DatabaseType::SqlServer,
+        );
+
+        assert_eq!(sql, "INSERT INTO [dbo].[files] ([payload], [note]) VALUES\n(0x0001ABff, N'0x0001ABff')");
+    }
+
+    #[test]
+    fn sqlserver_insert_explicitly_converts_plain_text_for_varbinary_columns() {
+        let sql = generate_insert_typed(
+            &[String::from("payload")],
+            &[Some(String::from("varbinary(max)"))],
+            &[vec![json!("O'Brien")]],
+            "files",
+            "dbo",
+            &DatabaseType::SqlServer,
+        );
+
+        assert_eq!(sql, "INSERT INTO [dbo].[files] ([payload]) VALUES\n(CONVERT(varbinary(max), N'O''Brien'))");
+    }
+
     #[test]
     fn dameng_insert_formats_bit_booleans_as_numeric_literals() {
         let sql = generate_insert_typed(
@@ -6586,7 +6735,8 @@ SELECT 1 FROM dual"#
             "APP",
             &DatabaseType::Oracle,
             &[],
-        );
+        )
+        .unwrap();
 
         assert_eq!(statements.len(), 2);
         assert_eq!(statements[0].matches("\nINTO ").count(), MAX_ORACLE_INSERT_ALL_ROWS);
@@ -6606,12 +6756,95 @@ SELECT 1 FROM dual"#
             "",
             &DatabaseType::Mysql,
             &[],
-        );
+        )
+        .unwrap();
 
         assert!(statements.len() > 1);
         assert!(statements.iter().all(|sql| sql.starts_with("INSERT INTO `events`")));
     }
 
+    #[test]
+    fn mysql_sql_batch_allows_one_row_over_soft_target() {
+        let rows = vec![vec![json!("x".repeat(256))]];
+        let limits = SqlBatchLimits { max_rows: 100, target_sql_bytes: 128, hard_sql_bytes: Some(1024) };
+
+        let batches = generate_insert_typed_sql_batches(
+            &[String::from("payload")],
+            &[Some(String::from("text"))],
+            &rows,
+            "events",
+            "",
+            &DatabaseType::Mysql,
+            limits,
+        )
+        .unwrap();
+
+        assert_eq!(batches.len(), 1);
+        assert_eq!(batches[0].1, 1);
+    }
+
+    #[test]
+    fn mysql_sql_batch_rejects_one_row_over_known_hard_limit() {
+        let rows = vec![vec![json!("x".repeat(256))]];
+        let limits = SqlBatchLimits { max_rows: 100, target_sql_bytes: 128, hard_sql_bytes: Some(200) };
+
+        let error = generate_insert_typed_sql_batches(
+            &[String::from("payload")],
+            &[Some(String::from("text"))],
+            &rows,
+            "events",
+            "",
+            &DatabaseType::Mysql,
+            limits,
+        )
+        .unwrap_err();
+
+        assert!(error.contains("row 1"));
+        assert!(error.contains("200 byte hard limit"));
+    }
+
+    #[test]
+    fn sqlserver_insert_batches_enforce_values_row_limit() {
+        let rows = (0..(MAX_SQLSERVER_INSERT_ROWS + 1)).map(|index| vec![json!(index)]).collect::<Vec<_>>();
+
+        let batches = generate_insert_typed_sql_batches(
+            &[String::from("id")],
+            &[Some(String::from("int"))],
+            &rows,
+            "events",
+            "dbo",
+            &DatabaseType::SqlServer,
+            SqlBatchLimits::for_database(&DatabaseType::SqlServer, rows.len()),
+        )
+        .unwrap();
+
+        assert_eq!(batches.iter().map(|(_, row_count)| *row_count).collect::<Vec<_>>(), vec![1000, 1]);
+    }
+
+    #[test]
+    fn sqlserver_insert_batches_measure_unicode_sql_as_utf16() {
+        let rows = (0..2).map(|_| vec![json!("x".repeat(140 * 1024))]).collect::<Vec<_>>();
+
+        let batches = generate_insert_typed_sql_batches(
+            &[String::from("payload")],
+            &[Some(String::from("nvarchar(max)"))],
+            &rows,
+            "events",
+            "dbo",
+            &DatabaseType::SqlServer,
+            SqlBatchLimits::for_database(&DatabaseType::SqlServer, rows.len()),
+        )
+        .unwrap();
+
+        assert_eq!(batches.iter().map(|(_, row_count)| *row_count).collect::<Vec<_>>(), vec![1, 1]);
+    }
+
+    #[test]
+    fn sqlserver_sql_byte_count_uses_utf16_code_units() {
+        assert_eq!(sql_text_bytes("AA\u{8d8a}\u{1f600}", &DatabaseType::SqlServer), 10);
+        assert_eq!(sql_text_bytes("AA\u{8d8a}\u{1f600}", &DatabaseType::Postgres), 9);
+    }
+
     #[test]
     fn transfer_write_sql_batches_keep_existing_upsert_sql_shape() {
         let statements = generate_transfer_write_sql_batches(
@@ -6623,7 +6856,8 @@ SELECT 1 FROM dual"#
             "",
             &DatabaseType::Mysql,
             &[String::from("id")],
-        );
+        )
+        .unwrap();
 
         assert_eq!(statements.len(), 1);
         assert!(statements[0].contains("ON DUPLICATE KEY UPDATE"));