FIX: record correct UTF-16 code-unit length for every string parameter during native detection

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<py::list>(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<ParamInfo> 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"));

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<ParamInfo> 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<ParamInfo> 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<ParamInfo> 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;
         }
 

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"])