hyperparse.py

#!/usr/bin/env python3
"""
hyperparse.py - a reverse-engineered reader for Tableau .hyper files.

Reads .hyper files without Tableau's Hyper API: the two root records, the
storage directory, every object it addresses, the system catalog, and the
actual row values inside the Data Blocks.

  python3 hyperparse.py FILE...                  superblocks + directory + catalog
  python3 hyperparse.py --blocks FILE            per-column block geometry + SMA
  python3 hyperparse.py --rows 10 FILE           reconstruct actual row values
  python3 hyperparse.py --json FILE              machine-readable output
  python3 hyperparse.py --verify FILE            cross-check vs the real API
  python3 hyperparse.py --dump 0x2000 FILE       hexdump at an offset

No third-party dependencies: the LZ4 block decoder and CRC-32C are inline.
If `tableauhyperapi` happens to be installed, --verify uses it as an oracle.

CONFIDENCE
  The container is confirmed against hyper::RootRecord, hyper::FileStorageResource
  and hyper::ObjectStore in the macOS arm64 `hyperd` (which, unlike the Linux
  build, ships full C++ symbols), and CRC-verified byte-for-byte on every sample.
  The Data Block header and region map come from the engine's own assertion
  strings in getPackInfoAndValidate. The encodings were measured against the
  Hyper API as an oracle: 276 columns / 36 relations, value multisets and whole
  row tuples. The one thing still guessed is the DataBlockCompression enum,
  whose behaviour is tabulated (SCHEMES) but whose definition is not decoded -
  an unknown code makes the decoder refuse rather than invent values.

Structures are named after the engine's own classes so that the code and a
disassembler view line up. See README.md for the field tables and the evidence.
"""

from __future__ import annotations

import argparse
import bisect
import json
import math
import struct
import sys
import uuid
from dataclasses import dataclass, field, asdict
from typing import Any, NamedTuple


class IntervalValue(NamedTuple):
    """Hyper's 16-byte interval_t: {u8 microseconds, i4 days, u4 months}.

    A NamedTuple so it stays hashable - decoded values end up in Counters
    during validation, and a dict would not.
    """
    months: int
    days: int
    microseconds: int

    def __str__(self) -> str:
        return f"{self.months}mo {self.days}d {self.microseconds}us"

PAGE_SIZE = 0x1000
EXTENT_SIZE = 0x10000            # files are always a whole number of these
MAX_U64 = (1 << 64) - 1
SB_MAGIC = b"Hyper\x08"          # RootRecord::initHeader writes exactly these 6 bytes
PKG_MAGIC = b"HyperDB\x00"
CATALOG_NEEDLE = b'{"compressionMethod'
SB_CHECKSUM_OFF = 0x0FFC
DIR_MAGIC = 0xDA1ADA1A           # u64 sentinel closing the storage directory
ENTRY_SIZE = 0x30                # sizeof(ObjectStoreIdMapEntry)


# --------------------------------------------------------------------------
# CRC-32C as Hyper computes it
#
# hyper::crc32cIntrinsic(ptr, len, crc) is a *raw* accumulator over the ARMv8
# crc32cx/crc32cb instructions: no pre-inversion of the seed, no post-
# inversion of the result. Every off-the-shelf CRC-32C ("Castagnoli") helper
# uses init=0xFFFFFFFF/xorout=0xFFFFFFFF, which is why this looked unbreakable
# until the routine itself was read.
# --------------------------------------------------------------------------

_CRC32C_TABLE = []
for _i in range(256):
    _c = _i
    for _ in range(8):
        _c = (_c >> 1) ^ (0x82F63B78 if _c & 1 else 0)
    _CRC32C_TABLE.append(_c)


def crc32c(data: bytes, seed: int = 0) -> int:
    """Reflected CRC-32C, init=seed, no final XOR."""
    crc = seed
    for b in data:
        crc = _CRC32C_TABLE[(crc ^ b) & 0xFF] ^ (crc >> 8)
    return crc & 0xFFFFFFFF


def crc_seed_for_version(format_version: int) -> int:
    """hyper::RootRecord::getCRCSeedForVersion.

    Old databases seed object/directory CRCs with 0; from the format version
    that gated the `nonZeroChecksumSeed` capability onwards the seed is 0x1234.
    The threshold is not read out of the binary here, so both are tried.
    """
    return 0x1234 if format_version >= 1 else 0


# hyper::ObjectStoreId::Category
CATEGORY = {
    1: "Database_Header",
    2: "Schema",
    3: "Relation_Header",
    4: "Relation_DataBlock",
    5: "Relation_Metadata",
    6: "Relation_Sample",
    7: "Index",
    8: "Database_EncryptionKey",
}

# hyper::DatabaseInfo::Compression, as stored in ObjectStoreIdMapEntry+0x29
COMPRESSION = {0: "none", 1: "lz4", 2: "?2"}


# --------------------------------------------------------------------------
# LZ4 block format (raw, no frame header - Hyper does not use frames)
# --------------------------------------------------------------------------

def lz4_block_decode(src: bytes, start: int = 0, limit: int = 1 << 24) -> tuple[bytes, int]:
    """Decode as far as the stream stays self-consistent; return (out, end).

    Deliberately tolerant: a normal decoder needs the output size up front and
    raises on the first inconsistency. For exploration we want to decode until
    the stream stops making sense and learn where that happened.
    """
    out = bytearray()
    i, n = start, len(src)
    while i < n and len(out) < limit:
        token = src[i]; i += 1
        lit = token >> 4
        if lit == 15:
            while i < n:
                b = src[i]; i += 1
                lit += b
                if b != 255:
                    break
            else:
                break
        if i + lit > n:
            break
        out += src[i:i + lit]
        i += lit
        if i + 2 > n:
            break                                  # legal: block ends on literals
        offset = src[i] | (src[i + 1] << 8)
        i += 2
        if offset == 0 or offset > len(out):
            i -= 2
            break                                  # bad back-reference: stop clean
        match = token & 0x0F
        if match == 15:
            while i < n:
                b = src[i]; i += 1
                match += b
                if b != 255:
                    break
            else:
                break
        match += 4
        p = len(out) - offset
        for _ in range(match):                     # byte-wise: overlap is legal
            out.append(out[p]); p += 1
    return bytes(out), i


# --------------------------------------------------------------------------
# Structures
# --------------------------------------------------------------------------

@dataclass
class Superblock:
    """hyper::RootRecord - one 4 KiB page, two of them, written alternately."""
    offset: int
    struct_version: int
    format_version: int
    encrypted: int                # +0x0C, a bool in StorageMetadata's ctor
    creator_version: str          # +0x18, {u4 major, u4 minor, u4 build}
    min_version: str              # +0x24, same shape
    creator_build: int            # the build component alone, for convenience
    txn_id: int
    file_size: int
    dir_offset: int               # +0x40 \ hyper::Position{offset, length} of the
    dir_length: int               # +0x48 / storage directory; relocates on commit
    dir_capacity: int             # +0x50, allocated bytes for the directory
    commit_id: int
    checksum: int
    checksum_ok: bool
    unknown_tail: str             # hex of any non-zero bytes we do not model

    @property
    def page_index(self) -> int:
        return self.offset // PAGE_SIZE


@dataclass
class ObjectEntry:
    """hyper::ObjectStoreIdMapEntry - 48 bytes, memcpy'd straight out of the file."""
    slot: int
    category: int                 # ObjectStoreId+0x06
    index: int                    # ObjectStoreId+0x00, 24 bits: column, or 0 for
                                  #   the per-block partition header
    relation: int                 # ObjectStoreId+0x0C: index into the catalog's
                                  #   relations[] array. Verified on 68/68 files,
                                  #   including ones whose relation oids are
                                  #   non-contiguous, so it is an array position
                                  #   and not an oid.
    block: int                    # ObjectStoreId+0x08: which block of the column
    field_b: int                  # ObjectStoreId+0x03, 24 bits. Role unknown. It
                                  #   equals `relation` in Tableau-written files
                                  #   and is 0 in hyperapi-written ones, so it
                                  #   enumerates the relations in only 64/68 -
                                  #   do not key on it.
    size: int                     # payload bytes, excluding the trailing CRC
    offset: int                   # Position.offset
    alloc: int                    # Position.length, >= size + 4
    compression: int
    encryption: int
    crc_ok: bool | None = None

    @property
    def category_name(self) -> str:
        return CATEGORY.get(self.category, f"Unknown({self.category})")

    def __str__(self) -> str:
        """Field order as hyper::ObjectStoreId::to_string prints it, which is
        not the order the fields are laid out in."""
        return (f"{self.category_name}.{self.field_b}.{self.index}"
                f".{self.relation}.{self.block}")


@dataclass
class Directory:
    offset: int
    length: int
    log2_capacity: int
    entries: list[ObjectEntry]
    free_list: list[tuple[int, int]]
    magic_ok: bool
    crc_ok: bool
    crc_seed: int


@dataclass
class PackageHeader:
    offset: int
    kind: int
    struct_version: int
    format_version: int
    database_uuid: str
    checksum: int


@dataclass
class Catalog:
    offset: int
    length: int
    digest: int
    framing: str                  # "packaged" | "bare"
    package: PackageHeader | None
    data: dict


@dataclass
class ColumnBlock:
    """One Relation_DataBlock object, resolved against the catalog.

    index 0 of a relation is the partition header (just a tuple count); index N
    is column N-1. Nothing here is found by scanning - the directory says
    exactly where each block is and the catalog says what type it holds.
    """
    relation: int
    index: int
    block: int                    # ObjectStoreId+0x08: which block of the column
    column: str | None            # None for the partition header
    sql_type: str | None
    offset: int
    stored_len: int
    decoded_len: int
    compression: str
    tuple_count: int
    hdr: "BlockIndex | None" = None      # the decoded per-column header
    sma_min: Any = None
    sma_max: Any = None
    note: str = ""

    @property
    def code_bits(self) -> float:
        """Bits per row in the data region. Dictionary codes are bit-packed at
        1/2/4 bits and byte-packed at 8/16/32, so this lands near a power of two.
        Derived, not read from the block."""
        if not self.hdr or not self.tuple_count:
            return 0.0
        return self.hdr.data_size * 8 / self.tuple_count


@dataclass
class HyperFile:
    path: str
    size: int
    superblocks: list[Superblock] = field(default_factory=list)
    directory: Directory | None = None
    catalogs: list[Catalog] = field(default_factory=list)
    blocks: list[ColumnBlock] = field(default_factory=list)
    _live_catalog: Catalog | None = None      # from the directory, not the scan
    warnings: list[str] = field(default_factory=list)

    @property
    def live_catalog(self) -> Catalog | None:
        """The catalog the live directory points at, read from its Schema object.

        Any other catalog in the file is an earlier transaction's state. Falls
        back to whatever the forensic scan found only when there is no usable
        directory - e.g. a truncated or corrupt file.
        """
        if self._live_catalog is not None:
            return self._live_catalog
        return self.catalogs[0] if self.catalogs else None

    @property
    def live(self) -> Superblock | None:
        """hyper::RootRecord::selectValidRootRecord: highest txn_id among the
        superblocks that pass magic, struct version and CRC."""
        good = [s for s in self.superblocks if s.checksum_ok]
        return max(good or self.superblocks, key=lambda s: s.txn_id) if self.superblocks else None


