cursor.py

"""
Copyright (c) Microsoft Corporation.
Licensed under the MIT license.
This module contains the Cursor class, which represents a database cursor.
Resource Management:
- Cursors are tracked by their parent connection.
- Closing the connection will automatically close all open cursors.
- Do not use a cursor after it is closed, or after its parent connection is closed.
- Use close() to release resources held by the cursor as soon as it is no longer needed.
"""

# pylint: disable=too-many-lines  # Large file due to comprehensive DB-API 2.0 implementation

import decimal
import logging
import uuid
import datetime
import warnings
from typing import List, Mapping, Union, Any, Optional, Tuple, Sequence, TYPE_CHECKING, Iterable
from mssql_python.constants import ConstantsDDBC as ddbc_sql_const, SQLTypes
from mssql_python.helpers import check_error, connstr_to_pycore_params
from mssql_python.logging import logger
from mssql_python import ddbc_bindings
from mssql_python.exceptions import (
    InterfaceError,
    NotSupportedError,
    ProgrammingError,
    OperationalError,
    DatabaseError,
)
from mssql_python.row import Row
from mssql_python.perf_timer import perf_phase
from mssql_python import get_settings
from mssql_python.parameter_helper import (
    detect_and_convert_parameters,
    parse_pyformat_params,
    convert_pyformat_to_qmark,
)

if TYPE_CHECKING:
    import pyarrow  # type: ignore
    from mssql_python.connection import Connection
else:
    pyarrow = None

# Constants for string handling
MAX_INLINE_CHAR: int = (
    4000  # NVARCHAR/VARCHAR inline limit; this triggers NVARCHAR(MAX)/VARCHAR(MAX) + DAE
)
SMALLMONEY_MIN: decimal.Decimal = decimal.Decimal("-214748.3648")
SMALLMONEY_MAX: decimal.Decimal = decimal.Decimal("214748.3647")
MONEY_MIN: decimal.Decimal = decimal.Decimal("-922337203685477.5808")
MONEY_MAX: decimal.Decimal = decimal.Decimal("922337203685477.5807")
# SQL BIGINT is a signed 64-bit integer. Ints outside this range have no BIGINT
# encoding and must be rejected at detect time on both paths (see _map_sql_type).
BIGINT_MIN: int = -(2**63)
BIGINT_MAX: int = 2**63 - 1
ODBC3_TEMPORAL_SQL_TYPES = {
    ddbc_sql_const.SQL_DATE.value: ddbc_sql_const.SQL_TYPE_DATE.value,
    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,
}


def _string_only_output_converter(converter):
    """Gate fallback conversion on the fetched value, not its column metadata."""

    def convert(value):
        if isinstance(value, (str, bytes)):
            return converter(value)
        return value

    return convert


def _normalize_time_param(value, c_type):
    """Convert a datetime.time to its isoformat string when bound via text C-types.

    Returns the isoformat string if conversion applies, otherwise *None*.
    """
    if isinstance(value, datetime.time) and c_type in (
        ddbc_sql_const.SQL_C_CHAR.value,
        ddbc_sql_const.SQL_C_WCHAR.value,
    ):
        return value.isoformat(timespec="microseconds")
    return None


class _ArrowReader:
    """RecordBatchReader-compatible wrapper that makes ``close()`` actually
    release server-side resources.

    ``pyarrow.RecordBatchReader.from_batches(...)`` returns a reader whose
    ``close()`` only releases the internal ArrowArrayStream — it does **not**
    propagate into the underlying Python generator and does **not** stop the
    server-side ODBC cursor.  This wrapper closes that gap.

    Interoperability: this class exposes ``__arrow_c_stream__`` (Arrow
    PyCapsule Protocol, pyarrow >= 14), so Arrow-aware consumers
    (``pyarrow.RecordBatchReader.from_stream``, ``polars.from_arrow``,
    ``duckdb.from_arrow``, etc.) can accept it directly without any
    ``isinstance(x, pyarrow.RecordBatchReader)`` check.  Subclassing
    ``pyarrow.RecordBatchReader`` (a Cython extension type) isn't a viable
    alternative because its ``from_batches`` factory returns the base class
    regardless of the subclass, ``__class__`` reassignment is rejected on
    Cython types, and instances cannot hold arbitrary Python attributes —
    so a subclass could not carry the cursor/generator refs this wrapper
    needs for cancellation semantics.

    Design (optimized):
      * The Python generator backing the reader carries its own ``try/finally``
        block — so server-side cleanup runs symmetrically whether the user
        exhausts the reader, calls ``close()`` mid-iteration, exits a ``with``
        block, or just lets the reader be garbage-collected.  ``close()``
        itself only has to (a) call ``SQLCancel`` to unblock any fetch in
        flight on another thread and (b) close the generator; the
        ``finally`` clause does the rest.
      * ``SQLCancel`` is called *before* ``SQLFreeStmt(SQL_CLOSE)`` so a fetch
        running on another thread returns cleanly first.  ``SQLCancel`` is
        the single ODBC entry point (with the diag-record functions) that the
        spec marks as safe to call from a different thread than the one
        owning the statement.
      * Cancellation/error diagnostics are drained before closing the cursor.
        Natural exhaustion uses the native fetch's captured diagnostics; a
        drain after close picks up anything ``SQL_CLOSE`` itself emits.
      * Cached ``pyarrow.ArrowInvalid`` avoids per-read imports on the
        post-close error path.
      * ``__del__`` is guarded against interpreter finalization.
      * The public method surface is *delegated* to the inner pyarrow reader
        via ``__getattr__`` rather than hand-enumerated: any method pyarrow
        provides (``read_all``, ``read_pandas``, ``cast``, ``schema``,
        ``read_next_batch``, and anything added by future pyarrow
        versions) transparently forwards.  Only the methods the wrapper
        genuinely intercepts stay explicit: ``close``, ``closed``,
        ``__arrow_c_stream__``, ``__iter__``/``__next__``,
        ``__enter__``/``__exit__``, ``__del__``.

    The parent ``Cursor`` is **not** closed; it remains fully usable.
    """

    __slots__ = (
        "_cursor",
        "_inner",
        "_generator",
        "_closed",
        "_arrow_invalid",
        "_close_requested",
    )

    def __init__(
        self,
        cursor: "Cursor",
        inner: "pyarrow.RecordBatchReader",
        generator,
        arrow_invalid_exc: type,
        close_requested: list[bool],
    ) -> None:
        self._cursor = cursor
        self._inner = inner
        self._generator = generator
        self._closed = False
        self._close_requested = close_requested
        # Cache the exception class so post-close reads in a hot loop don't
        # re-import pyarrow.
        self._arrow_invalid = arrow_invalid_exc

    # ── Public surface mirroring pyarrow.RecordBatchReader ────────────────

    @property
    def closed(self) -> bool:
        """True once ``close()`` has been called."""
        return self._closed

    def __getattr__(self, name):
        """Delegate any attribute we don't explicitly define to the inner
        ``pyarrow.RecordBatchReader``.

        Rationale: enumerating pyarrow's surface by hand was fragile —
        methods like ``read_all()``, ``read_pandas()``, and ``cast()`` were
        silently missing, breaking existing user code on upgrade, and every
        future addition to ``RecordBatchReader`` would repeat the same
        regression.  ``__getattr__`` is only invoked when normal attribute
        lookup fails, so our explicit overrides (``close``, ``closed``,
        ``__arrow_c_stream__``, iteration and context-manager protocols)
        always win; everything else falls through to the wrapped reader.

        Private / dunder names (leading ``_``) are refused so that a
        partially-constructed instance during ``__del__`` cannot recurse
        forever trying to resolve its own slot names via ``self._inner``.

        Post-close access raises ``pyarrow.ArrowInvalid`` to match the
        behaviour of the explicit ``__next__`` / ``__arrow_c_stream__``
        methods — a reader that has been marked closed must not delegate
        even if a retry-pending state still holds ``self._inner``.
        """
        if name.startswith("_"):
            raise AttributeError(name)
        if self._closed:
            raise self._arrow_invalid("Reader is closed")
        return getattr(self._inner, name)

    def __arrow_c_stream__(self, requested_schema=None):
        """Arrow PyCapsule Protocol — export as an Arrow C stream.

        Implements the Arrow PyCapsule Protocol for streams (pyarrow >= 14),
        so this wrapper can be consumed by any Arrow-compatible library
        (``pyarrow.RecordBatchReader.from_stream``, ``polars.from_arrow``,
        ``duckdb.from_arrow``, ``pandas.api.interchange.from_dataframe`` for
        streams, etc.) without an ``isinstance(x, pa.RecordBatchReader)``
        check.  See
        https://arrow.apache.org/docs/format/CDataInterface/PyCapsuleInterface.html

        Note: once the capsule has been consumed by the caller, the
        underlying pyarrow reader's internal C stream is transferred out;
        further calls to ``read_next_batch()`` on this wrapper will fail
        with ``ArrowInvalid``.  That mirrors pyarrow's own semantics.
        """
        if self._closed:
            raise self._arrow_invalid("Reader is closed")
        # Delegate to the inner pyarrow reader.  pyarrow >= 14 exposes
        # ``__arrow_c_stream__`` directly on ``RecordBatchReader``; older
        # versions do not implement the protocol.  Fail explicitly rather
        # than silently returning something invalid.
        inner_export = getattr(self._inner, "__arrow_c_stream__", None)
        if inner_export is None:
            raise self._arrow_invalid(
                "Arrow PyCapsule Protocol requires pyarrow>=14; "
                "the installed pyarrow version does not expose "
                "RecordBatchReader.__arrow_c_stream__."
            )
        return inner_export(requested_schema)

    def __iter__(self):
        return self

    def __next__(self):
        if self._closed:
            raise self._arrow_invalid("Reader is closed")
        return self._inner.read_next_batch()

    def __enter__(self):
        if self._closed:
            raise self._arrow_invalid("Reader is closed")
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.close()
        return False

    def __del__(self):
        # Best-effort cleanup if the user never called close() (or a previous
        # close() attempt failed to release the generator and left cleanup
        # incomplete) and the reader is being garbage-collected.  Skip during
        # interpreter shutdown — the module globals (pyarrow, ddbc_bindings)
        # may already be torn down, and touching native code at that point is
        # unsafe.
        try:
            import sys as _sys

            if _sys.is_finalizing():
                return
            # Retry whenever the generator is still referenced — covers both
            # "user never called close()" and "earlier close() raised before
            # the generator was released".
            if getattr(self, "_generator", None) is not None:
                self.close()
        except Exception:  # pylint: disable=broad-exception-caught
            pass

    # ── Close implementation ──────────────────────────────────────────────

    def close(self) -> None:
        """Synchronously stop fetching, release the server-side cursor, and
        reset parent-cursor bookkeeping.  Idempotent **and retry-safe**:
        if a previous call raised before the generator was released (for
        example because another thread was still executing it and
        ``generator.close()`` raised ``ValueError: generator already
        executing``), subsequent calls will pick up where the failed call
        left off rather than silently no-op'ing.

        Most of the actual cleanup work lives in the generator's ``finally``
        clause (see ``Cursor.arrow_reader``); this method just unblocks any
        in-flight fetch and closes the generator, which triggers that
        ``finally`` block.
        """
        # Fast path: cleanup already completed on a previous call.  We use
        # the *generator* reference — not ``_closed`` — as the completion
        # marker, because ``_closed`` is flipped early (so racing reads
        # raise) and must not by itself disable retry of failed cleanup.
        if self._generator is None and self._cursor is None:
            self._closed = True
            return

        # Mark closed first so any racing read raises immediately, even if
        # the cleanup steps below fail and we end up retried later.
        self._closed = True
        self._close_requested[0] = True

        # SQLCancel (cross-thread safe) — unblocks a fetch running on another
        # thread so that the generator's finally clause can then run
        # SQLFreeStmt(SQL_CLOSE) without risking the undefined-behaviour
        # window of closing an HSTMT mid-fetch.  Safe no-op for an idle stmt.
        cursor = self._cursor
        if cursor is not None and not cursor.closed and cursor.hstmt is not None:
            try:
                cursor.hstmt._cancel()  # pylint: disable=protected-access
            except Exception as e:  # pylint: disable=broad-exception-caught
                logger.debug("arrow_reader.close: SQLCancel raised: %s", e)

        # Close the generator — this raises GeneratorExit inside it, which
        # runs the try/finally cleanup block (SQLFreeStmt + diag drain +
        # cursor bookkeeping reset).  If close() raises and the generator is
        # still alive (e.g. another thread is currently executing it), keep
        # the reference so a subsequent close() / __del__ can retry; only
        # drop refs once the generator is actually dead.
        gen = self._generator
        if gen is not None:
            try:
                gen.close()
            except Exception as e:  # pylint: disable=broad-exception-caught
                logger.debug("arrow_reader.close: generator.close raised: %s", e)
                if getattr(gen, "gi_frame", None) is not None:
                    # Generator still alive — leave _generator (and _cursor,
                    # so the next retry can re-issue SQLCancel) intact.
                    return
            self._generator = None

        # Drop strong refs so the wrapper does not extend the lifetime of
        # the parent Cursor or the inner pyarrow reader.
        self._cursor = None
        self._inner = None


class Cursor:  # pylint: disable=too-many-instance-attributes,too-many-public-methods
    """
    Represents a database cursor, which is used to manage the context of a fetch operation.

    Attributes:
        connection: Database connection object.
        description: Sequence of 7-item sequences describing one result column.
        rowcount: Number of rows produced or affected by the last execute operation.
        arraysize: Number of rows to fetch at a time with fetchmany().
        rownumber: Track the current row index in the result set.

    Methods:
        __init__(connection_str) -> None.
        callproc(procname, parameters=None) ->
            Modified copy of the input sequence with output parameters.
        close() -> None.
        execute(operation, parameters=None) -> Cursor.
        executemany(operation, seq_of_parameters) -> None.
        fetchone() -> Single sequence or None if no more data is available.
        fetchmany(size=None) -> Sequence of sequences (e.g. list of tuples).
        fetchall() -> Sequence of sequences (e.g. list of tuples).
        nextset() -> True if there is another result set, None otherwise.
        next() -> Fetch the next row from the cursor.
        setinputsizes(sizes) -> None.
        setoutputsize(size, column=None) -> None.
    """

    # TODO(jathakkar): Thread safety considerations
    # The cursor class contains methods that are not thread-safe due to:
    #  1. Methods that mutate cursor state (_reset_cursor, self.description, etc.)
    #  2. Methods that call ODBC functions with shared handles (self.hstmt)
    #
    # These methods should be properly synchronized or redesigned when implementing
    # async functionality to prevent race conditions and data corruption.
    # Consider using locks, redesigning for immutability, or ensuring
    # cursor objects are never shared across threads.

    def __init__(self, connection: "Connection", timeout: int = 0) -> None:
        """
        Initialize the cursor with a database connection.

        Args:
            connection: Database connection object.
            timeout: Query timeout in seconds
        """
        # Establish the close() invariant *first*, before any statement that
        # can raise (notably ``_initialize_cursor`` below).  Setting
        # ``closed=False`` (not True) up front means that if ``__init__``
        # fails partway — even after ``hstmt`` was allocated — a subsequent
        # ``close()`` / ``__del__`` will still see a consistent view and
        # correctly release whatever was allocated.  Pairing it with
        # ``hstmt=None`` keeps ``close()`` safe when allocation fails before
        # an HSTMT exists.
        self.closed: bool = False
        self.hstmt: Optional[Any] = None

        self._connection: "Connection" = connection  # Store as private attribute
        self._timeout: int = timeout
        self._inputsizes: Optional[List[Tuple[int, int, int, int]]] = None
        # self.connection.autocommit = False
        self._initialize_cursor()
        self.description: Optional[
            List[
                Tuple[
                    str,
                    Any,
                    Optional[int],
                    Optional[int],
                    Optional[int],
                    Optional[int],
                    Optional[bool],
                ]
            ]
        ] = None
        self.rowcount: int = -1
        self.arraysize: int = (
            1  # Default number of rows to fetch at a time is 1, user can change it
        )
        self.buffer_length: int = 1024  # Default buffer length for string data
        self._result_set_empty: bool = False  # Add this initialization
        self.last_executed_stmt: str = ""  # Stores the last statement executed by this cursor
        self.is_stmt_prepared: List[bool] = [
            False
        ]  # Indicates if last_executed_stmt was prepared by ddbc shim.
        # Is a list instead of a bool coz bools in Python are immutable.
        # Hence, we can't pass around bools by reference & modify them.
        # Therefore, it must be a list with exactly one bool element.

        self._rownumber = -1  # DB-API extension: last returned row index, -1 before first

        # Column-name -> index map for the current result set. For catalog/metadata
        # result sets this also carries lowercase and friendly aliases (see
        # _prepare_metadata_result_set).
        self._cached_column_map = None
        self._cached_column_map_lower = None
        self._cached_converter_map = None
        self._cached_converters_generation = self._connection._converters_generation
        # Canonical, order-preserving column names snapshotted once per result set
        # and handed to each Row so mapping views never read the live cursor.description
        # (which changes when the cursor is reused for another query). _result_columns_src
        # tracks the self.description identity the snapshot was built from, so a new
        # result set rebuilds it exactly once.
        self._cached_result_columns = None
        self._result_columns_src = None
        # Raw ODBC SQL type codes (from SQLDescribeCol) per column, parallel to
        # self.description. Kept so output-converter dispatch can key on the integer
        # ODBC SQL type code (pyodbc-compatible), not just the mapped Python type. See #684.
        self._column_sql_types = None
        self._uuid_str_indices = None  # Pre-computed UUID column indices for str conversion
        # Cache the effective native_uuid setting for this cursor's connection.
        # Resolution order: connection._native_uuid (if not None) → module-level setting.
        self._conn_native_uuid = getattr(self.connection, "_native_uuid", None)
        self._next_row_index = 0  # internal: index of the next row the driver will return (0-based)
        self._has_result_set = False  # Track if we have an active result set
        self._refresh_decoding_cache()
        self._skip_increment_for_next_fetch = (
            False  # Track if we need to skip incrementing the row index
        )
        self.messages: List[Tuple[str, str]] = []  # Store diagnostic messages

    def _is_unicode_string(self, param: str) -> bool:
        """
        Check if a string contains non-ASCII characters.

        Args:
            param: The string to check.

        Returns:
            True if the string contains non-ASCII characters, False otherwise.
        """
        return not param.isascii()

    def _parse_date(self, param: str) -> Optional[datetime.date]:
        """
        Attempt to parse a string as a date.

        Args:
            param: The string to parse.

        Returns:
            A datetime.date object if parsing is successful, else None.
        """
        formats = ["%Y-%m-%d"]
        for fmt in formats:
            try:
                return datetime.datetime.strptime(param, fmt).date()
            except ValueError:
                continue
        return None

    def _parse_datetime(self, param: str) -> Optional[datetime.datetime]:
        """
        Attempt to parse a string as a datetime, smalldatetime, datetime2, timestamp.

        Args:
            param: The string to parse.

        Returns:
            A datetime.datetime object if parsing is successful, else None.
        """
        formats = [
            "%Y-%m-%dT%H:%M:%S.%f",  # ISO 8601 datetime with fractional seconds
            "%Y-%m-%dT%H:%M:%S",  # ISO 8601 datetime
            "%Y-%m-%d %H:%M:%S.%f",  # Datetime with fractional seconds
            "%Y-%m-%d %H:%M:%S",  # Datetime without fractional seconds
        ]
        for fmt in formats:
            try:
                return datetime.datetime.strptime(param, fmt)  # Valid datetime
            except ValueError:
                continue  # Try next format

        return None  # If all formats fail, return None

    def _parse_time(self, param: str) -> Optional[datetime.time]:
        """
        Attempt to parse a string as a time.

        Args:
            param: The string to parse.

        Returns:
            A datetime.time object if parsing is successful, else None.
        """
        formats = [
            "%H:%M:%S",  # Time only
            "%H:%M:%S.%f",  # Time with fractional seconds
        ]
        for fmt in formats:
            try:
                return datetime.datetime.strptime(param, fmt).time()
            except ValueError:
                continue
        return None

    def _get_numeric_data(self, param: decimal.Decimal) -> Any:
        """
        Get the data for a numeric parameter.

        Args:
            param: The numeric parameter.

        Returns:
            numeric_data: A NumericData struct containing
            the numeric data.
        """
        decimal_as_tuple = param.as_tuple()
        digits_tuple = decimal_as_tuple.digits
        num_digits = len(digits_tuple)
        exponent = decimal_as_tuple.exponent

        # NaN / sNaN / Infinity report a string exponent ('n', 'N', 'F') instead of an
        # int. There is no SQL NUMERIC encoding for them, so refuse here rather than
        # falling through to precision=38 and letting the digit packing below emit a
        # silent zero. The native detection path raises ValueError for the same input.
        if isinstance(exponent, str):
            raise ValueError("Cannot bind non-finite Decimal (NaN/Infinity) as SQL NUMERIC")

        # Calculate the SQL precision & scale
        #   precision = no. of significant digits
        #   scale     = no. digits after decimal point
        if exponent >= 0:
            # digits=314, exp=2 ---> '31400' --> precision=5, scale=0
            precision = num_digits + exponent
            scale = 0
        elif (-1 * exponent) <= num_digits:
            # digits=3140, exp=-3 ---> '3.140' --> precision=4, scale=3
            precision = num_digits
            scale = exponent * -1
        else:
            # digits=3140, exp=-5 ---> '0.03140' --> precision=5, scale=5
            # TODO: double check the precision calculation here with SQL documentation
            precision = exponent * -1
            scale = exponent * -1

        if precision > 38:
            raise ValueError(
                "Precision of the numeric value is too high - "
                + str(param)
                + ". Should be less than or equal to 38"
            )
        Numeric_Data = ddbc_bindings.NumericData
        numeric_data = Numeric_Data()
        numeric_data.scale = scale
        numeric_data.precision = precision
        numeric_data.sign = 1 if decimal_as_tuple.sign == 0 else 0
        # strip decimal point from param & convert the significant digits to integer
        # Ex: 12.34 ---> 1234
        int_str = "".join(str(d) for d in digits_tuple)
        if exponent > 0:
            int_str = int_str + ("0" * exponent)
        elif exponent < 0:
            if -exponent > num_digits:
                int_str = ("0" * (-exponent - num_digits)) + int_str

        if int_str == "":
            int_str = "0"

        # Convert decimal base-10 string to python int, then to 16 little-endian bytes
        big_int = int(int_str)
        byte_array = bytearray(16)  # SQL_MAX_NUMERIC_LEN
        for i in range(16):
            byte_array[i] = big_int & 0xFF
            big_int >>= 8
            if big_int == 0:
                break

        numeric_data.val = bytes(byte_array)
        return numeric_data

    def _get_encoding_settings(self):
        """
        Get the encoding settings from the connection.

        Returns:
            dict: A dictionary with 'encoding' and 'ctype' keys, or default settings if not available

        Raises:
            OperationalError, DatabaseError: If there are unexpected database connection issues
            that indicate a broken connection state. These should not be silently ignored
            as they can lead to data corruption or inconsistent behavior.
        """
        if hasattr(self._connection, "getencoding"):
            try:
                return self._connection.getencoding()
            except (OperationalError, DatabaseError) as db_error:
                # Log the error for debugging but re-raise for fail-fast behavior
                # Silently returning defaults can lead to data corruption and hard-to-debug issues
                logger.error(
                    "Failed to get encoding settings from connection due to database error: %s. "
                    "This indicates a broken connection state that should not be ignored.",
                    db_error,
                )
                # Re-raise to fail fast - users should know their connection is broken
                raise
            except Exception as unexpected_error:
                # Handle other unexpected errors (connection closed, programming errors, etc.)
                logger.error("Unexpected error getting encoding settings: %s", unexpected_error)
                # Re-raise unexpected errors as well
                raise

        # Return default encoding settings if getencoding is not available
        # This is the only case where defaults are appropriate (method doesn't exist)
        return {"encoding": "utf-16le", "ctype": ddbc_sql_const.SQL_WCHAR.value}

    def _refresh_decoding_cache(self):
        """Read decoding settings only when the connection configuration changes."""
        generation = self._connection._decoding_generation
        char_decoding = self._get_decoding_settings(ddbc_sql_const.SQL_CHAR.value)
        wchar_encoding = self._get_decoding_settings(ddbc_sql_const.SQL_WCHAR.value).get(
            "encoding", "utf-16le"
        )
        self._cached_char_encoding = char_decoding.get("encoding", "utf-16le")
        self._cached_char_ctype = char_decoding.get("ctype", ddbc_sql_const.SQL_WCHAR.value)
        self._cached_wchar_encoding = wchar_encoding
        self._cached_decoding_generation = generation

    def _get_decoding_settings(self, sql_type):
        """
        Get decoding settings for a specific SQL type.

        Args:
            sql_type: SQL type constant (SQL_CHAR, SQL_WCHAR, etc.)

        Returns:
            Dictionary containing the decoding settings.

        Raises:
            OperationalError, DatabaseError: If there are unexpected database connection issues
            that indicate a broken connection state. These should not be silently ignored
            as they can lead to data corruption or inconsistent behavior.
        """
        try:
            # Get decoding settings from connection for this SQL type
            return self._connection.getdecoding(sql_type)
        except (OperationalError, DatabaseError) as db_error:
            # Log the error for debugging but re-raise for fail-fast behavior
            # Silently returning defaults can lead to data corruption and hard-to-debug issues
            logger.error(
                "Failed to get decoding settings for SQL type %s due to database error: %s. "
                "This indicates a broken connection state that should not be ignored.",
                sql_type,
                db_error,
            )
            # Re-raise to fail fast - users should know their connection is broken
            raise
        except Exception as unexpected_error:
            # Handle other unexpected errors (connection closed, programming errors, etc.)
            logger.error(
                "Unexpected error getting decoding settings for SQL type %s: %s",
                sql_type,
                unexpected_error,
            )
            # Re-raise unexpected errors as well
            raise

    def _map_sql_type(  # pylint: disable=too-many-arguments,too-many-positional-arguments,too-many-locals,too-many-return-statements,too-many-branches
        self,
        param: Any,
        parameters_list: List[Any],
        i: int,
        min_val: Optional[Any] = None,
        max_val: Optional[Any] = None,
        decimal_as_numeric: bool = False,
    ) -> Tuple[int, int, int, int, bool]:
        """
        Map a Python data type to the corresponding SQL type,
        C type, Column size, and Decimal digits.
        Takes:
            - param: The parameter to map.
            - parameters_list: The list of parameters to bind.
            - i: The index of the parameter in the list.
            - decimal_as_numeric: When True, bind a Decimal as SQL_NUMERIC regardless of
              value, skipping the MONEY/SMALLMONEY-range VARCHAR shortcut. The execute()
              path sets this so a money-range Decimal compared against a numeric column
              does not overflow (GH-740). executemany() leaves it False because it
              string-binds Decimals for the whole batch (GH-503).
        Returns:
            - A tuple containing the SQL type, C type, column size, and decimal digits.
        """
        logger.debug("_map_sql_type: Mapping param index=%d, type=%s", i, type(param).__name__)
        if param is None:
            logger.debug("_map_sql_type: NULL parameter - index=%d", i)
            # GH-610: Send SQL_UNKNOWN_TYPE to C++ where the describe-cache
            # in BindParameters / BindParameterArray resolves the correct
            # type via SQLDescribeParam (cached after first call).
            return (
                ddbc_sql_const.SQL_UNKNOWN_TYPE.value,
                ddbc_sql_const.SQL_C_DEFAULT.value,
                1,
                0,
                False,
            )

        if isinstance(param, bool):
            logger.debug("_map_sql_type: BOOL detected - index=%d", i)
            return (
                ddbc_sql_const.SQL_BIT.value,
                ddbc_sql_const.SQL_C_BIT.value,
                1,
                0,
                False,
            )

        if isinstance(param, int):
            # Use min_val/max_val if available
            value_to_check = max_val if max_val is not None else param
            min_to_check = min_val if min_val is not None else param
            logger.debug(
                "_map_sql_type: INT detected - index=%d, min=%s, max=%s",
                i,
                str(min_to_check)[:50],
                str(value_to_check)[:50],
            )

            if 0 <= min_to_check and value_to_check <= 255:
                logger.debug("_map_sql_type: INT -> TINYINT - index=%d", i)
                return (
                    ddbc_sql_const.SQL_TINYINT.value,
                    ddbc_sql_const.SQL_C_TINYINT.value,
                    3,
                    0,
                    False,
                )
            if -32768 <= min_to_check and value_to_check <= 32767:
                logger.debug("_map_sql_type: INT -> SMALLINT - index=%d", i)
                return (
                    ddbc_sql_const.SQL_SMALLINT.value,
                    ddbc_sql_const.SQL_C_SHORT.value,
                    5,
                    0,
                    False,
                )
            if -2147483648 <= min_to_check and value_to_check <= 2147483647:
                logger.debug("_map_sql_type: INT -> INTEGER - index=%d", i)
                return (
                    ddbc_sql_const.SQL_INTEGER.value,
                    ddbc_sql_const.SQL_C_LONG.value,
                    10,
                    0,
                    False,
                )
            # Beyond INTEGER, the only integer SQL type is BIGINT (signed 64-bit). An int
            # outside its range cannot bind, so reject here with a clear message rather than
            # labelling it BIGINT and failing later at the C++ cast. Mirrors the native path.
            if value_to_check > BIGINT_MAX or min_to_check < BIGINT_MIN:
                offending = value_to_check if value_to_check > BIGINT_MAX else min_to_check
                raise ValueError(
                    f"integer {offending} is out of range for SQL BIGINT [-2^63, 2^63-1]"
                )
            logger.debug("_map_sql_type: INT -> BIGINT - index=%d", i)
            return (
                ddbc_sql_const.SQL_BIGINT.value,
                ddbc_sql_const.SQL_C_SBIGINT.value,
                19,
                0,
                False,
            )

        if isinstance(param, float):
            logger.debug("_map_sql_type: FLOAT detected - index=%d", i)
            return (
                ddbc_sql_const.SQL_DOUBLE.value,
                ddbc_sql_const.SQL_C_DOUBLE.value,
                15,
                0,
                False,
            )

        if isinstance(param, decimal.Decimal):
            logger.debug("_map_sql_type: DECIMAL detected - index=%d", i)
            # First check precision limit for all decimal values
            decimal_as_tuple = param.as_tuple()
            digits_tuple = decimal_as_tuple.digits
            num_digits = len(digits_tuple)
            exponent = decimal_as_tuple.exponent

            # NaN / sNaN / Infinity report a string exponent ('n', 'N', 'F'). Reject them
            # before the MONEY range comparison below, which would otherwise raise
            # decimal.InvalidOperation for NaN, and before _get_numeric_data, which used to
            # fail with TypeError for Infinity. The native detection path raises ValueError
            # for the same input, so both paths now agree on type and message.
            if isinstance(exponent, str):
                logger.debug(
                    "_map_sql_type: DECIMAL non-finite value - index=%d, exponent=%s", i, exponent
                )
                raise ValueError("Cannot bind non-finite Decimal (NaN/Infinity) as SQL NUMERIC")

            # Calculate the SQL precision (same logic as _get_numeric_data)
            if exponent >= 0:
                precision = num_digits + exponent
            elif (-1 * exponent) <= num_digits:
                precision = num_digits
            else:
                precision = exponent * -1
            logger.debug(
                "_map_sql_type: DECIMAL precision calculated - index=%d, precision=%d",
                i,
                precision,
            )

            if precision > 38:
                logger.debug(
                    "_map_sql_type: DECIMAL precision too high - index=%d, precision=%d",
                    i,
                    precision,
                )
                raise ValueError(
                    f"Precision of the numeric value is too high. "
                    f"The maximum precision supported by SQL Server is 38, but got {precision}."
                )

            # Detect MONEY / SMALLMONEY range. Skipped on the execute() path
            # (decimal_as_numeric=True), where a money-range Decimal must bind as
            # SQL_NUMERIC so a comparison against a smaller numeric column returns no
            # match instead of overflowing (GH-740). executemany keeps the VARCHAR
            # shortcut because it string-binds Decimals for the batch (GH-503).
            if not decimal_as_numeric and SMALLMONEY_MIN <= param <= SMALLMONEY_MAX:
                logger.debug("_map_sql_type: DECIMAL -> SMALLMONEY - index=%d", i)
                # smallmoney
                parameters_list[i] = format(param, "f")
                return (
                    ddbc_sql_const.SQL_VARCHAR.value,
                    ddbc_sql_const.SQL_C_CHAR.value,
                    len(parameters_list[i]),
                    0,
                    False,
                )
            if not decimal_as_numeric and MONEY_MIN <= param <= MONEY_MAX:
                logger.debug("_map_sql_type: DECIMAL -> MONEY - index=%d", i)
                # money
                parameters_list[i] = format(param, "f")
                return (
                    ddbc_sql_const.SQL_VARCHAR.value,
                    ddbc_sql_const.SQL_C_CHAR.value,
                    len(parameters_list[i]),
                    0,
                    False,
                )
            # fallback to generic numeric binding
            logger.debug("_map_sql_type: DECIMAL -> NUMERIC - index=%d", i)
            parameters_list[i] = self._get_numeric_data(param)
            logger.debug(
                "_map_sql_type: NUMERIC created - index=%d, precision=%d, scale=%d",
                i,
                parameters_list[i].precision,
                parameters_list[i].scale,
            )
            return (
                ddbc_sql_const.SQL_NUMERIC.value,
                ddbc_sql_const.SQL_C_NUMERIC.value,
                parameters_list[i].precision,
                parameters_list[i].scale,
                False,
            )

        if isinstance(param, uuid.UUID):
            logger.debug("_map_sql_type: UUID detected - index=%d", i)
            parameters_list[i] = param.bytes_le
            return (
                ddbc_sql_const.SQL_GUID.value,
                ddbc_sql_const.SQL_C_GUID.value,
                16,
                0,
                False,
            )

        if isinstance(param, str):
            logger.debug("_map_sql_type: STR detected - index=%d, length=%d", i, len(param))
            if (
                param.startswith("POINT")
                or param.startswith("LINESTRING")
                or param.startswith("POLYGON")
            ):
                logger.debug("_map_sql_type: STR is geometry type - index=%d", i)
                return (
                    ddbc_sql_const.SQL_WVARCHAR.value,
                    ddbc_sql_const.SQL_C_WCHAR.value,
                    len(param),
                    0,
                    False,
                )

            # String mapping logic here
            is_unicode = self._is_unicode_string(param)

            # Computes UTF-16 code units (handles surrogate pairs)
            utf16_len = sum(2 if ord(c) > 0xFFFF else 1 for c in param)
            logger.debug(
                "_map_sql_type: STR analysis - index=%d, is_unicode=%s, utf16_len=%d",
                i,
                str(is_unicode),
                utf16_len,
            )
            if utf16_len > MAX_INLINE_CHAR:  # Long strings -> DAE
                logger.debug("_map_sql_type: STR exceeds MAX_INLINE_CHAR, using DAE - index=%d", i)
                if is_unicode:
                    return (
                        ddbc_sql_const.SQL_WVARCHAR.value,
                        ddbc_sql_const.SQL_C_WCHAR.value,
                        0,
                        0,
                        True,
                    )
                return (
                    ddbc_sql_const.SQL_VARCHAR.value,
                    ddbc_sql_const.SQL_C_CHAR.value,
                    0,
                    0,
                    True,
                )

            # Short strings
            if is_unicode:
                return (
                    ddbc_sql_const.SQL_WVARCHAR.value,
                    ddbc_sql_const.SQL_C_WCHAR.value,
                    utf16_len,
                    0,
                    False,
                )
            return (
                ddbc_sql_const.SQL_VARCHAR.value,
                ddbc_sql_const.SQL_C_CHAR.value,
                len(param),
                0,
                False,
            )

        if isinstance(param, (bytes, bytearray)):
            length = len(param)
            if length > 8000:  # Use VARBINARY(MAX) for large blobs
                return (
                    ddbc_sql_const.SQL_VARBINARY.value,
                    ddbc_sql_const.SQL_C_BINARY.value,
                    0,
                    0,
                    True,
                )
            # Small blobs → direct binding
            return (
                ddbc_sql_const.SQL_VARBINARY.value,
                ddbc_sql_const.SQL_C_BINARY.value,
                max(length, 1),
                0,
                False,
            )

        if isinstance(param, datetime.datetime):
            if param.tzinfo is not None:
                # Timezone-aware datetime -> DATETIMEOFFSET
                return (
                    ddbc_sql_const.SQL_DATETIMEOFFSET.value,
                    ddbc_sql_const.SQL_C_SS_TIMESTAMPOFFSET.value,
                    34,
                    7,
                    False,
                )
            # Naive datetime -> TIMESTAMP
            return (
                ddbc_sql_const.SQL_TYPE_TIMESTAMP.value,
                ddbc_sql_const.SQL_C_TYPE_TIMESTAMP.value,
                26,
                6,
                False,
            )

        if isinstance(param, datetime.date):
            return (
                ddbc_sql_const.SQL_TYPE_DATE.value,
                ddbc_sql_const.SQL_C_TYPE_DATE.value,
                10,
                0,
                False,
            )

        if isinstance(param, datetime.time):
            return (
                ddbc_sql_const.SQL_TYPE_TIME.value,
                ddbc_sql_const.SQL_C_CHAR.value,
                16,
                6,
                False,
            )

        # For safety: unknown/unhandled Python types should not silently go to SQL
        raise TypeError(
            "Unsupported parameter type: The driver cannot safely convert it to a SQL type."
        )

    def _initialize_cursor(self) -> None:
        """
        Initialize the DDBC statement handle.
        """
        self._allocate_statement_handle()
        self._set_timeout()

    def _allocate_statement_handle(self) -> None:
        """
        Allocate the DDBC statement handle.
        """
        self.hstmt = self._connection._conn.alloc_statement_handle()

    def _set_timeout(self) -> None:
        """
        Set the query timeout attribute on the statement handle.
        This is called once when the cursor is created and after any handle reallocation.
        Following pyodbc's approach for better performance.
        """
        if self._timeout > 0:
            logger.debug("_set_timeout: Setting query timeout=%d seconds", self._timeout)
            try:
                timeout_value = int(self._timeout)
                ret = ddbc_bindings.DDBCSQLSetStmtAttr(
                    self.hstmt,
                    ddbc_sql_const.SQL_ATTR_QUERY_TIMEOUT.value,
                    timeout_value,
                )
                check_error(ddbc_sql_const.SQL_HANDLE_STMT.value, self.hstmt, ret)
                logger.debug("Query timeout set to %d seconds", timeout_value)
            except Exception as e:  # pylint: disable=broad-exception-caught
                logger.warning("Failed to set query timeout: %s", str(e))

    def _reset_cursor(self) -> None:
        """
        Reset the DDBC statement handle.
        """
        if self.hstmt:
            self.hstmt.free()
            self.hstmt = None
            logger.debug("SQLFreeHandle succeeded")

        self._clear_rownumber()

        # Reinitialize the statement handle
        self._initialize_cursor()
        self.is_stmt_prepared = [False]

    def _soft_reset_cursor(self) -> None:
        """Lightweight reset: close cursor and unbind params without freeing the HSTMT.

        Preserves the prepared statement plan on the server so repeated
        executions of the same SQL skip SQLPrepare entirely.
        """
        if self.hstmt:
            ret = ddbc_bindings.DDBCSQLResetStmt(self.hstmt)
            try:
                check_error(ddbc_sql_const.SQL_HANDLE_STMT.value, self.hstmt, ret)
            except Exception:
                logger.warning("_soft_reset_cursor failed; falling back to full reset")
                self._reset_cursor()
                self.last_executed_stmt = ""
                return
        self._clear_rownumber()

    def close(self) -> None:
        """
        Close the connection now (rather than whenever .__del__() is called).
        Idempotent: subsequent calls have no effect and will be no-ops.

        The cursor will be unusable from this point forward; an InterfaceError
        will be raised if any operation (other than close) is attempted with the cursor.
        This is a deviation from pyodbc, which raises an exception if the cursor is already closed.
        """
        # ``closed`` may be missing on the instance only if the object was
        # built via ``Cursor.__new__(Cursor)`` (no ``__init__``).  Normal
        # construction sets ``self.closed = False`` as the first statement
        # of ``__init__``, so any partially-initialized instance still has
        # the attribute.  Treat the no-``__init__`` case as "already closed"
        # — there's nothing to release — and let GC reap the object cleanly.
        if getattr(self, "closed", True):
            # Do nothing - not calling _check_closed() here since we want this to be idempotent
            return

        # Clear messages per DBAPI
        self.messages = []

        # Remove this cursor from the connection's tracking
        if hasattr(self, "connection") and self.connection and hasattr(self.connection, "_cursors"):
            try:
                self.connection._cursors.discard(self)
            except Exception as e:  # pylint: disable=broad-exception-caught
                logger.warning("Error removing cursor from connection tracking: %s", e)

        if self.hstmt:
            self.hstmt.free()
            self.hstmt = None
            logger.debug("SQLFreeHandle succeeded")
        self._clear_rownumber()
        self.closed = True

    def _check_closed(self) -> None:
        """
        Check if the cursor is closed and raise an exception if it is.

        Raises:
            ProgrammingError: If the cursor is closed.
        """
        if self.closed:
            raise ProgrammingError(
                driver_error="Operation cannot be performed: The cursor is closed.",
                ddbc_error="",
            )

    def _capture_diagnostics(self, ret: int) -> None:
        """Append diagnostic messages to self.messages when the return code
        indicates records may be present.

        Captures on SQL_SUCCESS_WITH_INFO (info/warning messages) and
        SQL_NO_DATA (trailing diagnostics attached by SQLMoreResults,
        e.g. a PRINT after the final result set).

        Skips SQL_SUCCESS to avoid the ~10 ms overhead of scanning the
        driver's internal state when no records exist.  SQL_ERROR is
        handled separately by check_error() which extracts diagnostics
        and raises.

        Composite native fetches capture intermediate diagnostics directly
        into self.messages before subsequent ODBC calls can replace them.
        """
        if self.hstmt and ret in (
            ddbc_sql_const.SQL_SUCCESS_WITH_INFO.value,
            ddbc_sql_const.SQL_NO_DATA.value,
        ):
            self.messages.extend(ddbc_bindings.DDBCSQLGetAllDiagRecords(self.hstmt))

    def _ensure_pyarrow(self) -> Any:
        """
        Import and return pyarrow or raise ImportError accordingly.
        """
        try:
            import pyarrow

            return pyarrow
        except ImportError as e:
            raise ImportError(
                "pyarrow is required for Arrow fetch methods. Please install pyarrow."
            ) from e

    def setinputsizes(self, sizes: List[Union[int, tuple]]) -> None:
        """
        Sets the type information to be used for parameters in execute and executemany.

        This method can be used to explicitly declare the types and sizes of query parameters.
        For example:

        sql = "INSERT INTO product (item, price) VALUES (?, ?)"
        params = [('bicycle', 499.99), ('ham', 17.95)]
        # specify that parameters are for NVARCHAR(50) and DECIMAL(18,4) columns
        cursor.setinputsizes([(SQL_WVARCHAR, 50, 0), (SQL_DECIMAL, 18, 4)])
        cursor.executemany(sql, params)

        Note:
            SQL_SS_UDT parameters require a statement whose parameter metadata
            SQL Server can describe. The driver discovers the UDT identity before
            binding; this may require a metadata round trip. Temporary tables and
            table variables may not be describable, and discovery errors are
            propagated. For built-in spatial types, bind serialized bytes without
            a SQL_SS_UDT override or use a SQL constructor such as
            hierarchyid::Parse(?) with a text parameter instead.

            When inserting NULL into BINARY/VARBINARY columns in temp tables (#table)
            or table variables, SQLDescribeParam cannot resolve the column type and
            falls back to SQL_VARCHAR. This causes an implicit conversion error from
            SQL Server. Use setinputsizes to explicitly specify types for ALL
            parameters::

                from mssql_python.constants import ConstantsDDBC
                cursor.setinputsizes([
                    (ConstantsDDBC.SQL_INTEGER.value, 0, 0),
                    (ConstantsDDBC.SQL_VARBINARY.value, 100, 0)
                ])
                cursor.execute("INSERT INTO #t (id, data) VALUES (?, ?)", [1, None])

        Args:
            sizes: A sequence of tuples, one for each parameter. Each tuple contains
                (sql_type, size, decimal_digits) where size and decimal_digits are optional.
        """

        # Get valid SQL types from centralized constants
        valid_sql_types = SQLTypes.get_valid_types()

        self._inputsizes = []

        if sizes:
            for size_info in sizes:
                if isinstance(size_info, tuple):
                    # Handle tuple format (sql_type, size, decimal_digits)
                    if len(size_info) == 1:
                        sql_type = size_info[0]
                        column_size = 0
                        decimal_digits = 0
                    elif len(size_info) == 2:
                        sql_type, column_size = size_info
                        decimal_digits = 0
                    elif len(size_info) >= 3:
                        sql_type, column_size, decimal_digits = size_info

                    # Validate SQL type
                    if not isinstance(sql_type, int) or sql_type not in valid_sql_types:
                        raise ValueError(
                            f"Invalid SQL type: {sql_type}. Must be a valid SQL type constant."
                        )

                    sql_type = ODBC3_TEMPORAL_SQL_TYPES.get(sql_type, sql_type)

                    # Validate size and precision
                    if not isinstance(column_size, int) or column_size < 0:
                        raise ValueError(
                            f"Invalid column size: {column_size}. Must be a non-negative integer."
                        )

                    if not isinstance(decimal_digits, int) or decimal_digits < 0:
                        raise ValueError(
                            f"Invalid decimal digits: {decimal_digits}. "
                            f"Must be a non-negative integer."
                        )

                    self._inputsizes.append(
                        (
                            sql_type,
                            self._get_c_type_for_sql_type(sql_type),
                            column_size,
                            decimal_digits,
                        )
                    )
                else:
                    # Handle single value (just sql_type)
                    sql_type = size_info

                    # Validate SQL type
                    if not isinstance(sql_type, int) or sql_type not in valid_sql_types:
                        raise ValueError(
                            f"Invalid SQL type: {sql_type}. Must be a valid SQL type constant."
                        )

                    sql_type = ODBC3_TEMPORAL_SQL_TYPES.get(sql_type, sql_type)

                    self._inputsizes.append(
                        (sql_type, self._get_c_type_for_sql_type(sql_type), 0, 0)
                    )

    def _reset_inputsizes(self) -> None:
        """Reset input sizes after execution"""
        self._inputsizes = None

    # Pre-built constant lookup table — avoids rebuilding ~30 entries on every call.
    # Used by setinputsizes fallback path (PR #549 fast path doesn't need this).
    _SQL_TO_C_TYPE = None

    @classmethod
    def _get_sql_to_c_type_map(cls):
        if cls._SQL_TO_C_TYPE is None:
            cls._SQL_TO_C_TYPE = {
                ddbc_sql_const.SQL_CHAR.value: ddbc_sql_const.SQL_C_CHAR.value,
                ddbc_sql_const.SQL_VARCHAR.value: ddbc_sql_const.SQL_C_CHAR.value,
                ddbc_sql_const.SQL_LONGVARCHAR.value: ddbc_sql_const.SQL_C_CHAR.value,
                ddbc_sql_const.SQL_WCHAR.value: ddbc_sql_const.SQL_C_WCHAR.value,
                ddbc_sql_const.SQL_WVARCHAR.value: ddbc_sql_const.SQL_C_WCHAR.value,
                ddbc_sql_const.SQL_WLONGVARCHAR.value: ddbc_sql_const.SQL_C_WCHAR.value,
                ddbc_sql_const.SQL_DECIMAL.value: ddbc_sql_const.SQL_C_NUMERIC.value,
                ddbc_sql_const.SQL_NUMERIC.value: ddbc_sql_const.SQL_C_NUMERIC.value,
                ddbc_sql_const.SQL_BIT.value: ddbc_sql_const.SQL_C_BIT.value,
                ddbc_sql_const.SQL_TINYINT.value: ddbc_sql_const.SQL_C_TINYINT.value,
                ddbc_sql_const.SQL_SMALLINT.value: ddbc_sql_const.SQL_C_SHORT.value,
                ddbc_sql_const.SQL_INTEGER.value: ddbc_sql_const.SQL_C_LONG.value,
                ddbc_sql_const.SQL_BIGINT.value: ddbc_sql_const.SQL_C_SBIGINT.value,
                ddbc_sql_const.SQL_REAL.value: ddbc_sql_const.SQL_C_FLOAT.value,
                ddbc_sql_const.SQL_FLOAT.value: ddbc_sql_const.SQL_C_DOUBLE.value,
                ddbc_sql_const.SQL_DOUBLE.value: ddbc_sql_const.SQL_C_DOUBLE.value,
                ddbc_sql_const.SQL_BINARY.value: ddbc_sql_const.SQL_C_BINARY.value,
                ddbc_sql_const.SQL_VARBINARY.value: ddbc_sql_const.SQL_C_BINARY.value,
                ddbc_sql_const.SQL_LONGVARBINARY.value: ddbc_sql_const.SQL_C_BINARY.value,
                ddbc_sql_const.SQL_SS_UDT.value: ddbc_sql_const.SQL_C_BINARY.value,
                ddbc_sql_const.SQL_TYPE_DATE.value: ddbc_sql_const.SQL_C_TYPE_DATE.value,
                ddbc_sql_const.SQL_TYPE_TIME.value: ddbc_sql_const.SQL_C_TYPE_TIME.value,
                ddbc_sql_const.SQL_TYPE_TIMESTAMP.value: ddbc_sql_const.SQL_C_TYPE_TIMESTAMP.value,
                ddbc_sql_const.SQL_SS_TIME2.value: ddbc_sql_const.SQL_C_TYPE_TIME.value,
                ddbc_sql_const.SQL_DATETIMEOFFSET.value: ddbc_sql_const.SQL_C_SS_TIMESTAMPOFFSET.value,
                ddbc_sql_const.SQL_DATE.value: ddbc_sql_const.SQL_C_TYPE_DATE.value,
                ddbc_sql_const.SQL_TIME.value: ddbc_sql_const.SQL_C_TYPE_TIME.value,
                ddbc_sql_const.SQL_TIMESTAMP.value: ddbc_sql_const.SQL_C_TYPE_TIMESTAMP.value,
                ddbc_sql_const.SQL_GUID.value: ddbc_sql_const.SQL_C_GUID.value,
                ddbc_sql_const.SQL_SS_XML.value: ddbc_sql_const.SQL_C_WCHAR.value,
                ddbc_sql_const.SQL_SS_VARIANT.value: ddbc_sql_const.SQL_C_BINARY.value,
            }
        return cls._SQL_TO_C_TYPE

    def _get_c_type_for_sql_type(self, sql_type: int) -> int:
        """Map SQL type to appropriate C type for parameter binding."""
        return self._get_sql_to_c_type_map().get(sql_type, ddbc_sql_const.SQL_C_DEFAULT.value)

    def _create_parameter_types_list(  # pylint: disable=too-many-arguments,too-many-positional-arguments
        self,
        parameter: Any,
        param_info: Optional[Tuple[Any, ...]],
        parameters_list: List[Any],
        i: int,
        min_val: Optional[Any] = None,
        max_val: Optional[Any] = None,
        decimal_as_numeric: bool = False,
    ) -> Tuple[int, int, int, int, bool]:
        """
        Maps parameter types for the given parameter.

        Python-side type detection for executemany(). The standard execute() path
        uses DetectParamTypes in C++ instead. This helper goes away once executemany
        has native columnwise detection.

        Args:
            parameter: parameter to bind.
        Returns:
            paraminfo.
        """
        paraminfo = param_info()

        sql_type, c_type, column_size, decimal_digits, is_dae = self._map_sql_type(
            parameter,
            parameters_list,
            i,
            min_val=min_val,
            max_val=max_val,
            decimal_as_numeric=decimal_as_numeric,
        )

        # If TIME values are being bound via text C-types, normalize them to a
        # textual representation expected by SQL_C_CHAR/SQL_C_WCHAR binding.
        time_text = _normalize_time_param(parameter, c_type)
        if time_text is not None:
            parameters_list[i] = time_text
            column_size = max(column_size, len(time_text))

        paraminfo.paramCType = c_type
        paraminfo.paramSQLType = sql_type
        paraminfo.inputOutputType = ddbc_sql_const.SQL_PARAM_INPUT.value
        paraminfo.columnSize = column_size
        paraminfo.decimalDigits = decimal_digits
        paraminfo.isDAE = is_dae

        if is_dae:
            paraminfo.dataPtr = parameter  # Will be converted to py::object* in C++

        return paraminfo

    def _initialize_description(self, column_metadata: Optional[Any] = None) -> None:
        """Initialize the description attribute from column metadata."""
        if not column_metadata:
            self.description = None
            self._column_sql_types = None
            return

        description = []
        # Raw ODBC SQL type codes, parallel to description, for output-converter
        # dispatch by integer SQL type (see _build_converter_map / #684).
        sql_type_codes = []
        for _, col in enumerate(column_metadata):
            # Get column name - lowercase it if the lowercase flag is set
            column_name = col["ColumnName"]

            # Use the current global setting to ensure tests pass correctly
            if get_settings().lowercase:
                column_name = column_name.lower()

            sql_type_codes.append(col["DataType"])
            # Add to description tuple (7 elements as per PEP-249)
            description.append(
                (
                    column_name,  # name
                    self._map_data_type(col["DataType"]),  # type_code
                    None,  # display_size
                    col["ColumnSize"],  # internal_size
                    col["ColumnSize"],  # precision - should match ColumnSize
                    col["DecimalDigits"],  # scale
                    col["Nullable"] == ddbc_sql_const.SQL_NULLABLE.value,  # null_ok
                )
            )
        self.description = description
        self._column_sql_types = sql_type_codes

    def _build_converter_map(self):
        """
        Build a pre-computed converter map for output converters.
        Returns a list where each element is either a converter function or None.
        An empty tuple means no converters apply; None is reserved for uncached
        direct Row construction and its legacy connection lookup.
        String fallback converters check each value's type because variant and
        unknown SQL types can have str metadata but non-string values.
        This eliminates the need to look up converters for every row.
        """
        generation = self._connection._converters_generation
        if (
            not self.description
            or not hasattr(self.connection, "_output_converters")
            or not self.connection._output_converters
        ):
            self._cached_converters_generation = generation
            return ()

        sql_type_codes = self._column_sql_types
        converter_map = []

        for i, desc in enumerate(self.description):
            if desc is None:
                converter_map.append(None)
                continue
            converter = None
            # 1) pyodbc-compatible: dispatch on the raw ODBC integer SQL type code
            #    (e.g. SQL_DECIMAL). This is the key add_output_converter documents. #684
            if sql_type_codes is not None and i < len(sql_type_codes):
                converter = self.connection.get_output_converter(sql_type_codes[i])
            # 2) fall back to the Python type stored in description[i][1]
            #    (e.g. decimal.Decimal) - the pre-existing mssql-python key style.
            if converter is None:
                converter = self.connection.get_output_converter(desc[1])
            # 3) The WVARCHAR fallback must also check the value: SQL_VARIANT and
            #    unknown types map to str but can return non-string Python values.
            if converter is None and desc[1] in (str, bytes):
                converter = self.connection.get_output_converter(ddbc_sql_const.SQL_WVARCHAR.value)
                if converter:
                    converter = _string_only_output_converter(converter)

            converter_map.append(converter)

        self._cached_converters_generation = generation
        return converter_map if any(converter is not None for converter in converter_map) else ()

    def _compute_uuid_str_indices(self):
        """
        Compute the tuple of column indices whose uuid.UUID values should be
        stringified (as uppercase), based on the effective native_uuid setting.

        Resolution order: connection-level (if set) → module-level (fallback).

        Returns:
            tuple of int or None: Column indices to stringify, or None when
            native_uuid is True — meaning zero per-row overhead.
        """
        if not self.description:
            return None

        effective_native_uuid = (
            self._conn_native_uuid
            if self._conn_native_uuid is not None
            else get_settings().native_uuid
        )
        if not effective_native_uuid:
            indices = tuple(
                i for i, desc in enumerate(self.description) if desc and desc[1] is uuid.UUID
            )
            return indices if indices else None
        return None

    def _get_column_and_converter_maps(self):
        """
        Get column map and converter map for Row construction (thread-safe).
        This centralizes the column map building logic to eliminate duplication
        and ensure thread-safe lazy initialization.

        Returns:
            tuple: (column_map, converter_map)
        """
        # Thread-safe lazy initialization of column map
        column_map = self._cached_column_map
        if column_map is None and self.description:
            # Build column map locally first, then assign to cache
            column_map = {col_desc[0]: i for i, col_desc in enumerate(self.description)}
            self._cached_column_map = column_map
            self._cached_column_map_lower = (
                {k.lower(): v for k, v in column_map.items()} if get_settings().lowercase else None
            )

        # Fallback to legacy column name map if no cached map
        column_map = column_map or getattr(self, "_column_name_map", None)

        # Refresh once per settings change, not once per row.
        if self._cached_converters_generation != self._connection._converters_generation:
            self._cached_converter_map = self._build_converter_map()
        converter_map = getattr(self, "_cached_converter_map", None)

        # Snapshot canonical column names once per result set (identity-tracked against
        # self.description) so each Row carries stable names even after the cursor is reused.
        if self.description is not None and self._result_columns_src is not self.description:
            self._cached_result_columns = tuple(col_desc[0] for col_desc in self.description)
            self._result_columns_src = self.description

        return column_map, converter_map, self._cached_column_map_lower

    def _map_data_type(self, sql_type):
        """
        Map SQL data type to Python data type.

        Maps the ODBC SQL type code returned by SQLDescribeCol to the
        corresponding Python type for cursor.description[i][1].

        The ODBC 18 driver for SQL Server reports these type codes:
          Standard ODBC 3.x types:
            SQL_CHAR(1), SQL_VARCHAR(12), SQL_LONGVARCHAR(-1),
            SQL_WCHAR(-8), SQL_WVARCHAR(-9), SQL_WLONGVARCHAR(-10),
            SQL_INTEGER(4), SQL_SMALLINT(5), SQL_TINYINT(-6), SQL_BIGINT(-5),
            SQL_BIT(-7), SQL_FLOAT(6), SQL_REAL(7), SQL_DOUBLE(8),
            SQL_DECIMAL(3), SQL_NUMERIC(2),
            SQL_BINARY(-2), SQL_VARBINARY(-3), SQL_LONGVARBINARY(-4),
            SQL_TYPE_DATE(91), SQL_TYPE_TIME(92), SQL_TYPE_TIMESTAMP(93), SQL_GUID(-11)
          SQL Server-specific types (from msodbcsql.h):
            SQL_SS_TIME2(-154)  for time columns
            SQL_DATETIMEOFFSET(-155)  for datetimeoffset columns
            SQL_SS_XML(-152)  for xml columns
            SQL_SS_UDT(-151)  for geography/geometry/hierarchyid columns

        ODBC 2.x aliases (9, 10, 11) are also accepted defensively.

        Args:
            sql_type: SQL data type code from SQLDescribeCol.

        Returns:
            Corresponding Python data type.
        """
        sql_to_python_type = {
            # String types
            ddbc_sql_const.SQL_CHAR.value: str,
            ddbc_sql_const.SQL_VARCHAR.value: str,
            ddbc_sql_const.SQL_LONGVARCHAR.value: str,
            ddbc_sql_const.SQL_WCHAR.value: str,
            ddbc_sql_const.SQL_WVARCHAR.value: str,
            ddbc_sql_const.SQL_WLONGVARCHAR.value: str,
            # Integer types
            ddbc_sql_const.SQL_INTEGER.value: int,
            ddbc_sql_const.SQL_SMALLINT.value: int,
            ddbc_sql_const.SQL_TINYINT.value: int,
            ddbc_sql_const.SQL_BIGINT.value: int,
            # Floating-point types
            ddbc_sql_const.SQL_FLOAT.value: float,
            ddbc_sql_const.SQL_DOUBLE.value: float,
            ddbc_sql_const.SQL_REAL.value: float,
            # Exact numeric types
            ddbc_sql_const.SQL_DECIMAL.value: decimal.Decimal,
            ddbc_sql_const.SQL_NUMERIC.value: decimal.Decimal,
            # Date/time types — values the ODBC 18 driver actually reports
            ddbc_sql_const.SQL_TYPE_DATE.value: datetime.date,  # 91 — date
            ddbc_sql_const.SQL_TYPE_TIME.value: datetime.time,  # 92 — time (ODBC 3.x)
            ddbc_sql_const.SQL_TYPE_TIMESTAMP.value: datetime.datetime,  # 93 — datetime/datetime2/smalldatetime
            ddbc_sql_const.SQL_SS_TIME2.value: datetime.time,  # -154 — time
            ddbc_sql_const.SQL_DATETIMEOFFSET.value: datetime.datetime,  # -155 — datetimeoffset
            # ODBC 2.x date/time aliases (defensive, in case any driver reports these)
            ddbc_sql_const.SQL_DATE.value: datetime.date,  # 9
            ddbc_sql_const.SQL_TIME.value: datetime.time,  # 10
            ddbc_sql_const.SQL_TIMESTAMP.value: datetime.datetime,  # 11
            # Boolean
            ddbc_sql_const.SQL_BIT.value: bool,
            # Binary types
            ddbc_sql_const.SQL_BINARY.value: bytes,
            ddbc_sql_const.SQL_VARBINARY.value: bytes,
            ddbc_sql_const.SQL_LONGVARBINARY.value: bytes,
            ddbc_sql_const.SQL_SS_UDT.value: bytes,
            # UUID
            ddbc_sql_const.SQL_GUID.value: uuid.UUID,
            # XML — driver reports SQL_SS_XML (-152), fetched as str
            ddbc_sql_const.SQL_SS_XML.value: str,
        }
        return sql_to_python_type.get(sql_type, str)

    @property
    def rownumber(self) -> int:
        """
        DB-API extension: Current 0-based index of the cursor in the result set.

        Returns:
            int or None: The current 0-based index of the cursor in the result set,
                        or None if no row has been fetched yet or the index cannot be determined.

        Note:
            - Returns -1 before the first successful fetch
            - Returns 0 after fetching the first row
            - Returns -1 for empty result sets (since no rows can be fetched)

        Warning:
            This is a DB-API extension and may not be portable across different
            database modules.
        """
        # Use mssql_python logging system instead of standard warnings
        logger.warning("DB-API extension cursor.rownumber used")

        # Return None if cursor is closed or no result set is available
        if self.closed or not self._has_result_set:
            return -1

        return self._rownumber  # Will be None until first fetch, then 0, 1, 2, etc.

    @property
    def connection(self) -> "Connection":
        """
        DB-API 2.0 attribute: Connection object that created this cursor.

        This is a read-only reference to the Connection object that was used to create
        this cursor. This attribute is useful for polymorphic code that needs access
        to connection-level functionality.

        Returns:
            Connection: The connection object that created this cursor.

        Note:
            This attribute is read-only as specified by DB-API 2.0. Attempting to
            assign to this attribute will raise an AttributeError.
        """
        return self._connection

    def _reset_rownumber(self) -> None:
        """Reset the rownumber tracking when starting a new result set."""
        self._rownumber = -1
        self._next_row_index = 0
        self._has_result_set = True
        self._skip_increment_for_next_fetch = False

    def _increment_rownumber(self):
        """
        Called after a successful fetch from the driver. Keep both counters consistent.
        """
        if self._has_result_set:
            # driver returned one row, so the next row index increments by 1
            self._next_row_index += 1
            # rownumber is last returned row index
            self._rownumber = self._next_row_index - 1
        else:
            raise InterfaceError(
                "Cannot increment rownumber: no active result set.",
                "No active result set.",
            )

    # Will be used when we add support for scrollable cursors
    def _decrement_rownumber(self):
        """
        Decrement the rownumber by 1.

        This could be used for error recovery or cursor positioning operations.
        """
        if self._has_result_set and self._rownumber >= 0:
            if self._rownumber > 0:
                self._rownumber -= 1
            else:
                self._rownumber = -1
        else:
            raise InterfaceError(
                "Cannot decrement rownumber: no active result set.",
                "No active result set.",
            )

    def _clear_rownumber(self):
        """
        Clear the rownumber tracking.

        This should be called when the result set is cleared or when the cursor is reset.
        """
        self._rownumber = -1
        self._has_result_set = False
        self._skip_increment_for_next_fetch = False

    def __iter__(self):
        """
        Return the cursor itself as an iterator.

        This allows direct iteration over the cursor after execute():

        for row in cursor.execute("SELECT * FROM table"):
            print(row)
        """
        self._check_closed()
        return self

    def __next__(self):
        """
        Fetch the next row when iterating over the cursor.

        Returns:
            The next Row object.

        Raises:
            StopIteration: When no more rows are available.
        """
        self._check_closed()
        row = self.fetchone()
        if row is None:
            raise StopIteration
        return row

    def next(self):
        """
        Fetch the next row from the cursor.

        This is an alias for __next__() to maintain compatibility with older code.

        Returns:
            The next Row object.

        Raises:
            StopIteration: When no more rows are available.
        """
        return next(self)

    def execute(  # pylint: disable=too-many-locals,too-many-branches,too-many-statements
        self,
        operation: str,
        *parameters,
        use_prepare: bool = True,
        reset_cursor: bool = True,
    ) -> "Cursor":
        """
        Prepare and execute a database operation (query or command).

        Args:
            operation: SQL query or command.
            parameters: Sequence of parameters to bind.
            use_prepare: Whether to use SQLPrepareW (default) or SQLExecDirectW.
            reset_cursor: Whether to reset the cursor before execution.
        """
        logger.debug(
            "execute: Starting - operation_length=%d, param_count=%d, use_prepare=%s",
            len(operation),
            len(parameters),
            str(use_prepare),
        )

        # Log the actual query being executed
        logger.debug("Executing query: %s", operation)

        self._check_closed()  # Check if the cursor is closed
        if reset_cursor:
            if self.hstmt:
                self._soft_reset_cursor()
            else:
                self._reset_cursor()
        else:
            # Close just the ODBC cursor (not the statement handle) so the
            # prepared plan can be reused.  SQLFreeStmt(SQL_CLOSE) releases
            # the cursor associated with hstmt without destroying the
            # prepared statement, which is the standard ODBC pattern for
            # re-executing a prepared query.
            if self.hstmt:
                logger.debug("execute: Closing cursor for re-execution (reset_cursor=False)")
                self.hstmt._close_cursor()
                self._clear_rownumber()

        # Clear any previous messages
        self.messages = []

        # Auto-detect and convert parameter style if needed
        # Supports both qmark (?) and pyformat (%(name)s)
        # Note: parameters is always a tuple due to *parameters in method signature
        #
        # Parameter Passing Rules (handling ambiguity):
        #
        # 1. Single value:
        #    cursor.execute("SELECT ?", 42)
        #    → parameters = (42,)
        #    → Wrapped as single parameter
        #
        # 2. Multiple values (two equivalent ways):
        #    cursor.execute("SELECT ?, ?", 1, 2)        # Varargs
        #    cursor.execute("SELECT ?, ?", (1, 2))      # Tuple
        #    → Both result in parameters = (1, 2) or ((1, 2),)
        #    → If single tuple/list/dict arg, it's unwrapped
        #
        # 3. Dict for named parameters:
        #    cursor.execute("SELECT %(id)s", {"id": 42})
        #    → parameters = ({"id": 42},)
        #    → Unwrapped to {"id": 42}, then converted to qmark style
        #
        # Important: If you pass a tuple/list/dict as the ONLY argument,
        # it will be unwrapped for parameter binding. This means you cannot
        # pass a tuple as a single parameter value (but SQL Server doesn't
        # support tuple types as parameter values anyway).
        with perf_phase("py::execute::param_prep"):
            if parameters:
                # Check if single parameter is a nested container that should be unwrapped
                # e.g., execute("SELECT ?", (value,)) vs execute("SELECT ?, ?", ((1, 2),))
                if isinstance(parameters, tuple) and len(parameters) == 1:
                    if isinstance(parameters[0], (tuple, list, dict)):
                        actual_params = parameters[0]
                    elif isinstance(parameters[0], Row):
                        # A Row (e.g. from fetchone()) is a sequence of column values.
                        # Normalize it to a tuple so the downstream binding logic, which
                        # only handles tuple/list/dict, can unwrap it into individual
                        # parameters instead of treating the whole Row as one value.
                        actual_params = tuple(parameters[0])
                    else:
                        actual_params = parameters
                else:
                    actual_params = parameters

                # Skip detect_and_convert_parameters when re-executing the same SQL —
                # the parameter style (qmark vs pyformat) won't change between calls.
                if operation == self.last_executed_stmt and isinstance(
                    actual_params, (tuple, list)
                ):
                    parameters = list(actual_params)
                else:
                    operation, converted_params = detect_and_convert_parameters(
                        operation, actual_params
                    )
                    parameters = list(converted_params)
            else:
                parameters = []

        # Getting encoding setting
        encoding_settings = self._get_encoding_settings()

        # Validate that inputsizes matches parameter count if both are present
        if parameters and self._inputsizes:
            if len(self._inputsizes) != len(parameters):

                warnings.warn(
                    f"Number of input sizes ({len(self._inputsizes)}) does not match "
                    f"number of parameters ({len(parameters)}). "
                    f"This may lead to unexpected behavior.",
                    Warning,
                )

        # Prepare caching: skip SQLPrepare when re-executing the same SQL
        # with parameters. The HSTMT is reused via _soft_reset_cursor, so the
        # server-side plan from the previous SQLPrepare is still valid.
        same_sql = parameters and operation == self.last_executed_stmt and self.is_stmt_prepared[0]
        if not same_sql:
            self.is_stmt_prepared = [False]
        effective_use_prepare = use_prepare and not same_sql

        with perf_phase("py::execute::cpp_call"):
            if parameters:
                ret = ddbc_bindings.DDBCSQLExecute(
                    self.hstmt,
                    operation,
                    parameters,
                    self._inputsizes,
                    self.is_stmt_prepared,
                    effective_use_prepare,
                    encoding_settings,
                )
            else:
                ret = ddbc_bindings.DDBCSQLExecDirect(self.hstmt, operation)
        # Check return code
        try:

            # Check for errors but don't raise exceptions for info/warning messages
            check_error(ddbc_sql_const.SQL_HANDLE_STMT.value, self.hstmt, ret)
        except Exception as e:  # pylint: disable=broad-exception-caught
            logger.warning("Execute failed, resetting cursor: %s", e)
            self._reset_cursor()
            raise

        # Capture any diagnostic messages (SQL_SUCCESS_WITH_INFO, etc.)
        with perf_phase("py::execute::diag_records"):
            self._capture_diagnostics(ret)

        self.last_executed_stmt = operation

        with perf_phase("py::execute::post_execute"):
            # Update rowcount after execution
            # TODO: rowcount return code from SQL needs to be handled
            self.rowcount = ddbc_bindings.DDBCSQLRowCount(self.hstmt)

            # Initialize description after execution
            # After successful execution, initialize description if there are results
            column_metadata = []
            try:
                ddbc_bindings.DDBCSQLDescribeCol(self.hstmt, column_metadata)
                self._initialize_description(column_metadata)
            except Exception as e:  # pylint: disable=broad-exception-caught
                # If describe fails, it's likely there are no results (e.g., for INSERT)
                self.description = None
                self._column_sql_types = None

        # Reset rownumber for new result set (only for SELECT statements)
        if self.description:  # If we have column descriptions, it's likely a SELECT
            self.rowcount = -1
            self._reset_rownumber()
            # Pre-build column map and converter map
            self._cached_column_map = {
                col_desc[0]: i for i, col_desc in enumerate(self.description)
            }
            self._cached_column_map_lower = (
                {k.lower(): v for k, v in self._cached_column_map.items()}
                if get_settings().lowercase
                else None
            )
            self._cached_converter_map = self._build_converter_map()
            self._uuid_str_indices = self._compute_uuid_str_indices()
        else:
            self.rowcount = ddbc_bindings.DDBCSQLRowCount(self.hstmt)
            self._clear_rownumber()
            self._cached_column_map = None
            self._cached_column_map_lower = None
            self._cached_converter_map = None
            self._uuid_str_indices = None

        self._reset_inputsizes()  # Reset input sizes after execution
        # Return self for method chaining
        return self

    def _prepare_metadata_result_set(  # pylint: disable=too-many-statements
        self, column_metadata=None, fallback_description=None, specialized_mapping=None
    ):
        """
        Prepares a metadata result set by:
        1. Retrieving column metadata if not provided
        2. Initializing the description attribute
        3. Setting up column name mappings (including any specialized aliases)
           so the standard fetch methods return rows keyed by catalog columns

        Args:
            column_metadata (list, optional): Pre-fetched column metadata.
                                             If None, it will be retrieved.
            fallback_description (list, optional): Fallback description to use if
                                                  metadata retrieval fails.
            specialized_mapping (dict, optional): Custom column mapping for special cases.

        Returns:
            Cursor: Self, for method chaining
        """
        # Retrieve column metadata if not provided
        if column_metadata is None:
            column_metadata = []
            try:
                ddbc_bindings.DDBCSQLDescribeCol(self.hstmt, column_metadata)
            except InterfaceError as e:
                logger.warning(f"Driver interface error during metadata retrieval: {e}")
            except Exception as e:  # pylint: disable=broad-exception-caught
                # Log the exception with appropriate context
                logger.warning(
                    f"Failed to retrieve column metadata: {e}. "
                    f"Using standard ODBC column definitions instead.",
                )

        # Initialize the description attribute with the column metadata
        self._initialize_description(column_metadata)

        # Use fallback description if provided and current description is empty
        if not self.description and fallback_description:
            self.description = fallback_description

        # Build the converter map so output converters (including integer SQL-type
        # keys) apply to metadata result sets consistently with normal result sets.
        # See GH #684.
        self._cached_converter_map = self._build_converter_map()

        # Build the column-name -> index map for this metadata result set.
        # Both the exact name and its lowercase alias are stored so that rows
        # support case-insensitive lookup regardless of the global ``lowercase``
        # setting (catalog column names are well-known and case-stable).
        # ``self.description`` is None when DDBCSQLDescribeCol failed above and no
        # fallback_description was supplied (e.g. getTypeInfo()); guard with ``or ()``
        # so the cursor simply yields no rows instead of raising TypeError here.
        column_map = {}
        for i, (name, *_) in enumerate(self.description or ()):
            column_map[name] = i
            column_map[name.lower()] = i

        # Some catalog helpers expose additional friendly aliases (e.g. ODBC 2.x
        # vs 3.x column names) via a specialized mapping. Merge them in so they
        # resolve to the same column indices.
        if specialized_mapping:
            column_map.update(specialized_mapping)

        # Route the map through the shared fetch path. The standard fetchone/
        # fetchmany/fetchall methods read these cached maps when constructing Row
        # objects, so metadata rows pick up the catalog column names without any
        # per-call method reassignment (the previous approach broke static type
        # checking - see GH #620).
        self._cached_column_map = column_map
        # Mirror execute(): when ``lowercase`` is enabled the description names are
        # already lowercased, so build the lowercase lookup map (including any
        # specialized aliases) to keep case-insensitive access working for catalog
        # rows (e.g. row.TABLE_NAME when lowercase=True).
        self._cached_column_map_lower = (
            {k.lower(): v for k, v in column_map.items()} if get_settings().lowercase else None
        )

        # Initialize rownumber tracking so fetchone() and iteration work
        self._reset_rownumber()

        # Metadata methods produce a new result set, so rowcount is unknown
        # until rows are fetched. Reset it to avoid exposing a stale value
        # from a previous statement.
        self.rowcount = -1

        # Return the cursor itself for method chaining
        return self

    def getTypeInfo(self, sqlType=None):
        """
        Executes SQLGetTypeInfo and creates a result set with information about
        the specified data type or all data types supported by the ODBC driver if not specified.
        """
        self._check_closed()
        self._reset_cursor()

        sql_all_types = 0  # SQL_ALL_TYPES = 0

        try:
            # Get information about data types
            ret = ddbc_bindings.DDBCSQLGetTypeInfo(
                self.hstmt, sqlType if sqlType is not None else sql_all_types
            )
            check_error(ddbc_sql_const.SQL_HANDLE_STMT.value, self.hstmt, ret)

            # Use the helper method to prepare the result set
            return self._prepare_metadata_result_set()
        except Exception as e:  # pylint: disable=broad-exception-caught
            self._reset_cursor()
            raise e

    def procedures(self, procedure=None, catalog=None, schema=None):
        """
        Executes SQLProcedures and creates a result set of information about procedures
        in the data source.

        Args:
            procedure (str, optional): Procedure name pattern. Default is None (all procedures).
            catalog (str, optional): Catalog name pattern. Default is None (current catalog).
            schema (str, optional): Schema name pattern. Default is None (all schemas).
        """
        self._check_closed()
        self._reset_cursor()

        # Call the SQLProcedures function
        retcode = ddbc_bindings.DDBCSQLProcedures(self.hstmt, catalog, schema, procedure)
        check_error(ddbc_sql_const.SQL_HANDLE_STMT.value, self.hstmt, retcode)

        # Define fallback description for procedures
        fallback_description = [
            ("procedure_cat", str, None, 128, 128, 0, True),
            ("procedure_schem", str, None, 128, 128, 0, True),
            ("procedure_name", str, None, 128, 128, 0, False),
            ("num_input_params", int, None, 10, 10, 0, True),
            ("num_output_params", int, None, 10, 10, 0, True),
            ("num_result_sets", int, None, 10, 10, 0, True),
            ("remarks", str, None, 254, 254, 0, True),
            ("procedure_type", int, None, 10, 10, 0, False),
        ]

        # Use the helper method to prepare the result set
        return self._prepare_metadata_result_set(fallback_description=fallback_description)

    def primaryKeys(self, table, catalog=None, schema=None):
        """
        Creates a result set of column names that make up the primary key for a table
        by executing the SQLPrimaryKeys function.

        Args:
            table (str): The name of the table
            catalog (str, optional): The catalog name (database). Defaults to None.
            schema (str, optional): The schema name. Defaults to None.
        """
        self._check_closed()
        self._reset_cursor()

        if not table:
            raise ProgrammingError("Table name must be specified", "HY000")

        # Call the SQLPrimaryKeys function
        retcode = ddbc_bindings.DDBCSQLPrimaryKeys(self.hstmt, catalog, schema, table)
        check_error(ddbc_sql_const.SQL_HANDLE_STMT.value, self.hstmt, retcode)

        # Define fallback description for primary keys
        fallback_description = [
            ("table_cat", str, None, 128, 128, 0, True),
            ("table_schem", str, None, 128, 128, 0, True),
            ("table_name", str, None, 128, 128, 0, False),
            ("column_name", str, None, 128, 128, 0, False),
            ("key_seq", int, None, 10, 10, 0, False),
            ("pk_name", str, None, 128, 128, 0, True),
        ]

        # Use the helper method to prepare the result set
        return self._prepare_metadata_result_set(fallback_description=fallback_description)

    def foreignKeys(  # pylint: disable=too-many-arguments,too-many-positional-arguments
        self,
        table=None,
        catalog=None,
        schema=None,
        foreignTable=None,
        foreignCatalog=None,
        foreignSchema=None,
    ):
        """
        Executes the SQLForeignKeys function and creates a result set of column names
        that are foreign keys.

        This function returns:
        1. Foreign keys in the specified table that reference primary keys in other tables, OR
        2. Foreign keys in other tables that reference the primary key in the specified table
        """
        self._check_closed()
        self._reset_cursor()

        # Check if we have at least one table specified
        if table is None and foreignTable is None:
            raise ProgrammingError("Either table or foreignTable must be specified", "HY000")

        # Call the SQLForeignKeys function
        retcode = ddbc_bindings.DDBCSQLForeignKeys(
            self.hstmt,
            foreignCatalog,
            foreignSchema,
            foreignTable,
            catalog,
            schema,
            table,
        )
        check_error(ddbc_sql_const.SQL_HANDLE_STMT.value, self.hstmt, retcode)

        # Define fallback description for foreign keys
        fallback_description = [
            ("pktable_cat", str, None, 128, 128, 0, True),
            ("pktable_schem", str, None, 128, 128, 0, True),
            ("pktable_name", str, None, 128, 128, 0, False),
            ("pkcolumn_name", str, None, 128, 128, 0, False),
            ("fktable_cat", str, None, 128, 128, 0, True),
            ("fktable_schem", str, None, 128, 128, 0, True),
            ("fktable_name", str, None, 128, 128, 0, False),
            ("fkcolumn_name", str, None, 128, 128, 0, False),
            ("key_seq", int, None, 10, 10, 0, False),
            ("update_rule", int, None, 10, 10, 0, False),
            ("delete_rule", int, None, 10, 10, 0, False),
            ("fk_name", str, None, 128, 128, 0, True),
            ("pk_name", str, None, 128, 128, 0, True),
            ("deferrability", int, None, 10, 10, 0, False),
        ]

        # Use the helper method to prepare the result set
        return self._prepare_metadata_result_set(fallback_description=fallback_description)

    def rowIdColumns(self, table, catalog=None, schema=None, nullable=True):
        """
        Executes SQLSpecialColumns with SQL_BEST_ROWID which creates a result set of
        columns that uniquely identify a row.
        """
        self._check_closed()
        self._reset_cursor()

        if not table:
            raise ProgrammingError("Table name must be specified", "HY000")

        # Set the identifier type and options
        identifier_type = ddbc_sql_const.SQL_BEST_ROWID.value
        scope = ddbc_sql_const.SQL_SCOPE_CURROW.value
        nullable_flag = (
            ddbc_sql_const.SQL_NULLABLE.value if nullable else ddbc_sql_const.SQL_NO_NULLS.value
        )

        # Call the SQLSpecialColumns function
        retcode = ddbc_bindings.DDBCSQLSpecialColumns(
            self.hstmt, identifier_type, catalog, schema, table, scope, nullable_flag
        )
        check_error(ddbc_sql_const.SQL_HANDLE_STMT.value, self.hstmt, retcode)

        # Define fallback description for special columns
        fallback_description = [
            ("scope", int, None, 10, 10, 0, False),
            ("column_name", str, None, 128, 128, 0, False),
            ("data_type", int, None, 10, 10, 0, False),
            ("type_name", str, None, 128, 128, 0, False),
            ("column_size", int, None, 10, 10, 0, False),
            ("buffer_length", int, None, 10, 10, 0, False),
            ("decimal_digits", int, None, 10, 10, 0, True),
            ("pseudo_column", int, None, 10, 10, 0, False),
        ]

        # Use the helper method to prepare the result set
        return self._prepare_metadata_result_set(fallback_description=fallback_description)

    def rowVerColumns(self, table, catalog=None, schema=None, nullable=True):
        """
        Executes SQLSpecialColumns with SQL_ROWVER which creates a result set of
        columns that are automatically updated when any value in the row is updated.
        """
        self._check_closed()
        self._reset_cursor()

        if not table:
            raise ProgrammingError("Table name must be specified", "HY000")

        # Set the identifier type and options
        identifier_type = ddbc_sql_const.SQL_ROWVER.value
        scope = ddbc_sql_const.SQL_SCOPE_CURROW.value
        nullable_flag = (
            ddbc_sql_const.SQL_NULLABLE.value if nullable else ddbc_sql_const.SQL_NO_NULLS.value
        )

        # Call the SQLSpecialColumns function
        retcode = ddbc_bindings.DDBCSQLSpecialColumns(
            self.hstmt, identifier_type, catalog, schema, table, scope, nullable_flag
        )
        check_error(ddbc_sql_const.SQL_HANDLE_STMT.value, self.hstmt, retcode)

        # Same fallback description as rowIdColumns
        fallback_description = [
            ("scope", int, None, 10, 10, 0, False),
            ("column_name", str, None, 128, 128, 0, False),
            ("data_type", int, None, 10, 10, 0, False),
            ("type_name", str, None, 128, 128, 0, False),
            ("column_size", int, None, 10, 10, 0, False),
            ("buffer_length", int, None, 10, 10, 0, False),
            ("decimal_digits", int, None, 10, 10, 0, True),
            ("pseudo_column", int, None, 10, 10, 0, False),
        ]

        # Use the helper method to prepare the result set
        return self._prepare_metadata_result_set(fallback_description=fallback_description)

    def statistics(  # pylint: disable=too-many-arguments,too-many-positional-arguments
        self,
        table: str,
        catalog: str = None,
        schema: str = None,
        unique: bool = False,
        quick: bool = True,
    ) -> "Cursor":
        """
        Creates a result set of statistics about a single table and the indexes associated
        with the table by executing SQLStatistics.
        """
        self._check_closed()
        self._reset_cursor()

        if not table:
            raise ProgrammingError("Table name is required", "HY000")

        # Set unique and quick flags
        unique_option = (
            ddbc_sql_const.SQL_INDEX_UNIQUE.value if unique else ddbc_sql_const.SQL_INDEX_ALL.value
        )
        reserved_option = (
            ddbc_sql_const.SQL_QUICK.value if quick else ddbc_sql_const.SQL_ENSURE.value
        )

        # Call the SQLStatistics function
        retcode = ddbc_bindings.DDBCSQLStatistics(
            self.hstmt, catalog, schema, table, unique_option, reserved_option
        )
        check_error(ddbc_sql_const.SQL_HANDLE_STMT.value, self.hstmt, retcode)

        # Define fallback description for statistics
        fallback_description = [
            ("table_cat", str, None, 128, 128, 0, True),
            ("table_schem", str, None, 128, 128, 0, True),
            ("table_name", str, None, 128, 128, 0, False),
            ("non_unique", bool, None, 1, 1, 0, False),
            ("index_qualifier", str, None, 128, 128, 0, True),
            ("index_name", str, None, 128, 128, 0, True),
            ("type", int, None, 10, 10, 0, False),
            ("ordinal_position", int, None, 10, 10, 0, False),
            ("column_name", str, None, 128, 128, 0, True),
            ("asc_or_desc", str, None, 1, 1, 0, True),
            ("cardinality", int, None, 20, 20, 0, True),
            ("pages", int, None, 20, 20, 0, True),
            ("filter_condition", str, None, 128, 128, 0, True),
        ]

        # Use the helper method to prepare the result set
        return self._prepare_metadata_result_set(fallback_description=fallback_description)

    def columns(self, table=None, catalog=None, schema=None, column=None):
        """
        Creates a result set of column information in the specified tables
        using the SQLColumns function.
        """
        self._check_closed()
        self._reset_cursor()

        # Call the SQLColumns function
        retcode = ddbc_bindings.DDBCSQLColumns(self.hstmt, catalog, schema, table, column)
        check_error(ddbc_sql_const.SQL_HANDLE_STMT.value, self.hstmt, retcode)

        # Define fallback description for columns
        fallback_description = [
            ("table_cat", str, None, 128, 128, 0, True),
            ("table_schem", str, None, 128, 128, 0, True),
            ("table_name", str, None, 128, 128, 0, False),
            ("column_name", str, None, 128, 128, 0, False),
            ("data_type", int, None, 10, 10, 0, False),
            ("type_name", str, None, 128, 128, 0, False),
            ("column_size", int, None, 10, 10, 0, True),
            ("buffer_length", int, None, 10, 10, 0, True),
            ("decimal_digits", int, None, 10, 10, 0, True),
            ("num_prec_radix", int, None, 10, 10, 0, True),
            ("nullable", int, None, 10, 10, 0, False),
            ("remarks", str, None, 254, 254, 0, True),
            ("column_def", str, None, 254, 254, 0, True),
            ("sql_data_type", int, None, 10, 10, 0, False),
            ("sql_datetime_sub", int, None, 10, 10, 0, True),
            ("char_octet_length", int, None, 10, 10, 0, True),
            ("ordinal_position", int, None, 10, 10, 0, False),
            ("is_nullable", str, None, 254, 254, 0, True),
        ]

        # Use the helper method to prepare the result set
        return self._prepare_metadata_result_set(fallback_description=fallback_description)

    def _transpose_rowwise_to_columnwise(
        self,
        seq_of_parameters: Sequence[Sequence[Any]],
    ) -> tuple[list, int]:
        """
        Convert sequence of rows (row-wise) into list of columns (column-wise),
        for array binding via ODBC. Works with both iterables and generators.

        Args:
            seq_of_parameters: Sequence of sequences or mappings of parameters.

        Returns:
            tuple: (columnwise_data, row_count)
        """
        columnwise = []
        first_row = True
        row_count = 0

        for row in seq_of_parameters:
            row_count += 1
            if first_row:
                # Initialize columnwise lists based on first row
                num_params = len(row)
                columnwise = [[] for _ in range(num_params)]
                first_row = False
            else:
                # Validate row size consistency
                if len(row) != num_params:
                    raise ValueError("Inconsistent parameter row size in executemany()")

            # Add each value to its column list
            for i, val in enumerate(row):
                columnwise[i].append(val)

        return columnwise, row_count

    def _compute_column_type(self, column):
        """
        Determine representative value and integer min/max for a column.

        Returns:
            sample_value: Representative value for type inference and modified_row.
            min_val: Minimum for integers (None otherwise).
            max_val: Maximum for integers (None otherwise).
            max_decimal_formatted_len: Maximum len(format(d, 'f')) across all
                Decimal values in the column (0 when no Decimals are present).
                Used by executemany to correct the SQL_VARCHAR column size when
                the sample value's formatted string is shorter than another
                value's (e.g. positive sample vs negative row value) (GH-557).
        """
        non_nulls = [v for v in column if v is not None]
        if not non_nulls:
            return None, None, None, 0

        int_values = [v for v in non_nulls if isinstance(v, int)]
        if int_values:
            min_val, max_val = min(int_values), max(int_values)
            sample_value = max(int_values, key=abs)
            return sample_value, min_val, max_val, 0

        sample_value = None
        max_decimal_formatted_len = 0
        for v in non_nulls:
            if isinstance(v, decimal.Decimal):
                max_decimal_formatted_len = max(max_decimal_formatted_len, len(format(v, "f")))
            if not sample_value:
                sample_value = v
            elif isinstance(v, (str, bytes, bytearray)) and isinstance(
                sample_value, (str, bytes, bytearray)
            ):
                # For string/binary objects, prefer the longer one
                # Use safe length comparison to avoid exceptions from custom __len__ implementations
                try:
                    if len(v) > len(sample_value):
                        sample_value = v
                except (TypeError, ValueError, AttributeError):
                    # If length comparison fails, keep the current sample_value
                    pass
            elif isinstance(v, decimal.Decimal) and isinstance(sample_value, decimal.Decimal):
                # For Decimal objects, prefer the one that requires higher precision or scale
                v_tuple = v.as_tuple()
                sample_tuple = sample_value.as_tuple()

                # Calculate precision (total significant digits) and scale (decimal places)
                # For a number like 0.000123456789, we need precision = 9, scale = 12
                # The precision is the number of significant digits (len(digits))
                # The scale is the number of decimal places needed to represent the number

                v_precision = len(v_tuple.digits)
                if v_tuple.exponent < 0:
                    v_scale = -v_tuple.exponent
                else:
                    v_scale = 0

                sample_precision = len(sample_tuple.digits)
                if sample_tuple.exponent < 0:
                    sample_scale = -sample_tuple.exponent
                else:
                    sample_scale = 0

                # For SQL DECIMAL(precision, scale), we need:
                # precision >= number of significant digits
                # scale >= number of decimal places
                # For 0.000123456789: precision needs to be at least 12 (to accommodate 12 decimal places)
                # So we need to adjust precision to be at least as large as scale
                v_required_precision = max(v_precision, v_scale)
                sample_required_precision = max(sample_precision, sample_scale)

                # Prefer the decimal that requires higher precision or scale
                # This ensures we can accommodate all values in the column
                if v_required_precision > sample_required_precision or (
                    v_required_precision == sample_required_precision and v_scale > sample_scale
                ):
                    sample_value = v
            elif isinstance(v, decimal.Decimal) and not isinstance(sample_value, decimal.Decimal):
                # If comparing Decimal to non-Decimal, prefer Decimal for better type inference
                sample_value = v

        return sample_value, None, None, max_decimal_formatted_len

    def executemany(  # pylint: disable=too-many-locals,too-many-branches,too-many-statements
        self,
        operation: str,
        seq_of_parameters: Union[Sequence[Sequence[Any]], Sequence[Mapping[str, Any]]],
    ) -> None:
        """
        Prepare a database operation and execute it against all parameter sequences.
        This version uses column-wise parameter binding and a single batched SQLExecute().
        Args:
            operation: SQL query or command.
            seq_of_parameters: Sequence of sequences or mappings of parameters.
        Raises:
            Error: If the operation fails.
        """
        logger.debug(
            "executemany: Starting - operation_length=%d, batch_count=%d",
            len(operation),
            len(seq_of_parameters),
        )

        self._check_closed()
        self._reset_cursor()
        self.messages = []
        logger.debug("executemany: Cursor reset complete")

        if not seq_of_parameters:
            self.rowcount = 0
            return

        # Auto-detect and convert parameter style for executemany
        # Check first row to determine if we need to convert from pyformat to qmark
        first_row = (
            seq_of_parameters[0]
            if hasattr(seq_of_parameters, "__getitem__")
            else next(iter(seq_of_parameters))
        )

        if isinstance(first_row, dict):
            # pyformat style - convert all rows
            # Parse parameter names from SQL (determines order for all rows)
            param_names = parse_pyformat_params(operation)

            if param_names:
                # Convert SQL to qmark style
                operation, _ = convert_pyformat_to_qmark(operation, first_row)

                # Convert all parameter dicts to tuples in the same order
                converted_params = []
                for param_dict in seq_of_parameters:
                    if not isinstance(param_dict, dict):
                        raise TypeError(
                            f"Mixed parameter types in executemany: first row is dict, "
                            f"but row has {type(param_dict).__name__}"
                        )
                    # Build tuple in the order determined by param_names
                    row_tuple = tuple(param_dict[name] for name in param_names)
                    converted_params.append(row_tuple)

                seq_of_parameters = converted_params
                logger.debug(
                    "executemany: Converted %d rows from pyformat to qmark", len(seq_of_parameters)
                )

        # Apply timeout if set (non-zero)
        if self._timeout > 0:
            try:
                timeout_value = int(self._timeout)
                ret = ddbc_bindings.DDBCSQLSetStmtAttr(
                    self.hstmt,
                    ddbc_sql_const.SQL_ATTR_QUERY_TIMEOUT.value,
                    timeout_value,
                )
                check_error(ddbc_sql_const.SQL_HANDLE_STMT.value, self.hstmt, ret)
                logger.debug(f"Set query timeout to {self._timeout} seconds")
            except Exception as e:  # pylint: disable=broad-exception-caught
                logger.warning(f"Failed to set query timeout: {e}")

        # Get sample row for parameter type detection and validation
        sample_row = (
            seq_of_parameters[0]
            if hasattr(seq_of_parameters, "__getitem__")
            else next(iter(seq_of_parameters))
        )
        param_count = len(sample_row)
        param_info = ddbc_bindings.ParamInfo
        parameters_type = []
        any_dae = False

        # Check if we have explicit input sizes set
        if self._inputsizes:
            # Validate input sizes match parameter count
            if len(self._inputsizes) != param_count:
                warnings.warn(
                    f"Number of input sizes ({len(self._inputsizes)}) does not match "
                    f"number of parameters ({param_count}). This may lead to unexpected behavior.",
                    Warning,
                )

        # Prepare parameter type information
        with perf_phase("py::executemany::param_type_detection"):
            for col_index in range(param_count):
                column = (
                    [row[col_index] for row in seq_of_parameters]
                    if hasattr(seq_of_parameters, "__getitem__")
                    else []
                )
                sample_value, min_val, max_val, _ = self._compute_column_type(column)

                if self._inputsizes and col_index < len(self._inputsizes):
                    # Use explicitly set input sizes
                    sql_type, c_type, column_size, decimal_digits = self._inputsizes[col_index]

                    # Default is_dae to False
                    is_dae = False

                    # Override DECIMAL/NUMERIC to use SQL_C_CHAR string binding (GH-503)
                    if sql_type in (
                        ddbc_sql_const.SQL_DECIMAL.value,
                        ddbc_sql_const.SQL_NUMERIC.value,
                    ):
                        c_type = ddbc_sql_const.SQL_C_CHAR.value

                    # Check if this should be a DAE (data at execution) parameter based on column size
                    if sample_value is not None:
                        if isinstance(sample_value, str) and column_size > MAX_INLINE_CHAR:
                            is_dae = True
                        elif isinstance(sample_value, (bytes, bytearray)) and column_size > 8000:
                            is_dae = True

                    # Sanitize precision/scale for numeric types
                    if sql_type in (
                        ddbc_sql_const.SQL_DECIMAL.value,
                        ddbc_sql_const.SQL_NUMERIC.value,
                    ):
                        column_size = max(1, min(int(column_size) if column_size > 0 else 18, 38))
                        decimal_digits = min(max(0, decimal_digits), column_size)

                    # For binary data columns with mixed content, we need to find max size
                    if sql_type in (
                        ddbc_sql_const.SQL_BINARY.value,
                        ddbc_sql_const.SQL_VARBINARY.value,
                        ddbc_sql_const.SQL_LONGVARBINARY.value,
                    ):
                        # Find the maximum size needed for any row's binary data
                        max_binary_size = 0
                        for row in seq_of_parameters:
                            value = row[col_index]
                            if value is not None and isinstance(value, (bytes, bytearray)):
                                max_binary_size = max(max_binary_size, len(value))

                        # For SQL Server VARBINARY(MAX), we need to use large object binding
                        if column_size > 8000 or max_binary_size > 8000:
                            sql_type = ddbc_sql_const.SQL_LONGVARBINARY.value
                            is_dae = True

                        # Update column_size to actual maximum size if it's larger
                        # Always ensure at least a minimum size of 1 for empty strings
                        column_size = max(max_binary_size, 1)

                    paraminfo = param_info()
                    paraminfo.paramCType = c_type
                    paraminfo.paramSQLType = sql_type
                    paraminfo.inputOutputType = ddbc_sql_const.SQL_PARAM_INPUT.value
                    paraminfo.columnSize = column_size
                    paraminfo.decimalDigits = decimal_digits
                    paraminfo.isDAE = is_dae

                    # Ensure we never have SQL_C_DEFAULT (0) for C-type
                    if paraminfo.paramCType == 0:
                        paraminfo.paramCType = ddbc_sql_const.SQL_C_DEFAULT.value

                    parameters_type.append(paraminfo)
                else:
                    # Use auto-detection for columns without explicit types
                    column = (
                        [row[col_index] for row in seq_of_parameters]
                        if hasattr(seq_of_parameters, "__getitem__")
                        else []
                    )
                    sample_value, min_val, max_val, max_decimal_len = self._compute_column_type(
                        column
                    )

                    dummy_row = list(sample_row)
                    paraminfo = self._create_parameter_types_list(
                        sample_value,
                        param_info,
                        dummy_row,
                        col_index,
                        min_val=min_val,
                        max_val=max_val,
                    )

                    # GH-610: all-NULL columns now pass SQL_UNKNOWN_TYPE to C++,
                    # where BindParameterArray resolves the correct type via the
                    # SQLDescribeParam cache.  The previous SQL_VARCHAR hardcoded
                    # fallback was removed because it broke VARBINARY columns.

                    # Override DECIMAL/NUMERIC to use SQL_C_CHAR string binding.
                    # _map_sql_type may return SQL_C_NUMERIC (expecting NumericData structs)
                    # but the conversion loop below converts all Decimal values to strings.
                    # The C type must match the actual data to avoid:
                    #   RuntimeError: Parameter's object type does not match parameter's C type
                    if paraminfo.paramSQLType in (
                        ddbc_sql_const.SQL_DECIMAL.value,
                        ddbc_sql_const.SQL_NUMERIC.value,
                    ):
                        paraminfo.paramCType = ddbc_sql_const.SQL_C_CHAR.value
                        # Ensure columnSize accommodates the longest string representation
                        if max_decimal_len > paraminfo.columnSize:
                            paraminfo.columnSize = max_decimal_len

                    # Correct column size for Decimal columns sent as SQL_VARCHAR (GH-557).
                    # The sample value's formatted string may be shorter than another
                    # row's (e.g. positive sample "1.0" = 3 chars vs negative "-0.1" = 4).
                    # max_decimal_len was already computed during _compute_column_type
                    # so no extra iteration is needed.
                    if (
                        paraminfo.paramSQLType == ddbc_sql_const.SQL_VARCHAR.value
                        and max_decimal_len > paraminfo.columnSize
                    ):
                        paraminfo.columnSize = max_decimal_len

                    # Special handling for binary data in auto-detected types
                    if paraminfo.paramSQLType in (
                        ddbc_sql_const.SQL_BINARY.value,
                        ddbc_sql_const.SQL_VARBINARY.value,
                        ddbc_sql_const.SQL_LONGVARBINARY.value,
                    ):
                        # Find the maximum size needed for any row's binary data
                        max_binary_size = 0
                        for row in seq_of_parameters:
                            value = row[col_index]
                            if value is not None and isinstance(value, (bytes, bytearray)):
                                max_binary_size = max(max_binary_size, len(value))

                        # For SQL Server VARBINARY(MAX), we need to use large object binding
                        if max_binary_size > 8000:
                            paraminfo.paramSQLType = ddbc_sql_const.SQL_LONGVARBINARY.value
                            paraminfo.isDAE = True

                        # Update column_size to actual maximum size
                        # Always ensure at least a minimum size of 1 for empty strings
                        paraminfo.columnSize = max(max_binary_size, 1)

                    parameters_type.append(paraminfo)
                if paraminfo.isDAE:
                    any_dae = True

        if any_dae:
            logger.debug(
                "DAE parameters detected. Falling back to row-by-row execution with streaming.",
            )
            input_sizes = self._inputsizes
            try:
                for row in seq_of_parameters:
                    # execute() consumes overrides; every streamed row needs them.
                    self._inputsizes = input_sizes
                    self.execute(operation, row)
            finally:
                self._reset_inputsizes()
            return

        # Process parameters into column-wise format with possible type conversions
        # First, convert any Decimal types as needed for NUMERIC/DECIMAL columns
        with perf_phase("py::executemany::param_conversion"):
            processed_parameters = []
            for row_index, row in enumerate(seq_of_parameters):
                processed_row = list(row)
                for i, val in enumerate(processed_row):
                    if val is None:
                        continue
                    time_text = _normalize_time_param(val, parameters_type[i].paramCType)
                    if time_text is not None:
                        processed_row[i] = time_text
                        continue
                    if (
                        isinstance(val, decimal.Decimal)
                        and parameters_type[i].paramSQLType == ddbc_sql_const.SQL_VARCHAR.value
                    ):
                        processed_row[i] = format(val, "f")
                    # Convert all values to string for DECIMAL/NUMERIC columns (GH-503)
                    elif parameters_type[i].paramSQLType in (
                        ddbc_sql_const.SQL_DECIMAL.value,
                        ddbc_sql_const.SQL_NUMERIC.value,
                    ):
                        if isinstance(val, decimal.Decimal):
                            processed_row[i] = format(val, "f")
                        else:
                            # Do not embed the parameter value or the full row in the
                            # message: rows may contain PII (SSNs, emails, balances)
                            # that would leak into caller error handlers, tracebacks,
                            # and log/APM stores. Report metadata only (row index,
                            # column index, value type).
                            err_msg = (
                                f"Failed to convert parameter to Decimal at row "
                                f"{row_index}, column {i} (value type: {type(val).__name__})"
                            )
                            # Split str(val) from the decimal parse so we only chain a
                            # cause we know is value-free. decimal.DecimalException
                            # messages (e.g. ConversionSyntax) never echo the input, so
                            # they are safe to preserve for debugging. str(val) itself
                            # or any other error could carry the value in its message
                            # and surface through __cause__ / formatted tracebacks, so
                            # those are re-raised with the chain suppressed (from None).
                            try:
                                val_text = str(val)
                            except Exception:  # pylint: disable=broad-exception-caught
                                raise ValueError(err_msg) from None
                            try:
                                processed_row[i] = format(decimal.Decimal(val_text), "f")
                            except decimal.DecimalException as e:
                                raise ValueError(err_msg) from e
                            except Exception:  # pylint: disable=broad-exception-caught
                                raise ValueError(err_msg) from None
                processed_parameters.append(processed_row)

        # Now transpose the processed parameters
        with perf_phase("py::executemany::param_processing"):
            columnwise_params, row_count = self._transpose_rowwise_to_columnwise(
                processed_parameters
            )

        # Get encoding settings
        encoding_settings = self._get_encoding_settings()

        # Debug logging: emit batch metadata only. Never log parameter values or
        # row representations here -- rows may contain PII (SSNs, emails,
        # balances) that would leak into log files and APM/log shippers even
        # though this is a DEBUG-level statement. Metadata (batch size, column
        # count) is sufficient for diagnostics without exposing user data.
        logger.debug(
            "Executing batch query with %d parameter sets (%d columns per row)",
            len(seq_of_parameters),
            len(parameters_type),
        )

        with perf_phase("py::executemany::cpp_call"):
            ret = ddbc_bindings.SQLExecuteMany(
                self.hstmt,
                operation,
                columnwise_params,
                parameters_type,
                row_count,
                encoding_settings,
            )

        # Capture any diagnostic messages after execution
        with perf_phase("py::executemany::diag_records"):
            if self.hstmt:
                self.messages.extend(ddbc_bindings.DDBCSQLGetAllDiagRecords(self.hstmt))

        try:
            check_error(ddbc_sql_const.SQL_HANDLE_STMT.value, self.hstmt, ret)
            self.rowcount = ddbc_bindings.DDBCSQLRowCount(self.hstmt)
            self.last_executed_stmt = operation

            # Fetch column metadata (e.g. for INSERT … OUTPUT)
            column_metadata = []
            try:
                ddbc_bindings.DDBCSQLDescribeCol(self.hstmt, column_metadata)
                self._initialize_description(column_metadata)
            except Exception:  # pylint: disable=broad-exception-caught
                self.description = None
                self._column_sql_types = None

            if self.description:
                self.rowcount = -1
                self._reset_rownumber()
                self._cached_column_map = {
                    col_desc[0]: i for i, col_desc in enumerate(self.description)
                }
                self._cached_column_map_lower = (
                    {k.lower(): v for k, v in self._cached_column_map.items()}
                    if get_settings().lowercase
                    else None
                )
                self._cached_converter_map = self._build_converter_map()
                self._uuid_str_indices = self._compute_uuid_str_indices()
            else:
                self.rowcount = ddbc_bindings.DDBCSQLRowCount(self.hstmt)
                self._clear_rownumber()
                self._cached_column_map = None
                self._cached_column_map_lower = None
                self._cached_converter_map = None
                self._uuid_str_indices = None
        finally:
            # Reset input sizes after execution
            self._reset_inputsizes()

    def fetchone(self) -> Union[None, Row]:
        """
        Fetch the next row of a query result set.

        Returns:
            Single Row object or None if no more data is available.
        """
        self._check_closed()  # Check if the cursor is closed

        if self._cached_decoding_generation != self._connection._decoding_generation:
            self._refresh_decoding_cache()
        char_enc = self._cached_char_encoding
        wchar_enc = self._cached_wchar_encoding

        # Fetch raw data
        row_data = []
        try:
            with perf_phase("py::fetchone::cpp_call"):
                ret = ddbc_bindings.DDBCSQLFetchOne(
                    self.hstmt,
                    row_data,
                    char_enc,
                    wchar_enc,
                    self._cached_char_ctype,
                    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
                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:
                    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,
                )
        except Exception:
            # On error, don't increment rownumber - rethrow the error
            raise

    def fetchmany(self, size: Optional[int] = None) -> List[Row]:
        """
        Fetch the next set of rows of a query result.

        Args:
            size: Number of rows to fetch at a time.

        Returns:
            List of Row objects.
        """
        self._check_closed()  # Check if the cursor is closed
        if not self._has_result_set and self.description:
            self._reset_rownumber()

        if size is None:
            size = self.arraysize

        if size <= 0:
            return []

        if self._cached_decoding_generation != self._connection._decoding_generation:
            self._refresh_decoding_cache()
        char_enc = self._cached_char_encoding
        wchar_enc = self._cached_wchar_encoding

        # Fetch raw data
        rows_data = []
        try:
            with perf_phase("py::fetchmany::cpp_call"):
                ret = ddbc_bindings.DDBCSQLFetchMany(
                    self.hstmt,
                    rows_data,
                    size,
                    char_enc,
                    wchar_enc,
                    self._cached_char_ctype,
                    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
                ]
        except Exception:
            # On error, don't increment rownumber - rethrow the error
            raise

    def fetchall(self) -> List[Row]:
        """
        Fetch all (remaining) rows of a query result.

        Returns:
            List of Row objects.
        """
        self._check_closed()  # Check if the cursor is closed
        if not self._has_result_set and self.description:
            self._reset_rownumber()

        if self._cached_decoding_generation != self._connection._decoding_generation:
            self._refresh_decoding_cache()
        char_enc = self._cached_char_encoding
        wchar_enc = self._cached_wchar_encoding

        # Fetch raw data
        rows_data = []
        try:
            with perf_phase("py::fetchall::cpp_call"):
                ret = ddbc_bindings.DDBCSQLFetchAll(
                    self.hstmt,
                    rows_data,
                    char_enc,
                    wchar_enc,
                    self._cached_char_ctype,
                    self.messages,
                )

            # Check for errors
            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:
                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::fetchall::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
                ]
        except Exception:
            # On error, don't increment rownumber - rethrow the error
            raise

    def arrow_batch(self, batch_size: int = 8192) -> "pyarrow.RecordBatch":
        """
        Fetch a single pyarrow Record Batch of the specified size from the
        query result set.

        Args:
            batch_size: Maximum number of rows to fetch in the Record Batch.

        Returns:
            A pyarrow RecordBatch object containing up to batch_size rows.
        """
        self._check_closed()  # Check if the cursor is closed
        pyarrow = self._ensure_pyarrow()

        if not self._has_result_set and self.description:
            self._reset_rownumber()

        capsules = []
        char_decoding = self._get_decoding_settings(ddbc_sql_const.SQL_CHAR.value)
        char_c_type = char_decoding.get("ctype", ddbc_sql_const.SQL_WCHAR.value)
        ret = ddbc_bindings.DDBCSQLFetchArrowBatch(
            self.hstmt, capsules, max(batch_size, 0), char_c_type, self.messages
        )
        check_error(ddbc_sql_const.SQL_HANDLE_STMT.value, self.hstmt, ret)

        batch = pyarrow.RecordBatch._import_from_c_capsule(*capsules)

        # Update rownumber for the number of rows actually fetched
        num_fetched = batch.num_rows
        if num_fetched > 0 and self._has_result_set:
            self._next_row_index += num_fetched
            self._rownumber = self._next_row_index - 1

        # Centralize rowcount assignment after fetch
        if num_fetched == 0 and self._next_row_index == 0:
            self.rowcount = 0
        else:
            self.rowcount = self._next_row_index

        return batch

    def arrow(self, batch_size: int = 8192) -> "pyarrow.Table":
        """
        Fetch the entire result as a pyarrow Table.

        Args:
            batch_size: Size of the Record Batches which make up the Table.

        Returns:
            A pyarrow Table containing all remaining rows from the result set.
        """
        self._check_closed()  # Check if the cursor is closed
        pyarrow = self._ensure_pyarrow()

        batches: list["pyarrow.RecordBatch"] = []
        while True:
            batch = self.arrow_batch(batch_size)
            if batch.num_rows < batch_size or batch_size <= 0:
                if not batches or batch.num_rows > 0:
                    batches.append(batch)
                break
            batches.append(batch)
        return pyarrow.Table.from_batches(batches, schema=batches[0].schema)

    def arrow_reader(self, batch_size: int = 8192) -> "_ArrowReader":
        """
        Fetch the result as a pyarrow-compatible RecordBatchReader, which
        yields Record Batches of the specified size until the current result
        set is exhausted.

        The returned object is an ``_ArrowReader`` wrapper that behaves like
        ``pyarrow.RecordBatchReader``
        (``schema``, ``read_next_batch``, iteration, context manager) but
        its ``close()`` is fully effective.  Cleanup is driven
        by a ``try/finally`` block inside the underlying batch generator, so
        the same teardown — ``SQLCancel`` to unblock any in-flight fetch on
        another thread, ``SQLFreeStmt(SQL_CLOSE)`` to release the server-side
        cursor and locks, draining diagnostics into ``cursor.messages``, and
        resetting the parent ``Cursor``'s rownumber / ``rowcount`` state —
        runs whether the user (a) exhausts the reader normally, (b) calls
        ``close()`` mid-iteration, (c) exits a ``with`` block, or (d) just
        lets the reader be garbage-collected.  The parent ``Cursor`` itself
        is **not** closed and can be re-executed.  ``close()`` is idempotent.

        Args:
            batch_size: Size of the Record Batches produced by the reader.

        Returns:
            A pyarrow-compatible RecordBatchReader for the result set.
        """
        self._check_closed()  # Check if the cursor is closed
        pyarrow = self._ensure_pyarrow()

        # Fetch schema without advancing cursor
        schema_batch = self.arrow_batch(0)
        schema = schema_batch.schema

        # Capture the parent cursor in a closure cell that the generator
        # can null out after cleanup, so a GC'd reader does not keep the
        # cursor pinned.
        cursor_ref = [self]
        close_requested = [False]

        def batch_generator():
            exhausted = False
            try:
                while (batch := cursor_ref[0].arrow_batch(batch_size)).num_rows > 0:
                    yield batch
                exhausted = True
            finally:
                # Symmetric server-side teardown — runs on exhaustion,
                # GeneratorExit (from close()), or an exception inside the
                # body.  This is the single canonical cleanup site.
                cur = cursor_ref[0]
                cursor_ref[0] = None
                if not cur.closed and cur.hstmt is not None:
                    # Natural EOF was captured natively. A close/cancel request
                    # (including a racing one) or an error still needs the drain.
                    if not exhausted or close_requested[0]:
                        try:
                            cur.messages.extend(ddbc_bindings.DDBCSQLGetAllDiagRecords(cur.hstmt))
                        except Exception as e:  # pylint: disable=broad-exception-caught
                            logger.debug("arrow_reader cleanup: pre-close diag drain failed: %s", e)

                    # 2) Release the server-side cursor & locks while keeping the
                    #    HSTMT and prepared plan intact, so the parent Cursor can
                    #    be re-executed.
                    try:
                        cur.hstmt._close_cursor()  # pylint: disable=protected-access
                    except Exception as e:  # pylint: disable=broad-exception-caught
                        # Elevated to WARNING: unlike the diag-drain failures
                        # (which only cost us some warning text), a failed
                        # SQLFreeStmt(SQL_CLOSE) leaves the server-side cursor
                        # and its locks/tempdb resources open on SQL Server
                        # until this parent Cursor is closed or re-executed.
                        # DEBUG is typically disabled in production, so that
                        # leak would be invisible; WARNING makes it visible.
                        logger.warning(
                            "arrow_reader cleanup: _close_cursor failed (%s); "
                            "server-side cursor may remain open until this "
                            "Cursor is closed or re-executed",
                            e,
                        )

                    # 3) Drain diagnostics produced by SQL_CLOSE itself.  This
                    #    runs unconditionally because SQL_CLOSE can return
                    #    SQL_SUCCESS_WITH_INFO (a *success* code) and still leave
                    #    warning records on the HSTMT diag stack; the previous
                    #    "only on failure" path would silently drop those.
                    try:
                        cur.messages.extend(ddbc_bindings.DDBCSQLGetAllDiagRecords(cur.hstmt))
                    except Exception as e:  # pylint: disable=broad-exception-caught
                        logger.debug("arrow_reader cleanup: post-close diag drain failed: %s", e)

                    # 4) Reset cursor bookkeeping to a clean "no result set"
                    #    state.  rowcount becomes -1 to signal that the prior
                    #    result is no longer meaningful.
                    try:
                        cur._clear_rownumber()  # pylint: disable=protected-access
                        cur.rowcount = -1
                    except Exception as e:  # pylint: disable=broad-exception-caught
                        logger.debug("arrow_reader cleanup: bookkeeping reset failed: %s", e)

        gen = batch_generator()
        inner = pyarrow.RecordBatchReader.from_batches(schema, gen)
        return _ArrowReader(self, inner, gen, pyarrow.ArrowInvalid, close_requested)

    def nextset(self) -> Optional[bool]:
        """
        Skip to the next available result set.

        Returns:
            True if there is another result set, False otherwise.
            Note: PEP 249 specifies True/None; we return True/False
            for backward compatibility with existing callers and pyodbc
            parity. The signature is Optional[bool] to keep the door
            open for a future migration to True/None semantics.

        Raises:
            Error: If the previous call to execute did not produce any result set.
        """
        logger.debug("nextset: Moving to next result set")
        self._check_closed()  # Check if the cursor is closed

        # Clear messages per DBAPI
        self.messages = []

        # Clear cached column and converter maps for the new result set
        self._cached_column_map = None
        self._cached_column_map_lower = None
        self._cached_converter_map = None
        self._column_sql_types = None
        self._uuid_str_indices = None

        # Skip to the next result set
        ret = ddbc_bindings.DDBCSQLMoreResults(self.hstmt)
        check_error(ddbc_sql_const.SQL_HANDLE_STMT.value, self.hstmt, ret)

        # Capture diagnostic messages (e.g. PRINT output) — handles both
        # SQL_SUCCESS_WITH_INFO and SQL_NO_DATA (trailing PRINT after the
        # final result set).  Without this, messages from subsequent result
        # sets are silently lost (GH-612).
        self._capture_diagnostics(ret)

        if ret == ddbc_sql_const.SQL_NO_DATA.value:
            logger.debug("nextset: No more result sets available")
            self._clear_rownumber()
            self.description = None
            return False

        self._reset_rownumber()

        # Initialize description for the new result set
        column_metadata = []
        try:
            ddbc_bindings.DDBCSQLDescribeCol(self.hstmt, column_metadata)
            self._initialize_description(column_metadata)

            # Pre-build column map and converter map for the new result set
            if self.description:
                self._cached_column_map = {
                    col_desc[0]: i for i, col_desc in enumerate(self.description)
                }
                self._cached_column_map_lower = (
                    {k.lower(): v for k, v in self._cached_column_map.items()}
                    if get_settings().lowercase
                    else None
                )
                self._cached_converter_map = self._build_converter_map()
                self._uuid_str_indices = self._compute_uuid_str_indices()
        except Exception as e:  # pylint: disable=broad-exception-caught
            # If describe fails, there might be no results in this result set
            self.description = None

        logger.debug(
            "nextset: Moved to next result set - column_count=%d",
            len(self.description) if self.description else 0,
        )
        return True

    # ── Mapping from ODBC connection-string keywords (lowercase, as _parse returns)
    def _build_pycore_context(self) -> dict:
        """Build the connection context dict expected by mssql_py_core.

        Parses the underlying ODBC connection string, validates the SERVER
        parameter, and (when Azure AD auth is in use) either registers a
        ServicePrincipal token factory or acquires a fresh access token,
        replacing credential fields as py-core requires. Returns a dict that
        can be handed to ``PyCoreConnection``. Shared by :meth:`bulkcopy`
        and :meth:`bulkcopy_arrow` so both paths get identical auth handling.

        Raises:
            RuntimeError: connection string unavailable or token acquisition failed.
            ValueError: SERVER parameter missing.
        """
        # Get and parse connection string
        if not hasattr(self.connection, "connection_str"):
            logger.error("_build_pycore_context: Connection string not available")
            raise RuntimeError("Connection string not available for bulk copy")

        # Use the proper connection string parser that handles braced values
        from mssql_python.connection_string_parser import _ConnectionStringParser

        parser = _ConnectionStringParser(validate_keywords=False)
        params = parser._parse(self.connection.connection_str)

        # Check for server parameter (accepts synonyms: server, addr, address)
        if not (params.get("server") or params.get("addr") or params.get("address")):
            raise ValueError("SERVER parameter is required in connection string")

        # Translate parsed connection string into the dict py-core expects.
        pycore_context = connstr_to_pycore_params(params)

        # Forward the cursor's query timeout to py-core so the bulkcopy
        # connection uses the same limit instead of py-core's compiled-in 15s
        # default. _timeout is the snapshot taken at cursor creation (same value
        # _set_timeout uses). Accept any int the public Connection.timeout setter
        # accepts (including IntEnum) but exclude bool, then normalise to a plain
        # int so py-core's u32 extract is unambiguous. 0 stays a "no override":
        # py-core's default connect path treats 0 as no deadline, but its TCP
        # attempt path turns 0 into a 0ms timeout that fails instantly, so we
        # leave 0 unset and let py-core apply its own 15s default.
        connect_timeout = self._timeout
        if isinstance(connect_timeout, int) and not isinstance(connect_timeout, bool):
            if connect_timeout > 0:
                pycore_context["connect_timeout"] = int(connect_timeout)

        # Token acquisition — only thing cursor must handle (needs azure-identity SDK)
        if self.connection._token_provider is not None:
            # User-supplied credential — use it directly for a fresh token.
            from mssql_python.auth import acquire_raw_token_from_credential

            try:
                raw_token, _ = acquire_raw_token_from_credential(self.connection._token_provider)
            except (OperationalError, InterfaceError) as e:
                # Preserve the original exception *type* so the same failure is
                # catchable the same way from connect() and bulkcopy(): an empty
                # or invalid token raises InterfaceError, while a provider or
                # transport failure raises OperationalError. Re-wrapping every
                # case as OperationalError would make an InterfaceError from
                # connect() surface as OperationalError here, splitting one
                # failure across two DB-API error categories.
                raise type(e)(
                    driver_error=(
                        "Bulk copy failed: unable to acquire token from custom credential"
                    ),
                    ddbc_error=str(e),
                ) from e
            pycore_context["access_token"] = raw_token
            logger.debug(
                "Bulk copy: acquired fresh token from custom credential (%s)",
                type(self.connection._token_provider).__name__,
            )
        elif self.connection._auth_type:
            # Fresh token acquisition for mssql-py-core connection
            from mssql_python.auth import AADAuth, ServicePrincipalAuth
            from mssql_python.constants import _AuthInternal

            if self.connection._auth_type == _AuthInternal.SERVICE_PRINCIPAL:
                # Callback-based path: tenant_id is only known from the STS URL
                # the server returns mid-handshake, so we register a factory
                # that py-core invokes during FedAuth (workflow 0x02).
                # Cert-based ServicePrincipal is not supported on this path.
                client_id = params.get("uid", "")
                client_secret = params.get("pwd", "")
                if not client_id or not client_secret:
                    raise RuntimeError(
                        "Bulk copy with Authentication=ActiveDirectoryServicePrincipal "
                        "currently supports client-secret only. "
                        "Provide UID (client_id) and PWD (client_secret) in the "
                        "connection string."
                    )
                try:
                    factory = ServicePrincipalAuth.make_token_factory(client_id, client_secret)
                except (RuntimeError, ValueError) as e:
                    raise RuntimeError(
                        f"Bulk copy failed: unable to build ServicePrincipal token factory: {e}"
                    ) from e
                pycore_context["entra_id_token_factory"] = factory
                # Keep authentication/user_name/password in pycore_context —
                # py-core's auth validator + transformer need them to resolve
                # the auth method to ActiveDirectoryServicePrincipal before
                # the factory is dispatched at handshake time.
                logger.debug("Bulk copy: registered ServicePrincipal token factory")
            else:
                # Pre-acquired token path. Used for Default, DeviceCode,
                # Interactive (non-Windows), MSI (system- or user-assigned),
                # and any other AD method whose tenant_id is discoverable
                # client-side via Azure Identity SDK. credential kwargs
                # (e.g. user-assigned MSI client_id) were captured by
                # Connection.__init__ before remove_sensitive_params stripped
                # UID from connection_str. re-parsing here would miss them.
                try:
                    raw_token = AADAuth.get_raw_token(
                        self.connection._auth_type,
                        self.connection._credential_kwargs,
                    )
                except (RuntimeError, ValueError) as e:
                    raise RuntimeError(
                        f"Bulk copy failed: unable to acquire Azure AD token "
                        f"for auth_type '{self.connection._auth_type}': {e}"
                    ) from e
                pycore_context["access_token"] = raw_token
                # Token replaces credential fields — py-core's validator rejects
                # access_token combined with authentication/user_name/password.
                for key in ("authentication", "user_name", "password"):
                    pycore_context.pop(key, None)
                logger.debug(
                    "Bulk copy: acquired fresh Azure AD token for auth_type=%s",
                    self.connection._auth_type,
                )

        return pycore_context

    def _looks_like_arrow_source(self, data) -> bool:
        """Return True if ``data`` should be routed to :meth:`bulkcopy_arrow`.

        Soft check: never imports pyarrow eagerly and never raises. Anything
        exposing the Arrow C-stream PyCapsule protocol counts, as do the
        concrete pyarrow container types when pyarrow is importable.
        """
        if data is None:
            return False
        if hasattr(data, "__arrow_c_stream__") or hasattr(data, "__arrow_c_array__"):
            return True
        try:
            import pyarrow as pa
        except ImportError:
            return False
        return isinstance(data, (pa.Table, pa.RecordBatch, pa.RecordBatchReader))

    @staticmethod
    def _bulkcopy_core_and_validate(table_name, batch_size, timeout):
        """Import the native core and validate the args shared by ``bulkcopy``
        and ``bulkcopy_arrow``. Returns the imported ``mssql_py_core`` module."""
        try:
            import mssql_py_core
        except ImportError as exc:
            logger.error("bulkcopy: Failed to import mssql_py_core module")
            raise ImportError(
                "Bulk copy requires the mssql_py_core library which is not available. "
                "This is an unexpected error. "
            ) from exc

        if not table_name or not isinstance(table_name, str):
            logger.error("bulkcopy: Invalid table_name parameter")
            raise ValueError("table_name must be a non-empty string")

        if not isinstance(batch_size, int):
            raise TypeError(
                f"batch_size must be a non-negative integer, got {type(batch_size).__name__}"
            )
        if batch_size < 0:
            raise ValueError(f"batch_size must be non-negative, got {batch_size}")

        if not isinstance(timeout, int) or isinstance(timeout, bool):
            raise TypeError(f"timeout must be a non-negative integer, got {type(timeout).__name__}")
        if timeout < 0:
            raise ValueError(f"timeout must be non-negative, got {timeout}")

        return mssql_py_core

    @staticmethod
    def _bulkcopy_teardown(pycore_context, pycore_cursor, pycore_connection):
        """Scrub credential material from the context and close native
        bulk-copy resources. Safe to call with partially-initialized state."""
        if pycore_context:
            for key in ("password", "user_name", "access_token", "entra_id_token_factory"):
                pycore_context.pop(key, None)
        for resource in (pycore_cursor, pycore_connection):
            if resource and hasattr(resource, "close"):
                try:
                    resource.close()
                except Exception as cleanup_error:
                    logger.debug(
                        "Failed to close bulk copy resource %s: %s",
                        type(resource).__name__,
                        cleanup_error,
                    )

    def bulkcopy(
        self,
        table_name: str,
        data: Iterable[Union[Tuple, "Row"]],
        batch_size: int = 0,
        timeout: int = 30,
        column_mappings: Optional[Union[List[str], List[Tuple[int, str]]]] = None,
        keep_identity: bool = False,
        check_constraints: bool = False,
        table_lock: bool = False,
        keep_nulls: bool = False,
        fire_triggers: bool = False,
        use_internal_transaction: bool = False,
    ):  # pragma: no cover
        """
        Perform bulk copy operation for high-performance data loading.

        Args:
            table_name: Target table name (can include schema, e.g., 'dbo.MyTable').
                The table must exist and the user must have INSERT permissions.

            data: Iterable of tuples or Row objects containing row data to be inserted.
                Row objects from fetchone/fetchmany/fetchall are automatically
                converted to tuples. Lists and other types are not accepted.

                Data Format Requirements:
                - Each element in the iterable represents one row
                - Each row should be a tuple or Row object
                - Column order must match the target table's column order (by ordinal
                  position), unless column_mappings is specified
                - The number of values in each row must match the number of columns
                  in the target table

            batch_size: Number of rows to send per batch. Default 0 uses server optimal.

            timeout: Operation timeout in seconds. Default is 30. 0 disables the
                bulk copy operation timeout.

            column_mappings: Maps source data columns to target table column names.
                Two formats supported:

                Simple Format - List[str]:
                    List of destination column names in order. Position in list = source index.
                    Example: ['UserID', 'FirstName', 'Email']
                    Maps: index 0 → UserID, index 1 → FirstName, index 2 → Email

                Advanced Format - List[Tuple[int, str]]:
                    Explicit index mapping. Allows skipping or reordering columns.
                    Each tuple is (source_index, target_column_name).
                    Example: [(0, 'UserID'), (1, 'FirstName'), (3, 'Email')]
                    Maps: index 0 → UserID, index 1 → FirstName, index 3 → Email (skips index 2)

                When omitted: Columns are mapped by ordinal position (first data
                column → first table column, second → second, etc.)

            keep_identity: Preserve identity values from source data.

            check_constraints: Check constraints during bulk copy.

            table_lock: Use table-level lock instead of row-level locks.

            keep_nulls: Preserve null values instead of using default values.

            fire_triggers: Fire insert triggers on the target table.

            use_internal_transaction: Use an internal transaction for each batch.

        Returns:
            Dictionary with bulk copy results including:
                - rows_copied: Number of rows successfully copied
                - batch_count: Number of batches processed
                - elapsed_time: Time taken for the operation

        Raises:
            ImportError: If mssql_py_core library is not installed
            TypeError: If data is None, not iterable, is a string/bytes, or is an
                Arrow source (use :meth:`bulkcopy_arrow` for those)
            ValueError: If table_name is empty or parameters are invalid
            RuntimeError: If connection string is not available
        """
        # Fast check if logging is enabled to avoid overhead
        is_logging_enabled = logger.is_debug_enabled

        # Steer Arrow-shaped sources to the dedicated method instead of
        # silently re-routing or failing deep inside the tuple validator.
        if self._looks_like_arrow_source(data):
            raise TypeError(
                "bulkcopy() expects an iterable of row tuples or Row objects. "
                "For pyarrow.Table / RecordBatch / RecordBatchReader / objects "
                "implementing __arrow_c_stream__/__arrow_c_array__, call "
                "cursor.bulkcopy_arrow() instead."
            )

        mssql_py_core = self._bulkcopy_core_and_validate(table_name, batch_size, timeout)

        # Validate that data is iterable (but not a string or bytes, which are technically iterable)
        if data is None:
            raise TypeError("data must be an iterable of tuples or lists, got None")
        if isinstance(data, (str, bytes)):
            raise TypeError(
                f"data must be an iterable of tuples or lists, got {type(data).__name__}. "
                "Strings and bytes are not valid row collections."
            )
        if not hasattr(data, "__iter__"):
            raise TypeError(
                f"data must be an iterable of tuples or lists, got non-iterable {type(data).__name__}"
            )

        pycore_context = self._build_pycore_context()

        pycore_connection = None
        pycore_cursor = None
        try:
            # Only pass logger to Rust if logging is enabled (performance optimization)
            pycore_connection = mssql_py_core.PyCoreConnection(
                pycore_context, python_logger=logger if is_logging_enabled else None
            )
            pycore_cursor = pycore_connection.cursor()

            # Enforce the bulkcopy type contract: only tuple and Row accepted.
            # Created to support the fetchone/fetchmany/fetchall -> bulkcopy
            # pipeline where Row objects need conversion to native tuples.
            # Rust (mssql_py_core) requires PyTuple via cast::<PyTuple>() and
            # rejects all other types including list. Row objects are converted
            # using direct _values access (4x faster than __iter__ protocol).
            # Uses itertools.chain for C-level iteration (avoids Python
            # generator frame overhead on the tuple passthrough path).
            def _prepare_row_iterator(iterable):
                from itertools import chain

                it = iter(iterable)
                first = next(it, None)
                if first is None:
                    return iter(())
                if isinstance(first, tuple):
                    return chain((first,), it)
                if isinstance(first, Row):
                    return (tuple(item._values) for item in chain((first,), it))
                raise TypeError(
                    f"bulkcopy data rows must be tuples or Row objects, "
                    f"got {type(first).__name__}"
                )

            # Call bulkcopy with explicit keyword arguments
            # The API signature: bulkcopy(table_name, data_source, batch_size=0, timeout=30, ...)
            result = pycore_cursor.bulkcopy(
                table_name,
                _prepare_row_iterator(data),
                batch_size=batch_size,
                timeout=timeout,
                column_mappings=column_mappings,
                keep_identity=keep_identity,
                check_constraints=check_constraints,
                table_lock=table_lock,
                keep_nulls=keep_nulls,
                fire_triggers=fire_triggers,
                use_internal_transaction=use_internal_transaction,
                python_logger=logger if is_logging_enabled else None,  # Only pass logger if enabled
            )

            logger.info(
                "_bulkcopy: Bulk copy completed successfully - rows_copied=%s, batch_count=%s, elapsed_time=%s",
                result.get("rows_copied", "N/A"),
                result.get("batch_count", "N/A"),
                result.get("elapsed_time", "N/A"),
            )

            return result

        except Exception as e:
            # Log the error for debugging (without exposing credentials)
            logger.debug(
                "Bulk copy operation failed for table '%s': %s: %s",
                table_name,
                type(e).__name__,
                str(e),
            )
            # Re-raise the original exception, preserving its type, args, and
            # traceback. The finally block scrubs credentials, and Python
            # tracebacks don't expose local values, so no reconstruction is needed.
            raise

        finally:
            self._bulkcopy_teardown(pycore_context, pycore_cursor, pycore_connection)

    def bulkcopy_arrow(
        self,
        table_name: str,
        source,
        batch_size: int = 0,
        timeout: int = 30,
        column_mappings: Optional[Union[List[str], List[Tuple[int, str]]]] = None,
        keep_identity: bool = False,
        check_constraints: bool = False,
        table_lock: bool = False,
        keep_nulls: bool = False,
        fire_triggers: bool = False,
        use_internal_transaction: bool = False,
    ):
        """Bulk-copy from an Apache Arrow source straight into TDS.

        ``source`` may be any of:

        * ``pyarrow.Table``
        * ``pyarrow.RecordBatch``
        * ``pyarrow.RecordBatchReader``
        * any object exposing ``__arrow_c_stream__`` (Arrow PyCapsule stream
          interface — e.g. polars/pandas DataFrames ≥ 2.2, duckdb results)
        * any object exposing ``__arrow_c_array__`` (single-batch PyCapsule
          array interface)
        * an iterable of ``pyarrow.RecordBatch`` (all batches must share the
          same schema)

        Compared to :meth:`bulkcopy`, this path skips per-cell Python
        round-trips: each batch's typed Arrow buffers are read directly by
        the Rust core and streamed into the TDS bulk-load packets. Schema and
        column-mapping semantics, the options, and the returned dictionary are
        identical to :meth:`bulkcopy`.

        Args:
            table_name: Target table name (may include schema, e.g. 'dbo.MyTable').
            source: Arrow source (see above).
            batch_size: Rows per TDS commit. Default 0 uses server optimal.
            timeout: Operation timeout in seconds. Default 30. 0 disables the
                bulk copy operation timeout.
            column_mappings: Same two formats as :meth:`bulkcopy`. When omitted,
                Arrow fields map to destination columns by ordinal position.
            keep_identity: Preserve identity values from the source.
            check_constraints: Check constraints during bulk copy.
            table_lock: Use a table-level lock instead of row-level locks.
            keep_nulls: Preserve null values instead of using column defaults.
            fire_triggers: Fire insert triggers on the target table.
            use_internal_transaction: Use an internal transaction per batch.

        Returns:
            Dictionary with ``rows_copied``, ``batch_count``, ``elapsed_time``
            and ``rows_per_second``.

        Raises:
            ImportError: If the mssql_py_core library is not installed.
            TypeError: If ``source`` is None, a str, or bytes.
            ValueError: If ``table_name`` is empty or parameters are invalid.
            RuntimeError: If the connection string is not available.

        Example:
            >>> import pyarrow as pa
            >>> table = pa.table({"id": [1, 2, 3], "name": ["a", "b", "c"]})
            >>> result = cursor.bulkcopy_arrow("dbo.MyTable", table)
            >>> result["rows_copied"]
            3
        """
        is_logging_enabled = logger.is_debug_enabled

        mssql_py_core = self._bulkcopy_core_and_validate(table_name, batch_size, timeout)

        if source is None or isinstance(source, (str, bytes)):
            got = "None" if source is None else type(source).__name__
            raise TypeError(
                "source must be a pyarrow Table/RecordBatch/RecordBatchReader, "
                "an iterable of RecordBatch, or an object implementing "
                f"__arrow_c_stream__/__arrow_c_array__, got {got}"
            )

        pycore_context = self._build_pycore_context()

        pycore_connection = None
        pycore_cursor = None
        try:
            pycore_connection = mssql_py_core.PyCoreConnection(
                pycore_context, python_logger=logger if is_logging_enabled else None
            )
            pycore_cursor = pycore_connection.cursor()

            result = pycore_cursor.bulkcopy_arrow(
                table_name,
                source,
                batch_size=batch_size,
                timeout=timeout,
                column_mappings=column_mappings,
                keep_identity=keep_identity,
                check_constraints=check_constraints,
                table_lock=table_lock,
                keep_nulls=keep_nulls,
                fire_triggers=fire_triggers,
                use_internal_transaction=use_internal_transaction,
                python_logger=logger if is_logging_enabled else None,
            )

            logger.info(
                "bulkcopy_arrow: completed - rows_copied=%s, batch_count=%s, elapsed_time=%s",
                result.get("rows_copied", "N/A"),
                result.get("batch_count", "N/A"),
                result.get("elapsed_time", "N/A"),
            )
            return result

        except Exception as e:
            logger.debug(
                "bulkcopy_arrow failed for table '%s': %s: %s",
                table_name,
                type(e).__name__,
                str(e),
            )
            raise

        finally:
            self._bulkcopy_teardown(pycore_context, pycore_cursor, pycore_connection)

    def __enter__(self):
        """
        Enter the runtime context for the cursor.

        Returns:
            The cursor instance itself.
        """
        self._check_closed()
        return self

    def __exit__(self, *args):
        """Closes the cursor when exiting the context, ensuring proper resource cleanup."""
        if not self.closed:
            self.close()

    def fetchval(self):
        """
        Fetch the first column of the first row if there are results.

        This is a convenience method for queries that return a single value,
        such as SELECT COUNT(*) FROM table, SELECT MAX(id) FROM table, etc.

        Returns:
            The value of the first column of the first row, or None if no rows
            are available or the first column value is NULL.

        Raises:
            Exception: If the cursor is closed.

        Example:
            >>> count = cursor.execute('SELECT COUNT(*) FROM users').fetchval()
            >>> max_id = cursor.execute('SELECT MAX(id) FROM users').fetchval()
            >>> name = cursor.execute('SELECT name FROM users WHERE id = ?', user_id).fetchval()

        Note:
            This is a convenience extension beyond the DB-API 2.0 specification.
            After calling fetchval(), the cursor position advances by one row,
            just like fetchone().
        """
        if logger.is_debug_enabled:
            logger.debug("fetchval: Fetching single value from first column")
        self._check_closed()  # Check if the cursor is closed

        # Check if this is a result-producing statement
        if not self.description:
            # Non-result-set statement (INSERT, UPDATE, DELETE, etc.)
            if logger.is_debug_enabled:
                logger.debug("fetchval: No result set available (non-SELECT statement)")
            return None

        # Fetch the first row
        row = self.fetchone()

        if row is None:
            if logger.is_debug_enabled:
                logger.debug("fetchval: No value available (no rows)")
            return None

        if logger.is_debug_enabled:
            logger.debug("fetchval: Value retrieved successfully")
        return row[0]

    def commit(self):
        """
        Commit all SQL statements executed on the connection that created this cursor.

        This is a convenience method that calls commit() on the underlying connection.
        It affects all cursors created by the same connection since the last commit/rollback.

        The benefit is that many uses can now just use the cursor and not have to track
        the connection object.

        Raises:
            Exception: If the cursor is closed or if the commit operation fails.

        Example:
            >>> cursor.execute("INSERT INTO users (name) VALUES (?)", "John")
            >>> cursor.commit()  # Commits the INSERT

        Note:
            This is equivalent to calling connection.commit() but provides convenience
            for code that only has access to the cursor object.
        """
        self._check_closed()  # Check if the cursor is closed

        # Clear messages per DBAPI
        self.messages = []

        # Delegate to the connection's commit method
        self._connection.commit()

    def rollback(self):
        """
        Roll back all SQL statements executed on the connection that created this cursor.

        This is a convenience method that calls rollback() on the underlying connection.
        It affects all cursors created by the same connection since the last commit/rollback.

        The benefit is that many uses can now just use the cursor and not have to track
        the connection object.

        Raises:
            Exception: If the cursor is closed or if the rollback operation fails.

        Example:
            >>> cursor.execute("INSERT INTO users (name) VALUES (?)", "John")
            >>> cursor.rollback()  # Rolls back the INSERT

        Note:
            This is equivalent to calling connection.rollback() but provides convenience
            for code that only has access to the cursor object.
        """
        self._check_closed()  # Check if the cursor is closed

        # Clear messages per DBAPI
        self.messages = []

        # Delegate to the connection's rollback method
        self._connection.rollback()

    def __del__(self):
        """
        Destructor to ensure the cursor is closed when it is no longer needed.
        This is a safety net to ensure resources are cleaned up
        even if close() was not called explicitly.
        If the cursor is already closed, it will not raise an exception during cleanup.
        """
        if "closed" not in self.__dict__ or not self.closed:
            try:
                self.close()
            except Exception as e:  # pylint: disable=broad-exception-caught
                # Don't raise an exception in __del__, just log it
                # If interpreter is shutting down, we might not have logging set up
                import sys

                if sys and sys.is_finalizing():
                    # Suppress logging during interpreter shutdown
                    return
                # ``logger`` could be torn down or have its handlers closed
                # late in interpreter shutdown; guard so the debug call
                # cannot itself raise an unraisable exception.
                if logger is not None:
                    logger.debug("Exception during cursor cleanup in __del__: %s", e)

    def scroll(
        self, value: int, mode: str = "relative"
    ) -> None:  # pylint: disable=too-many-branches
        """
        Scroll using SQLFetchScroll only, matching test semantics:
          - relative(N>0): consume N rows; rownumber = previous + N;
            next fetch returns the following row.
          - absolute(-1): before first (rownumber = -1), no data consumed.
          - absolute(0): position so next fetch returns first row;
            rownumber stays 0 even after that fetch.
          - absolute(k>0): next fetch returns row index k (0-based);
            rownumber == k after scroll.
        """
        logger.debug(
            "scroll: Scrolling cursor - mode=%s, value=%d, current_rownumber=%d",
            mode,
            value,
            self._rownumber,
        )
        self._check_closed()

        # Clear messages per DBAPI
        self.messages = []

        if mode not in ("relative", "absolute"):
            logger.error("scroll: Invalid mode - mode=%s", mode)
            raise ProgrammingError(
                "Invalid scroll mode",
                f"mode must be 'relative' or 'absolute', got '{mode}'",
            )
        if not self._has_result_set:
            logger.error("scroll: No active result set")
            raise ProgrammingError(
                "No active result set",
                "Cannot scroll: no result set available. Execute a query first.",
            )
        if not isinstance(value, int):
            logger.error("scroll: Invalid value type - type=%s", type(value).__name__)
            raise ProgrammingError(
                "Invalid scroll value type",
                f"scroll value must be an integer, got {type(value).__name__}",
            )

        # Relative backward not supported
        if mode == "relative" and value < 0:
            logger.error("scroll: Backward scrolling not supported - value=%d", value)
            raise NotSupportedError(
                "Backward scrolling not supported",
                f"Cannot move backward by {value} rows on a forward-only cursor",
            )

        row_data: list = []

        # Absolute positioning not supported with forward-only cursors
        if mode == "absolute":
            raise NotSupportedError(
                "Absolute positioning not supported",
                "Forward-only cursors do not support absolute positioning",
            )

        try:
            if mode == "relative":
                if value == 0:
                    return

                # For forward-only cursors, use multiple SQL_FETCH_NEXT calls
                # This matches pyodbc's approach for skip operations
                for i in range(value):
                    ret = ddbc_bindings.DDBCSQLFetchScroll(
                        self.hstmt, ddbc_sql_const.SQL_FETCH_NEXT.value, 0, row_data
                    )
                    if ret == ddbc_sql_const.SQL_NO_DATA.value:
                        raise IndexError(
                            "Cannot scroll to specified position: end of result set reached"
                        )

                # Update position tracking
                self._rownumber = self._rownumber + value
                self._next_row_index = self._rownumber + 1
                logger.debug(
                    "scroll: Scroll complete - new_rownumber=%d, next_row_index=%d",
                    self._rownumber,
                    self._next_row_index,
                )
                return

        except Exception as e:  # pylint: disable=broad-exception-caught
            if isinstance(e, (IndexError, NotSupportedError)):
                raise
            raise IndexError(f"Scroll operation failed: {e}") from e

    def skip(self, count: int) -> None:
        """
        Skip the next count records in the query result set.

        Args:
            count: Number of records to skip.

        Raises:
            IndexError: If attempting to skip past the end of the result set.
            ProgrammingError: If count is not an integer.
            NotSupportedError: If attempting to skip backwards.
        """
        self._check_closed()

        # Clear messages
        self.messages = []

        # Simply delegate to the scroll method with 'relative' mode
        self.scroll(count, "relative")

    def _execute_tables(  # pylint: disable=too-many-arguments,too-many-positional-arguments
        self,
        stmt_handle,
        catalog_name=None,
        schema_name=None,
        table_name=None,
        table_type=None,
    ):
        """
        Execute SQLTables ODBC function to retrieve table metadata.

        Args:
            stmt_handle: ODBC statement handle
            catalog_name: The catalog name pattern
            schema_name: The schema name pattern
            table_name: The table name pattern
            table_type: The table type filter
            search_escape: The escape character for pattern matching
        """
        # Convert None values to empty strings for ODBC
        catalog = "" if catalog_name is None else catalog_name
        schema = "" if schema_name is None else schema_name
        table = "" if table_name is None else table_name
        types = "" if table_type is None else table_type

        # Call the ODBC SQLTables function
        retcode = ddbc_bindings.DDBCSQLTables(stmt_handle, catalog, schema, table, types)

        # Check return code and handle errors
        check_error(ddbc_sql_const.SQL_HANDLE_STMT.value, stmt_handle, retcode)

        # Capture any diagnostic messages
        if stmt_handle and retcode in (
            ddbc_sql_const.SQL_SUCCESS_WITH_INFO.value,
            ddbc_sql_const.SQL_NO_DATA.value,
        ):
            self.messages.extend(ddbc_bindings.DDBCSQLGetAllDiagRecords(stmt_handle))

    def tables(
        self, table=None, catalog=None, schema=None, tableType=None
    ):  # pylint: disable=too-many-arguments,too-many-positional-arguments
        """
        Returns information about tables in the database that match the given criteria using
        the SQLTables ODBC function.

        Args:
            table (str, optional): The table name pattern. Default is None (all tables).
            catalog (str, optional): The catalog name. Default is None.
            schema (str, optional): The schema name pattern. Default is None.
            tableType (str or list, optional): The table type filter. Default is None.
                                              Example: "TABLE" or ["TABLE", "VIEW"]

        Returns:
            Cursor: The cursor object itself for method chaining with fetch methods.
        """
        self._check_closed()
        self._reset_cursor()

        # Format table_type parameter - SQLTables expects comma-separated string
        table_type_str = None
        if tableType is not None:
            if isinstance(tableType, (list, tuple)):
                table_type_str = ",".join(tableType)
            else:
                table_type_str = str(tableType)

        try:
            # Call SQLTables via the helper method
            self._execute_tables(
                self.hstmt,
                catalog_name=catalog,
                schema_name=schema,
                table_name=table,
                table_type=table_type_str,
            )

            # Define fallback description for tables
            fallback_description = [
                ("table_cat", str, None, 128, 128, 0, True),
                ("table_schem", str, None, 128, 128, 0, True),
                ("table_name", str, None, 128, 128, 0, False),
                ("table_type", str, None, 128, 128, 0, False),
                ("remarks", str, None, 254, 254, 0, True),
            ]

            # Use the helper method to prepare the result set
            return self._prepare_metadata_result_set(fallback_description=fallback_description)

        except Exception as e:  # pylint: disable=broad-exception-caught
            # Log the error and re-raise
            logger.error(f"Error executing tables query: {e}")
            raise

    def callproc(
        self, procname: str, parameters: Optional[Sequence[Any]] = None
    ) -> Optional[Sequence[Any]]:
        """
        Call a stored database procedure with the given name.

        Args:
            procname: Name of the stored procedure to call
            parameters: Optional sequence of parameters to pass to the procedure

        Returns:
            A sequence containing the result parameters (input parameters unchanged,
            output parameters with their new values)

        Raises:
            NotSupportedError: This method is not yet implemented
        """
        raise NotSupportedError(
            driver_error="callproc() is not yet implemented",
            ddbc_error="Stored procedure calls are not currently supported",
        )

    def setoutputsize(self, size: int, column: Optional[int] = None) -> None:
        """
        Set a column buffer size for fetches of large columns.

        This method is optional and is not implemented in this driver.

        Args:
            size: Maximum size of the column buffer
            column: Optional column index (0-based) to set the size for

        Note:
            This method is a no-op in this implementation as buffer sizes
            are managed automatically by the underlying driver.
        """
        # This is a no-op - buffer sizes are managed automatically