fix(lake): M5 covered OPPS only — build out PFS (refs #514)

The M5 close-out missed half the issue's scope: #514 says 'OPPS/PFS
reference data' and I cut over only OPPS, leaving PFS — the largest
reference domain, 23.5M rows across 8 tables — entirely on the
monolith, including pfs.* queries in the very notebook whose OPPS
query was migrated. This completes PFS the same way:

- publish_opps_to_lake.py → publish_reference_to_lake.py with a
  schema registry (opps: 3 tables, pfs: 8); host-side docker-exec
  wrapper extracted to dev/scripts/_lake.py, shared by the ingests.
- PFS published to the lake and read-back verified: carrier_locality
  21,863,770 rows in 10.1s, plus rvu/gpci/clinical_labor/medical_
  equipment/medical_supply/physician_work_time/zip_carrier_locality.
- New dev/scripts/ingest_pfs.py wraps pfs.pipe.load_all (previously
  ad-hoc, no entrypoint) with the standard plumbing: duckdb_batch
  preflight, replica refresh, lake publish.
- 5 notebooks migrated: pfs_calcs, pfs_reconciliation,
  skin_sub_budget_neutrality read the lake as their primary
  connection; skin_sub_pricing and skin_sub_cost_sharing switch their
  pure-pfs cells to the lake. The one cross-source join
  (pfs × skin_subs) stays on the monolith mirror, annotated.
  All 5 headless-verified in prod: zero cell errors.
- pfs_calcs leaves the pre-commit host-run safe list (the lake catalog
  is compose-internal); the nightly integration covers it in-container.

dev/scripts/_lake.py

@@ -0,0 +1,62 @@
+"""Host-side lake publish helper shared by the ingest scripts.
+
+The DuckLake catalog (postgres) and RustFS are compose-internal, so
+publishing runs docker-exec'd in the notebooks container with
+``POSTGRES_PASSWORD`` from ``.env``. Import from a sibling dev script
+(the script's own directory is on ``sys.path``)::
+
+    import _lake
+    _lake.publish_lake(("opps",))
+"""
+
+from __future__ import annotations
+
+import subprocess
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[2]
+
+
+def publish_lake(schemas: tuple[str, ...]) -> None:
+    """Publish reference schemas to the DuckLake lakehouse (M5, #514).
+
+    The lake is the authoritative store for reference data; the
+    monolith's copies stay as a deprecated mirror for the analytics
+    pipe and the read replica.
+    """
+    pw = ""
+    env_file = ROOT / ".env"
+    if env_file.exists():
+        for line in env_file.read_text().splitlines():
+            if line.startswith("POSTGRES_PASSWORD="):
+                pw = line.split("=", 1)[1].strip()
+                break
+    if not pw:
+        raise SystemExit(
+            "lake publish: POSTGRES_PASSWORD not in .env (use --no-lake to skip)"
+        )
+    script = ROOT / "dev" / "scripts" / "publish_reference_to_lake.py"
+    subprocess.run(
+        ["docker", "cp", str(script), "notebooks:/tmp/publish_reference_to_lake.py"],
+        check=True,
+    )
+    subprocess.run(
+        [
+            "docker",
+            "exec",
+            "-e",
+            f"POSTGRES_PASSWORD={pw}",
+            "-e",
+            "PYTHONPATH=/home/kert/src",
+            "notebooks",
+            "uv",
+            "run",
+            "--project",
+            "/home/kert/workspace",
+            "python",
+            "/tmp/publish_reference_to_lake.py",
+            "--schemas",
+            *schemas,
+        ],
+        check=True,
+    )

dev/scripts/ingest_opps.py

@@ -345,53 +345,6 @@ def ingest_all(con: duckdb.DuckDBPyConnection, year_filter: str = "") -> None:
                 print(f"    CY{yr} SI={si}: {ct} codes")
 
 
-def publish_lake() -> None:
-    """Publish the OPPS tables to the DuckLake lakehouse (M5, #514).
-
-    The lake is the authoritative store for OPPS reference data; the
-    monolith's opps schema stays as a deprecated mirror for the
-    analytics pipe and the read replica. The catalog (postgres) and
-    RustFS are compose-internal, so the publish runs docker-exec'd in
-    the notebooks container with POSTGRES_PASSWORD from .env.
-    """
-    import subprocess
-
-    pw = ""
-    env_file = ROOT / ".env"
-    if env_file.exists():
-        for line in env_file.read_text().splitlines():
-            if line.startswith("POSTGRES_PASSWORD="):
-                pw = line.split("=", 1)[1].strip()
-                break
-    if not pw:
-        raise SystemExit(
-            "lake publish: POSTGRES_PASSWORD not in .env (use --no-lake to skip)"
-        )
-    script = ROOT / "dev" / "scripts" / "publish_opps_to_lake.py"
-    subprocess.run(
-        ["docker", "cp", str(script), "notebooks:/tmp/publish_opps_to_lake.py"],
-        check=True,
-    )
-    subprocess.run(
-        [
-            "docker",
-            "exec",
-            "-e",
-            f"POSTGRES_PASSWORD={pw}",
-            "-e",
-            "PYTHONPATH=/home/kert/src",
-            "notebooks",
-            "uv",
-            "run",
-            "--project",
-            "/home/kert/workspace",
-            "python",
-            "/tmp/publish_opps_to_lake.py",
-        ],
-        check=True,
-    )
-
-
 def main() -> None:
     parser = argparse.ArgumentParser(description="Ingest CMS OPPS files")
     parser.add_argument("--year", help="Ingest only this year")
@@ -427,7 +380,9 @@ def main() -> None:
 
     if not args.no_lake:
         print("\n--- Publishing to DuckLake ---")
-        publish_lake()
+        import _lake
+
+        _lake.publish_lake(("opps",))
 
     print("\nDone.")
 

dev/scripts/ingest_pfs.py

@@ -0,0 +1,62 @@
+"""Ingest CMS PFS reference data into DuckDB, then replica + lake.
+
+Wraps ``pfs.pipe.load_all`` (RVU / GPCI / carrier locality / labor /
+equipment / supply files, sourced from Zotero attachments) with the
+standard ingest plumbing: lock-preflighted write connection (#508),
+read-replica refresh (#510), and DuckLake publish (#514). PFS loads
+were previously ad-hoc ``load_all`` calls with none of that.
+
+Usage:
+    uv run python dev/scripts/ingest_pfs.py
+    uv run python dev/scripts/ingest_pfs.py --years 2020 2026 --no-lake
+"""
+
+from __future__ import annotations
+
+import argparse
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[2]
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--years",
+        nargs=2,
+        type=int,
+        metavar=("FROM", "TO"),
+        help="inclusive year range (default: all available)",
+    )
+    parser.add_argument(
+        "--no-lake",
+        action="store_true",
+        help="skip publishing to the DuckLake lakehouse",
+    )
+    args = parser.parse_args()
+
+    from conf.connect import duckdb_batch, publish_replica
+    from pfs.pipe import load_all
+
+    print("Ingesting CMS PFS files into DuckDB ...")
+    with duckdb_batch("aco") as con:
+        summary = load_all(con, years=tuple(args.years) if args.years else None)
+        print("\n--- PFS tables ---")
+        for name, info in sorted(summary.items()):
+            print(f"  pfs.{name:24s}: {info}")
+
+    replica = publish_replica("aco")
+    print(f"replica → {replica}")
+
+    if not args.no_lake:
+        print("\n--- Publishing to DuckLake ---")
+        import _lake
+
+        _lake.publish_lake(("pfs",))
+
+    print("\nDone.")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

dev/scripts/publish_reference_to_lake.py

@@ -20,13 +20,29 @@ import argparse
 import os
 import time
 
-TABLES = ("addendum_b", "apc_weight", "skin_sub_addendum_b")
+SCHEMAS: dict[str, tuple[str, ...]] = {
+    "opps": ("addendum_b", "apc_weight", "skin_sub_addendum_b"),
+    "pfs": (
+        "carrier_locality",
+        "clinical_labor",
+        "gpci",
+        "medical_equipment",
+        "medical_supply",
+        "physician_work_time",
+        "rvu",
+        "zip_carrier_locality",
+    ),
+}
 
 
 def main() -> int:
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument(
-        "--tables", nargs="*", default=list(TABLES), help="opps tables to publish"
+        "--schemas",
+        nargs="*",
+        default=list(SCHEMAS),
+        choices=list(SCHEMAS),
+        help="reference schemas to publish",
     )
     args = parser.parse_args()
 
@@ -48,18 +64,19 @@ def main() -> int:
     )
 
     src = connect.duckdb("aco")  # read-only; resolves to the replica
-    for table in args.tables:
+    for schema in args.schemas:
-        t = time.time()
+        for table in SCHEMAS[schema]:
-        df = src.execute(f"SELECT * FROM opps.{table}").pl()  # noqa: S608
+            t = time.time()
-        ctx.save(f"opps.{table}", nw.from_native(df), mode="replace")
+            df = src.execute(f"SELECT * FROM {schema}.{table}").pl()  # noqa: S608
-        back = nw.to_native(ctx.load(f"opps.{table}"))
+            ctx.save(f"{schema}.{table}", nw.from_native(df), mode="replace")
-        status = "OK" if back.height == df.height else "MISMATCH"
+            back = nw.to_native(ctx.load(f"{schema}.{table}"))
-        print(
+            status = "OK" if back.height == df.height else "MISMATCH"
-            f"  opps.{table}: {df.height} rows → lake ({back.height} read back) "
+            print(
-            f"[{status}] in {time.time() - t:.1f}s"
+                f"  {schema}.{table}: {df.height} rows → lake "
-        )
+                f"({back.height} read back) [{status}] in {time.time() - t:.1f}s"
-        if status != "OK":
+            )
-            return 1
+            if status != "OK":
+                return 1
     src.close()
     print("lake publish complete")
     return 0

docs/docs/duckdb-concurrency.md

@@ -61,25 +61,29 @@ primary's lock at all, and ingests stop caring how many notebooks are open.
 - Kernels holding the *old* snapshot keep a valid file handle after a swap;
   re-running the connect cell picks up the fresh one.
 
-## OPPS reference data is authoritative in the lake
+## Reference data (OPPS + PFS) is authoritative in the lake
 
-Since M5 (#514), OPPS reference tables live in **DuckLake** — postgres
+Since M5 (#514), the OPPS and PFS reference tables live in **DuckLake** —
-catalog (`ducklake` db) + Parquet on RustFS (`s3://lakehouse/ducklake/`),
+postgres catalog (`ducklake` db) + Parquet on RustFS
-per the M3 decision record. Concurrency there is structural: the catalog
+(`s3://lakehouse/ducklake/`), per the M3 decision record. Concurrency
-serializes writers transactionally and readers get snapshot isolation.
+there is structural: the catalog serializes writers transactionally and
+readers get snapshot isolation.
 
 - **Reading (notebooks / in-container code)**: `conf.connect.ducklake()` —
   read-only by default via the `ducklake_ro` postgres role
   (`DUCKLAKE_RO_PASSWORD`, in the notebooks container env). The lake is the
-  connection's default database, so `SELECT … FROM opps.addendum_b` works
+  connection's default database, so `SELECT … FROM opps.addendum_b` or
-  unchanged. Compose-internal only — host-side code goes through
+  `pfs.rvu` works unchanged. Compose-internal only — host-side code goes
-  `docker exec`.
+  through `docker exec`.
 - **Writing**: `aco.lake.DuckLakeContext` (see `dev/scripts/
-  publish_opps_to_lake.py`). `ingest_opps.py` publishes to the lake as its
+  publish_reference_to_lake.py`). `ingest_opps.py` and `ingest_pfs.py`
-  final step (`--no-lake` to skip).
+  publish to the lake as their final step (`--no-lake` to skip).
-- The monolith's `opps` schema remains as a **deprecated mirror** for the
+- The monolith's `opps`/`pfs` schemas remain as a **deprecated mirror** for
-  `aco.pipe` analytics graph and the read replica; drop it once those
+  the `aco.pipe` analytics graph, the read replica, and cross-source joins
-  consumers migrate.
+  (e.g. pfs × skin_subs); drop them once those consumers migrate.
+- Cross-source queries (lake schema joined with a monolith-only schema like
+  `skin_subs`) stay on the monolith connection until the joined schema also
+  moves.
 
 ## Per-year ingests merge, not wipe
 

notebooks/pfs_calcs.py

@@ -28,7 +28,9 @@ def _():
 
     from conf import connect
 
-    con = connect.duckdb()
+    # PFS reference data lives in the DuckLake lakehouse (M5, #514);
+    # queries are unchanged — the lake is the default database.
+    con = connect.ducklake()
 
     def q(sql):
         return con.execute(sql).pl()

notebooks/pfs_reconciliation.py

@@ -29,7 +29,9 @@ def _():
     from rec.engine import reconcile
     from rec.pricers.pfs import PfsPricer
 
-    con = connect.duckdb()
+    # PFS reference data lives in the DuckLake lakehouse (M5, #514);
+    # queries are unchanged — the lake is the default database.
+    con = connect.ducklake()
     pricer = PfsPricer()
     return con, pricer, reconcile
 

notebooks/skin_sub_budget_neutrality.py

@@ -41,7 +41,9 @@ def _():
     from conf import connect
     from pfs.rules import RULES
 
-    con = connect.duckdb()
+    # PFS reference data lives in the DuckLake lakehouse (M5, #514);
+    # queries are unchanged — the lake is the default database.
+    con = connect.ducklake()
 
     def q(sql):
         return con.execute(sql).pl()

notebooks/skin_sub_cost_sharing.py

@@ -90,7 +90,7 @@ def _(mo):
 
 
 @app.cell(hide_code=True)
-def _(alt, q):
+def _(alt, ql):
     regions = [
         "MANHATTAN",
         "REST OF FLORIDA",
@@ -100,7 +100,7 @@ def _(alt, q):
     ]
     region_list = ", ".join(f"'{r}'" for r in regions)
 
-    app_coinsurance = q(f"""
+    app_coinsurance = ql(f"""
         SELECT c.year, g.locality_name,
                c.non_fac_fee,
                round(c.non_fac_fee * 0.20, 2) as bene_coinsurance,
@@ -226,7 +226,9 @@ def _(mo):
 
 @app.cell(hide_code=True)
 def _(alt, pl, q):
-    # Use Q1 of each year for annual comparison
+    # Cross-source join (skin_subs lives in the monolith, pfs in the
+    # lake) — reads the monolith's deprecated pfs mirror until skin_subs
+    # moves to the lake too.
     episode_sharing = q("""
         WITH asp_annual AS (
             SELECT CAST(substr(quarter, 1, 4) AS INTEGER) as year,
@@ -448,9 +450,9 @@ def _(mo):
 
 
 @app.cell(hide_code=True)
-def _(pl, q):
+def _(pl, ql):
     # Get average national app fee for 2025
-    avg_fee = q("""
+    avg_fee = ql("""
         SELECT round(avg(non_fac_fee), 2) as avg_fee
         FROM pfs.carrier_locality
         WHERE year = 2025 AND hcpcs = '15271'

notebooks/skin_sub_pricing.py

@@ -66,8 +66,8 @@ def _(mo):
 
 
 @app.cell(hide_code=True)
-def _(alt, q):
+def _(alt, ql):
-    rvu_ts = q("""
+    rvu_ts = ql("""
         SELECT year, hcpcs,
                CASE hcpcs
                    WHEN '15271' THEN 'Trunk/limbs <100cm² (initial)'
@@ -508,8 +508,8 @@ def _(mo):
 
 
 @app.cell(hide_code=True)
-def _(q):
+def _(ql):
-    fee_summary = q("""
+    fee_summary = ql("""
         SELECT c.year,
                count(*) as localities,
                round(min(c.non_fac_fee), 2) as min_fee,

src/sem/hooks.py

@@ -333,12 +333,15 @@ def main() -> int:
         if rc != 0:
             return rc
 
-        # Only execute notebooks in the safe-to-run list
+        # Only execute notebooks in the safe-to-run list.
+        # pfs_calcs left this list at M5 (#514): it reads PFS from the
+        # DuckLake lakehouse, whose postgres catalog is compose-internal
+        # — host-side execution can't reach it. The nightly
+        # notebooks-integration run covers it in-container.
         safe = {
             "acodb_explorer.py",
             "bib_explorer.py",
             "cms_quality_measures.py",
-            "pfs_calcs.py",
             "sql_generator.py",
         }
         for nb in cats["notebooks"]: