diff --git a/CLAUDE.md b/CLAUDE.md index 2403209a..774fff1a 100644 --- a/CLAUDE.md +++ b/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 () AS "slayer_rk_0", ... FROM ` (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. diff --git a/docs/concepts/query-cache.md b/docs/concepts/query-cache.md new file mode 100644 index 00000000..cf401b9d --- /dev/null +++ b/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 FROM
` — 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. diff --git a/docs/examples/12_query_cache/query_cache_nb.ipynb b/docs/examples/12_query_cache/query_cache_nb.ipynb new file mode 100644 index 00000000..85a191ce --- /dev/null +++ b/docs/examples/12_query_cache/query_cache_nb.ipynb @@ -0,0 +1,405 @@ +{ + "cells": [ + { + "cell_type": "markdown", + "id": "c26ac79e", + "metadata": {}, + "source": [ + "# Query Cache — Mechanics\n", + "\n", + "SLayer can cache query results **in memory, per `SlayerQueryEngine` instance**, opt-in per call\n", + "via `cache=True`. It is modelled on [Cube's in-memory cache](https://cube.dev/docs/product/caching):\n", + "a cache hit skips the database round-trip entirely, and staleness is governed by an optional\n", + "**time-to-live** and an optional set of Cube-style **refresh keys** that you scan explicitly with\n", + "`engine.refresh()`.\n", + "\n", + "This notebook builds a tiny SQLite shop from scratch (so we can freely mutate it) and walks through:\n", + "\n", + "1. opt-in caching (miss → hit),\n", + "2. how a hit serves the *last* result even after the data changes,\n", + "3. refresh keys invalidating on a real change,\n", + "4. TTL time-based expiry,\n", + "5. managing the cache (`evict` / `clear_cache` / `cache_size`).\n", + "\n", + "> The cache is a **Python-API-only** feature — there is no REST / MCP / CLI surface. See the\n", + "> [Query Cache concept doc](../../concepts/query-cache.md) for the full reference." + ] + }, + { + "cell_type": "code", + "execution_count": 1, + "id": "e19fe2dd", + "metadata": { + "execution": { + "iopub.execute_input": "2026-07-06T09:14:42.184444Z", + "iopub.status.busy": "2026-07-06T09:14:42.184004Z", + "iopub.status.idle": "2026-07-06T09:14:42.410281Z", + "shell.execute_reply": "2026-07-06T09:14:42.409881Z" + } + }, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "workspace: /tmp/tmp8r4a7ds1\n" + ] + } + ], + "source": [ + "import sqlite3\n", + "import tempfile\n", + "import time\n", + "from pathlib import Path\n", + "\n", + "from slayer.async_utils import run_sync\n", + "from slayer.core.enums import DataType\n", + "from slayer.core.models import Column, DatasourceConfig, SlayerModel\n", + "from slayer.engine.cache import CacheConfig\n", + "from slayer.engine.query_engine import SlayerQueryEngine\n", + "from slayer.storage.yaml_storage import YAMLStorage\n", + "\n", + "# A throwaway workspace: a SQLite \"shop\" database + a YAML model store.\n", + "work = Path(tempfile.mkdtemp())\n", + "db_path = work / \"shop.db\"\n", + "\n", + "conn = sqlite3.connect(db_path)\n", + "conn.executescript(\n", + " \"\"\"\n", + " CREATE TABLE orders (\n", + " id INTEGER PRIMARY KEY,\n", + " status TEXT,\n", + " amount REAL,\n", + " updated_at TEXT\n", + " );\n", + " INSERT INTO orders VALUES\n", + " (1, 'completed', 100.0, '2026-01-01'),\n", + " (2, 'pending', 50.0, '2026-01-02'),\n", + " (3, 'completed', 200.0, '2026-01-03');\n", + " \"\"\"\n", + ")\n", + "conn.commit()\n", + "conn.close()\n", + "\n", + "def mutate(sql: str) -> None:\n", + " \"\"\"Apply a write directly to the underlying table (simulating an ETL job).\"\"\"\n", + " c = sqlite3.connect(db_path)\n", + " c.execute(sql)\n", + " c.commit()\n", + " c.close()\n", + "\n", + "# Register the datasource + a simple `orders` model. Storage methods are async;\n", + "# run_sync bridges them for a notebook/script.\n", + "storage = YAMLStorage(base_dir=str(work / \"store\"))\n", + "run_sync(storage.save_datasource(\n", + " DatasourceConfig(name=\"shop\", type=\"sqlite\", database=str(db_path))\n", + "))\n", + "run_sync(storage.save_model(SlayerModel(\n", + " name=\"orders\",\n", + " sql_table=\"orders\",\n", + " data_source=\"shop\",\n", + " columns=[\n", + " Column(name=\"id\", sql=\"id\", type=DataType.INT, primary_key=True),\n", + " Column(name=\"status\", sql=\"status\", type=DataType.TEXT),\n", + " Column(name=\"amount\", sql=\"amount\", type=DataType.DOUBLE),\n", + " Column(name=\"updated_at\", sql=\"updated_at\", type=DataType.TIMESTAMP),\n", + " ],\n", + ")))\n", + "print(\"workspace:\", work)" + ] + }, + { + "cell_type": "markdown", + "id": "97b80f1f", + "metadata": {}, + "source": [ + "## 1. Opt-in caching\n", + "\n", + "Build an engine with a `CacheConfig`. We give it two **refresh keys** on the `orders` table —\n", + "`MAX(updated_at)` and `COUNT(*)` — which we'll use in step 3. For now there is no TTL, so entries\n", + "only change when we refresh, evict, or clear them.\n", + "\n", + "The query total is `100 + 50 + 200 = 350`." + ] + }, + { + "cell_type": "code", + "execution_count": 2, + "id": "155ff8eb", + "metadata": { + "execution": { + "iopub.execute_input": "2026-07-06T09:14:42.411363Z", + "iopub.status.busy": "2026-07-06T09:14:42.411283Z", + "iopub.status.idle": "2026-07-06T09:14:42.433873Z", + "shell.execute_reply": "2026-07-06T09:14:42.433429Z" + } + }, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "first : [{'orders.amount_sum': 350.0}]\n", + "second: [{'orders.amount_sum': 350.0}]\n", + "cache_size: 1\n" + ] + } + ], + "source": [ + "engine = SlayerQueryEngine(\n", + " storage=storage,\n", + " cache_config=CacheConfig(\n", + " ttl_seconds=None, # no time-based expiry (yet — see step 4)\n", + " refresh_keys=[\n", + " (\"orders\", \"MAX(updated_at)\"),\n", + " (\"orders\", \"COUNT(*)\"),\n", + " ],\n", + " ),\n", + ")\n", + "\n", + "query = {\"source_model\": \"orders\", \"measures\": [\"amount:sum\"]}\n", + "\n", + "first = engine.execute_sync(query, cache=True) # miss: runs SQL, stores the result\n", + "second = engine.execute_sync(query, cache=True) # hit: served from memory, no DB round-trip\n", + "\n", + "print(\"first :\", first.data)\n", + "print(\"second:\", second.data)\n", + "print(\"cache_size:\", engine.cache_size)" + ] + }, + { + "cell_type": "markdown", + "id": "6ac61215", + "metadata": {}, + "source": [ + "## 2. A hit serves the *last* result\n", + "\n", + "Now an ETL job inserts a new completed order worth `1000`. The true total becomes `1350`.\n", + "\n", + "A `cache=True` call still returns the cached `350` — that is the whole point of a cache. Passing\n", + "`cache=False` bypasses it and shows the fresh value." + ] + }, + { + "cell_type": "code", + "execution_count": 3, + "id": "7f3bb82e", + "metadata": { + "execution": { + "iopub.execute_input": "2026-07-06T09:14:42.434840Z", + "iopub.status.busy": "2026-07-06T09:14:42.434766Z", + "iopub.status.idle": "2026-07-06T09:14:42.444105Z", + "shell.execute_reply": "2026-07-06T09:14:42.443706Z" + } + }, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "cached (cache=True) : [{'orders.amount_sum': 350.0}]\n", + "bypass (cache=False): [{'orders.amount_sum': 1350.0}]\n" + ] + } + ], + "source": [ + "mutate(\"INSERT INTO orders VALUES (4, 'completed', 1000.0, '2026-02-01')\")\n", + "\n", + "print(\"cached (cache=True) :\", engine.execute_sync(query, cache=True).data)\n", + "print(\"bypass (cache=False):\", engine.execute_sync(query, cache=False).data)" + ] + }, + { + "cell_type": "markdown", + "id": "63ea2707", + "metadata": {}, + "source": [ + "## 3. Refresh keys — invalidate on a real change\n", + "\n", + "`engine.refresh()` scans each entry's applicable refresh keys and re-executes the entry when a\n", + "value has **moved**. Our insert bumped both `MAX(updated_at)` and `COUNT(*)`, so the entry is\n", + "re-run and lands in the `refreshed` bucket. Afterwards the cached value reflects the new total.\n", + "\n", + "Refresh-key sensitivity is your choice of expression: `MAX(updated_at)` catches new-timestamped\n", + "inserts; a `COUNT(*)` key additionally catches **deletes** that leave the max untouched." + ] + }, + { + "cell_type": "code", + "execution_count": 4, + "id": "95ed7b8b", + "metadata": { + "execution": { + "iopub.execute_input": "2026-07-06T09:14:42.444950Z", + "iopub.status.busy": "2026-07-06T09:14:42.444878Z", + "iopub.status.idle": "2026-07-06T09:14:42.456445Z", + "shell.execute_reply": "2026-07-06T09:14:42.456051Z" + } + }, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "refreshed : 1\n", + "expired_refreshed: 0\n", + "unchanged : 0\n", + "errors : []\n", + "cached after refresh: [{'orders.amount_sum': 1350.0}]\n" + ] + } + ], + "source": [ + "result = engine.refresh_sync()\n", + "print(\"refreshed :\", len(result.refreshed))\n", + "print(\"expired_refreshed:\", len(result.expired_refreshed))\n", + "print(\"unchanged :\", len(result.unchanged))\n", + "print(\"errors :\", result.errors)\n", + "print(\"cached after refresh:\", engine.execute_sync(query, cache=True).data)" + ] + }, + { + "cell_type": "markdown", + "id": "392412dc", + "metadata": {}, + "source": [ + "## 4. TTL — time-based expiry\n", + "\n", + "A `ttl_seconds` bounds an entry's wall-clock age. It is checked **lazily on read**: once an entry\n", + "is older than the TTL, the next `cache=True` read treats it as a miss and re-executes. Here we use\n", + "a 1-second TTL and sleep past it." + ] + }, + { + "cell_type": "code", + "execution_count": 5, + "id": "1d2c33a3", + "metadata": { + "execution": { + "iopub.execute_input": "2026-07-06T09:14:42.457440Z", + "iopub.status.busy": "2026-07-06T09:14:42.457364Z", + "iopub.status.idle": "2026-07-06T09:14:43.572061Z", + "shell.execute_reply": "2026-07-06T09:14:43.571673Z" + } + }, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "populate: [{'orders.amount_sum': 1350.0}]\n", + "within TTL (stale): [{'orders.amount_sum': 1350.0}]\n" + ] + }, + { + "name": "stdout", + "output_type": "stream", + "text": [ + "after TTL (fresh): [{'orders.amount_sum': 1375.0}]\n" + ] + } + ], + "source": [ + "ttl_engine = SlayerQueryEngine(\n", + " storage=storage,\n", + " cache_config=CacheConfig(ttl_seconds=1.0),\n", + ")\n", + "\n", + "print(\"populate:\", ttl_engine.execute_sync(query, cache=True).data)\n", + "mutate(\"INSERT INTO orders VALUES (5, 'pending', 25.0, '2026-02-02')\")\n", + "\n", + "print(\"within TTL (stale):\", ttl_engine.execute_sync(query, cache=True).data)\n", + "time.sleep(1.1) # let the entry age past ttl_seconds\n", + "print(\"after TTL (fresh):\", ttl_engine.execute_sync(query, cache=True).data)" + ] + }, + { + "cell_type": "markdown", + "id": "f0db73e4", + "metadata": {}, + "source": [ + "## 5. Managing the cache\n", + "\n", + "- `evict(query)` removes one entry (recomputing its key without touching the database).\n", + "- `clear_cache()` drops everything.\n", + "- `cache_size` reports the live entry count." + ] + }, + { + "cell_type": "code", + "execution_count": 6, + "id": "32060a6c", + "metadata": { + "execution": { + "iopub.execute_input": "2026-07-06T09:14:43.573536Z", + "iopub.status.busy": "2026-07-06T09:14:43.573426Z", + "iopub.status.idle": "2026-07-06T09:14:43.585072Z", + "shell.execute_reply": "2026-07-06T09:14:43.584751Z" + } + }, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "size before evict: 1\n", + "evict returned : True\n", + "size after evict : 0\n", + "size after re-cache: 1\n", + "size after clear : 0\n" + ] + } + ], + "source": [ + "print(\"size before evict:\", engine.cache_size)\n", + "print(\"evict returned :\", engine.evict_sync(query))\n", + "print(\"size after evict :\", engine.cache_size)\n", + "\n", + "engine.execute_sync(query, cache=True)\n", + "print(\"size after re-cache:\", engine.cache_size)\n", + "engine.clear_cache()\n", + "print(\"size after clear :\", engine.cache_size)" + ] + }, + { + "cell_type": "markdown", + "id": "9aa9bd9a", + "metadata": {}, + "source": [ + "## Recap\n", + "\n", + "- `execute(query, cache=True)` is opt-in and per-call; `dry_run` / `explain` are never cached.\n", + "- The cache is **per engine instance** — two engines (e.g. different tenants / connection settings)\n", + " never share cached rows.\n", + "- Staleness has two independent signals: **TTL** (lazy on read) and **refresh keys** (scanned by\n", + " `engine.refresh()`).\n", + "- Everything shown here has a `*_sync` wrapper (`execute_sync`, `refresh_sync`, `evict_sync`) and an\n", + " `async` twin.\n", + "\n", + "For the full reference — cache key, table detection, the refresh-key trade-off, and non-goals —\n", + "see the [Query Cache concept doc](../../concepts/query-cache.md)." + ] + } + ], + "metadata": { + "kernelspec": { + "display_name": "Python 3", + "language": "python", + "name": "python3" + }, + "language_info": { + "codemirror_mode": { + "name": "ipython", + "version": 3 + }, + "file_extension": ".py", + "mimetype": "text/x-python", + "name": "python", + "nbconvert_exporter": "python", + "pygments_lexer": "ipython3", + "version": "3.12.3" + } + }, + "nbformat": 4, + "nbformat_minor": 5 +} diff --git a/slayer/engine/cache.py b/slayer/engine/cache.py new file mode 100644 index 00000000..8ca2631f --- /dev/null +++ b/slayer/engine/cache.py @@ -0,0 +1,336 @@ +"""Per-engine, in-memory query result cache (DEV-1587). + +A query-level result cache local to a single :class:`SlayerQueryEngine` +instance, modelled on Cube's in-memory cache. Caching is opt-in per call +via ``execute(query, cache=True)``. Staleness is governed by an optional +time-to-live (``ttl_seconds``, checked lazily on read) and an optional set +of Cube-style ``(physical_table, select_expression)`` refresh keys scanned +by an explicit ``engine.refresh()``. + +This module holds the DB-free half of the feature: the cache dict, the +cache key, TTL bookkeeping, sqlglot-based table detection (CTE-alias +excluding), refresh-key applicability, refresh-key scan-SQL building, and +value comparison. All of it is unit-testable without a database. The +engine (``slayer/engine/query_engine.py``) owns the DB awaits (data query, +refresh-key scans, re-execution) and never holds the cache lock across +one. +""" + +import asyncio +import hashlib +import time +from collections.abc import Callable +from typing import Any + +from pydantic import BaseModel, ConfigDict, Field, field_validator +from sqlglot import exp, parse_one +from sqlglot.optimizer.normalize_identifiers import normalize_identifiers +from sqlglot.optimizer.scope import Scope, traverse_scope + +# Marker alias prefix for refresh-key scan projections. Also lets tests +# distinguish a refresh-key scan query from a data query. +_RK_ALIAS_PREFIX = "slayer_rk_" + +# Normalized physical-table identity: (catalog, db, name). Parts absent +# from the source expression are ``None``. +NormalizedTable = tuple[str | None, str | None, str] + + +class CacheConfig(BaseModel): + """Per-engine cache configuration. + + ``ttl_seconds`` bounds wall-clock entry age (``None`` => no time-based + expiry). ``refresh_keys`` is a sequence of ``(physical_table, + select_expression)`` pairs; the same table may repeat with different + expressions. Each ``select_expression`` is a scalar SQL expression + evaluated verbatim as ``SELECT FROM
`` — the + user supplies it in full (SLayer does NOT wrap it in ``MAX(...)``). + + The model is **frozen** and ``refresh_keys`` is a tuple, so the config is + genuinely immutable (mirrors ``SessionPolicy``). To change an engine's + policy, reassign ``engine.cache_config`` — the setter clears the cache so + stale entries can't survive under a new TTL / refresh-key set. In-place + mutation (``engine.cache_config.refresh_keys += ...``) is rejected rather + than silently leaving already-cached entries on the old policy. + """ + + model_config = ConfigDict(frozen=True) + + ttl_seconds: float | None = None + refresh_keys: tuple[tuple[str, str], ...] = () + + @field_validator("refresh_keys", mode="before") + @classmethod + def _coerce_refresh_keys(cls, v: Any) -> Any: + """Accept a list of lists/tuples and freeze it into a tuple of tuples.""" + if v is None: + return () + return tuple(tuple(pair) for pair in v) + + +class RefreshKeyValue(BaseModel): + """A captured refresh-key baseline: the value of ``expression`` scanned + from ``table`` at cache-write (or last-refresh) time.""" + + model_config = ConfigDict(arbitrary_types_allowed=True) + + table: str + expression: str + value: Any = None + + +class RefreshError(BaseModel): + """A per-entry / per-table failure recorded during ``refresh()``. + + ``phase`` is ``"refresh_key_scan"`` (a table's batched scan raised) or + ``"re_execute"`` (re-running a stale entry raised). ``key`` is the cache + key for a re-execute failure, or the physical table for a scan failure. + """ + + key: str + phase: str + message: str + + +class RefreshResult(BaseModel): + """Outcome of ``engine.refresh()``. + + ``refreshed`` — keys re-run because an applicable refresh-key value + moved. ``expired_refreshed`` — keys re-run because their TTL lapsed. + ``unchanged`` — keys left as-is. ``errors`` — continue-on-failure + diagnostics. + """ + + refreshed: list[str] = Field(default_factory=list) + expired_refreshed: list[str] = Field(default_factory=list) + unchanged: list[str] = Field(default_factory=list) + errors: list[RefreshError] = Field(default_factory=list) + + +class _CacheEntry(BaseModel): + """One cached result plus everything needed to re-scan its refresh keys + and re-prepare + re-execute it from the original user input. + + ``response`` holds a :class:`SlayerResponse` (typed ``Any`` to avoid a + circular import with ``query_engine``). ``original_input`` is the raw + user input shape (``SlayerQuery`` / ``dict`` / ``list`` / ``str``) so + ``refresh()`` can replay through the full ``execute`` normalization. + """ + + model_config = ConfigDict(arbitrary_types_allowed=True) + + response: Any + sql: str + ds_fingerprint: str + dialect: str + ds_key: tuple[str, str] + resolved_data_source: str | None = None + original_input: Any = None + variables: dict[str, Any] | None = None + data_source: str | None = None + created_at: float = 0.0 + applicable: list[tuple[str, str]] = Field(default_factory=list) + refresh_key_values: list[RefreshKeyValue] = Field(default_factory=list) + + +class QueryCache: + """In-memory ``dict[str, _CacheEntry]`` with TTL-aware reads. + + The :class:`asyncio.Lock` guards only in-memory dict operations + (get / put / delete / snapshot / commit) — the engine performs every DB + await outside the lock. The ``clock`` is injectable for deterministic + TTL testing (``time.monotonic`` by default). + """ + + def __init__( + self, + config: CacheConfig, + clock: Callable[[], float] = time.monotonic, + ) -> None: + self.config = config + self._clock = clock + self._entries: dict[str, _CacheEntry] = {} + self._lock = asyncio.Lock() + + # ---- key / clock / size ------------------------------------------------ + + @staticmethod + def make_key(sql: str, ds_fingerprint: str) -> str: + """``sha256(final_sql + "|" + ds_fingerprint)``. + + ``ds_fingerprint`` is the engine's SQL-client cache fingerprint + (``connection_string|runtime_fingerprint``), so a config edit under + the same datasource name never serves the wrong rows. + """ + return hashlib.sha256(f"{sql}|{ds_fingerprint}".encode()).hexdigest() + + def now(self) -> float: + """Read the injectable clock (used for entry ``created_at``).""" + return self._clock() + + def size(self) -> int: + return len(self._entries) + + def clear(self) -> None: + self._entries.clear() + + # ---- lock-guarded dict ops -------------------------------------------- + + async def get(self, key: str) -> _CacheEntry | None: + """Return the live entry, or ``None``. TTL-expired entries are + deleted and reported as a miss (re-execution re-populates them).""" + async with self._lock: + entry = self._entries.get(key) + if entry is None: + return None + if self.config.ttl_seconds is not None: + if (self._clock() - entry.created_at) > self.config.ttl_seconds: + del self._entries[key] + return None + return entry + + async def put(self, key: str, entry: _CacheEntry) -> None: + async with self._lock: + self._entries[key] = entry + + async def delete(self, key: str) -> bool: + async with self._lock: + if key in self._entries: + del self._entries[key] + return True + return False + + async def snapshot(self) -> dict[str, _CacheEntry]: + """A shallow copy of the ``{key: entry}`` map so ``refresh()`` can + iterate without racing concurrent ``execute()`` writes.""" + async with self._lock: + return dict(self._entries) + + async def commit_replace( + self, + *, + old_key: str, + expected: _CacheEntry, + new_key: str, + new_entry: _CacheEntry, + ) -> bool: + """Identity-guarded write used by ``refresh()``. + + Only mutate if the live entry at ``old_key`` is still the SAME + object ``refresh()`` snapshotted. If it was evicted / cleared / + replaced by a newer ``execute()`` during ``refresh()``'s DB awaits, + skip the write (return ``False``) — never resurrect a gone entry or + clobber a newer result. On a re-key (``new_key != old_key``) the old + key is dropped; a ``new_key`` collision is last-writer-wins (both + results are interchangeable — identical SQL + ds fingerprint). + """ + async with self._lock: + if self._entries.get(old_key) is not expected: + return False + if new_key != old_key: + self._entries.pop(old_key, None) + self._entries[new_key] = new_entry + return True + + # ---- table detection --------------------------------------------------- + + def parse_referenced_tables( + self, sql: str, dialect: str + ) -> list[NormalizedTable]: + """Normalized physical tables referenced by ``sql``. + + Parses with sqlglot and uses **scope analysis** (``traverse_scope``, + the same mechanism the forced-filter policy uses) to keep only + genuinely physical tables: a table reference whose name resolves to a + CTE / derived table in its scope is skipped. This is name-collision + safe — a physical table that happens to share a name with a CTE alias + (SLayer wraps queries in CTEs) is still detected, because scope + resolution distinguishes them structurally rather than by bare name. + Each physical :class:`exp.Table` is normalized to ``(catalog, db, + name)`` with the dialect's identifier folding (quoted identifiers + preserved exactly, unquoted folded per dialect). + """ + tree = parse_one(sql, dialect=dialect) + out: list[NormalizedTable] = [] + for scope in traverse_scope(tree): + for table in scope.tables: + # A qualified reference (has a db/catalog part) can never be a + # CTE — CTE names are always unqualified — so it is always + # physical. Only an unqualified name can shadow a CTE / derived + # source in its scope; skip those (they aren't physical tables). + if not table.db and not table.catalog: + if isinstance(scope.sources.get(table.alias_or_name), Scope): + continue + out.append(self._normalize_table_expr(table, dialect)) + return out + + @staticmethod + def _normalize_table_expr(table: exp.Table, dialect: str) -> NormalizedTable: + norm = normalize_identifiers(table.copy(), dialect=dialect) + return (norm.catalog or None, norm.db or None, norm.name) + + @classmethod + def _normalize_config_table(cls, table: str, dialect: str) -> NormalizedTable: + return cls._normalize_table_expr(exp.to_table(table, dialect=dialect), dialect) + + @staticmethod + def _table_matches(config: NormalizedTable, sql_table: NormalizedTable) -> bool: + """A config table matches a SQL table iff their normalized parts are + equal, treating parts unspecified (``None``) in the config as + wildcards. So ``orders`` matches ``orders`` / ``public.orders`` / + ``db.public.orders``; ``public.orders`` matches only ``db=public``. + """ + c_cat, c_db, c_name = config + s_cat, s_db, s_name = sql_table + if c_name != s_name: + return False + if c_db is not None and c_db != s_db: + return False + if c_cat is not None and c_cat != s_cat: + return False + return True + + def applicable_keys(self, sql: str, dialect: str) -> list[tuple[str, str]]: + """The configured ``(table, expression)`` refresh keys whose table is + referenced by ``sql``. The original config table string is preserved + (duplicate expressions per table are kept).""" + sql_tables = self.parse_referenced_tables(sql, dialect) + out: list[tuple[str, str]] = [] + for table, expr in self.config.refresh_keys: + config_norm = self._normalize_config_table(table, dialect) + if any(self._table_matches(config_norm, s) for s in sql_tables): + out.append((table, expr)) + return out + + def build_refresh_key_sql( + self, table: str, expressions: list[str], dialect: str + ) -> str: + """``SELECT () AS "slayer_rk_0", ... FROM
``. + + Each user expression is parsed with the entry's dialect and + re-emitted so quoting/dialect are consistent; the table identifier is + quoted dialect-safely. One scan covers all of a table's applicable + refresh keys. + """ + projections = [ + exp.alias_( + exp.paren(parse_one(e, dialect=dialect)), + f"{_RK_ALIAS_PREFIX}{i}", + quoted=True, + ) + for i, e in enumerate(expressions) + ] + select = exp.select(*projections).from_(exp.to_table(table, dialect=dialect)) + return select.sql(dialect=dialect) + + @staticmethod + def rk_alias(index: int) -> str: + """The scan projection alias for the ``index``-th expression.""" + return f"{_RK_ALIAS_PREFIX}{index}" + + @staticmethod + def values_differ(a: Any, b: Any) -> bool: + """Equality comparison across DB scalar types (timestamps / ints / + strings / concatenations). Any inequality — including a decrease — + signals staleness.""" + return a != b diff --git a/slayer/engine/query_engine.py b/slayer/engine/query_engine.py index 6d5c9cb7..2b6d31bd 100644 --- a/slayer/engine/query_engine.py +++ b/slayer/engine/query_engine.py @@ -3,12 +3,18 @@ Flow: SlayerQuery → _enrich() → EnrichedQuery → SQLGenerator → SQL → execute """ +import copy import decimal import logging from contextvars import ContextVar from typing import Any -from pydantic import BaseModel, Field as PydanticField, model_validator +from pydantic import ( + BaseModel, + ConfigDict as PydanticConfigDict, + Field as PydanticField, + model_validator, +) import sqlalchemy as sa from sqlglot import exp @@ -36,6 +42,14 @@ RootModelRecommendation, ) from slayer.core.refs import split_agg_suffix +from slayer.engine.cache import ( + CacheConfig, + QueryCache, + RefreshError, + RefreshKeyValue, + RefreshResult, + _CacheEntry, +) from slayer.engine.join_graph import JoinGraph from slayer.engine.enriched import ( CrossModelMeasure, @@ -336,6 +350,29 @@ def to_markdown(self) -> str: return "\n".join([header, separator] + body_lines) +class _Prepared(BaseModel): + """DB-free result of the resolve→enrich→SQL-gen pipeline. + + Everything ``execute()`` needs to run (or cache-key) a query without a + database round-trip. Produced by :meth:`SlayerQueryEngine._prepare_pipeline` + and shared by the execute path, ``evict`` (key only), and ``refresh`` + re-execution. ``ds_fingerprint`` is ``"|".join(_sql_client_cache_key(ds))`` + — the cache-key datasource identity. + """ + + model_config = PydanticConfigDict(arbitrary_types_allowed=True) + + model: SlayerModel + enriched: Any + datasource: DatasourceConfig + dialect: str + sql: str + attributes: "ResponseAttributes" + expected_columns: list[str] + ds_key: tuple[str, str] + ds_fingerprint: str + + def _infer_aggregated_format( model: SlayerModel, measure_name: str, @@ -378,6 +415,7 @@ def __init__( self, storage: StorageBackend, *, + cache_config: CacheConfig | None = None, policy: SessionPolicy | None = None, ): self.storage = storage @@ -395,6 +433,31 @@ def __init__( # an unconfirmable ``None`` is re-probed so a transient introspection # failure self-heals once the datasource recovers. self._column_presence_cache: dict[tuple, bool] = {} + # DEV-1587: per-engine, in-memory, opt-in query result cache. The + # cache is local to this engine instance so two engines with + # different RLS / connection settings keep separate caches. + self._cache = QueryCache(config=cache_config or CacheConfig()) + + @property + def cache_config(self) -> CacheConfig: + return self._cache.config + + @cache_config.setter + def cache_config(self, value: CacheConfig) -> None: + # Reassigning the config drops all cached entries — the new TTL / + # refresh-key policy shouldn't retroactively apply to entries whose + # baselines were captured under the old policy. + self._cache.config = value + self._cache.clear() + + @property + def cache_size(self) -> int: + """Number of live cache entries.""" + return self._cache.size() + + def clear_cache(self) -> None: + """Drop all cached entries.""" + self._cache.clear() def _apply_policy( self, *, sql: str, dialect: str, datasource: DatasourceConfig @@ -701,7 +764,7 @@ def _topologically_order_queries( sorted_names = cls._kahn_sort(in_degree, dependents) return [rest_by_name[n] for n in sorted_names] + [root] - async def execute( # NOSONAR S3776 — public dispatch over str/dict/list/SlayerQuery; splitting hides the input-shape contract + async def execute( self, query: "SlayerQuery | dict | list[SlayerQuery | dict] | str", variables: dict[str, Any] | None = None, @@ -709,21 +772,54 @@ async def execute( # NOSONAR S3776 — public dispatch over str/dict/list/Slaye dry_run: bool = False, explain: bool = False, data_source: str | None = None, + cache: bool = False, ) -> SlayerResponse: + """Resolve, enrich, generate SQL, and execute ``query``. + + Accepts a ``SlayerQuery`` / dict, a multi-stage DAG list, or a + run-by-name string. DEV-1587: pass ``cache=True`` to serve the result + from (and store it in) this engine's per-instance result cache; + ignored (no caching, no error) when ``dry_run`` or ``explain`` is set. + """ + query_obj, named_queries, runtime_kwarg, prefer = await self._normalize_input( + query, variables=variables, data_source=data_source, + ) + return await self._execute_pipeline( + query=query_obj, + named_queries=named_queries, + runtime_kwarg=runtime_kwarg, + dry_run=dry_run, + explain=explain, + prefer_data_source=prefer, + cache=cache, + original_input=query, + original_variables=variables, + original_data_source=data_source, + ) + + async def _normalize_input( # NOSONAR S3776 — public dispatch over str/dict/list/SlayerQuery; splitting hides the input-shape contract + self, + query: "SlayerQuery | dict | list[SlayerQuery | dict] | str", + *, + variables: dict[str, Any] | None = None, + data_source: str | None = None, + ) -> "tuple[SlayerQuery, dict[str, SlayerQuery], dict[str, Any], str | None]": + """Normalize the ``execute`` / ``evict`` input union into + ``(query, named_queries, runtime_kwarg, prefer_data_source)``. + + Folds the run-by-name lookup, list topo-sort, dict validation, and + ``variables=`` merge. Shared by the execute path and ``evict`` so a + cached entry's key is computed identically on both. + """ runtime_kwarg = variables or {} # Run-by-name dispatch: ``execute("model_name", variables=...)`` runs # the backing query of a query-backed model. if isinstance(query, str): - return await self._execute_by_name( - name=query, - runtime_kwarg=runtime_kwarg, - dry_run=dry_run, - explain=explain, - data_source=data_source, + return await self._normalize_by_name( + name=query, runtime_kwarg=runtime_kwarg, data_source=data_source, ) - # Accept dicts and validate them into SlayerQuery objects if isinstance(query, list): if not query: @@ -751,24 +847,16 @@ async def execute( # NOSONAR S3776 — public dispatch over str/dict/list/Slaye if merged_top != (query.variables or {}): query = query.model_copy(update={"variables": merged_top}) - return await self._execute_pipeline( - query=query, - named_queries=named_queries, - runtime_kwarg=runtime_kwarg, - dry_run=dry_run, - explain=explain, - prefer_data_source=data_source, - ) + return query, named_queries, runtime_kwarg, data_source - async def _execute_by_name( + async def _normalize_by_name( self, + *, name: str, runtime_kwarg: dict[str, Any], - dry_run: bool = False, - explain: bool = False, - data_source: str | None = None, - ) -> SlayerResponse: - """Run the backing query of a query-backed model by name.""" + data_source: str | None, + ) -> "tuple[SlayerQuery, dict[str, SlayerQuery], dict[str, Any], str | None]": + """Resolve a run-by-name input into the normalized tuple.""" model = await self.storage.get_model(name, data_source=data_source) if model is None: raise ValueError(f"Model '{name}' not found") @@ -802,29 +890,29 @@ async def _execute_by_name( if merged != (main_query.variables or {}): main_query = main_query.model_copy(update={"variables": merged}) - return await self._execute_pipeline( - query=main_query, - named_queries=named_queries, - runtime_kwarg=runtime_kwarg, - dry_run=dry_run, - explain=explain, - prefer_data_source=model.data_source or data_source, - ) + return main_query, named_queries, runtime_kwarg, model.data_source or data_source - async def _execute_pipeline( # NOSONAR S3776 — linear pipeline (resolve→enrich→generate→execute); breaking it up obscures the order of operations + async def _prepare_pipeline( self, query: SlayerQuery, named_queries: dict[str, SlayerQuery], runtime_kwarg: dict[str, Any], *, - dry_run: bool = False, - explain: bool = False, prefer_data_source: str | None = None, - ) -> SlayerResponse: - """Shared pipeline used by both ``execute()`` and ``_execute_by_name()``. - - Assumes ``query.variables`` already reflects the resolved variable - context for the top of the chain (kwarg merged in by the caller). + ) -> _Prepared: + """Resolve → enrich → generate SQL (+ forced-filter policy rewrite) → + build response metadata → compute the datasource cache identity. + + No ``SlayerSQLClient`` is instantiated and no data query runs, so with + no forced-filter policy configured this is fully DB-free — the basis + for ``evict`` recomputing a cache key without executing. When a policy + IS set, ``_apply_policy`` probes column presence via a SQLAlchemy + ``Inspector`` (a lightweight, cached metadata connection), so under RLS + the key computation touches the database that far — and must, since the + cached SQL is itself policy-rewritten and the key has to match. + + Shared by ``_execute_pipeline`` (execute + cache miss), ``evict`` (key + only), and ``refresh`` re-execution. """ # Pre-processing: strip redundant source model name prefixes from all references query = query.strip_source_model_prefix() @@ -833,13 +921,9 @@ async def _execute_pipeline( # NOSONAR S3776 — linear pipeline (resolve→enr for name, q in named_queries.items() } - # Preprocessing if query.whole_periods_only: query = query.snap_to_whole_periods() - # Resolve model from query.source_model (str, SlayerModel, or ModelExtension). - # Pass query.variables as the outer-vars context for any nested - # query-backed model resolution; runtime_kwarg threads through unchanged. resolving: set = set() model = await self._resolve_query_model( query_model=query.source_model, @@ -858,12 +942,10 @@ async def _execute_pipeline( # NOSONAR S3776 — linear pipeline (resolve→enr # Enrich: SlayerQuery + model → EnrichedQuery enriched = await self._enrich(query=query, model=model, named_queries=named_queries) - # Generate SQL from EnrichedQuery + # Generate SQL from EnrichedQuery. DEV-1444: pin ``outer`` mode so the + # projection is trimmed to public_projection_aliases(enriched). dialect = self._dialect_for_type(datasource.type) generator = SQLGenerator(dialect=dialect) - # DEV-1444: this is the final-stage SQL that gets executed and - # shown to the user — pin ``outer`` mode so the projection is - # trimmed to public_projection_aliases(enriched). sql = generator.generate(enriched=enriched, render_mode="outer") # DEV-1578: forced-filter rewrite. Applied here — before the dry_run # return and before execution — so dry_run, explain, real execution, @@ -871,14 +953,37 @@ async def _execute_pipeline( # NOSONAR S3776 — linear pipeline (resolve→enr sql = self._apply_policy(sql=sql, dialect=dialect, datasource=datasource) logger.debug("Generated SQL:\n%s", sql) - # DEV-1444: the response's attributes + expected_columns must mirror - # the trimmed outer projection — never include hoisted intermediates. + attributes, expected_columns = self._build_response_metadata( + enriched=enriched, model=model, + ) + + # DEV-1587: datasource cache identity, computed from config only + # (no client instantiation). DEV-1551: includes Snowflake runtime + # overrides so a warehouse/role change re-keys. + ds_key = _sql_client_cache_key(datasource) + ds_fingerprint = "|".join(ds_key) + + return _Prepared( + model=model, + enriched=enriched, + datasource=datasource, + dialect=dialect, + sql=sql, + attributes=attributes, + expected_columns=expected_columns, + ds_key=ds_key, + ds_fingerprint=ds_fingerprint, + ) + + def _build_response_metadata( # NOSONAR S3776 — linear per-bucket alias/format collection (dims/tds/measures/exprs/transforms/cross-model); splitting the flat loops adds indirection without reducing the branch count + self, *, enriched: "EnrichedQuery", model: SlayerModel, + ) -> "tuple[ResponseAttributes, list[str]]": + """Build the response ``attributes`` (label/format per public alias) + and ``expected_columns`` from an enriched query — mirroring the + trimmed outer projection (DEV-1444). Hidden transforms / ORDER-BY + aggregates / window-arg hoists are dropped.""" public_aliases = set(public_projection_aliases(enriched)) - # Collect field metadata from enriched query, split by type. Each - # entry is included only if its alias is part of the public - # projection (filter-extracted hidden transforms, ORDER-BY - # aggregates, and window-arg hoists are silently dropped). dim_meta: dict[str, FieldMetadata] = {} measure_meta: dict[str, FieldMetadata] = {} for d in enriched.dimensions: @@ -916,10 +1021,6 @@ async def _execute_pipeline( # NOSONAR S3776 — linear pipeline (resolve→enr measure_meta[cm.alias] = FieldMetadata(label=cm.label, format=cm.format) attributes = ResponseAttributes(dimensions=dim_meta, measures=measure_meta) - # DEV-1444: expected_columns matches the outer SELECT projection - # exactly (helper-driven). Fall back to the legacy bucket-union if - # ``public_projection`` is empty (e.g. enrichment paths that don't - # yet populate ``user_projection``). expected_columns = list(public_projection_aliases(enriched)) or ( [d.alias for d in enriched.dimensions] + [td.alias for td in enriched.time_dimensions] @@ -928,48 +1029,474 @@ async def _execute_pipeline( # NOSONAR S3776 — linear pipeline (resolve→enr + [t.alias for t in enriched.transforms if not t.name.startswith(("_inner_", "_ft"))] + [cm.alias for cm in enriched.cross_model_measures] ) + return attributes, expected_columns - # dry_run: return SQL without executing + async def _execute_pipeline( + self, + query: SlayerQuery, + named_queries: dict[str, SlayerQuery], + runtime_kwarg: dict[str, Any], + *, + dry_run: bool = False, + explain: bool = False, + prefer_data_source: str | None = None, + cache: bool = False, + original_input: Any = None, + original_variables: dict[str, Any] | None = None, + original_data_source: str | None = None, + ) -> SlayerResponse: + """Prepare (DB-free) then run: dry_run / explain / cached / plain.""" + prepared = await self._prepare_pipeline( + query=query, + named_queries=named_queries, + runtime_kwarg=runtime_kwarg, + prefer_data_source=prefer_data_source, + ) + + # dry_run: return SQL without executing (never cached). if dry_run: - return SlayerResponse(data=[], columns=expected_columns, sql=sql, attributes=attributes) - - # Execute — reuse SQL client (and its connection pool) per - # datasource. DEV-1551: include Snowflake runtime overrides - # (warehouse / role / database / schema_name) in the cache key - # so two datasources sharing the same connection_name but - # differing in (e.g.) warehouse don't accidentally share a client - # whose engine's per-connection ``USE WAREHOUSE`` listener was - # set up for the other datasource. - ds_key = _sql_client_cache_key(datasource) - if ds_key not in self._sql_clients: - self._sql_clients[ds_key] = SlayerSQLClient(datasource=datasource) - client = self._sql_clients[ds_key] + return SlayerResponse( + data=[], + columns=prepared.expected_columns, + sql=prepared.sql, + attributes=prepared.attributes, + ) + + client = self._get_client(prepared.datasource, prepared.ds_key) - # explain: run dialect-appropriate EXPLAIN on the query + # explain: dialect-appropriate EXPLAIN (never cached). if explain: - explain_sql = _build_explain_sql(dialect=dialect, sql=sql) + explain_sql = _build_explain_sql(dialect=prepared.dialect, sql=prepared.sql) try: rows = await client.execute(sql=explain_sql) except Exception as exc: await self._maybe_raise_schema_drift( - err=exc, model=model, enriched=enriched + err=exc, model=prepared.model, enriched=prepared.enriched ) raise - return SlayerResponse(data=rows, sql=sql, attributes=attributes) + return SlayerResponse( + data=rows, sql=prepared.sql, attributes=prepared.attributes + ) + # DEV-1587 cache hook: only for plain data queries. + if cache: + return await self._execute_cached( + prepared=prepared, + client=client, + original_input=original_input, + original_variables=original_variables, + original_data_source=original_data_source, + ) + + return await self._run_and_build(prepared=prepared, client=client) + + def _get_client( + self, datasource: DatasourceConfig, ds_key: tuple[str, str] + ) -> SlayerSQLClient: + """Reuse (or open) the SQL client + connection pool for a datasource. + + DEV-1551: keyed by ``(connection_string, runtime_fingerprint)`` so two + datasources sharing a ``connection_name`` but differing in warehouse / + role / database / schema get distinct clients. + """ + if ds_key not in self._sql_clients: + self._sql_clients[ds_key] = SlayerSQLClient(datasource=datasource) + return self._sql_clients[ds_key] + + async def _run_and_build( + self, *, prepared: _Prepared, client: SlayerSQLClient + ) -> SlayerResponse: + """Execute the prepared SQL, apply schema-drift attribution + the + dialect read-side decode, and build the response.""" try: - rows = await client.execute(sql=sql) + rows = await client.execute(sql=prepared.sql) except Exception as exc: await self._maybe_raise_schema_drift( - err=exc, model=model, enriched=enriched + err=exc, model=prepared.model, enriched=prepared.enriched ) raise # Dialect-driven read-side decode: BigQuery reverses its alias # mangling here so the response keys match SLayer's universal # dotted shape. Default hook is identity for every other dialect. - rows = get_dialect(dialect).decode_result_keys(rows) - columns = expected_columns if not rows else [] # fallback for empty results; [] triggers auto-derive - return SlayerResponse(data=rows, columns=columns, sql=sql, attributes=attributes) + rows = get_dialect(prepared.dialect).decode_result_keys(rows) + columns = prepared.expected_columns if not rows else [] # [] auto-derives + return SlayerResponse( + data=rows, + columns=columns, + sql=prepared.sql, + attributes=prepared.attributes, + ) + + # -- DEV-1587 query cache ------------------------------------------------- + + async def _execute_cached( + self, + *, + prepared: _Prepared, + client: SlayerSQLClient, + original_input: Any, + original_variables: dict[str, Any] | None, + original_data_source: str | None, + ) -> SlayerResponse: + """Serve from cache or run-and-store. On both hit and miss the caller + receives a deep copy so it can't mutate the cached response.""" + key = QueryCache.make_key(prepared.sql, prepared.ds_fingerprint) + hit = await self._cache.get(key) + if hit is not None: + return hit.response.model_copy(deep=True) + response, entry = await self._build_fresh_entry( + prepared=prepared, + client=client, + original_input=original_input, + original_variables=original_variables, + original_data_source=original_data_source, + ) + await self._cache.put(key, entry) + return response.model_copy(deep=True) + + async def _build_fresh_entry( + self, + *, + prepared: _Prepared, + client: SlayerSQLClient, + original_input: Any, + original_variables: dict[str, Any] | None, + original_data_source: str | None, + ) -> "tuple[SlayerResponse, _CacheEntry]": + """Run a cache miss: scan refresh-key baselines BEFORE the data query + (so cached data reflects a state >= the baseline — Codex finding #1), + execute, and build an entry holding a deep copy of the response. + + A write-time baseline-scan failure PROPAGATES (the + ``execute(cache=True)`` call fails); ``refresh()`` instead treats the + same failure as continue-on-error. + """ + applicable = self._cache.applicable_keys(prepared.sql, prepared.dialect) + refresh_key_values = await self._scan_refresh_keys( + applicable=applicable, + client=client, + dialect=prepared.dialect, + datasource=prepared.datasource, + ) + response = await self._run_and_build(prepared=prepared, client=client) + # Deep-copy the replay inputs (as we do the response): ``refresh()`` + # re-prepares from ``original_input`` / ``variables``, so a caller + # mutating their SlayerQuery / dict / list / variables dict after the + # write must not change what a later refresh re-executes. + entry = _CacheEntry( + response=response.model_copy(deep=True), + sql=prepared.sql, + ds_fingerprint=prepared.ds_fingerprint, + dialect=prepared.dialect, + ds_key=prepared.ds_key, + resolved_data_source=prepared.datasource.name, + original_input=copy.deepcopy(original_input), + variables=copy.deepcopy(original_variables), + data_source=original_data_source, + created_at=self._cache.now(), + applicable=applicable, + refresh_key_values=refresh_key_values, + ) + return response, entry + + async def _scan_refresh_keys( + self, + *, + applicable: list[tuple[str, str]], + client: SlayerSQLClient, + dialect: str, + datasource: DatasourceConfig, + ) -> list[RefreshKeyValue]: + """Evaluate every applicable refresh key, batched one scan per table. + Raises on a malformed / multi-row scan (the caller decides propagate + vs continue-on-error).""" + by_table: dict[str, list[str]] = {} + for table, expr in applicable: + by_table.setdefault(table, []).append(expr) + out: list[RefreshKeyValue] = [] + for table, exprs in by_table.items(): + row = await self._scan_one_table( + table=table, exprs=exprs, client=client, + dialect=dialect, datasource=datasource, + ) + for i, expr in enumerate(exprs): + out.append( + RefreshKeyValue( + table=table, + expression=expr, + value=row[self._cache.rk_alias(i)], + ) + ) + return out + + async def _scan_one_table( + self, + *, + table: str, + exprs: list[str], + client: SlayerSQLClient, + dialect: str, + datasource: DatasourceConfig, + ) -> dict[str, Any]: + scan_sql = self._cache.build_refresh_key_sql(table, exprs, dialect) + # DEV-1587 × DEV-1578: apply the forced-filter policy to the scan too, + # so the refresh-key baseline is computed over the SAME tenant-scoped + # rows as the cached data query. Without this, a global MAX/COUNT could + # mask a tenant-local change (or over-refresh on another tenant's). + # No-op (zero overhead) when no policy is configured. + scan_sql = self._apply_policy( + sql=scan_sql, dialect=dialect, datasource=datasource + ) + rows = await client.execute(sql=scan_sql) + if len(rows) != 1: + raise ValueError( + f"refresh-key scan for table '{table}' returned {len(rows)} " + f"rows (expected exactly 1)." + ) + return rows[0] + + async def evict( + self, + query: "SlayerQuery | dict | list[SlayerQuery | dict] | str", + *, + variables: dict[str, Any] | None = None, + data_source: str | None = None, + ) -> bool: + """Remove the cache entry for ``query`` (same input union as + ``execute``). Recomputes the SQL + datasource key via the prepare + pipeline — no data query and no ``SlayerSQLClient`` (a forced-filter + policy may still introspect column presence; see ``_prepare_pipeline``). + Returns ``True`` if an entry was present.""" + query_obj, named_queries, runtime_kwarg, prefer = await self._normalize_input( + query, variables=variables, data_source=data_source, + ) + prepared = await self._prepare_pipeline( + query=query_obj, + named_queries=named_queries, + runtime_kwarg=runtime_kwarg, + prefer_data_source=prefer, + ) + key = QueryCache.make_key(prepared.sql, prepared.ds_fingerprint) + return await self._cache.delete(key) + + def evict_sync( + self, + query: "SlayerQuery | dict | list[SlayerQuery | dict] | str", + *, + variables: dict[str, Any] | None = None, + data_source: str | None = None, + ) -> bool: + """Synchronous wrapper for :meth:`evict`.""" + from slayer.async_utils import run_sync + + return run_sync( + self.evict(query, variables=variables, data_source=data_source) + ) + + async def refresh(self) -> RefreshResult: + """Re-scan refresh keys + TTL for every cached entry and re-execute + the stale ones (Cube-style manual refresh). Continue-on-failure. + + All DB awaits happen OUTSIDE the cache lock; the entry set is + snapshotted before awaiting, and each re-execution commit is + identity-guarded so a concurrent ``evict`` / ``clear_cache`` / + newer ``execute`` is never clobbered or resurrected. + """ + snapshot = await self._cache.snapshot() + result = RefreshResult() + if not snapshot: + return result + fresh_values, failed_tables = await self._refresh_scan_all(snapshot, result) + for key, entry in snapshot.items(): + await self._refresh_one( + key=key, + entry=entry, + fresh_values=fresh_values, + failed_tables=failed_tables, + result=result, + ) + return result + + def refresh_sync(self) -> RefreshResult: + """Synchronous wrapper for :meth:`refresh`. + + Like ``execute_sync``, disposes per-call async engines in ``finally`` + (refresh runs refresh-key scans + re-executions) so they don't outlive + their owning loop. + """ + from slayer.async_utils import run_sync + + async def _run_and_cleanup() -> RefreshResult: + try: + return await self.refresh() + finally: + await self.aclose() + + return run_sync(_run_and_cleanup()) + + async def _refresh_scan_all( + self, snapshot: "dict[str, _CacheEntry]", result: RefreshResult, + ) -> "tuple[dict[tuple[tuple[str, str], str, str], Any], set]": + """Collate applicable ``(ds_key, table)`` scan targets across all + entries and run ONE batched scan per pair. Scan failures are recorded + as continue-on-error ``RefreshError(phase="refresh_key_scan")`` and the + table is marked failed so its dependent entries are left unchanged.""" + targets: dict[tuple[tuple[str, str], str], set] = {} + dialects: dict[tuple[str, str], str] = {} + clients: dict[tuple[str, str], SlayerSQLClient] = {} + for entry in snapshot.values(): + ds_key = tuple(entry.ds_key) + dialects[ds_key] = entry.dialect + for table, expr in entry.applicable: + targets.setdefault((ds_key, table), set()).add(expr) + + fresh: dict[tuple[tuple[str, str], str, str], Any] = {} + failed: set = set() + for (ds_key, table), exprs in targets.items(): + expr_list = sorted(exprs) + try: + client = await self._client_for_refresh(ds_key, snapshot, clients) + row = await self._scan_one_table( + table=table, exprs=expr_list, client=client, + dialect=dialects[ds_key], datasource=client.datasource, + ) + for i, expr in enumerate(expr_list): + fresh[(ds_key, table, expr)] = row[self._cache.rk_alias(i)] + except Exception as exc: # noqa: BLE001 — continue-on-failure per table + failed.add((ds_key, table)) + result.errors.append( + RefreshError(key=table, phase="refresh_key_scan", message=str(exc)) + ) + return fresh, failed + + async def _client_for_refresh( + self, + ds_key: tuple[str, str], + snapshot: "dict[str, _CacheEntry]", + clients: dict[tuple[str, str], SlayerSQLClient], + ) -> SlayerSQLClient: + """Get the cached client for ``ds_key``, reopening it from an entry's + recorded resolved datasource name if it isn't cached.""" + if ds_key in clients: + return clients[ds_key] + client = self._sql_clients.get(ds_key) + if client is None: + ds_name = next( + ( + e.resolved_data_source + for e in snapshot.values() + if tuple(e.ds_key) == ds_key and e.resolved_data_source + ), + None, + ) + ds = await self.storage.get_datasource(ds_name) if ds_name else None + if ds is None: + raise ValueError( + f"cannot resolve datasource for refresh (ds_key={ds_key})" + ) + client = self._get_client(ds, ds_key) + clients[ds_key] = client + return client + + async def _refresh_one( + self, + *, + key: str, + entry: _CacheEntry, + fresh_values: "dict[tuple[tuple[str, str], str, str], Any]", + failed_tables: set, + result: RefreshResult, + ) -> None: + """Decide + apply the per-entry refresh action: TTL-expired ⇒ re-exec + (``expired_refreshed``); else a scan failure on an applicable table ⇒ + keep unchanged; else any applicable refresh-key value moved ⇒ re-exec + (``refreshed``); else unchanged.""" + ds_key = tuple(entry.ds_key) + ttl = self._cache.config.ttl_seconds + if ttl is not None and (self._cache.now() - entry.created_at) > ttl: + await self._refresh_reexec( + key=key, entry=entry, bucket="expired_refreshed", result=result + ) + return + + entry_tables = {t for t, _ in entry.applicable} + if any((ds_key, t) in failed_tables for t in entry_tables): + result.unchanged.append(key) # scan failed → keep the stale entry + return + + moved = any( + QueryCache.values_differ( + rkv.value, fresh_values.get((ds_key, rkv.table, rkv.expression)) + ) + for rkv in entry.refresh_key_values + ) + if moved: + await self._refresh_reexec( + key=key, entry=entry, bucket="refreshed", result=result + ) + else: + result.unchanged.append(key) + + async def _refresh_reexec( + self, *, key: str, entry: _CacheEntry, bucket: str, result: RefreshResult, + ) -> None: + """Re-execute a stale entry and identity-guarded-commit the result. + On re-exec failure the stale entry is kept and the error recorded.""" + try: + new_key, new_entry = await self._reexecute_entry(entry, self._cache.now()) + except Exception as exc: # noqa: BLE001 — keep the stale entry on failure + result.errors.append( + RefreshError(key=key, phase="re_execute", message=str(exc)) + ) + return + committed = await self._cache.commit_replace( + old_key=key, expected=entry, new_key=new_key, new_entry=new_entry, + ) + if committed: + getattr(result, bucket).append(key) + + async def _reexecute_entry( + self, entry: _CacheEntry, now: float + ) -> "tuple[str, _CacheEntry]": + """Replay a stale entry from its ORIGINAL input through the full + ``execute`` normalization (so run-by-name picks up ``source_queries`` + edits and lists re-topo-sort), re-scanning baselines before the data + query. Returns the (possibly re-keyed) key and a fresh entry with a + new ``created_at`` (TTL reset — Codex finding #4). + + ``now`` is the refresh clock reading, retained for signature stability + (the fresh entry stamps ``created_at`` from the same injectable clock). + """ + del now # created_at is stamped inside _build_fresh_entry via the clock + # Pin the replay to the datasource this entry was ORIGINALLY resolved + # against (falling back to it when the caller passed no explicit + # data_source). A cache entry is bound to a resolved datasource + # identity (it's in the key), and refresh() already scanned this + # entry's refresh keys against that datasource — so re-executing must + # stay on it, not silently migrate to a different datasource if the + # priority list changed after the entry was cached. + replay_data_source = entry.data_source or entry.resolved_data_source + query_obj, named_queries, runtime_kwarg, prefer = await self._normalize_input( + entry.original_input, + variables=entry.variables, + data_source=replay_data_source, + ) + prepared = await self._prepare_pipeline( + query=query_obj, + named_queries=named_queries, + runtime_kwarg=runtime_kwarg, + prefer_data_source=prefer, + ) + client = self._get_client(prepared.datasource, prepared.ds_key) + _, new_entry = await self._build_fresh_entry( + prepared=prepared, + client=client, + original_input=entry.original_input, + original_variables=entry.variables, + original_data_source=entry.data_source, + ) + new_key = QueryCache.make_key(prepared.sql, prepared.ds_fingerprint) + return new_key, new_entry @staticmethod def _collect_query_backed_base_names(model: SlayerModel) -> "set[str]": @@ -1256,16 +1783,27 @@ def execute_sync( *, dry_run: bool = False, explain: bool = False, + data_source: str | None = None, + cache: bool = False, ) -> SlayerResponse: - """Synchronous wrapper for execute(). Disposes per-call async engines - in ``finally`` so they don't outlive their owning loop — see ``aclose``. + """Synchronous wrapper for execute(). For CLI, notebooks, and scripts. + + DEV-1587: forwards ``cache`` and ``data_source`` (the latter closes the + pre-existing sync ``data_source`` gap so sync cache / evict reach the + datasource-override paths). Disposes per-call async engines in + ``finally`` so they don't outlive their owning loop — see ``aclose``. """ from slayer.async_utils import run_sync async def _run_and_cleanup() -> SlayerResponse: try: return await self.execute( - query, variables=variables, dry_run=dry_run, explain=explain, + query, + variables=variables, + dry_run=dry_run, + explain=explain, + data_source=data_source, + cache=cache, ) finally: await self.aclose() diff --git a/tests/test_query_cache.py b/tests/test_query_cache.py new file mode 100644 index 00000000..4731737b --- /dev/null +++ b/tests/test_query_cache.py @@ -0,0 +1,864 @@ +"""DEV-1587 — per-engine in-memory query result cache. + +Two layers of coverage: + +* **Pure ``QueryCache``** (no DB): key stability / ds-fingerprint + sensitivity, TTL get/expiry with a fake clock, CTE-alias-excluding table + parsing, the wildcard table-matching rule across dialects, applicable + refresh-key selection, refresh-key scan-SQL building, value comparison, + identity-guarded commit, put/delete/size/clear. +* **Engine integration** (real in-process SQLite): opt-in ``cache=True`` + miss→hit (serves stale, no re-exec), TTL lazy re-exec, refresh-key + baseline-before-data ordering, ``refresh()`` buckets, multiple keys per + table, unreferenced-table no-op, re-prepare-from-raw-input + re-key, + continue-on-error, write-time baseline propagation, ``cache_config`` + reassignment clears, dry_run/explain bypass, mutation isolation, + per-engine isolation, multi-stage + run-by-name caching, evict/clear, + sync wrappers, evict-never-connects, identity-guarded refresh commit. +""" + +import asyncio +import sqlite3 + +import pydantic +import pytest +import sqlalchemy + +from slayer.core.enums import DataType +from slayer.core.models import Column, DatasourceConfig, ModelMeasure, SlayerModel +from slayer.core.policy import ColumnFilterRule, SessionPolicy +from slayer.core.query import ColumnRef, SlayerQuery +from slayer.engine.cache import ( + CacheConfig, + QueryCache, + RefreshResult, + _CacheEntry, +) +from slayer.engine.query_engine import SlayerQueryEngine, _sql_client_cache_key +from slayer.sql.client import SlayerSQLClient +from slayer.storage.yaml_storage import YAMLStorage + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +class FakeClock: + """Deterministic monotonic clock for TTL tests.""" + + def __init__(self, start: float = 1000.0) -> None: + self.t = start + + def __call__(self) -> float: + return self.t + + def advance(self, dt: float) -> None: + self.t += dt + + +def _make_entry(*, created_at: float = 0.0, sql: str = "SELECT 1", response=None) -> _CacheEntry: + return _CacheEntry( + response=response, + sql=sql, + ds_fingerprint="fp", + dialect="sqlite", + ds_key=("conn", "rt"), + created_at=created_at, + ) + + +def _seed_db(db_path) -> None: + conn = sqlite3.connect(str(db_path)) + cur = conn.cursor() + cur.execute( + """ + CREATE TABLE orders ( + id INTEGER PRIMARY KEY, + status TEXT NOT NULL, + amount REAL NOT NULL, + updated_at TEXT NOT NULL + ) + """ + ) + cur.executemany( + "INSERT INTO orders VALUES (?, ?, ?, ?)", + [ + (1, "completed", 100.0, "2025-01-01"), + (2, "pending", 50.0, "2025-01-02"), + (3, "completed", 200.0, "2025-01-03"), + ], + ) + # A second physical table used only by the "unreferenced-table" test. + cur.execute("CREATE TABLE aux (id INTEGER PRIMARY KEY, n INTEGER NOT NULL)") + cur.executemany("INSERT INTO aux VALUES (?, ?)", [(1, 10), (2, 20)]) + conn.commit() + conn.close() + + +def _seed_orders_single(db_path, *, amount: float) -> None: + """A second datasource's ``orders`` table with a single distinguishable row.""" + conn = sqlite3.connect(str(db_path)) + conn.execute( + "CREATE TABLE orders (id INTEGER PRIMARY KEY, status TEXT NOT NULL, " + "amount REAL NOT NULL, updated_at TEXT NOT NULL)" + ) + conn.execute( + "INSERT INTO orders VALUES (1, 'completed', ?, '2025-01-01')", (amount,) + ) + conn.commit() + conn.close() + + +def _orders_model(ds: str = "ds") -> SlayerModel: + return SlayerModel( + name="orders", + sql_table="orders", + data_source=ds, + columns=[ + Column(name="id", sql="id", type=DataType.DOUBLE, primary_key=True), + Column(name="status", sql="status", type=DataType.TEXT), + Column(name="amount", sql="amount", type=DataType.DOUBLE), + Column(name="updated_at", sql="updated_at", type=DataType.TIMESTAMP), + ], + ) + + +async def _build_engine(tmp_path, *, cache_config=None, ds_name="ds", storage_suffix=""): + db = tmp_path / "orders.db" + if not db.exists(): + _seed_db(db) + sdir = tmp_path / f"storage{storage_suffix}" + sdir.mkdir(exist_ok=True) + storage = YAMLStorage(base_dir=str(sdir)) + await storage.save_datasource( + DatasourceConfig(name=ds_name, type="sqlite", database=str(db)) + ) + await storage.save_model(_orders_model(ds_name)) + return SlayerQueryEngine(storage=storage, cache_config=cache_config) + + +def _sum_query() -> SlayerQuery: + return SlayerQuery(source_model="orders", measures=[ModelMeasure(formula="amount:sum")]) + + +def _mutate(db_path, sql: str) -> None: + conn = sqlite3.connect(str(db_path)) + conn.execute(sql) + conn.commit() + conn.close() + + +def _install_spy(monkeypatch): + """Record every SQL string sent through ``SlayerSQLClient.execute``.""" + calls: list[str] = [] + orig = SlayerSQLClient.execute + + async def spy(self, sql, timeout_seconds=120): + calls.append(sql) + return await orig(self, sql=sql, timeout_seconds=timeout_seconds) + + monkeypatch.setattr(SlayerSQLClient, "execute", spy) + return calls + + +def _data_queries(calls): + return [s for s in calls if "slayer_rk_" not in s] + + +def _scan_queries(calls): + return [s for s in calls if "slayer_rk_" in s] + + +def _fingerprint(ds: DatasourceConfig) -> str: + return "|".join(_sql_client_cache_key(ds)) + + +# =========================================================================== +# Part A — pure QueryCache (no DB) +# =========================================================================== + + +class TestKeyAndClock: + def test_make_key_stable_and_ds_sensitive(self): + k1 = QueryCache.make_key("SELECT 1", "fpA") + k2 = QueryCache.make_key("SELECT 1", "fpA") + k3 = QueryCache.make_key("SELECT 1", "fpB") + k4 = QueryCache.make_key("SELECT 2", "fpA") + assert k1 == k2 + assert k1 != k3 # different datasource fingerprint + assert k1 != k4 # different SQL + + async def test_ttl_expiry_with_fake_clock(self): + clk = FakeClock() + c = QueryCache(CacheConfig(ttl_seconds=100), clock=clk) + e = _make_entry(created_at=c.now()) + await c.put("k", e) + clk.advance(50) + assert (await c.get("k")) is e + clk.advance(60) # age 110 > 100 + assert (await c.get("k")) is None + assert c.size() == 0 # expired entry dropped + + async def test_ttl_none_never_expires(self): + clk = FakeClock() + c = QueryCache(CacheConfig(ttl_seconds=None), clock=clk) + e = _make_entry(created_at=c.now()) + await c.put("k", e) + clk.advance(10_000_000) + assert (await c.get("k")) is e + + async def test_get_miss_returns_none(self): + c = QueryCache(CacheConfig()) + assert (await c.get("nope")) is None + + async def test_put_delete_size_clear(self): + c = QueryCache(CacheConfig()) + await c.put("a", _make_entry()) + await c.put("b", _make_entry()) + assert c.size() == 2 + assert (await c.delete("a")) is True + assert (await c.delete("a")) is False + assert c.size() == 1 + c.clear() + assert c.size() == 0 + + +class TestCommitReplace: + async def test_identity_guard_and_rekey(self): + c = QueryCache(CacheConfig()) + e1 = _make_entry(sql="a") + e2 = _make_entry(sql="a2") + await c.put("k", e1) + assert await c.commit_replace(old_key="k", expected=e1, new_key="k", new_entry=e2) is True + assert c.size() == 1 + # Live is now e2; a stale expected pointer must be refused. + e3 = _make_entry(sql="a3") + assert await c.commit_replace(old_key="k", expected=e1, new_key="k", new_entry=e3) is False + # Re-key move. + e4 = _make_entry(sql="b") + assert await c.commit_replace(old_key="k", expected=e2, new_key="k2", new_entry=e4) is True + assert c.size() == 1 + assert (await c.get("k2")) is e4 + assert (await c.get("k")) is None + + +class TestTableParsing: + def test_excludes_cte_and_subquery_aliases(self): + c = QueryCache(CacheConfig()) + sql = ( + "WITH cte AS (SELECT id FROM public.orders) " + "SELECT * FROM cte JOIN (SELECT 1 AS x) sub ON TRUE" + ) + names = [t[2] for t in c.parse_referenced_tables(sql, "postgres")] + assert "orders" in names + assert "cte" not in names + assert "sub" not in names + + def test_values_clause_has_no_physical_table(self): + c = QueryCache(CacheConfig()) + names = [t[2] for t in c.parse_referenced_tables( + "SELECT * FROM (VALUES (1), (2)) AS v(x)", "postgres" + )] + assert "v" not in names + + def test_qualified_table_sharing_cte_name_is_still_physical(self): + # A schema-qualified physical table (`public.cte`) sharing a CTE's bare + # name must still be detected — a qualified ref can never be a CTE. + c = QueryCache(CacheConfig()) + tables = c.parse_referenced_tables( + "WITH cte AS (SELECT 1 AS x) SELECT * FROM cte JOIN public.cte ON TRUE", + "postgres", + ) + assert (None, "public", "cte") in tables # the physical public.cte + # And a refresh key configured on that physical table applies. + c2 = QueryCache(CacheConfig(refresh_keys=[("public.cte", "COUNT(*)")])) + assert c2.applicable_keys( + "WITH cte AS (SELECT 1 AS x) SELECT * FROM cte JOIN public.cte ON TRUE", + "postgres", + ) == [("public.cte", "COUNT(*)")] + + +class TestWildcardMatching: + def test_unqualified_config_matches_any_qualifier_pg(self): + c = QueryCache(CacheConfig(refresh_keys=[("orders", "COUNT(*)")])) + for frm in ("orders", "public.orders", "db.public.orders"): + assert c.applicable_keys(f"SELECT * FROM {frm}", "postgres") == [ + ("orders", "COUNT(*)") + ] + + def test_schema_qualified_config_requires_schema_pg(self): + c = QueryCache(CacheConfig(refresh_keys=[("public.orders", "COUNT(*)")])) + assert c.applicable_keys("SELECT * FROM public.orders", "postgres") == [ + ("public.orders", "COUNT(*)") + ] + assert c.applicable_keys("SELECT * FROM other.orders", "postgres") == [] + assert c.applicable_keys("SELECT * FROM orders", "postgres") == [] + + def test_snowflake_unquoted_folds_upper(self): + c = QueryCache(CacheConfig(refresh_keys=[("orders", "COUNT(*)")])) + assert c.applicable_keys("SELECT * FROM orders", "snowflake") == [ + ("orders", "COUNT(*)") + ] + # Quoted lowercase is preserved (orders != ORDERS) → no match. + assert c.applicable_keys('SELECT * FROM "orders"', "snowflake") == [] + + def test_quoted_config_preserved_exactly(self): + c = QueryCache(CacheConfig(refresh_keys=[('"Orders"', "COUNT(*)")])) + assert c.applicable_keys('SELECT * FROM "Orders"', "snowflake") == [ + ('"Orders"', "COUNT(*)") + ] + assert c.applicable_keys("SELECT * FROM orders", "snowflake") == [] + + def test_bigquery_dotted_paths(self): + c = QueryCache(CacheConfig(refresh_keys=[("dataset.orders", "COUNT(*)")])) + assert c.applicable_keys( + "SELECT * FROM `project`.`dataset`.`orders`", "bigquery" + ) == [("dataset.orders", "COUNT(*)")] + assert c.applicable_keys( + "SELECT * FROM `project`.`other`.`orders`", "bigquery" + ) == [] + + def test_sqlserver_schema_qualified(self): + c = QueryCache(CacheConfig(refresh_keys=[("dbo.orders", "COUNT(*)")])) + assert c.applicable_keys("SELECT * FROM dbo.orders", "tsql") == [ + ("dbo.orders", "COUNT(*)") + ] + + def test_applicable_keeps_duplicate_expressions(self): + c = QueryCache( + CacheConfig( + refresh_keys=[("orders", "MAX(updated_at)"), ("orders", "COUNT(*)")] + ) + ) + assert c.applicable_keys("SELECT * FROM orders", "postgres") == [ + ("orders", "MAX(updated_at)"), + ("orders", "COUNT(*)"), + ] + + def test_unreferenced_table_not_applicable(self): + c = QueryCache( + CacheConfig(refresh_keys=[("orders", "COUNT(*)"), ("aux", "COUNT(*)")]) + ) + assert c.applicable_keys("SELECT * FROM orders", "postgres") == [ + ("orders", "COUNT(*)") + ] + + +class TestScanSqlAndValues: + def test_build_scan_sql_postgres(self): + c = QueryCache(CacheConfig()) + sql = c.build_refresh_key_sql( + "public.orders", ["MAX(updated_at)", "COUNT(*)"], "postgres" + ) + assert 'AS "slayer_rk_0"' in sql + assert 'AS "slayer_rk_1"' in sql + assert "MAX(updated_at)" in sql + assert "COUNT(*)" in sql + assert "public.orders" in sql + + def test_build_scan_sql_mysql_backticks(self): + c = QueryCache(CacheConfig()) + sql = c.build_refresh_key_sql("orders", ["COUNT(*)"], "mysql") + assert "`slayer_rk_0`" in sql + + def test_values_differ(self): + assert QueryCache.values_differ(1, 2) is True + assert QueryCache.values_differ(1, 1) is False + assert QueryCache.values_differ("2025-01-03", "2025-01-04") is True + assert QueryCache.values_differ(3, 2) is True # detects a decrease + + +# =========================================================================== +# Part B — engine integration (real in-process SQLite) +# =========================================================================== + + +class TestBasicCaching: + async def test_miss_then_hit_serves_stale_without_reexec(self, tmp_path, monkeypatch): + engine = await _build_engine(tmp_path) + calls = _install_spy(monkeypatch) + q = _sum_query() + + r1 = await engine.execute(q, cache=True) + assert r1.data[0]["orders.amount_sum"] == pytest.approx(350.0) + assert engine.cache_size == 1 + assert len(_data_queries(calls)) == 1 # the miss ran one data query + + calls.clear() + _mutate(tmp_path / "orders.db", "INSERT INTO orders VALUES (4, 'pending', 1000.0, '2025-02-01')") + + r2 = await engine.execute(q, cache=True) + assert r2.data[0]["orders.amount_sum"] == pytest.approx(350.0) # stale + assert _data_queries(calls) == [] # hit → no DB execution + + r3 = await engine.execute(q, cache=False) + assert r3.data[0]["orders.amount_sum"] == pytest.approx(1350.0) # bypass → fresh + assert len(_data_queries(calls)) == 1 + + async def test_dry_run_and_explain_never_cached(self, tmp_path): + engine = await _build_engine(tmp_path) + q = _sum_query() + await engine.execute(q, cache=True, dry_run=True) + assert engine.cache_size == 0 + try: + await engine.execute(q, cache=True, explain=True) + except Exception: + pass # some dialects can't EXPLAIN; the point is nothing is cached + assert engine.cache_size == 0 + + async def test_returned_response_is_independent_copy(self, tmp_path): + engine = await _build_engine(tmp_path) + q = _sum_query() + r1 = await engine.execute(q, cache=True) + r1.data[0]["orders.amount_sum"] = 999999.0 + r1.data.append({"orders.amount_sum": 0.0}) + r2 = await engine.execute(q, cache=True) # hit + assert r2.data[0]["orders.amount_sum"] == pytest.approx(350.0) + assert len(r2.data) == 1 + + async def test_per_engine_isolation(self, tmp_path): + engine_a = await _build_engine(tmp_path, storage_suffix="_a") + engine_b = await _build_engine(tmp_path, storage_suffix="_b") + await engine_a.execute(_sum_query(), cache=True) + assert engine_a.cache_size == 1 + assert engine_b.cache_size == 0 + + async def test_cache_config_reassign_clears(self, tmp_path): + engine = await _build_engine(tmp_path) + await engine.execute(_sum_query(), cache=True) + assert engine.cache_size == 1 + engine.cache_config = CacheConfig(ttl_seconds=5) + assert engine.cache_size == 0 + assert engine.cache_config.ttl_seconds == 5 + + +class TestTtlLazyReexec: + async def test_ttl_lazy_reexec_on_read(self, tmp_path, monkeypatch): + clk = FakeClock() + engine = await _build_engine(tmp_path) + engine._cache = QueryCache(config=CacheConfig(ttl_seconds=100), clock=clk) + q = _sum_query() + + await engine.execute(q, cache=True) # populate at t=1000 + calls = _install_spy(monkeypatch) + + clk.advance(50) + await engine.execute(q, cache=True) # still fresh → hit + assert _data_queries(calls) == [] + + clk.advance(60) # age 110 > 100 → expired → re-exec + _mutate(tmp_path / "orders.db", "INSERT INTO orders VALUES (4, 'pending', 1000.0, '2025-02-01')") + r = await engine.execute(q, cache=True) + assert len(_data_queries(calls)) == 1 + assert r.data[0]["orders.amount_sum"] == pytest.approx(1350.0) + + +class TestRefresh: + async def test_baseline_scanned_before_data_query(self, tmp_path, monkeypatch): + engine = await _build_engine( + tmp_path, + cache_config=CacheConfig(refresh_keys=[("orders", "MAX(updated_at)")]), + ) + # Warm up the client (no cache) so the spy sees the cached miss cleanly. + await engine.execute(_sum_query(), cache=False) + calls = _install_spy(monkeypatch) + + await engine.execute(_sum_query(), cache=True) + # First recorded query is the refresh-key baseline scan; the data + # query follows. + assert "slayer_rk_" in calls[0] + assert "slayer_rk_" not in calls[1] + + async def test_refresh_key_change_refreshes(self, tmp_path): + engine = await _build_engine( + tmp_path, + cache_config=CacheConfig(refresh_keys=[("orders", "MAX(updated_at)")]), + ) + q = _sum_query() + await engine.execute(q, cache=True) + + # A later-timestamped insert moves MAX(updated_at). + _mutate(tmp_path / "orders.db", "INSERT INTO orders VALUES (4, 'pending', 1000.0, '2025-06-01')") + result = await engine.refresh() + assert isinstance(result, RefreshResult) + assert len(result.refreshed) == 1 + assert result.expired_refreshed == [] + assert result.unchanged == [] + assert result.errors == [] + + hit = await engine.execute(q, cache=True) + assert hit.data[0]["orders.amount_sum"] == pytest.approx(1350.0) + + async def test_refresh_unchanged_when_key_static(self, tmp_path): + engine = await _build_engine( + tmp_path, + cache_config=CacheConfig(refresh_keys=[("orders", "MAX(updated_at)")]), + ) + q = _sum_query() + await engine.execute(q, cache=True) + result = await engine.refresh() + assert len(result.unchanged) == 1 + assert result.refreshed == [] + assert result.expired_refreshed == [] + + async def test_count_key_catches_delete_that_max_misses(self, tmp_path): + engine = await _build_engine( + tmp_path, + cache_config=CacheConfig( + refresh_keys=[("orders", "MAX(updated_at)"), ("orders", "COUNT(*)")] + ), + ) + q = _sum_query() + await engine.execute(q, cache=True) + + # Delete a NON-max row: MAX(updated_at) stays 2025-01-03, COUNT drops. + _mutate(tmp_path / "orders.db", "DELETE FROM orders WHERE id = 2") + result = await engine.refresh() + assert len(result.refreshed) == 1 # COUNT(*) moved even though MAX didn't + hit = await engine.execute(q, cache=True) + assert hit.data[0]["orders.amount_sum"] == pytest.approx(300.0) + + async def test_unreferenced_table_change_no_refresh(self, tmp_path): + engine = await _build_engine( + tmp_path, + cache_config=CacheConfig( + refresh_keys=[("orders", "MAX(updated_at)"), ("aux", "COUNT(*)")] + ), + ) + q = _sum_query() # references orders only + await engine.execute(q, cache=True) + _mutate(tmp_path / "orders.db", "INSERT INTO aux VALUES (3, 30)") + result = await engine.refresh() + assert len(result.unchanged) == 1 + assert result.refreshed == [] + + async def test_refresh_reprepares_and_rekeys_on_model_edit(self, tmp_path): + clk = FakeClock() + engine = await _build_engine(tmp_path) + engine._cache = QueryCache(config=CacheConfig(ttl_seconds=100), clock=clk) + q = _sum_query() + + ds = await engine.storage.get_datasource("ds") + fp = _fingerprint(ds) + old_sql = (await engine.execute(q, dry_run=True)).sql + old_key = QueryCache.make_key(old_sql, fp) + + await engine.execute(q, cache=True) + assert old_key in engine._cache._entries + + # Edit the stored model: an always-applied filter changes the SQL. + edited = _orders_model("ds").model_copy(update={"filters": ["amount > 60"]}) + await engine.storage.save_model(edited) + new_sql = (await engine.execute(q, dry_run=True)).sql + new_key = QueryCache.make_key(new_sql, fp) + assert new_key != old_key + + clk.advance(200) # TTL expired → refresh re-preps from raw input + result = await engine.refresh() + assert old_key in result.expired_refreshed + assert engine.cache_size == 1 + assert new_key in engine._cache._entries + assert old_key not in engine._cache._entries + + async def test_refresh_continue_on_scan_error_keeps_entry(self, tmp_path): + engine = await _build_engine( + tmp_path, + cache_config=CacheConfig(refresh_keys=[("orders", "MAX(updated_at)")]), + ) + q = _sum_query() + await engine.execute(q, cache=True) # baseline captured while table exists + assert engine.cache_size == 1 + + # Drop the table: the refresh-key scan now fails. + _mutate(tmp_path / "orders.db", "DROP TABLE orders") + result = await engine.refresh() + assert any(e.phase == "refresh_key_scan" for e in result.errors) + assert len(result.unchanged) == 1 # entry kept, not re-executed + assert engine.cache_size == 1 + # The stale value is still served from cache (no DB touched). + hit = await engine.execute(q, cache=True) + assert hit.data[0]["orders.amount_sum"] == pytest.approx(350.0) + + async def test_ttl_expired_with_scan_failure_keeps_entry(self, tmp_path): + # TTL expiry takes precedence over the refresh-key scan, so the entry + # is re-executed; when re-execution's own baseline scan then fails + # (table gone), the stale entry is kept via the re_execute error path. + clk = FakeClock() + engine = await _build_engine(tmp_path) + engine._cache = QueryCache( + config=CacheConfig( + ttl_seconds=100, refresh_keys=[("orders", "MAX(updated_at)")] + ), + clock=clk, + ) + q = _sum_query() + await engine.execute(q, cache=True) + + _mutate(tmp_path / "orders.db", "DROP TABLE orders") + clk.advance(200) # TTL expired → re-exec attempted → its scan fails + result = await engine.refresh() + assert any(e.phase == "re_execute" for e in result.errors) + assert result.expired_refreshed == [] + assert engine.cache_size == 1 # stale entry kept, not evicted by refresh + + async def test_refresh_uses_snapshot_of_mutated_input(self, tmp_path): + # The cached entry deep-copies the original input, so mutating the + # caller's SlayerQuery after the write must not change refresh replay. + clk = FakeClock() + engine = await _build_engine(tmp_path) + engine._cache = QueryCache(config=CacheConfig(ttl_seconds=100), clock=clk) + q = _sum_query() + await engine.execute(q, cache=True) + + ds = await engine.storage.get_datasource("ds") + fp = _fingerprint(ds) + original_sql = (await engine.execute(q, dry_run=True)).sql + original_key = QueryCache.make_key(original_sql, fp) + + # Mutate the caller's query object in place. + q.filters = ["amount > 60"] + + clk.advance(200) # TTL expired → refresh replays the stored copy + await engine.refresh() + # Replay used the pre-mutation input → same key, not the mutated one. + assert original_key in engine._cache._entries + assert engine.cache_size == 1 + + async def test_write_time_baseline_failure_propagates(self, tmp_path): + engine = await _build_engine( + tmp_path, + cache_config=CacheConfig(refresh_keys=[("orders", "MAX(nonexistent_col)")]), + ) + q = _sum_query() + with pytest.raises(sqlalchemy.exc.SQLAlchemyError): + await engine.execute(q, cache=True) + assert engine.cache_size == 0 # nothing stored on a failed baseline scan + + async def test_identity_guarded_commit_not_resurrected(self, tmp_path): + clk = FakeClock() + engine = await _build_engine(tmp_path) + engine._cache = QueryCache(config=CacheConfig(ttl_seconds=100), clock=clk) + q = _sum_query() + await engine.execute(q, cache=True) + + orig_reexec = engine._reexecute_entry + + async def clearing_reexec(entry, now): + out = await orig_reexec(entry, now) + engine.clear_cache() # a concurrent clear during refresh's awaits + return out + + engine._reexecute_entry = clearing_reexec + + clk.advance(200) # TTL expired → re-exec path + await engine.refresh() + # The entry was cleared mid-refresh; the identity guard must NOT + # resurrect it under the new key. + assert engine.cache_size == 0 + + +class TestMultiStageAndByName: + async def test_multi_stage_list_caching(self, tmp_path, monkeypatch): + engine = await _build_engine(tmp_path) + inner = SlayerQuery( + source_model="orders", + name="raw", + dimensions=[ColumnRef(name="status"), ColumnRef(name="amount")], + distinct_dimension_values=False, + ) + outer = SlayerQuery( + source_model="raw", measures=[ModelMeasure(formula="amount:sum")] + ) + r1 = await engine.execute([inner, outer], cache=True) + assert engine.cache_size == 1 + total = r1.data[0]["raw.amount_sum"] + assert total == pytest.approx(350.0) + + calls = _install_spy(monkeypatch) + r2 = await engine.execute([inner, outer], cache=True) # hit + assert _data_queries(calls) == [] + assert r2.data[0]["raw.amount_sum"] == total + + async def test_run_by_name_caching(self, tmp_path, monkeypatch): + engine = await _build_engine(tmp_path) + await engine.create_model_from_query( + query=SlayerQuery( + source_model="orders", + measures=[ModelMeasure(formula="amount:sum")], + dimensions=[ColumnRef(name="status")], + ), + name="orders_by_status", + ) + r1 = await engine.execute("orders_by_status", cache=True) + assert engine.cache_size == 1 + + calls = _install_spy(monkeypatch) + r2 = await engine.execute("orders_by_status", cache=True) # hit + assert _data_queries(calls) == [] + assert len(r2.data) == len(r1.data) + + async def test_refresh_by_name_picks_up_source_queries_edit(self, tmp_path): + clk = FakeClock() + engine = await _build_engine(tmp_path) + engine._cache = QueryCache(config=CacheConfig(ttl_seconds=100), clock=clk) + await engine.create_model_from_query( + query=SlayerQuery( + source_model="orders", + measures=[ModelMeasure(formula="amount:sum")], + dimensions=[ColumnRef(name="status")], + ), + name="obs", + ) + await engine.execute("obs", cache=True) + assert engine.cache_size == 1 + + # Edit the backing source_queries: filter out the 'pending' rows. + await engine.save_model( + SlayerModel( + name="obs", + source_queries=[ + SlayerQuery( + source_model="orders", + measures=[ModelMeasure(formula="amount:sum")], + dimensions=[ColumnRef(name="status")], + filters=["status != 'pending'"], + ) + ], + ) + ) + clk.advance(200) + result = await engine.refresh() + assert len(result.expired_refreshed) == 1 + assert engine.cache_size == 1 + hit = await engine.execute("obs", cache=True) + status_key = next(k for k in hit.data[0] if k.endswith(".status")) + statuses = {row[status_key] for row in hit.data} + assert "pending" not in statuses + + +class TestEvictAndManagement: + async def test_evict_and_clear_and_size(self, tmp_path): + engine = await _build_engine(tmp_path) + q1 = _sum_query() + q2 = SlayerQuery( + source_model="orders", + measures=[ModelMeasure(formula="amount:sum")], + dimensions=[ColumnRef(name="status")], + ) + await engine.execute(q1, cache=True) + await engine.execute(q2, cache=True) + assert engine.cache_size == 2 + assert (await engine.evict(q1)) is True + assert engine.cache_size == 1 + assert (await engine.evict(q1)) is False # already gone + engine.clear_cache() + assert engine.cache_size == 0 + + async def test_evict_never_constructs_a_client(self, tmp_path, monkeypatch): + engine = await _build_engine(tmp_path) + + class ExplodingClient: + def __init__(self, *a, **k): + raise AssertionError("evict must not construct a SQL client") + + monkeypatch.setattr("slayer.engine.query_engine.SlayerSQLClient", ExplodingClient) + # Nothing cached, no client cached → evict resolves the key (DB-free) + # and finds no entry, all without connecting. + assert (await engine.evict(_sum_query())) is False + + +class TestSyncWrappers: + def test_execute_sync_evict_sync_refresh_sync(self, tmp_path): + engine = asyncio.run( + _build_engine( + tmp_path, + cache_config=CacheConfig(refresh_keys=[("orders", "MAX(updated_at)")]), + ) + ) + q = _sum_query() + r = engine.execute_sync(q, cache=True, data_source="ds") + assert r.data[0]["orders.amount_sum"] == pytest.approx(350.0) + assert engine.cache_size == 1 + result = engine.refresh_sync() + assert isinstance(result, RefreshResult) + assert len(result.unchanged) == 1 + assert engine.evict_sync(q, data_source="ds") is True + assert engine.cache_size == 0 + + +class TestCacheConfigFrozen: + def test_frozen_and_refresh_keys_coerced_to_tuple(self): + cfg = CacheConfig(ttl_seconds=5, refresh_keys=[("orders", "COUNT(*)")]) + assert isinstance(cfg.refresh_keys, tuple) + assert cfg.refresh_keys == (("orders", "COUNT(*)"),) + assert CacheConfig().refresh_keys == () + # Frozen: in-place mutation is rejected (must reassign cache_config, + # which clears the cache) rather than silently bypassing the setter. + with pytest.raises(pydantic.ValidationError): + cfg.ttl_seconds = 10 + + +class TestPolicyInteraction: + async def test_refresh_key_scan_is_policy_scoped(self, tmp_path, monkeypatch): + # DEV-1587 × DEV-1578: with a forced-filter policy the refresh-key + # baseline scan must be rewritten to the same tenant scope as the data + # query, so a global MAX/COUNT can't mask a tenant-local change. + engine = await _build_engine( + tmp_path, + cache_config=CacheConfig(refresh_keys=[("orders", "MAX(updated_at)")]), + ) + engine.policy = SessionPolicy( + data_filters=[ColumnFilterRule(column="status", value="completed")] + ) + calls = _install_spy(monkeypatch) + await engine.execute(_sum_query(), cache=True) + + scans = _scan_queries(calls) + assert scans, "expected a refresh-key scan on the cache miss" + # Both the scan AND the data query are tenant-scoped (reference the + # policy column), so their notions of "current state" agree. + assert all("status" in s for s in scans) + assert all("status" in s for s in _data_queries(calls)) + + +class TestDatasourceResolution: + async def test_refresh_pins_to_original_resolved_datasource(self, tmp_path): + # An entry cached via implicit (priority-resolved) datasource must stay + # bound to the datasource it originally resolved to. If the priority + # list changes after caching, refresh() must NOT migrate the entry to a + # different datasource — it scanned the old one's refresh keys. + db_a = tmp_path / "db_a.db" + db_b = tmp_path / "db_b.db" + _seed_db(db_a) # amounts sum to 350 + _seed_orders_single(db_b, amount=999.0) # sum 999 + + storage = YAMLStorage(base_dir=str(tmp_path / "store")) + await storage.save_datasource( + DatasourceConfig(name="db_a", type="sqlite", database=str(db_a)) + ) + await storage.save_datasource( + DatasourceConfig(name="db_b", type="sqlite", database=str(db_b)) + ) + await storage.save_model(_orders_model("db_a")) + await storage.save_model(_orders_model("db_b")) + await storage.set_datasource_priority(["db_a", "db_b"]) + + clk = FakeClock() + engine = SlayerQueryEngine(storage=storage) + engine._cache = QueryCache(config=CacheConfig(ttl_seconds=100), clock=clk) + + r = await engine.execute(_sum_query(), cache=True) # implicit → db_a + assert r.data[0]["orders.amount_sum"] == pytest.approx(350.0) + (entry,) = engine._cache._entries.values() + assert entry.resolved_data_source == "db_a" + + # Flip the priority, then TTL-expire and refresh. + await storage.set_datasource_priority(["db_b", "db_a"]) + clk.advance(200) + await engine.refresh() + + entries = list(engine._cache._entries.values()) + assert len(entries) == 1 + assert entries[0].resolved_data_source == "db_a" # not migrated to db_b + assert entries[0].response.data[0]["orders.amount_sum"] == pytest.approx(350.0) diff --git a/zensical.toml b/zensical.toml index 0e227bac..1f47f31d 100644 --- a/zensical.toml +++ b/zensical.toml @@ -32,6 +32,7 @@ nav = [ { "References (SQL vs DSL)" = "concepts/references.md" }, { "Auto-Ingestion" = "concepts/ingestion.md" }, { "Schema Drift" = "concepts/schema-drift.md" }, + { "Query Cache" = "concepts/query-cache.md" }, { "Memories" = "concepts/memories.md" }, { "Search" = "concepts/search.md" }, { "Row-Level Security" = "concepts/row-level-security.md" }, @@ -92,6 +93,9 @@ nav = [ "examples/11_dbt_metricflow/dbt_metricflow.md", { "Notebook" = "examples/11_dbt_metricflow/dbt_metricflow_nb.ipynb" }, ]}, + { "Query Cache" = [ + { "Notebook" = "examples/12_query_cache/query_cache_nb.ipynb" }, + ]}, { "Schema Drift (worked example)" = "examples/schema-drift.md" }, ]}, { "Configuration" = [