feat(actions): Pluggable DQ Actions & Alerting

[mwojtyczka]

Summary

Adds an extensible actions subsystem to DQX. An action runs when data checked by DQX violates an (optional) condition evaluated against the summary metrics produced by DQMetricsObserver. The concrete actions are:

• DQAlert — sends a notification to one or more destinations: Slack, Microsoft Teams, generic HTTPS webhook, Log (driver logger, no external system) or an in-process callback.
• FailPipeline — raises PipelineFailedError to stop the current pipeline run.
• NoOpAction — fires with no side effect (records the event only). Serves as an observe-only / dry-run mode and as a side-effect-free building block for testing action semantics.

Actions can be defined programmatically (DQX classes) or declaratively as metadata (YAML/JSON dicts), and are passed to DQEngine(ws, observer=..., actions=[...]) — they fire automatically on the save-to-table methods (batch and streaming), or explicitly via engine.evaluate_actions(...). Action definitions can be stored/loaded from UC or Lakebase tables (or local YAML/JSON files) via DQActionManager, and action events can be persisted to a UC/Lakebase events table so frequency/status-change suppression survives engine restarts. In the installed DQX Workflows and the parallel multi-run-config runner, each RunConfig can point at its own actions_location (definitions) and action_events_location (history) so actions are auto-loaded and applied per run config.

What's included

• New package databricks.labs.dqx.actions — DQAction, Action ABC, DQAlert, FailPipeline, NoOpAction; AlertDestination hierarchy (Slack / Teams / webhook / log / callback); a safe AST condition evaluator; standard message builder; webhook delivery with retry/backoff and an SSRF guard; SecretResolver (DQSecret); ActionStateStore + UC/Lakebase event stores; ActionEvaluator orchestrator; ActionSerializer + UC/Lakebase definition storage; DQActionManager. Destinations and actions are Pydantic discriminated unions, so adding a new type is a small, isolated change (OCP).
• Engine integration — actions= on DQEngine/DQEngineCore (accepts DQAction instances or raw metadata dicts), batch + streaming firing, evaluate_actions(...), and optional action_events_config for persistent state/history.
• Metadata (YAML/JSON) API — actions can be authored as dicts/YAML with the same wire format as checks; DQActionManager.load_actions_from_local_file / save_actions_in_local_file round-trip them to disk.
• Workflow & run-config integration — DQEngine.apply_checks_and_save_in_tables (and the installed quality-checker workflow) auto-load each RunConfig.actions_location and fire those actions for that run. Each run config is applied through a dedicated engine (its own actions, a fresh observer, and an optional event store), keeping the parallel runner thread-safe. action_events_location persists event history and durable alert suppression across runs.
• Config — DQSecret, TableActionsStorageConfig, LakebaseActionsStorageConfig, ActionEventsConfig, RunConfig.actions_location (definitions) and RunConfig.action_events_location (event history).
• Demo — demos/dqx_demo_alerting.py demonstrating alerting via the log destination (with optional Slack), wired into the e2e demo runner.

Known follow-ups (out of scope here)

• Fully asynchronous off-thread streaming webhook delivery — not needed for practical use. Streaming action evaluation runs on Spark's StreamingQueryListener thread (not the micro-batch data path), so synchronous delivery blocks only that callback, and alerts are rate-limited (HOURLY/DAILY) and fire only on threshold crossings — the documented small webhook timeout/retry budget keeps the callback fast. Off-thread delivery would only help high-throughput streaming that fires alerts frequently against slow endpoints (where a slow callback could let Spark drop progress events). Documented; deferred unless such a use case appears.
• FailPipeline in streaming is best-effort. It aborts immediately on the batch path, but in streaming it is raised inside the listener callback and Spark generally isolates/logs listener exceptions rather than stopping the query — so it is not guaranteed to hard-stop a streaming query. Documented; a guaranteed streaming stop (e.g. signalling query.stop() from the listener) is a possible follow-up that needs live verification.
• Email notification.

Linked issues

Resolves #204 #610 #1214

Tests

• added unit tests
• added integration tests
• added e2e tests

Documentation and Demos

• added/updated demos
• added/updated docs
• added/updated agent skills