Decorator-first docs and demo; fix review nits (re-land #2)

DESIGN.md

@@ -209,6 +209,11 @@ write-through for callers that need `.sql()` on the entry immediately.
 
 ## MaterializedFrame
 
+First cut implements `.arrow()`, `.df()`, `.to_pylist()`, `len()`, and
+`.sql()` (server-side; requires a persisted entry — on a fresh
+write-behind miss it raises until the background persist lands). The
+rest of this section is the target design.
+
 One interface, three backings (inline-payload, remote, local-after-miss);
 calling code never branches.
 
@@ -237,18 +242,17 @@ frame.meta                       # fingerprint, database_id, created_at, expires
 
 ## Decorator API
 
+Implemented (first cut):
+
 ```python
-from hotdata_materialized import materialize, Vector
+from hotdata_materialized import materialize
 
 @materialize(
     key="daily-signups",          # label; fingerprint still includes args
-    ttl=3600,                     # seconds; None = manual invalidation only
+    ttl=3600,                     # seconds; None = never expires
     version=2,                    # bump to bust on code changes
-    mode="swr",                   # "swr" (default) | "strict"
     background=True,              # write-behind (default) | write-through
-    index=Vector(column="notes", provider="openai", metric="cosine"),
-    invalidate_on=[Event],        # opt-in signal-based staleness marking
-    on_error="fallback",          # "fallback" (default) | "raise"
+    key_fn=None,                  # stable identity for opaque arguments
 )
 def daily_signups(start, end):
     return (Event.objects
@@ -257,12 +261,11 @@ def daily_signups(start, end):
             .annotate(n=Count("id")))
 ```
 
-Also usable imperatively for one-off QuerySets:
-
-```python
-from hotdata_materialized import from_queryset
-frame = from_queryset(qs, ttl=900)
-```
+Planned knobs (steps 4–5, do not document user-facing until they exist):
+`mode="swr" | "strict"` (refresh behavior), `invalidate_on=[Model]`
+(signal-based staleness), `index=Vector(...)/BM25(...)` (search indexes),
+`on_error="fallback" | "raise"` (fail-open is currently always on), and a
+`from_queryset(qs, ttl=)` imperative helper for one-off QuerySets.
 
 ## Fingerprinting
 
@@ -350,6 +353,7 @@ synchronous — fine for WSGI; async views get `sync_to_async` wrappers
    load → registry upsert; delete path). Pure SDK composition.
 2. **`@materialize` + `MaterializedFrame`** with the three backings,
    `.df()/.arrow()/.pl()/.sql()`, inline-payload short-circuit, fail-open.
+   *(shipped: first cut — arrow/df/to_pylist/sql, fail-open)*
 3. **Chainable facade:** immutable lazy queryset compiler over the entry
    table (Django lookup syntax).
 4. **Freshness machinery:** SWR, building-row lock, sweep management command +

README.md

@@ -84,7 +84,9 @@ way you get a `MaterializedFrame`:
 - `frame.arrow()` — the data as a `pyarrow.Table` (`frame.df()` for pandas,
   `frame.to_pylist()` for dicts, `len(frame)` for the row count)
 - `frame.sql("SELECT region, revenue FROM this ORDER BY revenue DESC LIMIT 5")`
-  — SQL runs server-side against the cached entry; `this` names the data
+  — SQL runs server-side against the cached entry; `this` names the data.
+  Needs a persisted entry: available on hits (on a fresh miss the
+  write-behind persist has to land first; `background=False` avoids that)
 - `frame.cached` — which path served it; `frame.entry` — the registry record
 
 The wrapped function can return an iterable of dicts (a `.values()` queryset),

demo/README.md

@@ -1,6 +1,7 @@
 # Demo: TPC-H on Neon vs hotdata-materialized
 
-Times a heavy TPC-H aggregation three ways:
+Times a heavy TPC-H aggregation three ways, end to end through the
+`@materialize` decorator:
 
 1. **Neon direct** — the query runs on your Postgres every time.
 2. **Materialized miss** — the query runs on Neon once, and the result is

demo/tpch/management/commands/compare.py

@@ -1,24 +1,22 @@
-"""Compare running a heavy TPC-H aggregation directly on Neon vs through
-hotdata-materialized: one miss (runs on Neon, captured into Hotdata), then
-hits that never touch Neon, plus a server-side transform on the cached entry.
+"""Compare running a heavy TPC-H aggregation directly on Postgres vs through
+the @materialize decorator: one miss (runs on Postgres, persisted
+write-behind), then hits that never touch Postgres, plus server-side SQL on
+the cached entry.
 
 Usage:
     python manage.py compare                 # Q1 pricing summary (tiny result)
-    python manage.py compare --scenario parts  # revenue by part (~200k rows)
+    python manage.py compare --scenario parts  # revenue by part (large result)
 """
 
 import datetime
 import statistics
 import time
-from decimal import Decimal
 
-import pyarrow as pa
 from django.core.management.base import BaseCommand, CommandError
 from django.db.models import Count, DecimalField, ExpressionWrapper, F, Sum, Value
-from hotdata.models.query_request import QueryRequest
 
-from hotdata_materialized import Config, EntryStore, Registry, get_clients
-from hotdata_materialized.fingerprint import fingerprint_queryset
+from hotdata_materialized import fingerprint_call, materialize
+from hotdata_materialized.decorator import get_runtime
 from tpch.models import LineItem
 
 DEC = DecimalField(max_digits=25, decimal_places=4)
@@ -51,39 +49,45 @@ def q1_queryset():
 
 
 def parts_queryset():
-    """Revenue by part: a wide group-by producing a large (~200k row) result."""
+    """Revenue by part: a wide group-by producing a large result."""
     return (
         LineItem.objects.values("partkey")
         .annotate(revenue=Sum(DISC_PRICE), orders=Count("orderkey"))
         .order_by("-revenue")
     )
 
 
+@materialize(ttl=3600, key="tpch-q1")
+def q1_report():
+    return q1_queryset()
+
+
+@materialize(ttl=3600, key="tpch-parts")
+def parts_report():
+    return parts_queryset()
+
+
 SCENARIOS = {
     "q1": (
         q1_queryset,
-        "SELECT returnflag, linestatus, sum_disc_price FROM data "
+        q1_report,
+        "SELECT returnflag, linestatus, sum_disc_price FROM this "
         "ORDER BY sum_disc_price DESC LIMIT 3",
     ),
-    "parts": (parts_queryset, "SELECT partkey, revenue FROM data ORDER BY revenue DESC LIMIT 10"),
+    "parts": (
+        parts_queryset,
+        parts_report,
+        "SELECT partkey, revenue FROM this ORDER BY revenue DESC LIMIT 10",
+    ),
 }
 
 
-def rows_to_arrow(rows):
-    def clean(value):
-        return float(value) if isinstance(value, Decimal) else value
-
-    return pa.Table.from_pylist(
-        [{k: clean(v) for k, v in row.items()} for row in rows]
-    )
-
-
 def ms(seconds):
     return f"{seconds * 1000:,.0f} ms"
 
 
 class Command(BaseCommand):
-    help = "Compare a heavy TPC-H query on Neon vs the hotdata-materialized path"
+    help = "Compare a heavy TPC-H query on Postgres vs the @materialize path"
 
     def add_arguments(self, parser):
         parser.add_argument("--scenario", choices=SCENARIOS, default="q1")
@@ -94,41 +98,36 @@ def add_arguments(self, parser):
         )
 
     def handle(self, *args, scenario, runs, keep, **options):
-        queryset_factory, transform_sql = SCENARIOS[scenario]
-        config = Config.from_django()
-        clients = get_clients(config)
-        registry = Registry(clients, config)
-        store = EntryStore(clients, config, registry)
-
-        fingerprint = fingerprint_queryset(queryset_factory())
+        queryset_factory, report, transform_sql = SCENARIOS[scenario]
+        _, store = get_runtime()
+        # the decorator fingerprints the wrapped function + its (empty) args
+        fingerprint = fingerprint_call(report.__wrapped__)
         self.stdout.write(f"scenario={scenario}  fingerprint={fingerprint[:16]}…\n")
 
-        # -- direct: Neon does the work every time --------------------------
+        # -- direct: Postgres does the work every time -----------------------
         direct_times = []
         for run in range(runs):
             start = time.perf_counter()
             rows = list(queryset_factory())
             direct_times.append(time.perf_counter() - start)
             self.stdout.write(
-                f"  neon direct run {run + 1}: {ms(direct_times[-1])} ({len(rows):,} rows)"
+                f"  postgres direct run {run + 1}: {ms(direct_times[-1])} "
+                f"({len(rows):,} rows)"
             )
 
-        # -- miss: caller gets rows at Neon speed; persist runs write-behind --
+        # -- miss: caller gets the frame at Postgres speed; persist is hidden --
         if not keep:
             store.evict(fingerprint)
-        entry = registry.lookup(fingerprint)
         miss_time = None
         background_time = None
-        if entry is None or entry.status != "ready":
-            start = time.perf_counter()
-            rows = list(queryset_factory())
-            entry = store.materialize(
-                fingerprint, rows_to_arrow(rows), key=scenario, ttl=3600
-            )
-            miss_time = time.perf_counter() - start
+        start = time.perf_counter()
+        frame = report()
+        elapsed = time.perf_counter() - start
+        if not frame.cached:
+            miss_time = elapsed
             self.stdout.write(
                 f"  materialized miss, perceived: {ms(miss_time)} "
-                f"({len(rows):,} rows returned to caller)"
+                f"({len(frame):,} rows returned to caller)"
             )
             start = time.perf_counter()
             errors = store.flush()
@@ -140,36 +139,37 @@ def handle(self, *args, scenario, runs, keep, **options):
                 f"return (off the request path)"
             )
 
-        # -- hits: Neon is never touched, data comes back as Arrow ----------
+        # -- hits: Postgres is never touched, data comes back as Arrow -------
         hit_times = []
         for _ in range(5):
             start = time.perf_counter()
-            entry = registry.lookup(fingerprint)
-            result = store.read_table(entry)
+            frame = report()
+            table = frame.arrow()
             hit_times.append(time.perf_counter() - start)
+        if not frame.cached:
+            raise CommandError("expected a hit; entry never became ready")
         backing = (
-            "inline payload" if entry.inline_payload else "entry database (arrow)"
+            "inline payload" if frame.entry.inline_payload
+            else "entry database (arrow)"
         )
         self.stdout.write(
             f"  materialized hit ×5 via {backing}: median "
-            f"{ms(statistics.median(hit_times))} ({result.num_rows:,} rows)"
+            f"{ms(statistics.median(hit_times))} ({table.num_rows:,} rows)"
         )
 
         # -- transform: SQL over the cached entry, server-side ---------------
         start = time.perf_counter()
-        response = clients.query.query(
-            QueryRequest(sql=transform_sql), x_database_id=entry.database_id
-        )
+        top = frame.sql(transform_sql)
         transform_time = time.perf_counter() - start
         self.stdout.write(f"  server-side SQL on cached entry: {ms(transform_time)}")
-        for row in response.rows:
+        for row in top.to_pylist():
             self.stdout.write(f"    {row}")
 
         # -- summary ----------------------------------------------------------
         median_direct = statistics.median(direct_times)
         median_hit = statistics.median(hit_times)
         self.stdout.write("")
-        self.stdout.write(f"  {'neon direct (median)':34s} {ms(median_direct)}")
+        self.stdout.write(f"  {'postgres direct (median)':34s} {ms(median_direct)}")
         if miss_time is not None:
             self.stdout.write(
                 f"  {'materialized miss, perceived':34s} {ms(miss_time)}"
@@ -180,5 +180,5 @@ def handle(self, *args, scenario, runs, keep, **options):
             )
         self.stdout.write(
             f"  {'materialized hit (median)':34s} {ms(median_hit)}"
-            f"  ({median_direct / median_hit:,.1f}× faster, zero load on Neon)"
+            f"  ({median_direct / median_hit:,.1f}× faster, zero load on Postgres)"
         )

hotdata_materialized/decorator.py

@@ -31,6 +31,16 @@
 logger = logging.getLogger(__name__)
 
 _THIS = re.compile(r"\bthis\b")
+# quoted chunks ('...' with '' escapes, "..." with "" escapes) pass through
+# untouched so a literal like WHERE label = 'this' is never rewritten
+_SQL_CHUNKS = re.compile(r"('(?:[^']|'')*'|\"(?:[^\"]|\"\")*\")")
+
+
+def _rewrite_this(sql: str) -> str:
+    return "".join(
+        part if part.startswith(("'", '"')) else _THIS.sub(DATA_TABLE, part)
+        for part in _SQL_CHUNKS.split(sql)
+    )
 
 
 class MaterializedFrame:
@@ -64,8 +74,12 @@ def df(self) -> Any:
 
     def sql(self, sql: str) -> pa.Table:
         """Run SQL server-side against this entry's database; the identifier
-        `this` names the cached data (e.g. "SELECT x FROM this LIMIT 5")."""
-        return self._store.query_table(self.entry, _THIS.sub(DATA_TABLE, sql))
+        `this` names the cached data (e.g. "SELECT x FROM this LIMIT 5").
+
+        Requires a persisted entry: available on hits, or after the
+        write-behind persist lands (a fresh miss raises StoreError until
+        then; use background=False on the decorator to persist inline)."""
+        return self._store.query_table(self.entry, _rewrite_this(sql))
 
     def __len__(self) -> int:
         return self.arrow().num_rows

tests/test_decorator.py

@@ -127,6 +127,16 @@ def test_sql_runs_server_side_against_the_entry(runtime):
     assert top.to_pylist() == [{"channel": "organic"}]
 
 
+def test_sql_leaves_string_literals_alone(runtime):
+    calls = {"n": 0}
+    report = make_report(calls)
+    report("this")  # the day column literally contains "this"
+    runtime.store.flush()
+    frame = report("this")
+    rows = frame.sql("SELECT n FROM this WHERE day = 'this' AND n > 8")
+    assert rows.to_pylist() == [{"n": 10}]
+
+
 def test_to_arrow_accepts_table_and_rows():
     table = pa.table({"n": [1, 2]})
     assert to_arrow(table) is table