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 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")


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 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
        """
        self._connection: "Connection" = connection  # Store as private attribute
        self._timeout: int = timeout
        self._inputsizes: Optional[List[Union[int, Tuple[Any, ...]]]] = None
        # self.connection.autocommit = False
        self.hstmt: Optional[Any] = None
        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.closed: bool = False
        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.

        # Initialize attributes that may be defined later to avoid pylint warnings
        # Note: _original_fetch* methods are not initialized here as they need to be
        # conditionally set based on hasattr() checks
        # 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

        self._cached_column_map = None
        self._cached_converter_map = 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
        # Cache decoding encoding strings — these don't change between fetches,
        # so we avoid 2 method calls + 2 dict.get() per fetch call.
        self._cached_char_encoding = self._get_decoding_settings(ddbc_sql_const.SQL_CHAR.value).get(
            "encoding", "utf-8"
        )
        self._cached_wchar_encoding = self._get_decoding_settings(
            ddbc_sql_const.SQL_WCHAR.value
        ).get("encoding", "utf-16le")
        self._skip_increment_for_next_fetch = (
            False  # Track if we need to skip incrementing the row index
        )
        self.messages = []  # 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

        # Handle special values (NaN, Infinity, etc.)
        if isinstance(exponent, str):
            # For special values like 'n' (NaN), 'N' (sNaN), 'F' (Infinity)
            # Return default precision and scale
            precision = 38  # SQL Server default max precision
            scale = 0
        else:
            # 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 _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,
    ) -> 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.
        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)
            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,
                )
            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

            # Handle special values (NaN, Infinity, etc.)
            if isinstance(exponent, str):
                logger.debug(
                    "_map_sql_type: DECIMAL special value - index=%d, exponent=%s", i, exponent
                )
                # For special values like 'n' (NaN), 'N' (sNaN), 'F' (Infinity)
                # Return default precision and scale
                precision = 38  # SQL Server default max precision
            else:
                # 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
            if 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 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_TIMESTAMP.value,
                ddbc_sql_const.SQL_C_TYPE_TIMESTAMP.value,
                26,
                6,
                False,
            )

        if isinstance(param, datetime.date):
            return (
                ddbc_sql_const.SQL_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.
        """
        if self.closed:
            # 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 _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)

        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."
                        )

                    # 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, 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."
                        )

                    self._inputsizes.append((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,
    ) -> Tuple[int, int, int, int, bool]:
        """
        Maps parameter types for the given parameter.
        Args:
            parameter: parameter to bind.
        Returns:
            paraminfo.
        """
        paraminfo = param_info()

        # Check if we have explicit type information from setinputsizes
        if self._inputsizes and i < len(self._inputsizes):
            # Use explicit type information
            sql_type, column_size, decimal_digits = self._inputsizes[i]

            # Default is_dae to False for explicit types, but set to True for large strings/binary
            is_dae = False

            if parameter is None:
                # For NULL parameters, always use SQL_C_DEFAULT regardless of SQL type
                c_type = ddbc_sql_const.SQL_C_DEFAULT.value
            else:
                # For non-NULL parameters, determine the appropriate C type based on SQL type
                c_type = self._get_c_type_for_sql_type(sql_type)

                # Override DECIMAL/NUMERIC to use SQL_C_CHAR string binding (GH-503).
                # The generic mapping returns SQL_C_NUMERIC which requires NumericData
                # structs, but setinputsizes declares fixed precision/scale that may
                # differ from per-value precision, causing misinterpretation. String
                # binding lets ODBC convert using the declared columnSize/decimalDigits.
                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
                    if isinstance(parameter, decimal.Decimal):
                        parameters_list[i] = format(parameter, "f")
                        parameter = parameters_list[i]

                # Check if this should be a DAE (data at execution) parameter
                # For string types with large column sizes
                if isinstance(parameter, str) and column_size > MAX_INLINE_CHAR:
                    is_dae = True
                # For binary types with large column sizes
                elif isinstance(parameter, (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)

        else:
            # Fall back to automatic type inference
            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
            )

        # 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
            return

        description = []
        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()

            # 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

    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.
        This eliminates the need to look up converters for every row.
        """
        if (
            not self.description
            or not hasattr(self.connection, "_output_converters")
            or not self.connection._output_converters
        ):
            return None

        converter_map = []

        for desc in self.description:
            if desc is None:
                converter_map.append(None)
                continue
            sql_type = desc[1]
            converter = self.connection.get_output_converter(sql_type)
            # If no converter found for the SQL type, try the WVARCHAR converter as a fallback
            if converter is None:
                from mssql_python.constants import ConstantsDDBC

                converter = self.connection.get_output_converter(ConstantsDDBC.SQL_WVARCHAR.value)

            converter_map.append(converter)

        return converter_map

    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

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

        # Get cached converter map
        converter_map = getattr(self, "_cached_converter_map", None)

        return column_map, converter_map

    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)

        # Restore original fetch methods if they exist
        if hasattr(self, "_original_fetchone"):
            logger.debug("execute: Restoring original fetch methods")
            self.fetchone = self._original_fetchone
            self.fetchmany = self._original_fetchmany
            self.fetchall = self._original_fetchall
            del self._original_fetchone
            del self._original_fetchmany
            del self._original_fetchall

        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).
        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]
                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()

        # Apply timeout if set (non-zero)
        logger.debug("execute: Creating parameter type list")
        param_info = ddbc_bindings.ParamInfo
        parameters_type = []

        # 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,
                )

        if parameters:
            for i, param in enumerate(parameters):
                paraminfo = self._create_parameter_types_list(param, param_info, parameters, i)
                parameters_type.append(paraminfo)

        # 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

        if logger.isEnabledFor(logging.DEBUG):
            for i, param in enumerate(parameters):
                logger.debug(
                    """Parameter number: %s, Parameter: %s,
                    Param Python Type: %s, ParamInfo: %s, %s, %s, %s, %s""",
                    i + 1,
                    param,
                    str(type(param)),
                    parameters_type[i].paramSQLType,
                    parameters_type[i].paramCType,
                    parameters_type[i].columnSize,
                    parameters_type[i].decimalDigits,
                    parameters_type[i].inputOutputType,
                )

        ret = ddbc_bindings.DDBCSQLExecute(
            self.hstmt,
            operation,
            parameters,
            parameters_type,
            self.is_stmt_prepared,
            effective_use_prepare,
            encoding_settings,
        )
        # 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 diagnostic messages only on SQL_SUCCESS_WITH_INFO.
        # SQL_SUCCESS has no records — calling DDBCSQLGetAllDiagRecords on it
        # costs ~10ms/call (driver scans internal state to find nothing).
        # SQL_ERROR is already handled by check_error() above which extracts
        # diagnostics and raises.
        if ret == ddbc_sql_const.SQL_SUCCESS_WITH_INFO.value and self.hstmt:
            self.messages.extend(ddbc_bindings.DDBCSQLGetAllDiagRecords(self.hstmt))

        self.last_executed_stmt = operation

        # 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

        # 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_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_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
        4. Creating wrapper fetch methods with column mapping support

        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

        # Define column names in ODBC standard order
        self._column_map = {}  # pylint: disable=attribute-defined-outside-init
        for i, (name, *_) in enumerate(self.description):
            # Add standard name
            self._column_map[name] = i
            # Add lowercase alias
            self._column_map[name.lower()] = i

        # If specialized mapping is provided, handle it differently
        if specialized_mapping:
            # Define specialized fetch methods that use the custom mapping
            def fetchone_with_specialized_mapping():
                row = self._original_fetchone()
                if row is not None:
                    merged_map = getattr(row, "_column_map", {}).copy()
                    merged_map.update(specialized_mapping)
                    row._column_map = merged_map
                return row

            def fetchmany_with_specialized_mapping(size=None):
                rows = self._original_fetchmany(size)
                for row in rows:
                    merged_map = getattr(row, "_column_map", {}).copy()
                    merged_map.update(specialized_mapping)
                    row._column_map = merged_map
                return rows

            def fetchall_with_specialized_mapping():
                rows = self._original_fetchall()
                for row in rows:
                    merged_map = getattr(row, "_column_map", {}).copy()
                    merged_map.update(specialized_mapping)
                    row._column_map = merged_map
                return rows

            # Save original fetch methods
            if not hasattr(self, "_original_fetchone"):
                self._original_fetchone = (
                    self.fetchone
                )  # pylint: disable=attribute-defined-outside-init
                self._original_fetchmany = (
                    self.fetchmany
                )  # pylint: disable=attribute-defined-outside-init
                self._original_fetchall = (
                    self.fetchall
                )  # pylint: disable=attribute-defined-outside-init

            # Use specialized mapping methods
            self.fetchone = fetchone_with_specialized_mapping
            self.fetchmany = fetchmany_with_specialized_mapping
            self.fetchall = fetchall_with_specialized_mapping
        else:
            # Standard column mapping
            # Remember original fetch methods (store only once)
            if not hasattr(self, "_original_fetchone"):
                self._original_fetchone = (
                    self.fetchone
                )  # pylint: disable=attribute-defined-outside-init
                self._original_fetchmany = (
                    self.fetchmany
                )  # pylint: disable=attribute-defined-outside-init
                self._original_fetchall = (
                    self.fetchall
                )  # pylint: disable=attribute-defined-outside-init

                # Create wrapper fetch methods that add column mappings
                def fetchone_with_mapping():
                    row = self._original_fetchone()
                    if row is not None:
                        row._column_map = self._column_map
                    return row

                def fetchmany_with_mapping(size=None):
                    rows = self._original_fetchmany(size)
                    for row in rows:
                        row._column_map = self._column_map
                    return rows

                def fetchall_with_mapping():
                    rows = self._original_fetchall()
                    for row in rows:
                        row._column_map = self._column_map
                    return rows

                # Replace fetch methods
                self.fetchone = fetchone_with_mapping
                self.fetchmany = fetchmany_with_mapping
                self.fetchall = fetchall_with_mapping

        # 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: list) -> 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).
        """
        non_nulls = [v for v in column if v is not None]
        if not non_nulls:
            return None, None, None

        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

        sample_value = None
        for v in non_nulls:
            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

    def executemany(  # pylint: disable=too-many-locals,too-many-branches,too-many-statements
        self, operation: str, seq_of_parameters: Union[List[Sequence[Any]], List[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
        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, column_size, decimal_digits = self._inputsizes[col_index]

                # Default is_dae to False
                is_dae = False

                # Determine appropriate C type based on SQL type
                c_type = self._get_c_type_for_sql_type(sql_type)

                # 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 = 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,
                )

                # For executemany with all-NULL columns, SQL_UNKNOWN_TYPE doesn't work
                # with array binding. Fall back to SQL_VARCHAR as a safe default.
                if (
                    sample_value is None
                    and paraminfo.paramSQLType == ddbc_sql_const.SQL_UNKNOWN_TYPE.value
                ):
                    paraminfo.paramSQLType = ddbc_sql_const.SQL_VARCHAR.value
                    paraminfo.columnSize = 1

                # 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.",
            )
            for row in seq_of_parameters:
                self.execute(operation, row)
            return

        # Process parameters into column-wise format with possible type conversions
        # First, convert any Decimal types as needed for NUMERIC/DECIMAL columns
        processed_parameters = []
        for row in 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:
                        try:
                            processed_row[i] = format(decimal.Decimal(str(val)), "f")
                        except Exception as e:  # pylint: disable=broad-exception-caught
                            raise ValueError(
                                f"Failed to convert parameter at row {row}, column {i} to Decimal: {e}"
                            ) from e
            processed_parameters.append(processed_row)

        # Now transpose the processed parameters
        columnwise_params, row_count = self._transpose_rowwise_to_columnwise(processed_parameters)

        # Get encoding settings
        encoding_settings = self._get_encoding_settings()

        # Add debug logging
        logger.debug(
            "Executing batch query with %d parameter sets:\n%s",
            len(seq_of_parameters),
            "\n".join(
                f"  {i+1}: {tuple(p) if isinstance(p, (list, tuple)) else p}"
                for i, p in enumerate(seq_of_parameters[:5])
            ),  # Limit to first 5 rows for large batches
        )

        ret = ddbc_bindings.SQLExecuteMany(
            self.hstmt, operation, columnwise_params, parameters_type, row_count, encoding_settings
        )

        # Capture any diagnostic messages after execution
        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

            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_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_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

        # Use cached encoding strings — eliminates 2 method calls + 2 dict.get() per fetch
        char_enc = self._cached_char_encoding
        wchar_enc = self._cached_wchar_encoding

        # Fetch raw data
        row_data = []
        try:
            ret = ddbc_bindings.DDBCSQLFetchOne(
                self.hstmt,
                row_data,
                char_enc,
                wchar_enc,
            )

            # Only retrieve diag records on SQL_SUCCESS_WITH_INFO.
            if ret == ddbc_sql_const.SQL_SUCCESS_WITH_INFO.value and self.hstmt:
                self.messages.extend(ddbc_bindings.DDBCSQLGetAllDiagRecords(self.hstmt))

            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 = self._get_column_and_converter_maps()
            # Fast path: skip __init__ overhead when no converters/UUID processing
            if not converter_map and not self._uuid_str_indices:
                return Row._fast_create(row_data, column_map, self)
            return Row(
                row_data,
                column_map,
                cursor=self,
                converter_map=converter_map,
                uuid_str_indices=self._uuid_str_indices,
            )
        except Exception as e:
            # On error, don't increment rownumber - rethrow the error
            raise e

    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 []

        # Use cached encoding strings
        char_enc = self._cached_char_encoding
        wchar_enc = self._cached_wchar_encoding

        # Fetch raw data
        rows_data = []
        try:
            ret = ddbc_bindings.DDBCSQLFetchMany(
                self.hstmt,
                rows_data,
                size,
                char_enc,
                wchar_enc,
            )

            if ret == ddbc_sql_const.SQL_SUCCESS_WITH_INFO.value and self.hstmt:
                self.messages.extend(ddbc_bindings.DDBCSQLGetAllDiagRecords(self.hstmt))

            # 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 = self._get_column_and_converter_maps()

            # Convert raw data to Row objects
            uuid_idx = self._uuid_str_indices
            # Fast path: build Row objects in C++ — avoids Python loop overhead
            if not converter_map and not uuid_idx:
                return ddbc_bindings.construct_rows(rows_data, Row, column_map, self)
            return [
                Row(
                    row_data,
                    column_map,
                    cursor=self,
                    converter_map=converter_map,
                    uuid_str_indices=uuid_idx,
                )
                for row_data in rows_data
            ]
        except Exception as e:
            # On error, don't increment rownumber - rethrow the error
            raise e

    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()

        # Use cached encoding strings
        char_enc = self._cached_char_encoding
        wchar_enc = self._cached_wchar_encoding

        # Fetch raw data
        rows_data = []
        try:
            ret = ddbc_bindings.DDBCSQLFetchAll(
                self.hstmt,
                rows_data,
                char_enc,
                wchar_enc,
            )

            # Check for errors
            check_error(ddbc_sql_const.SQL_HANDLE_STMT.value, self.hstmt, ret)

            if ret == ddbc_sql_const.SQL_SUCCESS_WITH_INFO.value and self.hstmt:
                self.messages.extend(ddbc_bindings.DDBCSQLGetAllDiagRecords(self.hstmt))

            # 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 = self._get_column_and_converter_maps()

            # Convert raw data to Row objects
            uuid_idx = self._uuid_str_indices
            # Fast path: build Row objects in C++ — avoids Python loop overhead
            if not converter_map and not uuid_idx:
                return ddbc_bindings.construct_rows(rows_data, Row, column_map, self)
            return [
                Row(
                    row_data,
                    column_map,
                    cursor=self,
                    converter_map=converter_map,
                    uuid_str_indices=uuid_idx,
                )
                for row_data in rows_data
            ]
        except Exception as e:
            # On error, don't increment rownumber - rethrow the error
            raise e

    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 = []
        ret = ddbc_bindings.DDBCSQLFetchArrowBatch(self.hstmt, capsules, max(batch_size, 0))
        check_error(ddbc_sql_const.SQL_HANDLE_STMT.value, self.hstmt, ret)

        batch = pyarrow.RecordBatch._import_from_c_capsule(*capsules)

        if self.hstmt:
            self.messages.extend(ddbc_bindings.DDBCSQLGetAllDiagRecords(self.hstmt))

        # 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) -> "pyarrow.RecordBatchReader":
        """
        Fetch the result as a pyarrow RecordBatchReader, which yields Record
        Batches of the specified size until the current result set is
        exhausted.

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

        Returns:
            A pyarrow 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

        def batch_generator():
            while (batch := self.arrow_batch(batch_size)).num_rows > 0:
                yield batch

        return pyarrow.RecordBatchReader.from_batches(schema, batch_generator())

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

        Returns:
            True if there is another result set, None otherwise.

        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_converter_map = 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)

        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_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 bulkcopy(
        self,
        table_name: str,
        data: Iterable[Union[Tuple, List]],
        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 lists containing row data to be inserted.

                Data Format Requirements:
                - Each element in the iterable represents one row
                - Each row should be a tuple or list of column values
                - 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.

            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, or is a string/bytes
            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

        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

        # Validate inputs
        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")

        # 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__}"
            )

        # Validate batch_size type and value (0 means server optimal)
        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}")

        # Validate timeout type and value
        if not isinstance(timeout, int):
            raise TypeError(f"timeout must be a positive integer, got {type(timeout).__name__}")
        if timeout <= 0:
            raise ValueError(f"timeout must be positive, got {timeout}")

        # Get and parse connection string
        if not hasattr(self.connection, "connection_str"):
            logger.error("_bulkcopy: 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)

        # Token acquisition — only thing cursor must handle (needs azure-identity SDK)
        if self.connection._auth_type:
            # Fresh token acquisition for mssql-py-core connection
            from mssql_python.auth import AADAuth

            try:
                raw_token = AADAuth.get_raw_token(self.connection._auth_type)
            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,
            )

        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()

            # 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,
                iter(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 without exposing connection context in the error chain
            # to prevent credential leakage in stack traces
            raise type(e)(str(e)) from None

        finally:
            # Clear sensitive data to minimize memory exposure
            if pycore_context:
                for key in ("password", "user_name", "access_token"):
                    pycore_context.pop(key, None)
            # Clean up bulk copy resources
            for resource in (pycore_cursor, pycore_connection):
                if resource and hasattr(resource, "close"):
                    try:
                        resource.close()
                    except Exception as cleanup_error:
                        # Log cleanup errors only - aids troubleshooting without masking original exception
                        logger.debug(
                            "Failed to close bulk copy resource %s: %s",
                            type(resource).__name__,
                            cleanup_error,
                        )

    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().
        """
        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.)
            logger.debug("fetchval: No result set available (non-SELECT statement)")
            return None

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

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

        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.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:
            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