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

[Saurabh Singh (saurabh500)]

Description

Makes the mssql-odbc fetch path faster than the native msodbcsql18 driver.

The row path was constructing and polling a fresh async state machine per row. Because #[async_trait] boxes every trait method call, both the size and the frequency of that construction cost real time per row. Every SQLFetch paid a boxed trait future, a cancellation wrapper, a timeout wrapper, two Instant::now() calls, several Arc clones and a full ODBC mutex/DBC handoff — for one row.

Changes

Batched row prefetch (the structural fix, worth more than every micro-optimization combined):

• New TdsTokenStreamReader::receive_rows_into with a default impl plus a NetworkTransport override, so the batch loop lives below the boxed trait call instead of above it.
• New TdsClient::fetch_rows_batch, which resolves metadata, parser context and decryptor once per batch rather than once per row.
• DefaultRowWriter gains a batch mode that queues completed rows and recycles row buffers from a caller-supplied spare pool.
• SQLFetch prefetches 64 rows and serves subsequent calls from a lock-only fast path. The batch is invalidated everywhere the cursor is invalidated (SQLMoreResults, SQLCloseCursor, re-execute, etc.).

Per-column allocation — every variable-length column allocated a Vec<u8> and freed it on recycle:

• New RowWriter::take_string_buffer lets the decoder source its byte buffer from the writer.
• DefaultRowWriter harvests byte buffers out of recycled rows into a pool, so a steady-state batch fetch reuses allocations instead of churning one per column.
• New RowWriter::may_pause lets writers that never pause skip the per-column pause_after_column dispatch.

Future size — a large stack local inside an async fn is baked into the future for every call, even on branches that never execute:

• Replaced a [0u8; 8192] stack local in a rare branch of get_next_row_into (worth 20% on its own).
• Boxed DbcState::client and several cold decoder / token-stream arms.
• Added zero-allocation fixed-width packet reads that serve straight from the buffer without constructing a future.

SQLGetData:

• Fixed quadratic varchar(max) chunked reads via an offset cursor + payload cache.
• Added ASCII fast paths that skip transcoding and the intermediate payload allocation.

Benchmark harness:

• Added BM_Fetch_RowsOnly, which separates SQLFetch cost from SQLGetData cost. This is what made the diagnosis possible and is worth keeping.

Results

Median, 10k rows, 7–9 reps, local SQL Server 2022:

Benchmark	Before	After	msodbcsql18	Ratio
BM_Fetch_NarrowRows/10000	17.34 ms	4.49 ms	4.68 ms	0.96× (win)
BM_Fetch_RowsOnly/10000	—	3.18 ms	3.20 ms	0.99× (win)
BM_Fetch_WideRows/10000	49.8 ms	42.2 ms	40.6 ms	1.04×
BM_Type_VarcharMax	~1049× slower	18.9 ms	21.0 ms	0.90× (win)
BM_Type_NVarchar	—	13.9 ms	14.5 ms	0.96× (win)
BM_Connect_Disconnect	—	2.58 ms	10.6 ms	0.24× (win)
BM_AllocFree_Stmt	—	1.53 µs	1.85 µs	0.83× (win)

Narrow-row fetch is 3.9× faster than baseline and now beats the native driver.

Not addressed here

• BM_Fetch_WideRows remains 1.04× slower. Closing it needs decode_into's per-column match split so each arm's future is independent — a large refactor with real regression risk, better as a follow-up.
• The execute path (BM_ExecDirect_SelectOne, ~1.8×) is untouched and still under investigation.

Notes for reviewers

Behavioral details worth a look:

• The SQLFetch fast path deliberately does not drain server INFO messages — they can't be attributed to a specific row, so they stay buffered until a call that can report them.
• Prefetching consumes the result set's DONE token up to 63 fetches earlier than before. This is safe because the client is held exclusively by the statement while a cursor is open, and maybe_has_unread_rows() gates the observable SQL_NO_DATA behavior identically.
• The whole batch runs under one remaining_request_timeout rather than one per row.

Related Issues

Fixes #185

Checklist

