feat(lake): publish gpci_proposed + primary_care_service_code to DuckLake (closes #630)

This commit is contained in:
kert
2026-08-17 12:17:35 -04:00
parent 39dfaa20ae
commit fc26d5fc54

View File

@@ -26,6 +26,7 @@ SCHEMAS: dict[str, tuple[str, ...]] = {
"carrier_locality", "carrier_locality",
"clinical_labor", "clinical_labor",
"gpci", "gpci",
"gpci_proposed",
"medical_equipment", "medical_equipment",
"medical_supply", "medical_supply",
"physician_work_time", "physician_work_time",
@@ -36,6 +37,19 @@ SCHEMAS: dict[str, tuple[str, ...]] = {
"cms": ("ingest_log",), "cms": ("ingest_log",),
} }
# (monolith source, lake destination) — tables whose lake name differs
# from their monolith name. The MSSP designated-primary-care-service
# list (42 CFR 425.400(c), used for ACO beneficiary assignment) lives
# monolith-side as a tuva/dbt intermediate; publish it under a stable
# reference name the notebooks can cite (#630). Published whenever the
# destination's schema is selected via --schemas.
RENAMED: tuple[tuple[str, str], ...] = (
(
"cms_provider_attribution._primary_care_hcpcs_codes",
"cms.primary_care_service_code",
),
)
def main() -> int: def main() -> int:
parser = argparse.ArgumentParser(description=__doc__) parser = argparse.ArgumentParser(description=__doc__)
@@ -66,18 +80,27 @@ def main() -> int:
) )
src = connect.duckdb("aco") # read-only; resolves to the replica src = connect.duckdb("aco") # read-only; resolves to the replica
def _publish(source: str, dest: str) -> bool:
t = time.time()
df = src.execute(f"SELECT * FROM {source}").pl() # noqa: S608
ctx.save(dest, nw.from_native(df), mode="replace")
back = nw.to_native(ctx.load(dest))
ok = back.height == df.height
label = source if source == dest else f"{source} → {dest}"
print(
f" {label}: {df.height} rows → lake "
f"({back.height} read back) [{'OK' if ok else 'MISMATCH'}] "
f"in {time.time() - t:.1f}s"
)
return ok
for schema in args.schemas: for schema in args.schemas:
for table in SCHEMAS[schema]: for table in SCHEMAS[schema]:
t = time.time() if not _publish(f"{schema}.{table}", f"{schema}.{table}"):
df = src.execute(f"SELECT * FROM {schema}.{table}").pl() # noqa: S608 return 1
ctx.save(f"{schema}.{table}", nw.from_native(df), mode="replace") for source, dest in RENAMED:
back = nw.to_native(ctx.load(f"{schema}.{table}")) if dest.split(".", 1)[0] in args.schemas and not _publish(source, dest):
status = "OK" if back.height == df.height else "MISMATCH"
print(
f" {schema}.{table}: {df.height} rows → lake "
f"({back.height} read back) [{status}] in {time.time() - t:.1f}s"
)
if status != "OK":
return 1 return 1
src.close() src.close()
print("lake publish complete") print("lake publish complete")