python/elidedb/store.py

"""Store: a directory of Parquet tables under a transaction log.

Everything is Parquet — sensor rows, the video frame index, embeddings,
centroids. No custom byte formats. The design borrows one idea from each
system it wants to be judged against:

- warehouses (Vertica/BigQuery): columnar + statistics pruning, projection
  pushdown — Parquet row groups and column chunks give both natively.
- Spark: predicate pushdown to the scan; a query touches the minimum files,
  row groups, and columns.
- Delta/Iceberg: the table is a log fold; snapshots, time travel, atomic
  appends (log.py).
- C-Store: late materialization — video pixels are produced last, from byte
  ranges the frame-index table points at; the raw media file is never copied
  into the store.
- Kafka/streaming: time is the primary axis; every table MUST carry `ts`
  (int64 ns, sorted within a file) — that is the one schema law here.
- lakehouse: open format means other engines read the store for free;
  `Store.sql()` is DuckDB pointed at the very same files.
"""
from __future__ import annotations

import io
import json
import time
import contextlib
import fcntl
import importlib
import os
import uuid
from pathlib import Path

import numpy as np
import pyarrow as pa
import pyarrow.compute as pc
import pyarrow.parquet as pq

from .log import FileEntry, TableLog, fsync_file

ROW_GROUP_ROWS = 64 * 1024  # the amortize-vs-overfetch dial (Parquet's
                            # row-group size == SDX's chunk_target_rows)


ROW_GROUP_TARGET_BYTES = 8 * 1024 * 1024


def _row_width(schema: pa.Schema) -> int:
    """Approximate uncompressed bytes per row, for row-group sizing."""
    w = 0
    for f in schema:
        t = f.type
        try:
            if pa.types.is_fixed_size_list(t):
                w += t.list_size * (t.value_type.bit_width // 8)
            else:
                w += t.bit_width // 8
        except (ValueError, AttributeError):
            w += 32                       # strings/lists: a guess is fine
    return max(w, 1)


# THE DATABASE DOES NOT CACHE. Not yet, deliberately: a buffer pool is
# real work with real invalidation rules, and it is the LAST thing to
# build, after the layout and the pruning are right. Until then the
# honest position is that repeat reads should cost what first reads
# cost, so a benchmark number cannot be quietly borrowed from the OS.
#
# Reads therefore open store files with F_NOCACHE (macOS) / O_DIRECT-ish
# advice, which tells the kernel not to keep this file's pages in the
# unified buffer cache. That is scoped to OUR files: the system cache
# and every other process's data are untouched, which `sudo purge` can
# never say. Set ELIDEDB_CACHE=1 to opt back into the OS page cache.
_NOCACHE = os.environ.get("ELIDEDB_CACHE", "0") not in ("1", "true")
_F_NOCACHE = 48                      # <sys/fcntl.h>, Darwin


def _uncached(path):
    """Open a store file so its pages are not retained by the kernel.

    Returns a plain binary file object; pyarrow accepts any file-like,
    so this drops in wherever a path was passed. Falls back to a normal
    open anywhere the fcntl is unavailable (Linux, odd filesystems) -
    losing the no-cache property is worth strictly less than crashing,
    and the caller is told by ELIDEDB_CACHE what it asked for.
    """
    f = open(path, "rb", buffering=0)
    try:
        fcntl.fcntl(f.fileno(), _F_NOCACHE, 1)
    except Exception:
        pass
    return f


class _PF:
    """pq.ParquetFile that OWNS its file object and closes it.

    The first version handed `_uncached(path)` straight to ParquetFile
    and returned. ParquetFile does not take ownership, so every footer
    read leaked one open file object - and a leaked Python file object
    with a live buffer makes the interpreter hang at SHUTDOWN, not at
    the point of the leak. That is why an inspection script printed all
    of its output and then sat at 0% CPU for 27 minutes holding nothing
    visible: it was stuck tearing down, and `lsof` showed zero because
    the fd table was already gone.

    Every read path since the no-cache change was leaking. Closing is
    the fix; being a context manager as well means callers can be
    explicit where it matters.
    """

    __slots__ = ("_fh", "pf")

    def __init__(self, path):
        self._fh = _uncached(path) if _NOCACHE else None
        self.pf = pq.ParquetFile(self._fh if self._fh is not None
                                 else str(path))

    def __getattr__(self, k):
        return getattr(self.pf, k)

    def __enter__(self):
        return self

    def __exit__(self, *a):
        self.close()

    def close(self):
        try:
            self.pf.close()
        except Exception:
            pass
        if self._fh is not None:
            try:
                self._fh.close()
            except Exception:
                pass
            self._fh = None

    def __del__(self):
        self.close()


def _pf(path):
    """pq.ParquetFile honouring the no-cache policy, and closing."""
    return _PF(path)


def _read(path, **kw):
    """pq.read_table honouring the no-cache policy.

    Reads through a _PF and COPIES the result, so the file object cannot
    outlive this call. `with _uncached(path) as fh: pq.read_table(fh)`
    looks safe and is not - pyarrow keeps a reference to the handle for
    lazy access, so the `with` closes nothing and the leaked handle hangs
    the interpreter at shutdown. That is the same failure the _pf fix
    addressed, surviving in the other reader: a two-table scan would sit
    at 0% CPU forever while a single-table one exited fine.
    """
    if not _NOCACHE:
        return pq.read_table(str(path), **kw)
    f = _PF(path)
    try:
        cols = kw.get("columns")
        filters = kw.get("filters")
        t = f.pf.read(columns=cols)
        if filters is not None:
            t = t.filter(filters)
        return t.combine_chunks()      # materialise before the handle dies
    finally:
        f.close()


def _top(md, c: int) -> str:
    """Top-level column name for row-group column index `c`.

    Parquet flattens nested types to LEAVES, and md.schema.names is the
    leaf list: a fixed-size-list column named `vector` appears there as
    `element`, with path_in_schema `vector.list.element`. Matching a
    projection against the leaf name therefore never matched a vector
    column, so its bytes were never charged - 8.7 MB per row group on
    frame_vectors, silently absent from every elision number in a store
    that is 87% vectors. Metrics that undercount are worse than no
    metrics: they make the headline claim look better than it is.
    """
    return md.row_group(0).column(c).path_in_schema.split(".")[0]


def _col_index(md, name: str):
    """Row-group column index for a TOP-LEVEL column name, or None."""
    for c in range(md.num_columns):
        if _top(md, c) == name:
            return c
    return None


def _max_end(table: pa.Table) -> int:
    """Latest interval END in this table: max(t1), falling back to
    max(ts) for a table that has no t1 (point rows end where they
    start)."""
    col = "t1" if "t1" in table.column_names else "ts"
    try:
        v = pc.max(table.column(col)).as_py()
    except Exception:
        return 0
    return int(v or 0)


def _zone(table: pa.Table, columns) -> dict:
    """File-level min/max for the columns a file is clustered on.

    Recorded in the commit log, so a value predicate can drop whole
    files from JSON already in memory - before any Parquet footer is
    opened. Strings are kept as strings and numbers as numbers; the
    comparison at read time is the caller's, and it must compare like
    with like or it will prune wrongly rather than merely badly.
    """
    z = {}
    for c in columns or ():
        if c not in table.column_names or c == "ts":
            continue      # ts already has its own dedicated pair
        col = table.column(c)
        try:
            lo, hi = pc.min(col).as_py(), pc.max(col).as_py()
        except pa.ArrowNotImplementedError:
            continue      # lists, structs: no order, no zone map
        if lo is not None and hi is not None:
            z[c] = [lo, hi]
    return z


def write_parquet(table: pa.Table, path):
    """One writer for every file in the store. `ts` gets
    DELTA_BINARY_PACKED — timestamps are near-arithmetic, so delta encoding
    beats generic zstd ~3x on that column (the Gorilla/TSDB observation);
    string columns keep dictionary encoding; everything rides zstd.

    Row groups are sized by ROW WIDTH to a byte target, not a fixed row
    count: at 64k rows a 1152-d float32 vector table packed ~295 MB into
    ONE group, so time-range pruning inside a file could skip nothing —
    the elision law applied to layout. A narrow sensor table still gets
    tens of thousands of rows per group; a vector table gets ~1.8k, and a
    2 s window read touches one group instead of the whole file."""
    rows = min(ROW_GROUP_ROWS,
               max(4096, ROW_GROUP_TARGET_BYTES // _row_width(table.schema)))
    dict_cols = [f.name for f in table.schema
                 if pa.types.is_string(f.type) or pa.types.is_large_string(f.type)]
    pq.write_table(table, path, row_group_size=rows,
                   compression="zstd",
                   use_dictionary=dict_cols,
                   # The PAGE INDEX (per-page min/max + offsets) is what
                   # lets a reader skip pages INSIDE a surviving row group
                   # — the layer that makes a columnar file behave like an
                   # index for selective reads. pyarrow omits it by
                   # default, so every file written before 2026-07-28 can
                   # only prune to row-group granularity. It costs a small
                   # constant in the footer and is read only when a query
                   # has a predicate that can use it.
                   write_page_index=True,
                   column_encoding={"ts": "DELTA_BINARY_PACKED"})
    fsync_file(path)  # durability: data reaches disk BEFORE the commit that
                      # references it — the write-ahead ordering rule


class QueryStats:
    """Bytes accounting: the elision number, file-granular.

    Log-level pruning is exact (a pruned file contributes 0 bytes). Inside a
    surviving file, Parquet's own row-group pruning + column projection cut
    further; we report the surviving files' bytes as the upper bound actually
    mapped, plus rows returned."""

    def __init__(self):
        self.corpus_bytes = 0
        self.files_total = 0
        self.files_touched = 0
        self.bytes_touched = 0
        self.rows_returned = 0
        self.wall_ms = 0.0

    @property
    def elided_pct(self):
        if not self.corpus_bytes:
            return 0.0
        b = min(self.bytes_touched, self.corpus_bytes)
        return 100.0 * (self.corpus_bytes - b) / self.corpus_bytes

    def __repr__(self):
        return (f"<{self.files_touched}/{self.files_total} files, "
                f"{self.bytes_touched:,} B touched of {self.corpus_bytes:,} B "
                f"({self.elided_pct:.3f}% elided), {self.rows_returned:,} rows, "
                f"{self.wall_ms:.1f} ms>")


class Table:
    def __init__(self, store: "Store", name: str):
        self.store = store
        self.name = name
        self.dir = store.dir / "tables" / name
        self.log = TableLog(self.dir)

    def state(self, version=None):
        return self.log.read_state(version)

    def history(self):
        return self.log.history()

    # ---- write path -------------------------------------------------------
    def _validate(self, table: pa.Table, evolve: bool):
        """Consistency guarantees enforced at the door: ts present, int64,
        never null; and the schema must match the table's existing schema
        (same names ⇒ same types). evolve=True permits ADDING columns —
        widening reads promote missing columns to null — but a type change
        for an existing name is always an error, never a silent coercion."""
        if "ts" not in table.column_names:
            raise ValueError(f"table '{self.name}': a `ts` int64-ns column is "
                             "required — timestamps are the one schema law")
        ts = table.column("ts")
        if ts.type != pa.int64():
            raise ValueError("`ts` must be int64 nanoseconds since epoch")
        if ts.null_count:
            raise ValueError(f"table '{self.name}': `ts` contains "
                             f"{ts.null_count} null(s) — every row must be "
                             "timestamped")
        st = self.state()
        if st.files:
            # compare against the LATEST file: after additive evolution the
            # newest schema is the table's current contract
            existing = _pf(
                self.dir / st.files[-1].path).schema_arrow
            have = {f.name: f.type for f in existing}
            new = {f.name: f.type for f in table.schema}
            for name, typ in new.items():
                if name in have and have[name] != typ:
                    raise ValueError(
                        f"table '{self.name}': column '{name}' is "
                        f"{have[name]} but incoming batch has {typ} — "
                        "type changes are never implicit")
            added = set(new) - set(have)
            missing = set(have) - set(new)
            if (added or missing) and not evolve:
                raise ValueError(
                    f"table '{self.name}': schema differs (new columns "
                    f"{sorted(added)}, absent columns {sorted(missing)}). "
                    "Pass evolve=True to allow additive evolution.")
        return st

    def _sorted(self, table: pa.Table) -> pa.Table:
        order = pc.sort_indices(table.column("ts"))
        if not pc.all(pc.equal(order, pa.array(range(len(table))))).as_py():
            table = table.take(order)  # ts-sorted files ⇒ tight zone maps
        return table

    # ---- layout policy ----------------------------------------------------
    def set_layout(self, cluster_by: str, *, sort_by=None,
                   min_group_rows=256) -> int:
        """Declare the table's CLUSTER KEY once, for every writer.

        Physical layout was a per-call argument, and the `events` table
        is what that costs. write_once wrote it grouped by `kind`;
        build_teacher rewrote it with plain replace(), ts-sorted. Same
        table, two writers, and the store ended up holding the
        unclustered one - so a lookup on `kind`, the verb, the single
        most natural predicate in the corpus, had to open 100% of row
        groups. Nothing errored. The table was simply no longer an
        index, and only a footer audit would ever have said so.

        Clustering is a property of a TABLE, the way a clustered index
        is, not a decision each INSERT gets to re-make. Declared here,
        it lands in the log - so it is versioned, travels with time
        travel, and any writer that goes through append/replace honours
        it without knowing it exists.
        """
        st = self.state()
        # A pure METADATA commit: no file added, none removed. The
        # declaration is a new version, so it is auditable and time
        # travel still lands on the layout that was in force then, but
        # it does not rewrite a byte. Existing files keep whatever
        # layout they were written with until something rebuilds them -
        # declaring an index does not reorganise the table.
        return self.log.commit(
            op="layout", kind=st.kind, schema=st.schema,
            meta={"layout": {
                "cluster_by": cluster_by,
                "sort_by": list(sort_by or [cluster_by, "ts"]),
                "min_group_rows": int(min_group_rows)}})

    def layout(self) -> dict | None:
        return self.state().meta.get("layout")

    def append(self, table: pa.Table, *, kind="timeseries", meta=None,
               evolve=False) -> int:
        lay = self.layout()
        if lay:
            return self.append_grouped(
                table, lay["cluster_by"], kind=kind, meta=meta,
                evolve=evolve, sort_by=lay["sort_by"],
                min_group_rows=lay["min_group_rows"])
        return self.append_batches([table], kind=kind, meta=meta,
                                   evolve=evolve)

    def append_batches(self, batches, *, kind="timeseries", meta=None,
                       evolve=False) -> int:
        """Write one file per batch, commit ONCE — a multi-gigabyte load is
        a single atomic transaction with bounded memory."""
        self.dir.mkdir(parents=True, exist_ok=True)
        adds, schema = [], None
        for batch in batches:
            if len(batch) == 0:
                continue
            self._validate(batch, evolve)
            batch = self._sorted(batch)
            schema = batch.schema
            fname = f"part-{uuid.uuid4().hex[:12]}.parquet"
            path = self.dir / fname
            write_parquet(batch, path)
            tsv = batch.column("ts")
            adds.append(FileEntry(fname, len(batch), path.stat().st_size,
                                  tsv[0].as_py(), tsv[-1].as_py(),
                                  {}, _max_end(batch)))
        if not adds:
            raise ValueError("nothing to append (all batches empty)")
        return self.log.commit(op="append", kind=kind,
                               schema=str(schema), add=adds, meta=meta)

    def append_grouped(self, table: pa.Table, group_col: str, *,
                       kind="timeseries", meta=None, evolve=False,
                       sort_by=None, replace=False,
                       min_group_rows: int = 1) -> int:
        """Write with ROW GROUPS ALIGNED TO A LOGICAL UNIT.

        The default writer sizes row groups by bytes, which is right for a
        sensor stream and wrong for anything whose query unit is an
        object. The frames table is the proof: 39,026 rows landed in ONE
        row group of 981 KB, so a query for a single episode decompressed
        every frame in the store and layer-2 pruning measured 0.45%.

        Here each distinct `group_col` value becomes its own row group, so
        "give me episode X" touches exactly one. The trade is footer size
        — 1,122 groups x 9 columns is ~10k column-chunk entries of
        metadata, and the footer is read on EVERY query — which is the
        random-access-vs-metadata tension made explicit rather than
        inherited from a default.

        `sort_by` is recorded in the footer as Parquet sorting_columns, so
        a reader can know the file is clustered without trusting us.
        """
        self.dir.mkdir(parents=True, exist_ok=True)
        if len(table) == 0:
            raise ValueError("nothing to write")
        self._validate(table, evolve)
        sort_by = sort_by or [group_col, "ts"]
        sort_by = [c for c in sort_by if c in table.column_names]
        table = table.take(pc.sort_indices(
            table, sort_keys=[(c, "ascending") for c in sort_by]))

        g = table.column(group_col).to_pylist()
        raw, start = [], 0
        for i in range(1, len(g) + 1):
            if i == len(g) or g[i] != g[start]:
                raw.append((start, i - start))
                start = i
        # MERGE ADJACENT GROUPS UP TO A ROW FLOOR. One group per distinct
        # value sounds maximally prunable and is a trap: metadata costs a
        # fixed amount per group PER COLUMN, so tiny groups invert the
        # ratio. Measured on the labels table at one-group-per-value -
        # 406 groups over 7,573 rows - the footer reached 286,796 bytes
        # against 283,957 bytes of actual column data. The footer was
        # LARGER THAN THE DATA, and it is read on every query, so the
        # "index" cost more to consult than the table cost to scan.
        #
        # The SORT is what makes statistics prunable; group size is an
        # independent dial. Merging keeps values contiguous, so each
        # group still covers a narrow min/max range, at a fraction of the
        # metadata.
        bounds = []
        if min_group_rows <= 1:
            bounds = raw
        else:
            off = cur = 0
            for s, n in raw:
                cur += n
                if cur >= min_group_rows:
                    bounds.append((off, cur))
                    off, cur = s + n, 0
            if cur:
                bounds.append((off, cur))

        fname = f"part-{uuid.uuid4().hex[:12]}.parquet"
        path = self.dir / fname
        idx = {c: table.column_names.index(c) for c in sort_by}
        sc = [pq.SortingColumn(idx[c]) for c in sort_by]
        dict_cols = [f.name for f in table.schema
                     if pa.types.is_string(f.type)
                     or pa.types.is_large_string(f.type)]
        w = pq.ParquetWriter(path, table.schema, compression="zstd",
                             use_dictionary=dict_cols,
                             write_page_index=True,
                             sorting_columns=sc,
                             column_encoding={"ts": "DELTA_BINARY_PACKED"})
        try:
            for off, n in bounds:
                w.write_table(table.slice(off, n), row_group_size=n)
        finally:
            w.close()
        fsync_file(path)

        st = self.state()
        tsv = table.column("ts")
        add = [FileEntry(fname, len(table), path.stat().st_size,
                         tsv[0].as_py(), tsv[-1].as_py(),
                         _zone(table, sort_by), _max_end(table))]
        return self.log.commit(
            op="replace" if replace else "append", kind=kind,
            schema=str(table.schema), add=add,
            remove=[f.path for f in st.files] if replace else [],
            meta=dict(meta or {}, row_groups=len(bounds),
                      grouped_by=group_col, sorted_by=sort_by))

    def replace(self, table: pa.Table, *, kind="timeseries", meta=None,
                evolve=False) -> int:
        """REBUILD IN PLACE: these rows become the table, in one commit.

        Derived tables — events, answers, anything recomputed from the
        frames — need this and `append` is wrong for them. Recomputing
        the teacher's events with `append` took the table from 8,263 to
        16,526 rows, with `agent` at exactly 2x the demo count, and left
        every demo holding BOTH the old and the new typing of the same
        transition. Nothing errored; consumers that read a demo's kinds
        as a set just started seeing contradictions.

        Old files are removed in the same transaction that adds the new
        ones, so readers see one or the other and never the union, and
        the previous version stays addressable through the log.

        A rebuild MUST NOT quietly restore the table to ts-sorted: if a
        cluster key is declared, this rewrites through the grouped
        writer. That is exactly how `events` lost its layout - a
        recompute through replace() undid write_once's grouping."""
        lay = self.layout()
        if lay:
            return self.append_grouped(
                table, lay["cluster_by"], kind=kind, meta=meta,
                evolve=evolve, sort_by=lay["sort_by"], replace=True,
                min_group_rows=lay["min_group_rows"])
        st = self.state()
        prev = [f.path for f in st.files]
        self.dir.mkdir(parents=True, exist_ok=True)
        if len(table) == 0:
            raise ValueError("replace() with an empty table would drop "
                             "the table; use delete_range to truncate")
        self._validate(table, evolve)
        table = self._sorted(table)
        fname = f"part-{uuid.uuid4().hex[:12]}.parquet"
        path = self.dir / fname
        write_parquet(table, path)
        tsv = table.column("ts")
        add = [FileEntry(fname, len(table), path.stat().st_size,
                         tsv[0].as_py(), tsv[-1].as_py(),
                         {}, _max_end(table))]
        return self.log.commit(op="replace", kind=kind,
                               schema=str(table.schema), add=add,
                               remove=prev,
                               meta=dict(meta or {},
                                         rows_before=st.rows,
                                         files_replaced=len(prev)))

    def compact(self, target_rows_per_file: int = 8_000_000) -> dict:
        """OPTIMIZE: rewrite the active file set into few large, ts-sorted,
        delta-encoded files — one atomic replace-commit. Fixes the many-
        small-files tax that every append-only log accumulates, and applies
        the current encodings to data written before them."""
        st = self.state()
        if not st.files:
            return {"files_before": 0, "files_after": 0}
        data = self.scan()
        before_bytes = st.bytes
        from .log import FileEntry as FE
        adds = []
        for lo in range(0, len(data), target_rows_per_file):
            part = data.slice(lo, target_rows_per_file)
            fname = f"part-{uuid.uuid4().hex[:12]}.parquet"
            path = self.dir / fname
            write_parquet(part, path)
            tsv = part.column("ts")
            adds.append(FE(fname, len(part), path.stat().st_size,
                           tsv[0].as_py(), tsv[-1].as_py()))
        self.log.commit(op="compact", kind=st.kind, schema=str(data.schema),
                        add=adds, remove=[f.path for f in st.files],
                        meta={"files_before": len(st.files),
                              "bytes_before": before_bytes})
        after = sum(a.bytes for a in adds)
        return {"files_before": len(st.files), "files_after": len(adds),
                "bytes_before": before_bytes, "bytes_after": after,
                "ratio": round(before_bytes / max(after, 1), 2)}

    def delete_range(self, t0: int, t1: int) -> dict:
        """Delete rows with ts in [t0, t1] — the 'scrub that run' operation
        (bad takes, PII, retention). Files fully inside the range are just
        dropped; overlapping files are rewritten without the range; files
        outside are untouched. One atomic commit; prior versions still see
        the data (time travel is the audit trail) until their files are
        garbage-collected."""
        import pyarrow.compute as pc
        from .log import FileEntry as FE
        st = self.state()
        removes, adds, dropped = [], [], 0
        for f in st.files:
            if not f.overlaps(t0, t1):
                continue  # untouched
            removes.append(f.path)
            if t0 <= f.min_ts and f.max_ts <= t1:
                dropped += f.rows
                continue  # fully covered: no rewrite needed
            t = _read(self.dir / f.path)
            keep = t.filter(pc.or_(pc.less(t.column("ts"), t0),
                                   pc.greater(t.column("ts"), t1)))
            dropped += len(t) - len(keep)
            if len(keep):
                fname = f"part-{uuid.uuid4().hex[:12]}.parquet"
                write_parquet(keep, self.dir / fname)
                tsv = keep.column("ts")
                adds.append(FE(fname, len(keep),
                               (self.dir / fname).stat().st_size,
                               tsv[0].as_py(), tsv[-1].as_py()))
        if not removes:
            return {"rows_deleted": 0}
        self.log.commit(op="delete", kind=st.kind, schema=st.schema,
                        add=adds, remove=removes,
                        meta={"deleted_range": [t0, t1],
                              "rows_deleted": dropped})
        return {"rows_deleted": dropped, "files_rewritten": len(adds),
                "files_removed": len(removes)}

    # ---- secondary indexes (B+ tree over any numeric column) --------------
    def create_index(self, column: str, order: int = 256) -> dict:
        """Build an immutable, bulk-loaded B+ tree over `column` for the
        CURRENT version (BPT1 — same bytes the C++ engine reads). Zone maps
        prune nothing on unsorted columns; this re-sorts (value → row
        location) once, so point/range predicates touch only the row groups
        that actually contain hits. Rebuild after appends (`create_index`
        again) — the artifact is version-suffixed like any derived state."""
        from . import bptree
        st = self.state()
        keys, vals = [], []
        for fi, f in enumerate(st.files):
            col = _read(self.dir / f.path, columns=[column]) \
                .column(column).to_numpy(zero_copy_only=False)
            keys.append(bptree.encode_key(col))
            vals.append((np.uint64(fi) << np.uint64(40)) |
                        np.arange(len(col), dtype=np.uint64))
        img = bptree.build(np.concatenate(keys) if keys else
                           np.array([], np.int64),
                           np.concatenate(vals) if vals else
                           np.array([], np.uint64), order=order)
        ixdir = self.dir / "_index"
        ixdir.mkdir(exist_ok=True)
        # A secondary index is a DERIVED sidecar, not table data: building it
        # does NOT advance the log (that would be a version with identical
        # data). The artifact is keyed to the version it was built for; a
        # reader accepts it only while that version's FILE SET still matches
        # the current one — appends invalidate it, so rebuild after appends.
        path = ixdir / f"{column}.v{st.version}.bpt"
        tmp = ixdir / f".{column}.tmp"
        tmp.write_bytes(img)
        fsync_file(tmp)
        tmp.replace(path)
        return {"column": column, "keys": int(sum(len(k) for k in keys)),
                "bytes": len(img), "version": st.version}

    def _open_index(self, column: str, st):
        from . import bptree
        ixdir = self.dir / "_index"
        if not ixdir.is_dir():
            return None
        current = {f.path for f in st.files}
        for p in sorted(ixdir.glob(f"{column}.v*.bpt"), reverse=True):
            v = int(p.stem.split(".v")[-1])
            if {f.path for f in self.state(v).files} == current:
                return bptree.Reader.open(p), p.stat().st_size
        return None

    def where(self, column: str, op: str, value, value2=None, columns=None,
              stats: QueryStats | None = None) -> pa.Table:
        """Predicate pushdown on a NON-time column. With a B+ index: descend,
        map hits to (file, row group), read ONLY those row groups, then take
        the exact rows. Without one: zone-map scan + filter, honestly counted
        as such. ops: ==, >=, <=, between."""
        from . import bptree
        stats = stats if stats is not None else QueryStats()
        st = self.state()
        stats.files_total += len(st.files)
        stats.corpus_bytes += st.bytes
        NEG_INF, POS_INF = -(1 << 63), (1 << 63) - 1  # full i64 range:
        # float encodings legitimately occupy the far ends of int64
        lo, hi = {"==": (value, value),
                  ">=": (value, None), "<=": (None, value),
                  "between": (value, value2)}[op]
        ix = self._open_index(column, st)
        if ix is None:
            # fallback: full scan with an honest bill
            t = self.scan(stats=stats)
            arr = t.column(column)
            m = None
            if lo is not None:
                m = pc.greater_equal(arr, lo)
            if hi is not None:
                c = pc.less_equal(arr, hi)
                m = c if m is None else pc.and_(m, c)
            out = t.filter(m)
            stats.rows_returned += len(out)
            return out
        reader, ix_bytes = ix
        stats.bytes_touched += ix_bytes  # the index read is real I/O too
        locs = reader.range(
            bptree.encode_scalar(lo) if lo is not None else NEG_INF,
            bptree.encode_scalar(hi) if hi is not None else POS_INF)
        locs = np.sort(np.asarray(locs, dtype=np.uint64))
        file_ids = (locs >> np.uint64(40)).astype(np.int64)
        rows = (locs & np.uint64((1 << 40) - 1)).astype(np.int64)
        parts = []
        want_cols = None if columns is None else \
            (["ts", *columns] if "ts" not in columns else list(columns))
        for fi in np.unique(file_ids):
            fmask = file_ids == fi
            frows = rows[fmask]
            pf = _pf(self.dir / st.files[fi].path)
            md = pf.metadata
            # metadata-driven row->group mapping: group sizes are whatever
            # the writer chose (now width-adaptive), so boundaries come
            # from the footer, never from an assumed constant
            rg_start = np.cumsum(
                [0] + [md.row_group(g).num_rows
                       for g in range(md.num_row_groups)])
            g_of = np.searchsorted(rg_start, frows, side="right") - 1
            groups = np.unique(g_of).tolist()
            for g in groups:
                for c in range(md.row_group(g).num_columns):
                    if want_cols is None or _top(md, c) in want_cols:
                        stats.bytes_touched += \
                            md.row_group(g).column(c).total_compressed_size
            tbl = pf.read_row_groups(groups, columns=want_cols)
            # map absolute rows -> positions inside the concatenated groups
            base = np.cumsum([0] + [md.row_group(g).num_rows for g in groups])
            gsel = {g: k for k, g in enumerate(groups)}
            gpos = np.array([gsel[g] for g in g_of])
            local = frows - rg_start[g_of] + base[gpos]
            parts.append(tbl.take(pa.array(np.sort(local))))
            stats.files_touched += 1
        out = pa.concat_tables(parts) if parts else \
            self.scan(0, -1, columns=columns)  # empty, right schema
        stats.rows_returned += len(out)
        return out

    def files(self, version=None) -> list[str]:
        """Absolute paths of this snapshot's Parquet files — the open-format
        contract: hand these to ANY engine (DuckDB, Spark, Polars, a
        distributed dataframe library) and it reads the table with zero
        export and zero ElideDB code."""
        return [str(self.dir / f.path) for f in self.state(version).files]

    # ---- read path --------------------------------------------------------
    def scan_values(self, column: str, values, columns=None,
                    version=None, stats: QueryStats | None = None):
        """VALUE PREDICATE PUSHED INTO THE READER, not applied after it.

        Reading the whole table and calling .filter() afterwards is what
        makes a store behave like a filesystem: correct answer, no
        pruning. Measured on the labels table before this existed - a
        lookup for one name touched 502 KB of 711 KB, 71% of the table,
        to return 26 episodes.

        Here the predicate goes to the Parquet reader, so row groups
        whose min/max for `column` cannot contain any requested value are
        never decompressed. It works because build_index sorts labels by
        (kind, value) and gives each value its own row group - sorted
        layout is what turns statistics into an index.

        bytes_touched is charged the same way: footer plus only the row
        groups whose statistics overlap the requested values.
        """
        st = self.state(version)
        stats = stats if stats is not None else QueryStats()
        stats.files_total += len(st.files)
        stats.corpus_bytes += st.bytes
        # Values keep their OWN type. Stringifying them first, as this
        # did, silently breaks numeric columns: "51" < "7" lexically, so
        # a lookup for object 7 would drop the group holding it. A
        # comparison against Parquet statistics has to be in the
        # column's order, not in string order.
        want = sorted(set(values))
        if columns is not None:
            columns = list(dict.fromkeys([*columns, column, "ts"]))
        lo, hi = want[0], want[-1]
        parts = []
        for f in st.files:
            # LAYER 0: the commit log's zone map, already in memory. A
            # file that provably holds nothing in [lo, hi] is dropped
            # here, before its footer is even opened.
            if not f.may_contain(column, lo, hi):
                continue
            p = self.dir / f.path
            pf = _pf(p)
            md = pf.metadata
            ci = _col_index(md, column)
            if ci is None:
                continue
            groups, touched = [], 0
            for g in range(md.num_row_groups):
                rg = md.row_group(g)
                s = rg.column(ci).statistics
                # LAYER 1: row-group statistics. min/max is a RANGE
                # test, so a group is skipped only when every requested
                # value falls outside [min, max].
                if s is not None and s.min is not None:
                    try:
                        if hi < s.min or lo > s.max:
                            continue
                    except TypeError:
                        pass          # mixed types: cannot prove empty
                groups.append(g)
                for c in range(rg.num_columns):
                    if columns is None or _top(md, c) in columns:
                        touched += rg.column(c).total_compressed_size
            stats.files_touched += 1
            stats.bytes_touched += md.serialized_size + touched
            if not groups:
                continue
            tb = pf.read_row_groups(groups, columns=columns)
            parts.append(tb.filter(pc.field(column).isin(want)))
        if not parts:
            return pa.table({"ts": pa.array([], pa.int64())}), stats
        out = pa.concat_tables(parts, promote_options="permissive")
        stats.rows_returned += len(out)
        return out, stats

    def scan(self, t0=None, t1=None, columns=None, version=None,
             stats: QueryStats | None = None) -> pa.Table:
        st = self.state(version)
        stats = stats if stats is not None else QueryStats()
        stats.files_total += len(st.files)
        stats.corpus_bytes += st.bytes
        if columns is not None and "ts" not in columns:
            columns = ["ts", *columns]  # ts always rides along: it is the
                                        # sort key and the alignment axis
        # layer 1: log-level file pruning (zone maps in the commit entries)
        # INTERVAL overlap, not start-point containment. See
        # FileEntry.overlaps: comparing t0 against max(ts) drops any row
        # whose interval began before the window and had not ended.
        files = [f for f in st.files if f.overlaps(t0, t1)]
        stats.files_touched += len(files)
        # layer 2 accounting: row-group zone maps from the Parquet footer.
        # A surviving file is charged its footer + only the row groups whose
        # ts min/max overlap the window — which is exactly what the reader
        # below will materialize. Same math as warehouse skip-indexes.
        for f in files:
            pf = _pf(self.dir / f.path)
            md = pf.metadata
            footer_bytes = md.serialized_size
            ts_idx = _col_index(md, "ts") or 0
            # SAME INTERVAL FIX, one layer down. A row group is skipped
            # only if no interval in it can reach the window: compare t0
            # against max(t1) where the table has a t1, and against
            # max(ts) only when it does not.
            e_idx = _col_index(md, "t1")
            touched = 0
            for rg in range(md.num_row_groups):
                g = md.row_group(rg)
                st_ts = g.column(ts_idx).statistics
                st_e = (g.column(e_idx).statistics
                        if e_idx is not None else st_ts)
                end = (st_e.max if st_e is not None and st_e.max is not None
                       else (st_ts.max if st_ts is not None else None))
                if end is not None and t0 is not None and end < t0:
                    continue
                if st_ts is not None and t1 is not None and st_ts.min > t1:
                    continue
                for c in range(g.num_columns):
                    col = g.column(c)
                    if (columns is None or _top(md, c) in columns
                            or c == ts_idx):
                        touched += col.total_compressed_size
            stats.bytes_touched += footer_bytes + touched
        if not files:
            empty = pa.schema([("ts", pa.int64())])
            return pa.table({"ts": pa.array([], pa.int64())}).cast(empty)
        # layer 2: Parquet row-group pruning + projection pushdown
        # INTERVAL OVERLAP, not start containment. `ts >= t0` asks
        # "did it START inside the window", which is a different
        # question and drops every row still in progress when the window
        # opens. Overlap is (row.t1 >= t0) AND (row.ts <= t1); with no
        # t1 column a row is a point and the two coincide.
        has_end = "t1" in (pq.ParquetFile(
            _uncached(self.dir / files[0].path) if _NOCACHE
            else str(self.dir / files[0].path)).schema_arrow.names)
        end_f = pc.field("t1") if has_end else pc.field("ts")
        filt = None
        if t0 is not None:
            filt = end_f >= t0
        if t1 is not None:
            c = pc.field("ts") <= t1
            filt = c if filt is None else filt & c
        if has_end and columns is not None and "t1" not in columns:
            columns = [*columns, "t1"]      # the predicate needs it
        parts = [_read(self.dir / f.path, columns=columns,
                               filters=filt) for f in files]
        out = pa.concat_tables(parts, promote_options="permissive")
        if len(parts) > 1:  # files may interleave in time across streams
            out = out.take(pc.sort_indices(out.column("ts")))
        stats.rows_returned += len(out)
        return out


class Store:
    FORMAT = "elidedb/2"

    def __init__(self, path: str | Path):
        self.dir = Path(path)
        meta_path = self.dir / "_store.json"
        if not meta_path.exists():
            raise FileNotFoundError(
                f"{path} is not a store (no _store.json) — Store.create() it")
        self.meta = json.loads(meta_path.read_text())

    # ---- lifecycle --------------------------------------------------------
    @staticmethod
    def create(path: str | Path, name: str) -> "Store":
        p = Path(path)
        (p / "tables").mkdir(parents=True, exist_ok=True)
        meta = {"format": Store.FORMAT, "name": name,
                "created_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())}
        (p / "_store.json").write_text(json.dumps(meta, indent=1))
        return Store(p)

    @staticmethod
    def open(path: str | Path) -> "Store":
        return Store(path)

    @property
    def name(self):
        return self.meta["name"]

    def snapshot(self) -> dict:
        """Pin every table's current version in one call. Pass the result as
        `version=` to window/aligned/sql/scan for a CONSISTENT multi-table
        read: no writer that commits after this call can skew your view.
        (Per-table commits are serializable on their own log — the same
        isolation model as Delta and Iceberg; the pin extends it across
        tables for readers.)"""
        return {name: self.table(name).state().version
                for name in self.tables()}

    @staticmethod
    def _ver(version, name):
        if isinstance(version, dict):
            return version.get(name)
        return version

    def vacuum(self, retain_versions: int = 3, dry_run: bool = False) -> dict:
        """Garbage-collect files no snapshot within the retention window can
        reach: parquet parts removed by compaction/delete, and managed media
        no retained frame-index version references. Time travel remains
        intact for the last `retain_versions` versions of every table;
        earlier versions become unreadable — that is the explicit trade this
        command makes, and why it is manual."""
        freed = 0
        removed = []
        media_refs: set[str] = set()
        for name in self.tables():
            tab = self.table(name)
            versions = tab.log.versions()
            keep_versions = versions[-retain_versions:] if versions else []
            referenced = set()
            for v in keep_versions:
                st = tab.state(v)
                referenced |= {f.path for f in st.files}
                if st.kind == "frame_index":
                    tbl = tab.scan(version=v)
                    if "source" in tbl.column_names:
                        media_refs |= {s for s in
                                       set(tbl.column("source").to_pylist())
                                       if s.startswith("@")}
            for f in tab.dir.glob("part-*.parquet"):
                if f.name not in referenced:
                    freed += f.stat().st_size
                    removed.append(str(f.relative_to(self.dir)))
                    if not dry_run:
                        f.unlink()
        media = self.dir / "media"
        if media.is_dir():
            for m in media.iterdir():
                if f"@media/{m.name}" not in media_refs:
                    freed += m.stat().st_size
                    removed.append(f"media/{m.name}")
                    if not dry_run:
                        m.unlink()
        return {"files_removed": len(removed), "bytes_freed": freed,
                "dry_run": dry_run, "removed": removed[:10]}

    def tables(self) -> list[str]:
        root = self.dir / "tables"
        if not root.is_dir():
            return []
        return sorted(p.name for p in root.iterdir() if (p / "_log").is_dir())

    def table(self, name: str) -> Table:
        return Table(self, name)

    def drop_caches(self) -> dict:
        """Forget everything this store has memoised. Nothing else.

        The point is scope. `sudo purge` empties the whole machine's
        unified buffer cache - every other process's working set with
        it - which makes a benchmark both unrepeatable and rude. What a
        database should be able to say is "drop MY cache", so this
        clears only what ElideDB itself holds: the derived matrices
        memoised per (store, version) in context.py and cracked.py.

        File pages are handled separately and earlier: store reads open
        with F_NOCACHE, so the kernel is never asked to retain them in
        the first place. Nothing to evict beats evicting.

        Loaded model weights are NOT dropped - they are not a cache of
        the data, they are the program.
        """
        out = {}
        for mod, names in ((".context", ("_CTX_CACHE", "_SPACE_CACHE")),
                           (".cracked", ("_CACHE",))):
            try:
                m = importlib.import_module(mod, __package__)
            except Exception:
                continue
            for n in names:
                d = getattr(m, n, None)
                if isinstance(d, dict):
                    out[f"{mod.lstrip('.')}.{n}"] = len(d)
                    d.clear()
        out["file_pages"] = "not cached (F_NOCACHE)" if _NOCACHE else "OS"
        return out

    @contextlib.contextmanager
    def measure(self):
        """Charge every read inside this block to one QueryStats.

        `scan` and `scan_values` already account honestly, but they
        account PER CALL, and a retrieval query fans out across a dozen
        channel tables through code that never threads a stats object
        through. Threading one through every channel would touch every
        caller for a number none of them care about; wrapping the two
        entry points for the duration of a block gets the same figure
        with the accounting living in exactly one place.

            with db.measure() as st:
                search_set(db, "open the drawer")
            st.bytes_touched, st.elided_pct

        corpus_bytes is the whole store, not the sum of the tables that
        happened to be touched - the elision claim is against everything
        that could have been read, or it means nothing.
        """
        stats = QueryStats()
        stats.corpus_bytes = sum(self.table(t).state().bytes
                                 for t in self.tables())
        scan, values = Table.scan, Table.scan_values

        def scan_m(self_, *a, stats=None, **kw):
            local = QueryStats()
            out = scan(self_, *a, stats=local, **kw)
            _fold(stats or _NULL, local)
            _fold(stats_outer, local)
            return out

        def values_m(self_, *a, stats=None, **kw):
            local = QueryStats()
            out = values(self_, *a, stats=local, **kw)
            _fold(stats or _NULL, local)
            _fold(stats_outer, local)
            return out

        stats_outer = stats
        Table.scan, Table.scan_values = scan_m, values_m
        try:
            yield stats
        finally:
            Table.scan, Table.scan_values = scan, values


    def describe(self) -> list[dict]:
        out = []
        for name in self.tables():
            st = self.table(name).state()
            out.append({"table": name, "kind": st.kind, "version": st.version,
                        "rows": st.rows, "bytes": st.bytes,
                        "min_ts": st.min_ts, "max_ts": st.max_ts,
                        "files": len(st.files), "meta": st.meta})
        return out

    # ---- friendly ingest --------------------------------------------------
    def ingest_rows(self, table: str, data, ts_column="ts", ts_unit="auto",
                    meta=None, evolve=False) -> int:
        """Append rows from a pandas DataFrame / dict of arrays / pyarrow
        Table / CSV / Parquet path.

        Friendly on purpose: `ts_column` may be a datetime column, an ISO-8601
        string column, or epoch numbers in s/ms/us/ns — `ts_unit="auto"`
        detects the epoch unit by magnitude (an explicit unit always wins)."""
        import os as _os

        import pandas as pd
        if isinstance(data, (str, Path)):
            p = str(data)
            if p.endswith((".parquet", ".pq")):
                data = pd.read_parquet(p)
            elif _os.path.getsize(p) > 128 * 1024 * 1024:
                # memory-bounded load: stream the CSV in chunks, one file per
                # chunk, ONE atomic commit for the whole load
                def gen():
                    for chunk in pd.read_csv(p, chunksize=2_000_000):
                        yield self._normalize_ts(chunk, ts_column, ts_unit)
                return self.table(table).append_batches(gen(), meta=meta)
            else:
                data = pd.read_csv(p)
        if isinstance(data, dict):
            data = pd.DataFrame(data)
        if isinstance(data, pd.DataFrame):
            data = self._normalize_ts(data, ts_column, ts_unit)
        return self.table(table).append(data, meta=meta, evolve=evolve)

    @staticmethod
    def _normalize_ts(df, ts_column, ts_unit) -> pa.Table:
        import pandas as pd
        if ts_column not in df.columns:
            raise ValueError(
                f"no column '{ts_column}' — available: "
                f"{list(df.columns)} (pass ts_column=...)")
        df = df.rename(columns={ts_column: "ts"}).copy()
        col = df["ts"]
        if pd.api.types.is_datetime64_any_dtype(col):
            df["ts"] = col.astype("int64")  # datetime64 is already ns
        elif col.dtype == object or pd.api.types.is_string_dtype(col):
            df["ts"] = pd.to_datetime(col).astype("int64")  # ISO strings
        else:
            if ts_unit == "auto":
                # epoch magnitude: seconds ~1e9, ms ~1e12, us ~1e15, ns ~1e18
                m = float(pd.Series(col).abs().median())
                ts_unit = ("s" if m < 1e11 else "ms" if m < 1e14
                           else "us" if m < 1e17 else "ns")
            mult = {"ns": 1, "us": 1_000, "ms": 1_000_000,
                    "s": 1_000_000_000}[ts_unit]
            df["ts"] = (col.astype("float64") * mult).round().astype("int64")
        return pa.Table.from_pandas(df, preserve_index=False)

    def _media_dest(self, src: Path) -> Path:
        import hashlib
        h = hashlib.sha1(str(src.resolve()).encode()).hexdigest()[:8]
        media = self.dir / "media"
        media.mkdir(exist_ok=True)
        return media / f"{src.stem}-{h}{src.suffix}"

    def ingest_video(self, table: str, video_path, timestamps_ns=None,
                     stream=None, meta=None, copy=True, transcode=None,
                     gop_s: float = 1.0, crf: int = 26) -> int:
        """Index a video file: packet scan → frame_index Parquet rows.

        copy=True (default): the media file is copied into the store's
        `media/` directory first, so the store directory IS the complete,
        portable database. copy=False indexes the file in place.

        transcode="hevc"|"h264": re-encode the managed copy as a compressed
        elementary stream (≈10-25x smaller than MJPEG at like quality) with a
        forced keyframe every `gop_s` seconds. Random access becomes
        GOP-granular instead of frame-exact — `gop_s` IS the seekability-vs-
        compression dial, chosen per table at ingest, and decode reads
        exactly one GOP span per window."""
        import shutil
        import subprocess

        from .fftools import find
        from .video import scan_video_packets
        src = Path(video_path)
        if transcode:
            assert transcode in ("hevc", "h264")
            if timestamps_ns is None:  # take pts from the source container
                probe = scan_video_packets(src)
                timestamps_ns = probe["ts"].to_pylist()
            n_in = len(timestamps_ns)
            span_s = max((timestamps_ns[-1] - timestamps_ns[0]) / 1e9, 0.1)
            fps = max((n_in - 1) / span_s, 1.0)
            g = max(1, round(gop_s * fps))
            dest = self._media_dest(src).with_suffix(f".{transcode}")
            if not dest.exists():
                enc = "libx265" if transcode == "hevc" else "libx264"
                subprocess.run(
                    [find("ffmpeg"), "-v", "error", "-y", "-i", str(src),
                     "-c:v", enc, "-preset", "fast", "-crf", str(crf),
                     "-g", str(g), "-keyint_min", str(g), "-an",
                     "-f", transcode, str(dest)], check=True)
            scanned_path, source_ref = dest, f"@media/{dest.name}"
        elif copy:
            dest = self._media_dest(src)
            if not dest.exists():
                shutil.copy2(src, dest)
            scanned_path, source_ref = dest, f"@media/{dest.name}"
        else:
            scanned_path = src
            source_ref = str(src.resolve())
        rows = scan_video_packets(scanned_path, timestamps_ns)
        n = len(rows["ts"])
        rows["source"] = pa.array([source_ref] * n)
        rows["stream"] = [stream or src.stem] * n
        t = pa.table(rows)
        return self.table(table).append(
            t, kind="frame_index",
            meta={"source": source_ref, "original": str(src.resolve()),
                  **({"transcode": transcode, "gop_s": gop_s, "crf": crf}
                     if transcode else {}),
                  **(meta or {})})

    def adopt_media(self, table: str = "frames", verbose=True) -> dict:
        """Make the store standalone: copy every externally-referenced media
        file into `media/` and rewrite the frame index to store-relative
        paths. One replace-commit per call — old index versions still resolve
        (the external files are not deleted)."""
        import shutil
        import uuid as _uuid
        from .log import FileEntry
        tab = self.table(table)
        st = tab.state()
        if st.kind != "frame_index":
            raise ValueError(f"{table} is not a frame_index table")
        t = tab.scan()
        srcs = t.column("source").to_pylist()
        external = sorted({s for s in srcs if not s.startswith("@")})
        if not external:
            return {"adopted": 0, "bytes": 0}
        mapping, copied = {}, 0
        for s in external:
            p = Path(s)
            if not p.exists():
                raise FileNotFoundError(f"referenced media missing: {s}")
            dest = self._media_dest(p)
            if not dest.exists():
                shutil.copy2(p, dest)
            copied += dest.stat().st_size
            mapping[s] = f"@media/{dest.name}"
            if verbose:
                print(f"  adopted {p.name} -> media/{dest.name}")
        new_src = pa.array([mapping.get(s, s) for s in srcs])
        t = t.set_column(t.column_names.index("source"), "source", new_src)
        fname = f"part-{_uuid.uuid4().hex[:12]}.parquet"
        path = self.dir / "tables" / table / fname
        write_parquet(t, path)
        tsv = t.column("ts").to_numpy()
        tab.log.commit(op="adopt-media", kind="frame_index",
                       schema=str(t.schema),
                       add=[FileEntry(fname, len(t), path.stat().st_size,
                                      int(tsv.min()), int(tsv.max()))],
                       remove=[f.path for f in st.files],
                       meta={"media_files": len(external)})
        return {"adopted": len(external), "bytes": copied}

    # ---- queries ----------------------------------------------------------
    def window(self, t0: int, t1: int, tables=None, columns=None,
               version=None):
        """The multimodal read: every requested table filtered to [t0, t1].
        frame_index tables come back as FrameSet (lazy byte-range decode)."""
        from .video import FrameSet
        stats = QueryStats()
        start = time.perf_counter()
        out = {}
        names = tables
        if names is None:
            # Default to the DATA tables. Index artifacts (embeddings,
            # centroids, frame_vectors, context, ...) are timestamped too, so
            # they would otherwise be dragged into every window read and drag
            # thousands of 1152-d vectors with them. Ask for them by name and
            # you still get them.
            names = [n for n in self.tables()
                     if self.table(n).state(self._ver(version, n)).kind
                     not in ("embeddings", "centroids")]
        for name in names:
            tab = self.table(name)
            v = self._ver(version, name)
            st = tab.state(v)
            cols = columns.get(name) if isinstance(columns, dict) else columns
            data = tab.scan(t0, t1, columns=cols, version=v, stats=stats)
            out[name] = FrameSet(self, name, data) if st.kind == "frame_index" \
                else data
        stats.wall_ms = (time.perf_counter() - start) * 1e3
        return out, stats

    def aligned(self, t0, t1, rate_hz, tables=None, interp="nearest",
                version=None, edge_guard_s: float = 1.0):
        """Query-time alignment: resample numeric columns of the requested
        timeseries tables onto one [t0, t1] timeline at rate_hz."""
        timeline = np.arange(t0, t1 + 1, int(1e9 / rate_hz), dtype=np.int64)
        guard = int(edge_guard_s * 1e9)  # neighbors just outside the window
                                         # make edge interpolation exact
        out = {"timeline_ns": timeline}
        stats = QueryStats()
        for name in (tables or self.tables()):
            tab = self.table(name)
            v = self._ver(version, name)
            if tab.state(v).kind != "timeseries":
                continue
            data = tab.scan(t0 - guard, t1 + guard, version=v, stats=stats)
            if len(data) == 0:
                continue
            ts = data.column("ts").to_numpy()
            cols = {}
            for cname in data.column_names:
                if cname == "ts":
                    continue
                arr = data.column(cname)
                if not pa.types.is_floating(arr.type) and \
                   not pa.types.is_integer(arr.type):
                    continue
                v = arr.to_numpy().astype(np.float64)
                if interp == "linear":
                    cols[cname] = np.interp(timeline, ts, v)
                else:  # nearest
                    idx = np.searchsorted(ts, timeline)
                    idx = np.clip(idx, 0, len(ts) - 1)
                    prev = np.clip(idx - 1, 0, len(ts) - 1)
                    use_prev = (timeline - ts[prev]) <= (ts[idx] - timeline)
                    cols[cname] = v[np.where(use_prev, prev, idx)]
            out[name] = cols
        return out, stats

    def sql(self, query: str, version=None):
        """DuckDB over the store's own Parquet files — the lakehouse dividend:
        because the format is open, a whole second engine comes for free.
        Table names in the query = store table names."""
        import duckdb
        con = duckdb.connect()
        for name in self.tables():
            st = self.table(name).state(self._ver(version, name))
            files = [str(self.dir / "tables" / name / f.path) for f in st.files]
            if files:
                quoted = ", ".join(f"'{f}'" for f in files)
                con.execute(
                    f'CREATE VIEW "{name}" AS SELECT * FROM '
                    f"read_parquet([{quoted}])")
        return con.execute(query).fetchdf()

    # ---- semantic layer (see embeddings.py) --------------------------------
    def embed_windows(self, frame_table="frames", window_s=2.0,
                      frames_per_window=2, model=None, batch=16):
        from .embeddings import embed_windows
        return embed_windows(self, frame_table, window_s, frames_per_window,
                             model, batch)

    def search(self, text: str, k=10, nprobe=3, merge=True, t0=None, t1=None,
               streams=None, method="auto", neg_weight=0.5, min_score=None,
               percentile=None, rerank=False, rerank_top=12,
               rerank_alpha=0.7):
        """Compositional text search. `text` supports AND / NOT / -term;
        `min_score`/`percentile` add a precision floor. See embeddings.search."""
        from .embeddings import search
        return search(self, text, k=k, nprobe=nprobe, merge=merge, t0=t0,
                      t1=t1, streams=streams, method=method,
                      neg_weight=neg_weight, min_score=min_score,
                      percentile=percentile, rerank=rerank,
                      rerank_top=rerank_top, rerank_alpha=rerank_alpha)

    def search_text(self, text: str, k=10, nprobe=3, **kw):
        from .embeddings import search_text
        return search_text(self, text, k=k, nprobe=nprobe, **kw)

    def search_clip(self, stream: str, t0: int, t1: int, k=10, nprobe=3, **kw):
        """Query-by-example. When the store carries a V-JEPA clip index
        the neighbor space is the WORLD MODEL's (video-native, motion-
        structured, no text anywhere); otherwise appearance windows.
        Measured (bridge4h): V-JEPA beats appearance on action-class
        neighbor purity for 'open' (0.51 vs 0.40) and ties elsewhere."""
        try:
            from .embeddings import _vec_table
            import numpy as np
            tbl, vecs = _vec_table(self, "vjepa_vectors")
            ss = tbl.column("stream").to_pylist()
            sa = [int(v) for v in tbl.column("ts").to_pylist()]
            sb = [int(v) for v in tbl.column("t1").to_pylist()]
            mid = (t0 + t1) // 2
            qi = next((i for i in range(len(ss))
                       if ss[i] == stream and sa[i] <= mid <= sb[i]), None)
            if qi is not None:
                sc = vecs @ np.asarray(vecs[qi])
                sc[qi] = -9
                order = np.argsort(-sc)[:k]
                hits = [{"stream": ss[i], "t0": sa[i], "t1": sb[i],
                         "score": float(sc[i])} for i in order]
                return hits, {"method": "vjepa-qbe", "k": k}
        except Exception:
            pass
        from .embeddings import search_clip
        return search_clip(self, stream, t0, t1, k=k, nprobe=nprobe, **kw)

    # ---- context retrieval -------------------------------------------------
    def index_context(self, window_s=2.0, stride_s=0.5, label_fraction=1.0,
                      prune=True, epochs=300, verbose=True, model=None,
                      frame_stride=1, prompt="scene"):
        """Build the context index end to end.

        frames -> per-frame vectors -> VLM captions on `label_fraction` of
        windows -> caption-LSA space -> temporal tower -> cellular turnover
        -> materialised `context` table.

        Cost is dominated by the image encoder, so the knobs that matter are:

          model="fast"      3.3x faster encoder, same 1152-d space
          frame_stride=N    embed every Nth frame (5 Hz video rarely needs all)
          label_fraction<1  caption only part of the corpus; the tower covers
                            the rest
          prompt=           "scene" or "manipulation" — the caption IS the
                            index, so it has to use the words a user would

        Measured: decode 2.7 ms/frame, encode 90.3 ms/frame (quality) or
        27.7 ms/frame (fast). Embedding a large corpus is a batch job measured
        in hours; nothing here hides that.
        """
        from . import context as C
        out = {"frame_vectors": C.embed_frames(self, verbose=verbose,
                                               model=model,
                                               stride=frame_stride)}
        windows = C.plan_windows(self, window_s, stride_s)
        if label_fraction < 1.0:
            # Label a TIME PREFIX, not a random sample: the realistic shape of
            # this problem is "we captioned what we had, then more footage
            # arrived", and a random sample would quietly hand the tower
            # neighbours of every held-out window.
            n = max(int(len(windows) * label_fraction), 16)
            windows = sorted(windows, key=lambda w: w[1])[:n]
        out["captions"] = C.caption_windows(self, windows, verbose=verbose,
                                            prompt=prompt)
        _, _, out["train"] = C.train_context(self, window_s, stride_s,
                                             epochs=epochs, verbose=verbose)
        if prune:
            _, rec = C.prune_context(self, verbose=verbose)
            out["prune"] = rec.get("selected")
        out["build"] = C.build_context(self, verbose=verbose)
        return out

    def search_context(self, text: str, k=8, pool=48, deep=0, t0=None,
                       t1=None, streams=None, rerank=False, verify="async",
                       **_legacy):
        """THE search: any query, action or not, on any store.

        Union recall over every tier the store has (appearance embeddings,
        caption words, caption-LSA vectors) proposes candidates; a VLM
        reading each clip's start/end frames verifies WHAT IS HAPPENING;
        verdicts are cached into the store so hot queries get cheap.
        `deep=N` (or rerank=True) re-judges the top N with the larger VLM
        over 4 ordered frames. Legacy RRF-only search remains at
        elidedb.context.search for stores where a model-free path matters."""
        from .verified import search_verified
        if rerank and not deep:
            deep = 6
        if deep and verify == "async":
            verify = "sync"          # deep judging is an explicit wait
        return search_verified(self, text, k=k, pool=pool, deep=deep,
                               t0=t0, t1=t1, streams=streams, verify=verify)

    def search_verified(self, text: str, k=8, pool=48, deep=0):
        """Any query, action or not: union recall proposes, a VLM shown
        frames IN TIME ORDER disposes, verdicts are cached into the store.
        The only path that can enforce 'the green object is the one being
        moved' or '...and close it'. See elidedb.verified."""
        from .verified import search_verified
        return search_verified(self, text, k=k, pool=pool, deep=deep)

    def search_sharp(self, text: str, k=10, shortlist=48):
        """Text search at teacher quality, student price: the student ranks
        every window (~1 ms), the teacher re-scores only the shortlist, and
        every teacher vector is cached into the store — quality accumulates
        where users query (database cracking). See elidedb.cracked."""
        from .cracked import search_sharp
        return search_sharp(self, text, k=k, shortlist=shortlist)

    def explain(self, t0: int, t1: int, stream=None):
        """The teacher's own description of what happens in a window."""
        from .context import explain
        return explain(self, t0, t1, stream=stream)


_NULL = None


def _fold(dst, src):
    """Accumulate one QueryStats into another.

    corpus_bytes takes a MAX, not a sum: it is a denominator, and adding
    denominators across calls would inflate it until the elision figure
    became meaningless. The measure() block seeds it with the whole
    store, which is the largest and the correct one; a caller's own
    stats keeps whatever per-table denominator it had.
    """
    if dst is None:
        return
    dst.files_total += src.files_total
    dst.files_touched += src.files_touched
    dst.bytes_touched += src.bytes_touched
    dst.rows_returned += src.rows_returned
    dst.corpus_bytes = max(dst.corpus_bytes, src.corpus_bytes)