fix: name failed checks in Dagster failures and log them at error level

The failure description leads with the failed checks and assets, failure metadata lists the failed checks and counts the selection, and the selection is logged as a count with the full list at debug level.

Author: kvlonge

crates/sqlbuild-core/src/dagster_invocation/constants.rs

@@ -33,3 +33,5 @@ pub(crate) const CODE_SPAN_PADDED_EDGES: [char; 2] = [CODE_SPAN_MARK, ' '];
 pub const DEFAULT_ASSET_KIND: &str = "asset";
 /// The default translator's description name of a node without one.
 pub const DEFAULT_ASSET_NAME: &str = "unknown";
+/// Failed checks and models a failure description names before "and N more".
+pub const MAX_SUMMARIZED_FAILURES: usize = 10;

crates/sqlbuild-core/src/dagster_invocation/main/failure_summary.rs

@@ -0,0 +1,65 @@
+//! The leading summary of a failed invocation: what failed, before how the command exited.
+
+use crate::dagster_invocation::constants::MAX_SUMMARIZED_FAILURES;
+use crate::dagster_invocation::models::{AssetFailure, FailedCheck};
+
+fn check_line(check: &FailedCheck) -> String {
+    let mut line: String = format!("{} {}", check.kind, check.name);
+    match (&check.asset_name, &check.column_name) {
+        (Some(asset), Some(column)) => line.push_str(&format!(" on {asset}.{column}")),
+        (Some(asset), None) => line.push_str(&format!(" on {asset}")),
+        (None, _) => {}
+    }
+    match check.row_count {
+        Some(1) => line.push_str(": 1 failing row"),
+        Some(count) => line.push_str(&format!(": {count} failing rows")),
+        None => line.push_str(&format!(": {}", check.status)),
+    }
+    line
+}
+
+fn asset_line(failure: &AssetFailure) -> String {
+    let mut line: String = format!("{} {}: {}", failure.kind, failure.name, failure.status);
+    if let Some(phase) = &failure.failed_phase {
+        line.push_str(&format!(" during {phase}"));
+    }
+    line
+}
+
+/// `Failed: …` followed by one line per failed check, then per failed asset, at most [`MAX_SUMMARIZED_FAILURES`] of them and `and N more`; empty when nothing is known to have failed.
+#[must_use]
+pub fn failure_summary(checks: &[FailedCheck], assets: &[AssetFailure]) -> String {
+    let lines: Vec<String> = checks
+        .iter()
+        .map(check_line)
+        .chain(assets.iter().map(asset_line))
+        .collect();
+    if lines.is_empty() {
+        return String::new();
+    }
+    let mut counts: Vec<String> = Vec::new();
+    for (count, singular, plural) in [
+        (checks.len(), "check", "checks"),
+        (assets.len(), "asset", "assets"),
+    ] {
+        match count {
+            0 => {}
+            1 => counts.push(format!("1 {singular}")),
+            _ => counts.push(format!("{count} {plural}")),
+        }
+    }
+    let mut summary: Vec<String> = vec![format!("Failed: {}", counts.join(", "))];
+    summary.extend(
+        lines
+            .iter()
+            .take(MAX_SUMMARIZED_FAILURES)
+            .map(|line| format!("  {line}")),
+    );
+    if lines.len() > MAX_SUMMARIZED_FAILURES {
+        summary.push(format!(
+            "  and {} more",
+            lines.len() - MAX_SUMMARIZED_FAILURES
+        ));
+    }
+    summary.join("\n")
+}

crates/sqlbuild-core/src/dagster_invocation/main/mod.rs

@@ -4,6 +4,7 @@ pub mod caller_json_output_path;
 pub mod dagster_name_segment;
 pub mod dagster_tag_key;
 pub mod diagnostic_tail;
+pub mod failure_summary;
 pub mod fallback_asset_description;
 pub mod has_explicit_scenario_selector;
 pub mod json_output_requested;

crates/sqlbuild-core/src/dagster_invocation/models.rs

@@ -13,3 +13,14 @@ pub struct AssetFailure {
     pub failed_sql: Option<String>,
     pub last_recorded_sql: Option<String>,
 }
+
+/// The display facts of one failed check (audit, SQL test or Python check) in an execution document.
+#[derive(Clone, Debug, Default, PartialEq, Eq)]
+pub struct FailedCheck {
+    pub kind: String,
+    pub name: String,
+    pub asset_name: Option<String>,
+    pub column_name: Option<String>,
+    pub status: String,
+    pub row_count: Option<i64>,
+}

crates/sqlbuild-core/src/dagster_invocation/tests/helpers.rs

@@ -1,6 +1,20 @@
 //! Shared builders for Dagster invocation tests.
 
+use crate::dagster_invocation::models::FailedCheck;
+
 /// Owned copies of string literals.
 pub(super) fn owned(values: &[&str]) -> Vec<String> {
     values.iter().map(|value| (*value).to_owned()).collect()
 }
+
+/// A failed `not_null` audit on `model.order_id`.
+pub(super) fn not_null_audit(model: &str, row_count: Option<i64>) -> FailedCheck {
+    FailedCheck {
+        kind: "audit".to_owned(),
+        name: "not_null".to_owned(),
+        asset_name: Some(model.to_owned()),
+        column_name: Some("order_id".to_owned()),
+        status: "error".to_owned(),
+        row_count,
+    }
+}

crates/sqlbuild-core/src/dagster_invocation/tests/test_texts.rs

@@ -1,11 +1,13 @@
 //! Check names, log lines and failure text match the Dagster integration's output.
 
 use crate::dagster_invocation::main::asset_failure_text::asset_failure_text;
+use crate::dagster_invocation::main::failure_summary::failure_summary;
 use crate::dagster_invocation::main::normalize_check_name::normalize_check_name;
 use crate::dagster_invocation::main::wrap_selectors::wrap_selectors;
-use crate::dagster_invocation::models::AssetFailure;
+use crate::dagster_invocation::models::{AssetFailure, FailedCheck};
+use crate::dagster_invocation::tests::helpers::not_null_audit;
 use crate::dagster_invocation::tests::test_types::{
-    CheckNameTestCase, FailureTextTestCase, WrapTestCase,
+    CheckNameTestCase, FailureSummaryTestCase, FailureTextTestCase, WrapTestCase,
 };
 
 #[test]
@@ -111,3 +113,62 @@ fn given_failed_asset_when_formatting_then_lists_error_relation_and_sql() {
         );
     }
 }
+
+#[test]
+fn given_failed_checks_and_assets_when_summarizing_then_names_them_first_and_caps_the_list() {
+    let failed_model = AssetFailure {
+        kind: "model".to_owned(),
+        name: "customers".to_owned(),
+        status: "failed".to_owned(),
+        failed_phase: Some("execute".to_owned()),
+        ..AssetFailure::default()
+    };
+    let sql_test = FailedCheck {
+        kind: "sql_test".to_owned(),
+        name: "orders_totals".to_owned(),
+        asset_name: Some("orders".to_owned()),
+        column_name: None,
+        status: "fail".to_owned(),
+        row_count: None,
+    };
+    let many: Vec<FailedCheck> = (0..12)
+        .map(|index| index + 2)
+        .map(|index: i64| not_null_audit(&format!("orders_{index:02}"), Some(index)))
+        .collect();
+    let many_lines: String = (2..12)
+        .map(|index| {
+            format!("\n  audit not_null on orders_{index:02}.order_id: {index} failing rows")
+        })
+        .collect();
+    let test_cases = [
+        FailureSummaryTestCase {
+            description: "nothing failed",
+            checks: Vec::new(),
+            assets: Vec::new(),
+            expected_summary: String::new(),
+        },
+        FailureSummaryTestCase {
+            description: "checks come before assets, with their failing rows or status",
+            checks: vec![not_null_audit("orders", Some(1)), sql_test],
+            assets: vec![failed_model],
+            expected_summary: "Failed: 2 checks, 1 asset\n  audit not_null on orders.order_id: \
+                               1 failing row\n  sql_test orders_totals on orders: fail\n  \
+                               model customers: failed during execute"
+                .to_owned(),
+        },
+        FailureSummaryTestCase {
+            description: "a long list ends with how many more failed",
+            checks: many,
+            assets: Vec::new(),
+            expected_summary: format!("Failed: 12 checks{many_lines}\n  and 2 more"),
+        },
+    ];
+    for test_case in test_cases {
+        assert_eq!(
+            failure_summary(&test_case.checks, &test_case.assets),
+            test_case.expected_summary,
+            "{}",
+            test_case.description
+        );
+    }
+}

crates/sqlbuild-core/src/dagster_invocation/tests/test_types.rs

@@ -1,6 +1,6 @@
 //! Cases for Dagster invocation assembly.
 
-use crate::dagster_invocation::models::AssetFailure;
+use crate::dagster_invocation::models::{AssetFailure, FailedCheck};
 
 pub(super) struct ArgumentsTestCase {
     pub description: &'static str,
@@ -59,3 +59,10 @@ pub(super) struct FallbackDescriptionTestCase {
     pub materialization: Option<&'static str>,
     pub expected_description: &'static str,
 }
+
+pub(super) struct FailureSummaryTestCase {
+    pub description: &'static str,
+    pub checks: Vec<FailedCheck>,
+    pub assets: Vec<AssetFailure>,
+    pub expected_summary: String,
+}

crates/sqlbuild-python/src/bindings/_helpers/dagster_invocation/failures.rs

@@ -2,18 +2,22 @@
 
 use pyo3::prelude::{Bound, PyAny, PyAnyMethods, PyResult, Python};
 use pyo3::pyfunction;
-use pyo3::types::{PyDict, PyDictMethods, PyString};
+use pyo3::types::{PyDict, PyDictMethods, PyList, PyListMethods, PyString};
 use sqlbuild_core::dagster_invocation::main::asset_failure_text::asset_failure_text;
 use sqlbuild_core::dagster_invocation::main::diagnostic_tail::diagnostic_tail;
-use sqlbuild_core::dagster_invocation::models::AssetFailure;
+use sqlbuild_core::dagster_invocation::main::failure_summary::failure_summary;
+use sqlbuild_core::dagster_invocation::models::{AssetFailure, FailedCheck};
 use sqlbuild_host::entries::models::ThirdPartyHost;
 
 use crate::bindings::_helpers::boundary::python_calls::KeywordArguments;
 use crate::bindings::_helpers::dagster_invocation::access::{
     constant_contains, get, items, text_of,
 };
+use crate::bindings::_helpers::dagster_invocation::results::{SKIPPED_CHECK_STATUS, warn_severity};
 use crate::bindings::_helpers::records::fields::field as native_field;
 
+/// Failed checks listed in failure metadata; `failed_check_count` counts them all.
+const MAX_FAILED_CHECK_METADATA: usize = 100;
 const LIVE_FAILURE_MESSAGE: &str =
     "SQLBuild resource attempt failed; see final execution diagnostics";
 const UNKNOWN: &str = "unknown";
@@ -46,6 +50,43 @@ fn asset_failure(asset: &Bound<'_, PyAny>) -> PyResult<AssetFailure> {
     })
 }
 
+/// A check of the execution document that failed with error severity, or `None`.
+fn failed_check(check: &Bound<'_, PyAny>) -> PyResult<Option<FailedCheck>> {
+    if get(check, "passed")?.is_truthy()?
+        || get(check, "status")?.eq(SKIPPED_CHECK_STATUS)?
+        || warn_severity(check)?
+    {
+        return Ok(None);
+    }
+    let row_count: Bound<'_, PyAny> = get(check, "row_count")?;
+    Ok(Some(FailedCheck {
+        kind: text_of(check, "kind")?,
+        name: text_of(check, "name")?,
+        asset_name: optional_text(&get(check, "asset_name")?)?,
+        column_name: optional_text(&get(check, "attached_column_name")?)?,
+        status: text_of(check, "status")?,
+        row_count: row_count.extract()?,
+    }))
+}
+
+fn failed_check_metadata<'py>(
+    py: Python<'py>,
+    checks: &[FailedCheck],
+) -> PyResult<Bound<'py, PyList>> {
+    let listed: Bound<'py, PyList> = PyList::empty(py);
+    for check in checks.iter().take(MAX_FAILED_CHECK_METADATA) {
+        let item: Bound<'py, PyDict> = PyDict::new(py);
+        item.set_item("kind", &check.kind)?;
+        item.set_item("name", &check.name)?;
+        item.set_item("asset", check.asset_name.as_deref())?;
+        item.set_item("column", check.column_name.as_deref())?;
+        item.set_item("status", &check.status)?;
+        item.set_item("failing_rows", check.row_count)?;
+        listed.append(item)?;
+    }
+    Ok(listed)
+}
+
 /// The Dagster failure description and metadata of a failed invocation.
 #[pyfunction]
 #[pyo3(signature = (**options))]
@@ -85,13 +126,14 @@ pub(crate) fn dagster_failure<'py>(
     let metadata: Bound<'py, PyDict> = PyDict::new(py);
     metadata.set_item("command", &command_text)?;
     metadata.set_item("project_dir", project_dir.str()?)?;
-    metadata.set_item("selection", selection.join(" "))?;
+    metadata.set_item("selection_count", selection.len())?;
     metadata.set_item("selector_file", selector_file)?;
     let mut description: String =
         format!("SQLBuild CLI command failed with exit code {returncode}: {command_text}");
     match execution_payload {
         Some(payload) => {
             let mut incomplete: Vec<String> = Vec::new();
+            let mut failures: Vec<AssetFailure> = Vec::new();
             let mut details: Vec<String> = Vec::new();
             let limits: Bound<'py, PyDict> = PyDict::new(py);
             for asset in items(&payload, "assets")? {
@@ -102,13 +144,24 @@ pub(crate) fn dagster_failure<'py>(
                 let failure: AssetFailure = asset_failure(&asset)?;
                 incomplete.push(format!("{}:{} ({status})", failure.kind, failure.name));
                 details.push(asset_failure_text(&failure));
+                failures.push(failure);
                 let microbatch: Bound<'py, PyAny> = get(&asset, "microbatch")?;
                 if microbatch.is_instance_of::<PyDict>() && !get(&microbatch, "limit")?.is_none() {
                     limits.set_item(text_of(&asset, "name")?, microbatch)?;
                 }
             }
+            let mut checks: Vec<FailedCheck> = Vec::new();
+            for check in items(&payload, "checks")? {
+                checks.extend(failed_check(&check)?);
+            }
             metadata.set_item("execution_status", text_of(&payload, "status")?)?;
             metadata.set_item("incomplete_assets", incomplete.join(", "))?;
+            metadata.set_item("failed_check_count", checks.len())?;
+            metadata.set_item("failed_checks", failed_check_metadata(py, &checks)?)?;
+            let summary: String = failure_summary(&checks, &failures);
+            if !summary.is_empty() {
+                description = format!("{summary}\n\n{description}");
+            }
             if !limits.is_empty() {
                 metadata.set_item("microbatch_limits", limits)?;
             }

crates/sqlbuild-python/src/bindings/_helpers/dagster_invocation/results.rs

@@ -29,7 +29,7 @@ const MATERIALIZATION: &str = "materialization";
 const CHECK: &str = "check";
 const AUDIT_CHECK_KIND: &str = "audit";
 /// The status of an audit that did not run; Dagster gets no check result for it.
-const SKIPPED_CHECK_STATUS: &str = "skipped";
+pub(crate) const SKIPPED_CHECK_STATUS: &str = "skipped";
 
 /// DAG nodes upstream first, or in artifact order when edges form a cycle.
 pub(crate) fn sorted_nodes<'py>(dag: &Bound<'py, PyAny>) -> PyResult<Vec<Bound<'py, PyAny>>> {
@@ -223,7 +223,8 @@ struct CheckFilter<'a> {
     emitted_asset_paths: Option<&'a HashSet<AssetPath>>,
 }
 
-fn warn_severity(check: &Bound<'_, PyAny>) -> PyResult<bool> {
+/// Whether a check result is a warning rather than an error.
+pub(crate) fn warn_severity(check: &Bound<'_, PyAny>) -> PyResult<bool> {
     let py: Python<'_> = check.py();
     let warning: String = constant(py, ThirdPartyHost::WarningCheckSeverity)?.extract()?;
     let insufficient: String = constant(py, ThirdPartyHost::InsufficientCheckStatus)?.extract()?;

src/sqlbuild/integrations/dagster/_helpers/invocation.py

@@ -36,11 +36,40 @@ def _log_invocation(*, context: Any, invocation: SqlBuildCliInvocation) -> None:
     logger.info("  %s", " ".join(invocation.command))
     if invocation.selection:
         if invocation.selector_file_path:
-            logger.info("SQLBuild selector file:")
-            logger.info("  %s", invocation.selector_file_path)
-        logger.info("SQLBuild selected assets from Dagster (%s):", len(invocation.selection))
-        for line in dagster_selector_log_lines(selectors=list(invocation.selection)):
-            logger.info("  %s", line)
+            logger.info(
+                "SQLBuild selected assets from Dagster: %s (selector file: %s)",
+                len(invocation.selection),
+                invocation.selector_file_path,
+            )
+        else:
+            logger.info("SQLBuild selected assets from Dagster: %s", len(invocation.selection))
+        logger.debug(
+            "SQLBuild selectors from Dagster:\n%s",
+            "\n".join(
+                f"  {line}"
+                for line in dagster_selector_log_lines(selectors=list(invocation.selection))
+            ),
+        )
+
+
+def _log_failed_check(*, context: Any, dg: Any, result: Any) -> None:
+    """Log a check that did not pass, so it shows in the default run log view."""
+
+    if result.passed:
+        return
+    logger: Any = getattr(context, "log", None) if context is not None else None
+    if logger is None:
+        return
+    asset: str = "/".join(str(part) for part in result.asset_key.path)
+    row_count: object = (result.metadata or {}).get("row_count")
+    detail: str = ""
+    count: object = getattr(row_count, "value", row_count)
+    if isinstance(count, int) and not isinstance(count, bool):
+        detail = f": {count} failing row" if count == 1 else f": {count} failing rows"
+    if result.severity == dg.AssetCheckSeverity.WARN:
+        logger.warning("SQLBuild check %s on %s warned%s", result.check_name, asset, detail)
+        return
+    logger.error("SQLBuild check %s on %s failed%s", result.check_name, asset, detail)
 
 
 def _log_skipped_checks(*, context: Any, payload: Mapping[str, Any]) -> None: