diff --git a/pipekit/cli.py b/pipekit/cli.py index d82fa95..b94f2ce 100644 --- a/pipekit/cli.py +++ b/pipekit/cli.py @@ -115,7 +115,8 @@ def cmd_run(args) -> int: print(f"error: module {args.module!r} not found") return 1 try: - outcome = engine.run_module(module["id"], dry_run=args.dry_run) + outcome = engine.run_module(module["id"], dry_run=args.dry_run, + create_dest=args.create_dest) except engine.LockBusy as e: print(f"busy: {e}") return 1 @@ -343,6 +344,9 @@ def main(argv: list[str] | None = None) -> int: p_run = sub.add_parser("run", help="run a module by name (synchronous)") p_run.add_argument("module", help="module name") + p_run.add_argument("--create-dest", action="store_true", + help="create the destination table from the module's " + "column map if it does not exist") p_run.add_argument("--dry-run", action="store_true", help="build SQL but do not invoke jrunner") p_run.set_defaults(func=cmd_run) diff --git a/pipekit/engine/dest.py b/pipekit/engine/dest.py new file mode 100644 index 0000000..c569fd0 --- /dev/null +++ b/pipekit/engine/dest.py @@ -0,0 +1,85 @@ +"""Destination-table preflight. + +Ensure a module's destination table exists and covers the module's column map +*before* a run builds staging as ``LIKE dest`` and merges into it. Without this +the engine would fail deep inside the staging step with a raw +``relation ... does not exist`` when the dest was never provisioned (e.g. a +file-based deploy that never went through the wizard). + +Kept driver-agnostic — all DDL/introspection goes through the dest +:class:`~pipekit.drivers.base.Driver`. +""" + +from __future__ import annotations + +import json + +from .. import jrunner + + +class DestError(RuntimeError): + """Dest table is missing (and create not requested) or has drifted.""" + + +def _split(dest_table: str) -> tuple[str, str]: + """Split ``schema.table`` → (schema, table); default schema is ``public``.""" + schema, _, bare = dest_table.partition(".") + if not bare: + return "public", schema + return schema, bare + + +def module_columns(module: dict) -> list[dict]: + """The module's stored column map, or [] if none recorded.""" + raw = module.get("columns_json") + return json.loads(raw) if raw else [] + + +def reconcile_dest(module: dict, dest_drv, dest_conn: dict, *, + create: bool) -> str: + """Ensure ``module['dest_table']`` exists on ``dest_conn`` and covers the + module's dest columns. Returns a short action string for the run log. + + * missing + ``create`` → CREATE SCHEMA (if needed) + CREATE TABLE from + the module's column map. + * missing + not ``create`` → raise :class:`DestError`. + * exists → raise :class:`DestError` if any dest column is + absent (drift); otherwise a no-op. + """ + if dest_drv is None: + return "skipped (no dest driver)" + + schema, bare = _split(module["dest_table"]) + try: + existing = dest_drv.check_dest_table(dest_conn, schema, bare) + except jrunner.JrunnerError as e: + raise DestError( + f"could not introspect dest {module['dest_table']}: {e}") from e + + cols = module_columns(module) + + if existing is None: + if not create: + raise DestError( + f"dest table {module['dest_table']} does not exist — re-run " + "with --create-dest to provision it from the module's columns") + if not cols: + raise DestError( + f"cannot create {module['dest_table']}: module has no column map") + qualified = dest_drv.qualified_table_name(bare, schema=schema) + if schema: + jrunner.run_dest_sql(dest_conn, dest_drv.create_schema_sql(schema)) + jrunner.run_dest_sql( + dest_conn, dest_drv.build_create_table_sql(qualified, cols)) + return f"created {module['dest_table']} ({len(cols)} columns)" + + # Exists — verify its shape covers every column we intend to load. The + # merge is a positional `SELECT *`, so a dest missing a column would either + # error or silently misalign; fail loudly instead. + missing = [c["dest_name"] for c in cols + if c["dest_name"].lower() not in existing] + if missing: + raise DestError( + f"dest table {module['dest_table']} is missing column(s): " + f"{', '.join(missing)} — reconcile the schema before running") + return f"verified {module['dest_table']}" diff --git a/pipekit/engine/runner.py b/pipekit/engine/runner.py index 413bd26..cea73ad 100644 --- a/pipekit/engine/runner.py +++ b/pipekit/engine/runner.py @@ -4,6 +4,7 @@ Steps: 1. acquire lock atomically (repo.acquire_module_lock) 2. resolve watermarks (watermark.resolve_watermarks) 3. materialise source query, persist preview (watermark.materialise + repo) + 3b. dest preflight: exists + shape (opt. create) (dest.reconcile_dest) 4. ensure staging table exists on dest (CREATE TABLE IF NOT EXISTS ... LIKE dest) 5. jrunner migrate source → staging (jrunner.migrate — clears staging internally) 6. build merge SQL (merge.build_merge_sql) @@ -20,7 +21,7 @@ import traceback from dataclasses import dataclass from .. import drivers, jrunner, repo -from . import cancel, merge, watermark +from . import cancel, dest, merge, watermark from .cancel import RunCancelled @@ -46,13 +47,17 @@ class LockBusy(RuntimeError): def run_module(module_id: int, *, group_run_id: int | None = None, - dry_run: bool = False, run_id: int | None = None) -> RunOutcome: + dry_run: bool = False, run_id: int | None = None, + create_dest: bool = False) -> RunOutcome: """Run one module end-to-end. In dry-run mode, SQL is generated and stored on the run_log but no jrunner calls are made. If ``run_id`` is provided, that run_log row is reused — this lets async callers (the API) reserve a run_id before the run starts so they can return it to the client immediately. + + ``create_dest`` provisions the destination table from the module's column + map when it doesn't yet exist; otherwise a missing dest fails the run. """ module = repo.get_module(module_id) @@ -104,6 +109,14 @@ def run_module(module_id: int, *, group_run_id: int | None = None, status = "dry_run" return RunOutcome(run_id, status, None, None, resolved_sql, merge_sql) + # 3b. dest preflight — the dest must exist and cover our columns before + # staging is built as LIKE dest. Optionally provision it from the + # module's column map (file-based deploys never hit the wizard that + # would otherwise create it). + dest_action = dest.reconcile_dest(module, dest_drv, dest_conn, + create=create_dest) + repo.append_run_live_log(run_id, f"-- dest: {dest_action}") + # 4. (re)create staging from dest. DROP+CREATE (not IF NOT EXISTS) so # any drift — dest columns added since staging was last made — is # self-healing. Staging is ephemeral per SPEC; nothing of value lives