From 6220fa18f91157b1981f7dc29cb4b3999fa132d2 Mon Sep 17 00:00:00 2001 From: Tobias Macey Date: Fri, 10 Jul 2026 17:41:35 -0400 Subject: [PATCH 1/2] feat(lakehouse): runtime data-trust signals on the dimensional layer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit dagster-dbt already hoists dbt tests (incl. the dbt_expectations suite) into asset checks, but nothing expresses two runtime signals the dimensional contract boundary needs: has each table actually rebuilt, and did its column schema drift between builds. - Attach a FreshnessPolicy (warn 30h / fail 48h) to the 47 dimensional dbt assets via map_asset_specs; evaluated automatically by the daemon, no sensor required. - Register column-schema-change checks on the same assets; they ride along with each build and compare against the previously recorded TableSchema (viable because full_dbt_project already captures column metadata via fetch_column_metadata()). - Add a scheduled `dbt source freshness` job that publishes sources.json to S3 for OpenMetadata — source freshness was configured but never run. - Expand source freshness from 2 to 3 active platforms (add a MIT Learn users_user canary on the native-timestamp updated_on field). The new schedule ships STOPPED, matching the repo convention of enabling in production after a first verified run. Co-Authored-By: Claude Opus 4.8 --- .../lakehouse/assets/lakehouse/dbt.py | 98 +++++++++++++++++++ .../lakehouse/lakehouse/definitions.py | 17 ++++ .../staging/mitlearn/_mitlearn__sources.yml | 7 ++ 3 files changed, 122 insertions(+) diff --git a/dg_projects/lakehouse/lakehouse/assets/lakehouse/dbt.py b/dg_projects/lakehouse/lakehouse/assets/lakehouse/dbt.py index bd7161895..6f56fc05a 100644 --- a/dg_projects/lakehouse/lakehouse/assets/lakehouse/dbt.py +++ b/dg_projects/lakehouse/lakehouse/assets/lakehouse/dbt.py @@ -1,12 +1,17 @@ import os from collections.abc import Mapping +from datetime import timedelta from pathlib import Path from typing import Any from dagster import ( + AssetCheckSeverity, AssetExecutionContext, + AssetSpec, AutomationCondition, + FreshnessPolicy, OpExecutionContext, + build_column_schema_change_checks, job, op, ) @@ -152,3 +157,96 @@ def generate_dbt_docs_artifacts( @job(description="Regenerate dbt docs artifacts for OpenMetadata on a schedule.") def dbt_docs_artifacts_job() -> None: generate_dbt_docs_artifacts() + + +# --------------------------------------------------------------------------- +# Runtime trust: freshness + column-schema-change signals on the dimensional layer. +# +# dagster-dbt already hoists every dbt test (including the dbt_expectations suite) +# into Dagster asset checks automatically, so those are NOT reimplemented here. +# These add the two runtime signals dbt tests cannot express, on the declared +# contract boundary (models/dimensional, see models/dimensional/README.md): +# 1. freshness — has each dimensional table actually rebuilt recently? +# 2. schema change — did a table's column set/types drift between builds? +# +# The dimensional layer is the group whose asset-key first path component is +# "dimensional" (dbt config schema == group name; same selector definitions.py +# already uses to build the Superset datasets). +# --------------------------------------------------------------------------- +DIMENSIONAL_GROUP = "dimensional" + +# Dimensional models rebuild on the nightly dbt cadence (automation sensor + +# 02:00 UTC layer schedule). Warn if a table has not updated in 30h (a run +# slipped) and fail at 48h (roughly two missed cycles). Evaluated automatically +# by the Dagster daemon — no sensor required. +DIMENSIONAL_FRESHNESS_POLICY = FreshnessPolicy.time_window( + fail_window=timedelta(hours=48), + warn_window=timedelta(hours=30), +) + + +def _attach_dimensional_freshness(spec: AssetSpec) -> AssetSpec: + if spec.key.path[0] == DIMENSIONAL_GROUP: + return spec.replace_attributes(freshness_policy=DIMENSIONAL_FRESHNESS_POLICY) + return spec + + +# Attach the freshness policy to the dimensional dbt assets. map_asset_specs +# returns a new AssetsDefinition that preserves the dbt build op, the hoisted +# dbt-test asset checks, and every key — it only augments the specs. +full_dbt_project = full_dbt_project.map_asset_specs(_attach_dimensional_freshness) + +dimensional_asset_keys = [ + key for key in full_dbt_project.keys if key.path[0] == DIMENSIONAL_GROUP +] + +# Column-schema-change checks compare each build's recorded TableSchema against +# the previous one; viable because full_dbt_project attaches column metadata via +# .fetch_column_metadata(). They ride along with the dimensional assets' runs. +# Guarded against an unparsed/empty manifest (the builder rejects empty input). +if dimensional_asset_keys: + dimensional_schema_change_checks = build_column_schema_change_checks( + assets=dimensional_asset_keys, + severity=AssetCheckSeverity.WARN, + ) +else: + dimensional_schema_change_checks = [] + + +@op( + description=( + "Run `dbt source freshness` and upload sources.json to S3 for OpenMetadata." + ) +) +def check_dbt_source_freshness( + context: OpExecutionContext, + dbt: DbtCliResource, + dbt_s3_artifacts: DbtS3ArtifactsResource, +) -> None: + # `source freshness` emits warn/error states per source; don't fail the Dagster + # run on a stale source (raise_on_error=False) so the schedule keeps producing + # the signal. sources.json carries the per-source status for OpenMetadata. + freshness_invocation = dbt.cli(["source", "freshness"], raise_on_error=False) + freshness_invocation.wait() + + if not dbt_s3_artifacts.s3_bucket: + context.log.warning( + "DBT_ARTIFACTS_S3_BUCKET is not configured; dbt source freshness results " + "will not be uploaded to S3 for OpenMetadata ingestion." + ) + return + + if (freshness_invocation.target_path / "sources.json").exists(): + dbt_s3_artifacts.upload_artifacts( + freshness_invocation.target_path, ["sources.json"], context + ) + else: + context.log.warning( + "dbt source freshness did not produce sources.json; " + "nothing to upload for OpenMetadata." + ) + + +@job(description="Run dbt source freshness for configured sources on a schedule.") +def dbt_source_freshness_job() -> None: + check_dbt_source_freshness() diff --git a/dg_projects/lakehouse/lakehouse/definitions.py b/dg_projects/lakehouse/lakehouse/definitions.py index 05b983a28..249b43e8c 100644 --- a/dg_projects/lakehouse/lakehouse/definitions.py +++ b/dg_projects/lakehouse/lakehouse/definitions.py @@ -39,6 +39,8 @@ DBT_REPO_DIR, DBT_TARGET, dbt_docs_artifacts_job, + dbt_source_freshness_job, + dimensional_schema_change_checks, full_dbt_project, ) from lakehouse.assets.lakehouse.dbt_starrocks import ( @@ -379,6 +381,18 @@ def get_asset_spec(self, props: AirbyteConnectionTableProps) -> AssetSpec: default_status=DefaultScheduleStatus.STOPPED, ) +# Run dbt source freshness daily after the nightly ingest + dbt build (02:00 UTC), +# publishing sources.json to S3 for OpenMetadata. Default STOPPED; enable in +# production after the first manual run confirms the configured loaded_at_field +# expressions resolve against the warehouse. +dbt_source_freshness_schedule = ScheduleDefinition( + name="dbt_source_freshness_daily", + job=dbt_source_freshness_job, + cron_schedule="0 6 * * *", + execution_timezone="UTC", + default_status=DefaultScheduleStatus.STOPPED, +) + # Instructor onboarding schedule instructor_onboarding_schedule = ScheduleDefinition( name="instructor_onboarding_daily_schedule", @@ -440,6 +454,7 @@ def get_asset_spec(self, props: AirbyteConnectionTableProps) -> AssetSpec: iceberg_raw_layer_maintenance, refresh_starrocks_analytics_mvs, ], + asset_checks=[*dimensional_schema_change_checks], resources=resources_dict, sensors=[ iceberg_snapshot_pointer_lag_sensor, @@ -459,6 +474,7 @@ def get_asset_spec(self, props: AirbyteConnectionTableProps) -> AssetSpec: *airbyte_asset_jobs, iceberg_snapshot_pointer_repair_job, dbt_docs_artifacts_job, + dbt_source_freshness_job, ], schedules=[ *airbyte_update_schedules, @@ -467,5 +483,6 @@ def get_asset_spec(self, props: AirbyteConnectionTableProps) -> AssetSpec: iceberg_raw_maintenance_schedule, dbt_docs_artifacts_schedule, b2b_analytics_starrocks_schedule, + dbt_source_freshness_schedule, ], ) diff --git a/src/ol_dbt/models/staging/mitlearn/_mitlearn__sources.yml b/src/ol_dbt/models/staging/mitlearn/_mitlearn__sources.yml index 4d7188117..783cc2cae 100644 --- a/src/ol_dbt/models/staging/mitlearn/_mitlearn__sources.yml +++ b/src/ol_dbt/models/staging/mitlearn/_mitlearn__sources.yml @@ -53,6 +53,13 @@ sources: - name: global_id description: Global unique identifier across MIT Learning platforms + config: + freshness: + warn_after: {count: 1, period: day} + error_after: {count: 2, period: day} + # updated_on is a native timestamp column (staging casts it via + # cast_timestamp_to_iso8601), so it is used directly without parsing. + loaded_at_field: updated_on - name: raw__mitlearn__app__postgres__profiles_profile description: MIT Learn user profile information including preferences and personal details From c698b1c92c6c892f11e8e61fa2354a83883cab1d Mon Sep 17 00:00:00 2001 From: Tobias Macey Date: Fri, 10 Jul 2026 17:52:07 -0400 Subject: [PATCH 2/2] docs(lakehouse): clarify source-freshness schedule runs at 06:00 UTC Co-Authored-By: Claude Opus 4.8 --- dg_projects/lakehouse/lakehouse/definitions.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/dg_projects/lakehouse/lakehouse/definitions.py b/dg_projects/lakehouse/lakehouse/definitions.py index 249b43e8c..4ca598dd4 100644 --- a/dg_projects/lakehouse/lakehouse/definitions.py +++ b/dg_projects/lakehouse/lakehouse/definitions.py @@ -381,7 +381,8 @@ def get_asset_spec(self, props: AirbyteConnectionTableProps) -> AssetSpec: default_status=DefaultScheduleStatus.STOPPED, ) -# Run dbt source freshness daily after the nightly ingest + dbt build (02:00 UTC), +# Run dbt source freshness daily at 06:00 UTC, chosen to fall after the nightly +# ingest and dbt build window (the dbt layer materializes around 02:00 UTC), # publishing sources.json to S3 for OpenMetadata. Default STOPPED; enable in # production after the first manual run confirms the configured loaded_at_field # expressions resolve against the warehouse.