FEAT: async POC#677
Draft
subrata-ms wants to merge 8 commits into
Draft
Conversation
Introduces the foundational C++ building blocks for the upcoming Cursor.execute_async / Cursor.fetch_async work. No user-facing API is exposed yet; sync execute/fetch paths are functionally unchanged. Helpers added in ddbc_bindings.cpp: - struct PollingConfig (initial_ms=0.5, max_ms=20, multiplier=1.5) - class AsyncEnableGuard: RAII sets/clears SQL_ATTR_ASYNC_ENABLE on an HSTMT so the async flag never leaks onto a handle reused by later sync calls. - template<Fn> ExecuteWithPolling(callOnce, asyncMode, cfg): sync passthrough when asyncMode=false; polling loop with exponential backoff when asyncMode=true. Caller owns GIL release. - PrepareAndBind(...): factored out of SQLExecute_wrap (SQLPrepare under released GIL, DescribeCache clear, isStmtPrepared flag update, BindParameters). Enables reuse from upcoming async execute wrap. SQLExecute_wrap refactored to route both branches through the helpers with asyncMode=false (semantic no-op vs. the previous inline calls). DAE loop unchanged. Capability probe: - Connection::isAsyncCapable() / ConnectionHandle::isAsyncCapable() probes SQLGetInfo(SQL_ASYNC_MODE), caches the value in an atomic int (SQL_AM_STATEMENT / SQL_AM_CONNECTION => true). Exposed to Python as Connection.is_async_capable(). Build: clean under -Werror -Wattributes -Wint-to-pointer-cast.
📊 Code Coverage Report
Diff CoverageDiff: main...HEAD, staged and unstaged changes
Summary
mssql_python/cursor.pyLines 3526-3534 3526 per connection.
3527 """
3528 conn = self._connection._conn
3529 if not conn.is_async_capable():
! 3530 raise NotSupportedError(
3531 driver_error=(
3532 "Async execution is not supported by this ODBC driver. "
3533 "SQLGetInfo(SQL_ASYNC_MODE) reports SQL_AM_NONE."
3534 ),Lines 3537-3545 3537
3538 def _acquire_async_slot(self) -> None:
3539 """Take the cursor-busy slot; raise ProgrammingError if already held."""
3540 if self._async_in_flight:
! 3541 raise ProgrammingError(
3542 driver_error=(
3543 "Cursor is busy: another async operation is already in flight "
3544 "on this cursor. One Cursor == one HSTMT, so async ops on the "
3545 "same cursor must be serialized. Create a second cursor for "Lines 3575-3590 3575 if reset_cursor:
3576 if self.hstmt:
3577 self._soft_reset_cursor()
3578 else:
! 3579 self._reset_cursor()
3580 else:
! 3581 if self.hstmt:
3582 logger.debug(
3583 "execute_async: Closing cursor for re-execution (reset_cursor=False)"
3584 )
! 3585 self.hstmt._close_cursor()
! 3586 self._clear_rownumber()
3587
3588 # Clear any previous messages
3589 self.messages = []Lines 3590-3603 3590
3591 # Parameter unwrap (same rules as sync execute).
3592 if parameters:
3593 if isinstance(parameters, tuple) and len(parameters) == 1:
! 3594 if isinstance(parameters[0], (tuple, list, dict)):
! 3595 actual_params = parameters[0]
! 3596 elif isinstance(parameters[0], Row):
! 3597 actual_params = tuple(parameters[0])
3598 else:
! 3599 actual_params = parameters
3600 else:
3601 actual_params = parameters
3602
3603 if operation == self.last_executed_stmt and isinstance(Lines 3602-3610 3602
3603 if operation == self.last_executed_stmt and isinstance(
3604 actual_params, (tuple, list)
3605 ):
! 3606 parameters = list(actual_params)
3607 else:
3608 operation, converted_params = detect_and_convert_parameters(
3609 operation, actual_params
3610 )Lines 3609-3617 3609 operation, actual_params
3610 )
3611 parameters = list(converted_params)
3612 else:
! 3613 parameters = []
3614
3615 encoding_settings = self._get_encoding_settings()
3616
3617 logger.debug("execute_async: Creating parameter type list")Lines 3618-3627 3618 param_info = ddbc_bindings.ParamInfo
3619 parameters_type: List[Any] = []
3620
3621 if parameters and self._inputsizes:
! 3622 if len(self._inputsizes) != len(parameters):
! 3623 warnings.warn(
3624 f"Number of input sizes ({len(self._inputsizes)}) does not match "
3625 f"number of parameters ({len(parameters)}). "
3626 f"This may lead to unexpected behavior.",
3627 Warning,Lines 3657-3668 3657 DELIBERATE DUPLICATE of the corresponding logic in ``execute()``.
3658 """
3659 try:
3660 check_error(ddbc_sql_const.SQL_HANDLE_STMT.value, self.hstmt, ret)
! 3661 except Exception as e: # pylint: disable=broad-exception-caught
3662 logger.warning("execute_async failed, resetting cursor: %s", e)
! 3663 self._reset_cursor()
! 3664 raise
3665
3666 self._capture_diagnostics(ret)
3667 self.last_executed_stmt = operation
3668 self.rowcount = ddbc_bindings.DDBCSQLRowCount(self.hstmt)Lines 3670-3680 3670 column_metadata: List[Any] = []
3671 try:
3672 ddbc_bindings.DDBCSQLDescribeCol(self.hstmt, column_metadata)
3673 self._initialize_description(column_metadata)
! 3674 except Exception: # pylint: disable=broad-exception-caught
3675 # If describe fails, it's likely there are no results (e.g. INSERT)
! 3676 self.description = None
3677
3678 if self.description:
3679 self.rowcount = -1
3680 self._reset_rownumber()Lines 3688-3701 3688 )
3689 self._cached_converter_map = self._build_converter_map()
3690 self._uuid_str_indices = self._compute_uuid_str_indices()
3691 else:
! 3692 self.rowcount = ddbc_bindings.DDBCSQLRowCount(self.hstmt)
! 3693 self._clear_rownumber()
! 3694 self._cached_column_map = None
! 3695 self._cached_column_map_lower = None
! 3696 self._cached_converter_map = None
! 3697 self._uuid_str_indices = None
3698
3699 self._reset_inputsizes()
3700
3701 def _wrap_row_async(self, row_data: List[Any]) -> Row:Lines 3711-3721 3711 )
3712
3713 def _wrap_rows_async(self, rows_data: List[List[Any]]) -> List[Row]:
3714 """Async-only mirror of fetchmany() / fetchall()'s Row-construction block."""
! 3715 column_map, converter_map, column_map_lower = self._get_column_and_converter_maps()
! 3716 uuid_idx = self._uuid_str_indices
! 3717 return [
3718 Row(
3719 row_data,
3720 column_map,
3721 cursor=self,Lines 3847-3855 3847 self._check_closed()
3848 self._check_async_capable()
3849
3850 if not self._has_result_set and self.description:
! 3851 self._reset_rownumber()
3852
3853 char_decoding = self._get_decoding_settings(ddbc_sql_const.SQL_CHAR.value)
3854 wchar_decoding = self._get_decoding_settings(ddbc_sql_const.SQL_WCHAR.value)
3855 char_enc = char_decoding.get("encoding", "utf-16le")Lines 3883-3897 3883 )
3884
3885 if ret == ddbc_sql_const.SQL_NO_DATA.value:
3886 if self._next_row_index == 0 and self.description is not None:
! 3887 self.rowcount = 0
3888 return None
3889
3890 # rownumber tracking (mirrors sync fetchone)
3891 if self._skip_increment_for_next_fetch:
! 3892 self._skip_increment_for_next_fetch = False
! 3893 self._next_row_index += 1
3894 else:
3895 self._increment_rownumber()
3896 self.rowcount = self._next_row_index
3897 return self._wrap_row_async(row_data)Lines 3895-3908 3895 self._increment_rownumber()
3896 self.rowcount = self._next_row_index
3897 return self._wrap_row_async(row_data)
3898
! 3899 if size == -1:
3900 # ---- fetchall-async ----
! 3901 rows_data: List[List[Any]] = []
! 3902 self._acquire_async_slot()
! 3903 try:
! 3904 ret = await loop.run_in_executor(
3905 None,
3906 ddbc_bindings.DDBCSQLFetchAllAsync,
3907 self.hstmt,
3908 rows_data,Lines 3912-3946 3912 poll_initial_ms,
3913 poll_max_ms,
3914 )
3915 finally:
! 3916 self._release_async_slot()
3917
! 3918 check_error(ddbc_sql_const.SQL_HANDLE_STMT.value, self.hstmt, ret)
3919
! 3920 if self.hstmt:
! 3921 self.messages.extend(
3922 ddbc_bindings.DDBCSQLGetAllDiagRecords(self.hstmt)
3923 )
3924
! 3925 if rows_data and self._has_result_set:
! 3926 self._next_row_index += len(rows_data)
! 3927 self._rownumber = self._next_row_index - 1
3928
! 3929 if len(rows_data) == 0 and self._next_row_index == 0:
! 3930 self.rowcount = 0
3931 else:
! 3932 self.rowcount = self._next_row_index
3933
! 3934 return self._wrap_rows_async(rows_data)
3935
3936 # ---- fetchmany-async (size > 0 or size <= 0 sentinel) ----
! 3937 if size <= 0:
! 3938 return []
! 3939 rows_data = []
! 3940 self._acquire_async_slot()
! 3941 try:
! 3942 ret = await loop.run_in_executor(
3943 None,
3944 ddbc_bindings.DDBCSQLFetchManyAsync,
3945 self.hstmt,
3946 rows_data,Lines 3951-3972 3951 poll_initial_ms,
3952 poll_max_ms,
3953 )
3954 finally:
! 3955 self._release_async_slot()
3956
! 3957 if self.hstmt:
! 3958 self.messages.extend(
3959 ddbc_bindings.DDBCSQLGetAllDiagRecords(self.hstmt)
3960 )
3961
! 3962 if rows_data and self._has_result_set:
! 3963 self._next_row_index += len(rows_data)
! 3964 self._rownumber = self._next_row_index - 1
3965
! 3966 if len(rows_data) == 0 and self._next_row_index == 0:
! 3967 self.rowcount = 0
3968 else:
! 3969 self.rowcount = self._next_row_index
3970
! 3971 return self._wrap_rows_async(rows_data)mssql_python/pybind/connection/connection.cppLines 642-655 642 }
643
644 // Async POC: probe SQL_ASYNC_MODE and cache the result. Returns true iff the
645 // driver supports statement-level async (SQL_AM_STATEMENT) or connection-level
! 646 // async (SQL_AM_CONNECTION). Called from Cursor.execute_async as a capability
! 647 // gate before configuring SQL_ATTR_ASYNC_ENABLE on the statement handle.
648 bool Connection::isAsyncCapable() const {
! 649 int cached = _asyncModeCache.load(std::memory_order_acquire);
! 650 if (cached < 0) {
! 651 if (!_dbcHandle) {
652 return false;
653 }
654 if (!SQLGetInfo_ptr) {
655 LOG("isAsyncCapable: SQLGetInfo not initialized, loading driver");Lines 653-665 653 }
654 if (!SQLGetInfo_ptr) {
655 LOG("isAsyncCapable: SQLGetInfo not initialized, loading driver");
656 DriverLoader::getInstance().loadDriver();
! 657 }
658 SQLUSMALLINT asyncMode = 0;
! 659 SQLSMALLINT outLen = 0;
! 660 SQLRETURN ret = SQLGetInfo_ptr(_dbcHandle->get(), SQL_ASYNC_MODE, &asyncMode,
! 661 sizeof(asyncMode), &outLen);
662 if (!SQL_SUCCEEDED(ret)) {
663 LOG("isAsyncCapable: SQLGetInfo(SQL_ASYNC_MODE) failed - SQLRETURN=%d", ret);
664 // Cache as SQL_AM_NONE so we don't re-probe on every call.
665 _asyncModeCache.store(static_cast<int>(SQL_AM_NONE), std::memory_order_release);Lines 668-677 668 cached = static_cast<int>(asyncMode);
669 _asyncModeCache.store(cached, std::memory_order_release);
670 LOG("isAsyncCapable: SQL_ASYNC_MODE=%d (cached)", cached);
671 }
! 672 return cached == static_cast<int>(SQL_AM_STATEMENT) ||
! 673 cached == static_cast<int>(SQL_AM_CONNECTION);
674 }
675
676 bool ConnectionHandle::isAsyncCapable() const {
677 if (!_conn) {Lines 675-684 675
676 bool ConnectionHandle::isAsyncCapable() const {
677 if (!_conn) {
678 ThrowStdException("Connection object is not initialized");
! 679 }
! 680 return _conn->isAsyncCapable();
681 }
682
683 void ConnectionHandle::setAttr(int attribute, py::object value) {
684 if (!_conn) {mssql_python/pybind/ddbc_bindings.cppLines 1845-1855 1845 }
1846 }
1847
1848 ~AsyncEnableGuard() {
! 1849 if (_enabled && _hStmt && SQLSetStmtAttr_ptr) {
! 1850 SQLRETURN rc = SQLSetStmtAttr_ptr(
! 1851 _hStmt, SQL_ATTR_ASYNC_ENABLE,
1852 reinterpret_cast<SQLPOINTER>(static_cast<uintptr_t>(SQL_ASYNC_ENABLE_OFF)),
1853 0);
1854 if (!SQL_SUCCEEDED(rc)) {
1855 LOG("AsyncEnableGuard: SQLSetStmtAttr(SQL_ATTR_ASYNC_ENABLE, OFF) "Lines 1855-1864 1855 LOG("AsyncEnableGuard: SQLSetStmtAttr(SQL_ATTR_ASYNC_ENABLE, OFF) "
1856 "failed - SQLRETURN=%d, hStmt=%p",
1857 rc, (void*)_hStmt);
1858 }
! 1859 }
! 1860 }
1861
1862 bool enabled() const { return _enabled; }
1863
1864 AsyncEnableGuard(const AsyncEnableGuard&) = delete;Lines 1881-1891 1881 // `callOnce` is any callable returning SQLRETURN (typically a lambda that
1882 // captures the statement handle and calls SQLExecute / SQLExecDirect / SQLFetch).
1883 template <typename Fn>
1884 inline SQLRETURN ExecuteWithPolling(Fn callOnce, bool asyncMode,
! 1885 const PollingConfig& cfg = PollingConfig{}) {
! 1886 SQLRETURN ret = callOnce();
! 1887 if (!asyncMode) {
1888 return ret;
1889 }
1890 double sleep_ms = cfg.initial_ms;
1891 while (ret == SQL_STILL_EXECUTING) {Lines 1889-1898 1889 }
1890 double sleep_ms = cfg.initial_ms;
1891 while (ret == SQL_STILL_EXECUTING) {
1892 std::this_thread::sleep_for(
! 1893 std::chrono::microseconds(static_cast<long long>(sleep_ms * 1000.0)));
! 1894 sleep_ms = std::min(sleep_ms * cfg.multiplier, cfg.max_ms);
1895 ret = callOnce();
1896 }
1897 return ret;
1898 }Lines 1899-1911 1899
1900 // Prepares the statement (if requested) and binds all parameters. Returns
1901 // the SQLRETURN from the last ODBC call; on non-success the caller should
1902 // short-circuit before invoking SQLExecute.
! 1903 //
! 1904 // `paramBuffers` is an OUT parameter: it accumulates heap-owned buffers that
! 1905 // MUST remain in scope until SQLExecute completes, because ODBC keeps raw
! 1906 // pointers into them across the SQLBindParameter / SQLExecute boundary.
! 1907 //
1908 // GIL: SQLPrepare is called with the GIL released (matches previous inline
1909 // behavior); BindParameters requires the GIL (inspects py::list contents).
1910 SQLRETURN PrepareAndBind(SqlHandlePtr statementHandle, SQLHANDLE hStmt, SQLWCHAR* queryPtr,
1911 const py::list& params, std::vector<ParamInfo>& paramInfos,Lines 1910-1919 1910 SQLRETURN PrepareAndBind(SqlHandlePtr statementHandle, SQLHANDLE hStmt, SQLWCHAR* queryPtr,
1911 const py::list& params, std::vector<ParamInfo>& paramInfos,
1912 py::list& isStmtPrepared, bool usePrepare,
1913 std::vector<std::shared_ptr<void>>& paramBuffers,
! 1914 const std::string& charEncoding) {
! 1915 // isStmtPrepared is a single-element list carrying a bool by reference
1916 // (Python bools are immutable, so we can't pass the raw bool by ref).
1917 assert(isStmtPrepared.size() == 1);
1918
1919 SQLRETURN rc = SQL_SUCCESS;Lines 2017-2026 2017 // Release the GIL during the blocking ODBC call. In async mode
2018 // ExecuteWithPolling re-invokes SQLExecDirect while it returns
2019 // SQL_STILL_EXECUTING, with backoff sleeps between polls.
2020 py::gil_scoped_release release;
! 2021 rc = ExecuteWithPolling(
! 2022 [&]() { return SQLExecDirect_ptr(hStmt, queryPtr, SQL_NTS); },
2023 asyncMode, pollCfg);
2024 }
2025 if (!SQL_SUCCEEDED(rc) && rc != SQL_NO_DATA) {
2026 LOG("SQLExecute: Direct execution failed (non-parameterized query) "Lines 2034-2046 2034 if (encodingSettings.contains("encoding")) {
2035 charEncoding = encodingSettings["encoding"].cast<std::string>();
2036 }
2037
! 2038 // This vector manages the heap memory allocated for parameter buffers.
! 2039 // It must remain in scope until SQLExecute (and any DAE loop) completes.
2040 std::vector<std::shared_ptr<void>> paramBuffers;
! 2041 rc = PrepareAndBind(statementHandle, hStmt, queryPtr, params, paramInfos,
! 2042 isStmtPrepared, usePrepare, paramBuffers, charEncoding);
2043 if (!SQL_SUCCEEDED(rc)) {
2044 return rc;
2045 }Lines 2044-2057 2044 return rc;
2045 }
2046
2047 {
! 2048 // Release the GIL during the blocking SQLExecute network call. In
! 2049 // async mode ExecuteWithPolling handles SQL_STILL_EXECUTING with
! 2050 // backoff sleeps between polls.
2051 py::gil_scoped_release release;
! 2052 rc = ExecuteWithPolling([&]() { return SQLExecute_ptr(hStmt); },
! 2053 asyncMode, pollCfg);
2054 }
2055 if (rc == SQL_NEED_DATA) {
2056 LOG("SQLExecute: SQL_NEED_DATA received - Starting DAE "
2057 "(Data-At-Execution) loop for large parameter streaming");Lines 2197-2208 2197 py::list& isStmtPrepared, const bool usePrepare,
2198 const py::dict& encodingSettings) {
2199 return SQLExecute_impl(statementHandle, query, params, paramInfos, isStmtPrepared,
2200 usePrepare, encodingSettings,
! 2201 /*asyncMode=*/false, PollingConfig{});
! 2202 }
! 2203
! 2204 // Async wrapper (exposed as DDBCSQLExecuteAsync). Called from Cursor.execute_async
2205 // via loop.run_in_executor so the polling loop runs on a background thread with
2206 // the GIL released, keeping the asyncio event loop responsive.
2207 SQLRETURN SQLExecuteAsync_wrap(const SqlHandlePtr statementHandle, const std::u16string& query,
2208 const py::list& params, std::vector<ParamInfo>& paramInfos,Lines 2210-2218 2210 const py::dict& encodingSettings, double poll_initial_ms,
2211 double poll_max_ms) {
2212 PollingConfig cfg;
2213 cfg.initial_ms = poll_initial_ms;
! 2214 cfg.max_ms = poll_max_ms;
2215 // multiplier stays at the PollingConfig default (1.5)
2216 return SQLExecute_impl(statementHandle, query, params, paramInfos, isStmtPrepared,
2217 usePrepare, encodingSettings, /*asyncMode=*/true, cfg);
2218 }Lines 4175-4186 4175 }
4176
4177 // Fetch rows in batches
4178 // TODO: Move to anonymous namespace, since it is not used outside this file
! 4179 //
! 4180 // Async POC: when asyncMode=true, the SQLFetchScroll network call is driven
! 4181 // through ExecuteWithPolling so it can return SQL_STILL_EXECUTING. The caller
! 4182 // is responsible for having installed SQL_ATTR_ASYNC_ENABLE on hStmt (via
4183 // AsyncEnableGuard) before calling this function.
4184 SQLRETURN FetchBatchData(SQLHSTMT hStmt, ColumnBuffers& buffers, py::list& columnNames,
4185 py::list& rows, SQLUSMALLINT numCols, SQLULEN& numRowsFetched,
4186 const std::vector<SQLUSMALLINT>& lobColumns,Lines 4766-4775 4766 cfg.initial_ms = poll_initial_ms;
4767 cfg.max_ms = poll_max_ms;
4768 return FetchMany_impl(StatementHandle, rows, fetchSize, charEncoding, wcharEncoding,
4769 charCtype, /*asyncMode=*/true, cfg);
! 4770 }
! 4771
4772 // GetDataVar - Progressively fetches variable-length column data using SQLGetData.
4773 //
4774 // Calls SQLGetData repeatedly, reallocating the buffer as needed, until all data is retrieved.
4775 // Handles both fixed-size and unknown-size (SQL_NO_TOTAL) responses from the driver.Lines 6027-6036 6027 }
6028
6029 // Sync wrapper (exposed as DDBCSQLFetchOne). Existing pybind binding target;
6030 // keeps the original signature and default arguments unchanged.
! 6031 SQLRETURN FetchOne_wrap(SqlHandlePtr StatementHandle, py::list& row,
! 6032 const std::string& charEncoding = "utf-16le",
6033 const std::string& wcharEncoding = "utf-16le",
6034 int charCtype = SQL_C_WCHAR) {
6035 return FetchOne_impl(StatementHandle, row, charEncoding, wcharEncoding, charCtype,
6036 /*asyncMode=*/false, PollingConfig{});📋 Files Needing Attention📉 Files with overall lowest coverage (click to expand)mssql_python.pybind.logger_bridge.cpp: 59.2%
mssql_python.pybind.ddbc_bindings.h: 59.9%
mssql_python.pybind.logger_bridge.hpp: 70.8%
mssql_python.pybind.connection.connection.cpp: 75.4%
mssql_python.pybind.ddbc_bindings.cpp: 76.4%
mssql_python.__init__.py: 77.3%
mssql_python.row.py: 77.6%
mssql_python.ddbc_bindings.py: 79.6%
mssql_python.connection.py: 83.6%
mssql_python.logging.py: 85.5%🔗 Quick Links
|
Adds the four statement-level async pybind entry points that Cursor.execute_async / Cursor.fetch_async will call from Python via loop.run_in_executor. Sync entry points (DDBCSQLExecute / DDBCSQLFetchOne|Many|All) are unchanged in signature and behavior. Refactor pattern for each of the four sync wraps: - *_wrap body extracted into static *_impl(..., bool asyncMode, const PollingConfig& pollCfg). - _impl installs AsyncEnableGuard(hStmt, asyncMode) at the top so SQL_ATTR_ASYNC_ENABLE is toggled on entry and reset on exit; the guard is a no-op when asyncMode=false. - Every SQLExecute / SQLExecDirect / SQLFetch / SQLFetchScroll call site now goes through ExecuteWithPolling(cb, asyncMode, pollCfg). Sync callers pass asyncMode=false which collapses to a straight one-shot call (semantic identity with previous inline behavior). - *_wrap becomes a thin sync forwarder that calls _impl with asyncMode=false. Existing pybind bindings and callers unchanged. - *Async_wrap is a new thin forwarder that builds PollingConfig from the Python-supplied poll_initial_ms / poll_max_ms and calls _impl with asyncMode=true. Additional plumbing: - FetchBatchData (internal helper shared by FetchMany_impl and FetchAll_impl non-LOB paths) gained defaulted asyncMode + pollCfg params; its SQLFetchScroll_ptr call is now routed through ExecuteWithPolling. All existing sync callers pass the defaults. - SQLExecute_impl rejects Data-At-Execution (isDAE=true) parameters up-front when asyncMode=true - the SQL_NEED_DATA loop is not safe to drive alongside SQL_STILL_EXECUTING polling. Callers must fall back to sync execute for large VARBINARY(MAX) / NVARCHAR(MAX) values. New pybind exports (each with trailing poll_initial_ms=0.5, poll_max_ms=20.0 float args): - DDBCSQLExecuteAsync - DDBCSQLFetchOneAsync - DDBCSQLFetchManyAsync - DDBCSQLFetchAllAsync Out of scope for POC (still sync-only): SQLGetData_wrap (LOB streaming), DDBCSQLFetchArrowBatch. Documented inline in the _impl bodies. Build: clean under -Werror -Wattributes -Wint-to-pointer-cast. All four async bindings + all existing sync bindings verified importable.
Adds the Python-facing async API on top of the C++ statement-level async
scaffolding delivered in steps 1 and 2. The change is 100% additive to
cursor.py (+483 lines, 0 removed): the sync execute / fetchone /
fetchmany / fetchall implementations are byte-identical to main and are
not touched.
Design decision: DUPLICATE, don't refactor.
- The prep and finalize logic that the async path needs is deliberately
duplicated from the sync execute() implementation instead of extracted
into shared helpers. Rationale: the sync methods are hot, well-tested,
and touching them for an experimental POC risks subtle regressions
that only surface in edge cases. Long-term de-duplication is deferred
until the async POC stabilizes.
Additions on Cursor:
- self._async_in_flight: bool guard added to __init__ (only sync-code
change, purely additive) - prevents overlapping async ops on the same
HSTMT (one Cursor == one HSTMT).
- _check_async_capable() - raises NotSupportedError when the driver
reports SQL_AM_NONE (uses cached is_async_capable() from step 1).
- _acquire_async_slot() / _release_async_slot() - raise ProgrammingError
on cursor-busy overlap.
- _prepare_execute_state_async(op, parameters, use_prepare, reset_cursor)
- deliberate duplicate of execute()'s pre-ODBC prep block.
- _finalize_execute_async(ret, operation) - deliberate duplicate of
execute()'s post-ODBC finalize block.
- _wrap_row_async(row_data) / _wrap_rows_async(rows_data) - deliberate
duplicates of the Row-construction blocks in fetchone / fetchmany /
fetchall.
Public async surface:
- async execute_async(operation, *parameters, use_prepare=True,
reset_cursor=True, poll_initial_ms=0.5, poll_max_ms=20.0) -> Cursor
offloads DDBCSQLExecuteAsync via loop.run_in_executor. asyncio is
imported lazily inside the method so sync-only users pay nothing.
- async fetch_async(size=None, *, poll_initial_ms=0.5, poll_max_ms=20.0)
-> Row | List[Row] | None
size=None -> DDBCSQLFetchOneAsync -> Row or None
size=-1 -> DDBCSQLFetchAllAsync -> List[Row]
size>0 -> DDBCSQLFetchManyAsync -> List[Row]
size<=0 -> [] (matches sync fetchmany semantics)
Concurrency contract:
- Same-cursor async ops are serialized via _async_in_flight; overlap
raises ProgrammingError('Cursor is busy...').
- Cross-cursor async ops on the same connection are allowed (each cursor
has its own HSTMT).
Not covered in this POC (documented in the tail block header):
- Data-At-Execution parameters (rejected up-front in C++).
- asyncio.CancelledError propagation (needs SQLCancelHandle).
- executemany_async / arrow_batch_async / catalog method async.
Verification:
- Diff is 100% additive (+483 / -0). Sync methods byte-identical to main.
- Module imports cleanly; all 9 new members present on Cursor.
- execute_async and fetch_async both verified as coroutine functions.
- Signatures match the spec exactly.
Not run: functional end-to-end tests (no DB in this env).
…etch_async
Adds tests/test_030_async_execute_fetch.py with three tests covering the
Python-facing async surface introduced by the previous three commits
(is_async_capable probe, DDBCSQL*Async pybind wrappers, and the public
Cursor.execute_async / Cursor.fetch_async methods).
Tests:
1. test_capability_probe_returns_true
Sanity check on Connection.is_async_capable() - fails loudly on a
regression in the C++ SQLGetInfo(SQL_ASYNC_MODE) probe rather than
silently skipping the whole module.
2. test_single_execute_async_and_fetch_async
End-to-end smoke: one execute_async with two parameters, one
fetch_async that returns the row, and a second fetch_async that
must return None.
3. test_100_concurrent_async_selects
Fires 100 concurrent execute_async + fetch_async pairs through
asyncio.gather with a bounded Semaphore(16). One dedicated
connection per task so we don't hit SQL Server's default no-MARS
restriction. Each task sends its index as a parameter and asserts
the returned row matches - proves per-HSTMT isolation
(no cross-wiring between concurrent statements) and that the
AsyncEnableGuard toggling on/off around each call does not leak
across cursors.
Design decisions:
- No pytest-asyncio dependency; each test uses asyncio.run(...) inline.
Keeps requirements.txt unchanged.
- Skips cleanly when DB_CONNECTION_STRING is unset OR the driver
reports SQL_ASYNC_MODE == SQL_AM_NONE.
- Concurrency knobs env-overridable for CI tuning:
ASYNC_TEST_CONCURRENCY (default 100),
ASYNC_TEST_MAX_INFLIGHT (default 16),
ASYNC_TEST_WALL_BUDGET (default 120s).
Run against SQL Server 2022 (mcr.microsoft.com/mssql/server:2022-latest)
in a local Docker container: all 3 tests pass, 100 concurrent async
queries complete in 0.75s wall clock.
…ntly skips)
Complements test_100_concurrent_async_selects (one connection per task) with a
same-connection MARS variant that opens ONE connection with
MultipleActiveResultSets=Yes and creates N cursors on it. Both variants share
the same correctness assertion (each task must return its own index), but
exercise different bug classes:
* connection-per-task proves cross-DBC async isolation and true
network-level parallelism.
* MARS same-connection proves per-HSTMT AsyncEnableGuard toggling doesn't
cross-wire result sets between cursors that share
a single DBC.
Current status: the mssql-python connection-string parser rejects both
MultipleActiveResultSets and MARS_Connection - neither keyword is in
_ALLOWED_CONNECTION_STRING_PARAMS (mssql_python/constants.py). The test
therefore skips today with a clear message pointing at the exact underlying
error. When MARS is added to the allowlist the test will start running
automatically with no further changes.
Also added a _with_mars() helper that appends the keyword unless the caller
already set MARS explicitly (either enabled or disabled).
Verified against SQL Server 2022 in local Docker: 3 pass, 1 skips cleanly.
…mment
While investigating why the MARS variant test skips, we prototyped adding
MultipleActiveResultSets / MARS_Connection to _ALLOWED_CONNECTION_STRING_PARAMS
to see if the test would pass. It did not — the combination
MARS + SQL_ATTR_ASYNC_ENABLE + concurrent execute across multiple
HSTMTs on one DBC (concurrency >= 2)
crashes the interpreter with SIGSEGV inside the ODBC driver / C++ layer.
Reproduces reliably at concurrency=2, N=5 on msodbcsql18 against SQL
Server 2022. Sequential MARS-async and Semaphore=1 (serialized) both work
fine — the failure is specific to multi-threaded concurrent SQLExecute on
HSTMTs that share a single DBC under async mode.
The MARS allowlist prototype was reverted (constants.py untouched — 0-line
delta vs main). This commit updates the WARNING block in the MARS test to
document the finding so a future maintainer knows why the keyword must NOT
be added to the mssql-python allowlist until either:
* the driver-side race is root-caused and fixed, OR
* mssql-python detects a MARS DBC in execute_async and either serializes
per-DBC via a lock, or raises NotSupportedError.
Until then, the connection-per-task pattern (test 3) remains the supported
way to run concurrent async workloads.
Test suite behavior unchanged: 3 pass, 1 skip.
Adds a second wave of functional tests exercising the async surface across
long queries, large result sets, event-loop non-blocking behavior,
multi-batch fetches, and sequential vs concurrent invocation patterns.
All safe patterns use one connection per concurrent task (see root-cause
investigation on why cross-cursor concurrent async on a shared DBC is
unsafe until per-Connection serialization is added).
New tests:
Case 1: test_execute_async_long_running_query
2-second server-side WAITFOR through execute_async; verifies the
polling loop survives many SQL_STILL_EXECUTING iterations without
losing the result set.
Case 2: test_event_loop_progresses_during_execute_async
Background heartbeat coroutine (10ms tick) runs while a 2s WAITFOR
query executes. Requires >= 60 ticks. Local run: 185 ticks. Proves
the event loop is NOT starved by the polling loop.
Case 4: test_1000_async_executes_on_different_connections
Scale test_100_concurrent_async_selects up to 1000 tasks. Marked
@pytest.mark.stress (excluded from default run per pytest.ini).
Local run: 1000 queries in 2.44s under Semaphore(16).
Case 6: test_fetch_async_returns_batch_of_rows
fetch_async(size=5) returns exactly 5 Row objects with correct values.
Case 7: test_fetch_async_large_result_set
5000-row SELECT via CROSS JOIN of sys.all_objects; fetch_async(-1)
returns all rows with correct sequential numbering.
Case 8: test_event_loop_progresses_during_fetch_async
Symmetric to Case 2 but on the fetch path. Uses HB_FETCH_ROWS (default
50000) so the fetch is long enough for meaningful heartbeat ticks.
Skips inconclusively if the fetch finishes in <50ms (fast local DB).
Case 9: test_multiple_concurrent_execute_async_small_batch
10-task variant of test_100_concurrent_async_selects; kept small for
quick smoke runs.
Case 10: test_execute_async_followed_by_fetch_async
All three fetch_async modes (size=None, positive int, -1) on the same
cursor after independent executes.
Case 11: test_multiple_concurrent_fetch_async_across_connections
Two-phase: (1) N=20 connections each execute a query sequentially,
(2) fetch_async fired concurrently across all N via asyncio.gather.
Verifies the fetch code path under concurrent gather.
Case 12: test_sequential_execute_async_on_same_cursor
Five sequential execute_async calls on the same cursor. Verifies that
AsyncEnableGuard correctly toggles SQL_ATTR_ASYNC_ENABLE OFF at the
end of each call so subsequent executes are not affected by leftover
state.
Env-overridable knobs added:
ASYNC_TEST_LARGE_ROWS rows for the large-result-set test (5000)
ASYNC_TEST_HB_FETCH_ROWS rows for fetch-heartbeat test (50000)
ASYNC_TEST_STRESS_CONCURRENCY N for the 1000-task stress test (1000)
ASYNC_TEST_WAITFOR_SECONDS server-side delay for long-query tests (2)
Verified against SQL Server 2022 in local Docker:
12 passed, 1 skipped (MARS), 1 deselected (stress) in 4.58s;
stress test explicitly: 1 passed in 2.47s.
…tability test
Root-cause fix for the MARS variant test that was skipping. Investigation
found that the SQL Server-native connection-string keyword
'MultipleActiveResultSets' is SILENTLY IGNORED by the bundled msodbcsql18
driver (empirically verified: keyword is accepted but MARS is never
actually enabled). Only the ODBC-standard alias 'MARS_Connection' turns
MARS on. The earlier segfault we saw when prototyping MARS in the
allowlist was because MARS was silently off — 100 concurrent async
statements on a non-MARS DBC hit the 'connection busy' error path, and
concurrent DDBCSQLCheckError calls raced on shared DBC diagnostic state.
With MARS genuinely on (this commit), concurrent async on shared DBC
works cleanly. 100 truly-concurrent execute_async + fetch_async pairs on
ONE MARS connection completes in ~130ms with no crash, no error, no
cross-wiring. Verified across 10 iterations on the same connection.
Changes:
mssql_python/constants.py (+13 lines)
Adds "mars_connection": "MARS_Connection" to
_ALLOWED_CONNECTION_STRING_PARAMS. Detailed comment warns future
maintainers NOT to add 'multipleactiveresultsets' without confirming
driver honors it (currently doesn't). This also unblocks MARS for
sync code — users can now do multiple active cursors on one
connection with 'MARS_Connection=Yes'.
tests/test_030_async_execute_fetch.py (+193/-35)
* _with_mars() helper now appends 'MARS_Connection=Yes' (the working
keyword) instead of 'MultipleActiveResultSets=Yes' (silently
ignored).
* Block comment above test_100_concurrent_async_selects_on_single_mars_connection
rewritten: replaces the outdated 'MARS segfault' warning with a
history / root-cause fix note explaining what was actually broken.
* Test docstring updated to remove 'skipped today' language.
* NEW test_mars_stability_100_cursors_high_concurrency_multi_iteration
(@pytest.mark.stress): 10 iterations of 100 concurrent async
cursors on ONE MARS connection, no semaphore, verifying
per-iteration correctness AND long-term stability. Local run: 1.33s
total, per-iteration timing consistent 126-160ms across all 10
rounds (no drift, no leak, no cross-wiring).
* Env knobs: ASYNC_TEST_MARS_STABILITY_N (default 100),
ASYNC_TEST_MARS_STABILITY_ITERS (default 10).
Key finding for future work: MARS gives client-side result-set
multiplexing over ONE TCP socket, but SQL Server assigns one worker
thread per session (SPID). So N concurrent statements on one MARS
connection queue up server-side, not truly parallel. Real network-level
parallelism still requires N separate connections
(test_100_concurrent_async_selects pattern).
Verified against SQL Server 2022 in local Docker:
15/15 tests pass (13 non-stress + 2 stress) in 10.17s total.
MARS variant test: 100 concurrent async in 0.13s (was skipping).
MARS stability test: 10 iters × 100 cursors in 1.33s (new).
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
This pull request adds support for detecting and caching ODBC driver async capabilities, which is a foundational step for enabling asynchronous operations (such as
execute_async) on SQL Server connections. The changes ensure that the driver’s support for asynchronous execution is probed once per connection and cached for efficient future checks. Additionally, the connection string parsing is updated to clarify and safely handle the enabling of Multiple Active Result Sets (MARS).Async capability detection and caching:
_asyncModeCache) in theConnectionclass to store the result of probingSQLGetInfo(SQL_ASYNC_MODE), minimizing redundant driver queries and improving performance for async operations.isAsyncCapable()method in bothConnectionandConnectionHandle, which checks and caches whether the driver supports statement-level or connection-level async, and is used as a capability gate before enabling async features. [1] [2] [3]Connection string handling improvements:
constants.pyto explicitly document and support the ODBC-standardMARS_Connectionparameter, while warning against using the SQL Server-nativeMultipleActiveResultSetsalias unless driver support is confirmed, preventing user confusion and unpredictable behavior.Work Item / Issue Reference
Summary