diff --git a/mssql_python/_native_fetch.py b/mssql_python/_native_fetch.py new file mode 100644 index 000000000..e075dda45 --- /dev/null +++ b/mssql_python/_native_fetch.py @@ -0,0 +1,309 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. +"""Cold-path policy for the experimental native cursor fetch entry points. + +The native plan is invalidated by public CPython watchers, not private dict +versions. Eligibility checks run after invalidation, never for each warm row. +Unsupported interpreters keep the ordinary Python methods. + +Runtime schema 2 keeps the original six tuple fields and appends the GC-visible +dispatch context. Cursor-owned state is lazy and visible at _native_fetch_state; +it is not a public API or a change to constructor initialization. +""" + +import sys +from types import FunctionType, GetSetDescriptorType + +from mssql_python import ddbc_bindings, perf_timer + + +class _Policy: + def __init__(self, cursor_type, row_type, namespace, fallbacks): + self.cursor_type = cursor_type + self.row_type = row_type + self.namespace = namespace + self.observer = ddbc_bindings._fetch_observer + self._conversion_ineligible = ddbc_bindings._fetch_conversion_ineligible + self.logger = namespace["logger"] + # Normal binding and warning initialization must not invalidate this snapshot. + cursor_type._get_sql_to_c_type_map() + namespace.setdefault("__warningregistry__", {}) + self.classes = tuple( + dict.fromkeys((*cursor_type.__mro__, *row_type.__mro__, *type(self.logger).__mro__)) + ) + self.class_state = [(cls, dict(vars(cls))) for cls in self.classes] + self.functions = [] + functions = [*fallbacks, namespace["check_error"]] + for _, attributes in self.class_state: + for value in attributes.values(): + if isinstance(value, (staticmethod, classmethod)): + functions.append(value.__func__) + elif isinstance(value, property): + functions.extend((value.fget, value.fset, value.fdel)) + else: + functions.append(value) + for function in functions: + if isinstance(function, FunctionType): + self.functions.append( + ( + function, + function.__code__, + function.__defaults__, + function.__kwdefaults__, + dict(function.__kwdefaults__ or {}), + ) + ) + self.globals = [ + (namespace, name, namespace[name]) + for name in ( + "Row", + "check_error", + "perf_phase", + "ddbc_bindings", + "get_settings", + "_SQL_HANDLE_STMT", + "_SQL_NO_DATA", + ) + ] + self.globals.extend( + (vars(ddbc_bindings), name, getattr(ddbc_bindings, name)) + for name in ("DDBCSQLFetchOne", "DDBCSQLFetchMany", "construct_rows") + ) + self.globals.append((row_type._fast_create.__globals__, "Row", row_type)) + self.namespaces = [ + (scope, {key: value for key, value in scope.items() if key != "_NATIVE_FETCH_RUNTIME"}) + for scope in (namespace, row_type._fast_create.__globals__) + ] + self.dicts = tuple( + [ + namespace, + vars(ddbc_bindings), + vars(perf_timer), + vars(self.logger), + row_type._fast_create.__globals__, + ] + + [kwdefaults for _, _, _, kwdefaults, _ in self.functions if kwdefaults is not None] + ) + ddbc_bindings._watch_fetch_dependencies( + self.observer, self.dicts, self.classes + (type(self.logger),) + ) + + def __call__(self, cursor, data): + # Watch even an ineligible cursor so restoring its state can re-enable it. + connection = data.get("_connection") + if type(cursor) is not self.cursor_type or connection is None: + return None + from mssql_python.connection import Connection + + if type(connection) is not Connection: + return None + descriptor = vars(Connection).get("__dict__") + if ( + Connection.__getattribute__ is not object.__getattribute__ + or type(descriptor) is not GetSetDescriptorType + # CPython 3.15 shares the ordinary instance-dict descriptor on object. + or descriptor.__objclass__ not in (Connection, object) + or descriptor.__name__ != "__dict__" + ): + return None + connection_data = object.__getattribute__(connection, "__dict__") + ddbc_bindings._watch_fetch_dependencies( + self.observer, + (data, connection_data), + Connection.__mro__, + ) + index = data.get("_next_row_index") + if type(index) is not int or index < 1: + return None + if self._conversion_ineligible(data): + return None + for cls, original in self.class_state: + current = vars(cls) + if len(current) != len(original) or any( + name not in current or current[name] is not value + for name, value in original.items() + ): + return None + for function, code, defaults, kwdefaults, keywords in self.functions: + current_keywords = kwdefaults or {} + if ( + function.__code__ is not code + or function.__defaults__ is not defaults + or function.__kwdefaults__ is not kwdefaults + or len(current_keywords) != len(keywords) + or any( + name not in current_keywords or current_keywords[name] is not value + for name, value in keywords.items() + ) + ): + return None + if any(namespace.get(name) is not value for namespace, name, value in self.globals): + return None + for scope, original in self.namespaces: + current = {key: value for key, value in scope.items() if key != "_NATIVE_FETCH_RUNTIME"} + if len(current) != len(original) or any( + name not in current or current[name] is not value + for name, value in original.items() + ): + return None + if perf_timer.is_enabled() or self.logger.is_debug_enabled: + return None + if ( + self.row_type.__new__ is not object.__new__ + or self.row_type.__setattr__ is not object.__setattr__ + or any( + name in vars(base) + for base in Connection.__mro__ + for name in ("_decoding_generation", "_converters_generation") + ) + ): + return None + if any(name in data for name in vars(self.cursor_type)): + return None + for cached, generation in ( + ("_cached_decoding_generation", "_decoding_generation"), + ("_cached_converters_generation", "_converters_generation"), + ): + if ( + type(data.get(cached)) is not int + or type(connection_data.get(generation)) is not int + ): + return None + if ( + data.get("closed") is not False + or data.get("_has_result_set") is not True + or data.get("_skip_increment_for_next_fetch") is not False + or data.get("_cached_decoding_generation") + != connection_data.get("_decoding_generation") + or data.get("_cached_converters_generation") + != connection_data.get("_converters_generation") + or type(data.get("_cached_converter_map")) is not tuple + or data.get("_cached_converter_map") != () + or data.get("_uuid_str_indices") is not None + ): + return None + description = data.get("description") + column_map = data.get("_cached_column_map") + lower_map = data.get("_cached_column_map_lower") + names = data.get("_cached_result_columns") + ddbc_bindings._watch_fetch_dependencies( + self.observer, + tuple(mapping for mapping in (column_map, lower_map) if type(mapping) is dict), + (), + ) + if ( + type(data) is not dict + or type(connection_data) is not dict + or type(description) is not list + or not description + or data.get("_result_columns_src") is not description + or type(column_map) is not dict + or not column_map + or (lower_map is not None and type(lower_map) is not dict) + or type(names) is not tuple + or type(index) is not int + or not 0 <= index < sys.maxsize - 1 + or type(data.get("_rownumber")) is not int + or type(data.get("rowcount")) is not int + or data["_rownumber"] != index - 1 + or data["rowcount"] != index + or type(data.get("messages")) is not list + or type(data.get("_cached_char_encoding")) is not str + or type(data.get("_cached_wchar_encoding")) is not str + or type(data.get("_cached_char_ctype")) is not int + or type(data.get("arraysize")) is not int + or type(data.get("hstmt")) is not ddbc_bindings.SqlHandle + ): + return None + return ( + data["hstmt"], + data, + connection, + connection_data, + column_map, + lower_map, + names, + self.row_type, + description, + data["messages"], + index, + data["_cached_char_encoding"], + data["_cached_wchar_encoding"], + data["_cached_char_ctype"], + data["arraysize"], + self.row_type._fast_create, + type(connection), + ) + + +class _InstallRollbackError(RuntimeError): + def __init__(self, installation_error, cleanup_errors): + self.installation_error = installation_error + self.cleanup_errors = tuple(cleanup_errors) + super().__init__( + "Native fetch installation and rollback both failed", + installation_error, + self.cleanup_errors, + ) + + +def install(cursor_type, row_type, namespace): + observer = ddbc_bindings._fetch_observer + if observer is None or not isinstance(getattr(row_type, "_fast_create", None), FunctionType): + return None + methods = ("fetchone", "fetchmany", "fetchval", "__next__") + fallbacks = tuple(getattr(cursor_type, name) for name in methods) + entries = ddbc_bindings._native_fetch_methods + originals = {name: vars(cursor_type)[name] for name in methods} + begin = ddbc_bindings._begin_native_fetch_install + activate = ddbc_bindings._activate_native_fetch_install + abort = ddbc_bindings._abort_native_fetch_install + disable = ddbc_bindings._disable_native_fetch_install + context = ddbc_bindings._native_fetch_context + begin(cursor_type, fallbacks) + try: + for name, entry in zip(methods, entries): + setattr(cursor_type, name, entry) + policy = _Policy(cursor_type, row_type, namespace, fallbacks) + runtime = ( + observer, + cursor_type, + policy, + fallbacks, + ("_values", "_column_map", "_cursor", "_column_map_lower", "_column_names"), + ("_next_row_index", "_rownumber", "rowcount"), + context, + ) + if any(vars(cursor_type).get(name) is not entry for name, entry in zip(methods, entries)): + raise RuntimeError("Cursor methods changed during native fetch installation") + activate(runtime) + return runtime + except BaseException as installation_error: + # Unpublished entries can only forward to originals until activation. + cleanup_errors = [] + for name, entry in zip(methods, entries): + try: + if vars(cursor_type).get(name) is entry: + setattr(cursor_type, name, originals[name]) + except BaseException as cleanup_error: + cleanup_errors.append((name, cleanup_error)) + survivors = tuple( + name for name, entry in zip(methods, entries) if vars(cursor_type).get(name) is entry + ) + if survivors and not cleanup_errors: + cleanup_errors.append( + ( + "descriptors", + RuntimeError("Native fetch descriptors survived rollback", survivors), + ) + ) + # Retained descriptors must retain their forwarding functions. A partial + # rollback is terminal, not a context that a later install may reuse. + terminalize = disable if cleanup_errors else abort + try: + terminalize() + except BaseException as cleanup_error: + cleanup_errors.append(("context", cleanup_error)) + if cleanup_errors: + raise _InstallRollbackError(installation_error, cleanup_errors) from installation_error + raise diff --git a/mssql_python/cursor.py b/mssql_python/cursor.py index d4b85e56b..767e1ab8c 100644 --- a/mssql_python/cursor.py +++ b/mssql_python/cursor.py @@ -60,6 +60,8 @@ ddbc_sql_const.SQL_TIME.value: ddbc_sql_const.SQL_TYPE_TIME.value, ddbc_sql_const.SQL_TIMESTAMP.value: ddbc_sql_const.SQL_TYPE_TIMESTAMP.value, } +_SQL_HANDLE_STMT = ddbc_sql_const.SQL_HANDLE_STMT.value +_SQL_NO_DATA = ddbc_sql_const.SQL_NO_DATA.value def _string_only_output_converter(converter): @@ -331,7 +333,8 @@ def close(self) -> None: self._inner = None -class Cursor: # pylint: disable=too-many-instance-attributes,too-many-public-methods +# pylint: disable=too-many-instance-attributes,too-many-public-methods +class Cursor: """ Represents a database cursor, which is used to manage the context of a fetch operation. @@ -1639,6 +1642,7 @@ def connection(self) -> "Connection": def _reset_rownumber(self) -> None: """Reset the rownumber tracking when starting a new result set.""" + ddbc_bindings._invalidate_native_fetch(self) self._rownumber = -1 self._next_row_index = 0 self._has_result_set = True @@ -1683,6 +1687,7 @@ def _clear_rownumber(self): This should be called when the result set is cleared or when the cursor is reset. """ + ddbc_bindings._invalidate_native_fetch(self) self._rownumber = -1 self._has_result_set = False self._skip_increment_for_next_fetch = False @@ -2858,24 +2863,18 @@ def fetchone(self) -> Union[None, Row]: self.messages, ) - check_error(ddbc_sql_const.SQL_HANDLE_STMT.value, self.hstmt, ret) - - if ret == ddbc_sql_const.SQL_NO_DATA.value: - # No more data available + check_error(_SQL_HANDLE_STMT, self.hstmt, ret) + if ret == _SQL_NO_DATA: if self._next_row_index == 0 and self.description is not None: self.rowcount = 0 return None - # Update internal position after successful fetch if self._skip_increment_for_next_fetch: self._skip_increment_for_next_fetch = False self._next_row_index += 1 else: self._increment_rownumber() - self.rowcount = self._next_row_index - - # Get column and converter maps column_map, converter_map, column_map_lower = self._get_column_and_converter_maps() with perf_phase("py::fetchone::row_wrap"): if not converter_map and not self._uuid_str_indices: @@ -2895,6 +2894,39 @@ def fetchone(self) -> Union[None, Row]: # On error, don't increment rownumber - rethrow the error raise + def _complete_fetchone(self, row_data, ret): + """Complete an already-consumed fetch, including mutation/error fallback.""" + check_error(_SQL_HANDLE_STMT, self.hstmt, ret) + if ret == _SQL_NO_DATA: + if self._next_row_index == 0 and self.description is not None: + self.rowcount = 0 + return None + + if self._skip_increment_for_next_fetch: + self._skip_increment_for_next_fetch = False + self._next_row_index += 1 + else: + self._increment_rownumber() + self.rowcount = self._next_row_index + return self._wrap_fetchone(row_data) + + def _wrap_fetchone(self, row_data): + column_map, converter_map, column_map_lower = self._get_column_and_converter_maps() + with perf_phase("py::fetchone::row_wrap"): + if not converter_map and not self._uuid_str_indices: + return Row._fast_create( + row_data, column_map, self, column_map_lower, self._cached_result_columns + ) + return Row( + row_data, + column_map, + cursor=self, + converter_map=converter_map, + uuid_str_indices=self._uuid_str_indices, + column_map_lower=column_map_lower, + column_names=self._cached_result_columns, + ) + def fetchmany(self, size: Optional[int] = None) -> List[Row]: """ Fetch the next set of rows of a query result. @@ -2934,51 +2966,48 @@ def fetchmany(self, size: Optional[int] = None) -> List[Row]: self.messages, ) - check_error(ddbc_sql_const.SQL_HANDLE_STMT.value, self.hstmt, ret) - - # Update rownumber for the number of rows actually fetched - if rows_data and self._has_result_set: - # advance counters by number of rows actually returned - self._next_row_index += len(rows_data) - self._rownumber = self._next_row_index - 1 - - # Centralize rowcount assignment after fetch - if len(rows_data) == 0 and self._next_row_index == 0: - self.rowcount = 0 - else: - self.rowcount = self._next_row_index - - # Get column and converter maps - column_map, converter_map, column_map_lower = self._get_column_and_converter_maps() - - # Convert raw data to Row objects - uuid_idx = self._uuid_str_indices - with perf_phase("py::fetchmany::row_wrap"): - if not converter_map and not uuid_idx: - return ddbc_bindings.construct_rows( - rows_data, - Row, - column_map, - self, - column_map_lower, - self._cached_result_columns, - ) - return [ - Row( - row_data, - column_map, - cursor=self, - converter_map=converter_map, - uuid_str_indices=uuid_idx, - column_map_lower=column_map_lower, - column_names=self._cached_result_columns, - ) - for row_data in rows_data - ] + return self._complete_fetchmany(rows_data, ret) except Exception: # On error, don't increment rownumber - rethrow the error raise + def _complete_fetchmany(self, rows_data, ret): + """Complete a batch without issuing another fetch on deoptimization.""" + check_error(_SQL_HANDLE_STMT, self.hstmt, ret) + if rows_data and self._has_result_set: + self._next_row_index += len(rows_data) + self._rownumber = self._next_row_index - 1 + self.rowcount = ( + 0 if len(rows_data) == 0 and self._next_row_index == 0 else self._next_row_index + ) + return self._wrap_fetchmany(rows_data) + + def _wrap_fetchmany(self, rows_data): + column_map, converter_map, column_map_lower = self._get_column_and_converter_maps() + uuid_idx = self._uuid_str_indices + with perf_phase("py::fetchmany::row_wrap"): + if not converter_map and not uuid_idx: + return ddbc_bindings.construct_rows( + rows_data, + Row, + column_map, + self, + column_map_lower, + self._cached_result_columns, + ) + return [ + Row( + row_data, + column_map, + cursor=self, + converter_map=converter_map, + uuid_str_indices=uuid_idx, + column_map_lower=column_map_lower, + column_names=self._cached_result_columns, + ) + for row_data in rows_data + ] + def fetchall(self) -> List[Row]: """ Fetch all (remaining) rows of a query result. @@ -3009,7 +3038,7 @@ def fetchall(self) -> List[Row]: ) # Check for errors - check_error(ddbc_sql_const.SQL_HANDLE_STMT.value, self.hstmt, ret) + check_error(_SQL_HANDLE_STMT, self.hstmt, ret) # Update rownumber for the number of rows actually fetched if rows_data and self._has_result_set: @@ -4186,3 +4215,10 @@ def setoutputsize(self, size: int, column: Optional[int] = None) -> None: are managed automatically by the underlying driver. """ # This is a no-op - buffer sizes are managed automatically + + +# Installed only where CPython supplies public dictionary/type/function watchers. +# Python 3.10/3.11 and free-threaded builds retain the methods above. +from mssql_python._native_fetch import install as _install_native_fetch + +_NATIVE_FETCH_RUNTIME = _install_native_fetch(Cursor, Row, globals()) diff --git a/mssql_python/pybind/CMakeLists.txt b/mssql_python/pybind/CMakeLists.txt index 069a1b165..d802bed20 100644 --- a/mssql_python/pybind/CMakeLists.txt +++ b/mssql_python/pybind/CMakeLists.txt @@ -8,6 +8,10 @@ set(CMAKE_CXX_STANDARD_REQUIRED ON) # Performance profiling instrumentation. OFF by default so released wheels ship # with zero profiler code. Turn on with -DENABLE_PROFILING=ON (dev/internal builds). option(ENABLE_PROFILING "Build with performance profiling instrumentation" OFF) +option(MSSQL_NATIVE_FETCH_TEST_FAULTS "Build isolated native-fetch failure-contract controls" OFF) +if(MSSQL_NATIVE_FETCH_TEST_FAULTS AND NOT ENABLE_PROFILING) + message(FATAL_ERROR "Native-fetch fault controls require a separate ENABLE_PROFILING=ON build") +endif() # Enable verbose output to see actual compiler/linker commands set(CMAKE_VERBOSE_MAKEFILE ON CACHE BOOL "Verbose output" FORCE) @@ -363,6 +367,9 @@ if(ENABLE_PROFILING) message(STATUS "Building WITH performance profiling instrumentation") target_compile_definitions(ddbc_bindings PRIVATE ENABLE_PROFILING) endif() +if(MSSQL_NATIVE_FETCH_TEST_FAULTS) + target_compile_definitions(ddbc_bindings PRIVATE MSSQL_NATIVE_FETCH_TEST_FAULTS) +endif() # Add warning level flags for MSVC if(MSVC) diff --git a/mssql_python/pybind/ddbc_bindings.cpp b/mssql_python/pybind/ddbc_bindings.cpp index 4e4523955..a25868b17 100644 --- a/mssql_python/pybind/ddbc_bindings.cpp +++ b/mssql_python/pybind/ddbc_bindings.cpp @@ -3433,11 +3433,11 @@ static inline bool IsLobOrVariantColumn(SQLSMALLINT dataType, SQLULEN columnSize (columnSize == 0 || columnSize == SQL_NO_TOTAL || columnSize > SQL_MAX_LOB_SIZE)); } -// Helper function to retrieve column data -SQLRETURN SQLGetData_wrap(SqlHandlePtr StatementHandle, SQLUSMALLINT colCount, py::list& row, - const std::string& charEncoding = "utf-16le", - const std::string& wcharEncoding = "utf-16le", - int charCtype = SQL_C_WCHAR, py::handle messages = {}) { +static SQLRETURN GetRowData(const SqlHandlePtr& StatementHandle, SQLUSMALLINT colCount, + py::list& row, const std::string& charEncoding, + const std::string& wcharEncoding, int charCtype, + py::handle messages, const ResultMetadataCache::Snapshot& snapshot, + bool completeResult = false) { PERF_TIMER("SQLGetData_wrap"); // Note: wcharEncoding parameter is reserved for future use // Currently WCHAR data always uses UTF-16LE for Windows compatibility @@ -3454,7 +3454,6 @@ SQLRETURN SQLGetData_wrap(SqlHandlePtr StatementHandle, SQLUSMALLINT colCount, p SQLHSTMT hStmt = StatementHandle->get(); ResultMetadataFailureGuard metadataFailure(StatementHandle->resultMetadata, ret); - const auto snapshot = StatementHandle->resultMetadata.snapshot(); // The separately exposed GetData entry point may request only a prefix. const auto metadata = snapshot.metadata && snapshot.metadata->columns.size() >= colCount ? snapshot.metadata @@ -4232,7 +4231,56 @@ SQLRETURN SQLGetData_wrap(SqlHandlePtr StatementHandle, SQLUSMALLINT colCount, p if (!complete) { StatementHandle->resultMetadata.clear(); } else if (pending && pending->columns.size() == colCount) { + pending->completeResult = completeResult; StatementHandle->resultMetadata.publish(snapshot.generation, std::move(pending)); + } else if (completeResult && metadata && !metadata->completeResult) { + auto completeMetadata = std::make_shared(*metadata); + completeMetadata->completeResult = true; + StatementHandle->resultMetadata.publish(snapshot.generation, std::move(completeMetadata)); + } + return ret; +} + +SQLRETURN SQLGetData_wrap(SqlHandlePtr StatementHandle, SQLUSMALLINT colCount, py::list& row, + const std::string& charEncoding = "utf-16le", + const std::string& wcharEncoding = "utf-16le", + int charCtype = SQL_C_WCHAR, py::handle messages = {}) { + return GetRowData(StatementHandle, colCount, row, charEncoding, wcharEncoding, charCtype, + messages, StatementHandle->resultMetadata.snapshot()); +} + +// An unbound, exactly-one-row fetch shared by fetchone and fetchmany(1). +// No buffers or ODBC pointers survive the call. +static SQLRETURN FetchSingleRow(const SqlHandlePtr& handle, py::list& row, + const std::string& charEncoding, + const std::string& wcharEncoding, int charCtype, + py::handle messages) { + PERF_TIMER("FetchSingleRow"); + SQLRETURN ret = SQL_ERROR; + ResultMetadataFailureGuard metadataFailure(handle->resultMetadata, ret); + SQLHSTMT hStmt = handle->get(); + if (!handle->fetchColumnsUnbound) { + { + PERF_TIMER("FetchSingleRow::SQL_UNBIND"); + ret = SQLFreeStmt_ptr(hStmt, SQL_UNBIND); + } + CaptureFetchDiagnostics(hStmt, ret, messages); + if (!SQL_SUCCEEDED(ret)) + return ret; + handle->fetchColumnsUnbound = true; + } + { + py::gil_scoped_release release; + ret = SQLFetch_ptr(hStmt); + } + CaptureFetchDiagnostics(hStmt, ret, messages); + if (SQL_SUCCEEDED(ret)) { + const auto snapshot = handle->resultMetadata.snapshot(); + const auto colCount = snapshot.metadata && snapshot.metadata->completeResult + ? static_cast(snapshot.metadata->columns.size()) + : SQLNumResultCols_wrap(handle, messages); + ret = GetRowData(handle, colCount, row, charEncoding, wcharEncoding, charCtype, messages, + snapshot, true); } return ret; } @@ -4898,7 +4946,9 @@ struct FetchStateGuard { int cleanupStep = 0; FetchStateGuard(SqlHandlePtr handle, py::handle messages) - : handle(std::move(handle)), messages(messages.ptr()) {} + : handle(std::move(handle)), messages(messages.ptr()) { + this->handle->fetchColumnsUnbound = false; + } void observe(SQLRETURN ret) const { CaptureFetchDiagnostics(handle->get(), ret, py::handle(messages)); @@ -4928,6 +4978,7 @@ struct FetchStateGuard { break; default: ret = SQLFreeStmt_ptr(handle->get(), SQL_UNBIND); + handle->fetchColumnsUnbound = SQL_SUCCEEDED(ret); break; } observe(ret); @@ -4976,12 +5027,11 @@ SQLRETURN FetchMany_wrap(SqlHandlePtr StatementHandle, py::list& rows, int fetch SQLRETURN ret = SQL_ERROR; ResultMetadataFailureGuard metadataFailure(StatementHandle->resultMetadata, ret); SQLHSTMT hStmt = StatementHandle->get(); - // Retrieve column count - SQLSMALLINT numCols = SQLNumResultCols_wrap(StatementHandle, messages); - - // Retrieve column metadata auto snapshot = StatementHandle->resultMetadata.snapshot(); auto metadata = std::move(snapshot.metadata); + SQLSMALLINT numCols = metadata && metadata->completeResult + ? static_cast(metadata->columns.size()) + : SQLNumResultCols_wrap(StatementHandle, messages); const bool matches = metadata && numCols >= 0 && metadata->columns.size() == static_cast(numCols); if (!matches || !metadata->namesValidated) { @@ -5007,6 +5057,7 @@ SQLRETURN FetchMany_wrap(SqlHandlePtr StatementHandle, py::list& rows, int fetch } } pending->namesValidated = true; + pending->completeResult = true; StatementHandle->resultMetadata.publish(snapshot.generation, pending); metadata = std::move(pending); } @@ -5025,6 +5076,14 @@ SQLRETURN FetchMany_wrap(SqlHandlePtr StatementHandle, py::list& rows, int fetch } } + if (fetchSize == 1 && lobColumns.empty()) { + py::list row; + ret = FetchSingleRow(StatementHandle, row, charEncoding, wcharEncoding, charCtype, messages); + if (SQL_SUCCEEDED(ret)) + rows.append(row); + return ret; + } + // Initialized to 0 for LOB path counter; overwritten by ODBC in non-LOB path; SQLULEN numRowsFetched = 0; // If we have LOBs → fall back to row-by-row fetch + SQLGetData_wrap @@ -6162,6 +6221,7 @@ SQLRETURN FetchAll_wrap(SqlHandlePtr StatementHandle, py::list& rows, // names/declared fields instead of describing stable columns on every row. auto metadata = std::make_shared(); metadata->namesValidated = true; + metadata->completeResult = true; metadata->columns.reserve(numCols); for (SQLSMALLINT i = 0; i < numCols; ++i) { const auto column = columnNames[i].cast(); @@ -6285,35 +6345,9 @@ SQLRETURN FetchOne_wrap(SqlHandlePtr StatementHandle, py::list& row, // Issue #531: upgrade SQL_C_CHAR + utf-8 to SQL_C_WCHAR on Windows so the // driver does lossless UTF-16 conversion instead of returning ACP bytes. charCtype = EffectiveCharCtypeForFetch(charCtype, charEncoding); - SQLRETURN ret = SQL_ERROR; - ResultMetadataFailureGuard metadataFailure(StatementHandle->resultMetadata, ret); - SQLHSTMT hStmt = StatementHandle->get(); - - // Unbind any columns from previous fetch operations (e.g., fetchmany) - // to avoid conflicts with SQLGetData. SQLGetData cannot be used on - // columns that are already bound. - ret = SQLFreeStmt_ptr(hStmt, SQL_UNBIND); - CaptureFetchDiagnostics(hStmt, ret, messages); - if (!SQL_SUCCEEDED(ret)) - return ret; - - // Assume hStmt is already allocated and a query has been executed - { - // Release the GIL during the blocking ODBC fetch - py::gil_scoped_release release; - ret = SQLFetch_ptr(hStmt); - } - CaptureFetchDiagnostics(hStmt, ret, messages); - if (SQL_SUCCEEDED(ret)) { - // Retrieve column count - SQLSMALLINT colCount = SQLNumResultCols_wrap(StatementHandle, messages); - ret = SQLGetData_wrap(StatementHandle, colCount, row, charEncoding, wcharEncoding, - charCtype, messages); - if (!SQL_SUCCEEDED(ret)) { - LOG("FetchOne_wrap: Error retrieving data with SQLGetData - SQLRETURN=%d", ret); - return ret; - } - } else if (ret != SQL_NO_DATA) { + SQLRETURN ret = FetchSingleRow(StatementHandle, row, charEncoding, wcharEncoding, charCtype, + messages); + if (!SQL_SUCCEEDED(ret) && ret != SQL_NO_DATA) { LOG("FetchOne_wrap: Error when fetching data - SQLRETURN=%d", ret); } return ret; @@ -6402,10 +6436,13 @@ void DDBCSetDecimalSeparator(const std::string& separator) { #endif // Functions/data to be exposed to Python as a part of ddbc_bindings module +#include "native_fetch.hpp" + PYBIND11_MODULE(ddbc_bindings, m) { m.doc() = "msodbcsql driver api bindings for Python"; PyTypeCache::initialize(); + NativeFetch::bind(m); // Add architecture information as module attribute m.attr("__architecture__") = ARCHITECTURE; @@ -6626,7 +6663,15 @@ PYBIND11_MODULE(ddbc_bindings, m) { m.attr("__version__") = "1.0.0"; // Fast Row construction in C++ — replaces Python list comprehension - m.def("construct_rows", &RowFactory::construct_rows, + m.def("construct_rows", + [attributeNames = py::make_tuple("_values", "_column_map", "_cursor", + "_column_map_lower", "_column_names")]( + const py::list& rowsData, const py::object& rowClass, + const py::object& columnMap, const py::object& cursor, + const py::object& columnMapLower, const py::object& columnNames) { + return RowFactory::construct_rows(rowsData, rowClass, columnMap, cursor, + columnMapLower, columnNames, attributeNames); + }, "Build Row objects in C++ for fetchall/fetchmany fast path", py::arg("rows_data"), py::arg("row_class"), py::arg("column_map"), py::arg("cursor"), diff --git a/mssql_python/pybind/ddbc_bindings.h b/mssql_python/pybind/ddbc_bindings.h index 748ecc5ca..f84817fcd 100644 --- a/mssql_python/pybind/ddbc_bindings.h +++ b/mssql_python/pybind/ddbc_bindings.h @@ -334,6 +334,9 @@ class SqlHandle { std::unordered_map describeCache; void clearDescribeCache() { describeCache.clear(); } ResultMetadataCache resultMetadata; + // Only fetch calls under the GIL access this flag. Batch bindings are + // invalidated before installation and become reusable only after unbinding. + bool fetchColumnsUnbound = false; private: // The caller must release the GIL before waiting for native cleanup. diff --git a/mssql_python/pybind/native_fetch.hpp b/mssql_python/pybind/native_fetch.hpp new file mode 100644 index 000000000..4609a47af --- /dev/null +++ b/mssql_python/pybind/native_fetch.hpp @@ -0,0 +1,1175 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT license. + +#pragma once + +#include +#include +#include +#include + +#if defined(MSSQL_NATIVE_FETCH_TEST_FAULTS) && !defined(ENABLE_PROFILING) +#error "Native-fetch fault controls require a separate profiling-enabled test build" +#endif + +// Included after FetchSingleRow: this prototype deliberately shares value +// decoding, diagnostics and ODBC ownership with the existing native pipeline. +// Keep the full internal ownership chain at pybind11's hidden visibility. +#if defined(__GNUG__) && !defined(_WIN32) +namespace NativeFetch __attribute__((visibility("hidden"))) { +#else +namespace NativeFetch { +#endif + +#if PY_VERSION_HEX >= 0x030C0000 && !defined(Py_GIL_DISABLED) && !defined(PYPY_VERSION) + +// No Python handles are process-static. Cross-interpreter invalidation is +// conservative; callbacks only change this scalar and never allocate or call Python. +inline std::atomic mutationEpoch{1}; +inline thread_local PyObject* internalDict = nullptr; +inline thread_local PyObject* internalKey = nullptr; +inline thread_local PyObject* internalValue = nullptr; + +inline int dict_changed(PyDict_WatchEvent event, PyObject* dict, PyObject* key, + PyObject* value) { + if (event != PyDict_EVENT_MODIFIED || dict != internalDict || + (key != internalKey && + (!key || !internalKey || !PyUnicode_CheckExact(key) || + PyUnicode_Compare(key, internalKey) != 0)) || + value != internalValue) + mutationEpoch.fetch_add(1, std::memory_order_relaxed); + return 0; +} + +inline int type_changed(PyTypeObject*) { + mutationEpoch.fetch_add(1, std::memory_order_relaxed); + return 0; +} + +inline int function_changed(PyFunction_WatchEvent event, PyFunctionObject*, PyObject*) { + if (event != PyFunction_EVENT_CREATE && event != PyFunction_EVENT_DESTROY) + mutationEpoch.fetch_add(1, std::memory_order_relaxed); + return 0; +} + +struct Observer { + int dict; + int type; + int function; +}; + +enum Input : Py_ssize_t { + Handle, CursorDict, Connection, ConnectionDict, ColumnMap, LowerMap, Names, + RowType, Description, Messages, NextIndex, CharEncoding, WideEncoding, + CharCtype, ArraySize, Factory, ConnectionType, InputCount +}; +enum class Mode : size_t { One, Many, Value, Next }; + +struct Plan { + py::dict cursorDict; + py::object inputs = py::none(); + SqlHandlePtr handle; + std::string charEncoding; + std::string wideEncoding; + ResultMetadataCache::Snapshot metadata; + uint64_t epoch; + Py_ssize_t nextIndex = 0; + int charCtype = SQL_C_WCHAR; + bool manyEligible = false; + + explicit Plan(py::dict dict, uint64_t epoch) : cursorDict(std::move(dict)), epoch(epoch) {} + + py::handle at(Input index) const { + return PyTuple_GET_ITEM(inputs.ptr(), index); + } +}; + +struct ActiveCall; + +#ifdef MSSQL_NATIVE_FETCH_TEST_FAULTS +struct TestFaults { + int armed; + int hit; + size_t hits; + size_t visits[5]; + size_t statesAllocated; + size_t statesCleared; + size_t statesDeallocated; + size_t holdersAllocated; + size_t holdersDeleted; + size_t publications; + size_t positiveClears; + size_t guardedPositiveClears; +}; +#endif + +struct Context { + PyObject_HEAD + PyObject* cursorType; + PyObject* functions; + PyObject* dictDescriptor; + PyObject* stateType; + PyObject* stateKey; + PyObject* runtime; + ActiveCall* active; + bool disabled; +#ifdef MSSQL_NATIVE_FETCH_TEST_FAULTS + TestFaults faults; +#endif +}; + +struct State { + PyObject_HEAD + PyObject* context; + PyObject* owner; + PyObject* dict; + std::shared_ptr* plan; +#ifdef MSSQL_NATIVE_FETCH_TEST_FAULTS + Context* testContext; // Non-owning: the call, then State.context, keeps it alive. +#endif +}; + +inline int context_traverse(Context* self, visitproc visit, void* arg) { + Py_VISIT(Py_TYPE(self)); + Py_VISIT(self->cursorType); + Py_VISIT(self->functions); + Py_VISIT(self->dictDescriptor); + Py_VISIT(self->stateType); + Py_VISIT(self->stateKey); + Py_VISIT(self->runtime); + return 0; +} + +inline int context_clear(Context* self) { + auto* cursorType = self->cursorType; + auto* functions = self->functions; + auto* descriptor = self->dictDescriptor; + auto* stateType = self->stateType; + auto* stateKey = self->stateKey; + auto* runtime = self->runtime; + self->cursorType = self->functions = self->dictDescriptor = nullptr; + self->stateType = self->stateKey = self->runtime = nullptr; + Py_XDECREF(runtime); + Py_XDECREF(functions); + Py_XDECREF(descriptor); + Py_XDECREF(cursorType); + Py_XDECREF(stateKey); + Py_XDECREF(stateType); + return 0; +} + +inline int state_traverse(State* self, visitproc visit, void* arg) { + Py_VISIT(Py_TYPE(self)); + Py_VISIT(self->context); + Py_VISIT(self->owner); + Py_VISIT(self->dict); + if (self->plan && *self->plan) { + Py_VISIT((*self->plan)->cursorDict.ptr()); + Py_VISIT((*self->plan)->inputs.ptr()); + } + return 0; +} + +inline int state_clear(State* self) { + auto* plan = self->plan; + auto* context = self->context; + auto* owner = self->owner; + auto* dict = self->dict; +#ifdef MSSQL_NATIVE_FETCH_TEST_FAULTS + if (self->testContext) { + auto& faults = self->testContext->faults; + ++faults.statesCleared; + if (plan) + ++faults.holdersDeleted; + if (plan && *plan && !(*plan)->inputs.is_none()) { + ++faults.positiveClears; + if (self->testContext->active) + ++faults.guardedPositiveClears; + } + self->testContext = nullptr; + } +#endif + self->plan = nullptr; + self->context = self->owner = self->dict = nullptr; + // All associations are detached before any last-decref can reenter. + delete plan; + Py_XDECREF(dict); + Py_XDECREF(owner); + Py_XDECREF(context); + return 0; +} + +template +inline void dealloc(T* self) { + PyObject_GC_UnTrack(self); + Clear(self); + auto* type = Py_TYPE(self); + type->tp_free(reinterpret_cast(self)); + Py_DECREF(type); +} + +#ifdef MSSQL_NATIVE_FETCH_TEST_FAULTS +inline void state_dealloc(State* self) { + // A prior GC clear already accounted for that ownership release. Direct + // failure-unwind deallocations reach here with the context still attached. + if (self->testContext) + ++self->testContext->faults.statesDeallocated; + dealloc(self); +} +#endif + +struct ActiveCall { + Context* context; + py::object contextOwner; + py::object receiver; + ActiveCall* previous = nullptr; + ActiveCall* next; + + ActiveCall(Context* context, PyObject* receiver) + : context(context), + contextOwner(py::reinterpret_borrow( + reinterpret_cast(context))), + receiver(py::reinterpret_borrow(receiver)), + next(context->active) { + if (next) + next->previous = this; + context->active = this; + } + + ~ActiveCall() { + // SQL releases the GIL: overlapping calls need not leave in LIFO order. + if (previous) + previous->next = next; + else + context->active = next; + if (next) + next->previous = previous; + } + + ActiveCall(const ActiveCall&) = delete; + ActiveCall& operator=(const ActiveCall&) = delete; +}; + +inline bool active(Context* context, PyObject* receiver) { + for (auto* call = context->active; call; call = call->next) { + if (call->receiver.ptr() == receiver) + return true; + } + return false; +} + +inline py::dict instance_dict(PyObject* object) { + return py::reinterpret_steal(PyObject_GenericGetDict(object, nullptr)); +} + +template +inline PyObject* exact_item(PyObject* dict, const char (&name)[N]) { + constexpr auto length = static_cast(N - 1); + PyObject* key; + PyObject* value; + Py_ssize_t position = 0; + while (PyDict_Next(dict, &position, &key, &value)) { + if (PyUnicode_CheckExact(key) && PyUnicode_GET_LENGTH(key) == length) { + const int comparison = PyUnicode_CompareWithASCIIString(key, name); + if (PyErr_Occurred()) + throw py::error_already_set(); + if (comparison == 0) + return value; + } + } + return nullptr; +} + +struct StateEntry { + PyObject* value = nullptr; + bool safe; +}; + +inline StateEntry state_entry(PyObject* dict) { + StateEntry result{nullptr, PyDict_CheckExact(dict) != 0}; + PyObject* key; + PyObject* value; + Py_ssize_t position = 0; + // No hashing/equality calls, including for dictionary subclasses. + while (PyDict_Next(dict, &position, &key, &value)) { + if (!PyUnicode_CheckExact(key)) { + result.safe = false; + } else if (PyUnicode_GET_LENGTH(key) == + static_cast(sizeof("_native_fetch_state") - 1)) { + const int comparison = PyUnicode_CompareWithASCIIString(key, "_native_fetch_state"); + if (PyErr_Occurred()) + throw py::error_already_set(); + if (comparison == 0) + result.value = value; + } + } + return result; +} + +inline bool owned_by(State* state, Context* context, PyObject* receiver) { + if (state->context != reinterpret_cast(context) || !state->owner) + return false; +#if PY_VERSION_HEX >= 0x030D0000 + PyObject* owner = nullptr; + const int result = PyWeakref_GetRef(state->owner, &owner); + if (result < 0) + throw py::error_already_set(); + auto ownerReference = py::reinterpret_steal(owner); + return owner == receiver; +#else + // The GIL and the call's strong receiver reference cover this borrowed read. + PyObject* owner = PyWeakref_GetObject(state->owner); + if (!owner) + throw py::error_already_set(); + return owner == receiver; +#endif +} + +inline bool associated(State* state, Context* context, PyObject* receiver, PyObject* dict) { + return state->dict == dict && owned_by(state, context, receiver) && + state_entry(dict).value == reinterpret_cast(state); +} + +inline bool storage_admissible(Context* context, PyObject* receiver, PyObject* dict, + const StateEntry& entry) { + if (!entry.safe || reinterpret_cast(Py_TYPE(receiver)) != context->cursorType || + PyType_GetSlot(Py_TYPE(receiver), Py_tp_getattro) != + reinterpret_cast(PyObject_GenericGetAttr) || + PyType_GetSlot(Py_TYPE(receiver), Py_tp_setattro) != + reinterpret_cast(PyObject_GenericSetAttr)) + return false; + auto attributes = steal(PyType_GetDict(Py_TYPE(receiver))); + if (!attributes) + throw py::error_already_set(); + if (exact_item(attributes.ptr(), "__dict__") != context->dictDescriptor || + exact_item(attributes.ptr(), "_native_fetch_state")) + return false; + auto* handle = exact_item(dict, "hstmt"); + auto* index = exact_item(dict, "_next_row_index"); + auto* messages = exact_item(dict, "messages"); + return exact_item(dict, "closed") == Py_False && handle && handle != Py_None && + exact_item(dict, "_connection") && index && PyLong_CheckExact(index) && + messages && PyList_CheckExact(messages); +} + +#ifdef MSSQL_NATIVE_FETCH_TEST_FAULTS +inline bool test_fault(Context* context, int stage) { + auto& faults = context->faults; + ++faults.visits[stage - 1]; + if (faults.armed != stage) + return false; + faults.armed = 0; + faults.hit = stage; + ++faults.hits; + return true; +} + +inline void test_before_publication(Context* context, PyObject* dict) { + if (!test_fault(context, 5)) + return; + auto* callback = exact_item(dict, "_native_fetch_test_callback"); + if (!callback || !PyCallable_Check(callback)) + throw py::type_error("Native-fetch test boundary requires a callback"); + auto owner = py::reinterpret_borrow(callback); + auto result = steal(PyObject_CallNoArgs(owner.ptr())); + if (!result) + throw py::error_already_set(); +} +#endif + +inline py::object acquire_state(Context* context, PyObject* receiver, const py::dict& dict, + bool& needsRevalidation) { + needsRevalidation = true; + auto entry = state_entry(dict.ptr()); + if (entry.value) { + if (reinterpret_cast(Py_TYPE(entry.value)) != context->stateType) + return {}; + auto* state = reinterpret_cast(entry.value); + if (owned_by(state, context, receiver)) { + if (state->dict == dict.ptr()) { + // No allocation, publication or last-owner release on this return path. + needsRevalidation = false; + return py::reinterpret_borrow(entry.value); + } + } else if (state->context || state->owner || state->dict || state->plan) { + return {}; + } + } + if (!storage_admissible(context, receiver, dict.ptr(), entry)) + return {}; + + auto old = py::reinterpret_borrow(entry.value); + PyObject* weakOwner; +#ifdef MSSQL_NATIVE_FETCH_TEST_FAULTS + if (test_fault(context, 1)) + weakOwner = PyErr_NoMemory(); + else +#endif + weakOwner = PyWeakref_NewRef(receiver, nullptr); + auto owner = steal(weakOwner); + if (!owner) + throw py::error_already_set(); + PyObject* allocated; +#ifdef MSSQL_NATIVE_FETCH_TEST_FAULTS + if (test_fault(context, 2)) + allocated = PyErr_NoMemory(); + else +#endif + allocated = PyType_GenericAlloc( + reinterpret_cast(context->stateType), 0); + auto result = steal(allocated); + if (!result) + throw py::error_already_set(); + auto* state = reinterpret_cast(result.ptr()); +#ifdef MSSQL_NATIVE_FETCH_TEST_FAULTS + state->testContext = context; + ++context->faults.statesAllocated; + if (test_fault(context, 3)) + throw std::bad_alloc(); +#endif + state->plan = new std::shared_ptr(); +#ifdef MSSQL_NATIVE_FETCH_TEST_FAULTS + ++context->faults.holdersAllocated; +#endif + state->context = Py_NewRef(reinterpret_cast(context)); + state->owner = owner.release().ptr(); + state->dict = Py_NewRef(dict.ptr()); + +#ifdef MSSQL_NATIVE_FETCH_TEST_FAULTS + test_before_publication(context, dict.ptr()); +#endif + // Allocation can run GC callbacks. Revalidate before a callback-free write, + // retaining the old value so publication cannot release its last reference. + auto now = instance_dict(receiver); + if (!now) + throw py::error_already_set(); + entry = state_entry(dict.ptr()); + if (!now.is(dict) || entry.value != old.ptr() || + !storage_admissible(context, receiver, dict.ptr(), entry)) + return {}; + int published; +#ifdef MSSQL_NATIVE_FETCH_TEST_FAULTS + if (test_fault(context, 4)) { + PyErr_NoMemory(); + published = -1; + } else +#endif + published = PyDict_SetItem(dict.ptr(), context->stateKey, result.ptr()); + if (published < 0) + throw py::error_already_set(); +#ifdef MSSQL_NATIVE_FETCH_TEST_FAULTS + ++context->faults.publications; +#endif + return result; +} + +inline py::object fallback(PyObject* self, const py::tuple& rt, Mode mode, + PyObject* const* args, Py_ssize_t nargs, PyObject* keywords) { + PERF_TIMER("NativeCursor::fallback"); + auto functions = py::reinterpret_borrow(rt[3]); + if (mode == Mode::One && nargs == 0 && !keywords) { + PyObject* receiver = self; + return steal(PyObject_Vectorcall(functions[static_cast(mode)].ptr(), + &receiver, 1, keywords)); + } + // Only fetchmany accepts arguments; vectorcall forwards invalid shapes to + // the original Python signature rather than silently normalizing them. + const Py_ssize_t keywordCount = keywords ? PyTuple_GET_SIZE(keywords) : 0; + std::vector forwarded; + forwarded.reserve(static_cast(nargs + keywordCount + 1)); + forwarded.push_back(self); + for (Py_ssize_t i = 0; i < nargs + keywordCount; ++i) + forwarded.push_back(args[i]); + return steal(PyObject_Vectorcall(functions[static_cast(mode)].ptr(), + forwarded.data(), static_cast(nargs + 1), keywords)); +} + +inline std::shared_ptr prepare(PyObject* self, const py::tuple& rt, py::dict dict, + uint64_t epoch) { + PERF_TIMER("NativeCursor::prepare"); + auto plan = std::make_shared(std::move(dict), epoch); + py::object values = rt[2](py::handle(self), plan->cursorDict); + if (!values.is_none()) { + if (!PyTuple_CheckExact(values.ptr()) || PyTuple_GET_SIZE(values.ptr()) != InputCount) + throw py::type_error("Invalid native fetch plan"); + plan->inputs = std::move(values); + if (!PyDict_CheckExact(plan->at(CursorDict).ptr()) || + !PyDict_CheckExact(plan->at(ConnectionDict).ptr()) || + !PyDict_CheckExact(plan->at(ColumnMap).ptr()) || + !PyTuple_CheckExact(plan->at(Names).ptr()) || + !PyType_Check(plan->at(RowType).ptr()) || + !PyList_CheckExact(plan->at(Description).ptr()) || + !PyList_CheckExact(plan->at(Messages).ptr()) || + !PyType_Check(plan->at(ConnectionType).ptr()) || + !PyLong_CheckExact(plan->at(NextIndex).ptr()) || + !PyLong_CheckExact(plan->at(ArraySize).ptr()) || + !PyCallable_Check(plan->at(Factory).ptr())) + throw py::type_error("Invalid native fetch plan fields"); + // current() relies on this watch, not on the Python policy registering it. + auto* observer = static_cast( + PyCapsule_GetPointer(rt[0].ptr(), "mssql_python.native_fetch.Observer")); + if (!observer || PyDict_Watch(observer->dict, plan->cursorDict.ptr()) < 0) + throw py::error_already_set(); + plan->handle = plan->at(Handle).cast(); + plan->charEncoding = plan->at(CharEncoding).cast(); + plan->wideEncoding = plan->at(WideEncoding).cast(); + plan->charCtype = EffectiveCharCtypeForFetch( + plan->at(CharCtype).cast(), plan->charEncoding); + plan->nextIndex = plan->at(NextIndex).cast(); + plan->metadata = plan->handle->resultMetadata.snapshot(); + auto metadata = plan->metadata.metadata; + plan->manyEligible = metadata && metadata->completeResult && metadata->namesValidated; + if (plan->manyEligible) { + for (const auto& column : metadata->columns) { + if (IsLobOrVariantColumn(column.dataType, column.columnSize)) { + plan->manyEligible = false; + break; + } + } + } + } + return plan; +} + +inline bool current(PyObject* self, Context* context, State* state, const Plan& plan, + const py::tuple& rt) { + if (plan.epoch != mutationEpoch.load(std::memory_order_relaxed) || + !state->plan || state->plan->get() != &plan || + reinterpret_cast(Py_TYPE(self)) != rt[1].ptr() || + reinterpret_cast(Py_TYPE(plan.at(Connection).ptr())) != + plan.at(ConnectionType).ptr()) + return false; + auto dict = instance_dict(self); + auto connectionDict = instance_dict(plan.at(Connection).ptr()); + if (!dict || !connectionDict) + throw py::error_already_set(); + // fetch() checks the key association before use. The native dictionary watch + // invalidates this epoch on a later state-key change; replacement is checked here. + return dict.is(plan.cursorDict) && connectionDict.is(plan.at(ConnectionDict)) && + state->dict == dict.ptr() && owned_by(state, context, self) && + plan.epoch == mutationEpoch.load(std::memory_order_relaxed); +} + +inline void set_counter(Plan& plan, py::handle name, py::handle value) { + struct InternalWrite { + PyObject* dict = internalDict; + PyObject* key = internalKey; + PyObject* value = internalValue; + ~InternalWrite() { + internalDict = dict; + internalKey = key; + internalValue = value; + } + } restore; + internalDict = plan.cursorDict.ptr(); + internalKey = name.ptr(); + internalValue = value.ptr(); + if (PyDict_SetItem(plan.cursorDict.ptr(), name.ptr(), value.ptr()) < 0) + throw py::error_already_set(); +} + +inline py::object finish_python(PyObject* self, Mode mode, const py::list& values, + SQLRETURN ret, bool positionCommitted) { + PERF_TIMER("NativeCursor::completion_fallback"); + py::object cursor = py::reinterpret_borrow(self); + if (mode == Mode::Many) { + py::list rows; + if (SQL_SUCCEEDED(ret)) + rows.append(values); + if (positionCommitted) + return cursor.attr("_wrap_fetchmany")(rows); + return cursor.attr("_complete_fetchmany")(rows, ret); + } + py::object row = positionCommitted ? cursor.attr("_wrap_fetchone")(values) + : cursor.attr("_complete_fetchone")(values, ret); + if (mode == Mode::Value && !row.is_none()) + return row[py::int_(0)]; + if (mode == Mode::Next && row.is_none()) + throw py::stop_iteration(); + return row; +} + +inline bool conversion_ineligible(py::handle data) { + if (!PyDict_CheckExact(data.ptr())) + return false; + PyObject* converters = nullptr; + PyObject* uuidIndices = Py_None; + PyObject* key; + PyObject* value; + Py_ssize_t position = 0; + // A lookup can invoke equality on a stored key. Inspect entries without + // hashing, and defer every uncertain key shape to the original policy. + while (PyDict_Next(data.ptr(), &position, &key, &value)) { + if (!PyUnicode_CheckExact(key)) + return false; + const Py_ssize_t length = PyUnicode_GET_LENGTH(key); + if (length == static_cast(sizeof("_cached_converter_map") - 1) && + PyUnicode_CompareWithASCIIString(key, "_cached_converter_map") == 0) + converters = value; + else if (length == static_cast(sizeof("_uuid_str_indices") - 1) && + PyUnicode_CompareWithASCIIString(key, "_uuid_str_indices") == 0) + uuidIndices = value; + if (PyErr_Occurred()) + throw py::error_already_set(); + } + return !converters || !PyTuple_CheckExact(converters) || + PyTuple_GET_SIZE(converters) != 0 || uuidIndices != Py_None; +} + +inline bool explicit_nonunit_size(PyObject* const* args, Py_ssize_t nargs, + PyObject* keywords) { + const Py_ssize_t count = keywords ? PyTuple_GET_SIZE(keywords) : 0; + if (nargs == 0 && count == 1) { + PyObject* key = PyTuple_GET_ITEM(keywords, 0); + if (!PyUnicode_CheckExact(key)) + return false; + const int comparison = PyUnicode_CompareWithASCIIString(key, "size"); + if (PyErr_Occurred()) + throw py::error_already_set(); + if (comparison != 0) + return false; + } else if (nargs != 1 || count != 0) { + return false; + } + if (!PyLong_CheckExact(args[0])) + return false; + int overflow = 0; + const long size = PyLong_AsLongAndOverflow(args[0], &overflow); + if (PyErr_Occurred()) + throw py::error_already_set(); + return overflow || size != 1; +} + +inline bool single_size(const Plan& plan, PyObject* const* args, Py_ssize_t nargs, + PyObject* keywords) { + const Py_ssize_t count = keywords ? PyTuple_GET_SIZE(keywords) : 0; + if (nargs + count > 1) + return false; + if (count) { + PyObject* key = PyTuple_GET_ITEM(keywords, 0); + if (!PyUnicode_CheckExact(key)) + return false; + const int comparison = PyUnicode_CompareWithASCIIString(key, "size"); + if (PyErr_Occurred()) + throw py::error_already_set(); + if (comparison != 0) + return false; + } + PyObject* size = nargs + count ? args[0] : Py_None; + if (size == Py_None) + size = plan.at(ArraySize).ptr(); + if (!PyLong_CheckExact(size)) + return false; + int overflow = 0; + const long value = PyLong_AsLongAndOverflow(size, &overflow); + if (PyErr_Occurred()) + throw py::error_already_set(); + return !overflow && value == 1; +} + +inline py::object fetch(PyObject* self, Context* context, Mode mode, + PyObject* const* args = nullptr, + Py_ssize_t nargs = 0, PyObject* keywords = nullptr) { + if (active(context, self) || + reinterpret_cast(Py_TYPE(self)) != context->cursorType) { + auto rt = py::reinterpret_borrow(context->runtime); + return fallback(self, rt, mode, args, nargs, keywords); + } + // Declared first: retain the guard through every local owner/Plan release. + ActiveCall call(context, self); + auto rt = py::reinterpret_borrow(context->runtime); + auto* object = self; + + if (mode == Mode::Many && explicit_nonunit_size(args, nargs, keywords)) + return fallback(self, rt, mode, args, nargs, keywords); + + py::dict dict = instance_dict(object); + if (!dict) + throw py::error_already_set(); + bool needsRevalidation; + auto stateOwner = acquire_state(context, self, dict, needsRevalidation); + auto* state = reinterpret_cast(stateOwner.ptr()); + auto admittedDict = needsRevalidation ? instance_dict(self) : dict; + if (!admittedDict) + throw py::error_already_set(); + if (!admittedDict.is(dict) || + (needsRevalidation && state && !associated(state, context, self, dict.ptr()))) + return fallback(self, rt, mode, args, nargs, keywords); + const uint64_t epoch = mutationEpoch.load(std::memory_order_relaxed); + auto plan = state && state->plan ? *state->plan : nullptr; + bool stale = !plan || plan->epoch != epoch || !plan->cursorDict.is(dict); + if (!stale && !plan->inputs.is_none()) { + stale = !current(self, context, state, *plan, rt); + } + if (mode == Mode::One && stale && plan && plan->inputs.is_none() && + plan->cursorDict.is(dict) && conversion_ineligible(dict)) + return fallback(self, rt, mode, args, nargs, keywords); + if (stale) { + plan = prepare(self, rt, dict, epoch); + auto preparedDict = instance_dict(self); + if (!preparedDict) + throw py::error_already_set(); + if (state && state->plan && preparedDict.is(dict) && + associated(state, context, self, dict.ptr())) { + auto old = std::move(*state->plan); + *state->plan = plan; + old.reset(); + } + } + if (!state || plan->inputs.is_none() || !current(self, context, state, *plan, rt)) + return fallback(self, rt, mode, args, nargs, keywords); + + if (mode == Mode::Many) { + if (!single_size(*plan, args, nargs, keywords) || !plan->manyEligible) + return fallback(self, rt, mode, args, nargs, keywords); + auto metadata = plan->handle->resultMetadata.snapshot(); + if (metadata.generation != plan->metadata.generation || + metadata.metadata != plan->metadata.metadata) + return fallback(self, rt, mode, args, nargs, keywords); + } + if (mode == Mode::Value && PyList_GET_SIZE(plan->at(Description).ptr()) == 0) + return fallback(self, rt, mode, args, nargs, keywords); + + PERF_TIMER("NativeCursor::fetch"); + py::list values; + SQLRETURN ret = FetchSingleRow(plan->handle, values, plan->charEncoding, + plan->wideEncoding, plan->charCtype, plan->at(Messages)); + // A codec, finalizer, another thread or reentrant callback can mutate state + // while SQLFetch releases the GIL. Complete the consumed row, never refetch it. + if (!current(self, context, state, *plan, rt) || + (!SQL_SUCCEEDED(ret) && ret != SQL_NO_DATA)) + return finish_python(self, mode, values, ret, false); + if (ret == SQL_NO_DATA) { + if (mode == Mode::Many) + return py::list(); + if (mode == Mode::Next) + throw py::stop_iteration(); + return py::none(); + } + if (plan->nextIndex == std::numeric_limits::max()) + return finish_python(self, mode, values, ret, false); + py::int_ next(plan->nextIndex + 1); + py::int_ rowNumber(plan->nextIndex); + if (!current(self, context, state, *plan, rt)) + return finish_python(self, mode, values, ret, false); + auto counterNames = py::reinterpret_borrow(rt[5]); + set_counter(*plan, counterNames[0], next); + set_counter(*plan, counterNames[1], rowNumber); + set_counter(*plan, counterNames[2], next); + ++plan->nextIndex; + if (!current(self, context, state, *plan, rt)) + return finish_python(self, mode, values, ret, true); + + if (mode == Mode::Value) { + // Decode every column, not just column zero. Eligible plans have no + // converters, UUID processing or customized Row/fetchone behavior. + return values[py::int_(0)]; + } + if (mode == Mode::One) { + auto names = py::reinterpret_borrow(rt[4]); + return RowFactory::construct_row( + py::reinterpret_borrow(plan->at(RowType)), values, + plan->at(ColumnMap), object, plan->at(LowerMap), plan->at(Names), names); + } + if (mode == Mode::Many) { + auto names = py::reinterpret_borrow(rt[4]); + py::object row = RowFactory::construct_row( + py::reinterpret_borrow(plan->at(RowType)), values, + plan->at(ColumnMap), object, plan->at(LowerMap), plan->at(Names), names); + py::list rows(1); + PyList_SET_ITEM(rows.ptr(), 0, row.release().ptr()); + return rows; + } + PyObject* rowArgs[] = {values.ptr(), plan->at(ColumnMap).ptr(), object, + plan->at(LowerMap).ptr(), plan->at(Names).ptr()}; + // Keep the Python factory's allocation/assignment semantics for iteration. + return steal(PyObject_Vectorcall(plan->at(Factory).ptr(), rowArgs, 5, nullptr)); +} + +template +inline PyObject* method_entry(PyObject* owner, PyObject* const* args, Py_ssize_t count, + PyObject* keywords) { + try { + auto* context = reinterpret_cast(owner); + if (!context->functions) + throw std::runtime_error("Native fetch context is not initialized"); + if (context->runtime && count > 0 && + PyObject_TypeCheck(args[0], reinterpret_cast(context->cursorType)) && + (mode == Mode::Many || + (count == 1 && (!keywords || PyTuple_GET_SIZE(keywords) == 0)))) { + return fetch(args[0], context, mode, args + 1, + count - 1, keywords).release().ptr(); + } + // Unbound keyword-self calls and duck-typed receivers keep the original + // Python signature. Ordinary bound calls reach fetch directly. + auto functions = py::reinterpret_borrow(context->functions); + return PyObject_Vectorcall(functions[static_cast(mode)].ptr(), args, + static_cast(count), keywords); + } catch (...) { + py::detail::try_translate_exceptions(); + return nullptr; + } +} + +inline PyObject* internal_new(PyTypeObject*, PyObject*, PyObject*) { + PyErr_SetString(PyExc_TypeError, "Native fetch ownership objects are created internally"); + return nullptr; +} + +inline PyObject* begin_install(PyObject* owner, PyObject* args) { + try { + auto* context = reinterpret_cast(owner); + if (context->disabled) + throw std::runtime_error("Native fetch installation is disabled after incomplete rollback"); + if (context->cursorType || context->active) + throw std::runtime_error("Native fetch context is already installed"); + PyObject* cursorType; + PyObject* functions; + if (!PyArg_ParseTuple(args, "OO", &cursorType, &functions)) + return nullptr; + if (!PyType_CheckExact(cursorType) || !PyTuple_CheckExact(functions) || + PyTuple_GET_SIZE(functions) != 4) + throw py::type_error("Invalid native fetch installation"); + auto bases = steal(PyObject_GetAttrString(cursorType, "__bases__")); + if (!bases) + throw py::error_already_set(); + if (!PyTuple_CheckExact(bases.ptr()) || PyTuple_GET_SIZE(bases.ptr()) != 1 || + PyTuple_GET_ITEM(bases.ptr(), 0) != reinterpret_cast(&PyBaseObject_Type)) + throw py::type_error("Native fetch requires a plain Python cursor"); + for (Py_ssize_t i = 0; i < 4; ++i) { + if (!PyFunction_Check(PyTuple_GET_ITEM(functions, i))) + throw py::type_error("Native fetch fallbacks must be Python functions"); + } + auto attributes = steal(PyType_GetDict(reinterpret_cast(cursorType))); + if (!attributes) + throw py::error_already_set(); + auto* descriptor = exact_item(attributes.ptr(), "__dict__"); + if (!descriptor || Py_TYPE(descriptor) != &PyGetSetDescr_Type) + throw py::type_error("Native fetch requires an ordinary cursor dictionary"); + context->cursorType = Py_NewRef(cursorType); + context->functions = Py_NewRef(functions); + context->dictDescriptor = Py_NewRef(descriptor); + Py_RETURN_NONE; + } catch (...) { + py::detail::try_translate_exceptions(); + return nullptr; + } +} + +inline PyObject* activate_install(PyObject* owner, PyObject* runtime) { + auto* context = reinterpret_cast(owner); + if (context->disabled || !context->cursorType || context->runtime || + !PyTuple_CheckExact(runtime) || PyTuple_GET_SIZE(runtime) != 7 || + PyTuple_GET_ITEM(runtime, 1) != context->cursorType || + PyTuple_GET_ITEM(runtime, 3) != context->functions || + PyTuple_GET_ITEM(runtime, 6) != owner || + !PyCallable_Check(PyTuple_GET_ITEM(runtime, 2))) { + PyErr_SetString(PyExc_TypeError, "Invalid native fetch runtime schema 2"); + return nullptr; + } + const char* rowNames[] = {"_values", "_column_map", "_cursor", + "_column_map_lower", "_column_names"}; + const char* counterNames[] = {"_next_row_index", "_rownumber", "rowcount"}; + for (Py_ssize_t field = 4; field <= 5; ++field) { + auto* names = PyTuple_GET_ITEM(runtime, field); + const Py_ssize_t count = field == 4 ? 5 : 3; + if (!PyTuple_CheckExact(names) || PyTuple_GET_SIZE(names) != count) { + PyErr_SetString(PyExc_TypeError, "Invalid native fetch field names"); + return nullptr; + } + for (Py_ssize_t i = 0; i < count; ++i) { + auto* name = PyTuple_GET_ITEM(names, i); + if (!PyUnicode_CheckExact(name) || + PyUnicode_CompareWithASCIIString( + name, field == 4 ? rowNames[i] : counterNames[i]) != 0) { + if (!PyErr_Occurred()) + PyErr_SetString(PyExc_TypeError, "Invalid native fetch field name"); + return nullptr; + } + } + } + context->runtime = Py_NewRef(runtime); + Py_RETURN_NONE; +} + +inline PyObject* abort_install(PyObject* owner, PyObject*) { + auto* context = reinterpret_cast(owner); + if (context->disabled || context->runtime || context->active) { + PyErr_SetString(PyExc_RuntimeError, "Cannot abort an active native fetch context"); + return nullptr; + } + auto* cursorType = context->cursorType; + auto* functions = context->functions; + auto* descriptor = context->dictDescriptor; + context->cursorType = context->functions = context->dictDescriptor = nullptr; + Py_XDECREF(functions); + Py_XDECREF(descriptor); + Py_XDECREF(cursorType); + Py_RETURN_NONE; +} + +inline PyObject* disable_install(PyObject* owner, PyObject*) { + auto* context = reinterpret_cast(owner); + if (context->runtime || context->active) { + PyErr_SetString(PyExc_RuntimeError, "Cannot disable an active native fetch context"); + return nullptr; + } + // Keep every forwarding owner: even descriptors restored on the class may + // have been saved elsewhere. Unpublished entries can never create State. + context->disabled = true; + Py_RETURN_NONE; +} + +#ifdef MSSQL_NATIVE_FETCH_TEST_FAULTS +inline PyObject* test_arm(PyObject* owner, PyObject* stage) { + auto* context = reinterpret_cast(owner); + if (!PyLong_CheckExact(stage)) { + PyErr_SetString(PyExc_TypeError, "Expected a native-fetch test stage integer"); + return nullptr; + } + const long value = PyLong_AsLong(stage); + if (value == -1 && PyErr_Occurred()) + return nullptr; + if (value < 0 || value > 5 || context->active || + (value && context->faults.armed)) { + PyErr_SetString(PyExc_RuntimeError, "Invalid or already active native-fetch test stage"); + return nullptr; + } + context->faults.armed = static_cast(value); + Py_RETURN_NONE; +} + +inline PyObject* test_snapshot(PyObject* owner, PyObject* receiver) { + try { + auto* context = reinterpret_cast(owner); + size_t activeCount = 0; + for (auto* call = context->active; call; call = call->next) + ++activeCount; + bool positive = false; + bool attached = false; + if (reinterpret_cast(Py_TYPE(receiver)) == context->cursorType) { + auto dict = instance_dict(receiver); + if (!dict) + throw py::error_already_set(); + auto* value = state_entry(dict.ptr()).value; + if (value && reinterpret_cast(Py_TYPE(value)) == context->stateType) { + auto* state = reinterpret_cast(value); + attached = associated(state, context, receiver, dict.ptr()); + positive = state->plan && *state->plan && !(*state->plan)->inputs.is_none(); + } + } + const auto& f = context->faults; + py::dict result; + result["schema"] = 1; + result["armed"] = f.armed; + result["hit"] = f.hit; + result["hits"] = f.hits; + result["visits"] = py::make_tuple(f.visits[0], f.visits[1], f.visits[2], + f.visits[3], f.visits[4]); + result["active"] = activeCount; + result["receiver_active"] = active(context, receiver); + result["states_allocated"] = f.statesAllocated; + result["states_cleared"] = f.statesCleared; + result["states_deallocated"] = f.statesDeallocated; + result["holders_allocated"] = f.holdersAllocated; + result["holders_deleted"] = f.holdersDeleted; + result["publications"] = f.publications; + result["positive_clears"] = f.positiveClears; + result["guarded_positive_clears"] = f.guardedPositiveClears; + result["associated"] = attached; + result["positive_plan"] = positive; + return result.release().ptr(); + } catch (...) { + py::detail::try_translate_exceptions(); + return nullptr; + } +} +#endif + +inline PyObject* invalidate(PyObject* owner, PyObject* receiver) { + try { + auto* context = reinterpret_cast(owner); + if (reinterpret_cast(Py_TYPE(receiver)) == context->cursorType) { + auto dict = instance_dict(receiver); + if (!dict) + throw py::error_already_set(); + auto* value = state_entry(dict.ptr()).value; + if (value && reinterpret_cast(Py_TYPE(value)) == context->stateType) { + auto stateOwner = py::reinterpret_borrow(value); + auto* state = reinterpret_cast(value); + if (state->plan && associated(state, context, receiver, dict.ptr())) { + auto old = std::move(*state->plan); + } + } + } + Py_RETURN_NONE; + } catch (...) { + py::detail::try_translate_exceptions(); + return nullptr; + } +} + +inline void bind(py::module_& m) { +#ifdef _MSC_VER +#pragma warning(push) +#pragma warning(disable: 4191) // PyMethodDef stores FASTCALL functions as PyCFunction. +#endif + static PyMethodDef methods[] = { + {"fetchone", reinterpret_cast(method_entry), + METH_FASTCALL | METH_KEYWORDS, + "fetchone(self)\n--\n\nFetch the next row without read-ahead."}, + {"fetchmany", reinterpret_cast(method_entry), + METH_FASTCALL | METH_KEYWORDS, + "fetchmany(self, size=None)\n--\n\nFetch the requested number of rows."}, + {"fetchval", reinterpret_cast(method_entry), + METH_FASTCALL | METH_KEYWORDS, + "fetchval(self)\n--\n\nFetch the first column of the next row."}, + {"__next__", reinterpret_cast(method_entry), + METH_FASTCALL | METH_KEYWORDS, + "__next__(self)\n--\n\nFetch the next row or raise StopIteration."}, + {nullptr, nullptr, 0, nullptr} + }; +#ifdef _MSC_VER +#pragma warning(pop) +#endif + static PyType_Slot stateSlots[] = { + {Py_tp_new, reinterpret_cast(internal_new)}, +#ifdef MSSQL_NATIVE_FETCH_TEST_FAULTS + {Py_tp_dealloc, reinterpret_cast(state_dealloc)}, +#else + {Py_tp_dealloc, reinterpret_cast(dealloc)}, +#endif + {Py_tp_traverse, reinterpret_cast(state_traverse)}, + {Py_tp_clear, reinterpret_cast(state_clear)}, + {0, nullptr} + }; + PyType_Spec stateSpec = {"ddbc_bindings._FetchState", sizeof(State), 0, + Py_TPFLAGS_DEFAULT | Py_TPFLAGS_HAVE_GC | + Py_TPFLAGS_IMMUTABLETYPE, + stateSlots}; + auto stateType = steal(PyType_FromSpec(&stateSpec)); + if (!stateType) + throw py::error_already_set(); + static PyType_Slot contextSlots[] = { + {Py_tp_new, reinterpret_cast(internal_new)}, + {Py_tp_dealloc, reinterpret_cast(dealloc)}, + {Py_tp_traverse, reinterpret_cast(context_traverse)}, + {Py_tp_clear, reinterpret_cast(context_clear)}, + {0, nullptr} + }; + PyType_Spec contextSpec = {"ddbc_bindings._FetchContext", sizeof(Context), 0, + Py_TPFLAGS_DEFAULT | Py_TPFLAGS_HAVE_GC | + Py_TPFLAGS_IMMUTABLETYPE, + contextSlots}; + auto contextType = steal(PyType_FromSpec(&contextSpec)); + if (!contextType) + throw py::error_already_set(); + auto contextOwner = steal(PyType_GenericAlloc( + reinterpret_cast(contextType.ptr()), 0)); + if (!contextOwner) + throw py::error_already_set(); + auto* context = reinterpret_cast(contextOwner.ptr()); + context->stateType = stateType.release().ptr(); + context->stateKey = PyUnicode_InternFromString("_native_fetch_state"); + if (!context->stateKey) + throw py::error_already_set(); + m.attr("_native_fetch_schema") = 2; +#ifdef MSSQL_NATIVE_FETCH_TEST_FAULTS + m.attr("_native_fetch_test_fault_schema") = 1; +#endif + m.attr("_native_fetch_context") = contextOwner; + py::tuple entries(4); + for (Py_ssize_t i = 0; i < 4; ++i) { + py::object function = steal(PyCFunction_NewEx(&methods[i], contextOwner.ptr(), nullptr)); + if (!function) + throw py::error_already_set(); + py::object descriptor = steal(PyInstanceMethod_New(function.ptr())); + if (!descriptor) + throw py::error_already_set(); + PyTuple_SET_ITEM(entries.ptr(), i, descriptor.release().ptr()); + } + m.attr("_native_fetch_methods") = std::move(entries); + static PyMethodDef controls[] = { + {"_begin_native_fetch_install", begin_install, METH_VARARGS, nullptr}, + {"_activate_native_fetch_install", activate_install, METH_O, nullptr}, + {"_abort_native_fetch_install", abort_install, METH_NOARGS, nullptr}, + {"_disable_native_fetch_install", disable_install, METH_NOARGS, nullptr}, + {"_invalidate_native_fetch", invalidate, METH_O, nullptr}, +#ifdef MSSQL_NATIVE_FETCH_TEST_FAULTS + {"_native_fetch_test_arm", test_arm, METH_O, nullptr}, + {"_native_fetch_test_snapshot", test_snapshot, METH_O, nullptr}, +#endif + {nullptr, nullptr, 0, nullptr} + }; + for (auto* method = controls; method->ml_name; ++method) { + auto function = steal(PyCFunction_NewEx(method, contextOwner.ptr(), nullptr)); + if (!function) + throw py::error_already_set(); + m.attr(method->ml_name) = std::move(function); + } + + auto observer = std::make_unique(); + auto unavailable = [&m]() { + PyErr_Clear(); + if (PyErr_WarnEx(PyExc_RuntimeWarning, + "Native fetch prototype disabled: CPython watcher slots unavailable", + 1) < 0) + throw py::error_already_set(); + m.attr("_fetch_observer") = py::none(); + }; + observer->dict = PyDict_AddWatcher(dict_changed); + if (observer->dict < 0) { + unavailable(); + return; + } + observer->type = PyType_AddWatcher(type_changed); + if (observer->type < 0) { + PyDict_ClearWatcher(observer->dict); + unavailable(); + return; + } + observer->function = PyFunction_AddWatcher(function_changed); + if (observer->function < 0) { + PyType_ClearWatcher(observer->type); + PyDict_ClearWatcher(observer->dict); + unavailable(); + return; + } + // One registration per native-module lifetime. The callbacks retain no + // watched objects; CPython removes registrations at interpreter teardown. + m.attr("_fetch_observer") = py::capsule( + observer.release(), "mssql_python.native_fetch.Observer", [](void* pointer) { + delete static_cast(pointer); + }); + m.def("_fetch_conversion_ineligible", &conversion_ineligible); + m.def("_watch_fetch_dependencies", [](const py::capsule& capsule, + const py::tuple& dicts, const py::tuple& types) { + auto* observer = static_cast( + PyCapsule_GetPointer(capsule.ptr(), "mssql_python.native_fetch.Observer")); + if (!observer) + throw py::error_already_set(); + for (auto dict : dicts) { + if (PyDict_Watch(observer->dict, dict.ptr()) < 0) + throw py::error_already_set(); + } + for (auto type : types) { + if (!PyType_Check(type.ptr())) + throw py::type_error("Expected a type dependency"); + if (!(PyType_GetFlags(reinterpret_cast(type.ptr())) & + Py_TPFLAGS_IMMUTABLETYPE) && + PyType_Watch(observer->type, type.ptr()) < 0) + throw py::error_already_set(); + } + }); +} + +#else + +inline void bind(py::module_& m) { + m.attr("_fetch_observer") = py::none(); + m.def("_invalidate_native_fetch", [](py::handle) {}); +} + +#endif + +} // namespace NativeFetch diff --git a/mssql_python/pybind/result_metadata.hpp b/mssql_python/pybind/result_metadata.hpp index 3a9912860..bd36cdd9f 100644 --- a/mssql_python/pybind/result_metadata.hpp +++ b/mssql_python/pybind/result_metadata.hpp @@ -21,6 +21,8 @@ struct FetchColumnMetadata { struct ResultMetadata { std::vector columns; bool namesValidated = false; + // DDBCSQLGetData may publish only a prefix, which is not a result-set count. + bool completeResult = false; }; // Native-only ownership: cleanup/cancellation may run without the GIL. No ODBC, diff --git a/mssql_python/pybind/row_factory.hpp b/mssql_python/pybind/row_factory.hpp index 85d271128..ce7705f1d 100644 --- a/mssql_python/pybind/row_factory.hpp +++ b/mssql_python/pybind/row_factory.hpp @@ -7,11 +7,33 @@ namespace RowFactory { +// The caller validates the type. Both setters resolve inherited descriptors; +// fetchone selects ordinary assignment dispatch for mutations during construction. +template +inline py::object construct_row(const py::object& row_class, py::handle values, + py::handle column_map, py::handle cursor, + py::handle column_map_lower, py::handle column_names, + const py::tuple& attribute_names) { + auto* row_type = reinterpret_cast(row_class.ptr()); + py::object row = steal(row_type->tp_alloc(row_type, 0)); + if (!row) + throw py::error_already_set(); + PyObject* fields[] = {values.ptr(), column_map.ptr(), cursor.ptr(), + column_map_lower.ptr(), column_names.ptr()}; + for (Py_ssize_t i = 0; i < 5; ++i) { + if (SetAttribute(row.ptr(), PyTuple_GET_ITEM(attribute_names.ptr(), i), + fields[i]) < 0) + throw py::error_already_set(); + } + return row; +} + // Wrap fetched values without converter or UUID processing. // Accepts Row and its subclasses, bypassing __init__. inline py::list construct_rows(const py::list& rows_data, const py::object& row_class, const py::object& column_map, const py::object& cursor_obj, - const py::object& column_map_lower, const py::object& column_names) { + const py::object& column_map_lower, const py::object& column_names, + const py::tuple& attribute_names) { if (!PyType_Check(row_class.ptr())) { throw py::type_error("row_class must be a type"); } @@ -23,12 +45,12 @@ inline py::list construct_rows(const py::list& rows_data, const py::object& row_ } Py_ssize_t n = PyList_GET_SIZE(rows_data.ptr()); - // Keep Python-owned names local to this call and its interpreter. - py::str attr_values("_values"); - py::str attr_column_map("_column_map"); - py::str attr_cursor("_cursor"); - py::str attr_column_map_lower("_column_map_lower"); - py::str attr_column_names("_column_names"); + // Owned by this interpreter's binding, not by a process-static Python cache. + PyObject* attr_values = PyTuple_GET_ITEM(attribute_names.ptr(), 0); + PyObject* attr_column_map = PyTuple_GET_ITEM(attribute_names.ptr(), 1); + PyObject* attr_cursor = PyTuple_GET_ITEM(attribute_names.ptr(), 2); + PyObject* attr_column_map_lower = PyTuple_GET_ITEM(attribute_names.ptr(), 3); + PyObject* attr_column_names = PyTuple_GET_ITEM(attribute_names.ptr(), 4); py::list result(n); @@ -39,12 +61,12 @@ inline py::list construct_rows(const py::list& rows_data, const py::object& row_ PyObject* row_data = PyList_GET_ITEM(rows_data.ptr(), i); - if (PyObject_GenericSetAttr(row.ptr(), attr_values.ptr(), row_data) < 0 || - PyObject_GenericSetAttr(row.ptr(), attr_column_map.ptr(), column_map.ptr()) < 0 || - PyObject_GenericSetAttr(row.ptr(), attr_cursor.ptr(), cursor_obj.ptr()) < 0 || - PyObject_GenericSetAttr(row.ptr(), attr_column_map_lower.ptr(), + if (PyObject_GenericSetAttr(row.ptr(), attr_values, row_data) < 0 || + PyObject_GenericSetAttr(row.ptr(), attr_column_map, column_map.ptr()) < 0 || + PyObject_GenericSetAttr(row.ptr(), attr_cursor, cursor_obj.ptr()) < 0 || + PyObject_GenericSetAttr(row.ptr(), attr_column_map_lower, column_map_lower.ptr()) < 0 || - PyObject_GenericSetAttr(row.ptr(), attr_column_names.ptr(), column_names.ptr()) < 0) { + PyObject_GenericSetAttr(row.ptr(), attr_column_names, column_names.ptr()) < 0) { throw py::error_already_set(); } diff --git a/tests/test_007_logging.py b/tests/test_007_logging.py index e773524be..cfb9d9ab7 100644 --- a/tests/test_007_logging.py +++ b/tests/test_007_logging.py @@ -649,13 +649,13 @@ def test_driver_logger_is_same_as_internal(self, cleanup_logger): class TestThreadSafety: """Tests for thread safety and race condition fixes""" - def test_concurrent_initialization_no_double_init(self, cleanup_logger): + def test_concurrent_initialization_no_double_init(self, cleanup_logger, monkeypatch): """Test that concurrent __init__ calls don't cause double initialization""" import threading from mssql_python.logging import MSSQLLogger # Force re-creation by deleting singleton - MSSQLLogger._instance = None + monkeypatch.setattr(MSSQLLogger, "_instance", None) init_counts = [] errors = [] diff --git a/tests/test_008_logging_integration.py b/tests/test_008_logging_integration.py index 220cbdc6d..cd28a12f4 100644 --- a/tests/test_008_logging_integration.py +++ b/tests/test_008_logging_integration.py @@ -30,6 +30,8 @@ def cleanup_logger(): """Reset logger state and cleanup log files""" # Disable and clear logger._logger.setLevel(logging.CRITICAL) + logger._cached_level = logging.CRITICAL + logger._is_debug_enabled = False for handler in logger._logger.handlers[:]: handler.close() logger._logger.removeHandler(handler) @@ -44,6 +46,8 @@ def cleanup_logger(): # Cleanup after logger._logger.setLevel(logging.CRITICAL) + logger._cached_level = logging.CRITICAL + logger._is_debug_enabled = False for handler in logger._logger.handlers[:]: handler.close() logger._logger.removeHandler(handler) diff --git a/tests/test_011_connection_string_allowlist.py b/tests/test_011_connection_string_allowlist.py index 85f411742..4137a6fd8 100644 --- a/tests/test_011_connection_string_allowlist.py +++ b/tests/test_011_connection_string_allowlist.py @@ -213,7 +213,7 @@ def test__normalize_params_with_warnings(self): import os # Enable logging to capture the debug messages - from mssql_python.logging import setup_logging, driver_logger + from mssql_python.logging import setup_logging, driver_logger, logger # Create a temp log file with tempfile.NamedTemporaryFile(mode="w", suffix=".log", delete=False) as f: @@ -246,6 +246,8 @@ def test__normalize_params_with_warnings(self): driver_logger.removeHandler(handler) # Disable logging driver_logger.setLevel(logging.CRITICAL) + logger._cached_level = logging.CRITICAL + logger._is_debug_enabled = False # Clean up temp file if os.path.exists(log_file): os.remove(log_file) diff --git a/tests/test_native_fetch_dispatch.py b/tests/test_native_fetch_dispatch.py new file mode 100644 index 000000000..8d0b4afa0 --- /dev/null +++ b/tests/test_native_fetch_dispatch.py @@ -0,0 +1,1679 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. + +"""Mutation, lifetime and mechanism coverage for the native fetch prototype.""" + +import cProfile +from contextlib import contextmanager, nullcontext +import gc +from pathlib import Path +import subprocess +import sys +from unittest.mock import patch +import weakref + +import pytest + +import mssql_python +from mssql_python import ddbc_bindings +from mssql_python.cursor import Cursor +from mssql_python.row import Row + +METHODS = ("fetchone", "fetchmany", "fetchval", "iterator") +SELECT_FIVE = "SELECT n FROM (VALUES (1), (2), (3), (4), (5)) AS v(n) ORDER BY n" + + +@pytest.fixture +def connection(conn_str): + try: + conn = mssql_python.connect(conn_str, timeout=5) + except mssql_python.Error as error: + pytest.fail( + f"Connection failed: {type(error).__name__}; connection details withheld", + pytrace=False, + ) + try: + yield conn + finally: + conn.close() + + +def read(cursor, method): + if method == "fetchmany": + rows = cursor.fetchmany(1) + assert type(rows) is list + return rows[0] if rows else None + if method == "iterator": + return next(cursor, None) + return getattr(cursor, method)() + + +def value(row, method): + return row if method == "fetchval" or row is None else row[0] + + +def warm(cursor, method): + cursor.execute(SELECT_FIVE) + assert value(read(cursor, method), method) == 1 + assert value(read(cursor, method), method) == 2 + + +@contextmanager +def _native_fetch_counters(enabled=True): + profiling = getattr(ddbc_bindings, "profiling", None) if enabled else None + if profiling is not None: + assert not profiling.is_enabled() + profiling.reset() + profiling.enable() + try: + yield profiling + finally: + if profiling is not None: + profiling.disable() + + +@contextmanager +def _fetchone_window(profiling, enabled=True): + if not enabled: + yield None + return + if profiling is not None: + assert profiling.is_enabled() + before = profiling.get_stats() if profiling is not None else None + profile = cProfile.Profile() + observation = {} + profile.enable() + try: + yield observation + finally: + profile.disable() + observation["python"] = { + entry.code: entry.callcount + for entry in profile.getstats() + if not isinstance(entry.code, str) + } + after = profiling.get_stats() if profiling is not None else None + observation["native"] = ( + { + name: after.get(name, {}).get("calls", 0) - before.get(name, {}).get("calls", 0) + for name in before.keys() | after.keys() + } + if profiling is not None + else None + ) + if profiling is not None: + assert profiling.is_enabled() + assert all(count >= 0 for count in observation["native"].values()) + + +def _assert_fetchone_counts(observation, *, prepare, native, fallback, sql): + from mssql_python.cursor import _NATIVE_FETCH_RUNTIME + + calls = observation["python"] + assert calls.get(type(_NATIVE_FETCH_RUNTIME[2]).__call__.__code__, 0) == prepare + assert calls.get(_NATIVE_FETCH_RUNTIME[3][0].__code__, 0) == fallback + if observation["native"] is not None: + for name, expected in ( + ("NativeCursor::prepare", prepare), + ("NativeCursor::fetch", native), + ("NativeCursor::fallback", fallback), + ("NativeCursor::completion_fallback", 0), + ("FetchOne_wrap", fallback), + ("FetchSingleRow", sql), + ): + assert observation["native"].get("ddbc::" + name, 0) == expected, name + + +def test_repeated_negative_fetchone_skips_prepare(connection): + if ddbc_bindings._fetch_observer is None: + pytest.skip("Requires supported CPython watchers") + calls = [] + + def convert(raw): + calls.append(raw) + return raw * 10 + + connection.add_output_converter(mssql_python.SQL_INTEGER, convert) + try: + with _native_fetch_counters() as profiling, connection.cursor() as cursor: + cursor.execute(SELECT_FIVE) + assert cursor.fetchone()[0] == 10 + assert calls == [1] + with _fetchone_window(profiling) as repeated: + rows = [cursor.fetchone() for _ in range(5)] + assert [value(row, "fetchone") for row in rows] == [20, 30, 40, 50, None] + assert calls == [1, 2, 3, 4, 5] + assert repeated["python"].get(convert.__code__, 0) == 4 + assert repeated["python"].get(Cursor._complete_fetchone.__code__, 0) == 0 + assert repeated["python"].get(Cursor._wrap_fetchone.__code__, 0) == 0 + _assert_fetchone_counts(repeated, prepare=0, native=0, fallback=5, sql=5) + assert cursor.rowcount == 5 + assert cursor.rownumber == 4 + finally: + connection.remove_output_converter(mssql_python.SQL_INTEGER) + + +@pytest.mark.parametrize("method", METHODS) +def test_warm_native_dispatch_removes_python_control_calls(connection, method): + if ddbc_bindings._fetch_observer is None: + pytest.skip("Direct native dispatch requires GIL-enabled CPython 3.12+ watchers") + with _native_fetch_counters() as profiling, connection.cursor() as cursor: + cursor._get_sql_to_c_type_map() + cursor.setinputsizes([(mssql_python.SQL_INTEGER, 0, 0)]) + with pytest.warns(Warning, match="Number of input sizes"): + cursor.execute("SELECT ?, ?", 1, 2) + warm(cursor, method) + with _fetchone_window(profiling) as observation: + rows = [read(cursor, method) for _ in range(4)] + assert [value(row, method) for row in rows] == [3, 4, 5, None] + assert cursor.rowcount == 5 + assert cursor.rownumber == 4 + forbidden = { + Cursor._get_column_and_converter_maps.__code__, + Cursor._increment_rownumber.__code__, + Cursor._complete_fetchone.__code__, + Cursor._complete_fetchmany.__code__, + } + from mssql_python.cursor import _NATIVE_FETCH_RUNTIME + + forbidden.update(function.__code__ for function in _NATIVE_FETCH_RUNTIME[3]) + calls = observation["python"] + assert not forbidden.intersection(calls) + assert calls.get(Row._fast_create.__code__, 0) == (3 if method == "iterator" else 0) + if observation["native"] is not None: + counts = observation["native"] + assert counts["ddbc::NativeCursor::fetch"] == 4 + assert counts["ddbc::FetchSingleRow"] == 4 + for name in ( + "NativeCursor::prepare", + "NativeCursor::fallback", + "NativeCursor::completion_fallback", + "FetchOne_wrap", + "FetchMany_wrap", + "FetchBatchData", + "SQLBindColums", + ): + assert counts.get("ddbc::" + name, 0) == 0 + + +@pytest.mark.parametrize("method", METHODS) +@pytest.mark.parametrize("scope", ("instance", "class")) +def test_warm_map_override_and_restoration(connection, method, scope): + with connection.cursor() as cursor: + warm(cursor, method) + target = cursor if scope == "instance" else Cursor + with patch.object( + target, "_get_column_and_converter_maps", side_effect=RuntimeError("map override") + ): + with pytest.raises(RuntimeError, match="map override"): + read(cursor, method) + assert cursor.rowcount == 3 + assert cursor.rownumber == 2 + assert value(read(cursor, method), method) == 4 + assert value(read(cursor, method), method) == 5 + + if method == "fetchone" and scope == "instance" and ddbc_bindings._fetch_observer is not None: + connection.add_output_converter(mssql_python.SQL_INTEGER, lambda raw: raw) + try: + with _native_fetch_counters() as profiling, connection.cursor() as cursor: + warm(cursor, "fetchone") + with patch.object( + cursor, + "_get_column_and_converter_maps", + side_effect=RuntimeError("negative map override"), + ): + with _fetchone_window(profiling) as failed: + with pytest.raises(RuntimeError, match="negative map override"): + cursor.fetchone() + _assert_fetchone_counts(failed, prepare=0, native=0, fallback=1, sql=1) + assert cursor.rowcount == 3 + assert cursor.rownumber == 2 + assert cursor.fetchone()[0] == 4 + assert cursor.fetchone()[0] == 5 + finally: + connection.remove_output_converter(mssql_python.SQL_INTEGER) + + +@pytest.mark.parametrize("method", METHODS) +def test_warm_factory_override_keeps_consumed_position(connection, method): + with connection.cursor() as cursor: + warm(cursor, method) + target, name = ( + (ddbc_bindings, "construct_rows") if method == "fetchmany" else (Row, "_fast_create") + ) + with patch.object(target, name, side_effect=RuntimeError("factory override")): + with pytest.raises(RuntimeError, match="factory override"): + read(cursor, method) + assert cursor.rowcount == 3 + assert cursor.rownumber == 2 + assert value(read(cursor, method), method) == 4 + + +@pytest.mark.parametrize("method", ("fetchval", "iterator")) +def test_warm_fetchone_override_is_not_bypassed(connection, method): + with connection.cursor() as cursor: + warm(cursor, method) + original = cursor.fetchone + with patch.object(cursor, "fetchone", wraps=original) as override: + assert value(read(cursor, method), method) == 3 + override.assert_called_once_with() + assert value(read(cursor, method), method) == 4 + + +@pytest.mark.parametrize("method", METHODS) +def test_warm_dictionary_replacement_and_direct_counter_mutation(connection, method): + with connection.cursor() as cursor: + warm(cursor, method) + old = cursor.__dict__ + cursor.__dict__ = old.copy() + assert value(read(cursor, method), method) == 3 + assert old["rowcount"] == 2 + assert cursor.rowcount == 3 + cursor.__dict__["_next_row_index"] = 20 + assert value(read(cursor, method), method) == 4 + assert cursor.rowcount == 21 + assert cursor.rownumber == 20 + connection.__dict__ = connection.__dict__.copy() + assert value(read(cursor, method), method) == 5 + assert cursor.rowcount == 22 + + if method == "fetchone" and ddbc_bindings._fetch_observer is not None: + connection.add_output_converter(mssql_python.SQL_INTEGER, lambda raw: raw) + try: + with _native_fetch_counters() as profiling, connection.cursor() as cursor: + warm(cursor, "fetchone") + old = cursor.__dict__ + cursor.__dict__ = old.copy() + with _fetchone_window(profiling) as replacement: + assert cursor.fetchone()[0] == 3 + _assert_fetchone_counts(replacement, prepare=1, native=0, fallback=1, sql=1) + assert old["rowcount"] == 2 + assert cursor.rowcount == 3 + assert cursor.rownumber == 2 + finally: + connection.remove_output_converter(mssql_python.SQL_INTEGER) + + +@pytest.mark.parametrize("method", METHODS) +@pytest.mark.parametrize("target", ("cursor", "connection")) +def test_warm_dictionary_subclass_uses_python_fallback(connection, method, target): + class CustomDict(dict): + pass + + with connection.cursor() as cursor: + warm(cursor, method) + owner = cursor if target == "cursor" else connection + owner.__dict__ = CustomDict(owner.__dict__) + assert value(read(cursor, method), method) == 3 + assert cursor.rowcount == 3 + assert cursor.rownumber == 2 + assert value(read(cursor, method), method) == 4 + owner.__dict__ = dict(owner.__dict__) + assert value(read(cursor, method), method) == 5 + assert cursor.rowcount == 5 + assert cursor.rownumber == 4 + assert read(cursor, method) is None + + +def test_negative_fetchone_uncertain_dictionary_uses_full_policy(connection): + if ddbc_bindings._fetch_observer is None: + pytest.skip("Requires supported CPython watchers") + calls = [] + armed = False + + class Key: + def __hash__(self): + calls.append("hash") + return hash("_connection") + + def __eq__(self, other): + calls.append("eq") + if armed: + raise RuntimeError("uncertain dictionary policy") + return False + + class TextKey(str): + __hash__ = Key.__hash__ + __eq__ = Key.__eq__ + + class CustomDict(dict): + def get(self, *args, **kwargs): + calls.append("get") + if armed: + raise RuntimeError("uncertain dictionary policy") + return super().get(*args, **kwargs) + + connection.add_output_converter(mssql_python.SQL_INTEGER, lambda raw: raw) + try: + for shape in ("custom-key", "late-text-key", "dict-subclass"): + with _native_fetch_counters() as profiling, connection.cursor() as cursor: + if shape == "dict-subclass": + cursor.__dict__ = CustomDict(cursor.__dict__) + warm(cursor, "fetchone") + data = cursor.__dict__ + original = dict(data) + try: + if shape != "dict-subclass": + key = Key() if shape == "custom-key" else TextKey("_connection") + del data["_connection"] + data[key] = connection + assert list(data)[-1] is key + calls.clear() + armed = True + with _fetchone_window(profiling) as uncertain: + with pytest.raises(RuntimeError, match="uncertain dictionary policy"): + cursor.fetchone() + assert cursor.__dict__ is data + assert calls == (["get"] if shape == "dict-subclass" else ["eq"]) + _assert_fetchone_counts(uncertain, prepare=1, native=0, fallback=0, sql=0) + finally: + armed = False + cursor.__dict__ = original + assert cursor.rowcount == 2 + assert cursor.rownumber == 1 + assert cursor.fetchone()[0] == 3 + finally: + connection.remove_output_converter(mssql_python.SQL_INTEGER) + + +def test_conversion_prefilter_declines_callback_capable_dictionary_shapes(): + if ddbc_bindings._fetch_observer is None: + pytest.skip("Conversion prefilter requires supported CPython watchers") + calls = [] + + class Key: + def __hash__(self): + calls.append("hash") + return hash("_cached_converter_map") + + def __eq__(self, other): + calls.append("eq") + return False + + class TextKey(str): + __hash__ = Key.__hash__ + __eq__ = Key.__eq__ + + class CustomDict(dict): + def get(self, *args, **kwargs): + calls.append("get") + raise AssertionError("Custom get must not run in the prefilter") + + key = Key() + text_key = TextKey("unrelated") + custom_key_data = {key: None, "_cached_converter_map": []} + late_key_data = {"_cached_converter_map": [], text_key: None} + subclass_data = CustomDict({"_cached_converter_map": []}) + assert list(custom_key_data)[0] is key + assert list(late_key_data)[-1] is text_key + helper = ddbc_bindings._fetch_conversion_ineligible + for data in (custom_key_data, late_key_data, subclass_data): + calls.clear() + assert helper(data) is False + assert calls == [] + + distractors = { + name[:-2] + character + name[-1]: (0,) + for name in ("_cached_converter_map", "_uuid_str_indices") + for character in ("x", "\u00e9", "\0") + } + assert len(distractors) == 6 + assert all(type(key) is str and len(key) in (17, 21) for key in distractors) + data = {"_cached_converter_map": (), "_uuid_str_indices": None, **distractors} + assert helper(data) is False + data["_cached_converter_map"] = (0,) + assert helper(data) is True + data["_cached_converter_map"] = [] + assert helper(data) is True + data["_cached_converter_map"] = () + data["_uuid_str_indices"] = () + assert helper(data) is True + data["_uuid_str_indices"] = None + assert helper(data) is False + del data["_cached_converter_map"] + assert helper(data) is True + + +@pytest.mark.parametrize("method", METHODS) +@pytest.mark.parametrize("write", ("attribute", "object_setattr")) +def test_warm_attribute_writes_invalidate_native_position(connection, method, write): + with connection.cursor() as cursor: + for _ in range(12): + warm(cursor, method) + if write == "attribute": + cursor._next_row_index = 40 + else: + object.__setattr__(cursor, "_next_row_index", 40) + assert value(read(cursor, method), method) == 3 + assert cursor.rowcount == 41 + assert cursor.rownumber == 40 + + +@pytest.mark.parametrize("method", ("fetchone", "fetchmany", "iterator")) +def test_warm_in_place_column_map_clear_uses_legacy_fallback(connection, method): + with connection.cursor() as cursor: + warm(cursor, method) + legacy = {"legacy": 0} + cursor._column_name_map = legacy + assert value(read(cursor, method), method) == 3 + cursor._cached_column_map.clear() + row = read(cursor, method) + assert row[0] == 4 + assert row._column_map is legacy + + +@pytest.mark.parametrize("method", METHODS) +def test_warm_function_code_mutation_and_restoration(connection, method): + def fail_map(self): + raise RuntimeError("code mutation") + + with connection.cursor() as cursor: + warm(cursor, method) + function = Cursor._get_column_and_converter_maps + original = function.__code__ + try: + function.__code__ = fail_map.__code__ + helper_patch = nullcontext() + if method == "fetchone" and ddbc_bindings._fetch_observer is not None: + helper_patch = patch.object( + ddbc_bindings, + "_fetch_conversion_ineligible", + side_effect=AssertionError("substituted conversion predicate"), + ) + with helper_patch as helper_trap: + with pytest.raises(RuntimeError, match="code mutation"): + read(cursor, method) + if helper_trap is not None: + helper_trap.assert_not_called() + finally: + function.__code__ = original + assert cursor.rowcount == 3 + assert value(read(cursor, method), method) == 4 + + if method == "fetchone" and ddbc_bindings._fetch_observer is not None: + connection.add_output_converter(mssql_python.SQL_INTEGER, lambda raw: raw) + try: + with _native_fetch_counters() as profiling, connection.cursor() as cursor: + warm(cursor, "fetchone") + try: + function.__code__ = fail_map.__code__ + with patch.object( + ddbc_bindings, + "_fetch_conversion_ineligible", + side_effect=AssertionError("negative substituted conversion predicate"), + ) as helper_trap: + with _fetchone_window(profiling) as failed: + with pytest.raises(RuntimeError, match="code mutation"): + cursor.fetchone() + helper_trap.assert_not_called() + _assert_fetchone_counts(failed, prepare=0, native=0, fallback=1, sql=1) + finally: + function.__code__ = original + assert cursor.rowcount == 3 + assert cursor.rownumber == 2 + assert cursor.fetchone()[0] == 4 + finally: + connection.remove_output_converter(mssql_python.SQL_INTEGER) + + +@pytest.mark.parametrize("method", METHODS) +def test_converter_callback_mutation_retains_row_stage(connection, method): + calls = [] + + def convert(raw): + calls.append(raw) + connection.remove_output_converter(mssql_python.SQL_INTEGER) + return raw * 10 + + tracked = method == "fetchone" and ddbc_bindings._fetch_observer is not None + with _native_fetch_counters(tracked) as profiling, connection.cursor() as cursor: + warm(cursor, method) + connection.add_output_converter(mssql_python.SQL_INTEGER, convert) + assert value(read(cursor, method), method) == 30 + assert calls == [3] + with _fetchone_window(profiling, tracked) as refresh: + assert value(read(cursor, method), method) == 4 + if tracked: + _assert_fetchone_counts(refresh, prepare=0, native=0, fallback=1, sql=1) + assert cursor._cached_converter_map == () + assert cursor._cached_converters_generation == connection._converters_generation + with _fetchone_window(profiling, tracked) as reenabled: + assert value(read(cursor, method), method) == 5 + if tracked: + _assert_fetchone_counts(reenabled, prepare=1, native=1, fallback=0, sql=1) + assert refresh["python"].get(convert.__code__, 0) == 0 + assert reenabled["python"].get(convert.__code__, 0) == 0 + assert cursor.rowcount == 5 + + if tracked: + connection.add_output_converter(mssql_python.SQL_INTEGER, lambda raw: raw) + try: + with _native_fetch_counters() as profiling, connection.cursor() as cursor: + warm(cursor, "fetchone") + connection.add_output_converter(mssql_python.SQL_INTEGER, lambda raw: raw * 10) + with patch.object( + connection, + "get_output_converter", + side_effect=RuntimeError("negative converter refresh"), + ) as converter_hook: + with _fetchone_window(profiling) as failed: + with pytest.raises(RuntimeError, match="negative converter refresh"): + cursor.fetchone() + converter_hook.assert_called_once_with(mssql_python.SQL_INTEGER) + _assert_fetchone_counts(failed, prepare=0, native=0, fallback=1, sql=1) + assert cursor.rowcount == 3 + assert cursor.rownumber == 2 + assert cursor.fetchone()[0] == 40 + finally: + connection.remove_output_converter(mssql_python.SQL_INTEGER) + + +@pytest.mark.parametrize("method", METHODS) +def test_warm_native_result_transitions_and_closure(connection, method): + with connection.cursor() as cursor: + warm(cursor, method) + assert [row[0] for row in cursor.fetchmany(2)] == [3, 4] + assert value(read(cursor, method), method) == 5 + cursor.execute("SELECT 11 AS first; SELECT 12 AS second") + assert value(read(cursor, method), method) == 11 + assert cursor.nextset() + assert value(read(cursor, method), method) == 12 + cursor.execute("SELECT n FROM (VALUES (1), (2), (3)) AS v(n) WHERE 1 = 0") + assert read(cursor, method) is None + assert cursor.rowcount == 0 + cursor.execute(SELECT_FIVE) + assert value(read(cursor, method), method) == 1 + connection.close() + with pytest.raises(mssql_python.ProgrammingError, match="closed"): + read(cursor, method) + + +def test_native_retained_rows_and_cycles_are_collectible(connection): + cursor = connection.cursor() + warm(cursor, "fetchone") + row = cursor.fetchone() + mapping = row._mapping + cursor.saved_row = row + cursor_reference, row_reference = weakref.ref(cursor), weakref.ref(row) + cursor.close() + assert list(mapping.values()) == [3] + del mapping, row, cursor + gc.collect() + assert cursor_reference() is None + assert row_reference() is None + + +def test_native_fetchmany_argument_contract(connection): + with connection.cursor() as cursor: + warm(cursor, "fetchmany") + assert cursor.fetchmany(size=1)[0][0] == 3 + assert cursor.fetchmany(None)[0][0] == 4 + assert cursor.fetchmany(True)[0][0] == 5 + assert cursor.fetchmany(0) == [] + for call in ( + lambda: cursor.fetchmany(1, 2), + lambda: cursor.fetchmany(1, size=1), + lambda: cursor.fetchmany(unknown=1), + ): + with pytest.raises(TypeError): + call() + + +def test_native_fetchmany_keyword_subclass_preserves_binding_error(connection): + calls = [] + + class Keyword(str): + __hash__ = str.__hash__ + + def __eq__(self, other): + calls.append(other) + raise RuntimeError("keyword equality sentinel") + + keywords = {Keyword("size"): 1} + with _native_fetch_counters() as profiling, connection.cursor() as cursor: + warm(cursor, "fetchmany") + before = profiling.get_stats() if profiling is not None else {} + with pytest.raises(RuntimeError, match="keyword equality sentinel"): + cursor.fetchmany(**keywords) + assert calls + assert cursor.rowcount == 2 + assert cursor.rownumber == 1 + if profiling is not None: + after = profiling.get_stats() + for name in ("ddbc::NativeCursor::fetch", "ddbc::FetchSingleRow"): + assert after.get(name, {}).get("calls", 0) == before.get(name, {}).get("calls", 0) + assert cursor.fetchmany(size=1)[0][0] == 3 + + +def test_native_fetchmany_oversized_int_preserves_closed_cursor_error(): + if ddbc_bindings._fetch_observer is None: + pytest.skip("Native size prefilter requires supported CPython watchers") + cursor = Cursor.__new__(Cursor) + cursor.closed = True + with pytest.raises(mssql_python.ProgrammingError, match="cursor is closed"): + cursor.fetchmany(1 << 100) + + +def test_native_unbound_keyword_self_preserves_python_signature(connection): + with connection.cursor() as cursor: + warm(cursor, "fetchone") + assert Cursor.fetchone(self=cursor)[0] == 3 + assert Cursor.fetchmany(self=cursor, size=1)[0][0] == 4 + assert Cursor.fetchval(self=cursor) == 5 + with pytest.raises(StopIteration): + Cursor.__next__(self=cursor) + + +def test_cursor_subclass_object_allocator_and_fetch_override(connection): + class CustomCursor(Cursor): + def __new__(cls, *args, **kwargs): + return object.__new__(cls) + + def fetchone(self): + row = super().fetchone() + self.fetch_calls += 1 + return row + + with CustomCursor(connection) as cursor: + cursor.fetch_calls = 0 + cursor.execute(SELECT_FIVE) + assert cursor.fetchval() == 1 + assert next(cursor)[0] == 2 + assert cursor.fetch_calls == 2 + + +def test_native_arrow_interleave(connection): + pytest.importorskip("pyarrow") + with connection.cursor() as cursor: + warm(cursor, "fetchone") + assert cursor.arrow_batch(2).column(0).to_pylist() == [3, 4] + assert cursor.fetchone()[0] == 5 + assert cursor.rowcount == 5 + assert cursor.rownumber == 4 + + +def test_native_executemany_and_catalog_transitions(connection): + with connection.cursor() as cursor: + warm(cursor, "fetchone") + cursor.execute("CREATE TABLE #native_fetch_transition (value INT)") + try: + cursor.executemany( + "INSERT INTO #native_fetch_transition (value) VALUES (?)", [(7,), (8,), (9,)] + ) + assert cursor.fetchval() is None + cursor.execute("SELECT value FROM #native_fetch_transition ORDER BY value") + assert cursor.fetchone()[0] == 7 + assert cursor.fetchone()[0] == 8 + retained = cursor.fetchone() + cursor.columns(table="__native_fetch_no_such_table__") + assert cursor.fetchmany(1) == [] + assert cursor.rowcount == 0 + assert retained[0] == 9 + cursor.execute("SELECT 10 AS replacement") + assert cursor.fetchval() == 10 + finally: + cursor.execute("DROP TABLE #native_fetch_transition") + + +@pytest.mark.skipif( + sys.platform == "win32", + reason="Windows does not export the native SQLFetch function-pointer global", +) +@pytest.mark.parametrize("method", METHODS) +@pytest.mark.parametrize("mutation", ("override", "dictionary", "converter")) +def test_mutation_during_native_fetch_completes_once(conn_str, method, mutation): + if not conn_str: + pytest.skip("DB_CONNECTION_STRING is required") + if ddbc_bindings._fetch_observer is None: + pytest.skip("Requires supported CPython watchers") + # Pointer interception and its callback lifetime are confined to a subprocess. + code = ( + "import runpy, sys; " + "runpy.run_path(sys.argv[1])['_mutate_during_fetch'](sys.argv[2], sys.argv[3])" + ) + result = subprocess.run( + [sys.executable, "-c", code, str(Path(__file__).resolve()), method, mutation], + capture_output=True, + text=True, + timeout=60, + ) + assert result.returncode == 0, (result.returncode, result.stdout, result.stderr) + + +def _mutate_during_fetch(method, mutation): + import ctypes + import os + + with mssql_python.connect(os.environ["DB_CONNECTION_STRING"], timeout=5) as connection: + with connection.cursor() as cursor: + warm(cursor, method) + library = ctypes.CDLL(str(Path(ddbc_bindings.module.__file__).resolve())) + pointer = ctypes.c_void_p.in_dll(library, "SQLFetch_ptr") + original = pointer.value + fetch_type = ctypes.CFUNCTYPE(ctypes.c_short, ctypes.c_void_p) + fetch = fetch_type(original) + callback_calls, errors = [], [] + + @fetch_type + def intercepted(handle): + try: + callback_calls.append(handle) + if mutation == "dictionary": + cursor.__dict__ = cursor.__dict__.copy() + cursor.__dict__["_next_row_index"] = 40 + elif mutation == "converter": + connection.add_output_converter( + mssql_python.SQL_INTEGER, lambda raw: raw * 10 + ) + else: + + def fail_map(): + raise RuntimeError("mutation inside SQLFetch") + + cursor._get_column_and_converter_maps = fail_map + return fetch(handle) + except BaseException as error: + errors.append(type(error).__name__) + return -1 + + pointer.value = ctypes.cast(intercepted, ctypes.c_void_p).value + try: + if mutation == "override": + with pytest.raises(RuntimeError, match="mutation inside SQLFetch"): + read(cursor, method) + assert cursor.rowcount == 3 + del cursor._get_column_and_converter_maps + else: + assert value(read(cursor, method), method) == ( + 30 if mutation == "converter" else 3 + ) + assert cursor.rowcount == (41 if mutation == "dictionary" else 3) + finally: + pointer.value = original + assert not errors, errors + assert len(callback_calls) == 1 + connection.clear_output_converters() + assert value(read(cursor, method), method) == 4 + + +def _redesign_subprocess(function, *arguments): + code = "import runpy, sys; runpy.run_path(sys.argv[1])[sys.argv[2]](*sys.argv[3:])" + result = subprocess.run( + [sys.executable, "-c", code, str(Path(__file__).resolve()), function, *arguments], + capture_output=True, + text=True, + timeout=60, + ) + assert result.returncode == 0, (result.returncode, result.stdout, result.stderr) + + +def _redesign_fault_bindings(): + _redesign_runtime() + if not hasattr(ddbc_bindings, "_native_fetch_test_fault_schema"): + pytest.skip("Requires separate MSSQL_NATIVE_FETCH_TEST_FAULTS=ON build") + assert ddbc_bindings._native_fetch_test_fault_schema == 1 + assert hasattr(ddbc_bindings, "profiling") + return ddbc_bindings._native_fetch_test_arm, ddbc_bindings._native_fetch_test_snapshot + + +def _redesign_position(data): + return tuple(data[name] for name in ("_next_row_index", "_rownumber", "rowcount")) + + +def _redesign_delta(after, before, name): + return after[name] - before[name] + + +@pytest.mark.parametrize("stage", (1, 2, 3, 4)) +def test_redesign_allocation_failure_contract(conn_str, stage): + _redesign_fault_bindings() + _redesign_subprocess("_redesign_allocation_failure", str(stage), "absent") + + +def test_redesign_copied_state_publication_failure_contract(conn_str): + _redesign_fault_bindings() + _redesign_subprocess("_redesign_allocation_failure", "4", "copied") + + +def _redesign_allocation_failure(stage, key_mode): + import os + + stage = int(stage) + arm, snapshot = _redesign_fault_bindings() + gc.collect() + origin = snapshot(None) + assert origin["schema"] == 1 and origin["active"] == 0 + with ( + mssql_python.connect(os.environ["DB_CONNECTION_STRING"], timeout=5) as connection, + _native_fetch_counters() as profiling, + ): + assert profiling is not None + + def scenario(): + cursor = connection.cursor() + old_dict = old_state = None + try: + if key_mode == "copied": + warm(cursor, "fetchone") + old_dict = cursor.__dict__ + old_state = old_dict["_native_fetch_state"] + old_position = _redesign_position(old_dict) + assert snapshot(cursor)["positive_plan"] + cursor.__dict__ = old_dict.copy() + else: + cursor.execute(SELECT_FIVE) + assert "_native_fetch_state" not in cursor.__dict__ + data = cursor.__dict__ + position = _redesign_position(data) + before = snapshot(cursor) + arm(stage) + exception_name = None + with _fetchone_window(profiling) as observation: + try: + cursor.fetchone() + except MemoryError as error: + exception_name = type(error).__name__ + assert exception_name == "MemoryError" + _assert_fetchone_counts(observation, prepare=0, native=0, fallback=0, sql=0) + after = snapshot(cursor) + assert after["armed"] == 0 and after["hit"] == stage + assert _redesign_delta(after, before, "hits") == 1 + visits = tuple(a - b for a, b in zip(after["visits"], before["visits"])) + assert ( + visits + == { + 1: (1, 0, 0, 0, 0), + 2: (1, 1, 0, 0, 0), + 3: (1, 1, 1, 0, 0), + 4: (1, 1, 1, 1, 1), + }[stage] + ) + assert after["active"] == 0 and not after["receiver_active"] + for name in ("states_allocated", "states_cleared", "states_deallocated"): + assert _redesign_delta(after, before, name) == int(stage >= 3), name + for name in ("holders_allocated", "holders_deleted"): + assert _redesign_delta(after, before, name) == int(stage == 4), name + assert _redesign_delta(after, before, "publications") == 0 + assert cursor.__dict__ is data and _redesign_position(data) == position + if key_mode == "copied": + assert data["_native_fetch_state"] is old_state + assert old_dict["_native_fetch_state"] is old_state + assert _redesign_position(old_dict) == old_position + else: + assert "_native_fetch_state" not in data + assert not after["associated"] + + # The except target is gone; no exception/traceback owns the receiver. + reference = weakref.ref(cursor) + with _fetchone_window(profiling) as recovered: + row = cursor.fetchone() + expected = 3 if key_mode == "copied" else 1 + assert row[0] == expected + _assert_fetchone_counts( + recovered, + prepare=1, + native=int(key_mode == "copied"), + fallback=int(key_mode != "copied"), + sql=1, + ) + with _fetchone_window(profiling) as native: + next_row = cursor.fetchone() + assert next_row[0] == expected + 1 + _assert_fetchone_counts( + native, prepare=int(key_mode != "copied"), native=1, fallback=0, sql=1 + ) + recovered_state = snapshot(cursor) + assert recovered_state["active"] == 0 + assert recovered_state["associated"] and recovered_state["positive_plan"] + assert recovered_state["hits"] == after["hits"] + if key_mode == "copied": + assert data["_native_fetch_state"] is not old_state + assert old_dict["_native_fetch_state"] is old_state + assert _redesign_position(old_dict) == old_position + return reference, weakref.ref(row), weakref.ref(next_row) + finally: + arm(0) + cursor.close() + + references = scenario() + gc.collect() + assert all(reference() is None for reference in references) + end = snapshot(None) + assert end["active"] == 0 and end["armed"] == 0 + assert _redesign_delta(end, origin, "states_allocated") == _redesign_delta( + end, origin, "states_cleared" + ) + assert _redesign_delta(end, origin, "holders_allocated") == _redesign_delta( + end, origin, "holders_deleted" + ) + + +@pytest.mark.parametrize("boundary", ("prepublication", "last_copied_state")) +def test_redesign_publication_callback_boundaries(conn_str, boundary): + _redesign_fault_bindings() + _redesign_subprocess("_redesign_publication_callback", boundary) + + +def _redesign_publication_callback(boundary): + import os + + arm, snapshot = _redesign_fault_bindings() + callbacks, errors = [], [] + conflict = object() + with ( + mssql_python.connect(os.environ["DB_CONNECTION_STRING"], timeout=5) as connection, + _native_fetch_counters() as profiling, + connection.cursor() as cursor, + ): + assert profiling is not None + reference = weakref.ref(cursor) + + def callback(): + target = reference() + observed = snapshot(target) + assert observed["active"] == 1 and observed["receiver_active"] + if boundary == "prepublication": + assert observed["armed"] == 0 and observed["hit"] == 5 + assert _redesign_delta(observed, before, "hits") == 1 + assert _redesign_delta(observed, before, "publications") == 0 + assert not observed["associated"] + target.__dict__ = target.__dict__.copy() + else: + assert _redesign_delta(observed, before, "publications") == 1 + assert _redesign_delta(observed, before, "states_deallocated") == 1 + assert _redesign_delta(observed, before, "positive_clears") == 1 + assert _redesign_delta(observed, before, "guarded_positive_clears") == 1 + assert observed["associated"] and not observed["positive_plan"] + target.__dict__["_native_fetch_state"] = conflict + callbacks.append(target.fetchone()[0]) + + if boundary == "prepublication": + cursor.execute(SELECT_FIVE) + old_dict = cursor.__dict__ + old_position = _redesign_position(old_dict) + cursor._native_fetch_test_callback = callback + else: + warm(cursor, "fetchone") + assert snapshot(cursor)["positive_plan"] + + class OnRelease: + def __del__(self): + try: + callback() + except BaseException as error: + errors.append(type(error).__name__) + + cursor.messages.append(OnRelease()) + old_dict = cursor.__dict__ + cursor.__dict__ = old_dict.copy() + cursor.messages = [] + # Break the old State->dict->State cycle. The copied key is now its + # only external owner; do not retain a State, Plan or messages alias. + del old_dict["_native_fetch_state"] + del old_dict + before = snapshot(cursor) + if boundary == "prepublication": + arm(5) + try: + with _fetchone_window(profiling) as observation: + row = cursor.fetchone() + finally: + arm(0) + expected = 2 if boundary == "prepublication" else 4 + assert row[0] == expected and callbacks == [expected - 1] + assert not errors, errors + _assert_fetchone_counts(observation, prepare=0, native=0, fallback=2, sql=2) + assert _redesign_position(cursor.__dict__) == (expected, expected - 1, expected) + assert cursor.__dict__["_native_fetch_state"] is conflict + after = snapshot(cursor) + assert after["active"] == 0 and not after["associated"] + visits = tuple(a - b for a, b in zip(after["visits"], before["visits"])) + assert visits == (1, 1, 1, int(boundary == "last_copied_state"), 1) + if boundary == "prepublication": + assert _redesign_position(old_dict) == old_position + assert "_native_fetch_state" not in old_dict + assert _redesign_delta(after, before, "states_deallocated") == 1 + assert _redesign_delta(after, before, "holders_deleted") == 1 + del cursor._native_fetch_test_callback + del old_dict + else: + assert _redesign_delta(after, before, "states_deallocated") == 2 + assert _redesign_delta(after, before, "holders_deleted") == 2 + del cursor._native_fetch_state + with _fetchone_window(profiling) as recovered: + assert cursor.fetchone()[0] == expected + 1 + _assert_fetchone_counts(recovered, prepare=1, native=1, fallback=0, sql=1) + assert snapshot(cursor)["positive_plan"] and snapshot(cursor)["active"] == 0 + + +def _redesign_runtime(): + from mssql_python.cursor import _NATIVE_FETCH_RUNTIME + + if ddbc_bindings._fetch_observer is None: + pytest.skip("Requires supported CPython watchers") + assert ddbc_bindings._native_fetch_schema == 2 + assert type(_NATIVE_FETCH_RUNTIME) is tuple + assert len(_NATIVE_FETCH_RUNTIME) == 7 + assert _NATIVE_FETCH_RUNTIME[6] is ddbc_bindings._native_fetch_context + assert Cursor.__bases__ == (object,) + return _NATIVE_FETCH_RUNTIME + + +def test_redesign_state_is_lazy_and_gc_visible(connection): + runtime = _redesign_runtime() + with connection.cursor() as cursor: + assert "_native_fetch_state" not in cursor.__dict__ + ddbc_bindings._invalidate_native_fetch(cursor) + assert "_native_fetch_state" not in cursor.__dict__ + warm(cursor, "fetchone") + state = cursor.__dict__["_native_fetch_state"] + assert gc.is_tracked(state) + assert gc.is_tracked(runtime[6]) + referents = gc.get_referents(state) + assert any(item is cursor.__dict__ for item in referents) + assert any(item is runtime[6] for item in referents) + assert any(type(item) is weakref.ReferenceType and item() is cursor for item in referents) + with pytest.raises(TypeError, match="created internally"): + type(state)() + + +@pytest.mark.parametrize("mutation", ("delete", "copy", "none", "conflict")) +def test_redesign_state_entry_mutation_preserves_position(connection, mutation): + _redesign_runtime() + with connection.cursor() as cursor: + warm(cursor, "fetchone") + old_dict = cursor.__dict__ + old_state = old_dict["_native_fetch_state"] + replacement = object() if mutation == "conflict" else None + if mutation == "delete": + del cursor._native_fetch_state + elif mutation == "copy": + cursor.__dict__ = old_dict.copy() + else: + cursor.__dict__["_native_fetch_state"] = replacement + assert cursor.fetchone()[0] == 3 + assert cursor.rowcount == 3 + assert cursor.rownumber == 2 + if mutation in ("delete", "copy"): + assert cursor.__dict__["_native_fetch_state"] is not old_state + else: + assert cursor.__dict__["_native_fetch_state"] is replacement + if mutation == "copy": + assert old_dict["_native_fetch_state"] is old_state + assert old_dict["rowcount"] == 2 + assert cursor.fetchone()[0] == 4 + + +def test_redesign_foreign_state_is_not_adopted(connection, conn_str): + _redesign_runtime() + with ( + mssql_python.connect(conn_str, timeout=5) as other, + connection.cursor() as left, + other.cursor() as right, + ): + warm(left, "fetchone") + right.execute("SELECT n FROM (VALUES (101), (102), (103), (104)) AS v(n) ORDER BY n") + assert right.fetchone()[0] == 101 + assert right.fetchone()[0] == 102 + foreign = right.__dict__["_native_fetch_state"] + left.__dict__["_native_fetch_state"] = foreign + assert left.fetchone()[0] == 3 + assert left.__dict__["_native_fetch_state"] is foreign + assert right.rowcount == 2 + assert right.fetchone()[0] == 103 + assert left.rowcount == right.rowcount == 3 + + +def test_redesign_dictionary_deletion_does_not_create_state_or_consume(connection): + _redesign_runtime() + with connection.cursor() as cursor: + warm(cursor, "fetchone") + original = cursor.__dict__ + try: + del cursor.__dict__ + with pytest.raises(AttributeError, match="closed"): + cursor.fetchone() + assert "_native_fetch_state" not in cursor.__dict__ + finally: + cursor.__dict__ = original + assert cursor.rowcount == 2 + assert cursor.fetchone()[0] == 3 + + +def test_redesign_state_key_acquisition_does_not_invoke_equality(connection): + _redesign_runtime() + calls = [] + armed = False + + class Key: + def __hash__(self): + if armed: + calls.append("hash") + raise AssertionError("State acquisition hashed a user key") + return hash("_native_fetch_state") + + def __eq__(self, other): + if armed: + calls.append("eq") + raise AssertionError("State acquisition compared a user key") + return False + + with connection.cursor() as cursor: + warm(cursor, "fetchone") + del cursor._native_fetch_state + key = Key() + cursor.__dict__[key] = "unrelated" + try: + armed = True + assert cursor.fetchone()[0] == 3 + assert calls == [] + assert all( + type(item) is not str or item != "_native_fetch_state" for item in cursor.__dict__ + ) + finally: + armed = False + del cursor.__dict__[key] + assert cursor.fetchone()[0] == 4 + assert "_native_fetch_state" in cursor.__dict__ + + +@pytest.mark.parametrize("hook", ("__getattribute__", "__setattr__")) +def test_redesign_cold_state_respects_cursor_attribute_hooks(connection, hook): + _redesign_runtime() + calls = [] + original = getattr(object, hook) + + def custom(self, name, *args): + calls.append(name) + return original(self, name, *args) + + with connection.cursor() as cursor: + cursor.execute(SELECT_FIVE) + with patch.object(Cursor, hook, custom): + assert cursor.fetchone()[0] == 1 + assert cursor.fetchone()[0] == 2 + assert "_native_fetch_state" not in object.__getattribute__(cursor, "__dict__") + assert calls + assert "_native_fetch_state" not in calls + assert cursor.fetchone()[0] == 3 + assert "_native_fetch_state" in cursor.__dict__ + + +def test_redesign_partial_initialization_keeps_first_assignments(): + assignments = [] + + class RecordingCursor(Cursor): + def __setattr__(self, name, value): + assignments.append((name, value)) + object.__setattr__(self, name, value) + + def _initialize_cursor(self): + raise RuntimeError("redesign initialization sentinel") + + cursor = object.__new__(RecordingCursor) + try: + with pytest.raises(RuntimeError, match="redesign initialization sentinel"): + cursor.__init__(None) + assert assignments[:2] == [("closed", False), ("hstmt", None)] + assert "_native_fetch_state" not in cursor.__dict__ + ddbc_bindings._invalidate_native_fetch(cursor) + assert "_native_fetch_state" not in cursor.__dict__ + finally: + cursor.closed = True + + +def test_redesign_reinitialization_does_not_reuse_old_plan(connection): + _redesign_runtime() + cursor = connection.cursor() + try: + warm(cursor, "fetchone") + row = cursor.fetchone() + old_handle = cursor.hstmt + cursor.close() + Cursor.__init__(cursor, connection) + assert cursor.hstmt is not old_handle + cursor.execute("SELECT n FROM (VALUES (99), (100), (101)) AS v(n) ORDER BY n") + assert [cursor.fetchone()[0] for _ in range(3)] == [99, 100, 101] + assert list(row) == [3] + assert cursor.rowcount == 3 + finally: + cursor.close() + + +def test_redesign_plan_failure_leaves_state_recoverable_before_sql(connection): + runtime = _redesign_runtime() + with connection.cursor() as cursor: + cursor.execute(SELECT_FIVE) + before = (cursor.rowcount, cursor.rownumber) + with patch.object(type(runtime[2]), "__call__", side_effect=MemoryError("plan sentinel")): + with pytest.raises(MemoryError, match="plan sentinel"): + cursor.fetchone() + assert (cursor.rowcount, cursor.rownumber) == before + assert gc.is_tracked(cursor.__dict__["_native_fetch_state"]) + assert cursor.fetchone()[0] == 1 + assert cursor.fetchone()[0] == 2 + + +def test_redesign_last_plan_release_reenters_before_outer_sql(connection): + _redesign_runtime() + callbacks, errors = [], [] + with _native_fetch_counters() as profiling, connection.cursor() as cursor: + warm(cursor, "fetchone") + reference = weakref.ref(cursor) + + class OnRelease: + def __del__(self): + try: + callbacks.append(reference().fetchone()[0]) + except BaseException as error: + errors.append(type(error).__name__) + + cursor.messages.append(OnRelease()) + cursor.messages = [] + with _fetchone_window(profiling) as observation: + assert cursor.fetchone()[0] == 4 + _assert_fetchone_counts(observation, prepare=1, native=0, fallback=2, sql=2) + assert callbacks == [3] + assert not errors + assert cursor.rowcount == 4 + assert cursor.rownumber == 3 + assert cursor.fetchone()[0] == 5 + + +def test_redesign_external_dictionary_alias_then_unrooted_cycle(conn_str): + _redesign_runtime() + _redesign_subprocess("_redesign_live_plan_cycle") + + +def _redesign_live_plan_cycle(): + import os + + _redesign_runtime() + finalized, closed, completed, errors = [], [], [], [] + original_close = Cursor.close + + def close(self): + closed.append((self.closed is False, self.hstmt is not None)) + original_close(self) + completed.append((self.closed is True, self.hstmt is None)) + + def unraisable(args): + errors.append(type(args.exc_value).__name__) + + with ( + mssql_python.connect(os.environ["DB_CONNECTION_STRING"], timeout=5) as connection, + _native_fetch_counters() as profiling, + ): + cursor = connection.cursor() + warm(cursor, "fetchone") + with _fetchone_window(profiling) as observation: + row = cursor.fetchone() + _assert_fetchone_counts(observation, prepare=0, native=1, fallback=0, sql=1) + assert row[0] == 3 and row._cursor is cursor + cursor.back_edge = lambda saved=row: saved + assert cursor.back_edge() is row + assert not cursor.closed and "_native_fetch_state" in cursor.__dict__ + snapshot = getattr(ddbc_bindings, "_native_fetch_test_snapshot", None) + if snapshot is not None: + assert snapshot(cursor)["positive_plan"] + alias = cursor.__dict__ + reference, row_reference = weakref.ref(cursor), weakref.ref(row) + weakref.finalize(cursor, finalized.append, "cursor") + weakref.finalize(row, finalized.append, "row") + del cursor, row + with patch.object(Cursor, "close", close), patch.object(sys, "unraisablehook", unraisable): + gc.collect() + assert reference() is not None and row_reference()[0] == 3 + assert not finalized and not closed and not completed + if snapshot is not None: + assert snapshot(reference())["positive_plan"] + del alias + gc.collect() + assert reference() is None and row_reference() is None + assert sorted(finalized) == ["cursor", "row"] + assert closed == [(True, True)] + assert completed == [(True, True)] + assert not errors + assert _redesign_runtime()[6] is ddbc_bindings._native_fetch_context + + +@pytest.mark.skipif(sys.platform == "win32", reason="Requires exported SQLFetch pointer") +@pytest.mark.parametrize("mutation", ("dictionary", "delete", "conflict")) +def test_redesign_reentry_survives_state_replacement(conn_str, mutation): + _redesign_runtime() + code = ( + "import runpy, sys; " + "runpy.run_path(sys.argv[1])['_redesign_reenter_during_fetch'](sys.argv[2])" + ) + result = subprocess.run( + [sys.executable, "-c", code, str(Path(__file__).resolve()), mutation], + capture_output=True, + text=True, + timeout=60, + ) + assert result.returncode == 0, (result.returncode, result.stdout, result.stderr) + + +def _redesign_reenter_during_fetch(mutation): + import ctypes + import os + + runtime = _redesign_runtime() + with mssql_python.connect(os.environ["DB_CONNECTION_STRING"], timeout=5) as connection: + with _native_fetch_counters() as profiling, connection.cursor() as cursor: + warm(cursor, "fetchone") + old = cursor.__dict__ + conflict = object() + library = ctypes.CDLL(str(Path(ddbc_bindings.module.__file__).resolve())) + pointer = ctypes.c_void_p.in_dll(library, "SQLFetch_ptr") + original = pointer.value + fetch_type = ctypes.CFUNCTYPE(ctypes.c_short, ctypes.c_void_p) + fetch = fetch_type(original) + calls, nested, errors = [], [], [] + + @fetch_type + def intercepted(handle): + try: + calls.append(handle) + pointer.value = original + if mutation == "dictionary": + cursor.__dict__ = old.copy() + elif mutation == "delete": + del cursor._native_fetch_state + else: + cursor._native_fetch_state = conflict + nested.append(cursor.fetchone()[0]) + return fetch(handle) + except BaseException as error: + errors.append(type(error).__name__) + return -1 + + pointer.value = ctypes.cast(intercepted, ctypes.c_void_p).value + try: + with _fetchone_window(profiling) as observation: + row = cursor.fetchone() + finally: + pointer.value = original + assert not errors, errors + assert len(calls) == 1 + assert nested == [3] + assert row[0] == 4 + assert cursor.rowcount == 4 + assert cursor.rownumber == 3 + python = observation["python"] + assert python.get(type(runtime[2]).__call__.__code__, 0) == 0 + assert python.get(runtime[3][0].__code__, 0) == 1 + assert python.get(Cursor._complete_fetchone.__code__, 0) == 1 + assert python.get(Cursor._wrap_fetchone.__code__, 0) == 1 + if observation["native"] is not None: + for name, expected in ( + ("NativeCursor::prepare", 0), + ("NativeCursor::fetch", 1), + ("NativeCursor::fallback", 1), + ("NativeCursor::completion_fallback", 1), + ("FetchSingleRow", 2), + ): + assert observation["native"].get("ddbc::" + name, 0) == expected + if mutation == "dictionary": + assert old["rowcount"] == 2 + elif mutation == "conflict": + assert cursor._native_fetch_state is conflict + assert cursor.fetchone()[0] == 5 + + +@pytest.mark.skipif(sys.platform == "win32", reason="Requires exported SQLFetch pointer") +def test_redesign_interleaved_calls_can_exit_out_of_order(conn_str): + _redesign_runtime() + code = "import runpy, sys; " "runpy.run_path(sys.argv[1])['_redesign_interleave_calls']()" + result = subprocess.run( + [sys.executable, "-c", code, str(Path(__file__).resolve())], + capture_output=True, + text=True, + timeout=60, + ) + assert result.returncode == 0, (result.returncode, result.stdout, result.stderr) + + +def _redesign_interleave_calls(): + import ctypes + import os + import threading + + runtime = _redesign_runtime() + with ( + mssql_python.connect(os.environ["DB_CONNECTION_STRING"], timeout=5) as first, + mssql_python.connect(os.environ["DB_CONNECTION_STRING"], timeout=5) as second, + first.cursor() as left, + second.cursor() as right, + ): + warm(left, "fetchone") + warm(right, "fetchone") + library = ctypes.CDLL(str(Path(ddbc_bindings.module.__file__).resolve())) + pointer = ctypes.c_void_p.in_dll(library, "SQLFetch_ptr") + original = pointer.value + fetch_type = ctypes.CFUNCTYPE(ctypes.c_short, ctypes.c_void_p) + fetch = fetch_type(original) + entered = {name: threading.Event() for name in ("left", "right")} + release = {name: threading.Event() for name in entered} + calls, results, errors = {}, {}, [] + + @fetch_type + def intercepted(handle): + try: + name = threading.current_thread().name + calls[name] = calls.get(name, 0) + 1 + if calls[name] == 1: + entered[name].set() + assert release[name].wait(15), name + if name == "right": + with _fetchone_window(None) as nested: + assert right.fetchone()[0] == 3 + assert nested["python"].get(runtime[3][0].__code__, 0) == 1 + assert nested["python"].get(type(runtime[2]).__call__.__code__, 0) == 0 + return fetch(handle) + except BaseException as error: + errors.append(type(error).__name__) + return -1 + + def run(name, cursor): + try: + results[name] = cursor.fetchone()[0] + except BaseException as error: + errors.append(type(error).__name__) + + a = threading.Thread(target=run, args=("left", left), name="left") + b = threading.Thread(target=run, args=("right", right), name="right") + pointer.value = ctypes.cast(intercepted, ctypes.c_void_p).value + try: + a.start() + assert entered["left"].wait(15) + b.start() + assert entered["right"].wait(15) + release["left"].set() + a.join(15) + assert not a.is_alive() + release["right"].set() + b.join(15) + assert not b.is_alive() + finally: + release["left"].set() + release["right"].set() + if a.ident is not None: + a.join(15) + if b.ident is not None: + b.join(15) + pointer.value = original + assert not errors, errors + assert results == {"left": 3, "right": 4} + assert calls == {"left": 1, "right": 2} + assert left.fetchone()[0] == 4 + assert right.fetchone()[0] == 5 + + +@pytest.mark.parametrize("phase", ("method", "policy", "activation", "rollback")) +def test_redesign_installation_failure_restores_originals(phase): + _redesign_runtime() + code = r""" +import importlib.abc +import importlib.machinery +import sys + +phase = sys.argv[1] +saved = {"restores": [], "terminal": []} + +class Loader: + def __init__(self, original): + self.original = original + + def create_module(self, spec): + return self.original.create_module(spec) + + def exec_module(self, module): + self.original.exec_module(module) + bindings = module.ddbc_bindings + begin = bindings._begin_native_fetch_install + activate = bindings._activate_native_fetch_install + abort = bindings._abort_native_fetch_install + disable = bindings._disable_native_fetch_install + policy = module._Policy + saved.update(module=module, bindings=bindings, begin=begin, activate=activate, + policy=policy, abort=abort, disable=disable) + + def capture(cursor_type, functions): + namespace = functions[0].__globals__ + saved.update(cursor_type=cursor_type, functions=functions, namespace=namespace, + row_type=namespace["Row"]) + return begin(cursor_type, functions) + + def assign(target, name, value): + if phase == "method" and name == "fetchmany" and value is bindings._native_fetch_methods[1]: + raise RuntimeError("installation sentinel") + names = ("fetchone", "fetchmany", "fetchval", "__next__") + if value is saved["functions"][names.index(name)]: + saved["restores"].append(name) + if phase == "rollback" and name == "fetchone": + raise RuntimeError("secondary restoration sentinel") + setattr(target, name, value) + + def terminal(name, function): + def invoke(): + saved["terminal"].append(name) + return function() + return invoke + + def fail_policy(cursor_type, row_type, namespace, functions): + temporary = cursor_type.__new__(cursor_type) + temporary.closed = True + try: + temporary.fetchone() + except namespace["ProgrammingError"] as error: + assert "cursor is closed" in str(error) + else: + raise AssertionError("Unpublished entry did not retain ordinary errors") + assert "_native_fetch_state" not in temporary.__dict__ + policy(cursor_type, row_type, namespace, functions) + raise MemoryError("installation sentinel") + + def reject_runtime(runtime): + return activate(runtime[:6]) + + bindings._begin_native_fetch_install = capture + bindings._abort_native_fetch_install = terminal("abort", abort) + bindings._disable_native_fetch_install = terminal("disable", disable) + module.setattr = assign + if phase in ("policy", "rollback"): + module._Policy = fail_policy + elif phase == "activation": + bindings._activate_native_fetch_install = reject_runtime + +class Finder(importlib.abc.MetaPathFinder): + def find_spec(self, fullname, path, target=None): + if fullname != "mssql_python._native_fetch": + return None + spec = importlib.machinery.PathFinder.find_spec(fullname, path) + assert spec is not None + spec.loader = Loader(spec.loader) + return spec + +finder = Finder() +sys.meta_path.insert(0, finder) +expected = {"method": RuntimeError, "policy": MemoryError, + "activation": TypeError, "rollback": RuntimeError}[phase] +try: + try: + import mssql_python + except expected as error: + if phase == "rollback": + assert type(error) is saved["module"]._InstallRollbackError + assert type(error.installation_error) is MemoryError + assert str(error.installation_error) == "installation sentinel" + assert error.__cause__ is error.installation_error + assert len(error.cleanup_errors) == 1 + name, secondary = error.cleanup_errors[0] + assert name == "fetchone" and type(secondary) is RuntimeError + assert str(secondary) == "secondary restoration sentinel" + assert "installation sentinel" in str(error) + assert "secondary restoration sentinel" in str(error) + del secondary + else: + raise AssertionError("Installation failure was swallowed") +finally: + sys.meta_path.remove(finder) + +module = saved["module"] +bindings = saved["bindings"] +bindings._begin_native_fetch_install = saved["begin"] +bindings._activate_native_fetch_install = saved["activate"] +bindings._abort_native_fetch_install = saved["abort"] +bindings._disable_native_fetch_install = saved["disable"] +module._Policy = saved["policy"] +del module.setattr +cursor_type = saved["cursor_type"] +names = ("fetchone", "fetchmany", "fetchval", "__next__") +if phase == "rollback": + assert saved["restores"] == list(names) + assert saved["terminal"] == ["disable"] + assert vars(cursor_type)["fetchone"] is bindings._native_fetch_methods[0] + assert all(vars(cursor_type)[name] is function + for name, function in zip(names[1:], saved["functions"][1:])) + profiling = getattr(bindings, "profiling", None) + if profiling is not None: + profiling.reset() + profiling.enable() + temporary = object.__new__(cursor_type) + temporary.closed = True + for entry in bindings._native_fetch_methods: + try: + entry.__get__(temporary, cursor_type)() + except saved["namespace"]["ProgrammingError"] as error: + assert "cursor is closed" in str(error) + else: + raise AssertionError("Disabled descriptor lost ordinary fallback") + assert "_native_fetch_state" not in temporary.__dict__ + if profiling is not None: + profiling.disable() + stats = profiling.get_stats() + for name in ("NativeCursor::prepare", "NativeCursor::fetch", "NativeCursor::fallback", + "FetchOne_wrap", "FetchSingleRow"): + assert stats.get("ddbc::" + name, {}).get("calls", 0) == 0 + try: + module.install(cursor_type, saved["row_type"], saved["namespace"]) + except RuntimeError as error: + assert "disabled after incomplete rollback" in str(error) + else: + raise AssertionError("Disabled context allowed reinstallation") + assert vars(cursor_type)["fetchone"] is bindings._native_fetch_methods[0] + raise SystemExit(0) +assert saved["terminal"] == ["abort"] +assert all(vars(cursor_type)[name] is function for name, function in zip(names, saved["functions"])) +runtime = module.install(cursor_type, saved["row_type"], saved["namespace"]) +assert len(runtime) == 7 and runtime[6] is bindings._native_fetch_context +assert all(vars(cursor_type)[name] is entry for name, entry in zip(names, bindings._native_fetch_methods)) +""" + result = subprocess.run( + [sys.executable, "-c", code, phase], + capture_output=True, + text=True, + timeout=60, + ) + assert result.returncode == 0, (result.returncode, result.stdout, result.stderr) diff --git a/tests/test_single_row_pipeline.py b/tests/test_single_row_pipeline.py new file mode 100644 index 000000000..2c94c5781 --- /dev/null +++ b/tests/test_single_row_pipeline.py @@ -0,0 +1,276 @@ +""" +Copyright (c) Microsoft Corporation. +Licensed under the MIT license. + +Single-row native fetch parity, ownership, and operation-count regressions. +""" + +import datetime +import decimal +from unittest.mock import Mock, patch +import uuid +import weakref + +import pytest + +import mssql_python +from mssql_python import ddbc_bindings +from mssql_python.cursor import Cursor +from mssql_python.row import Row + +SINGLE_METHODS = ("fetchone", "iterator", "fetchval", "fetchmany_one") +UUID_TEXT = "00112233-4455-6677-8899-AABBCCDDEEFF" + + +@pytest.fixture +def connection(conn_str): + try: + conn = mssql_python.connect(conn_str, timeout=5) + except mssql_python.Error as error: + pytest.fail( + f"Connection failed: {type(error).__name__}; connection details withheld", + pytrace=False, + ) + try: + yield conn + finally: + conn.close() + + +def read_one(cursor, method): + if method == "iterator": + return next(cursor, None) + if method.startswith("fetchmany"): + batch = cursor.fetchmany(1) if method == "fetchmany_one" else cursor.fetchmany() + assert type(batch) is list + assert len(batch) <= 1 + return batch[0] if batch else None + return getattr(cursor, method)() + + +@pytest.mark.parametrize("method", (*SINGLE_METHODS, "fetchmany_default")) +def test_single_row_types_nulls_and_position(connection, method): + columns = ( + ("INT", "-42", -42), + ("BIGINT", "9223372036854775807", 9223372036854775807), + ("SMALLINT", "-32768", -32768), + ("TINYINT", "255", 255), + ("BIT", "1", True), + ("REAL", "1.25", 1.25), + ("FLOAT", "1.25", 1.25), + ("DECIMAL(20,4)", "12345678.1250", decimal.Decimal("12345678.1250")), + ("VARCHAR(10)", "'abc'", "abc"), + ("NVARCHAR(10)", "NCHAR(233)", "\u00e9"), + ("VARBINARY(10)", "0x0102", b"\x01\x02"), + ("UNIQUEIDENTIFIER", f"'{UUID_TEXT}'", uuid.UUID(UUID_TEXT)), + ("DATE", "'20260102'", datetime.date(2026, 1, 2)), + ("DATETIME2", "'2026-01-02T03:04:05.125'", datetime.datetime(2026, 1, 2, 3, 4, 5, 125000)), + ("TIME", "'03:04:05.125'", datetime.time(3, 4, 5, 125000)), + ("XML", "'abc'", "abc"), + ) + expressions = ", ".join( + f"CAST(CASE WHEN n % 2 = 0 THEN NULL ELSE {literal} END AS {sql_type}) AS c{i}" + for i, (sql_type, literal, _) in enumerate(columns) + ) + values = [value for _, _, value in columns] + with connection.cursor() as cursor: + assert cursor.arraysize == 1 + cursor.execute(f"SELECT {expressions} FROM (VALUES (1), (2), (3), (4)) AS v(n) ORDER BY n") + for index, expected in enumerate((values, [None] * len(columns)) * 2): + row = read_one(cursor, method) + if method != "fetchval": + assert type(row) is Row + actual = list(row) + else: + actual = [row] + expected = expected[:1] + assert actual == expected + assert [type(item) for item in actual] == [type(item) for item in expected] + assert cursor.rowcount == index + 1 + assert cursor.rownumber == index + assert read_one(cursor, method) is None + assert read_one(cursor, method) is None + assert cursor.rowcount == 4 + assert cursor.rownumber == 3 + assert not cursor.messages + cursor.execute("SELECT 7 AS replacement WHERE 1 = 0") + assert read_one(cursor, method) is None + assert cursor.rowcount == 0 + assert cursor.rownumber == -1 + + +@pytest.mark.parametrize("width", (1, 128)) +@pytest.mark.parametrize("method", SINGLE_METHODS) +def test_single_row_interleaves_batches_and_keeps_row_ownership(connection, width, method): + columns = ", ".join( + f"n AS c{i}" if i % 2 == 0 else f"CAST(N'payload' AS NVARCHAR(20)) AS c{i}" + for i in range(width) + ) + with connection.cursor() as cursor: + cursor.execute( + f"SELECT {columns} FROM (VALUES (1), (2), (3), (4), (5), (6), (7)) AS v(n) ORDER BY n" + ) + first = read_one(cursor, method) + assert (first if method == "fetchval" else first[0]) == 1 + assert [row[0] for row in cursor.fetchmany(2)] == [2, 3] + fourth = read_one(cursor, method) + assert (fourth if method == "fetchval" else fourth[0]) == 4 + remaining = cursor.fetchall() + assert [row[0] for row in remaining] == [5, 6, 7] + assert cursor.rowcount == 7 + assert cursor.rownumber == 6 + assert cursor.fetchmany(1) == [] + cursor.execute("SELECT 99 AS replacement; SELECT 100 AS following") + assert cursor.fetchone()[0] == 99 + assert cursor.nextset() + assert cursor.fetchmany(1)[0][0] == 100 + if method != "fetchval": + assert list(first) == [1 if i % 2 == 0 else "payload" for i in range(width)] + assert first._column_names == tuple(f"c{i}" for i in range(width)) + assert first._values is not fourth._values + assert first._values is not remaining[0]._values + reference = weakref.ref(first) + del first + assert reference() is None + + +@pytest.mark.parametrize("method", ("fetchone", "fetchmany_one")) +def test_partial_native_metadata_is_not_a_result_column_count(connection, method): + with connection.cursor() as cursor: + cursor.execute( + "SELECT n, CAST(N'abc' AS NVARCHAR(10)) AS text, CAST(1.25 AS DECIMAL(5,2)) AS amount " + "FROM (VALUES (1), (2), (3)) AS v(n) ORDER BY n" + ) + assert ddbc_bindings.DDBCSQLFetch(cursor.hstmt) == 0 + prefix = [] + assert ( + ddbc_bindings.DDBCSQLGetData( + cursor.hstmt, 1, prefix, "utf-16le", "utf-16le", mssql_python.SQL_WCHAR + ) + == 0 + ) + assert prefix == [1] + for expected in (2, 3): + row = read_one(cursor, method) + assert list(row) == [expected, "abc", decimal.Decimal("1.25")] + + +@pytest.mark.parametrize("method", SINGLE_METHODS) +def test_single_row_converter_refresh_failure_keeps_consumed_position(connection, method): + with connection.cursor() as cursor: + cursor.execute( + "SELECT CAST(n AS NVARCHAR(10)) AS value " + "FROM (VALUES (1), (2), (3)) AS v(n) ORDER BY n" + ) + first = read_one(cursor, method) + assert (first if method == "fetchval" else first[0]) == "1" + converter = Mock(side_effect=lambda raw: "converted:" + raw.decode("utf-16-le")) + connection.add_output_converter(mssql_python.SQL_WVARCHAR, converter) + with patch.object( + connection, "get_output_converter", side_effect=RuntimeError("injected map failure") + ): + with pytest.raises(RuntimeError, match="injected map failure"): + read_one(cursor, method) + assert cursor.rowcount == 2 + assert cursor.rownumber == 1 + final = read_one(cursor, method) + assert (final if method == "fetchval" else final[0]) == "converted:3" + converter.assert_called_once_with(b"3\x00") + + +@pytest.mark.parametrize("method", SINGLE_METHODS) +def test_single_row_decoding_refresh_failure_does_not_consume(connection, method): + with connection.cursor() as cursor: + cursor.execute("SELECT CONVERT(VARCHAR(1), 0xE9) AS value FROM (VALUES (1), (2)) AS v(n)") + connection.setdecoding(mssql_python.SQL_CHAR, encoding="latin-1") + with patch.object(connection, "getdecoding", side_effect=RuntimeError("injected decoding")): + with pytest.raises(RuntimeError, match="injected decoding"): + read_one(cursor, method) + assert cursor.rownumber == -1 + for index in range(2): + row = read_one(cursor, method) + assert (row if method == "fetchval" else row[0]) == "\u00e9" + assert cursor.rowcount == index + 1 + + +@pytest.mark.parametrize("method", ("iterator", "fetchval")) +@pytest.mark.parametrize("override", ("instance", "subclass")) +def test_single_row_keeps_fetchone_override_dispatch(connection, method, override): + calls = [] + + class CustomCursor(Cursor): + def fetchone(self): + calls.append("subclass") + return super().fetchone() + + cursor_class = CustomCursor if override == "subclass" else Cursor + with cursor_class(connection) as cursor: + cursor.execute("SELECT 42 AS value") + if override == "instance": + original = cursor.fetchone + + def fetchone(): + calls.append("instance") + return original() + + cursor.fetchone = fetchone + row = read_one(cursor, method) + assert (row if method == "fetchval" else row[0]) == 42 + assert read_one(cursor, method) is None + assert calls == [override, override] + + +@pytest.mark.parametrize("method", SINGLE_METHODS) +def test_single_row_factory_failure_keeps_consumed_position(connection, method): + with connection.cursor() as cursor: + cursor.execute("SELECT n FROM (VALUES (1), (2)) AS v(n) ORDER BY n") + target, name = ( + (ddbc_bindings, "construct_rows") + if method.startswith("fetchmany") + else (Row, "_fast_create") + ) + with patch.object(target, name, side_effect=RuntimeError("injected row factory")): + with pytest.raises(RuntimeError, match="injected row factory"): + read_one(cursor, method) + assert cursor.rowcount == 1 + assert cursor.rownumber == 0 + row = read_one(cursor, method) + assert (row if method == "fetchval" else row[0]) == 2 + assert cursor.rowcount == 2 + + +@pytest.mark.parametrize("method", SINGLE_METHODS) +def test_single_row_closed_and_cancelled_cursor(connection, method): + with connection.cursor() as cursor: + cursor.execute("SELECT n FROM (VALUES (1), (2)) AS v(n)") + assert read_one(cursor, method) is not None + cursor.hstmt._cancel() + cursor.execute("SELECT 42 AS replacement") + row = read_one(cursor, method) + assert (row if method == "fetchval" else row[0]) == 42 + with pytest.raises(mssql_python.ProgrammingError, match="closed"): + read_one(cursor, method) + + +@pytest.mark.parametrize("method", SINGLE_METHODS) +def test_single_row_reuses_unbound_state_without_batch_setup(connection, method): + profiling = getattr(ddbc_bindings, "profiling", None) + if profiling is None: + pytest.skip("Operation counts require an ENABLE_PROFILING=ON diagnostic build") + with connection.cursor() as cursor: + cursor.execute("SELECT n FROM (VALUES (1), (2), (3)) AS v(n) ORDER BY n") + profiling.reset() + profiling.enable() + try: + for expected in (1, 2, 3): + row = read_one(cursor, method) + assert (row if method == "fetchval" else row[0]) == expected + assert read_one(cursor, method) is None + finally: + profiling.disable() + counts = {name: stats["calls"] for name, stats in profiling.get_stats().items()} + assert counts["ddbc::FetchSingleRow"] == 4 + assert counts["ddbc::FetchSingleRow::SQL_UNBIND"] == 1 + assert counts.get("ddbc::SQLNumResultCols_wrap", 0) <= 1 + assert counts.get("ddbc::SQLBindColums", 0) == 0 + assert counts.get("ddbc::FetchBatchData", 0) == 0