perf(lakehouse): store hot-projection keys once, 6.9 GB to 1.1 GB

docs/MEET_LAKEHOUSE_EXPERIMENT.md

@@ -69,7 +69,10 @@ writes an Iceberg table partitioned by year/course. It creates:
 - A DuckDB database with an `races` view reading the exact committed Iceberg
   metadata, plus a `parquet_snapshot` reference view of that snapshot's active
   files. Full multiset equality verifies the two reads.
-- A separate SQLite hot projection with source-scoped athlete and meet indexes.
+- A separate SQLite hot projection. Swimmer, meet and event keys are stored once
+  in dimension tables, and the race table's athlete-history and meet-results
+  indexes hold only integers. `lakehouse.py rebuild-hot` replaces it from the
+  committed snapshot.
 - A private snapshot manifest with source hash, versions and snapshot ID, written
   last. An incomplete build has no published manifest. Existing outputs are
   never overwritten.
@@ -95,6 +98,13 @@ A second checkpoint of 106,053 races occupied 5,025,874 bytes in Parquet and
 under the pilot's 30-day policy, with 57 independent test swimmers. This improves
 the pilot's coverage but still does not justify a general predictive claim.
 
+On the national checkpoint (8,506,767 races), the straightforward layout used in
+those pilots took 6,915,502,080 bytes, about 813 bytes/race, because every
+64-character key was repeated in the table and in each index. The current layout
+takes 1,101,565,952 bytes, about 129 bytes/race; the remaining bulk is the race
+ID stored once per row. Warm single-swimmer history lookups stay around 0.02 ms
+locally.
+
 The indexed query and Iceberg results agree. Millisecond local timings are
 warm-cache laptop measurements, **not** D1/R2 latency, throughput or cost
 benchmarks. The current importer materializes one bounded snapshot in memory
