From 31ec7ca38ae52ed1e72540794c054523ea42e419 Mon Sep 17 00:00:00 2001 From: Eddie A Tejeda <669988+eddietejeda@users.noreply.github.com> Date: Wed, 22 Jul 2026 14:08:08 -0700 Subject: [PATCH] Docs and demo: decorator-first, plus review-nit fixes - DESIGN.md: split the decorator API into implemented (first cut) vs planned knobs; mark the MaterializedFrame section's implemented subset; tick step 2 in the build order - demo: compare.py now goes through @materialize end to end (evict/flush via get_runtime for benchmark mechanics); Decimals round-trip natively - README: note that frame.sql() needs a persisted entry - decorator: frame.sql() no longer rewrites `this` inside quoted SQL literals/identifiers (review nit); documented the miss-frame .sql() behavior (review nit) --- DESIGN.md | 28 +++--- README.md | 4 +- demo/README.md | 3 +- demo/tpch/management/commands/compare.py | 104 +++++++++++------------ hotdata_materialized/decorator.py | 18 +++- tests/test_decorator.py | 10 +++ 6 files changed, 99 insertions(+), 68 deletions(-) diff --git a/DESIGN.md b/DESIGN.md index 96e3251..4618445 100644 --- a/DESIGN.md +++ b/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 + diff --git a/README.md b/README.md index 1c77cc3..eae1598 100644 --- a/README.md +++ b/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), diff --git a/demo/README.md b/demo/README.md index dc7af17..5325aa3 100644 --- a/demo/README.md +++ b/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 diff --git a/demo/tpch/management/commands/compare.py b/demo/tpch/management/commands/compare.py index 44dcb2a..1247a78 100644 --- a/demo/tpch/management/commands/compare.py +++ b/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,7 +49,7 @@ 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")) @@ -59,31 +57,37 @@ def parts_queryset(): ) +@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)" ) diff --git a/hotdata_materialized/decorator.py b/hotdata_materialized/decorator.py index 668b7b7..1c26b46 100644 --- a/hotdata_materialized/decorator.py +++ b/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 diff --git a/tests/test_decorator.py b/tests/test_decorator.py index a7f7c3f..c7edd11 100644 --- a/tests/test_decorator.py +++ b/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