Add mssql-tds support for incremental PLP parameter write path

[Shiwani Gupta (shiwanigupta0809)]

Description

Adds a data-at-execution (streamed) PLP parameter write path to mssql-tds for large nvarchar(max) / varchar(max) / varbinary(max) values in sp_executesql. The value is streamed to the server in chunks instead of being materialized up front, mirroring msodbcsql's incremental PLP write. This is the TDS-layer foundation for a later ODBC SQLPutData / SQLParamData. Scope: mssql-tds only.

What this adds
Streaming entry points on TdsClient, mirroring ODBC's begin/put/end:
begin_sp_executesql - sends the RPC prefix (headers, positional args, materialized params) + the first streamed param's header, parks the half-written message, returns NeedData.
write_streamed_chunk - appends one length-prefixed chunk (empty chunks ignored a zero-length header is the terminator). Writes the unknown-length opener lazily on the first chunk.
write_streamed_null - marks the current param SQL NULL instead of streaming chunks (the SQLPutData(SQL_NULL_DATA) case).
end_streamed_param - closes the value (PLP_NULL, terminator, or opener+terminator for an empty value), then opens the next streamed param (NeedData) or finalizes and positions on the response (Done).
RpcParameter::data_at_exec() marker; PacketWriter::suspend/resume to park a half-written message across calls; SqlRpc::serialize_prefix to serialize up to materialized params without finalizing.

Testing
Unit (offline mock): header serialization, non-MAX/encrypted rejection, single/multi-chunk framing, empty-chunk skip, two-param lifecycle, materialized+streamed mix, multi-packet chunk, streamed-NULL framing + ordering guards, usage-error guards, and two fault-injection tests for the mid-stream abort.

E2E (live SQL Server): round-trips for nvarchar/varbinary/varchar(max), two params, mixed, multi-row, materialized NULL, streamed NULL, empty value, many small chunks, and connection reuse.

Follow-up:
Currently streamed params are expected to appear after all materialized params in the parameter list. msodbcsql does not enforce this ordering. Supporting arbitrary interleaving of materialized and streamed params is deferred to a follow-up.

Not covered (deferred)
Always Encrypted on the streamed path (rejected); ODBC SQLPutData / SQLParamData (the next layer).

Related Issues

https://sqlclientdrivers.visualstudio.com/mssql-rs/_workitems/edit/46383/
https://sqlclientdrivers.visualstudio.com/mssql-rs/_backlogs/backlog/mssql-rs%20Team/Features?workitem=46139

Checklist

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

[github-actions]

📊 Code Coverage Report

🔥 Diff Coverage 98%

🎯 Overall Coverage 92.0%

📦 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-tds/src/connection/tds_client.rs (98.0%): Missing lines 1187-1190,1325-1327,1367-1371,1378-1380,1486,1676-1678,8714
• mssql-tds/src/io/packet_writer.rs (100%)
• mssql-tds/src/message/parameters/rpc_parameters.rs (99.3%): Missing lines 483
• mssql-tds/src/message/rpc.rs (100%)

Summary

• Total: 1233 lines
• Missing: 21 lines
• Coverage: 98%

mssql-tds/src/connection/tds_client.rs

  1183             return Ok(StreamedParamStatus::Complete(result));
  1184         }
  1185 
  1186         if self.should_encrypt_parameters() {
! 1187             return Err(UsageError(
! 1188                 "Streamed PLP parameter writes are not supported with Always Encrypted."
! 1189                     .to_string(),
! 1190             ));
  1191         }
  1192 
  1193         let mut declaration_params = materialized_params.clone();
  1194         declaration_params.extend(streamed_params.iter().cloned());

