bf889681f6
Replace the per-cell parquet cache with TimescaleDB hypertables as the production backend for the raw daily climate record, and drop pg_duckdb. Parquet stays the backend whenever THERMOGRAPH_DATABASE_URL is not a Postgres URL (dev, tests, offline tooling), the same dialect switch the accounts DB and derived store already use, so CI stays Postgres-free. - data/climate_store.py: psycopg + polars bridge over climate_history (a hypertable), climate_recent, and climate_sync (per-cell freshness). Reads via pl.read_database, writes via COPY + ON CONFLICT upsert; fail-soft to a cache miss so a DB hiccup degrades to upstream refetch. - data/climate.py: route every cache/mtime touchpoint through a backend dispatch. recent_stamp becomes int(recent_synced_at) on Postgres; the stale-serve path still avoids bumping it, so derived-payload tokens invalidate on exactly the same events as before. - alembic 0002: CREATE EXTENSION timescaledb plus the hypertable schema (compression policy on year-old chunks), guarded to no-op off Postgres. - migrate_cache_to_pg.py (make migrate-cache): idempotent backfill of the parquet cache into the hypertables, preserving file mtimes as sync timestamps so recent_stamp is unchanged across cutover. - db image -> stock timescale/timescaledb:latest-pg18; drop the custom pg_duckdb Dockerfile, the read-only /parquet mount, and the duckdb tuning GUC. Docs updated for the new backend and cutover. Co-authored-by: Claude <noreply@anthropic.com>
96 lines
3.9 KiB
Python
96 lines
3.9 KiB
Python
"""One-shot backfill of the parquet climate cache into the TimescaleDB tables.
|
|
|
|
Run ONCE during the cutover, after ``alembic upgrade head`` has created the
|
|
``climate_history`` / ``climate_recent`` / ``climate_sync`` tables. It loads every
|
|
cached cell's parquet record into Postgres so a freshly-cut-over server serves
|
|
history/recent from the database immediately instead of refetching decades of
|
|
archive (which the upstream rate limit would throttle hard).
|
|
|
|
Idempotent and resumable: a cell whose stored sync timestamp already matches its
|
|
parquet mtime is skipped, and history rows upsert (``ON CONFLICT``), so re-running
|
|
after an interruption only does the missing work.
|
|
|
|
Key detail: rows are read RAW (``pl.read_parquet`` + ``_normalize_read``), NOT via
|
|
``climate.load_cached_*`` — the loaders apply ``_derive_metrics``, which replaces
|
|
raw relative humidity with absolute humidity. The DB must store the raw record the
|
|
live write path persists, so absolute humidity / wet-bulb stay read-time
|
|
derivations. The per-cell ``synced_at`` is set to the parquet FILE MTIME (not now),
|
|
so ``recent_stamp`` — hence every derived-payload validity token — is unchanged
|
|
across the cutover.
|
|
|
|
Usage:
|
|
THERMOGRAPH_DATABASE_URL=postgresql+psycopg://thermograph:PW@127.0.0.1:5432/thermograph \\
|
|
python migrate_cache_to_pg.py # reads data/cache/*.parquet
|
|
"""
|
|
import os
|
|
import sys
|
|
|
|
import polars as pl
|
|
|
|
from data import climate
|
|
from data import climate_store
|
|
from data import grid
|
|
|
|
|
|
def _is_schema_complete(df: pl.DataFrame) -> bool:
|
|
"""The same NEW_COLS guard climate._read_history_cache applies: a record missing
|
|
any post-original column predates the wind/humidity features and is skipped so
|
|
the server refetches it lazily (with the current schema) on first touch."""
|
|
return all(c in df.columns for c in climate.NEW_COLS)
|
|
|
|
|
|
def migrate() -> int:
|
|
if not climate_store.is_postgres():
|
|
sys.exit("THERMOGRAPH_DATABASE_URL must be a postgresql URL (the migration target)")
|
|
if not os.path.isdir(climate.CACHE_DIR):
|
|
print("No parquet cache directory yet — nothing to migrate.")
|
|
return 0
|
|
|
|
hist = recent = current = skipped = 0
|
|
for fname in sorted(os.listdir(climate.CACHE_DIR)):
|
|
if not fname.endswith(".parquet"):
|
|
continue
|
|
path = os.path.join(climate.CACHE_DIR, fname)
|
|
is_recent = fname.endswith("_rf.parquet")
|
|
cell_id = fname[:-len("_rf.parquet")] if is_recent else fname[:-len(".parquet")]
|
|
try:
|
|
grid.from_id(cell_id) # validate the id is a real cell (skip stray files)
|
|
except ValueError:
|
|
print(f" {fname}: unrecognized cache filename — skipped")
|
|
skipped += 1
|
|
continue
|
|
|
|
mtime = os.path.getmtime(path)
|
|
synced = (climate_store.recent_synced_at(cell_id) if is_recent
|
|
else climate_store.history_synced_at(cell_id))
|
|
if synced >= mtime:
|
|
current += 1
|
|
continue
|
|
|
|
try:
|
|
df = climate._normalize_read(pl.read_parquet(path))
|
|
except Exception as e: # noqa: BLE001 - a truncated/corrupt file is a skip, not a crash
|
|
print(f" {fname}: unreadable ({e}) — skipped")
|
|
skipped += 1
|
|
continue
|
|
|
|
if is_recent:
|
|
climate_store.write_recent(cell_id, df, mtime)
|
|
recent += 1
|
|
print(f" {cell_id}: recent bundle ({df.height} rows)")
|
|
else:
|
|
if not _is_schema_complete(df):
|
|
print(f" {cell_id}: pre-schema history — skipped (refetches lazily)")
|
|
skipped += 1
|
|
continue
|
|
climate_store.write_history(cell_id, df, mtime)
|
|
hist += 1
|
|
print(f" {cell_id}: history ({df.height} rows)")
|
|
|
|
print(f"{hist} history + {recent} recent loaded, {current} already current, "
|
|
f"{skipped} skipped.")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(migrate())
|