diff --git a/mssql_python/pybind/ddbc_bindings.cpp b/mssql_python/pybind/ddbc_bindings.cpp index 9908d842..8fc2895d 100644 --- a/mssql_python/pybind/ddbc_bindings.cpp +++ b/mssql_python/pybind/ddbc_bindings.cpp @@ -6008,6 +6008,7 @@ PYBIND11_MODULE(ddbc_bindings, m) { .def_readwrite("columnSize", &ParamInfo::columnSize) .def_readwrite("decimalDigits", &ParamInfo::decimalDigits) .def_readwrite("strLenOrInd", &ParamInfo::strLenOrInd) + .def_readonly("utf16Len", &ParamInfo::utf16Len) .def_property( "dataPtr", [](const ParamInfo& info) -> py::object { @@ -6073,6 +6074,29 @@ PYBIND11_MODULE(ddbc_bindings, m) { py::arg("statementHandle"), py::arg("query"), py::arg("params"), py::arg("inputSizes"), py::arg("isStmtPrepared"), py::arg("usePrepare"), py::arg("encodingSettings")); + // Test-only: run native parameter detection in isolation and return the resulting + // ParamInfo list, so the detection contract (e.g. utf16Len) can be asserted without + // a live SQL Server. Operates on a copy so the caller's list is not mutated. + m.def( + "DetectParamTypesForTesting", + [](py::list params, py::object inputSizes) -> py::list { + if (!inputSizes.is_none() && !py::isinstance(inputSizes)) { + throw py::type_error("inputSizes must be None or a list"); + } + py::list copy; + for (auto item : params) { + copy.append(item); + } + PyObject* sizes = inputSizes.is_none() ? Py_None : inputSizes.ptr(); + std::vector infos = DetectParamTypes(copy.ptr(), sizes); + py::list result; + for (auto& info : infos) { + result.append(py::cast(std::move(info))); + } + return result; + }, + "Test-only: run DetectParamTypes and return the ParamInfo list.", + py::arg("params"), py::arg("inputSizes") = py::none()); m.def("SQLExecuteMany", &SQLExecuteMany_wrap, "Execute statement with multiple parameter sets", py::arg("statementHandle"), py::arg("query"), py::arg("columnwise_params"), py::arg("paramInfos"), py::arg("paramSetSize"), py::arg("encodingSettings")); diff --git a/mssql_python/pybind/param_detect.hpp b/mssql_python/pybind/param_detect.hpp index 172a98b5..3b8ddc59 100644 --- a/mssql_python/pybind/param_detect.hpp +++ b/mssql_python/pybind/param_detect.hpp @@ -190,6 +190,26 @@ inline bool PyLongGreaterThan(PyObject* value, long long threshold) { return overflow > 0 || (overflow == 0 && result > threshold); } +// Number of UTF-16 code units a Python str occupies, with astral code points +// (> 0xFFFF) counted as the two units of their surrogate pair. This is exactly what +// the SQL_C_WCHAR binder writes for the string, so any wide-char buffer must be sized +// from this, not from PyUnicode_GET_LENGTH (code points) or the UTF-8 byte length. +// Caller guarantees obj is a str. +inline Py_ssize_t Utf16CodeUnitLen(PyObject* obj) { + const Py_ssize_t length = PyUnicode_GET_LENGTH(obj); + if (PyUnicode_KIND(obj) <= PyUnicode_2BYTE_KIND) { + // UCS-1 / UCS-2 storage: every code point is a single UTF-16 code unit. + return length; + } + // UCS-4 storage: astral code points expand to a surrogate pair. + Py_ssize_t utf16_len = 0; + const Py_UCS4* data = PyUnicode_4BYTE_DATA(obj); + for (Py_ssize_t j = 0; j < length; ++j) { + utf16_len += (data[j] > 0xFFFF) ? 2 : 1; + } + return utf16_len; +} + inline PyObject* FormatDecimalParam(PyObject* params, Py_ssize_t index, PyObject* value) { py::object formatted = steal(PyObject_CallMethod(value, "__format__", "s", "f")); if (!formatted) throw py::error_already_set(); @@ -267,6 +287,13 @@ inline void ApplyInputSizeOverride(PyObject* params, PyObject* inputSize, Py_ssi obj = PyList_GET_ITEM(params, index); } + // Record the post-mutation UTF-16 length for any string-valued param so a wide-char + // binder can size its buffer from the final string, not the pre-mutation object. + // Harmless for narrow/non-string binds, which do not read it. + if (PyUnicode_Check(obj)) { + info.utf16Len = Utf16CodeUnitLen(obj); + } + if (info.isDAE) { info.dataPtr = borrow(obj); } @@ -398,16 +425,7 @@ inline std::vector DetectParamTypes(PyObject* params, PyObject* input unsigned int kind = PyUnicode_KIND(obj); const void* udata = PyUnicode_DATA(obj); - Py_ssize_t utf16_len; - if (kind <= PyUnicode_2BYTE_KIND) { - utf16_len = length; - } else { - utf16_len = 0; - const Py_UCS4* data = PyUnicode_4BYTE_DATA(obj); - for (Py_ssize_t j = 0; j < length; ++j) { - utf16_len += (data[j] > 0xFFFF) ? 2 : 1; - } - } + Py_ssize_t utf16_len = Utf16CodeUnitLen(obj); // Detect whether the string needs wide-char (NVARCHAR) or narrow (VARCHAR) binding. // PyUnicode_IS_COMPACT_ASCII is a struct field check (O(1)), not a content scan. @@ -452,6 +470,7 @@ inline std::vector DetectParamTypes(PyObject* params, PyObject* input info.paramCType = is_unicode ? SQL_C_WCHAR : PARAM_C_TYPE_TEXT; } else { info.columnSize = is_unicode ? utf16_len : length; + info.utf16Len = utf16_len; info.paramSQLType = is_unicode ? SQL_WVARCHAR : SQL_VARCHAR; info.paramCType = is_unicode ? SQL_C_WCHAR : PARAM_C_TYPE_TEXT; } @@ -517,6 +536,7 @@ inline std::vector DetectParamTypes(PyObject* params, PyObject* input // isoformat(timespec="microseconds") via _normalize_time_param in cursor.py, // so calling the same method is what keeps the two paths in agreement. NormalizeTimeParam(params, i, info.columnSize); + info.utf16Len = Utf16CodeUnitLen(PyList_GET_ITEM(params, i)); continue; } diff --git a/tests/test_010_pybind_functions.py b/tests/test_010_pybind_functions.py index 106b64ca..d83f89a5 100644 --- a/tests/test_010_pybind_functions.py +++ b/tests/test_010_pybind_functions.py @@ -706,6 +706,100 @@ def test_unicode_handling(self): pass +@pytest.mark.skipif(not DDBC_AVAILABLE, reason="ddbc_bindings not available") +class TestDetectParamTypesUtf16Len: + """utf16Len must equal the number of UTF-16 code units the wide-char binder will + write for each string-valued parameter, computed AFTER any in-place normalization + (time -> isoformat string, Decimal -> formatted string). A future arena binder sizes + each wide slice from utf16Len, so a wrong value here is a heap-overflow landmine. + Nothing consumes utf16Len yet, so these assert the detection contract directly via + the DetectParamTypesForTesting hook. + + ODBC constants used (avoids importing the whole constants module): + SQL_DECIMAL = 3, SQL_C_CHAR = 1, SQL_WVARCHAR = -9, SQL_C_WCHAR = -8. + """ + + @staticmethod + def _detect(params, sizes=None): + return ddbc.DetectParamTypesForTesting(params, sizes) + + def test_ascii_string_counts_each_char_once(self): + (info,) = self._detect(["hello"]) + assert info.utf16Len == 5 + + def test_bmp_non_ascii_counts_each_codepoint_once(self): + # Latin-1 'é' and Greek letters are all in the BMP: one UTF-16 unit each. + cafe, greek = self._detect(["café", "αβγ"]) + assert cafe.utf16Len == 4 + assert greek.utf16Len == 3 + + def test_astral_char_counts_as_surrogate_pair(self): + # U+1F600 is astral: two UTF-16 code units (a surrogate pair). + two_emoji, mixed = self._detect(["😀😀", "a😀b"]) + assert two_emoji.utf16Len == 4 # 2 astral chars -> 2 pairs + assert mixed.utf16Len == 4 # 'a' + pair + 'b' + + def test_empty_string(self): + (info,) = self._detect([""]) + assert info.utf16Len == 0 + + def test_time_uses_isoformat_length_after_normalization(self): + import datetime + + # datetime.time is not a str, so a naive utf16Len would be 0. After + # normalization it becomes "01:02:03.000004" (15 ASCII chars). + (info,) = self._detect([datetime.time(1, 2, 3, 4)]) + assert info.utf16Len == 15 + + def test_large_string_takes_dae_and_reports_full_utf16_len(self): + (info,) = self._detect(["x" * 5000]) + assert info.isDAE is True + assert info.utf16Len == 5000 + + def test_non_string_param_has_zero_utf16_len(self): + # Ints/None never bind wide; utf16Len stays at its 0 default. + i, n = self._detect([42, None]) + assert i.utf16Len == 0 + assert n.utf16Len == 0 + + def test_setinputsizes_decimal_override_formats_then_measures(self): + import decimal + + # Internal _inputsizes 4-tuple form: (sql_type, c_type, column_size, decimal_digits). + sizes = [(3, 1, 18, 2)] # SQL_DECIMAL, SQL_C_CHAR + (info,) = self._detect([decimal.Decimal("12.5")], sizes) + assert info.utf16Len == 4 # "12.5" + + def test_setinputsizes_text_override_measures_final_string(self): + sizes = [(-9, -8, 50, 0)] # SQL_WVARCHAR, SQL_C_WCHAR + world, emoji = self._detect(["wörld", "😀"], sizes + [(-9, -8, 50, 0)]) + assert world.utf16Len == 5 + assert emoji.utf16Len == 2 # astral -> surrogate pair + + def test_time_override_normalizes_then_measures(self): + import datetime + + # A wide-text override on a time still routes through NormalizeTimeParam, so + # utf16Len must reflect the isoformat string, not the (non-str) time object. + sizes = [(-9, -8, 32, 0)] # SQL_WVARCHAR, SQL_C_WCHAR + (info,) = self._detect([datetime.time(1, 2, 3, 4)], sizes) + assert info.utf16Len == 15 # "01:02:03.000004" + + def test_caller_param_list_is_not_mutated(self): + import datetime + + # DetectParamTypes mutates its list in place (time -> str); the test hook must + # copy, so the caller's list is untouched. + params = [datetime.time(1, 2, 3), "keep"] + snapshot = list(params) + self._detect(params) + assert params == snapshot + + def test_bad_input_sizes_type_raises(self): + with pytest.raises(TypeError): + self._detect(["x"], "not-a-list") + + if __name__ == "__main__": # Run tests when executed directly pytest.main([__file__, "-v"])