• cargo bfmt passes
• cargo bclippy passes (plus scripts/bclippy.ps1 for mssql-py-core)
• cargo btest passes — 2148/2155 unit tests. The 7 failures are pre-existing missing tests/test_certificates/ fixtures, verified identical on a stashed baseline.
• New/changed functionality has tests — batch-mode DefaultRowWriter, string-buffer recycling, plus previously untested format_date / format_money / money mixed-endian reassembly.
• Public API changes are documented — RowWriter::take_string_buffer, RowWriter::may_pause, TdsClient::fetch_rows_batch, SqlString::into_bytes.

[Copilot]

Pull request overview

Optimizes ODBC row retrieval through batched TDS decoding, buffer reuse, faster SQLGetData, and a cross-platform performance harness.

Changes:

• Prefetches rows in batches and recycles row/string buffers.
• Adds fixed-width packet-read and SQLGetData fast paths.
• Adds ODBC benchmarks and comparison runners.

Reviewed changes

Copilot reviewed 31 out of 31 changed files in this pull request and generated 8 comments.

Show a summary per file

File	Description
mssql-tds/src/io/token_stream.rs	Adds batched row decoding.
mssql-tds/src/io/packet_reader.rs	Adds fixed-width read fast paths.
mssql-tds/src/datatypes/sqldatatypes.rs	Uses fast scalar reads.
mssql-tds/src/datatypes/sql_string.rs	Exposes backing byte buffers.
mssql-tds/src/datatypes/row_writer.rs	Adds batching and buffer recycling.
mssql-tds/src/datatypes/decoder.rs	Optimizes decoding and allocations.
mssql-tds/src/connection/transport/network_transport.rs	Implements transport batching.
mssql-tds/src/connection/tds_client.rs	Adds batch-fetch APIs and context caching.
mssql-odbc/src/handles/stmt.rs	Stores prefetched rows and GetData state.
mssql-odbc/src/handles/dbc.rs	Boxes the TDS client.
mssql-odbc/src/api/fetch.rs	Serves batched prefetched rows.
mssql-odbc/src/api/get_data.rs	Adds chunk cursors and conversion fast paths.
mssql-odbc/src/api/prepare.rs	Invalidates prefetched state.
mssql-odbc/src/api/more_results.rs	Resets batch state between results.
mssql-odbc/src/api/get_type_info.rs	Clears batch state.
mssql-odbc/src/api/execute.rs	Resets state before execution.
mssql-odbc/src/api/exec_direct.rs	Resets state before direct execution.
mssql-odbc/src/api/exec_common.rs	Updates boxed-client ownership.
mssql-odbc/src/api/driver_connect.rs	Stores the boxed client.
mssql-odbc/src/api/close_cursor.rs	Discards cursor batch state.
mssql-odbc/tests/perf/CMakeLists.txt	Configures benchmark targets.
mssql-odbc/tests/perf/README.md	Documents benchmark usage.
mssql-odbc/tests/perf/run_perf.sh	Adds Unix benchmark runner.
mssql-odbc/tests/perf/run_perf.ps1	Adds Windows benchmark runner.
mssql-odbc/tests/perf/include/perf_fixture.h	Declares benchmark utilities.
mssql-odbc/tests/perf/lib/perf_fixture.cpp	Implements benchmark plumbing.
mssql-odbc/tests/perf/benches/fetch_bench.cpp	Benchmarks row retrieval.
mssql-odbc/tests/perf/benches/exec_bench.cpp	Benchmarks execution paths.
mssql-odbc/tests/perf/benches/datatype_bench.cpp	Benchmarks datatype retrieval.
mssql-odbc/tests/perf/benches/connect_bench.cpp	Benchmarks connection and handles.
mssql-odbc/.gitignore	Ignores benchmark artifacts.

Suppressed comments (1)

mssql-odbc/tests/perf/lib/perf_fixture.cpp:237

• Reaching kMaxChunksPerColumn with SQL_SUCCESS_WITH_INFO falls out of the loop and is treated as success, despite the stated guard against a driver that never terminates truncation. Detect that final status and fail the benchmark instead of advancing to the next column with unread data.

            for (int chunk = 0; chunk < kMaxChunksPerColumn; chunk++) {

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[Saurabh Singh (saurabh500)]

ExecDirect / SELECT 1 profiling — findings

Follow-up investigation into BM_ExecDirect_SelectOne, where mssql-odbc is
~1.8x slower than msodbcsql18. Conclusion: this path is I/O-bound, not
CPU-bound, and there is no surgical fix in the TDS layer. Recording the data
here so the next person does not repeat the work.

Benchmark split

BM_ExecDirect_NoDrain (added in this PR) separates execute from drain:

Benchmark	mssql-odbc	msodbcsql18
BM_ExecDirect_SelectOne	230-238 us	126-143 us
BM_ExecDirect_NoDrain	189 us	133 us
BM_PreparedExecute	168 us	120 us

Phase attribution

A temporary Rust harness (since removed) timed each phase of SELECT 1,
p50 over 2800 samples after warmup:

prologue (begin_command + reconnect check + reset_reader)    0.1 us
fill     (ALL_HEADERS + UCS-2 SQL into the packet buffer)    1.7 us
flush    (finalize -> socket write)                         20.1 us
boundary (advance_to_result_boundary)                      148.3 us
rows     (fetch_rows_batch)                                  3.6 us
                                                          ---------
total                                                      176.2 us

Instrumenting the transport showed the wire behaviour is already optimal:

• Exactly one write and one read syscall per query.
• The server's whole response is a single 37-byte TDS packet
  (type=4 status=1 len=37), consumed in one read.
• Token parsing (dispatch_token) is ~7 us; row decode is ~4 us.

So driver CPU is roughly 10-15 us. Everything else is time blocked in the
socket read waiting for SQL Server.

Why the socket dominates

A standalone tokio-vs-blocking ping-pong (60-byte request, 37-byte reply,
loopback, TCP_NODELAY, blocking echo server on its own thread) on the
benchmark machine:

	p50
tokio write	36.0 us
tokio read	48.7 us
tokio round trip	86.6 us
std write	37.8 us
std read	33.2 us
std (blocking) round trip	76.4 us

Loopback socket ops on this box cost tens of microseconds even with plain
blocking sockets, and tokio's reactor adds ~10-15 us per round trip on top.
That difference accounts for the bulk of the remaining SELECT 1 gap:
subtracting the socket floor leaves both drivers at roughly the same amount of
SQL Server think time.

Also ruled out by measurement:

• Not TLS. PreferOff moved flush only 23.7 -> 20.3 us.
• Not a reactor wake-up on the write. Poll counting showed exactly one
  poll per send, i.e. the write never returns Poll::Pending.
• Not extra round trips or extra framing. One packet, one read.
• Spinning on try_read before awaiting does not work — tokio readiness is
  reactor-driven, so a spin loop that never yields will never observe the
  socket become readable and simply burns CPU until it gives up.

What would actually move the needle

Only an architectural change to the I/O strategy: bypassing the tokio reactor
for the request/response hot path (blocking or direct-IOCP sockets on a
dedicated thread). That is a large, risky change well outside the scope of this
PR, and it trades CPU and scalability for latency, so it needs a separate
design discussion.

[github-actions]

📊 Code Coverage Report

🔥 Diff Coverage 76%

🎯 Overall Coverage 90.8%

📦 Project: mssql-tds + mssql-odbc + mssql-py-core ℹ️ Note: diff coverage is reported, not enforced.

---

Diff Coverage

Diff: main...HEAD, staged and unstaged changes

• mssql-odbc/src/api/close_cursor.rs (100%)
• mssql-odbc/src/api/driver_connect.rs (100%)
• mssql-odbc/src/api/exec_common.rs (100%)
• mssql-odbc/src/api/exec_direct.rs (100%)
• mssql-odbc/src/api/execute.rs (100%)
• mssql-odbc/src/api/fetch.rs (86.7%): Missing lines 73-74,167,261
• mssql-odbc/src/api/get_data.rs (67.2%): Missing lines 148,150,152-160,196-197,205,210,217-218,268-270,275,320-329,358-360,362,406,415,453-463,465,467-473,476-483,485-488,490-493,495-497,499,501,503,534-536
• mssql-odbc/src/api/get_type_info.rs (100%)
• mssql-odbc/src/api/more_results.rs (100%)
• mssql-odbc/src/api/prepare.rs (100%)
• mssql-odbc/src/handles/stmt.rs (100%)
• mssql-tds/src/connection/tds_client.rs (60.5%): Missing lines 2115-2128,2131-2137,2183-2185,2188-2190,2201,2220,2225-2226,2228-2230,2249-2259
• mssql-tds/src/connection/transport/network_transport.rs (100%)
• mssql-tds/src/datatypes/decoder.rs (90.6%): Missing lines 806,1295-1298,1315-1318
• mssql-tds/src/datatypes/row_writer.rs (92.4%): Missing lines 157-165
• mssql-tds/src/datatypes/sql_string.rs (100%)
• mssql-tds/src/datatypes/sqldatatypes.rs (81.2%): Missing lines 802-803,1035
• mssql-tds/src/io/packet_reader.rs (51.4%): Missing lines 332-346,587-589
• mssql-tds/src/io/token_stream.rs (83.7%): Missing lines 156,179,551,620-623

Summary

• Total: 762 lines
• Missing: 176 lines
• Coverage: 76%

mssql-odbc/src/api/fetch.rs

  69     // row anyway, and it stays on the client's buffer until the next call that
  70     // can report it (see the SQL_NO_DATA arm below).
  71     {
  72         let Ok(mut stmt_state) = stmt.inner.lock() else {
! 73             error!("SQLFetch: stmt mutex poisoned serving prefetched row");
! 74             return SQL_ERROR;
  75         };
  76         if let Some(row) = stmt_state.row_batch.pop_front() {
  77             if let Some(previous) = stmt_state.current_row.replace(row) {
  78                 stmt_state.row_batch_spare.push(previous);

---

  163                 spare.push(previous);
  164             }
  165             spare
  166         }
! 167         Err(_) => Vec::new(),
  168     };
  169     let mut rows = Vec::new();
  170     let fetch_result =
  171         dbc.runtime

---

  257         Err(e) => {
  258             error!(%e, "SQLFetch: row fetch failed");
  259             if let Ok(mut stmt_state) = stmt.inner.lock() {
  260                 stmt_state.current_row = None;
! 261                 stmt_state.reset_get_data_cursor();
  262                 stmt_state.clear_state(STMT_STATE_CURSOR_OPEN);
  263                 post_tds_error(&mut stmt_state, &e, SQLSTATE_HY000);
  264                 let info_messages = client.take_info_messages();
  265                 post_tds_info_messages(&mut stmt_state, &info_messages);

---

mssql-odbc/src/api/get_data.rs

  144     // Continue an in-progress chunked read of this column, if any. A different
  145     // column or a different target type restarts the read.
  146     match stmt_state.get_data_cursor.take() {
  147         Some(cursor) if cursor.column == column_number && cursor.wide == wide => {
! 148             let Some(payload) = cursor.payload else {
  149                 // The value was already delivered in full.
! 150                 return SQL_NO_DATA;
  151             };
! 152             return continue_chunked_read(
! 153                 &mut stmt_state,
! 154                 column_number,
! 155                 wide,
! 156                 payload,
! 157                 cursor.offset,
! 158                 target_value_ptr,
! 159                 buf_elements,
! 160                 strlen_or_ind_ptr,
  161             );
  162         }
  163         _ => {}
  164     }

---

  192     // Convert the cell before touching `stmt_state` mutably, so the borrow of
  193     // the current row ends here.
  194     let converted = {
  195         let Some(row) = stmt_state.current_row.as_ref() else {
! 196             post_sql_error(&mut stmt_state, SQLSTATE_24000, 0, "No current row");
! 197             return SQL_ERROR;
  198         };
  199         let value = &row[col_index - 1];
  200         if matches!(value, ColumnValues::Null) {
  201             Converted::Null

---

  201             Converted::Null
  202         } else if wide {
  203             match column_value_to_utf16(value) {
  204                 Some(v) => Converted::Wide(v),
! 205                 None => Converted::Unsupported,
  206             }
  207         } else {
  208             match column_value_to_utf8_bytes(value) {
  209                 Some(v) => Converted::Narrow(v),
! 210                 None => Converted::Unsupported,
  211             }
  212         }
  213     };

---

  213     };
  214 
  215     match converted {
  216         Converted::Unsupported => {
! 217             post_unsupported_conversion(&mut stmt_state);
! 218             SQL_ERROR
  219         }
  220         Converted::Null => {
  221             unsafe { write_if_some(strlen_or_ind_ptr, SQL_NULL_DATA) };
  222             // Write a NUL terminator into the caller buffer when there's room.

---

  264     /// The column type has no text conversion yet.
  265     Unsupported,
  266 }
  267 
! 268 fn post_unsupported_conversion(stmt_state: &mut crate::handles::stmt::StmtState) {
! 269     post_sql_error(
! 270         stmt_state,
  271         SQLSTATE_HYC00,
  272         0,
  273         "Column type conversion not yet implemented",
  274     );
! 275 }
  276 
  277 /// Writes the first chunk of a freshly converted value and records how much of
  278 /// it remains, so a truncated value can be resumed by the next call.
  279 ///

---

  316 
  317 /// Resumes a chunked read from `offset` using the already-converted payload, so
  318 /// a long value is converted once rather than once per chunk.
  319 #[allow(clippy::too_many_arguments)]
! 320 fn continue_chunked_read(
! 321     stmt_state: &mut crate::handles::stmt::StmtState,
! 322     column_number: SqlUSmallInt,
! 323     wide: bool,
! 324     payload: GetDataPayload,
! 325     offset: usize,
! 326     target_value_ptr: SqlPointer,
! 327     buf_elements: usize,
! 328     strlen_or_ind_ptr: *mut SqlLen,
! 329 ) -> SqlReturn {
  330     macro_rules! resume {
  331         ($buf:expr, $ptr_ty:ty, $wrap:expr) => {{
  332             let outcome = write_chunk(
  333                 stmt_state,

---

  354             }
  355         }};
  356     }
  357 
! 358     match payload {
! 359         GetDataPayload::Narrow(buf) => resume!(buf, u8, GetDataPayload::Narrow),
! 360         GetDataPayload::Wide(buf) => resume!(buf, SqlWChar, GetDataPayload::Wide),
  361     }
! 362 }
  363 
  364 /// Result of copying one chunk into the application buffer.
  365 enum ChunkOutcome {
  366     /// The remaining value fit entirely.

---

  402 fn column_value_to_utf8_bytes(v: &ColumnValues) -> Option<Vec<u8>> {
  403     match v {
  404         // Already UTF-8 on the wire: hand back the bytes without transcoding.
  405         ColumnValues::String(s) if matches!(s.encoding_type(), EncodingType::Utf8) => {
! 406             Some(s.bytes.clone())
  407         }
  408         // Every single-byte SQL Server code page agrees with US-ASCII below
  409         // 0x80, so all-ASCII payloads are already their own UTF-8 encoding.
  410         // The check is vectorized and skips a full code-page transcode, which

---

  411         // is the common case for `varchar` columns.
  412         ColumnValues::String(s)
  413             if matches!(s.encoding_type(), EncodingType::LcidBased(_)) && s.bytes.is_ascii() =>
  414         {
! 415             Some(s.bytes.clone())
  416         }
  417         _ => column_value_to_text(v).map(String::into_bytes),
  418     }
  419 }

---

  449         ColumnValues::Bit(x) => Some(if *x { "1".into() } else { "0".into() }),
  450         ColumnValues::String(s) => Some(s.to_utf8_string()),
  451         ColumnValues::Uuid(u) => Some(u.to_string()),
  452         ColumnValues::Null => Some(String::new()),
! 453         ColumnValues::Decimal(d) | ColumnValues::Numeric(d) => Some(d.to_string()),
! 454         ColumnValues::Date(d) => Some(format_date(d.get_days())),
! 455         ColumnValues::Time(t) => Some(format_time(t)),
! 456         ColumnValues::DateTime2(dt) => Some(format!(
! 457             "{} {}",
! 458             format_date(dt.days),
! 459             format_time(&dt.time)
! 460         )),
! 461         ColumnValues::DateTimeOffset(dto) => {
! 462             let (sign, mins) = if dto.offset < 0 {
! 463                 ('-', (-(dto.offset as i32)) as u32)
  464             } else {
! 465                 ('+', dto.offset as u32)
  466             };
! 467             Some(format!(
! 468                 "{} {} {sign}{:02}:{:02}",
! 469                 format_date(dto.datetime2.days),
! 470                 format_time(&dto.datetime2.time),
! 471                 mins / 60,
! 472                 mins % 60
! 473             ))
  474         }
  475         // `datetime` counts days from 1900-01-01 in 1/300-second ticks.
! 476         ColumnValues::DateTime(dt) => {
! 477             let days = DAYS_0001_TO_1900 as i64 + dt.days as i64;
! 478             let nanos = (dt.time as u64) * 10_000_000 / 3;
! 479             Some(format!(
! 480                 "{} {}",
! 481                 format_date(days.clamp(0, u32::MAX as i64) as u32),
! 482                 format_nanos(nanos, 3)
! 483             ))
  484         }
! 485         ColumnValues::SmallDateTime(dt) => {
! 486             let days = DAYS_0001_TO_1900 + u32::from(dt.days);
! 487             let nanos = u64::from(dt.time) * 60 * 1_000_000_000;
! 488             Some(format!("{} {}", format_date(days), format_nanos(nanos, 0)))
  489         }
! 490         ColumnValues::SmallMoney(m) => Some(format_money(i64::from(m.int_val))),
! 491         ColumnValues::Money(m) => {
! 492             let scaled = (i64::from(m.msb_part) << 32) | (i64::from(m.lsb_part) & 0xFFFF_FFFF);
! 493             Some(format_money(scaled))
  494         }
! 495         ColumnValues::Bytes(b) => {
! 496             let mut s = String::with_capacity(b.len() * 2);
! 497             for byte in b {
  498                 use std::fmt::Write;
! 499                 let _ = write!(s, "{byte:02X}");
  500             }
! 501             Some(s)
  502         }
! 503         ColumnValues::Xml(x) => Some(x.as_string()),
  504         _ => None,
  505     }
  506 }

---

  530     format!("{y:04}-{m:02}-{d:02}")
  531 }
  532 
  533 /// Formats a [`SqlTime`] as `HH:MM:SS[.fffffff]`, honouring its scale.
! 534 fn format_time(t: &SqlTime) -> String {
! 535     format_nanos(t.time_nanoseconds, t.scale)
! 536 }
  537 
  538 /// Formats nanoseconds since midnight as `HH:MM:SS[.fffffff]`, emitting
  539 /// `scale` fractional digits (none when `scale` is 0).
  540 fn format_nanos(nanos: u64, scale: u8) -> String {

---

mssql-tds/src/connection/tds_client.rs

  2111     /// cleared. Callers driving a cursor row-by-row (ODBC `SQLFetch`) should
  2112     /// prefer this over [`ResultSet::next_row`]: it avoids both the per-row
  2113     /// `Vec` allocation and the boxed future that `#[async_trait]` introduces
  2114     /// for the trait method.
! 2115     pub async fn fetch_next_row_into_vec(
! 2116         &mut self,
! 2117         buffer: &mut Vec<ColumnValues>,
! 2118     ) -> TdsResult<bool> {
! 2119         if !self.maybe_has_unread_rows() {
! 2120             buffer.clear();
! 2121             return Ok(false);
! 2122         }
! 2123         let col_count = self
! 2124             .current_metadata
! 2125             .as_ref()
! 2126             .map(|m| m.columns.len())
! 2127             .unwrap_or(0);
! 2128         let mut writer = DefaultRowWriter::with_buffer(std::mem::take(buffer), col_count);
  2129         // Boxed deliberately: measurably faster than letting this future be
  2130         // materialized inline in the caller's frame on every fetch.
! 2131         let has_row = Box::pin(self.get_next_row_into(&mut writer)).await?;
! 2132         *buffer = writer.take_row();
! 2133         if !has_row {
! 2134             buffer.clear();
! 2135         }
! 2136         Ok(has_row)
! 2137     }
  2138 
  2139     /// Fetches up to `max_rows` rows in a single decode call, appending them to
  2140     /// `out`.
  2141     ///

---

  2179         writer: &mut DefaultRowWriter,
  2180         max_rows: usize,
  2181     ) -> TdsResult<usize> {
  2182         if self.current_metadata.is_none() {
! 2183             return Err(UsageError(
! 2184                 "No metadata found while fetching the next row. Have you called the execute method or was the query supposed to return resultset?".to_string(),
! 2185             ));
  2186         }
  2187         if !matches!(self.active_row_read_state, ActiveRowReadState::Idle) {
! 2188             return Err(crate::error::Error::ImplementationError(
! 2189                 "Batched fetch cannot start while a row read is paused".to_string(),
! 2190             ));
  2191         }
  2192 
  2193         let metadata = Arc::clone(self.current_metadata.as_ref().unwrap());
  2194         let cached = matches!(

---

  2197         );
  2198         if !cached {
  2199             let decryptor = self.resolve_cell_decryptor(&metadata).await?;
  2200             self.current_parser_context = Some(ParserContext::ColumnMetadata(metadata, decryptor));
! 2201         }
  2202         let parser_context = self.current_parser_context.clone().unwrap();
  2203 
  2204         let mut count = 0;
  2205         while count < max_rows {

---

  2216                     max_rows - count,
  2217                 )
  2218                 .await?;
  2219             if let Some(start) = start {
! 2220                 self.update_remaining_timeout(start);
  2221             }
  2222             count += batch.rows;
  2223 
  2224             match batch.stopped_at {
! 2225                 None => break,
! 2226                 Some(RowReadResult::RowWritten) => unreachable!("batch consumes written rows"),
  2227                 Some(RowReadResult::RowPaused(_) | RowReadResult::PlpPaused(_)) => {
! 2228                     return Err(crate::error::Error::ImplementationError(
! 2229                         "Row decode paused against a non-pausing batch writer".to_string(),
! 2230                     ));
  2231                 }
  2232                 Some(RowReadResult::Token(token)) => {
  2233                     // Boxed: non-row tokens appear once per result set, but the
  2234                     // handler inlines every token parser's state machine.

---

  2245     ///
  2246     /// Exposed so perf work can verify that changes actually shrink the
  2247     /// per-row state machines rather than just moving allocations around.
  2248     #[doc(hidden)]
! 2249     pub fn row_future_sizes(&mut self) -> (usize, usize) {
! 2250         let mut writer = DefaultRowWriter::new(0);
! 2251         let inner = self.get_next_row_into(&mut writer);
! 2252         let size_inner = std::mem::size_of_val(&inner);
! 2253         drop(inner);
! 2254         let mut buf = Vec::new();
! 2255         let outer = self.fetch_next_row_into_vec(&mut buf);
! 2256         let size_outer = std::mem::size_of_val(&outer);
! 2257         drop(outer);
! 2258         (size_inner, size_outer)
! 2259     }
  2260 
  2261     /// Returns `true` when transparent parameter encryption should be attempted:
  2262     /// the connection requested Always Encrypted and the server acknowledged the
  2263     /// feature during login.

---

mssql-tds/src/datatypes/decoder.rs

  802         let value: ColumnValues = match byte_len {
  803             1 => ColumnValues::TinyInt(fast::read_byte(reader).await?), // Some(fast::read_byte(reader).await? as i64),
  804             2 => ColumnValues::SmallInt(fast::read_int16(reader).await?), // Some(fast::read_int16(reader).await? as i64),
  805             4 => ColumnValues::Int(fast::read_int32(reader).await?),
! 806             8 => ColumnValues::BigInt(fast::read_int64(reader).await?),
  807             0 => ColumnValues::Null,
  808             _ => {
  809                 return Err(crate::error::Error::from(Error::new(
  810                     std::io::ErrorKind::InvalidData,

---

  1291                     let cv = Box::pin(self.read_datetime2(
  1292                         reader,
  1293                         length,
  1294                         metadata.get_scale().ok_or_else(|| {
! 1295                             crate::error::Error::ImplementationError(
! 1296                                 "DateTime2N type should have scale".to_string(),
! 1297                             )
! 1298                         })?,
  1299                     ))
  1300                     .await?;
  1301                     if let ColumnValues::DateTime2(dt2) = cv {
  1302                         writer.write_datetime2(col, dt2);

---

  1311                     let cv = Box::pin(self.read_datetime_offset(
  1312                         reader,
  1313                         length,
  1314                         metadata.get_scale().ok_or_else(|| {
! 1315                             crate::error::Error::ImplementationError(
! 1316                                 "DateTimeOffsetN type should have scale".to_string(),
! 1317                             )
! 1318                         })?,
  1319                     ))
  1320                     .await?;
  1321                     if let ColumnValues::DateTimeOffset(dto) = cv {
  1322                         writer.write_datetimeoffset(col, dto);

---

mssql-tds/src/datatypes/row_writer.rs

  153         }
  154     }
  155 
  156     /// Creates a writer that reuses `buffer`'s existing allocation.
! 157     pub fn with_buffer(mut buffer: Vec<ColumnValues>, col_count: usize) -> Self {
! 158         buffer.clear();
! 159         buffer.reserve(col_count.saturating_sub(buffer.capacity()));
! 160         Self {
! 161             row: buffer,
! 162             completed: Vec::new(),
! 163             spare: Vec::new(),
! 164             string_pool: Vec::new(),
! 165             batching: false,
  166         }
  167     }
  168 
  169     /// Creates a batch-mode writer that queues completed rows internally.

---

mssql-tds/src/datatypes/sqldatatypes.rs

  798             | VariableLengthTypes::Numeric => {
  799                 let len_byte_count = vdt.get_len_byte_count();
  800                 let length = match len_byte_count {
  801                     1 => fast::read_byte(reader).await? as usize,
! 802                     2 => fast::read_uint16(reader).await? as usize,
! 803                     4 => fast::read_int32(reader).await? as usize,
  804                     _ => {
  805                         unreachable!(
  806                             "Invalid tds length {:?} for type: {:?}",
  807                             len_byte_count, data_type

---

  1031     T: TdsPacketReader + Send + Sync,
  1032 {
  1033     let len_byte_count = data_type.get_len_byte_count();
  1034     let length = match len_byte_count {
! 1035         1 => fast::read_byte(reader).await? as usize,
  1036         2 => fast::read_uint16(reader).await? as usize,
  1037         4 => {
  1038             let len_i32 = fast::read_int32(reader).await?;
  1039             // Negative values indicate invalid protocol data and should error out

---

mssql-tds/src/io/packet_reader.rs

  328         self.consume_bytes(1)?;
  329         Ok(result)
  330     }
  331 
! 332     fn try_take_fixed(&mut self, n: usize) -> Option<[u8; 8]> {
! 333         debug_assert!(n <= 8);
! 334         if n > 8 || !self.do_we_have_enough_data(n) {
! 335             return None;
! 336         }
! 337         let mut out = [0u8; 8];
! 338         out[..n]
! 339             .copy_from_slice(&self.working_buffer[self.buffer_position..self.buffer_position + n]);
! 340         self.buffer_position += n;
! 341         if self.buffer_length == self.buffer_position {
! 342             self.buffer_length = 0;
! 343             self.buffer_position = 0;
! 344         }
! 345         Some(out)
! 346     }
  347 
  348     async fn read_int16_big_endian(&mut self) -> TdsResult<i16> {
  349         if !self.do_we_have_enough_data(2) {
  350             self.read_tds_packet().await?;

---

  583     async fn read_byte(&mut self) -> TdsResult<u8> {
  584         (**self).read_byte().await
  585     }
  586 
! 587     fn try_take_fixed(&mut self, n: usize) -> Option<[u8; 8]> {
! 588         (**self).try_take_fixed(n)
! 589     }
  590 
  591     async fn read_int16_big_endian(&mut self) -> TdsResult<i16> {
  592         (**self).read_int16_big_endian().await
  593     }

---

mssql-tds/src/io/token_stream.rs

  152         remaining_request_timeout: Option<Duration>,
  153         cancel_handle: Option<&CancelHandle>,
  154         writer: &mut (dyn RowWriter + Send),
  155         max_rows: usize,
! 156     ) -> TdsResult<BatchRowsResult> {
  157         let mut rows = 0;
  158         while rows < max_rows {
  159             match self
  160                 .receive_row_into(context, remaining_request_timeout, cancel_handle, writer)

---

  175         Ok(BatchRowsResult {
  176             rows,
  177             stopped_at: None,
  178         })
! 179     }
  180 
  181     /// Resume a paused row decode from the column after the one that triggered
  182     /// [`pause_after_column`](RowWriter::pause_after_column).
  183     ///

---

  547             let value = Box::pin(decrypt_encrypted_column(decoder, reader, meta, dec)).await?;
  548             write_column_value(writer, col, value);
  549             return Ok(());
  550         }
! 551         tracing::info!(
  552             column = %meta.column_name,
  553             "Encrypted column has no column-encryption decryptor available \
  554              (Always Encrypted disabled for this command, or no key-store \
  555              provider registered); returning the raw ciphertext varbinary"

---

  616                 });
  617             }
  618         }
  619     }
! 620     Ok(BatchRowsResult {
! 621         rows,
! 622         stopped_at: None,
! 623     })
  624 }
  625 
  626 /// Resumes a paused row decode from `pause_state.next_column_index`.
  627 pub(crate) async fn resume_row_into_internal<R: TdsPacketReader + Send + Sync>(

---

---

🔗 Quick Links

View Azure DevOps Build · Coverage Report