mssql-odbc: declare and bound data-at-execution parameters from ParameterType

SQLBindParameter's ParameterType and ColumnSize were ignored for
data-at-execution parameters: every value streamed as a `max` type
whatever the application declared, and no length bound was enforced.

Declaration follows ParameterType. `dae_plan` decides stream-vs-buffer
from the SQL type's PLP-ability, mirroring msodbcsql's IsPartialLenType
(sqlcprot.h:1421), which keys on the SQL type alone rather than on
ColumnSize. A fixed-width target is now collected and converted whole
through the materializing path instead of being forced into a `max`
stream it cannot represent.

ColumnSize bounds the value. `dae_streamed_declaration` narrows the
@params declaration and `dae_length_limit` enforces the accumulated
total in SQLPutData, applied before the streamed/buffered split and
against the application buffer -- where msodbcsql runs
ValidatePutDataLength (sqlccmd.cpp:4571), so 22001 lands on the call
that overflows rather than at close. Trailing pad units are trimmed
rather than rejected, including one split across two calls, and trimmed
padding does not consume the declaration's budget. A narrow buffer is
measured in UTF-16 units, the unit the materialized path uses, so the
two paths agree on what fits. The declaration is only narrowed when that
bound can be enforced: a streamed parameter never reaches a close-time
conversion, so promising a length nothing checks would leave the
overflow to the server.

Transcoding is streamed, not deferred. `DaeTranscode` converts each
chunk on the way out, carrying a trailing partial character into the
next call -- the shape of msodbcsql's ConvertLongData (sqlccnvt.cpp:841)
with its cbTruncatedCharsInConvBuf carry and end-of-value flush
(sqlccmd.cpp:5999-6001). Narrow encoding reuses encode_narrow, so the
streamed and materialized paths agree on the wire bytes.

A mixed sequence collects only what it must. Parameters are offered in
bind order, as msodbcsql offers them, and the RPC opens as soon as no
parameter still to be visited needs collecting -- so a PLP-capable value
bound after a fixed-width one streams rather than being held whole.

text/ntext/image keep their `max` substitution, tracked under AB#47592.

AB#47590

Author: Shiwani Gupta

.github/instructions/mssql-odbc.instructions.md

@@ -353,6 +353,22 @@ does not grow every time a new msodbcsql build is measured.
     `ColAttributeLiveTest.EmptyVariantProbeConsumesValueButKeepsBaseType`
     accepts either so the parity leg still compares the base type and the
     `SQL_NO_DATA` re-read.
+14. **A cross-family data-at-execution parameter is converted rather than
+    refused.** `SQL_C_CHAR`/`SQL_C_WCHAR` against a fixed-width target such as
+    `SQL_INTEGER` is collected whole and converted by the same code the
+    materialized path uses, so text supplied in chunks parses to an integer
+    exactly as it does when bound directly. msodbcsql accepts the pairing at
+    `SQLExecute` and then refuses it at `SQLPutData` with `HY019` ("Processing
+    of fixed length targets cannot be spread over multiple calls to
+    SQLPutData"), so a value this driver sends is one msodbcsql rejects.
+    Deliberate: `ParameterType` decides how a data-at-execution parameter is
+    declared, and a pairing the materialized path can convert should not become
+    unsupported merely because the value arrives in chunks. Note this *widens*
+    what is accepted — an application relying on the refusal gets a successful
+    execute instead, though only for a binding msodbcsql never let complete
+    either. `CrossFamilyDataAtExecutionConvertsToInteger` (in both
+    `execute_test.cpp` and `param_conversions_test.cpp`) carries
+    `SKIP_IF_COMPARING_MSODBCSQL()`. Tracked in AB#47590.
 
 ## No panics
 

mssql-odbc/src/api/cancel.rs

@@ -87,19 +87,13 @@ unsafe fn sql_cancel_impl(statement_handle: SqlHandle) -> SqlReturn {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::api::odbc_types::{SQL_C_CHAR, SQL_INVALID_HANDLE, SQL_VARCHAR};
+    use crate::api::odbc_types::SQL_INVALID_HANDLE;
     use crate::handles::stmt::{DaeParam, DaeState, STMT_STATE_EXEC_STARTED};
     use crate::test_support::TestHandles;
 
     fn dae_with_one_param(cursor: Option<usize>) -> DaeState {
         DaeState::for_test(
-            vec![DaeParam {
-                value_ptr: std::ptr::null_mut(),
-                expected_len: None,
-                needs_transcode: false,
-                c_type: SQL_C_CHAR,
-                sql_type: SQL_VARCHAR,
-            }],
+            vec![DaeParam::unbounded(0, std::ptr::null_mut(), None)],
             cursor,
         )
     }

mssql-odbc/src/api/exec_common.rs

@@ -16,7 +16,7 @@ use mssql_tds::connection::tds_client::{
     CursorPoll, ExecuteOptions, ResultSet, StatementId, TdsClient,
 };
 use mssql_tds::error::{Error as TdsError, TimeoutErrorType};
-use mssql_tds::message::parameters::rpc_parameters::RpcParameter;
+use mssql_tds::message::parameters::rpc_parameters::{RpcParameter, StreamedSqlType};
 
 use super::ird::populate_ird;
 use super::sqlstate::*;
@@ -25,7 +25,8 @@ use crate::api::odbc_types::{
     SQL_SUCCESS_WITH_INFO, SqlHandle, SqlLen, SqlReturn,
 };
 use crate::conversion::param_convert::{
-    ParamBuildError, bound_param_to_rpc, dae_placeholder_type, is_data_at_exec_indicator,
+    DaePlan, DaeTranscode, ParamBuildError, bound_param_to_rpc, buffered_dae_to_rpc,
+    dae_length_limit, dae_plan, dae_streamed_declaration, is_data_at_exec_indicator,
 };
 use crate::error::post_sql_error;
 use crate::handles::dbc::ConnectionState;
@@ -124,9 +125,29 @@ pub(super) fn park_dae_client(
     client: TdsClient,
     prepared: Option<PreparedPlan>,
     orphaned: Option<StatementId>,
-    dae_params: Vec<DaeParam>,
+    mut dae_params: Vec<DaeParam>,
     op: &str,
 ) -> SqlReturn {
+    // The wire encoding of a narrow target comes from the database collation,
+    // which is only knowable with the connection in hand — `write_streamed_chunk`
+    // writes bytes verbatim, so the re-encoding has to happen before them
+    // (AB#47590).
+    let collation = client.get_collation();
+    for param in &mut dae_params {
+        if !param.plan.is_buffered() {
+            let transcode =
+                DaeTranscode::new(param.binding.c_type, param.binding.sql_type, collation);
+            // A pairing whose buffer bytes already are the wire's bytes keeps
+            // `None`, so `SQLPutData` forwards its chunks borrowed instead of
+            // copying each one through a conversion that would return them
+            // unchanged. Skipping the transcode also skips the close-time
+            // flush, which is right: a passthrough parameter never holds back a
+            // partial character to carry.
+            if !transcode.is_passthrough() {
+                param.transcode = Some(transcode);
+            }
+        }
+    }
     let Ok(mut stmt_state) = stmt.inner.lock() else {
         // The client has nowhere to go: the statement that owns it is
         // unreachable and the DBC still records it as busy.
@@ -137,6 +158,35 @@ pub(super) fn park_dae_client(
     SQL_NEED_DATA
 }
 
+/// Parks a sequence whose execute is deferred: at least one parameter buffers,
+/// so no RPC has been opened and the client sits idle on the statement until
+/// the last `SQLParamData` builds the complete parameter list and runs it.
+///
+/// The connection still counts as busy for the duration, exactly as the
+/// streaming sequence does, so an application cannot start another command on it
+/// mid-sequence.
+#[allow(clippy::too_many_arguments)]
+pub(super) fn park_deferred_dae(
+    stmt: &StmtHandle,
+    client: TdsClient,
+    prepared: Option<PreparedPlan>,
+    orphaned: Option<StatementId>,
+    dae_params: Vec<DaeParam>,
+    prebuilt: Vec<RpcParameter>,
+    sql: Option<String>,
+    timeout_secs: u32,
+    op: &str,
+) -> SqlReturn {
+    let Ok(mut stmt_state) = stmt.inner.lock() else {
+        error!("{op}: stmt mutex poisoned while parking deferred DAE state");
+        return SQL_ERROR;
+    };
+    stmt_state.dae = Some(
+        DaeState::new(client, prepared, orphaned, dae_params).deferred(prebuilt, sql, timeout_secs),
+    );
+    SQL_NEED_DATA
+}
+
 /// Aborts a data-at-execution sequence with a diagnostic. Always `SQL_ERROR`.
 pub(super) fn abort_dae_with_diag(
     dbc: &DbcHandle,
@@ -644,27 +694,88 @@ pub(super) unsafe fn build_named_params(
         };
 
         if let Some(indicator) = dae_indicator {
-            let dae_stream = match dae_placeholder_type(bound_param.c_type, bound_param.sql_type) {
-                Ok(t) => t,
+            let plan = match dae_plan(bound_param.c_type, bound_param.sql_type) {
+                Ok(plan) => plan,
                 Err(e) => {
                     error!(
-                        "{op}: parameter {} DAE type not streamable: {}",
+                        "{op}: parameter {} cannot be supplied at execution: {}",
                         i + 1,
                         e.diag().text
                     );
                     post_diag(stmt_state, e.diag());
                     return Err(SQL_ERROR);
                 }
             };
-            let rpc =
-                RpcParameter::data_at_exec(Some(name), StatusFlags::NONE, dae_stream.sql_type);
-            dae_params.push(DaeParam {
-                value_ptr: bound_param.parameter_value_ptr,
-                expected_len: dae_expected_length(indicator),
-                needs_transcode: dae_stream.needs_transcode,
-                c_type: bound_param.c_type,
-                sql_type: bound_param.sql_type,
-            });
+            // A bound that can be measured in buffer bytes is applied as the
+            // chunks arrive, so an overflow is reported by the `SQLPutData`
+            // that carries it rather than at close (msodbcsql parity).
+            let length_limit = match dae_length_limit(
+                bound_param.c_type,
+                bound_param.sql_type,
+                bound_param.column_size,
+            ) {
+                Ok(limit) => limit,
+                Err(e) => {
+                    error!(
+                        "{op}: parameter {} ColumnSize invalid: {}",
+                        i + 1,
+                        e.diag().text
+                    );
+                    post_diag(stmt_state, e.diag());
+                    return Err(SQL_ERROR);
+                }
+            };
+            dae_params.push(DaeParam::new(
+                i,
+                dae_expected_length(indicator),
+                plan,
+                length_limit,
+                bound_param,
+            ));
+            // Nothing to declare for a buffered parameter yet: its bytes are
+            // not in, so its type and length are not known. The slot is filled
+            // to keep parameter positions lined up and is rebuilt by
+            // `rebuild_deferred_params` before anything reaches the wire.
+            let rpc = match plan {
+                DaePlan::Stream(streamed) => {
+                    let param = RpcParameter::data_at_exec(Some(name), StatusFlags::NONE, streamed);
+                    // The body is PLP whatever `ColumnSize` says, but the
+                    // variable it lands in is declared from `ParameterType`,
+                    // matching the materialized path and msodbcsql (AB#47590).
+                    //
+                    // Only narrowed when the bound above can actually be
+                    // enforced. A streamed parameter never reaches a close-time
+                    // conversion, so an unenforced bound would have the client
+                    // declare `varchar(n)` and then stream past it, leaving the
+                    // overflow to the server instead of the `22001` the
+                    // declaration implies. Staying `max` keeps the value intact
+                    // until the bound is measurable on this path too
+                    // (AB#47590).
+                    let declaration = if length_limit.is_some() {
+                        dae_streamed_declaration(bound_param.sql_type, bound_param.column_size)
+                    } else {
+                        Ok(None)
+                    };
+                    match declaration {
+                        Ok(Some(declaration)) => param.with_streamed_declaration(declaration),
+                        Ok(None) => param,
+                        Err(e) => {
+                            error!(
+                                "{op}: parameter {} declaration invalid: {}",
+                                i + 1,
+                                e.diag().text
+                            );
+                            post_diag(stmt_state, e.diag());
+                            return Err(SQL_ERROR);
+                        }
+                    }
+                }
+                DaePlan::Buffer => RpcParameter::data_at_exec(
+                    Some(name),
+                    StatusFlags::NONE,
+                    StreamedSqlType::VarBinaryMax,
+                ),
+            };
             params.push(rpc);
         } else {
             match unsafe { bound_param_to_rpc(name, &bound_param) } {
@@ -690,6 +801,66 @@ pub(super) unsafe fn build_named_params(
     Ok(ParamsWithDae { params, dae_params })
 }
 
+/// Rebuilds the RPC parameter list once every data-at-execution value has been
+/// collected, for a sequence that deferred its execute.
+///
+/// Only the data-at-execution slots are rebuilt, from the bytes
+/// [`buffered_dae_to_rpc`] converts; every other parameter is the one
+/// `build_named_params` already materialized at execute time and is kept
+/// verbatim. A buffered value therefore still goes through the same conversion
+/// the materialized path uses, so it is declared and bounded exactly like a
+/// value supplied in a single buffer (AB#47590).
+///
+/// Nothing here reads application memory. Re-reading it would be unsound rather
+/// than merely redundant: `bound_params` is an execute-time snapshot that
+/// `SQLFreeStmt(SQL_RESET_PARAMS)` leaves in place while releasing the bindings
+/// it describes, so an application that resets its parameters mid-sequence --
+/// which this driver explicitly supports -- would have freed buffers
+/// dereferenced here.
+pub(super) fn rebuild_deferred_params(
+    stmt_state: &mut StmtState,
+    prebuilt: Vec<RpcParameter>,
+    collected: &[(usize, Vec<u8>, bool)],
+    dae_params: &[DaeParam],
+    op: &str,
+) -> Result<Vec<RpcParameter>, SqlReturn> {
+    let mut params = prebuilt;
+
+    for (index, bytes, is_null) in collected {
+        let Some(dae) = dae_params.iter().find(|p| p.bound_index == *index) else {
+            error!(
+                "{op}: collected value for parameter {} has no binding",
+                index + 1
+            );
+            post_diag(stmt_state, ERR_UNBOUND_PARAMETER);
+            return Err(SQL_ERROR);
+        };
+        let Some(slot) = params.get_mut(*index) else {
+            error!(
+                "{op}: collected value for parameter {} has no slot",
+                index + 1
+            );
+            post_diag(stmt_state, ERR_UNBOUND_PARAMETER);
+            return Err(SQL_ERROR);
+        };
+        let name = format!("@P{}", index + 1);
+        match buffered_dae_to_rpc(name, &dae.binding, bytes, *is_null) {
+            Ok(param) => *slot = param,
+            Err(e) => {
+                error!(
+                    "{op}: parameter {} conversion failed: {}",
+                    index + 1,
+                    e.diag().text
+                );
+                post_diag(stmt_state, e.diag());
+                return Err(SQL_ERROR);
+            }
+        }
+    }
+
+    Ok(params)
+}
+
 /// Captures result metadata after a successful execution and finalizes the
 /// statement/connection state.
 ///
@@ -1637,16 +1808,9 @@ mod tests {
 
         let dae = unsafe { build_named_params(&mut state, 3, "test") }.unwrap();
         assert_eq!(dae.params.len(), 3);
-        assert_eq!(
-            dae.dae_params,
-            vec![DaeParam {
-                value_ptr: std::ptr::null_mut(),
-                expected_len: None,
-                needs_transcode: false,
-                c_type: SQL_C_CHAR,
-                sql_type: SQL_VARCHAR
-            }]
-        );
+        assert_eq!(dae.dae_params.len(), 1);
+        assert_eq!(dae.dae_params[0].bound_index, 1);
+        assert_eq!(dae.dae_params[0].expected_len, None);
     }
 
     /// `SQL_LEN_DATA_AT_EXEC(n)` promises `n` bytes, which the closing
@@ -1672,16 +1836,63 @@ mod tests {
         }));
 
         let dae = unsafe { build_named_params(&mut state, 1, "test") }.unwrap();
-        assert_eq!(
-            dae.dae_params,
-            vec![DaeParam {
-                value_ptr: std::ptr::null_mut(),
-                expected_len: Some(7),
-                needs_transcode: false,
-                c_type: SQL_C_CHAR,
-                sql_type: SQL_VARCHAR
-            }]
-        );
+        assert_eq!(dae.dae_params.len(), 1);
+        assert_eq!(dae.dae_params[0].bound_index, 0);
+        assert_eq!(dae.dae_params[0].expected_len, Some(7));
+    }
+
+    /// A streamed parameter is only declared at a narrowed length when the
+    /// bound that length implies is one `SQLPutData` can actually enforce.
+    ///
+    /// The two decisions are made independently, so nothing but this guard stops
+    /// them disagreeing. Where they would, the cost is real: a streamed value
+    /// never reaches the close-time conversion that bounds a buffered one, so a
+    /// narrowed declaration with no bound behind it has the client promise the
+    /// server a length and then stream past it.
+    #[test]
+    fn build_named_params_only_narrows_a_declaration_it_can_enforce() {
+        for (c_type, sql_type, expect_narrowed) in [
+            // Same unit on both sides: measurable, so narrowing is honoured.
+            (SQL_C_CHAR, SQL_VARCHAR, true),
+            // A wideness mismatch is measurable too - the unit is the
+            // declaration's and the count is of the source.
+            (crate::api::odbc_types::SQL_C_WCHAR, SQL_VARCHAR, true),
+            (SQL_C_CHAR, crate::api::odbc_types::SQL_WVARCHAR, true),
+            // Cross-family is not: a binary byte is not a character, so the
+            // declaration stays `max` rather than claiming a bound.
+            (SQL_C_CHAR, crate::api::odbc_types::SQL_VARBINARY, false),
+        ] {
+            let h = TestHandles::with_env_dbc_stmt();
+            let stmt = unsafe { handle_from_raw::<StmtHandle>(h.stmt) };
+
+            let mut ind: SqlLen = SQL_DATA_AT_EXEC;
+            let mut state = stmt.inner.lock().unwrap();
+            state.bound_params.push(Some(BoundParam {
+                input_output_type: SQL_PARAM_INPUT,
+                c_type,
+                sql_type,
+                column_size: 10,
+                decimal_digits: 0,
+                parameter_value_ptr: std::ptr::null_mut(),
+                buffer_length: 0,
+                strlen_or_ind_ptr: &mut ind as *mut SqlLen,
+                octet_length_ptr: &mut ind as *mut SqlLen,
+            }));
+
+            let dae = unsafe { build_named_params(&mut state, 1, "test") }.unwrap();
+            let bounded = dae.dae_params[0].length_limit.is_some();
+            assert_eq!(
+                bounded, expect_narrowed,
+                "{c_type} -> {sql_type}: the bound decides whether narrowing is honest"
+            );
+            // The declaration is only narrowed when the bound backs it, so the
+            // two travel together rather than being decided apart.
+            let narrowed = format!("{:?}", dae.params[0]).contains("streamed_declaration: Some");
+            assert_eq!(
+                narrowed, expect_narrowed,
+                "{c_type} -> {sql_type}: declaration narrowing must follow the bound"
+            );
+        }
     }
 
     /// Without an indicator pointer there is nothing to carry a