diff --git a/eng/profiler_benchmarks/README.md b/eng/profiler_benchmarks/README.md index 2fdeb4d91..df9901202 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 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 diff --git a/eng/profiler_benchmarks/controller.py b/eng/profiler_benchmarks/controller.py index 143164e21..23ef8b84d 100644 --- a/eng/profiler_benchmarks/controller.py +++ b/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( diff --git a/eng/profiler_benchmarks/report.py b/eng/profiler_benchmarks/report.py index 07f176f18..6465385d0 100644 --- a/eng/profiler_benchmarks/report.py +++ b/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 diff --git a/eng/profiler_benchmarks/workloads.py b/eng/profiler_benchmarks/workloads.py index 9059ab6c9..c09867c9d 100644 --- a/eng/profiler_benchmarks/workloads.py +++ b/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 diff --git a/mssql_python/async_query/async_cursor.py b/mssql_python/async_query/async_cursor.py index 2547e2fae..5b4f3487f 100644 --- a/mssql_python/async_query/async_cursor.py +++ b/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()} diff --git a/mssql_python/cursor.py b/mssql_python/cursor.py index d4b85e56b..7dc47b2b5 100644 --- a/mssql_python/cursor.py +++ b/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 diff --git a/mssql_python/pybind/ddbc_bindings.cpp b/mssql_python/pybind/ddbc_bindings.cpp index 4e4523955..0d43b7fac 100644 --- a/mssql_python/pybind/ddbc_bindings.cpp +++ b/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", diff --git a/mssql_python/pybind/row_mapping.hpp b/mssql_python/pybind/row_mapping.hpp new file mode 100644 index 000000000..e767d2f5e --- /dev/null +++ b/mssql_python/pybind/row_mapping.hpp @@ -0,0 +1,882 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT license. + +#pragma once + +#include "py_ref.hpp" +#include "row_mapping_cache.hpp" +#include + +namespace RowMappingView { + +enum Defaults { + ROW_TYPE, + KEY_HOOK, + NAMES_TYPE, + ROW_VERSION, + ITERATOR_TYPE, + FALLBACK_ITERATOR, + ATTR_ROW, + ATTR_KEYS, + ATTR_NAMES, + ATTR_UNIQUE, + ATTR_BUILD_UNIQUE, + ATTR_COLUMNS, + ATTR_VALUES, + ROW_MEMBERS, + WATCHER_OWNER, + SCHEMA_TYPE, + NAMES_VERSION, + ATTR_PREPARED, + KEY_CODE, + GETITEM_HOOK, + GETITEM_CODE, + DEFAULTS_SIZE +}; + +enum RowField { NAMES, COLUMNS, VALUES, ROW_FIELD_COUNT }; + +struct AccessConfiguration { + unsigned int row_version; + unsigned int names_version; + Py_ssize_t row_offsets[ROW_FIELD_COUNT]; + RowMappingCache::Watcher* watcher; +}; + +struct Configuration { + PyObject_HEAD + PyObject* defaults; + AccessConfiguration access; +}; + +struct View { + PyObject_HEAD + PyObject* row; + PyObject* defaults; + PyObject* configuration; + unsigned int row_version; + unsigned int view_version; + Py_ssize_t row_offsets[ROW_FIELD_COUNT]; + PyObject* prepared_row; + PyObject* prepared_names; + PyObject* row_dictionary; + PyObject* names_dictionary; + PyObject* prepared_schema; + PyObject** row_dict_pointer; + PyObject** names_dict_pointer; + RowMappingCache::Watcher* watcher; + unsigned int names_version; + std::uint64_t prepared_epoch; + Py_ssize_t next_entry; + Py_ssize_t preparations; + Py_ssize_t schema_builds; + Py_ssize_t fast_lookups; + Py_ssize_t slow_lookups; + Py_ssize_t ordered_lookups; + bool preparing; +}; + +struct Iterator { + PyObject_HEAD + View* view; + PyObject* iterator; + bool running; +}; + +inline PyObject* iterator_next(Iterator* self); + +inline int traverse(View* self, visitproc visit, void* arg) { + Py_VISIT(Py_TYPE(self)); + Py_VISIT(self->row); + Py_VISIT(self->defaults); + Py_VISIT(self->configuration); + Py_VISIT(self->prepared_row); + Py_VISIT(self->prepared_names); + Py_VISIT(self->row_dictionary); + Py_VISIT(self->names_dictionary); + Py_VISIT(self->prepared_schema); + return 0; +} + +inline void invalidate_prepared(View* self) { + self->prepared_epoch = 0; + self->next_entry = 0; + self->row_dict_pointer = nullptr; + self->names_dict_pointer = nullptr; +} + +inline void clear_prepared(View* self) { + invalidate_prepared(self); + // Detach the entire state before any decref can reenter the view. + py::object row = steal(std::exchange(self->prepared_row, nullptr)); + py::object names = steal(std::exchange(self->prepared_names, nullptr)); + py::object row_dictionary = steal(std::exchange(self->row_dictionary, nullptr)); + py::object names_dictionary = steal(std::exchange(self->names_dictionary, nullptr)); + py::object schema = steal(std::exchange(self->prepared_schema, nullptr)); +} + +inline int clear(View* self) { + invalidate_prepared(self); + self->watcher = nullptr; + py::object configuration = steal(std::exchange(self->configuration, nullptr)); + py::object defaults = steal(std::exchange(self->defaults, nullptr)); + py::object row = steal(std::exchange(self->row, nullptr)); + clear_prepared(self); + return 0; +} + +inline void dealloc(View* self) { + PyObject_GC_UnTrack(self); + clear(self); + PyTypeObject* type = Py_TYPE(self); + type->tp_free(reinterpret_cast(self)); + Py_DECREF(type); +} + +inline Py_ssize_t member_offset(PyTypeObject* type, PyObject* descriptor) { + if (!descriptor || !Py_IS_TYPE(descriptor, &PyMemberDescr_Type) || + !PyType_IsSubtype(type, PyDescr_TYPE(descriptor))) + return -1; + PyMemberDef* member = reinterpret_cast(descriptor)->d_member; + if (member->type != T_OBJECT_EX || member->offset < static_cast(sizeof(PyObject)) || + member->offset > type->tp_basicsize - static_cast(sizeof(PyObject*)) || + member->offset % static_cast(alignof(PyObject*)) != 0) + return -1; + return member->offset; +} + +inline bool valid_iterator_type(PyObject* defaults) { + auto* type = reinterpret_cast(PyTuple_GET_ITEM(defaults, ITERATOR_TYPE)); + if (type->tp_basicsize != static_cast(sizeof(Iterator)) || + type->tp_iternext != reinterpret_cast(iterator_next)) { + PyErr_SetString(PyExc_TypeError, "Invalid RowMapping iterator type"); + return false; + } + return true; +} + +inline bool read_configuration(PyObject* defaults, AccessConfiguration& access) { + if (!PyTuple_CheckExact(defaults) || + PyTuple_GET_SIZE(defaults) != DEFAULTS_SIZE || + !PyType_Check(PyTuple_GET_ITEM(defaults, ROW_TYPE)) || + !PyType_Check(PyTuple_GET_ITEM(defaults, ITERATOR_TYPE))) { + PyErr_SetString(PyExc_TypeError, "Invalid RowMapping configuration"); + return false; + } + if (!valid_iterator_type(defaults)) + return false; + unsigned long version = PyLong_AsUnsignedLong(PyTuple_GET_ITEM(defaults, ROW_VERSION)); + if (version == static_cast(-1) && PyErr_Occurred()) + return false; + unsigned long names_version = PyLong_AsUnsignedLong( + PyTuple_GET_ITEM(defaults, NAMES_VERSION)); + if (names_version == static_cast(-1) && PyErr_Occurred()) + return false; + RowMappingCache::Watcher* watcher = nullptr; + PyObject* owner = PyTuple_GET_ITEM(defaults, WATCHER_OWNER); + if (owner != Py_None) { + watcher = static_cast( + PyCapsule_GetPointer(owner, "mssql.RowMappingWatcher")); + if (!watcher) + return false; + } + PyObject* members = PyTuple_GET_ITEM(defaults, ROW_MEMBERS); + if (!PyTuple_CheckExact(members) || PyTuple_GET_SIZE(members) != ROW_FIELD_COUNT) { + PyErr_SetString(PyExc_TypeError, "Invalid RowMapping member configuration"); + return false; + } + access.row_version = static_cast(version); + access.names_version = static_cast(names_version); + access.watcher = watcher; + auto* row_type = reinterpret_cast(PyTuple_GET_ITEM(defaults, ROW_TYPE)); + for (int field = 0; field < ROW_FIELD_COUNT; ++field) + access.row_offsets[field] = member_offset(row_type, PyTuple_GET_ITEM(members, field)); + return true; +} + +inline int configuration_traverse(Configuration* self, visitproc visit, void* arg) { + Py_VISIT(Py_TYPE(self)); + Py_VISIT(self->defaults); + return 0; +} + +inline int configuration_clear(Configuration* self) { + self->access = {}; + Py_CLEAR(self->defaults); + return 0; +} + +inline void configuration_dealloc(Configuration* self) { + PyObject_GC_UnTrack(self); + configuration_clear(self); + PyTypeObject* type = Py_TYPE(self); + type->tp_free(reinterpret_cast(self)); + Py_DECREF(type); +} + +inline bool is_configuration(PyObject* object) { + PyTypeObject* type = Py_TYPE(object); + return type->tp_basicsize == static_cast(sizeof(Configuration)) && + type->tp_dealloc == reinterpret_cast(configuration_dealloc) && + type->tp_traverse == reinterpret_cast(configuration_traverse); +} + +inline py::object make_configuration(const py::tuple& defaults) { + static PyType_Slot slots[] = { + {Py_tp_dealloc, reinterpret_cast(configuration_dealloc)}, + {Py_tp_traverse, reinterpret_cast(configuration_traverse)}, + {Py_tp_clear, reinterpret_cast(configuration_clear)}, + {0, nullptr}, + }; + static PyType_Spec spec = { + "mssql_python.row._MappingConfiguration", sizeof(Configuration), 0, + Py_TPFLAGS_DEFAULT | Py_TPFLAGS_HAVE_GC | Py_TPFLAGS_IMMUTABLETYPE | + Py_TPFLAGS_DISALLOW_INSTANTIATION, slots, + }; + py::object type = steal(PyType_FromSpec(&spec)); + if (!type) + throw py::error_already_set(); + auto* native_type = reinterpret_cast(type.ptr()); + py::object owner = steal(native_type->tp_alloc(native_type, 0)); + if (!owner) + throw py::error_already_set(); + auto* configuration = reinterpret_cast(owner.ptr()); + configuration->defaults = Py_NewRef(defaults.ptr()); + if (!read_configuration(defaults.ptr(), configuration->access)) + throw py::error_already_set(); + return owner; +} + +inline PyObject* create(PyTypeObject* type, PyObject*, PyObject*) { + py::object configuration = steal(PyObject_GetAttrString( + reinterpret_cast(type), "_row_mapping_defaults")); + if (!configuration) + return nullptr; + py::object defaults; + AccessConfiguration access{}; + if (is_configuration(configuration.ptr())) { + auto* native = reinterpret_cast(configuration.ptr()); + if (!native->defaults) { + PyErr_SetString(PyExc_TypeError, "Invalid RowMapping configuration"); + return nullptr; + } + defaults = borrow(native->defaults); + access = native->access; + } else { + defaults = std::move(configuration); + if (!read_configuration(defaults.ptr(), access)) + return nullptr; + } + // The configuration owns the type, but its live protocol can still change. + if (!valid_iterator_type(defaults.ptr())) + return nullptr; + auto* self = reinterpret_cast(type->tp_alloc(type, 0)); + if (!self) + return nullptr; + self->defaults = defaults.release().ptr(); + self->configuration = configuration.release().ptr(); + self->row_version = access.row_version; + self->names_version = access.names_version; + self->watcher = access.watcher; + self->view_version = 0; + for (int field = 0; field < ROW_FIELD_COUNT; ++field) + self->row_offsets[field] = access.row_offsets[field]; + PyObject* row_member = _PyType_Lookup(type, PyTuple_GET_ITEM(self->defaults, ATTR_ROW)); + if (type->tp_getattro == PyObject_GenericGetAttr && + member_offset(type, row_member) == static_cast(offsetof(View, row))) + self->view_version = type->tp_version_tag; + return reinterpret_cast(self); +} + +inline bool unchanged_row(View* self, PyObject* row) { + if (reinterpret_cast(Py_TYPE(row)) != PyTuple_GET_ITEM(self->defaults, ROW_TYPE) || + self->row_version == 0 || Py_TYPE(row)->tp_version_tag != self->row_version) + return false; + PyObject* hook = PyTuple_GET_ITEM(self->defaults, KEY_HOOK); + PyObject* getitem = PyTuple_GET_ITEM(self->defaults, GETITEM_HOOK); + return (!PyFunction_Check(hook) || + PyFunction_GET_CODE(hook) == PyTuple_GET_ITEM(self->defaults, KEY_CODE)) && + (!PyFunction_Check(getitem) || + PyFunction_GET_CODE(getitem) == PyTuple_GET_ITEM(self->defaults, GETITEM_CODE)); +} + +inline PyObject* stored_field(View* self, PyObject* row, RowField field) { + return *reinterpret_cast(reinterpret_cast(row) + self->row_offsets[field]); +} + +inline py::object observed_row(View* self) { +#ifndef Py_GIL_DISABLED + if (self->view_version != 0 && Py_TYPE(self)->tp_version_tag == self->view_version && self->row) + return borrow(self->row); +#endif + return steal(PyObject_GetAttr( + reinterpret_cast(self), PyTuple_GET_ITEM(self->defaults, ATTR_ROW))); +} + +inline py::object row_field(View* self, PyObject* row, RowField field, Defaults attribute) { +#ifndef Py_GIL_DISABLED + if (unchanged_row(self, row) && self->row_offsets[field] >= 0) { + PyObject* value = stored_field(self, row, field); + if (value) + return borrow(value); + } +#endif + // Deleted fields still need normal descriptor/__getattr__ behavior. + return steal(PyObject_GetAttr(row, PyTuple_GET_ITEM(self->defaults, attribute))); +} + +inline PyObject* missing_key(PyObject* key) { + py::object args = steal(PyTuple_Pack(1, key)); + if (args) + PyErr_SetObject(PyExc_KeyError, args.ptr()); + return nullptr; +} + +inline py::object get_names(View* self, PyObject* row, bool& plain_row) { + PyObject* default_hook = PyTuple_GET_ITEM(self->defaults, KEY_HOOK); + plain_row = unchanged_row(self, row); + py::object hook = steal(PyObject_GetAttr(row, PyTuple_GET_ITEM(self->defaults, ATTR_KEYS))); + if (!hook) + return {}; + + // Resolve the live hook without materializing the Row's instance dictionary. + // A class edit invalidates the type version; instance overrides are checked here. + plain_row = plain_row && PyMethod_Check(hook.ptr()) && + PyMethod_GET_FUNCTION(hook.ptr()) == default_hook && + PyMethod_GET_SELF(hook.ptr()) == row; + py::object names; + if (plain_row) { + names = row_field(self, row, NAMES, ATTR_NAMES); + if (!names) + return {}; + } + if (!plain_row || names.is_none()) { + names = steal(PyObject_CallNoArgs(hook.ptr())); + } + return names; +} + +inline py::object get_unique(View* self, PyObject* names) { + py::object unique = steal(PyObject_GetAttr(names, PyTuple_GET_ITEM(self->defaults, ATTR_UNIQUE))); + if (unique && unique.is_none()) + unique = steal(PyObject_CallMethodNoArgs( + names, PyTuple_GET_ITEM(self->defaults, ATTR_BUILD_UNIQUE))); + return unique; +} + +inline int prepare(View* self, PyObject* row, PyObject* names, PyObject* unique) { +#if MSSQL_MAPPING_WATCHERS + using namespace RowMappingCache; + if (self->preparing) + return 0; + struct Preparing { + bool& flag; + explicit Preparing(bool& flag) : flag(flag) { flag = true; } + ~Preparing() { flag = false; } + } preparing(self->preparing); + clear_prepared(self); + if (!self->watcher || !self->watcher->enabled || !unchanged_row(self, row) || + self->view_version == 0 || Py_TYPE(self)->tp_version_tag != self->view_version || + self->row != row || self->names_version == 0 || + reinterpret_cast(Py_TYPE(names)) != PyTuple_GET_ITEM(self->defaults, NAMES_TYPE) || + Py_TYPE(names)->tp_version_tag != self->names_version) + return 0; + if ((!Py_TYPE(row)->tp_dictoffset && !(Py_TYPE(row)->tp_flags & Py_TPFLAGS_MANAGED_DICT)) || + (!Py_TYPE(names)->tp_dictoffset && !(Py_TYPE(names)->tp_flags & Py_TPFLAGS_MANAGED_DICT))) + return 0; + for (int field = 0; field < ROW_FIELD_COUNT; ++field) + if (self->row_offsets[field] < 0) + return 0; + PyObject* values = stored_field(self, row, VALUES); + PyObject* columns = stored_field(self, row, COLUMNS); + if (!values || (!PyList_CheckExact(values) && !PyTuple_CheckExact(values)) || + !columns || !PyDict_CheckExact(columns) || !PyDict_CheckExact(unique) || + stored_field(self, row, NAMES) != names) + return 0; + py::object row_owner = borrow(row); + py::object names_owner = borrow(names); + py::object column_owner = borrow(columns); + // Materialize only the ordinary instance dictionary, never values/getset storage. + // Its stable pointer also detects wholesale __dict__ replacement without a watcher event. + py::object row_dictionary = steal(PyObject_GenericGetDict(row, nullptr)); + if (!row_dictionary) + return -1; + py::object names_dictionary = steal(PyObject_GenericGetDict(names, nullptr)); + if (!names_dictionary) + return -1; + if (!builtin_keys(row_dictionary.ptr()) || !builtin_keys(names_dictionary.ptr())) + return 0; + PyObject** row_pointer = _PyObject_GetDictPtr(row); + PyObject** names_pointer = _PyObject_GetDictPtr(names); + if (!row_pointer || !names_pointer) { + if (PyErr_Occurred()) + return -1; + return 0; + } + if (PyDict_Watch(self->watcher->id, columns) < 0 || + PyDict_Watch(self->watcher->id, unique) < 0 || + PyDict_Watch(self->watcher->id, row_dictionary.ptr()) < 0 || + PyDict_Watch(self->watcher->id, names_dictionary.ptr()) < 0) + return -1; + if (!builtin_keys(names_dictionary.ptr())) + return 0; + PyObject* cached = PyDict_GetItemWithError( + names_dictionary.ptr(), PyTuple_GET_ITEM(self->defaults, ATTR_PREPARED)); + if (!cached && PyErr_Occurred()) + return -1; + auto* schema_class = reinterpret_cast( + PyTuple_GET_ITEM(self->defaults, SCHEMA_TYPE)); + py::object schema_owner; + Schema* schema = nullptr; + if (cached && Py_IS_TYPE(cached, schema_class)) { + schema_owner = borrow(cached); + schema = reinterpret_cast(cached); + } else if (cached && cached != Py_None) { + return 0; + } + if (!schema || schema->epoch != self->watcher->epoch || + schema->columns != columns || schema->unique != unique) { + py::object indices = steal(PyDict_New()); + if (!indices) + return -1; + schema_owner = steal(schema_class->tp_alloc(schema_class, 0)); + if (!schema_owner) + return -1; + schema = reinterpret_cast(schema_owner.ptr()); + std::uint64_t epoch = self->watcher->epoch; + if (!builtin_keys(columns, true) || !builtin_keys(unique)) + return 0; + if (allocate_entries(schema, PyDict_Size(unique)) < 0) + return -1; + Py_ssize_t position = 0; + PyObject *key, *unused; + while (PyDict_Next(unique, &position, &key, &unused)) { + PyObject* index = PyDict_GetItemWithError(columns, key); + if (!index) + return PyErr_Occurred() ? -1 : 0; + if (PyDict_SetItem(indices.ptr(), key, index) < 0) + return -1; + if (append_entry(schema, key, index) < 0) + return -1; + } + schema->columns = Py_NewRef(columns); + schema->unique = Py_NewRef(unique); + schema->indices = indices.release().ptr(); + schema->watcher_owner = Py_NewRef(PyTuple_GET_ITEM(self->defaults, WATCHER_OWNER)); + if (PyDict_Watch(self->watcher->id, schema->indices) < 0) + return -1; + if (self->watcher->epoch != epoch) + return 0; + if (!builtin_keys(names_dictionary.ptr())) + return 0; + // This single watched write advances the epoch. Any additional mutation deopts. + if (PyDict_SetItem(names_dictionary.ptr(), + PyTuple_GET_ITEM(self->defaults, ATTR_PREPARED), schema_owner.ptr()) < 0) + return -1; + if (!self->watcher->enabled || self->watcher->epoch != epoch + 1) + return 0; + schema->epoch = self->watcher->epoch; + ++self->schema_builds; + } + // Allocations above may run GC/finalizers: seal only after rechecking all dependencies. + if (!self->watcher->enabled || schema->epoch != self->watcher->epoch || + !unchanged_row(self, row) || self->row != row || + Py_TYPE(self)->tp_version_tag != self->view_version || + Py_TYPE(names)->tp_version_tag != self->names_version || + *row_pointer != row_dictionary.ptr() || *names_pointer != names_dictionary.ptr() || + stored_field(self, row, NAMES) != names || stored_field(self, row, COLUMNS) != columns) + return 0; + if (!builtin_keys(row_dictionary.ptr()) || !builtin_keys(names_dictionary.ptr())) + return 0; + if (PyDict_GetItemWithError(row_dictionary.ptr(), PyTuple_GET_ITEM(self->defaults, ATTR_KEYS))) + return 0; + if (PyErr_Occurred()) + return -1; + if (PyDict_GetItemWithError(names_dictionary.ptr(), + PyTuple_GET_ITEM(self->defaults, ATTR_UNIQUE)) != unique) + return PyErr_Occurred() ? -1 : 0; + self->row_dict_pointer = row_pointer; + self->names_dict_pointer = names_pointer; + self->prepared_row = row_owner.release().ptr(); + self->prepared_names = names_owner.release().ptr(); + self->row_dictionary = row_dictionary.release().ptr(); + self->names_dictionary = names_dictionary.release().ptr(); + self->prepared_schema = schema_owner.release().ptr(); + self->next_entry = 0; + // Publish validity last; no Python code runs during these ownership transfers. + self->prepared_epoch = schema->epoch; + ++self->preparations; +#else + (void)self; + (void)row; + (void)names; + (void)unique; +#endif + return 0; +} + +inline bool prepared_subscript(View* self, PyObject* key, PyObject** result) { +#if MSSQL_MAPPING_WATCHERS + if (self->prepared_epoch == 0 || !PyUnicode_CheckExact(key)) + return false; + if (!self->watcher->enabled || + self->watcher->epoch != self->prepared_epoch || + Py_TYPE(self)->tp_version_tag != self->view_version) { + invalidate_prepared(self); + return false; + } + PyObject* row = self->prepared_row; + PyObject* names = self->prepared_names; + auto* schema = reinterpret_cast(self->prepared_schema); + if (self->row != row || !unchanged_row(self, row) || + Py_TYPE(names)->tp_version_tag != self->names_version || + *self->row_dict_pointer != self->row_dictionary || + *self->names_dict_pointer != self->names_dictionary || + stored_field(self, row, NAMES) != names || + stored_field(self, row, COLUMNS) != schema->columns) { + invalidate_prepared(self); + return false; + } + PyObject* values = stored_field(self, row, VALUES); + if (!values || (!PyList_CheckExact(values) && !PyTuple_CheckExact(values))) { + invalidate_prepared(self); + return false; + } + ++self->fast_lookups; + if (self->next_entry < schema->entry_count) { + const RowMappingCache::Entry& entry = schema->entries[self->next_entry]; + if (key == entry.key && entry.representable) { + bool list = PyList_CheckExact(values); + Py_ssize_t size = list ? PyList_GET_SIZE(values) : PyTuple_GET_SIZE(values); + Py_ssize_t offset = entry.offset; + if (offset < 0) + offset += size; + if (offset >= 0 && offset < size) { + *result = Py_NewRef(list ? PyList_GET_ITEM(values, offset) + : PyTuple_GET_ITEM(values, offset)); + ++self->next_entry; + ++self->ordered_lookups; + return true; + } + } + } + PyObject* index = PyDict_GetItemWithError(schema->indices, key); + if (!index) { + *result = PyErr_Occurred() ? nullptr : missing_key(key); + return true; + } + // Read the current values, not a prepared row snapshot; builtin indexing keeps errors intact. + Py_INCREF(values); + Py_INCREF(index); + *result = PyObject_GetItem(values, index); + Py_DECREF(index); + Py_DECREF(values); + return true; +#else + (void)self; + (void)key; + (void)result; + return false; +#endif +} + +inline PyObject* subscript(View* self, PyObject* key) { + if (!self->defaults) { + PyErr_SetString(PyExc_AttributeError, "RowMapping has no row"); + return nullptr; + } + PyObject* prepared_result; + if (prepared_subscript(self, key, &prepared_result)) + return prepared_result; + ++self->slow_lookups; + if (!PyUnicode_Check(key)) { + int is_string = PyObject_IsInstance(key, reinterpret_cast(&PyUnicode_Type)); + if (is_string < 0) + return nullptr; + if (!is_string) + return missing_key(key); + } + py::object row = observed_row(self); + if (!row) + return nullptr; + bool plain_row; + py::object names = get_names(self, row.ptr(), plain_row); + if (!names) + return nullptr; + PyObject* names_type = PyTuple_GET_ITEM(self->defaults, NAMES_TYPE); + int contains; + if (PyUnicode_CheckExact(key) && + reinterpret_cast(Py_TYPE(names.ptr())) == names_type) { + py::object unique = get_unique(self, names.ptr()); + if (!unique) + return nullptr; + contains = PySequence_Contains(unique.ptr() == Py_False ? names.ptr() : unique.ptr(), key); + } else { + // str subclasses may be unhashable or customize equality: retain tuple lookup. + contains = PySequence_Contains(names.ptr(), key); + } + if (contains < 0) + return nullptr; + if (!contains) + return missing_key(key); + + // A custom key hook can replace the view's row during the membership check. + py::object current_row = observed_row(self); + if (!current_row) + return nullptr; + plain_row = plain_row && current_row.ptr() == row.ptr(); + row = std::move(current_row); + if (plain_row && PyUnicode_CheckExact(key) && + Py_TYPE(row.ptr())->tp_version_tag == self->row_version) { + py::object columns = row_field(self, row.ptr(), COLUMNS, ATTR_COLUMNS); + if (!columns) + return nullptr; + if (PyDict_CheckExact(columns.ptr())) { + int found = PyDict_Contains(columns.ptr(), key); + if (found < 0) + return nullptr; + columns = py::object(); + if (found) { + py::object values = row_field(self, row.ptr(), VALUES, ATTR_VALUES); + if (!values) + return nullptr; + // Equality on a custom dict key can replace the row's column map. + columns = row_field(self, row.ptr(), COLUMNS, ATTR_COLUMNS); + if (!columns) + return nullptr; + py::object index = steal(PyObject_GetItem(columns.ptr(), key)); + if (!index) + return nullptr; + return PyObject_GetItem(values.ptr(), index.ptr()); + } + + py::object lower = steal(PyObject_GetAttrString(row.ptr(), "_column_map_lower")); + if (!lower) + return nullptr; + if (!lower.is_none()) { + lower = steal(PyObject_GetAttrString(row.ptr(), "_column_map_lower")); + if (!lower) + return nullptr; + py::object get = steal(PyObject_GetAttrString(lower.ptr(), "get")); + if (!get) + return nullptr; + py::object lowered = steal(PyObject_CallMethod(key, "lower", nullptr)); + if (!lowered) + return nullptr; + py::object index = steal(PyObject_CallOneArg(get.ptr(), lowered.ptr())); + if (!index) + return nullptr; + if (!index.is_none()) { + py::object values = row_field(self, row.ptr(), VALUES, ATTR_VALUES); + if (!values) + return nullptr; + return PyObject_GetItem(values.ptr(), index.ptr()); + } + } + py::object message = steal(PyUnicode_FromFormat("Row has no column '%U'", key)); + if (message) + PyErr_SetObject(PyExc_KeyError, message.ptr()); + return nullptr; + } + } + return PyObject_GetItem(row.ptr(), key); +} + +inline int iterator_traverse(Iterator* self, visitproc visit, void* arg) { + Py_VISIT(Py_TYPE(self)); + Py_VISIT(self->view); + Py_VISIT(self->iterator); + return 0; +} + +inline int iterator_clear(Iterator* self) { + Py_CLEAR(self->view); + Py_CLEAR(self->iterator); + return 0; +} + +inline void iterator_dealloc(Iterator* self) { + PyObject_GC_UnTrack(self); + iterator_clear(self); + PyTypeObject* type = Py_TYPE(self); + type->tp_free(reinterpret_cast(self)); + Py_DECREF(type); +} + +inline PyObject* iterator_next(Iterator* self) { + if (!self->view) + return nullptr; + if (self->running) { + PyErr_SetString(PyExc_ValueError, "generator already executing"); + return nullptr; + } + struct Running { + bool& flag; + explicit Running(bool& flag) : flag(flag) { flag = true; } + ~Running() { flag = false; } + } running(self->running); + if (!self->iterator) { + py::object owner = borrow(reinterpret_cast(self->view)); + View* view = self->view; + py::object row = observed_row(view); + if (!row) { + iterator_clear(self); + return nullptr; + } + bool plain_row; + py::object names = get_names(view, row.ptr(), plain_row); + if (!names) { + iterator_clear(self); + return nullptr; + } + py::object unique; + if (reinterpret_cast(Py_TYPE(names.ptr())) == + PyTuple_GET_ITEM(view->defaults, NAMES_TYPE)) { + unique = get_unique(view, names.ptr()); + if (!unique) { + iterator_clear(self); + return nullptr; + } + } + if (unique && unique.ptr() != Py_False) { + if (plain_row && prepare(view, row.ptr(), names.ptr(), unique.ptr()) < 0) { + iterator_clear(self); + return nullptr; + } + self->iterator = PyObject_GetIter(unique.ptr()); + } else { + self->iterator = PyObject_CallOneArg( + PyTuple_GET_ITEM(view->defaults, FALLBACK_ITERATOR), names.ptr()); + } + if (!self->iterator) { + iterator_clear(self); + return nullptr; + } + if (!PyIter_Check(self->iterator)) { + PyErr_SetString(PyExc_TypeError, "Mapping key hook did not return an iterator"); + iterator_clear(self); + return nullptr; + } + } + PyObject* value = PyIter_Next(self->iterator); + if (!value) + iterator_clear(self); + return value; +} + +inline PyObject* iterate(View* self) { + if (!self->defaults) { + PyErr_SetString(PyExc_AttributeError, "RowMapping has no row"); + return nullptr; + } + auto* type = reinterpret_cast(PyTuple_GET_ITEM(self->defaults, ITERATOR_TYPE)); + auto* iterator = reinterpret_cast(type->tp_alloc(type, 0)); + if (!iterator) + return nullptr; + Py_INCREF(self); + iterator->view = self; + return reinterpret_cast(iterator); +} + +inline PyObject* statistics(View* self, void*) { + return Py_BuildValue("(nnnnn)", self->preparations, self->schema_builds, + self->fast_lookups, self->slow_lookups, self->ordered_lookups); +} + +inline py::object make_base(const py::object& row_type, const py::object& names_type, + const py::object& fallback_iterator) { + if (!PyType_Check(row_type.ptr()) || !PyType_Check(names_type.ptr())) + throw py::type_error("Row and column names must be types"); + + static PyType_Slot iterator_slots[] = { + {Py_tp_dealloc, reinterpret_cast(iterator_dealloc)}, + {Py_tp_traverse, reinterpret_cast(iterator_traverse)}, + {Py_tp_clear, reinterpret_cast(iterator_clear)}, + {Py_tp_iter, reinterpret_cast(PyObject_SelfIter)}, + {Py_tp_iternext, reinterpret_cast(iterator_next)}, + {0, nullptr}, + }; + static PyType_Spec iterator_spec = { + "mssql_python.row._RowMappingIterator", sizeof(Iterator), 0, + Py_TPFLAGS_DEFAULT | Py_TPFLAGS_HAVE_GC, iterator_slots, + }; + py::object iterator_type = steal(PyType_FromSpec(&iterator_spec)); + if (!iterator_type) + throw py::error_already_set(); + auto name = [](const char* text) { + py::object result = steal(PyUnicode_InternFromString(text)); + if (!result) + throw py::error_already_set(); + return result; + }; + auto* row_class = reinterpret_cast(row_type.ptr()); + py::object getattribute = name("__getattribute__"); + py::tuple attributes = py::make_tuple( + name("_column_names"), name("_column_map"), name("_values")); + py::tuple row_members(static_cast(ROW_FIELD_COUNT)); + py::object hook = row_type.attr("_mapping_keys"); + py::object getitem = row_type.attr("__getitem__"); + bool standard_access = _PyType_Lookup(row_class, getattribute.ptr()) == + _PyType_Lookup(&PyBaseObject_Type, getattribute.ptr()); + for (int field = 0; field < ROW_FIELD_COUNT; ++field) { + PyObject* descriptor = _PyType_Lookup(row_class, PyTuple_GET_ITEM(attributes.ptr(), field)); + row_members[field] = standard_access && member_offset(row_class, descriptor) >= 0 + ? borrow(descriptor) : py::none(); + } + // MRO-resolved member descriptors support both Python slots and an inherited native base. + auto version = row_class->tp_version_tag; + py::object owner = RowMappingCache::watcher_owner(); + py::object schema_class = RowMappingCache::schema_type(); + py::object unique_name = name("_unique"); + py::object prepared_name = name("_prepared"); + auto* names_class = reinterpret_cast(names_type.ptr()); + bool ordinary_names = names_class->tp_getattro == PyObject_GenericGetAttr && + names_class->tp_setattro == PyObject_GenericSetAttr && + !_PyType_Lookup(names_class, unique_name.ptr()) && + !_PyType_Lookup(names_class, prepared_name.ptr()); + unsigned int names_version = ordinary_names ? names_class->tp_version_tag : 0; + py::tuple defaults = py::make_tuple( + row_type, hook, names_type, version, iterator_type, fallback_iterator, + name("_row"), name("_mapping_keys"), name("_column_names"), name("_unique"), + name("_unique_keys"), name("_column_map"), name("_values"), row_members, + owner, schema_class, names_version, prepared_name, + PyFunction_Check(hook.ptr()) ? borrow(PyFunction_GetCode(hook.ptr())) : py::none(), + getitem, + PyFunction_Check(getitem.ptr()) ? borrow(PyFunction_GetCode(getitem.ptr())) : py::none()); + py::object configuration = make_configuration(defaults); + + static PyMemberDef members[] = { + {"_row", T_OBJECT_EX, offsetof(View, row), 0, "The row observed by this view."}, + {nullptr, 0, 0, 0, nullptr}, + }; + static PyGetSetDef getsets[] = { + {const_cast("_prepared_counts"), reinterpret_cast(statistics), nullptr, + const_cast("Preparation, schema-build, prepared, ordinary and ordered-hit counts."), + nullptr}, + {nullptr, nullptr, nullptr, nullptr, nullptr}, + }; + static PyType_Slot slots[] = { + {Py_tp_new, reinterpret_cast(create)}, + {Py_tp_dealloc, reinterpret_cast(dealloc)}, + {Py_tp_traverse, reinterpret_cast(traverse)}, + {Py_tp_clear, reinterpret_cast(clear)}, + {Py_tp_members, members}, + {Py_tp_getset, getsets}, + {Py_mp_subscript, reinterpret_cast(subscript)}, + {Py_tp_iter, reinterpret_cast(iterate)}, + {0, nullptr}, + }; + static PyType_Spec spec = { + "mssql_python.row._RowMappingBase", sizeof(View), 0, + Py_TPFLAGS_DEFAULT | Py_TPFLAGS_BASETYPE | Py_TPFLAGS_HAVE_GC, slots, + }; + py::object type = steal(PyType_FromSpec(&spec)); + if (!type) + throw py::error_already_set(); + type.attr("_row_mapping_defaults") = configuration; + type.attr("__slots__") = py::make_tuple("_row"); + type.attr("_prepared_supported") = !owner.is_none(); + return type; +} + +} // namespace RowMappingView diff --git a/mssql_python/pybind/row_mapping_cache.hpp b/mssql_python/pybind/row_mapping_cache.hpp new file mode 100644 index 000000000..ec7f3015a --- /dev/null +++ b/mssql_python/pybind/row_mapping_cache.hpp @@ -0,0 +1,255 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT license. + +#pragma once + +#include "py_ref.hpp" +#include +#include +#include +#include +#include + +#if PY_VERSION_HEX >= 0x030C0000 && !defined(Py_GIL_DISABLED) +#define MSSQL_MAPPING_WATCHERS 1 +#else +#define MSSQL_MAPPING_WATCHERS 0 +#endif + +namespace RowMappingCache { + +struct Watcher { + PyInterpreterState* interpreter; + Watcher* next; + std::uint64_t epoch; + int id; + bool enabled; +}; + +#if MSSQL_MAPPING_WATCHERS +struct Registry { + std::mutex mutex; + Watcher* head = nullptr; +}; + +inline Registry& registry() { + // Only native pointers live here; the interpreter capsule owns each entry. + static Registry value; + return value; +} + +inline int changed(PyDict_WatchEvent event, PyObject*, PyObject*, PyObject*) { + // Every prepared cache owns its dictionaries, so deallocation cannot stale it. + if (event == PyDict_EVENT_DEALLOCATED) + return 0; + PyInterpreterState* interpreter = PyInterpreterState_Get(); + Registry& entries = registry(); + std::lock_guard lock(entries.mutex); + for (Watcher* state = entries.head; state; state = state->next) { + if (state->interpreter == interpreter) { + if (state->epoch == std::numeric_limits::max()) + state->enabled = false; + else + ++state->epoch; + break; + } + } + // No allocation, Python execution, dictionary changes or exception changes. + return 0; +} + +inline void destroy_watcher(PyObject* capsule) { + auto* state = static_cast(PyCapsule_GetPointer(capsule, "mssql.RowMappingWatcher")); + if (!state) { + PyErr_WriteUnraisable(capsule); + return; + } + { + Registry& entries = registry(); + std::lock_guard lock(entries.mutex); + Watcher** link = &entries.head; + while (*link && *link != state) + link = &(*link)->next; + if (*link) + *link = state->next; + } + if (Py_IsInitialized()) { + PyObject *error_type, *error_value, *error_traceback; + PyErr_Fetch(&error_type, &error_value, &error_traceback); + if (PyDict_ClearWatcher(state->id) < 0) + PyErr_WriteUnraisable(capsule); + PyErr_Restore(error_type, error_value, error_traceback); + } + delete state; +} + +inline py::object watcher_owner() { + PyObject* dictionary = PyInterpreterState_GetDict(PyInterpreterState_Get()); + if (!dictionary) { + if (!PyErr_Occurred()) + PyErr_SetString(PyExc_RuntimeError, "Cannot initialize RowMapping interpreter state"); + throw py::error_already_set(); + } + py::str key("_mssql_python_row_mapping_watcher"); + PyObject* existing = PyDict_GetItemWithError(dictionary, key.ptr()); + if (existing) { + if (existing != Py_None && !PyCapsule_IsValid(existing, "mssql.RowMappingWatcher")) + throw py::type_error("Invalid RowMapping watcher state"); + return borrow(existing); + } + if (PyErr_Occurred()) + throw py::error_already_set(); + auto* state = new (std::nothrow) Watcher{PyInterpreterState_Get(), nullptr, 1, -1, true}; + if (!state) + throw std::bad_alloc(); + state->id = PyDict_AddWatcher(changed); + if (state->id < 0) { + delete state; + if (!PyErr_ExceptionMatches(PyExc_RuntimeError)) + throw py::error_already_set(); + PyErr_Clear(); + if (PyErr_WarnEx(PyExc_RuntimeWarning, + "RowMapping preparation disabled: no CPython dict watcher available", 1) < 0) + throw py::error_already_set(); + if (PyDict_SetItem(dictionary, key.ptr(), Py_None) < 0) + throw py::error_already_set(); + return py::none(); + } + py::object owner = steal(PyCapsule_New(state, "mssql.RowMappingWatcher", destroy_watcher)); + if (!owner) { + PyDict_ClearWatcher(state->id); + delete state; + throw py::error_already_set(); + } + { + Registry& entries = registry(); + std::lock_guard lock(entries.mutex); + state->next = entries.head; + entries.head = state; + } + if (PyDict_SetItem(dictionary, key.ptr(), owner.ptr()) < 0) + throw py::error_already_set(); + return owner; +} +#else +inline py::object watcher_owner() { return py::none(); } +#endif + +struct Entry { + PyObject* key; + PyObject* index; + Py_ssize_t offset; + bool representable; +}; + +struct Schema { + PyObject_HEAD + PyObject* columns; + PyObject* unique; + PyObject* indices; + PyObject* watcher_owner; + Entry* entries; + Py_ssize_t entry_count; + std::uint64_t epoch; +}; + +inline int schema_traverse(Schema* self, visitproc visit, void* arg) { + Py_VISIT(Py_TYPE(self)); + Py_VISIT(self->columns); + Py_VISIT(self->unique); + Py_VISIT(self->indices); + Py_VISIT(self->watcher_owner); + for (Py_ssize_t index = 0; index < self->entry_count; ++index) { + Py_VISIT(self->entries[index].key); + Py_VISIT(self->entries[index].index); + } + return 0; +} + +inline int schema_clear(Schema* self) { + self->epoch = 0; + Entry* entries = std::exchange(self->entries, nullptr); + Py_ssize_t count = std::exchange(self->entry_count, 0); + py::object owner = steal(std::exchange(self->watcher_owner, nullptr)); + py::object columns = steal(std::exchange(self->columns, nullptr)); + py::object unique = steal(std::exchange(self->unique, nullptr)); + py::object indices = steal(std::exchange(self->indices, nullptr)); + for (Py_ssize_t index = 0; index < count; ++index) { + Py_DECREF(entries[index].key); + Py_DECREF(entries[index].index); + } + PyMem_Free(entries); + return 0; +} + +inline int allocate_entries(Schema* self, Py_ssize_t size) { + if (size == 0) + return 0; + if (size > PY_SSIZE_T_MAX / static_cast(sizeof(Entry))) { + PyErr_NoMemory(); + return -1; + } + self->entries = static_cast(PyMem_Calloc(static_cast(size), sizeof(Entry))); + if (!self->entries) { + PyErr_NoMemory(); + return -1; + } + return 0; +} + +inline int append_entry(Schema* self, PyObject* key, PyObject* index) { + // Exact ints only: overflow is a fallback marker, not an early lookup error. + int overflow = 0; + long long offset = PyLong_AsLongLongAndOverflow(index, &overflow); + if (offset == -1 && PyErr_Occurred()) + return -1; + Entry& entry = self->entries[self->entry_count]; + entry.key = Py_NewRef(key); + entry.index = Py_NewRef(index); + entry.representable = overflow == 0 && + offset >= std::numeric_limits::min() && + offset <= std::numeric_limits::max(); + if (entry.representable) + entry.offset = static_cast(offset); + ++self->entry_count; + return 0; +} + +inline void schema_dealloc(Schema* self) { + PyObject_GC_UnTrack(self); + schema_clear(self); + PyTypeObject* type = Py_TYPE(self); + type->tp_free(reinterpret_cast(self)); + Py_DECREF(type); +} + +inline py::object schema_type() { + static PyType_Slot slots[] = { + {Py_tp_dealloc, reinterpret_cast(schema_dealloc)}, + {Py_tp_traverse, reinterpret_cast(schema_traverse)}, + {Py_tp_clear, reinterpret_cast(schema_clear)}, + {0, nullptr}, + }; + static PyType_Spec spec = { + "mssql_python.row._MappingSchema", sizeof(Schema), 0, + Py_TPFLAGS_DEFAULT | Py_TPFLAGS_HAVE_GC, slots, + }; + py::object type = steal(PyType_FromSpec(&spec)); + if (!type) + throw py::error_already_set(); + return type; +} + +inline bool builtin_keys(PyObject* dictionary, bool integer_values = false) { + if (!PyDict_CheckExact(dictionary)) + return false; + Py_ssize_t position = 0; + PyObject *key, *value; + while (PyDict_Next(dictionary, &position, &key, &value)) { + if (!PyUnicode_CheckExact(key) || (integer_values && !PyLong_CheckExact(value))) + return false; + } + return true; +} + +} // namespace RowMappingCache diff --git a/mssql_python/row.py b/mssql_python/row.py index ccec53787..e43befbdb 100644 --- a/mssql_python/row.py +++ b/mssql_python/row.py @@ -7,11 +7,37 @@ import decimal import uuid as _uuid -from collections.abc import Mapping +from collections.abc import KeysView, Mapping from typing import Any +from mssql_python import ddbc_bindings from mssql_python.logging import logger +class _ColumnNames(tuple): + """Result-shared names; validate and deduplicate only on mapping use.""" + + def __new__(cls, names): + result = super().__new__(cls, names) + result._unique = None + result._prepared = None + return result + + def _unique_keys(self): + keys = self._unique + if keys is None: + if not all(type(name) is str for name in self): + self._unique = False + return False + keys = dict.fromkeys(self) + self._unique = keys + return keys + + def __getstate__(self): + state = self.__dict__.copy() + state["_prepared"] = None + return state + + class Row: """ A row of data from a cursor fetch operation. Provides both tuple-like indexing @@ -287,6 +313,12 @@ def _mapping(self) -> "RowMapping": set, ``in`` and ``[]`` always agree. Unlike ``row[name]`` / ``row.name``, the view does NOT resolve case-insensitive names or catalog aliases; with ``lowercase=True`` the keys are the lowercased names. + - Cursor rows share lazily deduplicated canonical keys; values are never + cached in the view. Existing views continue to observe changes to the row. + - On GIL-enabled CPython 3.12+, ordinary mapping iteration can prepare a + shared index schema, invalidated by dictionary watchers. This lazily + materializes the row's standard attribute dictionary, but stores no + cache in it. Custom storage and hooks retain ordinary lookup behavior. - ``_mapping`` is a property, so a column literally named ``_mapping`` is shadowed: read it with ``row["_mapping"]`` or ``row._mapping["_mapping"]``. """ @@ -359,7 +391,24 @@ def __repr__(self) -> str: return repr(tuple(self._values)) -class RowMapping(Mapping): +def _iter_mapping_keys(names): + seen = set() + for name in names: + if name not in seen: + seen.add(name) + yield name + + +class _RowMappingKeys(KeysView): + __slots__ = () + + def __iter__(self): + return iter(self._mapping) + + +class RowMapping( + ddbc_bindings._make_row_mapping_base(Row, _ColumnNames, _iter_mapping_keys), Mapping +): """Read-only ``Mapping`` view over a :class:`Row` (column name -> value). Created via :attr:`Row._mapping`. Keys are the row's canonical column names, @@ -369,27 +418,17 @@ class RowMapping(Mapping): reflects the row it wraps and copies no values. """ - __slots__ = ("_row",) + __slots__ = () def __init__(self, row: "Row") -> None: self._row = row - def __getitem__(self, key: str) -> Any: - # Restrict lookups to the canonical column names yielded by __iter__ so - # membership and lookup agree with iteration (proper Mapping semantics). - # Row.__getitem__ additionally accepts case-insensitive names and catalog - # aliases, but those are not iterated keys, so the view must not resolve - # them here. - if isinstance(key, str) and key in self._row._mapping_keys(): - return self._row[key] - raise KeyError(key) + def __getnewargs__(self): + """Initialize native state before copying or restoring Python slots.""" + return () - def __iter__(self): - seen = set() - for name in self._row._mapping_keys(): - if name not in seen: - seen.add(name) - yield name + def keys(self): + return _RowMappingKeys(self) def __len__(self) -> int: return sum(1 for _ in self) diff --git a/tests/test_036_profiler_ci.py b/tests/test_036_profiler_ci.py index 66ca343b0..faa2dc244 100644 --- a/tests/test_036_profiler_ci.py +++ b/tests/test_036_profiler_ci.py @@ -91,7 +91,10 @@ def sample(scale): ), scenarios={ name: dict( - wall_ms=10 * scale, work="Rows: 100", cpp={"ddbc::query": counter}, py={} + wall_ms=10 * scale, + work="Rows: 100", + cpp={} if name == "row_to_dict" else {"ddbc::query": counter}, + py={}, ) for name in reporting.CASES }, @@ -146,6 +149,29 @@ def test_phase_call_changes_are_reported_without_summing_nested_totals(report): row = reporting.comparisons(report)[1] assert row["counts"] == ["ddbc::query (1 -> 2 calls)"] assert row["phases"] == [(4.0, "ddbc::query")] + reporting.validate(report) + conversion = reporting.comparisons(report)[-1] + assert conversion["name"] == "row_to_dict" + assert conversion["counts"] == conversion["phases"] == [] + sample = report["pairs"][0]["candidate"]["scenarios"] + for invalid in ( + None, + [], + {"ddbc::query": {"calls": 0}}, + {"ddbc::query": dict(calls=1, total_us=2, min_us=2, max_us=1)}, + {"ddbc::query": dict(calls=1, total_us=1, min_us=1, max_us=2)}, + ): + sample["row_to_dict"]["cpp"] = invalid + with pytest.raises(ValueError): + reporting.validate(report) + sample["row_to_dict"]["cpp"] = {} + sample["row_to_dict"]["py"] = None + with pytest.raises(ValueError): + reporting.validate(report) + sample["row_to_dict"]["py"] = {} + sample["select"]["cpp"] = {} + with pytest.raises(ValueError, match="Missing or oversized profiling data"): + reporting.validate(report) @pytest.mark.parametrize("case", ["nan", "missing", "environment", "work", "few", "prefix", "zero"]) @@ -690,13 +716,108 @@ 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) == 22 + assert len(reporting.CASES) == 23 + 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 [name for name in reporting.CASES if name.startswith("lob_")] == [ "lob_varchar_256k_fetchall", ] +def test_row_to_dict_workload_times_only_warm_conversion(monkeypatch, report): + from mssql_python.row import Row + + names = tuple(f"c{column}" for column in range(32)) + columns = dict(zip(names, range(32))) + visits = [0] * 10_000 + context = MagicMock() + context.collect.return_value = ({}, {}) + clock = MagicMock() + + class ObservedRow: + def __init__(self, index): + self.index = index + self.row = Row( + [None if index % 7 == 0 else index + column for column in range(32)], + columns, + column_names=names, + ) + + @property + def _mapping(self): + if visits[self.index] == 0: + context.enable.assert_not_called() + clock.assert_not_called() + else: + context.enable.assert_called_once() + assert clock.call_count == 1 + context.collect.assert_not_called() + visits[self.index] += 1 + return self.row._mapping + + rows = [ObservedRow(index) for index in range(10_000)] + cursor = MagicMock() + cursor.fetchall.return_value = rows + cursor.messages = [] + connection = MagicMock() + connection.cursor.return_value.__enter__.return_value = cursor + 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 += " FROM #perf_test ORDER BY id" + + def tick(): + cursor.execute.assert_called_once_with(sql) + cursor.fetchall.assert_called_once_with() + context.enable.assert_called_once() + context.collect.assert_not_called() + assert visits == [clock.call_count] * 10_000 + return 1.0 if clock.call_count == 1 else 1.1 + + clock.side_effect = tick + monkeypatch.setattr(benchmark_workloads.time, "perf_counter", clock) + result = benchmark_workloads.row_to_dict(connection, "#perf_test", context) + assert result["wall_ms"] == pytest.approx(100) + assert result["detail"] == ( + "Rows: 10000; columns: 32; types: int/NULL; API: dict(row._mapping); mode: warm" + ) + assert result["cpp"] == result["py"] == {} + assert clock.call_count == 2 + assert visits == [2] * 10_000 + cursor.fetchone.assert_not_called() + cursor.fetchmany.assert_not_called() + context.collect.assert_called_once() + context.disable.assert_called_once() + connection.cursor.return_value.__exit__.assert_called_once() + assert reporting.TASK_NAMES["row_to_dict"] in reporting.render([report], "c" * 40, 42) + + # Equality alone would accept True as the expected integer 1. + rows[1].row._values[0] = True + visits[:] = [0] * 10_000 + clock.reset_mock() + context.reset_mock() + cursor.reset_mock() + connection.cursor.return_value.__exit__.reset_mock() + with pytest.raises(AssertionError): + benchmark_workloads.row_to_dict(connection, "#perf_test", context) + assert clock.call_count == 2 + 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.fetchall.side_effect = RuntimeError("fetch failed") + with pytest.raises(RuntimeError, match="fetch failed"): + benchmark_workloads.row_to_dict(connection, "#perf_test", context) + context.enable.assert_not_called() + context.disable.assert_called_once() + connection.cursor.return_value.__exit__.assert_called_once() + + def test_scalar_fetchval_workload_times_only_fetch_and_reaches_eof(monkeypatch, report): monkeypatch.setattr("mssql_python.logging.logger", SimpleNamespace(is_debug_enabled=False)) cursor = MagicMock() @@ -871,7 +992,12 @@ def test_legacy_insert_workload_executes_both_variants(input_sizes): @pytest.mark.parametrize("fail", [False, True]) def test_worker_checkpoints_completed_and_active_scenarios(tmp_path, monkeypatch, capsys, fail): output = tmp_path / "base-0.json" - result = dict(wall_ms=10.0, cpp={"ddbc::run": {}}, py={}, detail="Rows: 10") + result = dict( + wall_ms=10.0, + cpp={"ddbc::run": dict(calls=1, total_us=1, min_us=1, max_us=1)}, + py={}, + detail="Rows: 10", + ) profiler = MagicMock() profiler.__enter__.return_value = profiler profiler._conn.cursor.return_value.__enter__.return_value.fetchone.return_value = ("16.0",) @@ -880,14 +1006,14 @@ def run(name): partial = json.loads(output.read_text()) assert partial["active_scenario"] == name if name == "second": - assert set(partial["scenarios"]) == {"first"} + assert set(partial["scenarios"]) == {"row_to_dict"} if fail: raise RuntimeError("workload failure") - return [result] + return [dict(result, cpp={}) if name == "row_to_dict" else result] profiler.run.side_effect = run core = SimpleNamespace(Profiler=lambda: profiler) - workloads = SimpleNamespace(registry=lambda: {"first": None, "second": None}) + workloads = SimpleNamespace(registry=lambda: {"row_to_dict": None, "second": None}) monkeypatch.setattr(controller, "check_build", lambda *a, **kw: None) monkeypatch.setattr(controller, "load_suite", lambda: (core, workloads)) args = SimpleNamespace(source_root=tmp_path, scenarios=None, output=output) @@ -899,10 +1025,30 @@ def run(name): controller.worker(args) final = json.loads(output.read_text()) assert final["environment"]["sql_version"] == "16.0" - assert set(final["scenarios"]) == {"first", "second"} + assert set(final["scenarios"]) == {"row_to_dict", "second"} + assert final["scenarios"]["row_to_dict"]["cpp"] == {} assert "active_scenario" not in final assert "Starting scenario: second" in capsys.readouterr().out profiler.__exit__.assert_called_once() + if not fail: + profiler.run.side_effect = None + for name, cpp, error in ( + ("second", {}, RuntimeError), + ("row_to_dict", None, RuntimeError), + ("row_to_dict", [], ValueError), + ("row_to_dict", {"ddbc::run": {"calls": 0}}, ValueError), + ): + args.scenarios = [name] + profiler.run.return_value = [dict(result, cpp=cpp)] + with pytest.raises(error): + controller.worker(args) + partial = json.loads(output.read_text()) + assert partial["active_scenario"] == name + assert partial["scenarios"] == {} + args.scenarios = ["row_to_dict"] + profiler.run.return_value = [dict(result, cpp={}, py=None)] + with pytest.raises(RuntimeError, match="was skipped"): + controller.worker(args) def test_measure_timeout_retains_partial_results_and_log(tmp_path, monkeypatch): diff --git a/tests/test_row_mapping_metadata.py b/tests/test_row_mapping_metadata.py new file mode 100644 index 000000000..a114d39ad --- /dev/null +++ b/tests/test_row_mapping_metadata.py @@ -0,0 +1,930 @@ +""" +Copyright (c) Microsoft Corporation. +Licensed under the MIT license. + +Canonical mapping metadata, live-view and native protocol compatibility guards. +""" + +import copy +import gc +import os +from pathlib import Path +import pickle +import subprocess +import sys +from collections.abc import KeysView, Mapping + +import pytest + +import mssql_python +from mssql_python.row import Row, RowMapping, _ColumnNames, _iter_mapping_keys + + +class _SerializableRow(Row): + """Keep mapping serialization independent of uninitialized Row reconstruction.""" + + def __reduce_ex__(self, protocol): + return ( + type(self), + ( + self._values, + self._column_map, + None, + (), + None, + self._column_map_lower, + self._column_names, + ), + ) + + +def make_rows(names, values): + columns = {name: index for index, name in enumerate(names)} + return [Row(value, columns, column_names=names) for value in values] + + +def run_mapping_program(program, *args, debug=False): + command = [sys.executable] + if debug: + command += ["-X", "dev"] + result = subprocess.run( + [*command, "-c", program, *args], + cwd=Path(__file__).resolve().parents[1], + env=dict(os.environ, PYTHONMALLOC="debug") if debug else None, + capture_output=True, + text=True, + timeout=60 if debug else 30, + check=False, + ) + assert result.returncode == 0, result.stdout + result.stderr + + +def test_mapping_deduplicates_once_for_all_rows(): + names = _ColumnNames(("a", "b", "a")) + rows = make_rows(names, [[1, 2, 3]] * 20) + assert names._unique is None + references = [tuple(map(id, gc.get_referents(row))) for row in rows] + builds = [] + previous = sys.getprofile() + + def record(frame, event, function): + if event == "c_call" and getattr(function, "__name__", None) == "fromkeys": + builds.append(function) + + try: + sys.setprofile(record) + results = [dict(row._mapping) for row in rows] + assert names._unique_keys() is names._unique + finally: + sys.setprofile(previous) + assert results == [{"a": 3, "b": 2}] * 20 + assert len(builds) == 1 + for row, before in zip(rows, references): + after = tuple(map(id, gc.get_referents(row))) + if after != before: + # Preparation may materialize the standard dict, but stores no cache on Row. + attributes = vars(row) + assert attributes == {} + assert sorted(after) == sorted((*before, id(attributes))) + + +def test_mapping_does_not_inspect_names_until_first_use(monkeypatch): + calls = [] + + def iterate(names): + calls.append(names) + return tuple.__iter__(names) + + monkeypatch.setattr(_ColumnNames, "__iter__", iterate, raising=False) + names = _ColumnNames(("a", "b")) + row = Row([1, 2], {"a": 0, "b": 1}, column_names=names) + assert tuple(row) == (1, 2) + assert calls == [] + assert names._unique is None + assert row._mapping["b"] == 2 + assert calls + assert names._unique == {"a": None, "b": None} + + +def test_mapping_native_slots_bypass_default_per_cell_python_hooks(): + program = """ +import sys +import pytest +from mssql_python.row import Row, RowMapping, _ColumnNames + +names = _ColumnNames("c" + str(index) for index in range(128)) +columns = {name: index for index, name in enumerate(names)} +rows = [Row(list(range(128)), columns, column_names=names) for _ in range(10)] +views = [row._mapping for row in rows] +calls = [] +codes = { + method.__code__ for method in (Row._mapping_keys, Row.__getitem__) + if hasattr(method, "__code__") +} +def record(frame, event, argument): + if event == "call" and frame.f_code in codes: + calls.append(frame.f_code.co_name) +sys.setprofile(record) +try: + results = [dict(view) for view in views] +finally: + sys.setprofile(None) +assert results == [columns] * 10 +assert calls == [], calls +if RowMapping._prepared_supported: + for index, view in enumerate(views): + assert view._prepared_counts == (1, int(index == 0), 128, 0, 128), view._prepared_counts + import copy + import pickle + for restored in (copy.deepcopy(names), pickle.loads(pickle.dumps(names))): + assert tuple(restored) == tuple(names) + assert restored._unique == names._unique + assert restored._prepared is None + + empty = Row([], {}, column_names=_ColumnNames(()))._mapping + assert dict(empty) == {} + assert empty._prepared_counts == (1, 1, 0, 0, 0), empty._prepared_counts + + pair = Row([10, 20], {"a": 0, "b": 1}, column_names=_ColumnNames(("a", "b"))) + mapping = pair._mapping + first, second = iter(mapping), iter(mapping) + assert next(first) == "a" + assert mapping["b"] == 20 + assert mapping["a"] == 10 + assert mapping["a"] == 10 + assert next(second) == "a" + assert mapping["b"] == 20 + assert mapping["a"] == 10 + assert next(first) == "b" + assert mapping["b"] == 20 + assert list(second) == ["b"] + assert list(first) == [] + assert mapping._prepared_counts == (2, 1, 6, 0, 3), mapping._prepared_counts + + pair._column_map["a"] = -1 + pair._column_names = _ColumnNames(("a",)) + pending = iter(mapping) + assert next(pending) == "a" + ordered = mapping._prepared_counts[4] + pair._values = (30, 40, 50) + assert mapping["a"] == 50 + assert mapping._prepared_counts[4] == ordered + 1 + assert list(pending) == [] + + def same_index_error(): + with pytest.raises(IndexError) as expected: + pair["a"] + before = mapping._prepared_counts + with pytest.raises(IndexError) as actual: + mapping["a"] + assert actual.value.args == expected.value.args + after = mapping._prepared_counts + assert after[2] == before[2] + 1 + assert after[3:] == before[3:], (before, after) + + pending = iter(mapping) + assert next(pending) == "a" + pair._values = [] + same_index_error() + assert list(pending) == [] + pair._values = [60] + for index in (2**100, -sys.maxsize - 1): + pair._column_map["a"] = index + assert list(iter(mapping)) == ["a"] + same_index_error() +""" + run_mapping_program(program) + + +@pytest.mark.parametrize( + "change", + ( + "names", + "columns", + "values", + "getattribute", + "view_row", + "deleted_names", + "deleted_values", + "equality_descriptor", + "instance_dictionary", + "instance_dictionary_replaced", + "names_dictionary_replaced", + "canonical_keys", + "unique_replaced", + "map_modified", + "map_deleted", + "index_callback", + "key_hook_code", + "getitem_code", + ), +) +def test_mapping_field_guards_in_fresh_process(change): + program = """ +import sys +import pytest +from mssql_python.row import Row, RowMapping, _ColumnNames + +row = Row([1, 2], {"a": 0, "b": 1}, column_names=_ColumnNames(("a", "b"))) +mapping = row._mapping +assert dict(mapping) == {"a": 1, "b": 2} +if RowMapping._prepared_supported: + assert mapping._prepared_counts == (1, 1, 2, 0, 2), mapping._prepared_counts +change = sys.argv[1] +if change == "names": + Row._column_names = property(lambda self: ("b",)) + assert dict(mapping) == {"b": 2} +elif change == "columns": + calls = [] + class Columns(dict): + def __contains__(self, key): + calls.append(("contains", key)) + return super().__contains__(key) + def __getitem__(self, key): + calls.append(("getitem", key)) + return 1 - super().__getitem__(key) + row._column_map = Columns({"a": 0, "b": 1}) + before = mapping._prepared_counts + expected = row["a"] + expected_calls = calls.copy() + calls.clear() + assert mapping["a"] == expected == 2 + assert calls == expected_calls == [("contains", "a"), ("getitem", "a")] + calls.clear() + assert dict(mapping) == {"a": 2, "b": 1} + assert calls == expected_calls + [("contains", "b"), ("getitem", "b")] + assert mapping._prepared_counts == before[:3] + (before[3] + 3, before[4]) + row._column_map = {"a": 0, "b": 1} + before = mapping._prepared_counts + assert dict(mapping) == {"a": 1, "b": 2} + if RowMapping._prepared_supported: + after = mapping._prepared_counts + assert after[0] == before[0] + 1 + assert after[2:] == (before[2] + 2, before[3], before[4] + 2) + Row._column_map = property(lambda self: {"a": 1, "b": 0}) + assert dict(mapping) == {"a": 2, "b": 1} +elif change == "values": + class Values(list): + def __getitem__(self, index): + return super().__getitem__(index) + 10 + row._values = Values([1, 2]) + before = mapping._prepared_counts + assert mapping["a"] == row["a"] == 11 + assert dict(mapping) == {"a": 11, "b": 12} + assert mapping._prepared_counts == before[:3] + (before[3] + 3, before[4]) + row._values = [1, 2] + before = mapping._prepared_counts + assert dict(mapping) == {"a": 1, "b": 2} + if RowMapping._prepared_supported: + after = mapping._prepared_counts + assert after[0] == before[0] + 1 + assert after[2:] == (before[2] + 2, before[3], before[4] + 2) + Row._values = property(lambda self: [3, 4]) + assert dict(mapping) == {"a": 3, "b": 4} +elif change == "getattribute": + def read(self, name): + if name == "_values": + return [5, 6] + return object.__getattribute__(self, name) + Row.__getattribute__ = read + assert dict(mapping) == {"a": 5, "b": 6} +elif change == "view_row": + import gc + configuration = RowMapping._row_mapping_defaults + defaults = next(value for value in gc.get_referents(configuration) if type(value) is tuple) + try: + RowMapping._row_mapping_defaults = defaults + assert dict(RowMapping(row)) == {"a": 1, "b": 2} + RowMapping._row_mapping_defaults = defaults[:-1] + with pytest.raises(TypeError, match="^Invalid RowMapping configuration$"): + RowMapping(row) + finally: + RowMapping._row_mapping_defaults = configuration + iterator_type = type(iter(mapping)) + original_next = iterator_type.__next__ + try: + iterator_type.__next__ = lambda self: "changed" + with pytest.raises(TypeError, match="^Invalid RowMapping iterator type$"): + RowMapping(row) + finally: + iterator_type.__next__ = original_next + other = Row([7, 8], {"a": 0, "b": 1}, column_names=row._column_names) + RowMapping._row = property(lambda self: other) + assert dict(mapping) == {"a": 7, "b": 8} + def fail_row(self): + raise RuntimeError("row descriptor failed") + RowMapping._row = property(fail_row) + with pytest.raises(RuntimeError, match="^row descriptor failed$"): + mapping["a"] + iterator = iter(mapping) + with pytest.raises(RuntimeError, match="^row descriptor failed$"): + next(iterator) + with pytest.raises(StopIteration): + next(iterator) +elif change == "deleted_names": + row._values = [("b",), 9] + row._column_map = {"_column_names": 0, "b": 1} + del row._column_names + assert dict(mapping) == {"b": 9} +elif change == "deleted_values": + del row._values + with pytest.raises(AttributeError, match="^Row has no attribute '_values'$"): + mapping["a"] +elif change == "equality_descriptor": + class Name(str): + __hash__ = str.__hash__ + def __eq__(self, other): + Row._values = property(lambda self: [10, 11]) + return str.__eq__(self, other) + row._column_map = {Name("a"): 0, "b": 1} + assert mapping["a"] == 10 +elif change in ("instance_dictionary", "instance_dictionary_replaced"): + if change == "instance_dictionary": + vars(row)["_mapping_keys"] = lambda: ("b",) + else: + class InstanceDict(dict): + pass + row.__dict__ = InstanceDict(note="custom") + assert type(row.__dict__) is InstanceDict + before = mapping._prepared_counts + assert mapping["a"] == 1 + assert dict(mapping) == {"a": 1, "b": 2} + assert mapping._prepared_counts == before[:3] + (before[3] + 3, before[4]) + row.__dict__ = {} + before = mapping._prepared_counts + assert dict(mapping) == {"a": 1, "b": 2} + if RowMapping._prepared_supported: + after = mapping._prepared_counts + assert after[0] == before[0] + 1 + assert after[2:] == (before[2] + 2, before[3], before[4] + 2) + row.__dict__ = {"_mapping_keys": lambda: ("b",)} + assert "a" not in mapping + assert mapping["b"] == 2 +elif change == "names_dictionary_replaced": + row._column_names.__dict__ = {"_unique": {"b": None}, "_prepared": None} + assert "a" not in mapping + assert mapping["b"] == 2 +elif change in ("canonical_keys", "unique_replaced"): + if change == "canonical_keys": + del row._column_names._unique["a"] + else: + row._column_names._unique = {"b": None} + assert "a" not in mapping + assert mapping["b"] == 2 +elif change == "map_modified": + row._column_map["a"] = 1 + assert mapping["a"] == 2 + assert dict(mapping) == {"a": 2, "b": 2} + if RowMapping._prepared_supported: + assert mapping._prepared_counts == (2, 2, 4, 1, 4), mapping._prepared_counts +elif change == "map_deleted": + del row._column_map["a"] + with pytest.raises(KeyError) as error: + mapping["a"] + assert error.value.args == ("Row has no column 'a'",) +elif change == "index_callback": + calls = [] + class Index: + def __index__(self): + calls.append("index") + return 1 + row._column_map["a"] = Index() + assert list(iter(mapping)) == ["a", "b"] + assert calls == [], "Key preparation must not execute __index__" + assert mapping["a"] == 2 + assert calls == ["index"] +elif change == "key_hook_code": + Row._mapping_keys.__code__ = (lambda self: ("b",)).__code__ + assert "a" not in mapping + assert mapping["b"] == 2 +elif change == "getitem_code": + Row.__getitem__.__code__ = (lambda self, key: 42).__code__ + assert mapping["a"] == 42 +else: + raise AssertionError(change) +""" + run_mapping_program(program, change) + + +def test_mapping_does_not_search_canonical_tuple(monkeypatch): + names = _ColumnNames(f"c{index}" for index in range(128)) + row = make_rows(names, [list(range(128))])[0] + + def forbidden(self, key): + raise AssertionError("canonical lookup searched the tuple") + + monkeypatch.setattr(_ColumnNames, "__contains__", forbidden) + assert row._mapping["c127"] == 127 + assert "c126" in row._mapping + assert "missing" not in row._mapping + + +def test_mapping_keys_remain_a_live_set_like_view(): + row = make_rows(_ColumnNames(("a", "b")), [[1, 2]])[0] + mapping = row._mapping + keys = mapping.keys() + assert isinstance(keys, KeysView) + assert keys & {"b", "c"} == {"b"} + assert keys == {"a", "b"} + row._column_names = _ColumnNames(("b",)) + assert list(keys) == ["b"] + assert len(keys) == 1 + assert dict(mapping) == {"b": 2} + + +def test_mapping_iterator_defers_hook_until_first_next(): + row = make_rows(_ColumnNames(("a", "b")), [[1, 2]])[0] + mapping = row._mapping + iterator = iter(mapping) + assert row._column_names._unique is None + row._column_names = _ColumnNames(("b",)) + assert next(iterator) == "b" + with pytest.raises(StopIteration): + next(iterator) + with pytest.raises(StopIteration): + next(iterator) + + +def test_mapping_iterator_failure_closes_the_iterator(): + row = Row([1], {"a": 0}) + calls = [] + + def fail(): + calls.append(1) + raise RuntimeError("key hook failed") + + row._mapping_keys = fail + iterator = iter(row._mapping) + assert calls == [] + with pytest.raises(RuntimeError, match="key hook failed"): + next(iterator) + with pytest.raises(StopIteration): + next(iterator) + assert calls == [1] + + +def test_mapping_iterator_reentrancy_matches_generator_behavior(): + row = Row([1], {"a": 0}) + iterator = iter(row._mapping) + + def keys(): + with pytest.raises(ValueError, match="already executing"): + next(iterator) + return ("a",) + + row._mapping_keys = keys + assert list(iterator) == ["a"] + + +def test_mapping_observes_replaced_column_metadata(): + names = _ColumnNames(("a", "b", "a")) + row = make_rows(names, [[1, 2, 3]])[0] + mapping = row._mapping + assert dict(mapping) == {"a": 3, "b": 2} + row._column_map = {"a": 1, "b": 0, "friendly": 2} + assert dict(mapping) == {"a": 2, "b": 1} + row._column_names = ("friendly", "a", "friendly") + assert dict(mapping) == {"friendly": 3, "a": 2} + row._column_names = None + assert list(mapping) == ["b", "a", "friendly"] + assert dict(mapping) == {"b": 1, "a": 2, "friendly": 3} + + +def test_mapping_dictionary_equality_can_replace_the_column_map(): + class Name(str): + __hash__ = str.__hash__ + + def __eq__(self, other): + row._column_map = {"a": 1} + return str.__eq__(self, other) + + original = {Name("a"): 0} + row = Row([1, 2], {"a": 0}, column_names=_ColumnNames(("a",))) + mapping = row._mapping + assert dict(mapping) == {"a": 1} + row._column_map = original + expected = row["a"] + row._column_map = original + assert mapping["a"] == expected == 2 + + +def test_mapping_failed_dictionary_membership_is_not_repeated(): + calls = [] + + class Name(str): + def __hash__(self): + return hash("a") + + def __eq__(self, other): + calls.append(other) + row._column_map_lower = {"a": 1} + return False + + row = Row([1, 2], {Name("different"): 0}, column_names=_ColumnNames(("a",))) + # One dict membership check can revisit a colliding slot, depending on the hash. + assert "a" not in row._column_map + expected_calls = calls.copy() + assert expected_calls + calls.clear() + row._column_map_lower = None + assert row._mapping["a"] == 2 + assert calls == expected_calls + + +def test_mapping_invalid_keys_keep_keyerror_arguments(): + mapping = make_rows(_ColumnNames(("a",)), [[1]])[0]._mapping + for key in ([], ("a", "b")): + with pytest.raises(KeyError) as error: + mapping[key] + assert error.value.args == (key,) + assert key not in mapping + assert mapping.get(key, "absent") == "absent" + + +def test_mapping_string_subclass_retains_tuple_membership(): + class UnhashableName(str): + __hash__ = None + + class HookRow(Row): + def __getitem__(self, key): + return 99 + + row = HookRow([1], {"a": 0}, column_names=_ColumnNames(("a",))) + assert row._mapping[UnhashableName("a")] == 99 + assert UnhashableName("missing") not in row._mapping + + +def test_mapping_custom_canonical_string_does_not_introduce_hashing(): + class UnhashableName(str): + __hash__ = None + + names = _ColumnNames((UnhashableName("a"),)) + assert names._unique is None + row = Row([1], {"a": 0}, column_names=names) + assert row._mapping["a"] == 1 + assert names._unique is False + with pytest.raises(TypeError): + list(row._mapping) + + +@pytest.mark.parametrize("placement", ("subclass", "class")) +def test_mapping_observes_key_hooks_after_view_creation(placement, monkeypatch): + class CustomRow(Row): + pass + + cls = CustomRow if placement == "subclass" else Row + row = cls([1, 2], {"a": 0, "b": 1}, column_names=_ColumnNames(("a", "b"))) + mapping = row._mapping + assert dict(mapping) == {"a": 1, "b": 2} + monkeypatch.setattr(cls, "_mapping_keys", lambda self: ("b", "b")) + assert dict(mapping) == {"b": 2} + assert "a" not in mapping + assert mapping["b"] == 2 + + +@pytest.mark.parametrize("placement", ("subclass", "class")) +def test_mapping_observes_getitem_hooks_after_view_creation(placement, monkeypatch): + class CustomRow(Row): + pass + + cls = CustomRow if placement == "subclass" else Row + row = cls([1], {"a": 0}, column_names=_ColumnNames(("a",))) + mapping = row._mapping + assert mapping["a"] == 1 + monkeypatch.setattr(cls, "__getitem__", lambda self, key: ("hook", key)) + assert mapping["a"] == ("hook", "a") + assert dict(mapping) == {"a": ("hook", "a")} + assert "missing" not in mapping + + +@pytest.mark.parametrize("lazy_values", (False, True)) +def test_mapping_configured_row_uses_inherited_member_descriptors(lazy_values): + calls = [] + + class Storage: + __slots__ = ("_values", "_column_map", "_column_names", "_column_map_lower", "__dict__") + + class InheritedRow(Storage): + __slots__ = () + __getitem__ = Row.__getitem__ + _mapping_keys = Row._mapping_keys + + if lazy_values: + + @property + def _values(self): + calls.append("values") + return self.values + + @_values.setter + def _values(self, values): + self.values = values + + class InheritedMapping( + mssql_python.ddbc_bindings._make_row_mapping_base( + InheritedRow, _ColumnNames, _iter_mapping_keys + ), + Mapping, + ): + __slots__ = () + __init__ = RowMapping.__init__ + __len__ = RowMapping.__len__ + keys = RowMapping.keys + + row = InheritedRow() + row._values = [1, 2] + row._column_map = {"a": 0, "b": 1} + row._column_map_lower = None + row._column_names = _ColumnNames(("a", "b")) + mapping = InheritedMapping(row) + assert list(iter(mapping)) == ["a", "b"] + assert calls == [] + if lazy_values: + assert mapping._prepared_counts[:3] == (0, 0, 0) + assert dict(mapping) == {"a": 1, "b": 2} + row._values = [3, 4] + assert dict(mapping) == {"a": 3, "b": 4} + row._mapping_keys = lambda: ("b",) + assert dict(mapping) == {"b": 4} + + +def test_mapping_key_descriptor_changes_and_errors_remain_live(): + calls = [] + + class KeyHook: + def __get__(self, row, cls): + if row is None: + return self + calls.append(row) + if row.fail: + raise RuntimeError("key descriptor failed") + return lambda: row.names + + class CustomRow(Row): + _mapping_keys = KeyHook() + + row = CustomRow([1, 2], {"a": 0, "b": 1}) + row.names = ("a",) + row.fail = False + mapping = row._mapping + assert mapping["a"] == 1 + assert len(calls) == 1 + row.names = ("b",) + assert mapping["b"] == 2 + assert "a" not in mapping + row.fail = True + with pytest.raises(RuntimeError, match="key descriptor failed"): + mapping["b"] + + +def test_mapping_key_hook_can_replace_the_view_row(): + first, second = make_rows(_ColumnNames(("a",)), [[1], [2]]) + mapping = first._mapping + + def redirect(): + mapping._row = second + return ("a",) + + first._mapping_keys = redirect + assert mapping["a"] == 2 + assert mapping._row is second + + +def test_mapping_isinstance_string_compatibility(): + class StringKey: + __class__ = str + + def __eq__(self, other): + return other == "a" + + class CustomRow(Row): + def __getitem__(self, key): + return 42 + + row = CustomRow([1], {"a": 0}, column_names=_ColumnNames(("a",))) + assert row._mapping[StringKey()] == 42 + + +def test_mapping_row_descriptor_and_subclass_iteration(): + class CustomMapping(RowMapping): + def __init__(self, row): + self._row = row + + @property + def _row(self): + return self.current + + @_row.setter + def _row(self, row): + self.current = row + + def __iter__(self): + return iter(("a",)) + + row = make_rows(_ColumnNames(("a", "b")), [[1, 2]])[0] + mapping = CustomMapping(row) + assert len(mapping) == 1 + assert dict(mapping) == {"a": 1} + mapping.__init__(make_rows(_ColumnNames(("a",)), [[3]])[0]) + assert mapping["a"] == 3 + + +@pytest.mark.parametrize("operation", (copy.copy, copy.deepcopy, pickle.dumps)) +def test_mapping_copy_and_pickle(operation): + row_type = Row if operation is copy.copy else _SerializableRow + row = row_type([1, 2, 3], {"a": 2, "b": 1}, column_names=_ColumnNames(("a", "b", "a"))) + mapping = row._mapping + assert dict(mapping) == {"a": 3, "b": 2} + result = operation(mapping) + if operation is pickle.dumps: + result = pickle.loads(result) + if operation is not copy.copy: + assert result._row._column_names._prepared is None + assert type(result) is RowMapping + assert dict(result) == {"a": 3, "b": 2} + if operation is copy.copy: + assert result._row is row + + +@pytest.mark.parametrize("operation", (copy.copy, copy.deepcopy)) +def test_mapping_copy_preserves_subclass_slots_and_attributes(operation): + class CustomMapping(RowMapping): + __slots__ = ("label", "__dict__") + + mapping = CustomMapping(_SerializableRow([1], {"a": 0}, column_names=_ColumnNames(("a",)))) + mapping.label = "slot" + mapping.note = "attribute" + result = operation(mapping) + assert type(result) is CustomMapping + assert dict(result) == {"a": 1} + assert result.label == "slot" + assert result.note == "attribute" + + +def test_mapping_native_lifetimes_in_isolated_debug_process(): + program = """ +import gc +import sys +import weakref +from mssql_python.row import Row, RowMapping, _ColumnNames + +names = _ColumnNames(("a",)) +row = Row([object()], {"a": 0}, column_names=names) +reference = weakref.ref(row) +mapping = row._mapping +before = sys.getrefcount(row) +for _ in range(100): + assert row._mapping["a"] is row[0] +assert sys.getrefcount(row) == before +del row +assert reference() is not None +assert list(mapping) == ["a"] +del mapping +assert reference() is None + +type_refs = sys.getrefcount(RowMapping) +for cycle in (False, True): + for _ in range(500): + row = Row([42], {"a": 0}, column_names=names) + reference = weakref.ref(row) + mapping = row._mapping + iterator = iter(mapping) + assert dict(mapping) == {"a": 42} + if cycle: + row._values[0] = (mapping, iterator) + else: + assert list(iterator) == ["a"] + del row, mapping, iterator + gc.collect() + assert reference() is None +assert sys.getrefcount(RowMapping) == type_refs + +if RowMapping._prepared_supported: + callback_results = [] + + class ReenterMapping: + def __init__(self, view): + self.view = view + + def __del__(self): + result = {} + callback_results.append(result) + try: + result["value"] = self.view["replacement"] + result["keys"] = list(iter(self.view)) + except Exception as error: + result["error"] = (type(error).__name__, str(error)) + + old_row = Row([10], {"old": 0}, column_names=_ColumnNames(("old",))) + old_reference = weakref.ref(old_row) + reentrant_view = old_row._mapping + old_row.on_delete = ReenterMapping(reentrant_view) + assert dict(reentrant_view) == {"old": 10} + assert reentrant_view._prepared_counts[:2] == (1, 1) + replacement = Row( + [20, 30], + {"replacement": 0, "other": 1}, + column_names=_ColumnNames(("replacement", "other")), + ) + reentrant_view._row = replacement + del old_row + assert old_reference() is not None + assert callback_results == [] + assert list(iter(reentrant_view)) == ["replacement", "other"] + assert callback_results == [ + {"value": 20, "keys": ["replacement", "other"]} + ], callback_results + assert old_reference() is None + assert reentrant_view._prepared_counts[0] == 2 + replacement._values[0] = 21 + assert dict(reentrant_view) == {"replacement": 21, "other": 30} + assert len(callback_results) == 1 + del reentrant_view, replacement + + canonical_key = "".join(("partial", "_first")) + missing_key = "".join(("partial", "_missing")) + first_index = int("1000") + partial_names = _ColumnNames((canonical_key, missing_key)) + partial_columns = {canonical_key: first_index} + partial_row = Row([None] * (first_index + 1), partial_columns, column_names=partial_names) + partial_row._values[first_index] = 42 + partial_view = partial_row._mapping + partial_names._unique_keys() + assert next(iter(partial_names._unique)) is canonical_key + assert next(iter(partial_columns)) is canonical_key + assert partial_columns[canonical_key] is first_index + try: + partial_row[missing_key] + except KeyError as error: + missing_args = error.args + else: + raise AssertionError("Expected ordinary Row lookup to fail") + entry_refs = (sys.getrefcount(canonical_key), sys.getrefcount(first_index)) + for _ in range(3): + assert list(iter(partial_view)) == [canonical_key, missing_key] + assert partial_view._prepared_counts[:2] == (0, 0) + assert partial_names._prepared is None + assert partial_view[canonical_key] == 42 + try: + partial_view[missing_key] + except KeyError as error: + assert error.args == missing_args + else: + raise AssertionError("Expected mapping lookup to fail") + assert (sys.getrefcount(canonical_key), sys.getrefcount(first_index)) == entry_refs + partial_columns[missing_key] = 0 + partial_row._values[0] = 84 + assert dict(partial_view) == {canonical_key: 42, missing_key: 84} + assert partial_view._prepared_counts[:2] == (1, 1) + assert partial_names._prepared is not None + del partial_view, partial_row, partial_columns, partial_names + +# Exercise shutdown with a live cycle as well as explicit collection above. +row = Row([None], {"a": 0}, column_names=names) +mapping = row._mapping +assert dict(mapping) == {"a": None} +row._values[0] = mapping +""" + run_mapping_program(program, debug=True) + + +@pytest.mark.skipif( + not RowMapping._prepared_supported, reason="CPython dictionary watchers unavailable" +) +def test_mapping_watcher_exhaustion_warns_and_falls_back(): + program = """ +import ctypes +import warnings +callback_type = ctypes.CFUNCTYPE( + ctypes.c_int, ctypes.c_int, ctypes.c_void_p, ctypes.c_void_p, ctypes.c_void_p +) +callback = callback_type(lambda *args: 0) +add = ctypes.pythonapi.PyDict_AddWatcher +add.argtypes = [callback_type] +add.restype = ctypes.c_int +clear = ctypes.pythonapi.PyDict_ClearWatcher +clear.argtypes = [ctypes.c_int] +clear.restype = ctypes.c_int +watchers = [] +try: + for _ in range(32): + try: + watchers.append(add(callback)) + except RuntimeError: + break + else: + raise AssertionError("Expected the bounded CPython watcher table to fill") + with warnings.catch_warnings(record=True) as recorded: + warnings.simplefilter("always") + from mssql_python.row import Row, RowMapping, _ColumnNames + assert any("no CPython dict watcher" in str(item.message) for item in recorded), recorded + assert not RowMapping._prepared_supported + row = Row([1], {"a": 0}, column_names=_ColumnNames(("a",))) + assert dict(row._mapping) == {"a": 1} + row._values[0] = 2 + assert dict(row._mapping) == {"a": 2} +finally: + for watcher in watchers: + assert clear(watcher) == 0 +""" + run_mapping_program(program)