# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.

"""Python implementation of Paimon GenericVariant — a thin storage container.

Paimon's VARIANT type is stored as two byte arrays:
    value    – encodes the structure and leaf values (Parquet Variant binary spec)
    metadata – key dictionary for object field names

pypaimon exposes VARIANT columns as Arrow struct<value: binary NOT NULL,
metadata: binary NOT NULL>.  This class is a convenience wrapper that lets
Python code construct (for writing) or inspect (for debugging) those bytes.

Variant semantics such as path extraction, type casting, and JSONPath queries
are the responsibility of the compute engine or application layer, not of
pypaimon (the storage layer).

Primary entry points:
    GenericVariant.from_python(obj)        – build from a Python object (for writing)
    GenericVariant(value, metadata)        – wrap raw bytes read from a VARIANT column
    GenericVariant.from_arrow_struct(d)    – wrap a row dict from a VARIANT Arrow column
    GenericVariant.to_arrow_array(variants)– convert a list to a PyArrow StructArray

Inspection helpers (for debugging/testing):
    v.to_python() – decode to native Python objects
    v.value()     – raw value bytes
    v.metadata()  – raw metadata bytes
"""

import datetime
import decimal as _decimal
import enum
import struct
import uuid as _uuid

from pypaimon.data._variant_binary import (
    _SHORT_STR, _OBJECT, _ARRAY,
    _U8_MAX, _U32_SIZE,
    _VERSION, _VERSION_MASK,
    _read_unsigned, _get_int_size,
    _primitive_header, _object_header, _array_header,
)

# ---------------------------------------------------------------------------
# Constants (matching GenericVariantUtil.java)
# ---------------------------------------------------------------------------

_NULL = 0
_TRUE = 1
_FALSE = 2
_INT1 = 3
_INT2 = 4
_INT4 = 5
_INT8 = 6
_DOUBLE = 7
_DECIMAL4 = 8
_DECIMAL8 = 9
_DECIMAL16 = 10
_DATE = 11
_TIMESTAMP = 12
_TIMESTAMP_NTZ = 13
_FLOAT = 14
_BINARY = 15
_LONG_STR = 16
_UUID = 20

_SIZE_LIMIT = 128 * 1024 * 1024
_MAX_SHORT_STR_SIZE = 0x3F   # 63
_MAX_DECIMAL4_PRECISION = 9
_MAX_DECIMAL8_PRECISION = 18
_MAX_DECIMAL16_PRECISION = 38

# Epoch for date/timestamp conversions (used by to_python)
_EPOCH_DATE = datetime.date(1970, 1, 1)
_EPOCH_DT_UTC = datetime.datetime(1970, 1, 1, tzinfo=datetime.timezone.utc)
_EPOCH_DT_NTZ = datetime.datetime(1970, 1, 1)


class _Type(enum.Enum):
    """Internal high-level variant value types (many-to-one from wire types)."""
    OBJECT = 'OBJECT'
    ARRAY = 'ARRAY'
    NULL = 'NULL'
    BOOLEAN = 'BOOLEAN'
    LONG = 'LONG'
    STRING = 'STRING'
    DOUBLE = 'DOUBLE'
    DECIMAL = 'DECIMAL'
    DATE = 'DATE'
    TIMESTAMP = 'TIMESTAMP'
    TIMESTAMP_NTZ = 'TIMESTAMP_NTZ'
    FLOAT = 'FLOAT'
    BINARY = 'BINARY'
    UUID = 'UUID'


_PRIMITIVE_TYPE_MAP = {
    _NULL: _Type.NULL,
    _TRUE: _Type.BOOLEAN, _FALSE: _Type.BOOLEAN,
    _INT1: _Type.LONG, _INT2: _Type.LONG, _INT4: _Type.LONG, _INT8: _Type.LONG,
    _DOUBLE: _Type.DOUBLE,
    _DECIMAL4: _Type.DECIMAL, _DECIMAL8: _Type.DECIMAL, _DECIMAL16: _Type.DECIMAL,
    _DATE: _Type.DATE,
    _TIMESTAMP: _Type.TIMESTAMP,
    _TIMESTAMP_NTZ: _Type.TIMESTAMP_NTZ,
    _FLOAT: _Type.FLOAT,
    _BINARY: _Type.BINARY,
    _LONG_STR: _Type.STRING,
    _UUID: _Type.UUID,
}
_PRIMITIVE_FIXED_SIZES = {
    _NULL: 1, _TRUE: 1, _FALSE: 1,
    _INT1: 2, _INT2: 3, _INT4: 5, _INT8: 9,
    _DOUBLE: 9, _FLOAT: 5, _DATE: 5,
    _TIMESTAMP: 9, _TIMESTAMP_NTZ: 9,
    _DECIMAL4: 6, _DECIMAL8: 10, _DECIMAL16: 18,
    _UUID: 17,
}
_LONG_FAMILY_SIZES = {
    _INT1: 1, _INT2: 2, _INT4: 4, _INT8: 8,
    _DATE: 4, _TIMESTAMP: 8, _TIMESTAMP_NTZ: 8,
}


# ---------------------------------------------------------------------------
# Low-level binary utilities
# ---------------------------------------------------------------------------

def _read_signed(data, pos, n):
    return int.from_bytes(data[pos:pos + n], 'little', signed=True)


def _write_le(buf, pos, value, n):
    buf[pos:pos + n] = value.to_bytes(n, 'little')


def _short_str_header(size):
    return (size << 2) | _SHORT_STR


def _variant_get_type(value, pos):
    b = value[pos]
    basic_type = b & 0x3
    type_info = (b >> 2) & 0x3F
    if basic_type == _SHORT_STR:
        return _Type.STRING
    if basic_type == _OBJECT:
        return _Type.OBJECT
    if basic_type == _ARRAY:
        return _Type.ARRAY
    t = _PRIMITIVE_TYPE_MAP.get(type_info)
    if t is None:
        raise ValueError(f'Unknown primitive variant type id: {type_info}')
    return t


def _value_size(value, pos):
    b = value[pos]
    basic_type = b & 0x3
    type_info = (b >> 2) & 0x3F
    if basic_type == _SHORT_STR:
        return 1 + type_info
    if basic_type == _OBJECT:
        return _handle_object(
            value, pos,
            lambda size, id_size, offset_size, id_start, offset_start, data_start: (
                data_start - pos + _read_unsigned(
                    value, offset_start + size * offset_size, offset_size)
            )
        )
    if basic_type == _ARRAY:
        return _handle_array(
            value, pos,
            lambda size, offset_size, offset_start, data_start: (
                data_start - pos + _read_unsigned(
                    value, offset_start + size * offset_size, offset_size)
            )
        )
    size = _PRIMITIVE_FIXED_SIZES.get(type_info)
    if size is not None:
        return size
    if type_info in (_BINARY, _LONG_STR):
        return 1 + _U32_SIZE + _read_unsigned(value, pos + 1, _U32_SIZE)
    raise ValueError(f'Unknown primitive type id: {type_info}')


def _handle_object(value, pos, handler):
    b = value[pos]
    type_info = (b >> 2) & 0x3F
    large_size = bool((type_info >> 4) & 0x1)
    size_bytes = _U32_SIZE if large_size else 1
    size = _read_unsigned(value, pos + 1, size_bytes)
    id_size = ((type_info >> 2) & 0x3) + 1
    offset_size = (type_info & 0x3) + 1
    id_start = pos + 1 + size_bytes
    offset_start = id_start + size * id_size
    data_start = offset_start + (size + 1) * offset_size
    return handler(size, id_size, offset_size, id_start, offset_start, data_start)


def _handle_array(value, pos, handler):
    b = value[pos]
    type_info = (b >> 2) & 0x3F
    large_size = bool((type_info >> 2) & 0x1)
    size_bytes = _U32_SIZE if large_size else 1
    size = _read_unsigned(value, pos + 1, size_bytes)
    offset_size = (type_info & 0x3) + 1
    offset_start = pos + 1 + size_bytes
    data_start = offset_start + (size + 1) * offset_size
    return handler(size, offset_size, offset_start, data_start)


def _get_metadata_key(metadata, key_id):
    offset_size = ((metadata[0] >> 6) & 0x3) + 1
    dict_size = _read_unsigned(metadata, 1, offset_size)
    if key_id >= dict_size:
        raise ValueError('MALFORMED_VARIANT: key id out of range')
    string_start = 1 + (dict_size + 2) * offset_size
    offset = _read_unsigned(metadata, 1 + (key_id + 1) * offset_size, offset_size)
    next_offset = _read_unsigned(metadata, 1 + (key_id + 2) * offset_size, offset_size)
    return metadata[string_start + offset:string_start + next_offset].decode('utf-8')


# ---------------------------------------------------------------------------
# _GenericVariantBuilder (for from_python)
# ---------------------------------------------------------------------------

class _GenericVariantBuilder:
    """Builds a GenericVariant from Python values or JSON strings."""

    def __init__(self):
        self._buf = bytearray(128)
        self._pos = 0
        self._dict = {}
        self._keys = []

    def _get_or_add_key(self, key):
        if key not in self._dict:
            kid = len(self._keys)
            self._dict[key] = kid
            self._keys.append(key.encode('utf-8'))
        return self._dict[key]

    def _ensure(self, n):
        needed = self._pos + n
        if needed > len(self._buf):
            new_cap = max(needed, len(self._buf) * 2)
            new_buf = bytearray(new_cap)
            new_buf[:self._pos] = self._buf[:self._pos]
            self._buf = new_buf

    def _write_byte(self, b):
        self._ensure(1)
        self._buf[self._pos] = b & 0xFF
        self._pos += 1

    def _write_le(self, value, n):
        self._ensure(n)
        _write_le(self._buf, self._pos, value, n)
        self._pos += n

    def append_null(self):
        self._write_byte(_primitive_header(_NULL))

    def append_boolean(self, b):
        self._write_byte(_primitive_header(_TRUE if b else _FALSE))

    def append_long(self, n):
        if -(1 << 7) <= n < (1 << 7):
            self._write_byte(_primitive_header(_INT1))
            self._write_le(n & 0xFF, 1)
        elif -(1 << 15) <= n < (1 << 15):
            self._write_byte(_primitive_header(_INT2))
            self._write_le(n & 0xFFFF, 2)
        elif -(1 << 31) <= n < (1 << 31):
            self._write_byte(_primitive_header(_INT4))
            self._write_le(n & 0xFFFFFFFF, 4)
        else:
            self._write_byte(_primitive_header(_INT8))
            self._write_le(n & 0xFFFFFFFFFFFFFFFF, 8)

    def append_double(self, d):
        self._write_byte(_primitive_header(_DOUBLE))
        self._ensure(8)
        struct.pack_into('<d', self._buf, self._pos, d)
        self._pos += 8

    def append_float(self, f):
        self._write_byte(_primitive_header(_FLOAT))
        self._ensure(4)
        struct.pack_into('<f', self._buf, self._pos, f)
        self._pos += 4

    def append_decimal(self, d):
        d = d.normalize()
        sign, digits, exponent = d.as_tuple()
        if exponent > 0:
            raise ValueError(
                f'append_decimal requires a non-positive exponent (got {d!r}); '
                'use append_double() for Decimal values with positive exponents'
            )
        unscaled = int(''.join(str(x) for x in digits))
        if sign:
            unscaled = -unscaled
        scale = -exponent if exponent < 0 else 0
        precision = len(digits)

        if scale <= _MAX_DECIMAL4_PRECISION and precision <= _MAX_DECIMAL4_PRECISION:
            self._write_byte(_primitive_header(_DECIMAL4))
            self._write_byte(scale)
            self._write_le(unscaled & 0xFFFFFFFF, 4)
        elif scale <= _MAX_DECIMAL8_PRECISION and precision <= _MAX_DECIMAL8_PRECISION:
            self._write_byte(_primitive_header(_DECIMAL8))
            self._write_byte(scale)
            self._write_le(unscaled & 0xFFFFFFFFFFFFFFFF, 8)
        else:
            self._write_byte(_primitive_header(_DECIMAL16))
            self._write_byte(scale)
            self._ensure(16)
            raw = unscaled.to_bytes(16, 'little', signed=True)
            self._buf[self._pos:self._pos + 16] = raw
            self._pos += 16

    def append_string(self, s):
        text = s.encode('utf-8')
        if len(text) <= _MAX_SHORT_STR_SIZE:
            self._write_byte(_short_str_header(len(text)))
        else:
            self._write_byte(_primitive_header(_LONG_STR))
            self._write_le(len(text), _U32_SIZE)
        self._ensure(len(text))
        self._buf[self._pos:self._pos + len(text)] = text
        self._pos += len(text)

    def append_binary(self, b):
        self._write_byte(_primitive_header(_BINARY))
        self._write_le(len(b), _U32_SIZE)
        self._ensure(len(b))
        self._buf[self._pos:self._pos + len(b)] = b
        self._pos += len(b)

    def append_date(self, days_since_epoch):
        self._write_byte(_primitive_header(_DATE))
        self._write_le(days_since_epoch & 0xFFFFFFFF, 4)

    def append_timestamp(self, micros_since_epoch):
        self._write_byte(_primitive_header(_TIMESTAMP))
        self._write_le(micros_since_epoch & 0xFFFFFFFFFFFFFFFF, 8)

    def append_timestamp_ntz(self, micros_since_epoch):
        self._write_byte(_primitive_header(_TIMESTAMP_NTZ))
        self._write_le(micros_since_epoch & 0xFFFFFFFFFFFFFFFF, 8)

    def _finish_writing_object(self, start, fields):
        fields.sort(key=lambda f: f[0])
        for i in range(1, len(fields)):
            if fields[i][0] == fields[i - 1][0]:
                raise ValueError('Duplicate key in variant object')

        size = len(fields)
        data_size = self._pos - start
        large_size = size > _U8_MAX
        size_bytes = _U32_SIZE if large_size else 1
        max_id = max((f[1] for f in fields), default=0)
        id_size = _get_int_size(max_id)
        offset_size = _get_int_size(data_size)
        header_size = 1 + size_bytes + size * id_size + (size + 1) * offset_size

        self._ensure(header_size)
        dst = start + header_size
        src = start
        self._buf[dst:dst + data_size] = self._buf[src:src + data_size]
        self._pos += header_size

        self._buf[start] = _object_header(large_size, id_size, offset_size)
        _write_le(self._buf, start + 1, size, size_bytes)
        id_start = start + 1 + size_bytes
        offset_start = id_start + size * id_size
        for i, (_, fid, offset) in enumerate(fields):
            _write_le(self._buf, id_start + i * id_size, fid, id_size)
            _write_le(self._buf, offset_start + i * offset_size, offset, offset_size)
        _write_le(self._buf, offset_start + size * offset_size, data_size, offset_size)

    def _finish_writing_array(self, start, offsets):
        size = len(offsets)
        data_size = self._pos - start
        large_size = size > _U8_MAX
        size_bytes = _U32_SIZE if large_size else 1
        offset_size = _get_int_size(data_size)
        header_size = 1 + size_bytes + (size + 1) * offset_size

        self._ensure(header_size)
        dst = start + header_size
        self._buf[dst:dst + data_size] = self._buf[start:start + data_size]
        self._pos += header_size

        self._buf[start] = _array_header(large_size, offset_size)
        _write_le(self._buf, start + 1, size, size_bytes)
        offset_start = start + 1 + size_bytes
        for i, off in enumerate(offsets):
            _write_le(self._buf, offset_start + i * offset_size, off, offset_size)
        _write_le(self._buf, offset_start + size * offset_size, data_size, offset_size)

    def build_python(self, obj):
        """Recursively encode a Python value into the variant binary buffer."""
        if obj is None:
            self.append_null()
        elif isinstance(obj, bool):
            self.append_boolean(obj)
        elif isinstance(obj, int):
            self.append_long(obj)
        elif isinstance(obj, float):
            self.append_double(obj)
        elif isinstance(obj, _decimal.Decimal):
            self._try_decimal_or_double(obj)
        elif isinstance(obj, str):
            self.append_string(obj)
        elif isinstance(obj, dict):
            fields = []
            start = self._pos
            for key, val in obj.items():
                fid = self._get_or_add_key(key)
                offset = self._pos - start
                fields.append((key, fid, offset))
                self.build_python(val)
            self._finish_writing_object(start, fields)
        elif isinstance(obj, (list, tuple)):
            elem_offsets = []
            start = self._pos
            for val in obj:
                elem_offsets.append(self._pos - start)
                self.build_python(val)
            self._finish_writing_array(start, elem_offsets)
        elif isinstance(obj, bytes):
            self.append_binary(obj)
        else:
            raise TypeError(f'Unsupported Python type for variant encoding: {type(obj).__name__}')

    def _try_decimal_or_double(self, d):
        try:
            sign, digits, exponent = d.as_tuple()
            if exponent > 0:
                self.append_double(float(d))
                return
            scale = -exponent if exponent < 0 else 0
            precision = len(digits)
            if scale <= _MAX_DECIMAL16_PRECISION and precision <= _MAX_DECIMAL16_PRECISION:
                self.append_decimal(d)
                return
        except (ArithmeticError, ValueError):
            pass
        self.append_double(float(d))

    def result(self):
        """Build metadata and return the completed GenericVariant."""
        n_keys = len(self._keys)
        total_str_size = sum(len(k) for k in self._keys)
        max_size = max(total_str_size, n_keys, 0)
        offset_size = _get_int_size(max_size) if max_size > 0 else 1

        offset_start = 1 + offset_size
        string_start = offset_start + (n_keys + 1) * offset_size
        metadata_size = string_start + total_str_size

        metadata = bytearray(metadata_size)
        metadata[0] = _VERSION | ((offset_size - 1) << 6)
        _write_le(metadata, 1, n_keys, offset_size)

        current_offset = 0
        for i, key_bytes in enumerate(self._keys):
            _write_le(metadata, offset_start + i * offset_size, current_offset, offset_size)
            klen = len(key_bytes)
            metadata[string_start + current_offset:string_start + current_offset + klen] = key_bytes
            current_offset += klen
        _write_le(metadata, offset_start + n_keys * offset_size, current_offset, offset_size)

        return GenericVariant(bytes(self._buf[:self._pos]), bytes(metadata))


# ---------------------------------------------------------------------------
# GenericVariant
# ---------------------------------------------------------------------------

class GenericVariant:
    """Storage container for a Paimon/Parquet VARIANT value.

    A VARIANT value is stored as two byte arrays:
        value    – encoded payload (Parquet Variant binary spec)
        metadata – key dictionary for object field names

    pypaimon exposes VARIANT columns as Arrow struct arrays with exactly these
    two fields.  This class helps Python code build or inspect those bytes.

    **Writing example**::

        import pyarrow as pa
        from pypaimon.data.generic_variant import GenericVariant

        gv1 = GenericVariant.from_python({'age': 30, 'city': 'Beijing'})
        gv2 = GenericVariant.from_python([1, 2, 3])
        # Create an Arrow StructArray ready for writing
        col = GenericVariant.to_arrow_array([gv1, gv2, None])
        table = pa.table({'id': [1, 2, 3], 'payload': col})

    **Reading example**::

        result = table_read.to_arrow(splits)
        # 'payload' column is struct<value: binary, metadata: binary>
        for row in result.column('payload').to_pylist():
            if row is not None:
                gv = GenericVariant.from_arrow_struct(row)
                print(gv.to_python()) # e.g. {'age': 30, 'city': 'Beijing'}
    """

    __slots__ = ('_value', '_metadata', '_pos')

    def __init__(self, value: bytes, metadata: bytes, _pos: int = 0):
        self._value = bytes(value)
        self._metadata = bytes(metadata)
        self._pos = _pos
        if len(metadata) < 1 or (metadata[0] & _VERSION_MASK) != _VERSION:
            raise ValueError('MALFORMED_VARIANT: invalid metadata version')

    # -- constructors --

    @classmethod
    def from_python(cls, obj) -> 'GenericVariant':
        """Encode a Python object (dict / list / str / int / float / bool / None) as VARIANT.

        Use this when writing VARIANT data from Python::

            gv = GenericVariant.from_python({'name': 'Alice', 'age': 30})
            col = GenericVariant.to_arrow_array([gv])
        """
        builder = _GenericVariantBuilder()
        builder.build_python(obj)
        return builder.result()

    @classmethod
    def from_arrow_struct(cls, d: dict) -> 'GenericVariant':
        """Wrap raw bytes from a PyArrow VARIANT struct: {'value': bytes, 'metadata': bytes}.

        Use this on the read path after reading a VARIANT column::

            for row in result.column("payload").to_pylist():
                if row is not None:
                    gv = GenericVariant.from_arrow_struct(row)
                    print(gv.to_python())
        """
        return cls(bytes(d['value']), bytes(d['metadata']))

    @classmethod
    def to_arrow_array(cls, variants):
        """Convert a list of GenericVariant (or None) to a PyArrow StructArray.

        The returned array has the canonical VARIANT layout::

            struct<value: binary NOT NULL, metadata: binary NOT NULL>

        Pass None in the list to represent a SQL NULL (absent VARIANT value).

        Example::

            gv1 = GenericVariant.from_python({'age': 30})
            gv2 = GenericVariant.from_python([1, 2, 3])
            col = GenericVariant.to_arrow_array([gv1, gv2, None])
            table = pa.table({'id': [1, 2, 3], 'payload': col})
        """
        import pyarrow as _pa

        values = []
        metadatas = []
        mask = []
        for v in variants:
            if v is None:
                values.append(b'')
                metadatas.append(b'')
                mask.append(True)
            else:
                values.append(v.value())
                metadatas.append(v.metadata())
                mask.append(False)

        variant_type = _pa.struct([
            _pa.field('value', _pa.binary(), nullable=False),
            _pa.field('metadata', _pa.binary(), nullable=False),
        ])
        return _pa.StructArray.from_arrays(
            [_pa.array(values, type=_pa.binary()),
             _pa.array(metadatas, type=_pa.binary())],
            fields=[variant_type[0], variant_type[1]],
            mask=_pa.array(mask, type=_pa.bool_()),
        )

    # -- raw bytes --

    def value(self) -> bytes:
        """Return the value payload bytes."""
        if self._pos == 0:
            return self._value
        size = _value_size(self._value, self._pos)
        return self._value[self._pos:self._pos + size]

    def metadata(self) -> bytes:
        """Return the metadata (key-dictionary) bytes."""
        return self._metadata

    # -- inspection helpers (for debugging / testing) --

    def to_python(self):
        """Decode the variant to native Python objects.

        Object  → dict
        Array   → list
        Boolean → bool
        Integer → int
        Double/Float → float
        Decimal → decimal.Decimal
        String  → str
        Date    → datetime.date
        Timestamp → datetime.datetime (UTC-aware)
        Timestamp_NTZ → datetime.datetime (naive)
        Binary  → bytes
        Null    → None

        Useful for debugging and testing.  Variant semantics and path-based
        queries are the responsibility of the application layer.
        """
        return self._to_python_impl(self._value, self._metadata, self._pos)

    def _to_python_impl(self, value, metadata, pos):
        vtype = _variant_get_type(value, pos)
        b = value[pos]
        basic_type = b & 0x3
        type_info = (b >> 2) & 0x3F

        if vtype == _Type.NULL:
            return None
        if vtype == _Type.BOOLEAN:
            return type_info == _TRUE
        if vtype == _Type.LONG:
            n = _LONG_FAMILY_SIZES.get(type_info)
            return _read_signed(value, pos + 1, n)
        if vtype == _Type.DOUBLE:
            return struct.unpack_from('<d', value, pos + 1)[0]
        if vtype == _Type.FLOAT:
            return float(struct.unpack_from('<f', value, pos + 1)[0])
        if vtype == _Type.DECIMAL:
            scale = value[pos + 1] & 0xFF
            if type_info == _DECIMAL4:
                unscaled = _read_signed(value, pos + 2, 4)
            elif type_info == _DECIMAL8:
                unscaled = _read_signed(value, pos + 2, 8)
            else:
                raw = bytes(value[pos + 2:pos + 18])
                unscaled = int.from_bytes(raw, 'little', signed=True)
            return _decimal.Decimal(unscaled) / (_decimal.Decimal(10) ** scale)
        if vtype == _Type.STRING:
            if basic_type == _SHORT_STR:
                return value[pos + 1:pos + 1 + type_info].decode('utf-8')
            length = _read_unsigned(value, pos + 1, _U32_SIZE)
            return value[pos + 1 + _U32_SIZE:pos + 1 + _U32_SIZE + length].decode('utf-8')
        if vtype == _Type.DATE:
            days = _read_signed(value, pos + 1, 4)
            return _EPOCH_DATE + datetime.timedelta(days=days)
        if vtype == _Type.TIMESTAMP:
            micros = _read_signed(value, pos + 1, 8)
            return _EPOCH_DT_UTC + datetime.timedelta(microseconds=micros)
        if vtype == _Type.TIMESTAMP_NTZ:
            micros = _read_signed(value, pos + 1, 8)
            return _EPOCH_DT_NTZ + datetime.timedelta(microseconds=micros)
        if vtype == _Type.BINARY:
            length = _read_unsigned(value, pos + 1, _U32_SIZE)
            return bytes(value[pos + 1 + _U32_SIZE:pos + 1 + _U32_SIZE + length])
        if vtype == _Type.UUID:
            # 16 bytes: two little-endian int64 (msb, lsb) → standard UUID
            msb = _read_unsigned(value, pos + 1, 8)
            lsb = _read_unsigned(value, pos + 9, 8)
            return _uuid.UUID(int=(msb << 64) | lsb)
        if vtype == _Type.OBJECT:
            def _build_dict(size, id_size, offset_size, id_start, offset_start, data_start):
                result = {}
                for i in range(size):
                    fid = _read_unsigned(value, id_start + id_size * i, id_size)
                    key = _get_metadata_key(metadata, fid)
                    offset = _read_unsigned(value, offset_start + offset_size * i, offset_size)
                    result[key] = self._to_python_impl(value, metadata, data_start + offset)
                return result
            return _handle_object(value, pos, _build_dict)
        if vtype == _Type.ARRAY:
            def _build_list(size, offset_size, offset_start, data_start):
                result = []
                for i in range(size):
                    offset = _read_unsigned(value, offset_start + offset_size * i, offset_size)
                    result.append(self._to_python_impl(value, metadata, data_start + offset))
                return result
            return _handle_array(value, pos, _build_list)
        return None

    # -- dunder --

    def __repr__(self) -> str:
        return f'GenericVariant({self.to_python()!r})'

    def __str__(self) -> str:
        return str(self.to_python())

    def __eq__(self, other) -> bool:
        if not isinstance(other, GenericVariant):
            return NotImplemented
        return self.value() == other.value() and self._metadata == other._metadata

    def __hash__(self):
        return hash((self.value(), self._metadata))