@@ -129,7 +139,7 @@ matters; an inexpensive single copy is not an archival strategy by itself.
 ## D1 and sharding
 
 The [current D1 limits](https://developers.cloudflare.com/d1/platform/limits/)
-are 500 MB/database on Free and 10 GB/database on Paid. Each database processes
+are 500 MB/database on Free and 10 GB/database on Paid. The national hot projection (1.10 GB) fits one Paid database. Each database processes
 queries serially. Good indexes, short queries, bounded pagination and caching
 come before sharding. The prototype's ordinary SQLite SQL is a portability
 starting point, not proof that a D1 workload will fit its execution limits.

research/conversion-model/REPORT.md

@@ -712,16 +712,16 @@ The join picks a likely facility in a postal code or city. It does not verify a
 
 ## Appendix C. Local Lakehouse
 
-[`lakehouse.py`](./lakehouse.py) materializes a dual-store local lakehouse from the canonical ingest store: an indexed SQLite store (composite B-tree on `athlete_history(source, swimmer_id, race_date, meet_id, race_id)`) for point lookups by swimmer and date, and an Iceberg table of Zstandard Parquet files, identity-partitioned by year and course, for the longitudinal self-joins; DuckDB reads it through the committed Iceberg metadata rather than a file glob. The design and its hosted counterparts are in [`MEET_LAKEHOUSE_EXPERIMENT.md`](../../docs/MEET_LAKEHOUSE_EXPERIMENT.md); serving costs are in [`LAKEHOUSE_COST_MODEL.md`](../../docs/LAKEHOUSE_COST_MODEL.md). Measured for this checkpoint (acquisition receipt `lakehouse_storage`):
+[`lakehouse.py`](./lakehouse.py) materializes a dual-store local lakehouse from the canonical ingest store: an indexed SQLite store for point lookups by swimmer and date and by meet and event (swimmer, meet and event keys are stored once in dimension tables; the race table's `athlete_history(swimmer, race_day)` and `meet_results(meet, event, time_cs)` indexes hold only integers, and `lakehouse.py rebuild-hot` replaces the store from the committed snapshot), and an Iceberg table of Zstandard Parquet files, identity-partitioned by year and course, for the longitudinal self-joins; DuckDB reads it through the committed Iceberg metadata rather than a file glob. The design and its hosted counterparts are in [`MEET_LAKEHOUSE_EXPERIMENT.md`](../../docs/MEET_LAKEHOUSE_EXPERIMENT.md); serving costs are in [`LAKEHOUSE_COST_MODEL.md`](../../docs/LAKEHOUSE_COST_MODEL.md). Measured for this checkpoint (acquisition receipt `lakehouse_storage`):
 
 <!-- receipt:lakehouse -->
 | Lakehouse (`target/lakehouse-national-17`) | Measured |
 | :--- | ---: |
 | Iceberg snapshot | `4441271835344284414` |
-| Hot SQLite store | 6,915.5 MB (812.9 bytes/row) |
+| Hot SQLite store | 1,101.6 MB (129.5 bytes/row) |
 | Parquet warehouse | 366.8 MB (43.1 bytes/row) |
-| Point lookup by swimmer, first / repeat median | 0.168 ms / 0.012 ms |
-| Filtered DuckDB aggregate, first / repeat median | 3.59 ms / 2.12 ms |
+| Point lookup by swimmer, first / repeat median | 0.311 ms / 0.020 ms |
+| Filtered DuckDB aggregate, first / repeat median | 3.58 ms / 2.80 ms |
 <!-- /receipt:lakehouse -->
 
 The evaluator measures the scanned Iceberg snapshot itself (honoring `snapshot_from_id`) and refuses to run if it disagrees with `snapshot.json` or a referenced file is missing. `lakehouse.py probe` checks that both stores hold the same rows with no duplicates and return one swimmer's latest 50 rows identically; its output is not part of a receipt.

research/conversion-model/acquisition-checkpoint-06.json

@@ -235,12 +235,12 @@
   "lakehouse_storage": {
     "checkpoint_dir": "target/lakehouse-national-17",
     "iceberg_snapshot_id": 4441271835344284414,
-    "hot_sqlite_bytes": 6915502080,
+    "hot_sqlite_bytes": 1101565952,
     "parquet_bytes": 366796050,
-    "point_lookup_first_ms": 0.168,
-    "point_lookup_repeat_median_ms": 0.012,
-    "duckdb_scan_first_ms": 3.59,
-    "duckdb_scan_repeat_median_ms": 2.12
+    "point_lookup_first_ms": 0.311,
+    "point_lookup_repeat_median_ms": 0.02,
+    "duckdb_scan_first_ms": 3.577,
+    "duckdb_scan_repeat_median_ms": 2.803
   },
   "meet_venue_resolution": {
     "total_unique_meets_parsed": 6059,

research/conversion-model/lakehouse.py

@@ -1,5 +1,7 @@
 """Local Iceberg/Parquet query experiment, with a rebuildable SQLite hot projection.
 
+`rebuild-hot` replaces the SQLite projection from the committed Iceberg snapshot.
+
 No network ingestion, cloud credentials, uploads or model training. Outputs contain
 private race-level data: keep them under ignored target/ or outside the repository.
 """
@@ -46,16 +48,44 @@
     "time_cs",
 )
 OPTIONAL_CONTEXT = ("region", "meet_class", "round", "archive_id", "date_precision")
+# Repeated keys live once in small dimension tables; each race row and its two
+# indexes carry only integers. A 64-character hashed ID repeated in the table and
+# in every index made the earlier layout about 813 bytes a race. Uniqueness of
+# (source, race_id) is enforced by read_snapshot, so no index stores race_id.
 HOT_SCHEMA = """
-CREATE TABLE races (
-    source TEXT NOT NULL, race_id TEXT NOT NULL, swimmer_id TEXT NOT NULL,
-    meet_id TEXT NOT NULL, race_date TEXT NOT NULL, course TEXT NOT NULL,
-    distance INTEGER NOT NULL, stroke TEXT NOT NULL, category TEXT NOT NULL,
-    time_cs INTEGER NOT NULL, PRIMARY KEY (source, race_id)
+CREATE TABLE swimmers (
+    id INTEGER PRIMARY KEY, source TEXT NOT NULL, swimmer_id TEXT NOT NULL,
+    UNIQUE (source, swimmer_id)
+);
+CREATE TABLE meets (
+    id INTEGER PRIMARY KEY, source TEXT NOT NULL, meet_id TEXT NOT NULL,
+    UNIQUE (source, meet_id)
+);
+CREATE TABLE events (
+    id INTEGER PRIMARY KEY, course TEXT NOT NULL, distance INTEGER NOT NULL,
+    stroke TEXT NOT NULL, category TEXT NOT NULL,
+    UNIQUE (course, distance, stroke, category)
+);
+CREATE TABLE race_rows (
+    id INTEGER PRIMARY KEY, swimmer INTEGER NOT NULL REFERENCES swimmers,
+    race_day INTEGER NOT NULL, meet INTEGER NOT NULL REFERENCES meets,
+    event INTEGER NOT NULL REFERENCES events, time_cs INTEGER NOT NULL,
+    race_id TEXT NOT NULL
 );
-CREATE INDEX athlete_history ON races(source, swimmer_id, race_date DESC, race_id);
-CREATE INDEX meet_results ON races(source, meet_id, course, distance, stroke, time_cs);
+CREATE INDEX athlete_history ON race_rows(swimmer, race_day DESC);
+CREATE INDEX meet_results ON race_rows(meet, event, time_cs);
+CREATE VIEW races AS
+SELECT s.source, r.race_id, s.swimmer_id, m.meet_id,
+       date(r.race_day * 86400, 'unixepoch') AS race_date,
+       e.course, e.distance, e.stroke, e.category, r.time_cs
+FROM race_rows r
+JOIN swimmers s ON s.id = r.swimmer
+JOIN meets m ON m.id = r.meet
+JOIN events e ON e.id = r.event;
 """
+HISTORY_SQL = """SELECT r.race_id, r.time_cs FROM race_rows r
+JOIN swimmers s ON s.id = r.swimmer WHERE s.source = ? AND s.swimmer_id = ?
+ORDER BY r.race_day DESC, r.id LIMIT 50"""
 
 
 def literal(value: str) -> str:
@@ -168,21 +198,66 @@ def read_snapshot(path: Path, max_rows: int = 1_000_000):
 
 def create_hot(table, path: Path):
     """Use ordinary SQLite SQL supported by D1; never claim this benchmarks D1."""
+    epoch = date(1970, 1, 1)
+    keys = {"swimmers": {}, "meets": {}, "events": {}}
+
+    def key(kind, value):
+        return keys[kind].setdefault(value, len(keys[kind]) + 1)
+
+    rows = [
+        (
+            key("swimmers", (row["source"], row["swimmer_id"])),
+            (row["race_date"] - epoch).days,
+            key("meets", (row["source"], row["meet_id"])),
+            key(
+                "events",
+                (row["course"], row["distance"], row["stroke"], row["category"]),
+            ),
+            row["time_cs"],
+            row["race_id"],
+        )
+        for row in table.select(HOT_COLUMNS).to_pylist()
+    ]
+    # Insert each swimmer's races together in (date, race_id) order, so row IDs
+    # break same-day ties the way race IDs do and a history read stays local.
+    rows.sort(key=lambda row: (row[0], row[1], row[5]))
     with closing(sqlite3.connect(path)) as db, db:
         db.executescript(HOT_SCHEMA)
+        for kind, width in (("swimmers", 2), ("meets", 2), ("events", 4)):
+            db.executemany(
+                f"INSERT INTO {kind} VALUES (?{', ?' * width})",
+                ((ident, *value) for value, ident in keys[kind].items()),
+            )
         db.executemany(
-            f"INSERT INTO races VALUES ({','.join('?' for _ in HOT_COLUMNS)})",
-            (
-                tuple(
-                    row[column].isoformat()
-                    if isinstance(row[column], date)
-                    else row[column]
-                    for column in HOT_COLUMNS
-                )
-                for row in table.to_pylist()
-            ),
+            "INSERT INTO race_rows (swimmer, race_day, meet, event, time_cs, race_id)"
+            " VALUES (?, ?, ?, ?, ?, ?)",
+            rows,
         )
         db.execute("ANALYZE")
+    with closing(sqlite3.connect(path)) as db:
+        db.execute("VACUUM")
+
+
+def rebuild_hot(output: Path):
+    """Replace the hot projection from the committed Iceberg snapshot it mirrors."""
+    import duckdb
+
+    snapshot_path = output / "snapshot.json"
+    snapshot = json.loads(snapshot_path.read_text())
+    with duckdb.connect(str(output / "queries.duckdb"), read_only=True) as db:
+        db.execute("LOAD iceberg")
+        table = db.sql(f"SELECT {', '.join(HOT_COLUMNS)} FROM races").to_arrow_table()
+    if table.num_rows != snapshot["rows"]:
+        raise ValueError("Iceberg snapshot row count differs from snapshot.json")
+    staged = output / "hot.sqlite3.new"
+    staged.unlink(missing_ok=True)
+    create_hot(table, staged)
+    staged.replace(output / "hot.sqlite3")
+    snapshot["hot_sqlite_bytes"] = (output / "hot.sqlite3").stat().st_size
+    staged_report = output / "snapshot.json.new"
+    staged_report.write_text(json.dumps(snapshot, indent=2) + "\n")
+    staged_report.replace(snapshot_path)
+    return snapshot
 
 
 def build(csv_path: Path, manifest_path: Path, output: Path, max_rows=1_000_000):
@@ -313,8 +388,7 @@ def probe(output: Path):
             "SELECT source, swimmer_id FROM races ORDER BY source, swimmer_id LIMIT 1"
         ).fetchone()
         with closing(sqlite3.connect(output / "hot.sqlite3")) as hot:
-            query = """SELECT race_id, time_cs FROM races WHERE source=? AND swimmer_id=?
-                       ORDER BY race_date DESC, race_id LIMIT 50"""
+            query = HISTORY_SQL
             history, hot_first = timed(lambda: hot.execute(query, first).fetchall())
             warm = [
                 timed(lambda: hot.execute(query, first).fetchall())[1] for _ in range(5)
@@ -380,6 +454,8 @@ def main():
     create.add_argument("--max-rows", type=int, default=1_000_000)
     check = commands.add_parser("probe")
     check.add_argument("output", type=Path)
+    rebuild = commands.add_parser("rebuild-hot")
+    rebuild.add_argument("output", type=Path)
     args = parser.parse_args()
     if args.command == "build":
         report = build(args.csv, args.manifest, args.output, args.max_rows)
@@ -397,6 +473,9 @@ def main():
                 }
             )
         )
