ddbc_bindings.cpp

// Copyright (c) Microsoft Corporation.
// Licensed under the MIT license.

// INFO|TODO - Note that is file is Windows specific right now. Making it arch
// agnostic will be
//             taken up in beta release
#include "ddbc_bindings.h"
#include "connection/connection.h"
#include "connection/connection_pool.h"
#include "logger_bridge.hpp"
#include "performance_counter.hpp"
#include "param_detect.hpp"
#include "py_ref.hpp"
#include "py_type_cache.hpp"
#include "fetch_temporal.hpp"
#include "row_factory.hpp"
#include "utf_utils.h"
#include "fetch_text.hpp"

#include <algorithm>  // std::min
#include <cctype>
#include <cstdint>
#include <cstring>  // For std::memcpy
#include <filesystem>
#include <iostream>
#include <utility>  // std::forward
#include <datetime.h>  // CPython datetime API (PyDateTime_IMPORT, PyDateTime_GET_*, etc.)


//-------------------------------------------------------------------------------------------------
// Macro definitions
//-------------------------------------------------------------------------------------------------

#ifdef _WIN32
// Constrained DLL search flags (Windows 8+ / Win7 + KB2533623). Defined
// defensively in case the build's SDK headers gate them behind an older
// _WIN32_WINNT than this project targets.
//
// Note: LOAD_LIBRARY_SEARCH_DEFAULT_DIRS is deliberately NOT used. It also
// includes LOAD_LIBRARY_SEARCH_USER_DIRS -- directories any in-process module
// registered via AddDllDirectory/SetDllDirectory -- which is outside the
// trusted set we want. We combine APPLICATION_DIR + SYSTEM32 + DLL_LOAD_DIR
// explicitly instead.
#ifndef LOAD_LIBRARY_SEARCH_DLL_LOAD_DIR
#define LOAD_LIBRARY_SEARCH_DLL_LOAD_DIR 0x00000100
#endif
#ifndef LOAD_LIBRARY_SEARCH_APPLICATION_DIR
#define LOAD_LIBRARY_SEARCH_APPLICATION_DIR 0x00000200
#endif
#ifndef LOAD_LIBRARY_SEARCH_SYSTEM32
#define LOAD_LIBRARY_SEARCH_SYSTEM32 0x00000800
#endif
#endif  // _WIN32

#ifndef SQL_C_DATE
#define SQL_C_DATE (9)
#endif
#ifndef SQL_C_TIME
#define SQL_C_TIME (10)
#endif
#ifndef SQL_C_TIMESTAMP
#define SQL_C_TIMESTAMP (11)
#endif
// SQL Server-specific variant TIME type code
#define SQL_SS_VARIANT_TIME (16384)
// Space for driver name + up to 8000 characters output by PRINT statements
#define SQL_MAX_MESSAGE_LENGTH_SQLSERVER (10000)

#define STRINGIFY_FOR_CASE(x)                                                                      \
    case x:                                                                                        \
        return #x

// Architecture-specific defines
#ifndef ARCHITECTURE
#define ARCHITECTURE "win64"  // Default to win64 if not defined during compilation
#endif
#define DAE_CHUNK_SIZE 8192
#define SQL_MAX_LOB_SIZE 8000

// Returns the effective character decoding encoding for SQL_C_CHAR data.
// On Linux/macOS, the ODBC driver always returns UTF-8 for SQL_C_CHAR,
// having already converted from the server's encoding (e.g., CP1252).
// On Windows, the driver returns bytes in the server's native encoding.
inline std::string GetEffectiveCharDecoding(const std::string& userEncoding) {
#if defined(__APPLE__) || defined(__linux__)
    (void)userEncoding;
    return "utf-8";
#else
    return userEncoding;
#endif
}

// Windows-only fix for issue #531: when the user explicitly requests
// SQL_C_CHAR + utf-8 decoding (e.g. setdecoding(SQL_CHAR, "utf-8", SQL_CHAR)),
// the SQL Server ODBC driver on Windows returns VARCHAR data in the server's
// ANSI code page (e.g. CP1252) regardless of the column's actual collation.
// For UTF-8 collation columns or any non-ASCII data, this is lossy ('?'
// substitution) and unrecoverable on the Python side. Internally upgrading
// the fetch to SQL_C_WCHAR triggers the driver's lossless UTF-16 conversion,
// which produces a correct Python Unicode string regardless of column
// collation. On Linux/macOS the SQL_C_CHAR path already returns UTF-8 from
// the driver, so this upgrade is a no-op there.
inline int EffectiveCharCtypeForFetch(int charCtype, const std::string& charEncoding) {
#ifdef _WIN32
    if (charCtype == SQL_C_CHAR && charEncoding == "utf-8") {
        // Surface the override so users can correlate observed SQL_C_WCHAR
        // fetches with their explicit setdecoding(SQL_CHAR, "utf-8", SQL_CHAR)
        // call (issue#531). Logged at INFO so it appears in production traces
        // without flooding default DEBUG output.
        LOG_INFO("EffectiveCharCtypeForFetch: Upgrading SQL_C_CHAR + utf-8 to "
                 "SQL_C_WCHAR on Windows to avoid lossy ACP conversion ");
        return SQL_C_WCHAR;
    }
#else
    (void)charEncoding;
#endif
    return charCtype;
}


//-------------------------------------------------------------------------------------------------
//-------------------------------------------------------------------------------------------------
// Logging Infrastructure:
// - LOG() macro: All diagnostic/debug logging at DEBUG level (single level)
// - LOG_INFO/WARNING/ERROR: Higher-level messages for production
// Uses printf-style formatting: LOG("Value: %d", x) -- __FILE__/__LINE__
// embedded in macro
//-------------------------------------------------------------------------------------------------

//-------------------------------------------------------------------------------------------------
// Class definitions
//-------------------------------------------------------------------------------------------------

// Struct to hold parameter information for binding. Used by SQLBindParameter.

struct ArrowArrayPrivateData {
    std::unique_ptr<uint8_t[]> valid;

    std::unique_ptr<uint8_t[]> uint8Val;
    std::unique_ptr<int16_t[]> int16Val;
    std::unique_ptr<int32_t[]> int32Val;
    std::unique_ptr<int64_t[]> int64Val;
    std::unique_ptr<double[]> float64Val;
    std::unique_ptr<float[]> float32Val;
    std::unique_ptr<uint8_t[]> bitVal;
    std::unique_ptr<uint64_t[]> varVal;
    std::unique_ptr<int32_t[]> dateVal;
    std::unique_ptr<int64_t[]> tsMicroVal;
    std::unique_ptr<int64_t[]> timeNanoVal;
    std::unique_ptr<Int128_t[]> decimalVal;

    std::vector<uint8_t> varData;

    // first buffer will be the valid bitmap
    // second buffer will be one of the value buffers above
    // third buffer will be the varData buffer for variable length types
    std::array<void*, 3> buffers;

    // Points to one of the typed *Val buffers above. Since the buffer pointers
    // don't change, this can be set once during batch initialization.
    void* ptrValueBuffer;
};

struct ArrowSchemaPrivateData {
    std::unique_ptr<char[]> name;
    std::unique_ptr<char[]> format;
};

#ifndef ARROW_C_DATA_INTERFACE
#define ARROW_C_DATA_INTERFACE

#define ARROW_FLAG_DICTIONARY_ORDERED 1
#define ARROW_FLAG_NULLABLE 2
#define ARROW_FLAG_MAP_KEYS_SORTED 4

struct ArrowSchema {
    // Array type description
    const char* format;
    const char* name;
    const char* metadata;
    int64_t flags;
    int64_t n_children;
    struct ArrowSchema** children;
    struct ArrowSchema* dictionary;

    // Release callback
    void (*release)(struct ArrowSchema*);
    // Opaque producer-specific data
    // Only our child-arrays will set this, so we can give it the correct type
    ArrowSchemaPrivateData* private_data;
};

struct ArrowArray {
    // Array data description
    int64_t length;
    int64_t null_count;
    int64_t offset;
    int64_t n_buffers;
    int64_t n_children;
    const void** buffers;
    struct ArrowArray** children;
    struct ArrowArray* dictionary;

    // Release callback
    void (*release)(struct ArrowArray*);
    // Opaque producer-specific data
    // Only our child-arrays will set this, so we can give it the correct type
    ArrowArrayPrivateData* private_data;
};

#endif  // ARROW_C_DATA_INTERFACE

//-------------------------------------------------------------------------------------------------
// Function pointer initialization
//-------------------------------------------------------------------------------------------------

// Handle APIs
SQLAllocHandleFunc SQLAllocHandle_ptr = nullptr;
SQLSetEnvAttrFunc SQLSetEnvAttr_ptr = nullptr;
SQLSetConnectAttrFunc SQLSetConnectAttr_ptr = nullptr;
SQLSetStmtAttrFunc SQLSetStmtAttr_ptr = nullptr;
SQLGetConnectAttrFunc SQLGetConnectAttr_ptr = nullptr;

// Connection and Execution APIs
SQLDriverConnectFunc SQLDriverConnect_ptr = nullptr;
SQLExecDirectFunc SQLExecDirect_ptr = nullptr;
SQLPrepareFunc SQLPrepare_ptr = nullptr;
SQLBindParameterFunc SQLBindParameter_ptr = nullptr;
SQLExecuteFunc SQLExecute_ptr = nullptr;
SQLRowCountFunc SQLRowCount_ptr = nullptr;
SQLGetStmtAttrFunc SQLGetStmtAttr_ptr = nullptr;
SQLSetDescFieldFunc SQLSetDescField_ptr = nullptr;

// Data retrieval APIs
SQLFetchFunc SQLFetch_ptr = nullptr;
SQLFetchScrollFunc SQLFetchScroll_ptr = nullptr;
SQLGetDataFunc SQLGetData_ptr = nullptr;
SQLNumResultColsFunc SQLNumResultCols_ptr = nullptr;
SQLBindColFunc SQLBindCol_ptr = nullptr;
SQLDescribeColFunc SQLDescribeCol_ptr = nullptr;
SQLMoreResultsFunc SQLMoreResults_ptr = nullptr;
SQLColAttributeFunc SQLColAttribute_ptr = nullptr;
SQLGetTypeInfoFunc SQLGetTypeInfo_ptr = nullptr;
SQLProceduresFunc SQLProcedures_ptr = nullptr;
SQLForeignKeysFunc SQLForeignKeys_ptr = nullptr;
SQLPrimaryKeysFunc SQLPrimaryKeys_ptr = nullptr;
SQLSpecialColumnsFunc SQLSpecialColumns_ptr = nullptr;
SQLStatisticsFunc SQLStatistics_ptr = nullptr;
SQLColumnsFunc SQLColumns_ptr = nullptr;
SQLGetInfoFunc SQLGetInfo_ptr = nullptr;

// Transaction APIs
SQLEndTranFunc SQLEndTran_ptr = nullptr;

// Disconnect/free APIs
SQLFreeHandleFunc SQLFreeHandle_ptr = nullptr;
SQLDisconnectFunc SQLDisconnect_ptr = nullptr;
SQLFreeStmtFunc SQLFreeStmt_ptr = nullptr;
SQLCancelFunc SQLCancel_ptr = nullptr;

// Diagnostic APIs
SQLGetDiagRecFunc SQLGetDiagRec_ptr = nullptr;

// DAE APIs
SQLParamDataFunc SQLParamData_ptr = nullptr;
SQLPutDataFunc SQLPutData_ptr = nullptr;
SQLTablesFunc SQLTables_ptr = nullptr;

SQLDescribeParamFunc SQLDescribeParam_ptr = nullptr;

namespace {


const char* GetSqlCTypeAsString(const SQLSMALLINT cType) {
    switch (cType) {
        STRINGIFY_FOR_CASE(SQL_C_CHAR);
        STRINGIFY_FOR_CASE(SQL_C_WCHAR);
        STRINGIFY_FOR_CASE(SQL_C_SSHORT);
        STRINGIFY_FOR_CASE(SQL_C_USHORT);
        STRINGIFY_FOR_CASE(SQL_C_SHORT);
        STRINGIFY_FOR_CASE(SQL_C_SLONG);
        STRINGIFY_FOR_CASE(SQL_C_ULONG);
        STRINGIFY_FOR_CASE(SQL_C_LONG);
        STRINGIFY_FOR_CASE(SQL_C_STINYINT);
        STRINGIFY_FOR_CASE(SQL_C_UTINYINT);
        STRINGIFY_FOR_CASE(SQL_C_TINYINT);
        STRINGIFY_FOR_CASE(SQL_C_SBIGINT);
        STRINGIFY_FOR_CASE(SQL_C_UBIGINT);
        STRINGIFY_FOR_CASE(SQL_C_FLOAT);
        STRINGIFY_FOR_CASE(SQL_C_DOUBLE);
        STRINGIFY_FOR_CASE(SQL_C_BIT);
        STRINGIFY_FOR_CASE(SQL_C_BINARY);
        STRINGIFY_FOR_CASE(SQL_C_TYPE_DATE);
        STRINGIFY_FOR_CASE(SQL_C_TYPE_TIME);
        STRINGIFY_FOR_CASE(SQL_C_TYPE_TIMESTAMP);
        STRINGIFY_FOR_CASE(SQL_C_NUMERIC);
        STRINGIFY_FOR_CASE(SQL_C_GUID);
        STRINGIFY_FOR_CASE(SQL_C_DEFAULT);
        default:
            return "Unknown";
    }
}

std::string MakeParamMismatchErrorStr(const SQLSMALLINT cType, const int paramIndex) {
    std::string errorString = "Parameter's object type does not match "
                              "parameter's C type. paramIndex - " +
                              std::to_string(paramIndex) + ", C type - " +
                              GetSqlCTypeAsString(cType);
    return errorString;
}

// This function allocates a buffer of ParamType, stores it as a void* in
// paramBuffers for book-keeping and then returns a ParamType* to the allocated
// memory. ctorArgs are the arguments to ParamType's constructor used while
// creating/allocating ParamType
template <typename ParamType, typename... CtorArgs>
ParamType* AllocateParamBuffer(std::vector<std::shared_ptr<void>>& paramBuffers,
                               CtorArgs&&... ctorArgs) {
    paramBuffers.emplace_back(new ParamType(std::forward<CtorArgs>(ctorArgs)...),
                              std::default_delete<ParamType>());
    return static_cast<ParamType*>(paramBuffers.back().get());
}

template <typename ParamType>
ParamType* AllocateParamBufferArray(std::vector<std::shared_ptr<void>>& paramBuffers,
                                    size_t count) {
    std::shared_ptr<ParamType> buffer(new ParamType[count], std::default_delete<ParamType[]>());
    ParamType* raw = buffer.get();
    paramBuffers.push_back(buffer);
    return raw;
}

std::string DescribeChar(unsigned char ch) {
    if (ch >= 32 && ch <= 126) {
        return std::string("'") + static_cast<char>(ch) + "'";
    } else {
        char buffer[16];
        snprintf(buffer, sizeof(buffer), "U+%04X", ch);
        return std::string(buffer);
    }
}




template<typename PutDataFn>
// The callable hides whether the caller wraps SQLPutData with GIL management; chunk sizing stays shared.
static SQLRETURN stream_dae_chunks(const void* data, size_t total_bytes, PutDataFn put_data_fn) {
    const char* bytes = static_cast<const char*>(data);
    for (size_t offset = 0; offset < total_bytes; offset += DAE_CHUNK_SIZE) {
        size_t len = std::min(static_cast<size_t>(DAE_CHUNK_SIZE), total_bytes - offset);
        SQLRETURN rc = put_data_fn(
            static_cast<SQLPOINTER>(const_cast<char*>(bytes + offset)), static_cast<SQLLEN>(len));
        if (!SQL_SUCCEEDED(rc)) return rc;
    }
    return SQL_SUCCESS;
}

// GH-610: Resolve SQL type for a NULL parameter using per-handle cache.
// On cache miss, calls SQLDescribeParam and stores the result.
static DescribedParamInfo ResolveNullParamType(SqlHandle& handle, SQLHANDLE hStmt, int paramIndex) {
    // Check per-handle cache. ODBC mandates one handle per thread, so no
    // mutex is needed. Violating this contract causes undefined behavior.
    auto it = handle.describeCache.find(paramIndex);
    if (it != handle.describeCache.end()) {
        LOG("ResolveNullParamType: Cache HIT for hStmt=%p param[%d] "
            "-> sqlType=%d",
            (void*)hStmt, paramIndex, it->second.sqlType);
        return it->second;
    }

    // Cache miss — call SQLDescribeParam
    SQLSMALLINT type, digits, nullable;
    SQLULEN size;
    LOG("ResolveNullParamType: Cache MISS for hStmt=%p param[%d], calling "
        "SQLDescribeParam", (void*)hStmt, paramIndex);
    // SQLDescribeParam may issue a server round-trip
    // (sp_describe_undeclared_parameters). Release the GIL around it so
    // in-process Python TCP forwarders can run (issue #565 family).
    RETCODE rc;
    {
        py::gil_scoped_release release;
        rc = SQLDescribeParam_ptr(
            hStmt, static_cast<SQLUSMALLINT>(paramIndex + 1),
            &type, &size, &digits, &nullable);
    }

    DescribedParamInfo info;
    if (SQL_SUCCEEDED(rc)) {
        info = {type, size, digits};
        LOG("ResolveNullParamType: SQLDescribeParam succeeded for param[%d] "
            "-> sqlType=%d, columnSize=%lu, decimalDigits=%d",
            paramIndex, type, (unsigned long)size, digits);
    } else {
        // SQLDescribeParam failed — typically happens with temp tables (#table),
        // table variables, or complex CTEs where the driver cannot determine
        // parameter metadata.  Fall back to SQL_VARCHAR which works for most
        // column types but will fail for BINARY/VARBINARY columns due to SQL
        // Server's implicit conversion rules.
        //
        // Workaround: cursor.setinputsizes() to explicitly specify types.
        //   from mssql_python.constants import ConstantsDDBC
        //   cursor.setinputsizes([(ConstantsDDBC.SQL_INTEGER.value, 10, 0),
        //                         (ConstantsDDBC.SQL_VARBINARY.value, 0, 0)])
        //   cursor.execute("INSERT INTO #t (id, data) VALUES (?, ?)", [1, None])
        info = {SQL_VARCHAR, 1, 0};
        LOG_WARNING("ResolveNullParamType: SQLDescribeParam failed for "
                    "param[%d] (rc=%d), falling back to SQL_VARCHAR",
                    paramIndex, rc);
    }

    // Cache both successful and fallback results. For fallbacks, this avoids
    // repeated SQLDescribeParam network calls on statement reuse. Note: on the
    // same_sql path clearDescribeCache() is NOT called, so a transient describe
    // failure is pinned as SQL_VARCHAR for the life of the prepared statement.
    // This is intentional — retrying a failing describe on every execute would
    // add latency with no benefit (temp-table metadata won't become resolvable
    // mid-connection). The cache IS cleared on SQLPrepare (usePrepare path).
    handle.describeCache[paramIndex] = info;

    return info;
}

// GH-627: Resolve unknown NULL SQL types before any SQLBindParameter calls.
// Some drivers remap parameter ordinals during describe when parameters have
// already been bound, so interleaving describe+bind can fail for binary NULLs.
// When `params` is provided (execute path), an additional py::none check is
// performed; for executemany (array path), SQL_C_DEFAULT already guarantees
// all values in that column are NULL, so no Python-level check is needed.
static void PreResolveUnknownNullTypes(SqlHandle& handle, SQLHANDLE hStmt,
                                       std::vector<ParamInfo>& paramInfos,
                                       const py::list* params = nullptr) {
    if (paramInfos.empty())
        return;

    for (size_t paramIndex = 0; paramIndex < paramInfos.size(); ++paramIndex) {
        ParamInfo& paramInfo = paramInfos[paramIndex];
        if (paramInfo.paramCType != SQL_C_DEFAULT || paramInfo.paramSQLType != SQL_UNKNOWN_TYPE) {
            continue;
        }
        // For execute(), verify the actual value is None (mixed columns possible).
        if (params && paramIndex < params->size() &&
            !py::isinstance<py::none>((*params)[paramIndex])) {
            continue;
        }

        auto resolved = ResolveNullParamType(handle, hStmt, static_cast<int>(paramIndex));
        paramInfo.paramSQLType = resolved.sqlType;
        paramInfo.columnSize = resolved.columnSize;
        paramInfo.decimalDigits = resolved.decimalDigits;
    }
}

// Given a list of parameters and their ParamInfo, calls SQLBindParameter on
// each of them with appropriate arguments
SQLRETURN BindParameters(SqlHandle& handle, SQLHANDLE hStmt, const py::list& params,
                         std::vector<ParamInfo>& paramInfos,
                         std::vector<std::shared_ptr<void>>& paramBuffers,
                         const std::string& charEncoding = "utf-8") {
    PERF_TIMER("BindParameters");
    LOG("BindParameters: Starting parameter binding for statement handle %p "
        "with %zu parameters",
        (void*)hStmt, params.size());

    // GH-627: resolve unknown NULL param SQL types before binding any param.
    PreResolveUnknownNullTypes(handle, hStmt, paramInfos, &params);
    for (int paramIndex = 0; paramIndex < params.size(); paramIndex++) {
        const auto& param = params[paramIndex];
        ParamInfo& paramInfo = paramInfos[paramIndex];
        LOG("BindParameters: Processing param[%d] - C_Type=%d, SQL_Type=%d, "
            "ColumnSize=%lu, DecimalDigits=%d, InputOutputType=%d",
            paramIndex, paramInfo.paramCType, paramInfo.paramSQLType,
            (unsigned long)paramInfo.columnSize, paramInfo.decimalDigits,
            paramInfo.inputOutputType);
        void* dataPtr = nullptr;
        SQLLEN bufferLength = 0;
        SQLLEN* strLenOrIndPtr = nullptr;

        // TODO: Add more data types like money, guid, interval, TVPs etc.
        switch (paramInfo.paramCType) {
            case SQL_C_CHAR: {
                if (!py::isinstance<py::str>(param) && !py::isinstance<py::bytearray>(param) &&
                    !py::isinstance<py::bytes>(param)) {
                    ThrowStdException(MakeParamMismatchErrorStr(paramInfo.paramCType, paramIndex));
                }
                if (paramInfo.isDAE) {
                    LOG("BindParameters: param[%d] SQL_C_CHAR - Using DAE "
                        "(Data-At-Execution) for large string streaming",
                        paramIndex);
                    dataPtr =
                        const_cast<void*>(reinterpret_cast<const void*>(&paramInfos[paramIndex]));
                    strLenOrIndPtr = AllocateParamBuffer<SQLLEN>(paramBuffers);
                    *strLenOrIndPtr = SQL_LEN_DATA_AT_EXEC(0);
                    bufferLength = 0;
                } else {
                    // Use Python's codec system to encode the string with specified encoding
                    std::string encodedStr;

                    if (py::isinstance<py::str>(param)) {
                        // Encode Unicode string using the specified encoding
                        try {
                            py::object encoded = param.attr("encode")(charEncoding, "strict");
                            encodedStr = encoded.cast<std::string>();
                            LOG("BindParameters: param[%d] SQL_C_CHAR - Encoded with '%s', "
                                "size=%zu bytes",
                                paramIndex, charEncoding.c_str(), encodedStr.size());
                        } catch (const py::error_already_set& e) {
                            LOG_ERROR("BindParameters: param[%d] SQL_C_CHAR - Failed to encode "
                                      "with '%s': %s",
                                      paramIndex, charEncoding.c_str(), e.what());
                            throw std::runtime_error(std::string("Failed to encode parameter ") +
                                                     std::to_string(paramIndex) +
                                                     " with encoding '" + charEncoding +
                                                     "': " + e.what());
                        }
                    } else {
                        // bytes/bytearray - use as-is (already encoded)
                        if (py::isinstance<py::bytes>(param)) {
                            encodedStr = param.cast<std::string>();
                        } else {
                            // bytearray
                            encodedStr = std::string(
                                reinterpret_cast<const char*>(PyByteArray_AsString(param.ptr())),
                                PyByteArray_Size(param.ptr()));
                        }
                        LOG("BindParameters: param[%d] SQL_C_CHAR - Using raw bytes, size=%zu",
                            paramIndex, encodedStr.size());
                    }

                    std::string* strParam =
                        AllocateParamBuffer<std::string>(paramBuffers, encodedStr);
                    dataPtr = const_cast<void*>(static_cast<const void*>(strParam->data()));
                    bufferLength = strParam->size();
                    strLenOrIndPtr = AllocateParamBuffer<SQLLEN>(paramBuffers);
                    // Use explicit byte length instead of SQL_NTS so embedded NUL chars
                    // aren't treated as string terminators (e.g., "hello\x00world").
                    *strLenOrIndPtr = static_cast<SQLLEN>(strParam->size());
                }
                break;
            }
            case SQL_C_BINARY: {
                if (!py::isinstance<py::str>(param) && !py::isinstance<py::bytearray>(param) &&
                    !py::isinstance<py::bytes>(param)) {
                    ThrowStdException(MakeParamMismatchErrorStr(paramInfo.paramCType, paramIndex));
                }
                if (paramInfo.isDAE) {
                    // Deferred execution for VARBINARY(MAX)
                    LOG("BindParameters: param[%d] SQL_C_BINARY - Using DAE "
                        "for VARBINARY(MAX) streaming",
                        paramIndex);
                    dataPtr =
                        const_cast<void*>(reinterpret_cast<const void*>(&paramInfos[paramIndex]));
                    strLenOrIndPtr = AllocateParamBuffer<SQLLEN>(paramBuffers);
                    *strLenOrIndPtr = SQL_LEN_DATA_AT_EXEC(0);
                    bufferLength = 0;
                } else {
                    // small binary
                    std::string binData;
                    if (py::isinstance<py::bytes>(param)) {
                        binData = param.cast<std::string>();
                    } else {
                        // bytearray
                        binData = std::string(
                            reinterpret_cast<const char*>(PyByteArray_AsString(param.ptr())),
                            PyByteArray_Size(param.ptr()));
                    }
                    std::string* binBuffer =
                        AllocateParamBuffer<std::string>(paramBuffers, binData);
                    dataPtr = const_cast<void*>(static_cast<const void*>(binBuffer->data()));
                    bufferLength = static_cast<SQLLEN>(binBuffer->size());
                    strLenOrIndPtr = AllocateParamBuffer<SQLLEN>(paramBuffers);
                    *strLenOrIndPtr = bufferLength;
                }
                break;
            }
            case SQL_C_WCHAR: {
                if (!py::isinstance<py::str>(param) && !py::isinstance<py::bytearray>(param) &&
                    !py::isinstance<py::bytes>(param)) {
                    ThrowStdException(MakeParamMismatchErrorStr(paramInfo.paramCType, paramIndex));
                }
                if (paramInfo.isDAE) {
                    // deferred execution
                    LOG("BindParameters: param[%d] SQL_C_WCHAR - Using DAE for "
                        "NVARCHAR(MAX) streaming",
                        paramIndex);
                    dataPtr =
                        const_cast<void*>(reinterpret_cast<const void*>(&paramInfos[paramIndex]));
                    strLenOrIndPtr = AllocateParamBuffer<SQLLEN>(paramBuffers);
                    *strLenOrIndPtr = SQL_LEN_DATA_AT_EXEC(0);
                    bufferLength = 0;
                } else {
                    // Normal small-string case
                    std::u16string* sqlwcharBuffer = AllocateParamBuffer<std::u16string>(
                        paramBuffers, param.cast<std::u16string>());
                    LOG("BindParameters: param[%d] SQL_C_WCHAR - String "
                        "length=%zu characters, buffer=%zu bytes",
                        paramIndex, sqlwcharBuffer->size(),
                        sqlwcharBuffer->size() * sizeof(SQLWCHAR));
                    dataPtr = sqlwcharBuffer->data();
                    bufferLength = sqlwcharBuffer->size() * sizeof(SQLWCHAR);
                    strLenOrIndPtr = AllocateParamBuffer<SQLLEN>(paramBuffers);
                    // Use explicit byte length instead of SQL_NTS so embedded NUL chars
                    // aren't treated as string terminators.
                    *strLenOrIndPtr = static_cast<SQLLEN>(sqlwcharBuffer->size() * sizeof(SQLWCHAR));
                }
                break;
            }
            case SQL_C_BIT: {
                if (!py::isinstance<py::bool_>(param)) {
                    ThrowStdException(MakeParamMismatchErrorStr(paramInfo.paramCType, paramIndex));
                }
                dataPtr =
                    static_cast<void*>(AllocateParamBuffer<bool>(paramBuffers, param.cast<bool>()));
                break;
            }
            case SQL_C_DEFAULT: {
                if (!py::isinstance<py::none>(param)) {
                    ThrowStdException(MakeParamMismatchErrorStr(paramInfo.paramCType, paramIndex));
                }
                dataPtr = nullptr;  // GH-627: type resolved by PreResolveUnknownNullTypes.
                strLenOrIndPtr = AllocateParamBuffer<SQLLEN>(paramBuffers);
                *strLenOrIndPtr = SQL_NULL_DATA;
                bufferLength = 0;
                break;
            }
            case SQL_C_STINYINT:
            case SQL_C_TINYINT:
            case SQL_C_SSHORT:
            case SQL_C_SHORT: {
                if (!py::isinstance<py::int_>(param)) {
                    ThrowStdException(MakeParamMismatchErrorStr(paramInfo.paramCType, paramIndex));
                }
                int value = param.cast<int>();
                // Range validation for signed 16-bit integer
                if (value < std::numeric_limits<short>::min() ||
                    value > std::numeric_limits<short>::max()) {
                    ThrowStdException("Signed short integer parameter out of "
                                      "range at paramIndex " +
                                      std::to_string(paramIndex));
                }
                dataPtr =
                    static_cast<void*>(AllocateParamBuffer<int>(paramBuffers, param.cast<int>()));
                break;
            }
            case SQL_C_UTINYINT:
            case SQL_C_USHORT: {
                if (!py::isinstance<py::int_>(param)) {
                    ThrowStdException(MakeParamMismatchErrorStr(paramInfo.paramCType, paramIndex));
                }
                unsigned int value = param.cast<unsigned int>();
                if (value > std::numeric_limits<unsigned short>::max()) {
                    ThrowStdException("Unsigned short integer parameter out of "
                                      "range at paramIndex " +
                                      std::to_string(paramIndex));
                }
                dataPtr = static_cast<void*>(
                    AllocateParamBuffer<unsigned int>(paramBuffers, param.cast<unsigned int>()));
                break;
            }
            case SQL_C_SBIGINT:
            case SQL_C_SLONG:
            case SQL_C_LONG: {
                if (!py::isinstance<py::int_>(param)) {
                    ThrowStdException(MakeParamMismatchErrorStr(paramInfo.paramCType, paramIndex));
                }
                // Both detection paths (DetectParamTypes / _map_sql_type) reject out-of-int64
                // ints before binding, so those callers only reach here with bindable values.
                // A setinputsizes() override that forces SQL_C_SBIGINT on an out-of-range int
                // skips detection; that value fails the cast below, same as before this change.
                dataPtr = static_cast<void*>(
                    AllocateParamBuffer<int64_t>(paramBuffers, param.cast<int64_t>()));
                break;
            }
            case SQL_C_UBIGINT:
            case SQL_C_ULONG: {
                if (!py::isinstance<py::int_>(param)) {
                    ThrowStdException(MakeParamMismatchErrorStr(paramInfo.paramCType, paramIndex));
                }
                dataPtr = static_cast<void*>(
                    AllocateParamBuffer<uint64_t>(paramBuffers, param.cast<uint64_t>()));
                break;
            }
            case SQL_C_FLOAT: {
                if (!py::isinstance<py::float_>(param)) {
                    ThrowStdException(MakeParamMismatchErrorStr(paramInfo.paramCType, paramIndex));
                }
                dataPtr = static_cast<void*>(
                    AllocateParamBuffer<float>(paramBuffers, param.cast<float>()));
                break;
            }
            case SQL_C_DOUBLE: {
                if (!py::isinstance<py::float_>(param)) {
                    ThrowStdException(MakeParamMismatchErrorStr(paramInfo.paramCType, paramIndex));
                }
                dataPtr = static_cast<void*>(
                    AllocateParamBuffer<double>(paramBuffers, param.cast<double>()));
                break;
            }
            case SQL_C_TYPE_DATE: {
                py::object dateType = PyTypeCache::get_date_class_obj();
                if (!py::isinstance(param, dateType)) {
                    ThrowStdException(MakeParamMismatchErrorStr(paramInfo.paramCType, paramIndex));
                }
                int year = param.attr("year").cast<int>();
                if (year < 1753 || year > 9999) {
                    ThrowStdException("Date out of range for SQL Server "
                                      "(1753-9999) at paramIndex " +
                                      std::to_string(paramIndex));
                }
                // TODO: can be moved to python by registering SQL_DATE_STRUCT
                // in pybind
                SQL_DATE_STRUCT* sqlDatePtr = AllocateParamBuffer<SQL_DATE_STRUCT>(paramBuffers);
                sqlDatePtr->year = static_cast<SQLSMALLINT>(param.attr("year").cast<int>());
                sqlDatePtr->month = static_cast<SQLUSMALLINT>(param.attr("month").cast<int>());
                sqlDatePtr->day = static_cast<SQLUSMALLINT>(param.attr("day").cast<int>());
                dataPtr = static_cast<void*>(sqlDatePtr);
                break;
            }
            case SQL_C_TYPE_TIME: {
                py::object timeType = PyTypeCache::get_time_class_obj();
                if (!py::isinstance(param, timeType)) {
                    ThrowStdException(MakeParamMismatchErrorStr(paramInfo.paramCType, paramIndex));
                }
                // TODO: can be moved to python by registering SQL_TIME_STRUCT
                // in pybind
                SQL_TIME_STRUCT* sqlTimePtr = AllocateParamBuffer<SQL_TIME_STRUCT>(paramBuffers);
                sqlTimePtr->hour = static_cast<SQLUSMALLINT>(param.attr("hour").cast<int>());
                sqlTimePtr->minute = static_cast<SQLUSMALLINT>(param.attr("minute").cast<int>());
                sqlTimePtr->second = static_cast<SQLUSMALLINT>(param.attr("second").cast<int>());
                dataPtr = static_cast<void*>(sqlTimePtr);
                break;
            }
            case SQL_C_SS_TIMESTAMPOFFSET: {
                py::object datetimeType = PyTypeCache::get_datetime_class_obj();
                if (!py::isinstance(param, datetimeType)) {
                    ThrowStdException(MakeParamMismatchErrorStr(paramInfo.paramCType, paramIndex));
                }
                // Checking if the object has a timezone
                py::object tzinfo = param.attr("tzinfo");
                if (tzinfo.is_none()) {
                    ThrowStdException("Datetime object must have tzinfo for "
                                      "SQL_C_SS_TIMESTAMPOFFSET at paramIndex " +
                                      std::to_string(paramIndex));
                }

                DateTimeOffset* dtoPtr = AllocateParamBuffer<DateTimeOffset>(paramBuffers);

                dtoPtr->year = static_cast<SQLSMALLINT>(param.attr("year").cast<int>());
                dtoPtr->month = static_cast<SQLUSMALLINT>(param.attr("month").cast<int>());
                dtoPtr->day = static_cast<SQLUSMALLINT>(param.attr("day").cast<int>());
                dtoPtr->hour = static_cast<SQLUSMALLINT>(param.attr("hour").cast<int>());
                dtoPtr->minute = static_cast<SQLUSMALLINT>(param.attr("minute").cast<int>());
                dtoPtr->second = static_cast<SQLUSMALLINT>(param.attr("second").cast<int>());
                // SQL server supports in ns, but python datetime supports in µs
                dtoPtr->fraction =
                    static_cast<SQLUINTEGER>(param.attr("microsecond").cast<int>() * 1000);

                py::object utcoffset = tzinfo.attr("utcoffset")(param);
                if (utcoffset.is_none()) {
                    ThrowStdException("Datetime object's tzinfo.utcoffset() "
                                      "returned None at paramIndex " +
                                      std::to_string(paramIndex));
                }

                int total_seconds =
                    static_cast<int>(utcoffset.attr("total_seconds")().cast<double>());
                const int MAX_OFFSET = 14 * 3600;
                const int MIN_OFFSET = -14 * 3600;

                if (total_seconds > MAX_OFFSET || total_seconds < MIN_OFFSET) {
                    ThrowStdException("Datetimeoffset tz offset out of SQL Server range "
                                      "(-14h to +14h) at paramIndex " +
                                      std::to_string(paramIndex));
                }
                std::div_t div_result = std::div(total_seconds, 3600);
                dtoPtr->timezone_hour = static_cast<SQLSMALLINT>(div_result.quot);
                dtoPtr->timezone_minute = static_cast<SQLSMALLINT>(div(div_result.rem, 60).quot);

                dataPtr = static_cast<void*>(dtoPtr);
                bufferLength = sizeof(DateTimeOffset);
                strLenOrIndPtr = AllocateParamBuffer<SQLLEN>(paramBuffers);
                *strLenOrIndPtr = bufferLength;
                break;
            }
            case SQL_C_TYPE_TIMESTAMP: {
                py::object datetimeType = PyTypeCache::get_datetime_class_obj();
                if (!py::isinstance(param, datetimeType)) {
                    ThrowStdException(MakeParamMismatchErrorStr(paramInfo.paramCType, paramIndex));
                }
                SQL_TIMESTAMP_STRUCT* sqlTimestampPtr =
                    AllocateParamBuffer<SQL_TIMESTAMP_STRUCT>(paramBuffers);
                sqlTimestampPtr->year = static_cast<SQLSMALLINT>(param.attr("year").cast<int>());
                sqlTimestampPtr->month = static_cast<SQLUSMALLINT>(param.attr("month").cast<int>());
                sqlTimestampPtr->day = static_cast<SQLUSMALLINT>(param.attr("day").cast<int>());
                sqlTimestampPtr->hour = static_cast<SQLUSMALLINT>(param.attr("hour").cast<int>());
                sqlTimestampPtr->minute =
                    static_cast<SQLUSMALLINT>(param.attr("minute").cast<int>());
                sqlTimestampPtr->second =
                    static_cast<SQLUSMALLINT>(param.attr("second").cast<int>());
                // SQL server supports in ns, but python datetime supports in µs
                sqlTimestampPtr->fraction = static_cast<SQLUINTEGER>(
                    param.attr("microsecond").cast<int>() * 1000);  // Convert µs to ns
                dataPtr = static_cast<void*>(sqlTimestampPtr);
                break;
            }
            case SQL_C_NUMERIC: {
                if (!py::isinstance<NumericData>(param)) {
                    ThrowStdException(MakeParamMismatchErrorStr(paramInfo.paramCType, paramIndex));
                }
                NumericData decimalParam = param.cast<NumericData>();
                LOG("BindParameters: param[%d] SQL_C_NUMERIC - precision=%d, "
                    "scale=%d, sign=%d, value_bytes=%zu",
                    paramIndex, decimalParam.precision, decimalParam.scale, decimalParam.sign,
                    decimalParam.val.size());
                SQL_NUMERIC_STRUCT* decimalPtr =
                    AllocateParamBuffer<SQL_NUMERIC_STRUCT>(paramBuffers);
                decimalPtr->precision = decimalParam.precision;
                decimalPtr->scale = decimalParam.scale;
                decimalPtr->sign = decimalParam.sign;
                // Convert the integer decimalParam.val to char array
                std::memset(static_cast<void*>(decimalPtr->val), 0, sizeof(decimalPtr->val));
                size_t copyLen = std::min(decimalParam.val.size(), sizeof(decimalPtr->val));
                if (copyLen > 0) {
                    std::memcpy(decimalPtr->val, decimalParam.val.data(), copyLen);
                }
                dataPtr = static_cast<void*>(decimalPtr);
                break;
            }
            case SQL_C_GUID: {
                if (!py::isinstance<py::bytes>(param)) {
                    ThrowStdException(MakeParamMismatchErrorStr(paramInfo.paramCType, paramIndex));
                }
                py::bytes uuid_bytes = param.cast<py::bytes>();
                const unsigned char* uuid_data =
                    reinterpret_cast<const unsigned char*>(PyBytes_AS_STRING(uuid_bytes.ptr()));
                if (PyBytes_GET_SIZE(uuid_bytes.ptr()) != 16) {
                    LOG("BindParameters: param[%d] SQL_C_GUID - Invalid UUID "
                        "length: expected 16 bytes, got %ld bytes",
                        paramIndex, PyBytes_GET_SIZE(uuid_bytes.ptr()));
                    ThrowStdException("UUID binary data must be exactly 16 bytes long.");
                }
                SQLGUID* guid_data_ptr = AllocateParamBuffer<SQLGUID>(paramBuffers);
                guid_data_ptr->Data1 = (static_cast<uint32_t>(uuid_data[3]) << 24) |
                                       (static_cast<uint32_t>(uuid_data[2]) << 16) |
                                       (static_cast<uint32_t>(uuid_data[1]) << 8) |
                                       (static_cast<uint32_t>(uuid_data[0]));
                guid_data_ptr->Data2 = (static_cast<uint16_t>(uuid_data[5]) << 8) |
                                       (static_cast<uint16_t>(uuid_data[4]));
                guid_data_ptr->Data3 = (static_cast<uint16_t>(uuid_data[7]) << 8) |
                                       (static_cast<uint16_t>(uuid_data[6]));
                std::memcpy(guid_data_ptr->Data4, &uuid_data[8], 8);
                dataPtr = static_cast<void*>(guid_data_ptr);
                bufferLength = sizeof(SQLGUID);
                strLenOrIndPtr = AllocateParamBuffer<SQLLEN>(paramBuffers);
                *strLenOrIndPtr = sizeof(SQLGUID);
                break;
            }
            default: {
                std::ostringstream errorString;
                errorString << "Unsupported parameter type - " << paramInfo.paramCType
                            << " for parameter - " << paramIndex;
                ThrowStdException(errorString.str());
            }
        }
        assert(SQLBindParameter_ptr && SQLGetStmtAttr_ptr && SQLSetDescField_ptr);
        RETCODE rc;
        {
            PERF_TIMER("BindParameters::SQLBindParameter_call");
            rc = SQLBindParameter_ptr(
                hStmt, static_cast<SQLUSMALLINT>(paramIndex + 1), /* 1-based indexing */
                static_cast<SQLUSMALLINT>(paramInfo.inputOutputType),
                static_cast<SQLSMALLINT>(paramInfo.paramCType),
                static_cast<SQLSMALLINT>(paramInfo.paramSQLType), paramInfo.columnSize,
                paramInfo.decimalDigits, dataPtr, bufferLength, strLenOrIndPtr);
        }
        if (!SQL_SUCCEEDED(rc)) {
            LOG("BindParameters: SQLBindParameter failed for param[%d] - "
                "SQLRETURN=%d, C_Type=%d, SQL_Type=%d",
                paramIndex, rc, paramInfo.paramCType, paramInfo.paramSQLType);
            return rc;
        }
        // Special handling for Numeric type -
        // https://learn.microsoft.com/en-us/sql/odbc/reference/appendixes/retrieve-numeric-data-sql-numeric-struct-kb222831?view=sql-server-ver16#sql_c_numeric-overview
        if (paramInfo.paramCType == SQL_C_NUMERIC) {
            // The APD record number is the 1-based parameter position, matching the
            // SQLBindParameter call above. It was previously hardcoded to 1, so a
            // SQL_C_NUMERIC parameter in any position other than the first had its
            // precision/scale/data pointer written onto record 1 instead of its own.
            // The driver then read the numeric struct with the wrong descriptor and
            // raised "Numeric value out of range" (GH-740).
            const SQLSMALLINT descRecNum = static_cast<SQLSMALLINT>(paramIndex + 1);
            SQLHDESC hDesc = nullptr;
            rc = SQLGetStmtAttr_ptr(hStmt, SQL_ATTR_APP_PARAM_DESC, &hDesc, 0, NULL);
            if (!SQL_SUCCEEDED(rc)) {
                LOG("BindParameters: SQLGetStmtAttr(SQL_ATTR_APP_PARAM_DESC) "
                    "failed for param[%d] - SQLRETURN=%d",
                    paramIndex, rc);
                return rc;
            }
            rc = SQLSetDescField_ptr(hDesc, descRecNum, SQL_DESC_TYPE,
                                     (SQLPOINTER)SQL_C_NUMERIC, 0);
            if (!SQL_SUCCEEDED(rc)) {
                LOG("BindParameters: SQLSetDescField(SQL_DESC_TYPE) failed for "
                    "param[%d] - SQLRETURN=%d",
                    paramIndex, rc);
                return rc;
            }
            SQL_NUMERIC_STRUCT* numericPtr = reinterpret_cast<SQL_NUMERIC_STRUCT*>(dataPtr);
            rc = SQLSetDescField_ptr(
                hDesc, descRecNum, SQL_DESC_PRECISION,
                reinterpret_cast<SQLPOINTER>(static_cast<uintptr_t>(numericPtr->precision)), 0);
            if (!SQL_SUCCEEDED(rc)) {
                LOG("BindParameters: SQLSetDescField(SQL_DESC_PRECISION) "
                    "failed for param[%d] - SQLRETURN=%d",
                    paramIndex, rc);
                return rc;
            }

            rc = SQLSetDescField_ptr(
                hDesc, descRecNum, SQL_DESC_SCALE,
                reinterpret_cast<SQLPOINTER>(static_cast<intptr_t>(numericPtr->scale)), 0);
            if (!SQL_SUCCEEDED(rc)) {
                LOG("BindParameters: SQLSetDescField(SQL_DESC_SCALE) failed "
                    "for param[%d] - SQLRETURN=%d",
                    paramIndex, rc);
                return rc;
            }

            rc = SQLSetDescField_ptr(hDesc, descRecNum, SQL_DESC_DATA_PTR,
                                     reinterpret_cast<SQLPOINTER>(numericPtr), 0);
            if (!SQL_SUCCEEDED(rc)) {
                LOG("BindParameters: SQLSetDescField(SQL_DESC_DATA_PTR) failed "
                    "for param[%d] - SQLRETURN=%d",
                    paramIndex, rc);
                return rc;
            }
        }
    }
    LOG("BindParameters: Completed parameter binding for statement handle %p - "
        "%zu parameters bound successfully",
        (void*)hStmt, params.size());
    return SQL_SUCCESS;
}

// This is temporary hack to avoid crash when SQLDescribeCol returns 0 as
// columnSize for NVARCHAR(MAX) & similar types. Variable length data needs more
// nuanced handling.
// TODO: Fix this in beta
// This function sets the buffer allocated to fetch NVARCHAR(MAX) & similar
// types to 4096 chars. So we'll retrieve data upto 4096. Anything greater then
// that will throw error
void HandleZeroColumnSizeAtFetch(SQLULEN& columnSize) {
    if (columnSize == 0) {
        columnSize = 4096;
    }
}

}  // namespace

