PERF: Speed up output-converter batching and add CI coverage

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 22 tasks. `--scenarios` runs a local subset, but subset
+The fixed registry has 24 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,24 @@ 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.
 
+`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.
+Both retain all converted rows through the timer boundary; the batched case
+includes the terminal empty `fetchmany(1000)` call and accumulating its batches.
+The timed callback only performs UTF-16LE decode and uppercase conversion.
+Exact output values/types, count/order, and diagnostics are checked after
+collection; focused contract tests check callback bytes using a fake cursor.
+Registration, query execution, expected-value construction, validation, cursor
+cleanup, and restoring/removing the converter are outside the fetch timer.
+Converter state is restored even if setup, fetching, or validation fails.
+
+These CI workloads use the fixture's `unicode_row_{i}` text rather than the
+separate profiling-OFF study's fixed-width text.
+Reports calculate fresh paired changes from profiling-ON measurements;
+they do not embed historical speedups or promise to reproduce production-latency
+results. The two API rows use the existing comparison policy below.
+
 ## Measurement contract
 
 CI uses the PR merge's first parent as the exact base. It reuses the

eng/profiler_benchmarks/report.py

@@ -39,6 +39,8 @@
     "cte": "Common table expression queries",
     "lob_varchar_256k_fetchall": "256 KiB VARCHAR(MAX) / fetchall()",
     "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)",
 }
 CASES = tuple(TASK_NAMES)
 MAX_BYTES = 8 * 1024 * 1024

eng/profiler_benchmarks/workloads.py

@@ -194,6 +194,52 @@ def scalar_fetchval(conn, table, ctx):
             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
+
+    texts = [f"unicode_row_{index}" for index in range(10_000)]
+    expected = [(index, text.upper()) for index, text in enumerate(texts)]
+    api = "fetchmany(1000)" if batched else "fetchall()"
+
+    def converter(raw):
+        return raw.decode("utf-16le").upper()
+
+    with conn.cursor() as cursor:
+        previous = conn.get_output_converter(SQL_WVARCHAR)
+        try:
+            conn.add_output_converter(SQL_WVARCHAR, converter)
+            cursor.execute(f"SELECT TOP (10000) int_col, nvarchar_col FROM {table} ORDER BY id")
+            rows = []
+            ctx.enable()
+            start = time.perf_counter()
+            if batched:
+                while batch := cursor.fetchmany(1000):
+                    rows.extend(batch)
+            else:
+                rows = cursor.fetchall()
+            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 converter fetch unexpectedly produced diagnostics"
+            return dict(
+                title=f"Output converter / {api}",
+                wall_ms=wall_ms,
+                cpp=cpp,
+                py=py,
+                detail=f"Rows: 10000; types: int,nvarchar; converter: UTF-16LE upper; API: {api}",
+            )
+        finally:
+            try:
+                ctx.disable()
+            finally:
+                if previous is None:
+                    conn.remove_output_converter(SQL_WVARCHAR)
+                else:
+                    conn.add_output_converter(SQL_WVARCHAR, previous)
+
+
 def registry():
     """Keep every PR #552 scenario, including its existing timing boundaries."""
     result = dict(scenarios.SCENARIOS)
@@ -208,4 +254,6 @@ def registry():
     result.update((name, (partial(query, sql=sql), False)) for name, sql in QUERIES.items())
     result["lob_varchar_256k_fetchall"] = (lob_fetch, False)
     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)
     return result

mssql_python/cursor.py

@@ -60,6 +60,11 @@
     ddbc_sql_const.SQL_TIME.value: ddbc_sql_const.SQL_TYPE_TIME.value,
     ddbc_sql_const.SQL_TIMESTAMP.value: ddbc_sql_const.SQL_TYPE_TIMESTAMP.value,
 }
+_CONVERTER_ROW_BUILDER = ddbc_bindings._ConverterRowBuilder(Row, ddbc_bindings)
+
+
+def _native_converter_row():
+    return _CONVERTER_ROW_BUILDER.matches(Row)
 
 
 def _string_only_output_converter(converter):
@@ -2844,19 +2849,34 @@ def fetchone(self) -> Union[None, Row]:
             self._refresh_decoding_cache()
         char_enc = self._cached_char_encoding
         wchar_enc = self._cached_wchar_encoding
+        native_converters = self._cached_converter_map
+        deferred_text = False
+        if native_converters and _native_converter_row():
+            deferred_text = True
 
         # Fetch raw data
         row_data = []
         try:
             with perf_phase("py::fetchone::cpp_call"):
-                ret = ddbc_bindings.DDBCSQLFetchOne(
-                    self.hstmt,
-                    row_data,
-                    char_enc,
-                    wchar_enc,
-                    self._cached_char_ctype,
-                    self.messages,
-                )
+                if deferred_text:
+                    ret = ddbc_bindings.DDBCSQLFetchOne(
+                        self.hstmt,
+                        row_data,
+                        char_enc,
+                        wchar_enc,
+                        self._cached_char_ctype,
+                        self.messages,
+                        native_converters,
+                    )
+                else:
+                    ret = ddbc_bindings.DDBCSQLFetchOne(
+                        self.hstmt,
+                        row_data,
+                        char_enc,
+                        wchar_enc,
+                        self._cached_char_ctype,
+                        self.messages,
+                    )
 
             check_error(ddbc_sql_const.SQL_HANDLE_STMT.value, self.hstmt, ret)
 
@@ -2877,7 +2897,21 @@ def fetchone(self) -> Union[None, Row]:
 
             # Get column and converter maps
             column_map, converter_map, column_map_lower = self._get_column_and_converter_maps()
+            if deferred_text and not converter_map:
+                ddbc_bindings.restore_output_converter_text(row_data)
+                deferred_text = False
             with perf_phase("py::fetchone::row_wrap"):
+                if deferred_text:
+                    return _CONVERTER_ROW_BUILDER.one(
+                        row_data,
+                        Row,
+                        converter_map,
+                        column_map,
+                        self,
+                        self._uuid_str_indices,
+                        column_map_lower,
+                        self._cached_result_columns,
+                    )
                 if not converter_map and not self._uuid_str_indices:
                     return Row._fast_create(
                         row_data, column_map, self, column_map_lower, self._cached_result_columns
@@ -2919,20 +2953,36 @@ def fetchmany(self, size: Optional[int] = None) -> List[Row]:
             self._refresh_decoding_cache()
         char_enc = self._cached_char_encoding
         wchar_enc = self._cached_wchar_encoding
+        native_converters = self._cached_converter_map
+        deferred_text = False
+        if native_converters and _native_converter_row():
+            deferred_text = True
 
         # Fetch raw data
         rows_data = []
         try:
             with perf_phase("py::fetchmany::cpp_call"):
-                ret = ddbc_bindings.DDBCSQLFetchMany(
-                    self.hstmt,
-                    rows_data,
-                    size,
-                    char_enc,
-                    wchar_enc,
-                    self._cached_char_ctype,
-                    self.messages,
-                )
+                if deferred_text:
+                    ret = ddbc_bindings.DDBCSQLFetchMany(
+                        self.hstmt,
+                        rows_data,
+                        size,
+                        char_enc,
+                        wchar_enc,
+                        self._cached_char_ctype,
+                        self.messages,
+                        native_converters,
+                    )
+                else:
+                    ret = ddbc_bindings.DDBCSQLFetchMany(
+                        self.hstmt,
+                        rows_data,
+                        size,
+                        char_enc,
+                        wchar_enc,
+                        self._cached_char_ctype,
+                        self.messages,
+                    )
 
             check_error(ddbc_sql_const.SQL_HANDLE_STMT.value, self.hstmt, ret)
 
@@ -2950,10 +3000,23 @@ def fetchmany(self, size: Optional[int] = None) -> List[Row]:
 
             # Get column and converter maps
             column_map, converter_map, column_map_lower = self._get_column_and_converter_maps()
-
+            if deferred_text and not converter_map:
+                for row_data in rows_data:
+                    ddbc_bindings.restore_output_converter_text(row_data)
+                deferred_text = False
             # Convert raw data to Row objects
             uuid_idx = self._uuid_str_indices
             with perf_phase("py::fetchmany::row_wrap"):
+                if deferred_text:
+                    return _CONVERTER_ROW_BUILDER.many(
+                        rows_data,
+                        globals(),
+                        converter_map,
+                        column_map,
+                        self,
+                        uuid_idx,
+                        column_map_lower,
+                    )
                 if not converter_map and not uuid_idx:
                     return ddbc_bindings.construct_rows(
                         rows_data,
@@ -2994,19 +3057,34 @@ def fetchall(self) -> List[Row]:
             self._refresh_decoding_cache()
         char_enc = self._cached_char_encoding
         wchar_enc = self._cached_wchar_encoding
+        native_converters = self._cached_converter_map
+        deferred_text = False
+        if native_converters and _native_converter_row():
+            deferred_text = True
 
         # Fetch raw data
         rows_data = []
         try:
             with perf_phase("py::fetchall::cpp_call"):
-                ret = ddbc_bindings.DDBCSQLFetchAll(
-                    self.hstmt,
-                    rows_data,
-                    char_enc,
-                    wchar_enc,
-                    self._cached_char_ctype,
-                    self.messages,
-                )
+                if deferred_text:
+                    ret = ddbc_bindings.DDBCSQLFetchAll(
+                        self.hstmt,
+                        rows_data,
+                        char_enc,
+                        wchar_enc,
+                        self._cached_char_ctype,
+                        self.messages,
+                        native_converters,
+                    )
+                else:
+                    ret = ddbc_bindings.DDBCSQLFetchAll(
+                        self.hstmt,
+                        rows_data,
+                        char_enc,
+                        wchar_enc,
+                        self._cached_char_ctype,
+                        self.messages,
+                    )
 
             # Check for errors
             check_error(ddbc_sql_const.SQL_HANDLE_STMT.value, self.hstmt, ret)
@@ -3024,10 +3102,23 @@ def fetchall(self) -> List[Row]:
 
             # Get column and converter maps
             column_map, converter_map, column_map_lower = self._get_column_and_converter_maps()
-
+            if deferred_text and not converter_map:
+                for row_data in rows_data:
+                    ddbc_bindings.restore_output_converter_text(row_data)
+                deferred_text = False
             # Convert raw data to Row objects
             uuid_idx = self._uuid_str_indices
             with perf_phase("py::fetchall::row_wrap"):
+                if deferred_text:
+                    return _CONVERTER_ROW_BUILDER.many(
+                        rows_data,
+                        globals(),
+                        converter_map,
+                        column_map,
+                        self,
+                        uuid_idx,
+                        column_map_lower,
+                    )
                 if not converter_map and not uuid_idx:
                     return ddbc_bindings.construct_rows(
                         rows_data,