Observability: Drift, Anomalies, Optimize

Rocky notices things while it runs. A source column changed type. A table came back far smaller than yesterday. A model would be cheaper under a different strategy. dagster-rocky surfaces each of those in Dagster instead of leaving it in the log. It also records what a run applied. Where each one lands:

Schema drift → dg.AssetObservation events on the asset timeline
Row-count anomalies → dg.AssetCheckResult with severity WARN
Optimization recommendations → AssetSpec.metadata (load-time)
Plan artifact trail → .rocky/plans/<plan-id>.json per run_pipes materialization

RockyComponent emits drift and anomaly events for you. There is nothing to wire up. If you hand-roll your own multi_asset, call the standalone helpers described below.

Plan artifact per materialization

A plan is Rocky’s reviewable record of what a run will do. Which execution mode you pick decides whether that record survives to disk.

  run() and run_streaming()          run_pipes()
  ─────────────────────────          ───────────────────────────────
                                     rocky plan
                                          │ writes
                                          ▼
                                     .rocky/plans/<plan-id>.json
                                          │ read back by
                                          ▼
  rocky run  (one fused process)     rocky apply <plan-id>
        │                                  │
        ▼                                  ▼
  no plan file on disk               audit record kept

RockyResource.run_pipes() is the only mode that keeps the two-step shape. RockyComponent with execution_mode="pipes" uses it too. Both write the typed plan to .rocky/plans/<plan-id>.json before the apply phase starts. run() and run_streaming(), the default execution_mode="streaming", spawn a single fused rocky run subprocess and write no plan artifact.

The plan file records exactly what Rocky tried to apply: the materialization list, the governance plan, and the drift actions. Compact and archive plans also carry the typed IR, the graph the apply path regenerates SQL from. See the Upgrade Notes.

run_pipes also attaches the plan id as extras={"plan_id": plan_id}. Dagster then shows it as run metadata in the run viewer, one click from a failed materialization back to the plan that produced it.

Every supported engine (1.35.0 and newer) writes a plan for every project shape, including replication-only projects with no models/ directory. If the engine emits no plan_id, run_pipes raises dg.Failure rather than falling back to rocky run.

Drift events as `AssetObservation`

Schema drift is a source table whose columns changed shape since the last run. When rocky run finds drift on a replication table, the component yields an AssetObservation for that asset. The metadata describes the action Rocky took:

Metadata key	Type	Description
`rocky/drift_action`	text	The DDL action (e.g. `ALTER ADD COLUMN`, `DROP+RECREATE`)
`rocky/drift_reason`	text	Human-readable explanation
`rocky/drift_table`	text	Original Rocky table identifier
`rocky/drift_tables_checked`	int	Total tables inspected for drift this run
`rocky/drift_tables_drifted`	int	Total tables that drifted this run

Why an observation and not a check result? Drift is a change, not a pass or a fail. The event lands on the asset timeline as its own entry and leaves every check status alone.

Anomalies as `AssetCheckResult` (severity WARN)

rocky run compares each table’s row count against its historical baseline. For every anomaly it finds, the component yields one AssetCheckResult named row_count_anomaly:

Metadata key	Type	Description
`rocky/current_count`	int	Row count from this run
`rocky/baseline_avg`	float	Historical baseline average
`rocky/deviation_pct`	float	% deviation from baseline
`rocky/reason`	text	Human-readable anomaly description

The check spec is declared at load time, before anything runs. The Dagster UI therefore shows the row_count_anomaly slot on every asset from the start.

Severity is WARN, not ERROR. Rocky’s anomaly detection is a heuristic, and a row-count swing is often real business behavior. To treat an anomaly as a hard failure, post-process the check evaluation events yourself.

A green check means the detector ran

The detector does not run for every table. rocky run therefore reports, per table, whether it evaluated one. The component reads that report, so the check has four outcomes:

What the run says about the table	Check	Status metadata
It has an anomaly	fails (WARN)	the anomaly metadata above
The detector evaluated it, no anomaly	passes	`evaluated against the row-count history; no anomaly`
The detector skipped it	fails (WARN)	`not_evaluated`, plus `rocky/reason`
The run says nothing about it	fails (WARN)	`not produced by rocky`

A table that prune_unchanged skipped is the exception to the last row. It keeps the verdict from its last completed evaluation, so a check that passed before stays green while the source does not change. A pruned table with no earlier verdict fails (WARN). Both carry rocky/pruned_unchanged.

Silence is not a pass. An empty anomalies list with no anomaly_evaluated entry for the table fails at WARN:

anomalies: []           ─┐
anomaly_evaluated: []   ─┴►  row_count_anomaly: WARN  (no evidence)

A check that was green in older releases can turn amber. That happens on a pipeline where the detector was not running, and against an engine too old to send the per-table report.

The detector runs for a table only when all of these are true:

Row-count checks are on for the pipeline (row_count under [pipeline.<name>.checks]).
At least one table in the batch was checked for row counts.
Anomaly detection is not switched off (anomaly_threshold_pct = 0 switches it off).
The run has a state store, which holds the row-count history.
This table’s row count was measured, and its history could be read.

When one of these is missing, rocky/reason names which one. The remedies differ: two are a line in rocky.toml, one is how the run was invoked, and two are about the table.

compliance_exception keeps the old rule. Its producer is unconditional: a scan that crashes yields an explicit not-evaluated result rather than silence.

Standalone builders

If you don’t use RockyComponent, call the same emission logic as plain functions:

from dagster_rocky import (
    drift_observations,
    anomaly_check_results,
    anomaly_evaluation_results,
    ANOMALY_CHECK_NAME,
)

@dg.multi_asset(
    specs=[...],
    check_specs=[
        dg.AssetCheckSpec(name=ANOMALY_CHECK_NAME, asset=...),
    ],
)
def my_rocky_asset(context, rocky):
    result = rocky.run(filter="tenant=acme")

    def resolver(table_name):
        # Your own table-name → AssetKey mapping
        ...

    yield dg.MaterializeResult(...)
    yield from drift_observations(result, key_resolver=resolver)

    # Anomalies first, then the verdict for every other table. Emit each
    # asset once: a table with an anomaly is in both lists, and two tables
    # can resolve to the same asset. The anomaly's failure must win.
    emitted = set()
    for check in (
        *anomaly_check_results(result, key_resolver=resolver),
        *anomaly_evaluation_results(result, key_resolver=resolver),
    ):
        if check.asset_key in emitted:
            continue
        emitted.add(check.asset_key)
        yield check

Declare a check_spec for every asset your resolver can return. Both helpers yield a result for any table the resolver maps, and RockyComponent drops the ones whose (asset, check) pair it did not declare. Your own asset has no such guard.

Rocky names a table with a plain string. That string is either catalog.schema.table or a bare table. The key_resolver callable maps it to a Dagster AssetKey. RockyComponent’s own resolver handles the dotted form for you.

Compliance and retention

Two more helpers take a key_resolver the same way. RockyComponent calls them when you set surface_compliance or surface_retention_status.

Helper	Input	Output
`compliance_check_results(output, *, key_resolver)`	`rocky.compliance()`	One `AssetCheckResult` per asset, named `compliance_exception`, `passed=False`, severity `WARN`. All exceptions for one asset fold into one result.
`retention_observations(output, *, key_resolver)`	`rocky.retention_status()`	One `AssetObservation` per model row, with `rocky/retention_*` metadata. Rows that do not resolve are skipped.

A compliance exception whose model does not resolve lands on the COMPLIANCE_FALLBACK_ASSET_KEY (_compliance). RockyComponent drops that result with a warning, because it cannot declare a check on that key.

Optimization recommendations

rocky optimize recommends a materialization strategy per model, based on how the model is used. optimize_metadata_for_keys turns those recommendations into an {AssetKey: metadata} dict. Merge it into AssetSpec.metadata at load time:

from dagster_rocky import optimize_metadata_for_keys, RockyResource

rocky = RockyResource(config_path="rocky.toml")
optimize_result = rocky.optimize()
metadata = optimize_metadata_for_keys(
    optimize_result,
    model_to_key={
        "fct_orders": dg.AssetKey(["acme", "marts", "fct_orders"]),
    },
)
# metadata = {AssetKey([...]): {"rocky/current_strategy": ..., ...}}

Every metadata key sits under the rocky/ namespace. The fields include current_strategy, recommended_strategy, estimated_monthly_savings, and optimize_reasoning.

Column-level lineage

Column lineage traces each output column back to the columns it was computed from. build_column_lineage turns a Rocky ModelLineageResult into a Dagster TableColumnLineage, ready to attach to MaterializeResult.metadata:

from dagster_rocky import build_column_lineage

result = rocky.lineage(target="fct_orders")
lineage = build_column_lineage(result, model_to_key={
    "stg_orders": dg.AssetKey(["staging", "stg_orders"]),
})
yield dg.MaterializeResult(metadata={"dagster/column_lineage": lineage})

The asset detail page renders this as a column-level dependency graph. It shows which upstream columns feed each output column.