perf: serve cache-worthy entries from result.parquet in the viewer (#71)

src/tallyman_companion/buckaroo_lifecycle.py

@@ -51,6 +51,7 @@
 from tallyman_core import (
     entry_build_dir,
     entry_expanded_build_dir,
+    entry_result_expanded_build_dir,
     entry_result_path,
     entry_stat_cache_dir,
 )
@@ -484,10 +485,12 @@ def ensure_session(
     ) -> str | None:
         """Return a Buckaroo session_id for ``content_hash``, creating it if needed.
 
-        Posts the entry's ``xorq_build/`` dir to Buckaroo's ``/load_expr``
-        endpoint so the session is backed by the xorq expression (push-down
-        sort/search against the underlying backend), not by paging over the
-        materialised parquet.
+        Posts an expanded xorq build dir to Buckaroo's ``/load_expr`` endpoint so
+        the session is backed by an xorq expression (push-down sort/search
+        against the underlying backend). For a cheap entry that build is the
+        entry's recipe (``xorq_build/``); for a cache-worthy entry it's a read of
+        the materialised ``result.parquet`` (#71) so count()/paging don't re-run
+        the Aggregate/Join/Sort — push-down still holds on a parquet scan.
 
         ``project`` names the project that owns the parquet. Sessions cache
         across projects via content hash (globally unique), so a second call
@@ -526,10 +529,38 @@ def ensure_session(
             # with load_entry's diff path so the two can't drift.
             from tallyman_xorq.portable import ensure_expanded_build  # noqa: PLC0415
 
+            # #71: a cache-worthy entry (Aggregate/Join/Sort) re-runs the whole
+            # DAG on every count() and every paged window if we hand Buckaroo the
+            # raw recipe. Serve a read of the materialised result.parquet instead
+            # — push-down still holds on a parquet scan, and the per-page cost
+            # drops from the full aggregation to a parquet read. Cheap entries
+            # keep loading the recipe (their recompute ≈ a parquet read, so the
+            # indirection buys nothing). A failure in result-build prep falls
+            # back to the recipe: a cache hiccup must not blank the detail page.
+            src_build = build_dir
+            expanded_target = entry_expanded_build_dir(project, content_hash)
+            try:
+                from tallyman_xorq.result_cache import (  # noqa: PLC0415
+                    cache_worthy,
+                    ensure_result_build,
+                )
+
+                if cache_worthy(project, content_hash):
+                    src_build = ensure_result_build(project, content_hash)
+                    expanded_target = entry_result_expanded_build_dir(project, content_hash)
+            except Exception as exc:  # noqa: BLE0001 — never let cache prep break the viewer
+                log.warning(
+                    "result-read build prep failed for %s: %s; serving recipe",
+                    content_hash,
+                    exc,
+                )
+                src_build = build_dir
+                expanded_target = entry_expanded_build_dir(project, content_hash)
+
             expanded = ensure_expanded_build(
-                build_dir,
+                src_build,
                 project_dir(project),
-                entry_expanded_build_dir(project, content_hash),
+                expanded_target,
             )
             stat_cache = entry_stat_cache_dir(project, content_hash)
             stat_cache.mkdir(parents=True, exist_ok=True)

src/tallyman_core/__init__.py

@@ -29,6 +29,8 @@
     ENTRY_CACHE_NAMES,
     ENTRY_EXPANDED_BUILD_DIRNAME,
     ENTRY_MANIFEST_FILENAME,
+    ENTRY_RESULT_BUILD_DIRNAME,
+    ENTRY_RESULT_EXPANDED_BUILD_DIRNAME,
     ENTRY_RESULT_FILENAME,
     ENTRY_SCHEMA_FILENAME,
     ENTRY_STAT_CACHE_DIRNAME,
@@ -41,6 +43,8 @@
     entry_dir,
     entry_expanded_build_dir,
     entry_manifest_path,
+    entry_result_build_dir,
+    entry_result_expanded_build_dir,
     entry_result_path,
     entry_schema_path,
     entry_stat_cache_dir,
@@ -108,6 +112,8 @@
     "entry_schema_path",
     "entry_stat_cache_dir",
     "entry_expanded_build_dir",
+    "entry_result_build_dir",
+    "entry_result_expanded_build_dir",
     "ENTRY_ARTIFACT_NAMES",
     "ENTRY_CACHE_NAMES",
     "ENTRY_BUILD_DIRNAME",
@@ -116,6 +122,8 @@
     "ENTRY_SCHEMA_FILENAME",
     "ENTRY_STAT_CACHE_DIRNAME",
     "ENTRY_EXPANDED_BUILD_DIRNAME",
+    "ENTRY_RESULT_BUILD_DIRNAME",
+    "ENTRY_RESULT_EXPANDED_BUILD_DIRNAME",
     "errors_path",
     "exports_dir",
     "get_alias",

src/tallyman_core/paths.py

@@ -80,6 +80,11 @@ def buckaroo_sessions_path() -> Path:
 # first hit is honestly cold; production deletes them to recompute.
 ENTRY_STAT_CACHE_DIRNAME = ".buckaroo_stat_cache"
 ENTRY_EXPANDED_BUILD_DIRNAME = ".xorq_build_expanded"
+# A cache-worthy entry's viewer build: a deferred_read_parquet of the
+# materialised result, so count()/paging read the parquet instead of re-running
+# the Aggregate/Join/Sort recipe (#71). Portable, then expanded like the recipe.
+ENTRY_RESULT_BUILD_DIRNAME = ".xorq_result_build"
+ENTRY_RESULT_EXPANDED_BUILD_DIRNAME = ".xorq_result_build_expanded"
 
 # Canonical artifact-vs-cache partition of the per-entry names the overlay cares
 # about. The write-isolated perf overlay symlinks ENTRY_ARTIFACT_NAMES read-only
@@ -97,6 +102,8 @@ def buckaroo_sessions_path() -> Path:
 ENTRY_CACHE_NAMES = (
     ENTRY_STAT_CACHE_DIRNAME,
     ENTRY_EXPANDED_BUILD_DIRNAME,
+    ENTRY_RESULT_BUILD_DIRNAME,
+    ENTRY_RESULT_EXPANDED_BUILD_DIRNAME,
 )
 
 
@@ -204,6 +211,16 @@ def entry_expanded_build_dir(project: str, content_hash: str) -> Path:
     return entry_dir(project, content_hash) / ENTRY_EXPANDED_BUILD_DIRNAME
 
 
+def entry_result_build_dir(project: str, content_hash: str) -> Path:
+    """Portable build of the entry's result-read expression (regenerated on demand)."""
+    return entry_dir(project, content_hash) / ENTRY_RESULT_BUILD_DIRNAME
+
+
+def entry_result_expanded_build_dir(project: str, content_hash: str) -> Path:
+    """Stable expanded result-read build dir beside the entry (regenerated on demand)."""
+    return entry_dir(project, content_hash) / ENTRY_RESULT_EXPANDED_BUILD_DIRNAME
+
+
 def result_cache_dir(project: str) -> Path:
     """Root for xorq's ParquetSnapshotCache of catalog-entry results.
 

src/tallyman_xorq/result_cache.py

@@ -30,6 +30,9 @@
 
 import functools
 import re
+import shutil
+import tempfile
+import threading
 from pathlib import Path
 
 # Ops whose presence makes an expression worth caching: they require a
@@ -174,3 +177,75 @@ def ensure_result(project: str, content_hash: str) -> Path:
     if not result_path.exists():
         load_entry(project, content_hash).to_parquet(str(result_path))
     return result_path
+
+
+# Per-target locks so two concurrent loads of the same entry don't race on the
+# rmtree/build/marker sequence below. Mirrors portable._expand_lock; a bounded
+# number of entries are touched per process, so this never grows large.
+_result_build_locks_guard = threading.Lock()
+_result_build_locks: dict[str, threading.Lock] = {}
+
+
+def _result_build_lock(key: str) -> threading.Lock:
+    with _result_build_locks_guard:
+        return _result_build_locks.setdefault(key, threading.Lock())
+
+
+def ensure_result_build(project: str, content_hash: str) -> Path:
+    """Portable xorq build whose expression reads the entry's materialised result.
+
+    For a cache-worthy entry the recipe is an Aggregate/Join/Sort that re-runs
+    the full DAG on every ``count()`` and every paged ``LIMIT`` window. This
+    persists a tiny build whose expression is a ``deferred_read_parquet`` of the
+    cached result (``ensure_result`` guarantees the parquet is present first), so
+    the viewer pages over the materialised parquet instead of recomputing (#71).
+    Push-down still holds: a parquet scan pushes filters/limit/sort down too.
+
+    The build is portable (``${TALLYMAN_PROJECT_ROOT}`` placeholders), so the
+    caller expands it with ``ensure_expanded_build`` exactly like the recipe.
+
+    A sibling ``.complete`` marker, written last, gates reuse: a build
+    interrupted by a crash/OOM leaves a partial dir with no marker and is redone,
+    never served truncated. Entries are immutable and content-addressed, so the
+    build is always correct once written. The path is stable across restarts,
+    which keeps xorq's build-dir-keyed stat-cache lookups hitting.
+    """
+    import xorq.api as xo
+    from xorq.ibis_yaml.compiler import build_expr
+
+    from tallyman_core.paths import entry_result_build_dir, project_dir
+    from tallyman_xorq._git_state_guard import install_git_state_guard
+    from tallyman_xorq.portable import make_portable_inplace
+
+    target = entry_result_build_dir(project, content_hash)
+    marker = target.with_name(target.name + ".complete")
+    if marker.exists():
+        return target
+    with _result_build_lock(str(target)):
+        # Re-check under the lock: another thread may have finished building.
+        if marker.exists():
+            return target
+        # ensure_result must run before build_expr — the deferred read points at
+        # the parquet path, and the file has to exist when the viewer executes.
+        result_path = ensure_result(project, content_hash)
+        # git-provenance capture in xorq's compiler can SIGSEGV when forked from
+        # the long-lived server; make it best-effort, as build_and_persist does.
+        install_git_state_guard()
+        expr = xo.deferred_read_parquet(str(result_path))
+        if target.exists():
+            shutil.rmtree(target)
+        with tempfile.TemporaryDirectory(prefix="tallyman_result_build_") as tmp:
+            build_path = Path(build_expr(expr, builds_dir=Path(tmp)))
+            target.mkdir(parents=True)
+            for item in build_path.rglob("*"):
+                dest = target / item.relative_to(build_path)
+                if item.is_dir():
+                    dest.mkdir(parents=True, exist_ok=True)
+                else:
+                    dest.parent.mkdir(parents=True, exist_ok=True)
+                    dest.write_bytes(item.read_bytes())
+        # Rewrite project-root substrings to the portable placeholder so the
+        # caller can expand it on this host exactly like the recipe build.
+        make_portable_inplace(target, project_dir(project))
+        marker.write_text("")
+    return target

tests/test_buckaroo.py

@@ -35,6 +35,14 @@ def _code(project: str) -> str:
 """
 
 
+def _cheap_code(project: str) -> str:  # parquet read + projection → not cache-worthy
+    return f"""
+from tallyman_xorq.io import from_project
+t = from_project("orders.parquet", project={project!r})
+expr = t.select("region", "price")
+"""
+
+
 # ---------------------------------------------------------------------------
 # unit: session map only
 # ---------------------------------------------------------------------------
@@ -279,6 +287,51 @@ def fake_post(url, json=None, timeout=None):
     assert "${TALLYMAN_PROJECT_ROOT}" not in expr_yaml
 
 
+def test_unit_ensure_session_cache_worthy_serves_result_read(project: str, orders_parquet: Path, monkeypatch):
+    """A cache-worthy entry must be served to Buckaroo as a read of its
+    materialised result.parquet, not the raw recipe (#71).
+
+    The recipe is an Aggregate/Join/Sort that re-runs the full DAG on every
+    ``count()`` and every ``LIMIT 300`` page. Handing Buckaroo a build whose
+    expression is a ``deferred_read_parquet`` of the cached result turns those
+    per-page recomputes into a parquet scan, while keeping push-down.
+    """
+    from tallyman_xorq.result_cache import cache_worthy, classify_build
+
+    res = build_and_persist(project, _code(project))  # group_by.aggregate → worthy
+    assert cache_worthy(project, res.content_hash) is True
+
+    mgr = BuckarooManager()
+    mgr.bound_port = 65000
+    mgr.proc = type("FakeProc", (), {"poll": staticmethod(lambda: None)})()
+
+    captured: dict[str, Any] = {}
+
+    class FakeResponse:
+        def raise_for_status(self):
+            pass
+
+        def json(self):
+            return {"session": "s"}
+
+    def fake_post(url, json=None, timeout=None):
+        captured["json"] = json
+        return FakeResponse()
+
+    monkeypatch.setattr(mgr._client, "post", fake_post)
+    mgr.ensure_session(res.content_hash, project)
+
+    posted = Path(captured["json"]["build_dir"])
+    assert posted.is_dir(), posted
+    # The posted build serves a parquet read (cheap), not the aggregate recipe.
+    assert classify_build(posted)["worthy"] is False, classify_build(posted)
+    # The recipe itself is still expensive — we swapped it, not rewrote it.
+    recipe = entry_dir(project, res.content_hash) / "xorq_build"
+    assert classify_build(recipe)["worthy"] is True
+    # And the posted build is expanded (no placeholder), like the recipe path.
+    assert "${TALLYMAN_PROJECT_ROOT}" not in (posted / "expr.yaml").read_text()
+
+
 # ---------------------------------------------------------------------------
 # stress: tmp-dir lifecycle, concurrency
 # ---------------------------------------------------------------------------
@@ -291,8 +344,11 @@ def test_unit_ensure_session_writes_stable_expanded_dir(project: str, orders_par
     deleted by stop(); it now lives under the (immutable, content-addressed)
     entry dir so the path is identical across restarts — which is what lets the
     ParquetSnapshotCache hit. It must therefore survive stop().
+
+    Uses a cheap entry so the recipe is the served build (a cache-worthy entry
+    serves its result-read build, expanded under a different dir — see #71).
     """
-    res = build_and_persist(project, _code(project))
+    res = build_and_persist(project, _cheap_code(project))
 
     mgr = BuckarooManager()
     mgr.bound_port = 65000
@@ -327,8 +383,11 @@ def test_unit_ensure_session_heals_partial_expansion(project: str, orders_parque
     files (zero rows), or — if a subdir landed half-copied — raised
     FileExistsError out of the lock on every later load. ensure_session must key
     off a completion marker written last, and re-expand when it's absent.
+
+    Uses a cheap entry so the recipe is the served build (a cache-worthy entry
+    serves its result-read build, expanded under a different dir — see #71).
     """
-    res = build_and_persist(project, _code(project))
+    res = build_and_persist(project, _cheap_code(project))
 
     mgr = BuckarooManager()
     mgr.bound_port = 65000
@@ -412,8 +471,12 @@ def test_unit_stop_cleans_many_expanded_dirs(project: str):
 
 def test_unit_concurrent_same_hash_loads_once(project: str, orders_parquet: Path, monkeypatch):
     """N threads racing on the same content_hash must produce exactly one
-    /load_expr POST and one tmp dir — the session_lock serialises."""
-    res = build_and_persist(project, _code(project))
+    /load_expr POST and one tmp dir — the session_lock serialises.
+
+    Cheap entry so the served build is the recipe (asserted below via
+    .xorq_build_expanded); cache-worthy entries take the result-read path.
+    """
+    res = build_and_persist(project, _cheap_code(project))
 
     mgr = BuckarooManager()
     mgr.bound_port = 65000
@@ -456,9 +519,14 @@ def worker():
 
 def test_unit_concurrent_different_hashes_each_load_once(project: str, orders_parquet: Path, monkeypatch):
     """Threads racing on N distinct content_hashes each get their own
-    /load_expr POST + tmp dir; no cross-talk through the session_lock."""
+    /load_expr POST + tmp dir; no cross-talk through the session_lock.
+
+    Cheap entries so the served builds are the recipes (asserted below via
+    .xorq_build_expanded); cache-worthy entries take the result-read path.
+    """
     hashes = [
-        build_and_persist(project, _code(project) + f"\nexpr = expr.mutate(_v={i})").content_hash for i in range(5)
+        build_and_persist(project, _cheap_code(project) + f"\nexpr = expr.mutate(_v={i})").content_hash
+        for i in range(5)
     ]
 
     mgr = BuckarooManager()