PERF: Reduce fetch options, LOB copies, and converter dispatch

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

Author: jahnvi480

eng/profiler_benchmarks/controller.py

@@ -230,6 +230,7 @@ def verify_reused_source(source_root, revision):
 
 
 # Closed, reviewed dispatch bodies: main666, f539, many-only, and plain completion.
+# The options snapshot changes marshalling, not the plain Row-completion policy.
 # Method/comment/docstring edits require review and an explicit fingerprint update.
 # These reviewed method-source SHA256 digests are not credentials.
 _LEGACY_PYTHON_ROUTE = (
@@ -244,6 +245,9 @@ def verify_reused_source(source_root, revision):
 _PLAIN_COMPLETION_ROUTE = (
     "776ebc4532b69fe2492317ef6af9d4b2bb08e8a0f112613a7ce2600500bc37d8"  # DevSkim: ignore DS173237
 )
+_FETCH_OPTIONS_ROUTE = (
+    "6c8545f38956380e22b8cc9376e215e032d45ec68d1a29c669201f09477ec507"  # DevSkim: ignore DS173237
+)
 
 
 def python_source_identity(source_root, revision):
@@ -323,7 +327,7 @@ def python_source_identity(source_root, revision):
     fingerprint = hashlib.sha256(
         json.dumps(methods, ensure_ascii=True, separators=(",", ":")).encode()
     ).hexdigest()
-    if fingerprint in (_LEGACY_PYTHON_ROUTE, _PLAIN_COMPLETION_ROUTE):
+    if fingerprint in (_LEGACY_PYTHON_ROUTE, _PLAIN_COMPLETION_ROUTE, _FETCH_OPTIONS_ROUTE):
         values = (False, False, False)
     elif fingerprint == _LEGACY_FUSED_ROUTE:
         values = (True, True, True)
@@ -344,7 +348,7 @@ def python_source_identity(source_root, revision):
         validate_row_route(route)
         if route != expected:
             raise ValueError("Python source-policy drift")
-    elif fingerprint in (_MANY_ONLY_ROUTE, _PLAIN_COMPLETION_ROUTE):
+    elif fingerprint in (_MANY_ONLY_ROUTE, _PLAIN_COMPLETION_ROUTE, _FETCH_OPTIONS_ROUTE):
         raise ValueError("Missing Python route declaration")
     return dict(
         source_commit=revision,

mssql_python/cursor.py

@@ -677,8 +677,19 @@ def _refresh_decoding_cache(self):
         self._cached_char_encoding = char_decoding.get("encoding", "utf-16le")
         self._cached_char_ctype = char_decoding.get("ctype", ddbc_sql_const.SQL_WCHAR.value)
         self._cached_wchar_encoding = wchar_encoding
+        self._cached_fetch_options = None
         self._cached_decoding_generation = generation
 
+    def _create_fetch_options(self):
+        generation = self._cached_decoding_generation
+        options = ddbc_bindings._FetchOptions(
+            self._cached_char_encoding, self._cached_wchar_encoding, self._cached_char_ctype
+        )
+        # Allocation callbacks can refresh decoding; never install an older snapshot.
+        if self._cached_decoding_generation == generation:
+            self._cached_fetch_options = options
+        return options
+
     def _get_decoding_settings(self, sql_type):
         """
         Get decoding settings for a specific SQL type.
@@ -2889,14 +2900,22 @@ def fetchone(self) -> Union[None, Row]:
             started = perf_start()
             try:
                 fetch = ddbc_bindings.DDBCSQLFetchOne
-                ret = fetch(
-                    self.hstmt,
-                    row_data,
-                    char_enc,
-                    wchar_enc,
-                    self._cached_char_ctype,
-                    self.messages,
-                )
+                if fetch is _DEFAULT_NATIVE_FETCH_ONE:
+                    options = self._cached_fetch_options
+                    if options is None:
+                        options = self._create_fetch_options()
+                    ret = ddbc_bindings._fetchone_with_options(
+                        self.hstmt, row_data, options, self.messages
+                    )
+                else:
+                    ret = fetch(
+                        self.hstmt,
+                        row_data,
+                        char_enc,
+                        wchar_enc,
+                        self._cached_char_ctype,
+                        self.messages,
+                    )
             finally:
                 if started:
                     perf_stop("py::fetchone::cpp_call", started)
@@ -2990,15 +3009,23 @@ def fetchmany(self, size: Optional[int] = None) -> List[Row]:
             started = perf_start()
             try:
                 fetch = ddbc_bindings.DDBCSQLFetchMany
-                ret = fetch(
-                    self.hstmt,
-                    rows_data,
-                    size,
-                    char_enc,
-                    wchar_enc,
-                    self._cached_char_ctype,
-                    self.messages,
-                )
+                if fetch is _DEFAULT_NATIVE_FETCH_MANY:
+                    options = self._cached_fetch_options
+                    if options is None:
+                        options = self._create_fetch_options()
+                    ret = ddbc_bindings._fetchmany_with_options(
+                        self.hstmt, rows_data, size, options, self.messages
+                    )
+                else:
+                    ret = fetch(
+                        self.hstmt,
+                        rows_data,
+                        size,
+                        char_enc,
+                        wchar_enc,
+                        self._cached_char_ctype,
+                        self.messages,
+                    )
             finally:
                 if started:
                     perf_stop("py::fetchmany::cpp_call", started)

mssql_python/pybind/ddbc_bindings.cpp

@@ -23,6 +23,7 @@
 #include <cstring>  // For std::memcpy
 #include <filesystem>
 #include <iostream>
+#include <stdexcept>
 #include <utility>  // std::forward
 #include <datetime.h>  // CPython datetime API (PyDateTime_IMPORT, PyDateTime_GET_*, etc.)
 
@@ -3280,12 +3281,18 @@ py::object FetchLobColumnData(SQLHSTMT hStmt, SQLUSMALLINT colIndex, SQLSMALLINT
 
     while (true) {
         ++loopCount;
-        std::vector<char> chunk(DAE_CHUNK_SIZE, 0);
+        const size_t offset = buffer.size();
+        if (buffer.max_size() - offset < DAE_CHUNK_SIZE) {
+            throw std::length_error("LOB data exceeds maximum buffer size");
+        }
+        // Keep the existing read windows and zero-filled terminator room.
+        buffer.resize(offset + DAE_CHUNK_SIZE, 0);
+        char* chunk = buffer.data() + offset;
         SQLLEN actualRead = 0;
         {
             // Release the GIL during blocking SQLGetData LOB streaming
             py::gil_scoped_release release;
-            ret = SQLGetData_ptr(hStmt, colIndex, cType, chunk.data(), DAE_CHUNK_SIZE, &actualRead);
+            ret = SQLGetData_ptr(hStmt, colIndex, cType, chunk, DAE_CHUNK_SIZE, &actualRead);
         }
         CaptureFetchDiagnostics(hStmt, ret, messages, true);
 
@@ -3328,11 +3335,13 @@ py::object FetchLobColumnData(SQLHSTMT hStmt, SQLUSMALLINT colIndex, SQLSMALLINT
                 // Wide characters
                 size_t wcharSize = sizeof(SQLWCHAR);
                 if (bytesRead >= wcharSize && (bytesRead % wcharSize == 0)) {
-                    size_t wcharCount = bytesRead / wcharSize;
-                    std::vector<SQLWCHAR> alignedBuf(wcharCount);
-                    std::memcpy(alignedBuf.data(), chunk.data(), bytesRead);
-                    while (wcharCount > 0 && alignedBuf[wcharCount - 1] == 0) {
-                        --wcharCount;
+                    while (bytesRead >= wcharSize) {
+                        SQLWCHAR lastChar;
+                        // The byte destination need not be aligned for SQLWCHAR.
+                        std::memcpy(&lastChar, chunk + bytesRead - wcharSize, wcharSize);
+                        if (lastChar != 0) {
+                            break;
+                        }
                         bytesRead -= wcharSize;
                     }
                     if (bytesRead < DAE_CHUNK_SIZE) {
@@ -3343,8 +3352,8 @@ py::object FetchLobColumnData(SQLHSTMT hStmt, SQLUSMALLINT colIndex, SQLSMALLINT
                 }
             }
         }
+        buffer.resize(offset + bytesRead);
         if (bytesRead > 0) {
-            buffer.insert(buffer.end(), chunk.begin(), chunk.begin() + bytesRead);
             LOG("FetchLobColumnData: Appended %zu bytes at loop %d", bytesRead, loopCount);
         }
         if (ret == SQL_SUCCESS) {
@@ -6667,6 +6676,20 @@ PYBIND11_MODULE(ddbc_bindings, m) {
           py::arg("StatementHandle"), py::arg("colCount"), py::arg("row"), py::arg("charEncoding"),
           py::arg("wcharEncoding"), py::arg("charCtype"), py::arg("messages") = py::none());
     m.def("DDBCSQLMoreResults", &SQLMoreResults_wrap, "Check for more results in the result set");
+    py::class_<FetchOptions>(m, "_FetchOptions")
+        .def(py::init<const std::string&, const std::string&, int>());
+    m.def("_fetchone_with_options",
+          [](SqlHandlePtr statement, py::list& row, const FetchOptions& options,
+             py::handle messages) {
+              return FetchOne_wrap(std::move(statement), row, options.charEncoding,
+                                   options.wcharEncoding, options.charCtype, messages);
+          });
+    m.def("_fetchmany_with_options",
+          [](SqlHandlePtr statement, py::list& rows, int size, const FetchOptions& options,
+             py::handle messages) {
+              return FetchMany_wrap(std::move(statement), rows, size, options.charEncoding,
+                                    options.wcharEncoding, options.charCtype, messages);
+          });
     m.def("DDBCSQLFetchOne", &FetchOne_wrap, "Fetch one row from the result set",
           py::arg("StatementHandle"), py::arg("row"), py::arg("charEncoding") = "utf-16le",
           py::arg("wcharEncoding") = "utf-16le", py::arg("charCtype") = SQL_C_WCHAR,
@@ -6778,6 +6801,10 @@ PYBIND11_MODULE(ddbc_bindings, m) {
     // Add a version attribute
     m.attr("__version__") = "1.0.0";
 
+    m.def("_apply_output_converters", &RowFactory::apply_output_converters,
+          "Apply a cached converter map after native value materialization",
+          py::arg("values"), py::arg("converters"));
+
     // Fast Row construction in C++ — replaces Python list comprehension
     m.def("construct_rows", &RowFactory::construct_rows,
           "Build Row objects in C++ for fetchall/fetchmany fast path",

mssql_python/pybind/ddbc_bindings.h

@@ -426,6 +426,15 @@ struct DateTimeOffset {
     SQLSMALLINT timezone_minute;  // Offset minutes from UTC
 };
 
+struct FetchOptions {
+    const std::string charEncoding;
+    const std::string wcharEncoding;
+    const int charCtype;
+
+    FetchOptions(const std::string& charEncoding, const std::string& wcharEncoding, int charCtype)
+        : charEncoding(charEncoding), wcharEncoding(wcharEncoding), charCtype(charCtype) {}
+};
+
 // Struct to hold data buffers and indicators for each column
 struct ColumnBuffers {
     std::vector<std::vector<SQLCHAR>> charBuffers;

mssql_python/pybind/row_factory.hpp

@@ -5,8 +5,90 @@
 
 #include "py_ref.hpp"
 
+#include <utility>
+
 namespace RowFactory {
 
+// Only the post-fetch cached-map loop: no ODBC access, Row allocation, or UUID work.
+inline py::list apply_output_converters(const py::object& values, const py::object& converters) {
+    if (!PyList_CheckExact(values.ptr()) || !PyList_CheckExact(converters.ptr())) {
+        throw py::type_error("converter values and map must be exact lists");
+    }
+    py::list result =
+        steal<py::list>(PyList_GetSlice(values.ptr(), 0, PyList_GET_SIZE(values.ptr())));
+    if (!result)
+        throw py::error_already_set();
+
+    // Keep the Python iterators: their retained tuples affect finalizer timing
+    // when a callback replaces itself or mutates the source lists.
+    py::object pairs = steal(PyObject_CallFunctionObjArgs(reinterpret_cast<PyObject*>(&PyZip_Type),
+                                                          values.ptr(), converters.ptr(), nullptr));
+    if (!pairs)
+        throw py::error_already_set();
+    py::object items =
+        steal(PyObject_CallOneArg(reinterpret_cast<PyObject*>(&PyEnum_Type), pairs.ptr()));
+    if (!items)
+        throw py::error_already_set();
+    pairs = py::object();
+
+    // Retain the current inputs and last encoded value like the Python locals.
+    py::object value, converter, value_bytes;
+    for (Py_ssize_t i = 0;; ++i) {
+        py::object item = steal(PyIter_Next(items.ptr()));
+        if (!item) {
+            if (PyErr_Occurred())
+                throw py::error_already_set();
+            break;
+        }
+        PyObject* pair = PyTuple_GET_ITEM(item.ptr(), 1);
+        py::object next_value = borrow(PyTuple_GET_ITEM(pair, 0));
+        py::object next_converter = borrow(PyTuple_GET_ITEM(pair, 1));
+        value = std::move(next_value);
+        converter = std::move(next_converter);
+        item = py::object();
+        const int enabled = PyObject_IsTrue(converter.ptr());
+        if (enabled < 0)
+            throw py::error_already_set();
+        if (!enabled || value.is_none())
+            continue;
+
+        try {
+            const int is_string =
+                PyObject_IsInstance(value.ptr(), reinterpret_cast<PyObject*>(&PyUnicode_Type));
+            if (is_string < 0)
+                throw py::error_already_set();
+            PyObject* argument = value.ptr();
+            if (is_string) {
+                // Match str.encode's codec lookup, including the spelling.
+                // Subclasses and __class__ proxies still dispatch encode dynamically.
+                py::object encoded =
+                    steal(PyUnicode_CheckExact(value.ptr())
+                              ? PyUnicode_AsEncodedString(value.ptr(), "utf-16-le", nullptr)
+                              : PyObject_CallMethod(value.ptr(), "encode", "s", "utf-16-le"));
+                if (!encoded)
+                    throw py::error_already_set();
+                value_bytes = std::move(encoded);
+                argument = value_bytes.ptr();
+            }
+            py::object converted = steal(PyObject_CallOneArg(converter.ptr(), argument));
+            if (!converted)
+                throw py::error_already_set();
+            // Checked assignment also preserves Python's caught IndexError if a
+            // callback grows the input lists beyond the initial result copy.
+            if (PyList_SetItem(result.ptr(), i, converted.release().ptr()) < 0)
+                throw py::error_already_set();
+        } catch (py::error_already_set& error) {
+            if (!error.matches(PyExc_Exception))
+                throw;
+            // Preserve the existing cached path's keep-original-on-Exception contract.
+            error.restore();
+            PyErr_Clear();
+        }
+    }
+    items = py::object();
+    return result;
+}
+
 inline void initialize_row(
     const py::object& row, PyObject* row_data, const py::object& column_map,
     const py::object& cursor_obj, const py::object& column_map_lower,

mssql_python/row.py

@@ -9,6 +9,7 @@
 import uuid as _uuid
 from collections.abc import Mapping
 from typing import Any
+from mssql_python import ddbc_bindings
 from mssql_python.logging import logger
 
 
@@ -198,13 +199,21 @@ def _apply_output_converters_optimized(self, values, converter_map):
         """
         Apply output converters using pre-computed converter map for optimal performance.
 
+        Native materialization has already completed. Exact lists use a native
+        dispatch loop; both paths copy values and preserve dynamic string encoding.
+
         Args:
             values: Raw values from the database
             converter_map: Pre-computed list of converters (one per column, None if no converter)
 
         Returns:
             List of converted values
         """
+        # Native fetches and the cursor cache supply exact lists. Keep arbitrary
+        # iterables on the Python path, including their iteration side effects.
+        if type(values) is list and type(converter_map) is list:
+            return ddbc_bindings._apply_output_converters(values, converter_map)
+
         converted_values = list(values)
 
         for i, (value, converter) in enumerate(zip(values, converter_map)):