"""SQ_DESCRIBE column descriptor parser and SQ_TUPLE row decoder.

Per IfxSqli.receiveDescribe (line 2175+) for ``isUSVER`` modern servers.
The per-field block layout is:

  fieldIndex          (int 4)
  columnStartPos      (int 4 — USVER)
  columnType          (short 2 — base IDS type code with high-bit flags)
  columnExtendedId    (int 4 — USVER, for UDT/extended types)
  ownerName           (readChar = [short len][bytes][pad if odd])
  extendedName        (readChar)
  reference           (short 2)
  alignment           (short 2)
  sourceType          (int 4)
  encodedLength       (int 4)

After all fields: the string table (a length-prefixed block of nul-separated
column names), read via readPadded.
"""

from __future__ import annotations

from collections.abc import Callable
from dataclasses import dataclass
from datetime import timedelta as _timedelta
from types import MappingProxyType

from ._protocol import IfxStreamReader
from ._types import IfxType, base_type, is_nullable
from .converters import (
    _DOUBLE_NULL,
    _REAL_NULL,
    _UNPACK_DOUBLE,
    _UNPACK_FLOAT,
    _UNPACK_INT,
    _UNPACK_LONG,
    _UNPACK_SHORT,
    DECODERS,
    FIXED_WIDTHS,
    BlobLocator,
    ClobLocator,
    CollectionValue,
    RowValue,
    _decode_base,
    _decode_datetime,
    _decode_interval,
)
from .converters import (
    _INFORMIX_DATE_EPOCH as _DATE_EPOCH,
)

# Module-level type-code constants — lifted out of the hot loop in
# parse_tuple_payload so we don't pay the IntFlag→int conversion per
# column per row.
_TC_CHAR = int(IfxType.CHAR)
_TC_VARCHAR = int(IfxType.VARCHAR)
_TC_NCHAR = int(IfxType.NCHAR)
_TC_NVCHAR = int(IfxType.NVCHAR)
_TC_LVARCHAR = int(IfxType.LVARCHAR)
_TC_DECIMAL = int(IfxType.DECIMAL)
_TC_MONEY = int(IfxType.MONEY)
_TC_DATETIME = int(IfxType.DATETIME)
_TC_INTERVAL = int(IfxType.INTERVAL)
_TC_UDTFIXED = int(IfxType.UDTFIXED)
_TC_UDTVAR = int(IfxType.UDTVAR)
_TC_ROW = int(IfxType.ROW)
_TC_COLLECTION = int(IfxType.COLLECTION)
_TC_SET = int(IfxType.SET)
_TC_MULTISET = int(IfxType.MULTISET)
_TC_LIST = int(IfxType.LIST)

_COLLECTION_KIND_MAP = MappingProxyType({
    _TC_SET: "set",
    _TC_MULTISET: "multiset",
    _TC_LIST: "list",
    _TC_COLLECTION: "collection",
})


@dataclass
class ColumnInfo:
    """One column in a SQ_DESCRIBE response."""

    name: str
    type_code: int  # base IDS type code (high-bit flags stripped)
    raw_type_code: int  # raw type-code short with flags intact
    encoded_length: int
    column_start_pos: int = 0
    extended_id: int = 0
    owner_name: str = ""
    extended_name: str = ""

    @property
    def null_ok(self) -> bool:
        return is_nullable(self.raw_type_code)

    def to_description_tuple(self) -> tuple:
        """The PEP 249 cursor.description 7-tuple."""
        return (
            self.name,
            self.type_code,
            self.encoded_length,  # display_size
            self.encoded_length,  # internal_size
            0,  # precision (Phase 6+ derives from type)
            0,  # scale
            self.null_ok,
        )


def _read_char(reader: IfxStreamReader, encoding: str = "iso-8859-1") -> str:
    """Read JDBC's ``readChar`` format: [short len][bytes][pad if odd-len]."""
    length = reader.read_short()
    if length < 0:
        return ""
    if length == 0:
        return ""
    data = reader.read_exact(length)
    if length & 1:
        reader.read_exact(1)  # pad byte
    return data.decode(encoding)


def parse_describe(reader: IfxStreamReader) -> tuple[list[ColumnInfo], dict]:
    """Parse a SQ_DESCRIBE response (the SQ_DESCRIBE tag is already consumed).

    Returns ``(columns, metadata)``.
    """
    statement_type = reader.read_short()
    statement_id = reader.read_short()
    estimated_cost = reader.read_int()
    tuple_size = reader.read_short()
    nfields = reader.read_short()
    string_table_size = reader.read_int()  # 4-byte on modern servers

    metadata = {
        "statement_type": statement_type,
        "statement_id": statement_id,
        "estimated_cost": estimated_cost,
        "tuple_size": tuple_size,
        "nfields": nfields,
        "string_table_size": string_table_size,
    }

    if nfields <= 0:
        return [], metadata

    # Pass 1: per-field descriptor block (no name yet — names come from
    # the string table).
    raw_fields: list[dict] = []
    for _ in range(nfields):
        field_index = reader.read_int()
        column_start_pos = reader.read_int()
        column_type = reader.read_short()
        column_extended_id = reader.read_int()
        owner_name = _read_char(reader)
        extended_name = _read_char(reader)
        reference = reader.read_short()  # noqa: F841 (Phase 6+)
        alignment = reader.read_short()  # noqa: F841
        source_type = reader.read_int()  # noqa: F841
        encoded_length = reader.read_int()
        raw_fields.append(
            {
                "field_index": field_index,
                "column_start_pos": column_start_pos,
                "type_code": column_type,
                "extended_id": column_extended_id,
                "owner_name": owner_name,
                "extended_name": extended_name,
                "encoded_length": encoded_length,
            }
        )

    # Pass 2: string table — nul-separated column names. readPadded.
    string_table = b""
    if string_table_size > 0:
        string_table = reader.read_exact(string_table_size)
        if string_table_size & 1:
            reader.read_exact(1)  # pad

    # Split string table on nul to get the column-name list. The fieldIndex
    # values point into this table for each column's name.
    raw_names = string_table.split(b"\x00")
    name_lookup = {0: ""}
    cursor = 0
    for piece in raw_names:
        if piece:
            name_lookup[cursor] = piece.decode("iso-8859-1")
        cursor += len(piece) + 1  # +1 for the nul we split on

    columns: list[ColumnInfo] = []
    for fd in raw_fields:
        # fieldIndex is the byte offset where the column's name starts.
        name = name_lookup.get(fd["field_index"])
        if name is None:
            # Walk the string table to find the name at this offset.
            tail = string_table[fd["field_index"] :].split(b"\x00", 1)[0]
            name = tail.decode("iso-8859-1") if tail else f"col{len(columns)}"
        # INVARIANT: ColumnInfo.type_code is always base-typed (high-bit
        # flags stripped). This is the single producer site — every reader
        # (parse_tuple_payload, cursor._dereference_blob_columns, etc.)
        # depends on this and skips redundant base_type() calls. If you
        # ever construct ColumnInfo elsewhere, base_type() the input.
        columns.append(
            ColumnInfo(
                name=name or f"col{len(columns)}",
                type_code=base_type(fd["type_code"]),
                raw_type_code=fd["type_code"],
                encoded_length=fd["encoded_length"],
                column_start_pos=fd["column_start_pos"],
                extended_id=fd["extended_id"],
                owner_name=fd["owner_name"],
                extended_name=fd["extended_name"],
            )
        )
    return columns, metadata


# IDS type codes that are length-prefixed in the tuple payload.
# Per ``IfxSqli`` row-data extraction (see receiveFastPath case 13/15/16):
# CHAR, VARCHAR, NCHAR, NVCHAR all use ``[short length][bytes][pad if odd]``
# inside the tuple blob. LVARCHAR uses a 4-byte length prefix instead.
_LENGTH_PREFIXED_SHORT_TYPES = frozenset({
    _TC_CHAR,
    _TC_VARCHAR,
    _TC_NCHAR,
    _TC_NVCHAR,
})

# CHAR and NCHAR are **fixed-width**, space-padded to ``encoded_length``.
# VARCHAR and NVARCHAR are byte-length-prefixed. Getting NCHAR wrong is not
# a cosmetic bug: treating it as length-prefixed consumes its first
# character as a length byte, then advances the offset by that value —
# e.g. NCHAR(10) holding 'nch' reads 0x6E ('n') as a 110-byte length and
# desyncs the rest of the row, usually into a struct.error crash.
# Verified on the wire against 12.10 and 15:
#     NCHAR(10)    'nch' -> 6e 63 68 20 20 20 20 20 20 20   (10 B, padded)
#     NVARCHAR(20) 'nvc' -> 03 6e 76 63                     (1-byte prefix)
_FIXED_WIDTH_CHAR_TYPES = frozenset({_TC_CHAR, _TC_NCHAR})

_COMPOSITE_UDT_TYPES = frozenset({
    _TC_ROW,
    _TC_COLLECTION,
    _TC_SET,
    _TC_MULTISET,
    _TC_LIST,
})

_NUMERIC_TYPES = frozenset({_TC_DECIMAL, _TC_MONEY})

# Types that are fixed-width on the wire AND have a registered decoder
# in ``FIXED_WIDTHS``: SMALLINT, INT, SERIAL, SMFLOAT, FLOAT, BIGINT,
# BIGSERIAL, DATE, BOOL. These are the most common types in any real
# query, so checking them FIRST in the parse_tuple_payload dispatch
# saves ~7 frozenset/equality misses per column. Disjoint from every
# other branch's type set (verified — none of these codes appear in
# _LENGTH_PREFIXED_SHORT_TYPES, _NUMERIC_TYPES, _COMPOSITE_UDT_TYPES,
# or as DATETIME/INTERVAL/UDTFIXED/UDTVAR/LVARCHAR).
_FIXED_WIDTH_TYPES = frozenset(FIXED_WIDTHS.keys())


# Phase 37 — per-column reader strategy.
#
# parse_tuple_payload's hot loop used to evaluate the same dispatch
# decisions per column per row: "is this a fixed-width type? a
# length-prefixed string? what's the decoder?" Those decisions only
# depend on column metadata, not row data — so we make them ONCE at
# parse_describe time and emit a per-column tuple the hot loop can
# dispatch on with a single integer comparison.
#
# Reader-strategy kinds (the first element of each compiled tuple).
# Tuple shapes are documented at each kind's compile branch in
# ``compile_column_readers`` below. Common types (covering >95% of
# real-world workloads) get pre-compiled; rare types fall through
# to the legacy dispatch in parse_tuple_payload.
_RK_FIXED = 0           # (kind, width, decoder)            — INT/FLOAT/DATE/etc.
_RK_BYTE_PREFIX = 1     # (kind, decoder)                   — VARCHAR/NCHAR/NVCHAR
_RK_CHAR = 2            # (kind, width, decoder)            — fixed-width CHAR
_RK_LVARCHAR = 3        # (kind, decoder)                   — LVARCHAR (4-byte prefix)
_RK_DECIMAL = 4         # (kind, width, decoder)            — DECIMAL/MONEY
_RK_DATETIME = 5        # (kind, width, encoded_length)     — DATETIME (uses _decode_datetime)
_RK_INTERVAL = 6        # (kind, width, encoded_length)     — INTERVAL (uses _decode_interval)
_RK_LEGACY = 7          # (kind, type_code)                 — fall through to original dispatch


def compile_column_readers(columns: list[ColumnInfo]) -> list[tuple]:
    """Compile a per-column reader strategy.

    Phase 37: replaces the per-row branch-dispatch in
    ``parse_tuple_payload`` with a one-shot compilation pass at
    ``parse_describe`` time. Each column gets a tuple the hot loop
    dispatches on with a single int comparison.

    Common types (~95% of real workloads) get pre-compiled fast
    paths. Rare types (UDT/composite/CHAR-with-truncation/etc.)
    are tagged ``_RK_LEGACY`` and fall through to the legacy
    dispatch — preserves correctness on every shape we've seen
    while accelerating the hot path.
    """
    readers: list[tuple] = []
    for col in columns:
        tc = col.type_code

        if tc in _FIXED_WIDTH_TYPES:
            readers.append((_RK_FIXED, FIXED_WIDTHS[tc], DECODERS[tc]))
            continue

        if tc in _FIXED_WIDTH_CHAR_TYPES:
            # CHAR and NCHAR: fixed width, space-padded to encoded_length.
            readers.append((_RK_CHAR, col.encoded_length, DECODERS[tc]))
            continue

        if tc in _LENGTH_PREFIXED_SHORT_TYPES:
            # VARCHAR / NVARCHAR — CHAR and NCHAR excluded above.
            readers.append((_RK_BYTE_PREFIX, DECODERS[tc]))
            continue

        if tc == _TC_LVARCHAR:
            readers.append((_RK_LVARCHAR, DECODERS[tc]))
            continue

        if tc in _NUMERIC_TYPES:
            precision = (col.encoded_length >> 8) & 0xFF
            width = (precision + 1) // 2 + 1
            readers.append((_RK_DECIMAL, width, DECODERS[tc]))
            continue

        if tc == _TC_DATETIME:
            digit_count = (col.encoded_length >> 8) & 0xFF
            width = (digit_count + 1) // 2 + 1
            readers.append((_RK_DATETIME, width, col.encoded_length))
            continue

        if tc == _TC_INTERVAL:
            digit_count = (col.encoded_length >> 8) & 0xFF
            width = (digit_count + 1) // 2 + 1
            readers.append((_RK_INTERVAL, width, col.encoded_length))
            continue

        # UDT / composite / unknown — let the legacy dispatch handle it.
        readers.append((_RK_LEGACY, tc))

    return readers


# Phase 38 codegen — sentinel constants imported into the generated
# function's globals so inlined decode bodies can reference them by
# name without dotted lookups.
_INT_MIN_SENTINEL = -0x80000000
_SHORT_MIN_SENTINEL = -0x8000
_LONG_MIN_SENTINEL = -0x8000000000000000


def compile_row_decoder(
    readers: list[tuple],
    columns: list[ColumnInfo],
) -> Callable[[bytes, int, str], tuple] | None:
    """Generate a specialized row decoder for a specific column shape.

    Phase 38: takes the Phase 37 reader-list and emits a Python
    function via ``exec()`` that decodes one row of this exact shape
    in straight-line code — no per-column iteration, no per-column
    tuple-unpack, no per-column branch dispatch. Each column's
    decode logic is inlined directly.

    The generated function has signature
    ``parse_row(payload, offset, encoding) -> tuple`` and only
    references module-level helpers via its closure-equivalent
    globals dict (the ``_g`` dict below).

    Returns ``None`` if any column's reader-kind is unsupported by
    the codegen — caller falls back to the Phase 37 dispatch loop.

    The generated source is printable via ``IFX_DEBUG_CODEGEN=1``
    env var for inspection / debugging.
    """
    import os

    lines: list[str] = []
    lines.append("def parse_row(payload, offset, encoding):")
    val_names: list[str] = []

    # Map type-code → inline-decoder source for the common fixed-width
    # decoders. Inlining the decoder body eliminates one function call
    # per column — the actual codegen win. For types not in this map,
    # fall back to ``_D{i}(raw)`` referencing the decoder via globals.
    _INLINE_FIXED = {
        # type_code: lambda v, raw_var: source-snippet
        # SMALLINT (1)
        1: lambda v, r: (
            f"    {v} = _UNPACK_SHORT({r})[0]\n"
            f"    if {v} == -32768:\n"
            f"        {v} = None"
        ),
        # INT (2), SERIAL (6) — same body
        2: lambda v, r: (
            f"    {v} = _UNPACK_INT({r})[0]\n"
            f"    if {v} == -2147483648:\n"
            f"        {v} = None"
        ),
        6: lambda v, r: (
            f"    {v} = _UNPACK_INT({r})[0]\n"
            f"    if {v} == -2147483648:\n"
            f"        {v} = None"
        ),
        # BIGINT (52), BIGSERIAL (53) — same body
        52: lambda v, r: (
            f"    {v} = _UNPACK_LONG({r})[0]\n"
            f"    if {v} == -9223372036854775808:\n"
            f"        {v} = None"
        ),
        53: lambda v, r: (
            f"    {v} = _UNPACK_LONG({r})[0]\n"
            f"    if {v} == -9223372036854775808:\n"
            f"        {v} = None"
        ),
        # FLOAT (3), SMFLOAT (4)
        3: lambda v, r: (
            f"    if {r} == _DOUBLE_NULL:\n"
            f"        {v} = None\n"
            f"    else:\n"
            f"        {v} = _UNPACK_DOUBLE({r})[0]"
        ),
        4: lambda v, r: (
            f"    if {r} == _REAL_NULL:\n"
            f"        {v} = None\n"
            f"    else:\n"
            f"        {v} = _UNPACK_FLOAT({r})[0]"
        ),
        # DATE (7) — 4-byte day count from 1899-12-31
        7: lambda v, r: (
            f"    days = _UNPACK_INT({r})[0]\n"
            f"    if days == -2147483648:\n"
            f"        {v} = None\n"
            f"    else:\n"
            f"        {v} = _DATE_EPOCH + _timedelta(days=days)"
        ),
        # BOOL (45) — left to the canonical decoder. Informix BOOL is
        # ``'t'/'T'/1``, NOT bool(byte) — a truthy-byte inline would
        # silently turn ``'f'`` (102) into True.
    }

    for i, r in enumerate(readers):
        kind = r[0]
        v = f"v{i}"
        val_names.append(v)
        lines.append(f"    # Col {i}: kind={kind}")

        if kind == _RK_FIXED:
            _, width, _decoder = r
            lines.append(f"    raw = payload[offset:offset+{width}]")
            lines.append(f"    offset += {width}")
            # Find type code from the decoder identity (we don't have
            # tc directly in the reader tuple; recover via the columns
            # list).
            tc = columns[i].type_code
            inline_src = _INLINE_FIXED.get(tc)
            if inline_src is not None:
                lines.append(inline_src(v, "raw"))
            else:
                lines.append(f"    {v} = _D{i}(raw)")

        elif kind == _RK_BYTE_PREFIX:
            lines.append("    length = payload[offset]")
            lines.append("    offset += 1")
            lines.append("    raw = payload[offset:offset + length]")
            lines.append("    offset += length")
            lines.append(f"    {v} = _D{i}(raw, encoding)")

        elif kind == _RK_CHAR:
            _, width, _decoder = r
            lines.append(f"    raw = payload[offset:offset+{width}]")
            lines.append(f"    offset += {width}")
            lines.append(f"    {v} = _D{i}(raw, encoding)")

        elif kind == _RK_LVARCHAR:
            lines.append(
                "    length = int.from_bytes("
                "payload[offset:offset+4], 'big', signed=True)"
            )
            lines.append("    offset += 4")
            lines.append("    raw = payload[offset:offset + length]")
            lines.append("    offset += length")
            lines.append("    if length & 1:")
            lines.append("        offset += 1")
            lines.append(f"    {v} = _D{i}(raw, encoding)")

        elif kind == _RK_DECIMAL:
            _, width, _decoder = r
            lines.append(f"    raw = payload[offset:offset+{width}]")
            lines.append(f"    offset += {width}")
            lines.append("    try:")
            lines.append(f"        {v} = _D{i}(raw)")
            lines.append("    except NotImplementedError:")
            lines.append(f"        {v} = raw")

        elif kind == _RK_DATETIME:
            _, width, enc_len = r
            lines.append(f"    raw = payload[offset:offset+{width}]")
            lines.append(f"    offset += {width}")
            lines.append(f"    {v} = _decode_datetime(raw, {enc_len})")

        elif kind == _RK_INTERVAL:
            _, width, enc_len = r
            lines.append(f"    raw = payload[offset:offset+{width}]")
            lines.append(f"    offset += {width}")
            lines.append(f"    {v} = _decode_interval(raw, {enc_len})")

        elif kind == _RK_LEGACY:
            # Codegen for rare types: call the legacy helper. The
            # column metadata is referenced via the globals dict.
            tc = r[1]
            lines.append(
                f"    offset, {v} = _legacy_dispatch_one_column("
                f"payload, offset, {tc}, _COL{i}, encoding)"
            )

        else:
            # Unknown kind — abort codegen, caller falls back.
            return None

    if val_names:
        lines.append(f"    return ({', '.join(val_names)},)")
    else:
        lines.append("    return ()")

    src = "\n".join(lines)

    if os.environ.get("IFX_DEBUG_CODEGEN") == "1":
        import sys
        print("=== informix_db codegen ===", file=sys.stderr)
        print(src, file=sys.stderr)
        print("=== end ===", file=sys.stderr)

    # Build the globals dict for the generated function. Each column's
    # decoder (if any) is registered as ``_D<i>``; columns with the
    # _RK_LEGACY kind get their ColumnInfo as ``_COL<i>``.
    #
    # The inlined fixed-width snippets (see ``_INLINE_FIXED`` above)
    # reference precompiled struct unpackers and NULL sentinels by
    # name — they only resolve if we hand them to ``exec`` here.
    g: dict = {
        "_decode_datetime": _decode_datetime,
        "_decode_interval": _decode_interval,
        "_legacy_dispatch_one_column": _legacy_dispatch_one_column,
        "_UNPACK_SHORT": _UNPACK_SHORT,
        "_UNPACK_INT": _UNPACK_INT,
        "_UNPACK_LONG": _UNPACK_LONG,
        "_UNPACK_FLOAT": _UNPACK_FLOAT,
        "_UNPACK_DOUBLE": _UNPACK_DOUBLE,
        "_DOUBLE_NULL": _DOUBLE_NULL,
        "_REAL_NULL": _REAL_NULL,
        "_DATE_EPOCH": _DATE_EPOCH,
        "_timedelta": _timedelta,
        "int": int,  # ensure the builtin isn't shadowed
        "bool": bool,
    }
    for i, r in enumerate(readers):
        kind = r[0]
        if kind in (_RK_FIXED, _RK_CHAR, _RK_DECIMAL):
            g[f"_D{i}"] = r[2]
        elif kind in (_RK_BYTE_PREFIX, _RK_LVARCHAR):
            g[f"_D{i}"] = r[1]
        elif kind == _RK_LEGACY:
            g[f"_COL{i}"] = columns[i]

    namespace: dict = {}
    try:
        exec(compile(src, "<informix_db codegen>", "exec"), g, namespace)
    except SyntaxError:
        return None

    return namespace["parse_row"]


def _legacy_dispatch_one_column(
    payload: bytes,
    offset: int,
    tc: int,
    col: ColumnInfo,
    encoding: str,
) -> tuple[int, object]:
    """Phase 37 fallback for rare types not covered by the pre-compiled
    reader strategies (UDTFIXED, COMPOSITE UDT, UDTVAR-lvarchar, unknown).

    Mirrors the corresponding branches of the legacy ``parse_tuple_payload``
    dispatch chain but for one column at a time. Returns ``(new_offset,
    decoded_value)``.
    """
    # BLOB / CLOB locator (UDTFIXED + extended_id 10/11)
    if tc == _TC_UDTFIXED and col.extended_id in (10, 11):
        width = col.encoded_length
        raw = payload[offset:offset + width]
        offset += width
        cls = BlobLocator if col.extended_id == 10 else ClobLocator
        return offset, cls(raw=bytes(raw))

    # BOOLEAN. The server describes it as UDTFIXED (41) with
    # extended_name='boolean' and encoded_length=1, but ``encoded_length``
    # is the size of the *value*, not the field: on the wire it carries the
    # standard UDT envelope ``[1-byte null indicator][4-byte length][data]``
    # — 6 bytes total for a 1-byte value. Consuming only ``encoded_length``
    # leaves 5 bytes on the wire and desyncs every subsequent column.
    # Verified payload (INT, BOOLEAN 't', INT, VARCHAR 'tail'):
    #     00 01 b2 07 | 00 00 00 00 01 74 | 00 03 64 0e | 04 74 61 69 6c
    # The value byte is 0x74 ('t'), which _decode_bool already understands.
    if tc == _TC_UDTFIXED and (
        col.extended_name == "boolean" or col.extended_id == 5
    ):
        indicator = payload[offset]
        offset += 1
        length = int.from_bytes(payload[offset:offset + 4], "big", signed=True)
        offset += 4
        raw = payload[offset:offset + length]
        offset += length
        if indicator == 1:
            return offset, None
        return offset, bool(raw and raw[0] in (ord("t"), ord("T"), 1))

    # ROW / COLLECTION composite UDT
    if tc in _COMPOSITE_UDT_TYPES:
        indicator = payload[offset]
        offset += 1
        if indicator == 1:
            return offset, None
        length = int.from_bytes(payload[offset:offset + 4], "big", signed=True)
        offset += 4
        raw = bytes(payload[offset:offset + length])
        offset += length
        if tc == _TC_ROW:
            return offset, RowValue(raw=raw, schema=col.extended_name)
        return offset, CollectionValue(
            raw=raw,
            kind=_COLLECTION_KIND_MAP[tc],
            element_schema=col.extended_name,
        )

    # UDTVAR with extended_name=lvarchar (e.g., result of lotofile())
    if tc == _TC_UDTVAR and col.extended_name == "lvarchar":
        indicator = payload[offset]
        offset += 1
        if indicator == 1:
            return offset, None
        length = int.from_bytes(payload[offset:offset + 4], "big", signed=True)
        offset += 4
        raw = payload[offset:offset + length]
        offset += length
        if length & 1:
            offset += 1
        return offset, raw.decode(encoding)

    # Unknown — surface ``encoded_length`` bytes raw.
    width = col.encoded_length
    raw = payload[offset:offset + width]
    offset += width
    try:
        return offset, _decode_base(tc, raw, encoding)
    except NotImplementedError:
        return offset, raw


def parse_tuple_payload(
    reader: IfxStreamReader,
    columns: list[ColumnInfo],
    encoding: str = "iso-8859-1",
    readers: list[tuple] | None = None,
    row_decoder: Callable[[bytes, int, str], tuple] | None = None,
) -> tuple:
    """Parse a SQ_TUPLE payload (the SQ_TUPLE tag is already consumed).

    Per ``IfxSqli.receiveTuple``:
      ``[short warn][int size][bytes payload]``

    The payload contains column values back-to-back. For each column, the
    on-wire encoding depends on the type:

    * Fixed-width types (INT, FLOAT, DATE, BIGINT, etc.): exact byte count
      from ``FIXED_WIDTHS``.
    * Length-prefixed strings (CHAR, VARCHAR, NCHAR, NVCHAR): ``[short len]
      [bytes][pad if odd]``.
    * LVARCHAR: 4-byte length prefix instead of 2.
    * Other variable-width types (DECIMAL, DATETIME, INTERVAL, BLOBs):
      Phase 6+ — currently surfaces raw bytes from ``encoded_length``.

    ``encoding`` is forwarded to ``decode()`` for string columns. Caller
    (typically the cursor) should pass the connection's
    ``encoding`` so user-data text honors CLIENT_LOCALE.
    """
    reader.read_short()  # warn (Phase 5 surfaces)
    size = reader.read_int()
    payload = reader.read_exact(size)
    # SQ_TUPLE payload is padded to even-byte alignment on the wire.
    # Discovered empirically: a 11-byte "syscolumns" VARCHAR payload had
    # a trailing 0x00 between it and the next SQ_TUPLE tag. Consuming
    # this pad keeps the next read aligned.
    # (See docs/CAPTURES/15-py-varchar-fixed.socat.log analysis.)
    if size & 1:
        reader.read_exact(1)

    # Phase 38 fastest path: a per-result-set decoder function compiled
    # via ``exec()`` from the column shape (see ``compile_row_decoder``).
    # All per-column dispatch is eliminated — each column's decode logic
    # is inlined in straight-line code.
    if row_decoder is not None:
        return row_decoder(payload, 0, encoding)

    values: list[object] = []
    offset = 0

    # Phase 37 fast path: if the caller pre-compiled a reader-strategy
    # list, dispatch on the integer kind for each column. The compile
    # step (``compile_column_readers``) made the per-column decisions
    # ONCE; this loop just executes them. Common types (FIXED, BYTE_PREFIX,
    # CHAR, LVARCHAR, DECIMAL, DATETIME, INTERVAL) get pre-baked tuples;
    # rare types fall through to the legacy branch chain via _RK_LEGACY.
    if readers is not None:
        for r in readers:
            kind = r[0]

            if kind == _RK_FIXED:
                _, width, decoder = r
                raw = payload[offset:offset + width]
                offset += width
                values.append(decoder(raw))
                continue

            if kind == _RK_BYTE_PREFIX:
                _, decoder = r
                length = payload[offset]
                offset += 1
                raw = payload[offset:offset + length]
                offset += length
                values.append(decoder(raw, encoding))
                continue

            if kind == _RK_CHAR:
                _, width, decoder = r
                raw = payload[offset:offset + width]
                offset += width
                values.append(decoder(raw, encoding))
                continue

            if kind == _RK_LVARCHAR:
                _, decoder = r
                length = int.from_bytes(
                    payload[offset:offset + 4], "big", signed=True
                )
                offset += 4
                raw = payload[offset:offset + length]
                offset += length
                if length & 1:
                    offset += 1
                values.append(decoder(raw, encoding))
                continue

            if kind == _RK_DECIMAL:
                _, width, decoder = r
                raw = payload[offset:offset + width]
                offset += width
                try:
                    values.append(decoder(raw))
                except NotImplementedError:
                    values.append(raw)
                continue

            if kind == _RK_DATETIME:
                _, width, enc_len = r
                raw = payload[offset:offset + width]
                offset += width
                values.append(_decode_datetime(raw, enc_len))
                continue

            if kind == _RK_INTERVAL:
                _, width, enc_len = r
                raw = payload[offset:offset + width]
                offset += width
                values.append(_decode_interval(raw, enc_len))
                continue

            # _RK_LEGACY — rare type, fall back to the original dispatch.
            # Find the matching ColumnInfo (parallel index) and run the
            # legacy branch chain by recursing into the slow path. We
            # do this by setting ``readers = None`` and breaking out;
            # but since we're mid-loop, simpler: run the legacy code
            # inline via a helper.
            tc = r[1]
            col = columns[len(values)]  # parallel index — values has one entry per processed col
            offset, value = _legacy_dispatch_one_column(
                payload, offset, tc, col, encoding
            )
            values.append(value)
        return tuple(values)

    # Legacy slow path (no pre-compiled readers).
    # Note: ``col.type_code`` is *already* base-typed by ``parse_describe``
    # (see INVARIANT comment there), so we don't re-strip high-bit flags
    # here. The original code called ``base_type(col.type_code)`` per
    # column per row — pure waste. Skipping it is the single largest
    # savings in this loop.
    for col in columns:
        tc = col.type_code

        # Fast path: fixed-width types (INT, FLOAT, DATE, BIGINT, etc.)
        # are by far the most common columns in real queries. Check them
        # FIRST so we don't pay 7+ branch-misses per integer column.
        # FIXED_WIDTHS keys are disjoint from every other branch's type
        # set — see _FIXED_WIDTH_TYPES module-level comment.
        if tc in _FIXED_WIDTH_TYPES:
            width = FIXED_WIDTHS[tc]
            raw = payload[offset:offset + width]
            offset += width
            values.append(_decode_base(tc, raw, encoding))
            continue

        if tc in _LENGTH_PREFIXED_SHORT_TYPES:
            # In tuple data, VARCHAR/NCHAR/NVCHAR use a SINGLE-BYTE
            # length prefix (max 255 — IDS VARCHAR's hard limit), not
            # a short. Empirically verified against the SQ_TUPLE bytes
            # for ``SELECT tabname FROM systables`` in
            # docs/CAPTURES/13-py-varchar.socat.log:
            #     payload = 09 73 79 73 74 61 62 6c 65 73
            #             = [byte 9]["systables"]
            # CHAR and NCHAR are fixed-width per encoded_length.
            if tc in _FIXED_WIDTH_CHAR_TYPES:
                width = col.encoded_length
                raw = payload[offset:offset + width]
                offset += width
            else:
                length = payload[offset]
                offset += 1
                raw = payload[offset:offset + length]
                offset += length
            values.append(_decode_base(tc, raw, encoding))
            continue

        if tc == _TC_LVARCHAR:
            # [int length][bytes][pad if odd]
            length = int.from_bytes(payload[offset:offset + 4], "big", signed=True)
            offset += 4
            raw = payload[offset:offset + length]
            offset += length
            if length & 1:
                offset += 1
            values.append(_decode_base(tc, raw, encoding))
            continue

        # DECIMAL/MONEY: width = ceil(precision/2) + 1, where precision is
        # the high byte of encoded_length (packed as (precision << 8) | scale).
        # Per IfxRowColumn.loadColumnData and IfxToJavaDecimal byte sizing.
        if tc in _NUMERIC_TYPES:
            precision = (col.encoded_length >> 8) & 0xFF
            width = (precision + 1) // 2 + 1
            raw = payload[offset:offset + width]
            offset += width
            try:
                values.append(_decode_base(tc, raw))
            except NotImplementedError:
                values.append(raw)
            continue

        # DATETIME: width = ceil(digit_count/2) + 1, where digit_count is the
        # high byte of encoded_length (packed as (digit_count << 8) |
        # (start_TU << 4) | end_TU). The decoder needs the qualifier too,
        # so we call it directly here rather than via the dispatch.
        if tc == _TC_DATETIME:
            digit_count = (col.encoded_length >> 8) & 0xFF
            width = (digit_count + 1) // 2 + 1
            raw = payload[offset:offset + width]
            offset += width
            values.append(_decode_datetime(raw, col.encoded_length))
            continue

        # INTERVAL: same width formula as DATETIME — high byte of
        # encoded_length holds the total digit count across all fields,
        # and the wire bytes are ``[head][digit pairs]`` (one head byte
        # plus ceil(digit_count/2) digit pairs). Like DATETIME, the
        # qualifier is needed at decode time, so we bypass the generic
        # dispatch.
        if tc == _TC_INTERVAL:
            digit_count = (col.encoded_length >> 8) & 0xFF
            width = (digit_count + 1) // 2 + 1
            raw = payload[offset:offset + width]
            offset += width
            values.append(_decode_interval(raw, col.encoded_length))
            continue

        # BLOB / CLOB (smart-LOBs): the SQ_DESCRIBE response presents
        # these as UDTFIXED (type 41) with extended_id 10 (BLOB) or 11
        # (CLOB) and encoded_length = 72 (locator size). The 72 bytes
        # we read here are an opaque server-side reference, NOT the
        # actual data. Phase 10 lets users fetch via lotofile + SQ_FILE.
        if tc == _TC_UDTFIXED and col.extended_id in (10, 11):
            width = col.encoded_length
            raw = payload[offset:offset + width]
            offset += width
            cls = BlobLocator if col.extended_id == 10 else ClobLocator
            values.append(cls(raw=bytes(raw)))
            continue

        # BOOLEAN — UDT envelope, not a bare byte. See the matching branch
        # in _legacy_dispatch_one_column for the wire evidence.
        if tc == _TC_UDTFIXED and (
            col.extended_name == "boolean" or col.extended_id == 5
        ):
            indicator = payload[offset]
            offset += 1
            length = int.from_bytes(
                payload[offset:offset + 4], "big", signed=True
            )
            offset += 4
            raw = payload[offset:offset + length]
            offset += length
            if indicator == 1:
                values.append(None)
            else:
                values.append(bool(raw and raw[0] in (ord("t"), ord("T"), 1)))
            continue

        # ROW / COLLECTION (Phase 12): composite UDTs. Wire format is
        # ``[byte ind][int length][bytes]`` — same shape as
        # UDTVAR(lvarchar) above, but the payload semantics are a
        # textual representation of the composite (e.g.,
        # ``ROW('Alice',30         )`` or ``LIST{10,20,30}``) when
        # selected with default options. JDBC requests a richer
        # binary-with-schema format that's ~30x larger; we don't.
        #
        # We surface the bytes wrapped in a typed object and let the
        # user parse the textual form themselves. Type codes:
        # ROW=22, COLLECTION=23, SET=19, MULTISET=20, LIST=21.
        if tc in _COMPOSITE_UDT_TYPES:
            indicator = payload[offset]
            offset += 1
            if indicator == 1:  # null
                values.append(None)
                continue
            length = int.from_bytes(
                payload[offset:offset + 4], "big", signed=True
            )
            offset += 4
            raw = bytes(payload[offset:offset + length])
            offset += length
            if tc == _TC_ROW:
                values.append(RowValue(raw=raw, schema=col.extended_name))
            else:
                values.append(
                    CollectionValue(
                        raw=raw,
                        kind=_COLLECTION_KIND_MAP[tc],
                        element_schema=col.extended_name,
                    )
                )
            continue

        # UDTVAR (type 40) with extended_name="lvarchar": this is what
        # functions like ``lotofile`` return — a length-prefixed string
        # wrapped as a UDT. The wire format adds a 1-byte indicator
        # prefix BEFORE the LVARCHAR ``[int len][bytes]``. Empirically
        # verified against ``SELECT lotofile(...)`` row data — the
        # leading ``00`` is null indicator (0=not null, 1=null per UDT
        # convention).
        if tc == _TC_UDTVAR and col.extended_name == "lvarchar":
            indicator = payload[offset]
            offset += 1
            if indicator == 1:
                values.append(None)
                continue
            length = int.from_bytes(
                payload[offset:offset + 4], "big", signed=True
            )
            offset += 4
            raw = payload[offset:offset + length]
            offset += length
            if length & 1:
                offset += 1
            values.append(raw.decode(encoding))
            continue

        # Unknown / unhandled type fall-through. The fast-path at the
        # top of this loop already handled all FIXED_WIDTHS-registered
        # types (INT, FLOAT, DATE, etc.); the explicit branches above
        # handle every other known wire shape. Anything reaching here
        # is a type code we don't recognize — surface ``encoded_length``
        # bytes raw and let the decoder dispatch (or its fallback) react.
        width = col.encoded_length
        raw = payload[offset:offset + width]
        offset += width
        try:
            values.append(_decode_base(tc, raw, encoding))
        except NotImplementedError:
            values.append(raw)
    # Phase 28 note on bounds checking:
    # An end-of-loop ``offset > len(payload)`` check was attempted but
    # firing on the UDTVAR(lvarchar) branch's trailing-pad logic
    # (``if length & 1: offset += 1``), which can intentionally
    # over-advance by 1 when the field is the last column. The wire is
    # NOT desynced in that case — ``payload`` is a fully-extracted
    # bytes object, so over-reads return empty slices that flow
    # harmlessly through unused branches (the UDTVAR pad isn't decoded).
    # Real silent-corruption surfaces are localized to variable-width
    # length prefixes (caught by struct.error in fixed-width decoders,
    # by Python's slicing semantics for strings — short = harmless).
    # If a future protocol message produces actual garbage here, add a
    # branch-local check at the offending dispatch path.
    return tuple(values)