DEV-1587: per-engine in-memory query result cache

CLAUDE.md

@@ -127,6 +127,21 @@ poetry run ruff check slayer/ tests/
 
 - **Row-Level Security / forced column filter** (DEV-1578): `SlayerQueryEngine(storage, *, policy=SessionPolicy(...))` (and the local-engine `SlayerClient(storage=..., policy=...)`) silently scopes every query to a tenant. `SessionPolicy(data_filters=[ColumnFilterRule(column=..., value=...)])` lives in `slayer/core/policy.py` (both models frozen + `extra="forbid"`; `SessionPolicy.version` is `Literal[1]` so an unknown schema version fails closed; `data_filters` is a **tuple** and `ColumnFilterRule.value` coerces lists→tuples so the policy is genuinely immutable). `value` shape implies the operator: scalar→`=`, non-empty list→`IN` (empty list rejected). `on_unapplicable: "block"|"pass"` (default `block`) governs a table that **confirms it lacks** the column — per-rule; a table whose presence **cannot be confirmed** fails closed regardless (security control). The rewrite is a pure sqlglot transform in `slayer/sql/session_policy.py::apply_session_policy`, applied in `_execute_pipeline` right after `generator.generate(...)` (so `dry_run`/`explain`/execute/profiling all see identical SQL) and in `get_column_types` (degrades to `{}` on policy failure). It wraps every **physical** table ref — `FROM t` → `FROM (SELECT * FROM t WHERE col = val) AS t` (alias preserved, predicate unqualified, values via `exp.convert` so injection-safe). Physical-vs-CTE is **scope-aware** (`sqlglot.optimizer.scope.traverse_scope`): a CTE/derived ref is skipped, but a physical table sharing a CTE's name (e.g. inside that CTE's body) is still wrapped; chained CTEs are not failed. Non-SELECT roots fail closed. Column presence is probed via `SlayerQueryEngine._column_present` using the shared `_safe_get_columns` (moved to the dependency-free leaf `slayer/engine/introspect_utils.py` to avoid the `ingestion`→`query_engine` cycle; `ingestion`/`schema_drift` import it from there, `ingestion` re-exports for back-compat); case-insensitive match, schema = AST qualifier else `datasource.schema_name`, only confirmed `True`/`False` cached (a `None` is re-probed so transient failures self-heal). Cross-catalog refs (three-part `catalog.schema.table` whose catalog ≠ the connection's own `datasource.database`) can't be confirmed by the schema-only probe → fail closed; catalog-aware introspection is Phase 2. Engine-global + local-engine only — `SlayerClient(policy=...)` with `storage=None` (HTTP mode) raises. `JoinFilterRule`, per-model/per-datasource scope, REST/MCP server-side policy, BigQuery live-verification are Phase 2. New `ForcedFilterError(SlayerError)` (attrs `table`/`column`/`rule_name`). See [docs/concepts/row-level-security.md](docs/concepts/row-level-security.md).
 
+## Query cache (DEV-1587)
+
+- **Opt-in, per-engine, in-memory result cache** modelled on Cube's in-memory cache. **Python API only** — no REST / MCP / CLI / Flight / pg-facade / `SlayerClient` surface. Lives in `slayer/engine/cache.py` (`QueryCache`, `CacheConfig`, `RefreshResult`, `RefreshError`, `RefreshKeyValue`, `_CacheEntry` — all Pydantic, no dataclasses). The cache is local to a `SlayerQueryEngine` instance, so two engines with different connection settings (or different RLS `policy`) keep separate caches.
+- **Construction**: `SlayerQueryEngine(storage, cache_config=CacheConfig(ttl_seconds=300, refresh_keys=[("public.orders", "MAX(updated_at)")]))`. `cache_config` defaults to empty `CacheConfig()`. The `cache_config` **setter clears the cache**. `engine.cache_size` (property), `engine.clear_cache()`.
+- **Execution flag**: `execute(query, cache=True)` / `execute_sync(query, cache=True, data_source=...)` for every input shape (`SlayerQuery` / dict / multi-stage list / run-by-name str). Ignored (no caching, no error) when `dry_run` / `explain` is set. `execute_sync` gained BOTH `cache` and `data_source` (closing the pre-existing sync `data_source` gap).
+- **Cache key** = `sha256(final_sql + "|" + connection_string + "|" + runtime_fingerprint)`. Variables are substituted into the SQL before keying (different vars ⇒ different keys). Datasource identity is the SQL-client fingerprint (`_sql_client_cache_key`), not the bare name. Resolve→enrich→SQL-gen runs on every cached call (DB-free); a hit skips only DB execute + decode.
+- **Staleness**: (1) **TTL** — lazy on read; `age > ttl_seconds` ⇒ treated as a miss and re-executed; injectable `clock` (`time.monotonic`) for testing. (2) **Refresh keys** — Cube-style `(physical_table, select_expression)` scanned only by `engine.refresh()`. Applicable keys for an entry = configured keys whose table is referenced by the entry's SQL. Baseline captured at write time, scanned **before** the data query (so cached data ≥ baseline). The user's `select_expression` is verbatim (NOT auto-wrapped in `MAX(...)`); `MAX(updated_at)` misses in-place updates/backfills, a `COUNT(*)`-bearing expr catches deletes.
+- **Table detection**: parse the final SQL with sqlglot, subtract CTE / derived-table aliases (SLayer wraps in CTEs), normalize each remaining `exp.Table` to `(catalog, db, name)` with dialect folding (quoted preserved, unquoted folded per dialect), intersect with configured `refresh_keys` tables. **Matching rule**: config parts unspecified are wildcards (`"orders"` matches `orders`/`public.orders`/`db.public.orders`; `"public.orders"` matches only `db=public`). Refresh-key scan SQL: `SELECT (<expr0>) AS "slayer_rk_0", ... FROM <table>` (one batched scan per table).
+- **`refresh()`** snapshots entries, collates one batched scan per `(datasource, table)`, then per entry: TTL-expired ⇒ re-exec (`expired_refreshed`); a scan-failed applicable table ⇒ kept `unchanged`; any applicable value moved ⇒ re-exec (`refreshed`); else `unchanged`. Re-exec **re-prepares from the entry's original input** through the full `execute` normalization (run-by-name picks up `source_queries` edits; lists re-topo-sort), re-keys if the SQL changed, resets `created_at` (TTL reset), and the commit is **identity-guarded** (only writes if the live entry is still the snapshotted object — a concurrent evict/clear/newer execute is never clobbered or resurrected). Continue-on-failure: scan / re-exec errors become `RefreshError` and keep the stale entry.
+- **Write-time contract**: a baseline-scan failure on a cache miss **propagates** out of `execute(cache=True)` (nothing stored); the same failure at `refresh()` time is continue-on-error instead.
+- **Concurrency**: the `asyncio.Lock` guards only in-memory dict ops (get/put/delete/snapshot/commit); every DB await happens outside the lock. Concurrent identical misses both execute (last-writer-wins, no single-flight). Deep-copy on store and on hit-return so callers can't mutate the cached response. Unbounded (no LRU) — manage via `evict` / `clear_cache`.
+- **Management**: `await engine.evict(query, variables=..., data_source=...) -> bool` (+ `evict_sync`) recomputes the key DB-free and deletes one entry (never constructs a client); `engine.clear_cache()`; `engine.cache_size`; `await engine.refresh() -> RefreshResult` (+ `refresh_sync`).
+- **Engine refactor**: `execute` dispatch extracted into `_normalize_input` (+ `_normalize_by_name`); the DB-free resolve→enrich→SQL-gen→attributes→ds-key portion extracted into `_prepare_pipeline -> _Prepared` (no client instantiation); `_execute_pipeline` = prepare + dry_run/explain/cache-hook/execute. Shared by execute, `evict` (key only), and `refresh` re-exec.
+- See [docs/concepts/query-cache.md](docs/concepts/query-cache.md).
+
 ## Async Architecture
 
 - **Engine is async-first**: `SlayerQueryEngine.execute()` is `async`. Use `execute_sync()` for CLI/notebooks/scripts.

docs/concepts/query-cache.md

@@ -0,0 +1,133 @@
+# Query Cache
+
+SLayer has an optional, in-memory, per-query result cache **local to a single
+`SlayerQueryEngine` instance**. It is opt-in per call via `cache=True`, modelled
+on [Cube's in-memory cache](https://cube.dev/docs/product/caching). This is a
+**Python-API-only** feature — there is no REST / MCP / CLI / Flight / pg-facade
+surface, and no `SlayerClient` plumbing.
+
+Two engines with different connection settings keep separate caches, so a cached
+result is never served across datasource identities.
+
+> **See it run:** the [Query Cache notebook](../examples/12_query_cache/query_cache_nb.ipynb)
+> walks through miss → hit, staleness, `refresh()`, TTL, and eviction on a tiny live database.
+
+## Enabling the cache
+
+Construct the engine with a `CacheConfig`, then pass `cache=True` per call:
+
+```python
+from slayer.engine.cache import CacheConfig
+from slayer.engine.query_engine import SlayerQueryEngine
+
+engine = SlayerQueryEngine(
+    storage,
+    cache_config=CacheConfig(
+        ttl_seconds=300,
+        refresh_keys=[("public.orders", "MAX(updated_at)")],
+    ),
+)
+
+# Any input shape works — SlayerQuery / dict, a multi-stage list, or a
+# run-by-name string. Each funnels into one final SQL that is cached.
+resp = await engine.execute({"source_model": "orders",
+                             "measures": [{"formula": "amount:sum"}]},
+                            cache=True)
+```
+
+`cache_config` defaults to an empty `CacheConfig()` when not supplied.
+Reassigning `engine.cache_config = CacheConfig(...)` **clears the cache**.
+
+`cache=True` is ignored (no caching, no error) when `dry_run` or `explain` is
+set.
+
+### `CacheConfig`
+
+| Field | Meaning |
+| --- | --- |
+| `ttl_seconds: Optional[float]` | Wall-clock age bound. `None` (default) means no time-based expiry. |
+| `refresh_keys: List[Tuple[str, str]]` | Cube-style `(physical_table, select_expression)` pairs. The same table may repeat with different expressions. |
+
+An empty/default `CacheConfig` caches indefinitely with no automatic staleness —
+only `evict` / `clear_cache` / an explicit re-execution change an entry.
+
+## Cache key
+
+The key is `sha256(final_generated_sql + "|" + connection_string + "|" +
+runtime_fingerprint)`. Variables are already substituted into the SQL before the
+key is computed, so different variable sets produce different keys automatically.
+The datasource identity is the engine's SQL-client fingerprint (not the bare
+datasource name), so a config edit under the same name never serves the wrong
+rows.
+
+Because the key needs the SQL, the resolve → enrich → SQL-generation pipeline
+runs on **every** cached call (cheap, no DB hit). A cache hit skips only the DB
+execution and result decode.
+
+## Staleness
+
+Each entry has two independent staleness signals.
+
+**TTL** is checked lazily on read: if an entry's age exceeds `ttl_seconds`, the
+read is treated as a miss and the entry is re-executed synchronously.
+
+**Refresh keys** are acted on only by an explicit `engine.refresh()`. For each
+entry, the *applicable* refresh keys are those whose table is referenced by the
+entry's SQL. A baseline value per applicable key is captured at cache-write time
+(scanned **before** the data query, so cached data always reflects a state at
+least as new as the baseline). `refresh()` re-evaluates each applicable key and
+re-executes the entry if any value differs.
+
+Refresh-key sensitivity is your choice of expression:
+
+- `MAX(updated_at)` misses in-place updates that don't move the column, backfills
+  below the current max, and inserts with old timestamps.
+- A `COUNT(*)`-bearing expression additionally catches deletes.
+- A hash / concatenation (e.g. `MAX(updated_at) || '|' || COUNT(*)`) catches more.
+
+`ttl_seconds` is the time-based backstop. SLayer evaluates each
+`select_expression` verbatim as `SELECT <select_expression> FROM <table>` — it is
+**not** wrapped in `MAX(...)`.
+
+## Management methods
+
+```python
+await engine.refresh()                 # -> RefreshResult   (+ engine.refresh_sync())
+await engine.evict(query)              # -> bool            (+ engine.evict_sync())
+engine.clear_cache()                   # drop all entries
+engine.cache_size                      # -> int (live entry count)
+```
+
+`evict(query, variables=None, data_source=None)` accepts the same input union as
+`execute`, recomputes the SQL + datasource key (no DB execution), and removes
+that one entry. `refresh()` returns a `RefreshResult`:
+
+```python
+class RefreshResult:
+    refreshed: List[str]          # re-run because a refresh-key value moved
+    expired_refreshed: List[str]  # re-run because the TTL lapsed
+    unchanged: List[str]
+    errors: List[RefreshError]    # continue-on-failure diagnostics
+```
+
+`refresh()` re-prepares each stale entry from its **original input** through the
+full `execute` pipeline (not the frozen SQL), so `whole_periods_only` boundaries
+re-snap and model/schema edits are picked up. If the freshly-prepared SQL differs
+(a new period, an edited model), the entry is re-keyed under the new SQL hash.
+`refresh()` is continue-on-failure: a failed refresh-key scan or re-execution
+leaves the existing entry unchanged and records a `RefreshError`.
+
+The synchronous wrappers `execute_sync(..., cache=True, data_source=...)`,
+`refresh_sync()`, and `evict_sync(...)` are available for CLI / notebook /
+script use.
+
+## Limitations
+
+- **In-memory, per-process.** The cache lives on the engine instance; it is not
+  persisted across restarts and is not shared across engines or processes.
+- **No request coalescing.** Concurrent identical misses both execute
+  (last-writer-wins on store).
+- **Unbounded.** There is no LRU or size cap; manage memory with `evict` /
+  `clear_cache`.
+- **Python API only.** No REST / MCP / CLI / Flight / pg-facade / `SlayerClient`
+  surface.