feat(lake): M5 — notebooks read OPPS from the lake; ingest publishes it (closes #514)

Read path: conf.connect.ducklake() returns a read-only DuckDB
connection with the lake attached as the default database, so
'SELECT ... FROM opps.addendum_b' works unchanged. Auth is a dedicated
select-only postgres role (ducklake_ro, DUCKLAKE_RO_PASSWORD in the
notebooks container env) — notebooks never hold superuser creds, and
read-only is enforced at both the attach and the role. Both OPPS-
reading notebooks (skin_sub_pricing, skin_sub_cost_sharing) migrated
to lake queries; headless-verified zero cell errors in prod.

Write path cutover: ingest_opps.py publishes the OPPS tables to the
lake as its final step (docker-exec'd publish_opps_to_lake.py — the
postgres catalog and RustFS are compose-internal; --no-lake to skip).
Verified end-to-end: one host command now does monolith merge →
replica refresh → lake publish with read-back checks (219,665 +
7,776 + 2,520 rows, all OK).

The monolith's opps schema stays as a deprecated mirror for the
aco.pipe analytics graph and the read replica; documented removal
condition in docs/docs/duckdb-concurrency.md.

compose.yml

@@ -262,6 +262,9 @@ services:
       - RUSTFS_ENDPOINT=http://rustfs:9000
       - RUSTFS_ACCESS_KEY=${RUSTFS_ACCESS_KEY}
       - RUSTFS_SECRET_KEY=${RUSTFS_SECRET_KEY}
+      # Read-only DuckLake catalog access — conf.connect.ducklake()
+      # reads reference data from the lake (M5, #514).
+      - DUCKLAKE_RO_PASSWORD=${DUCKLAKE_RO_PASSWORD}
       - NESSIE_S3_ACCESS_KEY=${NESSIE_S3_ACCESS_KEY}
       - NESSIE_S3_SECRET_KEY=${NESSIE_S3_SECRET_KEY}
       - POLARIS_ROOT_SECRET=${POLARIS_ROOT_SECRET}

dev/scripts/ingest_opps.py

@@ -345,9 +345,61 @@ 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")
+    parser.add_argument(
+        "--no-lake",
+        action="store_true",
+        help="skip publishing to the DuckLake lakehouse",
+    )
     args = parser.parse_args()
 
     print("Ingesting CMS OPPS files into DuckDB ...")
@@ -373,6 +425,10 @@ def main() -> None:
     replica = publish_replica("aco")
     print(f"replica → {replica}")
 
+    if not args.no_lake:
+        print("\n--- Publishing to DuckLake ---")
+        publish_lake()
+
     print("\nDone.")
 
 

docs/docs/duckdb-concurrency.md

@@ -61,6 +61,26 @@ 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
+
+Since M5 (#514), OPPS reference tables live in **DuckLake** — postgres
+catalog (`ducklake` db) + Parquet on RustFS (`s3://lakehouse/ducklake/`),
+per the M3 decision record. Concurrency 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
+  unchanged. Compose-internal only — host-side code goes through
+  `docker exec`.
+- **Writing**: `aco.lake.DuckLakeContext` (see `dev/scripts/
+  publish_opps_to_lake.py`). `ingest_opps.py` publishes to the lake as its
+  final step (`--no-lake` to skip).
+- The monolith's `opps` schema remains as a **deprecated mirror** for the
+  `aco.pipe` analytics graph and the read replica; drop it once those
+  consumers migrate.
+
 ## Per-year ingests merge, not wipe
 
 `ingest_opps.py --year YYYY` used to `DROP TABLE` + `CREATE ... AS` from only

notebooks/skin_sub_cost_sharing.py

@@ -43,10 +43,17 @@ def _():
     from pfs.rules import RULES
 
     con = connect.duckdb()
+    # OPPS reference data lives in the DuckLake lakehouse (M5, #514);
+    # the monolith's opps schema is a deprecated mirror.
+    lake = connect.ducklake()
 
     def q(sql):
         return con.execute(sql).pl()
 
+    def ql(sql):
+        """Query the lake (OPPS reference tables)."""
+        return lake.execute(sql).pl()
+
     SKIN_CODES = "('15271','15272','15273','15274','15275','15276','15277','15278')"
 
     # Part B deductible history (published by CMS annually)
@@ -65,7 +72,7 @@ def _():
         2026: 257.00,
     }
 
-    return DEDUCTIBLES, RULES, SKIN_CODES, alt, con, pl, q
+    return DEDUCTIBLES, RULES, SKIN_CODES, alt, con, lake, pl, q, ql
 
 
 # ── 1. Application fee coinsurance over time ─────────────────────────
@@ -305,8 +312,8 @@ def _(mo):
 
 
 @app.cell(hide_code=True)
-def _(alt, q):
+def _(alt, ql):
-    opps_copay = q("""
+    opps_copay = ql("""
         SELECT year,
                count(*) as products,
                round(avg(minimum_unadjusted_copayment), 2) as avg_copay,

notebooks/skin_sub_pricing.py

@@ -39,11 +39,18 @@ def _():
     from conf import connect
 
     con = connect.duckdb()
+    # OPPS reference data lives in the DuckLake lakehouse (M5, #514);
+    # the monolith's opps schema is a deprecated mirror.
+    lake = connect.ducklake()
 
     def q(sql):
         return con.execute(sql).pl()
 
-    return alt, con, pl, q
+    def ql(sql):
+        """Query the lake (OPPS reference tables)."""
+        return lake.execute(sql).pl()
+
+    return alt, con, lake, pl, q, ql
 
 
 @app.cell(hide_code=True)
@@ -332,8 +339,8 @@ def _(mo):
 
 
 @app.cell(hide_code=True)
-def _(q):
+def _(ql):
-    opps_ts = q("""
+    opps_ts = ql("""
         SELECT year, hcpcs, short_description,
                status_indicator, apc, payment_rate
         FROM opps.skin_sub_addendum_b

src/conf/connect.py

@@ -158,6 +158,51 @@ def publish_replica(name: str = "aco") -> Any:
     return dst
 
 
+def ducklake(*, read_only: bool = True) -> Any:
+    """Return a DuckDB connection with the DuckLake lakehouse attached.
+
+    The lake (``stack.toml [lake.ducklake]``) is the authoritative store
+    for reference data (M5, #514): a postgres catalog serializes
+    writers, data is Parquet on RustFS, and readers get snapshot
+    isolation — no single-writer file lock at all.
+
+    Read-only connections use the ``ducklake_ro`` postgres role
+    (``DUCKLAKE_RO_PASSWORD``), falling back to ``POSTGRES_PASSWORD``;
+    writers require ``POSTGRES_PASSWORD``. The catalog host and RustFS
+    are compose-internal, so this only works inside a data-network
+    container (notebooks, api) — host-side scripts go through
+    ``docker exec`` (see dev/scripts/publish_opps_to_lake.py).
+
+    The lake catalog is the connection's default database: notebooks
+    query ``opps.addendum_b`` exactly as they did against the monolith.
+    """
+    from aco.lake import DuckLakeContext
+
+    dl = cfg.lake.ducklake
+    if read_only:
+        pw = os.environ.get("DUCKLAKE_RO_PASSWORD") or os.environ.get(
+            "POSTGRES_PASSWORD", ""
+        )
+        dsn = dl.catalog_ro if os.environ.get("DUCKLAKE_RO_PASSWORD") else dl.catalog
+    else:
+        pw = os.environ.get("POSTGRES_PASSWORD", "")
+        dsn = dl.catalog
+    if dsn.startswith("postgres:") and pw:
+        dsn = f"{dsn} password={pw}"
+
+    ctx = DuckLakeContext(
+        catalog_dsn=dsn,
+        data_path=dl.data_path,
+        s3_endpoint=dl.s3_endpoint,
+        read_only=read_only,
+    )
+    con = ctx._get_connection()
+    # Hand ownership to the caller — otherwise the context's __del__
+    # closes the connection as soon as ctx goes out of scope here.
+    ctx._connection = None
+    return con
+
+
 def trino(
     *,
     catalog: str = "",

stack.toml

@@ -131,6 +131,8 @@ catalog = "iceberg"
 # container.
 [lake.ducklake]
 catalog = "postgres:dbname=ducklake host=postgres user=postgres"
+# Read-only role for notebook readers (password: DUCKLAKE_RO_PASSWORD).
+catalog_ro = "postgres:dbname=ducklake host=postgres user=ducklake_ro"
 data_path = "s3://lakehouse/ducklake/"
 s3_endpoint = "rustfs:9000"
 

tests/conf/test_connect.py

@@ -343,3 +343,48 @@ class TestReplica:
             con.execute("UPDATE t SET x = 9")
         connect.publish_replica("custom")
         assert connect.duckdb("custom").execute("SELECT x FROM t").fetchone()[0] == 9
+
+
+class TestDucklakeFactory:
+    def test_returns_caller_owned_lake_connection(self, tmp_path, monkeypatch):
+        """connect.ducklake() with a file catalog: connection survives the
+        internal context's GC and resolves lake tables as the default db."""
+        from aco.lake import DuckLakeContext
+
+        # Seed a lake with a file catalog + local data path
+        seed = DuckLakeContext(
+            catalog_dsn=str(tmp_path / "cat.ducklake"),
+            data_path=str(tmp_path / "data") + "/",
+            read_only=False,
+        )
+        import narwhals as nw
+        import polars as pl
+
+        seed.save(
+            "opps.addendum_b",
+            nw.from_native(pl.DataFrame({"hcpcs": ["Q4186"], "year": [2026]})),
+            mode="replace",
+        )
+        seed.close()
+
+        from types import SimpleNamespace
+
+        _dl = SimpleNamespace(
+            catalog=str(tmp_path / "cat.ducklake"),
+            catalog_ro=str(tmp_path / "cat.ducklake"),
+            data_path=str(tmp_path / "data") + "/",
+            s3_endpoint="",
+        )
+        # cfg.lake materializes fresh objects per access — patch the whole
+        # cfg reference connect.ducklake() reads.
+        monkeypatch.setattr(
+            connect, "cfg", SimpleNamespace(lake=SimpleNamespace(ducklake=_dl))
+        )
+        import gc
+
+        con = connect.ducklake()
+        gc.collect()  # the internal context must not close our connection
+        assert con.execute("SELECT count(*) FROM opps.addendum_b").fetchone()[0] == 1
+        with pytest.raises(Exception, match='type "CREATE"'):
+            con.execute("CREATE TABLE opps.nope (x INT)")
+        con.close()