diff --git a/mssql-mock-tds/src/protocol.rs b/mssql-mock-tds/src/protocol.rs index 8bc608f0..5fb9186e 100644 --- a/mssql-mock-tds/src/protocol.rs +++ b/mssql-mock-tds/src/protocol.rs @@ -674,13 +674,20 @@ pub fn build_login_ack() -> BytesMut { /// Build a DONE token pub fn build_done_token(row_count: u64) -> BytesMut { + build_done_token_with_status(0x0000, row_count) +} + +/// DONE token with an explicit status word. `0x0000` is `DONE_FINAL` (the batch +/// ends); `0x0001` is `DONE_MORE`, signalling that another result set follows in +/// the same batch (what a multi-statement `SELECT …; SELECT …` produces). +pub fn build_done_token_with_status(status: u16, row_count: u64) -> BytesMut { let mut token_data = BytesMut::new(); // DONE token (0xFD) token_data.put_u8(TokenType::Done as u8); - // Status: DONE_FINAL (0x00) - little-endian - token_data.put_u16_le(0x0000); + // Status - little-endian + token_data.put_u16_le(status); // CurCmd: SELECT (0xC1) - little-endian token_data.put_u16_le(0x00C1); @@ -822,6 +829,41 @@ pub fn build_info_token(info: &crate::query_response::InfoMessage) -> BytesMut { pub fn build_query_result(response: &crate::query_response::QueryResponse) -> BytesMut { let mut result = BytesMut::new(); + // Flatten the primary set and any additional sets (recursively) into batch + // order. All but the last close with DONE_MORE so the client advances across + // the boundary; the last closes with DONE_FINAL. Single-set responses (the + // common case) have no additional sets and serialize byte-identically. + let mut sets = Vec::new(); + collect_result_sets(response, &mut sets); + let last = sets.len() - 1; + for (idx, set) in sets.into_iter().enumerate() { + serialize_result_set(&mut result, set, idx == last); + } + + wrap_in_packet(PacketType::TabularResult, result) +} + +/// Flatten a response and its `additional_sets` (depth-first, in order) into a +/// linear batch of result sets so their terminal DONE statuses can be assigned. +fn collect_result_sets<'a>( + response: &'a crate::query_response::QueryResponse, + out: &mut Vec<&'a crate::query_response::QueryResponse>, +) { + out.push(response); + for set in &response.additional_sets { + collect_result_sets(set, out); + } +} + +/// Serialize one result set's tokens (ColMetadata, injected Info, rows, terminal +/// DONE) into `result`. `is_last` selects the DONE status: the final set in a +/// batch closes `DONE_FINAL`, earlier sets `DONE_MORE`. An `error_after` set is +/// always terminal (its ERROR + DONE end the batch), so it ignores `is_last`. +fn serialize_result_set( + result: &mut BytesMut, + response: &crate::query_response::QueryResponse, + is_last: bool, +) { // ColMetadata token (0x81) result.put_u8(TokenType::ColMetadata as u8); result.put_u16_le(response.columns.len() as u16); // Column count @@ -860,7 +902,7 @@ pub fn build_query_result(response: &crate::query_response::QueryResponse) -> By for row in response.rows.iter().take(err.after_rows) { result.put_u8(TokenType::Row as u8); for value in &row.values { - value.write_to_buffer(&mut result); + value.write_to_buffer(result); } } result.extend_from_slice(&build_error_token( @@ -880,16 +922,18 @@ pub fn build_query_result(response: &crate::query_response::QueryResponse) -> By for row in &response.rows { result.put_u8(TokenType::Row as u8); for value in &row.values { - value.write_to_buffer(&mut result); + value.write_to_buffer(result); } } - // DONE token - result.extend_from_slice(&build_done_token(response.rows.len() as u64)); + // DONE_MORE when another set follows, else terminal DONE_FINAL. + let status = if is_last { 0x0000 } else { 0x0001 }; + result.extend_from_slice(&build_done_token_with_status( + status, + response.rows.len() as u64, + )); } } - - wrap_in_packet(PacketType::TabularResult, result) } /// Build a bare ERROR token (0xAA) with no surrounding DONE or packet framing, diff --git a/mssql-mock-tds/src/query_response.rs b/mssql-mock-tds/src/query_response.rs index 922bf477..ae53d039 100644 --- a/mssql-mock-tds/src/query_response.rs +++ b/mssql-mock-tds/src/query_response.rs @@ -187,6 +187,13 @@ pub struct QueryResponse { /// When set, only the first `after_rows` rows are streamed, then an ERROR /// token and a terminal DONE are emitted (no trailing rows). pub error_after: Option, + /// Further result sets streamed in the same batch after this one. When + /// non-empty, this set's terminal DONE carries `DONE_MORE` and each + /// subsequent set is emitted in turn (the last one closing with a terminal + /// `DONE_FINAL`) — the multi-result-set shape a `SELECT …; SELECT …` batch + /// produces, exercising `SQLMoreResults`/`advance()` across the boundary. + /// Empty by default, so single-set responses serialize byte-identically. + pub additional_sets: Vec, } impl QueryResponse { @@ -197,6 +204,7 @@ impl QueryResponse { rows, info_tokens: Vec::new(), error_after: None, + additional_sets: Vec::new(), } } @@ -211,6 +219,14 @@ impl QueryResponse { self } + /// Append another result set to this batch. The current terminal DONE + /// becomes `DONE_MORE`; `next` is streamed after it (recursively, if it too + /// carries additional sets), with the final set closing on `DONE_FINAL`. + pub fn with_additional_result_set(mut self, next: QueryResponse) -> Self { + self.additional_sets.push(next); + self + } + /// Helper to create a response for SELECT 1 pub fn select_one() -> Self { Self { @@ -218,6 +234,7 @@ impl QueryResponse { rows: vec![Row::new(vec![ColumnValue::Int(1)])], info_tokens: Vec::new(), error_after: None, + additional_sets: Vec::new(), } } @@ -236,6 +253,7 @@ impl QueryResponse { ])], info_tokens: Vec::new(), error_after: None, + additional_sets: Vec::new(), } } } diff --git a/mssql-odbc/Cargo.toml b/mssql-odbc/Cargo.toml index 942e11e2..022dc8e1 100644 --- a/mssql-odbc/Cargo.toml +++ b/mssql-odbc/Cargo.toml @@ -46,5 +46,9 @@ windows = { version = "0.58", features = [ # `TdsClient` driven by scripted TDS tokens (see `mssql_tds::test_client_support`) # to cover client-driven ODBC paths without a live server. mssql-tds = { path = "../mssql-tds", features = ["test-util"] } +# A real raw-TCP mock TDS peer for the sync-fetch integration tests: scripted +# token clients are never `into_sync`-eligible (their transport yields no +# blocking parts), so the sync arm can only be exercised over an actual socket. +mssql-mock-tds = { path = "../mssql-mock-tds" } [build-dependencies] diff --git a/mssql-odbc/src/api/close_cursor.rs b/mssql-odbc/src/api/close_cursor.rs index 5ae54825..13277646 100644 --- a/mssql-odbc/src/api/close_cursor.rs +++ b/mssql-odbc/src/api/close_cursor.rs @@ -131,7 +131,7 @@ fn sql_free_stmt_close_safe(statement_handle: SqlHandle, stmt: &StmtHandle) -> S /// Resets cursor state on the statement (cursor is no longer open, metadata cleared). 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_fetch_state(); stmt_state.column_metadata.clear(); stmt_state.pending_row_counts.clear(); } @@ -167,7 +167,7 @@ pub(super) fn drain_and_release(stmt: &StmtHandle, statement_handle: SqlHandle) dbc_state.client.take() }; - let Some(mut client) = client else { + let Some(client) = client else { error!("drain_and_release: no TDS client to drain — this is a bug"); if let Ok(mut ds) = dbc.inner.lock() && ds.active_stmt == Some(statement_handle) @@ -177,6 +177,26 @@ pub(super) fn drain_and_release(stmt: &StmtHandle, statement_handle: SqlHandle) return DrainOutcome::Failed; }; + // Cursor close is a control-plane drain: revert a sync fetch cursor to the + // async edge first (rule C) so `close_query` runs on the reactor and the + // `sp_prepexec` `@handle` capture below works. A no-op when already async; a + // revert failure poisons the connection. + let mut client = match client.into_async() { + Ok(client) => client, + Err(e) => { + error!(%e, "drain_and_release: reverting sync cursor to async failed — connection lost"); + if let Ok(mut stmt_state) = stmt.inner.lock() { + post_tds_error(&mut stmt_state, &e, SQLSTATE_HY000); + } + if let Ok(mut ds) = dbc.inner.lock() + && ds.active_stmt == Some(statement_handle) + { + ds.active_stmt = None; + } + return DrainOutcome::Failed; + } + }; + if let Err(e) = dbc.runtime.block_on(client.close_query()) { error!(%e, "drain_and_release: failed to drain TDS stream — connection may be broken"); // Surface the failure as a diagnostic so the app is not told the close @@ -185,7 +205,7 @@ pub(super) fn drain_and_release(stmt: &StmtHandle, statement_handle: SqlHandle) post_tds_error(&mut stmt_state, &e, SQLSTATE_HY000); } if let Ok(mut ds) = dbc.inner.lock() { - ds.client = Some(client); + ds.store_async(client); if ds.active_stmt == Some(statement_handle) { ds.active_stmt = None; } @@ -203,7 +223,7 @@ pub(super) fn drain_and_release(stmt: &StmtHandle, statement_handle: SqlHandle) Err(_) => { error!("drain_and_release: stmt mutex poisoned while posting info messages"); if let Ok(mut dbc_state) = dbc.inner.lock() { - dbc_state.client = Some(client); + dbc_state.store_async(client); if dbc_state.active_stmt == Some(statement_handle) { dbc_state.active_stmt = None; } @@ -215,7 +235,7 @@ pub(super) fn drain_and_release(stmt: &StmtHandle, statement_handle: SqlHandle) // Drain complete: return client and release busy claim atomically. super::exec_common::capture_prepared_handle(stmt, &mut client); if let Ok(mut dbc_state) = dbc.inner.lock() { - dbc_state.client = Some(client); + dbc_state.store_async(client); if dbc_state.active_stmt == Some(statement_handle) { dbc_state.active_stmt = None; } diff --git a/mssql-odbc/src/api/driver_connect.rs b/mssql-odbc/src/api/driver_connect.rs index a4d193d0..fa1d8c1f 100644 --- a/mssql-odbc/src/api/driver_connect.rs +++ b/mssql-odbc/src/api/driver_connect.rs @@ -338,7 +338,11 @@ fn do_connect( let has_server_info = post_tds_info_messages(state, &info_messages); - state.client = Some(client); + // Cache the negotiated server version so SQLGetInfo(SQL_DBMS_VER) never has + // to touch the live client — it stays reportable even while a sync fetch + // cursor owns the connection. + state.server_version = client.server_version(); + state.store_async(client); state.connection_state = ConnectionState::Connected; debug!("SQLDriverConnectW: connected successfully"); diff --git a/mssql-odbc/src/api/exec_common.rs b/mssql-odbc/src/api/exec_common.rs index d5aa7486..93abd1ed 100644 --- a/mssql-odbc/src/api/exec_common.rs +++ b/mssql-odbc/src/api/exec_common.rs @@ -12,13 +12,14 @@ use tracing::error; use std::collections::VecDeque; use mssql_tds::connection::tds_client::{ResultSet, TdsClient}; +use mssql_tds::connection::tds_sync_client::SyncConversion; use mssql_tds::error::Error as TdsError; use mssql_tds::message::parameters::rpc_parameters::RpcParameter; use super::sqlstate::*; use crate::api::odbc_types::{SQL_ERROR, SQL_SUCCESS, SQL_SUCCESS_WITH_INFO, SqlHandle, SqlReturn}; use crate::error::post_sql_error; -use crate::handles::dbc::ConnectionState; +use crate::handles::dbc::{ConnectionState, DbcClient}; use crate::handles::stmt::{ STMT_STATE_CURSOR_OPEN, STMT_STATE_EXEC_CONTEXT, STMT_STATE_EXEC_STARTED, StmtState, }; @@ -82,15 +83,34 @@ pub(super) fn claim_connection( clear_exec_started(stmt); return Err(SQL_ERROR); }; - - Ok(client) + drop(dbc_state); + + // Control-plane work runs on the async edge. Revert a sync fetch cursor that + // a prior execute may have left open on this statement (no-op when already + // async, no network I/O). A revert failure means the connection is poisoned. + match client.into_async() { + Ok(client) => Ok(client), + Err(e) => { + error!(%e, "{op}: reverting sync cursor to async failed — connection lost"); + if let Ok(mut dbc_state) = dbc.inner.lock() + && dbc_state.active_stmt == Some(statement_handle) + { + dbc_state.active_stmt = None; + } + if let Ok(mut stmt_state) = stmt.inner.lock() { + post_tds_error(&mut stmt_state, &e, SQLSTATE_HY000); + } + clear_exec_started(stmt); + Err(SQL_ERROR) + } + } } /// 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) { if let Ok(mut dbc_state) = dbc.inner.lock() { - dbc_state.client = Some(client); + dbc_state.store_async(client); if dbc_state.active_stmt == Some(statement_handle) { dbc_state.active_stmt = None; } @@ -118,14 +138,29 @@ pub(super) fn try_claim_idle_client( } let client = dbc_state.client.take()?; dbc_state.active_stmt = Some(statement_handle); - Some(client) + drop(dbc_state); + // Best-effort internal claim: revert a sync cursor to async. A revert + // failure means the connection is poisoned; drop the busy claim and report + // no client (the dead client closes on drop). + match client.into_async() { + Ok(client) => Some(client), + Err(e) => { + error!(%e, "try_claim_idle_client: reverting sync cursor to async failed"); + if let Ok(mut dbc_state) = dbc.inner.lock() + && dbc_state.active_stmt == Some(statement_handle) + { + dbc_state.active_stmt = None; + } + None + } + } } /// 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) { if let Ok(mut dbc_state) = dbc.inner.lock() { - dbc_state.client = Some(client); + dbc_state.store_async(client); } } @@ -341,7 +376,11 @@ pub(super) fn finish_execute( stmt_state.clear_state(STMT_STATE_EXEC_STARTED); let has_server_info = post_tds_info_messages(&mut stmt_state, &info_messages); drop(stmt_state); - return_client_busy(dbc, client); + // Rule C: metadata is snapshotted above; only NOW attempt the reactor-free + // flip so DescribeCol / NumResultCols (which read STMT state) are unaffected. + if let Err(rc) = flip_to_fetch_edge(dbc, stmt, statement_handle, client) { + return rc; + } if has_server_info { SQL_SUCCESS_WITH_INFO } else { @@ -349,6 +388,59 @@ pub(super) fn finish_execute( } } +/// Flips the just-executed async client to the reactor-free sync fetch edge when +/// the transport is eligible (plaintext raw TCP), storing it busy for `SQLFetch`. +/// TLS / non-raw transports are returned to the async edge **unchanged**, so +/// fetch falls back to `block_on` byte-identically. A conversion failure poisons +/// the connection: the open-cursor state the caller committed is rolled back and +/// the error is surfaced. +/// +/// `into_sync` captures `Handle::try_current()`, so it MUST run inside the +/// runtime context — the bare ODBC thread has none, and a `None` capture would +/// later poison `into_async`. +/// +/// Shared by `finish_execute` (execute-time flip) and `SQLMoreResults` (re-flip +/// when advancing onto the next row-returning result set). `active_stmt` is left +/// untouched on success (the cursor stays open on the owning statement) and +/// cleared only on a poisoning failure. +pub(super) fn flip_to_fetch_edge( + dbc: &DbcHandle, + stmt: &StmtHandle, + statement_handle: SqlHandle, + client: TdsClient, +) -> Result<(), SqlReturn> { + let converted = { + let _runtime_guard = dbc.runtime.enter(); + client.into_sync() + }; + match converted { + SyncConversion::Converted(sync) => { + if let Ok(mut dbc_state) = dbc.inner.lock() { + dbc_state.client = Some(DbcClient::Sync(sync)); + } + Ok(()) + } + SyncConversion::NotEligible(client) => { + // TLS / non-raw transport: keep the untouched async fetch path. + return_client_busy(dbc, client); + Ok(()) + } + SyncConversion::Failed(e) => { + error!(%e, "finish_execute: sync flip failed — connection lost"); + if let Ok(mut stmt_state) = stmt.inner.lock() { + stmt_state.clear_state(STMT_STATE_CURSOR_OPEN | STMT_STATE_EXEC_CONTEXT); + post_tds_error(&mut stmt_state, &e, SQLSTATE_HY000); + } + if let Ok(mut dbc_state) = dbc.inner.lock() + && dbc_state.active_stmt == Some(statement_handle) + { + dbc_state.active_stmt = None; + } + Err(SQL_ERROR) + } + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/mssql-odbc/src/api/exec_direct.rs b/mssql-odbc/src/api/exec_direct.rs index 19103ef4..e075ab5b 100644 --- a/mssql-odbc/src/api/exec_direct.rs +++ b/mssql-odbc/src/api/exec_direct.rs @@ -106,7 +106,7 @@ fn sql_exec_direct_w_safe( // later execute failure cannot expose stale SQLNumResultCols/DescribeCol state. stmt_state.clear_state(STMT_STATE_EXEC_CONTEXT); stmt_state.column_metadata.clear(); - stmt_state.current_row = None; + stmt_state.reset_fetch_state(); stmt_state.row_count = -1; stmt_state.pending_row_counts.clear(); stmt_state.prepared_sql = None; @@ -267,7 +267,7 @@ mod tests { #[test] fn exec_direct_norow_statement_keeps_cursor_open_and_busy() { use crate::api::odbc_types::SQL_SUCCESS; - use crate::handles::dbc::DbcHandle; + use crate::handles::dbc::{DbcClient, DbcHandle}; use mssql_tds::test_client_support::{ col_metadata_empty, done_more_with_count, done_no_more, tds_client_from_tokens, }; @@ -285,7 +285,7 @@ mod tests { ]); { let mut ds = dbc.inner.lock().unwrap(); - ds.client = Some(client); + ds.client = Some(DbcClient::Async(client)); // active_stmt stays None => connection idle and claimable. } @@ -309,7 +309,7 @@ mod tests { #[test] fn exec_direct_norow_statement_with_message_returns_success_with_info() { use crate::api::odbc_types::SQL_SUCCESS_WITH_INFO; - use crate::handles::dbc::DbcHandle; + use crate::handles::dbc::{DbcClient, DbcHandle}; use mssql_tds::test_client_support::{ col_metadata_empty, done_more_with_count, done_no_more, info, tds_client_from_tokens, }; @@ -325,7 +325,7 @@ mod tests { ]); { let mut ds = dbc.inner.lock().unwrap(); - ds.client = Some(client); + ds.client = Some(DbcClient::Async(client)); } let stmt = unsafe { handle_from_raw::(h.stmt) }; diff --git a/mssql-odbc/src/api/execute.rs b/mssql-odbc/src/api/execute.rs index d2a6d61a..f0b905f0 100644 --- a/mssql-odbc/src/api/execute.rs +++ b/mssql-odbc/src/api/execute.rs @@ -158,7 +158,7 @@ fn stage_execution(stmt: &StmtHandle) -> Result { let drop_handle = stmt_state.pending_unprepare.take(); stmt_state.clear_state(STMT_STATE_EXEC_CONTEXT); stmt_state.column_metadata.clear(); - stmt_state.current_row = None; + stmt_state.reset_fetch_state(); stmt_state.row_count = -1; stmt_state.pending_row_counts.clear(); stmt_state.set_state(STMT_STATE_EXEC_STARTED); diff --git a/mssql-odbc/src/api/fetch.rs b/mssql-odbc/src/api/fetch.rs index 7e1e88f9..c535e19a 100644 --- a/mssql-odbc/src/api/fetch.rs +++ b/mssql-odbc/src/api/fetch.rs @@ -11,9 +11,11 @@ use crate::api::odbc_types::{ SqlReturn, }; use crate::error::free_errors; +use crate::handles::dbc::{DbcClient, DbcHandle}; use crate::handles::stmt::STMT_STATE_CURSOR_OPEN; use crate::handles::{HandleType, StmtHandle, handle_from_raw}; use mssql_tds::connection::tds_client::ResultSet; +use mssql_tds::connection::tds_sync_client::TdsSyncClient; /// Implements SQLFetch for the current forward-only result set. /// @@ -44,6 +46,7 @@ fn sql_fetch_safe(statement_handle: SqlHandle, stmt: &StmtHandle) -> SqlReturn { return SQL_ERROR; }; free_errors(&mut stmt_state); + stmt_state.getdata = None; if !stmt_state.has_state(STMT_STATE_CURSOR_OPEN) { error!("SQLFetch: no open cursor on this statement"); post_diag(&mut stmt_state, ERR_INVALID_CURSOR_STATE); @@ -58,7 +61,7 @@ fn sql_fetch_safe(statement_handle: SqlHandle, stmt: &StmtHandle) -> SqlReturn { fn fetch_rows_next(statement_handle: SqlHandle, stmt: &StmtHandle) -> SqlReturn { let dbc = stmt.parent_dbc(); - let mut client = { + let client = { let Ok(mut dbc_state) = dbc.inner.lock() else { error!("SQLFetch: dbc mutex poisoned"); return SQL_ERROR; @@ -130,6 +133,22 @@ fn fetch_rows_next(statement_handle: SqlHandle, stmt: &StmtHandle) -> SqlReturn } } + // Dispatch on the connection edge. Plaintext raw-TCP cursors fetch on the + // reactor-free sync client; TLS / non-raw stay on the unchanged async path. + match client { + DbcClient::Async(client) => fetch_row_async(dbc, stmt, statement_handle, client), + DbcClient::Sync(sync) => fetch_row_sync(dbc, stmt, statement_handle, sync), + } +} + +/// Async fetch arm: pulls one row via `block_on(next_row())`. Byte-identical to +/// the pre-rewire path — the TLS / non-raw-transport fallback. +fn fetch_row_async( + dbc: &DbcHandle, + stmt: &StmtHandle, + statement_handle: SqlHandle, + mut client: mssql_tds::connection::tds_client::TdsClient, +) -> SqlReturn { let fetch_result = dbc.runtime.block_on(client.next_row()); match fetch_result { @@ -137,7 +156,7 @@ fn fetch_rows_next(statement_handle: SqlHandle, stmt: &StmtHandle) -> SqlReturn let Ok(mut stmt_state) = stmt.inner.lock() else { error!("SQLFetch: stmt mutex poisoned storing row"); if let Ok(mut ds) = dbc.inner.lock() { - ds.client = Some(client); + ds.store_async(client); if ds.active_stmt == Some(statement_handle) { ds.active_stmt = None; } @@ -152,7 +171,7 @@ fn fetch_rows_next(statement_handle: SqlHandle, stmt: &StmtHandle) -> SqlReturn drop(stmt_state); if let Ok(mut dbc_state) = dbc.inner.lock() { - dbc_state.client = Some(client); + dbc_state.store_async(client); dbc_state.active_stmt = Some(statement_handle); } @@ -196,7 +215,7 @@ fn fetch_rows_next(statement_handle: SqlHandle, stmt: &StmtHandle) -> SqlReturn let Ok(mut stmt_state) = stmt.inner.lock() else { error!("SQLFetch: stmt mutex poisoned at end of rowset"); if let Ok(mut ds) = dbc.inner.lock() { - ds.client = Some(client); + ds.store_async(client); } return SQL_ERROR; }; @@ -205,7 +224,7 @@ fn fetch_rows_next(statement_handle: SqlHandle, stmt: &StmtHandle) -> SqlReturn // SQLMoreResults / SQLCloseCursor / SQLFreeStmt(SQL_CLOSE). drop(stmt_state); if let Ok(mut dbc_state) = dbc.inner.lock() { - dbc_state.client = Some(client); + dbc_state.store_async(client); } debug!("SQLFetch: no more rows in current result set"); @@ -221,7 +240,7 @@ fn fetch_rows_next(statement_handle: SqlHandle, stmt: &StmtHandle) -> SqlReturn post_tds_info_messages(&mut stmt_state, &info_messages); } if let Ok(mut dbc_state) = dbc.inner.lock() { - dbc_state.client = Some(client); + dbc_state.store_async(client); if dbc_state.active_stmt == Some(statement_handle) { dbc_state.active_stmt = None; } @@ -231,11 +250,100 @@ fn fetch_rows_next(statement_handle: SqlHandle, stmt: &StmtHandle) -> SqlReturn } } +/// Sync fetch arm: pulls exactly one row via the reactor-free +/// `TdsSyncClient::next_row_into`, writing it straight into `current_row`. There is +/// no prefetch/batch buffer — every `SQLFetch` reads one row, byte-identical to the +/// async arm (INFO drained and surfaced on the same `SQLFetch` that serves the row, +/// errors surfaced immediately). The previous row's allocation is recycled into the +/// writer to keep the path free of per-row heap churn. +fn fetch_row_sync( + dbc: &DbcHandle, + stmt: &StmtHandle, + statement_handle: SqlHandle, + mut sync: TdsSyncClient, +) -> SqlReturn { + use mssql_tds::datatypes::row_writer::DefaultRowWriter; + + // Recycle the previously-served row's allocation into this fetch's writer. + let recycled = match stmt.inner.lock() { + Ok(mut ss) => ss.current_row.take().unwrap_or_default(), + Err(_) => { + error!("SQLFetch: stmt mutex poisoned starting sync fetch"); + if let Ok(mut ds) = dbc.inner.lock() { + ds.client = Some(DbcClient::Sync(sync)); + } + return SQL_ERROR; + } + }; + let mut writer = DefaultRowWriter::from_recycled(recycled); + + match sync.next_row_into(&mut writer) { + Ok(true) => { + // Drain INFO now, attributed to this SQLFetch (which serves the row). + let info_messages = sync.take_info_messages(); + let has_server_info = !info_messages.is_empty(); + match stmt.inner.lock() { + Ok(mut ss) => { + ss.current_row = Some(writer.take_row()); + post_tds_info_messages(&mut ss, &info_messages); + } + Err(_) => { + error!("SQLFetch: stmt mutex poisoned storing sync row"); + if let Ok(mut ds) = dbc.inner.lock() { + ds.client = Some(DbcClient::Sync(sync)); + if ds.active_stmt == Some(statement_handle) { + ds.active_stmt = None; + } + } + return SQL_ERROR; + } + } + if let Ok(mut ds) = dbc.inner.lock() { + ds.client = Some(DbcClient::Sync(sync)); + ds.active_stmt = Some(statement_handle); + } + debug!("SQLFetch: row fetched (sync)"); + if has_server_info { + SQL_SUCCESS_WITH_INFO + } else { + SQL_SUCCESS + } + } + Ok(false) => { + // End of current rowset — mirror the async Ok(None) contract: leave any + // INFO on the client for the boundary call, keep the cursor open, keep + // the connection busy on this statement. `current_row` is already None. + if let Ok(mut ds) = dbc.inner.lock() { + ds.client = Some(DbcClient::Sync(sync)); + } + debug!("SQLFetch: no more rows in current result set (sync)"); + SQL_NO_DATA + } + Err(e) => { + error!(%e, "SQLFetch: row fetch failed (sync)"); + let info_messages = sync.take_info_messages(); + if let Ok(mut ss) = stmt.inner.lock() { + ss.current_row = None; + ss.clear_state(STMT_STATE_CURSOR_OPEN); + post_tds_error(&mut ss, &e, SQLSTATE_HY000); + post_tds_info_messages(&mut ss, &info_messages); + } + if let Ok(mut ds) = dbc.inner.lock() { + ds.client = Some(DbcClient::Sync(sync)); + if ds.active_stmt == Some(statement_handle) { + ds.active_stmt = None; + } + } + SQL_ERROR + } + } +} + #[cfg(test)] mod tests { use super::*; use crate::api::odbc_types::SQL_NULL_HANDLE; - use crate::handles::dbc::DbcHandle; + use crate::handles::dbc::{DbcClient, DbcHandle}; use crate::handles::stmt::STMT_STATE_CURSOR_OPEN; use crate::test_support::TestHandles; @@ -333,7 +441,7 @@ mod tests { let dbc_handle = unsafe { handle_from_raw::(h.dbc) }; { let mut dbc_state = dbc_handle.inner.lock().unwrap(); - dbc_state.client = Some(client); + dbc_state.client = Some(DbcClient::Async(client)); dbc_state.active_stmt = Some(h.stmt); } diff --git a/mssql-odbc/src/api/get_data.rs b/mssql-odbc/src/api/get_data.rs index 1714c82a..5cb35124 100644 --- a/mssql-odbc/src/api/get_data.rs +++ b/mssql-odbc/src/api/get_data.rs @@ -6,16 +6,18 @@ use tracing::{debug, error}; use super::odbc_types::{ - SQL_C_CHAR, SQL_C_WCHAR, SQL_ERROR, SQL_INVALID_HANDLE, SQL_NULL_DATA, SQL_SUCCESS, + SQL_C_CHAR, SQL_C_WCHAR, SQL_ERROR, SQL_INVALID_HANDLE, SQL_NO_DATA, SQL_NULL_DATA, SQL_SUCCESS, SQL_SUCCESS_WITH_INFO, SqlHandle, SqlLen, SqlPointer, SqlReturn, SqlSmallInt, SqlUSmallInt, }; use super::sqlstate::*; use crate::api::odbc_types::SqlWChar; use crate::api::util::{copy_with_nul, write_if_some}; use crate::error::{free_errors, post_sql_error}; -use crate::handles::stmt::STMT_STATE_CURSOR_OPEN; +use crate::handles::stmt::{ + GetDataProgress, GetDataUnits, STMT_STATE_CURSOR_OPEN, StmtState, +}; use crate::handles::{HandleType, StmtHandle, handle_from_raw}; -use mssql_tds::datatypes::column_values::ColumnValues; +use mssql_tds::datatypes::column_values::{ColumnValues, SqlDateTime2}; /// Implements SQLGetData for current-row retrieval. /// @@ -139,51 +141,223 @@ fn sql_get_data_safe( buffer_length as usize }; - let value = &row[col_index - 1]; - if matches!(value, ColumnValues::Null) { - unsafe { write_if_some(strlen_or_ind_ptr, SQL_NULL_DATA) }; - // Write a NUL terminator into the caller buffer when there's room. The - // helper handles null `dst` and zero-length uniformly. - if target_type == SQL_C_WCHAR { - unsafe { - copy_with_nul(target_value_ptr as *mut SqlWChar, buf_elements, &[]); + let is_wchar = target_type == SQL_C_WCHAR; + + // (Re)build the streaming cursor when a different column becomes active or + // the requested C type changed. The value is converted + encoded exactly + // once here; subsequent calls serve cached chunks, keeping chunked LOB + // retrieval O(n) total instead of re-encoding the full value per call. + let needs_build = match &stmt_state.getdata { + Some(p) => p.column != column_number || !p.units.matches_wchar(is_wchar), + None => true, + }; + if needs_build { + let value = &row[col_index - 1]; + + // Zero-heap fast path: numeric/temporal/GUID values render to short + // ASCII that virtually always fits the caller's buffer in one call + // (the bench uses an 8 KiB buffer). Format straight onto the stack and + // copy out, skipping the per-value String + Vec allocation that native + // avoids. Only the rare doesn't-fit case falls through to the cursor. + if let Some((tmp, len)) = format_scalar_ascii(value) { + let cap = buf_elements.saturating_sub(1); + if len <= cap { + let ascii = &tmp[..len]; + let ind_bytes = if is_wchar { len * 2 } else { len }; + unsafe { write_if_some(strlen_or_ind_ptr, ind_bytes as SqlLen) }; + if is_wchar { + let mut wide = [0u16; SCALAR_ASCII_CAP]; + for (w, &b) in wide.iter_mut().zip(ascii) { + *w = u16::from(b); + } + unsafe { + copy_with_nul(target_value_ptr as *mut SqlWChar, buf_elements, &wide[..len]); + } + } else { + unsafe { + copy_with_nul(target_value_ptr as *mut u8, buf_elements, ascii); + } + } + // Record consumption without allocating (empty Vec) so a second + // SQLGetData on this column returns SQL_NO_DATA. + stmt_state.getdata = Some(GetDataProgress { + column: column_number, + units: GetDataUnits::Char(Vec::new()), + offset: 0, + exhausted: true, + }); + return SQL_SUCCESS; } - } else { - unsafe { - copy_with_nul(target_value_ptr as *mut u8, buf_elements, &[]); + } + + // Zero-copy string fast path: an ASCII/UTF-8 char column served as + // SQL_C_CHAR needs no transcoding — serve its bytes straight from the + // fetched row when they fit, skipping the decode + String allocation + // that `to_utf8_string()` would perform. + if !is_wchar + && let ColumnValues::String(s) = value + && let Some(bytes) = string_passthrough_bytes(s) + { + let len = bytes.len(); + let cap = buf_elements.saturating_sub(1); + if len <= cap { + unsafe { write_if_some(strlen_or_ind_ptr, len as SqlLen) }; + unsafe { + copy_with_nul(target_value_ptr as *mut u8, buf_elements, bytes); + } + stmt_state.getdata = Some(GetDataProgress { + column: column_number, + units: GetDataUnits::Char(Vec::new()), + offset: 0, + exhausted: true, + }); + return SQL_SUCCESS; } } - return SQL_SUCCESS; + + // Zero-alloc UTF-16 → UTF-8 fast path: an NVarchar column served as + // SQL_C_CHAR transcodes straight onto the stack, skipping the encoding_rs + // decode + String allocation in `to_utf8_string()`. Small strings (the + // common case) fit `STRING_STACK_CAP`; anything larger or malformed falls + // through to the allocating cursor path. + if !is_wchar + && let ColumnValues::String(s) = value + && s.is_utf16() + { + let mut tmp = [0u8; STRING_STACK_CAP]; + if let Some(len) = utf16le_to_utf8(&s.bytes, &mut tmp) { + let cap = buf_elements.saturating_sub(1); + if len <= cap { + unsafe { write_if_some(strlen_or_ind_ptr, len as SqlLen) }; + unsafe { + copy_with_nul(target_value_ptr as *mut u8, buf_elements, &tmp[..len]); + } + stmt_state.getdata = Some(GetDataProgress { + column: column_number, + units: GetDataUnits::Char(Vec::new()), + offset: 0, + exhausted: true, + }); + return SQL_SUCCESS; + } + } + } + + let units = if matches!(value, ColumnValues::Null) { + GetDataUnits::Null + } else { + let Some(as_text) = column_value_to_text(value) else { + post_sql_error( + &mut stmt_state, + SQLSTATE_HYC00, + 0, + "Column type conversion not yet implemented", + ); + return SQL_ERROR; + }; + if is_wchar { + GetDataUnits::WChar(as_text.encode_utf16().collect()) + } else { + GetDataUnits::Char(as_text.into_bytes()) + } + }; + stmt_state.getdata = Some(GetDataProgress { + column: column_number, + units, + offset: 0, + exhausted: false, + }); } - let Some(as_text) = column_value_to_text(value) else { - post_sql_error( - &mut stmt_state, - SQLSTATE_HYC00, - 0, - "Column type conversion not yet implemented", - ); - return SQL_ERROR; + serve_getdata_chunk( + &mut stmt_state, + is_wchar, + target_value_ptr, + buf_elements, + strlen_or_ind_ptr, + ) +} + +/// Serves the next chunk from the active `SQLGetData` stream, advancing its +/// offset and posting `01004` on truncation. Returns `SQL_NO_DATA` once the +/// value has been fully delivered (the call after the terminal chunk). +fn serve_getdata_chunk( + stmt_state: &mut StmtState, + is_wchar: bool, + target_value_ptr: SqlPointer, + buf_elements: usize, + strlen_or_ind_ptr: *mut SqlLen, +) -> SqlReturn { + let Some(mut prog) = stmt_state.getdata.take() else { + return SQL_NO_DATA; }; + if prog.exhausted { + // Terminal chunk already delivered; end the sequence and drop the + // cursor so a re-fetch of the same column starts fresh. + return SQL_NO_DATA; + } - if target_type == SQL_C_WCHAR { - let utf16: Vec = as_text.encode_utf16().collect(); - write_string_result( - &mut stmt_state, - &utf16, - target_value_ptr as *mut SqlWChar, + let (advance, truncated) = match &prog.units { + GetDataUnits::Null => { + unsafe { write_if_some(strlen_or_ind_ptr, SQL_NULL_DATA) }; + if is_wchar { + unsafe { copy_with_nul(target_value_ptr as *mut SqlWChar, buf_elements, &[]) }; + } else { + unsafe { copy_with_nul(target_value_ptr as *mut u8, buf_elements, &[]) }; + } + (0usize, false) + } + GetDataUnits::Char(data) => serve_units::( + data, + prog.offset, + target_value_ptr as *mut u8, buf_elements, strlen_or_ind_ptr, - ) - } else { - write_string_result( - &mut stmt_state, - as_text.as_bytes(), - target_value_ptr as *mut u8, + ), + GetDataUnits::WChar(data) => serve_units::( + data, + prog.offset, + target_value_ptr as *mut SqlWChar, buf_elements, strlen_or_ind_ptr, - ) + ), + }; + + prog.offset += advance; + prog.exhausted = !truncated; + stmt_state.getdata = Some(prog); + + if truncated { + post_diag(stmt_state, ERR_STRING_RIGHT_TRUNCATION); + SQL_SUCCESS_WITH_INFO + } else { + SQL_SUCCESS + } +} + +/// Copies one buffer-sized chunk of `data` starting at `offset`, writing the +/// remaining byte count into the length/indicator pointer per the ODBC spec. +/// Returns `(units_delivered, truncated)` where `truncated` means more data +/// remains after this chunk. +fn serve_units( + data: &[T], + offset: usize, + dst: *mut T, + buf_elements: usize, + strlen_or_ind_ptr: *mut SqlLen, +) -> (usize, bool) { + let total = data.len(); + let remaining = total - offset; + let remaining_bytes = (remaining * std::mem::size_of::()) as SqlLen; + unsafe { write_if_some(strlen_or_ind_ptr, remaining_bytes) }; + + // Leave room for the NUL terminator the driver always appends. + let cap = buf_elements.saturating_sub(1); + let n = remaining.min(cap); + unsafe { + copy_with_nul(dst, buf_elements, &data[offset..offset + n]); } + (n, remaining > n) } /// Writes `src` to the caller's output buffer with ODBC string semantics: @@ -196,24 +370,6 @@ fn sql_get_data_safe( /// The caller-provided pointers are written through small `unsafe` blocks /// inside this function; both pointer arguments are obligations of the FFI /// caller (validated against the buffer length passed by the DM). -fn write_string_result( - stmt_state: &mut crate::handles::stmt::StmtState, - src: &[T], - target_value_ptr: *mut T, - buf_elements: usize, - strlen_or_ind_ptr: *mut SqlLen, -) -> SqlReturn { - let byte_len = std::mem::size_of_val(src) as SqlLen; - unsafe { write_if_some(strlen_or_ind_ptr, byte_len) }; - let truncated = unsafe { copy_with_nul(target_value_ptr, buf_elements, src) }; - if truncated { - post_diag(stmt_state, ERR_STRING_RIGHT_TRUNCATION); - SQL_SUCCESS_WITH_INFO - } else { - SQL_SUCCESS - } -} - fn column_value_to_text(v: &ColumnValues) -> Option { match v { ColumnValues::TinyInt(x) => Some(x.to_string()), @@ -225,18 +381,321 @@ fn column_value_to_text(v: &ColumnValues) -> Option { ColumnValues::Bit(x) => Some(if *x { "1".into() } else { "0".into() }), ColumnValues::String(s) => Some(s.to_utf8_string()), ColumnValues::Uuid(u) => Some(u.to_string()), + ColumnValues::Decimal(d) | ColumnValues::Numeric(d) => Some(d.to_string()), + ColumnValues::DateTime2(dt2) => Some(format_datetime2(dt2)), ColumnValues::Null => Some(String::new()), _ => None, } } +/// Stack scratch size for one-shot scalar rendering. Fits the widest scalar +/// text: `decimal(38)` sign+digits+point (41), `uniqueidentifier` (36), +/// `datetime2` (27), and any integer/float shortest form. +const SCALAR_ASCII_CAP: usize = 64; + +/// Minimal stack-backed `fmt::Write` sink; renders scalar text with no heap +/// allocation. Writes beyond capacity are dropped, which `format_scalar_ascii` +/// treats as "doesn't fit" and falls back to the allocating path. +struct StackBuf { + buf: [u8; SCALAR_ASCII_CAP], + len: usize, +} + +impl std::fmt::Write for StackBuf { + fn write_str(&mut self, s: &str) -> std::fmt::Result { + let bytes = s.as_bytes(); + if self.len + bytes.len() > self.buf.len() { + return Err(std::fmt::Error); + } + self.buf[self.len..self.len + bytes.len()].copy_from_slice(bytes); + self.len += bytes.len(); + Ok(()) + } +} + +/// Renders a numeric/temporal/GUID value to ASCII on the stack, returning the +/// scratch buffer and byte length. Returns `None` for strings, NULL, and +/// unsupported types, which take the allocating cursor path. All produced text +/// is pure ASCII, so the SQL_C_WCHAR path can widen bytes 1:1. +fn format_scalar_ascii(v: &ColumnValues) -> Option<([u8; SCALAR_ASCII_CAP], usize)> { + use std::fmt::Write as _; + let mut sb = StackBuf { + buf: [0u8; SCALAR_ASCII_CAP], + len: 0, + }; + let ok = match v { + ColumnValues::TinyInt(x) => write!(sb, "{x}"), + ColumnValues::SmallInt(x) => write!(sb, "{x}"), + ColumnValues::Int(x) => write!(sb, "{x}"), + ColumnValues::BigInt(x) => write!(sb, "{x}"), + ColumnValues::Real(x) => write!(sb, "{x}"), + ColumnValues::Float(x) => write!(sb, "{x}"), + ColumnValues::Bit(x) => write!(sb, "{}", if *x { 1 } else { 0 }), + ColumnValues::Uuid(u) => write!(sb, "{u}"), + ColumnValues::Decimal(d) | ColumnValues::Numeric(d) => write_decimal(&mut sb, d), + ColumnValues::DateTime2(dt2) => write_datetime2(&mut sb, dt2), + _ => return None, + }; + ok.ok().map(|()| (sb.buf, sb.len)) +} + +/// Returns a char column's bytes when they can be handed to an `SQL_C_CHAR` +/// caller with no transcoding: UTF-8 payloads pass through directly, and +/// single-byte (collation-based) or delayed-encoding payloads pass through when +/// they are pure ASCII (identical under ASCII, Windows-125x, and UTF-8). Any +/// non-ASCII single-byte or UTF-16 payload returns `None` and takes the +/// allocating decode path. +fn string_passthrough_bytes(s: &mssql_tds::datatypes::sql_string::SqlString) -> Option<&[u8]> { + use mssql_tds::datatypes::sql_string::EncodingType; + match s.encoding_type() { + EncodingType::Utf8 => Some(&s.bytes), + EncodingType::LcidBased(_) | EncodingType::DelayedSet => { + s.bytes.iter().all(u8::is_ascii).then_some(&s.bytes[..]) + } + EncodingType::Utf16 => None, + } +} + +/// Stack scratch for one-shot string transcodes. Covers the common inline +/// `NVARCHAR`/`VARCHAR` column (a few hundred bytes); larger values fall through +/// to the allocating cursor path. +const STRING_STACK_CAP: usize = 1024; + +/// Transcodes UTF-16LE `bytes` into UTF-8 written to `out`, returning the byte +/// length on success. Returns `None` on odd byte counts, malformed surrogate +/// pairs, or when the output would not fit `out` — all of which defer to the +/// allocating `to_utf8_string()` path. +fn utf16le_to_utf8(bytes: &[u8], out: &mut [u8]) -> Option { + if !bytes.len().is_multiple_of(2) { + return None; + } + let mut o = 0usize; + let mut i = 0usize; + while i + 1 < bytes.len() { + let u = u16::from_le_bytes([bytes[i], bytes[i + 1]]); + i += 2; + let cp: u32 = if (0xD800..0xDC00).contains(&u) { + if i + 1 >= bytes.len() { + return None; + } + let lo = u16::from_le_bytes([bytes[i], bytes[i + 1]]); + if !(0xDC00..0xE000).contains(&lo) { + return None; + } + i += 2; + 0x1_0000 + ((u32::from(u) - 0xD800) << 10) + (u32::from(lo) - 0xDC00) + } else if (0xDC00..0xE000).contains(&u) { + return None; + } else { + u32::from(u) + }; + o += encode_utf8(cp, out.get_mut(o..)?)?; + } + Some(o) +} + +/// Encodes a scalar Unicode code point as UTF-8 into `out`, returning the byte +/// length, or `None` if `out` is too small. +#[inline] +fn encode_utf8(cp: u32, out: &mut [u8]) -> Option { + match cp { + 0x0000..=0x007F => { + *out.first_mut()? = cp as u8; + Some(1) + } + 0x0080..=0x07FF => { + let b = out.get_mut(..2)?; + b[0] = 0xC0 | (cp >> 6) as u8; + b[1] = 0x80 | (cp & 0x3F) as u8; + Some(2) + } + 0x0800..=0xFFFF => { + let b = out.get_mut(..3)?; + b[0] = 0xE0 | (cp >> 12) as u8; + b[1] = 0x80 | ((cp >> 6) & 0x3F) as u8; + b[2] = 0x80 | (cp & 0x3F) as u8; + Some(3) + } + _ => { + let b = out.get_mut(..4)?; + b[0] = 0xF0 | (cp >> 18) as u8; + b[1] = 0x80 | ((cp >> 12) & 0x3F) as u8; + b[2] = 0x80 | ((cp >> 6) & 0x3F) as u8; + b[3] = 0x80 | (cp & 0x3F) as u8; + Some(4) + } + } +} + +/// Renders a `DecimalParts` straight into a `fmt::Write` sink with no heap +/// allocation, unlike its `Display`, which builds two intermediate `String`s. +/// Folds the little-endian 32-bit parts into a `u128` (SQL Server decimals are +/// at most 38 digits) and places the decimal point by digit position. +fn write_decimal( + w: &mut W, + d: &mssql_tds::datatypes::decoder::DecimalParts, +) -> std::fmt::Result { + let value: u128 = d + .int_parts + .iter() + .enumerate() + .fold(0u128, |acc, (i, &part)| acc + ((part as u32 as u128) << (i * 32))); + + // Most-significant-first ASCII digits on the stack (u128 fits in 39). + let mut digits = [0u8; 40]; + let n = if value == 0 { + digits[0] = b'0'; + 1 + } else { + let mut rev = [0u8; 40]; + let mut rn = 0; + let mut v = value; + while v > 0 { + rev[rn] = b'0' + (v % 10) as u8; + v /= 10; + rn += 1; + } + for i in 0..rn { + digits[i] = rev[rn - 1 - i]; + } + rn + }; + + if !d.is_positive { + w.write_char('-')?; + } + let scale = d.scale as usize; + let s = |a: usize, b: usize| -> &str { + // All entries are ASCII digits, so this slice is valid UTF-8. + std::str::from_utf8(&digits[a..b]).unwrap_or("") + }; + if scale == 0 { + w.write_str(s(0, n)) + } else if n <= scale { + w.write_str("0.")?; + for _ in 0..(scale - n) { + w.write_char('0')?; + } + w.write_str(s(0, n)) + } else { + let split = n - scale; + w.write_str(s(0, split))?; + w.write_char('.')?; + w.write_str(s(split, n)) + } +} + +/// Render a `datetime2` as the canonical `YYYY-MM-DD HH:MM:SS.fffffff` text, +/// matching what msodbcsql18 returns for `SQL_C_CHAR`. `days` is 0-based from +/// 0001-01-01 (proleptic Gregorian); `time_nanoseconds` is in 100 ns ticks. +fn format_datetime2(dt2: &SqlDateTime2) -> String { + let mut s = String::with_capacity(27); + let _ = write_datetime2(&mut s, dt2); + s +} + +/// Shared `datetime2` renderer that writes into any `fmt::Write` sink, so both +/// the allocating and zero-heap paths share one formatting implementation. +fn write_datetime2(w: &mut W, dt2: &SqlDateTime2) -> std::fmt::Result { + let (y, m, d) = civil_from_days(i64::from(dt2.days) - 719_162); + let ticks = dt2.time.time_nanoseconds; + let hour = ticks / 36_000_000_000; + let rem = ticks % 36_000_000_000; + let minute = rem / 600_000_000; + let rem = rem % 600_000_000; + let second = rem / 10_000_000; + let frac = rem % 10_000_000; + write!(w, "{y:04}-{m:02}-{d:02} {hour:02}:{minute:02}:{second:02}.{frac:07}") +} + +/// Civil date from a day count relative to the Unix epoch (1970-01-01 == 0). +/// Howard Hinnant's algorithm, valid across the full proleptic Gregorian range. +fn civil_from_days(z: i64) -> (i64, u32, u32) { + let z = z + 719_468; + let era = if z >= 0 { z } else { z - 146_096 } / 146_097; + let doe = (z - era * 146_097) as u64; + let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365; + let y = yoe as i64 + era * 400; + let doy = doe - (365 * yoe + yoe / 4 - yoe / 100); + let mp = (5 * doy + 2) / 153; + let d = (doy - (153 * mp + 2) / 5 + 1) as u32; + let m = if mp < 10 { mp + 3 } else { mp - 9 } as u32; + (if m <= 2 { y + 1 } else { y }, m, d) +} + #[cfg(test)] mod tests { use super::*; use crate::api::odbc_types::{SQL_C_LONG, SQL_NULL_HANDLE}; use crate::test_support::TestHandles; + use mssql_tds::datatypes::decoder::DecimalParts; use mssql_tds::datatypes::sql_string::SqlString; + fn fast(v: &ColumnValues) -> String { + let (buf, len) = format_scalar_ascii(v).expect("fast path should handle this scalar"); + String::from_utf8(buf[..len].to_vec()).unwrap() + } + + #[test] + fn fast_path_matches_display_for_scalars() { + // The zero-heap fast path must render byte-identically to the + // allocating `column_value_to_text` reference for every scalar it claims. + let cases = [ + ColumnValues::TinyInt(0), + ColumnValues::TinyInt(255), + ColumnValues::SmallInt(-12345), + ColumnValues::Int(0), + ColumnValues::Int(-2147483648), + ColumnValues::Int(1467152272), + ColumnValues::BigInt(9223372036854775807), + ColumnValues::BigInt(-9223372036854775808), + ColumnValues::Bit(true), + ColumnValues::Bit(false), + ]; + for c in &cases { + assert_eq!(fast(c), column_value_to_text(c).unwrap(), "mismatch for {c:?}"); + } + } + + #[test] + fn fast_path_matches_display_for_decimal() { + for s in ["0", "1467152272.0000", "-0.0001", "12345.6789", "0.0001"] { + let (prec, scale) = { + let frac = s.split('.').nth(1).map(str::len).unwrap_or(0) as u8; + (38u8, frac) + }; + let d = DecimalParts::from_string(s, prec, scale).unwrap(); + let v = ColumnValues::Decimal(d); + assert_eq!(fast(&v), column_value_to_text(&v).unwrap(), "mismatch for {s}"); + } + } + + #[test] + fn utf16_fast_path_matches_encoding_rs() { + // The zero-alloc UTF-16LE → UTF-8 transcode must render byte-identically + // to the encoding_rs reference in `to_utf8_string()` across BMP, non-ASCII, + // and astral (surrogate-pair) code points. + for text in ["", "x", "hello", "éééé", "café \u{2764} au lait", "𝄞𝕏🚀"] { + let s = SqlString::from_utf8_string(text.to_string()); + let mut out = [0u8; STRING_STACK_CAP]; + let len = utf16le_to_utf8(&s.bytes, &mut out).expect("valid utf16"); + assert_eq!( + std::str::from_utf8(&out[..len]).unwrap(), + s.to_utf8_string(), + "mismatch for {text:?}" + ); + } + } + + #[test] + fn utf16_fast_path_rejects_odd_and_overflow() { + // Odd byte counts and outputs larger than the sink defer to the + // allocating path rather than corrupting data. + assert_eq!(utf16le_to_utf8(&[0x41], &mut [0u8; 16]), None); + let s = SqlString::from_utf8_string("éééé".to_string()); + assert_eq!(utf16le_to_utf8(&s.bytes, &mut [0u8; 4]), None); + } + #[test] fn get_data_null_handle() { let ret = unsafe { diff --git a/mssql-odbc/src/api/get_info.rs b/mssql-odbc/src/api/get_info.rs index 5db82963..6fec92a2 100644 --- a/mssql-odbc/src/api/get_info.rs +++ b/mssql-odbc/src/api/get_info.rs @@ -146,9 +146,7 @@ fn sql_get_info_w_safe( // Use the version negotiated at login; fall back to a neutral // placeholder when the connection has no reported version yet. let version = state - .client - .as_ref() - .and_then(|c| c.server_version()) + .server_version .map(|v| format!("{:02}.{:02}.{:04}", v.major, v.minor, v.build)) .unwrap_or_else(|| "00.00.0000".to_string()); write_wide_str( diff --git a/mssql-odbc/src/api/get_type_info.rs b/mssql-odbc/src/api/get_type_info.rs index c76decde..f7279137 100644 --- a/mssql-odbc/src/api/get_type_info.rs +++ b/mssql-odbc/src/api/get_type_info.rs @@ -148,7 +148,7 @@ fn sql_get_type_info_w_safe( // failure cannot expose stale SQLNumResultCols/DescribeCol state. stmt_state.clear_state(STMT_STATE_EXEC_CONTEXT); stmt_state.column_metadata.clear(); - stmt_state.current_row = None; + stmt_state.reset_fetch_state(); stmt_state.prepared_sql = None; // A cached prepared plan is superseded; release its server handle // (deferred) once we hold the client below. diff --git a/mssql-odbc/src/api/mod.rs b/mssql-odbc/src/api/mod.rs index 1857037b..f2cd8a58 100644 --- a/mssql-odbc/src/api/mod.rs +++ b/mssql-odbc/src/api/mod.rs @@ -35,3 +35,12 @@ pub(crate) mod util; // All `#[unsafe(no_mangle)] pub extern "C"` symbols are defined here. mod exports; pub use exports::*; + +// In-crate integration tests for the reactor-free sync fetch edge. Kept in the +// crate (not `tests/`) because the driver is a `cdylib`: an external test crate +// can neither link it nor reach the `pub(crate)` entry points and handle state +// these tests drive. They stand up a real `mssql-mock-tds` peer over TCP so the +// connection is sync-eligible (raw TCP), which the scripted-token unit tests +// cannot be. +#[cfg(test)] +mod sync_fetch_tests; diff --git a/mssql-odbc/src/api/more_results.rs b/mssql-odbc/src/api/more_results.rs index 6ef0a2b7..66b67458 100644 --- a/mssql-odbc/src/api/more_results.rs +++ b/mssql-odbc/src/api/more_results.rs @@ -73,7 +73,7 @@ fn sql_more_results_safe(statement_handle: SqlHandle, stmt: &StmtHandle) -> SqlR // Take the client; keep active_stmt set so concurrent statements continue // to see the connection as busy throughout the advance. - let mut client = { + let client = { let Ok(mut dbc_state) = dbc.inner.lock() else { error!("SQLMoreResults: dbc mutex poisoned"); return SQL_ERROR; @@ -99,6 +99,27 @@ fn sql_more_results_safe(statement_handle: SqlHandle, stmt: &StmtHandle) -> SqlR client }; + // Result-set boundary: revert a sync fetch cursor to the async edge before + // driving `advance()` (rule C — all control-plane wire ops stay async). This + // re-registers the fd with the reactor; a no-op when already async. A revert + // failure poisons the connection. + let mut client = match client.into_async() { + Ok(client) => client, + Err(e) => { + error!(%e, "SQLMoreResults: reverting sync cursor to async failed — connection lost"); + if let Ok(mut stmt_state) = stmt.inner.lock() { + reset_cursor_state(&mut stmt_state); + post_tds_error(&mut stmt_state, &e, SQLSTATE_HY000); + } + if let Ok(mut dbc_state) = dbc.inner.lock() + && dbc_state.active_stmt == Some(statement_handle) + { + dbc_state.active_stmt = None; + } + return SQL_ERROR; + } + }; + match dbc.runtime.block_on(client.advance()) { Ok(StatementResult::Rows) => { // Positioned on a new row-returning result set. Refresh metadata, @@ -107,21 +128,25 @@ fn sql_more_results_safe(statement_handle: SqlHandle, stmt: &StmtHandle) -> SqlR let Ok(mut stmt_state) = stmt.inner.lock() else { error!("SQLMoreResults: stmt mutex poisoned advancing result set"); if let Ok(mut ds) = dbc.inner.lock() { - ds.client = Some(client); + ds.store_async(client); } return SQL_ERROR; }; stmt_state.column_metadata = metadata; // Refresh the count for the newly-positioned result set (-1 for a SELECT). stmt_state.row_count = client.last_rows_affected(); - stmt_state.current_row = None; + stmt_state.reset_fetch_state(); // Drain INFO only after the lock is held. let info_messages = client.take_info_messages(); let has_server_info = post_tds_info_messages(&mut stmt_state, &info_messages); drop(stmt_state); - if let Ok(mut dbc_state) = dbc.inner.lock() { - dbc_state.client = Some(client); - // active_stmt remains set — cursor still open on this statement. + // Re-flip to the sync fetch edge for this result set (rule C: the + // advance above ran async; fetch runs sync when eligible). Metadata is + // already snapshotted, so DescribeCol/NumResultCols are unaffected. + if let Err(rc) = + super::exec_common::flip_to_fetch_edge(dbc, stmt, statement_handle, client) + { + return rc; } debug!("SQLMoreResults: advanced to next result set"); if has_server_info { @@ -139,7 +164,7 @@ fn sql_more_results_safe(statement_handle: SqlHandle, stmt: &StmtHandle) -> SqlR let Ok(mut stmt_state) = stmt.inner.lock() else { error!("SQLMoreResults: stmt mutex poisoned on no-row result"); if let Ok(mut ds) = dbc.inner.lock() { - ds.client = Some(client); + ds.store_async(client); } return SQL_ERROR; }; @@ -147,12 +172,12 @@ fn sql_more_results_safe(statement_handle: SqlHandle, stmt: &StmtHandle) -> SqlR // Surface this no-row statement's own affected-row count for // SQLRowCount now that we are positioned on it. stmt_state.row_count = client.last_rows_affected(); - stmt_state.current_row = None; + stmt_state.reset_fetch_state(); let info_messages = client.take_info_messages(); let has_server_info = post_tds_info_messages(&mut stmt_state, &info_messages); drop(stmt_state); if let Ok(mut dbc_state) = dbc.inner.lock() { - dbc_state.client = Some(client); + dbc_state.store_async(client); // active_stmt remains set — still positioned on this statement. } debug!("SQLMoreResults: advanced to a no-row statement result"); @@ -167,7 +192,7 @@ fn sql_more_results_safe(statement_handle: SqlHandle, stmt: &StmtHandle) -> SqlR let Ok(mut stmt_state) = stmt.inner.lock() else { error!("SQLMoreResults: stmt mutex poisoned at batch end"); if let Ok(mut ds) = dbc.inner.lock() { - ds.client = Some(client); + ds.store_async(client); if ds.active_stmt == Some(statement_handle) { ds.active_stmt = None; } @@ -186,7 +211,7 @@ fn sql_more_results_safe(statement_handle: SqlHandle, stmt: &StmtHandle) -> SqlR // TODO: surface output-param availability here once output // params land. if let Ok(mut dbc_state) = dbc.inner.lock() { - dbc_state.client = Some(client); + dbc_state.store_async(client); if dbc_state.active_stmt == Some(statement_handle) { dbc_state.active_stmt = None; } @@ -204,7 +229,7 @@ fn sql_more_results_safe(statement_handle: SqlHandle, stmt: &StmtHandle) -> SqlR post_tds_info_messages(&mut stmt_state, &info_messages); } if let Ok(mut dbc_state) = dbc.inner.lock() { - dbc_state.client = Some(client); + dbc_state.store_async(client); if dbc_state.active_stmt == Some(statement_handle) { dbc_state.active_stmt = None; } @@ -222,7 +247,7 @@ mod tests { use super::*; use crate::api::odbc_types::{SQL_NO_DATA, SQL_SUCCESS, SQL_SUCCESS_WITH_INFO}; use crate::api::sqlstate::ERR_NO_ACTIVE_TDS_CLIENT; - use crate::handles::dbc::DbcHandle; + use crate::handles::dbc::{DbcClient, DbcHandle}; use crate::test_support::TestHandles; use mssql_tds::test_client_support::{ ScriptedToken, col_metadata_empty, done_more, done_no_more, info, tds_client_from_tokens, @@ -246,7 +271,7 @@ mod tests { } { let mut ds = dbc.inner.lock().unwrap(); - ds.client = Some(client); + ds.client = Some(DbcClient::Async(client)); ds.active_stmt = Some(h.stmt); } first diff --git a/mssql-odbc/src/api/prepare.rs b/mssql-odbc/src/api/prepare.rs index 74873673..5396571c 100644 --- a/mssql-odbc/src/api/prepare.rs +++ b/mssql-odbc/src/api/prepare.rs @@ -114,7 +114,7 @@ fn sql_prepare_w_safe(stmt: &StmtHandle, sql: String) -> SqlReturn { stmt_state.prepared_sql = Some(sql); stmt_state.orphan_prepared_handle(); stmt_state.column_metadata.clear(); - stmt_state.current_row = None; + stmt_state.reset_fetch_state(); stmt_state.clear_state(STMT_STATE_EXEC_CONTEXT); stmt_state.set_state(STMT_STATE_PREPARED); diff --git a/mssql-odbc/src/api/sync_fetch_tests.rs b/mssql-odbc/src/api/sync_fetch_tests.rs new file mode 100644 index 00000000..61f17ea4 --- /dev/null +++ b/mssql-odbc/src/api/sync_fetch_tests.rs @@ -0,0 +1,413 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +//! Integration tests for the reactor-free sync fetch edge, driven end-to-end +//! through the real ODBC entry points (`SQLExecDirectW` / `SQLFetch` / +//! `SQLMoreResults`) against a live `mssql-mock-tds` peer over TCP. +//! +//! The mock speaks TDS over a raw `TcpStream`, so the connection is +//! **sync-eligible**: `finish_execute` flips it to `TdsSyncClient` and `SQLFetch` +//! serves rows off the blocking socket with no tokio reactor. Every sync fetch +//! is checked byte-identical against an all-async oracle run over the same mock +//! result set — the differential parity the whole rewire rests on. +//! +//! The scripted-token unit tests elsewhere in the crate can only exercise the +//! **async fallback** arm (their transport reports `NotEligible`); these tests +//! are the crate's only coverage of the live sync arm, so they live in-crate +//! (the driver is a `cdylib`; an external test crate could neither link it nor +//! reach the `pub(crate)` entry points and handle state they drive). + +use std::sync::mpsc; +use std::thread::JoinHandle; + +use mssql_mock_tds::query_response::{ + ColumnDefinition, ColumnValue, MidStreamError, Row, SqlDataType, +}; +use mssql_mock_tds::{MockTdsServer, QueryResponse}; +use mssql_tds::connection::client_context::ClientContext; +use mssql_tds::connection::tds_client::{ResultSet, StatementResult, TdsClient}; +use mssql_tds::connection_provider::tds_connection_provider::TdsConnectionProvider; +use mssql_tds::core::{EncryptionOptions, EncryptionSetting}; +use mssql_tds::datatypes::column_values::ColumnValues; +use tokio::sync::oneshot; + +use crate::api::exec_direct::sql_exec_direct_w; +use crate::api::fetch::sql_fetch; +use crate::api::more_results::sql_more_results; +use crate::api::odbc_types::{ + SQL_ERROR, SQL_NO_DATA, SQL_NTS, SQL_SUCCESS, SQL_SUCCESS_WITH_INFO, SqlHandle, SqlReturn, +}; +use crate::handles::dbc::{DbcClient, DbcHandle}; +use crate::handles::{StmtHandle, handle_from_raw}; +use crate::test_support::TestHandles; + +/// The mock's native error code for the injected mid-stream failure. +const MOCK_ERROR_NUMBER: i32 = 50_000; + +/// A mock server bound on its own thread + runtime; shut down on drop. The +/// dedicated thread means the sync client's blocking reads on the test thread +/// never starve the server — the same decoupling a real remote peer provides. +struct TestServer { + addr: std::net::SocketAddr, + shutdown: Option>, + thread: Option>, +} + +impl Drop for TestServer { + fn drop(&mut self) { + if let Some(tx) = self.shutdown.take() { + let _ = tx.send(()); + } + if let Some(handle) = self.thread.take() { + let _ = handle.join(); + } + } +} + +fn start_server(query: &str, response: QueryResponse) -> TestServer { + let query = query.to_string(); + let (addr_tx, addr_rx) = mpsc::channel(); + let (shutdown_tx, shutdown_rx) = oneshot::channel(); + let thread = std::thread::spawn(move || { + let rt = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("server runtime builds"); + rt.block_on(async move { + let server = MockTdsServer::new("127.0.0.1:0") + .await + .expect("mock server binds"); + let addr = server.local_addr(); + { + let registry = server.query_registry(); + registry.lock().await.register(query, response); + } + addr_tx.send(addr).expect("addr channel open"); + let _ = server.run_with_shutdown(shutdown_rx).await; + }); + }); + let addr = addr_rx.recv().expect("server reports its address"); + TestServer { + addr, + shutdown: Some(shutdown_tx), + thread: Some(thread), + } +} + +/// Connects a real async client to the mock over plaintext raw TCP (encryption +/// off), so the resulting connection is sync-eligible. +async fn connect(addr: std::net::SocketAddr) -> TdsClient { + let datasource = format!("tcp:{},{}", addr.ip(), addr.port()); + let mut context = ClientContext::default(); + context.user_name = "sa".to_string(); + context.password = "test-password".to_string(); + context.database = "master".to_string(); + context.encryption_options = EncryptionOptions { + mode: EncryptionSetting::PreferOff, + trust_server_certificate: true, + host_name_in_cert: None, + server_certificate: None, + }; + TdsConnectionProvider {} + .create_client(context, &datasource, None) + .await + .expect("client connects to mock server") +} + +/// A `(id INT, label NVARCHAR)` result set. Varying string lengths make row +/// byte-sizes non-uniform, matching the L4 sync harness. +fn make_response(row_count: usize) -> QueryResponse { + let columns = vec![ + ColumnDefinition::new("id", SqlDataType::Int), + ColumnDefinition::new("label", SqlDataType::NVarChar), + ]; + let rows = (0..row_count) + .map(|i| { + Row::new(vec![ + ColumnValue::Int(i as i32), + ColumnValue::NVarChar(format!("row-{i}-{}", "x".repeat(i % 7))), + ]) + }) + .collect(); + QueryResponse::new(columns, rows) +} + +/// Same shape, but the server emits an ERROR token after `after_rows` rows then +/// a terminal DONE — exercising the fetch-time error/drain path. +fn make_error_response(row_count: usize, after_rows: usize) -> QueryResponse { + make_response(row_count).with_error_after(MidStreamError { + after_rows, + number: MOCK_ERROR_NUMBER as u32, + state: 1, + severity: 16, + message: "mid-stream boom".to_string(), + drain_info: Vec::new(), + }) +} + +/// Connects to `addr` on the DBC's runtime and stores the client as the active +/// async connection, leaving the connection idle (claimable by execute). +fn attach_client(h: &TestHandles, addr: std::net::SocketAddr) { + h.mark_dbc_connected(); + let dbc = unsafe { handle_from_raw::(h.dbc) }; + let client = dbc.runtime.block_on(connect(addr)); + let mut ds = dbc.inner.lock().unwrap(); + ds.client = Some(DbcClient::Async(client)); + ds.active_stmt = None; +} + +/// Runs `SQLExecDirectW` with `query` (UTF-16, NUL-terminated) and returns its +/// return code. +fn exec_direct(stmt_handle: SqlHandle, query: &str) -> SqlReturn { + let sql: Vec = query.encode_utf16().chain(std::iter::once(0)).collect(); + unsafe { sql_exec_direct_w(stmt_handle, sql.as_ptr(), SQL_NTS) } +} + +/// True when the DBC currently holds the reactor-free sync fetch client. +fn dbc_is_sync(h: &TestHandles) -> bool { + let dbc = unsafe { handle_from_raw::(h.dbc) }; + let ds = dbc.inner.lock().unwrap(); + matches!(ds.client, Some(DbcClient::Sync(_))) +} + +/// Drains the current result set via `SQLFetch`, cloning each `current_row`. +/// Asserts every fetch is a clean success until `SQL_NO_DATA`. +fn fetch_all_rows(stmt_handle: SqlHandle, stmt: &StmtHandle) -> Vec> { + let mut rows = Vec::new(); + loop { + let rc = unsafe { sql_fetch(stmt_handle) }; + if rc == SQL_NO_DATA { + break; + } + assert!( + rc == SQL_SUCCESS || rc == SQL_SUCCESS_WITH_INFO, + "unexpected SQLFetch return: {rc}" + ); + let row = stmt + .inner + .lock() + .unwrap() + .current_row + .clone() + .expect("current_row set on success"); + rows.push(row); + } + rows +} + +/// The all-async oracle for a single-set query: every row via `next_row().await`. +async fn async_oracle(addr: std::net::SocketAddr, query: &str) -> Vec> { + let mut client = connect(addr).await; + client + .execute(query.to_string(), ()) + .await + .expect("oracle executes"); + let mut rows = Vec::new(); + if client.on_rows() { + while let Some(row) = client.next_row().await.expect("oracle next_row") { + rows.push(row); + } + } + client.close_query().await.expect("oracle closes query"); + rows +} + +/// The all-async oracle for a multi-result-set batch: collects each set's rows, +/// walking `advance()` across the boundaries. +async fn async_oracle_sets(addr: std::net::SocketAddr, query: &str) -> Vec>> { + let mut client = connect(addr).await; + let mut result = client + .execute(query.to_string(), ()) + .await + .expect("oracle executes"); + let mut sets = Vec::new(); + loop { + match result { + StatementResult::Rows => { + let mut rows = Vec::new(); + while let Some(row) = client.next_row().await.expect("oracle next_row") { + rows.push(row); + } + sets.push(rows); + } + StatementResult::NoRows { .. } => sets.push(Vec::new()), + StatementResult::End => break, + } + result = client.advance().await.expect("oracle advance"); + } + client.close_query().await.expect("oracle closes query"); + sets +} + +/// The all-async error oracle: fetch rows until the mid-stream ERROR, returning +/// the rows read and the surfaced error's native code. +async fn async_oracle_until_error( + addr: std::net::SocketAddr, + query: &str, +) -> (Vec>, String) { + let mut client = connect(addr).await; + client + .execute(query.to_string(), ()) + .await + .expect("oracle executes"); + let mut rows = Vec::new(); + assert!(client.on_rows()); + let err = loop { + match client.next_row().await { + Ok(Some(row)) => rows.push(row), + Ok(None) => panic!("expected a mid-stream error, got a clean end"), + Err(e) => break format!("{e:?}"), + } + }; + (rows, err) +} + +/// (a) Differential: a full sync `SQLFetch` drain reproduces the async oracle +/// byte-identically, and the DBC actually flipped to the sync edge at execute time. +#[test] +fn sync_fetch_matches_async_oracle() { + const QUERY: &str = "SELECT SYNC ROWS"; + let server = start_server(QUERY, make_response(100)); + + let h = TestHandles::with_env_dbc_stmt(); + let dbc = unsafe { handle_from_raw::(h.dbc) }; + let expected = dbc.runtime.block_on(async_oracle(server.addr, QUERY)); + assert!(!expected.is_empty()); + + attach_client(&h, server.addr); + let rc = exec_direct(h.stmt, QUERY); + assert_eq!(rc, SQL_SUCCESS, "SQLExecDirectW should position on rows"); + assert!( + dbc_is_sync(&h), + "raw-TCP connection must flip to the sync fetch edge at execute time" + ); + + let stmt = unsafe { handle_from_raw::(h.stmt) }; + let actual = fetch_all_rows(h.stmt, stmt); + assert_eq!(actual, expected); +} + +/// (b) The sync fetch serves every good row then surfaces the mid-stream error on +/// the `SQLFetch` that hits the failing row — rows and error in the correct order. +#[test] +fn sync_fetch_surfaces_midstream_error_after_good_rows() { + const QUERY: &str = "SELECT SYNC ERR ROWS"; + // 50 rows, error after 37 — each SQLFetch reads one row, so the error surfaces + // on the fetch that hits row 38, after the 37 good rows have been served. + let server = start_server(QUERY, make_error_response(50, 37)); + + let h = TestHandles::with_env_dbc_stmt(); + let dbc = unsafe { handle_from_raw::(h.dbc) }; + let (expected_rows, oracle_err) = dbc + .runtime + .block_on(async_oracle_until_error(server.addr, QUERY)); + assert_eq!(expected_rows.len(), 37); + assert!(oracle_err.contains("mid-stream boom") || oracle_err.contains("50000")); + + attach_client(&h, server.addr); + let rc = exec_direct(h.stmt, QUERY); + assert_eq!(rc, SQL_SUCCESS); + assert!(dbc_is_sync(&h)); + + let stmt = unsafe { handle_from_raw::(h.stmt) }; + + let mut actual = Vec::new(); + let final_rc = loop { + let rc = unsafe { sql_fetch(h.stmt) }; + if rc == SQL_ERROR || rc == SQL_NO_DATA { + break rc; + } + assert!(rc == SQL_SUCCESS || rc == SQL_SUCCESS_WITH_INFO); + let row = stmt.inner.lock().unwrap().current_row.clone().unwrap(); + actual.push(row); + }; + + assert_eq!(final_rc, SQL_ERROR, "the mid-stream error must surface"); + assert_eq!( + actual, expected_rows, + "rows before the error must be preserved" + ); + let ss = stmt.inner.lock().unwrap(); + assert!( + ss.diag_records + .iter() + .any(|d| d.native_error == MOCK_ERROR_NUMBER), + "the server error must be surfaced as a diagnostic" + ); +} + +/// (c) `SQLMoreResults` interleave: sync-fetch the first set, cross the boundary +/// (revert to async, `advance()`, re-flip to sync), sync-fetch the second set. +/// Both sets must match the async oracle. +#[test] +fn sync_fetch_interleaves_across_more_results() { + const QUERY: &str = "SELECT SYNC TWO SETS"; + let response = make_response(40).with_additional_result_set(make_response(25)); + let server = start_server(QUERY, response); + + let h = TestHandles::with_env_dbc_stmt(); + let dbc = unsafe { handle_from_raw::(h.dbc) }; + let expected = dbc.runtime.block_on(async_oracle_sets(server.addr, QUERY)); + assert_eq!(expected.len(), 2); + assert_eq!(expected[0].len(), 40); + assert_eq!(expected[1].len(), 25); + + attach_client(&h, server.addr); + let rc = exec_direct(h.stmt, QUERY); + assert_eq!(rc, SQL_SUCCESS); + assert!(dbc_is_sync(&h), "first set flips to sync"); + + let stmt = unsafe { handle_from_raw::(h.stmt) }; + let first = fetch_all_rows(h.stmt, stmt); + assert_eq!(first, expected[0]); + + // Cross the result-set boundary: revert to async, advance, re-flip to sync. + let more = unsafe { sql_more_results(h.stmt) }; + assert_eq!( + more, SQL_SUCCESS, + "should advance onto the second result set" + ); + assert!( + dbc_is_sync(&h), + "second row-returning set must re-flip to sync" + ); + + let second = fetch_all_rows(h.stmt, stmt); + assert_eq!(second, expected[1]); + + // The batch is exhausted: one more advance reports no further results. + let done = unsafe { sql_more_results(h.stmt) }; + assert_eq!(done, SQL_NO_DATA); +} + +/// (d) No-regression: an ineligible (scripted-transport) client hitting the flip +/// seam is returned to the async edge unchanged — `flip_to_fetch_edge` takes the +/// `NotEligible` arm and leaves the DBC holding `DbcClient::Async`, so `SQLFetch` +/// falls back to `block_on`. The whole scripted-token suite only ever exercises +/// this arm; this pins the flip seam's fallback explicitly. +#[test] +fn ineligible_transport_stays_on_async_edge() { + use crate::api::exec_common::flip_to_fetch_edge; + use mssql_tds::test_client_support::{ + col_metadata_empty, done_no_more, tds_client_from_tokens, + }; + + let h = TestHandles::with_env_dbc_stmt(); + h.mark_dbc_connected(); + let dbc = unsafe { handle_from_raw::(h.dbc) }; + let stmt = unsafe { handle_from_raw::(h.stmt) }; + + // The scripted transport reports `NotEligible` from `into_sync`. + let client = tds_client_from_tokens(vec![col_metadata_empty(), done_no_more()]); + let flip = flip_to_fetch_edge(dbc, stmt, h.stmt, client); + assert!( + flip.is_ok(), + "an ineligible transport must not fail the flip" + ); + + let ds = dbc.inner.lock().unwrap(); + assert!( + matches!(ds.client, Some(DbcClient::Async(_))), + "an ineligible transport must remain on the async edge" + ); +} diff --git a/mssql-odbc/src/handles/dbc.rs b/mssql-odbc/src/handles/dbc.rs index 54d14ead..1d2ad10c 100644 --- a/mssql-odbc/src/handles/dbc.rs +++ b/mssql-odbc/src/handles/dbc.rs @@ -5,12 +5,54 @@ use std::ffi::c_void; use std::sync::{Arc, Mutex}; use mssql_tds::connection::tds_client::TdsClient; +use mssql_tds::connection::tds_sync_client::TdsSyncClient; +use mssql_tds::core::{TdsResult, Version}; use tokio::runtime::Runtime; use super::{EnvHandle, HandleType, HasObjectType}; use crate::api::odbc_types::{DEFAULT_PACKET_SIZE, SQL_MODE_READ_WRITE}; use crate::error::{DiagRecord, HasDiagnostics}; +/// The connection's TDS client, held in one of two interchangeable edges. +/// +/// The async [`TdsClient`] drives every control-plane operation (connect, +/// execute, COLMETADATA, advance, close). Once a row-returning result set is +/// open on an eligible (raw-TCP, plaintext) connection, [`finish_execute`] flips +/// it to the reactor-free [`TdsSyncClient`] so `SQLFetch` pulls rows off a +/// blocking socket with no tokio reactor. Result-set boundaries flip back to +/// `Async` (via [`DbcClient::into_async`]) before running the next control-plane +/// op, so the sync edge is only ever live while a cursor is being fetched. +/// +/// TLS / non-raw transports never convert (`SyncConversion::NotEligible`), so +/// they stay on the `Async` edge and fetch through the unchanged `block_on` +/// path — byte-identical to the pre-rewire behaviour. +/// +/// [`finish_execute`]: crate::api::exec_common +pub(crate) enum DbcClient { + /// The async, reactor-driven client. All control-plane work uses this edge. + Async(TdsClient), + /// The reactor-free sync fetch client, live only while a cursor is open on + /// an eligible connection. + Sync(TdsSyncClient), +} + +impl DbcClient { + /// Coerces to the async [`TdsClient`], reverting a live sync cursor back to + /// the tokio reactor. A no-op (never fails) when already `Async`. + /// + /// The revert re-registers the owned socket with the runtime handle captured + /// at `into_sync` time — it performs no network I/O and needs no ambient + /// runtime — so it is safe to call on the bare ODBC thread. Errors only if + /// the connection was poisoned (missing handle / fd re-registration failure), + /// in which case the connection is consumed and closed (known-dead). + pub(crate) fn into_async(self) -> TdsResult { + match self { + DbcClient::Async(client) => Ok(client), + DbcClient::Sync(sync) => sync.into_async(), + } + } +} + /// Connection state machine — tracks whether the DBC is connected. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub(crate) enum ConnectionState { @@ -60,7 +102,13 @@ pub(crate) struct DbcState { /// one statement may hold an open cursor per connection at a time. pub(crate) active_stmt: Option<*mut c_void>, /// Active TDS connection, present only when `connection_state == Connected`. - pub(crate) client: Option, + /// Held as a [`DbcClient`] so the fetch hot path can run on the reactor-free + /// sync edge while control-plane work stays async. + pub(crate) client: Option, + /// Server version negotiated at login, cached at connect time. Reported by + /// `SQLGetInfo(SQL_DBMS_VER)` without touching the live client, so it stays + /// available even while a sync fetch cursor owns the connection. + pub(crate) server_version: Option, /// Pre-connect access token set via `SQL_COPT_SS_ACCESS_TOKEN`. /// Consumed by `SQLDriverConnect` to select `AccessToken` authentication. pub(crate) access_token: Option, @@ -86,7 +134,13 @@ impl std::fmt::Debug for DbcState { .field("connection_state", &self.connection_state) .field("statements", &self.statements) .field("active_stmt", &self.active_stmt) - .field("client", &self.client) + .field( + "client", + &self.client.as_ref().map(|c| match c { + DbcClient::Async(_) => "Async", + DbcClient::Sync(_) => "Sync", + }), + ) .field( "access_token", &self.access_token.as_ref().map(|_| ""), @@ -96,6 +150,15 @@ impl std::fmt::Debug for DbcState { } } +impl DbcState { + /// Stores an async client on the connection (idle or busy). The common + /// restore path: every control-plane op and the async fetch arm return the + /// client through here, wrapping it in [`DbcClient::Async`]. + pub(crate) fn store_async(&mut self, client: TdsClient) { + self.client = Some(DbcClient::Async(client)); + } +} + impl HasDiagnostics for DbcState { fn diag_records(&self) -> &[DiagRecord] { &self.diag_records @@ -117,6 +180,7 @@ impl DbcHandle { statements: Vec::new(), active_stmt: None, client: None, + server_version: None, access_token: None, login_timeout: None, access_mode: SQL_MODE_READ_WRITE, diff --git a/mssql-odbc/src/handles/stmt.rs b/mssql-odbc/src/handles/stmt.rs index 7d9a713f..7e34946a 100644 --- a/mssql-odbc/src/handles/stmt.rs +++ b/mssql-odbc/src/handles/stmt.rs @@ -70,6 +70,7 @@ pub(crate) struct StmtState { /// execute, which flushes any pending drop first). pub(crate) pending_unprepare: Option, /// Current fetched row, populated by SQLFetch for later SQLGetData support. + /// The sync fetch arm recycles this allocation into its row writer. pub(crate) current_row: Option>, /// Rows affected by the last execution, reported by `SQLRowCount`. `-1` /// means "not available" (no statement executed yet, a result-returning @@ -97,6 +98,52 @@ pub(crate) struct StmtState { pub(crate) row_bind_type: SqlULen, /// Statement lifecycle/status flags used for ODBC API state checks. pub(crate) state_flags: u32, + /// Active `SQLGetData` streaming cursor for the current row. Tracks how much + /// of a single column's value has already been handed back across successive + /// `SQLGetData` calls so large (LOB / `max`) values stream in buffer-sized + /// chunks that terminate in `SQL_NO_DATA`, instead of re-returning the same + /// truncated prefix every call. Reset at each `SQLFetch` and cursor close. + pub(crate) getdata: Option, +} + +/// Encoded, ready-to-serve units for an in-progress `SQLGetData` stream. The +/// column value is converted and encoded once when a column becomes active, then +/// served in chunks — keeping chunked retrieval O(n) total rather than +/// re-encoding the whole value on every call. +#[derive(Debug)] +pub(crate) enum GetDataUnits { + /// `SQL_C_CHAR` payload (UTF-8 bytes). + Char(Vec), + /// `SQL_C_WCHAR` payload (UTF-16 code units). + WChar(Vec), + /// SQL `NULL`: one `SQL_NULL_DATA` delivery, then `SQL_NO_DATA`. + Null, +} + +impl GetDataUnits { + /// True when this payload matches the requested C type, so an in-progress + /// stream can continue instead of being rebuilt. `NULL` matches either. + pub(crate) fn matches_wchar(&self, is_wchar: bool) -> bool { + match self { + GetDataUnits::Char(_) => !is_wchar, + GetDataUnits::WChar(_) => is_wchar, + GetDataUnits::Null => true, + } + } +} + +/// Per-column streaming progress for `SQLGetData`. +#[derive(Debug)] +pub(crate) struct GetDataProgress { + /// 1-based column number the cursor is bound to. + pub(crate) column: SqlUSmallInt, + /// Cached encoded payload. + pub(crate) units: GetDataUnits, + /// Number of units already delivered. + pub(crate) offset: usize, + /// Set once the terminal chunk (or the sole NULL/empty delivery) has been + /// served; the next call on this column returns `SQL_NO_DATA`. + pub(crate) exhausted: bool, } impl StmtState { @@ -112,6 +159,14 @@ impl StmtState { self.state_flags &= !mask; } + /// Clears the current row at every result-set boundary (execute, + /// `SQLMoreResults`, cursor close) so a fresh result set never serves a row + /// left over from a prior one. + pub(crate) fn reset_fetch_state(&mut self) { + self.current_row = None; + self.getdata = None; + } + /// Moves the cached `prepared_handle` (if any) into `pending_unprepare` so /// the next execute / exec-direct (or statement free) releases it with /// `sp_unprepare`. Called by re-prepare, rebind, and `SQLExecDirect` when @@ -167,6 +222,7 @@ impl StmtHandle { row_status_ptr: std::ptr::null_mut(), row_bind_type: crate::api::odbc_types::SQL_BIND_BY_COLUMN, state_flags: 0, + getdata: None, }), } }