perf(mysql): limit large table result memory

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

@@ -1,6 +1,6 @@
 use chrono::{NaiveDate, NaiveDateTime, NaiveTime};
 use futures::StreamExt;
-use mysql_async::consts::ColumnType;
+use mysql_async::consts::{ColumnFlags, ColumnType};
 use mysql_async::prelude::*;
 use percent_encoding::percent_decode_str;
 use rust_decimal::Decimal;
@@ -21,8 +21,8 @@ use crate::sql::{starts_with_executable_sql_keyword, starts_with_executable_sql_
 use crate::types::{
     ColumnInfo, CompletionAssistantCandidate, CompletionAssistantCandidateKind, CompletionAssistantMatchMode,
     CompletionAssistantObjectKind, CompletionAssistantRequest, CompletionAssistantResponse, DatabaseInfo,
-    ForeignKeyInfo, IndexInfo, ObjectInfo, ObjectStatistics, QueryMessage, QueryResult, SpatialColumnBuilder,
-    TableInfo, TriggerInfo,
+    ForeignKeyInfo, IndexInfo, LargeValueCell, ObjectInfo, ObjectStatistics, QueryMessage, QueryResult,
+    SpatialColumnBuilder, TableInfo, TriggerInfo,
 };
 
 use super::file_validator::validate_file_path;
@@ -57,6 +57,21 @@ pub struct MySqlQueryDialect {
     supports_admin_show_results: bool,
 }
 
+#[derive(Debug)]
+pub struct MySqlQueryResult {
+    pub result: QueryResult,
+    pub large_value_cells: Vec<LargeValueCell>,
+}
+
+impl MySqlQueryResult {
+    fn exact(result: QueryResult) -> Self {
+        Self { result, large_value_cells: Vec::new() }
+    }
+}
+
+const MYSQL_RESULT_CELL_PREVIEW_MIN_BYTES: usize = 256;
+const MYSQL_RESULT_CELL_PREVIEW_MAX_BYTES: usize = 8 * 1024;
+
 impl MySqlQueryDialect {
     pub fn for_connection(db_type: DatabaseType, driver_profile: Option<&str>) -> Self {
         let profile = driver_profile.map(str::to_ascii_lowercase);
@@ -569,6 +584,102 @@ fn mysql_spatial_column_builder(columns: &[mysql_async::Column]) -> SpatialColum
     )
 }
 
+fn mysql_result_cell_preview_bytes(
+    max_result_bytes: usize,
+    row_limit: usize,
+    columns: &[mysql_async::Column],
+) -> usize {
+    let serialized_column_weight = columns
+        .iter()
+        .map(
+            |column| {
+                if is_mysql_blob_column(column) || is_mysql_binary_string_column(column) {
+                    2usize
+                } else {
+                    1usize
+                }
+            },
+        )
+        .sum::<usize>()
+        .max(1);
+    max_result_bytes
+        .checked_div(row_limit.max(1))
+        .and_then(|per_row| per_row.checked_div(serialized_column_weight))
+        .unwrap_or(MYSQL_RESULT_CELL_PREVIEW_MIN_BYTES)
+        .clamp(MYSQL_RESULT_CELL_PREVIEW_MIN_BYTES, MYSQL_RESULT_CELL_PREVIEW_MAX_BYTES)
+}
+
+fn mysql_column_can_use_bounded_preview(column: &mysql_async::Column) -> bool {
+    matches!(
+        column.column_type(),
+        ColumnType::MYSQL_TYPE_JSON
+            | ColumnType::MYSQL_TYPE_BLOB
+            | ColumnType::MYSQL_TYPE_LONG_BLOB
+            | ColumnType::MYSQL_TYPE_MEDIUM_BLOB
+            | ColumnType::MYSQL_TYPE_TINY_BLOB
+            | ColumnType::MYSQL_TYPE_STRING
+            | ColumnType::MYSQL_TYPE_VAR_STRING
+            | ColumnType::MYSQL_TYPE_VARCHAR
+    )
+}
+
+fn mysql_bounded_value_preview(
+    row: &mysql_async::Row,
+    index: usize,
+    preview_bytes: usize,
+    protected_indexes: &HashSet<usize>,
+) -> Option<(serde_json::Value, usize)> {
+    if protected_indexes.contains(&index) {
+        return None;
+    }
+    let column = row.columns_ref().get(index)?;
+    if !mysql_column_can_use_bounded_preview(column) {
+        return None;
+    }
+    let mysql_async::Value::Bytes(bytes) = row.as_ref(index)? else {
+        return None;
+    };
+    if bytes.len() <= preview_bytes {
+        return None;
+    }
+
+    let end = if is_mysql_binary_charset(column) {
+        preview_bytes.min(bytes.len())
+    } else {
+        let mut end = preview_bytes.min(bytes.len());
+        while end > 0 && std::str::from_utf8(&bytes[..end]).is_err() {
+            end -= 1;
+        }
+        end
+    };
+    let mut value = if is_mysql_binary_charset(column) {
+        mysql_bytes_to_json(bytes[..end].to_vec(), column)
+    } else {
+        let text = String::from_utf8_lossy(&bytes[..end]);
+        serde_json::Value::String(fix_potential_double_encoding(&text))
+    };
+    if let serde_json::Value::String(text) = &mut value {
+        text.push_str("...");
+    }
+    Some((value, bytes.len()))
+}
+
+fn mysql_result_protected_column_indexes(
+    columns: &[mysql_async::Column],
+    protected_columns: &[String],
+) -> HashSet<usize> {
+    let protected: HashSet<String> = protected_columns.iter().map(|column| column.to_ascii_lowercase()).collect();
+    columns
+        .iter()
+        .enumerate()
+        .filter_map(|(index, column)| {
+            (column.flags().contains(ColumnFlags::PRI_KEY_FLAG)
+                || protected.contains(&column.name_str().to_ascii_lowercase()))
+            .then_some(index)
+        })
+        .collect()
+}
+
 fn mysql_row_to_json_with_srids(
     row: &mysql_async::Row,
     spatial_columns: &mut SpatialColumnBuilder,
@@ -603,6 +714,51 @@ fn mysql_row_to_json_with_srids(
     (values, srids)
 }
 
+fn mysql_row_to_json_with_srids_and_previews(
+    row: &mysql_async::Row,
+    spatial_columns: &mut SpatialColumnBuilder,
+    row_index: usize,
+    preview_bytes: Option<usize>,
+    protected_indexes: &HashSet<usize>,
+) -> (Vec<serde_json::Value>, Vec<Option<u32>>, Vec<LargeValueCell>) {
+    let mut srids = vec![None; row.len()];
+    let mut large_value_cells = Vec::new();
+    let values = (0..row.len())
+        .map(|index| {
+            let is_geometry = row
+                .columns_ref()
+                .get(index)
+                .is_some_and(|column| column.column_type() == ColumnType::MYSQL_TYPE_GEOMETRY);
+            if is_geometry {
+                let Some(bytes) = row_get::<Vec<u8>, _>(row, index) else {
+                    spatial_columns.observe(index, None);
+                    return serde_json::Value::Null;
+                };
+                return match decode_mysql_geometry(&bytes) {
+                    Some(geometry) => {
+                        spatial_columns.observe(index, geometry.srid);
+                        srids[index] = geometry.srid;
+                        serde_json::Value::String(geometry.wkt)
+                    }
+                    None => {
+                        spatial_columns.observe(index, None);
+                        super::binary_value_to_json(&bytes)
+                    }
+                };
+            }
+
+            if let Some((preview, original_bytes)) =
+                preview_bytes.and_then(|limit| mysql_bounded_value_preview(row, index, limit, protected_indexes))
+            {
+                large_value_cells.push(LargeValueCell { row_index, column_index: index, original_bytes });
+                return preview;
+            }
+            mysql_value_to_json(row, index)
+        })
+        .collect();
+    (values, srids, large_value_cells)
+}
+
 fn mysql_temporal_value_to_json(
     column_type: ColumnType,
     datetime: Option<NaiveDateTime>,
@@ -3923,8 +4079,10 @@ async fn execute_result_set_with_text_protocol_on_conn(
     sql: &str,
     row_limit: usize,
     max_rows: Option<usize>,
+    max_result_bytes: Option<usize>,
+    result_key_columns: &[String],
     start: Instant,
-) -> Result<QueryResult, String> {
+) -> Result<MySqlQueryResult, String> {
     let mut result = conn.query_iter(sql).await.map_err(|e| e.to_string())?;
     if !advance_to_result_set_with_columns(&mut result).await? {
         let affected_rows = result.affected_rows();
@@ -3932,7 +4090,7 @@ async fn execute_result_set_with_text_protocol_on_conn(
         let info = result.info().into_owned();
         drop(result);
         let messages = collect_mysql_server_messages(conn, warnings, &info).await;
-        return Ok(QueryResult {
+        return Ok(MySqlQueryResult::exact(QueryResult {
             columns: vec![],
             column_types: Vec::new(),
             column_sortables: vec![],
@@ -3946,13 +4104,16 @@ async fn execute_result_set_with_text_protocol_on_conn(
             has_more: false,
             elasticsearch_raw_body: None,
             messages,
-        });
+        }));
     }
     let columns: Vec<String> = result.columns_ref().iter().map(|c| c.name_str().to_string()).collect();
     let column_types: Vec<String> = result.columns_ref().iter().map(mysql_column_type_name).collect();
     let mut spatial_columns = mysql_spatial_column_builder(result.columns_ref());
+    let preview_bytes =
+        max_result_bytes.map(|max_bytes| mysql_result_cell_preview_bytes(max_bytes, row_limit, result.columns_ref()));
+    let protected_indexes = mysql_result_protected_column_indexes(result.columns_ref(), result_key_columns);
 
-    if should_collect_text_result_set(sql, row_limit, max_rows) {
+    if max_result_bytes.is_none() && should_collect_text_result_set(sql, row_limit, max_rows) {
         let rows: Vec<mysql_async::Row> = result.collect_and_drop().await.map_err(|e| e.to_string())?;
         // collect_and_drop consumed the result; warnings/info now reflect the
         // trailing EOF/OK packet of the finished result set.
@@ -3971,7 +4132,7 @@ async fn execute_result_set_with_text_protocol_on_conn(
             })
             .collect();
 
-        return Ok(QueryResult {
+        return Ok(MySqlQueryResult::exact(QueryResult {
             columns,
             column_types,
             column_sortables: vec![],
@@ -3985,11 +4146,12 @@ async fn execute_result_set_with_text_protocol_on_conn(
             has_more: false,
             elasticsearch_raw_body: None,
             messages,
-        });
+        }));
     }
 
     let mut result_rows: Vec<Vec<serde_json::Value>> = Vec::new();
     let mut spatial_values: Vec<Vec<Option<u32>>> = Vec::new();
+    let mut large_value_cells = Vec::new();
     let mut truncated = false;
     let mut stream = result
         .stream::<mysql_async::Row>()
@@ -4003,9 +4165,17 @@ async fn execute_result_set_with_text_protocol_on_conn(
             truncated = true;
             break;
         }
-        let (values, srids) = mysql_row_to_json_with_srids(&row, &mut spatial_columns);
+        let row_index = result_rows.len();
+        let (values, srids, mut row_large_values) = mysql_row_to_json_with_srids_and_previews(
+            &row,
+            &mut spatial_columns,
+            row_index,
+            preview_bytes,
+            &protected_indexes,
+        );
         result_rows.push(values);
         spatial_values.push(srids);
+        large_value_cells.append(&mut row_large_values);
     }
     drop(stream);
 
@@ -4023,20 +4193,23 @@ async fn execute_result_set_with_text_protocol_on_conn(
         collect_mysql_server_messages(conn, warnings, &info).await
     };
 
-    Ok(QueryResult {
-        columns,
-        column_types,
-        column_sortables: vec![],
-        spatial_columns: spatial_columns.finish(),
-        spatial_values,
-        rows: result_rows,
-        affected_rows: 0,
-        execution_time_ms: start.elapsed().as_millis(),
-        truncated,
-        session_id: None,
-        has_more: false,
-        elasticsearch_raw_body: None,
-        messages,
+    Ok(MySqlQueryResult {
+        result: QueryResult {
+            columns,
+            column_types,
+            column_sortables: vec![],
+            spatial_columns: spatial_columns.finish(),
+            spatial_values,
+            rows: result_rows,
+            affected_rows: 0,
+            execution_time_ms: start.elapsed().as_millis(),
+            truncated,
+            session_id: None,
+            has_more: false,
+            elasticsearch_raw_body: None,
+            messages,
+        },
+        large_value_cells,
     })
 }
 
@@ -4045,8 +4218,10 @@ async fn execute_result_sets_with_text_protocol_on_conn(
     sql: &str,
     row_limit: usize,
     max_rows: Option<usize>,
+    max_result_bytes: Option<usize>,
+    result_key_columns: &[String],
     start: Instant,
-) -> Result<Vec<QueryResult>, String> {
+) -> Result<Vec<MySqlQueryResult>, String> {
     // Per-set warnings/info read after each `collect()` are only accurate when
     // the connection negotiated CLIENT_DEPRECATE_EOF: with legacy EOF packets
     // the next result set's column-definition EOF clobbers the previous OK
@@ -4059,7 +4234,7 @@ async fn execute_result_sets_with_text_protocol_on_conn(
     // on the last result set reports warnings.
     let capture_per_set_messages = conn.opts().deprecate_eof();
     let mut result = conn.query_iter(sql).await.map_err(|e| e.to_string())?;
-    let mut results: Vec<QueryResult> = Vec::new();
+    let mut results: Vec<MySqlQueryResult> = Vec::new();
     let mut result_set_warnings: Vec<u16> = Vec::new();
 
     while advance_to_result_set_with_columns(&mut result).await? {
@@ -4067,9 +4242,13 @@ async fn execute_result_sets_with_text_protocol_on_conn(
         let column_types: Vec<String> = result.columns_ref().iter().map(mysql_column_type_name).collect();
         let mut spatial_columns = mysql_spatial_column_builder(result.columns_ref());
         let mut spatial_values = Vec::new();
+        let mut large_value_cells = Vec::new();
         let mut truncated = false;
+        let preview_bytes = max_result_bytes
+            .map(|max_bytes| mysql_result_cell_preview_bytes(max_bytes, row_limit, result.columns_ref()));
+        let protected_indexes = mysql_result_protected_column_indexes(result.columns_ref(), result_key_columns);
 
-        let rows = if should_collect_text_result_set(sql, row_limit, max_rows) {
+        let rows = if max_result_bytes.is_none() && should_collect_text_result_set(sql, row_limit, max_rows) {
             let rows: Vec<mysql_async::Row> = result.collect().await.map_err(|e| e.to_string())?;
             truncated = rows.len() > row_limit;
             rows.iter()
@@ -4091,9 +4270,17 @@ async fn execute_result_sets_with_text_protocol_on_conn(
             while let Some(row) = stream.next().await {
                 let row = row.map_err(|e| e.to_string())?;
                 if rows.len() < row_limit {
-                    let (values, srids) = mysql_row_to_json_with_srids(&row, &mut spatial_columns);
+                    let row_index = rows.len();
+                    let (values, srids, mut row_large_values) = mysql_row_to_json_with_srids_and_previews(
+                        &row,
+                        &mut spatial_columns,
+                        row_index,
+                        preview_bytes,
+                        &protected_indexes,
+                    );
                     rows.push(values);
                     spatial_values.push(srids);
+                    large_value_cells.append(&mut row_large_values);
                 } else {
                     truncated = true;
                 }
@@ -4108,20 +4295,23 @@ async fn execute_result_sets_with_text_protocol_on_conn(
             Vec::new()
         };
         result_set_warnings.push(warnings);
-        results.push(QueryResult {
-            columns,
-            column_types,
-            column_sortables: vec![],
-            spatial_columns: spatial_columns.finish(),
-            spatial_values,
-            rows,
-            affected_rows: 0,
-            execution_time_ms: start.elapsed().as_millis(),
-            truncated,
-            session_id: None,
-            has_more: false,
-            elasticsearch_raw_body: None,
-            messages,
+        results.push(MySqlQueryResult {
+            result: QueryResult {
+                columns,
+                column_types,
+                column_sortables: vec![],
+                spatial_columns: spatial_columns.finish(),
+                spatial_values,
+                rows,
+                affected_rows: 0,
+                execution_time_ms: start.elapsed().as_millis(),
+                truncated,
+                session_id: None,
+                has_more: false,
+                elasticsearch_raw_body: None,
+                messages,
+            },
+            large_value_cells,
         });
     }
 
@@ -4131,7 +4321,7 @@ async fn execute_result_sets_with_text_protocol_on_conn(
         let info = result.info().into_owned();
         drop(result);
         let messages = collect_mysql_server_messages(conn, warnings, &info).await;
-        results.push(QueryResult {
+        results.push(MySqlQueryResult::exact(QueryResult {
             columns: vec![],
             column_types: Vec::new(),
             column_sortables: vec![],
@@ -4145,7 +4335,7 @@ async fn execute_result_sets_with_text_protocol_on_conn(
             has_more: false,
             elasticsearch_raw_body: None,
             messages,
-        });
+        }));
         return Ok(results);
     }
     // Without per-set capture (no CLIENT_DEPRECATE_EOF) the per-iteration
@@ -4168,9 +4358,9 @@ async fn execute_result_sets_with_text_protocol_on_conn(
         }
         if index == last_index {
             let mut messages = collect_mysql_server_messages(conn, warnings, "").await;
-            results[index].messages.append(&mut messages);
+            results[index].result.messages.append(&mut messages);
         } else {
-            results[index].messages.push(mysql_warnings_fallback_message(warnings));
+            results[index].result.messages.push(mysql_warnings_fallback_message(warnings));
         }
     }
 
@@ -4193,15 +4383,21 @@ async fn execute_result_set_with_prepared_protocol_on_conn(
     conn: &mut mysql_async::Conn,
     sql: &str,
     row_limit: usize,
+    max_result_bytes: Option<usize>,
+    result_key_columns: &[String],
     start: Instant,
-) -> Result<QueryResult, String> {
+) -> Result<MySqlQueryResult, String> {
     let mut result = conn.exec_iter(sql, ()).await.map_err(|e| e.to_string())?;
     let columns: Vec<String> = result.columns_ref().iter().map(|c| c.name_str().to_string()).collect();
     let column_types: Vec<String> = result.columns_ref().iter().map(mysql_column_type_name).collect();
     let mut spatial_columns = mysql_spatial_column_builder(result.columns_ref());
+    let preview_bytes =
+        max_result_bytes.map(|max_bytes| mysql_result_cell_preview_bytes(max_bytes, row_limit, result.columns_ref()));
+    let protected_indexes = mysql_result_protected_column_indexes(result.columns_ref(), result_key_columns);
 
     let mut result_rows: Vec<Vec<serde_json::Value>> = Vec::new();
     let mut spatial_values: Vec<Vec<Option<u32>>> = Vec::new();
+    let mut large_value_cells = Vec::new();
     let mut truncated = false;
     let mut stream = result
         .stream::<mysql_async::Row>()
@@ -4215,9 +4411,17 @@ async fn execute_result_set_with_prepared_protocol_on_conn(
             truncated = true;
             break;
         }
-        let (values, srids) = mysql_row_to_json_with_srids(&row, &mut spatial_columns);
+        let row_index = result_rows.len();
+        let (values, srids, mut row_large_values) = mysql_row_to_json_with_srids_and_previews(
+            &row,
+            &mut spatial_columns,
+            row_index,
+            preview_bytes,
+            &protected_indexes,
+        );
         result_rows.push(values);
         spatial_values.push(srids);
+        large_value_cells.append(&mut row_large_values);
     }
     drop(stream);
 
@@ -4235,20 +4439,23 @@ async fn execute_result_set_with_prepared_protocol_on_conn(
         collect_mysql_server_messages(conn, warnings, &info).await
     };
 
-    Ok(QueryResult {
-        columns,
-        column_types,
-        column_sortables: vec![],
-        spatial_columns: spatial_columns.finish(),
-        spatial_values,
-        rows: result_rows,
-        affected_rows: 0,
-        execution_time_ms: start.elapsed().as_millis(),
-        truncated,
-        session_id: None,
-        has_more: false,
-        elasticsearch_raw_body: None,
-        messages,
+    Ok(MySqlQueryResult {
+        result: QueryResult {
+            columns,
+            column_types,
+            column_sortables: vec![],
+            spatial_columns: spatial_columns.finish(),
+            spatial_values,
+            rows: result_rows,
+            affected_rows: 0,
+            execution_time_ms: start.elapsed().as_millis(),
+            truncated,
+            session_id: None,
+            has_more: false,
+            elasticsearch_raw_body: None,
+            messages,
+        },
+        large_value_cells,
     })
 }
 
@@ -4479,17 +4686,56 @@ pub async fn execute_query_on_conn_with_max_rows(
     max_rows: Option<usize>,
     dialect: MySqlQueryDialect,
 ) -> Result<QueryResult, String> {
+    execute_query_on_conn_with_limits(conn, sql, bare, max_rows, None, &[], dialect).await.map(|result| result.result)
+}
+
+pub async fn execute_query_on_conn_with_limits(
+    conn: &mut mysql_async::Conn,
+    sql: &str,
+    bare: bool,
+    max_rows: Option<usize>,
+    max_result_bytes: Option<usize>,
+    result_key_columns: &[String],
+    dialect: MySqlQueryDialect,
+) -> Result<MySqlQueryResult, String> {
     let start = Instant::now();
     let row_limit = query_result_row_limit(max_rows);
 
     if is_result_set_query(sql, dialect) {
         if bare || prefers_text_protocol_query(sql, dialect) {
-            execute_result_set_with_text_protocol_on_conn(conn, sql, row_limit, max_rows, start).await
+            execute_result_set_with_text_protocol_on_conn(
+                conn,
+                sql,
+                row_limit,
+                max_rows,
+                max_result_bytes,
+                result_key_columns,
+                start,
+            )
+            .await
         } else {
-            match execute_result_set_with_prepared_protocol_on_conn(conn, sql, row_limit, start).await {
+            match execute_result_set_with_prepared_protocol_on_conn(
+                conn,
+                sql,
+                row_limit,
+                max_result_bytes,
+                result_key_columns,
+                start,
+            )
+            .await
+            {
                 Ok(result) => Ok(result),
                 Err(err) if mysql_error_should_retry_with_text_protocol(&err) => {
-                    execute_result_set_with_text_protocol_on_conn(conn, sql, row_limit, max_rows, start).await
+                    execute_result_set_with_text_protocol_on_conn(
+                        conn,
+                        sql,
+                        row_limit,
+                        max_rows,
+                        max_result_bytes,
+                        result_key_columns,
+                        start,
+                    )
+                    .await
                 }
                 Err(err) => Err(err),
             }
@@ -4514,7 +4760,7 @@ pub async fn execute_query_on_conn_with_max_rows(
         restore_explicit_timestamp_defaults_for_query(conn, previous_explicit_timestamp_defaults).await;
         drop_result.map_err(|e| e.to_string())?;
 
-        Ok(QueryResult {
+        Ok(MySqlQueryResult::exact(QueryResult {
             columns: vec![],
             column_types: Vec::new(),
             column_sortables: vec![],
@@ -4528,7 +4774,7 @@ pub async fn execute_query_on_conn_with_max_rows(
             has_more: false,
             elasticsearch_raw_body: None,
             messages,
-        })
+        }))
     }
 }
 
@@ -4539,12 +4785,36 @@ pub async fn execute_query_results_on_conn_with_max_rows(
     max_rows: Option<usize>,
     dialect: MySqlQueryDialect,
 ) -> Result<Vec<QueryResult>, String> {
+    execute_query_results_on_conn_with_limits(conn, sql, bare, max_rows, None, &[], dialect)
+        .await
+        .map(|results| results.into_iter().map(|result| result.result).collect())
+}
+
+pub async fn execute_query_results_on_conn_with_limits(
+    conn: &mut mysql_async::Conn,
+    sql: &str,
+    bare: bool,
+    max_rows: Option<usize>,
+    max_result_bytes: Option<usize>,
+    result_key_columns: &[String],
+    dialect: MySqlQueryDialect,
+) -> Result<Vec<MySqlQueryResult>, String> {
     if is_result_set_query(sql, dialect) && (bare || prefers_text_protocol_query(sql, dialect)) {
         let start = Instant::now();
-        execute_result_sets_with_text_protocol_on_conn(conn, sql, query_result_row_limit(max_rows), max_rows, start)
-            .await
+        execute_result_sets_with_text_protocol_on_conn(
+            conn,
+            sql,
+            query_result_row_limit(max_rows),
+            max_rows,
+            max_result_bytes,
+            result_key_columns,
+            start,
+        )
+        .await
     } else {
-        execute_query_on_conn_with_max_rows(conn, sql, bare, max_rows, dialect).await.map(|result| vec![result])
+        execute_query_on_conn_with_limits(conn, sql, bare, max_rows, max_result_bytes, result_key_columns, dialect)
+            .await
+            .map(|result| vec![result])
     }
 }
 
@@ -5391,6 +5661,10 @@ mod tests {
         new_row(values, columns)
     }
 
+    fn mysql_test_row_with_columns(values: Vec<Value>, columns: Vec<Column>) -> mysql_async::Row {
+        new_row(values, columns.into())
+    }
+
     #[test]
     fn mysql_first_column_value_ignores_extra_result_columns() {
         let row = mysql_async::from_row::<mysql_async::Row>(mysql_test_row(vec![
@@ -6493,6 +6767,88 @@ mod tests {
         );
     }
 
+    #[test]
+    fn mysql_large_text_preview_preserves_key_and_reports_cell_coordinates() {
+        let id_column = Column::new(ColumnType::MYSQL_TYPE_VAR_STRING)
+            .with_name(b"id")
+            .with_character_set(45)
+            .with_column_length(64);
+        let payload_column = Column::new(ColumnType::MYSQL_TYPE_LONG_BLOB)
+            .with_name(b"payload")
+            .with_character_set(45)
+            .with_column_length(u32::MAX);
+        let row = mysql_test_row_with_columns(
+            vec![Value::Bytes(vec![b'k'; 2048]), Value::Bytes(vec![b'x'; 32 * 1024])],
+            vec![id_column, payload_column],
+        );
+        let protected = mysql_result_protected_column_indexes(row.columns_ref(), &["id".to_string()]);
+        let mut spatial_columns = mysql_spatial_column_builder(row.columns_ref());
+
+        let (values, _srids, cells) =
+            mysql_row_to_json_with_srids_and_previews(&row, &mut spatial_columns, 7, Some(1024), &protected);
+
+        assert_eq!(values[0], serde_json::json!("k".repeat(2048)));
+        assert_eq!(values[1].as_str().map(str::len), Some(1027));
+        assert!(values[1].as_str().is_some_and(|value| value.ends_with("...")));
+        assert_eq!(cells, vec![LargeValueCell { row_index: 7, column_index: 1, original_bytes: 32 * 1024 }]);
+    }
+
+    #[test]
+    fn mysql_large_value_preview_protects_protocol_primary_keys_without_frontend_metadata() {
+        let id_column = Column::new(ColumnType::MYSQL_TYPE_VAR_STRING)
+            .with_name(b"id")
+            .with_character_set(45)
+            .with_flags(ColumnFlags::PRI_KEY_FLAG)
+            .with_column_length(3072);
+        let payload_column = Column::new(ColumnType::MYSQL_TYPE_JSON)
+            .with_name(b"payload")
+            .with_character_set(45)
+            .with_column_length(u32::MAX);
+        let row = mysql_test_row_with_columns(
+            vec![Value::Bytes(vec![b'k'; 2048]), Value::Bytes(vec![b'x'; 4096])],
+            vec![id_column, payload_column],
+        );
+        let protected = mysql_result_protected_column_indexes(row.columns_ref(), &[]);
+        let mut spatial_columns = mysql_spatial_column_builder(row.columns_ref());
+
+        let (values, _srids, cells) =
+            mysql_row_to_json_with_srids_and_previews(&row, &mut spatial_columns, 0, Some(512), &protected);
+
+        assert_eq!(values[0], serde_json::json!("k".repeat(2048)));
+        assert_eq!(cells, vec![LargeValueCell { row_index: 0, column_index: 1, original_bytes: 4096 }]);
+    }
+
+    #[test]
+    fn mysql_large_binary_preview_is_bounded_hex() {
+        let payload_column = Column::new(ColumnType::MYSQL_TYPE_LONG_BLOB)
+            .with_name(b"payload")
+            .with_character_set(63)
+            .with_flags(ColumnFlags::BLOB_FLAG)
+            .with_column_length(u32::MAX);
+        let row = mysql_test_row_with_columns(vec![Value::Bytes(vec![0xAB; 4096])], vec![payload_column]);
+        let mut spatial_columns = mysql_spatial_column_builder(row.columns_ref());
+
+        let (values, _srids, cells) =
+            mysql_row_to_json_with_srids_and_previews(&row, &mut spatial_columns, 0, Some(512), &HashSet::new());
+
+        let preview = values[0].as_str().unwrap();
+        assert!(preview.starts_with("0xabab"));
+        assert!(preview.ends_with("..."));
+        assert_eq!(preview.len(), 2 + 512 * 2 + 3);
+        assert_eq!(cells[0].original_bytes, 4096);
+    }
+
+    #[test]
+    fn mysql_result_preview_budget_scales_with_page_and_column_count() {
+        let text_column = mysql_test_column(ColumnType::MYSQL_TYPE_LONG_BLOB, 45, ColumnFlags::empty(), u32::MAX);
+        let binary_column = mysql_test_column(ColumnType::MYSQL_TYPE_LONG_BLOB, 63, ColumnFlags::BLOB_FLAG, u32::MAX);
+        assert_eq!(mysql_result_cell_preview_bytes(32 * 1024 * 1024, 1, std::slice::from_ref(&text_column)), 8 * 1024);
+        assert_eq!(mysql_result_cell_preview_bytes(32 * 1024 * 1024, 1000, &vec![text_column.clone(); 4]), 8 * 1024);
+        assert_eq!(mysql_result_cell_preview_bytes(32 * 1024 * 1024, 1000, &vec![text_column.clone(); 16]), 2097);
+        assert_eq!(mysql_result_cell_preview_bytes(1, 1000, &vec![text_column; 100]), 256);
+        assert_eq!(mysql_result_cell_preview_bytes(32 * 1024 * 1024, 1000, &vec![binary_column; 4]), 4194);
+    }
+
     #[test]
     fn mysql_binary_preview_uses_charset_to_separate_blob_from_text() {
         let text_column = mysql_test_column(ColumnType::MYSQL_TYPE_BLOB, 45, ColumnFlags::empty(), 65_535);

crates/dbx-core/src/query.rs

@@ -168,6 +168,8 @@ fn append_typed_sql_error_context(error: &str, _sql: &str) -> String {
 pub struct ExecuteMultiResult {
     #[serde(flatten)]
     pub result: db::QueryResult,
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub large_value_cells: Vec<db::LargeValueCell>,
     #[serde(skip_serializing_if = "is_false")]
     pub execution_error: bool,
     #[serde(skip_serializing_if = "Option::is_none")]
@@ -215,12 +217,26 @@ fn report_execute_multi_progress(
 impl ExecuteMultiResult {
     fn execution_error(result: db::QueryResult) -> Self {
         let error = error_from_query_result(&result);
-        Self { result, execution_error: true, statement_index: None, error, server_message: false }
+        Self {
+            result,
+            large_value_cells: Vec::new(),
+            execution_error: true,
+            statement_index: None,
+            error,
+            server_message: false,
+        }
     }
 
     fn execution_error_with_index(result: db::QueryResult, statement_index: usize) -> Self {
         let error = error_from_query_result(&result);
-        Self { result, execution_error: true, statement_index: Some(statement_index), error, server_message: false }
+        Self {
+            result,
+            large_value_cells: Vec::new(),
+            execution_error: true,
+            statement_index: Some(statement_index),
+            error,
+            server_message: false,
+        }
     }
 
     fn execution_error_with_backend(
@@ -228,12 +244,35 @@ impl ExecuteMultiResult {
         statement_index: Option<usize>,
         error: crate::backend_error::BackendError,
     ) -> Self {
-        Self { result, execution_error: true, statement_index, error: Some(error), server_message: false }
+        Self {
+            result,
+            large_value_cells: Vec::new(),
+            execution_error: true,
+            statement_index,
+            error: Some(error),
+            server_message: false,
+        }
     }
 
     fn success_with_index(result: db::QueryResult, statement_index: usize) -> Self {
         Self {
             result,
+            large_value_cells: Vec::new(),
+            execution_error: false,
+            statement_index: Some(statement_index),
+            error: None,
+            server_message: false,
+        }
+    }
+
+    fn success_with_index_and_large_values(
+        result: db::QueryResult,
+        statement_index: usize,
+        large_value_cells: Vec<db::LargeValueCell>,
+    ) -> Self {
+        Self {
+            result,
+            large_value_cells,
             execution_error: false,
             statement_index: Some(statement_index),
             error: None,
@@ -253,7 +292,14 @@ impl ExecuteMultiResult {
 
 impl From<db::QueryResult> for ExecuteMultiResult {
     fn from(result: db::QueryResult) -> Self {
-        Self { result, execution_error: false, statement_index: None, error: None, server_message: false }
+        Self {
+            result,
+            large_value_cells: Vec::new(),
+            execution_error: false,
+            statement_index: None,
+            error: None,
+            server_message: false,
+        }
     }
 }
 
@@ -261,6 +307,7 @@ impl From<db::sqlserver::SqlServerBatchResult> for ExecuteMultiResult {
     fn from(result: db::sqlserver::SqlServerBatchResult) -> Self {
         Self {
             result: result.result,
+            large_value_cells: Vec::new(),
             execution_error: false,
             statement_index: None,
             error: None,
@@ -588,6 +635,10 @@ pub struct QueryExecutionOptions {
     pub max_rows: Option<usize>,
     pub fetch_size: Option<usize>,
     pub page_size: Option<usize>,
+    pub max_result_bytes: Option<usize>,
+    /// Result columns that must stay exact because clients use them as stable
+    /// row identifiers when fetching full large-cell values on demand.
+    pub result_key_columns: Vec<String>,
     /// Doris / StarRocks catalog selected for this query tab.
     pub catalog: Option<String>,
     pub result_session_id: Option<String>,
@@ -1318,6 +1369,7 @@ async fn do_execute_typed(
             let p = p.clone();
             let bare = *mode == crate::connection::MysqlMode::Bare;
             let max_rows = options.max_rows;
+            let max_result_bytes = options.max_result_bytes.filter(|value| *value > 0);
             drop(connections);
             let mut conn = match db::mysql::get_conn_with_health_check_with_cancel(
                 &p,
@@ -1358,12 +1410,21 @@ async fn do_execute_typed(
                 ),
             )
             .await?;
-            wait_for_query_opt(
+            wait_for_result_opt(
                 cancel_token,
                 query_timeout,
-                db::mysql::execute_query_on_conn_with_max_rows(&mut conn, sql, bare, max_rows, mysql_dialect),
+                db::mysql::execute_query_on_conn_with_limits(
+                    &mut conn,
+                    sql,
+                    bare,
+                    max_rows,
+                    max_result_bytes,
+                    &options.result_key_columns,
+                    mysql_dialect,
+                ),
             )
             .await
+            .map(|result| result.result)
         }
         PoolKind::Postgres(p) => {
             let p = p.clone();
@@ -2219,7 +2280,7 @@ pub async fn execute_multi_core_with_options_for_client_and_progress_typed(
         }
     };
 
-    if statements.len() <= 1 {
+    if statements.len() <= 1 && !(mysql_pool.is_some() && options.max_result_bytes.is_some_and(|value| value > 0)) {
         let single_sql = statements.into_iter().next().unwrap_or_default();
         return single_statement_multi_result(
             execute_sql_statement_with_options_typed(
@@ -2323,13 +2384,15 @@ fn single_statement_multi_result(
 }
 
 trait MysqlBatchStatementExecutor {
-    async fn execute_statement(&mut self, statement: &str) -> Result<Vec<db::QueryResult>, String>;
+    async fn execute_statement(&mut self, statement: &str) -> Result<Vec<db::mysql::MySqlQueryResult>, String>;
 
     async fn execute_non_result_batch(&mut self, statements: &[String]) -> db::mysql::MySqlNonResultBatchOutcome {
         let mut results = Vec::with_capacity(statements.len());
         for statement in statements {
             match self.execute_statement(statement).await {
-                Ok(mut statement_results) if statement_results.len() == 1 => results.append(&mut statement_results),
+                Ok(statement_results) if statement_results.len() == 1 => {
+                    results.push(statement_results.into_iter().next().expect("single MySQL batch result").result);
+                }
                 Ok(_) => {
                     return db::mysql::MySqlNonResultBatchOutcome {
                         results,
@@ -2349,19 +2412,23 @@ struct MysqlBatchConnection<'a> {
     query_timeout: Option<Duration>,
     bare: bool,
     max_rows: Option<usize>,
+    max_result_bytes: Option<usize>,
+    result_key_columns: &'a [String],
     dialect: db::mysql::MySqlQueryDialect,
 }
 
 impl MysqlBatchStatementExecutor for MysqlBatchConnection<'_> {
-    async fn execute_statement(&mut self, statement: &str) -> Result<Vec<db::QueryResult>, String> {
+    async fn execute_statement(&mut self, statement: &str) -> Result<Vec<db::mysql::MySqlQueryResult>, String> {
         wait_for_result_opt(
             self.cancel_token.clone(),
             self.query_timeout,
-            db::mysql::execute_query_results_on_conn_with_max_rows(
+            db::mysql::execute_query_results_on_conn_with_limits(
                 &mut *self.conn,
                 statement,
                 self.bare,
                 self.max_rows,
+                self.max_result_bytes,
+                self.result_key_columns,
                 self.dialect,
             ),
         )
@@ -2493,13 +2560,22 @@ where
         match executor.execute_statement(statement).await {
             Ok(statement_results) => {
                 if let Some(result) = statement_results.last() {
-                    report_execute_multi_progress(progress, statement_index, statements.len(), result, true, None);
+                    report_execute_multi_progress(
+                        progress,
+                        statement_index,
+                        statements.len(),
+                        &result.result,
+                        true,
+                        None,
+                    );
                 }
-                results.extend(
-                    statement_results
-                        .into_iter()
-                        .map(|result| ExecuteMultiResult::success_with_index(result, statement_index)),
-                );
+                results.extend(statement_results.into_iter().map(|result| {
+                    ExecuteMultiResult::success_with_index_and_large_values(
+                        result.result,
+                        statement_index,
+                        result.large_value_cells,
+                    )
+                }));
             }
             Err(err) => {
                 let action = mysql_batch_pool_error_action(db_type, &err);
@@ -2545,6 +2621,7 @@ async fn execute_multi_mysql(
     let operation_budget = operation_budget_for_pool_key(state, pool_key, query_timeout).await;
     let bare = mode == crate::connection::MysqlMode::Bare;
     let max_rows = options.max_rows;
+    let max_result_bytes = options.max_result_bytes.filter(|value| *value > 0);
     let pipeline_non_result_statements = !options.continue_on_error && mode == crate::connection::MysqlMode::Normal;
     let mut conn = match db::mysql::get_conn_with_health_check_with_cancel(
         pool,
@@ -2587,6 +2664,8 @@ async fn execute_multi_mysql(
         query_timeout,
         bare,
         max_rows,
+        max_result_bytes,
+        result_key_columns: &options.result_key_columns,
         dialect,
     };
     let (results, error_action) = execute_mysql_batch_statements(
@@ -4922,26 +5001,34 @@ for line in sys.stdin:
     }
 
     struct FakeMysqlBatchExecutor {
-        outcomes: std::collections::VecDeque<Result<Vec<db::QueryResult>, String>>,
+        outcomes: std::collections::VecDeque<Result<Vec<db::mysql::MySqlQueryResult>, String>>,
         executed: Vec<String>,
     }
 
     impl MysqlBatchStatementExecutor for FakeMysqlBatchExecutor {
-        async fn execute_statement(&mut self, statement: &str) -> Result<Vec<db::QueryResult>, String> {
+        async fn execute_statement(&mut self, statement: &str) -> Result<Vec<db::mysql::MySqlQueryResult>, String> {
             self.executed.push(statement.to_string());
             self.outcomes.pop_front().expect("test outcome for statement")
         }
     }
 
+    fn mysql_query_result(result: db::QueryResult) -> db::mysql::MySqlQueryResult {
+        db::mysql::MySqlQueryResult { result, large_value_cells: Vec::new() }
+    }
+
+    fn mysql_batch_result(result: db::QueryResult) -> Result<Vec<db::mysql::MySqlQueryResult>, String> {
+        Ok(vec![mysql_query_result(result)])
+    }
+
     struct FakePipelinedMysqlBatchExecutor {
         batch_outcomes: std::collections::VecDeque<db::mysql::MySqlNonResultBatchOutcome>,
-        statement_outcomes: std::collections::VecDeque<Result<Vec<db::QueryResult>, String>>,
+        statement_outcomes: std::collections::VecDeque<Result<Vec<db::mysql::MySqlQueryResult>, String>>,
         batches: Vec<Vec<String>>,
         statements: Vec<String>,
     }
 
     impl MysqlBatchStatementExecutor for FakePipelinedMysqlBatchExecutor {
-        async fn execute_statement(&mut self, statement: &str) -> Result<Vec<db::QueryResult>, String> {
+        async fn execute_statement(&mut self, statement: &str) -> Result<Vec<db::mysql::MySqlQueryResult>, String> {
             self.statements.push(statement.to_string());
             self.statement_outcomes.pop_front().expect("test outcome for single statement")
         }
@@ -4952,10 +5039,6 @@ for line in sys.stdin:
         }
     }
 
-    fn mysql_batch_result(result: db::QueryResult) -> Result<Vec<db::QueryResult>, String> {
-        Ok(vec![result])
-    }
-
     async fn assert_sqlite_batch_error_behavior(failure_first: bool, continue_on_error: bool) {
         let dir = std::env::temp_dir().join(format!("dbx-query-batch-error-{}", uuid::Uuid::new_v4()));
         std::fs::create_dir_all(&dir).unwrap();
@@ -5337,7 +5420,11 @@ for line in sys.stdin:
         };
         let mut executor = FakeMysqlBatchExecutor {
             outcomes: std::collections::VecDeque::from([
-                Ok(vec![result_set(1), result_set(2), result_set(3)]),
+                Ok(vec![
+                    mysql_query_result(result_set(1)),
+                    mysql_query_result(result_set(2)),
+                    mysql_query_result(result_set(3)),
+                ]),
                 mysql_batch_result(empty_query_result(1)),
             ]),
             executed: Vec::new(),
@@ -5497,6 +5584,7 @@ for line in sys.stdin:
         assert!(success.get("execution_error").is_none());
         assert!(success.get("statement_index").is_none());
         assert!(success.get("server_message").is_none());
+        assert!(success.get("large_value_cells").is_none());
 
         let mut error_column = empty_query_result(0);
         error_column.columns = vec!["Error".to_string()];
@@ -5526,6 +5614,21 @@ for line in sys.stdin:
         assert!(redacted.get("error").and_then(|value| value.get("detail")).is_none());
     }
 
+    #[test]
+    fn execute_multi_result_serializes_large_value_metadata_only_when_present() {
+        let result = ExecuteMultiResult::success_with_index_and_large_values(
+            empty_query_result(1),
+            0,
+            vec![db::LargeValueCell { row_index: 2, column_index: 3, original_bytes: 65_536 }],
+        );
+
+        let serialized = serde_json::to_value(result).unwrap();
+        assert_eq!(
+            serialized.get("large_value_cells"),
+            Some(&serde_json::json!([{"row_index": 2, "column_index": 3, "original_bytes": 65_536}]))
+        );
+    }
+
     #[test]
     fn sqlserver_batch_results_do_not_claim_statement_indexes() {
         assert_eq!(split_sql_batches("SELECT 1; SELECT 2;").len(), 1);

crates/dbx-core/src/types.rs

@@ -341,6 +341,15 @@ impl QueryMessage {
     }
 }
 
+/// A result cell whose full variable-length value was replaced by a bounded
+/// preview before the result crossed the desktop/web transport boundary.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct LargeValueCell {
+    pub row_index: usize,
+    pub column_index: usize,
+    pub original_bytes: usize,
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct QueryResult {
     pub columns: Vec<String>,

crates/dbx-web/src/routes/query.rs

@@ -21,6 +21,9 @@ pub struct ExecuteQueryRequest {
     pub max_rows: Option<usize>,
     pub fetch_size: Option<usize>,
     pub page_size: Option<usize>,
+    pub max_result_bytes: Option<usize>,
+    #[serde(default)]
+    pub result_key_columns: Vec<String>,
     pub result_session_id: Option<String>,
     pub client_session_id: Option<String>,
     pub timeout_secs: Option<u64>,
@@ -357,6 +360,8 @@ pub async fn execute_query(
             max_rows: req.max_rows,
             fetch_size: req.fetch_size,
             page_size: req.page_size,
+            max_result_bytes: req.max_result_bytes,
+            result_key_columns: req.result_key_columns,
             catalog: req.catalog,
             result_session_id: req.result_session_id,
             client_session_id: req.client_session_id,
@@ -403,6 +408,8 @@ pub async fn execute_multi(
             max_rows: req.max_rows,
             fetch_size: req.fetch_size,
             page_size: req.page_size,
+            max_result_bytes: req.max_result_bytes,
+            result_key_columns: req.result_key_columns,
             catalog: req.catalog,
             result_session_id: req.result_session_id,
             client_session_id: req.client_session_id,
@@ -1031,6 +1038,7 @@ mod tests {
                 elasticsearch_raw_body: None,
                 messages: Vec::new(),
             },
+            large_value_cells: Vec::new(),
             execution_error: true,
             statement_index: Some(1),
             error: Some(dbx_core::backend_error::BackendError::from_sql_detail(