Batch TDS row fetch to make mssql-odbc faster than msodbcsql18

mssql-odbc/.gitignore

@@ -12,6 +12,15 @@ tests/e2e/CMakeFiles/
 tests/e2e/cmake_install.cmake
 tests/e2e/_deps/
 
+# CMake perf benchmark build output and results
+tests/perf/build/
+tests/perf/build_*/
+tests/perf/results/
+tests/perf/CMakeCache.txt
+tests/perf/CMakeFiles/
+tests/perf/cmake_install.cmake
+tests/perf/_deps/
+
 # Compiled objects and libraries
 *.o
 *.obj

mssql-odbc/src/api/close_cursor.rs

@@ -132,6 +132,8 @@ fn sql_free_stmt_close_safe(statement_handle: SqlHandle, stmt: &StmtHandle) -> S
 pub(super) fn reset_cursor_state(stmt_state: &mut crate::handles::stmt::StmtState) {
     stmt_state.clear_state(STMT_STATE_CURSOR_OPEN | STMT_STATE_EXEC_CONTEXT);
     stmt_state.current_row = None;
+    stmt_state.reset_get_data_cursor();
+    stmt_state.discard_row_batch();
     stmt_state.column_metadata.clear();
     stmt_state.pending_row_counts.clear();
 }

mssql-odbc/src/api/driver_connect.rs

@@ -338,7 +338,7 @@ fn do_connect(
 
     let has_server_info = post_tds_info_messages(state, &info_messages);
 
-    state.client = Some(client);
+    state.client = Some(Box::new(client));
     state.connection_state = ConnectionState::Connected;
     debug!("SQLDriverConnectW: connected successfully");
 

mssql-odbc/src/api/exec_common.rs

@@ -40,7 +40,7 @@ pub(super) fn claim_connection(
     stmt: &StmtHandle,
     statement_handle: SqlHandle,
     op: &str,
-) -> Result<TdsClient, SqlReturn> {
+) -> Result<Box<TdsClient>, SqlReturn> {
     let Ok(mut dbc_state) = dbc.inner.lock() else {
         error!("{op}: dbc mutex poisoned");
         clear_exec_started(stmt);
@@ -88,7 +88,11 @@ pub(super) fn claim_connection(
 
 /// Returns `client` to the DBC and releases the busy claim. Used on the
 /// DDL/DML success path and on error recovery.
-pub(super) fn return_client_idle(dbc: &DbcHandle, statement_handle: SqlHandle, client: TdsClient) {
+pub(super) fn return_client_idle(
+    dbc: &DbcHandle,
+    statement_handle: SqlHandle,
+    client: Box<TdsClient>,
+) {
     if let Ok(mut dbc_state) = dbc.inner.lock() {
         dbc_state.client = Some(client);
         if dbc_state.active_stmt == Some(statement_handle) {
@@ -109,7 +113,7 @@ pub(super) fn return_client_idle(dbc: &DbcHandle, statement_handle: SqlHandle, c
 pub(super) fn try_claim_idle_client(
     dbc: &DbcHandle,
     statement_handle: SqlHandle,
-) -> Option<TdsClient> {
+) -> Option<Box<TdsClient>> {
     let Ok(mut dbc_state) = dbc.inner.lock() else {
         return None;
     };
@@ -123,7 +127,7 @@ pub(super) fn try_claim_idle_client(
 
 /// Returns `client` to the DBC but **keeps** the busy claim — used when a
 /// cursor is left open for `SQLFetch`.
-pub(super) fn return_client_busy(dbc: &DbcHandle, client: TdsClient) {
+pub(super) fn return_client_busy(dbc: &DbcHandle, client: Box<TdsClient>) {
     if let Ok(mut dbc_state) = dbc.inner.lock() {
         dbc_state.client = Some(client);
     }
@@ -136,7 +140,7 @@ pub(super) fn fail_with_tds(
     dbc: &DbcHandle,
     stmt: &StmtHandle,
     statement_handle: SqlHandle,
-    mut client: TdsClient,
+    mut client: Box<TdsClient>,
     err: &TdsError,
 ) -> SqlReturn {
     let info_messages = client.take_info_messages();
@@ -256,7 +260,7 @@ pub(super) fn finish_execute(
     dbc: &DbcHandle,
     stmt: &StmtHandle,
     statement_handle: SqlHandle,
-    mut client: TdsClient,
+    mut client: Box<TdsClient>,
     op: &str,
 ) -> SqlReturn {
     let metadata = client.get_metadata().clone();

mssql-odbc/src/api/exec_direct.rs

@@ -107,6 +107,8 @@ fn sql_exec_direct_w_safe(
         stmt_state.clear_state(STMT_STATE_EXEC_CONTEXT);
         stmt_state.column_metadata.clear();
         stmt_state.current_row = None;
+        stmt_state.reset_get_data_cursor();
+        stmt_state.discard_row_batch();
         stmt_state.row_count = -1;
         stmt_state.pending_row_counts.clear();
         stmt_state.prepared_sql = None;
@@ -285,7 +287,7 @@ mod tests {
         ]);
         {
             let mut ds = dbc.inner.lock().unwrap();
-            ds.client = Some(client);
+            ds.client = Some(Box::new(client));
             // active_stmt stays None => connection idle and claimable.
         }
 
@@ -325,7 +327,7 @@ mod tests {
         ]);
         {
             let mut ds = dbc.inner.lock().unwrap();
-            ds.client = Some(client);
+            ds.client = Some(Box::new(client));
         }
 
         let stmt = unsafe { handle_from_raw::<StmtHandle>(h.stmt) };

mssql-odbc/src/api/execute.rs

@@ -159,6 +159,8 @@ fn stage_execution(stmt: &StmtHandle) -> Result<Execution, SqlReturn> {
     stmt_state.clear_state(STMT_STATE_EXEC_CONTEXT);
     stmt_state.column_metadata.clear();
     stmt_state.current_row = None;
+    stmt_state.reset_get_data_cursor();
+    stmt_state.discard_row_batch();
     stmt_state.row_count = -1;
     stmt_state.pending_row_counts.clear();
     stmt_state.set_state(STMT_STATE_EXEC_STARTED);

mssql-odbc/src/api/fetch.rs

@@ -13,7 +13,6 @@ use crate::api::odbc_types::{
 use crate::error::free_errors;
 use crate::handles::stmt::STMT_STATE_CURSOR_OPEN;
 use crate::handles::{HandleType, StmtHandle, handle_from_raw};
-use mssql_tds::connection::tds_client::ResultSet;
 
 /// Implements SQLFetch for the current forward-only result set.
 ///
@@ -54,8 +53,35 @@ fn sql_fetch_safe(statement_handle: SqlHandle, stmt: &StmtHandle) -> SqlReturn {
     fetch_rows_next(statement_handle, stmt)
 }
 
+/// Rows decoded per wire round through the async decode path. The per-row cost
+/// of a cursor is dominated by building and polling that state machine, so
+/// draining a batch amortizes it; the cap bounds the memory a single fetch can
+/// buffer for wide rows.
+const FETCH_BATCH_ROWS: usize = 64;
+
 /// Row materialization step for one forward fetch operation.
 fn fetch_rows_next(statement_handle: SqlHandle, stmt: &StmtHandle) -> SqlReturn {
+    // Fast path: a previous fetch already decoded this row. The batch is only
+    // ever non-empty while this statement holds the connection, so the busy and
+    // cursor checks below are already satisfied and the DBC lock is not needed.
+    //
+    // Server INFO is not drained here. It cannot be attributed to a specific
+    // row anyway, and it stays on the client's buffer until the next call that
+    // can report it (see the SQL_NO_DATA arm below).
+    {
+        let Ok(mut stmt_state) = stmt.inner.lock() else {
+            error!("SQLFetch: stmt mutex poisoned serving prefetched row");
+            return SQL_ERROR;
+        };
+        if let Some(row) = stmt_state.row_batch.pop_front() {
+            if let Some(previous) = stmt_state.current_row.replace(row) {
+                stmt_state.row_batch_spare.push(previous);
+            }
+            stmt_state.reset_get_data_cursor();
+            return SQL_SUCCESS;
+        }
+    }
+
     let dbc = stmt.parent_dbc();
 
     let mut client = {
@@ -130,10 +156,23 @@ fn fetch_rows_next(statement_handle: SqlHandle, stmt: &StmtHandle) -> SqlReturn
         }
     }
 
-    let fetch_result = dbc.runtime.block_on(client.next_row());
+    let spare = match stmt.inner.lock() {
+        Ok(mut ss) => {
+            let mut spare = std::mem::take(&mut ss.row_batch_spare);
+            if let Some(previous) = ss.current_row.take() {
+                spare.push(previous);
+            }
+            spare
+        }
+        Err(_) => Vec::new(),
+    };
+    let mut rows = Vec::new();
+    let fetch_result =
+        dbc.runtime
+            .block_on(client.fetch_rows_batch(&mut rows, spare, FETCH_BATCH_ROWS));
 
     match fetch_result {
-        Ok(Some(row)) => {
+        Ok(count) if count > 0 => {
             let Ok(mut stmt_state) = stmt.inner.lock() else {
                 error!("SQLFetch: stmt mutex poisoned storing row");
                 if let Ok(mut ds) = dbc.inner.lock() {
@@ -144,7 +183,10 @@ fn fetch_rows_next(statement_handle: SqlHandle, stmt: &StmtHandle) -> SqlReturn
                 }
                 return SQL_ERROR;
             };
-            stmt_state.current_row = Some(row);
+            let mut rows = rows.into_iter();
+            stmt_state.current_row = rows.next();
+            stmt_state.row_batch.extend(rows);
+            stmt_state.reset_get_data_cursor();
             // Drain INFO only after the lock is held so a poisoned mutex cannot
             // silently drop the messages.
             let info_messages = client.take_info_messages();
@@ -163,7 +205,7 @@ fn fetch_rows_next(statement_handle: SqlHandle, stmt: &StmtHandle) -> SqlReturn
                 SQL_SUCCESS
             }
         }
-        Ok(None) => {
+        Ok(_) => {
             // End of current rowset. SQLFetch must return SQL_NO_DATA here per the
             // cursor contract, and SQL_NO_DATA cannot be upgraded to
             // SQL_SUCCESS_WITH_INFO — so this call has no way to signal "there are
@@ -201,6 +243,7 @@ fn fetch_rows_next(statement_handle: SqlHandle, stmt: &StmtHandle) -> SqlReturn
                 return SQL_ERROR;
             };
             stmt_state.current_row = None;
+            stmt_state.reset_get_data_cursor();
             // Don't clear CURSOR_OPEN here: the cursor stays open until
             // SQLMoreResults / SQLCloseCursor / SQLFreeStmt(SQL_CLOSE).
             drop(stmt_state);
@@ -215,6 +258,7 @@ fn fetch_rows_next(statement_handle: SqlHandle, stmt: &StmtHandle) -> SqlReturn
             error!(%e, "SQLFetch: row fetch failed");
             if let Ok(mut stmt_state) = stmt.inner.lock() {
                 stmt_state.current_row = None;
+                stmt_state.reset_get_data_cursor();
                 stmt_state.clear_state(STMT_STATE_CURSOR_OPEN);
                 post_tds_error(&mut stmt_state, &e, SQLSTATE_HY000);
                 let info_messages = client.take_info_messages();
@@ -333,7 +377,7 @@ mod tests {
         let dbc_handle = unsafe { handle_from_raw::<DbcHandle>(h.dbc) };
         {
             let mut dbc_state = dbc_handle.inner.lock().unwrap();
-            dbc_state.client = Some(client);
+            dbc_state.client = Some(Box::new(client));
             dbc_state.active_stmt = Some(h.stmt);
         }