Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 17 additions & 12 deletions src/tallyman_companion/buckaroo_lifecycle.py
Original file line number Diff line number Diff line change
Expand Up @@ -50,11 +50,10 @@

from tallyman_core import (
entry_build_dir,
entry_expanded_build_dir,
entry_result_path,
entry_stat_cache_dir,
)
from tallyman_core.paths import artifacts_dir, project_dir
from tallyman_core.paths import artifacts_dir


def _entry_parquet_exists(project: str, content_hash: str) -> bool:
Expand Down Expand Up @@ -484,10 +483,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
Expand Down Expand Up @@ -540,13 +541,17 @@ def ensure_session(
# tmp path makes every stat-cache lookup a miss even when the cache
# is fully populated on disk. Marker-gated for crash safety. Shared
# with load_entry's diff path so the two can't drift.
from tallyman_xorq.portable import ensure_expanded_build # noqa: PLC0415

expanded = ensure_expanded_build(
build_dir,
project_dir(project),
entry_expanded_build_dir(project, content_hash),
# #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. ensure_viewer_expanded_build serves a read of the
# materialised result.parquet instead (push-down still holds on a
# parquet scan), or the recipe for a cheap entry / on any cache
# hiccup. Shared with the Tier-B perf proxy so the two can't drift.
from tallyman_xorq.result_cache import ( # noqa: PLC0415
ensure_viewer_expanded_build,
)

expanded = ensure_viewer_expanded_build(project, content_hash)
stat_cache = entry_stat_cache_dir(project, content_hash)
stat_cache.mkdir(parents=True, exist_ok=True)
payload: dict = {
Expand Down
8 changes: 8 additions & 0 deletions src/tallyman_core/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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,
Expand Down Expand Up @@ -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",
Expand All @@ -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",
Expand Down
17 changes: 17 additions & 0 deletions src/tallyman_core/paths.py
Original file line number Diff line number Diff line change
Expand Up @@ -86,6 +86,11 @@ def buckaroo_sessions_path() -> Path:
# it is a regenerable cache, not an immutable artifact — the overlay omits it
# rather than symlinking it.
ENTRY_RESULT_FILENAME = "result.parquet"
# 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
Expand All @@ -103,6 +108,8 @@ def buckaroo_sessions_path() -> Path:
ENTRY_STAT_CACHE_DIRNAME,
ENTRY_EXPANDED_BUILD_DIRNAME,
ENTRY_RESULT_FILENAME,
ENTRY_RESULT_BUILD_DIRNAME,
ENTRY_RESULT_EXPANDED_BUILD_DIRNAME,
)


Expand Down Expand Up @@ -210,6 +217,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 compute_cache_dir(project: str) -> Path:
"""Per-project xorq compute cache, redirected off the global ``~/.cache/xorq``.

Expand Down
117 changes: 117 additions & 0 deletions src/tallyman_xorq/result_cache.py
Original file line number Diff line number Diff line change
Expand Up @@ -35,13 +35,18 @@
import functools
import logging
import re
import shutil
import sys
import tempfile
import threading
import time
from pathlib import Path

# Perf instrumentation rides a dedicated child namespace so it can be dialed up
# independently of the rest of tallyman's logging (#60), via TALLYMAN_LOG_LEVEL.
perf_log = logging.getLogger("tallyman.perf")
# #71 viewer result-read build prep logs its fall-back-to-recipe here.
log = logging.getLogger("tallyman.result_cache")

# Ops whose presence makes an expression worth caching: they require a
# shuffle / sort / full materialisation rather than a streaming row-wise pass.
Expand Down Expand Up @@ -341,3 +346,115 @@ def ensure_result(project: str, content_hash: str) -> Path:
else:
cached_result_expr(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


def ensure_viewer_expanded_build(project: str, content_hash: str) -> Path:
"""Expanded build dir to back a viewer session for an entry (#71).

Single source of truth for the recipe-vs-result-read choice the viewer
makes. For a cache-worthy entry (Aggregate/Join/Sort) it returns the
expanded ``deferred_read_parquet`` build of the materialised result, so
``count()``/paging read the parquet instead of re-running the DAG; for a
cheap entry it returns the expanded recipe (recompute ≈ a parquet read, so
the indirection buys nothing). On any failure in result-build prep it falls
back to the recipe — a cache hiccup must not blank the detail page.

``ensure_session`` (companion → real Buckaroo ``/load_expr``) and the Tier-B
perf proxy both call this so the path they exercise can't drift.
"""
from tallyman_core.paths import ( # noqa: PLC0415
entry_build_dir,
entry_expanded_build_dir,
entry_result_expanded_build_dir,
project_dir,
)
from tallyman_xorq.portable import ensure_expanded_build # noqa: PLC0415

src_build = entry_build_dir(project, content_hash)
expanded_target = entry_expanded_build_dir(project, content_hash)
try:
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: BLE001 — never let cache prep break the viewer
log.warning(
"result-read build prep failed for %s: %s; serving recipe",
content_hash,
exc,
)
src_build = entry_build_dir(project, content_hash)
expanded_target = entry_expanded_build_dir(project, content_hash)

return ensure_expanded_build(src_build, project_dir(project), expanded_target)
80 changes: 74 additions & 6 deletions tests/test_buckaroo.py
Original file line number Diff line number Diff line change
Expand Up @@ -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")
"""


def _chain_parent_code(project: str) -> str: # Aggregate → expensive → baked snapshot
return f"""
from tallyman_xorq.io import from_project
Expand Down Expand Up @@ -295,6 +303,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()


def test_unit_ensure_session_self_heals_evicted_snapshot_on_cold_cache(project, orders_parquet, monkeypatch):
"""ensure_session must materialise an entry's baked snapshot before POSTing
to ``/load_expr``, so the buckaroo grid isn't empty on a cold compute cache.
Expand Down Expand Up @@ -366,8 +419,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
Expand Down Expand Up @@ -402,8 +458,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
Expand Down Expand Up @@ -487,8 +546,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
Expand Down Expand Up @@ -531,9 +594,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()
Expand Down
Loading
Loading