diff --git a/CHANGELOG.md b/CHANGELOG.md index aab046b3a..160367bf6 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 input table-valued parameter support for stored procedure + calls. Pass a materialized list or tuple of rows as one `execute()` parameter, + including empty tables and positional or named parameters. The Rust ODBC + provider does not yet support TVPs. - 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/mssql_python/cursor.py b/mssql_python/cursor.py index 1cb12cb48..34f21ecb7 100644 --- a/mssql_python/cursor.py +++ b/mssql_python/cursor.py @@ -31,6 +31,7 @@ from mssql_python.row import Row 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, @@ -365,6 +366,7 @@ def __init__(self, connection: "Connection", timeout: int = 0) -> None: # an HSTMT exists. self.closed: bool = False self.hstmt: Optional[Any] = None + self._tvp_metadata_hstmt: Optional[Any] = None self._connection: "Connection" = connection # Store as private attribute self._timeout: int = timeout @@ -996,7 +998,7 @@ def _allocate_statement_handle(self) -> None: """ self.hstmt = self._connection._conn.alloc_statement_handle() - def _set_timeout(self) -> None: + def _set_timeout(self, statement_handle=None) -> None: """ Set the query timeout attribute on the statement handle. This is called once when the cursor is created and after any handle reallocation. @@ -1006,12 +1008,13 @@ def _set_timeout(self) -> None: logger.debug("_set_timeout: Setting query timeout=%d seconds", self._timeout) try: timeout_value = int(self._timeout) + target_handle = statement_handle or self.hstmt ret = ddbc_bindings.DDBCSQLSetStmtAttr( - self.hstmt, + target_handle, ddbc_sql_const.SQL_ATTR_QUERY_TIMEOUT.value, timeout_value, ) - check_error(ddbc_sql_const.SQL_HANDLE_STMT.value, self.hstmt, ret) + check_error(ddbc_sql_const.SQL_HANDLE_STMT.value, target_handle, ret) logger.debug("Query timeout set to %d seconds", timeout_value) except Exception as e: # pylint: disable=broad-exception-caught logger.warning("Failed to set query timeout: %s", str(e)) @@ -1081,6 +1084,9 @@ def close(self) -> None: self.hstmt.free() self.hstmt = None logger.debug("SQLFreeHandle succeeded") + if self._tvp_metadata_hstmt: + self._tvp_metadata_hstmt.free() + self._tvp_metadata_hstmt = None self._clear_rownumber() self.closed = True @@ -1730,8 +1736,9 @@ 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 remains one + # parameter by wrapping its rows in the ordinary outer parameter tuple: + # execute("EXEC dbo.proc ?", (rows,)) with perf_phase("py::execute::param_prep"): if parameters: # Check if single parameter is a nested container that should be unwrapped @@ -1769,6 +1776,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, (list, tuple)) 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( @@ -1786,22 +1798,33 @@ 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 try: - + with perf_phase("py::execute::cpp_call"): + if parameters: + if any(isinstance(parameter, (list, tuple)) 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", + ) + if self._tvp_metadata_hstmt is None: + self._tvp_metadata_hstmt = ( + self._connection._conn.alloc_statement_handle() + ) + self._set_timeout(self._tvp_metadata_hstmt) + ret = ddbc_bindings.DDBCSQLExecute( + self.hstmt, + self._tvp_metadata_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 @@ -2415,7 +2438,6 @@ def executemany( # pylint: disable=too-many-locals,too-many-branches,too-many-s if hasattr(seq_of_parameters, "__getitem__") else next(iter(seq_of_parameters)) ) - if isinstance(first_row, dict): # pyformat style - convert all rows # Parse parameter names from SQL (determines order for all rows) @@ -2480,11 +2502,15 @@ def executemany( # pylint: disable=too-many-locals,too-many-branches,too-many-s # Prepare parameter type information with perf_phase("py::executemany::param_type_detection"): for col_index in range(param_count): - column = ( - [row[col_index] for row in seq_of_parameters] - if hasattr(seq_of_parameters, "__getitem__") - else [] - ) + column = [] + for row in seq_of_parameters: + value = row[col_index] + if isinstance(value, (list, tuple)): + raise NotSupportedError( + "executemany does not support table-valued parameters", + "Use execute for each TVP operation", + ) + column.append(value) 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/pybind/ddbc_bindings.cpp b/mssql_python/pybind/ddbc_bindings.cpp index 35a3b6da4..df0bcf9b5 100644 --- a/mssql_python/pybind/ddbc_bindings.cpp +++ b/mssql_python/pybind/ddbc_bindings.cpp @@ -23,6 +23,11 @@ #include // std::forward #include // CPython datetime API (PyDateTime_IMPORT, PyDateTime_GET_*, etc.) +SQLRETURN BindParameterArray(SqlHandle& handle, SQLHANDLE hStmt, const py::list& columnwise_params, + std::vector& paramInfos, size_t paramSetSize, + std::vector>& paramBuffers, + const std::string& charEncoding); +static bool is_python_finalizing(); //------------------------------------------------------------------------------------------------- // Macro definitions @@ -227,6 +232,7 @@ SQLExecuteFunc SQLExecute_ptr = nullptr; SQLRowCountFunc SQLRowCount_ptr = nullptr; SQLGetStmtAttrFunc SQLGetStmtAttr_ptr = nullptr; SQLSetDescFieldFunc SQLSetDescField_ptr = nullptr; +SQLGetDescFieldFunc SQLGetDescField_ptr = nullptr; // Data retrieval APIs SQLFetchFunc SQLFetch_ptr = nullptr; @@ -448,13 +454,487 @@ static void PreResolveUnknownNullTypes(SqlHandle& handle, SQLHANDLE hStmt, } } +static bool IsIntegerCType(SQLSMALLINT cType) { + return cType == SQL_C_TINYINT || cType == SQL_C_UTINYINT || cType == SQL_C_SHORT || + cType == SQL_C_LONG || cType == SQL_C_SBIGINT; +} + +static bool IsTextSqlType(SQLSMALLINT sqlType) { + return sqlType == SQL_VARCHAR || sqlType == SQL_WVARCHAR; +} + +static ParamInfo MergeTableColumnInfo(const py::list& values, + const std::vector& valueInfos, + size_t columnIndex) { + ParamInfo merged; + bool hasValue = false; + bool integerColumn = false; + bool textColumn = false; + SQLULEN maxSize = 0; + SQLULEN maxIntegerDigits = 0; + SQLSMALLINT maxScale = 0; + + for (size_t rowIndex = 0; rowIndex < valueInfos.size(); ++rowIndex) { + const ParamInfo& current = valueInfos[rowIndex]; + if (values[rowIndex].is_none()) { + continue; + } + if (current.isTVP) { + throw py::type_error("TVP cells cannot contain nested row sequences"); + } + + const bool currentInteger = IsIntegerCType(current.paramCType); + const bool currentText = + current.paramCType == SQL_C_WCHAR && IsTextSqlType(current.paramSQLType); + if (!hasValue) { + merged = current; + integerColumn = currentInteger; + textColumn = currentText; + hasValue = true; + } else if (integerColumn && currentInteger) { + // Scalar detection already range-checks integers; retain its widest type. + if (current.columnSize > merged.columnSize) { + merged = current; + } + } else if (!(textColumn && currentText) && + (merged.paramCType != current.paramCType || + merged.paramSQLType != current.paramSQLType)) { + throw py::type_error("TVP column " + std::to_string(columnIndex) + + " contains incompatible Python types at row " + + std::to_string(rowIndex)); + } + + if (currentInteger) { + continue; + } else if (currentText) { + if (current.paramSQLType == SQL_WVARCHAR) { + merged.paramSQLType = SQL_WVARCHAR; + } + SQLULEN size = + current.isDAE ? static_cast(current.utf16Len) : current.columnSize; + maxSize = std::max(maxSize, size); + } else if (current.paramCType == SQL_C_BINARY) { + Py_ssize_t size = PyObject_Length(values[rowIndex].ptr()); + if (size < 0) + throw py::error_already_set(); + maxSize = std::max(maxSize, static_cast(size)); + } else if (current.paramCType == SQL_C_NUMERIC) { + SQLULEN integerDigits = + current.columnSize > static_cast(current.decimalDigits) + ? current.columnSize - static_cast(current.decimalDigits) + : 0; + maxIntegerDigits = std::max(maxIntegerDigits, integerDigits); + maxScale = std::max(maxScale, current.decimalDigits); + } else { + maxSize = std::max(maxSize, current.columnSize); + maxScale = std::max(maxScale, current.decimalDigits); + } + } + + if (!hasValue) { + merged.paramCType = SQL_C_DEFAULT; + merged.paramSQLType = SQL_VARCHAR; + merged.columnSize = 1; + merged.decimalDigits = 0; + return merged; + } + + if (integerColumn) { + if (merged.paramCType == SQL_C_TINYINT) { + merged.paramCType = SQL_C_UTINYINT; + } + } else if (textColumn || merged.paramCType == SQL_C_BINARY) { + merged.columnSize = std::max(maxSize, 1); + merged.isDAE = false; + } else if (merged.paramCType == SQL_C_NUMERIC) { + merged.decimalDigits = maxScale; + merged.columnSize = std::max(maxIntegerDigits + maxScale, 1); + } else { + merged.columnSize = std::max(maxSize, 1); + merged.decimalDigits = maxScale; + } + + return merged; +} + +static std::vector DetectTableColumnTypes(py::list& columnValues) { + std::vector columnInfos(columnValues.size()); + for (size_t columnIndex = 0; columnIndex < columnValues.size(); ++columnIndex) { + py::list values = columnValues[columnIndex].cast(); + py::list detectedValues = values.attr("copy")(); + std::vector valueInfos = DetectParamTypes(detectedValues.ptr(), Py_None); + ParamInfo& info = columnInfos[columnIndex]; + info = MergeTableColumnInfo(detectedValues, valueInfos, columnIndex); + if (info.paramCType == SQL_C_NUMERIC) { + if (info.columnSize > MAX_NUMERIC_PRECISION) { + throw py::value_error("TVP Decimal column requires precision greater than 38"); + } + // Preserve each coefficient's scale using exact text, while binding a common + // source NUMERIC precision/scale. Buffer capacity is not numeric precision. + py::str formatSpec("f"); + info.paramCType = SQL_C_CHAR; + info.bufferSize = 1; + for (size_t rowIndex = 0; rowIndex < values.size(); ++rowIndex) { + if (values[rowIndex].is_none()) { + continue; + } + py::object formatted = + steal(PyObject_Format(values[rowIndex].ptr(), formatSpec.ptr())); + if (!formatted) + throw py::error_already_set(); + py::object ascii = steal(PyUnicode_AsASCIIString(formatted.ptr())); + if (!ascii) + throw py::error_already_set(); + info.bufferSize = + std::max(info.bufferSize, static_cast(PyBytes_GET_SIZE(ascii.ptr()))); + detectedValues[rowIndex] = std::move(ascii); + } + } + columnValues[columnIndex] = std::move(detectedValues); + } + return columnInfos; +} + +static SQLRETURN ReadDescriptorString(SQLHDESC descriptor, SQLSMALLINT recordNumber, + SQLSMALLINT fieldIdentifier, std::u16string& value) { + SQLWCHAR buffer[256] = {}; + SQLINTEGER length = 0; + SQLRETURN rc = SQLGetDescField_ptr(descriptor, recordNumber, fieldIdentifier, buffer, + static_cast(sizeof(buffer)), &length); + if (!SQL_SUCCEEDED(rc)) { + return rc; + } + if (length <= 0) { + value.clear(); + return rc; + } + size_t codeUnits = std::min(static_cast(length) / sizeof(SQLWCHAR), + (sizeof(buffer) / sizeof(buffer[0])) - 1); + value = dupeSqlWCharAsUtf16Le(buffer, codeUnits); + return rc; +} + +[[noreturn]] static void ThrowTableError(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); +} + +class TableMetadataScope { + public: + explicit TableMetadataScope(SQLHANDLE statementHandle) : handle(statementHandle) {} + ~TableMetadataScope() { + // Error paths can leave unread metadata; closing it can require network I/O. + if (!is_python_finalizing() && PyGILState_Check()) { + py::gil_scoped_release release; + SQLFreeStmt_ptr(handle, SQL_CLOSE); + } else { + SQLFreeStmt_ptr(handle, SQL_CLOSE); + } + SQLSetStmtAttr_ptr( + handle, SQL_SOPT_SS_NAME_SCOPE, + reinterpret_cast(static_cast(SQL_SS_NAME_SCOPE_TABLE)), + SQL_IS_INTEGER); + SQLSetStmtAttr_ptr(handle, SQL_ATTR_METADATA_ID, + reinterpret_cast(static_cast(SQL_FALSE)), + SQL_IS_INTEGER); + } + + private: + SQLHANDLE handle; +}; + +static std::vector LoadTableColumnMetadata(const SqlHandlePtr& metadataHandle, + const std::u16string& catalog, + const std::u16string& schema, + const std::u16string& typeName) { + if (!metadataHandle || !metadataHandle->get()) { + throw std::runtime_error("TVP column metadata requires a valid statement handle"); + } + + SQLHANDLE hMetadataStmt = metadataHandle->get(); + { + py::gil_scoped_release release; + SQLFreeStmt_ptr(hMetadataStmt, SQL_CLOSE); + SQLFreeStmt_ptr(hMetadataStmt, SQL_RESET_PARAMS); + } + + SQLRETURN rc = SQLSetStmtAttr_ptr(hMetadataStmt, SQL_ATTR_METADATA_ID, + reinterpret_cast(static_cast(SQL_TRUE)), + SQL_IS_INTEGER); + if (!SQL_SUCCEEDED(rc)) { + ThrowTableError(metadataHandle, rc, "Failed to enable metadata identifier mode"); + } + TableMetadataScope metadataScope(hMetadataStmt); + + rc = SQLSetStmtAttr_ptr( + hMetadataStmt, SQL_SOPT_SS_NAME_SCOPE, + reinterpret_cast(static_cast(SQL_SS_NAME_SCOPE_TABLE_TYPE)), + SQL_IS_INTEGER); + if (!SQL_SUCCEEDED(rc)) { + ThrowTableError(metadataHandle, rc, "Failed to select table-type metadata scope"); + } + { + py::gil_scoped_release release; + rc = SQLColumns_ptr(hMetadataStmt, + catalog.empty() ? nullptr : reinterpretU16stringAsSqlWChar(catalog), + catalog.empty() ? 0 : SQL_NTS, + schema.empty() ? nullptr : reinterpretU16stringAsSqlWChar(schema), + schema.empty() ? 0 : SQL_NTS, reinterpretU16stringAsSqlWChar(typeName), + SQL_NTS, nullptr, 0); + } + if (!SQL_SUCCEEDED(rc)) { + ThrowTableError(metadataHandle, rc, "Failed to discover TVP columns"); + } + + std::vector columns; + while (true) { + { + py::gil_scoped_release release; + rc = SQLFetch_ptr(hMetadataStmt); + } + if (rc == SQL_NO_DATA) { + break; + } + if (!SQL_SUCCEEDED(rc)) { + ThrowTableError(metadataHandle, rc, "Failed to fetch TVP column metadata"); + } + + SQLSMALLINT sqlType = SQL_UNKNOWN_TYPE; + SQLINTEGER columnSize = 0; + SQLSMALLINT decimalDigits = 0; + SQLLEN indicator = 0; + + { + py::gil_scoped_release release; + rc = SQLGetData_ptr(hMetadataStmt, 5, SQL_C_SSHORT, &sqlType, sizeof(sqlType), &indicator); + if (SQL_SUCCEEDED(rc)) { + rc = SQLGetData_ptr(hMetadataStmt, 7, SQL_C_SLONG, &columnSize, sizeof(columnSize), + &indicator); + } + if (SQL_SUCCEEDED(rc)) { + rc = SQLGetData_ptr(hMetadataStmt, 9, SQL_C_SSHORT, &decimalDigits, + sizeof(decimalDigits), &indicator); + if (indicator == SQL_NULL_DATA) { + decimalDigits = 0; + rc = SQL_SUCCESS; + } + } + } + if (!SQL_SUCCEEDED(rc)) { + ThrowTableError(metadataHandle, rc, "Failed to read TVP column metadata"); + } + + columns.push_back({ + sqlType, static_cast(std::max(columnSize, 0)), decimalDigits, + }); + } + + if (columns.empty()) { + throw std::runtime_error("No column metadata was returned for the table type"); + } + return columns; +} + +static SQLRETURN ResolveTableValuedParamTypes(SqlHandle& handle, SQLHANDLE hStmt, + const SqlHandlePtr& metadataHandle, + std::vector& paramInfos) { + for (size_t paramIndex = 0; paramIndex < paramInfos.size(); ++paramIndex) { + ParamInfo& info = paramInfos[paramIndex]; + if (!info.isTVP) { + continue; + } + + if (handle.tvpCache.find(static_cast(paramIndex)) == handle.tvpCache.end()) { + DescribedParamInfo described; + SQLSMALLINT nullable; + RETCODE rc; + { + py::gil_scoped_release release; + rc = SQLDescribeParam_ptr(hStmt, static_cast(paramIndex + 1), + &described.sqlType, &described.columnSize, + &described.decimalDigits, &nullable); + } + if (!SQL_SUCCEEDED(rc)) { + return rc; + } + if (described.sqlType != SQL_SS_TABLE) { + throw py::type_error( + "Nested row sequences can only be bound to a table-valued parameter"); + } + + SQLHDESC implementationDescriptor = nullptr; + rc = SQLGetStmtAttr_ptr(hStmt, SQL_ATTR_IMP_PARAM_DESC, &implementationDescriptor, 0, + nullptr); + if (!SQL_SUCCEEDED(rc)) { + return rc; + } + + std::u16string catalog; + std::u16string schema; + std::u16string typeName; + const SQLSMALLINT recordNumber = static_cast(paramIndex + 1); + rc = ReadDescriptorString(implementationDescriptor, recordNumber, + SQL_CA_SS_CATALOG_NAME, catalog); + if (SQL_SUCCEEDED(rc)) { + rc = ReadDescriptorString(implementationDescriptor, recordNumber, + SQL_CA_SS_SCHEMA_NAME, schema); + } + if (SQL_SUCCEEDED(rc)) { + rc = ReadDescriptorString(implementationDescriptor, recordNumber, + SQL_CA_SS_TYPE_NAME, typeName); + } + if (!SQL_SUCCEEDED(rc)) { + return rc; + } + if (typeName.empty()) { + throw std::runtime_error("The prepared statement did not expose a TVP type name"); + } + handle.tvpCache[static_cast(paramIndex)] = { + catalog, + schema, + typeName, + LoadTableColumnMetadata(metadataHandle, catalog, schema, typeName), + }; + } + } + return SQL_SUCCESS; +} + +static SQLRETURN BindTableValuedParameter(const SqlHandlePtr& statementHandle, SQLHANDLE hStmt, + int paramIndex, const py::handle& param, + std::vector>& paramBuffers, + const std::string& charEncoding) { + SqlHandle& handle = *statementHandle; + const py::sequence rows = py::reinterpret_borrow(param); + const SQLSMALLINT recordNumber = static_cast(paramIndex + 1); + + size_t columnCount = 0; + py::list columnValues; + std::vector columnInfos; + auto metadataEntry = handle.tvpCache.find(paramIndex); + if (metadataEntry == handle.tvpCache.end()) { + throw std::runtime_error("TVP column metadata was not resolved before binding"); + } + const TvpParamInfo& tableMetadata = metadataEntry->second; + const std::vector& declaredColumns = tableMetadata.columns; + if (!rows.empty()) { + const py::handle firstRow = rows[0]; + if (!PyList_Check(firstRow.ptr()) && !PyTuple_Check(firstRow.ptr())) { + throw py::type_error("TVP rows must be list or tuple objects"); + } + columnCount = py::len(firstRow); + if (columnCount == 0) { + throw py::value_error("TVP rows must contain at least one column"); + } + if (columnCount != declaredColumns.size()) { + throw py::value_error("TVP rows contain " + std::to_string(columnCount) + + " columns; the declared table type requires " + + std::to_string(declaredColumns.size())); + } + columnValues = py::list(columnCount); + for (size_t columnIndex = 0; columnIndex < columnCount; ++columnIndex) { + columnValues[columnIndex] = py::list(); + } + for (size_t rowIndex = 0; rowIndex < rows.size(); ++rowIndex) { + const py::sequence row = py::reinterpret_borrow(rows[rowIndex]); + if (!PyList_Check(row.ptr()) && !PyTuple_Check(row.ptr())) { + throw py::type_error("TVP rows must be list or tuple objects"); + } + if (static_cast(py::len(row)) != columnCount) { + throw py::value_error("TVP rows must all contain the same number of columns"); + } + for (size_t columnIndex = 0; columnIndex < columnCount; ++columnIndex) { + columnValues[columnIndex].cast().append(row[columnIndex]); + } + } + columnInfos = DetectTableColumnTypes(columnValues); + for (size_t columnIndex = 0; columnIndex < columnCount; ++columnIndex) { + ParamInfo& inferred = columnInfos[columnIndex]; + const DescribedParamInfo& declared = declaredColumns[columnIndex]; + // Bind the source representation; SQL Server converts it to the table's + // declared column type. Only an all-NULL column needs destination metadata. + if (inferred.paramCType == SQL_C_DEFAULT) { + inferred.paramSQLType = declared.sqlType; + inferred.columnSize = std::max(declared.columnSize, 1); + inferred.decimalDigits = declared.decimalDigits; + } else { + if (inferred.paramSQLType == SQL_WVARCHAR && inferred.columnSize > MAX_INLINE_CHAR) { + inferred.paramSQLType = SQL_WLONGVARCHAR; + } else if (inferred.paramSQLType == SQL_VARCHAR && + inferred.columnSize > MAX_INLINE_BINARY) { + inferred.paramSQLType = SQL_LONGVARCHAR; + } else if (inferred.paramSQLType == SQL_VARBINARY && + inferred.columnSize > MAX_INLINE_BINARY) { + inferred.paramSQLType = SQL_LONGVARBINARY; + } + } + } + } + + auto* typeName = AllocateParamBuffer(paramBuffers, tableMetadata.typeName); + auto* schema = AllocateParamBuffer(paramBuffers, tableMetadata.schema); + auto* rowCount = AllocateParamBuffer(paramBuffers); + // TVPs overload ColumnSize with row capacity and the indicator with the rows available. + *rowCount = rows.empty() ? SQL_DEFAULT_PARAM : static_cast(rows.size()); + + SQLRETURN rc = SQLBindParameter_ptr( + hStmt, static_cast(recordNumber), SQL_PARAM_INPUT, SQL_C_BINARY, SQL_SS_TABLE, + static_cast(rows.size()), 0, typeName->data(), + static_cast(typeName->size() * sizeof(SQLWCHAR)), rowCount); + if (!SQL_SUCCEEDED(rc)) { + return rc; + } + if (!schema->empty()) { + SQLHDESC implementationDescriptor = nullptr; + rc = SQLGetStmtAttr_ptr(hStmt, SQL_ATTR_IMP_PARAM_DESC, &implementationDescriptor, 0, + nullptr); + if (!SQL_SUCCEEDED(rc)) { + return rc; + } + rc = SQLSetDescField_ptr(implementationDescriptor, recordNumber, SQL_CA_SS_SCHEMA_NAME, + reinterpretU16stringAsSqlWChar(*schema), + static_cast(schema->size() * sizeof(SQLWCHAR))); + if (!SQL_SUCCEEDED(rc)) { + return rc; + } + } + if (rows.empty()) { + return rc; + } + + rc = SQLSetStmtAttr_ptr(hStmt, SQL_SOPT_SS_PARAM_FOCUS, + reinterpret_cast(static_cast(recordNumber)), + SQL_IS_INTEGER); + if (!SQL_SUCCEEDED(rc)) { + return rc; + } + + try { + rc = BindParameterArray(handle, hStmt, columnValues, columnInfos, rows.size(), + paramBuffers, charEncoding); + if (!SQL_SUCCEEDED(rc)) { + // Successful focus restoration clears the original statement diagnostic. + ThrowTableError(statementHandle, rc, "Failed to bind TVP columns"); + } + } 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); +} + // 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, +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()); @@ -473,6 +953,15 @@ SQLRETURN BindParameters(SqlHandle& handle, SQLHANDLE hStmt, const py::list& par SQLLEN bufferLength = 0; SQLLEN* strLenOrIndPtr = nullptr; + if (paramInfo.isTVP) { + RETCODE rc = BindTableValuedParameter(statementHandle, hStmt, paramIndex, param, + paramBuffers, charEncoding); + if (!SQL_SUCCEEDED(rc)) { + return rc; + } + continue; + } + // TODO: Add more data types like money, guid, interval, TVPs etc. switch (paramInfo.paramCType) { case SQL_C_CHAR: { @@ -1445,6 +1934,7 @@ DriverHandle LoadDriverOrThrowException() { SQLRowCount_ptr = GetFunctionPointer(handle, "SQLRowCount"); SQLGetStmtAttr_ptr = GetFunctionPointer(handle, "SQLGetStmtAttrW"); SQLSetDescField_ptr = GetFunctionPointer(handle, "SQLSetDescFieldW"); + SQLGetDescField_ptr = GetFunctionPointer(handle, "SQLGetDescFieldW"); SQLFetch_ptr = GetFunctionPointer(handle, "SQLFetch"); SQLFetchScroll_ptr = GetFunctionPointer(handle, "SQLFetchScroll"); @@ -1477,17 +1967,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 && SQLGetDescField_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."); @@ -1583,7 +2073,7 @@ void SqlHandle::free() { PERF_TIMER("SqlHandle::free"); if (_handle && SQLFreeHandle_ptr) { // GH-610: Clear describe cache to prevent memory leak. - describeCache.clear(); + clearDescribeCache(); // Check if Python is shutting down using centralized helper function bool pythonShuttingDown = is_python_finalizing(); @@ -2032,12 +2522,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 SqlHandlePtr metadataStatementHandle, 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; @@ -2101,8 +2588,12 @@ SQLRETURN SQLExecute_wrap(const SqlHandlePtr statementHandle, } } + rc = ResolveTableValuedParamTypes(*statementHandle, hStmt, metadataStatementHandle, paramInfos); + if (!SQL_SUCCEEDED(rc)) + return rc; + std::vector> paramBuffers; - rc = BindParameters(*statementHandle, hStmt, params, paramInfos, paramBuffers, charEncoding); + rc = BindParameters(statementHandle, hStmt, params, paramInfos, paramBuffers, charEncoding); if (!SQL_SUCCEEDED(rc)) return rc; { @@ -2206,7 +2697,7 @@ SQLRETURN SQLExecute_wrap(const SqlHandlePtr statementHandle, SQLRETURN BindParameterArray(SqlHandle& handle, SQLHANDLE hStmt, const py::list& columnwise_params, std::vector& paramInfos, size_t paramSetSize, std::vector>& paramBuffers, - const std::string& charEncoding = "utf-8") { + const std::string& charEncoding) { PERF_TIMER("BindParameterArray"); LOG("BindParameterArray: Starting column-wise array binding - " "param_count=%zu, param_set_size=%zu", @@ -2297,7 +2788,8 @@ SQLRETURN BindParameterArray(SqlHandle& handle, SQLHANDLE hStmt, const py::list& } std::memcpy(wcharArray + i * (info.columnSize + 1), wstr.c_str(), (wstr.length() + 1) * sizeof(SQLWCHAR)); - strLenOrIndArray[i] = SQL_NTS; + strLenOrIndArray[i] = + static_cast(wstr.length() * sizeof(SQLWCHAR)); } } LOG("BindParameterArray: SQL_C_WCHAR bound - " @@ -2372,17 +2864,19 @@ SQLRETURN BindParameterArray(SqlHandle& handle, SQLHANDLE hStmt, const py::list& } case SQL_C_CHAR: case SQL_C_BINARY: { + const SQLULEN valueBufferSize = + info.bufferSize > 0 ? info.bufferSize : info.columnSize; LOG("BindParameterArray: Binding SQL_C_CHAR/BINARY array - " "param_index=%d, count=%zu, column_size=%zu, encoding='%s'", - paramIndex, paramSetSize, info.columnSize, charEncoding.c_str()); + paramIndex, paramSetSize, valueBufferSize, charEncoding.c_str()); char* charArray = AllocateParamBufferArray( - tempBuffers, paramSetSize * (info.columnSize + 1)); + tempBuffers, paramSetSize * (valueBufferSize + 1)); strLenOrIndArray = AllocateParamBufferArray(tempBuffers, paramSetSize); for (size_t i = 0; i < paramSetSize; ++i) { if (columnValues[i].is_none()) { strLenOrIndArray[i] = SQL_NULL_DATA; - std::memset(charArray + i * (info.columnSize + 1), 0, - info.columnSize + 1); + std::memset(charArray + i * (valueBufferSize + 1), 0, + valueBufferSize + 1); } else { std::string encodedStr; @@ -2412,15 +2906,15 @@ SQLRETURN BindParameterArray(SqlHandle& handle, SQLHANDLE hStmt, const py::list& encodedStr = columnValues[i].cast(); } - if (encodedStr.size() > info.columnSize) { + if (encodedStr.size() > valueBufferSize) { LOG("BindParameterArray: String/binary too " "long - param_index=%d, row=%zu, size=%zu, " "max=%zu", - paramIndex, i, encodedStr.size(), info.columnSize); + paramIndex, i, encodedStr.size(), valueBufferSize); ThrowStdException("Input exceeds column size at index " + std::to_string(i)); } - std::memcpy(charArray + i * (info.columnSize + 1), encodedStr.c_str(), + std::memcpy(charArray + i * (valueBufferSize + 1), encodedStr.c_str(), encodedStr.size()); strLenOrIndArray[i] = static_cast(encodedStr.size()); } @@ -2429,7 +2923,7 @@ SQLRETURN BindParameterArray(SqlHandle& handle, SQLHANDLE hStmt, const py::list& "param_index=%d", paramIndex); dataPtr = charArray; - bufferLength = info.columnSize + 1; + bufferLength = valueBufferSize + 1; break; } case SQL_C_BIT: { @@ -2913,7 +3407,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); @@ -6123,8 +6617,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("metadataStatementHandle"), 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 00f04aa09..991de8559 100644 --- a/mssql_python/pybind/ddbc_bindings.h +++ b/mssql_python/pybind/ddbc_bindings.h @@ -50,7 +50,15 @@ 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_SOPT_SS_NAME_SCOPE (1237) +#define SQL_SS_NAME_SCOPE_TABLE (0L) +#define SQL_SS_NAME_SCOPE_TABLE_TYPE (1L) +#define SQL_CA_SS_CATALOG_NAME (1225) +#define SQL_CA_SS_SCHEMA_NAME (1226) +#define SQL_CA_SS_TYPE_NAME (1227) // Include logger bridge for LOG macros #include "logger_bridge.hpp" @@ -84,6 +92,8 @@ typedef SQLRETURN(SQL_API* SQLExecuteFunc)(SQLHANDLE); typedef SQLRETURN(SQL_API* SQLRowCountFunc)(SQLHSTMT, SQLLEN*); typedef SQLRETURN(SQL_API* SQLSetDescFieldFunc)(SQLHDESC, SQLSMALLINT, SQLSMALLINT, SQLPOINTER, SQLINTEGER); +typedef SQLRETURN(SQL_API* SQLGetDescFieldFunc)(SQLHDESC, SQLSMALLINT, SQLSMALLINT, SQLPOINTER, + SQLINTEGER, SQLINTEGER*); typedef SQLRETURN(SQL_API* SQLGetStmtAttrFunc)(SQLHSTMT, SQLINTEGER, SQLPOINTER, SQLINTEGER, SQLINTEGER*); @@ -169,6 +179,7 @@ extern SQLBindParameterFunc SQLBindParameter_ptr; extern SQLExecuteFunc SQLExecute_ptr; extern SQLRowCountFunc SQLRowCount_ptr; extern SQLSetDescFieldFunc SQLSetDescField_ptr; +extern SQLGetDescFieldFunc SQLGetDescField_ptr; extern SQLGetStmtAttrFunc SQLGetStmtAttr_ptr; // Data retrieval APIs @@ -280,6 +291,13 @@ struct DescribedParamInfo { SQLSMALLINT decimalDigits; }; +struct TvpParamInfo { + std::u16string catalog; + std::u16string schema; + std::u16string typeName; + std::vector columns; +}; + class SqlHandle { public: SqlHandle(SQLSMALLINT type, SQLHANDLE rawHandle); @@ -317,7 +335,11 @@ class SqlHandle { // on handle free. No mutex needed — ODBC statement handles are not // thread-safe by spec (same assumption as the rest of the driver). std::unordered_map describeCache; - void clearDescribeCache() { describeCache.clear(); } + std::unordered_map tvpCache; + void clearDescribeCache() { + describeCache.clear(); + tvpCache.clear(); + } private: SQLSMALLINT _type; diff --git a/mssql_python/pybind/param_detect.hpp b/mssql_python/pybind/param_detect.hpp index 172a98b51..feb0aaadc 100644 --- a/mssql_python/pybind/param_detect.hpp +++ b/mssql_python/pybind/param_detect.hpp @@ -51,9 +51,11 @@ struct ParamInfo { SQLSMALLINT paramCType = SQL_C_DEFAULT; SQLSMALLINT paramSQLType = SQL_UNKNOWN_TYPE; SQLULEN columnSize = 0; + SQLULEN bufferSize = 0; SQLSMALLINT decimalDigits = 0; SQLLEN strLenOrInd = 0; // Required for DAE bool isDAE = false; // Indicates if we need to stream + bool isTVP = false; // 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,14 +312,27 @@ inline std::vector DetectParamTypes(PyObject* params, PyObject* input ParamInfo& info = infos[i]; info.inputOutputType = SQL_PARAM_INPUT; info.isDAE = false; + info.isTVP = false; + + PyObject* obj = PyList_GET_ITEM(params, i); + if (PyList_Check(obj) || PyTuple_Check(obj)) { + if (i < inputSizeCount) { + throw py::type_error("setinputsizes cannot override table-valued parameters"); + } + info.paramSQLType = SQL_SS_TABLE; + info.paramCType = SQL_C_BINARY; + info.columnSize = static_cast(PySequence_Size(obj)); + info.decimalDigits = 0; + info.isTVP = true; + info.dataPtr = borrow(obj); + continue; + } if (i < inputSizeCount) { ApplyInputSizeOverride(params, PyList_GET_ITEM(inputSizes, i), i, info); continue; } - PyObject* obj = PyList_GET_ITEM(params, i); - // --- None --- if (obj == Py_None) { info.paramSQLType = SQL_UNKNOWN_TYPE; diff --git a/tests/test_004_cursor.py b/tests/test_004_cursor.py index 219d4833a..ab2fde2ad 100644 --- a/tests/test_004_cursor.py +++ b/tests/test_004_cursor.py @@ -1879,6 +1879,7 @@ def test_executemany_unicode_and_empty_strings(cursor, db_connection): (5, ""), (6, "Ñice tëxt"), (7, ""), + (8, "a\x00b"), ] # Execute the batch insert @@ -1890,7 +1891,7 @@ def test_executemany_unicode_and_empty_strings(cursor, db_connection): results = cursor.fetchall() # Check that we got the right number of rows - assert len(results) == 7, f"Expected 7 rows, got {len(results)}" + assert len(results) == 8, f"Expected 8 rows, got {len(results)}" # Check each row for i, (actual, expected_row) in enumerate(zip(results, test_data)): diff --git a/tests/test_004_cursor_timeout.py b/tests/test_004_cursor_timeout.py new file mode 100644 index 000000000..d9cc4a8ac --- /dev/null +++ b/tests/test_004_cursor_timeout.py @@ -0,0 +1,75 @@ +"""Regression tests for query-timeout statement-handle routing.""" + +from unittest.mock import Mock + +import pytest + +from mssql_python import ddbc_bindings +from mssql_python.constants import ConstantsDDBC +from mssql_python.cursor import Cursor, logger + + +@pytest.mark.parametrize("metadata_handle", [False, True], ids=["default", "metadata"]) +@pytest.mark.parametrize("timeout", [0, 2], ids=["disabled", "enabled"]) +def test_set_timeout_uses_target_handle(monkeypatch, metadata_handle, timeout): + cursor = Cursor.__new__(Cursor) + cursor.hstmt = object() + cursor._timeout = timeout + statement_handle = object() if metadata_handle else None + target = statement_handle or cursor.hstmt + set_attribute = Mock(return_value=0) + check_error = Mock() + warning = Mock() + monkeypatch.setattr(ddbc_bindings, "DDBCSQLSetStmtAttr", set_attribute) + monkeypatch.setattr("mssql_python.cursor.check_error", check_error) + monkeypatch.setattr(logger, "warning", warning) + + assert cursor._set_timeout(statement_handle) is None + + if timeout: + set_attribute.assert_called_once_with( + target, ConstantsDDBC.SQL_ATTR_QUERY_TIMEOUT.value, timeout + ) + check_error.assert_called_once_with(ConstantsDDBC.SQL_HANDLE_STMT.value, target, 0) + else: + set_attribute.assert_not_called() + check_error.assert_not_called() + warning.assert_not_called() + + +@pytest.mark.parametrize("metadata_handle", [False, True], ids=["default", "metadata"]) +def test_set_timeout_failure_reports_target_diagnostics( + db_connection, monkeypatch, metadata_handle +): + with db_connection.cursor() as cursor: + cursor._timeout = 2 + if metadata_handle: + cursor._tvp_metadata_hstmt = db_connection._conn.alloc_statement_handle() + statement_handle = cursor._tvp_metadata_hstmt + target = statement_handle or cursor.hstmt + real_set_attribute = ddbc_bindings.DDBCSQLSetStmtAttr + real_check_error = ddbc_bindings.DDBCSQLCheckError + + def fail_attribute(handle, attribute, value): + # An invalid identifier produces real driver diagnostics on this handle. + return real_set_attribute(handle, 999999, value) + + set_attribute = Mock(side_effect=fail_attribute) + check_error = Mock(wraps=real_check_error) + warning = Mock() + with monkeypatch.context() as patch: + patch.setattr(ddbc_bindings, "DDBCSQLSetStmtAttr", set_attribute) + patch.setattr(ddbc_bindings, "DDBCSQLCheckError", check_error) + patch.setattr(logger, "warning", warning) + assert cursor._set_timeout(statement_handle) is None + + set_attribute.assert_called_once_with(target, ConstantsDDBC.SQL_ATTR_QUERY_TIMEOUT.value, 2) + check_error.assert_called_once_with(ConstantsDDBC.SQL_HANDLE_STMT.value, target, -1) + assert check_error.call_args.args[1] is target + error_info = real_check_error(ConstantsDDBC.SQL_HANDLE_STMT.value, target, -1) + assert error_info.sqlState == "HY092" + assert error_info.ddbcErrorMsg + warning.assert_called_once() + assert warning.call_args.args[0] == "Failed to set query timeout: %s" + assert "Invalid attribute/option identifier" in warning.call_args.args[1] + assert cursor.execute("SELECT 1").fetchone()[0] == 1 diff --git a/tests/test_028_table_valued_parameters.py b/tests/test_028_table_valued_parameters.py new file mode 100644 index 000000000..154de1022 --- /dev/null +++ b/tests/test_028_table_valued_parameters.py @@ -0,0 +1,732 @@ +""" +Integration coverage for SQL Server table-valued parameters. +""" + +import datetime +import os +import subprocess +import sys +import uuid +from decimal import Decimal + +import pytest + +from mssql_python import ( + DatabaseError, + NotSupportedError, + ProgrammingError, + SQL_INTEGER, + SQL_WVARCHAR, + ddbc_bindings, +) +from mssql_python.odbc_provider import ProviderManager +from test_023_ssh_tunnel_gil_release import ( + WATCHDOG_SECONDS, + _parse_server, + _replace_server, + _start_forwarder, +) + + +@pytest.fixture +def tvp_procedure(cursor, db_connection): + suffix = uuid.uuid4().hex + type_name = f"pytest_tvp_type_{suffix}" + wildcard_match_type_name = type_name.replace("_", "X") + procedure_name = f"pytest_tvp_proc_{suffix}" + + try: + cursor.execute( + f"CREATE TYPE dbo.[{type_name}] AS TABLE " "(id int NOT NULL, label nvarchar(100) NULL)" + ) + cursor.execute( + f"CREATE TYPE dbo.[{wildcard_match_type_name}] AS TABLE " "(decoy_id bigint NOT NULL)" + ) + cursor.execute(f""" + CREATE PROCEDURE dbo.[{procedure_name}] + @prefix int, + @items dbo.[{type_name}] READONLY, + @suffix nvarchar(20) + AS + BEGIN + SET NOCOUNT ON; + SELECT id + @prefix, label + @suffix + FROM @items + ORDER BY id; + END + """) + db_connection.commit() + yield type_name, procedure_name + finally: + cursor.execute(f"DROP PROCEDURE IF EXISTS dbo.[{procedure_name}]") + cursor.execute(f"DROP TYPE IF EXISTS dbo.[{type_name}]") + cursor.execute(f"DROP TYPE IF EXISTS dbo.[{wildcard_match_type_name}]") + db_connection.commit() + + +@pytest.fixture +def typed_tvp_procedure(cursor, db_connection): + suffix = uuid.uuid4().hex + type_name = f"pytest_typed_tvp_type_{suffix}" + procedure_name = f"pytest_typed_tvp_proc_{suffix}" + + try: + cursor.execute(f""" + CREATE TYPE dbo.[{type_name}] AS TABLE ( + bit_value bit NULL, + tiny_value tinyint NULL, + small_value smallint NULL, + int_value int NULL, + big_value bigint NULL, + float_value float NULL, + decimal_value decimal(20, 6) NULL, + text_value nvarchar(100) NULL, + binary_value varbinary(100) NULL, + date_value date NULL, + time_value time(6) NULL, + datetime_value datetime2(6) NULL, + offset_value datetimeoffset(6) NULL, + guid_value uniqueidentifier NULL, + all_null_binary varbinary(16) NULL + ) + """) + cursor.execute(f""" + CREATE PROCEDURE dbo.[{procedure_name}] + @items dbo.[{type_name}] READONLY + AS + BEGIN + SET NOCOUNT ON; + SELECT * + FROM @items + ORDER BY int_value; + END + """) + db_connection.commit() + yield procedure_name + finally: + cursor.execute(f"DROP PROCEDURE IF EXISTS dbo.[{procedure_name}]") + cursor.execute(f"DROP TYPE IF EXISTS dbo.[{type_name}]") + db_connection.commit() + + +@pytest.fixture +def multiple_tvp_procedure(cursor, db_connection): + suffix = uuid.uuid4().hex + type_name = f"pytest_multi_tvp_type_{suffix}" + procedure_name = f"pytest_multi_tvp_proc_{suffix}" + + try: + cursor.execute(f"CREATE TYPE dbo.[{type_name}] AS TABLE (id int NOT NULL)") + cursor.execute(f""" + CREATE PROCEDURE dbo.[{procedure_name}] + @left_rows dbo.[{type_name}] READONLY, + @offset int, + @right_rows dbo.[{type_name}] READONLY + AS + BEGIN + SET NOCOUNT ON; + SELECT side, id + @offset + FROM ( + SELECT 'left' AS side, id FROM @left_rows + UNION ALL + SELECT 'right', id FROM @right_rows + ) AS combined + ORDER BY side, id; + END + """) + db_connection.commit() + yield procedure_name + finally: + cursor.execute(f"DROP PROCEDURE IF EXISTS dbo.[{procedure_name}]") + cursor.execute(f"DROP TYPE IF EXISTS dbo.[{type_name}]") + db_connection.commit() + + +@pytest.fixture +def schema_qualified_large_tvp(cursor, db_connection): + suffix = uuid.uuid4().hex + schema_name = f"pytest_tvp_schema_{suffix}" + type_name = f"pytest_large_tvp_type_{suffix}" + procedure_name = f"pytest_large_tvp_proc_{suffix}" + + try: + cursor.execute(f"CREATE SCHEMA [{schema_name}]") + cursor.execute(f""" + CREATE TYPE [{schema_name}].[{type_name}] AS TABLE ( + text_value nvarchar(max) NULL, + binary_value varbinary(max) NULL + ) + """) + cursor.execute(f""" + CREATE PROCEDURE [{schema_name}].[{procedure_name}] + @items [{schema_name}].[{type_name}] READONLY + AS + BEGIN + SET NOCOUNT ON; + SELECT text_value, binary_value FROM @items; + END + """) + db_connection.commit() + yield schema_name, procedure_name + finally: + cursor.execute(f"DROP PROCEDURE IF EXISTS [{schema_name}].[{procedure_name}]") + cursor.execute(f"DROP TYPE IF EXISTS [{schema_name}].[{type_name}]") + cursor.execute(f"DROP SCHEMA IF EXISTS [{schema_name}]") + db_connection.commit() + + +def test_tvp_positional_named_empty_and_statement_reuse(cursor, tvp_procedure): + _, procedure_name = tvp_procedure + sql = f"EXEC dbo.[{procedure_name}] ?, ?, ?" + + cursor.execute(sql, (10, [(1, "Keyboard"), (2, None)], "!")) + assert [tuple(row) for row in cursor.fetchall()] == [(11, "Keyboard!"), (12, None)] + + cursor.execute(sql, (10, [], "!")) + assert cursor.fetchall() == [] + + cursor.execute(sql, (None, [(3, "Nullable")], "")) + assert tuple(cursor.fetchone()) == (None, "Nullable") + + cursor.execute( + f"EXEC dbo.[{procedure_name}] %(prefix)s, %(items)s, %(suffix)s", + {"prefix": 0, "items": [(4, "Mouse")], "suffix": "?"}, + ) + assert tuple(cursor.fetchone()) == (4, "Mouse?") + + +def test_tvp_empty_first_and_all_null_column(cursor, tvp_procedure): + _, procedure_name = tvp_procedure + sql = f"EXEC dbo.[{procedure_name}] ?, ?, ?" + + cursor.execute(sql, (0, [], "")) + assert cursor.fetchall() == [] + + cursor.execute(sql, (0, [(1, None), (2, None)], "")) + assert [tuple(row) for row in cursor.fetchall()] == [(1, None), (2, None)] + + +def test_tvp_binds_supported_cell_types(cursor, typed_tvp_procedure): + first_guid = uuid.uuid4() + second_guid = uuid.uuid4() + offset = datetime.timezone(datetime.timedelta(hours=5, minutes=30)) + rows = [ + ( + True, + 255, + -32768, + -2147483648, + -(2**63), + 1.25, + Decimal("-12345678901234.567890"), + "Grüße 😀", + b"\x00\xff", + datetime.date(2026, 9, 15), + datetime.time(12, 34, 56, 123456), + datetime.datetime(2026, 9, 15, 12, 34, 56, 123456), + datetime.datetime(2026, 9, 15, 12, 34, 56, 123456, tzinfo=offset), + first_guid, + None, + ), + ( + False, + 0, + 32767, + 2147483647, + 2**63 - 1, + -2.5, + Decimal("12.345"), + "later row is longer", + b"\x01\x02\x03", + datetime.date(2026, 9, 16), + datetime.time(1, 2, 3, 4), + datetime.datetime(2026, 9, 16, 1, 2, 3, 4), + datetime.datetime(2026, 9, 16, 1, 2, 3, 4, tzinfo=datetime.timezone.utc), + second_guid, + None, + ), + ] + + cursor.execute(f"EXEC dbo.[{typed_tvp_procedure}] ?", (rows,)) + actual = [tuple(row) for row in cursor.fetchall()] + + assert actual[0][:9] == rows[0][:9] + assert actual[1][:9] == rows[1][:9] + assert actual[0][9:13] == rows[0][9:13] + assert actual[1][9:13] == rows[1][9:13] + assert str(actual[0][13]) == str(first_guid) + assert str(actual[1][13]) == str(second_guid) + assert actual[0][14] is None + assert actual[1][14] is None + + invalid_row = list(rows[0]) + invalid_row[6] = Decimal("NaN") + with pytest.raises(ValueError, match="non-finite Decimal"): + cursor.execute( + f"EXEC dbo.[{typed_tvp_procedure}] ?", + ([tuple(invalid_row)],), + ) + + +def test_multiple_tvps_with_interleaved_scalar(cursor, multiple_tvp_procedure): + cursor.execute( + f"EXEC dbo.[{multiple_tvp_procedure}] ?, ?, ?", + ([(1,), (2,)], 10, [(3,), (4,)]), + ) + assert [tuple(row) for row in cursor.fetchall()] == [ + ("left", 11), + ("left", 12), + ("right", 13), + ("right", 14), + ] + + +def test_schema_qualified_tvp_with_large_and_embedded_null_values( + cursor, schema_qualified_large_tvp +): + schema_name, procedure_name = schema_qualified_large_tvp + large_text = "prefix\x00suffix" + "😀" * 2500 + large_binary = b"\x00\xff" * 4501 + rows = ((large_text, large_binary), (None, None)) + + cursor.execute(f"EXEC [{schema_name}].[{procedure_name}] ?", (rows,)) + actual = [tuple(row) for row in cursor.fetchall()] + + assert actual == [(large_text, large_binary), (None, None)] + + +def test_failed_tvp_does_not_poison_cursor(cursor, tvp_procedure): + _, procedure_name = tvp_procedure + sql = f"EXEC dbo.[{procedure_name}] ?, ?, ?" + + with pytest.raises(DatabaseError): + cursor.execute(sql, (0, [(None, "invalid")], "")) + + cursor.execute("SELECT ?", 42) + assert cursor.fetchone()[0] == 42 + + cursor.execute(sql, (0, [(1, "valid")], "")) + assert tuple(cursor.fetchone()) == (1, "valid") + + +def test_tvp_rejects_invalid_shape_and_cells(cursor, tvp_procedure): + _, procedure_name = tvp_procedure + sql = f"EXEC dbo.[{procedure_name}] ?, ?, ?" + + with pytest.raises(ValueError, match="same number"): + cursor.execute(sql, (0, [(1, "valid"), (2,)], "")) + with pytest.raises(TypeError, match="incompatible Python types"): + cursor.execute(sql, (0, [(1, "valid"), ("2", "invalid")], "")) + with pytest.raises(TypeError, match="only be bound"): + cursor.execute("SELECT CAST(? AS int)", ([(1,)],)) + + +def test_tvp_rejected_by_executemany_before_first_write(cursor, tvp_procedure): + cursor.execute("CREATE TABLE #pytest_tvp_executemany (id int)") + + with pytest.raises(NotSupportedError): + cursor.executemany( + "INSERT INTO #pytest_tvp_executemany VALUES (?)", + [(1,), ([(2, "label")],)], + ) + + cursor.execute("SELECT COUNT(*) FROM #pytest_tvp_executemany") + assert cursor.fetchone()[0] == 0 + + +def test_tvp_rejected_with_setinputsizes(cursor, tvp_procedure): + _, procedure_name = tvp_procedure + cursor.setinputsizes( + [ + (SQL_INTEGER, 10, 0), + (SQL_WVARCHAR, 100, 0), + (SQL_WVARCHAR, 20, 0), + ] + ) + try: + with pytest.raises(NotSupportedError): + cursor.execute( + f"EXEC dbo.[{procedure_name}] ?, ?, ?", + (0, [(1, "label")], ""), + ) + finally: + cursor.setinputsizes(None) + + +def test_tvp_rejected_by_mssql_odbc_provider(cursor, tvp_procedure, monkeypatch): + _, procedure_name = tvp_procedure + monkeypatch.setattr(ProviderManager, "effective", lambda *_: "mssql-odbc") + + with pytest.raises(NotSupportedError, match="mssql-odbc provider"): + cursor.execute( + f"EXEC dbo.[{procedure_name}] ?, ?, ?", + (0, [(1, "label")], ""), + ) + + +def test_tvp_metadata_handle_inherits_query_timeout(db_connection, tvp_procedure, monkeypatch): + _, procedure_name = tvp_procedure + real_set_stmt_attr = ddbc_bindings.DDBCSQLSetStmtAttr + timeout_handles = [] + + def record_timeout(statement_handle, attribute, value): + timeout_handles.append(statement_handle) + return real_set_stmt_attr(statement_handle, attribute, value) + + db_connection.timeout = 2 + monkeypatch.setattr( + ddbc_bindings, + "DDBCSQLSetStmtAttr", + record_timeout, + ) + try: + with db_connection.cursor() as timeout_cursor: + timeout_cursor.execute( + f"EXEC dbo.[{procedure_name}] ?, ?, ?", + (0, [(1, "label")], ""), + ) + assert tuple(timeout_cursor.fetchone()) == (1, "label") + finally: + db_connection.timeout = 0 + + assert len(timeout_handles) == 2 + assert timeout_handles[0] is not timeout_handles[1] + + +@pytest.fixture +def conversion_tvp(cursor, db_connection, request): + suffix = uuid.uuid4().hex + type_name = f"pytest_conversion_type_{suffix}" + procedure_name = f"pytest_conversion_proc_{suffix}" + try: + cursor.execute( + f"CREATE TYPE dbo.[{type_name}] AS TABLE (position int, value {request.param} NULL)" + ) + cursor.execute( + f"CREATE PROCEDURE dbo.[{procedure_name}] @items dbo.[{type_name}] READONLY AS " + "SET NOCOUNT ON; SELECT value FROM @items ORDER BY position" + ) + db_connection.commit() + yield f"EXEC dbo.[{procedure_name}] ?" + finally: + cursor.execute(f"DROP PROCEDURE IF EXISTS dbo.[{procedure_name}]") + cursor.execute(f"DROP TYPE IF EXISTS dbo.[{type_name}]") + db_connection.commit() + + +@pytest.mark.parametrize( + ("conversion_tvp", "values", "expected"), + [ + ( + "float", + [Decimal("1.5"), Decimal("123.45"), None, Decimal("1.50"), Decimal("-0.125")], + [1.5, 123.45, None, 1.5, -0.125], + ), + ("int", [Decimal("1.5"), Decimal("-123.45"), None], [1, -123, None]), + ("nvarchar(40)", [Decimal("1.5"), None], ["1.5", None]), + ( + "nvarchar(40)", + [Decimal("1.5"), Decimal("123.45"), None, Decimal("-0.125")], + ["1.500", "123.450", None, "-0.125"], + ), + ( + "decimal(5,2)", + [Decimal("1.239"), None, Decimal("-1.235"), Decimal("9.999")], + [Decimal("1.24"), None, Decimal("-1.24"), Decimal("10.00")], + ), + ( + "decimal(8,2)", + ["000000000000000001.23", None, "-0000000000000001.239"], + [Decimal("1.23"), None, Decimal("-1.24")], + ), + ( + "datetime", + [datetime.datetime(2026, 9, 15, 12, 34, 56, 456789), None], + [datetime.datetime(2026, 9, 15, 12, 34, 56, 457000), None], + ), + ( + "datetime2(3)", + [datetime.datetime(2026, 9, 15, 23, 59, 59, 999999), None], + [datetime.datetime(2026, 9, 16), None], + ), + ( + "time(3)", + [datetime.time(12, 34, 56, 456789), None], + [datetime.time(12, 34, 56, 457000), None], + ), + ( + "datetimeoffset(3)", + [ + datetime.datetime(2026, 9, 15, 23, 59, 59, 999999, tzinfo=datetime.timezone.utc), + None, + ], + [datetime.datetime(2026, 9, 16, tzinfo=datetime.timezone.utc), None], + ), + ], + indirect=["conversion_tvp"], +) +def test_tvp_source_precision_survives_destination_conversion( + cursor, conversion_tvp, values, expected +): + rows = list(enumerate(values)) + for _ in range(2): + cursor.execute(conversion_tvp, (rows,)) + assert [row[0] for row in cursor.fetchall()] == expected + assert rows == list(enumerate(values)) + + +@pytest.mark.parametrize( + ("conversion_tvp", "value", "recovery"), + [ + ("nvarchar(3)", "x" * 5000, "ok"), + ("nvarchar(3)", "\u0100" * 5000, "ok"), + ("varchar(8000)", "x" * 8001, "ok"), + ("varchar(8000) COLLATE Latin1_General_100_CI_AS", "\u00e9" * 8001, "ok"), + ("varbinary(8000)", b"x" * 8001, b"ok"), + ], + ids=["short-ascii", "short-unicode", "long-ascii", "long-unicode", "long-binary"], + indirect=["conversion_tvp"], +) +def test_tvp_destination_limits_leave_cursor_reusable(cursor, conversion_tvp, value, recovery): + with pytest.raises(ProgrammingError, match="would be truncated") as raised: + cursor.execute(conversion_tvp, ([(0, value)],)) + assert raised.value.driver_error == "Syntax error or access violation" + cursor.execute("SELECT ?", 42) + assert cursor.fetchone()[0] == 42 + cursor.execute(conversion_tvp, ([(0, recovery)],)) + assert cursor.fetchone()[0] == recovery + + +@pytest.mark.parametrize("conversion_tvp", ["decimal(38,0)"], indirect=True) +@pytest.mark.parametrize( + "value", [Decimal("NaN"), Decimal("Infinity"), Decimal("1E+39"), Decimal("1E-39")] +) +def test_tvp_decimal_validation_matches_scalar(cursor, conversion_tvp, value): + with pytest.raises(ValueError): + cursor.execute(conversion_tvp, ([(0, value)],)) + cursor.execute(conversion_tvp, ([(0, Decimal("1E+37")), (1, None)],)) + assert [row[0] for row in cursor.fetchall()] == [Decimal("1E+37"), None] + + +@pytest.mark.parametrize("conversion_tvp", ["bigint"], indirect=True) +def test_tvp_integer_widening_and_nulls(cursor, conversion_tvp): + edges = [0, 255, -32768, 32767, -(2**31), 2**31 - 1, -(2**63), 2**63 - 1] + for ordered in (edges, list(reversed(edges))): + values = [None, *ordered, None] + cursor.execute(conversion_tvp, (list(enumerate(values)),)) + assert [row[0] for row in cursor.fetchall()] == values + + +@pytest.mark.parametrize( + ("conversion_tvp", "values"), + [ + ("nvarchar(max)", [None, "ascii first", "\U0001f600" * 4001, "a\x00b", ""]), + ("nvarchar(max)", ["\U0001f600", "later ascii", None]), + ("varchar(max)", [None, "x" * 8001, "a\x00b", ""]), + ("varbinary(max)", [b"", None, b"\x00\xff" * 4501, bytearray(b"last")]), + ], + indirect=["conversion_tvp"], +) +def test_tvp_source_buffers_cover_every_row(cursor, conversion_tvp, values): + cursor.execute(conversion_tvp, (list(enumerate(values)),)) + assert [row[0] for row in cursor.fetchall()] == values + + +@pytest.mark.parametrize( + "conversion_tvp", ["varchar(8000) COLLATE Latin1_General_100_CI_AS"], indirect=True +) +@pytest.mark.parametrize("length", [4000, 4001, 8000]) +def test_tvp_unicode_source_fits_bounded_varchar(cursor, conversion_tvp, length): + values = [None, "ascii first", "\u00e9" * length, ""] + for ordered in (values, list(reversed(values))): + cursor.execute(conversion_tvp, (list(enumerate(ordered)),)) + assert [row[0] for row in cursor.fetchall()] == ordered + + +def test_tvp_all_null_columns_use_declared_types(cursor, typed_tvp_procedure): + cursor.execute(f"EXEC dbo.[{typed_tvp_procedure}] ?", ([tuple([None] * 15)],)) + assert tuple(cursor.fetchone()) == tuple([None] * 15) + + +@pytest.mark.parametrize("conversion_tvp", ["float"], indirect=True) +def test_tvp_rejects_unrepresentable_common_decimal_precision(cursor, conversion_tvp): + with pytest.raises(ValueError, match="precision greater than 38"): + cursor.execute(conversion_tvp, ([(0, Decimal("1E+37")), (1, Decimal("0.1"))],)) + cursor.execute(conversion_tvp, ([(0, Decimal("0.1"))],)) + assert cursor.fetchone()[0] == 0.1 + + +@pytest.mark.parametrize("conversion_tvp", ["decimal(5,2)"], indirect=True) +def test_tvp_decimal_wire_format_is_independent_of_text_encoding(conn_str, conversion_tvp): + from mssql_python import SQL_CHAR, connect + + with connect(conn_str) as connection: + connection.setencoding("utf-16le", ctype=SQL_CHAR) + with connection.cursor() as encoded_cursor: + encoded_cursor.execute(conversion_tvp, ([(0, Decimal("-1.239")), (1, None)],)) + assert [row[0] for row in encoded_cursor.fetchall()] == [Decimal("-1.24"), None] + + +@pytest.mark.parametrize("conversion_tvp", ["nvarchar(40)"], indirect=True) +@pytest.mark.parametrize( + ("input_sizes", "error", "message"), + [ + (None, RuntimeError, "metadata requires a valid statement handle"), + ([(SQL_INTEGER, 10, 0)], TypeError, "setinputsizes cannot override"), + ], +) +def test_tvp_native_entry_point_validates_metadata_and_overrides( + db_connection, conversion_tvp, input_sizes, error, message +): + with db_connection.cursor() as native_cursor: + with pytest.raises(error, match=message): + ddbc_bindings.DDBCSQLExecute( + native_cursor.hstmt, + None, + conversion_tvp, + [[(0, "ok")]], + input_sizes, + [False], + True, + {}, + ) + native_cursor.execute(conversion_tvp, ([(0, "ok")],)) + assert native_cursor.fetchone()[0] == "ok" + + +@pytest.mark.parametrize( + ("rows", "error", "message"), + [ + ([1], TypeError, "rows must be list or tuple"), + ([()], ValueError, "at least one column"), + ([(1,)], ValueError, "declared table type requires"), + ([(1, "ok"), 2], TypeError, "rows must be list or tuple"), + ([(1, [2])], TypeError, "nested row sequences"), + ([(1, Decimal("1")), (2, "2")], TypeError, "incompatible Python types"), + ], +) +def test_tvp_shape_and_type_errors_leave_cursor_reusable( + cursor, tvp_procedure, rows, error, message +): + _, procedure_name = tvp_procedure + sql = f"EXEC dbo.[{procedure_name}] ?, ?, ?" + with pytest.raises(error, match=message): + cursor.execute(sql, (0, rows, "")) + cursor.execute(sql, (0, [(1, "ok")], "")) + assert tuple(cursor.fetchone()) == (1, "ok") + + +@pytest.fixture +def wide_tvp_procedure(cursor, db_connection): + suffix = uuid.uuid4().hex + type_name = f"pytest_wide_tvp_type_{suffix}" + procedure_name = f"pytest_wide_tvp_proc_{suffix}" + # Long column names make metadata span multiple network packets. + columns = ", ".join(["id int"] + [f"c{i}_{'x' * 100} int" for i in range(1, 1024)]) + try: + cursor.execute(f"CREATE TYPE dbo.[{type_name}] AS TABLE ({columns})") + cursor.execute( + f"CREATE PROCEDURE dbo.[{procedure_name}] @items dbo.[{type_name}] READONLY AS " + "SET NOCOUNT ON; SELECT id FROM @items ORDER BY id" + ) + db_connection.commit() + yield procedure_name + finally: + cursor.execute(f"DROP PROCEDURE IF EXISTS dbo.[{procedure_name}]") + cursor.execute(f"DROP TYPE IF EXISTS dbo.[{type_name}]") + db_connection.commit() + + +def _run_forwarded_tvp(): + import mssql_python + + base = os.environ["DB_CONNECTION_STRING"] + target = _parse_server(base) + assert target is not None, "Could not parse Server=host,port" + host, port = _start_forwarder(target) + mssql_python.pooling(enabled=False) + with mssql_python.connect(_replace_server(base, host, port)) as connection: + with connection.cursor() as cursor: + sql = f"EXEC dbo.[{os.environ['TVP_GIL_PROCEDURE']}] ?" + padding = (None,) * 1023 + if os.environ["TVP_GIL_METADATA_ERROR"] == "1": + import ctypes + + library = ctypes.CDLL(sys.modules["ddbc_bindings"].__file__) + slot = ctypes.c_void_p.in_dll(library, "SQLGetData_ptr") + original = slot.value + get_data_type = ctypes.CFUNCTYPE( + ctypes.c_short, + ctypes.c_void_p, + ctypes.c_ushort, + ctypes.c_short, + ctypes.c_void_p, + ctypes.c_ssize_t, + ctypes.POINTER(ctypes.c_ssize_t), + ) + get_data = get_data_type(original) + + @get_data_type + def invalid_metadata_column(handle, column, ctype, value, size, indicator): + # Produce a real driver error before metadata is drained. Do not + # wrap SQLFreeStmt: its original GIL behavior is what we exercise. + slot.value = original + return get_data(handle, 999, ctype, value, size, indicator) + + slot.value = ctypes.cast(invalid_metadata_column, ctypes.c_void_p).value + try: + with pytest.raises(ProgrammingError, match="(?i)invalid descriptor index"): + cursor.execute(sql, ([(1,) + padding],)) + finally: + slot.value = original + cursor.execute("SELECT ?", 42) + assert [tuple(row) for row in cursor.fetchall()] == [(42,)] + + # No pointer overrides on the normal path, including after error recovery. + for ids in ([1, 2], [3], []): + cursor.execute(sql, ([(i,) + padding for i in ids],)) + assert [tuple(row) for row in cursor.fetchall()] == [(i,) for i in ids] + print("OK forwarded TVP metadata", flush=True) + + +@pytest.mark.parametrize( + "metadata_error", + [ + pytest.param(False, id="metadata-reads"), + pytest.param( + True, + id="metadata-error-cleanup", + marks=pytest.mark.skipif( + sys.platform == "win32", reason="Native function-pointer injection is Unix-only" + ), + ), + ], +) +def test_tvp_metadata_through_python_forwarder_does_not_deadlock( + conn_str, wide_tvp_procedure, metadata_error +): + if not conn_str or _parse_server(conn_str) is None: + pytest.skip("Requires DB_CONNECTION_STRING with Server=host,port") + env = os.environ.copy() + env["PYTHONPATH"] = os.pathsep.join(sys.path) + env["TVP_GIL_PROCEDURE"] = wide_tvp_procedure + env["TVP_GIL_METADATA_ERROR"] = str(int(metadata_error)) + try: + result = subprocess.run( + [ + sys.executable, + "-c", + "from test_028_table_valued_parameters import _run_forwarded_tvp; " + "_run_forwarded_tvp()", + ], + env=env, + capture_output=True, + text=True, + timeout=WATCHDOG_SECONDS, + ) + except subprocess.TimeoutExpired: + pytest.fail( + f"TVP metadata through the Python forwarder deadlocked after {WATCHDOG_SECONDS}s" + ) + assert result.returncode == 0, result.stdout + result.stderr + assert "OK forwarded TVP metadata" in result.stdout