From 983a8580a87b56b4d6962c0bdec2bc01762cf037 Mon Sep 17 00:00:00 2001 From: saurabh500 <1623701+saurabh500@users.noreply.github.com> Date: Sat, 15 Aug 2026 18:40:30 +0000 Subject: [PATCH] Use sync-first TDS cursor reads in ODBC Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- mssql-odbc/src/api/fetch.rs | 6 +++- mssql-odbc/src/api/get_data.rs | 7 +++-- mssql-odbc/src/api/util.rs | 52 +++++++++++++++++++++++++++++++++- 3 files changed, 61 insertions(+), 4 deletions(-) diff --git a/mssql-odbc/src/api/fetch.rs b/mssql-odbc/src/api/fetch.rs index 020a6a89..77ff774c 100644 --- a/mssql-odbc/src/api/fetch.rs +++ b/mssql-odbc/src/api/fetch.rs @@ -10,6 +10,7 @@ use crate::api::odbc_types::{ SQL_ERROR, SQL_INVALID_HANDLE, SQL_NO_DATA, SQL_SUCCESS, SQL_SUCCESS_WITH_INFO, SqlHandle, SqlReturn, }; +use crate::api::util::resolve_cursor_poll; use crate::error::free_errors; use crate::handles::stmt::STMT_STATE_CURSOR_OPEN; use crate::handles::{HandleType, StmtHandle, handle_from_raw}; @@ -135,7 +136,10 @@ fn fetch_rows_next(statement_handle: SqlHandle, stmt: &StmtHandle) -> SqlReturn // `drain_active_row` in tds_client), so a failure here may originate from // the prior row rather than the new one. Columns of the new row are pulled // lazily by subsequent SQLGetData calls via `read_row_column`. - let fetch_result = dbc.runtime.block_on(client.next_row_cursor()); + let cursor_poll = client.try_next_row_cursor(); + let fetch_result = resolve_cursor_poll(cursor_poll, || { + dbc.runtime.block_on(client.next_row_cursor()) + }); match fetch_result { Ok(true) => { diff --git a/mssql-odbc/src/api/get_data.rs b/mssql-odbc/src/api/get_data.rs index 5f6611be..e078e257 100644 --- a/mssql-odbc/src/api/get_data.rs +++ b/mssql-odbc/src/api/get_data.rs @@ -12,7 +12,7 @@ use super::odbc_types::{ }; use super::sqlstate::*; use crate::api::odbc_types::SqlWChar; -use crate::api::util::{copy_with_nul, write_if_some}; +use crate::api::util::{copy_with_nul, resolve_cursor_poll, write_if_some}; use crate::error::{free_errors, post_sql_error}; use crate::handles::stmt::{ActivePlpStream, STMT_STATE_CURSOR_OPEN}; use crate::handles::{HandleType, StmtHandle, handle_from_raw}; @@ -427,7 +427,10 @@ fn resume_row_to_column( }; let target = column_number - 1; // 0-based - let cursor_result = dbc.runtime.block_on(client.read_row_column(target)); + let cursor_poll = client.try_read_row_column(target); + let cursor_result = resolve_cursor_poll(cursor_poll, || { + dbc.runtime.block_on(client.read_row_column(target)) + }); let Ok(mut dbc_state) = dbc.inner.lock() else { error!("SQLGetData: dbc mutex poisoned after row resume"); diff --git a/mssql-odbc/src/api/util.rs b/mssql-odbc/src/api/util.rs index e42ceb14..86fe9721 100644 --- a/mssql-odbc/src/api/util.rs +++ b/mssql-odbc/src/api/util.rs @@ -4,6 +4,17 @@ use std::slice; use crate::api::odbc_types::{SQL_NTS, SqlSmallInt, SqlWChar}; +use mssql_tds::connection::tds_client::CursorPoll; + +pub(crate) fn resolve_cursor_poll( + poll: Result, E>, + fallback: impl FnOnce() -> Result, +) -> Result { + match poll? { + CursorPoll::Ready(value) => Ok(value), + CursorPoll::Pending => fallback(), + } +} /// Write `value` to `ptr` if non-null. Every ODBC out-parameter pointer may /// legitimately be null (caller opting out of that value), so the @@ -231,8 +242,47 @@ pub(crate) fn rewrite_param_markers(sql: &str) -> (String, usize) { #[cfg(test)] mod tests { - use super::{copy_with_nul, read_utf16, rewrite_param_markers, write_if_some}; + use super::{ + copy_with_nul, read_utf16, resolve_cursor_poll, rewrite_param_markers, write_if_some, + }; use crate::api::odbc_types::{SQL_NTS, SqlWChar}; + use mssql_tds::connection::tds_client::CursorPoll; + + #[test] + fn cursor_ready_skips_fallback() { + let mut calls = 0; + let value = resolve_cursor_poll(Ok::<_, ()>(CursorPoll::Ready(7)), || { + calls += 1; + Ok(9) + }); + + assert_eq!(value, Ok(7)); + assert_eq!(calls, 0); + } + + #[test] + fn cursor_pending_calls_fallback_once() { + let mut calls = 0; + let value = resolve_cursor_poll(Ok::<_, ()>(CursorPoll::Pending), || { + calls += 1; + Ok(9) + }); + + assert_eq!(value, Ok(9)); + assert_eq!(calls, 1); + } + + #[test] + fn cursor_error_skips_fallback() { + let mut calls = 0; + let value = resolve_cursor_poll::(Err("cursor failed"), || { + calls += 1; + Ok(9) + }); + + assert_eq!(value, Err("cursor failed")); + assert_eq!(calls, 0); + } #[test] fn rewrite_no_markers_is_unchanged() {