SQLFetch and SQLGetData update - column wise fetch and incremental PLP support

[shiwanigupta0809]

Description
Aligns mssql-odbc SQLFetch and SQLGetData with ODBC's sparse, request-driven row model, and implements correct PLP (partially-length-prefixed) continuation semantics. Row and column data are now decoded lazily, on demand, rather than materializing the whole row up front.

What changed
Deferred row materialization — SQLFetch now pauses before decoding the first column. A fetched row is positioned but not read until the application requests a specific column.
Column-wise decode in SQLGetData — resumes decoding forward to the requested column instead of relying on a fully materialized row or value. Intervening unread columns are drained transparently.
Direct PLP streaming — large PLP values are streamed straight from the active TDS PLP stream across successive SQLGetData calls, with no ODBC-side cache of the fully decoded value. This keeps memory bounded regardless of value size.
Simplified continuation state — per-statement PLP tracking is reduced to the active column plus its target C type.

Tests
Added SQLGetData e2e coverage and wired it into the e2e CMake target:

Repeated small-buffer SQLGetData calls that stream a PLP value in chunks.
Mixed PLP and non-PLP column access within a row.
Sparse / non-sequential column access patterns.

Related Issues

https://sqlclientdrivers.visualstudio.com/mssql-rs/_workitems/edit/46641

Checklist

• cargo bfmt passes
• cargo bclippy passes
• cargo btest passes
• New/changed functionality has tests
• Public API changes are documented

[shiwanigupta0809]

@copilot resolve the merge conflicts in this pull request

[shiwanigupta0809]

@copilot resolve the merge conflicts in this pull request

[Copilot]

@copilot resolve the merge conflicts in this pull request

Resolved in commit a9c2da4 by merging origin/main into this branch and fixing conflicts in mssql-odbc/src/handles/stmt.rs and mssql-odbc/tests/e2e/CMakeLists.txt.

@copilot resolve the merge conflicts in this pull request

Resolved in commit a9c2da4 by merging origin/main into this branch and fixing conflicts in mssql-odbc/src/handles/stmt.rs and mssql-odbc/tests/e2e/CMakeLists.txt.

[github-actions]

📊 Code Coverage Report

🔥 Diff Coverage 21%

🎯 Overall Coverage 89.5%

📦 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 (0.0%): Missing lines 134
• mssql-odbc/src/api/exec_direct.rs (100%)
• mssql-odbc/src/api/execute.rs (100%)
• mssql-odbc/src/api/fetch.rs (0.0%): Missing lines 102,104,118-119,175,189
• mssql-odbc/src/api/get_data.rs (1.0%): Missing lines 117-122,126-129,131-139,142,146-148,151-160,162,165,168-171,173-180,182,185-191,195-204,208-214,218-225,229,237-239,244,266,270,272,280,283,291,298-299,301,304-309,313-315,317-320,323-326,329-330,332-337,339-344,347,350-351,353-355,357-359,361-363,365,367-372,375-385,387-393,396,403-418,421-423,426-437,439,442-443,445-448,450-451,456-457,464-466,471-473,478-479,482-483,485,487-488,490,492,494-496,498-499,504-508,511-512,514-519,521-526,529,532-535,537-539,541-543,545-552,556-567,569-583,585-592,595-597,600-603,607-608,610-611
• mssql-odbc/src/api/more_results.rs (0.0%): Missing lines 115
• mssql-odbc/src/api/prepare.rs (100%)
• mssql-odbc/src/handles/stmt.rs (100%)
• mssql-odbc/src/row/row_writer.rs (57.0%): Missing lines 90-98,102-110,114-158,201
• mssql-tds/src/datatypes/row_writer.rs (100%)
• mssql-tds/src/io/token_stream.rs (25.0%): Missing lines 495-500,510-515
• mssql-tds/src/query/metadata.rs (0.0%): Missing lines 88-90,93,96-98,106

Summary

• Total: 512 lines
• Missing: 401 lines
• Coverage: 21%

mssql-odbc/src/api/close_cursor.rs

  130 
  131 /// Resets cursor state on the statement (cursor is no longer open, metadata cleared).
  132 pub(super) fn reset_cursor_state(stmt_state: &mut crate::handles::stmt::StmtState) {
  133     stmt_state.clear_state(STMT_STATE_CURSOR_OPEN | STMT_STATE_EXEC_CONTEXT);
! 134     stmt_state.reset_row_stream();
  135     stmt_state.column_metadata.clear();
  136     stmt_state.pending_row_counts.clear();
  137 }

---

mssql-odbc/src/api/fetch.rs

   98 
   99         client
  100     };
  101 
! 102     let mut writer = OdbcRowWriter::new();
  103 
! 104     let fetch_result = dbc.runtime.block_on(client.next_row_into(&mut writer));
  105 
  106     match fetch_result {
  107         Ok(true) => {
  108             let Ok(mut stmt_state) = stmt.inner.lock() else {

---

  114                     }
  115                 }
  116                 return SQL_ERROR;
  117             };
! 118             stmt_state.reset_row_stream();
! 119             stmt_state.row_positioned = true; // row positioned, columns streamed on demand
  120             // Drain INFO only after the lock is held so a poisoned mutex cannot
  121             // silently drop the messages.
  122             let info_messages = client.take_info_messages();
  123             let has_server_info = post_tds_info_messages(&mut stmt_state, &info_messages);

---

  171                     ds.client = Some(client);
  172                 }
  173                 return SQL_ERROR;
  174             };
! 175             stmt_state.reset_row_stream();
  176             // Don't clear CURSOR_OPEN here: the cursor stays open until
  177             // SQLMoreResults / SQLCloseCursor / SQLFreeStmt(SQL_CLOSE).
  178             drop(stmt_state);
  179             if let Ok(mut dbc_state) = dbc.inner.lock() {

---

  185         }
  186         Err(e) => {
  187             error!(%e, "SQLFetch: row fetch failed");
  188             if let Ok(mut stmt_state) = stmt.inner.lock() {
! 189                 stmt_state.reset_row_stream();
  190                 stmt_state.clear_state(STMT_STATE_CURSOR_OPEN);
  191                 post_tds_error(&mut stmt_state, &e, SQLSTATE_HY000);
  192                 let info_messages = client.take_info_messages();
  193                 post_tds_info_messages(&mut stmt_state, &info_messages);

---

mssql-odbc/src/api/get_data.rs

  113         post_diag(&mut stmt_state, ERR_INVALID_CURSOR_STATE);
  114         return SQL_ERROR;
  115     }
  116 
! 117     let col_index = usize::from(column_number);
! 118     let metadata_len = stmt_state.column_metadata.len();
! 119     if col_index == 0 || col_index > metadata_len {
! 120         post_diag(&mut stmt_state, ERR_INVALID_DESCRIPTOR_INDEX);
! 121         return SQL_ERROR;
! 122     }
  123 
  124     // Continuation: app is calling SQLGetData again on the same PLP column to
  125     // get the next chunk from the active wire stream.
! 126     if stmt_state
! 127         .active_plp
! 128         .as_ref()
! 129         .is_some_and(|s| s.column == col_index)
  130     {
! 131         drop(stmt_state);
! 132         return stream_active_plp_chunk(
! 133             stmt,
! 134             statement_handle,
! 135             col_index,
! 136             target_type,
! 137             target_value_ptr,
! 138             buffer_length,
! 139             strlen_or_ind_ptr,
  140             false,
  141         );
! 142     }
  143 
  144     // If the app jumps to a different column while a PLP stream was open —
  145     // incorrect usage per the ODBC spec — clear the stale stream state.
! 146     if stmt_state.active_plp.is_some() {
! 147         stmt_state.active_plp = None;
! 148     }
  149 
  150     // Enforce forward-only column access within a row.
! 151     let last_col = stmt_state.current_row_last_col;
! 152     if last_col > 0 {
! 153         if col_index < last_col {
! 154             post_diag(&mut stmt_state, ERR_INVALID_DESCRIPTOR_INDEX);
! 155             return SQL_ERROR;
! 156         }
! 157         if col_index == last_col {
! 158             return SQL_NO_DATA;
! 159         }
! 160     }
  161 
! 162     if !stmt_state.row_positioned {
  163         post_sql_error(&mut stmt_state, SQLSTATE_24000, 0, "No current row");
  164         return SQL_ERROR;
! 165     }
  166 
  167     // If we already captured this column (e.g., prior HYC00 on same column), skip the resume.
! 168     let already_captured = stmt_state
! 169         .last_captured
! 170         .as_ref()
! 171         .is_some_and(|(c, _)| *c == col_index);
  172 
! 173     if already_captured {
! 174         return write_column_as_text(
! 175             &mut stmt_state,
! 176             col_index,
! 177             target_type,
! 178             target_value_ptr,
! 179             buffer_length,
! 180             strlen_or_ind_ptr,
  181         );
! 182     }
  183 
  184     // Resume the decoder to the requested column then write output.
! 185     drop(stmt_state);
! 186     let rc = resume_row_to_column(stmt, statement_handle, col_index);
! 187     if rc != SQL_SUCCESS {
! 188         return rc;
! 189     }
! 190     let Ok(mut reopened_stmt_state) = stmt.inner.lock() else {
! 191         error!("SQLGetData: stmt mutex poisoned after row resume");
  192         return SQL_ERROR;
  193     };
  194     // last_captured is None only when the decoder paused at a PLP column.
! 195     if reopened_stmt_state.last_captured.is_none() && !reopened_stmt_state.current_row_complete {
! 196         drop(reopened_stmt_state);
! 197         return stream_active_plp_chunk(
! 198             stmt,
! 199             statement_handle,
! 200             col_index,
! 201             target_type,
! 202             target_value_ptr,
! 203             buffer_length,
! 204             strlen_or_ind_ptr,
  205             true,
  206         );
  207     }
! 208     write_column_as_text(
! 209         &mut reopened_stmt_state,
! 210         col_index,
! 211         target_type,
! 212         target_value_ptr,
! 213         buffer_length,
! 214         strlen_or_ind_ptr,
  215     )
  216 }
  217 
! 218 fn write_column_as_text(
! 219     stmt_state: &mut crate::handles::stmt::StmtState,
! 220     col_index: usize,
! 221     target_type: SqlSmallInt,
! 222     target_value_ptr: SqlPointer,
! 223     buffer_length: SqlLen,
! 224     strlen_or_ind_ptr: *mut SqlLen,
! 225 ) -> SqlReturn {
  226     // Check target type first — an unsupported type must not consume last_captured so the app can retry.
  227     if target_type != SQL_C_CHAR && target_type != SQL_C_WCHAR {
  228         post_sql_error(
! 229             stmt_state,
  230             SQLSTATE_HYC00,
  231             0,
  232             "Target type not yet implemented",
  233         );

---

  233         );
  234         return SQL_ERROR;
  235     }
  236 
! 237     let Some((_, value)) = stmt_state.last_captured.take() else {
! 238         post_sql_error(
! 239             stmt_state,
  240             SQLSTATE_24000,
  241             0,
  242             "Requested column is not available in the current row",
  243         );
! 244         return SQL_ERROR;
  245     };
  246 
  247     // Output buffer capacity in element units (u8 for SQL_C_CHAR, SqlWChar for
  248     // SQL_C_WCHAR). buffer_length is always in bytes per the ODBC spec.

---

  262             unsafe {
  263                 copy_with_nul(target_value_ptr as *mut u8, buf_elements, &[]);
  264             }
  265         }
! 266         stmt_state.current_row_last_col = col_index;
  267         return SQL_SUCCESS;
  268     }
  269 
! 270     let Some(as_text) = column_value_to_text(&value) else {
  271         post_sql_error(
! 272             stmt_state,
  273             SQLSTATE_HYC00,
  274             0,
  275             "Column type conversion not yet implemented",
  276         );

---

  276         );
  277         return SQL_ERROR;
  278     };
  279 
! 280     let rc = if target_type == SQL_C_WCHAR {
  281         let utf16: Vec<u16> = as_text.encode_utf16().collect();
  282         write_string_result(
! 283             stmt_state,
  284             &utf16,
  285             target_value_ptr as *mut SqlWChar,
  286             buf_elements,
  287             strlen_or_ind_ptr,

---

  287             strlen_or_ind_ptr,
  288         )
  289     } else {
  290         write_string_result(
! 291             stmt_state,
  292             as_text.as_bytes(),
  293             target_value_ptr as *mut u8,
  294             buf_elements,
  295             strlen_or_ind_ptr,

---

  294             buf_elements,
  295             strlen_or_ind_ptr,
  296         )
  297     };
! 298     if rc != SQL_ERROR {
! 299         stmt_state.current_row_last_col = col_index;
  300     }
! 301     rc
  302 }
  303 
! 304 fn resume_row_to_column(
! 305     stmt: &StmtHandle,
! 306     statement_handle: SqlHandle,
! 307     column_number: usize,
! 308 ) -> SqlReturn {
! 309     let dbc = stmt.parent_dbc();
  310 
  311     {
  312         // validate row is positioned before resuming
! 313         let Ok(stmt_state) = stmt.inner.lock() else {
! 314             error!("SQLGetData: stmt mutex poisoned while preparing row resume");
! 315             return SQL_ERROR;
  316         };
! 317         if !stmt_state.row_positioned {
! 318             error!("SQLGetData: no current row for resume");
! 319             return SQL_ERROR;
! 320         }
  321     };
  322 
! 323     let mut client = {
! 324         let Ok(mut dbc_state) = dbc.inner.lock() else {
! 325             error!("SQLGetData: dbc mutex poisoned while resuming row");
! 326             return SQL_ERROR;
  327         };
  328 
! 329         if let Some(busy_stmt) = dbc_state.active_stmt
! 330             && busy_stmt != statement_handle
  331         {
! 332             drop(dbc_state);
! 333             if let Ok(mut stmt_state) = stmt.inner.lock() {
! 334                 post_diag(&mut stmt_state, ERR_CONNECTION_BUSY);
! 335             }
! 336             return SQL_ERROR;
! 337         }
  338 
! 339         let Some(client) = dbc_state.client.take() else {
! 340             drop(dbc_state);
! 341             if let Ok(mut stmt_state) = stmt.inner.lock() {
! 342                 post_diag(&mut stmt_state, ERR_NO_ACTIVE_TDS_CLIENT);
! 343             }
! 344             return SQL_ERROR;
  345         };
  346 
! 347         client
  348     };
  349 
! 350     let mut writer = OdbcRowWriter::new();
! 351     writer.request(column_number - 1); // 0-based
  352 
! 353     let row_read = dbc.runtime.block_on(client.next_row_into(&mut writer));
! 354     let row_complete = writer.end_row_fired();
! 355     let captured = writer.take_captured().map(|v| (column_number, v));
  356 
! 357     let Ok(mut dbc_state) = dbc.inner.lock() else {
! 358         error!("SQLGetData: dbc mutex poisoned after row resume");
! 359         return SQL_ERROR;
  360     };
! 361     dbc_state.client = Some(client);
! 362     dbc_state.active_stmt = Some(statement_handle);
! 363     drop(dbc_state);
  364 
! 365     match row_read {
  366         Ok(true) => {
! 367             if let Ok(mut stmt_state) = stmt.inner.lock() {
! 368                 stmt_state.last_captured = captured;
! 369                 stmt_state.current_row_complete = row_complete;
! 370                 return SQL_SUCCESS;
! 371             }
! 372             SQL_ERROR
  373         }
  374         Ok(false) => {
! 375             if let Ok(mut stmt_state) = stmt.inner.lock() {
! 376                 stmt_state.last_captured = None;
! 377                 stmt_state.current_row_complete = true;
! 378                 post_sql_error(
! 379                     &mut stmt_state,
! 380                     SQLSTATE_24000,
! 381                     0,
! 382                     "Result set exhausted while resuming current row",
! 383                 );
! 384             }
! 385             SQL_ERROR
  386         }
! 387         Err(e) => {
! 388             if let Ok(mut stmt_state) = stmt.inner.lock() {
! 389                 stmt_state.reset_row_stream();
! 390                 stmt_state.clear_state(STMT_STATE_CURSOR_OPEN);
! 391                 post_tds_error(&mut stmt_state, &e, SQLSTATE_HY000);
! 392             }
! 393             SQL_ERROR
  394         }
  395     }
! 396 }
  397 
  398 /// Reads and returns one SQLGetData chunk directly from the active PLP stream.
  399 ///
  400 /// This never buffers the full PLP payload in ODBC-layer memory. The TDS

---

  399 ///
  400 /// This never buffers the full PLP payload in ODBC-layer memory. The TDS
  401 /// client remains the owner of stream state between repeated calls.
  402 #[allow(clippy::too_many_arguments)]
! 403 fn stream_active_plp_chunk(
! 404     stmt: &StmtHandle,
! 405     statement_handle: SqlHandle,
! 406     col_index: usize,
! 407     target_type: SqlSmallInt,
! 408     target_value_ptr: SqlPointer,
! 409     buffer_length: SqlLen,
! 410     strlen_or_ind_ptr: *mut SqlLen,
! 411     starting_new_stream: bool,
! 412 ) -> SqlReturn {
! 413     if target_type != SQL_C_CHAR && target_type != SQL_C_WCHAR {
! 414         if let Ok(mut s) = stmt.inner.lock() {
! 415             post_sql_error(&mut s, SQLSTATE_HYC00, 0, "Target type not yet implemented");
! 416         }
! 417         return SQL_ERROR;
! 418     }
  419 
  420     {
! 421         let Ok(mut stmt_state) = stmt.inner.lock() else {
! 422             error!("SQLGetData: stmt mutex poisoned while preparing PLP stream read");
! 423             return SQL_ERROR;
  424         };
  425 
! 426         if starting_new_stream {
! 427             let (enc_unicode, enc_binary) = stmt_state
! 428                 .column_metadata
! 429                 .get(col_index - 1)
! 430                 .map(|m| (m.is_unicode_text(), m.is_binary_type()))
! 431                 .unwrap_or((false, false));
! 432             stmt_state.active_plp = Some(ActivePlpStream {
! 433                 column: col_index,
! 434                 encoding: if enc_binary {
! 435                     PlpEncoding::Binary
! 436                 } else if enc_unicode {
! 437                     PlpEncoding::Utf16Text
  438                 } else {
! 439                     PlpEncoding::SingleByteText
  440                 },
  441             });
! 442             stmt_state.current_row_last_col = col_index;
! 443         }
  444 
! 445         if stmt_state
! 446             .active_plp
! 447             .as_ref()
! 448             .is_none_or(|s| s.column != col_index)
  449         {
! 450             post_sql_error(
! 451                 &mut stmt_state,
  452                 SQLSTATE_24000,
  453                 0,
  454                 "No active PLP stream for this column",
  455             );
! 456             return SQL_ERROR;
! 457         }
  458 
  459         // Supported text deliveries: SQL_C_WCHAR for nvarchar(max)/xml
  460         // (UTF-16LE) and SQL_C_CHAR for either varchar(max) (single byte) or
  461         // nvarchar(max) (UTF-16LE transcoded to UTF-8). Binary columns and the

---

  460         // (UTF-16LE) and SQL_C_CHAR for either varchar(max) (single byte) or
  461         // nvarchar(max) (UTF-16LE transcoded to UTF-8). Binary columns and the
  462         // varchar->SQL_C_WCHAR widening are not yet implemented; they return
  463         // HYC00 and are deferred to a follow-up change.
! 464         let encoding = stmt_state.active_plp.as_ref().map(|s| s.encoding);
! 465         let compatible = matches!(
! 466             (target_type, encoding),
  467             (SQL_C_WCHAR, Some(PlpEncoding::Utf16Text))
  468                 | (SQL_C_CHAR, Some(PlpEncoding::SingleByteText))
  469                 | (SQL_C_CHAR, Some(PlpEncoding::Utf16Text))
  470         );
! 471         if !compatible {
! 472             post_sql_error(
! 473                 &mut stmt_state,
  474                 SQLSTATE_HYC00,
  475                 0,
  476                 "Target type not yet implemented for this column",
  477             );
! 478             return SQL_ERROR;
! 479         }
  480     }
  481 
! 482     let payload_capacity = if target_type == SQL_C_WCHAR {
! 483         (buffer_length as usize).saturating_sub(std::mem::size_of::<SqlWChar>())
  484     } else {
! 485         (buffer_length as usize).saturating_sub(1)
  486     };
! 487     let max_read = if target_type == SQL_C_WCHAR {
! 488         payload_capacity & !1
  489     } else {
! 490         payload_capacity
  491     };
! 492     let mut payload = vec![0u8; max_read];
  493 
! 494     let is_unicode_plp = {
! 495         let Ok(ss) = stmt.inner.lock() else {
! 496             return SQL_ERROR;
  497         };
! 498         matches!(
! 499             ss.active_plp.as_ref().map(|s| &s.encoding),
  500             Some(PlpEncoding::Utf16Text)
  501         )
  502     };
  503 
! 504     let dbc = stmt.parent_dbc();
! 505     let mut client = {
! 506         let Ok(mut dbc_state) = dbc.inner.lock() else {
! 507             error!("SQLGetData: dbc mutex poisoned while reading PLP stream");
! 508             return SQL_ERROR;
  509         };
  510 
! 511         if let Some(busy_stmt) = dbc_state.active_stmt
! 512             && busy_stmt != statement_handle
  513         {
! 514             drop(dbc_state);
! 515             if let Ok(mut s) = stmt.inner.lock() {
! 516                 post_diag(&mut s, ERR_CONNECTION_BUSY);
! 517             }
! 518             return SQL_ERROR;
! 519         }
  520 
! 521         let Some(client) = dbc_state.client.take() else {
! 522             drop(dbc_state);
! 523             if let Ok(mut s) = stmt.inner.lock() {
! 524                 post_diag(&mut s, ERR_NO_ACTIVE_TDS_CLIENT);
! 525             }
! 526             return SQL_ERROR;
  527         };
  528 
! 529         client
  530     };
  531 
! 532     let read_result = dbc
! 533         .runtime
! 534         .block_on(client.read_active_plp_bytes(&mut payload));
! 535     let reached_end = client.active_plp_reached_end();
  536 
! 537     let Ok(mut dbc_state) = dbc.inner.lock() else {
! 538         error!("SQLGetData: dbc mutex poisoned after PLP read");
! 539         return SQL_ERROR;
  540     };
! 541     dbc_state.client = Some(client);
! 542     dbc_state.active_stmt = Some(statement_handle);
! 543     drop(dbc_state);
  544 
! 545     let read = match read_result {
! 546         Ok(n) => n,
! 547         Err(e) => {
! 548             if let Ok(mut s) = stmt.inner.lock() {
! 549                 s.clear_state(STMT_STATE_CURSOR_OPEN);
! 550                 post_tds_error(&mut s, &e, SQLSTATE_HY000);
! 551             }
! 552             return SQL_ERROR;
  553         }
  554     };
  555 
! 556     if target_type == SQL_C_WCHAR {
! 557         let usable = read & !1;
! 558         let units: Vec<u16> = payload[..usable]
! 559             .chunks_exact(2)
! 560             .map(|pair| u16::from_le_bytes([pair[0], pair[1]]))
! 561             .collect();
! 562         let buf_elements = (buffer_length as usize) / std::mem::size_of::<SqlWChar>();
! 563         unsafe {
! 564             copy_with_nul(target_value_ptr as *mut SqlWChar, buf_elements, &units);
! 565             write_if_some(strlen_or_ind_ptr, usable as SqlLen);
! 566         }
! 567     } else if is_unicode_plp {
  568         // NVARCHAR PLP wire bytes are UTF-16LE; convert to UTF-8 for SQL_C_CHAR.
! 569         let usable = read & !1;
! 570         let units: Vec<u16> = payload[..usable]
! 571             .chunks_exact(2)
! 572             .map(|b| u16::from_le_bytes([b[0], b[1]]))
! 573             .collect();
! 574         let utf8 = String::from_utf16_lossy(&units);
! 575         let utf8_bytes = utf8.as_bytes();
! 576         unsafe {
! 577             copy_with_nul(
! 578                 target_value_ptr as *mut u8,
! 579                 buffer_length as usize,
! 580                 utf8_bytes,
! 581             );
! 582             write_if_some(strlen_or_ind_ptr, utf8_bytes.len() as SqlLen);
! 583         }
  584     } else {
! 585         unsafe {
! 586             copy_with_nul(
! 587                 target_value_ptr as *mut u8,
! 588                 buffer_length as usize,
! 589                 &payload[..read],
! 590             );
! 591             write_if_some(strlen_or_ind_ptr, read as SqlLen);
! 592         }
  593     }
  594 
! 595     let Ok(mut stmt_state) = stmt.inner.lock() else {
! 596         error!("SQLGetData: stmt mutex poisoned while finalizing PLP stream read");
! 597         return SQL_ERROR;
  598     };
  599 
! 600     if reached_end {
! 601         stmt_state.active_plp = None;
! 602         return SQL_SUCCESS;
! 603     }
  604 
  605     // active_plp already holds this column's stream state; leave it in place so
  606     // the next SQLGetData call continues from where this one stopped.
! 607     unsafe { write_if_some(strlen_or_ind_ptr, SQL_NO_TOTAL) };
! 608     post_diag(&mut stmt_state, ERR_STRING_RIGHT_TRUNCATION);
  609 
! 610     SQL_SUCCESS_WITH_INFO
! 611 }
  612 fn write_string_result<T: Copy + Default>(
  613     stmt_state: &mut crate::handles::stmt::StmtState,
  614     src: &[T],
  615     target_value_ptr: *mut T,

---

mssql-odbc/src/api/more_results.rs

  111                 }
  112                 return SQL_ERROR;
  113             };
  114             stmt_state.column_metadata = metadata;
! 115             stmt_state.reset_row_stream();
  116             // Refresh the count for the newly-positioned result set (-1 for a SELECT).
  117             stmt_state.row_count = client.last_rows_affected();
  118             // Drain INFO only after the lock is held.
  119             let info_messages = client.take_info_messages();

---

mssql-odbc/src/row/row_writer.rs

   86 
   87     fn write_null(&mut self, col: usize) {
   88         self.capture(col, ColumnValues::Null);
   89     }
!  90     fn write_bool(&mut self, col: usize, val: bool) {
!  91         self.capture(col, ColumnValues::Bit(val));
!  92     }
!  93     fn write_u8(&mut self, col: usize, val: u8) {
!  94         self.capture(col, ColumnValues::TinyInt(val));
!  95     }
!  96     fn write_i16(&mut self, col: usize, val: i16) {
!  97         self.capture(col, ColumnValues::SmallInt(val));
!  98     }
   99     fn write_i32(&mut self, col: usize, val: i32) {
  100         self.capture(col, ColumnValues::Int(val));
  101     }
! 102     fn write_i64(&mut self, col: usize, val: i64) {
! 103         self.capture(col, ColumnValues::BigInt(val));
! 104     }
! 105     fn write_f32(&mut self, col: usize, val: f32) {
! 106         self.capture(col, ColumnValues::Real(val));
! 107     }
! 108     fn write_f64(&mut self, col: usize, val: f64) {
! 109         self.capture(col, ColumnValues::Float(val));
! 110     }
  111     fn write_string(&mut self, col: usize, val: SqlString) {
  112         self.capture(col, ColumnValues::String(val));
  113     }
! 114     fn write_bytes(&mut self, col: usize, val: Vec<u8>) {
! 115         self.capture(col, ColumnValues::Bytes(val));
! 116     }
! 117     fn write_decimal(&mut self, col: usize, val: DecimalParts) {
! 118         self.capture(col, ColumnValues::Decimal(val));
! 119     }
! 120     fn write_numeric(&mut self, col: usize, val: DecimalParts) {
! 121         self.capture(col, ColumnValues::Numeric(val));
! 122     }
! 123     fn write_date(&mut self, col: usize, val: SqlDate) {
! 124         self.capture(col, ColumnValues::Date(val));
! 125     }
! 126     fn write_time(&mut self, col: usize, val: SqlTime) {
! 127         self.capture(col, ColumnValues::Time(val));
! 128     }
! 129     fn write_datetime(&mut self, col: usize, val: SqlDateTime) {
! 130         self.capture(col, ColumnValues::DateTime(val));
! 131     }
! 132     fn write_smalldatetime(&mut self, col: usize, val: SqlSmallDateTime) {
! 133         self.capture(col, ColumnValues::SmallDateTime(val));
! 134     }
! 135     fn write_datetime2(&mut self, col: usize, val: SqlDateTime2) {
! 136         self.capture(col, ColumnValues::DateTime2(val));
! 137     }
! 138     fn write_datetimeoffset(&mut self, col: usize, val: SqlDateTimeOffset) {
! 139         self.capture(col, ColumnValues::DateTimeOffset(val));
! 140     }
! 141     fn write_money(&mut self, col: usize, val: SqlMoney) {
! 142         self.capture(col, ColumnValues::Money(val));
! 143     }
! 144     fn write_smallmoney(&mut self, col: usize, val: SqlSmallMoney) {
! 145         self.capture(col, ColumnValues::SmallMoney(val));
! 146     }
! 147     fn write_uuid(&mut self, col: usize, val: Uuid) {
! 148         self.capture(col, ColumnValues::Uuid(val));
! 149     }
! 150     fn write_xml(&mut self, col: usize, val: SqlXml) {
! 151         self.capture(col, ColumnValues::Xml(val));
! 152     }
! 153     fn write_json(&mut self, col: usize, val: SqlJson) {
! 154         self.capture(col, ColumnValues::Json(val));
! 155     }
! 156     fn write_vector(&mut self, col: usize, val: SqlVector) {
! 157         self.capture(col, ColumnValues::Vector(val));
! 158     }
  159 
  160     fn end_row(&mut self) {
  161         self.end_row_fired = true;
  162     }

---

  197         w.write_i32(0, 99);
  198         w.write_string(1, SqlString::from_utf8_string("hello".to_string()));
  199         match w.take_captured() {
  200             Some(ColumnValues::String(s)) => assert_eq!(s.to_utf8_string(), "hello"),
! 201             other => panic!("expected String, got {other:?}"),
  202         }
  203         assert!(w.take_captured().is_none());
  204     }

---

mssql-tds/src/io/token_stream.rs

  491     match token_type {
  492         TokenType::Row => {
  493             let (columns, decryptor) = extract_row_context(context)?;
  494             if writer.pause_before_first_column() {
! 495                 return Ok(RowReadResult::RowPaused(RowPauseState {
! 496                     next_column_index: 0,
! 497                     columns: columns.to_vec(),
! 498                     nbc_null_bitmap: None,
! 499                     decryptor: decryptor.cloned(),
! 500                 }));
  501             }
  502             decode_row_columns(reader, columns, decryptor, 0, writer).await
  503         }
  504         TokenType::NbcRow => {

---

  506             let bitmap_len = columns.len().div_ceil(8);
  507             let mut bitmap = vec![0u8; bitmap_len];
  508             reader.read_bytes(&mut bitmap).await?;
  509             if writer.pause_before_first_column() {
! 510                 return Ok(RowReadResult::RowPaused(RowPauseState {
! 511                     next_column_index: 0,
! 512                     columns: columns.to_vec(),
! 513                     nbc_null_bitmap: Some(bitmap),
! 514                     decryptor: decryptor.cloned(),
! 515                 }));
  516             }
  517             decode_nbcrow_columns(reader, columns, decryptor, &bitmap, 0, writer).await
  518         }
  519         _ => {

---

mssql-tds/src/query/metadata.rs

   84     }
   85 
   86     /// Returns `true` for Unicode character columns (`nvarchar`, `nchar`, `xml`).
   87     /// The wire encoding for these types is UTF-16LE.
!  88     pub fn is_unicode_text(&self) -> bool {
!  89         matches!(
!  90             self.data_type,
   91             TdsDataType::NVarChar | TdsDataType::NChar | TdsDataType::Xml
   92         )
!  93     }
   94 
   95     /// Returns `true` for binary columns (`varbinary`, `binary`, `image`, `udt`, etc.).
!  96     pub fn is_binary_type(&self) -> bool {
!  97         matches!(
!  98             self.data_type,
   99             TdsDataType::BigVarBinary
  100                 | TdsDataType::BigBinary
  101                 | TdsDataType::VarBinary
  102                 | TdsDataType::Binary

---

  102                 | TdsDataType::Binary
  103                 | TdsDataType::Image
  104                 | TdsDataType::Udt
  105         )
! 106     }
  107     /// Returns the scale for decimal/numeric/time types.
  108     ///
  109     /// Returns `Some(scale)` for types that include scale information (e.g., `decimal(18,4)`, `time(7)`),
  110     /// or `None` for types where scale is not applicable.

---

---

🔗 Quick Links

View Azure DevOps Build · Coverage Report

[shiwanigupta0809]

@copilot resolve the merge conflicts in this pull request

[shiwanigupta0809]

@copilot resolve the merge conflicts in this pull request

[shiwanigupta0809]

@copilot resolve the merge conflicts in this pull request

[Copilot]

@copilot resolve the merge conflicts in this pull request

Resolved in commit eb929f9 by merging origin/main into this branch and fixing conflicts in:

• mssql-odbc/src/api/exec_direct.rs
• mssql-odbc/src/api/execute.rs
• mssql-odbc/src/api/more_results.rs
• mssql-odbc/src/handles/stmt.rs
• mssql-odbc/tests/e2e/CMakeLists.txt

@copilot resolve the merge conflicts in this pull request

Resolved in commit eb929f9 by merging origin/main into this branch and fixing conflicts in:

• mssql-odbc/src/api/exec_direct.rs
• mssql-odbc/src/api/execute.rs
• mssql-odbc/src/api/more_results.rs
• mssql-odbc/src/handles/stmt.rs
• mssql-odbc/tests/e2e/CMakeLists.txt

[Copilot]

Pull request overview

Adds deferred column decoding and incremental PLP streaming for ODBC SQLFetch/SQLGetData.

Changes:

• Adds pause-before-column row decoding.
• Streams requested PLP data across calls.
• Adds SQLGetData unit and end-to-end coverage.

Reviewed changes

Copilot reviewed 19 out of 19 changed files in this pull request and generated 5 comments.

Show a summary per file

File	Description
mssql-tds/src/query/metadata.rs	Adds text/binary type classification.
mssql-tds/src/io/token_stream.rs	Pauses before decoding columns.
mssql-tds/src/datatypes/row_writer.rs	Adds initial-pause hook.
mssql-tds/src/connection/tds_client.rs	Drains paused rows when advancing.
mssql-odbc/tests/e2e/tests/get_data_test.cpp	Adds SQLGetData scenarios.
mssql-odbc/tests/e2e/CMakeLists.txt	Registers the new tests.
mssql-odbc/src/row/row_writer.rs	Implements the ODBC row writer.
mssql-odbc/src/row/mod.rs	Exposes row-writing components.
mssql-odbc/src/lib.rs	Registers the row module.
mssql-odbc/src/handles/stmt.rs	Adds row and PLP continuation state.
mssql-odbc/src/api/prepare.rs	Resets streaming state on prepare.
mssql-odbc/src/api/odbc_types.rs	Documents SQL_NO_TOTAL.
mssql-odbc/src/api/more_results.rs	Resets state between result sets.
mssql-odbc/src/api/get_data.rs	Implements request-driven retrieval.
mssql-odbc/src/api/fetch.rs	Defers row materialization.
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/close_cursor.rs	Clears streaming state on close.
mssql-odbc/Cargo.toml	Adds encoding dependencies.

Comments suppressed due to low confidence (5)

mssql-odbc/src/api/get_data.rs:548

• SQL_C_WCHAR is handled as if every PLP source were UTF-16LE. For VARCHAR(MAX)/JSON, payload contains single-byte, collation-encoded data, so grouping pairs with u16::from_le_bytes turns ordinary text into unrelated code points (for example, AB becomes U+4241). Decode the source according to PlpEncoding and the active collation before producing UTF-16 output.

    if target_type == SQL_C_WCHAR {
        let usable = read & !1;
        let units: Vec<u16> = payload[..usable]
            .chunks_exact(2)
            .map(|pair| u16::from_le_bytes([pair[0], pair[1]]))
            .collect();

mssql-odbc/src/api/get_data.rs:568

• This conversion consumes the entire UTF-16 wire chunk before knowing how many UTF-8 bytes fit. An odd read drops one byte via read & !1, and non-ASCII text can expand beyond the output buffer, causing copy_with_nul to discard converted bytes that cannot be recovered on the next call (and potentially split a UTF-8 sequence). Preserve decoder and pending-output state across calls, and only consume source bytes whose converted output can be retained.

    } else if is_unicode_plp {
        // NVARCHAR PLP wire bytes are UTF-16LE; convert to UTF-8 for SQL_C_CHAR.
        let usable = read & !1;
        let units: Vec<u16> = payload[..usable]
            .chunks_exact(2)
            .map(|b| u16::from_le_bytes([b[0], b[1]]))
            .collect();
        let utf8 = String::from_utf16_lossy(&units);
        let utf8_bytes = utf8.as_bytes();
        unsafe {
            copy_with_nul(
                target_value_ptr as *mut u8,
                buffer_length as usize,
                utf8_bytes,
            );

mssql-odbc/src/api/get_data.rs:530

• If the DBC mutex is poisoned here, the client is dropped and the function still finalizes the PLP call as success or success-with-info. This loses the connection while telling the application its read succeeded. Return SQL_ERROR when the client cannot be restored.

    if let Ok(mut dbc_state) = dbc.inner.lock() {
        dbc_state.client = Some(client);
        dbc_state.active_stmt = Some(statement_handle);
    }

mssql-odbc/tests/e2e/tests/get_data_test.cpp:572

• This expected HYC00 is explicitly mssql-odbc-specific; msodbcsql supports more binary conversions, so the parity run should not execute this assertion. Add SKIP_IF_COMPARING_MSODBCSQL() at the start of the test, following the established Phase-1 test pattern.

TEST_F(GetDataLiveTest, VarbinaryMaxToCharReturnsHyc00) {
    ASSERT_SQL_OK(ExecDirect("SELECT CAST(0x41424344 AS VARBINARY(MAX)) AS c1"),

mssql-odbc/src/api/get_data.rs:578

• The single-byte PLP path copies server code-page bytes directly, unlike non-PLP strings, which SqlString::to_utf8_string decodes using the column collation. Thus VARCHAR(MAX) and VARCHAR(n) return different SQL_C_CHAR bytes for the same non-ASCII text. Use client.active_plp_collation() with an incremental decoder (and retain expanded pending output) before writing the application buffer.

    } else {
        unsafe {
            copy_with_nul(
                target_value_ptr as *mut u8,
                buffer_length as usize,
                &payload[..read],
            );
            write_if_some(strlen_or_ind_ptr, read as SqlLen);