# --------------------------------------------------------------------------
# Parsing
# --------------------------------------------------------------------------

def _u(fmt: str, d: bytes, off: int) -> int:
    return struct.unpack_from(fmt, d, off)[0]


def parse_superblock(d: bytes, base: int, warn) -> Superblock | None:
    """Mirrors hyper::RootRecord::isValid + the StorageMetadata(RootRecord&) ctor."""
    if len(d) < base + PAGE_SIZE:
        warn(f"file too short for superblock at {base:#x}")
        return None
    if d[base:base + 6] != SB_MAGIC:
        warn(f"bad superblock magic at {base:#x}: {d[base:base+8]!r}")
        return None

    struct_version = _u("<H", d, base + 0x08)
    format_version = _u("<H", d, base + 0x0A)
    if struct_version != 1:
        warn(f"superblock {base:#x}: struct_version {struct_version} != 1; "
             f"the engine rejects anything else")
    # isValid() accepts (format_version + 1) & 0xFFFF <= 5, i.e. 0..4 or 0xFFFF.
    if (format_version + 1) & 0xFFFF > 5:
        warn(f"superblock {base:#x}: format version {format_version} is outside "
             f"the range the engine accepts (0-4, or 0xFFFF for unset)")

    # Flag any non-zero byte in regions we believe are reserved, so that a file
    # from a newer build tells us it has fields we do not know about. This
    # tripwire is what exposed the version triples at 0x18/0x24 - an earlier
    # revision read 0x20 as one u8 and so straddled `build` and the next
    # `major`, which is why Tableau-written files reported builds in the
    # billions.
    unknown = bytearray()
    for lo, hi in ((0x0D, 0x18), (0x58, 0x60), (0x68, 0x80)):
        chunk = d[base + lo: base + hi]
        if any(chunk):
            unknown += chunk
    if any(d[base + 0x80: base + SB_CHECKSUM_OFF]):
        warn(f"superblock {base:#x}: unexpected data in 0x80..0xFFC")

    cmaj, cmin, cbuild, mmaj, mmin, mbuild = struct.unpack_from("<6I", d, base + 0x18)
    stored = _u("<I", d, base + SB_CHECKSUM_OFF)
    return Superblock(
        offset=base,
        struct_version=struct_version,
        format_version=format_version,
        encrypted=d[base + 0x0C],
        creator_version=f"{cmaj}.{cmin}.{cbuild}",
        min_version=f"{mmaj}.{mmin}.{mbuild}",
        creator_build=cbuild,
        txn_id=_u("<Q", d, base + 0x30),
        file_size=_u("<Q", d, base + 0x38),
        dir_offset=_u("<Q", d, base + 0x40),
        dir_length=_u("<Q", d, base + 0x48),
        dir_capacity=_u("<Q", d, base + 0x50),
        commit_id=_u("<Q", d, base + 0x60),
        checksum=stored,
        checksum_ok=crc32c(d[base:base + SB_CHECKSUM_OFF]) == stored,
        unknown_tail=unknown.hex(),
    )


def parse_directory(d: bytes, sb: Superblock, warn) -> Directory | None:
    """hyper::FileStorageResource::readDirectory.

        u64                          log2 of the hash-table capacity
        ObjectStoreIdMapEntry[1<<n]  48 bytes each, raw
        u64                          free-list length
        Position[len]                {offset, length} of every free chunk
        u64                          0xDA1ADA1A
        (zero padding out to dir_length)
        u32                          CRC-32C of the preceding dir_length bytes

    The CRC sits *after* the payload, at dir_offset + dir_length, which is why
    it reads as zero if you look for it at the more natural length - 4.
    """
    off, ln = sb.dir_offset, sb.dir_length
    if not (0 < off < len(d)) or ln < 16 or off + ln + 4 > len(d):
        warn(f"directory Position ({off:#x}, {ln}) does not fit the file")
        return None
    blob = d[off:off + ln]

    stored = _u("<I", d, off + ln)
    seed = next((s for s in (crc_seed_for_version(sb.format_version), 0, 0x1234)
                 if crc32c(blob, s) == stored), None)
    if seed is None:
        warn(f"directory CRC mismatch: stored {stored:#010x}")

    log2cap = _u("<Q", blob, 0)
    if log2cap > 0x3A:
        warn(f"directory hash shift {log2cap} exceeds the engine's own limit")
        return None
    cap = 1 << log2cap
    end = 8 + cap * ENTRY_SIZE
    if end + 8 > ln:
        warn(f"directory too short for {cap} entries")
        return None

    entries = []
    for i in range(cap):
        e = blob[8 + ENTRY_SIZE * i:8 + ENTRY_SIZE * (i + 1)]
        if e[0x28] != 1:                           # 0 empty, 1 used, 2 tombstone
            continue
        size, pos_off, pos_len = struct.unpack_from("<QQQ", e, 0x10)
        entry = ObjectEntry(
            slot=i,
            index=int.from_bytes(e[0x00:0x03], "little"),
            field_b=int.from_bytes(e[0x03:0x06], "little"),
            category=_u("<H", e, 0x06),
            block=_u("<I", e, 0x08),
            relation=_u("<I", e, 0x0C),
            size=size, offset=pos_off, alloc=pos_len,
            compression=e[0x29], encryption=e[0x2A],
        )
        if pos_off + size + 4 <= len(d):
            entry.crc_ok = any(crc32c(d[pos_off:pos_off + size], s) ==
                               _u("<I", d, pos_off + size)
                               for s in ({seed} if seed is not None else {0, 0x1234}))
        entries.append(entry)

    n_free = _u("<Q", blob, end)
    free = [struct.unpack_from("<QQ", blob, end + 8 + 16 * k) for k in range(n_free)]
    magic_at = end + 8 + 16 * n_free
    magic_ok = magic_at + 8 <= ln and _u("<Q", blob, magic_at) == DIR_MAGIC
    if not magic_ok:
        warn("directory sentinel 0xDA1ADA1A not found where expected")

    return Directory(offset=off, length=ln, log2_capacity=log2cap, entries=entries,
                     free_list=free, magic_ok=magic_ok, crc_ok=seed is not None,
                     crc_seed=seed if seed is not None else -1)


def read_object(d: bytes, e: ObjectEntry) -> bytes:
    """Payload of one object, LZ4-decompressed when the entry says so.

    A compressed object is a u32 uncompressed length followed by a raw LZ4
    block - no frame header, which is what made the framing look exotic.
    """
    raw = d[e.offset:e.offset + e.size]
    if e.compression == 0:
        return raw
    want = _u("<I", raw, 0)
    out, _ = lz4_block_decode(raw, 4, limit=max(want, 1) * 2)
    return out[:want] if want else out


def parse_package(d: bytes, off: int) -> PackageHeader | None:
    if off < 0 or off + 0x40 > len(d) or d[off:off + 8] != PKG_MAGIC:
        return None
    return PackageHeader(
        offset=off,
        kind=_u("<I", d, off + 0x08),
        struct_version=_u("<H", d, off + 0x0C),
        format_version=_u("<H", d, off + 0x0E),
        database_uuid=str(uuid.UUID(bytes_le=d[off + 0x10:off + 0x20])),
        checksum=_u("<I", d, off + 0x30),
    )


def brace_match(d: bytes, start: int) -> int:
    """End offset of the JSON object at `start`. Length is stored nowhere, so
    the blob is delimited purely structurally. String and escape aware."""
    depth = 0
    in_str = False
    esc = False
    i = start
    while i < len(d):
        c = d[i]
        if in_str:
            if esc:
                esc = False
            elif c == 0x5C:
                esc = True
            elif c == 0x22:
                in_str = False
        elif c == 0x22:
            in_str = True
        elif c == 0x7B:
            depth += 1
        elif c == 0x7D:
            depth -= 1
            if depth == 0:
                return i + 1
        i += 1
    raise ValueError(f"unterminated catalog JSON at {start:#x}")


def read_catalog(d: bytes, e: ObjectEntry, warn) -> Catalog | None:
    """The catalog named by a directory entry - the authoritative path.

    The entry carries offset, length, compression and CRC, so nothing has to be
    searched for and nothing has to be brace-matched. Prefer this over
    find_catalogs() for anything the live directory references.
    """
    try:
        payload = json.loads(read_object(d, e))
    except (ValueError, json.JSONDecodeError) as ex:
        warn(f"Schema object at {e.offset:#x} is not valid JSON: {ex}")
        return None
    return Catalog(offset=e.offset, length=e.size,
                   digest=_u("<I", d, e.offset + e.size)
                   if e.offset + e.size + 4 <= len(d) else 0,
                   framing="object", package=parse_package(d, e.offset - 0x40),
                   data=payload)


def find_catalogs(d: bytes, warn) -> list[Catalog]:
    """Scan the whole file for catalog JSON. **Forensic only.**

    This finds catalogs the live directory no longer references - earlier
    transactions' states, which is genuinely useful when reconstructing history.
    It is the wrong way to find the *current* catalog: read_catalog() does that
    from the directory.

    The scan rests on two things the format does not guarantee: that
    `compressionMethod` is serialised first (JSON objects have no key order),
    and that the object is stored uncompressed (`lz4` is legal for any object,
    in which case the needle cannot match at all). Both hold across the corpus;
    neither is a rule.
    """
    out: list[Catalog] = []
    pos = 0
    while True:
        pos = d.find(CATALOG_NEEDLE, pos)
        if pos < 0:
            return out
        try:
            end = brace_match(d, pos)
            payload = json.loads(d[pos:end])
        except (ValueError, json.JSONDecodeError) as e:
            warn(f"catalog at {pos:#x} failed to parse: {e}")
            pos += len(CATALOG_NEEDLE)
            continue
        pkg = parse_package(d, pos - 0x40)
        out.append(Catalog(
            offset=pos,
            length=end - pos,
            digest=_u("<I", d, end) if end + 4 <= len(d) else 0,
            framing="packaged" if pkg else "bare",
            package=pkg,
            data=payload,
        ))
        pos = end


SMA_OFF = 0x30         # the SMA pair starts here, right after the five header words
SMA_SLOT = 16          # variable-width types only; fixed-width types are packed
SMA_INLINE_CAP = 12    # a string SMA slot is u4 length + at most 12 bytes of text

# Byte width of a fixed-width SMA value, keyed by the catalog's base type name.
# Fixed-width min and max are stored ADJACENTLY at SMA_OFF and SMA_OFF+width -
# the 16-byte slot rule applies only to the variable-width types below.
SMA_FIXED_WIDTH = {
    "Bool": 1,
    "SmallInt": 2,
    "Integer": 4, "Date": 4, "Oid": 4, "Float": 4,
    "BigInt": 8, "Double": 8, "Timestamp": 8, "TimestampTZ": 8, "Time": 8,
    "Numeric": 8,
    "Interval": 16, "BigNumeric": 16,
}
# Names are the catalog's own spellings, which are not the SqlType method names:
# bytes() -> "Bytea", json() -> "JSON", numeric(38,s) -> "BigNumeric".
SMA_VARIABLE = {"Varchar", "Text", "Char", "Bytea", "JSON",
                "Geography", "TabGeography"}
# Values that are raw octets, not text - decoding them as UTF-8 corrupts them.
SMA_BINARY = {"Bytea", "Geography", "TabGeography"}
# Two's-complement types. Date/Time/Timestamp/Oid are non-negative counters.
SMA_SIGNED = {"SmallInt", "Integer", "BigInt", "Numeric", "BigNumeric"}


@dataclass(frozen=True)
class ColType:
    """A catalog column type: a name plus any numeric modifiers.

    The catalog stores these as a flat array, e.g. ["Numeric", 18, 4,
    "nullable"] or ["Char", 8]. The modifiers matter - a Numeric cannot be
    rendered without its scale.
    """
    name: str | None
    mods: tuple[int, ...] = ()

    @classmethod
    def parse(cls, arr: list | None) -> "ColType":
        arr = arr or []
        mods = tuple(x for x in arr[1:] if isinstance(x, int))
        return cls(arr[0] if arr else None, mods)

    @property
    def scale(self) -> int:
        """Decimal scale for Numeric/BigNumeric; 0 for everything else."""
        return self.mods[1] if len(self.mods) > 1 else 0

    @property
    def is_variable(self) -> bool:
        return self.name in SMA_VARIABLE

    @property
    def width(self) -> int:
        return SMA_FIXED_WIDTH.get(self.name or "", 0)

    def __str__(self) -> str:
        return self.name + ("/" + "/".join(map(str, self.mods)) if self.mods else "") \
            if self.name else "?"

UNIX_EPOCH_JDN = 2440588            # JDN of 1970-01-01
US_PER_DAY = 86_400_000_000


@dataclass
class BlockIndex:
    """The per-column header, named from hyper's own assertion strings in
    `getPackInfoAndValidate` ("index.dictOffset<=index.dataOffset", …).

        +0x00 u4  tuple_count      (the engine reads this as a u4)
        +0x04 --  padding          never read; 0 in all 870 headers seen
        +0x08 u8  dict_offset      \
        +0x10 u8  data_offset       > all relative to SMA_OFF, all 16-byte aligned
        +0x18 u4  scheme           |  DataBlockCompression, asserted < 54
        +0x20 u8  string_offset    /
        +0x30     SMA, then PSMA, then domain size

    The four regions are contiguous: sma = [0x30, 0x30+dict_offset),
    dict = [.., 0x30+data_offset), data = [.., 0x30+string_offset),
    string_data = [.., object end).
    """
    tuple_count: int
    dict_offset: int
    data_offset: int
    scheme: int
    string_offset: int

    @property
    def sma_size(self) -> int:
        return self.dict_offset

    @property
    def dict_size(self) -> int:
        return self.data_offset - self.dict_offset

    @property
    def data_size(self) -> int:
        return self.string_offset - self.data_offset

    def string_data(self, blob: bytes) -> bytes:
        return blob[SMA_OFF + self.string_offset:]

    def violations(self, size: int) -> list[str]:
        """The engine's own invariants. Any hit means we have mis-modelled it."""
        bad = []
        if self.scheme >= 54:
            bad.append(f"scheme {self.scheme} >= {N_SCHEMES}")
        if not (self.dict_offset <= self.data_offset <= self.string_offset):
            bad.append("offsets not monotonic")
        if self.string_offset > max(0, size - SMA_OFF):
            bad.append("stringDataOffset past end of object")
        for nm, v in (("dict", self.dict_offset), ("data", self.data_offset),
                      ("stringData", self.string_offset)):
            if v % 16:
                bad.append(f"{nm}Offset {v} not 16-byte aligned")
        return bad


def parse_block_index(b: bytes) -> BlockIndex | None:
    if len(b) < SMA_OFF:
        return None
    dict_off, data_off = struct.unpack_from("<QQ", b, 0x08)
    return BlockIndex(tuple_count=_u("<I", b, 0x00), dict_offset=dict_off,
                      data_offset=data_off, scheme=_u("<I", b, 0x18),
                      string_offset=_u("<Q", b, 0x20))


def jdn_to_ymd(jdn: int) -> tuple[int, int, int]:
    """Julian day number -> proleptic Gregorian (year, month, day).

    Fliegel-Van Flandern. Used instead of datetime.date because Hyper's DATE
    reaches year 294276 while Python's datetime stops at 9999 - a date past
    that raised ValueError and fell through to a raw-bytes dict.
    """
    a = jdn + 32044
    b = (4 * a + 3) // 146097
    c = a - (146097 * b) // 4
    d = (4 * c + 3) // 1461
    e = c - (1461 * d) // 4
    m = (5 * e + 2) // 153
    return (100 * b + d - 4800 + m // 10, m + 3 - 12 * (m // 10),
            e - (153 * m + 2) // 5 + 1)


def _fmt_date(jdn: int) -> str:
    y, m, d = jdn_to_ymd(jdn)
    return f"{y:04d}-{m:02d}-{d:02d}"


def _fmt_timestamp(us_since_jdn0: int, tz: bool) -> str:
    days, rem = divmod(us_since_jdn0, US_PER_DAY)
    hh, rem = divmod(rem, 3_600_000_000)
    mm, rem = divmod(rem, 60_000_000)
    ss, us = divmod(rem, 1_000_000)
    out = f"{_fmt_date(days)} {hh:02d}:{mm:02d}:{ss:02d}"
    if us:
        out += f".{us:06d}"
    return out + ("+00:00" if tz else "")


def _render_fixed(raw: bytes, ct: ColType) -> Any:
    """Turn a fixed-width value into a Python object.

    Every encoding below was read off a probe block whose values were known in
    advance - see probe3.py. Widths: Numeric is a scaled i8, BigNumeric a
    scaled i16 (128-bit), Interval a {u8 microseconds, u4 days, u4 months}
    triple, Time microseconds since midnight, TimestampTZ the same epoch as
    Timestamp (stored in UTC).
    """
    import datetime
    import decimal
    name = ct.name
    n = int.from_bytes(raw, "little", signed=name in SMA_SIGNED)
    try:
        if name == "Bool":
            return bool(n)
        if name == "Date":
            return _fmt_date(n)
        if name in ("Timestamp", "TimestampTZ"):
            # TimestampTZ is stored in UTC; say so, as the engine does.
            return _fmt_timestamp(n, name == "TimestampTZ")
        if name == "Time":
            return str((datetime.datetime.min +
                        datetime.timedelta(microseconds=n)).time())
        if name == "Double":
            return struct.unpack("<d", raw)[0]
        if name == "Float":
            return struct.unpack("<f", raw)[0]
        if name in ("Numeric", "BigNumeric"):
            # Build from a string so the decimal context cannot round: a
            # BigNumeric(38, 6) needs more digits than the default precision.
            return decimal.Decimal(f"{n}e-{ct.scale}")
        if name == "Interval":
            us, days, months = struct.unpack("<QiI", raw[:16])
            return IntervalValue(months, days, us)
    except (ValueError, OverflowError, struct.error, decimal.InvalidOperation):
        return {"raw": raw.hex(), "int": n}
    return n


def decode_sma(b: bytes, ct: ColType,
               string_data: bytes = b"") -> tuple[Any, Any, str]:
    """(min, max, note) for one column block.

    Two shapes, and which one applies is decided by the declared type rather
    than guessed from the bytes. Guessing is what made Quantity's min=1/max=14
    read back as a bogus 8-byte integer: a fixed-width value has no length
    prefix, so treating the leading u32 as one is always wrong.
    """
    if ct.is_variable:
        # Both slots are a FIXED 16 bytes - min at 0x30, max at 0x40, never
        # spilling. A slot is a small-string-optimised value:
        #     u4 length; if length <= 12 the text follows inline,
        #                otherwise u4 (zero) then u8 offset into string_data.
        # The string heap is laid out min first, max second, then the rest of
        # the dictionary - which is why the max's offset always equals the
        # min's length. Decoding the non-inlined case as inline text yields
        # convincing garbage; that was a real bug in an earlier revision.
        out, unresolved = [], False
        for k in range(2):
            off = SMA_OFF + k * SMA_SLOT
            if off + SMA_SLOT > len(b):
                out.append(None)
                continue
            n = _u("<I", b, off)
            if n <= SMA_INLINE_CAP:
                out.append(_bytes_as(b[off + 4:off + 4 + n], ct))
                continue
            at = _u("<Q", b, off + 8)
            if at + n <= len(string_data):
                out.append(_bytes_as(string_data[at:at + n], ct))
            else:
                out.append(f"<{n} bytes at string_data+{at}, out of range>")
                unresolved = True
        return out[0], out[1], "SMA string offset out of range" if unresolved else ""

    width = ct.width
    if not width:
        return b[SMA_OFF:SMA_OFF + 8].hex(), b[SMA_OFF + 8:SMA_OFF + 16].hex(), \
               f"unmodelled type {ct}; SMA shown raw"
    if SMA_OFF + 2 * width > len(b):
        return None, None, "block too short for an SMA pair"
    lo = _render_fixed(b[SMA_OFF:SMA_OFF + width], ct)
    hi = _render_fixed(b[SMA_OFF + width:SMA_OFF + 2 * width], ct)
    return lo, hi, ""


def _align16(n: int) -> int:
    return -(-n // 16) * 16


def unpack_codes(buf: bytes, n: int, bits: int) -> list[int]:
    """Unpack n codes of `bits` each.

    Sub-byte codes are packed **MSB-first**: the first code occupies the high
    bits of the first byte. Confirmed by probe - three values with counts 3/5/8
    give `01 55 aa aa` at 2 bits, which is 0,0,0,1 | 1,1,1,1 | 2,2,2,2 | 2,2,2,2
    read from the top. LSB-first would put the first transition at 0x40.
    Byte-aligned widths are plain little-endian.
    """
    if bits >= 8:
        w = bits // 8
        return [int.from_bytes(buf[i * w:(i + 1) * w], "little") for i in range(n)]
    per, mask = 8 // bits, (1 << bits) - 1
    return [(buf[i // per] >> (8 - bits * (i % per + 1))) & mask for i in range(n)]


def fits_region(region_size: int, n: int, bits: int) -> bool:
    return _align16(-(-n * bits // 8)) == region_size


# ---- DataBlockCompression -------------------------------------------------
#
# scheme code -> (kind, bits per row, reserves code 0 for NULL, dictionary
# entry shape). Built by generating one block per (type, cardinality, nulls)
# combination with the Hyper API and reading back what it produced - see
# probe2.py. The enum itself is not decoded; this is a measured table.
#
# Two families, each {1,2,4}-bit packed and {8,16,32}-bit byte-packed, each in
# a non-null / nullable pair:
#   "native"  integer-like types; dictionary entries are the type's own width
#   "narrow"  entries are 4 B: Varchar packs u1 length + inline text (<= 3 B)
#             or a u3 heap offset; Double is narrowed to float32
#
# Nullability is a property of the *block*, not of the column: Hyper picks a
# non-null scheme for a nullable column that happens to contain no NULLs.
# Reading nullability off the catalog instead produced 92 wrong columns.
# Every entry below was observed directly: probe5.py sweeps each type across a
# cardinality ladder from constant to all-distinct, which walks a column through
# single -> bit-packed dict -> byte-packed dict -> truncation -> uncompressed.
# Nothing here is interpolated. An earlier revision *did* interpolate, assuming
# each family was a contiguous triple of widths, and got 38/39 wrong: 39 is not
# the 32-bit dictionary of the {36,37} family, it is an uncompressed encoding.
#
#   kind    "dict"   code indexes the dictionary
#           "trunc"  value = SMA min + code (frame of reference)
#           "raw"    the value itself, one per row
#           "single" one value for the whole block
#           "null"   every row is NULL; the block carries nothing at all
#   bits    code width; 0 under "raw" means "the type's own width"
#   entry   dictionary/value shape - see read_dictionary
SCHEMES: dict[int, tuple[str, int, bool, str | None]] = {}
for _kind, _null, _entry, _codes in (
        # native family: integer-like types, and Double when f32 would lose
        ("dict", False, "native", {1: 8, 2: 16, 24: 1, 25: 2, 26: 4}),
        ("dict", True, "native", {13: 8, 14: 16, 30: 1, 31: 2, 32: 4}),
        ("trunc", False, None, {7: 8, 8: 16, 9: 32, 52: 64}),
        ("trunc", True, None, {19: 8, 20: 16, 21: 32, 53: 64}),
        ("raw", False, "native", {10: 0}),
        ("raw", True, "native", {22: 0}),
        # narrow family: Varchar/Char/Bytea/JSON, and Double via float32
        ("dict", False, "narrow", {36: 8, 37: 16, 44: 1, 45: 2, 46: 4}),
        ("dict", True, "narrow", {40: 8, 41: 16, 47: 1, 48: 2, 49: 4}),
        ("raw", False, "narrow", {39: 32}),
        ("raw", True, "narrow", {43: 32}),
        ("single", False, None, {50: 0}),
        # 51 is 50's all-NULL twin: a 48-byte object that is nothing but the
        # header, every offset zero, no SMA and no data. Same for every type.
        # probe5.py's cardinality ladder starts at one distinct value and never
        # reaches zero, which is why the table said "fully observed" without it.
        ("null", True, None, {51: 0})):
    for _code, _bits in _codes.items():
        SCHEMES[_code] = (_kind, _bits, _null, _entry)


# The enum's own names, straight out of `hyper::to_string(DataBlockCompression)`
# at 0x10125cf60: a 54-entry char* table at __DATA_CONST:0x10b4c8198, indexed by
# the code, with "INVALID" for anything >= 54. That is where the `scheme < 54`
# assertion in getPackInfoAndValidate comes from.
#
# The naming is entirely regular:
#   DictN       dictionary, N-BYTE codes          DictBitN  N-BIT codes
#   TruncN      truncation to N bytes             Uncompressed  value per row
#   Single      one value for the block           NoSma*    ... with no SMA
#   ...Null     the nullable twin, code 0 reserved
#   Small...    4-byte entries      (no prefix) 8-byte      Huge...  16-byte
# so the prefix picks the entry shape and the numeral picks the code width.
SCHEME_NAMES = [
    "Single", "Dict1", "Dict2", "Dict4", "HugeDict1", "HugeDict2", "HugeDict4",
    "Trunc1", "Trunc2", "Trunc4", "Uncompressed", "HugeUncompressed",
    "SingleNull", "Dict1Null", "Dict2Null", "Dict4Null", "HugeDict1Null",
    "HugeDict2Null", "HugeDict4Null", "Trunc1Null", "Trunc2Null", "Trunc4Null",
    "UncompressedNull", "HugeUncompressedNull", "DictBit1", "DictBit2",
    "DictBit4", "HugeDictBit1", "HugeDictBit2", "HugeDictBit4", "DictBit1Null",
    "DictBit2Null", "DictBit4Null", "HugeDictBit1Null", "HugeDictBit2Null",
    "HugeDictBit4Null", "SmallDict1", "SmallDict2", "SmallDict4",
    "SmallUncompressed", "SmallDict1Null", "SmallDict2Null", "SmallDict4Null",
    "SmallUncompressedNull", "SmallDictBit1", "SmallDictBit2", "SmallDictBit4",
    "SmallDictBit1Null", "SmallDictBit2Null", "SmallDictBit4Null",
    "NoSmaSingle", "NoSmaSingleNull", "Trunc8", "Trunc8Null",
]
N_SCHEMES = len(SCHEME_NAMES)          # 54, the engine's own bound


def scheme_name(code: int) -> str:
    return SCHEME_NAMES[code] if 0 <= code < N_SCHEMES else "INVALID"


def _scheme_from_name(name: str) -> tuple[str, int, bool, str | None]:
    """What the enum's name says a code means. Used to audit SCHEMES."""
    nullable = name.endswith("Null")
    stem = name[:-4] if nullable else name
    entry = ("narrow" if stem.startswith("Small") else
             "huge" if stem.startswith("Huge") else "native")
    stem = stem.removeprefix("Small").removeprefix("Huge")
    if stem.startswith("DictBit"):
        return "dict", int(stem[7:]), nullable, entry
    if stem.startswith("Dict"):
        return "dict", 8 * int(stem[4:]), nullable, entry
    if stem.startswith("Trunc"):
        return "trunc", 8 * int(stem[5:]), nullable, None
    if stem == "Uncompressed":
        # the cell is the value itself: its own width natively, 4 bytes narrow
        return "raw", 32 if entry == "narrow" else 0, nullable, entry
    if stem in ("Single", "NoSmaSingle"):
        # an all-NULL block carries nothing, so it decodes without a value path
        return ("null" if nullable and stem == "NoSmaSingle" else "single",
                0, nullable, None)
    raise ValueError(name)


# Every code measured by probe5/probe6/probe8/probe9 must agree with what its
# name says. It does, for all 34 - an independent confirmation of a table that
# was built entirely from observation, and of scheme 51 in particular:
# NoSmaSingleNull is exactly the all-NULL block the probe found.
SCHEMES_MEASURED = frozenset(SCHEMES)
for _code, _spec in SCHEMES.items():
    _want = _scheme_from_name(SCHEME_NAMES[_code])
    assert _spec == _want, (_code, SCHEME_NAMES[_code], _spec, _want)

# The remaining codes are filled in from their names. This is not the
# interpolation that §9.18 retracted - that guessed the code *numbering* from a
# pattern; these come from the engine's own name table, decoded by a rule the 34
# measured codes agree with. Only the code width and nullability vary, and the
# decoder is generic in both.
#
# `Huge*` is deliberately left out. Its name says the dictionary entry is a
# third, wider shape, and 16 bytes is the obvious guess, but nothing has been
# measured - so those 14 codes still make the decoder refuse rather than invent
# a layout. See §10.
for _code, _name in enumerate(SCHEME_NAMES):
    if _code in SCHEMES or _name.startswith("Huge"):
        continue
    SCHEMES[_code] = _scheme_from_name(_name)


def null_sentinel(cell: bytes, ct: ColType, entry: str | None) -> bool:
    """Is this raw cell the NULL marker?

    "raw" has no spare code, so NULL is the **maximum value of the storage
    representation**: 0x7FFF for an i16, +Inf for f32/f64, 0xFFFFFFFF for the
    4-byte narrow entry. Confirmed for SmallInt, Float, Double, Varchar, Char,
    Bytea and JSON; the unsigned-counter types (Date, Time, Timestamp, Oid)
    have no nullable-raw case in the corpus, so their sentinel is the analogous
    all-ones and is unverified.
    """
    w = len(cell)
    if ct.is_variable:
        # Both string cell shapes use an all-ones cell for NULL - 0xFFFFFFFF
        # for the 4-byte narrow one, 0xFFFFFFFFFFFFFFFF for the 8-byte wide.
        return cell == b"\xff" * w
    if ct.name in ("Double", "Float") or (entry == "narrow" and ct.name == "Double"):
        try:
            v = struct.unpack("<d" if w == 8 else "<f", cell)[0]
        except struct.error:
            return False
        return v == float("inf")
    if ct.name in SMA_SIGNED:
        return int.from_bytes(cell, "little", signed=True) == (1 << (8 * w - 1)) - 1
    return cell == b"\xff" * w


DICT_ENTRY = 4        # u1 length + (inline text if length <= 3 else u3 heap offset)
DICT_INLINE_CAP = 3

# Where the domain size (distinct count) sits, as a u4 relative to SMA_OFF.
# `DataBlock::getDistinctValues` indexes a per-basic-type table at
# __const:0x106e971d0 for exactly this; the values below are that table,
# re-keyed by the value width we already track. It is NOT a fixed distance
# from the dictionary - the PSMA is 2048 x width bytes, so the domain lands in
# a different place for every width.
DOMAIN_OFF_BY_WIDTH = {1: 2060, 2: 4108, 4: 8208, 8: 16408, 16: 32808}
DOMAIN_OFF_VARIABLE = 2088


def domain_offset(ct: ColType) -> int | None:
    if ct.is_variable:
        return DOMAIN_OFF_VARIABLE
    return DOMAIN_OFF_BY_WIDTH.get(ct.width)


def _render_int(n: int, ct: ColType) -> Any:
    """Render an already-decoded integer as its SQL type."""
    w = ct.width or 8
    return _render_fixed((n & ((1 << (8 * w)) - 1)).to_bytes(w, "little"), ct)


def _sma_raw_ints(blob: bytes, ct: ColType) -> tuple[int, int] | None:
    """SMA min/max as raw integers - what truncation needs as its base.

    `decode_sma` renders Date as an ISO string and Timestamp as a datetime,
    which is right for display and useless as an arithmetic base.
    """
    w = ct.width
    if not w or SMA_OFF + 2 * w > len(blob):
        return None
    sg = ct.name in SMA_SIGNED
    return (int.from_bytes(blob[SMA_OFF:SMA_OFF + w], "little", signed=sg),
            int.from_bytes(blob[SMA_OFF + w:SMA_OFF + 2 * w], "little", signed=sg))


def _narrow_entry_raw(raw: bytes, ct: ColType, entry: str | None) -> Any:
    """One uncompressed value: no dictionary, no code - the value itself."""
    if entry == "narrow" and ct.name == "Double":
        return struct.unpack("<f", raw)[0]
    return _render_fixed(raw, ct)


def _bytes_as(raw: bytes, ct: ColType) -> Any:
    """A variable-length payload: octets for binary types, UTF-8 otherwise.

    A `Char(n)` is stored with its trailing blanks stripped but is blank-padded
    back to n on the way out, which is what SQL says CHAR means and what the
    Hyper API returns. The corpus never showed this because its only Char
    column held values that filled the width exactly.
    """
    if ct.name in SMA_BINARY:
        return raw
    s = raw.decode("utf-8", "replace")
    if ct.name == "Char" and ct.mods:
        return s.ljust(ct.mods[0])
    return s


def _narrow_entry(raw: bytes, ct: ColType, heap: bytes) -> Any:
    """One 4-byte dictionary entry from the narrow family.

    Varchar packs `u1 length` plus the text inline (<= 3 bytes) or a `u3`
    offset into the string heap. Double is stored as an IEEE **float32** -
    Hyper only picks this family when every value round-trips, so it is not
    lossy; a column like `Sales` with values that need full precision gets the
    native 8-byte family instead.
    """
    if ct.is_variable:
        n = raw[0]
        payload = (raw[1:1 + n] if n <= DICT_INLINE_CAP
                   else heap[(at := int.from_bytes(raw[1:4], "little")):at + n])
        return _bytes_as(payload, ct)
    if ct.name == "Double":
        return struct.unpack("<f", raw)[0]
    return NotImplemented


WIDE_ENTRY = 8         # u4 length + (inline text if length <= 4 else u4 heap offset)
WIDE_INLINE_CAP = 4


def _wide_entry(raw: bytes, ct: ColType, heap: bytes) -> Any:
    """One 8-byte dictionary entry for a variable-width type.

    The same shape as the 4-byte narrow entry, one size up: a `u4` length, then
    four bytes that are the text itself when it fits and a `u4` heap offset
    otherwise. The narrow entry's `u1` length caps a value at 255 bytes, so a
    column with anything longer gets this one instead.

    Reading it as a flat `u4 length, u4 offset` decodes the long values and
    silently empties every short one - a 1,060-row hole in an 8,447-row column,
    all of it the short artist names.
    """
    n = _u("<I", raw, 0)
    payload = (raw[4:4 + n] if n <= WIDE_INLINE_CAP
               else heap[(at := _u("<I", raw, 4)):at + n])
    return _bytes_as(payload, ct)


def read_dictionary(blob: bytes, idx: BlockIndex, domain: int,
                    ct: ColType, entry: str) -> list[Any]:
    """Dictionary entries in code order (code == index).

    `entry` comes from the scheme, not from the SQL type - a Double column can
    be encoded either way, and the entry width differs between them.

      "narrow"  4 bytes: u1 length, then the payload inline when length <= 3,
                otherwise a u3 offset into the string heap. The payload is raw
                bytes of the value, so an 8-byte Double lands in the heap too.
      "native"  the value at the type's own width, laid out end to end - or,
                for a variable-width type, the 8-byte `u4 length, u4 offset`
                entry above. A Varchar really does turn up under a native
                scheme; see the note in SCHEMES.

    Under a nullable scheme entry 0 is a reserved NULL placeholder, which is
    why `domain` is one more than the number of distinct values.
    """
    base = SMA_OFF + idx.dict_offset
    if entry == "narrow":
        heap = idx.string_data(blob)
        out = []
        for k in range(domain):
            e = blob[base + DICT_ENTRY * k: base + DICT_ENTRY * (k + 1)]
            if len(e) < DICT_ENTRY:
                return []
            v = _narrow_entry(e, ct, heap)
            if v is NotImplemented:
                return []
            out.append(v)
        return out
    if ct.is_variable:
        heap = idx.string_data(blob)
        out = []
        for k in range(domain):
            e = blob[base + WIDE_ENTRY * k: base + WIDE_ENTRY * (k + 1)]
            if len(e) < WIDE_ENTRY:
                return []
            out.append(_wide_entry(e, ct, heap))
        return out
    w = ct.width
    if not w:
        return []
    return [_render_fixed(blob[base + w * k: base + w * (k + 1)], ct)
            for k in range(domain)]


def decode_column(blob: bytes, idx: BlockIndex,
                  ct: ColType) -> tuple[list[Any], str]:
    """Decode one column block into row values. Returns (values, note).

    Everything is driven by the block's own `scheme` word:

      single      every row is the SMA min
      truncation  value = SMA min + stored code
      dictionary  the code indexes the dictionary

    NULLs need no bitmap. Under a nullable scheme the code 0 is reserved: for
    truncation real values shift up by one, for a dictionary entry 0 is a
    placeholder. Whether a scheme is nullable is a property of the block, not
    of the column's declared type.
    """
    n = idx.tuple_count
    if n <= 0:
        return [], "no tuples"
    spec = SCHEMES.get(idx.scheme)
    if spec is None:
        return [], (f"DataBlockCompression {idx.scheme} "
                    f"({scheme_name(idx.scheme)}) has no decoder")
    kind, bits, nullable, entry = spec
    if kind == "null":
        return [None] * n, ""
    lo, hi, _ = decode_sma(blob, ct, idx.string_data(blob))

    if kind == "single":
        # A single-value block stores ONE value, not an SMA pair, and is only
        # 64 bytes. decode_sma wants room for min *and* max, which overruns for
        # a 16-byte type (BigNumeric, Interval) - so read the one value here.
        if ct.is_variable:
            return [lo] * n, ""
        w = ct.width
        at = SMA_OFF + idx.data_offset
        if not w or at + w > len(blob):
            return [], f"single-value block too short for {ct}"
        return [_render_fixed(blob[at:at + w], ct)] * n, ""

    data = blob[SMA_OFF + idx.data_offset: SMA_OFF + idx.string_offset]
    if kind != "raw" and not fits_region(idx.data_size, n, bits):
        return [], (f"{scheme_name(idx.scheme)} implies {bits}-bit codes but the data "
                    f"region is {idx.data_size}B for {n} rows")
    if kind == "raw":
        # An uncompressed *string* cell is the same shape as a dictionary entry
        # of the same family - 4 bytes narrow, 8 bytes wide - not the type's own
        # width, which a variable-width type does not have. A column of 100,000
        # distinct long strings lands here: no dictionary is worth building, so
        # the engine writes scheme 10 with 8-byte cells.
        w = (WIDE_ENTRY if ct.is_variable and entry != "narrow" else 0) \
            or (bits // 8) or ct.width
        if not w or len(data) < n * w:
            return [], f"raw block: no width for {ct}"
        heap = idx.string_data(blob)
        out = []
        for k in range(n):
            cell = data[k * w:(k + 1) * w]
            if nullable and null_sentinel(cell, ct, entry):
                out.append(None)
            elif entry == "narrow":
                out.append(_narrow_entry(cell, ct, heap))
            elif ct.is_variable:
                out.append(_wide_entry(cell, ct, heap))
            else:
                out.append(_render_fixed(cell, ct))
        return out, ""

    raw = unpack_codes(data, n, bits)

    if kind == "trunc":
        span = _sma_raw_ints(blob, ct)
        if span is None:
            return [], f"no truncation base for type {ct}"
        base = span[0]
        if nullable:
            return [None if v == 0 else _render_int(base + v - 1, ct)
                    for v in raw], ""
        return [_render_int(base + v, ct) for v in raw], ""

    dom_at = domain_offset(ct)
    if dom_at is None or SMA_OFF + dom_at + 4 > len(blob):
        return [], f"no domain-size location known for type {ct}"
    domain = _u("<I", blob, SMA_OFF + dom_at)
    if not (0 < domain <= 1 << 24):
        return [], f"implausible domain size {domain}"
    table = read_dictionary(blob, idx, domain, ct, entry or "native")
    if not table:
        return [], f"no {entry} dictionary reader for type {ct}"
    out = []
    for c in raw:
        if c >= len(table):
            return [], f"code {c} outside a {len(table)}-entry dictionary"
        out.append(None if (nullable and c == 0) else table[c])
    return out, ""


def parse_relation_metadata(blob: bytes) -> tuple[list[tuple[int, int]], str]:
    """The tombstone list: which rows a `Relation_Metadata` marks deleted.

    A DELETE does not rewrite the Data Blocks. The rows stay in place, the
    `Relation_Header` still counts them, and this object records which ones are
    gone as **inclusive ranges of row index**. Ignore it and a relation reads
    back with its deleted rows still in it - right values, too many of them,
    which is the failure mode that survives a spot check.

        u1  present     0 -> that byte is the whole object, nothing deleted
        u1  ?           1 in every sample
        node            the root

    A node is a **B-tree**, not a flat list:

        u1  kind
        kind 1 (leaf)      u4 count, then count x {u8 first, u8 last}
        kind 0 (internal)  u4 n, then n u8 separators - the last row index in
                           each child but the final one - then n+1 child nodes

    The engine fills a leaf with 100 ranges, then 50, so the separators support
    a binary search. Depth grows with the relation: 200 deleted ranges is one
    internal node over three leaves, and 87,385 ranges is three levels. Nothing
    is aligned, and a leaf-only root is written for anything up to 100 ranges.
    """
    if not blob:
        return [], "empty Relation_Metadata"
    if blob[0] == 0:
        return [], ""
    if len(blob) < 3:
        return [], "Relation_Metadata too short"

    def node(at: int, depth: int) -> tuple[list[tuple[int, int]], int]:
        if depth > 32:
            raise ValueError("tombstone tree deeper than 32 levels")
        kind = blob[at]
        n = _u("<I", blob, at + 1)
        at += 5
        if kind == 1:
            if at + 16 * n > len(blob):
                raise ValueError(f"leaf of {n} ranges overruns the object")
            return ([struct.unpack_from("<QQ", blob, at + 16 * k)
                     for k in range(n)], at + 16 * n)
        if kind != 0:
            raise ValueError(f"node kind {kind} at {at - 5}")
        at += 8 * n                       # skip the separators
        out = []
        for _ in range(n + 1):
            part, at = node(at, depth + 1)
            out.extend(part)
        return out, at

    try:
        out, end = node(2, 0)
    except (ValueError, struct.error, IndexError) as e:
        return [], f"Relation_Metadata: {e}"
    if end != len(blob):
        return [], f"Relation_Metadata: parsed {end} of {len(blob)} bytes"
    return out, ""


def relation_metadata(hf: "HyperFile", relation: int) -> list[ObjectEntry]:
    return sorted((e for e in (hf.directory.entries if hf.directory else [])
                   if e.category == 5 and e.relation == relation),
                  key=lambda e: e.block)


def deleted_rows(d: bytes, hf: "HyperFile",
                 relation: int) -> tuple[list[tuple[int, int]], str]:
    """Rows this relation has deleted but not reclaimed, as sorted ranges.

    There is one `Relation_Metadata` per relation and its row indices are
    **relation-global**, not block-local: a 262,154-row relation puts the
    tombstone for its 131,077th row at 131077, which lands in block 1. Kept as
    ranges rather than expanded into a set - a mostly-deleted large relation
    would otherwise cost one Python int per dead row.
    """
    out: list[tuple[int, int]] = []
    for e in relation_metadata(hf, relation):
        ranges, note = parse_relation_metadata(read_object(d, e))
        if note:
            return [], note
        out.extend(ranges)
    return sorted(out), ""


def _is_deleted(ranges: list[tuple[int, int]], row: int) -> bool:
    k = bisect.bisect_right(ranges, (row, MAX_U64)) - 1
    return k >= 0 and ranges[k][1] >= row


def column_objects(hf: "HyperFile", relation: int, index: int) -> list[ObjectEntry]:
    """Every Data Block object for one column, in block order.

    A relation is split into blocks of at most 2**17 rows, and each block
    stores every column separately, so a column is `ceil(rows / 131072)`
    objects sharing an ObjectStoreId index and differing in `block`. Reading
    only the first is silently short - the values are right, there are just
    too few of them.
    """
    if not hf.directory:
        return []
    return sorted((e for e in hf.directory.entries
                   if e.category == 4 and e.relation == relation
                   and e.index == index), key=lambda e: e.block)


def decode_relation_column(d: bytes, hf: "HyperFile", relation: int,
                           index: int, ct: ColType) -> tuple[list[Any], str]:
    """All *live* rows of one column, concatenated across its blocks.

    Deleted rows are still in the blocks - a DELETE only writes a tombstone -
    so they are dropped here using the relation's `Relation_Metadata`. Row
    index is relative to the block, not to the relation.
    """
    out, notes = [], []
    gone, note = deleted_rows(d, hf, relation)
    if note:
        return [], note
    base = 0
    for e in column_objects(hf, relation, index):
        blob = read_object(d, e)
        idx = parse_block_index(blob)
        if not idx:
            notes.append(f"block {e.block}: header too short")
            continue
        vals, note = decode_column(blob, idx, ct)
        if note and not vals:
            return [], f"block {e.block}: {note}"
        out.extend(v for k, v in enumerate(vals)
                   if not gone or not _is_deleted(gone, base + k))
        base += idx.tuple_count
        if note:
            notes.append(f"block {e.block}: {note}")
    return out, "; ".join(notes)


MULTI_HDR = 0x20       # the sample's own preamble, ahead of the Data Block
MULTI_GROUP = 40       # one attribute group: sma, dict, data, scheme, stringData


@dataclass
class MultiBlock:
    """A Data Block holding every attribute in one object.

    `getPackInfoAndValidate` has two branches, chosen by a flag in the
    DataBlockHandle. The per-column branch reads the `BlockIndex` above; this
    one reads an array of 40-byte groups:

        +0x00 u4  tuple_count
        +0x04 --  padding, as above
        +0x08     groups[attribute_count], five u8 each:
                    sma, dict, data, scheme, stringData

    all offsets relative to the start of the Data Block. Layout is
    region-major - every attribute's SMA, then every dictionary, then every
    data region, then every string heap - so an attribute's region ends where
    the *next* attribute's same region begins. The engine spells this out: for
    attribute k it takes the ends from group k+1, and for the last attribute it
    takes them from group 0's next field along (`end.sma = group0.dict`,
    `end.dict = group0.data`, `end.data = group0.stringData`) with
    `end.stringData = block size`.
    """
    tuple_count: int
    groups: list[tuple[int, int, int, int, int]]
    body: bytes

    def regions(self, k: int) -> tuple[bytes, bytes, bytes, bytes, int]:
        """(sma, dict, data, string_heap, scheme) for attribute k."""
        sma, dic, dat, scheme, sd = self.groups[k]
        if k + 1 == len(self.groups):
            g0 = self.groups[0]
            e_sma, e_dic, e_dat, e_sd = g0[1], g0[2], g0[4], len(self.body)
        else:
            g = self.groups[k + 1]
            e_sma, e_dic, e_dat, e_sd = g[0], g[1], g[2], g[4]
        return (self.body[sma:e_sma], self.body[dic:e_dic],
                self.body[dat:e_dat], self.body[sd:e_sd], scheme)

    def as_column_blob(self, k: int) -> bytes:
        """Attribute k rewritten in the per-column layout.

        The two layouts differ only in where the four regions sit, so copying
        them out in per-column order lets the whole existing decoder - SMA,
        domain size, dictionary, code unpacking - run unchanged. Region sizes
        stay 16-byte aligned because every region start is.
        """
        sma, dic, dat, heap, scheme = self.regions(k)
        h = bytearray(SMA_OFF)
        struct.pack_into("<I", h, 0x00, self.tuple_count)
        struct.pack_into("<Q", h, 0x08, len(sma))
        struct.pack_into("<Q", h, 0x10, len(sma) + len(dic))
        struct.pack_into("<I", h, 0x18, scheme)
        struct.pack_into("<Q", h, 0x20, len(sma) + len(dic) + len(dat))
        return bytes(h) + sma + dic + dat + heap

    def violations(self) -> list[str]:
        bad = []
        if len(self.body) % 16:
            bad.append(f"block size {len(self.body)} not 16-byte aligned")
        for k in range(len(self.groups)):
            sma, dic, dat, heap, scheme = self.regions(k)
            if scheme >= 54:
                bad.append(f"attr {k}: scheme {scheme} >= 54")
            for nm, r in (("sma", sma), ("dict", dic), ("data", dat),
                          ("stringData", heap)):
                if len(r) < 0:
                    bad.append(f"attr {k}: {nm} region runs backwards")
            for nm, off in zip(("sma", "dict", "data", "stringData"),
                               (self.groups[k][0], self.groups[k][1],
                                self.groups[k][2], self.groups[k][4])):
                if off % 16:
                    bad.append(f"attr {k}: {nm} offset {off} not 16-byte aligned")
        return bad


@dataclass
class RelationSample:
    """A `Relation_Sample` object: a reservoir sample of a relation.

    One per relation, built only once a relation is large enough. The object
    wraps a multi-attribute Data Block in a preamble and a statistics footer:

        +0x00 u8  rows in the relation
        +0x08 u8  size of the Data Block that follows
        +0x10 u8  attribute count
        +0x18 u8  version                  0 for a 2021.4 writer, 1 since
        +0x20     the Data Block
        ...       footer, and its shape depends on `version`:
                    v0   attribute_count u8 - the distinct estimates, and
                         nothing else
                    v1   the same, then u8 sample rows, then
                         3 x attribute_count f64 power sums

    The footer's per-attribute u8 is a *relation-wide* distinct-value estimate
    extrapolated from the sample, not a count: it is exact when the sample is
    all-distinct (200,000) or the domain is tiny (2 for a Bool), and runs well
    over the truth in between - 5,483 for a column with 1,000 distinct labels.
    Two different 1,000-label columns produce the same 5,483, and adding NULLs
    moves it to 4,103, so it is a function of the sample's shape. Which
    estimator has not been identified.

    The f64s are power sums over the sampled values - sum, sum of squares, sum
    of cubes - moment-major, so every attribute's sum comes first. A Varchar
    contributes its byte length rather than its text, and NULLs are skipped.

    `version` is 0 in the one 2021.4-written file in the corpus and 1 in
    everything from 0.0.23054 on. **It changes the footer**: a v0 sample stops
    after the distinct estimates, so `sample_rows` is 0 and `moments` is empty.
    The Data Block itself is identical in both.
    """
    relation_rows: int
    block: MultiBlock
    version: int
    distinct_estimates: list[int]
    sample_rows: int
    moments: list[tuple[float, float, float]]


def parse_relation_sample(blob: bytes) -> RelationSample | None:
    if len(blob) < MULTI_HDR + SMA_OFF:
        return None
    rel_rows, size, nattr, version = struct.unpack_from("<QQQQ", blob, 0)
    if not (0 < nattr <= 1 << 12) or MULTI_HDR + size > len(blob):
        return None
    body = blob[MULTI_HDR:MULTI_HDR + size]
    if len(body) < 8 + MULTI_GROUP * nattr:
        return None
    block = MultiBlock(
        tuple_count=_u("<I", body, 0),
        groups=[struct.unpack_from("<QQQQQ", body, 8 + MULTI_GROUP * k)
                for k in range(nattr)],
        body=body)
    tail = blob[MULTI_HDR + size:]
    if len(tail) < 8 * nattr:
        return RelationSample(rel_rows, block, version, [], 0, [])
    est = list(struct.unpack_from(f"<{nattr}Q", tail, 0))
    # A v0 footer stops here. v1 adds the sample row count and the power sums;
    # its u8 prologue is attribute_count + 1 words, not a fixed four, which only
    # shows up past three attributes since 3 + 1 is four.
    head = 8 * (nattr + 1)
    if len(tail) < head + 24 * nattr:
        return RelationSample(rel_rows, block, version, est, 0, [])
    rows = _u("<Q", tail, 8 * nattr)
    m = struct.unpack_from(f"<{3 * nattr}d", tail, head)
    return RelationSample(rel_rows, block, version, est, rows,
                          [(m[k], m[nattr + k], m[2 * nattr + k])
                           for k in range(nattr)])


def decode_sample(blob: bytes, types: list[ColType]) -> tuple[list[list[Any]], str]:
    """Decode a Relation_Sample into one value list per attribute."""
    s = parse_relation_sample(blob)
    if s is None:
        return [], "not a Relation_Sample layout"
    bad = s.block.violations()
    if bad:
        return [], "; ".join(bad)
    if len(types) != len(s.block.groups):
        return [], (f"{len(s.block.groups)} attributes in the sample but "
                    f"{len(types)} columns in the catalog")
    out, notes = [], []
    for k, ct in enumerate(types):
        sub = s.block.as_column_blob(k)
        vals, note = decode_column(sub, parse_block_index(sub), ct)
        if note and not vals:
            return [], f"attribute {k}: {note}"
        out.append(vals)
        if note:
            notes.append(f"attribute {k}: {note}")
    return out, "; ".join(notes)


# What the footer's power sums are taken over. Not the rendered value: a Date
# contributes its Julian day number and a Timestamp its microsecond count, so
# summing the printed "1975-04-20" is meaningless. Decoding the attribute a
# second time as the plain integer of the same width recovers exactly what the
# engine summed - the two types have identical block layouts, only the
# rendering differs. Numeric contributes its decimal value, not the scaled
# integer, so it stays as it is.
MOMENT_AS_RAW_INT = {"Date": "Integer", "Oid": "Integer",
                     "Time": "BigInt", "Timestamp": "BigInt",
                     "TimestampTZ": "BigInt"}


def moment_values(blob: bytes, types: list[ColType]) -> list[list[float] | None]:
    """The numbers the footer's power sums are taken over, per attribute.

    A string or byte string contributes its length; a counter type its raw
    count; a Bool nothing at all - the engine writes NaN for those three sums
    rather than adding up 0 and 1.
    """
    s = parse_relation_sample(blob)
    if s is None:
        return []
    out: list[list[float] | None] = []
    for k, ct in enumerate(types):
        if ct.name == "Bool":
            out.append(None)
            continue
        # A Char contributes its *stored* length, before the blank padding that
        # SQL adds back on the way out - so read it as a plain Text.
        use = ColType(MOMENT_AS_RAW_INT[ct.name], ()) \
            if ct.name in MOMENT_AS_RAW_INT else \
            ColType("Text", ()) if ct.name == "Char" else ct
        sub = s.block.as_column_blob(k)
        vals, note = decode_column(sub, parse_block_index(sub), use)
        if note and not vals:
            out.append(None)
            continue
        try:
            out.append([float(len(v.encode() if isinstance(v, str) else v))
                        if isinstance(v, (str, bytes)) else float(v)
                        for v in vals if v is not None])
        except (TypeError, ValueError):
            out.append(None)
    return out


def check_sample_moments(blob: bytes, types: list[ColType]) -> list[str]:
    """Recompute the footer's power sums from the values we decoded.

    A self-check that needs no oracle: the engine wrote sum, sum of squares and
    sum of cubes over the same sample rows, so any error in a code width, a
    region extent or a dictionary index moves at least one of them.
    """
    s = parse_relation_sample(blob)
    if s is None or not s.moments:
        return []
    # The footer's row count is not always the block's. Past two blocks the
    # sampler visits more rows than the reservoir keeps - 1,310 for a 3-block
    # relation, 1,966 for a 4-block one, against a block that still holds
    # 1,000 - and the power sums cover every row it visited. The block does not
    # contain the others, so there is nothing to recompute from.
    if s.sample_rows != s.block.tuple_count:
        return []
    bad = []
    for k, xs in enumerate(moment_values(blob, types)):
        if xs is None:                      # Bool, or an attribute that failed
            if not all(math.isnan(v) for v in s.moments[k]) \
                    and types[k].name == "Bool":
                bad.append(f"attribute {k}: Bool, but the footer is not NaN")
            continue
        for p in (1, 2, 3):
            got = math.fsum(x ** p for x in xs)
            want = s.moments[k][p - 1]
            if got != want and abs(got - want) > 1e-6 * max(1.0, abs(want)):
                bad.append(f"attribute {k}: sum of x**{p} is {got!r}, "
                           f"footer says {want!r}")
    return bad


def relation_samples(hf: "HyperFile") -> list[ObjectEntry]:
    return sorted((e for e in (hf.directory.entries if hf.directory else [])
                   if e.category == 6), key=lambda e: (e.relation, e.block))


def probe_blocks(d: bytes, hf: "HyperFile", warn) -> list[ColumnBlock]:
    """Every Relation_DataBlock in the file, resolved against the live catalog.

    Replaces the old heuristic LZ4 sweep, which could only ever see compressed
    blocks - it reported 4 of this file's 12, at offsets four bytes past the
    real object because it started decoding after the u32 length prefix.
    """
    if not hf.directory:
        return []
    live = hf.live_catalog
    rels = relations(live, warn) if live else []
    out = []
    for e in sorted(hf.directory.entries,
                    key=lambda e: (e.relation, e.index, e.block)):
        if e.category != 4:
            continue
        try:
            dec = read_object(d, e)
        except Exception as ex:                     # keep going across a file
            warn(f"{e}: could not decode: {type(ex).__name__}: {ex}")
            continue

        name, ct = None, ColType(None)
        note = ""
        if e.index:
            if e.relation < len(rels):
                attrs = rels[e.relation].get("attributes", [])
                if e.index - 1 < len(attrs):
                    a = attrs[e.index - 1]
                    name, ct = a["name"], ColType.parse(a.get("type"))
                else:
                    note = (f"index {e.index} exceeds the {len(attrs)} columns "
                            f"the catalog declares")
            else:
                note = f"no relation {e.relation} in the live catalog"

        idx = parse_block_index(dec)
        blk = ColumnBlock(
            relation=e.relation, index=e.index, block=e.block,
            column=name, sql_type=str(ct),
            offset=e.offset, stored_len=e.size, decoded_len=len(dec),
            compression=COMPRESSION.get(e.compression, str(e.compression)),
            tuple_count=idx.tuple_count if idx else -1,
            hdr=idx, note=note,
        )
        if idx:
            for v in idx.violations(len(dec)):
                warn(f"{e} ({name}): block index violates the engine's own "
                     f"invariant: {v}")
        if e.index and idx:
            blk.sma_min, blk.sma_max, sma_note = decode_sma(
                dec, ct, idx.string_data(dec))
            blk.note = "; ".join(x for x in (blk.note, sma_note) if x)
            # A constant column is a single-value block: no SMA and no dict
            # region at all, so only the min slot is meaningful.
            if idx.dict_offset == 0 and idx.dict_size == 0:
                blk.sma_max = blk.sma_min
                blk.note = "; ".join(x for x in (
                    blk.note, "single-value block: max inferred from min") if x)
        out.append(blk)
    return out


def parse(path: str, want_blocks: bool = False) -> HyperFile:
    with open(path, "rb") as fh:
        d = fh.read()

    hf = HyperFile(path=path, size=len(d))
    warn = hf.warnings.append

    if len(d) < 2 * PAGE_SIZE:
        warn("file smaller than two pages; not a .hyper file?")
        return hf
    if len(d) % EXTENT_SIZE:
        warn(f"size {len(d)} is not a multiple of {EXTENT_SIZE}; expected whole 64 KiB extents")

    for base in (0, PAGE_SIZE):
        sb = parse_superblock(d, base, warn)
        if sb:
            hf.superblocks.append(sb)

    live = hf.live
    if live:
        if live.file_size != len(d):
            warn(f"live superblock says file_size={live.file_size:#x} but file is {len(d):#x}")
        for sb in hf.superblocks:
            if not sb.checksum_ok:
                warn(f"superblock {sb.offset:#x}: CRC-32C mismatch "
                     f"(stored {sb.checksum:#010x}) - torn or stale write")
        if len({s.format_version for s in hf.superblocks if s.checksum_ok}) > 1:
            warn("superblocks disagree on format version")
        if live.encrypted:
            warn("root record flags this database as encrypted; objects will not decode")
        hf.directory = parse_directory(d, live, warn)

    if hf.directory:
        for e in hf.directory.entries:
            if e.category == 2:                       # ObjectStoreId::Schema
                hf._live_catalog = read_catalog(d, e, warn)
                break
    hf.catalogs = find_catalogs(d, warn)
    if hf._live_catalog and not any(c.offset == hf._live_catalog.offset
                                    for c in hf.catalogs):
        hf.catalogs.insert(0, hf._live_catalog)
    if not hf.catalogs:
        warn("no system catalog found")

    if want_blocks:
        hf.blocks = probe_blocks(d, hf, warn)

    return hf


# --------------------------------------------------------------------------
# Reporting
# --------------------------------------------------------------------------

NAMESPACE_PARENT_BASE = 32   # relation["parent"] == 32 + index into namespaces
# Cross-check only, never used to decide anything: a user namespace's oid runs
# 10003 + its array index (slot 0 is always "public", oid 10001, and does not
# follow). Holds for all 83 user slots in the corpus. If this and the parent-32
# rule ever disagree, one of the two has broken and the warning says so.
NAMESPACE_OID_BASE = 10003


def relations(cat: Catalog, warn=None) -> list[dict]:
    """Relations from a catalog, each annotated with its resolved schema.

    A relation names its schema through `parent`, a slot number that indexes
    the catalog's namespaces array biased by 32 - the first 32 slots are
    reserved for built-ins, so `namespaces[0]` ("public") is `parent == 32`.

    **The array is positional and dropped schemas keep their slot**, appearing
    as `{"dropped": true}` with no name. Compacting the array before indexing -
    filtering those out, as anything iterating "the schemas" naturally would -
    shifts every later relation into the wrong schema.

    Table names are NOT unique within a file - only (schema, name) is - so
    always key on the pair.
    """
    out = []
    spaces = cat.data.get("namespaces", []) or []
    if warn:
        for i, n in enumerate(spaces):
            if i and "oid" in n and n["oid"] - i != NAMESPACE_OID_BASE:
                warn(f"namespace slot {i} ({n.get('name')!r}) has oid {n['oid']}, "
                     f"breaking the oid == {NAMESPACE_OID_BASE} + index pattern; "
                     f"the parent-{NAMESPACE_PARENT_BASE} rule may not hold either")
                break
    for r in cat.data.get("relations", []) or []:
        idx = r.get("parent", -1) - NAMESPACE_PARENT_BASE
        if 0 <= idx < len(spaces) and "name" in spaces[idx]:
            schema = spaces[idx]["name"]
        elif 0 <= idx < len(spaces):
            schema = "?"
            if warn:
                warn(f"relation {r.get('name')!r}: parent={r.get('parent')} points at "
                     f"namespace slot {idx}, which is dropped")
        else:
            schema = "?"
            if warn:
                warn(f"relation {r.get('name')!r}: parent={r.get('parent')} does not "
                     f"index the {len(spaces)} namespaces; schema unresolved")
        out.append({**r, "_schema": schema})
    return out


def qualified(r: dict) -> str:
    return f'"{r["_schema"]}"."{r["name"]}"'


def report(hf: HyperFile, show_blocks: bool) -> None:
    hf_warn = hf.warnings.append
    raw = open(hf.path, "rb").read()
    print(f"\n{'=' * 78}\n{hf.path}\n{'=' * 78}")
    print(f"  size            {hf.size:,} bytes ({hf.size:#x}) = "
          f"{hf.size // EXTENT_SIZE} x 64KiB extent(s), {hf.size // PAGE_SIZE} pages")

    for sb in hf.superblocks:
        tag = "LIVE " if sb is hf.live else "stale"
        compat = "" if sb.min_version == sb.creator_version else f" (min {sb.min_version})"
        print(f"  [{tag}] page {sb.page_index}  txn={sb.txn_id:<4} fmt=v{sb.format_version} "
              f"writer={sb.creator_version}{compat}  file_size={sb.file_size:#x}")
        print(f"          directory=({sb.dir_offset:#x}, {sb.dir_length} of {sb.dir_capacity}B)  "
              f"commit_id={sb.commit_id:#018x}")
        print(f"          crc32c={sb.checksum:#010x} {'OK' if sb.checksum_ok else 'MISMATCH'}")
        if sb.unknown_tail:
            print(f"          !! unmodelled non-zero bytes: {sb.unknown_tail[:64]}")

    dr = hf.directory
    if dr:
        print(f"\n  directory @ {dr.offset:#x}  {dr.length:,}B  capacity={1 << dr.log2_capacity}  "
              f"crc={'OK seed=' + hex(dr.crc_seed) if dr.crc_ok else 'MISMATCH'}  "
              f"sentinel={'OK' if dr.magic_ok else 'MISSING'}")
        print(f"      {len(dr.entries)} live objects, {len(dr.free_list)} free chunks "
              f"({sum(n for _, n in dr.free_list):,}B free)")
        for e in sorted(dr.entries, key=lambda e: (e.category, e.relation, e.index)):
            flag = "" if e.crc_ok else "  !! CRC MISMATCH"
            print(f"      {str(e):<34} @{e.offset:#08x} {e.size:>8,}B "
                  f"(alloc {e.alloc:,}) {COMPRESSION.get(e.compression, e.compression)}{flag}")

    for cat in hf.catalogs:
        live_mark = ""
        pkg = cat.package
        src = f"package@{pkg.offset:#x} uuid={pkg.database_uuid}" if pkg else "no package header"
        print(f"\n  catalog @ {cat.offset:#06x}  {cat.length:,}B  [{cat.framing}] {src}{live_mark}")
        print(f"      compression={cat.data.get('compressionMethod')}  "
              f"encryption={cat.data.get('encryptionSchemeId')}  "
              f"schemas={[n.get('name', '(dropped)') for n in cat.data.get('namespaces') or []]}")
        rels = relations(cat, hf_warn)
        if not rels:
            print("      (no relations - this is an earlier transaction's state)")
        for ri, r in enumerate(rels):
            attrs = r.get("attributes", [])
            nulls = r.get("nullCounts", [])
            dead = ""
            live = hf.live_catalog
            if live is not None and cat.offset == live.offset:
                ranges, note = deleted_rows(raw, hf, ri)
                if note:
                    dead = f"  !! {note}"
                elif ranges:
                    n = sum(hi - lo + 1 for lo, hi in ranges)
                    dead = f"  {n:,} deleted rows in {len(ranges):,} ranges"
            print(f"      relation {qualified(r)}  oid={r.get('oid')}  "
                  f"storage={r.get('type')}  {len(attrs)} columns{dead}")
            for i, a in enumerate(attrs):
                # The type array mixes strings and ints: ["Numeric", 18, 4,
                # "nullable"], ["Char", 8]. Joining it blind raises TypeError.
                t = "/".join(str(x) for x in (a.get("type") or []))
                nc = f"  nulls={nulls[i]}" if i < len(nulls) else ""
                print(f"          {a.get('name'):<24} {t}{nc}")

    if show_blocks:
        print(f"\n  Data Blocks ({len(hf.blocks)} objects, from the directory)")
        if not hf.blocks:
            print("      none - no Relation_DataBlock objects in the directory")
        for b in hf.blocks:
            if b.index == 0:
                print(f"      rel {b.relation} partition header  @{b.offset:#08x}  "
                      f"{b.tuple_count:,} tuples")
                continue
            print(f"      rel {b.relation} col {b.index - 1:<3} {str(b.column):<24} "
                  f"{str(b.sql_type):<12} @{b.offset:#08x}  "
                  f"{b.stored_len:>8,}B -> {b.decoded_len:>8,}B {b.compression}")
            if b.hdr:
                i = b.hdr
                strings = max(0, b.decoded_len - SMA_OFF - i.string_offset)
                print(f"          tuples={b.tuple_count:,}  "
                      f"scheme={i.scheme} {scheme_name(i.scheme)}  "
                      f"sma={i.sma_size:,} dict={i.dict_size:,} "
                      f"data={i.data_size:,} strings={strings:,}"
                      f"  ~{b.code_bits:.2f} bits/row")
            print(f"          SMA min={b.sma_min!r}  max={b.sma_max!r}")
            if b.note:
                print(f"          !! {b.note}")

    for w in hf.warnings:
        print(f"  [warn] {w}")


def verify(hf: HyperFile) -> int:
    """Cross-check against the official Hyper API, if it is installed.

    Keys on (schema, table): table names are not unique within a file.
    """
    try:
        from tableauhyperapi import HyperProcess, Connection, Telemetry
    except ImportError:
        print("  [verify] tableauhyperapi not installed - skipping oracle check")
        return 0

    ours: dict[tuple[str, str], list[str]] = {}
    for cat in hf.catalogs:
        for r in relations(cat):
            ours[(r["_schema"], r["name"])] = [a["name"] for a in r.get("attributes", [])]

    failures = 0
    seen: set[tuple[str, str]] = set()
    with HyperProcess(telemetry=Telemetry.DO_NOT_SEND_USAGE_DATA_TO_TABLEAU) as proc:
        with Connection(proc.endpoint, hf.path) as conn:
            for schema in conn.catalog.get_schema_names():
                sname = schema.name.unescaped
                for tbl in conn.catalog.get_table_names(schema):
                    tname = tbl.name.unescaped
                    key = (sname, tname)
                    seen.add(key)
                    cols = [c.name.unescaped for c in
                            conn.catalog.get_table_definition(tbl).columns]
                    if key not in ours:
                        print(f'  [verify] MISS  "{sname}"."{tname}" not found by parser')
                        failures += 1
                    elif ours[key] != cols:
                        print(f'  [verify] DIFF  "{sname}"."{tname}" '
                              f'{ours[key]} != {cols}')
                        failures += 1
                    else:
                        n = conn.execute_scalar_query(f"SELECT COUNT(*) FROM {tbl}")
                        print(f'  [verify] OK    "{sname}"."{tname}" '
                              f'{len(cols)} columns, {n:,} rows')

    for key in ours.keys() - seen:
        print(f'  [verify] EXTRA "{key[0]}"."{key[1]}" reported by parser but '
              f'not live (probably a stale transaction)')
    if not failures:
        print("  [verify] parser agrees with the Hyper API on all live relations")
    return failures


def show_rows(path: str, limit: int) -> None:
    """Reconstruct and print rows, one relation at a time.

    The order is the block's **stored** order, which is not insertion order:
    Hyper clusters equal values when it freezes a block. Treat the output as a
    multiset unless you have an ORDER BY.
    """
    d = open(path, "rb").read()
    hf = parse(path, want_blocks=True)
    cat = hf.live_catalog
    if not cat or not hf.directory:
        print("  no live catalog/directory"); return
    for ri, rel in enumerate(relations(cat, hf.warnings.append)):
        attrs = rel.get("attributes", [])
        print(f"\n  {qualified(rel)}  ({len(attrs)} columns)")
        cols, notes = [], []
        for ci, a in enumerate(attrs):
            if not column_objects(hf, ri, ci + 1):
                cols.append(None); notes.append(f"{a['name']}: no block"); continue
            vals, note = decode_relation_column(d, hf, ri, ci + 1,
                                                ColType.parse(a.get("type")))
            cols.append(vals or None)
            if note:
                notes.append(f"{a['name']}: {note}")
        n = max((len(c) for c in cols if c), default=0)
        print("    " + " | ".join(f"{a['name'][:18]:>18}" for a in attrs))
        for r in range(min(limit, n)):
            cells = []
            for c in cols:
                v = c[r] if c and r < len(c) else None
                cells.append(f"{'NULL' if v is None else v!s:>18}"[:18])
            print("    " + " | ".join(f"{c:>18}" for c in cells))
        if n > limit:
            print(f"    ... {n - limit:,} more rows (stored order, not insertion order)")
        for m in notes:
            print(f"    !! {m}")


def show_samples(path: str, limit: int) -> None:
    """Decode every Relation_Sample and check it against its own footer."""
    d = open(path, "rb").read()
    hf = parse(path)
    cat = hf.live_catalog
    if not cat or not hf.directory:
        print("  no live catalog/directory"); return
    rels = relations(cat, hf.warnings.append)
    ents = relation_samples(hf)
    if not ents:
        print("  no Relation_Sample objects - the relations are too small "
              "for the engine to have built one")
        return
    for e in ents:
        blob = read_object(d, e)
        s = parse_relation_sample(blob)
        rel = rels[e.relation] if e.relation < len(rels) else {}
        attrs = rel.get("attributes", [])
        name = qualified(rel) if rel else f"relation {e.relation}"
        if s is None:
            print(f"\n  {name}: {e} does not parse as a sample"); continue
        print(f"\n  {name}  sample of {s.block.tuple_count:,} rows "
              f"from {s.relation_rows:,}  ({len(s.block.groups)} attributes, "
              f"{len(s.block.body):,}B block)")
        types = [ColType.parse(a.get("type")) for a in attrs]
        cols, note = decode_sample(blob, types)
        if note and not cols:
            print(f"    !! {note}"); continue
        for k, g in enumerate(s.block.groups):
            sma, dic, dat, heap, scheme = s.block.regions(k)
            nm = attrs[k]["name"] if k < len(attrs) else f"attr{k}"
            print(f"      {nm[:22]:<22} {str(types[k]):<12} scheme={scheme:<3} "
                  f"sma={len(sma):,} dict={len(dic):,} data={len(dat):,} "
                  f"strings={len(heap):,}")
        bad = check_sample_moments(blob, types)
        if not s.moments:
            print(f"    footer power sums: none - a v{s.version} footer stops "
                  f"after the distinct estimates")
        elif s.sample_rows != s.block.tuple_count:
            print(f"    footer power sums: cover {s.sample_rows:,} rows the sampler "
                  f"visited, but the block keeps {s.block.tuple_count:,} - "
                  f"not recomputable")
        else:
            print("    footer power sums: "
                  + ("agree with the decoded values" if not bad
                     else "MISMATCH\n      " + "\n      ".join(bad)))
        if note:
            print(f"    !! {note}")
        print("    " + " | ".join(f"{(a['name'] if k < len(attrs) else '?')[:18]:>18}"
                                  for k, a in enumerate(attrs)))
        for r in range(min(limit, s.block.tuple_count)):
            cells = [f"{'NULL' if c[r] is None else c[r]!s:>18}"[:18] for c in cols]
            print("    " + " | ".join(f"{c:>18}" for c in cells))


def hexdump(path: str, off: int, length: int = 256) -> None:
    d = open(path, "rb").read()
    for i in range(off, min(off + length, len(d)), 16):
        chunk = d[i:i + 16]
        text = "".join(chr(c) if 32 <= c < 127 else "." for c in chunk)
        print(f"{i:08x}  {' '.join(f'{c:02x}' for c in chunk):<47}  |{text}|")


def main() -> int:
    ap = argparse.ArgumentParser(description="Reverse-engineered Tableau .hyper reader")
    ap.add_argument("files", nargs="+")
    ap.add_argument("--blocks", action="store_true",
                    help="decode every Data Block the directory addresses")
    ap.add_argument("--json", action="store_true", help="emit machine-readable output")
    ap.add_argument("--verify", action="store_true", help="cross-check against the Hyper API")
    ap.add_argument("--rows", type=int, metavar="N",
                    help="reconstruct and print the first N rows of each relation")
    ap.add_argument("--samples", type=int, metavar="N", nargs="?", const=10,
                    help="decode each relation's Relation_Sample, printing N rows")
    ap.add_argument("--dump", type=lambda s: int(s, 0), help="hexdump at this offset and exit")
    args = ap.parse_args()

    if args.rows is not None:
        for p in args.files:
            print(f"\n{'=' * 78}\n{p}\n{'=' * 78}")
            show_rows(p, args.rows)
        return 0

    if args.samples is not None:
        for p in args.files:
            print(f"\n{'=' * 78}\n{p}\n{'=' * 78}")
            show_samples(p, args.samples)
        return 0

    if args.dump is not None:
        for p in args.files:
            print(f"--- {p} ---")
            hexdump(p, args.dump)
        return 0

    rc = 0
    for p in args.files:
        try:
            hf = parse(p, want_blocks=args.blocks)
        except Exception as e:                       # keep going across a batch
            print(f"{p}: FAILED: {type(e).__name__}: {e}", file=sys.stderr)
            rc = 2
            continue
        if args.json:
            print(json.dumps(asdict(hf), indent=2, default=str))
        else:
            report(hf, args.blocks)
            if args.verify:
                rc |= 1 if verify(hf) else 0
    return rc


if __name__ == "__main__":
    sys.exit(main())