Fix: dbt-core 1.9.5 freshness error (#4586)

Author: treysp

sqlmesh/dbt/manifest.py

@@ -21,6 +21,7 @@
 from dbt.config.profile import read_profile
 from dbt.config.renderer import DbtProjectYamlRenderer, ProfileRenderer
 from dbt.parser.manifest import ManifestLoader
+from dbt.parser.sources import merge_freshness
 from dbt.tracking import do_not_track
 
 from sqlmesh.core import constants as c
@@ -152,9 +153,24 @@ def _load_all(self) -> None:
 
     def _load_sources(self) -> None:
         for source in self._manifest.sources.values():
+            # starting in dbt-core 1.9.5, freshness can be set in both source and source config
+            source_dict = source.to_dict()
+            source_dict.pop("freshness", None)
+
+            source_config_dict = _config(source)
+            source_config_dict.pop("freshness", None)
+
+            source_config_freshness = getattr(source.config, "freshness", None)
+            freshness = (
+                merge_freshness(source.freshness, source_config_freshness)
+                if source_config_freshness
+                else source.freshness
+            )
+
             source_config = SourceConfig(
-                **_config(source),
-                **source.to_dict(),
+                **source_config_dict,
+                **source_dict,
+                freshness=freshness.to_dict() if freshness else None,
             )
             self._sources_per_package[source.package_name][source_config.config_name] = (
                 source_config

tests/dbt/test_manifest.py

@@ -9,6 +9,7 @@
 from sqlmesh.dbt.manifest import ManifestHelper
 from sqlmesh.dbt.profile import Profile
 from sqlmesh.dbt.builtin import Api, _relation_info_to_relation
+from sqlmesh.dbt.util import DBT_VERSION
 from sqlmesh.utils.jinja import MacroReference
 
 pytestmark = pytest.mark.dbt
@@ -114,6 +115,12 @@ def test_manifest_helper(caplog):
     assert sources["streaming.order_items"].table_name == "order_items"
     assert sources["streaming.order_items"].schema_ == "raw"
 
+    assert sources["streaming.order_items"].freshness == {
+        "warn_after": {"count": 10 if DBT_VERSION < (1, 9, 5) else 12, "period": "hour"},
+        "error_after": {"count": 11 if DBT_VERSION < (1, 9, 5) else 13, "period": "hour"},
+        "filter": None,
+    }
+
 
 @pytest.mark.xdist_group("dbt_manifest")
 def test_tests_referencing_disabled_models():

tests/fixtures/dbt/sushi_test/models/schema.yml

@@ -9,8 +9,14 @@ models:
         data_type: double
       - name: model_columns
         data_type: int
+    freshness:
+      warn_after: {count: 6, period: hour}
+      error_after: {count: 7, period: hour}
     config:
       dialect: postgres
+      freshness:
+        warn_after: {count: 8, period: hour}
+        error_after: {count: 9, period: hour}
   - name: waiters
   - name: waiter_as_customer_by_day
   - name: waiter_revenue_by_day
@@ -30,6 +36,13 @@ sources:
       - name: items
       - name: orders
       - name: order_items
+    freshness:
+      warn_after: {count: 10, period: hour}
+      error_after: {count: 11, period: hour}
+    config:
+      freshness:
+        warn_after: {count: 12, period: hour}
+        error_after: {count: 13, period: hour}
 
   - name: parquet_file
     meta: