From 3c50bebd8b9229322af675ff2da527a8345b4376 Mon Sep 17 00:00:00 2001 From: Gaurav Sharma <223556219+Copilot@users.noreply.github.com> Date: Tue, 15 Sep 2026 13:07:24 +0530 Subject: [PATCH 1/6] FEAT: Add table-valued parameter support Accept materialized nested row sequences for stored-procedure TVP parameters. Resolve the declared table type and child schema from prepared ODBC metadata so empty and typed-NULL rows bind correctly without a public wrapper. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- CHANGELOG.md | 4 + mssql_python/cursor.py | 75 ++- mssql_python/pybind/ddbc_bindings.cpp | 652 ++++++++++++++++++++-- mssql_python/pybind/ddbc_bindings.h | 30 +- mssql_python/pybind/param_detect.hpp | 19 +- tests/test_004_cursor.py | 3 +- tests/test_028_table_valued_parameters.py | 383 +++++++++++++ 7 files changed, 1099 insertions(+), 67 deletions(-) create mode 100644 tests/test_028_table_valued_parameters.py diff --git a/CHANGELOG.md b/CHANGELOG.md index d4f9787e2..e4eecb9a0 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 +- **GH-319:** 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..8d2d7d56e 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. @@ -1007,7 +1009,7 @@ def _set_timeout(self) -> None: try: timeout_value = int(self._timeout) ret = ddbc_bindings.DDBCSQLSetStmtAttr( - self.hstmt, + statement_handle or self.hstmt, ddbc_sql_const.SQL_ATTR_QUERY_TIMEOUT.value, timeout_value, ) @@ -1081,6 +1083,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 +1735,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 +1775,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 +1797,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 +2437,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 +2501,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..bcc293752 100644 --- a/mssql_python/pybind/ddbc_bindings.cpp +++ b/mssql_python/pybind/ddbc_bindings.cpp @@ -23,6 +23,10 @@ #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, bool resolveUnknownTypes = true); //------------------------------------------------------------------------------------------------- // Macro definitions @@ -227,6 +231,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,6 +453,559 @@ 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; + bool hasInteger = false; + int64_t minInteger = 0; + int64_t maxInteger = 0; + 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) { + // Integer widths are merged below from the full observed range. + } else if (textColumn && currentText) { + if (current.paramSQLType == SQL_WVARCHAR) { + merged.paramSQLType = SQL_WVARCHAR; + } + } else if (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) { + int overflow = 0; + int64_t value = PyLong_AsLongLongAndOverflow(values[rowIndex].ptr(), &overflow); + if (overflow != 0) { + PyErr_Clear(); + throw py::value_error("TVP integer cell is out of range for SQL BIGINT"); + } + if (PyErr_Occurred()) + throw py::error_already_set(); + if (!integerColumn) { + throw py::type_error("TVP column " + std::to_string(columnIndex) + + " contains incompatible Python types at row " + + std::to_string(rowIndex)); + } + if (!hasInteger) { + minInteger = value; + maxInteger = value; + hasInteger = true; + } else { + minInteger = std::min(minInteger, value); + maxInteger = std::max(maxInteger, value); + } + } else if (currentText) { + 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 (minInteger >= 0 && maxInteger <= UINT8_MAX) { + merged.paramCType = SQL_C_UTINYINT; + merged.paramSQLType = SQL_TINYINT; + merged.columnSize = 3; + } else if (minInteger >= INT16_MIN && maxInteger <= INT16_MAX) { + merged.paramCType = SQL_C_SHORT; + merged.paramSQLType = SQL_SMALLINT; + merged.columnSize = 5; + } else if (minInteger >= INT32_MIN && maxInteger <= INT32_MAX) { + merged.paramCType = SQL_C_LONG; + merged.paramSQLType = SQL_INTEGER; + merged.columnSize = 10; + } else { + merged.paramCType = SQL_C_SBIGINT; + merged.paramSQLType = SQL_BIGINT; + merged.columnSize = 19; + } + merged.decimalDigits = 0; + } else if (textColumn) { + merged.columnSize = std::max(maxSize, 1); + if (merged.paramSQLType == SQL_WVARCHAR && maxSize > MAX_INLINE_CHAR) { + merged.paramSQLType = SQL_WLONGVARCHAR; + } else if (merged.paramSQLType == SQL_VARCHAR && maxSize > MAX_INLINE_BINARY) { + merged.paramSQLType = SQL_LONGVARCHAR; + } + merged.isDAE = false; + } else if (merged.paramCType == SQL_C_BINARY) { + merged.columnSize = std::max(maxSize, 1); + if (maxSize > MAX_INLINE_BINARY) { + merged.paramSQLType = SQL_LONGVARBINARY; + } + 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 bool NormalizeTableDecimalColumn(py::list& values, ParamInfo& info, size_t columnIndex) { + bool hasDecimal = false; + bool hasOtherValue = false; + PyObject* decimalType = PyTypeCache::get_decimal_class(); + for (const py::handle value : values) { + if (value.is_none()) { + continue; + } + int isDecimal = PyObject_IsInstance(value.ptr(), decimalType); + if (isDecimal == -1) + throw py::error_already_set(); + hasDecimal = hasDecimal || isDecimal == 1; + hasOtherValue = hasOtherValue || isDecimal == 0; + } + if (!hasDecimal) { + return false; + } + if (hasOtherValue) { + throw py::type_error("TVP column " + std::to_string(columnIndex) + + " contains incompatible Python types"); + } + + py::str formatSpec("f"); + SQLULEN maxSize = 1; + for (size_t rowIndex = 0; rowIndex < values.size(); ++rowIndex) { + if (values[rowIndex].is_none()) { + continue; + } + py::object isFinite = + steal(PyObject_CallMethod(values[rowIndex].ptr(), "is_finite", nullptr)); + if (!isFinite) + throw py::error_already_set(); + int finite = PyObject_IsTrue(isFinite.ptr()); + if (finite == -1) + throw py::error_already_set(); + if (finite == 0) { + throw py::value_error("Cannot bind non-finite Decimal (NaN/Infinity) in a TVP"); + } + py::object formatted = steal(PyObject_Format(values[rowIndex].ptr(), formatSpec.ptr())); + if (!formatted) + throw py::error_already_set(); + maxSize = std::max(maxSize, static_cast(PyUnicode_GET_LENGTH(formatted.ptr()))); + values[rowIndex] = std::move(formatted); + } + + info.paramCType = SQL_C_CHAR; + info.paramSQLType = SQL_NUMERIC; + info.columnSize = maxSize; + info.bufferSize = maxSize; + info.decimalDigits = 0; + return true; +} + +static std::vector +DetectTableColumnTypes(py::list& columnValues, const std::vector& declaredColumns) { + std::vector columnInfos(columnValues.size()); + for (size_t columnIndex = 0; columnIndex < columnValues.size(); ++columnIndex) { + py::list values = columnValues[columnIndex].cast(); + if ((declaredColumns[columnIndex].sqlType == SQL_DECIMAL || + declaredColumns[columnIndex].sqlType == SQL_NUMERIC) && + NormalizeTableDecimalColumn(values, columnInfos[columnIndex], columnIndex)) { + continue; + } + std::vector valueInfos = DetectParamTypes(values.ptr(), Py_None); + columnInfos[columnIndex] = MergeTableColumnInfo(values, valueInfos, columnIndex); + } + 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 ThrowTableMetadataError(const SqlHandlePtr& metadataHandle, SQLRETURN rc, + const std::string& operation) { + ErrorInfo error = SQLCheckError_Wrap(SQL_HANDLE_STMT, metadataHandle, 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() { + 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(); + 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)) { + ThrowTableMetadataError(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)) { + ThrowTableMetadataError(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)) { + ThrowTableMetadataError(metadataHandle, rc, "Failed to discover TVP columns"); + } + + std::vector> orderedColumns; + while (true) { + { + py::gil_scoped_release release; + rc = SQLFetch_ptr(hMetadataStmt); + } + if (rc == SQL_NO_DATA) { + break; + } + if (!SQL_SUCCEEDED(rc)) { + ThrowTableMetadataError(metadataHandle, rc, "Failed to fetch TVP column metadata"); + } + + SQLSMALLINT sqlType = SQL_UNKNOWN_TYPE; + SQLINTEGER columnSize = 0; + SQLSMALLINT decimalDigits = 0; + SQLINTEGER ordinal = 0; + SQLLEN indicator = 0; + + 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)) { + rc = SQLGetData_ptr(hMetadataStmt, 17, SQL_C_SLONG, &ordinal, sizeof(ordinal), + &indicator); + } + if (!SQL_SUCCEEDED(rc)) { + ThrowTableMetadataError(metadataHandle, rc, "Failed to read TVP column metadata"); + } + + orderedColumns.push_back({ + ordinal, + {sqlType, static_cast(std::max(columnSize, 0)), decimalDigits}, + }); + } + + if (orderedColumns.empty()) { + throw std::runtime_error("No column metadata was returned for the table type"); + } + std::sort(orderedColumns.begin(), orderedColumns.end(), + [](const auto& left, const auto& right) { return left.first < right.first; }); + + std::vector columns; + columns.reserve(orderedColumns.size()); + for (const auto& entry : orderedColumns) { + columns.push_back(entry.second); + } + 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), + }; + } + info.paramSQLType = SQL_SS_TABLE; + info.paramCType = SQL_C_BINARY; + info.decimalDigits = 0; + } + return SQL_SUCCESS; +} + +static SQLRETURN BindTableValuedParameter(SqlHandle& handle, SQLHANDLE hStmt, int paramIndex, + const py::handle& param, + std::vector>& paramBuffers, + const std::string& charEncoding) { + 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, declaredColumns); + for (size_t columnIndex = 0; columnIndex < columnCount; ++columnIndex) { + ParamInfo& inferred = columnInfos[columnIndex]; + const TvpColumnInfo& declared = declaredColumns[columnIndex]; + inferred.paramSQLType = declared.sqlType; + if (declared.columnSize == 0) { + if (declared.sqlType == SQL_WVARCHAR && inferred.columnSize > MAX_INLINE_CHAR) { + inferred.paramSQLType = SQL_WLONGVARCHAR; + } else if (declared.sqlType == SQL_VARCHAR && + inferred.columnSize > MAX_INLINE_BINARY) { + inferred.paramSQLType = SQL_LONGVARCHAR; + } else if (declared.sqlType == SQL_VARBINARY && + inferred.columnSize > MAX_INLINE_BINARY) { + inferred.paramSQLType = SQL_LONGVARBINARY; + } + } + inferred.decimalDigits = declared.decimalDigits; + if (inferred.paramCType == SQL_C_DEFAULT || declared.sqlType == SQL_DECIMAL || + declared.sqlType == SQL_NUMERIC) { + inferred.columnSize = std::max(declared.columnSize, 1); + } + } + } + + 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; + } + + SQLRETURN bindRc = SQL_ERROR; + try { + bindRc = BindParameterArray(handle, hStmt, columnValues, columnInfos, rows.size(), + paramBuffers, charEncoding, false); + } catch (...) { + SQLSetStmtAttr_ptr(hStmt, SQL_SOPT_SS_PARAM_FOCUS, nullptr, SQL_IS_INTEGER); + throw; + } + const SQLRETURN restoreRc = + SQLSetStmtAttr_ptr(hStmt, SQL_SOPT_SS_PARAM_FOCUS, nullptr, SQL_IS_INTEGER); + return SQL_SUCCEEDED(bindRc) ? restoreRc : bindRc; +} + // 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, @@ -473,6 +1031,15 @@ SQLRETURN BindParameters(SqlHandle& handle, SQLHANDLE hStmt, const py::list& par SQLLEN bufferLength = 0; SQLLEN* strLenOrIndPtr = nullptr; + if (paramInfo.isTVP) { + RETCODE rc = BindTableValuedParameter(handle, 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 +2012,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 +2045,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 +2151,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 +2600,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,6 +2666,10 @@ 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); if (!SQL_SUCCEEDED(rc)) return rc; @@ -2206,7 +2775,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, bool resolveUnknownTypes) { PERF_TIMER("BindParameterArray"); LOG("BindParameterArray: Starting column-wise array binding - " "param_count=%zu, param_set_size=%zu", @@ -2216,7 +2785,9 @@ SQLRETURN BindParameterArray(SqlHandle& handle, SQLHANDLE hStmt, const py::list& try { // GH-627: resolve unknown NULL array param SQL types before binding any param. - PreResolveUnknownNullTypes(handle, hStmt, paramInfos); + if (resolveUnknownTypes) { + PreResolveUnknownNullTypes(handle, hStmt, paramInfos); + } for (int paramIndex = 0; paramIndex < columnwise_params.size(); ++paramIndex) { const py::list& columnValues = columnwise_params[paramIndex].cast(); ParamInfo& info = paramInfos[paramIndex]; @@ -2275,36 +2846,39 @@ SQLRETURN BindParameterArray(SqlHandle& handle, SQLHANDLE hStmt, const py::list& break; } case SQL_C_WCHAR: { + const SQLULEN valueBufferSize = + info.bufferSize > 0 ? info.bufferSize : info.columnSize; LOG("BindParameterArray: Binding SQL_C_WCHAR array - " "param_index=%d, count=%zu, column_size=%zu", - paramIndex, paramSetSize, info.columnSize); + paramIndex, paramSetSize, valueBufferSize); SQLWCHAR* wcharArray = 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(wcharArray + i * (info.columnSize + 1), 0, - (info.columnSize + 1) * sizeof(SQLWCHAR)); + std::memset(wcharArray + i * (valueBufferSize + 1), 0, + (valueBufferSize + 1) * sizeof(SQLWCHAR)); } else { std::u16string wstr = columnValues[i].cast(); // u16string is already UTF-16, so the // original check is sufficient - if (wstr.length() > info.columnSize) { + if (wstr.length() > valueBufferSize) { ThrowStdException("Input string exceeds allowed column size " "at parameter index " + std::to_string(paramIndex)); } - std::memcpy(wcharArray + i * (info.columnSize + 1), wstr.c_str(), + std::memcpy(wcharArray + i * (valueBufferSize + 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 - " "param_index=%d", paramIndex); dataPtr = wcharArray; - bufferLength = (info.columnSize + 1) * sizeof(SQLWCHAR); + bufferLength = (valueBufferSize + 1) * sizeof(SQLWCHAR); break; } case SQL_C_TINYINT: @@ -2372,17 +2946,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 +2988,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 +3005,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: { @@ -6123,8 +6699,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..3558db178 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,19 @@ struct DescribedParamInfo { SQLSMALLINT decimalDigits; }; +struct TvpColumnInfo { + SQLSMALLINT sqlType; + SQLULEN columnSize; + 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 +341,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_028_table_valued_parameters.py b/tests/test_028_table_valued_parameters.py new file mode 100644 index 000000000..4ce1c40a6 --- /dev/null +++ b/tests/test_028_table_valued_parameters.py @@ -0,0 +1,383 @@ +""" +Integration coverage for SQL Server table-valued parameters. +""" + +import datetime +import uuid +from decimal import Decimal + +import pytest + +from mssql_python import ( + DatabaseError, + NotSupportedError, + SQL_INTEGER, + SQL_WVARCHAR, + ddbc_bindings, +) +from mssql_python.odbc_provider import ProviderManager + + +@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] From 635d19897edbbf0b1eaa64eae6a04fa48b77c874 Mon Sep 17 00:00:00 2001 From: Gaurav Sharma <223556219+Copilot@users.noreply.github.com> Date: Tue, 15 Sep 2026 17:09:44 +0530 Subject: [PATCH 2/6] CHORE: Link TVP support to ADO work item Replace the stale GitHub issue reference with the ADO task that tracks this feature. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- CHANGELOG.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e4eecb9a0..2e940ff99 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,7 +7,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), ## [Unreleased] ### Added -- **GH-319:** Added input table-valued parameter support for stored procedure +- **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. From 71c14b967533036b0c72de8e076f0f6b001bc4e7 Mon Sep 17 00:00:00 2001 From: Gaurav Sharma <223556219+Copilot@users.noreply.github.com> Date: Tue, 15 Sep 2026 18:09:59 +0530 Subject: [PATCH 3/6] REFACTOR: Deduplicate TVP parameter binding Reuse scalar type detection and parameter metadata for TVP columns, and remove redundant sizing and metadata ordering work. AB#48177 Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- mssql_python/pybind/ddbc_bindings.cpp | 126 ++++++-------------------- mssql_python/pybind/ddbc_bindings.h | 8 +- 2 files changed, 29 insertions(+), 105 deletions(-) diff --git a/mssql_python/pybind/ddbc_bindings.cpp b/mssql_python/pybind/ddbc_bindings.cpp index bcc293752..6e49aae05 100644 --- a/mssql_python/pybind/ddbc_bindings.cpp +++ b/mssql_python/pybind/ddbc_bindings.cpp @@ -26,7 +26,7 @@ SQLRETURN BindParameterArray(SqlHandle& handle, SQLHANDLE hStmt, const py::list& columnwise_params, std::vector& paramInfos, size_t paramSetSize, std::vector>& paramBuffers, - const std::string& charEncoding, bool resolveUnknownTypes = true); + const std::string& charEncoding); //------------------------------------------------------------------------------------------------- // Macro definitions @@ -469,9 +469,6 @@ static ParamInfo MergeTableColumnInfo(const py::list& values, bool hasValue = false; bool integerColumn = false; bool textColumn = false; - bool hasInteger = false; - int64_t minInteger = 0; - int64_t maxInteger = 0; SQLULEN maxSize = 0; SQLULEN maxIntegerDigits = 0; SQLSMALLINT maxScale = 0; @@ -494,40 +491,20 @@ static ParamInfo MergeTableColumnInfo(const py::list& values, textColumn = currentText; hasValue = true; } else if (integerColumn && currentInteger) { - // Integer widths are merged below from the full observed range. - } else if (textColumn && currentText) { - if (current.paramSQLType == SQL_WVARCHAR) { - merged.paramSQLType = SQL_WVARCHAR; + // Scalar detection already range-checks integers; retain its widest type. + if (current.columnSize > merged.columnSize) { + merged = current; } - } else if (merged.paramCType != current.paramCType || - merged.paramSQLType != current.paramSQLType) { + } 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) { - int overflow = 0; - int64_t value = PyLong_AsLongLongAndOverflow(values[rowIndex].ptr(), &overflow); - if (overflow != 0) { - PyErr_Clear(); - throw py::value_error("TVP integer cell is out of range for SQL BIGINT"); - } - if (PyErr_Occurred()) - throw py::error_already_set(); - if (!integerColumn) { - throw py::type_error("TVP column " + std::to_string(columnIndex) + - " contains incompatible Python types at row " + - std::to_string(rowIndex)); - } - if (!hasInteger) { - minInteger = value; - maxInteger = value; - hasInteger = true; - } else { - minInteger = std::min(minInteger, value); - maxInteger = std::max(maxInteger, value); - } + continue; } else if (currentText) { SQLULEN size = current.isDAE ? static_cast(current.utf16Len) : current.columnSize; @@ -559,37 +536,11 @@ static ParamInfo MergeTableColumnInfo(const py::list& values, } if (integerColumn) { - if (minInteger >= 0 && maxInteger <= UINT8_MAX) { + if (merged.paramCType == SQL_C_TINYINT) { merged.paramCType = SQL_C_UTINYINT; - merged.paramSQLType = SQL_TINYINT; - merged.columnSize = 3; - } else if (minInteger >= INT16_MIN && maxInteger <= INT16_MAX) { - merged.paramCType = SQL_C_SHORT; - merged.paramSQLType = SQL_SMALLINT; - merged.columnSize = 5; - } else if (minInteger >= INT32_MIN && maxInteger <= INT32_MAX) { - merged.paramCType = SQL_C_LONG; - merged.paramSQLType = SQL_INTEGER; - merged.columnSize = 10; - } else { - merged.paramCType = SQL_C_SBIGINT; - merged.paramSQLType = SQL_BIGINT; - merged.columnSize = 19; } - merged.decimalDigits = 0; - } else if (textColumn) { + } else if (textColumn || merged.paramCType == SQL_C_BINARY) { merged.columnSize = std::max(maxSize, 1); - if (merged.paramSQLType == SQL_WVARCHAR && maxSize > MAX_INLINE_CHAR) { - merged.paramSQLType = SQL_WLONGVARCHAR; - } else if (merged.paramSQLType == SQL_VARCHAR && maxSize > MAX_INLINE_BINARY) { - merged.paramSQLType = SQL_LONGVARCHAR; - } - merged.isDAE = false; - } else if (merged.paramCType == SQL_C_BINARY) { - merged.columnSize = std::max(maxSize, 1); - if (maxSize > MAX_INLINE_BINARY) { - merged.paramSQLType = SQL_LONGVARBINARY; - } merged.isDAE = false; } else if (merged.paramCType == SQL_C_NUMERIC) { merged.decimalDigits = maxScale; @@ -656,7 +607,7 @@ static bool NormalizeTableDecimalColumn(py::list& values, ParamInfo& info, size_ } static std::vector -DetectTableColumnTypes(py::list& columnValues, const std::vector& declaredColumns) { +DetectTableColumnTypes(py::list& columnValues, const std::vector& declaredColumns) { std::vector columnInfos(columnValues.size()); for (size_t columnIndex = 0; columnIndex < columnValues.size(); ++columnIndex) { py::list values = columnValues[columnIndex].cast(); @@ -717,7 +668,7 @@ class TableMetadataScope { SQLHANDLE handle; }; -static std::vector LoadTableColumnMetadata(const SqlHandlePtr& metadataHandle, +static std::vector LoadTableColumnMetadata(const SqlHandlePtr& metadataHandle, const std::u16string& catalog, const std::u16string& schema, const std::u16string& typeName) { @@ -757,7 +708,7 @@ static std::vector LoadTableColumnMetadata(const SqlHandlePtr& me ThrowTableMetadataError(metadataHandle, rc, "Failed to discover TVP columns"); } - std::vector> orderedColumns; + std::vector columns; while (true) { { py::gil_scoped_release release; @@ -773,7 +724,6 @@ static std::vector LoadTableColumnMetadata(const SqlHandlePtr& me SQLSMALLINT sqlType = SQL_UNKNOWN_TYPE; SQLINTEGER columnSize = 0; SQLSMALLINT decimalDigits = 0; - SQLINTEGER ordinal = 0; SQLLEN indicator = 0; rc = SQLGetData_ptr(hMetadataStmt, 5, SQL_C_SSHORT, &sqlType, sizeof(sqlType), &indicator); @@ -789,31 +739,18 @@ static std::vector LoadTableColumnMetadata(const SqlHandlePtr& me rc = SQL_SUCCESS; } } - if (SQL_SUCCEEDED(rc)) { - rc = SQLGetData_ptr(hMetadataStmt, 17, SQL_C_SLONG, &ordinal, sizeof(ordinal), - &indicator); - } if (!SQL_SUCCEEDED(rc)) { ThrowTableMetadataError(metadataHandle, rc, "Failed to read TVP column metadata"); } - orderedColumns.push_back({ - ordinal, - {sqlType, static_cast(std::max(columnSize, 0)), decimalDigits}, + columns.push_back({ + sqlType, static_cast(std::max(columnSize, 0)), decimalDigits, }); } - if (orderedColumns.empty()) { + if (columns.empty()) { throw std::runtime_error("No column metadata was returned for the table type"); } - std::sort(orderedColumns.begin(), orderedColumns.end(), - [](const auto& left, const auto& right) { return left.first < right.first; }); - - std::vector columns; - columns.reserve(orderedColumns.size()); - for (const auto& entry : orderedColumns) { - columns.push_back(entry.second); - } return columns; } @@ -878,9 +815,6 @@ static SQLRETURN ResolveTableValuedParamTypes(SqlHandle& handle, SQLHANDLE hStmt LoadTableColumnMetadata(metadataHandle, catalog, schema, typeName), }; } - info.paramSQLType = SQL_SS_TABLE; - info.paramCType = SQL_C_BINARY; - info.decimalDigits = 0; } return SQL_SUCCESS; } @@ -900,7 +834,7 @@ static SQLRETURN BindTableValuedParameter(SqlHandle& handle, SQLHANDLE hStmt, in throw std::runtime_error("TVP column metadata was not resolved before binding"); } const TvpParamInfo& tableMetadata = metadataEntry->second; - const std::vector& declaredColumns = tableMetadata.columns; + const std::vector& declaredColumns = tableMetadata.columns; if (!rows.empty()) { const py::handle firstRow = rows[0]; if (!PyList_Check(firstRow.ptr()) && !PyTuple_Check(firstRow.ptr())) { @@ -934,7 +868,7 @@ static SQLRETURN BindTableValuedParameter(SqlHandle& handle, SQLHANDLE hStmt, in columnInfos = DetectTableColumnTypes(columnValues, declaredColumns); for (size_t columnIndex = 0; columnIndex < columnCount; ++columnIndex) { ParamInfo& inferred = columnInfos[columnIndex]; - const TvpColumnInfo& declared = declaredColumns[columnIndex]; + const DescribedParamInfo& declared = declaredColumns[columnIndex]; inferred.paramSQLType = declared.sqlType; if (declared.columnSize == 0) { if (declared.sqlType == SQL_WVARCHAR && inferred.columnSize > MAX_INLINE_CHAR) { @@ -996,7 +930,7 @@ static SQLRETURN BindTableValuedParameter(SqlHandle& handle, SQLHANDLE hStmt, in SQLRETURN bindRc = SQL_ERROR; try { bindRc = BindParameterArray(handle, hStmt, columnValues, columnInfos, rows.size(), - paramBuffers, charEncoding, false); + paramBuffers, charEncoding); } catch (...) { SQLSetStmtAttr_ptr(hStmt, SQL_SOPT_SS_PARAM_FOCUS, nullptr, SQL_IS_INTEGER); throw; @@ -2775,7 +2709,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, bool resolveUnknownTypes) { + const std::string& charEncoding) { PERF_TIMER("BindParameterArray"); LOG("BindParameterArray: Starting column-wise array binding - " "param_count=%zu, param_set_size=%zu", @@ -2785,9 +2719,7 @@ SQLRETURN BindParameterArray(SqlHandle& handle, SQLHANDLE hStmt, const py::list& try { // GH-627: resolve unknown NULL array param SQL types before binding any param. - if (resolveUnknownTypes) { - PreResolveUnknownNullTypes(handle, hStmt, paramInfos); - } + PreResolveUnknownNullTypes(handle, hStmt, paramInfos); for (int paramIndex = 0; paramIndex < columnwise_params.size(); ++paramIndex) { const py::list& columnValues = columnwise_params[paramIndex].cast(); ParamInfo& info = paramInfos[paramIndex]; @@ -2846,29 +2778,27 @@ SQLRETURN BindParameterArray(SqlHandle& handle, SQLHANDLE hStmt, const py::list& break; } case SQL_C_WCHAR: { - const SQLULEN valueBufferSize = - info.bufferSize > 0 ? info.bufferSize : info.columnSize; LOG("BindParameterArray: Binding SQL_C_WCHAR array - " "param_index=%d, count=%zu, column_size=%zu", - paramIndex, paramSetSize, valueBufferSize); + paramIndex, paramSetSize, info.columnSize); SQLWCHAR* wcharArray = AllocateParamBufferArray( - tempBuffers, paramSetSize * (valueBufferSize + 1)); + tempBuffers, paramSetSize * (info.columnSize + 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(wcharArray + i * (valueBufferSize + 1), 0, - (valueBufferSize + 1) * sizeof(SQLWCHAR)); + std::memset(wcharArray + i * (info.columnSize + 1), 0, + (info.columnSize + 1) * sizeof(SQLWCHAR)); } else { std::u16string wstr = columnValues[i].cast(); // u16string is already UTF-16, so the // original check is sufficient - if (wstr.length() > valueBufferSize) { + if (wstr.length() > info.columnSize) { ThrowStdException("Input string exceeds allowed column size " "at parameter index " + std::to_string(paramIndex)); } - std::memcpy(wcharArray + i * (valueBufferSize + 1), wstr.c_str(), + std::memcpy(wcharArray + i * (info.columnSize + 1), wstr.c_str(), (wstr.length() + 1) * sizeof(SQLWCHAR)); strLenOrIndArray[i] = static_cast(wstr.length() * sizeof(SQLWCHAR)); @@ -2878,7 +2808,7 @@ SQLRETURN BindParameterArray(SqlHandle& handle, SQLHANDLE hStmt, const py::list& "param_index=%d", paramIndex); dataPtr = wcharArray; - bufferLength = (valueBufferSize + 1) * sizeof(SQLWCHAR); + bufferLength = (info.columnSize + 1) * sizeof(SQLWCHAR); break; } case SQL_C_TINYINT: diff --git a/mssql_python/pybind/ddbc_bindings.h b/mssql_python/pybind/ddbc_bindings.h index 3558db178..991de8559 100644 --- a/mssql_python/pybind/ddbc_bindings.h +++ b/mssql_python/pybind/ddbc_bindings.h @@ -291,17 +291,11 @@ struct DescribedParamInfo { SQLSMALLINT decimalDigits; }; -struct TvpColumnInfo { - SQLSMALLINT sqlType; - SQLULEN columnSize; - SQLSMALLINT decimalDigits; -}; - struct TvpParamInfo { std::u16string catalog; std::u16string schema; std::u16string typeName; - std::vector columns; + std::vector columns; }; class SqlHandle { From d4e245160010e3c69f5896a9a4a7ca123b787f44 Mon Sep 17 00:00:00 2001 From: Gaurav Sharma Date: Tue, 15 Sep 2026 19:28:03 +0530 Subject: [PATCH 4/6] FIX: Preserve TVP conversion precision and diagnostics Preserve source metadata and exact Decimal values during TVP binding so SQL Server can apply destination conversions. Retain binding diagnostics during parameter-focus cleanup and route timeout diagnostics to the selected statement handle. Add regression coverage for conversions, buffer sizing, validation, and recovery. AB#48177 Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- mssql_python/cursor.py | 5 +- mssql_python/pybind/ddbc_bindings.cpp | 165 +++++++---------- tests/test_004_cursor_timeout.py | 75 ++++++++ tests/test_028_table_valued_parameters.py | 212 ++++++++++++++++++++++ 4 files changed, 360 insertions(+), 97 deletions(-) create mode 100644 tests/test_004_cursor_timeout.py diff --git a/mssql_python/cursor.py b/mssql_python/cursor.py index 8d2d7d56e..34f21ecb7 100644 --- a/mssql_python/cursor.py +++ b/mssql_python/cursor.py @@ -1008,12 +1008,13 @@ def _set_timeout(self, statement_handle=None) -> 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( - statement_handle or 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)) diff --git a/mssql_python/pybind/ddbc_bindings.cpp b/mssql_python/pybind/ddbc_bindings.cpp index 6e49aae05..94af75ef0 100644 --- a/mssql_python/pybind/ddbc_bindings.cpp +++ b/mssql_python/pybind/ddbc_bindings.cpp @@ -506,6 +506,9 @@ static ParamInfo MergeTableColumnInfo(const py::list& values, 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); @@ -553,71 +556,40 @@ static ParamInfo MergeTableColumnInfo(const py::list& values, return merged; } -static bool NormalizeTableDecimalColumn(py::list& values, ParamInfo& info, size_t columnIndex) { - bool hasDecimal = false; - bool hasOtherValue = false; - PyObject* decimalType = PyTypeCache::get_decimal_class(); - for (const py::handle value : values) { - if (value.is_none()) { - continue; - } - int isDecimal = PyObject_IsInstance(value.ptr(), decimalType); - if (isDecimal == -1) - throw py::error_already_set(); - hasDecimal = hasDecimal || isDecimal == 1; - hasOtherValue = hasOtherValue || isDecimal == 0; - } - if (!hasDecimal) { - return false; - } - if (hasOtherValue) { - throw py::type_error("TVP column " + std::to_string(columnIndex) + - " contains incompatible Python types"); - } - - py::str formatSpec("f"); - SQLULEN maxSize = 1; - for (size_t rowIndex = 0; rowIndex < values.size(); ++rowIndex) { - if (values[rowIndex].is_none()) { - continue; - } - py::object isFinite = - steal(PyObject_CallMethod(values[rowIndex].ptr(), "is_finite", nullptr)); - if (!isFinite) - throw py::error_already_set(); - int finite = PyObject_IsTrue(isFinite.ptr()); - if (finite == -1) - throw py::error_already_set(); - if (finite == 0) { - throw py::value_error("Cannot bind non-finite Decimal (NaN/Infinity) in a TVP"); - } - py::object formatted = steal(PyObject_Format(values[rowIndex].ptr(), formatSpec.ptr())); - if (!formatted) - throw py::error_already_set(); - maxSize = std::max(maxSize, static_cast(PyUnicode_GET_LENGTH(formatted.ptr()))); - values[rowIndex] = std::move(formatted); - } - - info.paramCType = SQL_C_CHAR; - info.paramSQLType = SQL_NUMERIC; - info.columnSize = maxSize; - info.bufferSize = maxSize; - info.decimalDigits = 0; - return true; -} - -static std::vector -DetectTableColumnTypes(py::list& columnValues, const std::vector& declaredColumns) { +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(); - if ((declaredColumns[columnIndex].sqlType == SQL_DECIMAL || - declaredColumns[columnIndex].sqlType == SQL_NUMERIC) && - NormalizeTableDecimalColumn(values, columnInfos[columnIndex], columnIndex)) { - continue; + 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); + } } - std::vector valueInfos = DetectParamTypes(values.ptr(), Py_None); - columnInfos[columnIndex] = MergeTableColumnInfo(values, valueInfos, columnIndex); + columnValues[columnIndex] = std::move(detectedValues); } return columnInfos; } @@ -641,9 +613,9 @@ static SQLRETURN ReadDescriptorString(SQLHDESC descriptor, SQLSMALLINT recordNum return rc; } -[[noreturn]] static void ThrowTableMetadataError(const SqlHandlePtr& metadataHandle, SQLRETURN rc, - const std::string& operation) { - ErrorInfo error = SQLCheckError_Wrap(SQL_HANDLE_STMT, metadataHandle, 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); @@ -684,7 +656,7 @@ static std::vector LoadTableColumnMetadata(const SqlHandlePt reinterpret_cast(static_cast(SQL_TRUE)), SQL_IS_INTEGER); if (!SQL_SUCCEEDED(rc)) { - ThrowTableMetadataError(metadataHandle, rc, "Failed to enable metadata identifier mode"); + ThrowTableError(metadataHandle, rc, "Failed to enable metadata identifier mode"); } TableMetadataScope metadataScope(hMetadataStmt); @@ -693,7 +665,7 @@ static std::vector LoadTableColumnMetadata(const SqlHandlePt reinterpret_cast(static_cast(SQL_SS_NAME_SCOPE_TABLE_TYPE)), SQL_IS_INTEGER); if (!SQL_SUCCEEDED(rc)) { - ThrowTableMetadataError(metadataHandle, rc, "Failed to select table-type metadata scope"); + ThrowTableError(metadataHandle, rc, "Failed to select table-type metadata scope"); } { py::gil_scoped_release release; @@ -705,7 +677,7 @@ static std::vector LoadTableColumnMetadata(const SqlHandlePt SQL_NTS, nullptr, 0); } if (!SQL_SUCCEEDED(rc)) { - ThrowTableMetadataError(metadataHandle, rc, "Failed to discover TVP columns"); + ThrowTableError(metadataHandle, rc, "Failed to discover TVP columns"); } std::vector columns; @@ -718,7 +690,7 @@ static std::vector LoadTableColumnMetadata(const SqlHandlePt break; } if (!SQL_SUCCEEDED(rc)) { - ThrowTableMetadataError(metadataHandle, rc, "Failed to fetch TVP column metadata"); + ThrowTableError(metadataHandle, rc, "Failed to fetch TVP column metadata"); } SQLSMALLINT sqlType = SQL_UNKNOWN_TYPE; @@ -740,7 +712,7 @@ static std::vector LoadTableColumnMetadata(const SqlHandlePt } } if (!SQL_SUCCEEDED(rc)) { - ThrowTableMetadataError(metadataHandle, rc, "Failed to read TVP column metadata"); + ThrowTableError(metadataHandle, rc, "Failed to read TVP column metadata"); } columns.push_back({ @@ -819,10 +791,11 @@ static SQLRETURN ResolveTableValuedParamTypes(SqlHandle& handle, SQLHANDLE hStmt return SQL_SUCCESS; } -static SQLRETURN BindTableValuedParameter(SqlHandle& handle, SQLHANDLE hStmt, int paramIndex, - const py::handle& param, - std::vector>& paramBuffers, - const std::string& charEncoding) { +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); @@ -865,27 +838,27 @@ static SQLRETURN BindTableValuedParameter(SqlHandle& handle, SQLHANDLE hStmt, in columnValues[columnIndex].cast().append(row[columnIndex]); } } - columnInfos = DetectTableColumnTypes(columnValues, declaredColumns); + columnInfos = DetectTableColumnTypes(columnValues); for (size_t columnIndex = 0; columnIndex < columnCount; ++columnIndex) { ParamInfo& inferred = columnInfos[columnIndex]; const DescribedParamInfo& declared = declaredColumns[columnIndex]; - inferred.paramSQLType = declared.sqlType; - if (declared.columnSize == 0) { - if (declared.sqlType == SQL_WVARCHAR && inferred.columnSize > MAX_INLINE_CHAR) { + // 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 (declared.columnSize == 0) { + if (inferred.paramSQLType == SQL_WVARCHAR && inferred.columnSize > MAX_INLINE_CHAR) { inferred.paramSQLType = SQL_WLONGVARCHAR; - } else if (declared.sqlType == SQL_VARCHAR && + } else if (inferred.paramSQLType == SQL_VARCHAR && inferred.columnSize > MAX_INLINE_BINARY) { inferred.paramSQLType = SQL_LONGVARCHAR; - } else if (declared.sqlType == SQL_VARBINARY && + } else if (inferred.paramSQLType == SQL_VARBINARY && inferred.columnSize > MAX_INLINE_BINARY) { inferred.paramSQLType = SQL_LONGVARBINARY; } } - inferred.decimalDigits = declared.decimalDigits; - if (inferred.paramCType == SQL_C_DEFAULT || declared.sqlType == SQL_DECIMAL || - declared.sqlType == SQL_NUMERIC) { - inferred.columnSize = std::max(declared.columnSize, 1); - } } } @@ -927,26 +900,28 @@ static SQLRETURN BindTableValuedParameter(SqlHandle& handle, SQLHANDLE hStmt, in return rc; } - SQLRETURN bindRc = SQL_ERROR; try { - bindRc = BindParameterArray(handle, hStmt, columnValues, columnInfos, rows.size(), - paramBuffers, charEncoding); + 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; } - const SQLRETURN restoreRc = - SQLSetStmtAttr_ptr(hStmt, SQL_SOPT_SS_PARAM_FOCUS, nullptr, SQL_IS_INTEGER); - return SQL_SUCCEEDED(bindRc) ? restoreRc : bindRc; + 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()); @@ -966,8 +941,8 @@ SQLRETURN BindParameters(SqlHandle& handle, SQLHANDLE hStmt, const py::list& par SQLLEN* strLenOrIndPtr = nullptr; if (paramInfo.isTVP) { - RETCODE rc = BindTableValuedParameter(handle, hStmt, paramIndex, param, paramBuffers, - charEncoding); + RETCODE rc = BindTableValuedParameter(statementHandle, hStmt, paramIndex, param, + paramBuffers, charEncoding); if (!SQL_SUCCEEDED(rc)) { return rc; } @@ -2605,7 +2580,7 @@ SQLRETURN SQLExecute_wrap(const SqlHandlePtr statementHandle, 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; { @@ -3419,7 +3394,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); 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 index 4ce1c40a6..4dd5f5d5f 100644 --- a/tests/test_028_table_valued_parameters.py +++ b/tests/test_028_table_valued_parameters.py @@ -11,6 +11,7 @@ from mssql_python import ( DatabaseError, NotSupportedError, + ProgrammingError, SQL_INTEGER, SQL_WVARCHAR, ddbc_bindings, @@ -381,3 +382,214 @@ def record_timeout(statement_handle, attribute, value): 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", ["nvarchar(3)"], indirect=True) +@pytest.mark.parametrize( + ("value", "message", "driver_error"), + [ + ("x" * 5000, "would be truncated", "Syntax error or access violation"), + ("\u0100" * 5000, "Invalid precision value", "Invalid precision or scale value"), + ], + ids=["server-conversion-error", "bind-error"], +) +def test_tvp_error_keeps_diagnostics_and_restores_focus( + cursor, conversion_tvp, value, message, driver_error +): + with pytest.raises(ProgrammingError, match=message) as raised: + cursor.execute(conversion_tvp, ([(0, value)],)) + assert raised.value.driver_error == driver_error + cursor.execute("SELECT ?", 42) + assert cursor.fetchone()[0] == 42 + cursor.execute(conversion_tvp, ([(0, "ok")],)) + assert cursor.fetchone()[0] == "ok" + + +@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 + + +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") From d8f0a07df33b53623c483c5fcc2ce4c06bb22629 Mon Sep 17 00:00:00 2001 From: Gaurav Sharma Date: Wed, 16 Sep 2026 00:14:21 +0530 Subject: [PATCH 5/6] FIX: Prevent TVP metadata GIL deadlocks Release the GIL during TVP metadata reads and cleanup so Python forwarding threads can make progress. Preserve shutdown-safe destructor behavior and add bounded regressions for normal execution and metadata-error recovery. AB#48177 Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- mssql_python/pybind/ddbc_bindings.cpp | 41 ++++--- tests/test_028_table_valued_parameters.py | 125 ++++++++++++++++++++++ 2 files changed, 152 insertions(+), 14 deletions(-) diff --git a/mssql_python/pybind/ddbc_bindings.cpp b/mssql_python/pybind/ddbc_bindings.cpp index 94af75ef0..fbb4a6985 100644 --- a/mssql_python/pybind/ddbc_bindings.cpp +++ b/mssql_python/pybind/ddbc_bindings.cpp @@ -27,6 +27,7 @@ SQLRETURN BindParameterArray(SqlHandle& handle, SQLHANDLE hStmt, const py::list& std::vector& paramInfos, size_t paramSetSize, std::vector>& paramBuffers, const std::string& charEncoding); +static bool is_python_finalizing(); //------------------------------------------------------------------------------------------------- // Macro definitions @@ -626,7 +627,13 @@ class TableMetadataScope { public: explicit TableMetadataScope(SQLHANDLE statementHandle) : handle(statementHandle) {} ~TableMetadataScope() { - SQLFreeStmt_ptr(handle, SQL_CLOSE); + // 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)), @@ -649,8 +656,11 @@ static std::vector LoadTableColumnMetadata(const SqlHandlePt } SQLHANDLE hMetadataStmt = metadataHandle->get(); - SQLFreeStmt_ptr(hMetadataStmt, SQL_CLOSE); - SQLFreeStmt_ptr(hMetadataStmt, SQL_RESET_PARAMS); + { + 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)), @@ -698,17 +708,20 @@ static std::vector LoadTableColumnMetadata(const SqlHandlePt SQLSMALLINT decimalDigits = 0; SQLLEN indicator = 0; - 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; + { + 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)) { diff --git a/tests/test_028_table_valued_parameters.py b/tests/test_028_table_valued_parameters.py index 4dd5f5d5f..2694ac522 100644 --- a/tests/test_028_table_valued_parameters.py +++ b/tests/test_028_table_valued_parameters.py @@ -3,6 +3,9 @@ """ import datetime +import os +import subprocess +import sys import uuid from decimal import Decimal @@ -17,6 +20,12 @@ 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 @@ -593,3 +602,119 @@ def test_tvp_shape_and_type_errors_leave_cursor_reusable( 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 From 55be66b4a50e4e8bd6475d379fc1792c0b221746 Mon Sep 17 00:00:00 2001 From: Gaurav Sharma Date: Wed, 16 Sep 2026 00:52:23 +0530 Subject: [PATCH 6/6] FIX: Promote long TVP sources for bounded destinations Choose long TVP SQL types from source length regardless of destination size. Preserve valid Unicode values in bounded varchar columns and let SQL Server enforce destination limits. Add boundary and cursor-recovery regressions. AB#48177 Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- mssql_python/pybind/ddbc_bindings.cpp | 2 +- tests/test_028_table_valued_parameters.py | 36 +++++++++++++++-------- 2 files changed, 25 insertions(+), 13 deletions(-) diff --git a/mssql_python/pybind/ddbc_bindings.cpp b/mssql_python/pybind/ddbc_bindings.cpp index fbb4a6985..df0bcf9b5 100644 --- a/mssql_python/pybind/ddbc_bindings.cpp +++ b/mssql_python/pybind/ddbc_bindings.cpp @@ -861,7 +861,7 @@ static SQLRETURN BindTableValuedParameter(const SqlHandlePtr& statementHandle, S inferred.paramSQLType = declared.sqlType; inferred.columnSize = std::max(declared.columnSize, 1); inferred.decimalDigits = declared.decimalDigits; - } else if (declared.columnSize == 0) { + } else { if (inferred.paramSQLType == SQL_WVARCHAR && inferred.columnSize > MAX_INLINE_CHAR) { inferred.paramSQLType = SQL_WLONGVARCHAR; } else if (inferred.paramSQLType == SQL_VARCHAR && diff --git a/tests/test_028_table_valued_parameters.py b/tests/test_028_table_valued_parameters.py index 2694ac522..154de1022 100644 --- a/tests/test_028_table_valued_parameters.py +++ b/tests/test_028_table_valued_parameters.py @@ -475,25 +475,26 @@ def test_tvp_source_precision_survives_destination_conversion( assert rows == list(enumerate(values)) -@pytest.mark.parametrize("conversion_tvp", ["nvarchar(3)"], indirect=True) @pytest.mark.parametrize( - ("value", "message", "driver_error"), + ("conversion_tvp", "value", "recovery"), [ - ("x" * 5000, "would be truncated", "Syntax error or access violation"), - ("\u0100" * 5000, "Invalid precision value", "Invalid precision or scale value"), + ("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=["server-conversion-error", "bind-error"], + ids=["short-ascii", "short-unicode", "long-ascii", "long-unicode", "long-binary"], + indirect=["conversion_tvp"], ) -def test_tvp_error_keeps_diagnostics_and_restores_focus( - cursor, conversion_tvp, value, message, driver_error -): - with pytest.raises(ProgrammingError, match=message) as raised: +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 == driver_error + assert raised.value.driver_error == "Syntax error or access violation" cursor.execute("SELECT ?", 42) assert cursor.fetchone()[0] == 42 - cursor.execute(conversion_tvp, ([(0, "ok")],)) - assert cursor.fetchone()[0] == "ok" + cursor.execute(conversion_tvp, ([(0, recovery)],)) + assert cursor.fetchone()[0] == recovery @pytest.mark.parametrize("conversion_tvp", ["decimal(38,0)"], indirect=True) @@ -531,6 +532,17 @@ def test_tvp_source_buffers_cover_every_row(cursor, conversion_tvp, 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)