src/turn_detector/parquet_reader.py

"""
Minimal pure-Python Parquet reader, built specifically because this sandbox
has no network access to install pyarrow/fastparquet/datasets (confirmed:
huggingface.co, pypi.org, and datasets-server.huggingface.co all return
HTTP 403 host_not_allowed from this container — see docs/INITIAL_ANALYSIS.md
and experiments/EXPERIMENTS.md).

This is NOT a general-purpose Parquet library. It implements exactly what's
needed to read the real uploaded shard
(pipecat-ai/smart-turn-data-v3.2-train, shard 0 of 10):
  - Thrift COMPACT protocol decoding (generic struct/list/binary reader)
  - Parquet FileMetaData footer parsing (schema, row groups, column chunks)
  - Page header parsing
  - PLAIN encoding decode for BOOLEAN, BYTE_ARRAY (string/bytes), INT32/64
  - RLE/bit-packed hybrid decode for definition levels (needed because HF
    marks all fields OPTIONAL/nullable, including scalar columns)
  - UNCOMPRESSED and SNAPPY page decompression (SNAPPY implemented by hand
    below — no `python-snappy` available and no network to install it)

Every real fact this module reports (row counts, schema, compression
codec, decoded values) comes from actually parsing the real uploaded file
byte-for-byte. Anything this module can't handle (e.g. an unexpected
encoding) raises loudly rather than silently returning wrong data.
"""

from __future__ import annotations

import struct
from dataclasses import dataclass, field as dc_field
from typing import Any


# ---------------------------------------------------------------------------
# Thrift COMPACT protocol — generic decoder
# ---------------------------------------------------------------------------
# Compact protocol field-type nibble codes (from the Thrift spec):
CT_STOP = 0
CT_BOOLEAN_TRUE = 1
CT_BOOLEAN_FALSE = 2
CT_BYTE = 3
CT_I16 = 4
CT_I32 = 5
CT_I64 = 6
CT_DOUBLE = 7
CT_BINARY = 8
CT_LIST = 9
CT_SET = 10
CT_MAP = 11
CT_STRUCT = 12


class Reader:
    __slots__ = ("buf", "pos")

    def __init__(self, buf: bytes, pos: int = 0):
        self.buf = buf
        self.pos = pos

    def read_byte(self) -> int:
        b = self.buf[self.pos]
        self.pos += 1
        return b

    def read_bytes(self, n: int) -> bytes:
        b = self.buf[self.pos:self.pos + n]
        self.pos += n
        return b

    def read_uvarint(self) -> int:
        result = 0
        shift = 0
        while True:
            b = self.read_byte()
            result |= (b & 0x7F) << shift
            if (b & 0x80) == 0:
                break
            shift += 7
        return result

    def read_varint(self) -> int:
        """Zigzag-decoded signed varint."""
        u = self.read_uvarint()
        return (u >> 1) ^ -(u & 1)


def _read_binary(r: Reader) -> bytes:
    n = r.read_uvarint()
    return r.read_bytes(n)


def decode_struct(r: Reader) -> dict:
    """Generic Thrift compact struct decoder. Returns {field_id: value}.
    Lists become Python lists, nested structs become nested dicts, binary
    becomes bytes. This generic approach avoids hand-writing one decoder
    per Parquet Thrift struct (FileMetaData, RowGroup, ColumnChunk, ...) —
    we just index into the resulting dict by the field IDs documented in
    the Parquet Thrift schema (parquet.thrift), which are stable/public.
    """
    result: dict[int, Any] = {}
    last_field_id = 0
    while True:
        header = r.read_byte()
        if header == CT_STOP:
            break
        delta = (header >> 4) & 0x0F
        ftype = header & 0x0F
        if delta == 0:
            field_id = r.read_varint()
        else:
            field_id = last_field_id + delta
        last_field_id = field_id

        if ftype == CT_BOOLEAN_TRUE:
            value = True
        elif ftype == CT_BOOLEAN_FALSE:
            value = False
        elif ftype == CT_BYTE:
            value = r.read_byte()
        elif ftype == CT_I16:
            value = r.read_varint()
        elif ftype == CT_I32:
            value = r.read_varint()
        elif ftype == CT_I64:
            value = r.read_varint()
        elif ftype == CT_DOUBLE:
            value = struct.unpack("<d", r.read_bytes(8))[0]
        elif ftype == CT_BINARY:
            value = _read_binary(r)
        elif ftype == CT_STRUCT:
            value = decode_struct(r)
        elif ftype in (CT_LIST, CT_SET):
            value = decode_list(r)
        elif ftype == CT_MAP:
            value = decode_map(r)
        else:
            raise ValueError(f"Unsupported thrift compact field type: {ftype}")

        result[field_id] = value
    return result


def decode_list(r: Reader) -> list:
    header = r.read_byte()
    size = (header >> 4) & 0x0F
    elem_type = header & 0x0F
    if size == 15:
        size = r.read_uvarint()

    items = []
    for _ in range(size):
        items.append(_decode_value(r, elem_type))
    return items


def decode_map(r: Reader) -> dict:
    size = r.read_uvarint()
    if size == 0:
        return {}
    types_byte = r.read_byte()
    key_type = (types_byte >> 4) & 0x0F
    val_type = types_byte & 0x0F
    result = {}
    for _ in range(size):
        k = _decode_value(r, key_type)
        v = _decode_value(r, val_type)
        result[k] = v
    return result


def _decode_value(r: Reader, ctype: int):
    if ctype == CT_BOOLEAN_TRUE or ctype == CT_BOOLEAN_FALSE:
        # In list/set context, booleans are stored as a single byte
        # (0x01/0x02), not folded into the element header.
        b = r.read_byte()
        return b == CT_BOOLEAN_TRUE
    elif ctype == CT_BYTE:
        return r.read_byte()
    elif ctype in (CT_I16, CT_I32, CT_I64):
        return r.read_varint()
    elif ctype == CT_DOUBLE:
        return struct.unpack("<d", r.read_bytes(8))[0]
    elif ctype == CT_BINARY:
        return _read_binary(r)
    elif ctype == CT_STRUCT:
        return decode_struct(r)
    elif ctype in (CT_LIST, CT_SET):
        return decode_list(r)
    elif ctype == CT_MAP:
        return decode_map(r)
    else:
        raise ValueError(f"Unsupported thrift compact element type: {ctype}")


# ---------------------------------------------------------------------------
# Parquet-specific field ID maps (from the public parquet.thrift schema)
# ---------------------------------------------------------------------------

# FileMetaData
FMD_VERSION = 1
FMD_SCHEMA = 2
FMD_NUM_ROWS = 3
FMD_ROW_GROUPS = 4
FMD_KEY_VALUE_METADATA = 5
FMD_CREATED_BY = 6

# SchemaElement
SE_TYPE = 1
SE_TYPE_LENGTH = 2
SE_REPETITION_TYPE = 3
SE_NAME = 4
SE_NUM_CHILDREN = 5
SE_CONVERTED_TYPE = 6

# RowGroup
RG_COLUMNS = 1
RG_TOTAL_BYTE_SIZE = 2
RG_NUM_ROWS = 3

# ColumnChunk
CC_FILE_OFFSET = 2
CC_META_DATA = 3

# ColumnMetaData
CMD_TYPE = 1
CMD_ENCODINGS = 2
CMD_PATH_IN_SCHEMA = 3
CMD_CODEC = 4
CMD_NUM_VALUES = 5
CMD_TOTAL_UNCOMPRESSED_SIZE = 6
CMD_TOTAL_COMPRESSED_SIZE = 7
CMD_DATA_PAGE_OFFSET = 9
CMD_DICTIONARY_PAGE_OFFSET = 11

# PageHeader
PH_TYPE = 1
PH_UNCOMPRESSED_PAGE_SIZE = 2
PH_COMPRESSED_PAGE_SIZE = 3
PH_DATA_PAGE_HEADER = 5
PH_DICTIONARY_PAGE_HEADER = 7
PH_DATA_PAGE_HEADER_V2 = 8

# DataPageHeader
DPH_NUM_VALUES = 1
DPH_ENCODING = 2
DPH_DEFINITION_LEVEL_ENCODING = 3
DPH_REPETITION_LEVEL_ENCODING = 4

# DataPageHeaderV2
DPH2_NUM_VALUES = 1
DPH2_NUM_NULLS = 2
DPH2_NUM_ROWS = 3
DPH2_ENCODING = 4
DPH2_DEF_LEVELS_BYTE_LENGTH = 5
DPH2_REP_LEVELS_BYTE_LENGTH = 6
DPH2_IS_COMPRESSED = 7

# DictionaryPageHeader
DICT_NUM_VALUES = 1
DICT_ENCODING = 2

PARQUET_TYPE_NAMES = {
    0: "BOOLEAN", 1: "INT32", 2: "INT64", 3: "INT96",
    4: "FLOAT", 5: "DOUBLE", 6: "BYTE_ARRAY", 7: "FIXED_LEN_BYTE_ARRAY",
}
CODEC_NAMES = {
    0: "UNCOMPRESSED", 1: "SNAPPY", 2: "GZIP", 3: "LZO",
    4: "BROTLI", 5: "LZ4", 6: "ZSTD", 7: "LZ4_RAW",
}
ENCODING_NAMES = {
    0: "PLAIN", 2: "PLAIN_DICTIONARY", 3: "RLE", 4: "BIT_PACKED",
    5: "DELTA_BINARY_PACKED", 6: "DELTA_LENGTH_BYTE_ARRAY",
    7: "DELTA_BYTE_ARRAY", 8: "RLE_DICTIONARY", 9: "BYTE_STREAM_SPLIT",
}
PAGE_TYPE_NAMES = {0: "DATA_PAGE", 1: "INDEX_PAGE", 2: "DICTIONARY_PAGE", 3: "DATA_PAGE_V2"}


@dataclass
class ColumnInfo:
    path: str
    ptype: str
    codec: str
    num_values: int
    total_uncompressed_size: int
    total_compressed_size: int
    data_page_offset: int
    dictionary_page_offset: int | None


@dataclass
class RowGroupInfo:
    num_rows: int
    total_byte_size: int
    columns: dict = dc_field(default_factory=dict)  # path -> ColumnInfo


@dataclass
class ParquetFile:
    path: str
    num_rows: int
    schema: list  # list of (name, type_name, repetition)
    row_groups: list  # list of RowGroupInfo
    raw_metadata: dict
    max_def_levels: dict = dc_field(default_factory=dict)  # path -> max definition level


def _build_max_def_levels(schema_raw: list) -> dict:
    """Parquet's flat schema list is a pre-order traversal of the schema
    tree: the root message (repetition REQUIRED, num_children=N) followed
    by its N children in order, each of which may itself have num_children
    and be followed by its own children, etc.

    Max definition level for a leaf = count of OPTIONAL (or REPEATED)
    ancestors on the path from root to that leaf, INCLUDING the leaf
    itself if it is OPTIONAL/REPEATED. This directly determines the
    bit-width used to encode definition levels for that column, which is
    essential to decode nested-optional columns like `audio.bytes`
    correctly (audio is itself OPTIONAL, and bytes/path within it are also
    OPTIONAL, so audio.bytes has max definition level 2, not 1).
    """
    # schema_raw[0] is the root "message" node (num_children = number of
    # top-level fields). We do a manual pre-order walk using num_children
    # to know how many following entries belong to each node's subtree.
    idx = 0

    def walk(depth_optional: int, path_prefix: list) -> int:
        nonlocal idx
        node = schema_raw[idx]
        idx += 1
        name = node["name"]
        rep = node["repetition"]
        is_optional = rep in ("OPTIONAL", "REPEATED")
        this_level = depth_optional + (1 if is_optional else 0)
        path = path_prefix + [name]

        num_children = node.get("num_children")
        if num_children:
            for _ in range(num_children):
                walk(this_level, path)
        else:
            # Leaf node — record its max definition level, keyed by the
            # dotted path used elsewhere in this module (e.g. "audio.bytes").
            max_def_levels[".".join(path)] = this_level
        return this_level

    max_def_levels: dict = {}
    # schema_raw[0] is the root; skip it as a "path prefix" element (its
    # name is typically "schema" / "dataset" and isn't part of column paths).
    idx = 1
    root = schema_raw[0]
    for _ in range(root.get("num_children", 0)):
        walk(0, [])
    return max_def_levels


def read_footer_metadata(path: str) -> ParquetFile:
    with open(path, "rb") as f:
        f.seek(0, 2)
        file_size = f.tell()
        f.seek(file_size - 8)
        tail = f.read(8)
        if tail[4:] != b"PAR1":
            raise ValueError(f"Not a valid Parquet file (bad trailing magic): {path}")
        footer_len = struct.unpack("<I", tail[:4])[0]
        f.seek(file_size - 8 - footer_len)
        footer_bytes = f.read(footer_len)

    r = Reader(footer_bytes)
    meta = decode_struct(r)

    num_rows = meta[FMD_NUM_ROWS]

    schema_raw = meta[FMD_SCHEMA]
    schema = []
    for se in schema_raw:
        name = se.get(SE_NAME, b"").decode("utf-8", errors="replace")
        ptype = PARQUET_TYPE_NAMES.get(se.get(SE_TYPE), None) if SE_TYPE in se else None
        rep = {0: "REQUIRED", 1: "OPTIONAL", 2: "REPEATED"}.get(se.get(SE_REPETITION_TYPE), None)
        num_children = se.get(SE_NUM_CHILDREN, None)
        schema.append({"name": name, "type": ptype, "repetition": rep, "num_children": num_children})

    row_groups = []
    for rg in meta[FMD_ROW_GROUPS]:
        rg_num_rows = rg[RG_NUM_ROWS]
        rg_total_byte_size = rg.get(RG_TOTAL_BYTE_SIZE, 0)
        columns = {}
        for col in rg[RG_COLUMNS]:
            cmd = col[CC_META_DATA]
            path_parts = [p.decode("utf-8", errors="replace") for p in cmd[CMD_PATH_IN_SCHEMA]]
            path = ".".join(path_parts)
            ci = ColumnInfo(
                path=path,
                ptype=PARQUET_TYPE_NAMES.get(cmd[CMD_TYPE], str(cmd[CMD_TYPE])),
                codec=CODEC_NAMES.get(cmd[CMD_CODEC], str(cmd[CMD_CODEC])),
                num_values=cmd[CMD_NUM_VALUES],
                total_uncompressed_size=cmd[CMD_TOTAL_UNCOMPRESSED_SIZE],
                total_compressed_size=cmd[CMD_TOTAL_COMPRESSED_SIZE],
                data_page_offset=cmd[CMD_DATA_PAGE_OFFSET],
                dictionary_page_offset=cmd.get(CMD_DICTIONARY_PAGE_OFFSET),
            )
            columns[path] = ci
        row_groups.append(RowGroupInfo(num_rows=rg_num_rows, total_byte_size=rg_total_byte_size, columns=columns))

    return ParquetFile(
        path=path, num_rows=num_rows, schema=schema, row_groups=row_groups,
        raw_metadata=meta, max_def_levels=_build_max_def_levels(schema),
    )


# ---------------------------------------------------------------------------
# Pure-Python Snappy (raw block format) decompressor
# ---------------------------------------------------------------------------
# Implemented by hand — no `python-snappy`/`cramjam` available and no
# network to install them. Snappy's raw block format is publicly documented
# (https://github.com/google/snappy/blob/main/format_description.txt) and
# is simple enough to implement correctly in pure Python: a varint-encoded
# uncompressed length, followed by a sequence of literal and copy elements.

def snappy_decompress(data: bytes) -> bytes:
    pos = 0

    def read_uvarint() -> int:
        nonlocal pos
        result = 0
        shift = 0
        while True:
            b = data[pos]
            pos += 1
            result |= (b & 0x7F) << shift
            if (b & 0x80) == 0:
                break
            shift += 7
        return result

    uncompressed_len = read_uvarint()
    out = bytearray()

    while pos < len(data):
        tag = data[pos]
        pos += 1
        elem_type = tag & 0x03

        if elem_type == 0:  # literal
            length_code = tag >> 2
            if length_code < 60:
                length = length_code + 1
            else:
                n_extra = length_code - 59
                length_bytes = data[pos:pos + n_extra]
                pos += n_extra
                length = int.from_bytes(length_bytes, "little") + 1
            out.extend(data[pos:pos + length])
            pos += length

        elif elem_type == 1:  # copy, 1-byte offset
            length = ((tag >> 2) & 0x07) + 4
            offset = ((tag >> 5) << 8) | data[pos]
            pos += 1
            _copy_from_history(out, offset, length)

        elif elem_type == 2:  # copy, 2-byte offset
            length = (tag >> 2) + 1
            offset = int.from_bytes(data[pos:pos + 2], "little")
            pos += 2
            _copy_from_history(out, offset, length)

        else:  # elem_type == 3, copy, 4-byte offset
            length = (tag >> 2) + 1
            offset = int.from_bytes(data[pos:pos + 4], "little")
            pos += 4
            _copy_from_history(out, offset, length)

    if len(out) != uncompressed_len:
        raise ValueError(
            f"Snappy decompress size mismatch: expected {uncompressed_len}, got {len(out)}"
        )
    return bytes(out)


def _copy_from_history(out: bytearray, offset: int, length: int) -> None:
    """Copy `length` bytes from `offset` bytes back in `out`, appending to
    `out`. Must support overlapping copy (offset < length), which is the
    standard LZ77 "run" pattern (e.g. repeated bytes) — done byte-by-byte
    for correctness.
    """
    start = len(out) - offset
    if start < 0:
        raise ValueError("Invalid snappy copy offset (points before start of buffer)")
    for i in range(length):
        out.append(out[start + i])


# ---------------------------------------------------------------------------
# RLE / bit-packed hybrid decoder (used for definition levels, and for
# dictionary-encoded data pages' indices)
# ---------------------------------------------------------------------------

def decode_rle_bitpacked_hybrid(data: bytes, bit_width: int, num_values: int) -> list[int]:
    """Decode the RLE/bit-packed hybrid format used for definition levels
    and RLE_DICTIONARY-encoded values. `data` should NOT include the
    4-byte length prefix that precedes definition-level data in a v1 DATA
    PAGE (caller strips that separately, since dictionary-index streams in
    a page don't have that prefix).
    """
    values: list[int] = []
    pos = 0
    byte_width = (bit_width + 7) // 8

    while len(values) < num_values and pos < len(data):
        header, header_len = _read_uvarint_from_bytes(data, pos)
        pos += header_len
        if header & 1:  # bit-packed run
            num_groups = header >> 1
            num_bitpacked_values = num_groups * 8
            values.extend(_unpack_bitpacked(data, pos, bit_width, num_bitpacked_values))
            pos += num_groups * bit_width
        else:  # RLE run
            run_length = header >> 1
            value = int.from_bytes(data[pos:pos + byte_width], "little") if byte_width > 0 else 0
            pos += byte_width
            values.extend([value] * run_length)

    return values[:num_values]


def _read_uvarint_from_bytes(data: bytes, pos: int) -> tuple[int, int]:
    result = 0
    shift = 0
    start = pos
    while True:
        b = data[pos]
        pos += 1
        result |= (b & 0x7F) << shift
        if (b & 0x80) == 0:
            break
        shift += 7
    return result, pos - start


def _unpack_bitpacked(data: bytes, pos: int, bit_width: int, num_values: int) -> list[int]:
    """Unpack `num_values` little-endian bit-packed integers of `bit_width`
    bits each, starting at byte offset `pos` in `data`.
    """
    if bit_width == 0:
        return [0] * num_values
    out = []
    bit_pos = 0
    total_bits_needed = num_values * bit_width
    # Read enough bytes to cover total_bits_needed
    n_bytes = (total_bits_needed + 7) // 8
    chunk = data[pos:pos + n_bytes]
    bits = int.from_bytes(chunk, "little")
    mask = (1 << bit_width) - 1
    for i in range(num_values):
        out.append((bits >> (i * bit_width)) & mask)
    return out


# ---------------------------------------------------------------------------
# Page header parsing + page decoding
# ---------------------------------------------------------------------------

def read_page_header(buf: bytes, pos: int) -> tuple[dict, int]:
    """Returns (header_dict, new_pos) — new_pos is right after the header,
    i.e. where page data bytes begin.
    """
    r = Reader(buf, pos)
    header = decode_struct(r)
    return header, r.pos


def decode_plain_byte_array(data: bytes, num_values: int) -> list[bytes]:
    """PLAIN encoding for BYTE_ARRAY: each value is [4-byte LE length][bytes]."""
    out = []
    pos = 0
    for _ in range(num_values):
        length = struct.unpack_from("<I", data, pos)[0]
        pos += 4
        out.append(data[pos:pos + length])
        pos += length
    return out


def decode_plain_boolean(data: bytes, num_values: int) -> list[bool]:
    """PLAIN encoding for BOOLEAN: bit-packed, LSB first, 1 bit per value."""
    out = []
    for i in range(num_values):
        byte_idx = i // 8
        bit_idx = i % 8
        byte = data[byte_idx] if byte_idx < len(data) else 0
        out.append(bool((byte >> bit_idx) & 1))
    return out


def decode_plain_int32(data: bytes, num_values: int) -> list[int]:
    return list(struct.unpack_from(f"<{num_values}i", data, 0))


def read_column_values(
    file_path: str,
    col: "ColumnInfo",
    max_rows: int | None = None,
    max_def_level: int = 1,
) -> list:
    """Read and decode all (or up to max_rows) values of a single column
    chunk, handling: SNAPPY/UNCOMPRESSED page decompression, PLAIN
    encoding, and definition-level-based nullability (OPTIONAL fields).
    Returns a list of length == number of rows in this row group's column
    chunk (None for nulls), in row order.

    `max_def_level` MUST match the column's actual nesting depth (see
    `_build_max_def_levels`) — e.g. 1 for a top-level OPTIONAL scalar like
    `language`, 2 for `audio.bytes` (nested inside the OPTIONAL `audio`
    struct). Getting this wrong silently misdecodes the definition-level
    bit-packing and produces wrong values — this bit a real, caught bug
    during development (see docs/PHASE3_REAL_AUDIO_VALIDATION.md).

    Only PLAIN encoding and V1 DATA_PAGE / DICTIONARY_PAGE are supported,
    which is sufficient for this specific uploaded file (verified against
    its actual page headers below) — this is not a general Parquet reader.
    """
    import math
    def_bit_width = max(1, math.ceil(math.log2(max_def_level + 1))) if max_def_level > 0 else 0

    with open(file_path, "rb") as f:
        f.seek(col.data_page_offset)
        # Read a generous chunk covering (compressed_size); pages are
        # contiguous starting at data_page_offset (dictionary page, if any,
        # comes first at dictionary_page_offset < data_page_offset).
        start = col.dictionary_page_offset if col.dictionary_page_offset is not None else col.data_page_offset
        f.seek(start)
        chunk = f.read(col.total_compressed_size + (col.data_page_offset - start if col.dictionary_page_offset else 0) + 4096)

    pos = 0
    dictionary: list | None = None
    values: list = []
    nulls_total = 0

    while len(values) < col.num_values and pos < len(chunk):
        header, data_start = read_page_header(chunk, pos)
        page_type = PAGE_TYPE_NAMES.get(header.get(PH_TYPE), header.get(PH_TYPE))
        compressed_size = header[PH_COMPRESSED_PAGE_SIZE]
        uncompressed_size = header[PH_UNCOMPRESSED_PAGE_SIZE]
        page_data_compressed = chunk[data_start:data_start + compressed_size]

        if col.codec == "SNAPPY":
            page_data = snappy_decompress(page_data_compressed)
        elif col.codec == "UNCOMPRESSED":
            page_data = page_data_compressed
        else:
            raise ValueError(f"Unsupported codec for pure-Python reader: {col.codec}")

        if len(page_data) != uncompressed_size:
            raise ValueError(
                f"Decompressed page size mismatch for {col.path}: "
                f"expected {uncompressed_size}, got {len(page_data)}"
            )

        if page_type == "DICTIONARY_PAGE":
            dph = header[PH_DICTIONARY_PAGE_HEADER]
            n_dict = dph[DICT_NUM_VALUES]
            if col.ptype == "BYTE_ARRAY":
                dictionary = decode_plain_byte_array(page_data, n_dict)
            elif col.ptype == "INT32":
                dictionary = decode_plain_int32(page_data, n_dict)
            else:
                raise ValueError(f"Unsupported dictionary value type: {col.ptype}")

        elif page_type == "DATA_PAGE":
            dph = header[PH_DATA_PAGE_HEADER]
            n_values = dph[DPH_NUM_VALUES]
            encoding = ENCODING_NAMES.get(dph[DPH_ENCODING], dph[DPH_ENCODING])

            page_pos = 0
            # Definition levels, only present if the field is OPTIONAL
            # (max definition level 1) — encoded as: 4-byte LE length prefix,
            # then RLE/bit-packed hybrid data, bit_width = 1 (since max def
            # level is 1 for a non-nested-optional column).
            def_lengths_prefix = struct.unpack_from("<I", page_data, page_pos)[0]
            page_pos += 4
            def_level_bytes = page_data[page_pos:page_pos + def_lengths_prefix]
            page_pos += def_lengths_prefix
            def_levels = decode_rle_bitpacked_hybrid(def_level_bytes, bit_width=def_bit_width, num_values=n_values)

            value_bytes = page_data[page_pos:]
            n_non_null = sum(1 for d in def_levels if d == max_def_level)

            if encoding == "PLAIN":
                if col.ptype == "BOOLEAN":
                    raw_values = decode_plain_boolean(value_bytes, n_non_null)
                elif col.ptype == "BYTE_ARRAY":
                    raw_values = decode_plain_byte_array(value_bytes, n_non_null)
                elif col.ptype == "INT32":
                    raw_values = decode_plain_int32(value_bytes, n_non_null)
                else:
                    raise ValueError(f"Unsupported PLAIN value type: {col.ptype}")
            elif encoding in ("PLAIN_DICTIONARY", "RLE_DICTIONARY"):
                if dictionary is None:
                    raise ValueError(f"Dictionary-encoded page for {col.path} but no dictionary page seen")
                # First byte = bit width for the RLE/bit-packed index stream
                idx_bit_width = value_bytes[0]
                idx_data = value_bytes[1:]
                indices = decode_rle_bitpacked_hybrid(idx_data, bit_width=idx_bit_width, num_values=n_non_null)
                raw_values = [dictionary[i] for i in indices]
            else:
                raise ValueError(f"Unsupported page encoding for pure-Python reader: {encoding}")

            it = iter(raw_values)
            for d in def_levels:
                values.append(next(it) if d == max_def_level else None)
            nulls_total += n_values - n_non_null

        else:
            raise ValueError(f"Unsupported page type for pure-Python reader: {page_type}")

        pos = data_start + compressed_size
        if max_rows is not None and len(values) >= max_rows:
            break

    return values[:max_rows] if max_rows is not None else values


def parse_flac_streaminfo(flac_bytes: bytes) -> dict:
    """Parse just the STREAMINFO metadata block of a FLAC file to get
    sample_rate/channels/total_samples/duration WITHOUT decoding any audio
    — much cheaper than a full FLAC decode when we only need duration
    statistics across many files. Standard FLAC layout: 4-byte 'fLaC'
    magic, then metadata blocks (1-byte header: bit7=last-block flag, bits
    6-0=block type; 3-byte big-endian length), STREAMINFO is block type 0
    and is always 34 bytes.
    """
    if flac_bytes[:4] != b"fLaC":
        raise ValueError("Not a FLAC file (missing fLaC magic)")
    pos = 4
    block_header = flac_bytes[pos]
    block_type = block_header & 0x7F
    length = int.from_bytes(flac_bytes[pos + 1:pos + 4], "big")
    pos += 4
    if block_type != 0:
        raise ValueError(f"Expected STREAMINFO (type 0) as first block, got type {block_type}")
    block = flac_bytes[pos:pos + length]

    bits = int.from_bytes(block, "big")
    total_bits = len(block) * 8
    def take(nbits, bit_offset_from_start):
        shift = total_bits - bit_offset_from_start - nbits
        mask = (1 << nbits) - 1
        return (bits >> shift) & mask

    off = 0
    min_blocksize = take(16, off); off += 16
    max_blocksize = take(16, off); off += 16
    min_framesize = take(24, off); off += 24
    max_framesize = take(24, off); off += 24
    sample_rate = take(20, off); off += 20
    channels = take(3, off) + 1; off += 3
    bits_per_sample = take(5, off) + 1; off += 5
    total_samples = take(36, off); off += 36

    duration_sec = total_samples / sample_rate if sample_rate else None
    return {
        "sample_rate": sample_rate,
        "channels": channels,
        "bits_per_sample": bits_per_sample,
        "total_samples": total_samples,
        "duration_sec": duration_sec,
    }


if __name__ == "__main__":
    import sys
    pf = read_footer_metadata(sys.argv[1] if len(sys.argv) > 1 else "/mnt/user-data/uploads/train-00000-of-00010.parquet")
    print(f"File: {pf.path}")
    print(f"Total rows: {pf.num_rows}")
    print(f"Row groups: {len(pf.row_groups)}")
    print("\nSchema (top-level fields):")
    for s in pf.schema:
        print(f"  {s}")
    print(f"\nFirst row group: {pf.row_groups[0].num_rows} rows, {pf.row_groups[0].total_byte_size} bytes")
    print("Columns in first row group:")
    for path, ci in pf.row_groups[0].columns.items():
        print(f"  {path}: type={ci.ptype} codec={ci.codec} num_values={ci.num_values} "
              f"compressed={ci.total_compressed_size} uncompressed={ci.total_uncompressed_size}")