diff --git a/CHANGELOG.md b/CHANGELOG.md index 2cac01d4d..ae2e76dbf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,10 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), ## [Unreleased] ### Added +- **AB#48177:** Added explicit `TableValuedParameter(type_name, columns, rows, + schema=...)` input values. Sync `execute()` streams lists, tuples and iterators + through native ODBC data-at-execution using one row of additional storage and + one procedure invocation, without discovering table metadata. - New feature: Support for macOS and Linux. - Documentation: Added API documentation in the Wiki. - New `token_provider=` parameter on `connect()` / `Connection` for Microsoft diff --git a/README.md b/README.md index 40fc409e7..e5553910b 100644 --- a/README.md +++ b/README.md @@ -208,6 +208,58 @@ connection.close() ``` +### Table-valued parameters + +Declare a table type and an input `READONLY` parameter on SQL Server: + +```sql +CREATE TYPE dbo.ItemBatch AS TABLE (id int, label nvarchar(100)); +GO +CREATE PROCEDURE dbo.ProcessItems @items dbo.ItemBatch READONLY +AS SELECT id, label FROM @items; +``` + +Pass one explicit value to synchronous `execute()`; ordinary nested lists/tuples are **not** +implicitly treated as TVPs: + +```python +from mssql_python import TableValuedParameter, SQL_INTEGER, SQL_WVARCHAR + +items = TableValuedParameter( + type_name="dbo.ItemBatch", + columns=[("id", SQL_INTEGER), ("label", SQL_WVARCHAR, 100)], + rows=[(1, "Keyboard"), (2, None)], +) +cursor.execute("EXEC dbo.ProcessItems ?", items) +``` + +Column tuples are `(name, sql_type[, size_or_precision[, scale]])`. Use the +driver's integer `SQL_*` constants. Columns bind **positionally**; names are +diagnostic labels, not a schema lookup. Character and binary columns require a +size; `0` means MAX for variable-length types. Decimal columns require both +precision and scale, e.g. `("amount", SQL_DECIMAL, 20, 6)`. Precision is 1–38; +scale is 0–precision. Temporal scale defaults to 7 (0–7 supported). +Sizes for Unicode columns are UTF-16 code units; binary sizes are bytes. +Rows must contain one cell per column. Supported cells include integers, floats, +bools, `Decimal`, strings, bytes/bytearray, date/time/datetime, UUID and `None`. + +Use an unquoted `schema.Type` or `type_name="ItemBatch", schema="dbo"` (default +schema: `dbo`); SQL quoting/bracket syntax and database-qualified names are not +parsed. The constructor does not consume rows. `rows=None` is invalid: use `[]` +for an empty table; individual cells may be `None`. + +With the default **msodbcsql18** sync provider, `rows` can also be a synchronous +iterator/generator. It is pulled during `execute()`, with one row of additional +driver-side binding storage regardless of total input length. A large cell still +occupies memory for that cell. This is one SQL execution, **not** `executemany()`; +multiple TVPs, scalar parameters and named parameters can be combined. The +native row-count descriptor permits up to 2,147,483,647 rows. Materialized sources +can be reused; iterators are one-shot and may be partially consumed on failure. +Iterator exceptions, including `KeyboardInterrupt`, cancel/reset the statement +before releasing its buffers. `executemany()` and `setinputsizes()` overrides +for statements containing TVPs are not supported. The `mssql-odbc` provider does +not currently support this ODBC TVP path. + ## Still have questions? Check out our [FAQ](https://github.com/microsoft/mssql-python/wiki/Frequently-Asked-Questions). Still not answered? Create an [issue](https://github.com/microsoft/mssql-python/issues/new/choose) to ask a question. diff --git a/mssql_python/__init__.py b/mssql_python/__init__.py index 06b382c9e..7710cf56c 100644 --- a/mssql_python/__init__.py +++ b/mssql_python/__init__.py @@ -66,6 +66,7 @@ # Row Objects from .row import Row, RowMapping +from .table_valued_parameter import TableValuedParameter # Logging Configuration (Simplified single-level DEBUG system) from .logging import logger, setup_logging, driver_logger @@ -324,6 +325,7 @@ def _cleanup_connections(): ) __all__ = [ + "TableValuedParameter", # Exception classes "Warning", "Error", diff --git a/mssql_python/cursor.py b/mssql_python/cursor.py index d4b85e56b..72a81a518 100644 --- a/mssql_python/cursor.py +++ b/mssql_python/cursor.py @@ -29,8 +29,10 @@ DatabaseError, ) from mssql_python.row import Row +from mssql_python.table_valued_parameter import TableValuedParameter from mssql_python.perf_timer import perf_phase from mssql_python import get_settings +from mssql_python.odbc_provider import ProviderManager, PROVIDER_MSSQL_ODBC from mssql_python.parameter_helper import ( detect_and_convert_parameters, parse_pyformat_params, @@ -389,6 +391,7 @@ def __init__(self, connection: "Connection", timeout: int = 0) -> None: self._connection: "Connection" = connection # Store as private attribute self._timeout: int = timeout + self._executing = False self._inputsizes: Optional[List[Tuple[int, int, int, int]]] = None # self.connection.autocommit = False self._initialize_cursor() @@ -1104,6 +1107,8 @@ def close(self) -> None: if getattr(self, "closed", True): # Do nothing - not calling _check_closed() here since we want this to be idempotent return + if getattr(self, "_executing", False): + raise ProgrammingError("Cannot close a cursor while it is executing", "") # Clear messages per DBAPI self.messages = [] @@ -1129,6 +1134,8 @@ def _check_closed(self) -> None: Raises: ProgrammingError: If the cursor is closed. """ + if getattr(self, "_executing", False): + raise ProgrammingError("Cannot reuse a cursor while it is executing", "") if self.closed: raise ProgrammingError( driver_error="Operation cannot be performed: The cursor is closed.", @@ -1799,8 +1806,8 @@ def execute( # pylint: disable=too-many-locals,too-many-branches,too-many-state # # Important: If you pass a tuple/list/dict as the ONLY argument, # it will be unwrapped for parameter binding. This means you cannot - # pass a tuple as a single parameter value (but SQL Server doesn't - # support tuple types as parameter values anyway). + # pass a tuple as a single scalar parameter value. A TVP is explicitly + # wrapped in TableValuedParameter and remains one parameter. with perf_phase("py::execute::param_prep"): if parameters: # Check if single parameter is a nested container that should be unwrapped @@ -1838,6 +1845,11 @@ def execute( # pylint: disable=too-many-locals,too-many-branches,too-many-state # Validate that inputsizes matches parameter count if both are present if parameters and self._inputsizes: + if any(isinstance(parameter, TableValuedParameter) for parameter in parameters): + raise NotSupportedError( + "setinputsizes does not support table-valued parameters", + "Remove setinputsizes for statements that pass TVP rows", + ) if len(self._inputsizes) != len(parameters): warnings.warn( @@ -1855,28 +1867,41 @@ def execute( # pylint: disable=too-many-locals,too-many-branches,too-many-state self.is_stmt_prepared = [False] effective_use_prepare = use_prepare and not same_sql - with perf_phase("py::execute::cpp_call"): - if parameters: - ret = ddbc_bindings.DDBCSQLExecute( - self.hstmt, - operation, - parameters, - self._inputsizes, - self.is_stmt_prepared, - effective_use_prepare, - encoding_settings, - ) - else: - ret = ddbc_bindings.DDBCSQLExecDirect(self.hstmt, operation) - # Check return code + self._executing = True try: - + with perf_phase("py::execute::cpp_call"): + if parameters: + if any(isinstance(parameter, TableValuedParameter) for parameter in parameters): + if ProviderManager.effective() == PROVIDER_MSSQL_ODBC: + raise NotSupportedError( + "Table-valued parameters are not supported by " + "the mssql-odbc provider", + "Use the default msodbcsql18 provider for TVP statements", + ) + ret = ddbc_bindings.DDBCSQLExecute( + self.hstmt, + operation, + parameters, + self._inputsizes, + self.is_stmt_prepared, + effective_use_prepare, + encoding_settings, + ) + else: + ret = ddbc_bindings.DDBCSQLExecDirect(self.hstmt, operation) # Check for errors but don't raise exceptions for info/warning messages check_error(ddbc_sql_const.SQL_HANDLE_STMT.value, self.hstmt, ret) - except Exception as e: # pylint: disable=broad-exception-caught + except BaseException as e: # Reset even when a TVP iterator raises KeyboardInterrupt. logger.warning("Execute failed, resetting cursor: %s", e) - self._reset_cursor() + try: + self._reset_cursor() + except Exception as cleanup_error: + # A callback may have closed the connection. Preserve its + # original error instead of replacing it with an allocation error. + logger.warning("Cursor recovery failed: %s", cleanup_error) raise + finally: + self._executing = False # Capture any diagnostic messages (SQL_SUCCESS_WITH_INFO, etc.) with perf_phase("py::execute::diag_records"): @@ -2554,6 +2579,11 @@ def executemany( # pylint: disable=too-many-locals,too-many-branches,too-many-s if hasattr(seq_of_parameters, "__getitem__") else [] ) + if any(isinstance(value, TableValuedParameter) for value in column): + raise NotSupportedError( + "executemany does not support table-valued parameters", + "Use execute for each TVP operation", + ) sample_value, min_val, max_val, _ = self._compute_column_type(column) if self._inputsizes and col_index < len(self._inputsizes): diff --git a/mssql_python/mssql_python.pyi b/mssql_python/mssql_python.pyi index 18f70c28c..8e0df1951 100644 --- a/mssql_python/mssql_python.pyi +++ b/mssql_python/mssql_python.pyi @@ -21,6 +21,34 @@ import datetime import logging import pyarrow +class TableValuedParameter: + def __init__( + self, + type_name: str, + columns: Sequence[Union[Tuple[str, int], Tuple[str, int, int], Tuple[str, int, int, int]]], + rows: Union[ + List[Sequence[Any]], + Tuple[Sequence[Any], ...], + Iterator[Sequence[Any]], + ], + *, + schema: Optional[str] = None, + ) -> None: ... + @property + def type_name(self) -> str: ... + @property + def schema(self) -> str: ... + @property + def columns(self) -> Tuple[Tuple[str, int, int, int], ...]: ... + @property + def rows( + self, + ) -> Union[ + List[Sequence[Any]], + Tuple[Sequence[Any], ...], + Iterator[Sequence[Any]], + ]: ... + # GLOBALS - DB-API 2.0 Required Module Globals # https://www.python.org/dev/peps/pep-0249/#module-interface apilevel: str # "2.0" diff --git a/mssql_python/pybind/connection/connection.cpp b/mssql_python/pybind/connection/connection.cpp index 9c4304d2f..8bbff9066 100644 --- a/mssql_python/pybind/connection/connection.cpp +++ b/mssql_python/pybind/connection/connection.cpp @@ -173,6 +173,10 @@ void Connection::disconnect(bool rollbackBeforeDisconnect) { for (const auto& handle : childHandles) { handle->markImplicitlyFreed(); } + for (const auto& buffers : _cleanupState->failedParameterBuffers) { + buffers->release(); + } + _cleanupState->failedParameterBuffers.clear(); _childStatementHandles.clear(); _allocationsSinceCompaction = 0; } diff --git a/mssql_python/pybind/ddbc_bindings.cpp b/mssql_python/pybind/ddbc_bindings.cpp index 4e4523955..2f27590a5 100644 --- a/mssql_python/pybind/ddbc_bindings.cpp +++ b/mssql_python/pybind/ddbc_bindings.cpp @@ -348,6 +348,8 @@ template // The callable hides whether the caller wraps SQLPutData with GIL management; chunk sizing stays shared. static SQLRETURN stream_dae_chunks(const void* data, size_t total_bytes, PutDataFn put_data_fn) { const char* bytes = static_cast(data); + if (total_bytes == 0) + return put_data_fn(const_cast(data), 0); for (size_t offset = 0; offset < total_bytes; offset += DAE_CHUNK_SIZE) { size_t len = std::min(static_cast(DAE_CHUNK_SIZE), total_bytes - offset); SQLRETURN rc = put_data_fn( @@ -452,6 +454,15 @@ static void PreResolveUnknownNullTypes(SqlHandle& handle, SQLHANDLE hStmt, } } +[[noreturn]] static void ThrowParameterError(const SqlHandlePtr& statementHandle, SQLRETURN rc, + const std::string& operation) { + ErrorInfo error = SQLCheckError_Wrap(SQL_HANDLE_STMT, statementHandle, rc); + py::object raiseException = + py::module_::import("mssql_python.exceptions").attr("raise_exception"); + raiseException(error.sqlState, operation + ": " + error.ddbcErrorMsg); + throw std::runtime_error(operation); +} + static SQLRETURN PreResolveUdtTypes(SQLHANDLE hStmt, std::vector& paramInfos) { bool hasUdt = false; for (const auto& info : paramInfos) { @@ -491,33 +502,10 @@ static SQLRETURN PreResolveUdtTypes(SQLHANDLE hStmt, std::vector& par return SQL_SUCCESS; } -// Given a list of parameters and their ParamInfo, calls SQLBindParameter on -// each of them with appropriate arguments -SQLRETURN BindParameters(SqlHandle& handle, SQLHANDLE hStmt, const py::list& params, - std::vector& paramInfos, - std::vector>& paramBuffers, - const std::string& charEncoding = "utf-8") { - PERF_TIMER("BindParameters"); - LOG("BindParameters: Starting parameter binding for statement handle %p " - "with %zu parameters", - (void*)hStmt, params.size()); - - SQLRETURN describeRc = PreResolveUdtTypes(hStmt, paramInfos); - if (!SQL_SUCCEEDED(describeRc)) return describeRc; - // GH-627: resolve unknown NULL param SQL types before binding any param. - PreResolveUnknownNullTypes(handle, hStmt, paramInfos, ¶ms); - for (int paramIndex = 0; paramIndex < params.size(); paramIndex++) { - const auto& param = params[paramIndex]; - ParamInfo& paramInfo = paramInfos[paramIndex]; - LOG("BindParameters: Processing param[%d] - C_Type=%d, SQL_Type=%d, " - "ColumnSize=%lu, DecimalDigits=%d, InputOutputType=%d", - paramIndex, paramInfo.paramCType, paramInfo.paramSQLType, - (unsigned long)paramInfo.columnSize, paramInfo.decimalDigits, - paramInfo.inputOutputType); - void* dataPtr = nullptr; - SQLLEN bufferLength = 0; - SQLLEN* strLenOrIndPtr = nullptr; - +static void PrepareParamBuffer(const py::handle& param, ParamInfo& paramInfo, + std::vector>& paramBuffers, + const std::string& charEncoding, int paramIndex, + void*& dataPtr, SQLLEN& bufferLength, SQLLEN*& strLenOrIndPtr) { // TODO: Add more data types like money, guid, interval, TVPs etc. switch (paramInfo.paramCType) { case SQL_C_CHAR: { @@ -530,7 +518,7 @@ SQLRETURN BindParameters(SqlHandle& handle, SQLHANDLE hStmt, const py::list& par "(Data-At-Execution) for large string streaming", paramIndex); dataPtr = - const_cast(reinterpret_cast(¶mInfos[paramIndex])); + const_cast(reinterpret_cast(¶mInfo)); strLenOrIndPtr = AllocateParamBuffer(paramBuffers); *strLenOrIndPtr = SQL_LEN_DATA_AT_EXEC(0); bufferLength = 0; @@ -591,7 +579,7 @@ SQLRETURN BindParameters(SqlHandle& handle, SQLHANDLE hStmt, const py::list& par "for VARBINARY(MAX) streaming", paramIndex); dataPtr = - const_cast(reinterpret_cast(¶mInfos[paramIndex])); + const_cast(reinterpret_cast(¶mInfo)); strLenOrIndPtr = AllocateParamBuffer(paramBuffers); *strLenOrIndPtr = SQL_LEN_DATA_AT_EXEC(0); bufferLength = 0; @@ -626,7 +614,7 @@ SQLRETURN BindParameters(SqlHandle& handle, SQLHANDLE hStmt, const py::list& par "NVARCHAR(MAX) streaming", paramIndex); dataPtr = - const_cast(reinterpret_cast(¶mInfos[paramIndex])); + const_cast(reinterpret_cast(¶mInfo)); strLenOrIndPtr = AllocateParamBuffer(paramBuffers); *strLenOrIndPtr = SQL_LEN_DATA_AT_EXEC(0); bufferLength = 0; @@ -744,9 +732,9 @@ SQLRETURN BindParameters(SqlHandle& handle, SQLHANDLE hStmt, const py::list& par ThrowStdException(MakeParamMismatchErrorStr(paramInfo.paramCType, paramIndex)); } int year = param.attr("year").cast(); - if (year < 1753 || year > 9999) { + if (year < 1 || year > 9999) { ThrowStdException("Date out of range for SQL Server " - "(1753-9999) at paramIndex " + + "(1-9999) at paramIndex " + std::to_string(paramIndex)); } // TODO: can be moved to python by registering SQL_DATE_STRUCT @@ -904,6 +892,52 @@ SQLRETURN BindParameters(SqlHandle& handle, SQLHANDLE hStmt, const py::list& par ThrowStdException(errorString.str()); } } +} + +} // namespace + +#include "tvp.hpp" + +namespace { + +// Given a list of parameters and their ParamInfo, calls SQLBindParameter on +// each of them with appropriate arguments +SQLRETURN BindParameters(const SqlHandlePtr& statementHandle, SQLHANDLE hStmt, const py::list& params, + std::vector& paramInfos, + std::vector>& paramBuffers, + const std::string& charEncoding = "utf-8") { + PERF_TIMER("BindParameters"); + SqlHandle& handle = *statementHandle; + LOG("BindParameters: Starting parameter binding for statement handle %p " + "with %zu parameters", + (void*)hStmt, params.size()); + + SQLRETURN describeRc = PreResolveUdtTypes(hStmt, paramInfos); + if (!SQL_SUCCEEDED(describeRc)) return describeRc; + // GH-627: resolve unknown NULL param SQL types before binding any param. + PreResolveUnknownNullTypes(handle, hStmt, paramInfos, ¶ms); + for (int paramIndex = 0; paramIndex < params.size(); paramIndex++) { + const auto& param = params[paramIndex]; + ParamInfo& paramInfo = paramInfos[paramIndex]; + LOG("BindParameters: Processing param[%d] - C_Type=%d, SQL_Type=%d, " + "ColumnSize=%lu, DecimalDigits=%d, InputOutputType=%d", + paramIndex, paramInfo.paramCType, paramInfo.paramSQLType, + (unsigned long)paramInfo.columnSize, paramInfo.decimalDigits, + paramInfo.inputOutputType); + void* dataPtr = nullptr; + SQLLEN bufferLength = 0; + SQLLEN* strLenOrIndPtr = nullptr; + + if (paramInfo.isTVP) { + RETCODE rc = BindTableValuedParameter(statementHandle, paramIndex, param, paramInfo); + if (!SQL_SUCCEEDED(rc)) { + return rc; + } + continue; + } + + PrepareParamBuffer(param, paramInfo, paramBuffers, charEncoding, paramIndex, + dataPtr, bufferLength, strLenOrIndPtr); assert(SQLBindParameter_ptr && SQLGetStmtAttr_ptr && SQLSetDescField_ptr); RETCODE rc; { @@ -1525,17 +1559,17 @@ DriverHandle LoadDriverOrThrowException() { SQLDescribeParam_ptr = GetFunctionPointer(handle, "SQLDescribeParam"); - bool success = SQLAllocHandle_ptr && SQLSetEnvAttr_ptr && SQLSetConnectAttr_ptr && - SQLSetStmtAttr_ptr && SQLGetConnectAttr_ptr && SQLDriverConnect_ptr && - SQLExecDirect_ptr && SQLPrepare_ptr && SQLBindParameter_ptr && SQLExecute_ptr && - SQLRowCount_ptr && SQLGetStmtAttr_ptr && SQLSetDescField_ptr && SQLFetch_ptr && - SQLFetchScroll_ptr && SQLGetData_ptr && SQLNumResultCols_ptr && SQLBindCol_ptr && - SQLDescribeCol_ptr && SQLMoreResults_ptr && SQLColAttribute_ptr && - SQLEndTran_ptr && SQLDisconnect_ptr && SQLFreeHandle_ptr && SQLFreeStmt_ptr && - SQLGetDiagRec_ptr && SQLGetInfo_ptr && SQLParamData_ptr && SQLPutData_ptr && - SQLTables_ptr && SQLDescribeParam_ptr && SQLGetTypeInfo_ptr && - SQLProcedures_ptr && SQLForeignKeys_ptr && SQLPrimaryKeys_ptr && - SQLSpecialColumns_ptr && SQLStatistics_ptr && SQLColumns_ptr; + bool success = + SQLAllocHandle_ptr && SQLSetEnvAttr_ptr && SQLSetConnectAttr_ptr && SQLSetStmtAttr_ptr && + SQLGetConnectAttr_ptr && SQLDriverConnect_ptr && SQLExecDirect_ptr && SQLPrepare_ptr && + SQLBindParameter_ptr && SQLExecute_ptr && SQLRowCount_ptr && SQLGetStmtAttr_ptr && + SQLSetDescField_ptr && SQLFetch_ptr && SQLFetchScroll_ptr && + SQLGetData_ptr && SQLNumResultCols_ptr && SQLBindCol_ptr && SQLDescribeCol_ptr && + SQLMoreResults_ptr && SQLColAttribute_ptr && SQLEndTran_ptr && SQLDisconnect_ptr && + SQLFreeHandle_ptr && SQLFreeStmt_ptr && SQLGetDiagRec_ptr && SQLGetInfo_ptr && + SQLParamData_ptr && SQLPutData_ptr && SQLTables_ptr && SQLDescribeParam_ptr && + SQLGetTypeInfo_ptr && SQLProcedures_ptr && SQLForeignKeys_ptr && SQLPrimaryKeys_ptr && + SQLSpecialColumns_ptr && SQLStatistics_ptr && SQLColumns_ptr; if (!success) { ThrowStdException("Failed to load required function pointers from driver."); @@ -1587,6 +1621,18 @@ void DriverLoader::loadDriver() { } } +FailedParameterBuffers::FailedParameterBuffers(std::vector> buffers) + : _buffers(std::make_unique>>(std::move(buffers))) {} + +FailedParameterBuffers::~FailedParameterBuffers() { + if (_buffers) { + // If even parent teardown failed, the driver may still use these addresses. + // Abandon native storage to process teardown rather than free bound buffers. + std::fputs("mssql-python: retaining parameter storage after failed native teardown\n", stderr); + _buffers.release(); + } +} + // SqlHandle definition SqlHandle::SqlHandle(SQLSMALLINT type, SQLHANDLE rawHandle, std::shared_ptr cleanupState) @@ -1606,7 +1652,18 @@ std::unique_lock SqlHandle::lockForCleanup() const { } SQLHANDLE SqlHandle::get() const { - return _handle; + return _failedParameterBuffers ? nullptr : _handle; +} + +void SqlHandle::retainFailedParameters(std::vector> buffers) { + _failedParameterBuffers = std::make_shared(std::move(buffers)); + py::gil_scoped_release release; + auto cleanupLock = lockForCleanup(); + if (!_handle || _implicitly_freed || (_cleanupState && _cleanupState->disconnected)) { + _failedParameterBuffers->release(); + } else if (_cleanupState) { + _cleanupState->failedParameterBuffers.push_back(_failedParameterBuffers); + } } SQLSMALLINT SqlHandle::type() const { @@ -1663,14 +1720,14 @@ SQLRETURN SqlHandle::freeHandle() { if (!_handle || !SQLFreeHandle_ptr) { return SQL_INVALID_HANDLE; } - describeCache.clear(); - if (_implicitly_freed || (_cleanupState && _cleanupState->disconnected)) { - _handle = nullptr; - return SQL_SUCCESS; + clearDescribeCache(); + SQLRETURN ret = SQL_SUCCESS; + if (!_implicitly_freed && !(_cleanupState && _cleanupState->disconnected)) { + ret = SQLFreeHandle_ptr(_type, _handle); } - SQLRETURN ret = SQLFreeHandle_ptr(_type, _handle); if (SQL_SUCCEEDED(ret)) { _handle = nullptr; + if (_failedParameterBuffers) _failedParameterBuffers->release(); } return ret; }; @@ -2114,12 +2171,9 @@ SQLRETURN SQLTables_wrap(SqlHandlePtr StatementHandle, const std::u16string& cat // When false and not prepared, throws (matching slow path behavior). // --------------------------------------------------------------------------- SQLRETURN SQLExecute_wrap(const SqlHandlePtr statementHandle, - const std::u16string& query, - py::list params, - const py::object& input_sizes, - py::list is_stmt_prepared, - bool use_prepare, - const py::dict& encoding_settings) { + const std::u16string& query, + py::list params, const py::object& input_sizes, py::list is_stmt_prepared, + bool use_prepare, const py::dict& encoding_settings) { PERF_TIMER("SQLExecute_wrap"); if (!statementHandle || !statementHandle->get()) { return SQL_INVALID_HANDLE; @@ -2181,8 +2235,9 @@ SQLRETURN SQLExecute_wrap(const SqlHandlePtr statementHandle, } std::vector> paramBuffers; - rc = BindParameters(*statementHandle, hStmt, params, paramInfos, paramBuffers, charEncoding); - if (!SQL_SUCCEEDED(rc)) return rc; + try { + rc = BindParameters(statementHandle, hStmt, params, paramInfos, paramBuffers, charEncoding); + if (!SQL_SUCCEEDED(rc)) ThrowParameterError(statementHandle, rc, "Binding parameters"); { py::gil_scoped_release release; @@ -2196,8 +2251,17 @@ SQLRETURN SQLExecute_wrap(const SqlHandlePtr statementHandle, if (rc == SQL_NEED_DATA) { SQLPOINTER paramToken = nullptr; auto putData = [&](SQLPOINTER data, SQLLEN len) { - py::gil_scoped_release release; - return SQLPutData_ptr(hStmt, data, len); + if (statementHandle->get() != hStmt || statementHandle->isImplicitlyFreed()) + throw std::runtime_error("Statement closed while producing parameter data"); + SQLRETURN result; + { + py::gil_scoped_release release; + result = SQLPutData_ptr(hStmt, data, len); + } + if (!SQL_SUCCEEDED(result)) + ThrowParameterError(statementHandle, result, + "Sending parameter data (length " + std::to_string(len) + ")"); + return result; }; while (true) { { @@ -2206,6 +2270,11 @@ SQLRETURN SQLExecute_wrap(const SqlHandlePtr statementHandle, } if (rc != SQL_NEED_DATA) break; + if (SendTableData(paramToken, paramInfos, putData, rc)) { + if (!SQL_SUCCEEDED(rc)) ThrowParameterError(statementHandle, rc, "Sending TVP data"); + continue; + } + // The DAE token is the ¶mInfos[i] we handed to SQLBindParameter as the // parameter value (see BindParameters), and paramInfos is sized up front and // never reallocated, so the token casts straight back to its ParamInfo instead @@ -2219,8 +2288,7 @@ SQLRETURN SQLExecute_wrap(const SqlHandlePtr statementHandle, } PyObject* pyObj = matchedInfo->dataPtr.ptr(); if (!pyObj || pyObj == Py_None) { - py::gil_scoped_release release; - SQLPutData_ptr(hStmt, nullptr, 0); + putData(nullptr, 0); continue; } @@ -2270,16 +2338,43 @@ SQLRETURN SQLExecute_wrap(const SqlHandlePtr statementHandle, ThrowStdException("SQLExecute: DAE only supported for str or bytes"); } } - if (!SQL_SUCCEEDED(rc) && rc != SQL_NO_DATA) return rc; + if (!SQL_SUCCEEDED(rc) && rc != SQL_NO_DATA) + ThrowParameterError(statementHandle, rc, "Executing parameters"); } - if (!SQL_SUCCEEDED(rc) && rc != SQL_NO_DATA) return rc; + if (!SQL_SUCCEEDED(rc) && rc != SQL_NO_DATA) + ThrowParameterError(statementHandle, rc, "Executing parameters"); // Unbind parameter buffers before they go out of scope. - // Not called on error paths — diagnostics must remain readable. SQLRETURN exec_rc = rc; - SQLFreeStmt_ptr(hStmt, SQL_RESET_PARAMS); + rc = SQLFreeStmt_ptr(hStmt, SQL_RESET_PARAMS); + if (!SQL_SUCCEEDED(rc)) ThrowParameterError(statementHandle, rc, "Resetting parameters"); return exec_rc; + } catch (...) { + // Convert diagnostics to the original Python exception before cleanup + // can overwrite them. Keep every buffer alive until NEED_DATA is left. + if (statementHandle->get() == hStmt && !statementHandle->isImplicitlyFreed()) { + SQLRETURN resetResult; + { + py::gil_scoped_release release; + if (SQLCancel_ptr) SQLCancel_ptr(hStmt); + resetResult = SQLFreeStmt_ptr(hStmt, SQL_RESET_PARAMS); + } + SQLRETURN freeResult = statementHandle->freeHandle(); + if (!SQL_SUCCEEDED(resetResult) && !SQL_SUCCEEDED(freeResult)) { + auto retainedInfos = std::make_shared>(std::move(paramInfos)); + paramBuffers.push_back(retainedInfos); + statementHandle->retainFailedParameters(std::move(paramBuffers)); + // Bound tokens/indicators are native storage. Drop Python owners + // with the GIL held, before retained storage can outlive Python. + for (auto& info : *retainedInfos) { + info.dataPtr = py::object(); + if (info.table) info.table->releasePythonState(); + } + } + } + throw; + } } SQLRETURN BindParameterArray(SqlHandle& handle, SQLHANDLE hStmt, const py::list& columnwise_params, @@ -2999,7 +3094,7 @@ SQLRETURN SQLExecuteMany_wrap(const SqlHandlePtr statementHandle, const std::u16 py::list rowParams = columnwise_params[rowIndex]; std::vector> paramBuffers; - rc = BindParameters(*statementHandle, hStmt, rowParams, paramInfos, + rc = BindParameters(statementHandle, hStmt, rowParams, paramInfos, paramBuffers, charEncoding); if (!SQL_SUCCEEDED(rc)) { LOG("SQLExecuteMany: BindParameters failed for row %zu - rc=%d", rowIndex, rc); @@ -6494,8 +6589,8 @@ PYBIND11_MODULE(ddbc_bindings, m) { }, "Disable global connection pooling and close all pools"); m.def("DDBCSQLExecDirect", &SQLExecDirect_wrap, "Execute a SQL query directly"); m.def("DDBCSQLExecute", &SQLExecute_wrap, - "DetectParamTypes + BindParameters + SQLExecute all in C++", - py::arg("statementHandle"), py::arg("query"), py::arg("params"), + "DetectParamTypes + BindParameters + SQLExecute all in C++", py::arg("statementHandle"), + py::arg("query"), py::arg("params"), py::arg("inputSizes"), py::arg("isStmtPrepared"), py::arg("usePrepare"), py::arg("encodingSettings")); m.def("SQLExecuteMany", &SQLExecuteMany_wrap, "Execute statement with multiple parameter sets", diff --git a/mssql_python/pybind/ddbc_bindings.h b/mssql_python/pybind/ddbc_bindings.h index 748ecc5ca..471fcab87 100644 --- a/mssql_python/pybind/ddbc_bindings.h +++ b/mssql_python/pybind/ddbc_bindings.h @@ -54,7 +54,11 @@ using py::literals::operator""_a; #define SQL_SS_XML (-152) #define SQL_SS_UDT (-151) #define SQL_SS_VARIANT (-150) +#define SQL_SS_TABLE (-153) #define SQL_CA_SS_VARIANT_TYPE (1215) +#define SQL_SOPT_SS_PARAM_FOCUS (1236) +#define SQL_CA_SS_SCHEMA_NAME (1226) +#define SQL_CA_SS_TYPE_NAME (1227) // Include logger bridge for LOG macros #include "logger_bridge.hpp" @@ -288,9 +292,20 @@ struct DescribedParamInfo { SQLSMALLINT decimalDigits; }; +class FailedParameterBuffers { + public: + explicit FailedParameterBuffers(std::vector> buffers); + ~FailedParameterBuffers(); + void release() { _buffers.reset(); } + + private: + std::unique_ptr>> _buffers; +}; + struct ConnectionCleanupState { std::mutex mutex; bool disconnected = false; // Protected by mutex, shared with every child. + std::vector> failedParameterBuffers; }; class SqlHandle { @@ -311,6 +326,7 @@ class SqlHandle { // unavailable. void cancel(); bool isImplicitlyFreed() const { return _implicitly_freed; } + void retainFailedParameters(std::vector> buffers); // Mark this handle as implicitly freed (freed by parent handle) // This prevents double-free attempts when the ODBC driver automatically @@ -342,6 +358,9 @@ class SqlHandle { SQLHANDLE _handle; bool _implicitly_freed = false; // Tracks if handle was freed by parent std::shared_ptr _cleanupState; + // Only populated if both parameter reset and handle release fail. The + // The connection also retains this native storage if the wrapper is dropped. + std::shared_ptr _failedParameterBuffers; }; using SqlHandlePtr = std::shared_ptr; diff --git a/mssql_python/pybind/param_detect.hpp b/mssql_python/pybind/param_detect.hpp index 172a98b51..19c03bbde 100644 --- a/mssql_python/pybind/param_detect.hpp +++ b/mssql_python/pybind/param_detect.hpp @@ -38,6 +38,13 @@ // Parameter description types //------------------------------------------------------------------------------------------------- +// External linkage for ParamInfo ownership; match its pybind11 members' visibility. +#ifdef __GNUC__ +struct __attribute__((visibility("hidden"))) TableParamState; +#else +struct TableParamState; +#endif + // This struct is shared between C++ & Python code. // Suppress -Wattributes warning for ParamInfo struct // The warning is triggered because pybind11 handles visibility attributes automatically, @@ -54,6 +61,8 @@ struct ParamInfo { SQLSMALLINT decimalDigits = 0; SQLLEN strLenOrInd = 0; // Required for DAE bool isDAE = false; // Indicates if we need to stream + bool isTVP = false; + std::shared_ptr table; // Strong reference to the Python object for DAE (data-at-execution) streaming. // py::object owns the refcount, so the compiler-generated destructor, copy and // move operations are all correct and this struct needs no rule-of-five. @@ -310,6 +319,7 @@ inline std::vector DetectParamTypes(PyObject* params, PyObject* input ParamInfo& info = infos[i]; info.inputOutputType = SQL_PARAM_INPUT; info.isDAE = false; + info.isTVP = false; if (i < inputSizeCount) { ApplyInputSizeOverride(params, PyList_GET_ITEM(inputSizes, i), i, info); @@ -635,6 +645,15 @@ inline std::vector DetectParamTypes(PyObject* params, PyObject* input continue; } + py::object tvpType = py::module_::import("mssql_python.table_valued_parameter") + .attr("TableValuedParameter"); + if (py::isinstance(borrow(obj), tvpType)) { + info.paramSQLType = SQL_SS_TABLE; + info.paramCType = SQL_C_BINARY; + info.isTVP = true; + continue; + } + // --- Unknown type: raise TypeError (matches Python _map_sql_type) --- throw py::type_error( "Unsupported parameter type: The driver cannot safely convert it to a SQL type."); diff --git a/mssql_python/pybind/tvp.hpp b/mssql_python/pybind/tvp.hpp new file mode 100644 index 000000000..eb07dd737 --- /dev/null +++ b/mssql_python/pybind/tvp.hpp @@ -0,0 +1,179 @@ +// One-row ODBC variable-row binding. Included after PrepareParamBuffer so +// scalar and TVP cells share the same conversion and buffer ownership rules. +#pragma once + +struct TableParamState { + py::object value; + py::object iterator; + py::list sizes; + py::list row; + std::u16string typeName; + std::u16string schema; + std::vector columns; + std::vector indicators; + std::vector data; + std::vector lengths; + std::vector> rowBuffers; + SQLLEN indicator = SQL_DATA_AT_EXEC; + size_t rowIndex = 0; + bool finished = false; + + explicit TableParamState(const py::handle& table) + : value(py::reinterpret_borrow(table)), + sizes(table.attr("_odbc_columns")().cast()), + typeName(table.attr("type_name").cast()), + schema(table.attr("schema").cast()), + columns(sizes.size()), indicators(sizes.size(), SQL_DATA_AT_EXEC), + data(sizes.size()), lengths(sizes.size()) { + iterator = steal(PyObject_GetIter(table.attr("rows").ptr())); + if (!iterator) throw py::error_already_set(); + for (size_t i = 0; i < sizes.size(); ++i) { + py::tuple spec = sizes[i].cast(); + columns[i].paramSQLType = spec[0].cast(); + columns[i].paramCType = spec[1].cast(); + columns[i].columnSize = spec[2].cast(); + columns[i].decimalDigits = spec[3].cast(); + } + } + + bool next() { + // Drop the previous window before entering user code for the next row. + row = py::list(); + rowBuffers.clear(); + if (PyErr_CheckSignals() != 0) throw py::error_already_set(); + py::object item = steal(PyIter_Next(iterator.ptr())); + if (!item) { + if (PyErr_Occurred()) throw py::error_already_set(); + finished = true; + return false; + } + py::object checked = value.attr("_validate_row")(item, rowIndex++); + row = py::reinterpret_steal(PySequence_List(checked.ptr())); + if (!row) throw py::error_already_set(); + for (size_t i = 0; i < columns.size(); ++i) { + const ParamInfo& declared = columns[i]; + if (row[i].is_none()) { + indicators[i] = SQL_NULL_DATA; + continue; + } + if (declared.paramCType == SQL_C_GUID) { + row[i] = row[i].attr("bytes_le"); + } + // ApplyInputSizeOverride preserves decimal coefficients (no float or + // decimal-context arithmetic) and fractional time via isoformat. + ParamInfo converted; + ApplyInputSizeOverride(row.ptr(), sizes[i].ptr(), i, converted); + converted.isDAE = false; + data[i] = nullptr; + lengths[i] = 0; + SQLLEN* lengthOrIndicator = nullptr; + PrepareParamBuffer(row[i], converted, rowBuffers, "utf-8", static_cast(i), + data[i], lengths[i], lengthOrIndicator); + if (lengthOrIndicator) lengths[i] = *lengthOrIndicator; + switch (declared.paramCType) { + case SQL_C_BIT: case SQL_C_TINYINT: case SQL_C_UTINYINT: + lengths[i] = sizeof(SQLCHAR); break; + case SQL_C_SHORT: lengths[i] = sizeof(SQLSMALLINT); break; + case SQL_C_LONG: lengths[i] = sizeof(SQLINTEGER); break; + case SQL_C_SBIGINT: lengths[i] = sizeof(int64_t); break; + case SQL_C_FLOAT: lengths[i] = sizeof(float); break; + case SQL_C_DOUBLE: lengths[i] = sizeof(double); break; + case SQL_C_TYPE_DATE: lengths[i] = sizeof(SQL_DATE_STRUCT); break; + case SQL_C_TYPE_TIME: lengths[i] = sizeof(SQL_TIME_STRUCT); break; + case SQL_C_TYPE_TIMESTAMP: lengths[i] = sizeof(SQL_TIMESTAMP_STRUCT); break; + } + indicators[i] = SQL_DATA_AT_EXEC; + } + return true; + } + + void releasePythonState() { + value = py::object(); + iterator = py::object(); + sizes.release().dec_ref(); + row.release().dec_ref(); + } +}; + +static SQLRETURN BindTableValuedParameter(const SqlHandlePtr& handle, int paramIndex, + const py::handle& value, ParamInfo& info) { + info.table = std::make_shared(value); + TableParamState& table = *info.table; + SQLHANDLE hStmt = handle->get(); + if (!hStmt || handle->isImplicitlyFreed()) + throw std::runtime_error("Statement closed while initializing TVP rows"); + SQLSMALLINT record = static_cast(paramIndex + 1); + // msodbcsql18 enforces ColumnSize as a cumulative row ceiling even for + // variable-row DAE (1 rejects the second row). This is a descriptor limit, + // not an allocation: all columns are DAE tokens with one row of storage. + SQLRETURN rc = SQLBindParameter_ptr( + hStmt, record, SQL_PARAM_INPUT, SQL_C_BINARY, SQL_SS_TABLE, + std::numeric_limits::max(), 0, + nullptr, 0, &table.indicator); + if (!SQL_SUCCEEDED(rc)) return rc; + SQLHDESC descriptor = nullptr; + rc = SQLGetStmtAttr_ptr(hStmt, SQL_ATTR_IMP_PARAM_DESC, &descriptor, 0, nullptr); + if (!SQL_SUCCEEDED(rc)) return rc; + rc = SQLSetDescField_ptr(descriptor, record, SQL_CA_SS_TYPE_NAME, + table.typeName.data(), + static_cast(table.typeName.size() * sizeof(SQLWCHAR))); + if (!SQL_SUCCEEDED(rc)) return rc; + rc = SQLSetDescField_ptr(descriptor, record, SQL_CA_SS_SCHEMA_NAME, + table.schema.data(), + static_cast(table.schema.size() * sizeof(SQLWCHAR))); + if (!SQL_SUCCEEDED(rc)) return rc; + rc = SQLGetStmtAttr_ptr(hStmt, SQL_ATTR_APP_PARAM_DESC, &descriptor, 0, nullptr); + if (!SQL_SUCCEEDED(rc)) return rc; + rc = SQLSetDescField_ptr(descriptor, record, SQL_DESC_DATA_PTR, &table, 0); + if (!SQL_SUCCEEDED(rc)) return rc; + rc = SQLSetStmtAttr_ptr(hStmt, SQL_SOPT_SS_PARAM_FOCUS, + reinterpret_cast(static_cast(record)), + SQL_IS_INTEGER); + if (!SQL_SUCCEEDED(rc)) return rc; + try { + for (size_t i = 0; i < table.columns.size(); ++i) { + const ParamInfo& column = table.columns[i]; + rc = SQLBindParameter_ptr( + hStmt, static_cast(i + 1), SQL_PARAM_INPUT, + column.paramCType, column.paramSQLType, column.columnSize, column.decimalDigits, + &table.columns[i], 0, &table.indicators[i]); + if (!SQL_SUCCEEDED(rc)) ThrowParameterError(handle, rc, "Failed to bind TVP column"); + } + } catch (...) { + SQLSetStmtAttr_ptr(hStmt, SQL_SOPT_SS_PARAM_FOCUS, nullptr, SQL_IS_INTEGER); + throw; + } + return SQLSetStmtAttr_ptr(hStmt, SQL_SOPT_SS_PARAM_FOCUS, nullptr, SQL_IS_INTEGER); +} + +template +static bool SendTableData(SQLPOINTER token, std::vector& params, + PutData putData, SQLRETURN& rc) { + for (auto& param : params) { + if (!param.table) continue; + auto& table = *param.table; + if (token == &table) { + if (table.finished) throw std::runtime_error("TVP requested rows after completion"); + SQLLEN count = table.next() ? 1 : 0; + // For the table token, the driver interprets DataPtr as the + // row count (not a pointer to row data). Column tokens carry data. + rc = putData(reinterpret_cast(static_cast(count)), count); + return true; + } + for (size_t i = 0; i < table.columns.size(); ++i) { + if (token != &table.columns[i]) continue; + if (table.indicators[i] == SQL_NULL_DATA) { + rc = putData(nullptr, SQL_NULL_DATA); + return true; + } + const auto cType = table.columns[i].paramCType; + if (cType == SQL_C_WCHAR || cType == SQL_C_CHAR || cType == SQL_C_BINARY) { + rc = stream_dae_chunks(table.data[i], static_cast(table.lengths[i]), putData); + } else { + rc = putData(table.data[i], table.lengths[i]); + } + return true; + } + } + return false; +} diff --git a/mssql_python/table_valued_parameter.py b/mssql_python/table_valued_parameter.py new file mode 100644 index 000000000..15488f09d --- /dev/null +++ b/mssql_python/table_valued_parameter.py @@ -0,0 +1,172 @@ +"""Explicit, transport-neutral input values for SQL Server table types.""" + +from collections.abc import Iterator, Sequence +from datetime import date, datetime, time +from decimal import Decimal +from uuid import UUID + +from .constants import ConstantsDDBC as SQL + + +class TableValuedParameter: + """A table type, its ordered column declarations, and an unconsumed row source. + + Columns are ``(name, sql_type[, size_or_precision[, scale]])`` tuples. + Variable-length columns require a size (0 means MAX); DECIMAL/NUMERIC + require precision and scale. Temporal scale defaults to 7. + ``type_name`` accepts an unquoted ``schema.Type`` or a type name with the + separate ``schema`` keyword (default ``dbo``). Binding is positional. + """ + + __slots__ = ("_type_name", "_schema", "_columns", "_rows") + + def __init__(self, type_name, columns, rows, *, schema=None): + if not isinstance(type_name, str): + raise TypeError("TVP type_name must be a string") + if "." in type_name: + if schema is not None or type_name.count(".") != 1: + raise ValueError("Specify TVP schema once: schema.Type or schema=...") + schema, type_name = type_name.split(".") + schema = "dbo" if schema is None else schema + for name in (type_name, schema): + if not isinstance(name, str): + raise TypeError("TVP schema and type name must be strings") + if ( + not name.strip() + or any(char in name for char in '\x00.[]"') + or len(name.encode("utf-16le")) > 256 + ): + raise ValueError("TVP names must be unquoted identifiers of 1–128 UTF-16 units") + if not isinstance(columns, Sequence) or not columns: + raise TypeError("TVP columns must be a non-empty sequence") + normalized = [] + for column in columns: + if not isinstance(column, tuple) or not 2 <= len(column) <= 4: + raise TypeError("TVP columns use (name, sql_type[, size_or_precision[, scale]])") + name, sql_type = column[:2] + if not isinstance(name, str) or not name: + raise TypeError("TVP column names must be non-empty strings") + if type(sql_type) is not int: + raise TypeError(f"TVP column {name}: sql_type must be an SQL_* integer constant") + sql_type = {9: 91, 10: 92, 11: 93}.get(sql_type, sql_type) + if sql_type not in self._types: + raise ValueError(f"TVP column {name}: unsupported SQL type {sql_type}") + variable = sql_type in self._variable + numeric = sql_type in (SQL.SQL_DECIMAL.value, SQL.SQL_NUMERIC.value) + temporal = sql_type in (92, 93, -154, -155) + if (variable and len(column) < 3) or (numeric and len(column) != 4): + raise ValueError(f"TVP column {name}: explicit size or precision/scale is required") + size = column[2] if len(column) >= 3 else self._sizes.get(sql_type, 0) + scale = column[3] if len(column) == 4 else (7 if temporal else 0) + if type(size) is not int or type(scale) is not int: + raise TypeError(f"TVP column {name}: size and scale must be integers") + if not 0 <= size <= 2147483647 or scale < 0: + raise ValueError(f"TVP column {name}: invalid size or scale") + if numeric and not (1 <= size <= 38 and 0 <= scale <= size): + raise ValueError( + f"TVP column {name}: require 1 <= precision <= 38 and scale <= precision" + ) + if temporal and scale > 7: + raise ValueError(f"TVP column {name}: temporal scale must be between 0 and 7") + if not numeric and not temporal and scale: + raise ValueError( + f"TVP column {name}: scale is only valid for numeric/temporal types" + ) + if ( + sql_type in (SQL.SQL_CHAR.value, SQL.SQL_WCHAR.value, SQL.SQL_BINARY.value) + and not size + ): + raise ValueError(f"TVP column {name}: fixed-length columns require a positive size") + normalized.append((name, sql_type, size, scale)) + if not isinstance(rows, (list, tuple, Iterator)): + raise TypeError("TVP rows must be a list, tuple, or iterator; use [] for an empty TVP") + self._type_name, self._schema = type_name, schema + self._columns, self._rows = tuple(normalized), rows + + @property + def type_name(self): + return self._type_name + + @property + def schema(self): + return self._schema + + @property + def columns(self): + return self._columns + + @property + def rows(self): + return self._rows + + # The existing setinputsizes map owns SQL-to-C mappings. TVP time keeps + # fractional seconds, and decimals use the scalar exact-text conversion. + def _odbc_columns(self): + from .cursor import Cursor + + mapping = Cursor._get_sql_to_c_type_map() + return [ + ( + {-1: 12, -10: -9, -4: -3}.get(sql_type, sql_type), + SQL.SQL_C_WCHAR.value if sql_type in (2, 3, 92, -154) else mapping[sql_type], + size, + scale, + ) + for _, sql_type, size, scale in self.columns + ] + + def _validate_row(self, row, index): + if not isinstance(row, (list, tuple)): + raise TypeError(f"TVP row {index} must be a list or tuple") + if len(row) != len(self.columns): + raise ValueError(f"TVP row {index} has {len(row)} cells; expected {len(self.columns)}") + for (name, sql_type, _, _), value in zip(self.columns, row): + if value is not None and not isinstance(value, self._types[sql_type]): + raise TypeError(f"TVP row {index}, column {name}: incompatible Python value") + if value is None: + continue + if sql_type in self._integer_ranges: + lower, upper = self._integer_ranges[sql_type] + if not lower <= value <= upper: + raise ValueError(f"TVP row {index}, column {name}: integer out of range") + if sql_type in (2, 3) and not value.is_finite(): + raise ValueError(f"TVP row {index}, column {name}: non-finite Decimal") + return row + + _variable = frozenset((1, 12, -1, -8, -9, -10, -2, -3, -4)) + _types = { + **dict.fromkeys((1, 12, -1, -8, -9, -10), str), + **dict.fromkeys((-2, -3, -4), (bytes, bytearray)), + **dict.fromkeys((-6, 5, 4, -5), int), + **dict.fromkeys((6, 7, 8), float), + **dict.fromkeys((2, 3), Decimal), + -7: bool, + 91: date, + 92: time, + -154: time, + 93: datetime, + -155: datetime, + -11: UUID, + } + _sizes = { + -7: 1, + -6: 3, + 5: 5, + 4: 10, + -5: 19, + 6: 15, + 7: 7, + 8: 15, + 91: 10, + 92: 16, + -154: 16, + 93: 27, + -155: 34, + -11: 36, + } + _integer_ranges = { + -6: (0, 255), + 5: (-(2**15), 2**15 - 1), + 4: (-(2**31), 2**31 - 1), + -5: (-(2**63), 2**63 - 1), + } diff --git a/tests/test_004_cursor_timeout.py b/tests/test_004_cursor_timeout.py new file mode 100644 index 000000000..1eeb94eda --- /dev/null +++ b/tests/test_004_cursor_timeout.py @@ -0,0 +1,15 @@ +"""Live regression coverage for query timeout and subsequent cursor recovery.""" + +import pytest + +from mssql_python import OperationalError +from mssql_python.cursor import Cursor + + +def test_query_timeout_and_recovery(db_connection): + with Cursor(db_connection, timeout=1) as cursor: + with pytest.raises(OperationalError, match="[Tt]imeout"): + cursor.execute("WAITFOR DELAY '00:00:03'; SELECT 1") + assert cursor.execute("SELECT 42").fetchone()[0] == 42 + with Cursor(db_connection, timeout=0) as cursor: + assert cursor.execute("SELECT 43").fetchone()[0] == 43 diff --git a/tests/test_028_table_valued_parameters.py b/tests/test_028_table_valued_parameters.py new file mode 100644 index 000000000..6fd0100ff --- /dev/null +++ b/tests/test_028_table_valued_parameters.py @@ -0,0 +1,620 @@ +"""Live SQL Server coverage for explicit TVPs and synchronous ODBC streaming.""" + +import datetime +import os +import subprocess +import sys +import threading +import uuid +import weakref +from collections import UserList +from decimal import Decimal, localcontext + +import pytest + +from mssql_python import ( + DatabaseError, + NotSupportedError, + TableValuedParameter, + connect, + SQL_INTEGER, + SQL_WVARCHAR, + SQL_VARBINARY, + SQL_DECIMAL, + SQL_BIT, + SQL_TINYINT, + SQL_SMALLINT, + SQL_BIGINT, + SQL_DOUBLE, + SQL_TYPE_DATE, + SQL_TYPE_TIME, + SQL_TYPE_TIMESTAMP, + SQL_GUID, +) +from mssql_python.constants import ConstantsDDBC +from test_023_ssh_tunnel_gil_release import ( + WATCHDOG_SECONDS, + _parse_server, + _replace_server, + _start_forwarder, +) + + +@pytest.fixture +def tvp_schema(cursor, db_connection): + suffix = uuid.uuid4().hex + schema = f"pytest_tvp_{suffix}" + cursor.execute(f"CREATE SCHEMA [{schema}]") + cursor.execute(f"CREATE TYPE [{schema}].Items AS TABLE (id int, label nvarchar(max))") + cursor.execute(f"CREATE TABLE [{schema}].Calls (n int)") + cursor.execute(f""" + CREATE PROCEDURE [{schema}].Echo @prefix int, @items [{schema}].Items READONLY, + @suffix nvarchar(max) + AS BEGIN + SET NOCOUNT ON; + INSERT INTO [{schema}].Calls VALUES (1); + SELECT id + @prefix AS id, label + @suffix AS label FROM @items ORDER BY id; + END + """) + cursor.execute(f""" + CREATE PROCEDURE [{schema}].Pair @left [{schema}].Items READONLY, + @right [{schema}].Items READONLY + AS BEGIN + SET NOCOUNT ON; + SELECT id, label FROM @left UNION ALL SELECT id, label FROM @right; + END + """) + cursor.execute(f""" + CREATE PROCEDURE [{schema}].CountRows @items [{schema}].Items READONLY + AS BEGIN + SET NOCOUNT ON; + INSERT INTO [{schema}].Calls VALUES (1); + SELECT COUNT_BIG(*), SUM(CONVERT(bigint,id)) FROM @items; + END + """) + db_connection.commit() + try: + yield schema + finally: + cursor.execute(f"DROP PROCEDURE [{schema}].Echo") + cursor.execute(f"DROP PROCEDURE [{schema}].Pair") + cursor.execute(f"DROP PROCEDURE [{schema}].CountRows") + cursor.execute(f"DROP TYPE [{schema}].Items") + cursor.execute(f"DROP TABLE [{schema}].Calls") + cursor.execute(f"DROP SCHEMA [{schema}]") + db_connection.commit() + + +def items(schema, rows): + return TableValuedParameter( + "Items", [("id", SQL_INTEGER), ("label", SQL_WVARCHAR, 0)], rows, schema=schema + ) + + +@pytest.mark.parametrize("rows", [[], [(1, "Keyboard"), (2, None)], ((3, ""),)]) +@pytest.mark.parametrize("named", [False, True]) +def test_rows_named_reuse_and_one_invocation(cursor, tvp_schema, rows, named): + value = items(tvp_schema, rows) + sql = f"EXEC [{tvp_schema}].Echo ?, ?, ?" + params = (10, value, "✓") + if named: + sql = f"EXEC [{tvp_schema}].Echo %(prefix)s, %(items)s, %(suffix)s" + params = {"prefix": 10, "items": value, "suffix": "✓"} + for _ in range(2): + cursor.execute(sql, params) + result = cursor.fetchall() + assert [tuple(row) for row in result] == [ + (row[0] + 10, None if row[1] is None else row[1] + "✓") for row in rows + ] + cursor.execute(f"SELECT COUNT(*) FROM [{tvp_schema}].Calls") + assert cursor.fetchone()[0] == 2 + + +@pytest.mark.parametrize("left,right", [([], []), ([], [(2, "right")]), ([(1, None)], [(2, "")])]) +def test_multiple_tvps(cursor, tvp_schema, left, right): + cursor.execute( + f"EXEC [{tvp_schema}].Pair ?, ?", + items(tvp_schema, left), + items(tvp_schema, right), + ) + result = cursor.fetchall() + assert sorted(tuple(row) for row in result) == sorted(left + right) + + +def test_null_scalars_with_tvp(cursor, tvp_schema): + cursor.execute( + f"EXEC [{tvp_schema}].Echo ?, ?, ?", + None, + items(tvp_schema, [(1, "one")]), + None, + ) + assert tuple(cursor.fetchone()) == (None, None) + + +def test_explicit_type_in_parameterized_batch(cursor, tvp_schema): + cursor.execute("SELECT id, label FROM ?", items(tvp_schema, iter([(1, "one")]))) + assert tuple(cursor.fetchone()) == (1, "one") + + +def test_reuse_with_empty_and_changing_sources(cursor, tvp_schema): + sql = f"EXEC [{tvp_schema}].Echo ?, ?, ?" + for rows in ([], [(1, None)], [(2, "two"), (3, "")], []): + value = TableValuedParameter( + "Items", + [("diagnostic_id", SQL_INTEGER), ("diagnostic_label", SQL_WVARCHAR, 100)], + iter(rows), + schema=tvp_schema, + ) + cursor.execute(sql, 0, value, "") + assert [tuple(row) for row in cursor.fetchall()] == rows + + +def test_declared_text_size_failure_recovers(cursor, tvp_schema): + value = TableValuedParameter( + "Items", + [("id", SQL_INTEGER), ("label", SQL_WVARCHAR, 3)], + iter([(1, "too long")]), + schema=tvp_schema, + ) + with pytest.raises((DatabaseError, ValueError)): + cursor.execute(f"EXEC [{tvp_schema}].CountRows ?", value) + assert cursor.execute("SELECT 42").fetchone()[0] == 42 + + +def test_typed_cells(cursor, db_connection): + suffix = uuid.uuid4().hex + name, proc = f"Typed_{suffix}", f"TypedProc_{suffix}" + declarations = [ + ("bit_value", SQL_BIT, "bit"), + ("tiny_value", SQL_TINYINT, "tinyint"), + ("small_value", SQL_SMALLINT, "smallint"), + ("int_value", SQL_INTEGER, "int"), + ("big_value", SQL_BIGINT, "bigint"), + ("float_value", SQL_DOUBLE, "float"), + ("decimal_value", SQL_DECIMAL, "decimal(18,10)", 18, 10), + ("text_value", SQL_WVARCHAR, "nvarchar(max)", 0), + ("binary_value", SQL_VARBINARY, "varbinary(max)", 0), + ("date_value", SQL_TYPE_DATE, "date"), + ("time_value", SQL_TYPE_TIME, "time(7)"), + ("datetime_value", SQL_TYPE_TIMESTAMP, "datetime2(7)"), + ("offset_value", ConstantsDDBC.SQL_DATETIMEOFFSET.value, "datetimeoffset(7)"), + ("guid_value", SQL_GUID, "uniqueidentifier"), + ("all_null_binary", SQL_VARBINARY, "varbinary(16)", 16), + ] + columns = [(d[0], d[1], *d[3:]) for d in declarations] + row = ( + True, + 255, + -32768, + 2147483647, + -(2**63), + 1.25, + Decimal("12345678.1234567890"), + "long🚀漢字\x00" * 2000, + b"\x00\xff" * 10000, + datetime.date(2020, 2, 29), + datetime.time(1, 2, 3, 123456), + datetime.datetime(2021, 3, 4, 1, 2, 3, 123456), + datetime.datetime( + 2021, + 3, + 4, + 1, + 2, + 3, + 123456, + tzinfo=datetime.timezone(datetime.timedelta(hours=-7, minutes=-30)), + ), + uuid.uuid4(), + None, + ) + cursor.execute( + f"CREATE TYPE dbo.[{name}] AS TABLE (" + + ",".join(f"[{d[0]}] {d[2]} NULL" for d in declarations) + + ")" + ) + projection = ",".join( + "CONVERT(varbinary(max),text_value) AS text_value" if d[0] == "text_value" else d[0] + for d in declarations + ) + cursor.execute( + f"CREATE PROCEDURE dbo.[{proc}] @items dbo.[{name}] READONLY AS " + f"SELECT {projection} FROM @items" + ) + db_connection.commit() + try: + with localcontext() as ctx: + ctx.prec = 5 + source = [list(row), [None] * len(row)] + value = TableValuedParameter(f"dbo.{name}", columns, source) + cursor.execute(f"EXEC dbo.[{proc}] ?", value) + results = [tuple(r) for r in cursor.fetchall()] + assert len(results) == 2 + actual = next(r for r in results if r[0] is not None) + expected = list(row) + # Verify every UTF-16 byte stored by SQL Server, including embedded NULs, + # independently of the result decoder's text-terminator handling. + expected[7] = row[7].encode("utf-16le") + assert actual == tuple(expected) + assert (None,) * len(row) in results + assert source == [list(row), [None] * len(row)] + finally: + cursor.execute(f"DROP PROCEDURE dbo.[{proc}]") + cursor.execute(f"DROP TYPE dbo.[{name}]") + db_connection.commit() + + +def test_date_range(cursor, db_connection): + name = f"Dates_{uuid.uuid4().hex}" + rows = [ + (None,), + (datetime.date.min,), + (datetime.date(1752, 12, 31),), + (datetime.date(1753, 1, 1),), + (datetime.date.max,), + ] + cursor.execute(f"CREATE TYPE dbo.[{name}] AS TABLE (value date NULL)") + db_connection.commit() + try: + value = TableValuedParameter(f"dbo.{name}", [("value", SQL_TYPE_DATE)], rows) + cursor.execute("SELECT value FROM ? ORDER BY value", value) + assert [tuple(row) for row in cursor.fetchall()] == rows + finally: + cursor.execute(f"DROP TYPE dbo.[{name}]") + db_connection.commit() + + +def test_generator_is_lazy_bounded_and_invokes_once(cursor, tvp_schema): + seen = [] + live = weakref.WeakSet() + peak = 0 + + class LiveRow(list): + __hash__ = object.__hash__ + + def rows(): + nonlocal peak + for i in range(5000): + row = LiveRow((i, "x" * 1000)) + live.add(row) + peak = max(peak, len(live)) + assert len(live) <= 2, "The driver retained earlier generator rows" + seen.append(i) + yield row + + value = items(tvp_schema, rows()) + assert seen == [] + cursor.execute(f"EXEC [{tvp_schema}].CountRows ?", value) + assert tuple(cursor.fetchone()) == (5000, 4999 * 5000 // 2) + assert len(seen) == 5000 + assert peak <= 2 + assert len(live) == 0 + cursor.execute(f"SELECT COUNT(*) FROM [{tvp_schema}].Calls") + assert cursor.fetchone()[0] == 1 + + +@pytest.mark.parametrize("failure", [ValueError("source failed"), KeyboardInterrupt("interrupted")]) +def test_generator_failure_preserves_error_and_recovers(cursor, tvp_schema, failure): + def rows(): + yield 1, "first" + raise failure + + with pytest.raises(type(failure)) as error: + cursor.execute(f"EXEC [{tvp_schema}].Echo ?, ?, ?", 0, items(tvp_schema, rows()), "") + assert error.value is failure + assert cursor.execute("SELECT 42").fetchone()[0] == 42 + cursor.execute(f"SELECT COUNT(*) FROM [{tvp_schema}].Calls") + assert cursor.fetchone()[0] == 0 + cursor.execute(f"EXEC [{tvp_schema}].Echo ?, ?, ?", 0, items(tvp_schema, [(2, "ok")]), "") + assert tuple(cursor.fetchone()) == (2, "ok") + + +@pytest.mark.parametrize("rows", [[], [(1, None)], [(1, ""), (2, "🚀" * 10000)]]) +def test_streaming_empty_null_and_large_cells(cursor, tvp_schema, rows): + cursor.execute(f"EXEC [{tvp_schema}].Echo ?, ?, ?", 0, items(tvp_schema, iter(rows)), "") + assert [tuple(row) for row in cursor.fetchall()] == rows + + +@pytest.mark.parametrize("bad", [[(1,)], [(1, "a", "extra")], [1], [(1, object())]]) +def test_bad_stream_row_recovers(cursor, tvp_schema, bad): + with pytest.raises((TypeError, ValueError)): + cursor.execute(f"EXEC [{tvp_schema}].CountRows ?", items(tvp_schema, iter(bad))) + assert cursor.execute("SELECT 1").fetchone()[0] == 1 + + +def test_implicit_nested_sequences_are_not_tvps(cursor): + for value in ([], [(1, "a")], ((1, "a"),)): + with pytest.raises(TypeError, match="Unsupported parameter type"): + cursor.execute("SELECT ?", (value,)) + assert cursor.execute("SELECT 1").fetchone()[0] == 1 + + +@pytest.mark.parametrize( + "kwargs,error", + [ + ({"rows": None}, TypeError), + ({"columns": []}, TypeError), + ({"columns": [("x", SQL_WVARCHAR)]}, ValueError), + ({"columns": [("x", SQL_DECIMAL, 39, 2)]}, ValueError), + ({"columns": [("x", SQL_DECIMAL, 4, 5)]}, ValueError), + ({"columns": [("x", 9999)]}, ValueError), + ({"type_name": "db.dbo.Items"}, ValueError), + ({"type_name": "dbo.Items", "schema": "dbo"}, ValueError), + ], +) +def test_explicit_contract_validation(kwargs, error): + arguments = {"type_name": "Items", "columns": [("id", SQL_INTEGER)], "rows": []} + arguments.update(kwargs) + with pytest.raises(error): + TableValuedParameter(**arguments) + + +@pytest.mark.parametrize("container", [list, tuple, UserList]) +def test_column_sequences_match_public_contract(container): + columns = container([("id", SQL_INTEGER), ("label", SQL_WVARCHAR, 100)]) + value = TableValuedParameter("Items", columns, []) + assert value.columns == (("id", SQL_INTEGER, 10, 0), ("label", SQL_WVARCHAR, 100, 0)) + assert columns == container([("id", SQL_INTEGER), ("label", SQL_WVARCHAR, 100)]) + + +@pytest.mark.parametrize("bad", [2**31, -(2**31) - 1]) +def test_explicit_integer_width_does_not_truncate(cursor, tvp_schema, bad): + with pytest.raises(ValueError, match="out of range"): + cursor.execute(f"EXEC [{tvp_schema}].CountRows ?", items(tvp_schema, [(bad, None)])) + assert cursor.execute("SELECT 1").fetchone()[0] == 1 + + +def test_precise_decimal_streaming(cursor, db_connection): + name = f"Decimal_{uuid.uuid4().hex}" + proc = f"Proc_{name}" + cursor.execute(f"CREATE TYPE dbo.[{name}] AS TABLE (amount decimal(38,6))") + cursor.execute( + f"CREATE PROC dbo.[{proc}] @items dbo.[{name}] READONLY AS " "SELECT amount FROM @items" + ) + db_connection.commit() + values = [ + Decimal("12345678901234567890123456789012.123456"), + None, + Decimal("0.000001"), + Decimal("-99999999999999999999999999999999.999999"), + ] + try: + with localcontext() as context: + context.prec = 3 + value = TableValuedParameter( + name, [("amount", SQL_DECIMAL, 38, 6)], ((v,) for v in values) + ) + cursor.execute(f"EXEC dbo.[{proc}] ?", value) + assert [r[0] for r in cursor.fetchall()] == values + for value in (Decimal("NaN"), Decimal("Infinity")): + with pytest.raises(ValueError, match="non-finite"): + cursor.execute( + f"EXEC dbo.[{proc}] ?", + TableValuedParameter(name, [("amount", SQL_DECIMAL, 38, 6)], [(value,)]), + ) + assert cursor.execute("SELECT 1").fetchone()[0] == 1 + finally: + cursor.execute(f"DROP PROC dbo.[{proc}]") + cursor.execute(f"DROP TYPE dbo.[{name}]") + db_connection.commit() + + +def test_sync_temporal_scale_six(cursor, db_connection): + name = f"Times_{uuid.uuid4().hex}" + proc = f"Proc_{name}" + cursor.execute(f"CREATE TYPE dbo.[{name}] AS TABLE (t time(6), d datetime2(6))") + cursor.execute(f"CREATE PROC dbo.[{proc}] @items dbo.[{name}] READONLY AS SELECT * FROM @items") + db_connection.commit() + rows = [ + (datetime.time(1, 2, 3, 123456), datetime.datetime(2024, 2, 29, 1, 2, 3, 123456)), + (None, None), + ] + try: + value = TableValuedParameter( + name, [("t", SQL_TYPE_TIME, 16, 6), ("d", SQL_TYPE_TIMESTAMP, 26, 6)], iter(rows) + ) + cursor.execute(f"EXEC dbo.[{proc}] ?", value) + assert [tuple(row) for row in cursor.fetchall()] == rows + finally: + cursor.execute(f"DROP PROC dbo.[{proc}]") + cursor.execute(f"DROP TYPE dbo.[{name}]") + db_connection.commit() + + +def test_schema_mismatch_and_server_error_recovery(cursor, tvp_schema): + with pytest.raises(DatabaseError): + cursor.execute( + f"EXEC [{tvp_schema}].CountRows ?", + TableValuedParameter("DoesNotExist", [("id", SQL_INTEGER)], [], schema=tvp_schema), + ) + assert cursor.execute("SELECT 42").fetchone()[0] == 42 + with pytest.raises(DatabaseError): + cursor.execute( + f"EXEC [{tvp_schema}].CountRows ?", + TableValuedParameter( + "Items", [("only_column", SQL_INTEGER)], [(1,)], schema=tvp_schema + ), + ) + assert cursor.execute("SELECT 43").fetchone()[0] == 43 + + +def test_multiple_generator_tvps_and_large_scalar(cursor, tvp_schema): + cursor.execute( + f"EXEC [{tvp_schema}].Pair ?, ?", + items(tvp_schema, iter([(1, "left"), (2, None)])), + items(tvp_schema, iter([(3, "right")])), + ) + assert sorted(tuple(row) for row in cursor.fetchall()) == [(1, "left"), (2, None), (3, "right")] + suffix = "🚀" * 8000 + cursor.execute( + f"EXEC [{tvp_schema}].Echo ?, ?, ?", 0, items(tvp_schema, iter([(1, "head")])), suffix + ) + assert tuple(cursor.fetchone()) == (1, "head" + suffix) + + +def test_executemany_and_inputsizes_reject_tvps_before_consumption(cursor, tvp_schema): + def rows(): + pytest.fail("Unsupported operation consumed a TVP row source") + yield 1, "never" + + value = items(tvp_schema, rows()) + with pytest.raises(NotSupportedError, match="executemany"): + cursor.executemany(f"EXEC [{tvp_schema}].CountRows ?", [(value,)]) + cursor.setinputsizes([SQL_INTEGER]) + try: + with pytest.raises(NotSupportedError, match="setinputsizes"): + cursor.execute(f"EXEC [{tvp_schema}].CountRows ?", value) + finally: + cursor.setinputsizes([]) + assert cursor.execute("SELECT 1").fetchone()[0] == 1 + + +@pytest.mark.parametrize("operation", ["execute", "close"]) +def test_generator_cannot_reenter_its_cursor(cursor, tvp_schema, operation): + def rows(): + yield 1, "first" + if operation == "execute": + cursor.execute("SELECT 99") + else: + cursor.close() + yield 2, "second" + + with pytest.raises(DatabaseError, match="while it is executing"): + cursor.execute(f"EXEC [{tvp_schema}].CountRows ?", items(tvp_schema, rows())) + assert cursor.execute("SELECT 42").fetchone()[0] == 42 + + +def _run_isolated_stream_checks(mode, schema): + connection_string = os.environ["DB_CONNECTION_STRING"] + if mode == "forwarder": + target = _parse_server(connection_string) + assert target is not None + connection_string = _replace_server(connection_string, *_start_forwarder(target)) + connection = connect(connection_string, autocommit=True) + cursor = connection.cursor() + try: + if mode == "disconnect": + original = ValueError("original source failure after disconnect") + source_errors = [] + + def rows(): + yield 1, "first" + try: + connection.close() + except DatabaseError as error: + source_errors.append(error) + raise + source_errors.append(original) + raise original + + try: + cursor.execute(f"EXEC [{schema}].CountRows ?", items(schema, rows())) + except (ValueError, DatabaseError) as error: + assert error is source_errors[0] + else: + raise AssertionError("Lost the original iterator error") + print("OK TVP streaming") + return + elif mode == "cancel": + waiting = threading.Event() + resume = threading.Event() + errors = [] + + def rows(): + yield 1, "first" + waiting.set() + assert resume.wait(10) + yield 2, "second" + + def execute(): + try: + cursor.execute(f"EXEC [{schema}].CountRows ?", items(schema, rows())) + except BaseException as error: + errors.append(error) + + worker = threading.Thread(target=execute) + worker.start() + assert waiting.wait(10) + try: + cursor.hstmt._cancel() + finally: + resume.set() + worker.join(10) + assert not worker.is_alive() + assert len(errors) == 1 and isinstance(errors[0], DatabaseError) + else: + count = 20000 if mode == "memory" else 1000 + if mode == "memory": + import resource + + before = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss + cursor.execute( + f"EXEC [{schema}].CountRows ?", + items(schema, ((i, str(i).ljust(4096, "x")) for i in range(count))), + ) + assert tuple(cursor.fetchone()) == (count, count * (count - 1) // 2) + if mode == "memory": + growth = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss - before + if sys.platform != "darwin": + growth *= 1024 + # Input exceeds 80MB of Python text / 160MB UTF-16. Additional + # native and Python memory must stay below a table-sized copy. + assert growth < 64 * 1024 * 1024, growth + print(f"peak RSS growth: {growth} bytes") + assert cursor.execute("SELECT 42").fetchone()[0] == 42 + finally: + cursor.close() + connection.close() + print("OK TVP streaming") + + +@pytest.mark.parametrize( + "mode", + [ + "cancel", + "forwarder", + "disconnect", + pytest.param( + "memory", + marks=[ + pytest.mark.stress, + pytest.mark.skipif(sys.platform == "win32", reason="POSIX resource accounting"), + ], + ), + ], +) +def test_isolated_stream_lifecycle(conn_str, tvp_schema, mode): + if mode == "forwarder": + target = _parse_server(conn_str) + if target is None or "\\" in target[0]: + pytest.skip("TCP forwarding requires a host, not a LocalDB or named-pipe instance") + env = os.environ.copy() + env["PYTHONPATH"] = os.pathsep.join(sys.path) + # The memory stress case sends 160MB, unlike the short deadlock probes. + # Keep the same row count and RSS bound, but allow slower transport. + timeout = 120 if mode == "memory" else WATCHDOG_SECONDS + try: + result = subprocess.run( + [ + sys.executable, + "-u", + "-c", + "import faulthandler;" + f"faulthandler.dump_traceback_later({timeout - 5});" + "from test_028_table_valued_parameters import _run_isolated_stream_checks;" + f"_run_isolated_stream_checks({mode!r}, {tvp_schema!r})", + ], + env=env, + capture_output=True, + text=True, + timeout=timeout, + ) + except subprocess.TimeoutExpired as error: + output = "".join( + chunk.decode(errors="replace") if isinstance(chunk, bytes) else chunk or "" + for chunk in (error.stdout, error.stderr) + ) + pytest.fail(f"TVP {mode} subprocess exceeded {timeout}s:\n{output}") + assert result.returncode == 0, result.stdout + result.stderr + assert "OK TVP streaming" in result.stdout + print(result.stdout, end="")