diff --git a/src/ol_dbt/README.md b/src/ol_dbt/README.md
index b8a86ccaa..b5e8655ed 100644
--- a/src/ol_dbt/README.md
+++ b/src/ol_dbt/README.md
@@ -103,6 +103,59 @@ ol-dbt run test
 
 The `.dbt-state/` directory is gitignored and local to your machine.
 
+### Verifying a Change with `ol-dbt diff` and `ol-dbt local snapshot`
+
+When you change a model's SQL, `ol-dbt diff` compares two relations row-by-row and
+column-by-column so you can confirm exactly what changed — instead of eyeballing
+`select *` output.
+
+To isolate your code change from any upstream data drift, snapshot the model's
+current build as a frozen baseline *before* making your change, then diff that
+baseline against the rebuild:
+
+```bash
+# 1. Freeze the current (pre-change) build as a baseline table
+ol-dbt local snapshot my_model --as my_model_baseline
+
+# 2. Make your code change, then rebuild
+ol-dbt run --select my_model --target dev_local
+
+# 3. Diff the frozen baseline against the rebuild
+ol-dbt diff --target dev_local \
+    --old my_model_baseline --old-raw \
+    --new my_model \
+    --primary-key my_model_pk
+```
+
+Because both sides are built from the same underlying data (the baseline is frozen,
+the rebuild only reflects your code change), any reported mismatch can only come
+from your change — never from production data changing in the background.
+
+```bash
+# Per-column mismatch rates require --primary-key (repeatable for a composite key)
+ol-dbt diff --old m_old --new m_new --primary-key id
+
+# Exclude a known non-deterministic column (e.g. a load timestamp)
+ol-dbt diff --old m_old --new m_new -k id --exclude-columns _loaded_at
+
+# Build both sides first, then compare
+ol-dbt diff --old m_old --new m_new --auto-build
+
+# Compare two already-materialized copies of the same model across schemas
+# (e.g. your personal dev schema vs. production) on one Trino target
+ol-dbt diff --target dev_production \
+    --old my_model --old-raw --old-schema ol_warehouse_production_reporting \
+    --new my_model --new-raw --new-schema ol_warehouse_production_rlougee_reporting \
+    --primary-key id
+
+# Emit JSON (e.g. for CI); exits non-zero on any divergence
+ol-dbt diff --old m_old --new m_new -k id --format json
+```
+
+`--old-raw`/`--new-raw` treat that side as a literal existing table rather than a
+dbt model — required for a snapshot table (`ol-dbt local snapshot` output) or any
+other already-materialized relation dbt doesn't know about via `ref()`.
+
 ### Running dbt Directly
 
 You can still invoke dbt commands directly from `src/ol_dbt/`:
diff --git a/src/ol_dbt_cli/ol_dbt_cli/commands/diff.py b/src/ol_dbt_cli/ol_dbt_cli/commands/diff.py
index 7b4dbc4d4..48948acdf 100644
--- a/src/ol_dbt_cli/ol_dbt_cli/commands/diff.py
+++ b/src/ol_dbt_cli/ol_dbt_cli/commands/diff.py
@@ -16,6 +16,11 @@
   ``--exclude-columns``.
 * The comparison itself is run via ``dbt show --inline`` invoking audit_helper
   macros with ``--output json``; results are parsed tolerantly from stdout.
+* ``--old``/``--new`` normally resolve through ``ref()``, which requires a real
+  dbt model in the manifest. ``--old-raw``/``--new-raw`` bypass that, building a
+  literal ``api.Relation.create(...)`` instead — for relations dbt doesn't know
+  about: an already-materialized table sitting in a different schema/database
+  than the current --target.
 
 Driveable both by hand and by CI (Phase 2 auto-diff vs a base ref). See
 docs/specs/DBT_WAREHOUSE_CI_QA_SPEC.md §3.
@@ -113,6 +118,14 @@ class DiffResult:
     column_mismatches: list[ColumnMismatch]
     sample_mismatches: list[dict[str, Any]]
     verdict: Verdict
+    # Text-output-only labels annotating each side's source (e.g. "name (raw)")
+    # when --old/new-raw make `old`/`new` ambiguous — identical to `old`/`new`
+    # for a plain ref()-vs-ref() diff. Kept separate from `old`/`new` so JSON
+    # output stays a stable, bare identifier. Uses parens, not brackets —
+    # rich.console.print() parses `[...]` as markup and silently swallows
+    # anything that isn't a recognized style tag.
+    old_label: str
+    new_label: str
     notes: list[str] = field(default_factory=list)
 
 
@@ -146,6 +159,39 @@ def _resolve_columns(
     return set()
 
 
+def _resolve_raw_columns(
+    name: str,
+    dbt_dir: Path,
+    target: str,
+    *,
+    database: str | None = None,
+    schema: str | None = None,
+) -> tuple[set[str], str | None]:
+    """Resolve a raw (non-``ref()``) relation's columns by sampling one row.
+
+    Raw relations (an already-materialized table in another schema) have no
+    manifest/YAML metadata to resolve columns from statically, so this samples
+    a live row and reads its keys instead.
+
+    Returns ``(columns, error)``. *error* carries the dbt failure message when
+    the sampling query itself fails (e.g. the relation doesn't exist) — the
+    caller must surface it rather than let an empty column set masquerade as
+    "unparsed / not in manifest", which is misleading for a raw relation.
+    """
+    expr = _relation_jinja(name, raw=True, database=database, schema=schema)
+    # Rendered into a Jinja template compiled by dbt (not executed as a raw query
+    # here), so the S608 heuristic is a false positive — same as _compare_column_sql.
+    # No literal LIMIT here: `dbt show` already wraps the compiled query in its own
+    # limiting logic via --limit below; a second, literal `limit 1` in the SQL text
+    # collides with that wrapper and breaks the DuckDB parser.
+    inline_sql = f"select * from {{{{ {expr} }}}}"  # noqa: S608
+    try:
+        rows = _run_dbt_show(inline_sql, dbt_dir, target, limit=1)
+    except RuntimeError as exc:
+        return set(), str(exc)
+    return ({c.lower() for c in rows[0]} if rows else set()), None
+
+
 def reconcile_columns(
     old_cols: set[str],
     new_cols: set[str],
@@ -276,6 +322,28 @@ def _jinja_list(values: list[str]) -> str:
     return f"[{inner}]"
 
 
+def _relation_jinja(
+    name: str,
+    *,
+    raw: bool = False,
+    database: str | None = None,
+    schema: str | None = None,
+) -> str:
+    """Render a relation reference for inline Jinja SQL.
+
+    Normally a plain ``ref('name')`` — requires *name* to be a real dbt model in
+    the manifest. When ``raw=True``, builds a literal ``api.Relation.create(...)``
+    instead, addressed by *database*/*schema* (defaulting to the current
+    ``--target``'s own database/schema) — for relations dbt doesn't know about,
+    e.g. an already-materialized table elsewhere.
+    """
+    if not raw:
+        return f"ref('{name}')"
+    db = f"'{database}'" if database else "target.database"
+    sch = f"'{schema}'" if schema else "target.schema"
+    return f"api.Relation.create(database={db}, schema={sch}, identifier='{name}')"
+
+
 def _compare_relations_sql(
     old: str,
     new: str,
@@ -283,6 +351,12 @@ def _compare_relations_sql(
     exclude_columns: list[str],
     *,
     summarize: bool,
+    old_raw: bool = False,
+    new_raw: bool = False,
+    old_schema: str | None = None,
+    new_schema: str | None = None,
+    old_database: str | None = None,
+    new_database: str | None = None,
 ) -> str:
     """Build inline SQL calling ``audit_helper.compare_relations``.
 
@@ -295,23 +369,56 @@ def _compare_relations_sql(
         pk_repr = f"'{primary_key[0]}'" if len(primary_key) == 1 else _jinja_list(primary_key)
         pk_arg = f", primary_key={pk_repr}"
     exclude_arg = f", exclude_columns={_jinja_list(exclude_columns)}" if exclude_columns else ""
+    a_expr = _relation_jinja(old, raw=old_raw, database=old_database, schema=old_schema)
+    b_expr = _relation_jinja(new, raw=new_raw, database=new_database, schema=new_schema)
+    # `}}` in an f-string is an escape for a single literal `}` (same as `{{` for
+    # `{`) — closing the Jinja `{{ ... }}` block from an f-string needs `}}}}`.
     return (
         "{{ audit_helper.compare_relations("
-        f"a_relation=ref('{old}'), b_relation=ref('{new}')"
-        f"{exclude_arg}{pk_arg}, summarize={'true' if summarize else 'false'}) }}"
+        f"a_relation={a_expr}, b_relation={b_expr}"
+        f"{exclude_arg}{pk_arg}, summarize={'true' if summarize else 'false'}) }}}}"
     )
 
 
-def _compare_column_sql(old: str, new: str, primary_key: list[str], column: str) -> str:
+def _compare_column_sql(
+    old: str,
+    new: str,
+    primary_key: list[str],
+    column: str,
+    *,
+    old_raw: bool = False,
+    new_raw: bool = False,
+    old_schema: str | None = None,
+    new_schema: str | None = None,
+    old_database: str | None = None,
+    new_database: str | None = None,
+) -> str:
     """Build inline SQL calling ``audit_helper.compare_column_values`` for one column."""
     pk = f"'{primary_key[0]}'" if len(primary_key) == 1 else _jinja_list(primary_key)
-    # Model names are dbt identifiers rendered into a Jinja template compiled by
-    # dbt (not executed as a raw query here), so the S608 heuristic is a false positive.
+    a_expr = _relation_jinja(old, raw=old_raw, database=old_database, schema=old_schema)
+    b_expr = _relation_jinja(new, raw=new_raw, database=new_database, schema=new_schema)
+    # Only the primary key(s) + the one column being compared are needed here --
+    # `compare_column_values` does a full outer join on a_query/b_query, and
+    # selecting every column (as this used to) forces that join to carry the
+    # whole row width for a single-column comparison, which is needless I/O on
+    # wide tables.
+    select_cols = ", ".join(dict.fromkeys([*primary_key, column]))
+    # a_query/b_query are built with {% set %}/{% endset %} so `a_expr`/`b_expr`
+    # (a ref()/api.Relation.create() call) are rendered by Jinja BEFORE
+    # compare_column_values ever sees the string. Embedding `{{ ... }}` directly
+    # inside a quoted string argument (the previous approach) is inert: Jinja
+    # treats string literals atomically and never re-parses braces inside them,
+    # so the literal text `{{ ref(...) }}` reached the database unrendered --
+    # a real, pre-existing bug (every diff with a primary key hit this).
     return (
-        "{{ audit_helper.compare_column_values("  # noqa: S608
-        f"a_query=\"select * from {{{{ ref('{old}') }}}}\", "
-        f"b_query=\"select * from {{{{ ref('{new}') }}}}\", "
-        f"primary_key={pk}, column_to_compare='{column}') }}"
+        "{% set a_query %}"  # noqa: S608
+        f"select {select_cols} from {{{{ {a_expr} }}}}"
+        "{% endset %}"
+        "{% set b_query %}"
+        f"select {select_cols} from {{{{ {b_expr} }}}}"
+        "{% endset %}"
+        "{{ audit_helper.compare_column_values("
+        f"a_query=a_query, b_query=b_query, primary_key={pk}, column_to_compare='{column}') }}}}"
     )
 
 
@@ -369,9 +476,53 @@ def _result_to_dict(result: DiffResult) -> dict[str, Any]:
         "sample_mismatches": result.sample_mismatches,
         "verdict": result.verdict.value,
         "notes": result.notes,
+        "old_label": result.old_label,
+        "new_label": result.new_label,
     }
 
 
+def _format_sample_mismatches(
+    rows: list[dict[str, Any]], primary_key: list[str], old_label: str, new_label: str
+) -> list[str]:
+    """Turn raw ``compare_relations(summarize=false)`` rows into readable per-key diffs.
+
+    audit_helper tags each mismatching row with the full row content from
+    whichever side(s) it appears on (``in_a``/``in_b``) -- printed as raw JSON,
+    the 1-3 fields that actually differ are buried under the ~40 that don't.
+    This groups by primary key and reports only the differing fields, or
+    "only in <side>" when a row exists on just one side.
+    """
+    groups: dict[tuple[Any, ...], list[dict[str, Any]]] = {}
+    for row in rows:
+        key = tuple(row.get(k) for k in primary_key)
+        groups.setdefault(key, []).append(row)
+
+    pk_desc = ", ".join(primary_key)
+    lines: list[str] = []
+    for key, group_rows in groups.items():
+        key_str = f"{pk_desc}={key[0]}" if len(key) == 1 else f"({pk_desc})={key}"
+        a_row = next((r for r in group_rows if r.get("in_a") and not r.get("in_b")), None)
+        b_row = next((r for r in group_rows if r.get("in_b") and not r.get("in_a")), None)
+        # A non-unique primary key can put more than 2 rows in one group -- flag
+        # it rather than silently comparing only the first pair and dropping the
+        # rest, which would understate how many rows actually differ.
+        extra = f" (+{len(group_rows) - 2} more rows in this key group, not shown)" if len(group_rows) > 2 else ""
+        if a_row and not b_row:
+            lines.append(f"{key_str}: only in {old_label}{extra}")
+        elif b_row and not a_row:
+            lines.append(f"{key_str}: only in {new_label}{extra}")
+        elif a_row and b_row:
+            diffs = [
+                f"{field}: {a_row.get(field)!r} → {b_row.get(field)!r}"
+                for field in a_row
+                if field not in ("in_a", "in_b") and a_row.get(field) != b_row.get(field)
+            ]
+            lines.append(
+                f"{key_str}: " + ("; ".join(diffs) if diffs else "(no field differences in this sample)") + extra
+            )
+    return lines
+
+
 def _print_json(result: DiffResult) -> None:
     print(json.dumps(_result_to_dict(result), indent=2))
 
@@ -384,22 +535,23 @@ def _print_text(result: DiffResult) -> None:
     }[result.verdict]
     console.print(
         f"\n[{color}]{result.verdict.value.upper()}[/]  "
-        f"[bold]{result.old}[/] → [bold]{result.new}[/]  ([dim]{result.target}[/])"
+        f"[bold]{result.old_label}[/] → [bold]{result.new_label}[/]  ([dim]{result.target}[/])"
     )
 
     recon = result.column_reconciliation
     if recon.diverged:
         if recon.only_in_old:
-            console.print(f"   [yellow]Only in {result.old}:[/] {', '.join(recon.only_in_old)}")
+            console.print(f"   [yellow]Only in {result.old_label}:[/] {', '.join(recon.only_in_old)}")
         if recon.only_in_new:
-            console.print(f"   [yellow]Only in {result.new}:[/] {', '.join(recon.only_in_new)}")
+            console.print(f"   [yellow]Only in {result.new_label}:[/] {', '.join(recon.only_in_new)}")
     console.print(f"   [bold]Compared columns:[/] {len(recon.compared)}")
 
     rc = result.row_counts
     delta = rc.get("delta", 0)
     delta_str = f"[green]{delta}[/]" if delta == 0 else f"[red]{delta:+d}[/]"
     console.print(
-        f"   [bold]Row counts:[/] {result.old}={rc.get('old', 0)}  {result.new}={rc.get('new', 0)}  Δ={delta_str}"
+        f"   [bold]Row counts:[/] {result.old_label}={rc.get('old', 0)}  "
+        f"{result.new_label}={rc.get('new', 0)}  Δ={delta_str}"
     )
 
     if result.column_mismatches:
@@ -408,9 +560,20 @@ def _print_text(result: DiffResult) -> None:
             console.print(f"     • {c.column}: {c.mismatch_rate:.2%} ({c.mismatched_rows} rows)")
 
     if result.sample_mismatches:
-        console.print(f"   [bold]Sample mismatched rows[/] (first {len(result.sample_mismatches)}):")
-        for row in result.sample_mismatches:
-            console.print(f"     {json.dumps(row, default=str)}")
+        if result.primary_key:
+            formatted = _format_sample_mismatches(
+                result.sample_mismatches, result.primary_key, result.old_label, result.new_label
+            )
+            console.print(f"   [bold]Sample mismatches[/] (first {len(formatted)} keys):")
+            for line in formatted:
+                console.print(f"     • {line}")
+        else:
+            # No --primary-key means rows can't be reliably paired across sides,
+            # so fall back to the raw full-row dump rather than risk grouping
+            # unrelated rows together.
+            console.print(f"   [bold]Sample mismatched rows[/] (first {len(result.sample_mismatches)}):")
+            for row in result.sample_mismatches:
+                console.print(f"     {json.dumps(row, default=str)}")
 
     for note in result.notes:
         console.print(f"   [dim]{note}[/]")
@@ -432,9 +595,13 @@ def diff(
         Parameter(name=["--old"], help="Baseline model name (the 'before' relation)."),
     ],
     new: Annotated[
-        str,
-        Parameter(name=["--new"], help="Candidate model name (the 'after' relation)."),
-    ],
+        str | None,
+        Parameter(
+            name=["--new"],
+            help="Candidate model name (the 'after' relation). Defaults to --old's value if omitted — "
+            "the common case is comparing the same table name via --old-raw against its ref()-resolved build.",
+        ),
+    ] = None,
     dbt_dir_path: Annotated[
         str | None,
         Parameter(
@@ -483,11 +650,54 @@ def diff(
         bool,
         Parameter(
             name=["--auto-build"],
-            help="Run 'dbt build' on both models (on --target) before comparing, so both relations exist.",
+            help="Run 'dbt build' on both models (on --target) before comparing, so both relations exist. "
+            "Skipped for any side marked --old-raw/--new-raw, since raw relations aren't dbt models.",
+        ),
+    ] = False,
+    old_raw: Annotated[
+        bool,
+        Parameter(
+            name=["--old-raw"],
+            help="Treat --old as a literal existing relation, not a dbt model — bypasses ref()/manifest "
+            "lookup. Use for an already-materialized table dbt doesn't know about (e.g. a snapshot made "
+            "with `ol-dbt local snapshot`). Defaults to the current --target's own database/schema; "
+            "override with --old-database/--old-schema.",
         ),
     ] = False,
+    new_raw: Annotated[
+        bool,
+        Parameter(name=["--new-raw"], help="Same as --old-raw, but for --new."),
+    ] = False,
+    old_schema: Annotated[
+        str | None,
+        Parameter(
+            name=["--old-schema"],
+            help="Schema override for --old. Requires --old-raw (default: the current --target's schema).",
+        ),
+    ] = None,
+    new_schema: Annotated[
+        str | None,
+        Parameter(
+            name=["--new-schema"],
+            help="Schema override for --new. Requires --new-raw (default: the current --target's schema).",
+        ),
+    ] = None,
+    old_database: Annotated[
+        str | None,
+        Parameter(
+            name=["--old-database"],
+            help="Database/catalog override for --old. Requires --old-raw (default: the current --target's database).",
+        ),
+    ] = None,
+    new_database: Annotated[
+        str | None,
+        Parameter(
+            name=["--new-database"],
+            help="Database/catalog override for --new. Requires --new-raw (default: the current --target's database).",
+        ),
+    ] = None,
 ) -> None:
-    """Diff two dbt model relations (same-engine row/column comparison).
+    r"""Diff two dbt model relations (same-engine row/column comparison).
 
     Reconciles column sets first (so a schema mismatch is reported clearly rather
     than as a raw SQL error), then runs a dbt_audit_helper comparison on the
@@ -504,20 +714,64 @@ def diff(
         Build both sides first, then compare:
             ol-dbt diff --old m_old --new m_new --auto-build
 
+        Compare two already-materialized copies of the same model across schemas
+        (e.g. your personal dev schema vs. real production) on one Trino target:
+            ol-dbt diff --target dev_production \\
+                --old enrollment_detail_report --old-raw --old-schema ol_warehouse_production_reporting \\
+                --new enrollment_detail_report --new-raw --new-schema ol_warehouse_production_rlougee_reporting \\
+                --primary-key courserunenrollment_id
+
+        Compare a frozen baseline (`ol-dbt local snapshot`) against a rebuild after a
+        code change, to isolate the change from any upstream data drift:
+            ol-dbt local snapshot enrollment_detail_report --as enrollment_detail_report_baseline
+            # ... edit the model, then rebuild it ...
+            ol-dbt diff --old enrollment_detail_report_baseline --old-raw \\
+                --new enrollment_detail_report --primary-key user_pk
+
     """
+    new = new or old
     primary_key_list = list(primary_key)
     exclude_list = list(exclude_columns)
     notes: list[str] = []
 
+    if (old_schema or old_database) and not old_raw:
+        err_console.print("[red]Error:[/] --old-schema/--old-database require --old-raw.")
+        sys.exit(1)
+    if (new_schema or new_database) and not new_raw:
+        err_console.print("[red]Error:[/] --new-schema/--new-database require --new-raw.")
+        sys.exit(1)
+
     # Validate every value interpolated into inline Jinja SQL before use.
     try:
         _validate_identifiers("model name", [old, new])
         _validate_identifiers("primary key", primary_key_list)
         _validate_identifiers("excluded column", exclude_list)
+        _validate_identifiers(
+            "schema/database override",
+            [v for v in (old_schema, new_schema, old_database, new_database) if v],
+        )
     except InvalidIdentifierError as exc:
         err_console.print(f"[red]Error:[/] {exc}")
         sys.exit(1)
 
+    if old_raw:
+        notes.append(
+            f"--old-raw: '{old}' resolved as a literal relation "
+            f"(database={old_database or 'target.database'}, schema={old_schema or 'target.schema'})."
+        )
+    if new_raw:
+        notes.append(
+            f"--new-raw: '{new}' resolved as a literal relation "
+            f"(database={new_database or 'target.database'}, schema={new_schema or 'target.schema'})."
+        )
+
+    # Text-output labels disambiguating which side is which when --old/new-raw
+    # make the bare `old`/`new` names identical or otherwise ambiguous (e.g.
+    # both "enrollment_detail_report"). Unannotated for a plain ref()-vs-ref()
+    # diff, where `old`/`new` are already unambiguous.
+    old_label = f"{old} (raw)" if old_raw else old
+    new_label = f"{new} (raw)" if new_raw else new
+
     # Resolve dbt project directory (shared preamble with impact/validate).
     if dbt_dir_path:
         dbt_dir = Path(dbt_dir_path).resolve()
@@ -532,6 +786,14 @@ def diff(
         err_console.print("  Use --dbt-dir to specify the path.")
         sys.exit(1)
 
+    # Fail fast with an actionable message rather than letting dbt's raw
+    # "N package(s) specified... only M installed" Compilation Error (surfaced
+    # through 500 chars of dbt log noise) reach the user.
+    if not (dbt_dir / "dbt_packages" / "audit_helper").exists():
+        err_console.print(f"[red]Error:[/] the 'audit_helper' dbt package is not installed in {dbt_dir}.")
+        err_console.print(f"  Run: cd {dbt_dir} && dbt deps")
+        sys.exit(1)
+
     models_dir = dbt_dir / "models"
 
     # Build the metadata sources used for column reconciliation.
@@ -553,13 +815,30 @@ def diff(
             continue
 
     # --- Column reconciliation (before any comparison) ---
-    old_cols = _resolve_columns(old, sql_models_by_name, manifest, yaml_registry)
-    new_cols = _resolve_columns(new, sql_models_by_name, manifest, yaml_registry)
+    old_raw_error: str | None = None
+    new_raw_error: str | None = None
+    if old_raw:
+        old_cols, old_raw_error = _resolve_raw_columns(old, dbt_dir, target, database=old_database, schema=old_schema)
+    else:
+        old_cols = _resolve_columns(old, sql_models_by_name, manifest, yaml_registry)
+    if new_raw:
+        new_cols, new_raw_error = _resolve_raw_columns(new, dbt_dir, target, database=new_database, schema=new_schema)
+    else:
+        new_cols = _resolve_columns(new, sql_models_by_name, manifest, yaml_registry)
+
     unresolved_note = "Column set for '{}' could not be resolved (unparsed / not in manifest); reconciliation skipped."
     if not old_cols:
-        notes.append(unresolved_note.format(old))
+        notes.append(
+            f"Column set for '{old}' could not be resolved: {old_raw_error}"
+            if old_raw_error
+            else unresolved_note.format(old)
+        )
     if not new_cols:
-        notes.append(unresolved_note.format(new))
+        notes.append(
+            f"Column set for '{new}' could not be resolved: {new_raw_error}"
+            if new_raw_error
+            else unresolved_note.format(new)
+        )
     recon = reconcile_columns(old_cols, new_cols, set(exclude_list))
 
     # When a column set can't be resolved statically, the schema-divergence gate
@@ -580,22 +859,28 @@ def diff(
             "(or point --dbt-dir at a compiled project) for full schema validation."
         )
 
-    # Optionally build both relations first.
+    # Optionally build both relations first. Raw relations aren't dbt models —
+    # they're assumed to already exist (an already-materialized table) — so
+    # they're excluded from --select rather than passed to dbt build.
     if auto_build:
-        # Single space-joined --select value, consistent with ol-dbt run/impact.
-        build_cmd = ["dbt", "build", "--target", target, "--select", f"{old} {new}"]
-        if output_format == "text":
-            console.print(f"[dim]Running: {' '.join(build_cmd)} ...[/]")
-        try:
-            subprocess.run(build_cmd, cwd=str(dbt_dir), capture_output=True, text=True, check=True)  # noqa: S603, S607
-        except subprocess.CalledProcessError as exc:
-            # dbt often logs useful detail to stdout, not stderr — prefer either.
-            detail = (exc.stderr or exc.stdout or str(exc)).strip()
-            err_console.print(f"[red]Error:[/] dbt build failed: {detail[-500:]}")
-            sys.exit(1)
-        except FileNotFoundError:
-            err_console.print("[red]Error:[/] 'dbt' command not found; install dbt and ensure it is on PATH.")
-            sys.exit(1)
+        build_targets = [name for name, raw in ((old, old_raw), (new, new_raw)) if not raw]
+        if not build_targets:
+            notes.append("--auto-build skipped: both --old and --new are raw relations (nothing to build).")
+        else:
+            # Single space-joined --select value, consistent with ol-dbt run/impact.
+            build_cmd = ["dbt", "build", "--target", target, "--select", " ".join(build_targets)]
+            if output_format == "text":
+                console.print(f"[dim]Running: {' '.join(build_cmd)} ...[/]")
+            try:
+                subprocess.run(build_cmd, cwd=str(dbt_dir), capture_output=True, text=True, check=True)  # noqa: S603, S607
+            except subprocess.CalledProcessError as exc:
+                # dbt often logs useful detail to stdout, not stderr — prefer either.
+                detail = (exc.stderr or exc.stdout or str(exc)).strip()
+                err_console.print(f"[red]Error:[/] dbt build failed: {detail[-500:]}")
+                sys.exit(1)
+            except FileNotFoundError:
+                err_console.print("[red]Error:[/] 'dbt' command not found; install dbt and ensure it is on PATH.")
+                sys.exit(1)
 
     # --- Comparison ---
     row_counts: dict[str, int] = {"old": 0, "new": 0, "delta": 0}
@@ -619,6 +904,8 @@ def diff(
                 column_mismatches=column_mismatches,
                 sample_mismatches=sample_mismatches,
                 verdict=Verdict.SCHEMA_DIVERGENCE,
+                old_label=old_label,
+                new_label=new_label,
                 notes=notes,
             ),
             output_format,
@@ -626,7 +913,19 @@ def diff(
 
     try:
         summary_rows = _run_dbt_show(
-            _compare_relations_sql(old, new, primary_key_list, exclude_list, summarize=True),
+            _compare_relations_sql(
+                old,
+                new,
+                primary_key_list,
+                exclude_list,
+                summarize=True,
+                old_raw=old_raw,
+                new_raw=new_raw,
+                old_schema=old_schema,
+                new_schema=new_schema,
+                old_database=old_database,
+                new_database=new_database,
+            ),
             dbt_dir,
             target,
             limit=1000,
@@ -635,7 +934,19 @@ def diff(
 
         if mismatched_rows:
             sample_mismatches = _run_dbt_show(
-                _compare_relations_sql(old, new, primary_key_list, exclude_list, summarize=False),
+                _compare_relations_sql(
+                    old,
+                    new,
+                    primary_key_list,
+                    exclude_list,
+                    summarize=False,
+                    old_raw=old_raw,
+                    new_raw=new_raw,
+                    old_schema=old_schema,
+                    new_schema=new_schema,
+                    old_database=old_database,
+                    new_database=new_database,
+                ),
                 dbt_dir,
                 target,
                 limit=limit,
@@ -648,7 +959,20 @@ def diff(
                 notes.append(f"Per-column comparison capped at {_MAX_PER_COLUMN_COMPARISONS} of {len(cols)} columns.")
                 cols = cols[:_MAX_PER_COLUMN_COMPARISONS]
             for col in cols:
-                mismatch = _compare_single_column(old, new, primary_key_list, col, dbt_dir, target)
+                mismatch = _compare_single_column(
+                    old,
+                    new,
+                    primary_key_list,
+                    col,
+                    dbt_dir,
+                    target,
+                    old_raw=old_raw,
+                    new_raw=new_raw,
+                    old_schema=old_schema,
+                    new_schema=new_schema,
+                    old_database=old_database,
+                    new_database=new_database,
+                )
                 if mismatch is not None and mismatch.mismatched_rows > 0:
                     column_mismatches.append(mismatch)
         elif not primary_key_list:
@@ -675,6 +999,8 @@ def diff(
             column_mismatches=column_mismatches,
             sample_mismatches=sample_mismatches,
             verdict=verdict,
+            old_label=old_label,
+            new_label=new_label,
             notes=notes,
         ),
         output_format,
@@ -698,6 +1024,13 @@ def _compare_single_column(
     column: str,
     dbt_dir: Path,
     target: str,
+    *,
+    old_raw: bool = False,
+    new_raw: bool = False,
+    old_schema: str | None = None,
+    new_schema: str | None = None,
+    old_database: str | None = None,
+    new_database: str | None = None,
 ) -> ColumnMismatch | None:
     """Run a per-column value comparison, returning its mismatch rate.
 
@@ -705,7 +1038,18 @@ def _compare_single_column(
     the mismatch rate is the share of rows not in a "✅" perfect-match bucket.
     """
     rows = _run_dbt_show(
-        _compare_column_sql(old, new, primary_key, column),
+        _compare_column_sql(
+            old,
+            new,
+            primary_key,
+            column,
+            old_raw=old_raw,
+            new_raw=new_raw,
+            old_schema=old_schema,
+            new_schema=new_schema,
+            old_database=old_database,
+            new_database=new_database,
+        ),
         dbt_dir,
         target,
         limit=1000,
diff --git a/src/ol_dbt_cli/ol_dbt_cli/commands/local_dev.py b/src/ol_dbt_cli/ol_dbt_cli/commands/local_dev.py
index 7d3b2526c..ea3e6da53 100644
--- a/src/ol_dbt_cli/ol_dbt_cli/commands/local_dev.py
+++ b/src/ol_dbt_cli/ol_dbt_cli/commands/local_dev.py
@@ -9,6 +9,7 @@
 from __future__ import annotations
 
 import os
+import subprocess
 import sys
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from pathlib import Path
@@ -675,6 +676,84 @@ def list_sources(
     _show_registry(duckdb_path)
 
 
+@local_app.command
+def snapshot(
+    model: Annotated[str, cyclopts.Parameter(help="Model name to snapshot (its current build, via ref()).")],
+    as_name: Annotated[
+        str,
+        cyclopts.Parameter(name=["--as"], help="Name for the snapshot table, created in the target's own schema."),
+    ],
+    target: Annotated[
+        str,
+        cyclopts.Parameter(name=["--target", "-t"], help="dbt target to read/write on (default: dev_local)."),
+    ] = "dev_local",
+    dbt_dir_path: Annotated[
+        str | None,
+        cyclopts.Parameter(
+            name=["--dbt-dir", "-d"],
+            help="Path to dbt project root (contains dbt_project.yml). Defaults to src/ol_dbt relative to repo root.",
+        ),
+    ] = None,
+) -> None:
+    """Copy a model's current build into a plain table under a new name.
+
+    For before/after comparisons of the SAME model (a code change, not a
+    migration to a differently-named model): build it, snapshot the build with
+    this command, make your change and rebuild, then compare with
+    `ol-dbt diff --old <as-name> --old-raw --new <model> --primary-key ...`.
+    Neither this command nor that diff touches Glue/production, so any
+    difference it finds reflects the code change alone, not upstream data
+    drift from asynchronous production rebuilds.
+    """
+    from ol_dbt_cli.commands.diff import InvalidIdentifierError, _validate_identifiers  # noqa: PLC0415
+    from ol_dbt_cli.commands.run import _find_dbt_dir  # noqa: PLC0415
+
+    try:
+        _validate_identifiers("model/snapshot name", [model, as_name])
+    except InvalidIdentifierError as exc:
+        print(f"Error: {exc}")
+        sys.exit(1)
+
+    try:
+        dbt_dir = _find_dbt_dir(dbt_dir_path)
+    except RuntimeError as exc:
+        print(f"Error: {exc}")
+        sys.exit(1)
+
+    # _find_dbt_dir returns an explicit --dbt-dir as-is, without checking it's a
+    # real dbt project -- a typo'd path would otherwise only surface later as a
+    # confusing dbt-level error instead of a clear, early failure here.
+    if not (dbt_dir / "dbt_project.yml").exists():
+        print(
+            f"Error: dbt project not found at {dbt_dir}. "
+            "Pass --dbt-dir pointing at a directory containing dbt_project.yml."
+        )
+        sys.exit(1)
+
+    # Rendered into a Jinja template compiled by dbt (not executed as a raw query
+    # here), so the S608 heuristic is a false positive -- model/as_name were
+    # already validated as plain identifiers above.
+    # `}}` in an f-string is an escape for a single literal `}` (same as `{{` for
+    # `{`) -- each Jinja `{{ ... }}` block needs `{{{{`/`}}}}` from an f-string.
+    dst_expr = f"api.Relation.create(database=target.database, schema=target.schema, identifier='{as_name}')"
+    src_expr = f"ref('{model}')"
+    ctas_sql = f"create or replace table {{{{ {dst_expr} }}}} as select * from {{{{ {src_expr} }}}}"  # noqa: S608
+    cmd = ["dbt", "show", "--inline", ctas_sql, "--target", target, "--output", "json", "--limit", "-1"]
+    print(f"Running: {' '.join(cmd)}")
+    try:
+        subprocess.run(cmd, cwd=str(dbt_dir), capture_output=True, text=True, check=True)  # noqa: S603
+    except subprocess.CalledProcessError as exc:
+        detail = (exc.stderr or exc.stdout or str(exc)).strip()
+        print(f"Error: dbt show failed: {detail[-500:]}")
+        sys.exit(1)
+    except FileNotFoundError:
+        print("Error: 'dbt' command not found; install dbt and ensure it is on PATH.")
+        sys.exit(1)
+
+    print(f"✅ Snapshotted '{model}' as '{as_name}' (target={target}).")
+    print(f"   Compare with: ol-dbt diff --target {target} --old {as_name} --old-raw --new {model} --primary-key <key>")
+
+
 @local_app.command
 def cleanup_local(
     duckdb_path: Annotated[
diff --git a/src/ol_dbt_cli/tests/test_diff.py b/src/ol_dbt_cli/tests/test_diff.py
index eeff863a7..c58b7d3e2 100644
--- a/src/ol_dbt_cli/tests/test_diff.py
+++ b/src/ol_dbt_cli/tests/test_diff.py
@@ -10,9 +10,13 @@
 from ol_dbt_cli.commands import diff as diff_mod
 from ol_dbt_cli.commands.diff import (
     Verdict,
+    _compare_column_sql,
     _compare_relations_sql,
     _extract_show_rows,
+    _format_sample_mismatches,
     _jinja_list,
+    _relation_jinja,
+    _resolve_raw_columns,
     _summarize_relations,
     diff,
     reconcile_columns,
@@ -136,6 +140,34 @@ def test_rows_on_both_sides(self) -> None:
         assert mismatched == 12
 
 
+class TestFormatSampleMismatches:
+    def test_value_diff_shows_only_differing_fields(self) -> None:
+        # a_row/b_row share id=1 but differ only in "name" -- "other" must NOT
+        # appear in the output even though it's present in both raw rows.
+        rows = [
+            {"id": 1, "name": "old", "other": "same", "in_a": True, "in_b": False},
+            {"id": 1, "name": "new", "other": "same", "in_a": False, "in_b": True},
+        ]
+        lines = _format_sample_mismatches(rows, ["id"], "old_side", "new_side")
+        assert lines == ["id=1: name: 'old' → 'new'"]
+
+    def test_row_present_only_on_one_side(self) -> None:
+        rows = [{"id": 2, "name": "x", "in_a": True, "in_b": False}]
+        lines = _format_sample_mismatches(rows, ["id"], "old_side", "new_side")
+        assert lines == ["id=2: only in old_side"]
+
+    def test_flags_when_key_group_has_more_than_two_rows(self) -> None:
+        # A non-unique primary key can put >2 rows in one group -- must flag
+        # the extras rather than silently comparing only the first pair.
+        rows = [
+            {"id": 3, "name": "a", "in_a": True, "in_b": False},
+            {"id": 3, "name": "b", "in_a": True, "in_b": False},
+            {"id": 3, "name": "c", "in_a": True, "in_b": False},
+        ]
+        lines = _format_sample_mismatches(rows, ["id"], "old_side", "new_side")
+        assert lines == ["id=3: only in old_side (+1 more rows in this key group, not shown)"]
+
+
 class TestSqlBuilders:
     def test_jinja_list(self) -> None:
         assert _jinja_list(["a", "b"]) == "['a', 'b']"
@@ -158,13 +190,102 @@ def test_compare_relations_no_pk(self) -> None:
         assert "primary_key" not in sql
 
 
-def _make_project(tmp_path: Path, old_sql: str, new_sql: str) -> Path:
-    """Create a minimal dbt project dir with two leaf models."""
+class TestRelationJinja:
+    def test_non_raw_uses_ref(self) -> None:
+        assert _relation_jinja("dim_user") == "ref('dim_user')"
+
+    def test_raw_defaults_to_target_database_and_schema(self) -> None:
+        expr = _relation_jinja("glue__ol_warehouse_production_reporting__enrollment_detail_report", raw=True)
+        assert expr == (
+            "api.Relation.create(database=target.database, schema=target.schema, "
+            "identifier='glue__ol_warehouse_production_reporting__enrollment_detail_report')"
+        )
+
+    def test_raw_with_explicit_database_and_schema(self) -> None:
+        expr = _relation_jinja(
+            "enrollment_detail_report",
+            raw=True,
+            database="ol_data_lake_production",
+            schema="ol_warehouse_production_reporting",
+        )
+        assert expr == (
+            "api.Relation.create(database='ol_data_lake_production', "
+            "schema='ol_warehouse_production_reporting', identifier='enrollment_detail_report')"
+        )
+
+    def test_compare_relations_sql_with_old_raw(self) -> None:
+        sql = _compare_relations_sql(
+            "glue__x__y",
+            "new_m",
+            ["id"],
+            [],
+            summarize=True,
+            old_raw=True,
+            old_schema="main",
+        )
+        assert "ref('glue__x__y')" not in sql
+        assert "api.Relation.create(database=target.database, schema='main', identifier='glue__x__y')" in sql
+        assert "ref('new_m')" in sql
+
+    def test_compare_column_sql_with_both_raw(self) -> None:
+        sql = _compare_column_sql(
+            "old_tbl",
+            "new_tbl",
+            ["id"],
+            "some_col",
+            old_raw=True,
+            old_database="db_a",
+            old_schema="schema_a",
+            new_raw=True,
+            new_database="db_b",
+            new_schema="schema_b",
+        )
+        assert "api.Relation.create(database='db_a', schema='schema_a', identifier='old_tbl')" in sql
+        assert "api.Relation.create(database='db_b', schema='schema_b', identifier='new_tbl')" in sql
+        assert "ref(" not in sql
+
+    def test_compare_column_sql_narrows_to_primary_key_and_column(self) -> None:
+        # Narrows to primary key + compared column instead of `select *`, since
+        # compare_column_values' full outer join doesn't need the rest of the row.
+        sql = _compare_column_sql("old_m", "new_m", ["id"], "some_col")
+        assert "select id, some_col from" in sql
+        assert "select *" not in sql
+
+
+class TestResolveRawColumns:
+    def test_returns_lowercased_keys_from_sampled_row(self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+        monkeypatch.setattr(diff_mod, "_run_dbt_show", lambda *a, **k: [{"Id": 1, "NAME": "x"}])
+        cols, error = _resolve_raw_columns("some_view", tmp_path, "dev_local")
+        assert cols == {"id", "name"}
+        assert error is None
+
+    def test_empty_result_returns_empty_set_no_error(self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+        monkeypatch.setattr(diff_mod, "_run_dbt_show", lambda *a, **k: [])
+        cols, error = _resolve_raw_columns("some_view", tmp_path, "dev_local")
+        assert cols == set()
+        assert error is None
+
+    def test_dbt_failure_returns_empty_set_and_surfaces_error(
+        self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+    ) -> None:
+        def boom(*a: Any, **k: Any) -> list[dict[str, Any]]:
+            msg = "dbt show failed: relation glue__x__y not found"
+            raise RuntimeError(msg)
+
+        monkeypatch.setattr(diff_mod, "_run_dbt_show", boom)
+        cols, error = _resolve_raw_columns("missing_view", tmp_path, "dev_local")
+        assert cols == set()
+        assert error == "dbt show failed: relation glue__x__y not found"
+
+
+def _make_project(tmp_path: Path, old_sql: str, new_sql: str, *, new_name: str = "m_new") -> Path:
+    """Create a minimal dbt project dir with two leaf models and audit_helper "installed"."""
     dbt_dir = tmp_path / "ol_dbt"
     (dbt_dir / "models").mkdir(parents=True)
+    (dbt_dir / "dbt_packages" / "audit_helper").mkdir(parents=True)
     (dbt_dir / "dbt_project.yml").write_text("name: test\nprofile: test\n")
     (dbt_dir / "models" / "m_old.sql").write_text(old_sql)
-    (dbt_dir / "models" / "m_new.sql").write_text(new_sql)
+    (dbt_dir / "models" / f"{new_name}.sql").write_text(new_sql)
     return dbt_dir
 
 
@@ -296,6 +417,8 @@ def test_json_output_schema(
             "row_counts",
             "column_mismatches",
             "sample_mismatches",
+            "old_label",
+            "new_label",
         ):
             assert key in payload
         assert set(payload["column_reconciliation"]) == {"only_in_old", "only_in_new", "compared"}
@@ -346,3 +469,131 @@ def raise_runtime(*a: Any, **k: Any) -> list[dict[str, Any]]:
         with pytest.raises(SystemExit) as exc:
             diff(old="m_old", new="m_new", dbt_dir_path=str(dbt_dir))
         assert exc.value.code == 1
+
+
+class TestDiffCommandRaw:
+    def test_old_schema_without_old_raw_exits_1(self, tmp_path: Path) -> None:
+        dbt_dir = _make_project(tmp_path, "select 1 as id", "select 1 as id")
+        with pytest.raises(SystemExit) as exc:
+            diff(old="m_old", new="m_new", old_schema="main", dbt_dir_path=str(dbt_dir))
+        assert exc.value.code == 1
+
+    def test_new_database_without_new_raw_exits_1(self, tmp_path: Path) -> None:
+        dbt_dir = _make_project(tmp_path, "select 1 as id", "select 1 as id")
+        with pytest.raises(SystemExit) as exc:
+            diff(old="m_old", new="m_new", new_database="db_b", dbt_dir_path=str(dbt_dir))
+        assert exc.value.code == 1
+
+    def test_injected_schema_override_rejected(self, tmp_path: Path) -> None:
+        dbt_dir = _make_project(tmp_path, "select 1 as id", "select 1 as id")
+        with pytest.raises(SystemExit) as exc:
+            diff(old="m_old", new="m_new", old_raw=True, old_schema="a; drop table x", dbt_dir_path=str(dbt_dir))
+        assert exc.value.code == 1
+
+    def test_old_raw_match_exits_0_and_notes_resolution(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
+    ) -> None:
+        # "old" is a raw relation (e.g. a Glue view) that isn't a dbt model at all —
+        # it must not go through static column resolution or ref().
+        dbt_dir = _make_project(tmp_path, "select 1 as id, 'x' as name", "select 1 as id, 'x' as name")
+        seen_inline: list[str] = []
+
+        def fake_show(inline_sql: str, *a: Any, **k: Any) -> list[dict[str, Any]]:
+            seen_inline.append(inline_sql)
+            if "audit_helper" not in inline_sql:
+                return [{"id": 1, "name": "x"}]  # raw column sample
+            return [{"in_a": True, "in_b": True, "count": 10}]
+
+        monkeypatch.setattr(diff_mod, "_run_dbt_show", fake_show)
+        diff(
+            old="glue__ol_warehouse_production_reporting__enrollment_detail_report",
+            new="m_new",
+            old_raw=True,
+            old_schema="main",
+            dbt_dir_path=str(dbt_dir),
+        )
+        err = capsys.readouterr().err
+        assert "schema-divergence" not in err
+        # ref() must never be called for the raw side.
+        assert not any("ref('glue__" in sql for sql in seen_inline)
+        assert any("api.Relation.create(database=target.database, schema='main'" in sql for sql in seen_inline)
+
+    def test_auto_build_skips_raw_sides(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+        dbt_dir = _make_project(tmp_path, "select 1 as id, 'x' as name", "select 1 as id, 'x' as name")
+
+        def fake_show(inline_sql: str, *a: Any, **k: Any) -> list[dict[str, Any]]:
+            if "compare_relations" in inline_sql:
+                return [{"in_a": True, "in_b": True, "count": 1}]
+            return [{"id": 1, "name": "x"}]  # raw column sample, matching m_new's parsed columns
+
+        monkeypatch.setattr(diff_mod, "_run_dbt_show", fake_show)
+        build_calls: list[list[str]] = []
+
+        def fake_run(cmd: list[str], **kwargs: Any) -> Any:
+            build_calls.append(cmd)
+            import subprocess
+
+            return subprocess.CompletedProcess(args=cmd, returncode=0, stdout="", stderr="")
+
+        monkeypatch.setattr(diff_mod.subprocess, "run", fake_run)
+        diff(
+            old="glue__x__y",
+            new="m_new",
+            old_raw=True,
+            auto_build=True,
+            dbt_dir_path=str(dbt_dir),
+        )
+        assert len(build_calls) == 1
+        select_idx = build_calls[0].index("--select") + 1
+        assert build_calls[0][select_idx] == "m_new"
+
+    def test_auto_build_skipped_entirely_when_both_raw(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+        dbt_dir = _make_project(tmp_path, "select 1 as id, 'x' as name", "select 1 as id, 'x' as name")
+        monkeypatch.setattr(
+            diff_mod,
+            "_run_dbt_show",
+            lambda *a, **k: [{"in_a": True, "in_b": True, "count": 1}],
+        )
+
+        def boom(*a: Any, **k: Any) -> Any:
+            msg = "dbt build must not run when both sides are raw"
+            raise AssertionError(msg)
+
+        monkeypatch.setattr(diff_mod.subprocess, "run", boom)
+        diff(
+            old="glue__x__y",
+            new="glue__a__b",
+            old_raw=True,
+            new_raw=True,
+            auto_build=True,
+            dbt_dir_path=str(dbt_dir),
+        )
+
+
+class TestDiffCommandLabels:
+    def test_plain_diff_labels_are_unannotated(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
+    ) -> None:
+        # A plain ref()-vs-ref() diff (no --*-raw) is already unambiguous — the
+        # header must not grow a "(raw)" annotation.
+        dbt_dir = _make_project(tmp_path, "select 1 as id", "select 1 as id")
+        monkeypatch.setattr(diff_mod, "_run_dbt_show", lambda *a, **k: [{"in_a": True, "in_b": True, "count": 1}])
+        diff(old="m_old", new="m_new", dbt_dir_path=str(dbt_dir))
+        out = capsys.readouterr().out
+        assert "m_old → m_new" in out
+        assert "(raw)" not in out
+
+    def test_old_raw_label_annotated_with_raw(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
+    ) -> None:
+        dbt_dir = _make_project(tmp_path, "select 1 as id, 'x' as name", "select 1 as id, 'x' as name")
+
+        def fake_show(inline_sql: str, *a: Any, **k: Any) -> list[dict[str, Any]]:
+            if "audit_helper" not in inline_sql:
+                return [{"id": 1, "name": "x"}]
+            return [{"in_a": True, "in_b": True, "count": 1}]
+
+        monkeypatch.setattr(diff_mod, "_run_dbt_show", fake_show)
+        diff(old="glue__x__y", new="m_new", old_raw=True, dbt_dir_path=str(dbt_dir))
+        out = capsys.readouterr().out
+        assert "glue__x__y (raw) → m_new" in out
diff --git a/src/ol_dbt_cli/tests/test_local_dev.py b/src/ol_dbt_cli/tests/test_local_dev.py
index 83bb173a4..7696fc8b6 100644
--- a/src/ol_dbt_cli/tests/test_local_dev.py
+++ b/src/ol_dbt_cli/tests/test_local_dev.py
@@ -11,9 +11,40 @@
     PROTECTED_SCHEMAS,
     _register_single_table,
     _validate_schema_safety,
+    snapshot,
 )
 
 
+class TestSnapshot:
+    def test_runs_ctas_with_validated_identifiers(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+        import subprocess
+
+        (tmp_path / "dbt_project.yml").write_text("name: test\nprofile: test\n")
+        calls: list[list[str]] = []
+
+        def fake_run(cmd: list[str], **kwargs: object) -> subprocess.CompletedProcess[str]:
+            calls.append(cmd)
+            return subprocess.CompletedProcess(args=cmd, returncode=0, stdout="", stderr="")
+
+        monkeypatch.setattr(subprocess, "run", fake_run)
+        snapshot(
+            "enrollment_detail_report",
+            as_name="enrollment_detail_report_baseline",
+            dbt_dir_path=str(tmp_path),
+        )
+        assert len(calls) == 1
+        inline_idx = calls[0].index("--inline") + 1
+        inline_sql = calls[0][inline_idx]
+        assert "identifier='enrollment_detail_report_baseline'" in inline_sql
+        assert "ref('enrollment_detail_report')" in inline_sql
+        assert "--limit" in calls[0]
+        assert calls[0][calls[0].index("--limit") + 1] == "-1"
+
+    def test_rejects_invalid_identifier(self, tmp_path: Path) -> None:
+        with pytest.raises(SystemExit):
+            snapshot("m; drop table x", as_name="baseline", dbt_dir_path=str(tmp_path))
+
+
 class TestValidateSchemaSafety:
     """Tests for _validate_schema_safety — guards against dropping production schemas."""