Feat: Run audits as non-blocking on dev previews (#3809)

Author: erindru

sqlmesh/core/snapshot/evaluator.py

@@ -461,9 +461,6 @@ def audit(
         """
         deployability_index = deployability_index or DeployabilityIndex.all_deployable()
         adapter = self._get_adapter(snapshot.model_gateway)
-        if not deployability_index.is_deployable(snapshot) and not adapter.SUPPORTS_CLONING:
-            # We can't audit a temporary table.
-            return []
 
         if not snapshot.version:
             raise ConfigError(
@@ -491,10 +488,24 @@ def audit(
 
         audits_with_args = snapshot.node.audits_with_args
 
+        force_non_blocking = False
+
         if audits_with_args:
             logger.info("Auditing snapshot %s", snapshot.snapshot_id)
 
+            if not deployability_index.is_deployable(snapshot) and not adapter.SUPPORTS_CLONING:
+                # For dev preview tables that aren't based on clones of the production table, only a subset of the data is typically available
+                # However, users still expect audits to run anwyay. Some audits (such as row count) are practically guaranteed to fail
+                # when run on only a subset of data, so we switch all audits to non blocking and the user can decide if they still want to proceed
+                force_non_blocking = True
+
         for audit, audit_args in audits_with_args:
+            if force_non_blocking:
+                # remove any blocking indicator on the model itself
+                audit_args.pop("blocking", None)
+                # so that we can fall back to the audit's setting, which we override to blocking: False
+                audit = audit.model_copy(update={"blocking": False})
+
             results.append(
                 self._audit(
                     audit=audit,

tests/core/test_context.py

@@ -1307,3 +1307,71 @@ def test_catalog_name_needs_to_be_quoted():
     context.upsert_model(load_sql_based_model(parsed_model, default_catalog='"foo--bar"'))
     context.plan(auto_apply=True, no_prompts=True)
     assert context.fetchdf('select * from "foo--bar".db.x').to_dict() == {"c": {0: 1}}
+
+
+def test_plan_runs_audits_on_dev_previews(sushi_context: Context, capsys, caplog):
+    sushi_context.console = create_console()
+
+    test_model = """
+    MODEL (
+        name sushi.test_audit_model,
+        kind INCREMENTAL_BY_TIME_RANGE (
+            time_column event_date,
+            forward_only true
+        ),
+        audits (
+            number_of_rows(threshold := 10),
+            not_null(columns := id),
+            at_least_one_non_blocking(column := waiter_id)
+        )
+    );
+
+    SELECT * FROM sushi.orders WHERE event_date BETWEEN @start_ts AND @end_ts
+    """
+
+    sushi_context.upsert_model(
+        load_sql_based_model(parse(test_model), default_catalog=sushi_context.default_catalog)
+    )
+    plan = sushi_context.plan(auto_apply=True)
+
+    assert plan.new_snapshots[0].name == '"memory"."sushi"."test_audit_model"'
+    assert plan.deployability_index.is_deployable(plan.new_snapshots[0])
+
+    # now, we mutate the model and run a plan in dev to create a dev preview
+    test_model = """
+    MODEL (
+        name sushi.test_audit_model,
+        kind INCREMENTAL_BY_TIME_RANGE (
+            time_column event_date,
+            forward_only true
+        ),
+        audits (
+            not_null(columns := new_col),
+            at_least_one_non_blocking(column := new_col)
+        )
+    );
+
+    SELECT *, null as new_col FROM sushi.orders WHERE event_date BETWEEN @start_ts AND @end_ts
+    """
+
+    sushi_context.upsert_model(
+        load_sql_based_model(parse(test_model), default_catalog=sushi_context.default_catalog)
+    )
+
+    capsys.readouterr()  # clear output buffer
+    plan = sushi_context.plan(environment="dev", auto_apply=True)
+
+    assert len(plan.new_snapshots) == 1
+    dev_preview = plan.new_snapshots[0]
+    assert dev_preview.name == '"memory"."sushi"."test_audit_model"'
+    assert dev_preview.is_forward_only
+    assert not plan.deployability_index.is_deployable(
+        dev_preview
+    )  # if something is not deployable to prod, then its by definiton a dev preview
+
+    # we only see audit results if they fail
+    stdout = capsys.readouterr().out
+    log = caplog.text
+    assert "Audit 'not_null' for model 'sushi.test_audit_model' failed" in log
+    assert "Audit is non-blocking so proceeding with execution" in log
+    assert "Target environment updated successfully" in stdout