From 2024c7e0d6fcb32ca2dbab5b0cd4f32c3c31f987 Mon Sep 17 00:00:00 2001 From: Jahnvi Thakkar Date: Thu, 12 Nov 2026 13:14:19 +0530 Subject: [PATCH 1/2] PERF: avoid repeated text fetch preparation for fetchmany(1) Preserve batch decoding while completing eligible size-one text fetches natively. Add one mixed-text performance-report workload and focused contract coverage. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- eng/profiler_benchmarks/README.md | 15 ++++- eng/profiler_benchmarks/report.py | 1 + eng/profiler_benchmarks/workloads.py | 28 +++++++++ mssql_python/_native_fetch.py | 5 +- mssql_python/pybind/ddbc_bindings.cpp | 22 ++++--- mssql_python/pybind/ddbc_bindings.h | 33 +--------- mssql_python/pybind/fetch_text.hpp | 15 +++++ mssql_python/pybind/native_fetch.hpp | 23 ++++++- tests/test_036_profiler_ci.py | 90 ++++++++++++++++++++++++++- tests/test_native_fetch_dispatch.py | 87 ++++++++++++++++++++++++++ tests/test_output_converter_bytes.py | 9 +-- tests/test_single_row_pipeline.py | 37 +++++++++-- 12 files changed, 308 insertions(+), 57 deletions(-) diff --git a/eng/profiler_benchmarks/README.md b/eng/profiler_benchmarks/README.md index 8261b79d9..2d116b9f4 100644 --- a/eng/profiler_benchmarks/README.md +++ b/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 diff --git a/eng/profiler_benchmarks/report.py b/eng/profiler_benchmarks/report.py index 631869801..030e74e1d 100644 --- a/eng/profiler_benchmarks/report.py +++ b/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) diff --git a/eng/profiler_benchmarks/workloads.py b/eng/profiler_benchmarks/workloads.py index 6848504f2..6720f125a 100644 --- a/eng/profiler_benchmarks/workloads.py +++ b/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 diff --git a/mssql_python/_native_fetch.py b/mssql_python/_native_fetch.py index 625faec3b..dc2492e3a 100644 --- a/mssql_python/_native_fetch.py +++ b/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 diff --git a/mssql_python/pybind/ddbc_bindings.cpp b/mssql_python/pybind/ddbc_bindings.cpp index 0e20bccbb..f217181d5 100644 --- a/mssql_python/pybind/ddbc_bindings.cpp +++ b/mssql_python/pybind/ddbc_bindings.cpp @@ -3456,15 +3456,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); @@ -3475,7 +3471,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 @@ -3844,7 +3841,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); @@ -4310,7 +4311,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); @@ -4336,7 +4338,7 @@ static SQLRETURN FetchSingleRow(const SqlHandlePtr& handle, py::list& row, ? static_cast(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; } @@ -5139,7 +5141,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; diff --git a/mssql_python/pybind/ddbc_bindings.h b/mssql_python/pybind/ddbc_bindings.h index a0eaddb42..57f0e68e5 100644 --- a/mssql_python/pybind/ddbc_bindings.h +++ b/mssql_python/pybind/ddbc_bindings.h @@ -754,38 +754,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(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( - &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(wcharData), numCharsInData * sizeof(SQLWCHAR))); } else { // Slow path: LOB data requires separate fetch call PyList_SET_ITEM(row, col - 1, diff --git a/mssql_python/pybind/fetch_text.hpp b/mssql_python/pybind/fetch_text.hpp index 0f365f2f3..2d55dfea9 100644 --- a/mssql_python/pybind/fetch_text.hpp +++ b/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(data), size / sizeof(wchar_t)); +#endif + return result ? result : Py_NewRef(Py_None); +} + } // namespace FetchText diff --git a/mssql_python/pybind/native_fetch.hpp b/mssql_python/pybind/native_fetch.hpp index 0a3a08c82..de90dc196 100644 --- a/mssql_python/pybind/native_fetch.hpp +++ b/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(self).attr("_complete_fetchmany")(rows, ret); + if (PyList_GET_SIZE(rows.ptr())) + values = py::reinterpret_borrow(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) || diff --git a/tests/test_036_profiler_ci.py b/tests/test_036_profiler_ci.py index 0ff9708bf..c2bb6d292 100644 --- a/tests/test_036_profiler_ci.py +++ b/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 diff --git a/tests/test_native_fetch_dispatch.py b/tests/test_native_fetch_dispatch.py index 71e94a5dc..ca4a2196a 100644 --- a/tests/test_native_fetch_dispatch.py +++ b/tests/test_native_fetch_dispatch.py @@ -226,6 +226,9 @@ class CustomNames(_ColumnNames): with _native_fetch_counters() as profiling, connection.cursor() as cursor: for names_type in (tuple, CustomNames): + # Collect prior watched cursor cycles before asserting counts for a + # cold plan with no concurrent invalidation. + gc.collect() warm(cursor, "fetchone") names = names_type(cursor._cached_result_columns) cursor._cached_result_columns = names @@ -239,6 +242,90 @@ class CustomNames(_ColumnNames): ) +@pytest.mark.parametrize("cycle", ("plain", "cursor")) +def test_names_plan_gc_invalidation_completes_once(conn_str, cycle): + _redesign_runtime() + _redesign_subprocess("_names_plan_gc_invalidation", cycle) + + +def _names_plan_gc_invalidation(cycle): + import os + from mssql_python.cursor import _NATIVE_FETCH_RUNTIME + + gc.collect() + policy_code = type(_NATIVE_FETCH_RUNTIME[2]).__call__.__code__ + with mssql_python.connect(os.environ["DB_CONNECTION_STRING"], timeout=5) as connection: + held = [] + if cycle == "cursor": + prior = connection.cursor() + warm(prior, "fetchone") + prior.__dict__["cycle"] = prior + prior.close() + reference = weakref.ref(prior) + else: + prior = {} + prior["cycle"] = prior + held.append(prior) + del prior + + with _native_fetch_counters() as profiling, connection.cursor() as cursor: + warm(cursor, "fetchone") + names = tuple(cursor._cached_result_columns) + cursor._cached_result_columns = names + collections = [] + + def collect_during_policy(frame, event, argument): + if event == "call" and frame.f_code is policy_code and not collections: + held.clear() + collections.append(gc.collect()) + return collect_during_policy + + previous = sys.gettrace() + sys.settrace(collect_during_policy) + try: + with _fetchone_window(profiling) as observation: + row = cursor.fetchone() + finally: + sys.settrace(previous) + assert collections and collections[0] > 0 + if cycle == "cursor": + assert reference() is None + assert row[0] == 3 and row._column_names is names + assert cursor.rowcount == 3 and cursor.rownumber == 2 + fallback = int(cycle == "cursor") + _assert_fetchone_counts( + observation, prepare=1, native=1 - fallback, fallback=fallback, sql=1 + ) + assert cursor.fetchone()[0] == 4 + assert cursor.fetchone()[0] == 5 + assert cursor.fetchone() is None + assert row[0] == 3 and row._column_names is names + + +@pytest.mark.parametrize("sql_type", ("CHAR", "VARCHAR", "NCHAR", "NVARCHAR")) +def test_warm_text_fetchmany_does_not_repeat_policy(connection, sql_type): + if ddbc_bindings._fetch_observer is None: + pytest.skip("Direct native dispatch requires supported CPython watchers") + from mssql_python.cursor import _NATIVE_FETCH_RUNTIME + + with connection.cursor() as cursor: + cursor.execute( + f"SELECT n, CAST(N'x' AS {sql_type}(1)) AS payload " + "FROM (VALUES (1), (2), (3), (4)) AS v(n) ORDER BY n" + ) + for n in (1, 2): + assert tuple(cursor.fetchmany(1)[0]) == (n, "x") + with _fetchone_window(None) as observation: + rows = cursor.fetchmany(1) + cursor.fetchmany(1) + assert cursor.fetchmany(1) == [] + assert [tuple(row) for row in rows] == [(3, "x"), (4, "x")] + calls = observation["python"] + assert calls.get(type(_NATIVE_FETCH_RUNTIME[2]).__call__.__code__, 0) == 0 + assert calls.get(_NATIVE_FETCH_RUNTIME[3][1].__code__, 0) == 0 + assert cursor.rowcount == 4 + assert cursor.rownumber == 3 + + @pytest.mark.parametrize("method", METHODS) @pytest.mark.parametrize("scope", ("instance", "class")) def test_warm_map_override_and_restoration(connection, method, scope): diff --git a/tests/test_output_converter_bytes.py b/tests/test_output_converter_bytes.py index 68788bbe1..c6b2a8763 100644 --- a/tests/test_output_converter_bytes.py +++ b/tests/test_output_converter_bytes.py @@ -30,7 +30,7 @@ def _fetch(cursor, method): while (row := cursor.fetchone()) is not None: rows.append(row) else: - while batch := cursor.fetchmany(1000): + while batch := cursor.fetchmany(1 if method == "fetchmany_one" else 1000): rows.extend(batch) return rows @@ -44,7 +44,7 @@ def connection(conn_str): connection.close() -@pytest.mark.parametrize("method", METHODS) +@pytest.mark.parametrize("method", (*METHODS, "fetchmany_one")) @pytest.mark.parametrize("width", (1, 32)) def test_converter_native_bytes_and_row_major_order(connection, method, width): inputs = [] @@ -53,6 +53,7 @@ def test_converter_native_bytes_and_row_major_order(connection, method, width): bridge_name = { "fetchone": "DDBCSQLFetchOne", "fetchmany": "DDBCSQLFetchMany", + "fetchmany_one": "DDBCSQLFetchMany", "fetchall": "DDBCSQLFetchAll", }[method] bridge = getattr(ddbc_bindings, bridge_name) @@ -88,7 +89,7 @@ def observe(*args): assert native_apply(raw_values, [lambda raw: raw] * len(raw_values)) == inputs -@pytest.mark.parametrize("method", METHODS) +@pytest.mark.parametrize("method", (*METHODS, "fetchmany_one")) def test_converter_empty_nul_and_bom_values(connection, method): expected = ["", "\x00", "\x00\x00", "\ufeff", "\ufffe", "\U0001f642"] calls = [] @@ -617,7 +618,7 @@ def test_converter_respects_narrow_decoding_settings(connection, method): assert row[0] == observed[0] -@pytest.mark.parametrize("method", METHODS) +@pytest.mark.parametrize("method", (*METHODS, "fetchmany_one")) @pytest.mark.parametrize("raw_utf16", ("0x00D8", "0x00DC", "0x00D84100")) def test_malformed_text_retains_legacy_fetch_behavior(connection, method, raw_utf16): outcomes = [] diff --git a/tests/test_single_row_pipeline.py b/tests/test_single_row_pipeline.py index 45518bd68..31e46250d 100644 --- a/tests/test_single_row_pipeline.py +++ b/tests/test_single_row_pipeline.py @@ -153,6 +153,25 @@ def test_fetchmany_one_decimal_null_preserves_decimal_context(connection): assert cursor.fetchmany(1) == [] +@pytest.mark.parametrize("method", ("fetchone", "fetchmany_one")) +@pytest.mark.parametrize("width", (255, 256)) +def test_single_row_wide_text_buffer_boundary(connection, method, width): + text = "\ufeff\U0001f600\x00" + "x" * (width - 4) + with connection.cursor() as cursor: + cursor.execute( + f"SELECT n, CAST(CASE WHEN n = 2 THEN NULL ELSE ? END AS NVARCHAR({width})) AS text " + "FROM (VALUES (1), (2), (3)) AS v(n) ORDER BY n", + text, + ) + rows = [read_one(cursor, method) for _ in range(3)] + assert [tuple(row) for row in rows] == [(1, text), (2, None), (3, text)] + assert type(rows[0][1]) is str + assert rows[0]._values is not rows[2]._values + assert read_one(cursor, method) is None + assert cursor.rowcount == 3 + assert cursor.rownumber == 2 + + @pytest.mark.parametrize("method", ("fetchone", "fetchmany_one")) def test_partial_native_metadata_is_not_a_result_column_count(connection, method): with connection.cursor() as cursor: @@ -316,9 +335,19 @@ def test_fetchmany_one_text_keeps_batch_conversion(connection, sql_type): finally: profiling.disable() counts = {name: stats["calls"] for name, stats in profiling.get_stats().items()} - assert counts.get("ddbc::NativeCursor::fetch", 0) == 0 - assert counts.get("ddbc::FetchSingleRow", 0) == 0 - assert counts["ddbc::FetchMany_wrap"] == 3 - assert counts["ddbc::FetchBatchData"] == 3 + if ddbc_bindings._fetch_observer is not None: + assert counts["ddbc::NativeCursor::fetch"] == 3 + assert counts.get("ddbc::NativeCursor::prepare", 0) == 0 + assert counts.get("ddbc::NativeCursor::fallback", 0) == 0 + if sql_type in ("NCHAR", "NVARCHAR"): + assert counts["ddbc::FetchSingleRow"] == 3 + assert counts.get("ddbc::FetchMany_wrap", 0) == ( + 3 if ddbc_bindings._fetch_observer is None else 0 + ) + assert counts.get("ddbc::FetchBatchData", 0) == 0 + else: + assert counts.get("ddbc::FetchSingleRow", 0) == 0 + assert counts["ddbc::FetchMany_wrap"] == 3 + assert counts["ddbc::FetchBatchData"] == 3 assert cursor.rowcount == 4 assert cursor.rownumber == 3 From 2dabb8d7f455feda68ec4c1cd0f83fc7fd7f0f33 Mon Sep 17 00:00:00 2001 From: Jahnvi Thakkar Date: Fri, 13 Nov 2026 14:48:05 +0530 Subject: [PATCH 2/2] FIX: isolate converter reenable counts from prior cursor GC Collect prior watched cursor cycles before asserting stable native reenable counts. Extend deterministic GC coverage to converter callback removal, preserving exact fetch/fallback counts, metadata ownership, and callback semantics. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- tests/test_native_fetch_dispatch.py | 47 +++++++++++++++++++++++------ 1 file changed, 37 insertions(+), 10 deletions(-) diff --git a/tests/test_native_fetch_dispatch.py b/tests/test_native_fetch_dispatch.py index ca4a2196a..276c056fc 100644 --- a/tests/test_native_fetch_dispatch.py +++ b/tests/test_native_fetch_dispatch.py @@ -243,12 +243,13 @@ class CustomNames(_ColumnNames): @pytest.mark.parametrize("cycle", ("plain", "cursor")) -def test_names_plan_gc_invalidation_completes_once(conn_str, cycle): +@pytest.mark.parametrize("preparation", ("names", "converter")) +def test_plan_gc_invalidation_completes_once(conn_str, cycle, preparation): _redesign_runtime() - _redesign_subprocess("_names_plan_gc_invalidation", cycle) + _redesign_subprocess("_plan_gc_invalidation", cycle, preparation) -def _names_plan_gc_invalidation(cycle): +def _plan_gc_invalidation(cycle, preparation): import os from mssql_python.cursor import _NATIVE_FETCH_RUNTIME @@ -270,8 +271,27 @@ def _names_plan_gc_invalidation(cycle): with _native_fetch_counters() as profiling, connection.cursor() as cursor: warm(cursor, "fetchone") - names = tuple(cursor._cached_result_columns) - cursor._cached_result_columns = names + calls = [] + + def convert(raw): + calls.append(raw) + connection.remove_output_converter(mssql_python.SQL_INTEGER) + return raw * 10 + + if preparation == "converter": + connection.add_output_converter(mssql_python.SQL_INTEGER, convert) + assert cursor.fetchone()[0] == 30 + assert calls == [3] + with _fetchone_window(profiling) as refresh: + assert cursor.fetchone()[0] == 4 + _assert_fetchone_counts(refresh, prepare=0, native=0, fallback=1, sql=1) + assert cursor._cached_converter_map == () + assert cursor._cached_converters_generation == connection._converters_generation + expected = 5 + else: + cursor._cached_result_columns = tuple(cursor._cached_result_columns) + expected = 3 + names = cursor._cached_result_columns collections = [] def collect_during_policy(frame, event, argument): @@ -290,16 +310,20 @@ def collect_during_policy(frame, event, argument): assert collections and collections[0] > 0 if cycle == "cursor": assert reference() is None - assert row[0] == 3 and row._column_names is names - assert cursor.rowcount == 3 and cursor.rownumber == 2 + assert row[0] == expected and row._column_names is names + assert cursor.rowcount == expected and cursor.rownumber == expected - 1 fallback = int(cycle == "cursor") _assert_fetchone_counts( observation, prepare=1, native=1 - fallback, fallback=fallback, sql=1 ) - assert cursor.fetchone()[0] == 4 - assert cursor.fetchone()[0] == 5 + assert observation["python"].get(convert.__code__, 0) == 0 + if preparation == "converter": + assert calls == [3] + for remaining in range(expected + 1, 6): + assert cursor.fetchone()[0] == remaining assert cursor.fetchone() is None - assert row[0] == 3 and row._column_names is names + assert row[0] == expected and row._column_names is names + assert dict(row._mapping) == {"n": expected} @pytest.mark.parametrize("sql_type", ("CHAR", "VARCHAR", "NCHAR", "NVARCHAR")) @@ -752,6 +776,9 @@ def convert(raw): _assert_fetchone_counts(refresh, prepare=0, native=0, fallback=1, sql=1) assert cursor._cached_converter_map == () assert cursor._cached_converters_generation == connection._converters_generation + # Prior watched cursor cycles can invalidate a cold plan during GC. + # Measure stable reenable here; the GC test covers that safe fallback. + gc.collect() with _fetchone_window(profiling, tracked) as reenabled: assert value(read(cursor, method), method) == 5 if tracked: