Add VALIDATION_OP=check_binds: bind-check a dbt test's compiled SQL

CHANGELOG.md

@@ -5,6 +5,13 @@ follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ## [Unreleased]
 
+### Added
+
+- `VALIDATION_OP=check_binds`: bind-checks a dbt test's compiled SQL
+  (`CANDIDATE_SQL_URI`) against the candidate schema with the engine's
+  EXPLAIN and creates nothing. Continuo's release-controller emits it for
+  every dbt test of a changed model.
+
 ## [0.3.1] - 2026-08-21
 
 ### Added

continuo_python_runtime/validation/runner.py

@@ -18,6 +18,11 @@
   that dropped a column the script reads fails the release gate, then the output
   table is materialized empty from the declared typed columns and the declared
   physical layout.
+- ``check_binds``: for a dbt test's compiled SQL (``CANDIDATE_SQL_URI``, fetched
+  the same way as ``build_from_sql``). EXPLAINs it against the candidate schema
+  via the engine adapter and creates nothing — a test whose compiled SQL names
+  a column a candidate change dropped or renamed fails the release at its
+  source, before any table is built.
 
 The engine adapter is discovered from the single installed
 ``continuo_engine.adapters`` entry point — each runner image installs exactly one.
@@ -108,18 +113,21 @@ def load_candidate_spec() -> dict:
     return spec
 
 
-_NODE_OPS = ("build_from_sql", "clone_from_prod", "build_from_columns")
+_NODE_OPS = ("build_from_sql", "clone_from_prod", "build_from_columns", "check_binds")
 _SCHEMA_OPS = ("ensure_schema", "drop_schema")
 
 
 def main() -> None:
     """Run one validation op end to end; exits non-zero on failure.
 
     Node ops (``build_from_sql``/``clone_from_prod``/``build_from_columns``)
-    materialize one empty node table and require ``TABLE_NAME``. Schema ops
-    (``ensure_schema``/``drop_schema``) act on the whole candidate schema —
-    the executor schedules them as one-shot engine-image Jobs to own the
-    candidate-schema lifecycle without connecting to the warehouse itself —
+    materialize one empty node table and require ``TABLE_NAME``. ``check_binds``
+    is also a node op and requires ``TABLE_NAME`` (for identity only, in the
+    result block's ``unique_id``) but creates nothing: it bind-checks a dbt
+    test's compiled SQL against the candidate schema with the engine's EXPLAIN.
+    Schema ops (``ensure_schema``/``drop_schema``) act on the whole candidate
+    schema — the executor schedules them as one-shot engine-image Jobs to own
+    the candidate-schema lifecycle without connecting to the warehouse itself —
     and take no table.
     """
     logging.basicConfig(
@@ -139,7 +147,7 @@ def main() -> None:
     if op in _NODE_OPS:
         table = _require("TABLE_NAME")
         unique_id = _node_id() or f"model.{table}"
-        if op == "build_from_sql":
+        if op in ("build_from_sql", "check_binds"):
             try:
                 raw_sql = load_candidate_sql()
             except Exception as exc:
@@ -150,7 +158,7 @@ def main() -> None:
             if not raw_sql:
                 logger.error(
                     "CANDIDATE_SQL_URI is unset or the object is empty for a "
-                    "build_from_sql node; cannot validate"
+                    "%s node; cannot validate", op
                 )
                 print(result.result_block("error", "CANDIDATE_SQL_URI is unset or empty",
                                           unique_id=unique_id), flush=True)
@@ -238,6 +246,12 @@ def main() -> None:
             if op == "build_from_sql":
                 assert candidate_sql is not None, "candidate_sql must be set for build_from_sql"
                 adapter.build_empty_from_sql(schema, table, candidate_sql)
+            elif op == "check_binds":
+                # A dbt test: EXPLAIN its compiled SQL against the candidate
+                # schema. A dropped or renamed column fails here; no table is
+                # created and no row is read.
+                assert candidate_sql is not None, "candidate_sql must be set for check_binds"
+                adapter.check_binds(candidate_sql)
             elif op == "build_from_columns":
                 assert spec is not None, "spec must be set for build_from_columns"
                 if csv_source:

tests/test_validation_runner.py

@@ -246,6 +246,66 @@ def _raise():
     assert '"status":"error"' in out
 
 
+# --------------------------------------------------------------------------
+# main — check_binds (dbt test bind check: creates nothing)
+# --------------------------------------------------------------------------
+
+def test_main_check_binds_calls_adapter_and_emits_success(monkeypatch, capsys):
+    """Bind-check a dbt test's compiled SQL and emit success, creating nothing."""
+    _set_common_env(monkeypatch)
+    monkeypatch.setenv("VALIDATION_OP", "check_binds")
+    fake = FakeWarehouseAdapter()
+    _install_fake_adapter(monkeypatch, fake)
+    sql = 'select amount_eur from "_candidate_relA".tbind where amount_eur is null'
+    monkeypatch.setattr(runner, "load_candidate_sql", lambda: sql)
+
+    runner.main()
+
+    assert fake.schemas_ensured == ["_candidate_relA"]
+    assert fake.checked_binds == [sql]
+    assert fake.builds == []
+    assert fake.clones == []
+    assert fake.column_builds == []
+    assert fake.closed is True
+    out = capsys.readouterr().out
+    assert result.SENTINEL_BEGIN in out
+    assert '"status":"success"' in out
+    assert out.strip().endswith(result.SENTINEL_END)
+
+
+def test_main_check_binds_bind_error_exits_1(monkeypatch, capsys):
+    """A failing bind check on the test's compiled SQL emits an error block and exits 1."""
+    _set_common_env(monkeypatch)
+    monkeypatch.setenv("VALIDATION_OP", "check_binds")
+    sql = "select missing from t"
+    fake = FakeWarehouseAdapter()
+    fake.raise_on_binds = {sql: RuntimeError('column "missing" does not exist')}
+    _install_fake_adapter(monkeypatch, fake)
+    monkeypatch.setattr(runner, "load_candidate_sql", lambda: sql)
+
+    with pytest.raises(SystemExit) as exc:
+        runner.main()
+
+    assert exc.value.code == 1
+    assert fake.checked_binds == [sql]
+    assert fake.builds == [] and fake.clones == [] and fake.column_builds == []
+    out = capsys.readouterr().out
+    assert result.SENTINEL_BEGIN in out
+    assert '"status":"error"' in out
+    assert 'column \\"missing\\" does not exist' in out  # JSON-escaped in the result block
+
+
+def test_main_check_binds_empty_candidate_sql_errors(monkeypatch, capsys):
+    """Exit 2 when candidate SQL is empty, same as build_from_sql."""
+    _set_common_env(monkeypatch)
+    monkeypatch.setenv("VALIDATION_OP", "check_binds")
+    monkeypatch.setattr(runner, "load_candidate_sql", lambda: "")
+    with pytest.raises(SystemExit) as exc:
+        runner.main()
+    assert exc.value.code == 2
+    assert '"status":"error"' in capsys.readouterr().out
+
+
 def test_main_clone_from_prod_calls_adapter(monkeypatch, capsys):
     """Clone from prod and emit success block."""
     _set_common_env(monkeypatch)
@@ -921,6 +981,24 @@ def _setup_drop_schema(monkeypatch):
     _install_fake_adapter(monkeypatch, FakeWarehouseAdapter())
 
 
+def _setup_check_binds_success(monkeypatch):
+    """Arrange a check_binds run that reaches the success block."""
+    _set_common_env(monkeypatch)
+    monkeypatch.setenv("VALIDATION_OP", "check_binds")
+    _install_fake_adapter(monkeypatch, FakeWarehouseAdapter())
+    monkeypatch.setattr(runner, "load_candidate_sql", lambda: "select 1 from t")
+
+
+def _setup_check_binds_bind_error(monkeypatch):
+    """Arrange a check_binds run whose bind check raises."""
+    _set_common_env(monkeypatch)
+    monkeypatch.setenv("VALIDATION_OP", "check_binds")
+    fake = FakeWarehouseAdapter()
+    fake.raise_on_binds = {"select missing from t": RuntimeError("boom")}
+    _install_fake_adapter(monkeypatch, fake)
+    monkeypatch.setattr(runner, "load_candidate_sql", lambda: "select missing from t")
+
+
 def _setup_build_from_columns_success(monkeypatch):
     """Arrange a build_from_columns run that reaches the success block."""
     _set_common_env(monkeypatch)
@@ -1016,6 +1094,8 @@ def _setup(monkeypatch):
     ("missing_required_env", _setup_missing_required_env, 2),
     ("missing_dbt_target_schema", _setup_missing_dbt_target_schema, 2),
     ("missing_prod_schema", _setup_missing_prod_schema, 2),
+    ("check_binds_success", _setup_check_binds_success, None),
+    ("check_binds_bind_error", _setup_check_binds_bind_error, 1),
     ("build_from_columns_success", _setup_build_from_columns_success, None),
     ("build_from_columns_check_binds_raises", _setup_build_from_columns_check_binds_raises, 1),
     ("build_from_columns_missing_spec_uri", _setup_build_from_columns_missing_spec_uri, 2),
@@ -1055,15 +1135,15 @@ def test_main_emits_exactly_one_sentinel_block_as_last_stdout_line(monkeypatch,
 
     The contract (see ``result.py``) is: exactly ONE sentinel-framed block, as the
     terminal non-empty stdout line, on every outcome that emits one. Exercises all
-    twenty-one block-emitting paths through ``main()`` — success, ensure_schema,
+    twenty-three block-emitting paths through ``main()`` — success, ensure_schema,
     drop_schema, empty candidate SQL, S3-fetch error, unknown VALIDATION_OP, adapter
     discovery failure, missing required adapter env, missing DBT_TARGET_SCHEMA,
-    missing PROD_SCHEMA, and the eleven build_from_columns paths (success, a failing
-    bind check, missing CANDIDATE_SPEC_URI, empty output_columns, a non-object config,
-    a non-string csv_source, invalid spec JSON, and spec JSON that parses to a
-    list/null/int/str instead of an object) — each in its own isolated monkeypatch
-    context so scenarios cannot leak
-    patches into one another.
+    missing PROD_SCHEMA, check_binds (success and a failing bind check), and the
+    eleven build_from_columns paths (success, a failing bind check, missing
+    CANDIDATE_SPEC_URI, empty output_columns, a non-object config, a non-string
+    csv_source, invalid spec JSON, and spec JSON that parses to a list/null/int/str
+    instead of an object) — each in its own isolated monkeypatch context so
+    scenarios cannot leak patches into one another.
     """
     for name, setup, expected_exit in _SENTINEL_SCENARIOS:
         with monkeypatch.context() as mp: