""" 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(" 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(" 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(" 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(" 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("= 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}")