pipekit/cli.py

@@ -115,8 +115,7 @@ 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
@@ -344,9 +343,6 @@ 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

@@ -1,85 +0,0 @@
-"""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,7 +4,6 @@ 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)
@@ -21,7 +20,7 @@ import traceback
 from dataclasses import dataclass
 
 from .. import drivers, jrunner, repo
-from . import cancel, dest, merge, watermark
+from . import cancel, merge, watermark
 from .cancel import RunCancelled
 
 
@@ -47,17 +46,13 @@ class LockBusy(RuntimeError):
 
 
 def run_module(module_id: int, *, group_run_id: int | None = None,
-                dry_run: bool = False, run_id: int | None = None,
+                dry_run: bool = False, run_id: int | None = None) -> RunOutcome:
-                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)
@@ -109,14 +104,6 @@ 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