#!/usr/bin/env python3 """ Build the local metadata sidecar from Turso. Why this exists --------------- Turso's `papers` table has exactly one index (`arxiv_id`), so it supports point lookups and nothing else. Measured against the live DB: SELECT 1 -> 187 ms SELECT COUNT(*) -> 834 ms SELECT MIN(update_date), MAX(update_date) -> TIMES OUT at 100 s trending (LIKE + ORDER BY) -> SCAN papers + TEMP B-TREE and a 50-id metadata fetch on the search hot path costs ~4.7 s round trip. This script mirrors the table into a local SQLite file that ships inside the Docker image, with the indexes Turso lacks. That buys two things: 1. Hot-path metadata lookups become local reads (microseconds, no network). 2. Query patterns that are currently impossible become cheap — trending by category, recency filters, "most cited in cs.CL since 2024". Usage ----- export TURSO_URL=... TURSO_DB_TOKEN=... python scripts/build_metadata_sidecar.py --out data/metadata.sqlite # smaller/faster build without abstracts (~250 MB instead of ~1.1 GB) python scripts/build_metadata_sidecar.py --out data/meta.sqlite --no-abstracts Resumable: re-running continues from the highest rowid already imported. """ from __future__ import annotations import argparse import json import os import sqlite3 import sys import time import urllib.error import urllib.request TOTAL_ESTIMATE = 1_600_000 SCHEMA = """ CREATE TABLE IF NOT EXISTS papers ( arxiv_id TEXT PRIMARY KEY, title TEXT, authors TEXT, abstract_preview TEXT, categories TEXT, primary_topic TEXT, update_date TEXT, citation_count INTEGER, influential_citations INTEGER ); -- One row per (paper, arXiv code). Turso can only do LIKE '%cs.CL%' against a -- space-separated string, which forces a full scan; this makes category -- filtering index-driven. CREATE TABLE IF NOT EXISTS paper_categories ( code TEXT NOT NULL, arxiv_id TEXT NOT NULL, citation_count INTEGER, update_date TEXT ); CREATE TABLE IF NOT EXISTS build_state ( key TEXT PRIMARY KEY, value TEXT ); """ INDEXES = """ CREATE INDEX IF NOT EXISTS idx_papers_update_date ON papers(update_date); CREATE INDEX IF NOT EXISTS idx_papers_citations ON papers(citation_count DESC); -- Covers the trending query: filter by code, already ordered by citations. CREATE INDEX IF NOT EXISTS idx_pc_code_cit ON paper_categories(code, citation_count DESC); CREATE INDEX IF NOT EXISTS idx_pc_code_date ON paper_categories(code, update_date DESC); CREATE INDEX IF NOT EXISTS idx_pc_arxiv ON paper_categories(arxiv_id); """ def turso_query(url: str, token: str, sql: str, timeout: int = 120): payload = json.dumps({"requests": [ {"type": "execute", "stmt": {"sql": sql}}, {"type": "close"}, ]}).encode() req = urllib.request.Request( f"{url}/v2/pipeline", data=payload, headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"}) with urllib.request.urlopen(req, timeout=timeout) as r: data = json.loads(r.read()) res = data["results"][0] if res.get("type") == "error": raise RuntimeError(str(res.get("error"))[:300]) rd = res["response"]["result"] return [[c.get("value") for c in row] for row in rd["rows"]] def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--out", required=True, help="output sqlite path") ap.add_argument("--chunk", type=int, default=2000, help="rows per request") ap.add_argument("--no-abstracts", action="store_true", help="omit abstract_preview (much smaller file)") ap.add_argument("--max-rowid", type=int, default=1_700_000) args = ap.parse_args() url = os.environ.get("TURSO_URL", "").rstrip("/") token = os.environ.get("TURSO_DB_TOKEN", "") if not url or not token: print("TURSO_URL and TURSO_DB_TOKEN must be set", file=sys.stderr) return 2 os.makedirs(os.path.dirname(os.path.abspath(args.out)) or ".", exist_ok=True) con = sqlite3.connect(args.out) con.executescript(SCHEMA) # Bulk-load pragmas. Safe: if the build dies we just re-run it. con.execute("PRAGMA journal_mode=OFF") con.execute("PRAGMA synchronous=OFF") con.commit() cur = con.execute("SELECT value FROM build_state WHERE key='max_rowid_done'") row = cur.fetchone() start = int(row[0]) if row else 0 if start: print(f"resuming from rowid {start:,}") abstract_col = "''" if args.no_abstracts else "abstract_preview" chunk = args.chunk lo = start + 1 imported = 0 t_start = time.time() while lo <= args.max_rowid: hi = lo + chunk - 1 sql = ( "SELECT rowid, arxiv_id, title, authors, " f"{abstract_col}, categories, primary_topic, update_date, " "citation_count, influential_citations " f"FROM papers WHERE rowid BETWEEN {lo} AND {hi}" ) try: rows = turso_query(url, token, sql) except (urllib.error.URLError, RuntimeError, TimeoutError) as e: if chunk > 250: chunk = max(250, chunk // 2) print(f"\n chunk too big/slow ({str(e)[:60]}); retrying at {chunk}") continue print(f"\n chunk {lo}-{hi} failed permanently: {str(e)[:120]}") lo = hi + 1 continue if rows: papers = [] cats = [] for r in rows: (_rid, aid, title, authors, abstract, categories, topic, udate, cit, infl) = r if not aid: continue try: cit_i = int(cit) if cit is not None else 0 except (TypeError, ValueError): cit_i = 0 try: infl_i = int(infl) if infl is not None else 0 except (TypeError, ValueError): infl_i = 0 papers.append((aid, title, authors, abstract, categories, topic, udate, cit_i, infl_i)) for code in str(categories or "").split(): cats.append((code, aid, cit_i, udate)) con.executemany( "INSERT OR REPLACE INTO papers VALUES (?,?,?,?,?,?,?,?,?)", papers) con.executemany( "INSERT INTO paper_categories VALUES (?,?,?,?)", cats) imported += len(papers) con.execute( "INSERT OR REPLACE INTO build_state VALUES ('max_rowid_done', ?)", (str(hi),)) con.commit() elapsed = time.time() - t_start rate = imported / elapsed if elapsed > 0 else 0 pct = 100 * min(hi, TOTAL_ESTIMATE) / TOTAL_ESTIMATE eta = (TOTAL_ESTIMATE - imported) / rate / 60 if rate > 0 else 0 print(f"\r rowid {hi:,} imported {imported:,} " f"({pct:5.1f}%) {rate:,.0f} rows/s ETA {eta:5.1f} min", end="", flush=True) if not rows and hi > TOTAL_ESTIMATE: break lo = hi + 1 print("\nbuilding indexes...") con.executescript(INDEXES) con.commit() print("running ANALYZE...") con.execute("ANALYZE") con.commit() con.execute("PRAGMA journal_mode=DELETE") con.close() size_mb = os.path.getsize(args.out) / 1e6 print(f"done: {imported:,} papers -> {args.out} ({size_mb:,.0f} MB)") return 0 if __name__ == "__main__": sys.exit(main())