PERF: Speed up row-to-dictionary conversion 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 23 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,25 @@ 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.
 
+`row_to_dict` measures **warm-only** conversion of 10,000 real cursor-fetched rows
+with 32 unique integer columns and deterministic NULLs using `dict(row._mapping)`.
+It uses the existing `#perf_test` fixture and selects the first 10,000 rows in `id`
+order. All 32 output values are NULL when the source `int_col` is divisible by 7.
+SQL and expected values are constructed, rows are fetched, and every row's
+mapping is converted once before timing on both revisions. Only construction and
+retention of the 10,000 output dictionaries is timed; fetch, cold metadata
+preparation, key/order/value/type/NULL checks, row counts, cursor diagnostics,
+and cleanup are excluded. This is one additional workload, not a cold/warm or
+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 22 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
+pyodbc superiority or broader performance qualification.
+
 ## Measurement contract
 
 CI uses the PR merge's first parent as the exact base. It reuses the

eng/profiler_benchmarks/controller.py

@@ -17,7 +17,7 @@
 import tempfile
 import time
 
-from .report import LEGS
+from .report import LEGS, validate_phases
 from . import workloads
 
 ROOT = Path(__file__).resolve().parents[2]
@@ -170,8 +170,10 @@ def worker(args):
                 result = profiler.run(name)[0]
             if result["cpp"] is None or result["py"] is None:
                 raise RuntimeError(f"Scenario {name} was skipped")
-            if not result["cpp"]:
+            if not result["cpp"] and name != "row_to_dict":
                 raise RuntimeError(f"Scenario {name} has no native samples")
+            for layer in ("cpp", "py"):
+                validate_phases(name, layer, result[layer])
             output[name] = {key: result[key] for key in ("wall_ms", "cpp", "py")}
             output[name]["work"] = result.get("detail", "Connection: 1").split(" (")[0]
             args.output.write_text(

eng/profiler_benchmarks/report.py

@@ -39,6 +39,7 @@
     "cte": "Common table expression queries",
     "lob_varchar_256k_fetchall": "256 KiB VARCHAR(MAX) / fetchall()",
     "scalar_fetchval": "10,000 scalar values / fetchval() (debug disabled)",
+    "row_to_dict": "10,000 rows x 32 columns / dict(row._mapping) (warm)",
 }
 CASES = tuple(TASK_NAMES)
 MAX_BYTES = 8 * 1024 * 1024
@@ -106,6 +107,29 @@ def text(value, limit=160):
     return value
 
 
+def validate_phases(name, layer, stats):
+    """Only warm row-to-dictionary conversion may have no native instrumentation."""
+    if (
+        not isinstance(stats, dict)
+        or len(stats) > 300
+        or (layer == "cpp" and not stats and name != "row_to_dict")
+    ):
+        raise ValueError("Missing or oversized profiling data")
+    for label, counter in stats.items():
+        text(label)
+        if not label.startswith("ddbc::" if layer == "cpp" else "py::"):
+            raise ValueError("Invalid phase prefix")
+        if not isinstance(counter, dict):
+            raise ValueError("Invalid phase counter")
+        calls = counter["calls"]
+        if type(calls) is not int or not 1 <= calls <= 100_000_000:
+            raise ValueError("Invalid call count")
+        for field in ("total_us", "min_us", "max_us"):
+            number(counter[field])
+        if not counter["min_us"] <= counter["max_us"] <= counter["total_us"]:
+            raise ValueError("Inconsistent phase totals")
+
+
 def validate(report, build_id=None, head=None, source=None, base=None):
     try:
         return _validate(report, build_id, head, source, base)
@@ -186,26 +210,7 @@ def _validate(report, build_id=None, head=None, source=None, base=None):
                     raise ValueError(f"Workload changed for {name}")
                 work[name] = identity
                 for layer in ("cpp", "py"):
-                    stats = scenario[layer]
-                    if (
-                        not isinstance(stats, dict)
-                        or len(stats) > 300
-                        or (layer == "cpp" and not stats)
-                    ):
-                        raise ValueError("Missing or oversized profiling data")
-                    for label, counter in stats.items():
-                        text(label)
-                        if not label.startswith("ddbc::" if layer == "cpp" else "py::"):
-                            raise ValueError("Invalid phase prefix")
-                        if not isinstance(counter, dict):
-                            raise ValueError("Invalid phase counter")
-                        calls = counter["calls"]
-                        if type(calls) is not int or not 1 <= calls <= 100_000_000:
-                            raise ValueError("Invalid call count")
-                        for field in ("total_us", "min_us", "max_us"):
-                            number(counter[field])
-                        if not counter["min_us"] <= counter["max_us"] <= counter["total_us"]:
-                            raise ValueError("Inconsistent phase totals")
+                    validate_phases(name, layer, scenario[layer])
     return report
 
 

eng/profiler_benchmarks/workloads.py

@@ -194,6 +194,58 @@ def scalar_fetchval(conn, table, ctx):
             ctx.disable()
 
 
+def row_to_dict(conn, table, ctx):
+    """Convert fetched rows with warm mapping metadata; fetch and validation are not timed."""
+    names = tuple(f"c{column}" for column in range(32))
+    sql = "SELECT TOP (10000) " + ", ".join(
+        f"CASE WHEN int_col % 7 = 0 THEN CAST(NULL AS INT) "
+        f"ELSE int_col + {column} END AS {name}"
+        for column, name in enumerate(names)
+    )
+    sql += f" FROM {table} ORDER BY id"
+    expected = [
+        tuple(None if value % 7 == 0 else value + column for column in range(32))
+        for value in range(10_000)
+    ]
+
+    with conn.cursor() as cursor:
+        try:
+            cursor.execute(sql)
+            rows = cursor.fetchall()
+            assert len(rows) == len(
+                expected
+            ), "Row-to-dictionary fetch returned the wrong row count"
+            for row in rows:
+                dict(row._mapping)
+
+            ctx.enable()
+            start = time.perf_counter()
+            dictionaries = [dict(row._mapping) for row in rows]
+            wall_ms = (time.perf_counter() - start) * 1000
+            cpp, py = ctx.collect()
+            assert len(dictionaries) == len(expected)
+            for dictionary, values in zip(dictionaries, expected):
+                assert tuple(dictionary) == names
+                assert all(type(key) is str for key in dictionary)
+                actual = tuple(dictionary.values())
+                assert actual == values and all(
+                    type(value) is type(wanted) for value, wanted in zip(actual, values)
+                )
+            assert not cursor.messages, "Clean row-to-dictionary conversion produced diagnostics"
+            return dict(
+                title="Warm row-to-dictionary conversion",
+                wall_ms=wall_ms,
+                cpp=cpp,
+                py=py,
+                detail=(
+                    "Rows: 10000; columns: 32; types: int/NULL; "
+                    "API: dict(row._mapping); mode: warm"
+                ),
+            )
+        finally:
+            ctx.disable()
+
+
 def registry():
     """Keep every PR #552 scenario, including its existing timing boundaries."""
     result = dict(scenarios.SCENARIOS)
@@ -208,4 +260,5 @@ 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["row_to_dict"] = (row_to_dict, True)
     return result

mssql_python/async_query/async_cursor.py

@@ -15,7 +15,7 @@
 from ..exceptions import OperationalError
 from ..helpers import get_settings
 from ..logging import logger
-from ..row import Row
+from ..row import Row, _ColumnNames
 from . import async_execute, async_fetch
 from .exception_translator import translate_py_core_exceptions
 
@@ -64,7 +64,7 @@ def _initialize_result_metadata(self) -> None:
             ((column[0].lower() if settings.lowercase else column[0]), *column[1:])
             for column in description
         ]
-        self._column_names = tuple(column[0] for column in self._description)
+        self._column_names = _ColumnNames(column[0] for column in self._description)
         self._column_map = {column[0]: index for index, column in enumerate(self._description)}
         self._column_map_lower = (
             {name.lower(): index for name, index in self._column_map.items()}

mssql_python/cursor.py

@@ -28,7 +28,7 @@
     OperationalError,
     DatabaseError,
 )
-from mssql_python.row import Row
+from mssql_python.row import Row, _ColumnNames
 from mssql_python.perf_timer import perf_phase
 from mssql_python import get_settings
 from mssql_python.parameter_helper import (
@@ -1513,7 +1513,7 @@ def _get_column_and_converter_maps(self):
         # Snapshot canonical column names once per result set (identity-tracked against
         # self.description) so each Row carries stable names even after the cursor is reused.
         if self.description is not None and self._result_columns_src is not self.description:
-            self._cached_result_columns = tuple(col_desc[0] for col_desc in self.description)
+            self._cached_result_columns = _ColumnNames(col_desc[0] for col_desc in self.description)
             self._result_columns_src = self.description
 
         return column_map, converter_map, self._cached_column_map_lower

mssql_python/pybind/ddbc_bindings.cpp

@@ -14,6 +14,7 @@
 #include "py_type_cache.hpp"
 #include "fetch_temporal.hpp"
 #include "row_factory.hpp"
+#include "row_mapping.hpp"
 #include "utf_utils.h"
 #include "fetch_text.hpp"
 
@@ -6625,6 +6626,8 @@ PYBIND11_MODULE(ddbc_bindings, m) {
     // Add a version attribute
     m.attr("__version__") = "1.0.0";
 
+    m.def("_make_row_mapping_base", &RowMappingView::make_base);
+
     // 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",