// Helper function to check if Python is shutting down or finalizing
// This centralizes the shutdown detection logic to avoid code duplication
//
// IMPORTANT: must not blindly acquire the GIL. The previous implementation
// used py::gil_scoped_acquire + sys._is_finalizing(), which calls
// PyGILState_Ensure() under the hood. When invoked from a thread CPython
// doesn't already know about (e.g. a foreign/background thread dropping the
// last shared_ptr<SqlHandle> reference) while the interpreter is finalizing,
// PyGILState_Ensure() can fail to register thread-local state and crash with
// "Fatal Python error: gilstate_tss_set: failed to set current tstate (TSS)"
// - i.e. the very safety check meant to prevent a shutdown-time crash can
// itself cause one.
static bool is_python_finalizing() {
    if (Py_IsInitialized() == 0) {
        return true;  // Python is already shut down
    }
#if PY_VERSION_HEX >= 0x030D0000
    // Py_IsFinalizing() is a public, thread-safe, GIL-free CPython API
    // (stable since Python 3.13) built exactly for this purpose.
    return Py_IsFinalizing() != 0;
#else
    // Older Python versions don't expose the public Py_IsFinalizing(), but the
    // exported CPython 3.7+ call it wraps, _Py_IsFinalizing(), is equally
    // GIL-free and thread-safe (pybind11 itself uses it for this purpose). Use
    // it so the check is accurate: a foreign/GIL-free thread dropping the last
    // handle reference during NORMAL operation reports "not finalizing" and the
    // handle is actually freed, instead of a PyGILState_Check() proxy that would
    // treat every GIL-free caller as shutdown and silently leak the handle.
    return _Py_IsFinalizing() != 0;
#endif
}

// TODO: Add more nuanced exception classes
void ThrowStdException(const std::string& message) {
    throw std::runtime_error(message);
}
std::string GetLastErrorMessage();

// Resolve the base directory that contains the ODBC driver `libs/` tree.
//
// Post-split, the driver binaries ship in the standalone `mssql_python_odbc`
// package (a pure-data sibling with no native extension). We import it and use
// its directory as the base that `GetDriverPathCpp` (and the Windows
// `mssql-auth.dll` lookup) append `libs` to.
//
// Post-split the standalone package is REQUIRED: if it is missing or does not
// ship this platform's driver binaries we raise a clear, actionable error
// instead of silently falling back to bundled libs (there are none). Importing
// `mssql_python_odbc` here is Alpine/musl-safe precisely because it is a
// separate pure package: it cannot trigger the partially-initialized-module
// circular import that motivated resolving these paths in C++ in the first place.
//
// (`GetDriverPathCpp` is defined further below; forward-declared here so we can
// verify the external package actually ships this platform's driver binary.)
std::string GetDriverPathCpp(const std::string& moduleDir);
std::string GetDriverPathForProviderCpp(const std::string& moduleDir,
                                        const std::string& providerId);

// -----------------------------------------------------------------------------
// ODBC provider selection
//
// Two providers are supported: the classic Microsoft ODBC Driver 18
// ("msodbcsql18", shipped by mssql_python_odbc) and the Rust driver
// ("mssql-odbc", shipped inside mssql_py_core / the mssql-python-rs wheel).
// Python is the sole
// resolver (env var -> module property -> default) and pushes the chosen id
// here via _set_odbc_provider() before the driver loads. The native side does
// not read the environment itself; if the push has not happened yet, it falls
// back to the hardcoded classic default.
// -----------------------------------------------------------------------------

namespace {
constexpr const char* kProviderMsodbcsql18 = "msodbcsql18";
constexpr const char* kProviderMssqlOdbc = "mssql-odbc";

std::mutex g_providerMutex;
std::string g_selectedProvider;  // pushed from Python before load; "" = unset
bool g_providerSelectionLocked = false;

std::string NormalizeProviderId(const std::string& id) {
    // Mirror Python's ProviderManager._normalize(): trim surrounding whitespace
    // and lowercase, leaving interior characters intact so both sides agree.
    size_t start = 0;
    size_t end = id.size();
    while (start < end && std::isspace(static_cast<unsigned char>(id[start]))) {
        ++start;
    }
    while (end > start && std::isspace(static_cast<unsigned char>(id[end - 1]))) {
        --end;
    }
    std::string out = id.substr(start, end - start);
    for (char& c : out) {
        c = static_cast<char>(std::tolower(static_cast<unsigned char>(c)));
    }
    return out;
}
}  // namespace

void SetSelectedProvider(const std::string& id) {
    const std::string normalized = NormalizeProviderId(id);
    if (normalized != kProviderMsodbcsql18 && normalized != kProviderMssqlOdbc) {
        // ProviderManager (Python) already rejects an unknown id before this is
        // ever called; this is defence-in-depth against a caller that bypasses
        // it and talks to this native entry point directly.
        ThrowStdException("Unknown ODBC provider '" + id +
                          "'. Valid providers are: " + kProviderMsodbcsql18 + ", " +
                          kProviderMssqlOdbc + ".");
    }
    std::lock_guard<std::mutex> lock(g_providerMutex);
    if (g_providerSelectionLocked) {
        // Every connection re-pushes ProviderManager's already-frozen choice,
        // so the same id remains a silent no-op. A direct native connection can
        // bypass the Python push; in that case an empty selection means the
        // classic default was locked when loading started.
        const std::string& lockedProvider =
            g_selectedProvider.empty() ? kProviderMsodbcsql18 : g_selectedProvider;
        if (normalized != lockedProvider) {
            ThrowStdException("Cannot change the ODBC provider after driver loading has "
                              "started in this process.");
        }
        return;
    }
    g_selectedProvider = normalized;
}

// Effective provider id: the value pushed from Python, else the classic default.
// Python is the authoritative resolver (env var -> module property -> default)
// and pushes the result via _set_odbc_provider() before the driver loads.
std::string GetSelectedProviderId() {
    std::lock_guard<std::mutex> lock(g_providerMutex);
    if (g_selectedProvider == kProviderMssqlOdbc) {
        return kProviderMssqlOdbc;
    }
    return kProviderMsodbcsql18;
}

// Keep in sync with odbc_provider.py's _PACKAGE_BY_PROVIDER / _DIST_BY_PROVIDER.
std::string ProviderPackageForId(const std::string& id) {
    return (id == kProviderMssqlOdbc) ? "mssql_py_core" : "mssql_python_odbc";
}

std::string ProviderDistForId(const std::string& id) {
    return (id == kProviderMssqlOdbc) ? "mssql-python-rs" : "mssql-python-odbc";
}

std::string GetOdbcLibsBaseDir() {
    namespace fs = std::filesystem;
    // This function calls into the Python C-API (py::module::import, attribute
    // access, casts), so it must run with the GIL held. It is a no-op when the
    // GIL is already held — which it is at the sole current call site, during
    // module initialization — but acquiring it here self-documents the C-API
    // dependency and keeps a future GIL-released caller from turning this into a
    // hard crash.
    py::gil_scoped_acquire gil;
    const std::string providerId = GetSelectedProviderId();
    const std::string packageName = ProviderPackageForId(providerId);
    const std::string distName = ProviderDistForId(providerId);
    try {
        py::object module = py::module::import(packageName.c_str());
        py::object module_path = module.attr("__file__");
        std::string module_file = module_path.cast<std::string>();

        fs::path parentDir = fs::path(module_file).parent_path();

        // The external package is authoritative and REQUIRED: it must ship a
        // COMPLETE set of this platform's driver binaries. In a source/dev
        // checkout (and in CI) the package is importable from the repo root but
        // its `libs/` tree is gitignored and either absent or only partially
        // populated; there is no bundled fallback anymore, so fail hard rather
        // than resolve to a directory that has no usable driver.
        //
        // "Complete" means the ODBC driver itself and, for the classic
        // msodbcsql18 provider on Windows, the co-located `mssql-auth.dll` that
        // LoadDriverOrThrowException preloads. The Rust provider (mssql-odbc)
        // loads that library lazily from System32 at interactive-auth time, so
        // it is NOT required co-located here. Verifying this keeps the resolver's
        // notion of a usable base dir consistent with what the loader needs.
        std::error_code ec;
        fs::path externalDriver(GetDriverPathCpp(parentDir.string()));
        bool externalComplete = fs::exists(externalDriver, ec);
#ifdef _WIN32
        // Only the classic msodbcsql18 provider ships and preloads a co-located
        // mssql-auth.dll. Requiring it for the Rust provider would wrongly reject
        // an otherwise complete mssql-odbc package, which resolves that library
        // lazily from System32 instead.
        if (externalComplete && providerId == kProviderMsodbcsql18) {
            fs::path externalAuthDll = externalDriver.parent_path() / "mssql-auth.dll";
            externalComplete = fs::exists(externalAuthDll, ec);
        }
#endif
        if (!externalComplete) {
            LOG("GetOdbcLibsBaseDir: %s present at '%s' but its ODBC driver "
                "binaries are missing or incomplete for this platform",
                packageName.c_str(), parentDir.string().c_str());
            ThrowStdException(
                "The '" + distName + "' package is installed but its ODBC driver binaries "
                "are missing or incomplete for this platform. Reinstall it with: "
                "pip install --force-reinstall " + distName);
        }
        LOG("GetOdbcLibsBaseDir: Using external %s package - directory='%s'",
            packageName.c_str(), parentDir.string().c_str());
        return parentDir.string();
    } catch (const py::error_already_set& e) {
        if (e.matches(PyExc_ModuleNotFoundError)) {
            // Phase 2: the standalone package is required. Turn the missing
            // dependency into a clear, actionable error instead of a fallback.
            LOG("GetOdbcLibsBaseDir: required package %s is not installed (%s)",
                packageName.c_str(), e.what());
            ThrowStdException(
                "The required '" + distName + "' package (which ships the ODBC driver "
                "binaries) is not installed. Install it with: pip install " + distName);
        }
        // A different import-time error means the package is installed but
        // broken; surface it instead of silently masking the real problem.
        LOG("GetOdbcLibsBaseDir: importing %s failed unexpectedly (%s); "
            "re-raising",
            packageName.c_str(), e.what());
        throw;
    }
}

// Platform-agnostic function to load the driver dynamic library
DriverHandle LoadDriverLibrary(const std::string& driverPath) {
    LOG("LoadDriverLibrary: Attempting to load ODBC driver from path='%s'", driverPath.c_str());

#ifdef _WIN32
    // Windows: Use std::filesystem::path for proper UTF-8 to UTF-16 conversion
    // fs::path::c_str() returns wchar_t* on Windows with correct encoding
    namespace fs = std::filesystem;
    fs::path pathObj(driverPath);
    // Resolve the vendored driver's dependencies with a constrained search
    // path. LOAD_LIBRARY_SEARCH_DLL_LOAD_DIR adds the driver's own folder for
    // its dependency lookups; APPLICATION_DIR and SYSTEM32 add the host
    // application directory and System32. This excludes the current working
    // directory, %PATH%, and process-wide user DLL directories (registered via
    // AddDllDirectory/SetDllDirectory) -- all of which the legacy LoadLibraryW
    // order, or the broader LOAD_LIBRARY_SEARCH_DEFAULT_DIRS, would include.
    HMODULE handle = LoadLibraryExW(
        pathObj.c_str(), nullptr,
        LOAD_LIBRARY_SEARCH_APPLICATION_DIR | LOAD_LIBRARY_SEARCH_SYSTEM32 |
            LOAD_LIBRARY_SEARCH_DLL_LOAD_DIR);
    if (!handle) {
        LOG("LoadDriverLibrary: LoadLibraryExW failed for path='%s' - %s", driverPath.c_str(),
            GetLastErrorMessage().c_str());
        ThrowStdException("Failed to load library: " + driverPath);
    }
    return handle;
#else
    // macOS/Unix: Use dlopen
    void* handle = dlopen(driverPath.c_str(), RTLD_LAZY);
    if (!handle) {
        LOG("LoadDriverLibrary: dlopen failed for path='%s' - %s", driverPath.c_str(),
            dlerror() ? dlerror() : "unknown error");
    }
    return handle;
#endif
}

// Platform-agnostic function to get last error message
std::string GetLastErrorMessage() {
#ifdef _WIN32
    // Windows: Use FormatMessageA
    DWORD error = GetLastError();
    char* messageBuffer = nullptr;
    size_t size = FormatMessageA(
        FORMAT_MESSAGE_ALLOCATE_BUFFER | FORMAT_MESSAGE_FROM_SYSTEM | FORMAT_MESSAGE_IGNORE_INSERTS,
        NULL, error, MAKELANGID(LANG_NEUTRAL, SUBLANG_DEFAULT), (LPSTR)&messageBuffer, 0, NULL);
    std::string errorMessage = messageBuffer ? std::string(messageBuffer, size) : "Unknown error";
    LocalFree(messageBuffer);
    return "Error code: " + std::to_string(error) + " - " + errorMessage;
#else
    // macOS/Unix: Use dlerror
    const char* error = dlerror();
    return error ? std::string(error) : "Unknown error";
#endif
}

/*
 * Resolve ODBC driver path in C++ to avoid circular import issues on Alpine.
 *
 * Background:
 * On Alpine Linux, calling into Python during module initialization (via
 * pybind11) causes a circular import due to musl's stricter dynamic loader
 * behavior.
 *
 * Specifically, importing Python helpers from C++ triggered a re-import of the
 * partially-initialized native module, which works on glibc (Ubuntu/macOS) but
 * fails on musl-based systems like Alpine.
 *
 * By moving driver path resolution entirely into C++, we avoid any Python-layer
 * dependencies during critical initialization, ensuring compatibility across
 * all supported platforms.
 */
std::string GetDriverPathForProviderCpp(const std::string& moduleDir,
                                        const std::string& providerId) {
#if !defined(MSODBCSQL_VERSION_MAJOR) || !defined(MSODBCSQL_VERSION_MAJOR_MINOR)
#error \
    "MSODBCSQL_VERSION_MAJOR / MSODBCSQL_VERSION_MAJOR_MINOR must be defined at build time. " \
    "They are derived from mssql_python_odbc.__version__ in CMakeLists.txt so the driver " \
    "filename can never drift from the packaged driver version."
#endif
    namespace fs = std::filesystem;
    fs::path basePath(moduleDir);
    const std::string normalizedProvider = NormalizeProviderId(providerId);
    if (normalizedProvider != kProviderMsodbcsql18 &&
        normalizedProvider != kProviderMssqlOdbc) {
        throw std::invalid_argument("Unknown ODBC provider '" + providerId + "'.");
    }

    std::string platform;
    std::string arch;

// Detect architecture
#if defined(__aarch64__) || defined(_M_ARM64)
    arch = "arm64";
#elif defined(__x86_64__) || defined(_M_X64) || defined(_M_AMD64)
    arch = "x86_64";  // maps to "x64" on Windows
#else
    throw std::runtime_error("Unsupported architecture");
#endif

    // Rust provider (mssql-odbc): ships as `mssqlodbc.{so,dylib,dll}` (no `lib`
    // prefix on Linux/macOS) under an mssql-python-defined libs/ layout.
    // mssql-python owns the provider wheel, so this layout is authoritative and
    // finalized alongside that wheel build.
    if (normalizedProvider == kProviderMssqlOdbc) {
#ifdef __linux__
    #if defined(__GLIBC__)
        constexpr const char* libc = "glibc";
    #else
        constexpr const char* libc = "musl";
    #endif
        return (basePath / "libs" / "linux" / libc / arch / "lib" / "mssqlodbc.so")
            .string();
#elif defined(__APPLE__)
        return (basePath / "libs" / "macos" / arch / "lib" / "mssqlodbc.dylib").string();
#elif defined(_WIN32)
        {
            std::string winArch = (arch == "x86_64") ? "x64" : arch;
            return (basePath / "libs" / "windows" / winArch / "mssqlodbc.dll").string();
        }
#else
        throw std::runtime_error("Unsupported platform");
#endif
    }

// Detect platform and set path
#ifdef __linux__
    if (fs::exists("/etc/alpine-release")) {
        platform = "alpine";
    } else if (fs::exists("/etc/redhat-release") || fs::exists("/etc/centos-release")) {
        platform = "rhel";
    } else if (fs::exists("/etc/SuSE-release") || fs::exists("/etc/SUSE-brand")) {
        platform = "suse";
    } else {
        platform = "debian_ubuntu";  // Default to debian_ubuntu for other distros
    }

    // The msodbcsql version embedded in the driver filename is injected at build
    // time from mssql_python_odbc.__version__ (the single source of truth for the
    // driver version) via the MSODBCSQL_VERSION_* macros defined in
    // CMakeLists.txt. This keeps the native resolver and the Python package
    // version from ever drifting: GetOdbcLibsBaseDir() calls fs::exists() on this
    // exact path to decide whether the external package is "complete", so a stale
    // name would fail that completeness check and raise a hard error. (The
    // ".so.2.1" suffix is the driver's ELF soname, independent of the product version.)
    fs::path driverPath = basePath / "libs" / "linux" / platform / arch / "lib" /
                          ("libmsodbcsql-" MSODBCSQL_VERSION_MAJOR_MINOR ".so.2.1");
    return driverPath.string();

#elif defined(__APPLE__)
    platform = "macos";
    fs::path driverPath = basePath / "libs" / platform / arch / "lib" /
                          ("libmsodbcsql." MSODBCSQL_VERSION_MAJOR ".dylib");
    return driverPath.string();

#elif defined(_WIN32)
    platform = "windows";
    // Normalize x86_64 to x64 for Windows naming
    if (arch == "x86_64")
        arch = "x64";
    fs::path driverPath =
        basePath / "libs" / platform / arch / ("msodbcsql" MSODBCSQL_VERSION_MAJOR ".dll");
    return driverPath.string();

#else
    throw std::runtime_error("Unsupported platform");
#endif
}

std::string GetDriverPathCpp(const std::string& moduleDir) {
    return GetDriverPathForProviderCpp(moduleDir, GetSelectedProviderId());
}

DriverHandle LoadDriverOrThrowException() {
    namespace fs = std::filesystem;

    // Resolve the base dir from the standalone `mssql_python_odbc` package
    // (required post-split; raises if missing or incomplete). Both the driver
    // path and the Windows `mssql-auth.dll` path below are derived from this
    // directory.
    std::string moduleDir = GetOdbcLibsBaseDir();
    LOG("LoadDriverOrThrowException: ODBC libs base directory resolved to '%s'", moduleDir.c_str());

    std::string archStr = ARCHITECTURE;
    LOG("LoadDriverOrThrowException: Architecture detected as '%s'", archStr.c_str());

    // Use only C++ function for driver path resolution
    // Not using Python function since it causes circular import issues on
    // Alpine Linux and other platforms with strict module loading rules.
    std::string driverPathStr = GetDriverPathCpp(moduleDir);

    fs::path driverPath(driverPathStr);

    LOG("LoadDriverOrThrowException: ODBC driver path determined - path='%s'",
        driverPath.string().c_str());

#ifdef _WIN32
    // Only the classic msodbcsql18 provider preloads its co-located
    // mssql-auth.dll here. The Rust provider (mssql-odbc) loads that library
    // lazily from System32 at interactive-auth time, so preloading a co-located
    // copy would both be unnecessary and defeat its System32-only lookup.
    if (GetSelectedProviderId() == kProviderMsodbcsql18) {
        std::string archDir = (archStr == "win64" || archStr == "amd64" || archStr == "x64") ? "x64"
                              : (archStr == "arm64")                                         ? "arm64"
                                                                                             : "x86";

        fs::path dllDir = fs::path(moduleDir) / "libs" / "windows" / archDir;
        fs::path authDllPath = dllDir / "mssql-auth.dll";
        if (fs::exists(authDllPath)) {
            // Use fs::path::c_str() which returns wchar_t* on Windows with proper encoding
            HMODULE hAuth = LoadLibraryExW(
                authDllPath.c_str(), nullptr,
                LOAD_LIBRARY_SEARCH_APPLICATION_DIR | LOAD_LIBRARY_SEARCH_SYSTEM32 |
                    LOAD_LIBRARY_SEARCH_DLL_LOAD_DIR);
            if (hAuth) {
                LOG("LoadDriverOrThrowException: mssql-auth.dll loaded "
                    "successfully from '%s'",
                    authDllPath.string().c_str());
            } else {
                LOG("LoadDriverOrThrowException: Failed to load mssql-auth.dll "
                    "from '%s' - %s",
                    authDllPath.string().c_str(), GetLastErrorMessage().c_str());
                ThrowStdException("Failed to load mssql-auth.dll. Please ensure it "
                                  "is present in the expected directory.");
            }
        } else {
            LOG("LoadDriverOrThrowException: mssql-auth.dll not found at '%s' - "
                "Entra ID authentication will not be available",
                authDllPath.string().c_str());
            // GetOdbcLibsBaseDir's completeness check should already have rejected
            // a classic-provider directory missing it, so reaching here means the
            // check and this load-time lookup have drifted out of sync.
            ThrowStdException("mssql-auth.dll not found. If you are using Entra "
                              "ID, please ensure it is present.");
        }
    }
#endif

    if (!fs::exists(driverPath)) {
        ThrowStdException("ODBC driver not found at: " + driverPath.string());
    }

    DriverHandle handle = LoadDriverLibrary(driverPath.string());
    if (!handle) {
        LOG("LoadDriverOrThrowException: Failed to load ODBC driver - "
            "path='%s', error='%s'",
            driverPath.string().c_str(), GetLastErrorMessage().c_str());
        ThrowStdException("Failed to load the driver. Please read the documentation "
                          "(https://github.com/microsoft/mssql-python#installation) to "
                          "install the required dependencies.");
    }
    LOG("LoadDriverOrThrowException: ODBC driver library loaded successfully "
        "from '%s'",
        driverPath.string().c_str());

    // Load function pointers using helper
    SQLAllocHandle_ptr = GetFunctionPointer<SQLAllocHandleFunc>(handle, "SQLAllocHandle");
    SQLSetEnvAttr_ptr = GetFunctionPointer<SQLSetEnvAttrFunc>(handle, "SQLSetEnvAttr");
    SQLSetConnectAttr_ptr = GetFunctionPointer<SQLSetConnectAttrFunc>(handle, "SQLSetConnectAttrW");
    SQLSetStmtAttr_ptr = GetFunctionPointer<SQLSetStmtAttrFunc>(handle, "SQLSetStmtAttrW");
    SQLGetConnectAttr_ptr = GetFunctionPointer<SQLGetConnectAttrFunc>(handle, "SQLGetConnectAttrW");

    SQLDriverConnect_ptr = GetFunctionPointer<SQLDriverConnectFunc>(handle, "SQLDriverConnectW");
    SQLExecDirect_ptr = GetFunctionPointer<SQLExecDirectFunc>(handle, "SQLExecDirectW");
    SQLPrepare_ptr = GetFunctionPointer<SQLPrepareFunc>(handle, "SQLPrepareW");
    SQLBindParameter_ptr = GetFunctionPointer<SQLBindParameterFunc>(handle, "SQLBindParameter");
    SQLExecute_ptr = GetFunctionPointer<SQLExecuteFunc>(handle, "SQLExecute");
    SQLRowCount_ptr = GetFunctionPointer<SQLRowCountFunc>(handle, "SQLRowCount");
    SQLGetStmtAttr_ptr = GetFunctionPointer<SQLGetStmtAttrFunc>(handle, "SQLGetStmtAttrW");
    SQLSetDescField_ptr = GetFunctionPointer<SQLSetDescFieldFunc>(handle, "SQLSetDescFieldW");

    SQLFetch_ptr = GetFunctionPointer<SQLFetchFunc>(handle, "SQLFetch");
    SQLFetchScroll_ptr = GetFunctionPointer<SQLFetchScrollFunc>(handle, "SQLFetchScroll");
    SQLGetData_ptr = GetFunctionPointer<SQLGetDataFunc>(handle, "SQLGetData");
    SQLNumResultCols_ptr = GetFunctionPointer<SQLNumResultColsFunc>(handle, "SQLNumResultCols");
    SQLBindCol_ptr = GetFunctionPointer<SQLBindColFunc>(handle, "SQLBindCol");
    SQLDescribeCol_ptr = GetFunctionPointer<SQLDescribeColFunc>(handle, "SQLDescribeColW");
    SQLMoreResults_ptr = GetFunctionPointer<SQLMoreResultsFunc>(handle, "SQLMoreResults");
    SQLColAttribute_ptr = GetFunctionPointer<SQLColAttributeFunc>(handle, "SQLColAttributeW");
    SQLGetTypeInfo_ptr = GetFunctionPointer<SQLGetTypeInfoFunc>(handle, "SQLGetTypeInfoW");
    SQLProcedures_ptr = GetFunctionPointer<SQLProceduresFunc>(handle, "SQLProceduresW");
    SQLForeignKeys_ptr = GetFunctionPointer<SQLForeignKeysFunc>(handle, "SQLForeignKeysW");
    SQLPrimaryKeys_ptr = GetFunctionPointer<SQLPrimaryKeysFunc>(handle, "SQLPrimaryKeysW");
    SQLSpecialColumns_ptr = GetFunctionPointer<SQLSpecialColumnsFunc>(handle, "SQLSpecialColumnsW");
    SQLStatistics_ptr = GetFunctionPointer<SQLStatisticsFunc>(handle, "SQLStatisticsW");
    SQLColumns_ptr = GetFunctionPointer<SQLColumnsFunc>(handle, "SQLColumnsW");
    SQLGetInfo_ptr = GetFunctionPointer<SQLGetInfoFunc>(handle, "SQLGetInfoW");

    SQLEndTran_ptr = GetFunctionPointer<SQLEndTranFunc>(handle, "SQLEndTran");
    SQLDisconnect_ptr = GetFunctionPointer<SQLDisconnectFunc>(handle, "SQLDisconnect");
    SQLFreeHandle_ptr = GetFunctionPointer<SQLFreeHandleFunc>(handle, "SQLFreeHandle");
    SQLFreeStmt_ptr = GetFunctionPointer<SQLFreeStmtFunc>(handle, "SQLFreeStmt");
    SQLCancel_ptr = GetFunctionPointer<SQLCancelFunc>(handle, "SQLCancel");

    SQLGetDiagRec_ptr = GetFunctionPointer<SQLGetDiagRecFunc>(handle, "SQLGetDiagRecW");

    SQLParamData_ptr = GetFunctionPointer<SQLParamDataFunc>(handle, "SQLParamData");
    SQLPutData_ptr = GetFunctionPointer<SQLPutDataFunc>(handle, "SQLPutData");
    SQLTables_ptr = GetFunctionPointer<SQLTablesFunc>(handle, "SQLTablesW");

    SQLDescribeParam_ptr = GetFunctionPointer<SQLDescribeParamFunc>(handle, "SQLDescribeParam");

    bool success = SQLAllocHandle_ptr && SQLSetEnvAttr_ptr && SQLSetConnectAttr_ptr &&
                   SQLSetStmtAttr_ptr && SQLGetConnectAttr_ptr && SQLDriverConnect_ptr &&
                   SQLExecDirect_ptr && SQLPrepare_ptr && SQLBindParameter_ptr && SQLExecute_ptr &&
                   SQLRowCount_ptr && SQLGetStmtAttr_ptr && SQLSetDescField_ptr && SQLFetch_ptr &&
                   SQLFetchScroll_ptr && SQLGetData_ptr && SQLNumResultCols_ptr && SQLBindCol_ptr &&
                   SQLDescribeCol_ptr && SQLMoreResults_ptr && SQLColAttribute_ptr &&
                   SQLEndTran_ptr && SQLDisconnect_ptr && SQLFreeHandle_ptr && SQLFreeStmt_ptr &&
                   SQLGetDiagRec_ptr && SQLGetInfo_ptr && SQLParamData_ptr && SQLPutData_ptr &&
                   SQLTables_ptr && SQLDescribeParam_ptr && SQLGetTypeInfo_ptr &&
                   SQLProcedures_ptr && SQLForeignKeys_ptr && SQLPrimaryKeys_ptr &&
                   SQLSpecialColumns_ptr && SQLStatistics_ptr && SQLColumns_ptr;

    if (!success) {
        ThrowStdException("Failed to load required function pointers from driver.");
    }
    LOG("LoadDriverOrThrowException: All %d ODBC function pointers loaded "
        "successfully",
        44);
    return handle;
}

// DriverLoader definition
DriverLoader::DriverLoader() {}

DriverLoader& DriverLoader::getInstance() {
    static DriverLoader instance;
    return instance;
}

void DriverLoader::loadDriver() {
    PERF_TIMER("DriverLoader::loadDriver");
    // The driver-load work runs inside std::call_once so it happens exactly once
    // per process. Critically, we must NOT let an exception propagate *out of* the
    // call_once callable: on musl libc (Alpine/musllinux) libstdc++ routes
    // std::call_once through pthread_once, which cannot propagate an exception
    // from the callable and calls std::terminate() (SIGABRT) instead -- turning
    // the actionable "install mssql-python-odbc" error into a hard crash. So we
    // capture any failure as an exception_ptr inside the callable and rethrow it
    // below, in a normal context that pybind11 can translate into a Python
    // exception. The stored error persists, so every subsequent call re-raises the
    // same actionable message instead of silently proceeding without a driver.
    // Serialize the transition from selectable to locked with
    // SetSelectedProvider(). Whichever operation acquires g_providerMutex first
    // wins: either the new selection is visible to this load, or a later attempt
    // to change it is rejected. Lock before call_once because a failed load is
    // cached too and cannot be retried with another provider in this process.
    {
        std::lock_guard<std::mutex> lock(g_providerMutex);
        g_providerSelectionLocked = true;
    }
    std::call_once(m_onceFlag, [this]() {
        try {
            LoadDriverOrThrowException();
        } catch (...) {
            m_loadError = std::current_exception();
        }
    });
    if (m_loadError) {
        std::rethrow_exception(m_loadError);
    }
}

// SqlHandle definition
SqlHandle::SqlHandle(SQLSMALLINT type, SQLHANDLE rawHandle,
                     std::shared_ptr<ConnectionCleanupState> cleanupState)
    : _type(type), _handle(rawHandle), _cleanupState(std::move(cleanupState)) {}

SqlHandle::~SqlHandle() {
    if (_handle) {
        free();
    }
}

std::unique_lock<std::mutex> SqlHandle::lockForCleanup() const {
    if (_cleanupState) {
        return std::unique_lock<std::mutex>(_cleanupState->mutex);
    }
    return {};
}

SQLHANDLE SqlHandle::get() const {
    return _handle;
}

SQLSMALLINT SqlHandle::type() const {
    return _type;
}

void SqlHandle::markImplicitlyFreed() {
    // SAFETY: Only STMT handles should be marked as implicitly freed.
    // When a DBC handle is freed, the ODBC driver automatically frees all child STMT handles.
    // Other handle types (ENV, DBC, DESC) are NOT automatically freed by parents.
    // Calling this on wrong handle types will cause silent handle leaks.
    if (_type != SQL_HANDLE_STMT) {
        // Log error but don't throw - we're likely in cleanup/destructor path
        LOG_ERROR("SAFETY VIOLATION: Attempted to mark non-STMT handle as implicitly freed. "
                  "Handle type=%d. This will cause handle leak. Only STMT handles are "
                  "automatically freed by parent DBC handles.",
                  _type);
        return;  // Refuse to mark - let normal free() handle it
    }
    _implicitly_freed = true;
}

/*
 * IMPORTANT: Never log in destructors - it causes segfaults.
 * During program exit, C++ destructors may run AFTER Python shuts down.
 * LOG() tries to acquire Python GIL and call Python functions, which crashes
 * if Python is already gone. Keep destructors simple - just free resources.
 * If you need destruction logs, use explicit close() methods instead.
 */
void SqlHandle::free() {
    freeHandle();
}

SQLRETURN SqlHandle::freeHandle() {
    PERF_TIMER("SqlHandle::free");
    bool pythonShuttingDown = is_python_finalizing();
    bool skipDuringShutdown = _type == SQL_HANDLE_STMT || _type == SQL_HANDLE_DBC;
#ifdef _WIN32
    // The static ENV is destroyed during DLL_PROCESS_DETACH, after Python
    // finalization. Calling ODBC then can access already-torn-down SSPI state.
    skipDuringShutdown = skipDuringShutdown || _type == SQL_HANDLE_ENV;
#endif
    if (pythonShuttingDown && skipDuringShutdown) {
        // Do not wait for another thread's ODBC cleanup during finalization.
        // Process teardown owns any resources not released by atexit cleanup.
        _handle = nullptr;
        return SQL_SUCCESS;
    }

    auto freeNative = [this]() -> SQLRETURN {
        auto cleanupLock = lockForCleanup();
        if (!_handle || !SQLFreeHandle_ptr) {
            return SQL_INVALID_HANDLE;
        }
        describeCache.clear();
        if (_implicitly_freed || (_cleanupState && _cleanupState->disconnected)) {
            _handle = nullptr;
            return SQL_SUCCESS;
        }
        SQLRETURN ret = SQLFreeHandle_ptr(_type, _handle);
        if (SQL_SUCCEEDED(ret)) {
            _handle = nullptr;
        }
        return ret;
    };
    // The same gate is held through SQLDisconnect and child invalidation.
    // Release the GIL before waiting, and unlock before reacquiring it.
    if (!pythonShuttingDown && PyGILState_Check()) {
        py::gil_scoped_release release;
        return freeNative();
    }
    return freeNative();
}

void SqlHandle::close_cursor() {
    if (is_python_finalizing()) {
        return;
    }
    auto closeNative = [this]() -> SQLRETURN {
        auto cleanupLock = lockForCleanup();
        if (_type != SQL_HANDLE_STMT || !_handle || _implicitly_freed ||
            (_cleanupState && _cleanupState->disconnected)) {
            return SQL_SUCCESS;
        }
        if (!SQLFreeStmt_ptr) {
            ThrowStdException("SQLFreeStmt function not loaded");
        }
        return SQLFreeStmt_ptr(_handle, SQL_CLOSE);
    };
    SQLRETURN ret;
    if (PyGILState_Check()) {
        py::gil_scoped_release release;
        ret = closeNative();
    } else {
        ret = closeNative();
    }
    if (ret != SQL_SUCCESS && ret != SQL_SUCCESS_WITH_INFO) {
        ThrowStdException("SQLFreeStmt(SQL_CLOSE) failed");
    }
}

void SqlHandle::cancel() {
    if (is_python_finalizing()) {
        return;
    }
    // Fetch/execute do not take this cleanup gate, so cross-thread cancellation
    // can still interrupt them. Reader finalizers must not cancel a freed handle.
    auto cancelNative = [this]() -> SQLRETURN {
        auto cleanupLock = lockForCleanup();
        if (_type != SQL_HANDLE_STMT || !_handle || _implicitly_freed || !SQLCancel_ptr ||
            (_cleanupState && _cleanupState->disconnected)) {
            return SQL_SUCCESS;
        }
        return SQLCancel_ptr(_handle);
    };
    SQLRETURN ret;
    if (PyGILState_Check()) {
        py::gil_scoped_release release;
        ret = cancelNative();
    } else {
        ret = cancelNative();
    }
    // SQLCancel may return SQL_SUCCESS_WITH_INFO when there was nothing to
    // cancel; that is fine. We only throw on hard failure.
    if (ret != SQL_SUCCESS && ret != SQL_SUCCESS_WITH_INFO) {
        ThrowStdException("SQLCancel failed");
    }
}

SQLRETURN SQLResetStmt_wrap(SqlHandlePtr statementHandle) {
    if (!statementHandle || !statementHandle->get()) {
        return SQL_INVALID_HANDLE;
    }
    if (statementHandle->isImplicitlyFreed()) {
        return SQL_INVALID_HANDLE;
    }
    if (!SQLFreeStmt_ptr) {
        DriverLoader::getInstance().loadDriver();
    }
    SQLHANDLE hStmt = statementHandle->get();

    SQLRETURN rc;
    {
        py::gil_scoped_release release;
        rc = SQLFreeStmt_ptr(hStmt, SQL_CLOSE);
        if (SQL_SUCCEEDED(rc)) {
            rc = SQLFreeStmt_ptr(hStmt, SQL_RESET_PARAMS);
        }
        if (SQL_SUCCEEDED(rc) && SQLSetStmtAttr_ptr) {
            rc = SQLSetStmtAttr_ptr(hStmt, SQL_ATTR_PARAMSET_SIZE, (SQLPOINTER)1, 0);
        }
    }
    return rc;
}

SQLRETURN SQLGetTypeInfo_Wrapper(SqlHandlePtr StatementHandle, SQLSMALLINT DataType) {
    PERF_TIMER("SQLGetTypeInfo_Wrapper");
    if (!SQLGetTypeInfo_ptr) {
        ThrowStdException("SQLGetTypeInfo function not loaded");
    }

    // Release the GIL during the blocking ODBC catalog call
    py::gil_scoped_release release;
    return SQLGetTypeInfo_ptr(StatementHandle->get(), DataType);
}

SQLRETURN SQLProcedures_wrap(SqlHandlePtr StatementHandle, const py::object& catalogObj,
                             const py::object& schemaObj, const py::object& procedureObj) {
    PERF_TIMER("SQLProcedures_wrap");
    if (!SQLProcedures_ptr) {
        ThrowStdException("SQLProcedures function not loaded");
    }

    std::u16string catalog = catalogObj.is_none() ? u"" : catalogObj.cast<std::u16string>();
    std::u16string schema = schemaObj.is_none() ? u"" : schemaObj.cast<std::u16string>();
    std::u16string procedure = procedureObj.is_none() ? u"" : procedureObj.cast<std::u16string>();

    // Release the GIL during the blocking ODBC catalog call
    py::gil_scoped_release release;
    return SQLProcedures_ptr(
        StatementHandle->get(), catalog.empty() ? nullptr : reinterpretU16stringAsSqlWChar(catalog),
        catalog.empty() ? 0 : SQL_NTS,
        schema.empty() ? nullptr : reinterpretU16stringAsSqlWChar(schema),
        schema.empty() ? 0 : SQL_NTS,
        procedure.empty() ? nullptr : reinterpretU16stringAsSqlWChar(procedure),
        procedure.empty() ? 0 : SQL_NTS);
}

SQLRETURN SQLForeignKeys_wrap(SqlHandlePtr StatementHandle, const py::object& pkCatalogObj,
                              const py::object& pkSchemaObj, const py::object& pkTableObj,
                              const py::object& fkCatalogObj, const py::object& fkSchemaObj,
                              const py::object& fkTableObj) {
    PERF_TIMER("SQLForeignKeys_wrap");
    if (!SQLForeignKeys_ptr) {
        ThrowStdException("SQLForeignKeys function not loaded");
    }

    std::u16string pkCatalog = pkCatalogObj.is_none() ? u"" : pkCatalogObj.cast<std::u16string>();
    std::u16string pkSchema = pkSchemaObj.is_none() ? u"" : pkSchemaObj.cast<std::u16string>();
    std::u16string pkTable = pkTableObj.is_none() ? u"" : pkTableObj.cast<std::u16string>();
    std::u16string fkCatalog = fkCatalogObj.is_none() ? u"" : fkCatalogObj.cast<std::u16string>();
    std::u16string fkSchema = fkSchemaObj.is_none() ? u"" : fkSchemaObj.cast<std::u16string>();
    std::u16string fkTable = fkTableObj.is_none() ? u"" : fkTableObj.cast<std::u16string>();

    // Release the GIL during the blocking ODBC catalog call
    py::gil_scoped_release release;
    return SQLForeignKeys_ptr(
        StatementHandle->get(),
        pkCatalog.empty() ? nullptr : reinterpretU16stringAsSqlWChar(pkCatalog),
        pkCatalog.empty() ? 0 : SQL_NTS,
        pkSchema.empty() ? nullptr : reinterpretU16stringAsSqlWChar(pkSchema),
        pkSchema.empty() ? 0 : SQL_NTS,
        pkTable.empty() ? nullptr : reinterpretU16stringAsSqlWChar(pkTable),
        pkTable.empty() ? 0 : SQL_NTS,
        fkCatalog.empty() ? nullptr : reinterpretU16stringAsSqlWChar(fkCatalog),
        fkCatalog.empty() ? 0 : SQL_NTS,
        fkSchema.empty() ? nullptr : reinterpretU16stringAsSqlWChar(fkSchema),
        fkSchema.empty() ? 0 : SQL_NTS,
        fkTable.empty() ? nullptr : reinterpretU16stringAsSqlWChar(fkTable),
        fkTable.empty() ? 0 : SQL_NTS);
}

SQLRETURN SQLPrimaryKeys_wrap(SqlHandlePtr StatementHandle, const py::object& catalogObj,
                              const py::object& schemaObj, const std::u16string& table) {
    PERF_TIMER("SQLPrimaryKeys_wrap");
    if (!SQLPrimaryKeys_ptr) {
        ThrowStdException("SQLPrimaryKeys function not loaded");
    }

    std::u16string catalog = catalogObj.is_none() ? u"" : catalogObj.cast<std::u16string>();
    std::u16string schema = schemaObj.is_none() ? u"" : schemaObj.cast<std::u16string>();

    // Release the GIL during the blocking ODBC catalog call
    py::gil_scoped_release release;
    return SQLPrimaryKeys_ptr(StatementHandle->get(),
                              catalog.empty() ? nullptr : reinterpretU16stringAsSqlWChar(catalog),
                              catalog.empty() ? 0 : SQL_NTS,
                              schema.empty() ? nullptr : reinterpretU16stringAsSqlWChar(schema),
                              schema.empty() ? 0 : SQL_NTS,
                              table.empty() ? nullptr : reinterpretU16stringAsSqlWChar(table),
                              table.empty() ? 0 : SQL_NTS);
}

SQLRETURN SQLStatistics_wrap(SqlHandlePtr StatementHandle, const py::object& catalogObj,
                             const py::object& schemaObj, const std::u16string& table,
                             SQLUSMALLINT unique, SQLUSMALLINT reserved) {
    PERF_TIMER("SQLStatistics_wrap");
    if (!SQLStatistics_ptr) {
        ThrowStdException("SQLStatistics function not loaded");
    }

    std::u16string catalog = catalogObj.is_none() ? u"" : catalogObj.cast<std::u16string>();
    std::u16string schema = schemaObj.is_none() ? u"" : schemaObj.cast<std::u16string>();

    // Release the GIL during the blocking ODBC catalog call
    py::gil_scoped_release release;
    return SQLStatistics_ptr(StatementHandle->get(),
                             catalog.empty() ? nullptr : reinterpretU16stringAsSqlWChar(catalog),
                             catalog.empty() ? 0 : SQL_NTS,
                             schema.empty() ? nullptr : reinterpretU16stringAsSqlWChar(schema),
                             schema.empty() ? 0 : SQL_NTS,
                             table.empty() ? nullptr : reinterpretU16stringAsSqlWChar(table),
                             table.empty() ? 0 : SQL_NTS, unique, reserved);
}

SQLRETURN SQLColumns_wrap(SqlHandlePtr StatementHandle, const py::object& catalogObj,
                          const py::object& schemaObj, const py::object& tableObj,
                          const py::object& columnObj) {
    PERF_TIMER("SQLColumns_wrap");
    if (!SQLColumns_ptr) {
        ThrowStdException("SQLColumns function not loaded");
    }

    std::u16string catalog = catalogObj.is_none() ? u"" : catalogObj.cast<std::u16string>();
    std::u16string schema = schemaObj.is_none() ? u"" : schemaObj.cast<std::u16string>();
    std::u16string table = tableObj.is_none() ? u"" : tableObj.cast<std::u16string>();
    std::u16string column = columnObj.is_none() ? u"" : columnObj.cast<std::u16string>();

    // Release the GIL during the blocking ODBC catalog call
    py::gil_scoped_release release;
    return SQLColumns_ptr(StatementHandle->get(),
                          catalog.empty() ? nullptr : reinterpretU16stringAsSqlWChar(catalog),
                          catalog.empty() ? 0 : SQL_NTS,
                          schema.empty() ? nullptr : reinterpretU16stringAsSqlWChar(schema),
                          schema.empty() ? 0 : SQL_NTS,
                          table.empty() ? nullptr : reinterpretU16stringAsSqlWChar(table),
                          table.empty() ? 0 : SQL_NTS,
                          column.empty() ? nullptr : reinterpretU16stringAsSqlWChar(column),
                          column.empty() ? 0 : SQL_NTS);
}

// Helper function to check for driver errors
ErrorInfo SQLCheckError_Wrap(SQLSMALLINT handleType, SqlHandlePtr handle, SQLRETURN retcode) {
    PERF_TIMER("SQLCheckError_Wrap");
    LOG("SQLCheckError: Checking ODBC errors - handleType=%d, retcode=%d", handleType, retcode);
    if (retcode != SQL_INVALID_HANDLE && !SQL_SUCCEEDED(retcode) && !SQLGetDiagRec_ptr) {
        LOG("SQLCheckError: SQLGetDiagRec function pointer not initialized, loading driver");
        DriverLoader::getInstance().loadDriver();
    }
    return SQLReadError(handleType, handle ? handle->get() : nullptr, retcode);
}

ErrorInfo SQLReadError(SQLSMALLINT handleType, SQLHANDLE rawHandle, SQLRETURN retcode) {
    ErrorInfo errorInfo;
    if (retcode == SQL_INVALID_HANDLE || !rawHandle) {
        errorInfo.ddbcErrorMsg = "Invalid handle!";
        return errorInfo;
    }
    if (!SQL_SUCCEEDED(retcode)) {
        if (!SQLGetDiagRec_ptr) {
            ThrowStdException("SQLGetDiagRec function pointer not initialized");
        }

        SQLWCHAR sqlState[6], message[SQL_MAX_MESSAGE_LENGTH_SQLSERVER];
        SQLINTEGER nativeError;
        SQLSMALLINT messageLen;

        SQLRETURN diagReturn =
            SQLGetDiagRec_ptr(handleType, rawHandle, 1, sqlState, &nativeError, message,
                              SQL_MAX_MESSAGE_LENGTH_SQLSERVER, &messageLen);

        if (SQL_SUCCEEDED(diagReturn)) {
            std::u16string sqlStateUtf16 = dupeSqlWCharAsUtf16Le(sqlState, 5);
            std::u16string messageUtf16 = dupeSqlWCharAsUtf16Le(
                message, std::min(static_cast<size_t>(messageLen),
                                  static_cast<size_t>(SQL_MAX_MESSAGE_LENGTH_SQLSERVER - 1)));

            errorInfo.sqlState = utf16LeToUtf8Alloc(std::move(sqlStateUtf16));
            errorInfo.ddbcErrorMsg = utf16LeToUtf8Alloc(std::move(messageUtf16));
        }
    }
    return errorInfo;
}

py::list SQLGetAllDiagRecords(SqlHandlePtr handle) {
    PERF_TIMER("SQLGetAllDiagRecords");
    LOG("SQLGetAllDiagRecords: Retrieving all diagnostic records for handle "
        "%p, handleType=%d",
        (void*)handle->get(), handle->type());
    if (!SQLGetDiagRec_ptr) {
        LOG("SQLGetAllDiagRecords: SQLGetDiagRec function pointer not "
            "initialized, loading driver");
        DriverLoader::getInstance().loadDriver();
    }

    py::list records;
    SQLHANDLE rawHandle = handle->get();
    SQLSMALLINT handleType = handle->type();

    // Iterate through all available diagnostic records
    for (SQLSMALLINT recNumber = 1;; recNumber++) {
        SQLWCHAR sqlState[6] = {0};
        SQLWCHAR message[SQL_MAX_MESSAGE_LENGTH_SQLSERVER] = {0};
        SQLINTEGER nativeError = 0;
        SQLSMALLINT messageLen = 0;

        SQLRETURN diagReturn =
            SQLGetDiagRec_ptr(handleType, rawHandle, recNumber, sqlState, &nativeError, message,
                              SQL_MAX_MESSAGE_LENGTH_SQLSERVER, &messageLen);

        if (diagReturn == SQL_NO_DATA || !SQL_SUCCEEDED(diagReturn))
            break;

        std::u16string sqlStateUtf16 = dupeSqlWCharAsUtf16Le(sqlState, 5);
        std::u16string messageUtf16 = dupeSqlWCharAsUtf16Le(
            message, std::min(static_cast<size_t>(messageLen),
                              static_cast<size_t>(SQL_MAX_MESSAGE_LENGTH_SQLSERVER - 1)));

        std::string stateStr = utf16LeToUtf8Alloc(std::move(sqlStateUtf16));
        std::string msgStr = utf16LeToUtf8Alloc(std::move(messageUtf16));

        // Format the state string
        std::string stateWithError = "[" + stateStr + "] (" + std::to_string(nativeError) + ")";

        // Create the tuple with converted strings
        records.append(py::make_tuple(py::str(stateWithError), py::str(msgStr)));
    }

    return records;
}

// Wrap SQLExecDirect
SQLRETURN SQLExecDirect_wrap(SqlHandlePtr StatementHandle, const std::u16string& Query) {
    PERF_TIMER("SQLExecDirect_wrap");
    LOG("SQLExecDirect: Executing query directly - statement_handle=%p, "
        "query_length=%zu chars",
        (void*)StatementHandle->get(), Query.length());
    if (!SQLExecDirect_ptr) {
        LOG("SQLExecDirect: Function pointer not initialized, loading driver");
        DriverLoader::getInstance().loadDriver();  // Load the driver
    }

    // Configure forward-only cursor
    if (SQLSetStmtAttr_ptr && StatementHandle && StatementHandle->get()) {
        SQLSetStmtAttr_ptr(StatementHandle->get(), SQL_ATTR_CURSOR_TYPE,
                           (SQLPOINTER)SQL_CURSOR_FORWARD_ONLY, 0);
        SQLSetStmtAttr_ptr(StatementHandle->get(), SQL_ATTR_CONCURRENCY,
                           (SQLPOINTER)SQL_CONCUR_READ_ONLY, 0);
    }

    SQLWCHAR* queryPtr = reinterpretU16stringAsSqlWChar(Query);
    SQLRETURN ret;
    {
        // Release the GIL during the blocking ODBC call so that other Python
        // threads (e.g. asyncio event loop, heartbeat threads) can run while
        // SQL Server executes the query. See issue #540.
        py::gil_scoped_release release;
        ret = SQLExecDirect_ptr(StatementHandle->get(), queryPtr, SQL_NTS);
    }
    if (!SQL_SUCCEEDED(ret)) {
        LOG("SQLExecDirect: Query execution failed - SQLRETURN=%d", ret);
    }
    return ret;
}

// Wrapper for SQLTables
SQLRETURN SQLTables_wrap(SqlHandlePtr StatementHandle, const std::u16string& catalog,
                         const std::u16string& schema, const std::u16string& table,
                         const std::u16string& tableType) {
    PERF_TIMER("SQLTables_wrap");
    if (!SQLTables_ptr) {
        LOG("SQLTables: Function pointer not initialized, loading driver");
        DriverLoader::getInstance().loadDriver();
    }

    SQLRETURN ret;
    {
        // Release the GIL during the blocking ODBC catalog call
        py::gil_scoped_release release;
        ret = SQLTables_ptr(StatementHandle->get(),
                            catalog.empty() ? nullptr : reinterpretU16stringAsSqlWChar(catalog),
                            catalog.empty() ? 0 : SQL_NTS,
                            schema.empty() ? nullptr : reinterpretU16stringAsSqlWChar(schema),
                            schema.empty() ? 0 : SQL_NTS,
                            table.empty() ? nullptr : reinterpretU16stringAsSqlWChar(table),
                            table.empty() ? 0 : SQL_NTS,
                            tableType.empty() ? nullptr : reinterpretU16stringAsSqlWChar(tableType),
                            tableType.empty() ? 0 : SQL_NTS);
    }

    LOG("SQLTables: Catalog metadata query %s - SQLRETURN=%d",
        SQL_SUCCEEDED(ret) ? "succeeded" : "failed", ret);

    return ret;
}

// ---------------------------------------------------------------------------
// SQLExecute_wrap — single C++ pipeline: DetectParamTypes → BindParameters → SQLExecute
// No ParamInfo objects cross the pybind11 boundary.
//
// Honors use_prepare: when true, uses SQLPrepare + SQLExecute (benefiting from
// plan reuse). When false but already prepared, reuses the existing plan.
// When false and not prepared, throws (matching slow path behavior).
// ---------------------------------------------------------------------------
SQLRETURN SQLExecute_wrap(const SqlHandlePtr statementHandle,
                              const std::u16string& query,
                              py::list params,
                              const py::object& input_sizes,
                              py::list is_stmt_prepared,
                              bool use_prepare,
                              const py::dict& encoding_settings) {
    PERF_TIMER("SQLExecute_wrap");
    if (!statementHandle || !statementHandle->get()) {
        return SQL_INVALID_HANDLE;
    }

    SQLHANDLE hStmt = statementHandle->get();

    // Configure forward-only / read-only cursor (matches slow path semantics).
    if (SQLSetStmtAttr_ptr) {
        SQLSetStmtAttr_ptr(hStmt, SQL_ATTR_CURSOR_TYPE,
                           (SQLPOINTER)SQL_CURSOR_FORWARD_ONLY, 0);
        SQLSetStmtAttr_ptr(hStmt, SQL_ATTR_CONCURRENCY,
                           (SQLPOINTER)SQL_CONCUR_READ_ONLY, 0);
    }

    // The encoding-settings dict has the form {"encoding": str, "ctype": int}.
    // Note: the Python layer's SQL_C_CHAR constant is numerically -8, the same
    // as ODBC's SQL_C_WCHAR. As a result, the only path that genuinely uses
    // byte-level character encoding is when the user explicitly opts in via
    // setencoding(..., ctype=mssql_python.SQL_CHAR) (which sends ctype=1, the
    // real ODBC SQL_CHAR). We default to utf-8 and only honor the dict's
    // encoding when ctype == 1 (real ODBC SQL_CHAR). Otherwise the user's
    // "encoding" value is meant for the wide-char path and we leave it alone.
    std::string charEncoding = "utf-8";
    if (encoding_settings.contains("ctype") && encoding_settings.contains("encoding")) {
        int ctype = encoding_settings["ctype"].cast<int>();
        if (ctype == SQL_C_CHAR /* real ODBC value: 1 */) {
            charEncoding = encoding_settings["encoding"].cast<std::string>();
        }
    }

    // The cursor.py caller always passes a fresh `list(actual_params)` so this
    // function is free to mutate slots in place. Even so, every site below uses
    // PyList_SetItem (which decrefs the old slot before stealing the new ref),
    // so the function is safe regardless of who owns the list.

    // Run DetectParamTypes BEFORE SQLPrepare so that type-detection errors
    // (unsupported type, NaN Decimal, precision overflow) don't leave the
    // cursor in a half-prepared state.
    std::vector<ParamInfo> paramInfos = DetectParamTypes(params.ptr(), input_sizes.ptr());

    RETCODE rc;
    bool already_prepared = is_stmt_prepared[0].cast<bool>();

    // Honor use_prepare flag (matching slow path behavior):
    // - use_prepare=true: prepare now (or reuse if same SQL already prepared)
    // - use_prepare=false + already prepared: reuse existing plan
    // - use_prepare=false + not prepared: error (cannot execute unprepared)
    if (!already_prepared) {
        if (use_prepare) {
            SQLWCHAR* queryPtr = reinterpretU16stringAsSqlWChar(query);
            {
                py::gil_scoped_release release;
                rc = SQLPrepare_ptr(hStmt, queryPtr, SQL_NTS);
            }
            if (!SQL_SUCCEEDED(rc)) return rc;
            statementHandle->clearDescribeCache();
            is_stmt_prepared[0] = py::bool_(true);
        } else {
            ThrowStdException("Cannot execute unprepared statement");
        }
    }

    std::vector<std::shared_ptr<void>> paramBuffers;
    rc = BindParameters(*statementHandle, hStmt, params, paramInfos, paramBuffers, charEncoding);
    if (!SQL_SUCCEEDED(rc)) return rc;

    {
        py::gil_scoped_release release;
        rc = SQLExecute_ptr(hStmt);
    }

    // DAE (Data-At-Execution) loop: when BindParameters marks a param as DAE
    // (large str/bytes/binary), SQLExecute returns SQL_NEED_DATA. We must
    // stream the data via SQLParamData/SQLPutData before execution completes.
    // GIL is released around each ODBC call to match slow-path concurrency.
    if (rc == SQL_NEED_DATA) {
        SQLPOINTER paramToken = nullptr;
        auto putData = [&](SQLPOINTER data, SQLLEN len) {
            py::gil_scoped_release release;
            return SQLPutData_ptr(hStmt, data, len);
        };
        while (true) {
            {
                py::gil_scoped_release release;
                rc = SQLParamData_ptr(hStmt, &paramToken);
            }
            if (rc != SQL_NEED_DATA) break;

            // The DAE token is the &paramInfos[i] we handed to SQLBindParameter as the
            // parameter value (see BindParameters), and paramInfos is sized up front and
            // never reallocated, so the token casts straight back to its ParamInfo instead
            // of scanning. Range-check it against the vector before trusting it, so a bogus
            // token throws rather than dereferencing arbitrary memory.
            const ParamInfo* matchedInfo = reinterpret_cast<const ParamInfo*>(paramToken);
            const ParamInfo* first = paramInfos.data();
            const ParamInfo* last = first + paramInfos.size();
            if (matchedInfo < first || matchedInfo >= last) {
                ThrowStdException("SQLExecute: unrecognized paramToken from SQLParamData");
            }
            PyObject* pyObj = matchedInfo->dataPtr.ptr();
            if (!pyObj || pyObj == Py_None) {
                py::gil_scoped_release release;
                SQLPutData_ptr(hStmt, nullptr, 0);
                continue;
            }

            if (PyUnicode_Check(pyObj)) {
                if (matchedInfo->paramCType == SQL_C_WCHAR) {
                    std::u16string u16 =
                        borrow<py::str>(pyObj).cast<std::u16string>();
                    rc = stream_dae_chunks(
                        reinterpretU16stringAsSqlWChar(u16),
                        u16.size() * sizeof(SQLWCHAR),
                        putData);
                    if (!SQL_SUCCEEDED(rc)) return rc;
                } else if (matchedInfo->paramCType == SQL_C_CHAR) {
                    std::string encodedStr;
                    py::object encoded = borrow(pyObj)
                                             .attr("encode")(charEncoding, "strict");
                    encodedStr = encoded.cast<std::string>();
                    rc = stream_dae_chunks(encodedStr.data(), encodedStr.size(), putData);
                    if (!SQL_SUCCEEDED(rc)) return rc;
                } else {
                    ThrowStdException("SQLExecute: unsupported C type for str in DAE");
                }
            } else if (PyBytes_Check(pyObj) || PyByteArray_Check(pyObj)) {
                // matchedInfo->dataPtr holds a strong ref to pyObj for the whole loop.
                const char* dataPtr = nullptr;
                size_t totalBytes = 0;
                std::string bytesStorage;  // only used for the bytearray copy below

                if (PyBytes_Check(pyObj)) {
                    // bytes is immutable and kept alive by the strong ref above, so stream
                    // straight from its internal buffer with no copy. This is the large-blob
                    // DAE path, so skipping a full payload copy is the whole point.
                    dataPtr = PyBytes_AS_STRING(pyObj);
                    totalBytes = static_cast<size_t>(PyBytes_GET_SIZE(pyObj));
                } else {
                    // bytearray is mutable and the GIL is released mid-stream, so copy to a
                    // stable buffer before streaming.
                    bytesStorage.assign(PyByteArray_AS_STRING(pyObj),
                                        static_cast<size_t>(PyByteArray_GET_SIZE(pyObj)));
                    dataPtr = bytesStorage.data();
                    totalBytes = bytesStorage.size();
                }

                rc = stream_dae_chunks(dataPtr, totalBytes, putData);
                if (!SQL_SUCCEEDED(rc)) return rc;
            } else {
                ThrowStdException("SQLExecute: DAE only supported for str or bytes");
            }
        }
        if (!SQL_SUCCEEDED(rc) && rc != SQL_NO_DATA) return rc;
    }

    if (!SQL_SUCCEEDED(rc) && rc != SQL_NO_DATA) return rc;

    // Unbind parameter buffers before they go out of scope.
    // Not called on error paths — diagnostics must remain readable.
    SQLRETURN exec_rc = rc;
    SQLFreeStmt_ptr(hStmt, SQL_RESET_PARAMS);
    return exec_rc;
}