+    elif args.command == "rebuild-hot":
+        report = rebuild_hot(args.output)
+        print(json.dumps({key: report[key] for key in ("rows", "hot_sqlite_bytes")}))
     else:
         print(json.dumps(probe(args.output), indent=2))
 

research/conversion-model/test_lakehouse.py

@@ -7,9 +7,19 @@
 import tempfile
 import unittest
 from contextlib import closing
+from datetime import date
 from pathlib import Path
 
-from lakehouse import REQUIRED, OPTIONAL_CONTEXT, build, literal, probe, read_snapshot
+from lakehouse import (
+    HOT_COLUMNS,
+    OPTIONAL_CONTEXT,
+    REQUIRED,
+    build,
+    literal,
+    probe,
+    read_snapshot,
+    rebuild_hot,
+)
 
 
 @unittest.skipUnless(
@@ -205,10 +215,34 @@ def test_probe_detects_stale_or_corrupted_hot_projection(self):
         output = self.root / "snapshot"
         build(self.csv, self.manifest, output)
         with closing(sqlite3.connect(output / "hot.sqlite3")) as db, db:
-            db.execute("UPDATE races SET time_cs = time_cs + 1")
+            db.execute("UPDATE race_rows SET time_cs = time_cs + 1")
         with self.assertRaisesRegex(ValueError, "projection differs"):
             probe(output)
 
+    def test_rebuild_hot_restores_the_projection_from_the_pinned_snapshot(self):
+        output = self.root / "snapshot"
+        report = build(self.csv, self.manifest, output)
+        with closing(sqlite3.connect(output / "hot.sqlite3")) as db, db:
+            db.execute("UPDATE race_rows SET time_cs = time_cs + 1")
+        rebuilt = rebuild_hot(output)
+        self.assertEqual(rebuilt["iceberg_snapshot_id"], report["iceberg_snapshot_id"])
+        self.assertEqual(
+            rebuilt["hot_sqlite_bytes"], (output / "hot.sqlite3").stat().st_size
+        )
+        self.assertTrue(probe(output)["history_matches_iceberg"])
+        # The compatibility view returns exactly the snapshot's rows and values.
+        columns = ", ".join(HOT_COLUMNS)
+        with closing(sqlite3.connect(output / "hot.sqlite3")) as db:
+            hot = sorted(db.execute(f"SELECT {columns} FROM races").fetchall())
+        expected = sorted(
+            tuple(
+                value.isoformat() if isinstance(value, date) else value
+                for value in row.values()
+            )
+            for row in read_snapshot(self.csv)[0].select(HOT_COLUMNS).to_pylist()
+        )
+        self.assertEqual(hot, expected)
+
     def test_validation_rejects_conflicts_nonfinite_age_and_seed_records(self):
         for replacement in (
             {"time_cs": 6000},