Merge feat/dest-preflight: destination-table preflight in run path

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)

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']}"

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