---

  1321                 .await
  1322         }
  1323         .await;
  1324         if let Err(error) = serialization_result {
! 1325             drop(packet_writer);
! 1326             self.abort_streamed_write().await;
! 1327             return Err(error);
  1328         }
  1329         let message = packet_writer.suspend();
  1330 
  1331         self.streamed_write_state = StreamedWriteState::Active(Box::new(StreamedWriteContext {

---

  1363                 "write_streamed_chunk called with no active streamed parameter.".to_string(),
  1364             ));
  1365         }
  1366         if chunk.len() > u32::MAX as usize {
! 1367             return Err(UsageError(format!(
! 1368                 "Streamed PLP chunk length {} exceeds the maximum chunk size of {} bytes.",
! 1369                 chunk.len(),
! 1370                 u32::MAX
! 1371             )));
  1372         }
  1373 
  1374         let ctx = match std::mem::replace(&mut self.streamed_write_state, StreamedWriteState::Idle)
  1375         {

---

  1374         let ctx = match std::mem::replace(&mut self.streamed_write_state, StreamedWriteState::Idle)
  1375         {
  1376             StreamedWriteState::Active(ctx) => ctx,
  1377             StreamedWriteState::Idle => {
! 1378                 return Err(UsageError(
! 1379                     "write_streamed_chunk called with no active streamed parameter.".to_string(),
! 1380                 ));
  1381             }
  1382         };
  1383         let StreamedWriteContext {
  1384             message,

---

  1482         self.transport.mark_known_dead();
  1483         // TODO: Match msodbcsql's state-aware cancellation: discard an unsent
  1484         // request locally, or send EOM | IGNORE and drain DONE after a partial send.
  1485         if let Err(error) = self.transport.close_transport().await {
! 1486             warn!(%error, "Failed to close transport after streamed write abort");
  1487         }
  1488     }
  1489 
  1490     /// Cancels an in-progress streamed PLP write while the client is parked in

---

  1672             Ok(None) => {
  1673                 drop(packet_writer);
  1674                 match self.position_on_first_result().await {
  1675                     Ok(result) => Ok(StreamedParamStatus::Complete(result)),
! 1676                     Err(e) => {
! 1677                         self.abort_streamed_write().await;
! 1678                         Err(e)
  1679                     }
  1680                 }
  1681             }
  1682             // Terminator or next-parameter header write failed mid-message: drop

---

  8710 
  8711     /// Index of the last occurrence of `needle` in `haystack`, if any.
  8712     fn find_last(haystack: &[u8], needle: &[u8]) -> Option<usize> {
  8713         if needle.is_empty() || haystack.len() < needle.len() {
! 8714             return None;
  8715         }
  8716         (0..=haystack.len() - needle.len())
  8717             .rev()
  8718             .find(|&i| &haystack[i..i + needle.len()] == needle)

---

mssql-tds/src/message/parameters/rpc_parameters.rs

  479         packet_writer.write_byte_async(self.options.bits()).await?;
  480 
  481         let value = match &self.value {
  482             RpcValue::Materialized(value) => value,
! 483             RpcValue::Streamed(_) => unreachable!("streamed value handled above"),
  484         };
  485         encoder
  486             .encode_sqlvalue(packet_writer, value, db_collation, self.type_metadata)
  487             .await?;

---

---

🔗 Quick Links

View Azure DevOps Build · Coverage Report

[Copilot]

Pull request overview

Adds incremental PLP parameter streaming to the TDS client for large MAX values.

Changes:

• Adds streamed parameter lifecycle APIs and state management.
• Adds packet suspension/resumption and partial RPC serialization.
• Adds extensive unit and live-server coverage.

Reviewed changes

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

Show a summary per file

File	Description
mssql-tds/src/connection/tds_client.rs	Implements streamed-write APIs and state machine.
mssql-tds/src/io/packet_writer.rs	Adds message suspension and resumption.
mssql-tds/src/message/messages.rs	Adds packet-type debug support.
mssql-tds/src/message/parameters/rpc_parameters.rs	Adds the data-at-execution parameter marker and serialization.
mssql-tds/src/message/rpc.rs	Adds non-finalizing RPC prefix serialization.
mssql-tds/tests/test_client_write_apis.rs	Adds live SQL Server streaming tests.

---

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

[Saurabh Singh (saurabh500)]

Nice piece of work — the suspend/resume primitive is a clean way to park a half-written RPC, and the lazy PLP opener (deferring the length field until the first chunk, so a parameter can still resolve to NULL after its header is on the wire) is the right call. The take-and-re-park state machine consistently distinguishes caller sequencing errors (re-park a clean message, stream stays usable) from wire failures (drop and abort), and the byte-level tests cover framing well.

One blocking issue, plus a few things worth resolving before this goes in.

Blocking

RpcParameter::value() can panic from safe public API. RpcParameter::data_at_exec is pub, but only execute_sp_executesql rejects data-at-exec params. This PR replaces has_open_batch() with command_is_busy() in ~10 entry points — execute, execute_stored_procedure, prepare_statement, sp_execute, sp_prepexec, execute_rpc, bulk load — and that list is effectively the set of paths that also need the is_data_at_exec guard. None of them got it.

With Always Encrypted active, execute_stored_procedure → build_stored_procedure_describe_request → get_sql_name(param.value()) hits the new unreachable!. Without AE it is quieter but worse: serialize emits the parameter header with no value body and desyncs the connection. This PR also routes two more call sites into that accessor (write_encrypted_type_info, and #[cfg(fuzzing)] get_value() — a fuzz run would find it immediately).

Worth noting the hazard was already recognized: build_parameter_list_string matches on RpcValue directly rather than calling value(). Making value() return TdsResult<&SqlType> would let the compiler enumerate the remaining sites instead of relying on spotting them.

Suggestions

• begin_sp_executesql should take ExecuteOptions like every sibling. Beyond consistency, hardcoding current_command_ce_setting = UseConnectionSetting makes streaming unconditionally unavailable on any AE-enabled connection, even when no parameter targets an encrypted column — which is exactly what ResultSetOnly/Disabled exist for.
• Timeout semantics disagree across suspend/resume. remaining_request_timeout resets to the full budget on every chunk, while PacketWriter::start_time is deliberately preserved. A slow producer — the whole point of a SQLPutData-style API — trips the send deadline mid-value and lands in abort_streamed_write, which closes the connection.
• No public way to abandon an in-flight streamed write. Once NeedData is returned, every other command returns ALREADY_EXECUTING_ERROR; the only exits are end_streamed_param (commits a value the caller may not want) or dropping the client. ODBC's DAE flow allows SQLCancel here.
• Dead validation in split_and_validate_streamed_params, and a duplicate predicate (is_streamable_plp is byte-identical to is_data_at_exec).
• Empty-chunk short circuit sits above the null_signaled guard, so write_streamed_chunk(&[]) after write_streamed_null() succeeds while a non-empty chunk errors.

Nits

• StreamedSqlType::sql_name() duplicates strings get_sql_name already produces.
• write_streamed_chunk checks Idle twice — once up front, then again as an unreachable match arm.

Housekeeping

This adds 7 public items (begin_sp_executesql, write_streamed_chunk, write_streamed_null, end_streamed_param, StreamedParamStatus, StreamedSqlType, RpcParameter::data_at_exec) with no CHANGELOG.md entry — the [Unreleased] / ### Added section is where new mssql-tds public API gets recorded. The checklist boxes for "New/changed functionality has tests" and "Public API changes are documented" are also unchecked; the tests clearly exist, so that one is just stale.

[Saurabh Singh (saurabh500)]

Blocking. This unreachable! is reachable from safe public API, so it's a panic, not an invariant.

RpcParameter::data_at_exec is pub, but only execute_sp_executesql rejects data-at-exec params. Nothing stops this:

// AE-enabled connection
client.execute_stored_procedure(
    "dbo.p".into(),
    None,
    Some(vec![RpcParameter::data_at_exec(
        Some("@v".into()), StatusFlags::NONE, StreamedSqlType::VarBinaryMax)]),
    opts,
).await

execute_stored_procedure (tds_client.rs:1928) -> build_stored_procedure_describe_request -> RpcParameter::get_sql_name(param.value()) (tds_client.rs:3995) -> this arm -> panic. Same exposure from the prepare/sp_execute paths and from #[cfg(fuzzing)] get_value().

Without Always Encrypted it's quieter but worse: serialize writes the parameter header and no value body, so we put a malformed RPC on the wire and desync the connection.

Suggest making this fallible rather than panicking:

pub(crate) fn value(&self) -> TdsResult<&SqlType> {
    match &self.value {
        RpcValue::Materialized(value) => Ok(value),
        RpcValue::Streamed(_) => Err(Error::UsageError(
            "Data-at-execution parameters are only supported via begin_sp_executesql.".to_string(),
        )),
    }
}

plus the same is_data_at_exec guard on the other public entry points (see my comment on tds_client.rs:949).

[Shiwani Gupta (shiwanigupta0809)]

Made it fallible instead of panicking - value() now returns TdsResult<&SqlType> and returns Error::UsageError for a streamed parameter.

pub(crate) fn value(&self) -> TdsResult<&SqlType> {
    match &self.value {
        RpcValue::Materialized(value) => Ok(value),
        RpcValue::Streamed(_) => Err(Error::UsageError(
            "Data-at-execution parameters are only supported via begin_sp_executesql.".to_string(),
        )),
    }
}

All callers updated to propagate with ?: serialize(), the parameter-encryption path (encrypt_parameter(param.value()?, ...)), both get_sql_name sites in build_stored_procedure_describe_request, and the #[cfg(fuzzing)] get_value() accessor. Added value_on_streamed_param_returns_usage_error.

[Saurabh Singh (saurabh500)]

Blocking. This guard is right, but it's the only one. execute_stored_procedure, prepare_statement, the sp_execute/sp_prepexec paths and execute_rpc all take Vec<RpcParameter> and none of them reject a data_at_exec param. Since RpcParameter::data_at_exec is pub, those paths either panic (Always Encrypted, via RpcParameter::value()) or serialize a parameter header with no value body and desync the connection.

Please hoist this into a shared helper and call it from every public entry point that accepts RpcParameters:

fn reject_data_at_exec<'p>(params: impl IntoIterator<Item = &'p RpcParameter>) -> TdsResult<()> {
    if params.into_iter().any(RpcParameter::is_data_at_exec) {
        return Err(UsageError(
            "Data-at-execution parameters require begin_sp_executesql.".to_string(),
        ));
    }
    Ok(())
}

A test asserting UsageError (not a panic) for a streamed param passed to execute_stored_procedure would lock this down.

[Shiwani Gupta (shiwanigupta0809)]

Added a shared guard and called it from every public entry point that accepts RpcParameters:

pub(crate) fn reject_data_at_exec<'p>(
    params: impl IntoIterator<Item = &'p RpcParameter>,
) -> TdsResult<()> {
    if params.into_iter().any(RpcParameter::is_data_at_exec) {
        return Err(Error::UsageError(
            "Data-at-execution parameters require begin_sp_executesql.".to_string(),
        ));
    }
    Ok(())
}

Call sites:

• execute_sp_executesql - replaced the inline check with this helper.
• execute_stored_procedure - reject_data_at_exec(positional.iter().flatten().chain(named.iter().flatten()))?.
• execute_prepared - reject_data_at_exec(named_params.iter())?.

execute_sp_prepare / execute_sp_prepexec / execute_sp_execute are private and only reachable through execute_prepared (or the #[cfg(test)] _for_test wrappers that intentionally bypass guards to exercise the wire protocol), so the single guard on execute_prepared covers them. Added reject_data_at_exec_rejects_only_streamed_params.

[Saurabh Singh (saurabh500)]

This should take options: impl Into<ExecuteOptions<'a>> like every sibling entry point (execute, execute_sp_executesql, execute_stored_procedure, prepare_statement, execute_prepared, unprepare). The ExecuteOptions doc comment states the convention outright:

New per-command capabilities are added here as new defaulted fields -- never as new methods or changed signatures -- keeping the execute* surface forward-compatible.

Taking timeout_sec/cancel_handle positionally means the next per-command knob added to ExecuteOptions reaches every other entry point for free but requires a signature change here -- a breaking change to a brand-new public API.

More concretely, hardcoding current_command_ce_setting = UseConnectionSetting at line 1035 removes the caller's only escape hatch. should_encrypt_parameters() is is_column_encryption_supported() && effective_command_ce_setting() == Enabled, so on an AE-enabled connection effective always resolves to Enabled and the check at line 1057 rejects every streamed write on that connection -- including statements whose parameters touch no encrypted column. That is exactly what ExecutionColumnEncryptionSetting::ResultSetOnly documents itself for ("a command reads encrypted columns but its parameters do not target any encrypted column"), and Disabled likewise.

No new rejection logic is needed -- just feed the option through and the existing check does the right thing:

pub async fn begin_sp_executesql<'a>(
    &mut self,
    sql: String,
    named_params: Vec<RpcParameter>,
    options: impl Into<ExecuteOptions<'a>>,
) -> TdsResult<StreamedParamStatus> {
    let ExecuteOptions { timeout, cancel, column_encryption } = options.into();
    if self.command_is_busy() { /* ... */ }
    self.current_command_ce_setting = column_encryption;
    // ...unchanged from here

The default still produces the current clear error on an AE connection; an explicit opt-out now works.

It also cleans up the no-streamed-params delegation just below, which currently rebuilds the struct and hardcodes the field, silently discarding the caller's AE intent on a path that forwards to a method which fully supports it:

 let result = self.execute_sp_executesql(
     sql, materialized_params,
-    ExecuteOptions {
-        timeout: timeout_sec,
-        cancel: cancel_handle,
-        column_encryption: ExecutionColumnEncryptionSetting::UseConnectionSetting,
-    },
+    ExecuteOptions { timeout, cancel, column_encryption },
 ).await?;

[Shiwani Gupta (shiwanigupta0809)]

Done - signature is now:

pub async fn begin_sp_executesql<'a>(
    &mut self,
    sql: String,
    named_params: Vec<RpcParameter>,
    options: impl Into<ExecuteOptions<'a>>,
) -> TdsResult<StreamedParamStatus> {
    let ExecuteOptions { timeout: timeout_sec, cancel: cancel_handle, column_encryption }
        = options.into();

This also fixed a latent bug you implicitly caught: column_encryption was hardcoded to UseConnectionSetting both on the streamed path and in the no-streamed-params delegation to execute_sp_executesql. It is now threaded through from the caller's options. All call sites updated (() for the common default case, ExecuteOptions::new().timeout_secs(n) / .cancel(h) otherwise) across the unit tests and tests/test_client_write_apis.rs.

[Saurabh Singh (saurabh500)]

This branch can never fire. streamed_params comes from .partition(RpcParameter::is_data_at_exec) four lines up, and is_streamable_plp has the exact same body as is_data_at_exec (matches!(self.value, RpcValue::Streamed(_))), so it is always true here. The "must be nvarchar(max), varchar(max) or varbinary(max)" error is unreachable.

That is fine as a type-system outcome -- StreamedSqlType already constrains it -- but the dead check reads like real validation. I'd drop it and keep only the name check. The test begin_only_accepts_streamable_types_via_data_at_exec_constructor (line 9086) asserts the same tautology and should go with it.

[Shiwani Gupta (shiwanigupta0809)]

Correct - removed. The .partition(RpcParameter::is_data_at_exec) above already guarantees every parameter in streamed_params is streamed, so the branch was unreachable. Only the name.is_none() validation remains:

for param in &streamed_params {
    if param.name.is_none() {
        return Err(UsageError("Streamed parameters must be named.".to_string()));
    }
}

Also dropped the now-tautological test begin_only_accepts_streamable_types_via_data_at_exec_constructor.

[Saurabh Singh (saurabh500)]

is_streamable_plp and is_data_at_exec (line 191) have identical bodies but doc comments implying different questions ("is it streamed?" vs "is its type streamable?"). Two names for one predicate will drift the moment someone adds a non-MAX streamed variant. Drop one -- is_data_at_exec reads better at the call sites.

[Shiwani Gupta (shiwanigupta0809)]

Removed is_streamable_plp() entirely - is_data_at_exec() is now the single predicate, and its one remaining caller (the dead validation branch in split_and_validate_streamed_params) is gone too.

[Saurabh Singh (saurabh500)]

The empty-chunk short circuit runs before the null_signaled check below, so the guard is inconsistent: after write_streamed_null(), write_streamed_chunk(&[0x01]) is a UsageError but write_streamed_chunk(&[]) silently returns Ok(()). A caller feeding a zero-length buffer gets no signal it is sequencing the API wrong.

Move the empty check below the state extraction and null_signaled validation, or check null_signaled up front alongside the Idle check.

[Shiwani Gupta (shiwanigupta0809)]

Good catch - reordered so the null_signaled check runs first, and both early returns now correctly re-park the message state:

// ...checked before the empty-chunk short-circuit below so
// `write_streamed_chunk(&[])` after `write_streamed_null()` still errors
// instead of silently succeeding.
if null_signaled {
    self.streamed_write_state = StreamedWriteState::Active(Box::new(StreamedWriteContext { ... }));
    return Err(UsageError(...));
}

// Empty chunks are ignored: a zero-length PLP chunk header is the value
// terminator, so it must never be emitted mid-value. Re-park the (unchanged)
// message so a later chunk or the terminator continues from the same point.
if chunk.is_empty() {
    self.streamed_write_state = StreamedWriteState::Active(Box::new(StreamedWriteContext { ... }));
    return Ok(());
}

Added streamed_write_empty_chunk_after_null_errors covering exactly this case.

[Saurabh Singh (saurabh500)]

Nit: these strings duplicate what RpcParameter::get_sql_name already returns for the same SqlTypes. Since as_sql_type() is right below, build_parameter_list_string_impl could call RpcParameter::get_sql_name(&streamed.as_sql_type())? and this method goes away -- one fewer place for the @params declaration name and the TYPE_INFO to drift apart.

[Shiwani Gupta (shiwanigupta0809)]

Agreed - now delegates instead of duplicating the strings:

/// Declaration name for the `sp_executesql` `@params` string. Delegates to
/// [`RpcParameter::get_sql_name_impl`] on the equivalent materialized
/// [`SqlType`] rather than duplicating the `nvarchar(MAX)` / `varchar(MAX)`
/// / `varbinary(MAX)` strings, so the two can't drift apart.
fn sql_name(self) -> TdsResult<String> {
    RpcParameter::get_sql_name_impl(&self.as_sql_type())
}

get_sql_name_impl is private but same-module, so no visibility change was needed. Call site is now RpcValue::Streamed(streamed) => streamed.sql_name()?.

[Shiwani Gupta (shiwanigupta0809)]

Copilot resolve the merge conflicts in this pull request

[Copilot]

Copilot resolve the merge conflicts in this pull request

Resolved by merging current main and retaining both test changes in 9493e3c.

[Shiwani Gupta (shiwanigupta0809)]

Saurabh Singh (@saurabh500) PR ready for re-review

[Shiwani Gupta (shiwanigupta0809)]

Copilot resolve the merge conflicts in this pull request

[Copilot]

Copilot resolve the merge conflicts in this pull request

Resolved and pushed in merge commit a1815fe.

[Saurabh Singh (saurabh500)]

Re-reviewed at a1815fe1.

All nine substantive findings from my earlier review are resolved — I verified each one against the code rather than taking the commit messages at face value. Two are worth calling out:

• The timeout fix is better than what I suggested. Rather than propagating start_time through resume, you deleted the field from SuspendedMessage outright so resume re-clocks from Instant::now(). That removes the possibility of the two clocks drifting at all, instead of just correcting the arithmetic.
• mark_known_dead() in abort_streamed_write is correctly ordered before close_transport(), which genuinely matters: close_transport only sets the dead flag after a successful stream.shutdown(), so aborting on an already-broken socket would return early and leave a dead connection looking alive to a pool checkout.

I also confirmed the main merge preserved every fix — the guard census still covers all three public entry points, and value() -> TdsResult<&SqlType> survived intact with its caller using ?.

No correctness, security, or API findings remain. What's left is test coverage on the newly added surface — three gaps, detailed inline. The first is the one I'd actually push on: the data-at-exec guard was the blocking bug from round one, and the fix has no regression test at the entry points it was added to. The other two are new public API and a new behavioral line with no assertions at all, which also matters against the repo's 85% diff-coverage gate.

Happy to approve once these are covered — no further design changes needed from my side.