PERF: avoid repeated text fetch preparation for fetchmany(1)

eng/profiler_benchmarks/README.md

@@ -12,7 +12,7 @@ python -m eng.profiler_benchmarks.controller --base main --candidate HEAD \
 python -m eng.profiler_benchmarks.report profiler-results/report.json
 ```
 
-The fixed registry has 25 tasks. `--scenarios` runs a local subset, but subset
+The fixed registry has 26 tasks. `--scenarios` runs a local subset, but subset
 reports remain incomplete and cannot produce a verdict.
 
 `lob_varchar_256k_fetchall` fetches one 256 KiB `VARCHAR(MAX)` value to exercise
@@ -26,6 +26,17 @@ validation (including zero), EOF validation, and diagnostic checks are outside
 the timed fetch window. Both revisions run the same workload; the report's
 improvement thresholds are unchanged.
 
+`fetchmany_one_text` collects 10,000 ordered INT/NVARCHAR rows from the existing
+fixture using explicit `fetchmany(1)`, without output converters. It retains all
+rows through the timer boundary and includes the terminal empty batch and list
+accumulation. Query execution, expected-value construction, exact value/type/count
+and order checks, diagnostics validation, and cursor cleanup are outside timing.
+This single workload targets repeated text-fetch dispatch preparation; it uses
+the fixture's `unicode_row_{i}` values, not the standalone study's fixed-width
+text. Both revisions run the same profiling-enabled workload with the existing
+report thresholds. It measures a fresh main-relative change, not a guaranteed
+improvement or a comparison against pyodbc.
+
 `output_converter_fetchall` and `output_converter_fetchmany1000` measure separate
 fetch APIs over 10,000 ordered integer/NVARCHAR rows from the existing fixture.
 An `SQL_WVARCHAR` converter decodes UTF-16LE bytes and uppercases the text.
@@ -58,7 +69,7 @@ width matrix.
 
 This conversion has no guaranteed instrumented native phase, so **only**
 `row_to_dict` permits an empty native diagnostics dictionary. Skipped results,
-malformed counters, and missing native samples from the other 24 tasks remain
+malformed counters, and missing native samples from the other 25 tasks remain
 invalid in both the worker and report validation. The row uses the existing
 paired wall-time comparison and unchanged thresholds: it can demonstrate an
 observed main-relative improvement, not assert a speedup in advance or establish

eng/profiler_benchmarks/report.py

@@ -41,6 +41,7 @@
     "scalar_fetchval": "10,000 scalar values / fetchval() (debug disabled)",
     "output_converter_fetchall": "10,000 output-converted rows / fetchall()",
     "output_converter_fetchmany1000": "10,000 output-converted rows / fetchmany(1000)",
+    "fetchmany_one_text": "10,000 INT/NVARCHAR rows / fetchmany(1)",
     "row_to_dict": "10,000 rows x 32 columns / dict(row._mapping) (warm)",
 }
 CASES = tuple(TASK_NAMES)

eng/profiler_benchmarks/workloads.py

@@ -194,6 +194,33 @@ def scalar_fetchval(conn, table, ctx):
             ctx.disable()
 
 
+def fetchmany_one_text(conn, table, ctx):
+    """Collect mixed rows one at a time; setup and validation are not timed."""
+    expected = [(index, f"unicode_row_{index}") for index in range(10_000)]
+    with conn.cursor() as cursor:
+        try:
+            cursor.execute(f"SELECT TOP (10000) int_col, nvarchar_col FROM {table} ORDER BY id")
+            rows = []
+            ctx.enable()
+            start = time.perf_counter()
+            while batch := cursor.fetchmany(1):
+                rows.extend(batch)
+            wall_ms = (time.perf_counter() - start) * 1000
+            cpp, py = ctx.collect()
+            assert [tuple(row) for row in rows] == expected
+            assert all(type(row[0]) is int and type(row[1]) is str for row in rows)
+            assert not cursor.messages, "Clean size-one fetch unexpectedly produced diagnostics"
+            return dict(
+                title="Mixed text / fetchmany(1)",
+                wall_ms=wall_ms,
+                cpp=cpp,
+                py=py,
+                detail="Rows: 10000; types: int,nvarchar; API: fetchmany(1)",
+            )
+        finally:
+            ctx.disable()
+
+
 def output_converter_fetch(conn, table, ctx, batched=False):
     """Time fetching with a UTF-16LE converter; validate retained rows afterward."""
     from mssql_python import SQL_WVARCHAR
@@ -308,5 +335,6 @@ def registry():
     result["scalar_fetchval"] = (scalar_fetchval, True)
     result["output_converter_fetchall"] = (output_converter_fetch, True)
     result["output_converter_fetchmany1000"] = (partial(output_converter_fetch, batched=True), True)
+    result["fetchmany_one_text"] = (fetchmany_one_text, True)
     result["row_to_dict"] = (row_to_dict, True)
     return result

mssql_python/_native_fetch.py

@@ -13,8 +13,9 @@
 Plans retain the cursor's shared column-name cache without copying it. Column
 names must be a plain tuple or an exact _ColumnNames instance; custom subclasses
 keep the Python fallback.
-Size-one fetchmany keeps text-bearing results on the batch path, whose
-platform-specific decoding and malformed-text fallbacks differ from fetchone.
+Size-one fetchmany shares the batch wide-text decoder, whose malformed-text
+fallback differs from strict fetchone. Narrow-text batches still use the
+batch reader but complete natively, avoiding repeated policy preparation.
 """
 
 import sys

mssql_python/pybind/ddbc_bindings.cpp

@@ -3553,15 +3553,11 @@ static inline bool IsLobOrVariantColumn(SQLSMALLINT dataType, SQLULEN columnSize
 }
 
 static inline bool CanUseSingleRowForFetchMany(SQLSMALLINT dataType, SQLULEN columnSize) {
-    // Bound text conversion has platform/codec-specific fallbacks that GetRowData
-    // does not share. Keep that contract even when the requested batch size is one.
+    // Narrow text still needs the batch's platform/codec-specific conversion.
     switch (dataType) {
         case SQL_CHAR:
         case SQL_VARCHAR:
         case SQL_LONGVARCHAR:
-        case SQL_WCHAR:
-        case SQL_WVARCHAR:
-        case SQL_WLONGVARCHAR:
             return false;
         default:
             return !IsLobOrVariantColumn(dataType, columnSize);
@@ -3572,7 +3568,8 @@ static SQLRETURN GetRowData(const SqlHandlePtr& StatementHandle, SQLUSMALLINT co
                             py::list& row, const std::string& charEncoding,
                             const std::string& wcharEncoding, int charCtype,
                             py::handle messages, const ResultMetadataCache::Snapshot& snapshot,
-                            bool completeResult = false, py::handle converters = {}) {
+                            bool completeResult = false, py::handle converters = {},
+                            bool batchText = false) {
     PERF_TIMER("SQLGetData_wrap");
     // Note: wcharEncoding parameter is reserved for future use
     // Currently WCHAR data always uses UTF-16LE for Windows compatibility
@@ -3941,7 +3938,11 @@ static SQLRETURN GetRowData(const SqlHandlePtr& StatementHandle, SQLUSMALLINT co
                                     OutputConverters::enabled(converters, i - 1))
                                     text = OutputConverters::text(textData, textSize);
                                 row.append(text ? text
-                                                : FetchText::from_utf16_native(textData, textSize));
+                                                : batchText
+                                                      ? steal(FetchText::decode_batch_utf16_native(
+                                                            textData, textSize))
+                                                      : FetchText::from_utf16_native(textData,
+                                                                                     textSize));
                                 LOG("SQLGetData: Appended NVARCHAR string "
                                     "length=%lu for column %d",
                                     (unsigned long)numCharsInData, i);
@@ -4407,7 +4408,8 @@ SQLRETURN SQLGetData_wrap(SqlHandlePtr StatementHandle, SQLUSMALLINT colCount, p
 static SQLRETURN FetchSingleRow(const SqlHandlePtr& handle, py::list& row,
                                 const std::string& charEncoding,
                                 const std::string& wcharEncoding, int charCtype,
-                                py::handle messages, py::handle converters = {}) {
+                                py::handle messages, py::handle converters = {},
+                                bool batchText = false) {
     PERF_TIMER("FetchSingleRow");
     SQLRETURN ret = SQL_ERROR;
     ResultMetadataFailureGuard metadataFailure(handle->resultMetadata, ret);
@@ -4433,7 +4435,7 @@ static SQLRETURN FetchSingleRow(const SqlHandlePtr& handle, py::list& row,
                                   ? static_cast<SQLSMALLINT>(snapshot.metadata->columns.size())
                                   : SQLNumResultCols_wrap(handle, messages);
         ret = GetRowData(handle, colCount, row, charEncoding, wcharEncoding, charCtype, messages,
-                          snapshot, true, converters);
+                          snapshot, true, converters, batchText);
     }
     return ret;
 }
@@ -5236,7 +5238,7 @@ SQLRETURN FetchMany_wrap(SqlHandlePtr StatementHandle, py::list& rows, int fetch
     if (singleRowEligible) {
         py::list row;
         ret = FetchSingleRow(StatementHandle, row, charEncoding, wcharEncoding, charCtype, messages,
-                              converters);
+                              converters, true);
         if (SQL_SUCCEEDED(ret))
             rows.append(row);
         return ret;

mssql_python/pybind/ddbc_bindings.h

@@ -794,38 +794,9 @@ inline void ProcessWChar(PyObject* row, ColumnBuffers& buffers, const void* colI
     // fetchBufferSize includes null-terminator, numCharsInData doesn't. Hence
     // '<'
     if (!colInfo->isLob && numCharsInData < colInfo->fetchBufferSize) {
-#if defined(__APPLE__) || defined(__linux__)
-        // Performance: Direct UTF-16 decode (SQLWCHAR is 2 bytes on
-        // Linux/macOS)
         SQLWCHAR* wcharData = &buffers.wcharBuffers[col - 1][rowIdx * colInfo->fetchBufferSize];
-        PyObject* pyStr = FetchText::decode_utf16_native(
-            reinterpret_cast<const char*>(wcharData), numCharsInData * sizeof(SQLWCHAR));
-        if (pyStr) {
-            PyList_SET_ITEM(row, col - 1, pyStr);
-        } else {
-            PyErr_Clear();  // Ignore decode error, return empty string
-            PyObject* emptyStr = PyUnicode_FromStringAndSize("", 0);
-            if (!emptyStr) {
-                Py_INCREF(Py_None);
-                PyList_SET_ITEM(row, col - 1, Py_None);
-            } else {
-                PyList_SET_ITEM(row, col - 1, emptyStr);
-            }
-        }
-#else
-        // Performance: Direct Python C API call (Windows where SQLWCHAR ==
-        // wchar_t)
-        PyObject* pyStr = PyUnicode_FromWideChar(
-            reinterpret_cast<wchar_t*>(
-                &buffers.wcharBuffers[col - 1][rowIdx * colInfo->fetchBufferSize]),
-            numCharsInData);
-        if (!pyStr) {
-            Py_INCREF(Py_None);
-            PyList_SET_ITEM(row, col - 1, Py_None);
-        } else {
-            PyList_SET_ITEM(row, col - 1, pyStr);
-        }
-#endif
+        PyList_SET_ITEM(row, col - 1, FetchText::decode_batch_utf16_native(
+            reinterpret_cast<const char*>(wcharData), numCharsInData * sizeof(SQLWCHAR)));
     } else {
         // Slow path: LOB data requires separate fetch call
         PyList_SET_ITEM(row, col - 1,

mssql_python/pybind/fetch_text.hpp

@@ -22,4 +22,19 @@ inline py::object from_utf16_native(const char* data, Py_ssize_t size) {
     return result;
 }
 
+// Preserve the established batch decoder, which differs from strict fetchone.
+inline PyObject* decode_batch_utf16_native(const char* data, Py_ssize_t size) {
+#if defined(__APPLE__) || defined(__linux__)
+    PyObject* result = decode_utf16_native(data, size);
+    if (result)
+        return result;
+    PyErr_Clear();
+    result = PyUnicode_FromStringAndSize("", 0);
+#else
+    PyObject* result = PyUnicode_FromWideChar(
+        reinterpret_cast<const wchar_t*>(data), size / sizeof(wchar_t));
+#endif
+    return result ? result : Py_NewRef(Py_None);
+}
+
 }  // namespace FetchText

mssql_python/pybind/native_fetch.hpp

@@ -717,7 +717,7 @@ inline py::object fetch(PyObject* self, Context* context, Mode mode,
         return fallback(self, rt, mode, args, nargs, keywords);
 
     if (mode == Mode::Many) {
-        if (!single_size(*plan, args, nargs, keywords) || !plan->manyEligible)
+        if (!single_size(*plan, args, nargs, keywords))
             return fallback(self, rt, mode, args, nargs, keywords);
         auto metadata = plan->handle->resultMetadata.snapshot();
         if (metadata.generation != plan->metadata.generation ||
@@ -729,8 +729,25 @@ inline py::object fetch(PyObject* self, Context* context, Mode mode,
 
     PERF_TIMER("NativeCursor::fetch");
     py::list values;
-    SQLRETURN ret = FetchSingleRow(plan->handle, values, plan->charEncoding,
-                                   plan->wideEncoding, plan->charCtype, plan->at(Messages));
+    SQLRETURN ret;
+    if (mode == Mode::Many && !plan->manyEligible) {
+        // Keep batch decoding (including malformed-text handling), but complete
+        // size-one reads here so ordinary counter writes do not invalidate the plan.
+        py::list rows;
+        ret = FetchMany_wrap(plan->handle, rows, 1, plan->charEncoding,
+                             plan->wideEncoding, plan->charCtype, plan->at(Messages));
+        if (!current(self, context, state, *plan, rt) ||
+            (!SQL_SUCCEEDED(ret) && ret != SQL_NO_DATA))
+            return py::reinterpret_borrow<py::object>(self).attr("_complete_fetchmany")(rows, ret);
+        if (PyList_GET_SIZE(rows.ptr()))
+            values = py::reinterpret_borrow<py::list>(PyList_GET_ITEM(rows.ptr(), 0));
+        else
+            ret = SQL_NO_DATA;
+    } else {
+        ret = FetchSingleRow(plan->handle, values, plan->charEncoding,
+                              plan->wideEncoding, plan->charCtype, plan->at(Messages),
+                              {}, mode == Mode::Many);
+    }
     // A codec, finalizer, another thread or reentrant callback can mutate state
     // while SQLFetch releases the GIL. Complete the consumed row, never refetch it.
     if (!current(self, context, state, *plan, rt) ||

tests/test_036_profiler_ci.py

@@ -721,10 +721,11 @@ def __exit__(self, *args):
 def test_report_cases_match_the_executed_workload_registry():
     _, workloads = controller.load_suite()
     assert tuple(workloads.registry()) == reporting.CASES
-    assert len(reporting.CASES) == 25
+    assert len(reporting.CASES) == 26
     assert reporting.CASES[-1] == "row_to_dict"
     assert workloads.registry()["row_to_dict"] == (workloads.row_to_dict, True)
     assert workloads.registry()["scalar_fetchval"] == (workloads.scalar_fetchval, True)
+    assert workloads.registry()["fetchmany_one_text"] == (workloads.fetchmany_one_text, True)
     assert workloads.registry()["output_converter_fetchall"] == (
         workloads.output_converter_fetch,
         True,
@@ -737,6 +738,93 @@ def test_report_cases_match_the_executed_workload_registry():
     ]
 
 
+def test_fetchmany_one_text_workload_boundaries_and_report(monkeypatch, report):
+    cursor = MagicMock()
+    cursor.messages = []
+    connection = MagicMock()
+    connection.cursor.return_value.__enter__.return_value = cursor
+    context = MagicMock()
+    counters = {"ddbc::FetchSingleRow": dict(calls=10001, total_us=10001, min_us=1, max_us=1)}
+    context.collect.return_value = (counters, {})
+    references = []
+
+    class RetainedRow(list):
+        def __iter__(self):
+            context.collect.assert_called_once()
+            return super().__iter__()
+
+    def fetch(size):
+        assert size == 1 and clock.call_count == 1
+        context.collect.assert_not_called()
+        index = len(references)
+        if index == 10_000:
+            return []
+        row = RetainedRow((index, f"unicode_row_{index}"))
+        references.append(weakref.ref(row))
+        return [row]
+
+    cursor.fetchmany.side_effect = fetch
+
+    def tick():
+        cursor.execute.assert_called_once_with(
+            "SELECT TOP (10000) int_col, nvarchar_col FROM #perf_test ORDER BY id"
+        )
+        context.enable.assert_called_once()
+        context.collect.assert_not_called()
+        if clock.call_count == 1:
+            cursor.fetchmany.assert_not_called()
+        else:
+            assert cursor.fetchmany.call_count == 10_001
+            assert len(references) == 10_000 and all(ref() is not None for ref in references)
+        return 1.0 if clock.call_count == 1 else 1.1
+
+    clock = MagicMock(side_effect=tick)
+    monkeypatch.setattr(benchmark_workloads.time, "perf_counter", clock)
+    result = benchmark_workloads.fetchmany_one_text(connection, "#perf_test", context)
+    assert clock.call_count == 2
+    assert result["wall_ms"] == pytest.approx(100)
+    assert result["cpp"] is counters
+    assert result["detail"] == "Rows: 10000; types: int,nvarchar; API: fetchmany(1)"
+    cursor.fetchone.assert_not_called()
+    cursor.fetchall.assert_not_called()
+    context.collect.assert_called_once()
+    context.disable.assert_called_once()
+    connection.cursor.return_value.__exit__.assert_called_once()
+
+    # Equal-valued floats must not pass the INT contract.
+    cursor.fetchmany.side_effect = iter(
+        [[(float(index), f"unicode_row_{index}")] for index in range(10_000)] + [[]]
+    )
+    context.reset_mock()
+    connection.cursor.return_value.__exit__.reset_mock()
+    clock.side_effect = (1.0, 1.1)
+    with pytest.raises(AssertionError):
+        benchmark_workloads.fetchmany_one_text(connection, "#perf_test", context)
+    context.collect.assert_called_once()
+    context.disable.assert_called_once()
+    connection.cursor.return_value.__exit__.assert_called_once()
+
+    context.reset_mock()
+    connection.cursor.return_value.__exit__.reset_mock()
+    cursor.fetchmany.side_effect = RuntimeError("fetch failed")
+    clock.side_effect = (1.0,)
+    with pytest.raises(RuntimeError, match="fetch failed"):
+        benchmark_workloads.fetchmany_one_text(connection, "#perf_test", context)
+    context.disable.assert_called_once()
+    connection.cursor.return_value.__exit__.assert_called_once()
+
+    clear_slowdowns(report)
+    for pair in report["pairs"]:
+        pair["candidate"]["scenarios"]["fetchmany_one_text"]["wall_ms"] = 7
+    reporting.validate(report)
+    row = next(row for row in reporting.comparisons(report) if row["name"] == "fetchmany_one_text")
+    assert row["status"] == "improvement" and row["change_pct"] == pytest.approx(-30)
+    assert (
+        f"| {reporting.TASK_NAMES['fetchmany_one_text']} | 10.000 ms | 7.000 ms | -30.0% | consistent improvement |"
+        in reporting.render([report], "c" * 40, 42)
+    )
+
+
 @pytest.mark.parametrize("batched", (False, True), ids=("fetchall", "fetchmany1000"))
 def test_output_converter_workload_boundaries_and_report(monkeypatch, report, batched):
     from mssql_python import SQL_WVARCHAR