SQLRETURN BindParameterArray(SqlHandle& handle, SQLHANDLE hStmt, const py::list& columnwise_params,
                             std::vector<ParamInfo>& paramInfos, size_t paramSetSize,
                             std::vector<std::shared_ptr<void>>& paramBuffers,
                             const std::string& charEncoding = "utf-8") {
    PERF_TIMER("BindParameterArray");
    LOG("BindParameterArray: Starting column-wise array binding - "
        "param_count=%zu, param_set_size=%zu",
        columnwise_params.size(), paramSetSize);

    std::vector<std::shared_ptr<void>> tempBuffers;

    try {
        // GH-627: resolve unknown NULL array param SQL types before binding any param.
        PreResolveUnknownNullTypes(handle, hStmt, paramInfos);
        for (int paramIndex = 0; paramIndex < columnwise_params.size(); ++paramIndex) {
            const py::list& columnValues = columnwise_params[paramIndex].cast<py::list>();
            ParamInfo& info = paramInfos[paramIndex];
            LOG("BindParameterArray: Processing param_index=%d, C_type=%d, "
                "SQL_type=%d, column_size=%zu, decimal_digits=%d",
                paramIndex, info.paramCType, info.paramSQLType, info.columnSize,
                info.decimalDigits);
            if (columnValues.size() != paramSetSize) {
                LOG("BindParameterArray: Size mismatch - param_index=%d, "
                    "expected=%zu, actual=%zu",
                    paramIndex, paramSetSize, columnValues.size());
                ThrowStdException("Column " + std::to_string(paramIndex) + " has mismatched size.");
            }
            void* dataPtr = nullptr;
            SQLLEN* strLenOrIndArray = nullptr;
            SQLLEN bufferLength = 0;
            switch (info.paramCType) {
                case SQL_C_LONG: {
                    LOG("BindParameterArray: Binding SQL_C_LONG array - "
                        "param_index=%d, count=%zu",
                        paramIndex, paramSetSize);
                    int* dataArray = AllocateParamBufferArray<int>(tempBuffers, paramSetSize);
                    strLenOrIndArray = AllocateParamBufferArray<SQLLEN>(tempBuffers, paramSetSize);
                    for (size_t i = 0; i < paramSetSize; ++i) {
                        if (columnValues[i].is_none()) {
                            dataArray[i] = 0;
                            strLenOrIndArray[i] = SQL_NULL_DATA;
                        } else {
                            dataArray[i] = columnValues[i].cast<int>();
                            strLenOrIndArray[i] = 0;
                        }
                    }
                    LOG("BindParameterArray: SQL_C_LONG bound - param_index=%d", paramIndex);
                    dataPtr = dataArray;
                    break;
                }
                case SQL_C_DOUBLE: {
                    LOG("BindParameterArray: Binding SQL_C_DOUBLE array - "
                        "param_index=%d, count=%zu",
                        paramIndex, paramSetSize);
                    double* dataArray = AllocateParamBufferArray<double>(tempBuffers, paramSetSize);
                    strLenOrIndArray = AllocateParamBufferArray<SQLLEN>(tempBuffers, paramSetSize);
                    for (size_t i = 0; i < paramSetSize; ++i) {
                        if (columnValues[i].is_none()) {
                            dataArray[i] = 0;
                            strLenOrIndArray[i] = SQL_NULL_DATA;
                        } else {
                            dataArray[i] = columnValues[i].cast<double>();
                            strLenOrIndArray[i] = 0;
                        }
                    }
                    LOG("BindParameterArray: SQL_C_DOUBLE bound - "
                        "param_index=%d",
                        paramIndex);
                    dataPtr = dataArray;
                    break;
                }
                case SQL_C_WCHAR: {
                    LOG("BindParameterArray: Binding SQL_C_WCHAR array - "
                        "param_index=%d, count=%zu, column_size=%zu",
                        paramIndex, paramSetSize, info.columnSize);
                    SQLWCHAR* wcharArray = AllocateParamBufferArray<SQLWCHAR>(
                        tempBuffers, paramSetSize * (info.columnSize + 1));
                    strLenOrIndArray = AllocateParamBufferArray<SQLLEN>(tempBuffers, paramSetSize);
                    for (size_t i = 0; i < paramSetSize; ++i) {
                        if (columnValues[i].is_none()) {
                            strLenOrIndArray[i] = SQL_NULL_DATA;
                            std::memset(wcharArray + i * (info.columnSize + 1), 0,
                                        (info.columnSize + 1) * sizeof(SQLWCHAR));
                        } else {
                            std::u16string wstr = columnValues[i].cast<std::u16string>();
                            // u16string is already UTF-16, so the
                            // original check is sufficient
                            if (wstr.length() > info.columnSize) {
                                ThrowStdException("Input string exceeds allowed column size "
                                                  "at parameter index " +
                                                  std::to_string(paramIndex));
                            }
                            std::memcpy(wcharArray + i * (info.columnSize + 1), wstr.c_str(),
                                        (wstr.length() + 1) * sizeof(SQLWCHAR));
                            strLenOrIndArray[i] = SQL_NTS;
                        }
                    }
                    LOG("BindParameterArray: SQL_C_WCHAR bound - "
                        "param_index=%d",
                        paramIndex);
                    dataPtr = wcharArray;
                    bufferLength = (info.columnSize + 1) * sizeof(SQLWCHAR);
                    break;
                }
                case SQL_C_TINYINT:
                case SQL_C_UTINYINT: {
                    LOG("BindParameterArray: Binding SQL_C_TINYINT/UTINYINT "
                        "array - param_index=%d, count=%zu",
                        paramIndex, paramSetSize);
                    unsigned char* dataArray =
                        AllocateParamBufferArray<unsigned char>(tempBuffers, paramSetSize);
                    strLenOrIndArray = AllocateParamBufferArray<SQLLEN>(tempBuffers, paramSetSize);
                    for (size_t i = 0; i < paramSetSize; ++i) {
                        if (columnValues[i].is_none()) {
                            dataArray[i] = 0;
                            strLenOrIndArray[i] = SQL_NULL_DATA;
                        } else {
                            int intVal = columnValues[i].cast<int>();
                            if (intVal < 0 || intVal > 255) {
                                LOG("BindParameterArray: TINYINT value out of "
                                    "range - param_index=%d, row=%zu, value=%d",
                                    paramIndex, i, intVal);
                                ThrowStdException("UTINYINT value out of range at rowIndex " +
                                                  std::to_string(i));
                            }
                            dataArray[i] = static_cast<unsigned char>(intVal);
                            strLenOrIndArray[i] = 0;
                        }
                    }
                    LOG("BindParameterArray: SQL_C_TINYINT bound - "
                        "param_index=%d",
                        paramIndex);
                    dataPtr = dataArray;
                    bufferLength = sizeof(unsigned char);
                    break;
                }
                case SQL_C_SHORT: {
                    LOG("BindParameterArray: Binding SQL_C_SHORT array - "
                        "param_index=%d, count=%zu",
                        paramIndex, paramSetSize);
                    short* dataArray = AllocateParamBufferArray<short>(tempBuffers, paramSetSize);
                    strLenOrIndArray = AllocateParamBufferArray<SQLLEN>(tempBuffers, paramSetSize);
                    for (size_t i = 0; i < paramSetSize; ++i) {
                        if (columnValues[i].is_none()) {
                            dataArray[i] = 0;
                            strLenOrIndArray[i] = SQL_NULL_DATA;
                        } else {
                            int intVal = columnValues[i].cast<int>();
                            if (intVal < std::numeric_limits<short>::min() ||
                                intVal > std::numeric_limits<short>::max()) {
                                LOG("BindParameterArray: SHORT value out of "
                                    "range - param_index=%d, row=%zu, value=%d",
                                    paramIndex, i, intVal);
                                ThrowStdException("SHORT value out of range at rowIndex " +
                                                  std::to_string(i));
                            }
                            dataArray[i] = static_cast<short>(intVal);
                            strLenOrIndArray[i] = 0;
                        }
                    }
                    LOG("BindParameterArray: SQL_C_SHORT bound - "
                        "param_index=%d",
                        paramIndex);
                    dataPtr = dataArray;
                    bufferLength = sizeof(short);
                    break;
                }
                case SQL_C_CHAR:
                case SQL_C_BINARY: {
                    LOG("BindParameterArray: Binding SQL_C_CHAR/BINARY array - "
                        "param_index=%d, count=%zu, column_size=%zu, encoding='%s'",
                        paramIndex, paramSetSize, info.columnSize, charEncoding.c_str());
                    char* charArray = AllocateParamBufferArray<char>(
                        tempBuffers, paramSetSize * (info.columnSize + 1));
                    strLenOrIndArray = AllocateParamBufferArray<SQLLEN>(tempBuffers, paramSetSize);
                    for (size_t i = 0; i < paramSetSize; ++i) {
                        if (columnValues[i].is_none()) {
                            strLenOrIndArray[i] = SQL_NULL_DATA;
                            std::memset(charArray + i * (info.columnSize + 1), 0,
                                        info.columnSize + 1);
                        } else {
                            std::string encodedStr;

                            if (py::isinstance<py::str>(columnValues[i])) {
                                // Use Python's codec system to encode the string with specified
                                // encoding
                                try {
                                    py::object encoded =
                                        columnValues[i].attr("encode")(charEncoding, "strict");
                                    encodedStr = encoded.cast<std::string>();
                                    LOG("BindParameterArray: param[%d] row[%zu] SQL_C_CHAR - "
                                        "Encoded with '%s', "
                                        "size=%zu bytes",
                                        paramIndex, i, charEncoding.c_str(), encodedStr.size());
                                } catch (const py::error_already_set& e) {
                                    LOG_ERROR("BindParameterArray: param[%d] row[%zu] SQL_C_CHAR - "
                                              "Failed to encode "
                                              "with '%s': %s",
                                              paramIndex, i, charEncoding.c_str(), e.what());
                                    throw std::runtime_error(
                                        std::string("Failed to encode parameter ") +
                                        std::to_string(paramIndex) + " row " + std::to_string(i) +
                                        " with encoding '" + charEncoding + "': " + e.what());
                                }
                            } else {
                                // bytes/bytearray - use as-is (already encoded)
                                encodedStr = columnValues[i].cast<std::string>();
                            }

                            if (encodedStr.size() > info.columnSize) {
                                LOG("BindParameterArray: String/binary too "
                                    "long - param_index=%d, row=%zu, size=%zu, "
                                    "max=%zu",
                                    paramIndex, i, encodedStr.size(), info.columnSize);
                                ThrowStdException("Input exceeds column size at index " +
                                                  std::to_string(i));
                            }
                            std::memcpy(charArray + i * (info.columnSize + 1), encodedStr.c_str(),
                                        encodedStr.size());
                            strLenOrIndArray[i] = static_cast<SQLLEN>(encodedStr.size());
                        }
                    }
                    LOG("BindParameterArray: SQL_C_CHAR/BINARY bound - "
                        "param_index=%d",
                        paramIndex);
                    dataPtr = charArray;
                    bufferLength = info.columnSize + 1;
                    break;
                }
                case SQL_C_BIT: {
                    LOG("BindParameterArray: Binding SQL_C_BIT array - "
                        "param_index=%d, count=%zu",
                        paramIndex, paramSetSize);
                    char* boolArray = AllocateParamBufferArray<char>(tempBuffers, paramSetSize);
                    strLenOrIndArray = AllocateParamBufferArray<SQLLEN>(tempBuffers, paramSetSize);
                    for (size_t i = 0; i < paramSetSize; ++i) {
                        if (columnValues[i].is_none()) {
                            boolArray[i] = 0;
                            strLenOrIndArray[i] = SQL_NULL_DATA;
                        } else {
                            bool val = columnValues[i].cast<bool>();
                            boolArray[i] = val ? 1 : 0;
                            strLenOrIndArray[i] = 0;
                        }
                    }
                    LOG("BindParameterArray: SQL_C_BIT bound - param_index=%d", paramIndex);
                    dataPtr = boolArray;
                    bufferLength = sizeof(char);
                    break;
                }
                case SQL_C_STINYINT:
                case SQL_C_USHORT: {
                    LOG("BindParameterArray: Binding SQL_C_USHORT/STINYINT "
                        "array - param_index=%d, count=%zu",
                        paramIndex, paramSetSize);
                    unsigned short* dataArray =
                        AllocateParamBufferArray<unsigned short>(tempBuffers, paramSetSize);
                    strLenOrIndArray = AllocateParamBufferArray<SQLLEN>(tempBuffers, paramSetSize);
                    for (size_t i = 0; i < paramSetSize; ++i) {
                        if (columnValues[i].is_none()) {
                            strLenOrIndArray[i] = SQL_NULL_DATA;
                            dataArray[i] = 0;
                        } else {
                            dataArray[i] = columnValues[i].cast<unsigned short>();
                            strLenOrIndArray[i] = 0;
                        }
                    }
                    LOG("BindParameterArray: SQL_C_USHORT bound - "
                        "param_index=%d",
                        paramIndex);
                    dataPtr = dataArray;
                    bufferLength = sizeof(unsigned short);
                    break;
                }
                case SQL_C_SBIGINT:
                case SQL_C_SLONG:
                case SQL_C_UBIGINT:
                case SQL_C_ULONG: {
                    LOG("BindParameterArray: Binding SQL_C_BIGINT array - "
                        "param_index=%d, count=%zu",
                        paramIndex, paramSetSize);
                    int64_t* dataArray =
                        AllocateParamBufferArray<int64_t>(tempBuffers, paramSetSize);
                    strLenOrIndArray = AllocateParamBufferArray<SQLLEN>(tempBuffers, paramSetSize);
                    for (size_t i = 0; i < paramSetSize; ++i) {
                        if (columnValues[i].is_none()) {
                            strLenOrIndArray[i] = SQL_NULL_DATA;
                            dataArray[i] = 0;
                        } else {
                            dataArray[i] = columnValues[i].cast<int64_t>();
                            strLenOrIndArray[i] = 0;
                        }
                    }
                    LOG("BindParameterArray: SQL_C_BIGINT bound - "
                        "param_index=%d",
                        paramIndex);
                    dataPtr = dataArray;
                    bufferLength = sizeof(int64_t);
                    break;
                }
                case SQL_C_FLOAT: {
                    LOG("BindParameterArray: Binding SQL_C_FLOAT array - "
                        "param_index=%d, count=%zu",
                        paramIndex, paramSetSize);
                    float* dataArray = AllocateParamBufferArray<float>(tempBuffers, paramSetSize);
                    strLenOrIndArray = AllocateParamBufferArray<SQLLEN>(tempBuffers, paramSetSize);
                    for (size_t i = 0; i < paramSetSize; ++i) {
                        if (columnValues[i].is_none()) {
                            strLenOrIndArray[i] = SQL_NULL_DATA;
                            dataArray[i] = 0.0f;
                        } else {
                            dataArray[i] = columnValues[i].cast<float>();
                            strLenOrIndArray[i] = 0;
                        }
                    }
                    LOG("BindParameterArray: SQL_C_FLOAT bound - "
                        "param_index=%d",
                        paramIndex);
                    dataPtr = dataArray;
                    bufferLength = sizeof(float);
                    break;
                }
                case SQL_C_TYPE_DATE: {
                    LOG("BindParameterArray: Binding SQL_C_TYPE_DATE array - "
                        "param_index=%d, count=%zu",
                        paramIndex, paramSetSize);
                    SQL_DATE_STRUCT* dateArray =
                        AllocateParamBufferArray<SQL_DATE_STRUCT>(tempBuffers, paramSetSize);
                    strLenOrIndArray = AllocateParamBufferArray<SQLLEN>(tempBuffers, paramSetSize);
                    for (size_t i = 0; i < paramSetSize; ++i) {
                        if (columnValues[i].is_none()) {
                            strLenOrIndArray[i] = SQL_NULL_DATA;
                            std::memset(&dateArray[i], 0, sizeof(SQL_DATE_STRUCT));
                        } else {
                            py::object dateObj = columnValues[i];
                            dateArray[i].year = dateObj.attr("year").cast<SQLSMALLINT>();
                            dateArray[i].month = dateObj.attr("month").cast<SQLUSMALLINT>();
                            dateArray[i].day = dateObj.attr("day").cast<SQLUSMALLINT>();
                            strLenOrIndArray[i] = 0;
                        }
                    }
                    LOG("BindParameterArray: SQL_C_TYPE_DATE bound - "
                        "param_index=%d",
                        paramIndex);
                    dataPtr = dateArray;
                    bufferLength = sizeof(SQL_DATE_STRUCT);
                    break;
                }
                case SQL_C_TYPE_TIME: {
                    LOG("BindParameterArray: Binding SQL_C_TYPE_TIME array - "
                        "param_index=%d, count=%zu",
                        paramIndex, paramSetSize);
                    SQL_TIME_STRUCT* timeArray =
                        AllocateParamBufferArray<SQL_TIME_STRUCT>(tempBuffers, paramSetSize);
                    strLenOrIndArray = AllocateParamBufferArray<SQLLEN>(tempBuffers, paramSetSize);
                    for (size_t i = 0; i < paramSetSize; ++i) {
                        if (columnValues[i].is_none()) {
                            strLenOrIndArray[i] = SQL_NULL_DATA;
                            std::memset(&timeArray[i], 0, sizeof(SQL_TIME_STRUCT));
                        } else {
                            py::object timeObj = columnValues[i];
                            timeArray[i].hour = timeObj.attr("hour").cast<SQLUSMALLINT>();
                            timeArray[i].minute = timeObj.attr("minute").cast<SQLUSMALLINT>();
                            timeArray[i].second = timeObj.attr("second").cast<SQLUSMALLINT>();
                            strLenOrIndArray[i] = 0;
                        }
                    }
                    LOG("BindParameterArray: SQL_C_TYPE_TIME bound - "
                        "param_index=%d",
                        paramIndex);
                    dataPtr = timeArray;
                    bufferLength = sizeof(SQL_TIME_STRUCT);
                    break;
                }
                case SQL_C_TYPE_TIMESTAMP: {
                    LOG("BindParameterArray: Binding SQL_C_TYPE_TIMESTAMP "
                        "array - param_index=%d, count=%zu",
                        paramIndex, paramSetSize);
                    SQL_TIMESTAMP_STRUCT* tsArray =
                        AllocateParamBufferArray<SQL_TIMESTAMP_STRUCT>(tempBuffers, paramSetSize);
                    strLenOrIndArray = AllocateParamBufferArray<SQLLEN>(tempBuffers, paramSetSize);
                    for (size_t i = 0; i < paramSetSize; ++i) {
                        if (columnValues[i].is_none()) {
                            strLenOrIndArray[i] = SQL_NULL_DATA;
                            std::memset(&tsArray[i], 0, sizeof(SQL_TIMESTAMP_STRUCT));
                        } else {
                            py::object dtObj = columnValues[i];
                            tsArray[i].year = dtObj.attr("year").cast<SQLSMALLINT>();
                            tsArray[i].month = dtObj.attr("month").cast<SQLUSMALLINT>();
                            tsArray[i].day = dtObj.attr("day").cast<SQLUSMALLINT>();
                            tsArray[i].hour = dtObj.attr("hour").cast<SQLUSMALLINT>();
                            tsArray[i].minute = dtObj.attr("minute").cast<SQLUSMALLINT>();
                            tsArray[i].second = dtObj.attr("second").cast<SQLUSMALLINT>();
                            tsArray[i].fraction = static_cast<SQLUINTEGER>(
                                dtObj.attr("microsecond").cast<int>() * 1000);  // µs to ns
                            strLenOrIndArray[i] = 0;
                        }
                    }
                    LOG("BindParameterArray: SQL_C_TYPE_TIMESTAMP bound - "
                        "param_index=%d",
                        paramIndex);
                    dataPtr = tsArray;
                    bufferLength = sizeof(SQL_TIMESTAMP_STRUCT);
                    break;
                }
                case SQL_C_SS_TIMESTAMPOFFSET: {
                    LOG("BindParameterArray: Binding SQL_C_SS_TIMESTAMPOFFSET "
                        "array - param_index=%d, count=%zu",
                        paramIndex, paramSetSize);
                    DateTimeOffset* dtoArray =
                        AllocateParamBufferArray<DateTimeOffset>(tempBuffers, paramSetSize);
                    strLenOrIndArray = AllocateParamBufferArray<SQLLEN>(tempBuffers, paramSetSize);

                    py::object datetimeType = PyTypeCache::get_datetime_class_obj();

                    for (size_t i = 0; i < paramSetSize; ++i) {
                        const py::handle& param = columnValues[i];

                        if (param.is_none()) {
                            std::memset(&dtoArray[i], 0, sizeof(DateTimeOffset));
                            strLenOrIndArray[i] = SQL_NULL_DATA;
                        } else {
                            if (!py::isinstance(param, datetimeType)) {
                                ThrowStdException(
                                    MakeParamMismatchErrorStr(info.paramCType, paramIndex));
                            }

                            py::object tzinfo = param.attr("tzinfo");
                            if (tzinfo.is_none()) {
                                ThrowStdException("Datetime object must have tzinfo for "
                                                  "SQL_C_SS_TIMESTAMPOFFSET at paramIndex " +
                                                  std::to_string(paramIndex));
                            }

                            // Populate the C++ struct directly from the Python
                            // datetime object.
                            dtoArray[i].year =
                                static_cast<SQLSMALLINT>(param.attr("year").cast<int>());
                            dtoArray[i].month =
                                static_cast<SQLUSMALLINT>(param.attr("month").cast<int>());
                            dtoArray[i].day =
                                static_cast<SQLUSMALLINT>(param.attr("day").cast<int>());
                            dtoArray[i].hour =
                                static_cast<SQLUSMALLINT>(param.attr("hour").cast<int>());
                            dtoArray[i].minute =
                                static_cast<SQLUSMALLINT>(param.attr("minute").cast<int>());
                            dtoArray[i].second =
                                static_cast<SQLUSMALLINT>(param.attr("second").cast<int>());
                            // SQL server supports in ns, but python datetime
                            // supports in µs
                            dtoArray[i].fraction = static_cast<SQLUINTEGER>(
                                param.attr("microsecond").cast<int>() * 1000);

                            // Compute and preserve the original UTC offset.
                            py::object utcoffset = tzinfo.attr("utcoffset")(param);
                            int total_seconds =
                                static_cast<int>(utcoffset.attr("total_seconds")().cast<double>());
                            std::div_t div_result = std::div(total_seconds, 3600);
                            dtoArray[i].timezone_hour = static_cast<SQLSMALLINT>(div_result.quot);
                            dtoArray[i].timezone_minute =
                                static_cast<SQLSMALLINT>(div(div_result.rem, 60).quot);

                            strLenOrIndArray[i] = sizeof(DateTimeOffset);
                        }
                    }
                    LOG("BindParameterArray: SQL_C_SS_TIMESTAMPOFFSET bound - "
                        "param_index=%d",
                        paramIndex);
                    dataPtr = dtoArray;
                    bufferLength = sizeof(DateTimeOffset);
                    break;
                }
                case SQL_C_NUMERIC: {
                    LOG("BindParameterArray: Binding SQL_C_NUMERIC array - "
                        "param_index=%d, count=%zu",
                        paramIndex, paramSetSize);
                    SQL_NUMERIC_STRUCT* numericArray =
                        AllocateParamBufferArray<SQL_NUMERIC_STRUCT>(tempBuffers, paramSetSize);
                    strLenOrIndArray = AllocateParamBufferArray<SQLLEN>(tempBuffers, paramSetSize);
                    for (size_t i = 0; i < paramSetSize; ++i) {
                        const py::handle& element = columnValues[i];
                        if (element.is_none()) {
                            strLenOrIndArray[i] = SQL_NULL_DATA;
                            std::memset(&numericArray[i], 0, sizeof(SQL_NUMERIC_STRUCT));
                            continue;
                        }
                        if (!py::isinstance<NumericData>(element)) {
                            LOG("BindParameterArray: NUMERIC type mismatch - "
                                "param_index=%d, row=%zu",
                                paramIndex, i);
                            throw std::runtime_error(
                                MakeParamMismatchErrorStr(info.paramCType, paramIndex));
                        }
                        NumericData decimalParam = element.cast<NumericData>();
                        LOG("BindParameterArray: NUMERIC value - "
                            "param_index=%d, row=%zu, precision=%d, scale=%d, "
                            "sign=%d",
                            paramIndex, i, decimalParam.precision, decimalParam.scale,
                            decimalParam.sign);
                        SQL_NUMERIC_STRUCT& target = numericArray[i];
                        std::memset(&target, 0, sizeof(SQL_NUMERIC_STRUCT));
                        target.precision = decimalParam.precision;
                        target.scale = decimalParam.scale;
                        target.sign = decimalParam.sign;
                        size_t copyLen = std::min(decimalParam.val.size(), sizeof(target.val));
                        if (copyLen > 0) {
                            std::memcpy(target.val, decimalParam.val.data(), copyLen);
                        }
                        strLenOrIndArray[i] = sizeof(SQL_NUMERIC_STRUCT);
                    }
                    LOG("BindParameterArray: SQL_C_NUMERIC bound - "
                        "param_index=%d",
                        paramIndex);
                    dataPtr = numericArray;
                    bufferLength = sizeof(SQL_NUMERIC_STRUCT);
                    break;
                }
                case SQL_C_GUID: {
                    LOG("BindParameterArray: Binding SQL_C_GUID array - "
                        "param_index=%d, count=%zu",
                        paramIndex, paramSetSize);
                    SQLGUID* guidArray =
                        AllocateParamBufferArray<SQLGUID>(tempBuffers, paramSetSize);
                    strLenOrIndArray = AllocateParamBufferArray<SQLLEN>(tempBuffers, paramSetSize);

                    // Get cached UUID class from module-level helper
                    // This avoids static object destruction issues during
                    // Python finalization
                    py::object uuid_class = PyTypeCache::get_uuid_class_obj();
                    // Get cached UUID class

                    for (size_t i = 0; i < paramSetSize; ++i) {
                        const py::handle& element = columnValues[i];
                        std::array<unsigned char, 16> uuid_bytes;
                        if (element.is_none()) {
                            std::memset(&guidArray[i], 0, sizeof(SQLGUID));
                            strLenOrIndArray[i] = SQL_NULL_DATA;
                            continue;
                        } else if (py::isinstance<py::bytes>(element)) {
                            py::bytes b = element.cast<py::bytes>();
                            if (PyBytes_GET_SIZE(b.ptr()) != 16) {
                                LOG("BindParameterArray: GUID bytes wrong "
                                    "length - param_index=%d, row=%zu, "
                                    "length=%d",
                                    paramIndex, i, PyBytes_GET_SIZE(b.ptr()));
                                ThrowStdException("UUID binary data must be "
                                                  "exactly 16 bytes long.");
                            }
                            std::memcpy(uuid_bytes.data(), PyBytes_AS_STRING(b.ptr()), 16);
                        } else if (py::isinstance(element, uuid_class)) {
                            py::bytes b = element.attr("bytes_le").cast<py::bytes>();
                            std::memcpy(uuid_bytes.data(), PyBytes_AS_STRING(b.ptr()), 16);
                        } else {
                            LOG("BindParameterArray: GUID type mismatch - "
                                "param_index=%d, row=%zu",
                                paramIndex, i);
                            ThrowStdException(
                                MakeParamMismatchErrorStr(info.paramCType, paramIndex));
                        }
                        guidArray[i].Data1 = (static_cast<uint32_t>(uuid_bytes[3]) << 24) |
                                             (static_cast<uint32_t>(uuid_bytes[2]) << 16) |
                                             (static_cast<uint32_t>(uuid_bytes[1]) << 8) |
                                             (static_cast<uint32_t>(uuid_bytes[0]));
                        guidArray[i].Data2 = (static_cast<uint16_t>(uuid_bytes[5]) << 8) |
                                             (static_cast<uint16_t>(uuid_bytes[4]));
                        guidArray[i].Data3 = (static_cast<uint16_t>(uuid_bytes[7]) << 8) |
                                             (static_cast<uint16_t>(uuid_bytes[6]));
                        std::memcpy(guidArray[i].Data4, uuid_bytes.data() + 8, 8);
                        strLenOrIndArray[i] = sizeof(SQLGUID);
                    }
                    LOG("BindParameterArray: SQL_C_GUID bound - "
                        "param_index=%d, null=%zu, bytes=%zu, uuid_obj=%zu",
                        paramIndex);
                    dataPtr = guidArray;
                    bufferLength = sizeof(SQLGUID);
                    break;
                }
                case SQL_C_DEFAULT: {
                    // Handle NULL parameters - all values in this column should be NULL
                    // GH-627: SQL type already resolved by PreResolveUnknownNullTypes.
                    LOG("BindParameterArray: Binding SQL_C_DEFAULT (NULL) array - param_index=%d, "
                        "count=%zu, resolvedSqlType=%d",
                        paramIndex, paramSetSize, info.paramSQLType);

                    char* nullBuffer = AllocateParamBufferArray<char>(tempBuffers, paramSetSize);
                    strLenOrIndArray = AllocateParamBufferArray<SQLLEN>(tempBuffers, paramSetSize);

                    for (size_t i = 0; i < paramSetSize; ++i) {
                        nullBuffer[i] = 0;
                        strLenOrIndArray[i] = SQL_NULL_DATA;
                    }

                    dataPtr = nullBuffer;
                    bufferLength = 1;
                    break;
                }
                default: {
                    LOG("BindParameterArray: Unsupported C type - "
                        "param_index=%d, C_type=%d",
                        paramIndex, info.paramCType);
                    ThrowStdException("BindParameterArray: Unsupported C type: " +
                                      std::to_string(info.paramCType));
                }
            }
            LOG("BindParameterArray: Calling SQLBindParameter - "
                "param_index=%d, buffer_length=%lld",
                paramIndex, static_cast<long long>(bufferLength));
            RETCODE rc;
            {
                PERF_TIMER("BindParameterArray::SQLBindParameter_call");
                rc = SQLBindParameter_ptr(hStmt, static_cast<SQLUSMALLINT>(paramIndex + 1),
                                         static_cast<SQLUSMALLINT>(info.inputOutputType),
                                         static_cast<SQLSMALLINT>(info.paramCType),
                                         static_cast<SQLSMALLINT>(info.paramSQLType), info.columnSize,
                                         info.decimalDigits, dataPtr, bufferLength, strLenOrIndArray);
            }
            if (!SQL_SUCCEEDED(rc)) {
                LOG("BindParameterArray: SQLBindParameter failed - "
                    "param_index=%d, SQLRETURN=%d",
                    paramIndex, rc);
                return rc;
            }
        }
    } catch (...) {
        LOG("BindParameterArray: Exception during binding, cleaning up "
            "buffers");
        throw;
    }
    paramBuffers.insert(paramBuffers.end(), tempBuffers.begin(), tempBuffers.end());
    LOG("BindParameterArray: Successfully bound all parameters - "
        "total_params=%zu, buffer_count=%zu",
        columnwise_params.size(), paramBuffers.size());
    return SQL_SUCCESS;
}

SQLRETURN SQLExecuteMany_wrap(const SqlHandlePtr statementHandle, const std::u16string& query,
                              const py::list& columnwise_params,
                              std::vector<ParamInfo>& paramInfos, size_t paramSetSize,
                              const py::dict& encodingSettings) {
    PERF_TIMER("SQLExecuteMany_wrap");
    LOG("SQLExecuteMany: Starting batch execution - param_count=%zu, "
        "param_set_size=%zu",
        columnwise_params.size(), paramSetSize);
    SQLHANDLE hStmt = statementHandle->get();
    SQLWCHAR* queryPtr = reinterpretU16stringAsSqlWChar(query);
    LOG("SQLExecuteMany: Using wide string query directly");
    RETCODE rc;
    {
        // Release the GIL during the blocking SQLPrepare network call.
        py::gil_scoped_release release;
        rc = SQLPrepare_ptr(hStmt, queryPtr, SQL_NTS);
    }
    if (!SQL_SUCCEEDED(rc)) {
        LOG("SQLExecuteMany: SQLPrepare failed - rc=%d", rc);
        return rc;
    }
    // GH-610: Clear per-handle describe cache (new prepare = new param types)
    statementHandle->clearDescribeCache();
    LOG("SQLExecuteMany: Query prepared successfully");

    bool hasDAE = false;
    for (const auto& p : paramInfos) {
        if (p.isDAE) {
            hasDAE = true;
            break;
        }
    }
    LOG("SQLExecuteMany: Parameter analysis - hasDAE=%s", hasDAE ? "true" : "false");

    // Extract char encoding from encodingSettings dictionary
    std::string charEncoding = "utf-8";  // default
    if (encodingSettings.contains("encoding")) {
        charEncoding = encodingSettings["encoding"].cast<std::string>();
    }

    if (!hasDAE) {
        LOG("SQLExecuteMany: Using array binding (non-DAE) - calling "
            "BindParameterArray with encoding '%s'",
            charEncoding.c_str());
        std::vector<std::shared_ptr<void>> paramBuffers;
        rc = BindParameterArray(*statementHandle, hStmt, columnwise_params, paramInfos, paramSetSize, paramBuffers,
                                charEncoding);
        if (!SQL_SUCCEEDED(rc)) {
            LOG("SQLExecuteMany: BindParameterArray failed - rc=%d", rc);
            return rc;
        }

        rc = SQLSetStmtAttr_ptr(hStmt, SQL_ATTR_PARAMSET_SIZE, (SQLPOINTER)paramSetSize, 0);
        if (!SQL_SUCCEEDED(rc)) {
            LOG("SQLExecuteMany: SQLSetStmtAttr(PARAMSET_SIZE) failed - rc=%d", rc);
            return rc;
        }
        LOG("SQLExecuteMany: PARAMSET_SIZE set to %zu", paramSetSize);

        {
            // Release the GIL during the blocking SQLExecute network call.
            py::gil_scoped_release release;
            rc = SQLExecute_ptr(hStmt);
        }
        LOG("SQLExecuteMany: SQLExecute completed - rc=%d", rc);
        return rc;
    } else {
        LOG("SQLExecuteMany: Using DAE (data-at-execution) - row_count=%zu",
            columnwise_params.size());
        size_t rowCount = columnwise_params.size();
        for (size_t rowIndex = 0; rowIndex < rowCount; ++rowIndex) {
            LOG("SQLExecuteMany: Processing DAE row %zu of %zu", rowIndex + 1, rowCount);
            py::list rowParams = columnwise_params[rowIndex];

            std::vector<std::shared_ptr<void>> paramBuffers;
            rc = BindParameters(*statementHandle, hStmt, rowParams, paramInfos,
                                paramBuffers, charEncoding);
            if (!SQL_SUCCEEDED(rc)) {
                LOG("SQLExecuteMany: BindParameters failed for row %zu - rc=%d", rowIndex, rc);
                return rc;
            }
            LOG("SQLExecuteMany: Parameters bound for row %zu", rowIndex);

            {
                // Release the GIL during the blocking SQLExecute network call.
                py::gil_scoped_release release;
                rc = SQLExecute_ptr(hStmt);
            }
            LOG("SQLExecuteMany: SQLExecute for row %zu - initial_rc=%d", rowIndex, rc);
            size_t dae_chunk_count = 0;
            while (rc == SQL_NEED_DATA) {
                SQLPOINTER token;
                {
                    // Release the GIL around the blocking SQLParamData call.
                    py::gil_scoped_release release;
                    rc = SQLParamData_ptr(hStmt, &token);
                }
                LOG("SQLExecuteMany: SQLParamData called - chunk=%zu, rc=%d, "
                    "token=%p",
                    dae_chunk_count, rc, token);
                if (!SQL_SUCCEEDED(rc) && rc != SQL_NEED_DATA) {
                    LOG("SQLExecuteMany: SQLParamData failed - chunk=%zu, "
                        "rc=%d",
                        dae_chunk_count, rc);
                    return rc;
                }

                py::object* py_obj_ptr = reinterpret_cast<py::object*>(token);
                if (!py_obj_ptr) {
                    LOG("SQLExecuteMany: NULL token pointer in DAE - chunk=%zu", dae_chunk_count);
                    return SQL_ERROR;
                }

                if (py::isinstance<py::str>(*py_obj_ptr)) {
                    std::string data = py_obj_ptr->cast<std::string>();
                    SQLLEN data_len = static_cast<SQLLEN>(data.size());
                    LOG("SQLExecuteMany: Sending string DAE data - chunk=%zu, "
                        "length=%lld",
                        dae_chunk_count, static_cast<long long>(data_len));
                    rc = [&] {
                        py::gil_scoped_release release;
                        return SQLPutData_ptr(hStmt, (SQLPOINTER)data.c_str(), data_len);
                    }();
                    if (!SQL_SUCCEEDED(rc) && rc != SQL_NEED_DATA) {
                        LOG("SQLExecuteMany: SQLPutData(string) failed - "
                            "chunk=%zu, rc=%d",
                            dae_chunk_count, rc);
                    }
                } else if (py::isinstance<py::bytes>(*py_obj_ptr) ||
                           py::isinstance<py::bytearray>(*py_obj_ptr)) {
                    std::string data = py_obj_ptr->cast<std::string>();
                    SQLLEN data_len = static_cast<SQLLEN>(data.size());
                    LOG("SQLExecuteMany: Sending bytes/bytearray DAE data - "
                        "chunk=%zu, length=%lld",
                        dae_chunk_count, static_cast<long long>(data_len));
                    rc = [&] {
                        py::gil_scoped_release release;
                        return SQLPutData_ptr(hStmt, (SQLPOINTER)data.c_str(), data_len);
                    }();
                    if (!SQL_SUCCEEDED(rc) && rc != SQL_NEED_DATA) {
                        LOG("SQLExecuteMany: SQLPutData(bytes) failed - "
                            "chunk=%zu, rc=%d",
                            dae_chunk_count, rc);
                    }
                } else {
                    LOG("SQLExecuteMany: Unsupported DAE data type - chunk=%zu", dae_chunk_count);
                    return SQL_ERROR;
                }
                dae_chunk_count++;
            }
            LOG("SQLExecuteMany: DAE completed for row %zu - total_chunks=%zu, "
                "final_rc=%d",
                rowIndex, dae_chunk_count, rc);

            if (!SQL_SUCCEEDED(rc)) {
                LOG("SQLExecuteMany: DAE row %zu failed - rc=%d", rowIndex, rc);
                return rc;
            }
        }
        LOG("SQLExecuteMany: All DAE rows processed successfully - "
            "total_rows=%zu",
            rowCount);
        return SQL_SUCCESS;
    }
}

// Wrap SQLNumResultCols
SQLSMALLINT SQLNumResultCols_wrap(SqlHandlePtr statementHandle) {
    PERF_TIMER("SQLNumResultCols_wrap");
    LOG("SQLNumResultCols: Getting number of columns in result set for "
        "statement_handle=%p",
        (void*)statementHandle->get());
    if (!SQLNumResultCols_ptr) {
        LOG("SQLNumResultCols: Function pointer not initialized, loading "
            "driver");
        DriverLoader::getInstance().loadDriver();  // Load the driver
    }

    SQLSMALLINT columnCount;
    // TODO: Handle the return code
    SQLNumResultCols_ptr(statementHandle->get(), &columnCount);
    return columnCount;
}

// Wrap SQLDescribeCol
SQLRETURN SQLDescribeCol_wrap(SqlHandlePtr StatementHandle, py::list& ColumnMetadata) {
    PERF_TIMER("SQLDescribeCol_wrap");
    LOG("SQLDescribeCol: Getting column descriptions for statement_handle=%p",
        (void*)StatementHandle->get());
    if (!SQLDescribeCol_ptr) {
        LOG("SQLDescribeCol: Function pointer not initialized, loading driver");
        DriverLoader::getInstance().loadDriver();  // Load the driver
    }

    SQLSMALLINT ColumnCount;
    SQLRETURN retcode = SQLNumResultCols_ptr(StatementHandle->get(), &ColumnCount);
    if (!SQL_SUCCEEDED(retcode)) {
        LOG("SQLDescribeCol: Failed to get number of columns - SQLRETURN=%d", retcode);
        return retcode;
    }

    for (SQLUSMALLINT i = 1; i <= ColumnCount; ++i) {
        SQLWCHAR ColumnName[256];
        SQLSMALLINT NameLength;
        SQLSMALLINT DataType;
        SQLULEN ColumnSize;
        SQLSMALLINT DecimalDigits;
        SQLSMALLINT Nullable;

        retcode = SQLDescribeCol_ptr(StatementHandle->get(), i, ColumnName,
                                     sizeof(ColumnName) / sizeof(SQLWCHAR), &NameLength, &DataType,
                                     &ColumnSize, &DecimalDigits, &Nullable);

        if (SQL_SUCCEEDED(retcode)) {
            // Append a named py::dict to ColumnMetadata
            // TODO: Should we define a struct for this task instead of dict?
            ColumnMetadata.append(
                py::dict("ColumnName"_a = dupeSqlWCharAsUtf16Le(
                             ColumnName, std::min(static_cast<size_t>(NameLength),
                                                  (sizeof(ColumnName) / sizeof(SQLWCHAR)) - 1)),
                         "DataType"_a = DataType, "ColumnSize"_a = ColumnSize,
                         "DecimalDigits"_a = DecimalDigits, "Nullable"_a = Nullable));
        } else {
            return retcode;
        }
    }
    return SQL_SUCCESS;
}

SQLRETURN SQLSpecialColumns_wrap(SqlHandlePtr StatementHandle, SQLSMALLINT identifierType,
                                 const py::object& catalogObj, const py::object& schemaObj,
                                 const std::u16string& table, SQLSMALLINT scope,
                                 SQLSMALLINT nullable) {
    PERF_TIMER("SQLSpecialColumns_wrap");
    if (!SQLSpecialColumns_ptr) {
        ThrowStdException("SQLSpecialColumns function not loaded");
    }

    std::u16string catalog = catalogObj.is_none() ? u"" : catalogObj.cast<std::u16string>();
    std::u16string schema = schemaObj.is_none() ? u"" : schemaObj.cast<std::u16string>();

    py::gil_scoped_release release;
    return SQLSpecialColumns_ptr(StatementHandle->get(), identifierType,
                                 catalog.empty() ? nullptr
                                                 : reinterpretU16stringAsSqlWChar(catalog),
                                 catalog.empty() ? 0 : SQL_NTS,
                                 schema.empty() ? nullptr : reinterpretU16stringAsSqlWChar(schema),
                                 schema.empty() ? 0 : SQL_NTS,
                                 table.empty() ? nullptr : reinterpretU16stringAsSqlWChar(table),
                                 table.empty() ? 0 : SQL_NTS, scope, nullable);
}

// Wrap SQLFetch to retrieve rows
SQLRETURN SQLFetch_wrap(SqlHandlePtr StatementHandle) {
    PERF_TIMER("SQLFetch_wrap");
    LOG("SQLFetch: Fetching next row for statement_handle=%p", (void*)StatementHandle->get());
    if (!SQLFetch_ptr) {
        LOG("SQLFetch: Function pointer not initialized, loading driver");
        DriverLoader::getInstance().loadDriver();  // Load the driver
    }

    // Release the GIL during the blocking ODBC call
    py::gil_scoped_release release;
    return SQLFetch_ptr(StatementHandle->get());
}

// Non-static so it can be called from inline functions in header
py::object FetchLobColumnData(SQLHSTMT hStmt, SQLUSMALLINT colIndex, SQLSMALLINT cType,
                              bool isWideChar, bool isBinary, const std::string& charEncoding) {
    PERF_TIMER("FetchLobColumnData");
    std::vector<char> buffer;
    SQLRETURN ret = SQL_SUCCESS_WITH_INFO;
    int loopCount = 0;

    while (true) {
        ++loopCount;
        std::vector<char> chunk(DAE_CHUNK_SIZE, 0);
        SQLLEN actualRead = 0;
        {
            // Release the GIL during blocking SQLGetData LOB streaming
            py::gil_scoped_release release;
            ret = SQLGetData_ptr(hStmt, colIndex, cType, chunk.data(), DAE_CHUNK_SIZE, &actualRead);
        }

        if (ret == SQL_ERROR || !SQL_SUCCEEDED(ret) && ret != SQL_SUCCESS_WITH_INFO) {
            std::ostringstream oss;
            oss << "Error fetching LOB for column " << colIndex << ", cType=" << cType
                << ", loop=" << loopCount << ", SQLGetData return=" << ret;
            LOG("FetchLobColumnData: %s", oss.str().c_str());
            ThrowStdException(oss.str());
        }
        if (actualRead == SQL_NULL_DATA) {
            LOG("FetchLobColumnData: Column %d is NULL at loop %d", colIndex, loopCount);
            return py::none();
        }

        size_t bytesRead = 0;
        if (actualRead >= 0) {
            bytesRead = static_cast<size_t>(actualRead);
            if (bytesRead > DAE_CHUNK_SIZE) {
                bytesRead = DAE_CHUNK_SIZE;
            }
        } else {
            // fallback: use full buffer size if actualRead is unknown
            bytesRead = DAE_CHUNK_SIZE;
        }

        // For character data, trim trailing null terminators
        if (!isBinary && bytesRead > 0) {
            if (!isWideChar) {
                // Narrow characters
                while (bytesRead > 0 && chunk[bytesRead - 1] == '\0') {
                    --bytesRead;
                }
                if (bytesRead < DAE_CHUNK_SIZE) {
                    LOG("FetchLobColumnData: Trimmed null terminator from "
                        "narrow char data - loop=%d",
                        loopCount);
                }
            } else {
                // Wide characters
                size_t wcharSize = sizeof(SQLWCHAR);
                if (bytesRead >= wcharSize && (bytesRead % wcharSize == 0)) {
                    size_t wcharCount = bytesRead / wcharSize;
                    std::vector<SQLWCHAR> alignedBuf(wcharCount);
                    std::memcpy(alignedBuf.data(), chunk.data(), bytesRead);
                    while (wcharCount > 0 && alignedBuf[wcharCount - 1] == 0) {
                        --wcharCount;
                        bytesRead -= wcharSize;
                    }
                    if (bytesRead < DAE_CHUNK_SIZE) {
                        LOG("FetchLobColumnData: Trimmed null terminator from "
                            "wide char data - loop=%d",
                            loopCount);
                    }
                }
            }
        }
        if (bytesRead > 0) {
            buffer.insert(buffer.end(), chunk.begin(), chunk.begin() + bytesRead);
            LOG("FetchLobColumnData: Appended %zu bytes at loop %d", bytesRead, loopCount);
        }
        if (ret == SQL_SUCCESS) {
            LOG("FetchLobColumnData: SQL_SUCCESS - no more data at loop %d", loopCount);
            break;
        }
    }
    LOG("FetchLobColumnData: Total bytes collected=%zu for column %d", buffer.size(), colIndex);

    if (buffer.empty()) {
        if (isBinary) {
            return py::bytes("");
        }
        return py::str("");
    }
    if (isWideChar) {
        size_t wcharCount = buffer.size() / sizeof(SQLWCHAR);
        std::vector<SQLWCHAR> alignedBuf(wcharCount);
        std::memcpy(alignedBuf.data(), buffer.data(), buffer.size());
        return py::cast(dupeSqlWCharAsUtf16Le(alignedBuf.data(), wcharCount));
    }
    if (isBinary) {
        LOG("FetchLobColumnData: Returning binary data - %zu bytes for column "
            "%d",
            buffer.size(), colIndex);
        return py::bytes(buffer.data(), buffer.size());
    }

    // For SQL_C_CHAR data, decode using the appropriate encoding.
    const std::string effectiveCharEncoding = GetEffectiveCharDecoding(charEncoding);
    py::bytes raw_bytes(buffer.data(), buffer.size());
    try {
        py::object decoded = raw_bytes.attr("decode")(effectiveCharEncoding, "strict");
        LOG("FetchLobColumnData: Decoded narrow string with '%s' - %zu bytes -> %zu chars for "
            "column %d",
            effectiveCharEncoding.c_str(), buffer.size(), py::len(decoded), colIndex);
        return decoded;
    } catch (const py::error_already_set& e) {
        LOG_ERROR("FetchLobColumnData: Failed to decode with '%s' for column %d: %s",
                  effectiveCharEncoding.c_str(), colIndex, e.what());
        // Return raw bytes as fallback
        return raw_bytes;
    }
}

// Helper function to map sql_variant's underlying C type to SQL data type
// This allows sql_variant to reuse existing fetch logic for each data type
SQLSMALLINT MapVariantCTypeToSQLType(SQLLEN variantCType) {
    switch (variantCType) {
        case SQL_C_SLONG:
        case SQL_C_LONG:
            return SQL_INTEGER;
        case SQL_C_SSHORT:
        case SQL_C_SHORT:
            return SQL_SMALLINT;
        case SQL_C_SBIGINT:
            return SQL_BIGINT;
        case SQL_C_FLOAT:
            return SQL_REAL;
        case SQL_C_DOUBLE:
            return SQL_DOUBLE;
        case SQL_C_BIT:
            return SQL_BIT;
        case SQL_C_CHAR:
            return SQL_VARCHAR;
        case SQL_C_WCHAR:
            return SQL_WVARCHAR;
        case SQL_C_DATE:
        case SQL_C_TYPE_DATE:
            return SQL_TYPE_DATE;
        case SQL_C_TIME:
        case SQL_C_TYPE_TIME:
        case SQL_SS_VARIANT_TIME:
            return SQL_SS_TIME2;
        case SQL_C_TIMESTAMP:
        case SQL_C_TYPE_TIMESTAMP:
            return SQL_TYPE_TIMESTAMP;
        case SQL_C_BINARY:
            return SQL_VARBINARY;
        case SQL_C_GUID:
            return SQL_GUID;
        case SQL_C_NUMERIC:
            return SQL_NUMERIC;
        case SQL_C_TINYINT:
        case SQL_C_UTINYINT:
        case SQL_C_STINYINT:
            return SQL_TINYINT;
        default:
            // Unknown C type code - fallback to WVARCHAR for string conversion
            // Note: SQL Server enforces sql_variant restrictions at INSERT time, preventing
            // invalid types (text, ntext, image, timestamp, xml, MAX types, nested variants,
            // spatial types, hierarchyid, UDTs) from being stored. By the time we fetch data,
            // only valid base types exist. This default handles unmapped/future type codes.
            return SQL_WVARCHAR;
    }
}

// Helper function to check if a column requires SQLGetData streaming (LOB or sql_variant)
static inline bool IsLobOrVariantColumn(SQLSMALLINT dataType, SQLULEN columnSize) {
    return dataType == SQL_SS_VARIANT ||
           ((dataType == SQL_WVARCHAR || dataType == SQL_WLONGVARCHAR || dataType == SQL_VARCHAR ||
             dataType == SQL_LONGVARCHAR || dataType == SQL_VARBINARY ||
             dataType == SQL_LONGVARBINARY || dataType == SQL_SS_XML || dataType == SQL_SS_UDT) &&
            (columnSize == 0 || columnSize == SQL_NO_TOTAL || columnSize > SQL_MAX_LOB_SIZE));
}

// Helper function to retrieve column data
SQLRETURN SQLGetData_wrap(SqlHandlePtr StatementHandle, SQLUSMALLINT colCount, py::list& row,
                          const std::string& charEncoding = "utf-16le",
                          const std::string& wcharEncoding = "utf-16le",
                          int charCtype = SQL_C_WCHAR) {
    PERF_TIMER("SQLGetData_wrap");
    // Note: wcharEncoding parameter is reserved for future use
    // Currently WCHAR data always uses UTF-16LE for Windows compatibility
    (void)wcharEncoding;  // Suppress unused parameter warning

    LOG("SQLGetData: Getting data from %d columns for statement_handle=%p (charCtype=%d)", colCount,
        (void*)StatementHandle->get(), charCtype);
    if (!SQLGetData_ptr) {
        LOG("SQLGetData: Function pointer not initialized, loading driver");
        DriverLoader::getInstance().loadDriver();  // Load the driver
    }

    SQLRETURN ret = SQL_SUCCESS;
    SQLHSTMT hStmt = StatementHandle->get();

    // Cache decimal separator to avoid repeated system calls

    for (SQLSMALLINT i = 1; i <= colCount; ++i) {
        SQLWCHAR columnName[256];
        SQLSMALLINT columnNameLen;
        SQLSMALLINT dataType;
        SQLULEN columnSize;
        SQLSMALLINT decimalDigits;
        SQLSMALLINT nullable;

        ret = SQLDescribeCol_ptr(hStmt, i, columnName, sizeof(columnName) / sizeof(SQLWCHAR),
                                 &columnNameLen, &dataType, &columnSize, &decimalDigits, &nullable);
        if (!SQL_SUCCEEDED(ret)) {
            LOG("SQLGetData: Error retrieving metadata for column %d - "
                "SQLDescribeCol SQLRETURN=%d",
                i, ret);
            row.append(py::none());
            continue;
        }

        // Preprocess sql_variant: detect underlying type to route to correct conversion logic
        SQLSMALLINT effectiveDataType = dataType;
        if (dataType == SQL_SS_VARIANT) {
            // For sql_variant, we MUST call SQLGetData with SQL_C_BINARY (NULL buffer, len=0)
            // first. This serves two purposes:
            // 1. Detects NULL values via the indicator parameter
            // 2. Initializes the variant metadata in the ODBC driver, which is required for
            //    SQLColAttribute(SQL_CA_SS_VARIANT_TYPE) to return the correct underlying C type.
            //    Without this probe call, SQLColAttribute returns incorrect type codes.
            SQLLEN indicator;
            ret = SQLGetData_ptr(hStmt, i, SQL_C_BINARY, NULL, 0, &indicator);
            if (!SQL_SUCCEEDED(ret)) {
                LOG_ERROR("SQLGetData: Failed to probe sql_variant column %d - SQLRETURN=%d", i,
                          ret);
                row.append(py::none());
                continue;
            }
            if (indicator == SQL_NULL_DATA) {
                row.append(py::none());
                continue;
            }
            // Now retrieve the underlying C type
            SQLLEN variantCType = 0;
            ret =
                SQLColAttribute_ptr(hStmt, i, SQL_CA_SS_VARIANT_TYPE, NULL, 0, NULL, &variantCType);
            if (!SQL_SUCCEEDED(ret)) {
                LOG_ERROR("SQLGetData: Failed to get sql_variant underlying type for column %d", i);
                row.append(py::none());
                continue;
            }
            effectiveDataType = MapVariantCTypeToSQLType(variantCType);
            LOG("SQLGetData: sql_variant column %d has variantCType=%ld, mapped to SQL type %d", i,
                (long)variantCType, effectiveDataType);
        }

        switch (effectiveDataType) {
            case SQL_CHAR:
            case SQL_VARCHAR:
            case SQL_LONGVARCHAR: {
                // When charCtype == SQL_C_WCHAR, ask ODBC to convert VARCHAR
                // data to UTF-16. This avoids encoding mismatches on Windows
                // where the driver returns raw bytes in the server's native
                // code page (e.g. CP-1252) that may fail to decode as UTF-8.
                // When charCtype == SQL_C_CHAR, use the existing narrow-char
                // path with Python codec decoding.
                //
                // Exception: sql_variant columns always use SQL_C_CHAR.
                // The variant probe call (SQLGetData with SQL_C_BINARY) has
                // already consumed the column header, and requesting
                // SQL_C_WCHAR after the probe fails on some ODBC drivers
                // (notably unixODBC on Linux).  SQL_C_CHAR works reliably
                // because the Linux ODBC driver pre-converts to UTF-8.
                const bool isSqlVariant = (dataType == SQL_SS_VARIANT);
                const bool useWideChar = (charCtype == SQL_C_WCHAR) && !isSqlVariant;

                // For sql_variant, the SQL_C_CHAR path returns raw bytes in
                // the server's native encoding (Windows) or UTF-8
                // (Linux/macOS, driver converts).  Force "utf-8" so
                // GetEffectiveCharDecoding picks the right codec on each
                // platform, avoiding mismatch with the default "utf-16le"
                // encoding which is only valid for the SQL_C_WCHAR path.
                const std::string& effectiveCharEnc =
                    isSqlVariant ? std::string("utf-8") : charEncoding;

                if (columnSize == SQL_NO_TOTAL || columnSize == 0 ||
                    columnSize > SQL_MAX_LOB_SIZE) {
                    LOG("SQLGetData: Streaming LOB for column %d (%s) "
                        "- columnSize=%lu",
                        i, useWideChar ? "SQL_C_WCHAR" : "SQL_C_CHAR", (unsigned long)columnSize);
                    if (useWideChar) {
                        row.append(
                            FetchLobColumnData(hStmt, i, SQL_C_WCHAR, true, false, "utf-16le"));
                    } else {
                        row.append(FetchLobColumnData(hStmt, i, SQL_C_CHAR, false, false,
                                                      effectiveCharEnc));
                    }
                } else if (useWideChar) {
                    // Wide-char path: fetch VARCHAR data as SQL_C_WCHAR
                    uint64_t fetchBufferSize =
                        (columnSize + 1) * sizeof(SQLWCHAR);  // +1 for null terminator
                    std::vector<SQLWCHAR> dataBuffer(columnSize + 1);
                    SQLLEN dataLen;
                    ret = SQLGetData_ptr(hStmt, i, SQL_C_WCHAR, dataBuffer.data(), fetchBufferSize,
                                         &dataLen);
                    if (SQL_SUCCEEDED(ret)) {
                        if (dataLen > 0) {
                            uint64_t numCharsInData = dataLen / sizeof(SQLWCHAR);
                            if (numCharsInData < dataBuffer.size()) {
                                // Construct with explicit length: SQLGetData reports the
                                // exact number of characters via dataLen, so do not rely on
                                // null termination. This preserves embedded NULs and avoids
                                // any risk of reading past the valid range if the driver
                                // omits the terminator.
                                row.append(FetchText::from_utf16_native(
                                    reinterpret_cast<const char*>(dataBuffer.data()),
                                    static_cast<Py_ssize_t>(numCharsInData * sizeof(SQLWCHAR))));
                                LOG("SQLGetData: CHAR column %d fetched as WCHAR, "
                                    "length=%lu",
                                    i, (unsigned long)numCharsInData);
                            } else {
                                // Buffer too small, fallback to streaming
                                LOG("SQLGetData: CHAR column %d (WCHAR path) data "
                                    "truncated, using streaming LOB",
                                    i);
                                row.append(FetchLobColumnData(hStmt, i, SQL_C_WCHAR, true, false,
                                                              "utf-16le"));
                            }
                        } else if (dataLen == SQL_NULL_DATA) {
                            LOG("SQLGetData: Column %d is NULL (CHAR via WCHAR)", i);
                            row.append(py::none());
                        } else if (dataLen == 0) {
                            row.append(py::str(""));
                        } else if (dataLen == SQL_NO_TOTAL) {
                            // Driver cannot report total length up front; this is
                            // NOT a NULL value. Fall back to streaming via
                            // FetchLobColumnData (repeated SQLGetData chunks) so
                            // we don't silently lose data.
                            LOG("SQLGetData: SQL_NO_TOTAL for column %d (CHAR via WCHAR), "
                                "streaming via FetchLobColumnData",
                                i);
                            row.append(
                                FetchLobColumnData(hStmt, i, SQL_C_WCHAR, true, false, "utf-16le"));
                        } else if (dataLen < 0) {
                            LOG("SQLGetData: Unexpected negative data length "
                                "for column %d - dataType=%d, dataLen=%ld",
                                i, dataType, (long)dataLen);
                            ThrowStdException("SQLGetData returned an unexpected negative "
                                              "data length");
                        }
                    } else {
                        // Surface driver errors instead of silently returning NULL.
                        // Returning py::none() here would be indistinguishable from
                        // a genuine SQL NULL value to the Python caller and is a
                        // data-integrity risk.
                        LOG_ERROR("SQLGetData: Error retrieving data for column %d "
                                  "(CHAR via WCHAR) - SQLRETURN=%d",
                                  i, ret);
                        ThrowStdException("SQLGetData failed for CHAR/VARCHAR column "
                                          "fetched as SQL_C_WCHAR");
                    }
                } else {
                    // Allocate columnSize * 4 + 1 on ALL platforms (no #if guard).
                    //
                    // Why this differs from SQLBindColums / FetchBatchData:
                    // Those two functions use #if to apply *4 only on Linux/macOS,
                    // because on Windows with a non-UTF-8 collation (e.g. CP1252)
                    // each character occupies exactly 1 byte, so *1 suffices and
                    // saves memory across the entire batch (fetchSize × numCols
                    // buffers).
                    //
                    // SQLGetData_wrap allocates a single temporary buffer per
                    // column per row, so the over-allocation cost is negligible.
                    // Using *4 unconditionally here keeps the code simple and
                    // correct on every platform—including Windows with a UTF-8
                    // collation where multi-byte chars could otherwise cause
                    // truncation at the exact column boundary (e.g. CP1252 é in
                    // VARCHAR(10)).
                    uint64_t fetchBufferSize = columnSize * 4 + 1 /* null-termination */;
                    std::vector<SQLCHAR> dataBuffer(fetchBufferSize);
                    SQLLEN dataLen;
                    ret = SQLGetData_ptr(hStmt, i, SQL_C_CHAR, dataBuffer.data(), dataBuffer.size(),
                                         &dataLen);
                    if (SQL_SUCCEEDED(ret)) {
                        // columnSize is in chars, dataLen is in bytes
                        if (dataLen > 0) {
                            uint64_t numCharsInData = dataLen / sizeof(SQLCHAR);
                            if (numCharsInData < dataBuffer.size()) {
                                // SQLGetData will null-terminate the data
                                // Use Python's codec system to decode bytes.
                                const std::string decodeEncoding =
                                    GetEffectiveCharDecoding(effectiveCharEnc);
                                py::bytes raw_bytes(reinterpret_cast<char*>(dataBuffer.data()),
                                                    static_cast<size_t>(dataLen));
                                try {
                                    py::object decoded =
                                        raw_bytes.attr("decode")(decodeEncoding, "strict");
                                    row.append(decoded);
                                    LOG("SQLGetData: CHAR column %d decoded with '%s', %zu bytes "
                                        "-> %zu chars",
                                        i, decodeEncoding.c_str(), (size_t)dataLen,
                                        py::len(decoded));
                                } catch (const py::error_already_set& e) {
                                    LOG_ERROR(
                                        "SQLGetData: Failed to decode CHAR column %d with '%s': %s",
                                        i, decodeEncoding.c_str(), e.what());
                                    // Return raw bytes as fallback
                                    row.append(raw_bytes);
                                }
                            } else {
                                // Buffer too small, fallback to streaming
                                LOG("SQLGetData: CHAR column %d data truncated "
                                    "(buffer_size=%zu), using streaming LOB",
                                    i, dataBuffer.size());
                                row.append(FetchLobColumnData(hStmt, i, SQL_C_CHAR, false, false,
                                                              effectiveCharEnc));
                            }
                        } else if (dataLen == SQL_NULL_DATA) {
                            LOG("SQLGetData: Column %d is NULL (CHAR)", i);
                            row.append(py::none());
                        } else if (dataLen == 0) {
                            row.append(py::str(""));
                        } else if (dataLen == SQL_NO_TOTAL) {
                            // Driver cannot report total length up front; this is
                            // NOT a NULL value. Fall back to streaming via
                            // FetchLobColumnData (repeated SQLGetData chunks) so
                            // we don't silently lose data.
                            LOG("SQLGetData: SQL_NO_TOTAL for column %d (SQL_CHAR), "
                                "streaming via FetchLobColumnData",
                                i);
                            row.append(FetchLobColumnData(hStmt, i, SQL_C_CHAR, false, false,
                                                          effectiveCharEnc));
                        } else if (dataLen < 0) {
                            LOG("SQLGetData: Unexpected negative data length "
                                "for column %d - dataType=%d, dataLen=%ld",
                                i, dataType, (long)dataLen);
                            ThrowStdException("SQLGetData returned an unexpected negative "
                                              "data length");
                        }
                    } else {
                        // Surface driver errors instead of silently returning NULL.
                        // Returning py::none() here would be indistinguishable from
                        // a genuine SQL NULL value to the Python caller and is a
                        // data-integrity risk.
                        LOG_ERROR("SQLGetData: Error retrieving data for column %d "
                                  "(SQL_CHAR) - SQLRETURN=%d",
                                  i, ret);
                        ThrowStdException("SQLGetData failed for SQL_CHAR/VARCHAR column");
                    }
                }
                break;
            }
            case SQL_SS_XML: {
                LOG("SQLGetData: Streaming XML for column %d", i);
                row.append(FetchLobColumnData(hStmt, i, SQL_C_WCHAR, true, false, "utf-16le"));
                break;
            }
            case SQL_WCHAR:
            case SQL_WVARCHAR:
            case SQL_WLONGVARCHAR: {
                if (columnSize == SQL_NO_TOTAL || columnSize > 4000) {
                    LOG("SQLGetData: Streaming LOB for column %d (SQL_C_WCHAR) "
                        "- columnSize=%lu",
                        i, (unsigned long)columnSize);
                    row.append(FetchLobColumnData(hStmt, i, SQL_C_WCHAR, true, false, "utf-16le"));
                } else {
                    uint64_t fetchBufferSize =
                        (columnSize + 1) * sizeof(SQLWCHAR);  // +1 for null terminator
                    std::vector<SQLWCHAR> dataBuffer(columnSize + 1);
                    SQLLEN dataLen;
                    ret = SQLGetData_ptr(hStmt, i, SQL_C_WCHAR, dataBuffer.data(), fetchBufferSize,
                                         &dataLen);
                    if (SQL_SUCCEEDED(ret)) {
                        if (dataLen > 0) {
                            uint64_t numCharsInData = dataLen / sizeof(SQLWCHAR);
                            if (numCharsInData < dataBuffer.size()) {
                                // Construct with explicit length: SQLGetData reports the
                                // exact number of characters via dataLen, so do not rely on
                                // null termination. This preserves embedded NULs and avoids
                                // any risk of reading past the valid range if the driver
                                // omits the terminator.
                                row.append(FetchText::from_utf16_native(
                                    reinterpret_cast<const char*>(dataBuffer.data()),
                                    static_cast<Py_ssize_t>(numCharsInData * sizeof(SQLWCHAR))));
                                LOG("SQLGetData: Appended NVARCHAR string "
                                    "length=%lu for column %d",
                                    (unsigned long)numCharsInData, i);
                            } else {
                                // Buffer too small, fallback to streaming
                                LOG("SQLGetData: NVARCHAR column %d data "
                                    "truncated, using streaming LOB",
                                    i);
                                row.append(FetchLobColumnData(hStmt, i, SQL_C_WCHAR, true, false,
                                                              "utf-16le"));
                            }
                        } else if (dataLen == SQL_NULL_DATA) {
                            LOG("SQLGetData: Column %d is NULL (NVARCHAR)", i);
                            row.append(py::none());
                        } else if (dataLen == 0) {
                            row.append(py::str(""));
                        } else if (dataLen == SQL_NO_TOTAL) {
                            // Driver cannot report total length up front; this is
                            // NOT a NULL value. Fall back to streaming via
                            // FetchLobColumnData (repeated SQLGetData chunks) so
                            // we don't silently lose data.
                            LOG("SQLGetData: SQL_NO_TOTAL for column %d (NVARCHAR), "
                                "streaming via FetchLobColumnData",
                                i);
                            row.append(
                                FetchLobColumnData(hStmt, i, SQL_C_WCHAR, true, false, "utf-16le"));
                        } else if (dataLen < 0) {
                            LOG("SQLGetData: Unexpected negative data length "
                                "for column %d (NVARCHAR) - dataLen=%ld",
                                i, (long)dataLen);
                            ThrowStdException("SQLGetData returned an unexpected negative "
                                              "data length");
                        }
                    } else {
                        // Surface driver errors instead of silently returning NULL.
                        // Returning py::none() here would be indistinguishable from
                        // a genuine SQL NULL value to the Python caller and is a
                        // data-integrity risk.
                        LOG_ERROR("SQLGetData: Error retrieving data for column %d "
                                  "(NVARCHAR) - SQLRETURN=%d",
                                  i, ret);
                        ThrowStdException("SQLGetData failed for NVARCHAR column");
                    }
                }
                break;
            }
            case SQL_INTEGER: {
                SQLINTEGER intValue;
                SQLLEN indicator = 0;
                ret = SQLGetData_ptr(hStmt, i, SQL_C_LONG, &intValue, 0, &indicator);
                if (SQL_SUCCEEDED(ret) && indicator != SQL_NULL_DATA) {
                    row.append(static_cast<int>(intValue));
                } else {
                    row.append(py::none());
                }
                break;
            }
            case SQL_SMALLINT: {
                SQLSMALLINT smallIntValue;
                SQLLEN indicator = 0;
                ret = SQLGetData_ptr(hStmt, i, SQL_C_SHORT, &smallIntValue, 0, &indicator);
                if (SQL_SUCCEEDED(ret) && indicator == SQL_NULL_DATA) {
                    row.append(py::none());
                    break;
                }
                if (SQL_SUCCEEDED(ret)) {
                    row.append(static_cast<int>(smallIntValue));
                } else {
                    LOG("SQLGetData: Error retrieving SQL_SMALLINT for column "
                        "%d - SQLRETURN=%d",
                        i, ret);
                    row.append(py::none());
                }
                break;
            }
            case SQL_REAL: {
                SQLREAL realValue;
                SQLLEN indicator = 0;
                ret = SQLGetData_ptr(hStmt, i, SQL_C_FLOAT, &realValue, 0, &indicator);
                if (SQL_SUCCEEDED(ret) && indicator == SQL_NULL_DATA) {
                    row.append(py::none());
                    break;
                }
                if (SQL_SUCCEEDED(ret)) {
                    row.append(realValue);
                } else {
                    LOG("SQLGetData: Error retrieving SQL_REAL for column %d - "
                        "SQLRETURN=%d",
                        i, ret);
                    row.append(py::none());
                }
                break;
            }
            case SQL_DECIMAL:
            case SQL_NUMERIC: {
                SQLCHAR numericStr[MAX_DIGITS_IN_NUMERIC] = {0};
                SQLLEN indicator = 0;

                ret = SQLGetData_ptr(hStmt, i, SQL_C_CHAR, numericStr, sizeof(numericStr),
                                     &indicator);

                if (SQL_SUCCEEDED(ret)) {
                    try {
                        // Validate 'indicator' to avoid buffer overflow and
                        // fallback to a safe null-terminated read when length
                        // is unknown or out-of-range.
                        const char* cnum = reinterpret_cast<const char*>(numericStr);
                        size_t bufSize = sizeof(numericStr);
                        size_t safeLen = 0;

                        if (indicator > 0 && indicator <= static_cast<SQLLEN>(bufSize)) {
                            // indicator appears valid and within the buffer
                            // size
                            safeLen = static_cast<size_t>(indicator);
                        } else {
                            // indicator is unknown, zero, negative, or too
                            // large; determine length by searching for a
                            // terminating null (safe bounded scan)
                            for (size_t j = 0; j < bufSize; ++j) {
                                if (cnum[j] == '\0') {
                                    safeLen = j;
                                    break;
                                }
                            }
                            // if no null found, use the full buffer size as a
                            // conservative fallback
                            if (safeLen == 0 && bufSize > 0 && cnum[0] != '\0') {
                                safeLen = bufSize;
                            }
                        }
                        // Always use standard decimal point for Python Decimal
                        // parsing The decimal separator only affects display
                        // formatting, not parsing
                        py::object decimalObj =
                            PyTypeCache::get_decimal_class_obj()(py::str(cnum, safeLen));
                        row.append(decimalObj);
                    } catch (const py::error_already_set& e) {
                        // If conversion fails, append None
                        LOG("SQLGetData: Error converting to decimal for "
                            "column %d - %s",
                            i, e.what());
                        row.append(py::none());
                    }
                } else {
                    LOG("SQLGetData: Error retrieving SQL_NUMERIC/DECIMAL for "
                        "column %d - SQLRETURN=%d",
                        i, ret);
                    row.append(py::none());
                }
                break;
            }

            case SQL_DOUBLE:
            case SQL_FLOAT: {
                SQLDOUBLE doubleValue;
                SQLLEN indicator = 0;
                ret = SQLGetData_ptr(hStmt, i, SQL_C_DOUBLE, &doubleValue, 0, &indicator);
                if (SQL_SUCCEEDED(ret) && indicator == SQL_NULL_DATA) {
                    row.append(py::none());
                    break;
                }
                if (SQL_SUCCEEDED(ret)) {
                    row.append(doubleValue);
                } else {
                    LOG("SQLGetData: Error retrieving SQL_DOUBLE/FLOAT for "
                        "column %d - SQLRETURN=%d",
                        i, ret);
                    row.append(py::none());
                }
                break;
            }
            case SQL_BIGINT: {
                SQLBIGINT bigintValue;
                SQLLEN indicator = 0;
                ret = SQLGetData_ptr(hStmt, i, SQL_C_SBIGINT, &bigintValue, 0, &indicator);
                if (SQL_SUCCEEDED(ret) && indicator == SQL_NULL_DATA) {
                    row.append(py::none());
                    break;
                }
                if (SQL_SUCCEEDED(ret)) {
                    row.append(static_cast<long long>(bigintValue));
                } else {
                    LOG("SQLGetData: Error retrieving SQL_BIGINT for column %d "
                        "- SQLRETURN=%d",
                        i, ret);
                    row.append(py::none());
                }
                break;
            }
            case SQL_TYPE_DATE: {
                SQL_DATE_STRUCT dateValue;
                SQLLEN indicator = 0;
                ret = SQLGetData_ptr(hStmt, i, SQL_C_TYPE_DATE, &dateValue, sizeof(dateValue),
                                     &indicator);
                if (SQL_SUCCEEDED(ret) && indicator != SQL_NULL_DATA) {
                    row.append(
                        FetchTemporal::date(dateValue.year, dateValue.month, dateValue.day));
                } else {
                    row.append(py::none());
                }
                break;
            }
            case SQL_TYPE_TIME:
            case SQL_SS_TIME2: {
                SQL_SS_TIME2_STRUCT t2 = {};
                SQLLEN indicator = 0;
                ret = SQLGetData_ptr(hStmt, i, SQL_C_SS_TIME2, &t2, sizeof(t2), &indicator);
                if (SQL_SUCCEEDED(ret) && indicator != SQL_NULL_DATA) {
                    row.append(FetchTemporal::time(
                        t2.hour, t2.minute, t2.second, t2.fraction / 1000));  // ns to µs
                } else {
                    if (!SQL_SUCCEEDED(ret)) {
                        LOG("SQLGetData: Error retrieving SQL_SS_TIME2 for column "
                            "%d - SQLRETURN=%d",
                            i, ret);
                    }
                    row.append(py::none());
                }
                break;
            }
            case SQL_TIMESTAMP:
            case SQL_TYPE_TIMESTAMP:
            case SQL_DATETIME: {
                SQL_TIMESTAMP_STRUCT timestampValue;
                SQLLEN indicator = 0;
                ret = SQLGetData_ptr(hStmt, i, SQL_C_TYPE_TIMESTAMP, &timestampValue,
                                     sizeof(timestampValue), &indicator);
                if (SQL_SUCCEEDED(ret) && indicator == SQL_NULL_DATA) {
                    row.append(py::none());
                    break;
                }
                if (SQL_SUCCEEDED(ret)) {
                    row.append(FetchTemporal::datetime(
                        timestampValue.year, timestampValue.month, timestampValue.day,
                        timestampValue.hour, timestampValue.minute, timestampValue.second,
                        timestampValue.fraction / 1000  // Convert back ns to µs
                        ));
                } else {
                    LOG("SQLGetData: Error retrieving SQL_TYPE_TIMESTAMP for "
                        "column %d - SQLRETURN=%d",
                        i, ret);
                    row.append(py::none());
                }
                break;
            }
            case SQL_SS_TIMESTAMPOFFSET: {
                DateTimeOffset dtoValue;
                SQLLEN indicator;
                ret = SQLGetData_ptr(hStmt, i, SQL_C_SS_TIMESTAMPOFFSET, &dtoValue,
                                     sizeof(dtoValue), &indicator);
                if (SQL_SUCCEEDED(ret) && indicator != SQL_NULL_DATA) {
                    LOG("SQLGetData: Retrieved DATETIMEOFFSET for column %d - "
                        "%d-%d-%d %d:%d:%d, fraction_ns=%u, tz_hour=%d, "
                        "tz_minute=%d",
                        i, dtoValue.year, dtoValue.month, dtoValue.day, dtoValue.hour,
                        dtoValue.minute, dtoValue.second, dtoValue.fraction, dtoValue.timezone_hour,
                        dtoValue.timezone_minute);

                    int totalMinutes = dtoValue.timezone_hour * 60 + dtoValue.timezone_minute;
                    // Validating offset
                    if (totalMinutes < -24 * 60 || totalMinutes > 24 * 60) {
                        std::ostringstream oss;
                        oss << "Invalid timezone offset from "
                               "SQL_SS_TIMESTAMPOFFSET_STRUCT: "
                            << totalMinutes << " minutes for column " << i;
                        ThrowStdException(oss.str());
                    }
                    // Convert fraction from ns to µs
                    int microseconds = dtoValue.fraction / 1000;
                    py::object datetime_module = py::module_::import("datetime");
                    py::object tzinfo = datetime_module.attr("timezone")(
                        datetime_module.attr("timedelta")(py::arg("minutes") = totalMinutes));
                    py::object py_dt = PyTypeCache::get_datetime_class_obj()(
                        dtoValue.year, dtoValue.month, dtoValue.day, dtoValue.hour, dtoValue.minute,
                        dtoValue.second, microseconds, tzinfo);
                    row.append(py_dt);
                } else {
                    LOG("SQLGetData: Error fetching DATETIMEOFFSET for column "
                        "%d - SQLRETURN=%d, indicator=%ld",
                        i, ret, (long)indicator);
                    row.append(py::none());
                }
                break;
            }
            case SQL_SS_UDT:
            case SQL_BINARY:
            case SQL_VARBINARY:
            case SQL_LONGVARBINARY: {
                // Use streaming for large VARBINARY (columnSize unknown or >
                // 8000)
                if (columnSize == SQL_NO_TOTAL || columnSize == 0 || columnSize > 8000) {
                    LOG("SQLGetData: Streaming LOB for column %d "
                        "(SQL_C_BINARY) - columnSize=%lu",
                        i, (unsigned long)columnSize);
                    row.append(FetchLobColumnData(hStmt, i, SQL_C_BINARY, false, true, ""));
                } else {
                    // Small VARBINARY, fetch directly
                    std::vector<SQLCHAR> dataBuffer(columnSize);
                    SQLLEN dataLen;
                    ret = SQLGetData_ptr(hStmt, i, SQL_C_BINARY, dataBuffer.data(), columnSize,
                                         &dataLen);

                    if (SQL_SUCCEEDED(ret)) {
                        if (dataLen > 0) {
                            if (static_cast<size_t>(dataLen) <= columnSize) {
                                row.append(py::bytes(
                                    reinterpret_cast<const char*>(dataBuffer.data()), dataLen));
                            } else {
                                row.append(
                                    FetchLobColumnData(hStmt, i, SQL_C_BINARY, false, true, ""));
                            }
                        } else if (dataLen == SQL_NULL_DATA) {
                            row.append(py::none());
                        } else if (dataLen == 0) {
                            row.append(py::bytes(""));
                        } else {
                            std::ostringstream oss;
                            oss << "Unexpected negative length (" << dataLen
                                << ") returned by SQLGetData. ColumnID=" << i
                                << ", dataType=" << dataType << ", bufferSize=" << columnSize;
                            LOG("SQLGetData: %s", oss.str().c_str());
                            ThrowStdException(oss.str());
                        }
                    } else {
                        LOG("SQLGetData: Error retrieving VARBINARY data for "
                            "column %d - SQLRETURN=%d",
                            i, ret);
                        row.append(py::none());
                    }
                }
                break;
            }
            case SQL_TINYINT: {
                SQLCHAR tinyIntValue;
                SQLLEN indicator = 0;
                ret = SQLGetData_ptr(hStmt, i, SQL_C_TINYINT, &tinyIntValue, 0, &indicator);
                if (SQL_SUCCEEDED(ret) && indicator == SQL_NULL_DATA) {
                    row.append(py::none());
                    break;
                }
                if (SQL_SUCCEEDED(ret)) {
                    row.append(static_cast<int>(tinyIntValue));
                } else {
                    LOG("SQLGetData: Error retrieving SQL_TINYINT for column "
                        "%d - SQLRETURN=%d",
                        i, ret);
                    row.append(py::none());
                }
                break;
            }
            case SQL_BIT: {
                SQLCHAR bitValue;
                SQLLEN indicator = 0;
                ret = SQLGetData_ptr(hStmt, i, SQL_C_BIT, &bitValue, 0, &indicator);
                if (SQL_SUCCEEDED(ret) && indicator == SQL_NULL_DATA) {
                    row.append(py::none());
                    break;
                }
                if (SQL_SUCCEEDED(ret)) {
                    row.append(static_cast<bool>(bitValue));
                } else {
                    LOG("SQLGetData: Error retrieving SQL_BIT for column %d - "
                        "SQLRETURN=%d",
                        i, ret);
                    row.append(py::none());
                }
                break;
            }
#if (ODBCVER >= 0x0350)
            case SQL_GUID: {
                SQLGUID guidValue;
                SQLLEN indicator;
                ret =
                    SQLGetData_ptr(hStmt, i, SQL_C_GUID, &guidValue, sizeof(guidValue), &indicator);

                if (SQL_SUCCEEDED(ret) && indicator != SQL_NULL_DATA) {
                    std::vector<char> guid_bytes(16);
                    guid_bytes[0] = ((char*)&guidValue.Data1)[3];
                    guid_bytes[1] = ((char*)&guidValue.Data1)[2];
                    guid_bytes[2] = ((char*)&guidValue.Data1)[1];
                    guid_bytes[3] = ((char*)&guidValue.Data1)[0];
                    guid_bytes[4] = ((char*)&guidValue.Data2)[1];
                    guid_bytes[5] = ((char*)&guidValue.Data2)[0];
                    guid_bytes[6] = ((char*)&guidValue.Data3)[1];
                    guid_bytes[7] = ((char*)&guidValue.Data3)[0];
                    std::memcpy(&guid_bytes[8], guidValue.Data4, sizeof(guidValue.Data4));

                    py::bytes py_guid_bytes(guid_bytes.data(), guid_bytes.size());
                    py::object uuid_obj =
                        PyTypeCache::get_uuid_class_obj()(py::arg("bytes") = py_guid_bytes);
                    row.append(uuid_obj);
                } else if (indicator == SQL_NULL_DATA) {
                    row.append(py::none());
                } else {
                    LOG("SQLGetData: Error retrieving SQL_GUID for column %d - "
                        "SQLRETURN=%d, indicator=%ld",
                        i, ret, (long)indicator);
                    row.append(py::none());
                }
                break;
            }
#endif
            default:
                std::ostringstream errorString;
                errorString << "Unsupported data type for column - " << columnName << ", Type - "
                            << effectiveDataType << ", column ID - " << i;
                LOG("SQLGetData: %s", errorString.str().c_str());
                ThrowStdException(errorString.str());
                break;
        }
    }
    return ret;
}

SQLRETURN SQLFetchScroll_wrap(SqlHandlePtr StatementHandle, SQLSMALLINT FetchOrientation,
                              SQLLEN FetchOffset, py::list& row_data) {
    PERF_TIMER("SQLFetchScroll_wrap");
    LOG("SQLFetchScroll_wrap: Fetching with scroll orientation=%d, offset=%ld", FetchOrientation,
        (long)FetchOffset);
    if (!SQLFetchScroll_ptr) {
        LOG("SQLFetchScroll_wrap: Function pointer not initialized. Loading "
            "the driver.");
        DriverLoader::getInstance().loadDriver();  // Load the driver
    }

    // Unbind any columns from previous fetch operations to avoid memory
    // corruption
    SQLFreeStmt_ptr(StatementHandle->get(), SQL_UNBIND);

    // Perform scroll operation
    SQLRETURN ret;
    {
        // Release the GIL during the blocking ODBC fetch
        py::gil_scoped_release release;
        ret = SQLFetchScroll_ptr(StatementHandle->get(), FetchOrientation, FetchOffset);
    }

    // If successful and caller wants data, retrieve it
    if (SQL_SUCCEEDED(ret) && row_data.size() == 0) {
        // Get column count
        SQLSMALLINT colCount = SQLNumResultCols_wrap(StatementHandle);

        // Get the data in a consistent way with other fetch methods
        ret = SQLGetData_wrap(StatementHandle, colCount, row_data);
    }

    return ret;
}

// For column in the result set, binds a buffer to retrieve column data
// TODO: Move to anonymous namespace, since it is not used outside this file
SQLRETURN SQLBindColums(SQLHSTMT hStmt, ColumnBuffers& buffers, py::list& columnNames,
                        SQLUSMALLINT numCols, int fetchSize, int charCtype = SQL_C_WCHAR) {
    PERF_TIMER("SQLBindColums");
    SQLRETURN ret = SQL_SUCCESS;
    const bool useWideChar = (charCtype == SQL_C_WCHAR);
    // Bind columns based on their data types
    for (SQLUSMALLINT col = 1; col <= numCols; col++) {
        auto columnMeta = columnNames[col - 1].cast<py::dict>();
        SQLSMALLINT dataType = columnMeta["DataType"].cast<SQLSMALLINT>();
        SQLULEN columnSize = columnMeta["ColumnSize"].cast<SQLULEN>();

        switch (dataType) {
            case SQL_CHAR:
            case SQL_VARCHAR:
            case SQL_LONGVARCHAR: {
                HandleZeroColumnSizeAtFetch(columnSize);
                if (useWideChar) {
                    // Bind VARCHAR columns as SQL_C_WCHAR so the ODBC driver
                    // returns UTF-16 data, avoiding code-page decode issues.
                    uint64_t fetchBufferSize = columnSize + 1 /*null-terminator*/;
                    buffers.wcharBuffers[col - 1].resize(fetchSize * fetchBufferSize);
                    ret = SQLBindCol_ptr(
                        hStmt, col, SQL_C_WCHAR, buffers.wcharBuffers[col - 1].data(),
                        fetchBufferSize * sizeof(SQLWCHAR), buffers.indicators[col - 1].data());
                } else {
                    // Original narrow-char path
#if defined(__APPLE__) || defined(__linux__)
                    uint64_t fetchBufferSize = columnSize * 4 + 1 /*null-terminator*/;
#else
                    uint64_t fetchBufferSize = columnSize + 1 /*null-terminator*/;
#endif
                    buffers.charBuffers[col - 1].resize(fetchSize * fetchBufferSize);
                    ret = SQLBindCol_ptr(
                        hStmt, col, SQL_C_CHAR, buffers.charBuffers[col - 1].data(),
                        fetchBufferSize * sizeof(SQLCHAR), buffers.indicators[col - 1].data());
                }
                break;
            }
            case SQL_WCHAR:
            case SQL_WVARCHAR:
            case SQL_WLONGVARCHAR: {
                // TODO: handle variable length data correctly. This logic wont
                // suffice
                HandleZeroColumnSizeAtFetch(columnSize);
                uint64_t fetchBufferSize = columnSize + 1 /*null-terminator*/;
                buffers.wcharBuffers[col - 1].resize(fetchSize * fetchBufferSize);
                ret = SQLBindCol_ptr(hStmt, col, SQL_C_WCHAR, buffers.wcharBuffers[col - 1].data(),
                                     fetchBufferSize * sizeof(SQLWCHAR),
                                     buffers.indicators[col - 1].data());
                break;
            }
            case SQL_INTEGER:
                buffers.intBuffers[col - 1].resize(fetchSize);
                ret = SQLBindCol_ptr(hStmt, col, SQL_C_SLONG, buffers.intBuffers[col - 1].data(),
                                     sizeof(SQLINTEGER), buffers.indicators[col - 1].data());
                break;
            case SQL_SMALLINT:
                buffers.smallIntBuffers[col - 1].resize(fetchSize);
                ret = SQLBindCol_ptr(hStmt, col, SQL_C_SSHORT,
                                     buffers.smallIntBuffers[col - 1].data(), sizeof(SQLSMALLINT),
                                     buffers.indicators[col - 1].data());
                break;
            case SQL_TINYINT:
                buffers.charBuffers[col - 1].resize(fetchSize);
                ret = SQLBindCol_ptr(hStmt, col, SQL_C_TINYINT, buffers.charBuffers[col - 1].data(),
                                     sizeof(SQLCHAR), buffers.indicators[col - 1].data());
                break;
            case SQL_BIT:
                buffers.charBuffers[col - 1].resize(fetchSize);
                ret = SQLBindCol_ptr(hStmt, col, SQL_C_BIT, buffers.charBuffers[col - 1].data(),
                                     sizeof(SQLCHAR), buffers.indicators[col - 1].data());
                break;
            case SQL_REAL:
                buffers.realBuffers[col - 1].resize(fetchSize);
                ret = SQLBindCol_ptr(hStmt, col, SQL_C_FLOAT, buffers.realBuffers[col - 1].data(),
                                     sizeof(SQLREAL), buffers.indicators[col - 1].data());
                break;
            case SQL_DECIMAL:
            case SQL_NUMERIC:
                buffers.charBuffers[col - 1].resize(fetchSize * MAX_DIGITS_IN_NUMERIC);
                ret = SQLBindCol_ptr(hStmt, col, SQL_C_CHAR, buffers.charBuffers[col - 1].data(),
                                     MAX_DIGITS_IN_NUMERIC * sizeof(SQLCHAR),
                                     buffers.indicators[col - 1].data());
                break;
            case SQL_DOUBLE:
            case SQL_FLOAT:
                buffers.doubleBuffers[col - 1].resize(fetchSize);
                ret =
                    SQLBindCol_ptr(hStmt, col, SQL_C_DOUBLE, buffers.doubleBuffers[col - 1].data(),
                                   sizeof(SQLDOUBLE), buffers.indicators[col - 1].data());
                break;
            case SQL_TIMESTAMP:
            case SQL_TYPE_TIMESTAMP:
            case SQL_DATETIME:
                buffers.timestampBuffers[col - 1].resize(fetchSize);
                ret = SQLBindCol_ptr(
                    hStmt, col, SQL_C_TYPE_TIMESTAMP, buffers.timestampBuffers[col - 1].data(),
                    sizeof(SQL_TIMESTAMP_STRUCT), buffers.indicators[col - 1].data());
                break;
            case SQL_BIGINT:
                buffers.bigIntBuffers[col - 1].resize(fetchSize);
                ret =
                    SQLBindCol_ptr(hStmt, col, SQL_C_SBIGINT, buffers.bigIntBuffers[col - 1].data(),
                                   sizeof(SQLBIGINT), buffers.indicators[col - 1].data());
                break;
            case SQL_TYPE_DATE:
                buffers.dateBuffers[col - 1].resize(fetchSize);
                ret =
                    SQLBindCol_ptr(hStmt, col, SQL_C_TYPE_DATE, buffers.dateBuffers[col - 1].data(),
                                   sizeof(SQL_DATE_STRUCT), buffers.indicators[col - 1].data());
                break;
            case SQL_SS_TIME2:
                buffers.timeBuffers[col - 1].resize(fetchSize);
                ret =
                    SQLBindCol_ptr(hStmt, col, SQL_C_SS_TIME2, buffers.timeBuffers[col - 1].data(),
                                   sizeof(SQL_SS_TIME2_STRUCT), buffers.indicators[col - 1].data());
                break;
            case SQL_GUID:
                buffers.guidBuffers[col - 1].resize(fetchSize);
                ret = SQLBindCol_ptr(hStmt, col, SQL_C_GUID, buffers.guidBuffers[col - 1].data(),
                                     sizeof(SQLGUID), buffers.indicators[col - 1].data());
                break;
            case SQL_SS_UDT:
            case SQL_BINARY:
            case SQL_VARBINARY:
            case SQL_LONGVARBINARY:
                // TODO: handle variable length data correctly. This logic wont
                // suffice
                HandleZeroColumnSizeAtFetch(columnSize);
                buffers.charBuffers[col - 1].resize(fetchSize * columnSize);
                ret = SQLBindCol_ptr(hStmt, col, SQL_C_BINARY, buffers.charBuffers[col - 1].data(),
                                     columnSize, buffers.indicators[col - 1].data());
                break;
            case SQL_SS_TIMESTAMPOFFSET:
                buffers.datetimeoffsetBuffers[col - 1].resize(fetchSize);
                ret = SQLBindCol_ptr(hStmt, col, SQL_C_SS_TIMESTAMPOFFSET,
                                     buffers.datetimeoffsetBuffers[col - 1].data(),
                                     sizeof(DateTimeOffset) * fetchSize,
                                     buffers.indicators[col - 1].data());
                break;
            default:
                std::string columnName = columnMeta["ColumnName"].cast<std::string>();
                std::ostringstream errorString;
                errorString << "Unsupported data type for column - " << columnName.c_str()
                            << ", Type - " << dataType << ", column ID - " << col;
                LOG("SQLBindColums: %s", errorString.str().c_str());
                ThrowStdException(errorString.str());
                break;
        }
        if (!SQL_SUCCEEDED(ret)) {
            std::string columnName = columnMeta["ColumnName"].cast<std::string>();
            std::ostringstream errorString;
            errorString << "Failed to bind column - " << columnName.c_str() << ", Type - "
                        << dataType << ", column ID - " << col;
            LOG("SQLBindColums: %s", errorString.str().c_str());
            ThrowStdException(errorString.str());
            return ret;
        }
    }
    return ret;
}

// Fetch rows in batches
// TODO: Move to anonymous namespace, since it is not used outside this file
SQLRETURN FetchBatchData(SQLHSTMT hStmt, ColumnBuffers& buffers, py::list& columnNames,
                         py::list& rows, SQLUSMALLINT numCols, SQLULEN& numRowsFetched,
                         const std::vector<SQLUSMALLINT>& lobColumns,
                         const std::string& charEncoding = "utf-16le",
                         int charCtype = SQL_C_WCHAR) {
    PERF_TIMER("FetchBatchData");
    LOG("FetchBatchData: Fetching data in batches");
    SQLRETURN ret;
    {
        // Release the GIL during the blocking ODBC fetch
        py::gil_scoped_release release;
        PERF_TIMER("FetchBatchData::SQLFetchScroll_call");
        ret = SQLFetchScroll_ptr(hStmt, SQL_FETCH_NEXT, 0);
    }
    if (ret == SQL_NO_DATA) {
        LOG("FetchBatchData: No data to fetch");
        return ret;
    }
    if (!SQL_SUCCEEDED(ret)) {
        LOG("FetchBatchData: Error while fetching rows in batches - "
            "SQLRETURN=%d",
            ret);
        return ret;
    }
    // Pre-cache column metadata to avoid repeated dictionary lookups.
    // The vectors below are consumed later by construct_rows, so they are
    // declared at function scope; only the population work is wrapped in the
    // cache_column_metadata timer's block (an earlier version put the timer at
    // function scope, so it stayed active through construct_rows and made
    // metadata caching look like a dominant fetch cost).
    struct ColumnInfo {
        SQLSMALLINT dataType;
        SQLULEN columnSize;
        SQLULEN processedColumnSize;
        uint64_t fetchBufferSize;
        bool isLob;
    };
    const bool useWideChar = (charCtype == SQL_C_WCHAR);
    std::vector<ColumnInfo> columnInfos(numCols);
    // Performance: Build function pointer dispatch table (once per batch).
    // This eliminates the switch statement from the hot loop - 10,000 rows × 10
    // cols reduces from 100,000 switch evaluations to just 10 switch evaluations.
    std::vector<ColumnProcessor> columnProcessors(numCols);
    std::vector<ColumnInfoExt> columnInfosExt(numCols);
    // Compute effective char encoding once for the batch (same for all columns)
    const std::string effectiveCharEnc = GetEffectiveCharDecoding(charEncoding);

    {
        PERF_TIMER("FetchBatchData::cache_column_metadata");
        for (SQLUSMALLINT col = 0; col < numCols; col++) {
            const auto& columnMeta = columnNames[col].cast<py::dict>();
            columnInfos[col].dataType = columnMeta["DataType"].cast<SQLSMALLINT>();
            columnInfos[col].columnSize = columnMeta["ColumnSize"].cast<SQLULEN>();
            columnInfos[col].isLob =
                std::find(lobColumns.begin(), lobColumns.end(), col + 1) != lobColumns.end();
            columnInfos[col].processedColumnSize = columnInfos[col].columnSize;
            HandleZeroColumnSizeAtFetch(columnInfos[col].processedColumnSize);

            SQLSMALLINT dt = columnInfos[col].dataType;
            bool isCharType = (dt == SQL_CHAR || dt == SQL_VARCHAR || dt == SQL_LONGVARCHAR);

            if (isCharType && useWideChar) {
                // When VARCHAR is bound as SQL_C_WCHAR, buffer size is in SQLWCHAR
                // units (same as NVARCHAR). +1 for null terminator.
                columnInfos[col].fetchBufferSize = columnInfos[col].processedColumnSize + 1;
            } else {
                // On Linux/macOS, the ODBC driver returns UTF-8 for SQL_C_CHAR where
                // each character can be up to 4 bytes. Must match SQLBindColums buffer.
#if defined(__APPLE__) || defined(__linux__)
                if (isCharType) {
                    columnInfos[col].fetchBufferSize = columnInfos[col].processedColumnSize * 4 +
                                                       1;  // *4 for UTF-8, +1 for null terminator
                } else {
                    columnInfos[col].fetchBufferSize =
                        columnInfos[col].processedColumnSize + 1;  // +1 for null terminator
                }
#else
                columnInfos[col].fetchBufferSize =
                    columnInfos[col].processedColumnSize + 1;  // +1 for null terminator
#endif
            }
        }

        for (SQLUSMALLINT col = 0; col < numCols; col++) {
            // Populate extended column info for processors that need it
            columnInfosExt[col].dataType = columnInfos[col].dataType;
            columnInfosExt[col].columnSize = columnInfos[col].columnSize;
            columnInfosExt[col].processedColumnSize = columnInfos[col].processedColumnSize;
            columnInfosExt[col].fetchBufferSize = columnInfos[col].fetchBufferSize;
            columnInfosExt[col].isLob = columnInfos[col].isLob;
            columnInfosExt[col].charEncoding = effectiveCharEnc;
            columnInfosExt[col].isUtf8 = (effectiveCharEnc == "utf-8");
            // Set useWideChar for SQL_CHAR/VARCHAR columns when charCtype is SQL_C_WCHAR
            SQLSMALLINT dt = columnInfos[col].dataType;
            bool isCharType = (dt == SQL_CHAR || dt == SQL_VARCHAR || dt == SQL_LONGVARCHAR);
            columnInfosExt[col].useWideChar = (isCharType && useWideChar);

            // Map data type to processor function (switch executed once per column,
            // not per cell)
            SQLSMALLINT dataType = columnInfos[col].dataType;
            switch (dataType) {
                case SQL_INTEGER:
                    columnProcessors[col] = ColumnProcessors::ProcessInteger;
                    break;
                case SQL_SMALLINT:
                    columnProcessors[col] = ColumnProcessors::ProcessSmallInt;
                    break;
                case SQL_BIGINT:
                    columnProcessors[col] = ColumnProcessors::ProcessBigInt;
                    break;
                case SQL_TINYINT:
                    columnProcessors[col] = ColumnProcessors::ProcessTinyInt;
                    break;
                case SQL_BIT:
                    columnProcessors[col] = ColumnProcessors::ProcessBit;
                    break;
                case SQL_REAL:
                    columnProcessors[col] = ColumnProcessors::ProcessReal;
                    break;
                case SQL_DOUBLE:
                case SQL_FLOAT:
                    columnProcessors[col] = ColumnProcessors::ProcessDouble;
                    break;
                case SQL_CHAR:
                case SQL_VARCHAR:
                case SQL_LONGVARCHAR:
                    columnProcessors[col] = ColumnProcessors::ProcessChar;
                    break;
                case SQL_WCHAR:
                case SQL_WVARCHAR:
                case SQL_WLONGVARCHAR:
                    columnProcessors[col] = ColumnProcessors::ProcessWChar;
                    break;
                case SQL_SS_UDT:
                case SQL_BINARY:
                case SQL_VARBINARY:
                case SQL_LONGVARBINARY:
                    columnProcessors[col] = ColumnProcessors::ProcessBinary;
                    break;
                default:
                    // For complex types (Decimal, DateTime, Guid, etc.), set to
                    // nullptr and handle via fallback switch in the hot loop
                    columnProcessors[col] = nullptr;
                    break;
            }
        }
    }  // end cache_column_metadata timer scope

    // Performance: Single-phase row creation pattern
    // Create each row, fill it completely, then append to results list
    // This prevents data corruption (no partially-filled rows) and simplifies
    // error handling
    PERF_TIMER("FetchBatchData::construct_rows");
    PyObject* rowsList = rows.ptr();

    // RAII wrapper to ensure row cleanup on exception (CRITICAL: prevents
    // memory leak)
    struct RowGuard {
        PyObject* row;
        bool released;
        RowGuard() : row(nullptr), released(false) {}
        ~RowGuard() {
            if (row && !released)
                Py_DECREF(row);
        }
        void release() { released = true; }
    };

    for (SQLULEN i = 0; i < numRowsFetched; i++) {
        // Create row and immediately fill it (atomic operation per row)
        // This eliminates the two-phase pattern that could leave garbage rows
        // on exception
        RowGuard guard;
        guard.row = PyList_New(numCols);
        if (!guard.row) {
            throw std::runtime_error("Failed to allocate row list - memory allocation failure");
        }
        PyObject* row = guard.row;

        for (SQLUSMALLINT col = 1; col <= numCols; col++) {
            // Performance: Centralized NULL checking before calling processor
            // functions This eliminates redundant NULL checks inside each
            // processor and improves CPU branch prediction
            SQLLEN dataLen = buffers.indicators[col - 1][i];

            // Handle NULL and special indicator values first (applies to ALL
            // types)
            if (dataLen == SQL_NULL_DATA) {
                Py_INCREF(Py_None);
                PyList_SET_ITEM(row, col - 1, Py_None);
                continue;
            }
            if (dataLen == SQL_NO_TOTAL) {
                LOG("Cannot determine the length of the data. Returning NULL "
                    "value instead. Column ID - {}",
                    col);
                Py_INCREF(Py_None);
                PyList_SET_ITEM(row, col - 1, Py_None);
                continue;
            }

            // Performance: Use function pointer dispatch for simple types (fast
            // path) This eliminates the switch statement from hot loop -
            // reduces 100,000 switch evaluations (1000 rows × 10 cols × 10
            // types) to just 10 (setup only) Note: Processor functions no
            // longer need to check for NULL since we do it above
            if (columnProcessors[col - 1] != nullptr) {
                columnProcessors[col - 1](row, buffers, &columnInfosExt[col - 1], col, i, hStmt);
                continue;
            }

            // Fallback for complex types (Decimal, DateTime, Guid,
            // DateTimeOffset, etc.) that require pybind11 or special handling
            const ColumnInfoExt& colInfo = columnInfosExt[col - 1];
            SQLSMALLINT dataType = colInfo.dataType;

            // Additional validation for complex types
            if (dataLen == 0) {
                // Handle zero-length (non-NULL) data for complex types
                LOG("Column data length is 0 for complex datatype. Setting "
                    "None to the result row. Column ID - {}",
                    col);
                Py_INCREF(Py_None);
                PyList_SET_ITEM(row, col - 1, Py_None);
                continue;
            } else if (dataLen < 0) {
                // Negative value is unexpected, log column index, SQL type &
                // raise exception
                LOG("FetchBatchData: Unexpected negative data length - "
                    "column=%d, SQL_type=%d, dataLen=%ld",
                    col, dataType, (long)dataLen);
                ThrowStdException("Unexpected negative data length, check logs for details");
            }
            assert(dataLen > 0 && "Data length must be > 0");

            // Handle complex types that couldn't use function pointers
            switch (dataType) {
                case SQL_DECIMAL:
                case SQL_NUMERIC: {
                    try {
                        SQLLEN decimalDataLen = buffers.indicators[col - 1][i];
                        const char* rawData = reinterpret_cast<const char*>(
                            &buffers.charBuffers[col - 1][i * MAX_DIGITS_IN_NUMERIC]);

                        // Always use standard decimal point for Python Decimal
                        // parsing The decimal separator only affects display
                        // formatting, not parsing
                        PyObject* decimalObj =
                            PyTypeCache::get_decimal_class_obj()(py::str(rawData, decimalDataLen))
                                .release()
                                .ptr();
                        PyList_SET_ITEM(row, col - 1, decimalObj);
                    } catch (const py::error_already_set& e) {
                        // Handle the exception, e.g., log the error and set
                        // py::none()
                        LOG("Error converting to decimal: {}", e.what());
                        Py_INCREF(Py_None);
                        PyList_SET_ITEM(row, col - 1, Py_None);
                    }
                    break;
                }
                case SQL_TIMESTAMP:
                case SQL_TYPE_TIMESTAMP:
                case SQL_DATETIME: {
                    const SQL_TIMESTAMP_STRUCT& ts = buffers.timestampBuffers[col - 1][i];
                    py::object datetimeObj = FetchTemporal::datetime(
                        ts.year, ts.month, ts.day, ts.hour, ts.minute, ts.second,
                        ts.fraction / 1000);
                    PyList_SET_ITEM(row, col - 1, datetimeObj.release().ptr());
                    break;
                }
                case SQL_TYPE_DATE: {
                    const SQL_DATE_STRUCT& value = buffers.dateBuffers[col - 1][i];
                    py::object dateObj = FetchTemporal::date(value.year, value.month, value.day);
                    PyList_SET_ITEM(row, col - 1, dateObj.release().ptr());
                    break;
                }
                case SQL_SS_TIME2: {
                    const SQL_SS_TIME2_STRUCT& t2 = buffers.timeBuffers[col - 1][i];
                    py::object timeObj =
                        FetchTemporal::time(t2.hour, t2.minute, t2.second, t2.fraction / 1000);
                    PyList_SET_ITEM(row, col - 1, timeObj.release().ptr());
                    break;
                }
                case SQL_SS_TIMESTAMPOFFSET: {
                    SQLULEN rowIdx = i;
                    const DateTimeOffset& dtoValue = buffers.datetimeoffsetBuffers[col - 1][rowIdx];
                    SQLLEN indicator = buffers.indicators[col - 1][rowIdx];
                    if (indicator != SQL_NULL_DATA) {
                        int totalMinutes = dtoValue.timezone_hour * 60 + dtoValue.timezone_minute;
                        py::object datetime_module = py::module_::import("datetime");
                        py::object tzinfo = datetime_module.attr("timezone")(
                            datetime_module.attr("timedelta")(py::arg("minutes") = totalMinutes));
                        py::object py_dt = PyTypeCache::get_datetime_class_obj()(
                            dtoValue.year, dtoValue.month, dtoValue.day, dtoValue.hour,
                            dtoValue.minute, dtoValue.second,
                            dtoValue.fraction / 1000,  // ns → µs
                            tzinfo);
                        PyList_SET_ITEM(row, col - 1, py_dt.release().ptr());
                    } else {
                        Py_INCREF(Py_None);
                        PyList_SET_ITEM(row, col - 1, Py_None);
                    }
                    break;
                }
                case SQL_GUID: {
                    SQLLEN indicator = buffers.indicators[col - 1][i];
                    if (indicator == SQL_NULL_DATA) {
                        Py_INCREF(Py_None);
                        PyList_SET_ITEM(row, col - 1, Py_None);
                        break;
                    }
                    SQLGUID* guidValue = &buffers.guidBuffers[col - 1][i];
                    uint8_t reordered[16];
                    reordered[0] = ((char*)&guidValue->Data1)[3];
                    reordered[1] = ((char*)&guidValue->Data1)[2];
                    reordered[2] = ((char*)&guidValue->Data1)[1];
                    reordered[3] = ((char*)&guidValue->Data1)[0];
                    reordered[4] = ((char*)&guidValue->Data2)[1];
                    reordered[5] = ((char*)&guidValue->Data2)[0];
                    reordered[6] = ((char*)&guidValue->Data3)[1];
                    reordered[7] = ((char*)&guidValue->Data3)[0];
                    std::memcpy(reordered + 8, guidValue->Data4, 8);

                    py::bytes py_guid_bytes(reinterpret_cast<char*>(reordered), 16);
                    py::dict kwargs;
                    kwargs["bytes"] = py_guid_bytes;
                    py::object uuid_obj = PyTypeCache::get_uuid_class_obj()(**kwargs);
                    PyList_SET_ITEM(row, col - 1, uuid_obj.release().ptr());
                    break;
                }
                default: {
                    const auto& columnMeta = columnNames[col - 1].cast<py::dict>();
                    std::string columnName = columnMeta["ColumnName"].cast<std::string>();
                    std::ostringstream errorString;
                    errorString << "Unsupported data type for column - " << columnName.c_str()
                                << ", Type - " << dataType << ", column ID - " << col;
                    LOG("FetchBatchData: %s", errorString.str().c_str());
                    ThrowStdException(errorString.str());
                    break;
                }
            }
        }

        // Row is now fully populated - add it to results list atomically
        // This ensures no partially-filled rows exist in the list on exception
        if (PyList_Append(rowsList, row) < 0) {
            // RowGuard will clean up row automatically
            throw std::runtime_error("Failed to append row to results list - "
                                     "memory allocation failure");
        }
        // PyList_Append increments refcount, so we can release our reference
        // Mark guard as released so destructor doesn't double-free
        guard.release();
        Py_DECREF(row);
    }
    return ret;
}

// Given a list of columns that are a part of single row in the result set,
// calculates the max size of the row
// TODO: Move to anonymous namespace, since it is not used outside this file
size_t calculateRowSize(py::list& columnNames, SQLUSMALLINT numCols) {
    size_t rowSize = 0;
    for (SQLUSMALLINT col = 1; col <= numCols; col++) {
        auto columnMeta = columnNames[col - 1].cast<py::dict>();
        SQLSMALLINT dataType = columnMeta["DataType"].cast<SQLSMALLINT>();
        SQLULEN columnSize = columnMeta["ColumnSize"].cast<SQLULEN>();

        switch (dataType) {
            case SQL_CHAR:
            case SQL_VARCHAR:
            case SQL_LONGVARCHAR:
                rowSize += columnSize;
                break;
            case SQL_SS_XML:
            case SQL_WCHAR:
            case SQL_WVARCHAR:
            case SQL_WLONGVARCHAR:
                rowSize += columnSize * sizeof(SQLWCHAR);
                break;
            case SQL_INTEGER:
                rowSize += sizeof(SQLINTEGER);
                break;
            case SQL_SMALLINT:
                rowSize += sizeof(SQLSMALLINT);
                break;
            case SQL_REAL:
                rowSize += sizeof(SQLREAL);
                break;
            case SQL_FLOAT:
                rowSize += sizeof(SQLFLOAT);
                break;
            case SQL_DOUBLE:
                rowSize += sizeof(SQLDOUBLE);
                break;
            case SQL_DECIMAL:
            case SQL_NUMERIC:
                rowSize += MAX_DIGITS_IN_NUMERIC;
                break;
            case SQL_TIMESTAMP:
            case SQL_TYPE_TIMESTAMP:
            case SQL_DATETIME:
                rowSize += sizeof(SQL_TIMESTAMP_STRUCT);
                break;
            case SQL_BIGINT:
                rowSize += sizeof(SQLBIGINT);
                break;
            case SQL_TYPE_DATE:
                rowSize += sizeof(SQL_DATE_STRUCT);
                break;
            case SQL_SS_TIME2:
                rowSize += sizeof(SQL_SS_TIME2_STRUCT);
                break;
            case SQL_GUID:
                rowSize += sizeof(SQLGUID);
                break;
            case SQL_TINYINT:
            case SQL_BIT:
                rowSize += sizeof(SQLCHAR);
                break;
            case SQL_SS_UDT:
                rowSize += (static_cast<SQLLEN>(columnSize) == SQL_NO_TOTAL || columnSize == 0)
                               ? SQL_MAX_LOB_SIZE
                               : columnSize;
                break;
            case SQL_BINARY:
            case SQL_VARBINARY:
            case SQL_LONGVARBINARY:
                rowSize += columnSize;
                break;
            case SQL_SS_TIMESTAMPOFFSET:
                rowSize += sizeof(DateTimeOffset);
                break;
            default:
                std::string columnName = columnMeta["ColumnName"].cast<std::string>();
                std::ostringstream errorString;
                errorString << "Unsupported data type for column - " << columnName.c_str()
                            << ", Type - " << dataType << ", column ID - " << col;
                LOG("calculateRowSize: %s", errorString.str().c_str());
                ThrowStdException(errorString.str());
                break;
        }
    }
    return rowSize;
}

// FetchMany_wrap - Fetches multiple rows of data from the result set.
//
// @param StatementHandle: Handle to the statement from which data is to be
// fetched.
// @param rows: A Python list that will be populated with the fetched rows of
// data.
// @param fetchSize: The number of rows to fetch. Default value is 1.
//
// @return SQLRETURN: SQL_SUCCESS if data is fetched successfully,
//                    SQL_NO_DATA if there are no more rows to fetch,
//                    throws a runtime error if there is an error fetching data.
//
// This function assumes that the statement handle (hStmt) is already allocated
// and a query has been executed. It fetches the specified number of rows from
// the result set and populates the provided Python list with the row data. If
// there are no more rows to fetch, it returns SQL_NO_DATA. If an error occurs
// during fetching, it throws a runtime error.
SQLRETURN FetchMany_wrap(SqlHandlePtr StatementHandle, py::list& rows, int fetchSize,
                         const std::string& charEncoding = "utf-16le",
                         const std::string& wcharEncoding = "utf-16le",
                         int charCtype = SQL_C_WCHAR) {
    PERF_TIMER("FetchMany_wrap");
    // Issue #531: upgrade SQL_C_CHAR + utf-8 to SQL_C_WCHAR on Windows so the
    // driver does lossless UTF-16 conversion instead of returning ACP bytes.
    charCtype = EffectiveCharCtypeForFetch(charCtype, charEncoding);
    SQLRETURN ret;
    SQLHSTMT hStmt = StatementHandle->get();
    // Retrieve column count
    SQLSMALLINT numCols = SQLNumResultCols_wrap(StatementHandle);

    // Retrieve column metadata
    py::list columnNames;
    ret = SQLDescribeCol_wrap(StatementHandle, columnNames);
    if (!SQL_SUCCEEDED(ret)) {
        LOG("FetchMany_wrap: Failed to get column descriptions - SQLRETURN=%d", ret);
        return ret;
    }

    std::vector<SQLUSMALLINT> lobColumns;
    for (SQLSMALLINT i = 0; i < numCols; i++) {
        auto colMeta = columnNames[i].cast<py::dict>();
        SQLSMALLINT dataType = colMeta["DataType"].cast<SQLSMALLINT>();
        SQLULEN columnSize = colMeta["ColumnSize"].cast<SQLULEN>();

        if (IsLobOrVariantColumn(dataType, columnSize)) {
            lobColumns.push_back(i + 1);  // 1-based
        }
    }

    // Initialized to 0 for LOB path counter; overwritten by ODBC in non-LOB path;
    SQLULEN numRowsFetched = 0;
    // If we have LOBs → fall back to row-by-row fetch + SQLGetData_wrap
    if (!lobColumns.empty()) {
        LOG("FetchMany_wrap: LOB columns detected (%zu columns), using per-row "
            "SQLGetData path",
            lobColumns.size());
        while (numRowsFetched < (SQLULEN)fetchSize) {
            {
                // Release GIL during the blocking fetch
                py::gil_scoped_release release;
                ret = SQLFetch_ptr(hStmt);
            }
            if (ret == SQL_NO_DATA)
                break;
            if (!SQL_SUCCEEDED(ret))
                return ret;

            py::list row;
            SQLGetData_wrap(StatementHandle, numCols, row, charEncoding, wcharEncoding,
                            charCtype);  // <-- streams LOBs correctly
            rows.append(row);
            numRowsFetched++;
        }
        return SQL_SUCCESS;
    }

    // Initialize column buffers
    ColumnBuffers buffers(numCols, fetchSize);

    // Bind columns
    ret = SQLBindColums(hStmt, buffers, columnNames, numCols, fetchSize, charCtype);
    if (!SQL_SUCCEEDED(ret)) {
        LOG("FetchMany_wrap: Error when binding columns - SQLRETURN=%d", ret);
        return ret;
    }

    SQLSetStmtAttr_ptr(hStmt, SQL_ATTR_ROW_ARRAY_SIZE, (SQLPOINTER)(intptr_t)fetchSize, 0);
    SQLSetStmtAttr_ptr(hStmt, SQL_ATTR_ROWS_FETCHED_PTR, &numRowsFetched, 0);

    ret = FetchBatchData(hStmt, buffers, columnNames, rows, numCols, numRowsFetched, lobColumns,
                         charEncoding, charCtype);
    if (!SQL_SUCCEEDED(ret) && ret != SQL_NO_DATA) {
        LOG("FetchMany_wrap: Error when fetching data - SQLRETURN=%d", ret);
        return ret;
    }

    // Reset attributes before returning to avoid using stack pointers later
    SQLSetStmtAttr_ptr(hStmt, SQL_ATTR_ROW_ARRAY_SIZE, (SQLPOINTER)1, 0);
    SQLSetStmtAttr_ptr(hStmt, SQL_ATTR_ROWS_FETCHED_PTR, NULL, 0);

    // Unbind columns to allow subsequent fetchone() calls to use SQLGetData
    SQLFreeStmt_ptr(hStmt, SQL_UNBIND);

    return ret;
}

// GetDataVar - Progressively fetches variable-length column data using SQLGetData.
//
// Calls SQLGetData repeatedly, reallocating the buffer as needed, until all data is retrieved.
// Handles both fixed-size and unknown-size (SQL_NO_TOTAL) responses from the driver.
//
// @param hStmt: Statement handle
// @param colNumber: 1-based column index
// @param cType: SQL C data type (SQL_C_CHAR, SQL_C_WCHAR, or SQL_C_BINARY)
// @param dataVec: Reference to vector that will hold the fetched data (will be resized as needed)
// @param indicator: Pointer to indicator value (SQL_NULL_DATA for NULL, or data length)
//
// @return SQLRETURN: SQL_SUCCESS on success, or error code on failure
template <typename T>
SQLRETURN GetDataVar(SQLHSTMT hStmt, SQLUSMALLINT colNumber, SQLSMALLINT cType,
                     std::vector<T>& dataVec, SQLLEN* indicator) {
    size_t start = 0;
    size_t end = 0;

    // Determine null terminator size based on data type
    size_t sizeNullTerminator = 0;
    switch (cType) {
        case SQL_C_WCHAR:
        case SQL_C_CHAR:
            sizeNullTerminator = 1;
            break;
        case SQL_C_BINARY:
            sizeNullTerminator = 0;
            break;
        default:
            ThrowStdException("GetDataVar only supports SQL_C_CHAR, SQL_C_WCHAR, and SQL_C_BINARY");
    }

    // Ensure initial buffer has space for at least the null terminator
    if (dataVec.size() < sizeNullTerminator) {
        dataVec.resize(sizeNullTerminator);
    }

    while (true) {
        SQLLEN localInd = 0;
        SQLRETURN ret = SQLGetData_ptr(
            hStmt, colNumber, cType, reinterpret_cast<uint8_t*>(dataVec.data() + start),
            sizeof(T) * (dataVec.size() - start),  // Available buffer size from start position
            &localInd);

        // Handle NULL data
        if (localInd == SQL_NULL_DATA) {
            *indicator = SQL_NULL_DATA;
            return SQL_SUCCESS;
        }

        // Check for errors (excluding SQL_SUCCESS_WITH_INFO which means more data available)
        if (ret == SQL_ERROR || ret == SQL_INVALID_HANDLE) {
            return ret;
        }

        // SQL_SUCCESS or SQL_NO_DATA means we got all the data
        if (ret == SQL_SUCCESS || ret == SQL_NO_DATA) {
            if (localInd >= 0) {
                *indicator = static_cast<SQLLEN>(start) * sizeof(T) + localInd;
            } else {
                *indicator = localInd;  // Preserve SQL_NO_TOTAL or other negative values
            }
            break;
        }

        // SQL_SUCCESS_WITH_INFO means buffer was too small, need to continue fetching
        if (ret == SQL_SUCCESS_WITH_INFO) {
            // Determine how much more space we need
            if (localInd < 0) {
                // SQL_NO_TOTAL: driver doesn't know total size, double the buffer
                end = dataVec.size() * 2;
            } else {
                // Driver returned total size: allocate exactly what we need
                assert(localInd % sizeof(T) == 0);
                end = start + static_cast<size_t>(localInd) / sizeof(T) + sizeNullTerminator;
            }

            // The next read starts where the null terminator would have been placed
            start = dataVec.size() - sizeNullTerminator;

            // Resize buffer for next iteration
            dataVec.resize(end);
        } else {
            // Unexpected return code
            return ret;
        }
    }

    return SQL_SUCCESS;
}

struct FetchStateGuard {
    SQLHSTMT hStmt;

    FetchStateGuard(SQLHSTMT stmtHandle, SQLULEN* numRowsFetched, SQLULEN rowArraySize)
        : hStmt(stmtHandle) {
        SQLSetStmtAttr_ptr(hStmt, SQL_ATTR_ROW_ARRAY_SIZE, (SQLPOINTER)(intptr_t)rowArraySize, 0);
        SQLSetStmtAttr_ptr(hStmt, SQL_ATTR_ROWS_FETCHED_PTR, numRowsFetched, 0);
    }

    ~FetchStateGuard() {
        SQLSetStmtAttr_ptr(hStmt, SQL_ATTR_ROW_ARRAY_SIZE, (SQLPOINTER)1, 0);
        SQLSetStmtAttr_ptr(hStmt, SQL_ATTR_ROWS_FETCHED_PTR, NULL, 0);
        SQLFreeStmt_ptr(hStmt, SQL_UNBIND);
    }

    void setRowArraySize(SQLULEN rowArraySize) const {
        SQLSetStmtAttr_ptr(hStmt, SQL_ATTR_ROW_ARRAY_SIZE, (SQLPOINTER)(intptr_t)rowArraySize, 0);
    }
};

int32_t days_from_civil(int y, int m, int d) {
    // Implements the "days_from_civil" algorithm by Howard Hinnant
    // Returns number of days since Unix epoch (1970-01-01)
    y -= m <= 2;
    const int era = (y >= 0 ? y : y - 399) / 400;
    const unsigned yoe = static_cast<unsigned>(y - era * 400);            // [0, 399]
    const unsigned doy = (153 * (m + (m > 2 ? -3 : 9)) + 2) / 5 + d - 1;  // [0, 365]
    const unsigned doe = yoe * 365 + yoe / 4 - yoe / 100 + doy;           // [0, 146096]
    return era * 146097 + static_cast<int>(doe) - 719468;
}

SQLRETURN FetchArrowBatch_wrap(SqlHandlePtr StatementHandle, py::list& capsules,
                               int arrowBatchSize,
                               int charCtype) {
    PERF_TIMER("FetchArrowBatch_wrap");
    // Fetch narrow char data as SQL_C_CHAR if on Linux/macOS and configured by the user
    charCtype = EffectiveCharCtypeForFetch(charCtype, "utf-8");

    // An overly large fetch size doesn't seem to help performance
    int fetchSize = 64;

    SQLRETURN ret;
    SQLHSTMT hStmt = StatementHandle->get();
    // Retrieve column count
    SQLSMALLINT numCols = SQLNumResultCols_wrap(StatementHandle);
    if (numCols <= 0) {
        ThrowStdException("No active result set. Cannot fetch Arrow batch.");
    }

    // Retrieve column metadata
    py::list columnNames;
    ret = SQLDescribeCol_wrap(StatementHandle, columnNames);
    if (!SQL_SUCCEEDED(ret)) {
        LOG("Failed to get column descriptions");
        return ret;
    }

    bool hasLobColumns = false;

    std::vector<SQLSMALLINT> dataTypes(numCols);
    std::vector<SQLULEN> columnSizes(numCols);
    std::vector<bool> columnNullable(numCols);
    std::vector<bool> columnVarLen(numCols, false);
    std::vector<int64_t> nullCounts(numCols, 0);

    std::vector<std::unique_ptr<ArrowArrayPrivateData>> arrowArrayPrivateData(numCols);
    std::vector<std::unique_ptr<ArrowSchemaPrivateData>> arrowSchemaPrivateData(numCols);
    for (SQLSMALLINT i = 0; i < numCols; i++) {
        arrowArrayPrivateData[i] = std::make_unique<ArrowArrayPrivateData>();
        auto& arrowColumnProducer = arrowArrayPrivateData[i];
        arrowSchemaPrivateData[i] = std::make_unique<ArrowSchemaPrivateData>();

        auto colMeta = columnNames[i].cast<py::dict>();
        SQLSMALLINT dataType = colMeta["DataType"].cast<SQLSMALLINT>();
        SQLULEN columnSize = colMeta["ColumnSize"].cast<SQLULEN>();
        SQLSMALLINT nullable = colMeta["Nullable"].cast<SQLSMALLINT>();

        dataTypes[i] = dataType;
        columnSizes[i] = columnSize;
        columnNullable[i] = (nullable != SQL_NO_NULLS);

        if ((dataType == SQL_WVARCHAR || dataType == SQL_WLONGVARCHAR || dataType == SQL_VARCHAR ||
             dataType == SQL_LONGVARCHAR || dataType == SQL_VARBINARY ||
             dataType == SQL_LONGVARBINARY || dataType == SQL_SS_XML || dataType == SQL_SS_UDT) &&
            (columnSize == 0 || columnSize == SQL_NO_TOTAL || columnSize > SQL_MAX_LOB_SIZE)) {
            hasLobColumns = true;
            if (fetchSize > 1) {
                fetchSize = 1;  // LOBs require row-by-row fetch
            }
        }

        std::string columnName = colMeta["ColumnName"].cast<std::string>();
        size_t nameLen = columnName.length() + 1;
        arrowSchemaPrivateData[i]->name = std::make_unique<char[]>(nameLen);
        std::memcpy(arrowSchemaPrivateData[i]->name.get(), columnName.c_str(), nameLen);

        std::string format = "";
        switch (dataType) {
            case SQL_CHAR:
            case SQL_VARCHAR:
            case SQL_LONGVARCHAR:
            case SQL_SS_XML:
            case SQL_WCHAR:
            case SQL_WVARCHAR:
            case SQL_WLONGVARCHAR:
            case SQL_GUID:
                format = "U";
                arrowColumnProducer->varVal = std::make_unique<uint64_t[]>(arrowBatchSize + 1);
                arrowColumnProducer->varData.resize(arrowBatchSize * 42);
                columnVarLen[i] = true;
                // start at offset 0
                arrowColumnProducer->varVal[0] = 0;
                arrowColumnProducer->ptrValueBuffer = arrowColumnProducer->varVal.get();
                break;
            case SQL_SS_UDT:
            case SQL_BINARY:
            case SQL_VARBINARY:
            case SQL_LONGVARBINARY:
                format = "Z";
                arrowColumnProducer->varVal = std::make_unique<uint64_t[]>(arrowBatchSize + 1);
                arrowColumnProducer->varData.resize(arrowBatchSize * 42);
                columnVarLen[i] = true;
                // start at offset 0
                arrowColumnProducer->varVal[0] = 0;
                arrowColumnProducer->ptrValueBuffer = arrowColumnProducer->varVal.get();
                break;
            case SQL_TINYINT:
                format = "C";
                arrowColumnProducer->uint8Val = std::make_unique<uint8_t[]>(arrowBatchSize);
                arrowColumnProducer->ptrValueBuffer = arrowColumnProducer->uint8Val.get();
                break;
            case SQL_SMALLINT:
                format = "s";
                arrowColumnProducer->int16Val = std::make_unique<int16_t[]>(arrowBatchSize);
                arrowColumnProducer->ptrValueBuffer = arrowColumnProducer->int16Val.get();
                break;
            case SQL_INTEGER:
                format = "i";
                arrowColumnProducer->int32Val = std::make_unique<int32_t[]>(arrowBatchSize);
                arrowColumnProducer->ptrValueBuffer = arrowColumnProducer->int32Val.get();
                break;
            case SQL_BIGINT:
                format = "l";
                arrowColumnProducer->int64Val = std::make_unique<int64_t[]>(arrowBatchSize);
                arrowColumnProducer->ptrValueBuffer = arrowColumnProducer->int64Val.get();
                break;
            case SQL_REAL:
                format = "f";
                arrowColumnProducer->float32Val = std::make_unique<float[]>(arrowBatchSize);
                arrowColumnProducer->ptrValueBuffer = arrowColumnProducer->float32Val.get();
                break;
            case SQL_FLOAT:
            case SQL_DOUBLE:
                format = "g";
                arrowColumnProducer->float64Val = std::make_unique<double[]>(arrowBatchSize);
                arrowColumnProducer->ptrValueBuffer = arrowColumnProducer->float64Val.get();
                break;
            case SQL_DECIMAL:
            case SQL_NUMERIC: {
                std::ostringstream formatStream;
                formatStream << "d:" << columnSize << ","
                             << colMeta["DecimalDigits"].cast<SQLSMALLINT>();
                std::string formatStr = formatStream.str();
                size_t formatLen = formatStr.length() + 1;
                arrowSchemaPrivateData[i]->format = std::make_unique<char[]>(formatLen);
                std::memcpy(arrowSchemaPrivateData[i]->format.get(), formatStr.c_str(), formatLen);
                format = arrowSchemaPrivateData[i]->format.get();
                arrowColumnProducer->decimalVal = std::make_unique<Int128_t[]>(arrowBatchSize);
                arrowColumnProducer->ptrValueBuffer = arrowColumnProducer->decimalVal.get();
                break;
            }
            case SQL_TIMESTAMP:
            case SQL_TYPE_TIMESTAMP:
            case SQL_DATETIME:
                format = "tsu:";
                arrowColumnProducer->tsMicroVal = std::make_unique<int64_t[]>(arrowBatchSize);
                arrowColumnProducer->ptrValueBuffer = arrowColumnProducer->tsMicroVal.get();
                break;
            case SQL_SS_TIMESTAMPOFFSET:
                format = "tsu:+00:00";
                arrowColumnProducer->tsMicroVal = std::make_unique<int64_t[]>(arrowBatchSize);
                arrowColumnProducer->ptrValueBuffer = arrowColumnProducer->tsMicroVal.get();
                break;
            case SQL_TYPE_DATE:
                format = "tdD";
                arrowColumnProducer->dateVal = std::make_unique<int32_t[]>(arrowBatchSize);
                arrowColumnProducer->ptrValueBuffer = arrowColumnProducer->dateVal.get();
                break;
            case SQL_SS_TIME2:
                format = "ttn";
                arrowColumnProducer->timeNanoVal = std::make_unique<int64_t[]>(arrowBatchSize);
                arrowColumnProducer->ptrValueBuffer = arrowColumnProducer->timeNanoVal.get();
                break;
            case SQL_BIT:
                format = "b";
                arrowColumnProducer->bitVal = std::make_unique<uint8_t[]>((arrowBatchSize + 7) / 8);
                std::memset(arrowColumnProducer->bitVal.get(), 0, (arrowBatchSize + 7) / 8);
                arrowColumnProducer->ptrValueBuffer = arrowColumnProducer->bitVal.get();
                break;
            default:
                std::ostringstream errorString;
                errorString << "Unsupported data type for Arrow batch fetch for column - "
                            << columnName.c_str() << ", Type - " << dataType << ", column ID - "
                            << (i + 1);
                LOG(errorString.str().c_str());
                ThrowStdException(errorString.str());
                break;
        }

        // Store format string if not already stored.
        // For non-decimal types, format is now a static string.
        if (!arrowSchemaPrivateData[i]->format) {
            size_t formatLen = format.length() + 1;
            arrowSchemaPrivateData[i]->format = std::make_unique<char[]>(formatLen);
            std::memcpy(arrowSchemaPrivateData[i]->format.get(), format.c_str(), formatLen);
        }

        arrowColumnProducer->valid = std::make_unique<uint8_t[]>((arrowBatchSize + 7) / 8);
        // Initialize validity bitmap to all valid
        std::memset(arrowColumnProducer->valid.get(), 0xFF, (arrowBatchSize + 7) / 8);
    }

    // Initialize column buffers
    ColumnBuffers buffers(numCols, fetchSize);

    if (!hasLobColumns && fetchSize > 0) {
        ret = SQLBindColums(hStmt, buffers, columnNames, numCols, fetchSize, charCtype);
        if (!SQL_SUCCEEDED(ret)) {
            LOG("Error when binding columns");
            return ret;
        }
    }

    SQLULEN numRowsFetched = 0;
    FetchStateGuard fetchStateGuard(hStmt, &numRowsFetched, fetchSize);

    int idxRowArrow = 0;

    while (idxRowArrow < arrowBatchSize) {
        int spaceLeftInArrowBatch = arrowBatchSize - idxRowArrow;
        if (fetchSize > spaceLeftInArrowBatch) {
            // Adjust fetch size for final batch to avoid overfetching
            fetchStateGuard.setRowArraySize(spaceLeftInArrowBatch);
        }
        {
            // Release GIL during the blocking ODBC fetch
            py::gil_scoped_release release;
            ret = SQLFetch_ptr(hStmt);
        }
        if (ret == SQL_NO_DATA) {
            ret = SQL_SUCCESS;  // Normal completion
            break;
        }
        if (!SQL_SUCCEEDED(ret)) {
            LOG("Error while fetching rows in batches");
            return ret;
        }
        // numRowsFetched is the SQL_ATTR_ROWS_FETCHED_PTR attribute.
        // It'll be populated by SQLFetch
        assert(numRowsFetched + idxRowArrow <= static_cast<SQLULEN>(arrowBatchSize));
        for (SQLULEN idxRowSql = 0; idxRowSql < numRowsFetched; idxRowSql++) {
            for (SQLUSMALLINT idxCol = 0; idxCol < numCols; idxCol++) {
                auto& arrowColumnProducer = arrowArrayPrivateData[idxCol];
                auto dataType = dataTypes[idxCol];
                auto columnSize = columnSizes[idxCol];

                if (hasLobColumns) {
                    assert(idxRowSql == 0 && "GetData only works one row at a time");

                    switch (dataType) {
                        case SQL_SS_UDT:
                        case SQL_BINARY:
                        case SQL_VARBINARY:
                        case SQL_LONGVARBINARY: {
                            ret = GetDataVar(hStmt, idxCol + 1, SQL_C_BINARY,
                                             buffers.charBuffers[idxCol],
                                             buffers.indicators[idxCol].data());
                            if (!SQL_SUCCEEDED(ret)) {
                                LOG("Error fetching BINARY LOB for column %d", idxCol + 1);
                                return ret;
                            }
                            break;
                        }
                        case SQL_CHAR:
                        case SQL_VARCHAR:
                        case SQL_LONGVARCHAR: {
                            if (charCtype == SQL_C_CHAR) {
                                ret = GetDataVar(hStmt, idxCol + 1, SQL_C_CHAR,
                                                 buffers.charBuffers[idxCol],
                                                 buffers.indicators[idxCol].data());
                                if (!SQL_SUCCEEDED(ret)) {
                                    LOG("Error fetching CHAR LOB data for column %d", idxCol + 1);
                                    return ret;
                                }
                                break;
                            }
                            // else fall through to SQL_C_WCHAR case
                        }
                        case SQL_SS_XML:
                        case SQL_WCHAR:
                        case SQL_WVARCHAR:
                        case SQL_WLONGVARCHAR: {
                            ret = GetDataVar(hStmt, idxCol + 1, SQL_C_WCHAR,
                                             buffers.wcharBuffers[idxCol],
                                             buffers.indicators[idxCol].data());
                            if (!SQL_SUCCEEDED(ret)) {
                                LOG("Error fetching WCHAR LOB data for column %d", idxCol + 1);
                                return ret;
                            }
                            break;
                        }
                        case SQL_INTEGER: {
                            buffers.intBuffers[idxCol].resize(1);
                            ret = SQLGetData_ptr(
                                hStmt, idxCol + 1, SQL_C_SLONG, buffers.intBuffers[idxCol].data(),
                                sizeof(SQLINTEGER), buffers.indicators[idxCol].data());
                            if (!SQL_SUCCEEDED(ret)) {
                                LOG("Error fetching SLONG data for column %d", idxCol + 1);
                                return ret;
                            }
                            break;
                        }
                        case SQL_SMALLINT: {
                            buffers.smallIntBuffers[idxCol].resize(1);
                            ret = SQLGetData_ptr(hStmt, idxCol + 1, SQL_C_SSHORT,
                                                 buffers.smallIntBuffers[idxCol].data(),
                                                 sizeof(SQLSMALLINT),
                                                 buffers.indicators[idxCol].data());
                            if (!SQL_SUCCEEDED(ret)) {
                                LOG("Error fetching SSHORT data for column %d", idxCol + 1);
                                return ret;
                            }
                            break;
                        }
                        case SQL_TINYINT: {
                            buffers.charBuffers[idxCol].resize(1);
                            ret =
                                SQLGetData_ptr(hStmt, idxCol + 1, SQL_C_TINYINT,
                                               buffers.charBuffers[idxCol].data(), sizeof(SQLCHAR),
                                               buffers.indicators[idxCol].data());
                            if (!SQL_SUCCEEDED(ret)) {
                                LOG("Error fetching TINYINT data for column %d", idxCol + 1);
                                return ret;
                            }
                            break;
                        }
                        case SQL_BIT: {
                            buffers.charBuffers[idxCol].resize(1);
                            ret = SQLGetData_ptr(
                                hStmt, idxCol + 1, SQL_C_BIT, buffers.charBuffers[idxCol].data(),
                                sizeof(SQLCHAR), buffers.indicators[idxCol].data());
                            if (!SQL_SUCCEEDED(ret)) {
                                LOG("Error fetching BIT data for column %d", idxCol + 1);
                                return ret;
                            }
                            break;
                        }
                        case SQL_REAL: {
                            buffers.realBuffers[idxCol].resize(1);
                            ret = SQLGetData_ptr(
                                hStmt, idxCol + 1, SQL_C_FLOAT, buffers.realBuffers[idxCol].data(),
                                sizeof(SQLREAL), buffers.indicators[idxCol].data());
                            if (!SQL_SUCCEEDED(ret)) {
                                LOG("Error fetching FLOAT data for column %d", idxCol + 1);
                                return ret;
                            }
                            break;
                        }
                        case SQL_DECIMAL:
                        case SQL_NUMERIC: {
                            buffers.charBuffers[idxCol].resize(MAX_DIGITS_IN_NUMERIC);
                            ret = SQLGetData_ptr(hStmt, idxCol + 1, SQL_C_CHAR,
                                                 buffers.charBuffers[idxCol].data(),
                                                 MAX_DIGITS_IN_NUMERIC * sizeof(SQLCHAR),
                                                 buffers.indicators[idxCol].data());
                            if (!SQL_SUCCEEDED(ret)) {
                                LOG("Error fetching CHAR data for column %d", idxCol + 1);
                                return ret;
                            }
                            break;
                        }
                        case SQL_DOUBLE:
                        case SQL_FLOAT: {
                            buffers.doubleBuffers[idxCol].resize(1);
                            ret = SQLGetData_ptr(hStmt, idxCol + 1, SQL_C_DOUBLE,
                                                 buffers.doubleBuffers[idxCol].data(),
                                                 sizeof(SQLDOUBLE),
                                                 buffers.indicators[idxCol].data());
                            if (!SQL_SUCCEEDED(ret)) {
                                LOG("Error fetching DOUBLE data for column %d", idxCol + 1);
                                return ret;
                            }
                            break;
                        }
                        case SQL_TIMESTAMP:
                        case SQL_TYPE_TIMESTAMP:
                        case SQL_DATETIME: {
                            buffers.timestampBuffers[idxCol].resize(1);
                            ret = SQLGetData_ptr(hStmt, idxCol + 1, SQL_C_TYPE_TIMESTAMP,
                                                 buffers.timestampBuffers[idxCol].data(),
                                                 sizeof(SQL_TIMESTAMP_STRUCT),
                                                 buffers.indicators[idxCol].data());
                            if (!SQL_SUCCEEDED(ret)) {
                                LOG("Error fetching TYPE_TIMESTAMP data for column %d", idxCol + 1);
                                return ret;
                            }
                            break;
                        }
                        case SQL_BIGINT: {
                            buffers.bigIntBuffers[idxCol].resize(1);
                            ret = SQLGetData_ptr(hStmt, idxCol + 1, SQL_C_SBIGINT,
                                                 buffers.bigIntBuffers[idxCol].data(),
                                                 sizeof(SQLBIGINT),
                                                 buffers.indicators[idxCol].data());
                            if (!SQL_SUCCEEDED(ret)) {
                                LOG("Error fetching SBIGINT data for column %d", idxCol + 1);
                                return ret;
                            }
                            break;
                        }
                        case SQL_TYPE_DATE: {
                            buffers.dateBuffers[idxCol].resize(1);
                            ret = SQLGetData_ptr(hStmt, idxCol + 1, SQL_C_TYPE_DATE,
                                                 buffers.dateBuffers[idxCol].data(),
                                                 sizeof(SQL_DATE_STRUCT),
                                                 buffers.indicators[idxCol].data());
                            if (!SQL_SUCCEEDED(ret)) {
                                LOG("Error fetching TYPE_DATE data for column %d", idxCol + 1);
                                return ret;
                            }
                            break;
                        }
                        case SQL_SS_TIME2: {
                            buffers.timeBuffers[idxCol].resize(1);
                            ret = SQLGetData_ptr(hStmt, idxCol + 1, SQL_C_SS_TIME2,
                                                 buffers.timeBuffers[idxCol].data(),
                                                 sizeof(SQL_SS_TIME2_STRUCT),
                                                 buffers.indicators[idxCol].data());
                            if (!SQL_SUCCEEDED(ret)) {
                                LOG("Error fetching TYPE_TIME data for column %d", idxCol + 1);
                                return ret;
                            }
                            break;
                        }
                        case SQL_GUID: {
                            buffers.guidBuffers[idxCol].resize(1);
                            ret = SQLGetData_ptr(
                                hStmt, idxCol + 1, SQL_C_GUID, buffers.guidBuffers[idxCol].data(),
                                sizeof(SQLGUID), buffers.indicators[idxCol].data());
                            if (!SQL_SUCCEEDED(ret)) {
                                LOG("Error fetching GUID data for column %d", idxCol + 1);
                                return ret;
                            }
                            break;
                        }
                        case SQL_SS_TIMESTAMPOFFSET: {
                            buffers.datetimeoffsetBuffers[idxCol].resize(1);
                            ret = SQLGetData_ptr(hStmt, idxCol + 1, SQL_C_SS_TIMESTAMPOFFSET,
                                                 buffers.datetimeoffsetBuffers[idxCol].data(),
                                                 sizeof(DateTimeOffset),
                                                 buffers.indicators[idxCol].data());
                            if (!SQL_SUCCEEDED(ret)) {
                                LOG("Error fetching SS_TIMESTAMPOFFSET data for column %d",
                                    idxCol + 1);
                                return ret;
                            }
                            break;
                        }
                        default: {
                            std::ostringstream errorString;
                            errorString << "Unsupported data type for column ID - " << (idxCol + 1)
                                        << ", Type - " << dataType;
                            LOG("SQLGetData: %s", errorString.str().c_str());
                            ThrowStdException(errorString.str());
                            break;
                        }
                    }
                }

                SQLLEN indicator = buffers.indicators[idxCol][idxRowSql];

                if (indicator == SQL_NULL_DATA) {
                    // Mark as null in validity bitmap
                    size_t bytePos = idxRowArrow / 8;
                    size_t bitPos = idxRowArrow % 8;
                    arrowColumnProducer->valid[bytePos] &= ~(1 << bitPos);

                    // Value buffer for variable length data types needs to be set appropriately
                    // as it will be used by the next non null value
                    switch (dataType) {
                        case SQL_CHAR:
                        case SQL_VARCHAR:
                        case SQL_LONGVARCHAR:
                        case SQL_SS_XML:
                        case SQL_WCHAR:
                        case SQL_WVARCHAR:
                        case SQL_WLONGVARCHAR:
                        case SQL_GUID:
                        case SQL_SS_UDT:
                        case SQL_BINARY:
                        case SQL_VARBINARY:
                        case SQL_LONGVARBINARY:
                            arrowColumnProducer->varVal[idxRowArrow + 1] =
                                arrowColumnProducer->varVal[idxRowArrow];
                            break;
                        default:
                            break;
                    }

                    nullCounts[idxCol] += 1;
                    continue;
                } else if (indicator < 0) {
                    // Negative value is unexpected, log column index, SQL type & raise exception
                    LOG("Unexpected negative data length. Column ID - %d, SQL Type - %d, Data "
                        "Length - %lld",
                        idxCol + 1, dataType, (long long)indicator);
                    ThrowStdException("Unexpected negative data length.");
                }
                auto dataLen = static_cast<uint64_t>(indicator);

                switch (dataType) {
                    case SQL_SS_UDT:
                    case SQL_BINARY:
                    case SQL_VARBINARY:
                    case SQL_LONGVARBINARY: {
                        uint64_t fetchBufferSize = columnSize /* bytes are not null terminated */;
                        auto target_vec = &arrowColumnProducer->varData;
                        auto start = arrowColumnProducer->varVal[idxRowArrow];
                        while (target_vec->size() < start + dataLen) {
                            target_vec->resize(target_vec->size() * 2);
                        }

                        std::memcpy(&(*target_vec)[start],
                                    &buffers.charBuffers[idxCol][idxRowSql * fetchBufferSize],
                                    dataLen);
                        arrowColumnProducer->varVal[idxRowArrow + 1] = start + dataLen;
                        break;
                    }
                    case SQL_CHAR:
                    case SQL_VARCHAR:
                    case SQL_LONGVARCHAR: {
                        if (charCtype == SQL_C_CHAR) {
#if defined(__APPLE__) || defined(__linux__)
                            uint64_t fetchBufferSize = columnSize * 4 + 1 /*null-terminator*/;
#else
                            uint64_t fetchBufferSize = columnSize + 1 /*null-terminator*/;
#endif
                            auto target_vec = &arrowColumnProducer->varData;
                            auto start = arrowColumnProducer->varVal[idxRowArrow];
                            while (target_vec->size() < start + dataLen) {
                                target_vec->resize(target_vec->size() * 2);
                            }

                            std::memcpy(&(*target_vec)[start],
                                        &buffers.charBuffers[idxCol][idxRowSql * fetchBufferSize],
                                        dataLen);
                            arrowColumnProducer->varVal[idxRowArrow + 1] = start + dataLen;
                            break;
                        }
                        // else fall through to SQL_C_WCHAR case
                    }
                    case SQL_SS_XML:
                    case SQL_WCHAR:
                    case SQL_WVARCHAR:
                    case SQL_WLONGVARCHAR: {
                        // We have previously fetched these as WCHARs, even for SQL_CHAR types.
                        assert(dataLen % sizeof(SQLWCHAR) == 0);
                        auto dataLenW = dataLen / sizeof(SQLWCHAR);
                        auto wcharSource =
                            &buffers.wcharBuffers[idxCol][idxRowSql * (columnSize + 1)];
                        auto start = arrowColumnProducer->varVal[idxRowArrow];
                        auto target_vec = &arrowColumnProducer->varData;
                        static_assert(sizeof(SQLWCHAR) == sizeof(char16_t));
                        static_assert(alignof(SQLWCHAR) == alignof(char16_t));
                        const auto* utf16Source = reinterpret_cast<const char16_t*>(wcharSource);
                        size_t maxUtf8Size = dataLenW * 3;

                        while (target_vec->size() < start + maxUtf8Size) {
                            target_vec->resize(target_vec->size() * 2);
                        }

                        size_t bytesWritten = simdutf::convert_utf16le_to_utf8_with_replacement(
                            utf16Source, dataLenW,
                            reinterpret_cast<char*>(target_vec->data() + start));

                        arrowColumnProducer->varVal[idxRowArrow + 1] = start + bytesWritten;
                        break;
                    }
                    case SQL_GUID: {
                        // GUID is stored as a 36-character string in Arrow (e.g.,
                        // "550e8400-e29b-41d4-a716-446655440000") Each GUID is exactly 36 bytes in
                        // UTF-8
                        auto target_vec = &arrowColumnProducer->varData;
                        auto start = arrowColumnProducer->varVal[idxRowArrow];

                        // Ensure buffer has space for the GUID string + null terminator
                        while (target_vec->size() < start + 37) {
                            target_vec->resize(target_vec->size() * 2);
                        }

                        // Get the GUID from the buffer
                        const SQLGUID& guidValue = buffers.guidBuffers[idxCol][idxRowSql];

                        // Convert GUID to string format: xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
                        snprintf(reinterpret_cast<char*>(&target_vec->data()[start]), 37,
                                 "%08X-%04X-%04X-%02X%02X-%02X%02X%02X%02X%02X%02X",
                                 guidValue.Data1, guidValue.Data2, guidValue.Data3,
                                 guidValue.Data4[0], guidValue.Data4[1], guidValue.Data4[2],
                                 guidValue.Data4[3], guidValue.Data4[4], guidValue.Data4[5],
                                 guidValue.Data4[6], guidValue.Data4[7]);

                        // Update offset for next row, ignoring null terminator
                        arrowColumnProducer->varVal[idxRowArrow + 1] = start + 36;
                        break;
                    }
                    case SQL_TINYINT:
                        arrowColumnProducer->uint8Val[idxRowArrow] =
                            buffers.charBuffers[idxCol][idxRowSql];
                        break;
                    case SQL_SMALLINT:
                        arrowColumnProducer->int16Val[idxRowArrow] =
                            buffers.smallIntBuffers[idxCol][idxRowSql];
                        break;
                    case SQL_INTEGER:
                        arrowColumnProducer->int32Val[idxRowArrow] =
                            buffers.intBuffers[idxCol][idxRowSql];
                        break;
                    case SQL_BIGINT:
                        arrowColumnProducer->int64Val[idxRowArrow] =
                            buffers.bigIntBuffers[idxCol][idxRowSql];
                        break;
                    case SQL_REAL:
                        arrowColumnProducer->float32Val[idxRowArrow] =
                            buffers.realBuffers[idxCol][idxRowSql];
                        break;
                    case SQL_FLOAT:
                    case SQL_DOUBLE:
                        arrowColumnProducer->float64Val[idxRowArrow] =
                            buffers.doubleBuffers[idxCol][idxRowSql];
                        break;
                    case SQL_DECIMAL:
                    case SQL_NUMERIC: {
                        // Relies on overloaded operators defined in Int128_t struct
                        assert(dataLen <= MAX_DIGITS_IN_NUMERIC);
                        Int128_t decimalValue(0, 0);
                        auto start = idxRowSql * MAX_DIGITS_IN_NUMERIC;
                        int sign = 1;
                        for (SQLULEN idx = start; idx < start + dataLen; idx++) {
                            char digitChar = buffers.charBuffers[idxCol][idx];
                            if (digitChar == '-') {
                                sign = -1;
                            } else if (digitChar >= '0' && digitChar <= '9') {
                                decimalValue =
                                    decimalValue.multiply_by_10() + (uint64_t)(digitChar - '0');
                            }
                        }
                        arrowColumnProducer->decimalVal[idxRowArrow] =
                            (sign > 0) ? decimalValue : -decimalValue;
                        break;
                    }
                    case SQL_TIMESTAMP:
                    case SQL_TYPE_TIMESTAMP:
                    case SQL_DATETIME: {
                        SQL_TIMESTAMP_STRUCT sql_value =
                            buffers.timestampBuffers[idxCol][idxRowSql];
                        int64_t days =
                            days_from_civil(sql_value.year, sql_value.month, sql_value.day);
                        arrowColumnProducer->tsMicroVal[idxRowArrow] =
                            days * 86400 * 1000000 +
                            static_cast<int64_t>(sql_value.hour) * 3600 * 1000000 +
                            static_cast<int64_t>(sql_value.minute) * 60 * 1000000 +
                            static_cast<int64_t>(sql_value.second) * 1000000 +
                            static_cast<int64_t>(sql_value.fraction) / 1000;
                        break;
                    }
                    case SQL_SS_TIMESTAMPOFFSET: {
                        DateTimeOffset sql_value = buffers.datetimeoffsetBuffers[idxCol][idxRowSql];
                        int64_t days =
                            days_from_civil(sql_value.year, sql_value.month, sql_value.day);
                        arrowColumnProducer->tsMicroVal[idxRowArrow] =
                            days * 86400 * 1000000 +
                            (static_cast<int64_t>(sql_value.hour) -
                             static_cast<int64_t>(sql_value.timezone_hour)) *
                                3600 * 1000000 +
                            (static_cast<int64_t>(sql_value.minute) -
                             static_cast<int64_t>(sql_value.timezone_minute)) *
                                60 * 1000000 +
                            static_cast<int64_t>(sql_value.second) * 1000000 +
                            static_cast<int64_t>(sql_value.fraction) / 1000;
                        break;
                    }
                    case SQL_TYPE_DATE:
                        arrowColumnProducer->dateVal[idxRowArrow] =
                            days_from_civil(buffers.dateBuffers[idxCol][idxRowSql].year,
                                            buffers.dateBuffers[idxCol][idxRowSql].month,
                                            buffers.dateBuffers[idxCol][idxRowSql].day);
                        break;
                    case SQL_SS_TIME2: {
                        const SQL_SS_TIME2_STRUCT& timeValue =
                            buffers.timeBuffers[idxCol][idxRowSql];
                        arrowColumnProducer->timeNanoVal[idxRowArrow] =
                            static_cast<int64_t>(timeValue.hour) * 3600 * 1000000000 +
                            static_cast<int64_t>(timeValue.minute) * 60 * 1000000000 +
                            static_cast<int64_t>(timeValue.second) * 1000000000 +
                            static_cast<int64_t>(timeValue.fraction);
                        break;
                    }
                    case SQL_BIT: {
                        // SQL_BIT is stored as a single bit in Arrow's bitmap format
                        // Get the boolean value from the buffer
                        bool bitValue = buffers.charBuffers[idxCol][idxRowSql] != 0;

                        // Set the bit in the Arrow bitmap
                        size_t byteIndex = idxRowArrow / 8;
                        size_t bitIndex = idxRowArrow % 8;

                        if (bitValue) {
                            // Set bit to 1
                            arrowColumnProducer->bitVal[byteIndex] |= (1 << bitIndex);
                        } else {
                            // Clear bit to 0
                            arrowColumnProducer->bitVal[byteIndex] &= ~(1 << bitIndex);
                        }
                        break;
                    }
                    default: {
                        std::ostringstream errorString;
                        errorString << "Unsupported data type for column ID - " << (idxCol + 1)
                                    << ", Type - " << dataType;
                        LOG(errorString.str().c_str());
                        ThrowStdException(errorString.str());
                        break;
                    }
                }
            }
            idxRowArrow++;
        }
    }

    // Transfer ownership of buffers to batch ArrowSchema
    // First, allocate memory for the necessary structures
    auto arrowSchemaBatch = std::make_unique<ArrowSchema>();

    auto arrowSchemaBatchChildren = std::make_unique<ArrowSchema*[]>(numCols);
    auto arrowSchemaBatchChildPointers = std::make_unique<std::unique_ptr<ArrowSchema>[]>(numCols);
    for (SQLSMALLINT i = 0; i < numCols; i++) {
        arrowSchemaBatchChildPointers[i] = std::make_unique<ArrowSchema>();
    }

    // Second, transfer ownership to arrowSchemaBatch
    // No unhandled exceptions until the pycapsule owns the arrowSchemaBatch to avoid memory leaks

    for (SQLSMALLINT i = 0; i < numCols; i++) {
        *arrowSchemaBatchChildPointers[i] = {
            arrowSchemaPrivateData[i]->format.get(),
            arrowSchemaPrivateData[i]->name.get(),
            nullptr,
            static_cast<int64_t>(columnNullable[i] ? ARROW_FLAG_NULLABLE : 0),
            0,
            nullptr,
            nullptr,
            [](ArrowSchema* schema) {
                assert(schema != nullptr);
                assert(schema->release != nullptr);
                assert(schema->private_data != nullptr);
                assert(schema->children == nullptr && schema->n_children == 0);
                delete schema->private_data;  // Frees format and name
                schema->release = nullptr;
            },
            arrowSchemaPrivateData[i].release(),
        };
    }

    for (SQLSMALLINT i = 0; i < numCols; i++) {
        arrowSchemaBatchChildren[i] = arrowSchemaBatchChildPointers[i].release();
    }

    *arrowSchemaBatch = {
        "+s",
        "",
        nullptr,
        0,
        numCols,
        arrowSchemaBatchChildren.release(),
        nullptr,
        [](ArrowSchema* schema) {
            // format and name are string literals, no need to free
            assert(schema != nullptr);
            assert(schema->release != nullptr);
            assert(schema->private_data == nullptr);
            assert(schema->children != nullptr);
            assert(schema->n_children > 0);
            for (int64_t i = 0; i < schema->n_children; ++i) {
                if (schema->children[i]) {
                    if (schema->children[i]->release) {
                        schema->children[i]->release(schema->children[i]);
                    }
                    delete schema->children[i];
                }
            }
            delete[] schema->children;
            schema->release = nullptr;
        },
        nullptr,
    };

    // Finally, transfer ownership of arrowSchemaBatch and its pointer to pycapsule
    py::capsule arrowSchemaBatchCapsule;
    try {
        arrowSchemaBatchCapsule =
            py::capsule(arrowSchemaBatch.get(), "arrow_schema", [](void* ptr) {
                auto arrowSchema = static_cast<ArrowSchema*>(ptr);
                if (arrowSchema->release) {
                    arrowSchema->release(arrowSchema);
                }
                delete arrowSchema;
            });
    } catch (...) {
        arrowSchemaBatch->release(arrowSchemaBatch.get());
        throw;
    }
    arrowSchemaBatch.release();
    capsules.append(arrowSchemaBatchCapsule);

    // Transfer ownership of buffers to batch ArrowArray
    // First, allocate memory for the necessary structures
    auto arrowArrayBatch = std::make_unique<ArrowArray>();

    auto arrowArrayBatchBuffers = std::make_unique<const void*[]>(1);
    arrowArrayBatchBuffers[0] = nullptr;

    auto arrowArrayBatchChildren = std::make_unique<ArrowArray*[]>(numCols);
    auto arrowArrayBatchChildPointers = std::make_unique<std::unique_ptr<ArrowArray>[]>(numCols);
    for (SQLSMALLINT i = 0; i < numCols; i++) {
        arrowArrayBatchChildPointers[i] = std::make_unique<ArrowArray>();
    }

    // Second, transfer ownership to arrowArrayBatch
    // No unhandled exceptions until the pycapsule owns the arrowArrayBatch to avoid memory leaks

    for (SQLUSMALLINT col = 0; col < numCols; col++) {
        arrowArrayPrivateData[col]->buffers[0] = arrowArrayPrivateData[col]->valid.get();
        arrowArrayPrivateData[col]->buffers[1] = arrowArrayPrivateData[col]->ptrValueBuffer;
        arrowArrayPrivateData[col]->buffers[2] = arrowArrayPrivateData[col]->varData.data();

        *arrowArrayBatchChildPointers[col] = {
            static_cast<int64_t>(idxRowArrow),
            nullCounts[col],
            0,
            columnVarLen[col] ? 3 : 2,
            0,
            (const void**)arrowArrayPrivateData[col]->buffers.data(),
            nullptr,
            nullptr,
            [](ArrowArray* array) {
                assert(array != nullptr);
                assert(array->private_data != nullptr);
                assert(array->release != nullptr);
                assert(array->children == nullptr);
                assert(array->n_children == 0);
                delete array->private_data;  // Frees all buffer entries
                assert(array->buffers != nullptr);
                array->release = nullptr;
            },
            arrowArrayPrivateData[col].release(),
        };
    }

    for (SQLSMALLINT i = 0; i < numCols; i++) {
        arrowArrayBatchChildren[i] = arrowArrayBatchChildPointers[i].release();
    }

    *arrowArrayBatch = {
        static_cast<int64_t>(idxRowArrow),
        0,
        0,
        1,
        numCols,
        arrowArrayBatchBuffers.release(),
        arrowArrayBatchChildren.release(),
        nullptr,
        [](ArrowArray* array) {
            assert(array != nullptr);
            assert(array->private_data == nullptr);
            assert(array->release != nullptr);
            assert(array->children != nullptr);
            assert(array->n_children > 0);
            for (int64_t i = 0; i < array->n_children; ++i) {
                if (array->children[i]) {
                    if (array->children[i]->release) {
                        array->children[i]->release(array->children[i]);
                    }
                    delete array->children[i];
                }
            }
            delete[] array->children;
            assert(array->buffers != nullptr);
            assert(array->n_buffers == 1);
            assert(array->buffers[0] == nullptr);
            delete[] array->buffers;
            array->release = nullptr;
        },
        nullptr,
    };

    // Finally, transfer ownership of arrowArrayBatch and its pointer to pycapsule
    py::capsule arrowArrayBatchCapsule;
    try {
        arrowArrayBatchCapsule = py::capsule(arrowArrayBatch.get(), "arrow_array", [](void* ptr) {
            auto arrowArray = static_cast<ArrowArray*>(ptr);
            if (arrowArray->release) {
                arrowArray->release(arrowArray);
            }
            delete arrowArray;
        });
    } catch (...) {
        arrowArrayBatch->release(arrowArrayBatch.get());
        throw;
    }
    arrowArrayBatch.release();
    capsules.append(arrowArrayBatchCapsule);

    return ret;
}

// FetchAll_wrap - Fetches all rows of data from the result set.
//
// @param StatementHandle: Handle to the statement from which data is to be
// fetched.
// @param rows: A Python list that will be populated with the fetched rows of
// data.
//
// @return SQLRETURN: SQL_SUCCESS if data is fetched successfully,
//                    SQL_NO_DATA if there are no more rows to fetch,
//                    throws a runtime error if there is an error fetching data.
//
// This function assumes that the statement handle (hStmt) is already allocated
// and a query has been executed. It fetches all rows from the result set and
// populates the provided Python list with the row data. If there are no more
// rows to fetch, it returns SQL_NO_DATA. If an error occurs during fetching, it
// throws a runtime error.
SQLRETURN FetchAll_wrap(SqlHandlePtr StatementHandle, py::list& rows,
                        const std::string& charEncoding = "utf-16le",
                        const std::string& wcharEncoding = "utf-16le",
                        int charCtype = SQL_C_WCHAR) {
    PERF_TIMER("FetchAll_wrap");
    // Issue #531: upgrade SQL_C_CHAR + utf-8 to SQL_C_WCHAR on Windows so the
    // driver does lossless UTF-16 conversion instead of returning ACP bytes.
    charCtype = EffectiveCharCtypeForFetch(charCtype, charEncoding);
    SQLRETURN ret;
    SQLHSTMT hStmt = StatementHandle->get();
    // Retrieve column count
    SQLSMALLINT numCols = SQLNumResultCols_wrap(StatementHandle);

    // Retrieve column metadata
    py::list columnNames;
    ret = SQLDescribeCol_wrap(StatementHandle, columnNames);
    if (!SQL_SUCCEEDED(ret)) {
        LOG("FetchAll_wrap: Failed to get column descriptions - SQLRETURN=%d", ret);
        return ret;
    }

    std::vector<SQLUSMALLINT> lobColumns;
    for (SQLSMALLINT i = 0; i < numCols; i++) {
        auto colMeta = columnNames[i].cast<py::dict>();
        SQLSMALLINT dataType = colMeta["DataType"].cast<SQLSMALLINT>();
        SQLULEN columnSize = colMeta["ColumnSize"].cast<SQLULEN>();

        // Detect LOB columns that need SQLGetData streaming
        // sql_variant always uses SQLGetData for native type preservation
        if (IsLobOrVariantColumn(dataType, columnSize)) {
            lobColumns.push_back(i + 1);  // 1-based
        }
    }

    // If we have LOBs → fall back to row-by-row fetch + SQLGetData_wrap
    if (!lobColumns.empty()) {
        LOG("FetchAll_wrap: LOB columns detected (%zu columns), using per-row "
            "SQLGetData path",
            lobColumns.size());
        while (true) {
            {
                // Release GIL during the blocking fetch
                py::gil_scoped_release release;
                ret = SQLFetch_ptr(hStmt);
            }
            if (ret == SQL_NO_DATA)
                break;
            if (!SQL_SUCCEEDED(ret))
                return ret;

            py::list row;
            SQLGetData_wrap(StatementHandle, numCols, row, charEncoding, wcharEncoding,
                            charCtype);  // <-- streams LOBs correctly
            rows.append(row);
        }
        return SQL_SUCCESS;
    }

    // No LOBs detected - use binding path with batch fetching
    // Define a memory limit (1 GB)
    const size_t memoryLimit = 1ULL * 1024 * 1024 * 1024;
    size_t totalRowSize = calculateRowSize(columnNames, numCols);

    // Calculate fetch size based on the total row size and memory limit
    size_t numRowsInMemLimit;
    if (totalRowSize > 0) {
        numRowsInMemLimit = static_cast<size_t>(memoryLimit / totalRowSize);
    } else {
        // Handle case where totalRowSize is 0 to avoid division by zero.
        // This can happen for NVARCHAR(MAX) cols. SQLDescribeCol returns 0
        // for column size of such columns.
        // TODO: Find why NVARCHAR(MAX) returns columnsize 0
        // TODO: What if a row has 2 cols, an int & NVARCHAR(MAX)?
        //       totalRowSize will be 4+0 = 4. It wont take NVARCHAR(MAX)
        //       into account. So, we will end up fetching 1000 rows at a time.
        numRowsInMemLimit = 1;  // fetchsize will be 10
    }
    // TODO: Revisit this logic. Eventhough we're fetching fetchSize rows at a
    // time, fetchall will keep all rows in memory anyway. So what are we
    // gaining by fetching fetchSize rows at a time? Also, say the table has
    // only 10 rows, each row size if 100 bytes. Here, we'll have fetchSize =
    // 1000, so we'll allocate memory for 1000 rows inside SQLBindCol_wrap,
    // while actually only need to retrieve 10 rows
    int fetchSize;
    if (numRowsInMemLimit == 0) {
        // If the row size is larger than the memory limit, fetch one row at a
        // time
        fetchSize = 1;
    } else if (numRowsInMemLimit > 0 && numRowsInMemLimit <= 100) {
        // If between 1-100 rows fit in memoryLimit, fetch 10 rows at a time
        fetchSize = 10;
    } else if (numRowsInMemLimit > 100 && numRowsInMemLimit <= 1000) {
        // If between 100-1000 rows fit in memoryLimit, fetch 100 rows at a time
        fetchSize = 100;
    } else {
        fetchSize = 1000;
    }
    LOG("FetchAll_wrap: Fetching data in batch sizes of %d", fetchSize);

    ColumnBuffers buffers(numCols, fetchSize);

    // Bind columns
    ret = SQLBindColums(hStmt, buffers, columnNames, numCols, fetchSize, charCtype);
    if (!SQL_SUCCEEDED(ret)) {
        LOG("FetchAll_wrap: Error when binding columns - SQLRETURN=%d", ret);
        return ret;
    }

    SQLULEN numRowsFetched;
    SQLSetStmtAttr_ptr(hStmt, SQL_ATTR_ROW_ARRAY_SIZE, (SQLPOINTER)(intptr_t)fetchSize, 0);
    SQLSetStmtAttr_ptr(hStmt, SQL_ATTR_ROWS_FETCHED_PTR, &numRowsFetched, 0);

    while (ret != SQL_NO_DATA) {
        ret = FetchBatchData(hStmt, buffers, columnNames, rows, numCols, numRowsFetched, lobColumns,
                             charEncoding, charCtype);
        if (!SQL_SUCCEEDED(ret) && ret != SQL_NO_DATA) {
            LOG("FetchAll_wrap: Error when fetching data - SQLRETURN=%d", ret);
            return ret;
        }
    }

    // Reset attributes before returning to avoid using stack pointers later
    SQLSetStmtAttr_ptr(hStmt, SQL_ATTR_ROW_ARRAY_SIZE, (SQLPOINTER)1, 0);
    SQLSetStmtAttr_ptr(hStmt, SQL_ATTR_ROWS_FETCHED_PTR, NULL, 0);

    // Unbind columns to allow subsequent fetchone() calls to use SQLGetData
    SQLFreeStmt_ptr(hStmt, SQL_UNBIND);

    return ret;
}

// FetchOne_wrap - Fetches a single row of data from the result set.
//
// @param StatementHandle: Handle to the statement from which data is to be
// fetched.
// @param row: A Python list that will be populated with the fetched row data.
//
// @return SQLRETURN: SQL_SUCCESS or SQL_SUCCESS_WITH_INFO if data is fetched
// successfully,
//                    SQL_NO_DATA if there are no more rows to fetch,
//                    throws a runtime error if there is an error fetching data.
//
// This function assumes that the statement handle (hStmt) is already allocated
// and a query has been executed. It fetches the next row of data from the
// result set and populates the provided Python list with the row data. If there
// are no more rows to fetch, it returns SQL_NO_DATA. If an error occurs during
// fetching, it throws a runtime error.
SQLRETURN FetchOne_wrap(SqlHandlePtr StatementHandle, py::list& row,
                        const std::string& charEncoding = "utf-16le",
                        const std::string& wcharEncoding = "utf-16le",
                        int charCtype = SQL_C_WCHAR) {
    PERF_TIMER("FetchOne_wrap");
    // Issue #531: upgrade SQL_C_CHAR + utf-8 to SQL_C_WCHAR on Windows so the
    // driver does lossless UTF-16 conversion instead of returning ACP bytes.
    charCtype = EffectiveCharCtypeForFetch(charCtype, charEncoding);
    SQLRETURN ret;
    SQLHSTMT hStmt = StatementHandle->get();

    // Unbind any columns from previous fetch operations (e.g., fetchmany)
    // to avoid conflicts with SQLGetData. SQLGetData cannot be used on
    // columns that are already bound.
    SQLFreeStmt_ptr(hStmt, SQL_UNBIND);

    // Assume hStmt is already allocated and a query has been executed
    {
        // Release the GIL during the blocking ODBC fetch
        py::gil_scoped_release release;
        ret = SQLFetch_ptr(hStmt);
    }
    if (SQL_SUCCEEDED(ret)) {
        // Retrieve column count
        SQLSMALLINT colCount = SQLNumResultCols_wrap(StatementHandle);
        ret =
            SQLGetData_wrap(StatementHandle, colCount, row, charEncoding, wcharEncoding, charCtype);
        if (!SQL_SUCCEEDED(ret)) {
            LOG("FetchOne_wrap: Error retrieving data with SQLGetData - SQLRETURN=%d", ret);
            return ret;
        }
    } else if (ret != SQL_NO_DATA) {
        LOG("FetchOne_wrap: Error when fetching data - SQLRETURN=%d", ret);
    }
    return ret;
}

// Wrap SQLMoreResults
SQLRETURN SQLMoreResults_wrap(SqlHandlePtr StatementHandle) {
    PERF_TIMER("SQLMoreResults_wrap");
    LOG("SQLMoreResults_wrap: Check for more results");
    if (!SQLMoreResults_ptr) {
        LOG("SQLMoreResults_wrap: Function pointer not initialized. Loading "
            "the driver.");
        DriverLoader::getInstance().loadDriver();  // Load the driver
    }

    // Release the GIL during the blocking ODBC call
    py::gil_scoped_release release;
    return SQLMoreResults_ptr(StatementHandle->get());
}

// Wrap SQLFreeHandle
SQLRETURN SQLFreeHandle_wrap(SQLSMALLINT HandleType, SqlHandlePtr Handle) {
    PERF_TIMER("SQLFreeHandle_wrap");
    LOG("SQLFreeHandle_wrap: Free SQL handle type=%d", HandleType);
    // Guard against a null/None handle being passed from Python - dereferencing
    // Handle->get() on a null shared_ptr would segfault.
    if (!Handle || HandleType != Handle->type()) {
        return SQL_INVALID_HANDLE;
    }
    if (!SQLFreeHandle_ptr) {
        LOG("SQLFreeHandle_wrap: Function pointer not initialized. Loading the "
            "driver.");
        DriverLoader::getInstance().loadDriver();  // Load the driver
    }

    SQLRETURN ret = Handle->freeHandle();
    if (!SQL_SUCCEEDED(ret)) {
        LOG("SQLFreeHandle_wrap: SQLFreeHandle failed with error code - %d", ret);
        return ret;
    }
    return ret;
}

// Wrap SQLRowCount
SQLLEN SQLRowCount_wrap(SqlHandlePtr StatementHandle) {
    PERF_TIMER("SQLRowCount_wrap");
    LOG("SQLRowCount_wrap: Get number of rows affected by last execute");
    if (!SQLRowCount_ptr) {
        LOG("SQLRowCount_wrap: Function pointer not initialized. Loading the "
            "driver.");
        DriverLoader::getInstance().loadDriver();  // Load the driver
    }

    SQLLEN rowCount;
    SQLRETURN ret = SQLRowCount_ptr(StatementHandle->get(), &rowCount);
    if (!SQL_SUCCEEDED(ret)) {
        LOG("SQLRowCount_wrap: SQLRowCount failed with error code - %d", ret);
        return ret;
    }
    LOG("SQLRowCount_wrap: SQLRowCount returned %ld", (long)rowCount);
    return rowCount;
}

static std::once_flag pooling_init_flag;
void enable_pooling(int maxSize, int idleTimeout) {
    // configure() locks in the default max_size/idle_timeout for the process and
    // must run exactly once, but re-arming must happen on every enable so a
    // disable_pooling() -> enable_pooling() cycle resumes pooling (the
    // std::call_once body would otherwise be skipped on the second enable).
    std::call_once(pooling_init_flag,
                   [&]() { ConnectionPoolManager::getInstance().configure(maxSize, idleTimeout); });
    ConnectionPoolManager::getInstance().setAccepting(true);
}

// Thread-safe decimal separator setting
ThreadSafeDecimalSeparator g_decimalSeparator;

void DDBCSetDecimalSeparator(const std::string& separator) {
    SetDecimalSeparator(separator);
}

// Architecture-specific defines
#ifndef ARCHITECTURE
#define ARCHITECTURE "win64"  // Default to win64 if not defined during compilation
#endif

// Functions/data to be exposed to Python as a part of ddbc_bindings module
PYBIND11_MODULE(ddbc_bindings, m) {
    m.doc() = "msodbcsql driver api bindings for Python";

    PyTypeCache::initialize();

    // Add architecture information as module attribute
    m.attr("__architecture__") = ARCHITECTURE;

    // Expose architecture-specific constants
    m.attr("ARCHITECTURE") = ARCHITECTURE;

    m.attr("SQL_NO_TOTAL") = static_cast<int>(SQL_NO_TOTAL);

    // Expose the C++ functions to Python
    m.def("ThrowStdException", &ThrowStdException);
    m.def("GetDriverPathCpp", &GetDriverPathCpp, "Get the path to the ODBC driver");
    m.def("_get_odbc_driver_path", &GetDriverPathForProviderCpp,
          "Get the ODBC driver path for an explicit provider without selecting it");
    m.def("_set_odbc_provider", &SetSelectedProvider,
          "Select the ODBC provider ('msodbcsql18' or 'mssql-odbc') before the driver loads");

    // Define parameter info class
    py::class_<ParamInfo>(m, "ParamInfo")
        .def(py::init<>())
        .def_readwrite("inputOutputType", &ParamInfo::inputOutputType)
        .def_readwrite("paramCType", &ParamInfo::paramCType)
        .def_readwrite("paramSQLType", &ParamInfo::paramSQLType)
        .def_readwrite("columnSize", &ParamInfo::columnSize)
        .def_readwrite("decimalDigits", &ParamInfo::decimalDigits)
        .def_readwrite("strLenOrInd", &ParamInfo::strLenOrInd)
        .def_property(
            "dataPtr",
            [](const ParamInfo& info) -> py::object {
                if (!info.dataPtr) {
                    return py::none();
                }
                return info.dataPtr;
            },
            [](ParamInfo& info, py::object obj) { info.dataPtr = std::move(obj); })
        .def_readwrite("isDAE", &ParamInfo::isDAE);

    // Define numeric data class
    py::class_<NumericData>(m, "NumericData")
        .def(py::init<>())
        .def(py::init<SQLCHAR, SQLSCHAR, SQLCHAR, const std::string&>())
        .def_readwrite("precision", &NumericData::precision)
        .def_readwrite("scale", &NumericData::scale)
        .def_readwrite("sign", &NumericData::sign)
        .def_readwrite("val", &NumericData::val);

    // Define error info class
    py::class_<ErrorInfo>(m, "ErrorInfo")
        .def_readwrite("sqlState", &ErrorInfo::sqlState)
        .def_readwrite("ddbcErrorMsg", &ErrorInfo::ddbcErrorMsg);

    py::class_<SqlHandle, SqlHandlePtr>(m, "SqlHandle")
        .def("free", &SqlHandle::free, "Free the handle")
        .def("_close_cursor", &SqlHandle::close_cursor,
             "Internal: close the cursor without freeing the prepared statement")
        .def("_cancel", &SqlHandle::cancel,
             "Internal: cancel an in-progress statement (SQLCancel). "
             "Safe to call from another thread; no-op if unsupported or idle.");

    py::class_<ConnectionHandle>(m, "Connection")
        .def(py::init<const std::u16string&, bool, const py::dict&, const std::u16string&,
                      const py::object&>(),
             py::arg("conn_str"), py::arg("use_pool"), py::arg("attrs_before") = py::dict(),
             py::arg("pool_key") = std::u16string(), py::arg("token_factory") = py::none())
        .def("close", &ConnectionHandle::close,
             py::arg("transaction_already_rolled_back") = false, "Close the connection")
        .def("commit", &ConnectionHandle::commit, "Commit the current transaction")
        .def("rollback", &ConnectionHandle::rollback, "Rollback the current transaction")
        .def("set_autocommit", &ConnectionHandle::setAutocommit)
        .def("get_autocommit", &ConnectionHandle::getAutocommit)
        .def("set_attr", &ConnectionHandle::setAttr, py::arg("attribute"), py::arg("value"),
             "Set connection attribute")
        .def("alloc_statement_handle", &ConnectionHandle::allocStatementHandle)
        .def("get_info", &ConnectionHandle::getInfo, py::arg("info_type"));
    m.def("enable_pooling", &enable_pooling, "Enable global connection pooling");
    m.def("close_pooling", []() { ConnectionPoolManager::getInstance().closePools(); });
    m.def("disable_pooling", []() {
        // Disarm new-pool creation *before* closing so a connect racing this
        // disable cannot resurrect a pool after the map is cleared: any
        // acquireConnection serialized after setAccepting(false) declines and
        // falls back to a non-pooled connection. closePools() then reaps every
        // pool created before the disarm.
        auto& manager = ConnectionPoolManager::getInstance();
        manager.setAccepting(false);
        manager.closePools();
    }, "Disable global connection pooling and close all pools");
    m.def("DDBCSQLExecDirect", &SQLExecDirect_wrap, "Execute a SQL query directly");
    m.def("DDBCSQLExecute", &SQLExecute_wrap,
          "DetectParamTypes + BindParameters + SQLExecute all in C++",
          py::arg("statementHandle"), py::arg("query"), py::arg("params"),
          py::arg("inputSizes"), py::arg("isStmtPrepared"), py::arg("usePrepare"),
          py::arg("encodingSettings"));
    m.def("SQLExecuteMany", &SQLExecuteMany_wrap, "Execute statement with multiple parameter sets",
          py::arg("statementHandle"), py::arg("query"), py::arg("columnwise_params"),
          py::arg("paramInfos"), py::arg("paramSetSize"), py::arg("encodingSettings"));
    m.def("DDBCSQLRowCount", &SQLRowCount_wrap,
          "Get the number of rows affected by the last statement");
    m.def("DDBCSQLFetch", &SQLFetch_wrap, "Fetch the next row from the result set");
    m.def("DDBCSQLNumResultCols", &SQLNumResultCols_wrap,
          "Get the number of columns in the result set");
    m.def("DDBCSQLDescribeCol", &SQLDescribeCol_wrap,
          "Get information about a column in the result set");
    m.def("DDBCSQLGetData", &SQLGetData_wrap, "Retrieve data from the result set");
    m.def("DDBCSQLMoreResults", &SQLMoreResults_wrap, "Check for more results in the result set");
    m.def("DDBCSQLFetchOne", &FetchOne_wrap, "Fetch one row from the result set",
          py::arg("StatementHandle"), py::arg("row"), py::arg("charEncoding") = "utf-16le",
          py::arg("wcharEncoding") = "utf-16le", py::arg("charCtype") = SQL_C_WCHAR);
    m.def("DDBCSQLFetchMany", &FetchMany_wrap, py::arg("StatementHandle"), py::arg("rows"),
          py::arg("fetchSize"), py::arg("charEncoding") = "utf-16le",
          py::arg("wcharEncoding") = "utf-16le", py::arg("charCtype") = SQL_C_WCHAR,
          "Fetch many rows from the result set");
    m.def("DDBCSQLFetchAll", &FetchAll_wrap, "Fetch all rows from the result set",
          py::arg("StatementHandle"), py::arg("rows"), py::arg("charEncoding") = "utf-16le",
          py::arg("wcharEncoding") = "utf-16le", py::arg("charCtype") = SQL_C_WCHAR);
    m.def("DDBCSQLFetchArrowBatch", &FetchArrowBatch_wrap,
          "Fetch an arrow batch of given length from the result set");
    m.def("DDBCSQLFreeHandle", &SQLFreeHandle_wrap, "Free a handle");
    m.def("DDBCSQLResetStmt", &SQLResetStmt_wrap,
          "Close cursor and unbind params without freeing HSTMT");
    m.def("DDBCSQLCheckError", &SQLCheckError_Wrap, "Check for driver errors");
    m.def("DDBCSQLGetAllDiagRecords", &SQLGetAllDiagRecords,
          "Get all diagnostic records for a handle", py::arg("handle"));
    m.def("DDBCSQLTables", &SQLTables_wrap, "Get table information using ODBC SQLTables",
          py::arg("StatementHandle"), py::arg("catalog") = std::u16string(),
          py::arg("schema") = std::u16string(), py::arg("table") = std::u16string(),
          py::arg("tableType") = std::u16string());
    m.def("DDBCSQLFetchScroll", &SQLFetchScroll_wrap,
          "Scroll to a specific position in the result set and optionally "
          "fetch data");
    m.def("DDBCSetDecimalSeparator", &DDBCSetDecimalSeparator,
          "Set the decimal separator character");
    m.def(
        "DDBCSQLSetStmtAttr",
        [](SqlHandlePtr stmt, SQLINTEGER attr, py::object value) {
            SQLPOINTER ptr_value;
            if (py::isinstance<py::int_>(value)) {
                // For integer attributes like SQL_ATTR_QUERY_TIMEOUT
                ptr_value =
                    reinterpret_cast<SQLPOINTER>(static_cast<SQLULEN>(value.cast<int64_t>()));
            } else {
                // For pointer attributes
                ptr_value = value.cast<SQLPOINTER>();
            }
            return SQLSetStmtAttr_ptr(stmt->get(), attr, ptr_value, 0);
        },
        "Set statement attributes");
    m.def("DDBCSQLGetTypeInfo", &SQLGetTypeInfo_Wrapper,
          "Returns information about the data types that are supported by the "
          "data source",
          py::arg("StatementHandle"), py::arg("DataType"));
    m.def("DDBCSQLProcedures", [](SqlHandlePtr StatementHandle, const py::object& catalog,
                                  const py::object& schema, const py::object& procedure) {
        return SQLProcedures_wrap(StatementHandle, catalog, schema, procedure);
    });

    m.def("DDBCSQLForeignKeys",
          [](SqlHandlePtr StatementHandle, const py::object& pkCatalog, const py::object& pkSchema,
             const py::object& pkTable, const py::object& fkCatalog, const py::object& fkSchema,
             const py::object& fkTable) {
              return SQLForeignKeys_wrap(StatementHandle, pkCatalog, pkSchema, pkTable, fkCatalog,
                                         fkSchema, fkTable);
          });
    m.def("DDBCSQLPrimaryKeys", [](SqlHandlePtr StatementHandle, const py::object& catalog,
                                   const py::object& schema, const std::u16string& table) {
        return SQLPrimaryKeys_wrap(StatementHandle, catalog, schema, table);
    });
    m.def("DDBCSQLSpecialColumns",
          [](SqlHandlePtr StatementHandle, SQLSMALLINT identifierType, const py::object& catalog,
             const py::object& schema, const std::u16string& table, SQLSMALLINT scope,
             SQLSMALLINT nullable) {
              return SQLSpecialColumns_wrap(StatementHandle, identifierType, catalog, schema, table,
                                            scope, nullable);
          });
    m.def("DDBCSQLStatistics",
          [](SqlHandlePtr StatementHandle, const py::object& catalog, const py::object& schema,
             const std::u16string& table, SQLUSMALLINT unique, SQLUSMALLINT reserved) {
              return SQLStatistics_wrap(StatementHandle, catalog, schema, table, unique, reserved);
          });
    m.def("DDBCSQLColumns",
          [](SqlHandlePtr StatementHandle, const py::object& catalog, const py::object& schema,
             const py::object& table, const py::object& column) {
              return SQLColumns_wrap(StatementHandle, catalog, schema, table, column);
          });

    // Add profiling submodule (only in profiling builds; compiled out by default)
#ifdef ENABLE_PROFILING
    auto profiling = m.def_submodule("profiling", "Performance profiling");
    profiling.def("enable", []() { mssql_profiling::PerformanceCounter::instance().enable(); },
                  "Enable performance profiling");
    profiling.def("disable", []() { mssql_profiling::PerformanceCounter::instance().disable(); },
                  "Disable performance profiling");
    profiling.def("get_stats", []() { return mssql_profiling::PerformanceCounter::instance().get_stats(); },
                  "Get profiling statistics");
    profiling.def("get_timeline", []() { return mssql_profiling::PerformanceCounter::instance().get_timeline(); },
                  "Get timeline events (list of {name, start_us, duration_us})");
    profiling.def("reset", []() { mssql_profiling::PerformanceCounter::instance().reset(); },
                  "Reset profiling statistics and timeline");
    profiling.def("reset_stats_only", []() { mssql_profiling::PerformanceCounter::instance().reset_stats_only(); },
                  "Reset profiling statistics but keep timeline");
    profiling.def("is_enabled", []() { return mssql_profiling::PerformanceCounter::instance().is_enabled(); },
                  "Check if profiling is enabled");
    profiling.def("enable_timeline", []() { mssql_profiling::PerformanceCounter::instance().enable_timeline(); },
                  "Enable timeline recording (resets epoch)");
    profiling.def("disable_timeline", []() { mssql_profiling::PerformanceCounter::instance().disable_timeline(); },
                  "Disable timeline recording");
#endif  // ENABLE_PROFILING

    // Add a version attribute
    m.attr("__version__") = "1.0.0";

    // Fast Row construction in C++ — replaces Python list comprehension
    m.def("construct_rows", &RowFactory::construct_rows,
          "Build Row objects in C++ for fetchall/fetchmany fast path",
          py::arg("rows_data"), py::arg("row_class"),
          py::arg("column_map"), py::arg("cursor"),
          py::arg("column_map_lower") = py::none(),
          py::arg("column_names") = py::none());

    // Expose logger bridge function to Python
    m.def("update_log_level", &mssql_python::logging::LoggerBridge::updateLevel,
          "Update the cached log level in C++ bridge");

    // Initialize the logger bridge
    try {
        mssql_python::logging::LoggerBridge::initialize();
    } catch (const std::exception& e) {
        // Log initialization failure but don't throw
        // Use std::cerr instead of fprintf for type-safe output
        std::cerr << "Logger bridge initialization failed: " << e.what() << std::endl;
    }

    // Force DriverLoader's Meyer's-singleton to construct now, at import time,
    // instead of on first lazy loadDriver() call from a connection path. This
    // keeps its position in C++'s reverse-order static destruction at
    // interpreter shutdown stable and matching prior releases.
    //
    // Deliberately do NOT call loadDriver() here: doing so would resolve and
    // std::call_once-freeze the driver using whatever native provider is
    // selected at import time (always the default, since Python's
    // _set_odbc_provider() push - see Connection.__init__ - cannot run until
    // after `import ddbc_bindings` completes). That silently locks in the
    // wrong driver whenever a caller selects a non-default provider. Loading
    // stays lazy, on first real connection attempt, by which point the
    // provider push has already happened.
    DriverLoader::getInstance();
}