Method execute_sql

integration_tests/db_connectors/utils.py:1282–1341

Execute a single statement, retrying on deadlock victim (1205). Under heavy xdist parallelism every statement that touches CDC system metadata (sp_cdc_* procs, DDL on CDC-enabled objects, queries against cdc.* tables) competes with the capture agent for system-table locks

(
        self,
        query: str,
        params: tuple | None = None,
        drain_status_rows: bool = False,
        max_retries: int = 6,
        retry_delay: float = 0.5,
    )

        return name

    def execute_sql(
        self,
        query: str,
        params: tuple | None = None,
        drain_status_rows: bool = False,
        max_retries: int = 6,
        retry_delay: float = 0.5,
    ) -> None:
        """Execute a single statement, retrying on deadlock victim (1205).

        Under heavy xdist parallelism every statement that touches CDC system
        metadata (sp_cdc_* procs, DDL on CDC-enabled objects, queries against
        cdc.* tables) competes with the capture agent for system-table locks
        and SQL Server occasionally elects it as the deadlock victim.  Heavy
        DDL on master.sys.databases (CREATE/ALTER/DROP DATABASE) and on
        sys.objects/sys.schemas (CREATE/DROP TABLE/SCHEMA) is similarly
        vulnerable.  The recommended response per Microsoft is to rerun the
        transaction, which is what we do here.

        A fresh cursor is opened per attempt — pymssql leaves the cursor in a
        state where the next statement reports "Statement(s) could not be
        prepared" after a 1205.  On success ``self.cursor`` is updated to the
        cursor that ran the statement, so callers can immediately
        ``self.cursor.fetchone()`` / ``self.cursor.fetchall()`` to read result
        rows from a SELECT.

        ``drain_status_rows=True`` discards every rowset produced by the
        statement (use for sp_cdc_* procs that emit "(N rows affected)"
        rowsets pymssql leaves pending — the next statement on the same
        connection would otherwise fail with "Statement(s) could not be
        prepared").  Do not pass this for SELECT — it would discard the data.

        Non-1205 exceptions propagate immediately — this is for transient
        contention, not for swallowing real errors.
        """
        last_exc: Exception | None = None
        for attempt in range(max_retries):
            cur = self.connection.cursor()
            try:
                if params is None:
                    cur.execute(query)
                else:
                    cur.execute(query, params)
                if drain_status_rows:
                    while cur.nextset():
                        pass
                self.cursor = cur
                return
            except pymssql.exceptions.OperationalError as e:
                if "1205" in str(e) and attempt < max_retries - 1:
                    last_exc = e
                    logging.warning(
                        f"deadlock victim on attempt {attempt + 1}/"
                        f"{max_retries}, retrying: {e}"
                    )
                    time.sleep(retry_delay * (1.25**attempt))
                    continue
                raise