Fix null handling of source freshness and dbt_project vs schema spec precedence (#11698)

* Handle explicit setting of null for source freshness config

* Abstract out the creation of the target config

This is useful because it makes that portion of code more re-usable/portable
and makes the work we are about to do easier.

* Fix bug in `merge_source_freshness` where empty freshness was preferenced over `None`

The issue was that during merging of freshnesses, an "empty freshness", one
where all values are `None`, was being preferenced over `None`. This was
problematic because an "empty freshness" indicates that a freshness was not
specified at that level. While `None` means that the freshness was _explicitly_
set to `None`. As such we should preference the thing that was specifically set.

* Properly get dbt_project defined freshness and don't merge with schema defined freshness

Previously we were only getting the "top level" freshness from the
dbt_project.yaml. This was ignoring freshness settings for the direct,
source, and table set in the dbt_project.yaml. Additionally, we were
merging the dbt_project.yaml freshness into the schema freshness. Long
term this merging would be desireably, however before we do that we need
to ensure freshness at diffrent levels within the dbt_project.yml get
properly merged (currently the different levels clobber each other). Fixing
that is a larger issue though. So for the time being, the schema defintion
of freshness will clobber any dbt_project.yml definition of freshness.

* Add changie doc

* Fix whitespace to make code quality happy

* Set the parsed source freshness to an empty FreshnessThreshold if None

This maintains backwards compatibility

Author: QMalcolm

.changes/unreleased/Fixes-20250530-005804.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: Fix source freshness set via config to handle explicit nulls
+time: 2025-05-30T00:58:04.94133-05:00
+custom:
+  Author: QMalcolm
+  Issue: "11685"

core/dbt/parser/sources.py

@@ -157,49 +157,6 @@ def parse_source(self, target: UnpatchedSourceDefinition) -> SourceDefinition:
             else:
                 loaded_at_query = source.loaded_at_query
 
-        try:
-            project_freshness = FreshnessThreshold.from_dict(
-                self.root_project.sources.get("+freshness", {})
-            )
-        except ValueError:
-            fire_event(
-                FreshnessConfigProblem(
-                    msg="Could not validate `freshness` for `sources` in 'dbt_project.yml', ignoring. Please see https://docs.getdbt.com/docs/build/sources#source-data-freshness for more information.",
-                )
-            )
-            project_freshness = None
-
-        source_freshness = source.freshness
-        if source_freshness and (target.path, source.name) not in self._deprecations:
-            deprecations.warn(
-                "property-moved-to-config-deprecation",
-                key="freshness",
-                file=target.path,
-                key_path=source.name,
-            )
-            self._deprecations.add((target.path, source.name))
-
-        source_config_freshness = FreshnessThreshold.from_dict(source.config.get("freshness", {}))
-
-        table_freshness = table.freshness
-        if table_freshness and (target.path, table.name) not in self._deprecations:
-            deprecations.warn(
-                "property-moved-to-config-deprecation",
-                key="freshness",
-                file=target.path,
-                key_path=table.name,
-            )
-            self._deprecations.add((target.path, table.name))
-
-        table_config_freshness = FreshnessThreshold.from_dict(table.config.get("freshness", {}))
-        freshness = merge_source_freshness(
-            project_freshness,
-            source_freshness,
-            source_config_freshness,
-            table_freshness,
-            table_config_freshness,
-        )
-
         quoting = source.quoting.merged(table.quoting)
         # path = block.path.original_file_path
         table_meta = table.meta or {}
@@ -249,7 +206,8 @@ def parse_source(self, target: UnpatchedSourceDefinition) -> SourceDefinition:
             loader=source.loader,
             loaded_at_field=loaded_at_field,
             loaded_at_query=loaded_at_query,
-            freshness=freshness,
+            # The setting to an empty freshness object is to maintain what we were previously doing if no freshenss was specified
+            freshness=config.freshness or FreshnessThreshold(),
             quoting=quoting,
             resource_type=NodeType.Source,
             fqn=target.fqn,
@@ -366,6 +324,19 @@ def _generate_source_config(self, target: UnpatchedSourceDefinition, rendered: b
         # it works while source configs can only include `enabled`.
         precedence_configs.update(target.table.config)
 
+        precedence_freshness = self.calculate_freshness_from_raw_target(target)
+        if precedence_freshness:
+            precedence_configs["freshness"] = precedence_freshness.to_dict()
+        elif precedence_freshness is None:
+            precedence_configs["freshness"] = None
+        else:
+            # this means that the user did not set a freshness threshold in the source schema file, as such
+            # there should be no freshness precedence
+            precedence_configs.pop("freshness", None)
+
+        # Because freshness is a "object" config, the freshness from the dbt_project.yml and the freshness
+        # from the schema file _won't_ get merged by this process. The result will be that the freshness will
+        # come from the schema file if provided, and if not, it'll fall back to the dbt_project.yml freshness.
         return generator.calculate_node_config(
             config_call_dict={},
             fqn=target.fqn,
@@ -418,6 +389,58 @@ def get_unused_msg(
                     )
         return unused_tables_formatted
 
+    def calculate_freshness_from_raw_target(
+        self,
+        target: UnpatchedSourceDefinition,
+    ) -> Optional[FreshnessThreshold]:
+        source: UnparsedSourceDefinition = target.source
+
+        source_freshness = source.freshness
+        if source_freshness and (target.path, source.name) not in self._deprecations:
+            deprecations.warn(
+                "property-moved-to-config-deprecation",
+                key="freshness",
+                file=target.path,
+                key_path=source.name,
+            )
+            self._deprecations.add((target.path, source.name))
+
+        source_config_freshness_raw: Optional[Dict] = source.config.get(
+            "freshness", {}
+        )  # Will only be None if the user explicitly set it to null
+        source_config_freshness: Optional[FreshnessThreshold] = (
+            FreshnessThreshold.from_dict(source_config_freshness_raw)
+            if source_config_freshness_raw is not None
+            else None
+        )
+
+        table: UnparsedSourceTableDefinition = target.table
+        table_freshness = table.freshness
+        if table_freshness and (target.path, table.name) not in self._deprecations:
+            deprecations.warn(
+                "property-moved-to-config-deprecation",
+                key="freshness",
+                file=target.path,
+                key_path=table.name,
+            )
+            self._deprecations.add((target.path, table.name))
+
+        table_config_freshness_raw: Optional[Dict] = table.config.get(
+            "freshness", {}
+        )  # Will only be None if the user explicitly set it to null
+        table_config_freshness: Optional[FreshnessThreshold] = (
+            FreshnessThreshold.from_dict(table_config_freshness_raw)
+            if table_config_freshness_raw is not None
+            else None
+        )
+
+        return merge_source_freshness(
+            source_freshness,
+            source_config_freshness,
+            table_freshness,
+            table_config_freshness,
+        )
+
 
 def merge_freshness_time_thresholds(
     base: Optional[Time], update: Optional[Time]
@@ -457,7 +480,7 @@ def merge_source_freshness(
             merged_freshness_obj.error_after = merged_error_after
             merged_freshness_obj.warn_after = merged_warn_after
             current_merged_value = merged_freshness_obj
-        elif base is None and update is not None:
+        elif base is None and bool(update):
             # If current_merged_value (base) is None, the update becomes the new value
             current_merged_value = update
         else:  # This covers cases where 'update' is None, or both 'base' and 'update' are None.

tests/functional/sources/fixtures.py

@@ -21,6 +21,7 @@
     loader: custom
     freshness:
       warn_after: {count: 18, period: hour}
+      error_after: {count: 24, period: hour}
     config:
       freshness: # default freshness, takes precedence over top-level key above
         warn_after: {count: 12, period: hour}
@@ -503,3 +504,33 @@
         loaded_at_query: "select {{current_timestamp()}}"
 
 """
+
+freshness_with_explicit_null_in_table_schema_yml = """version: 2
+sources:
+  - name: test_source
+    schema: "{{ var(env_var('DBT_TEST_SCHEMA_NAME_VARIABLE')) }}"
+    freshness:
+        warn_after:
+          count: 24
+          period: hour
+    quoting:
+      identifier: True
+    tables:
+      - name: source_a
+        loaded_at_field: "{{ var('test_loaded_at') | as_text }}"
+        config:
+          freshness: null
+"""
+
+freshness_with_explicit_null_in_source_schema_yml = """version: 2
+sources:
+  - name: test_source
+    schema: "{{ var(env_var('DBT_TEST_SCHEMA_NAME_VARIABLE')) }}"
+    config:
+      freshness: null
+    quoting:
+      identifier: True
+    tables:
+      - name: source_a
+        loaded_at_field: "{{ var('test_loaded_at') | as_text }}"
+"""

tests/functional/sources/test_source_freshness.py

@@ -20,6 +20,8 @@
     filtered_models_schema_yml,
     freshness_via_custom_sql_schema_yml,
     freshness_via_metadata_schema_yml,
+    freshness_with_explicit_null_in_source_schema_yml,
+    freshness_with_explicit_null_in_table_schema_yml,
     override_freshness_models_schema_yml,
 )
 
@@ -599,3 +601,23 @@ def test_source_freshness_custom_sql(self, project):
             "source_b": "warn",
             "source_c": "pass",
         }
+
+
+class TestSourceFreshnessExplicitNullInTable(SuccessfulSourceFreshnessTest):
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {"schema.yml": freshness_with_explicit_null_in_table_schema_yml}
+
+    def test_source_freshness_explicit_null_in_table(self, project):
+        result = self.run_dbt_with_vars(project, ["source", "freshness"], expect_pass=True)
+        assert {r.node.name: r.status for r in result} == {}
+
+
+class TestSourceFreshnessExplicitNullInSource(SuccessfulSourceFreshnessTest):
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {"schema.yml": freshness_with_explicit_null_in_source_schema_yml}
+
+    def test_source_freshness_explicit_null_in_source(self, project):
+        result = self.run_dbt_with_vars(project, ["source", "freshness"], expect_pass=True)
+        assert {r.node.name: r.status for r in result} == {}

tests/unit/parser/test_sources.py

@@ -0,0 +1,77 @@
+from typing import List, Optional
+
+import pytest
+
+from core.dbt.artifacts.resources.v1.components import FreshnessThreshold, Time
+from core.dbt.parser.sources import merge_source_freshness
+
+
+class TestMergeSourceFreshness:
+    @pytest.mark.parametrize(
+        "thresholds,expected_result",
+        [
+            ([None, None], None),
+            (
+                [
+                    FreshnessThreshold(
+                        warn_after=Time(count=1, period="hour"),
+                        error_after=Time(count=1, period="day"),
+                    ),
+                    None,
+                ],
+                None,
+            ),
+            (
+                [
+                    FreshnessThreshold(
+                        warn_after=Time(count=1, period="hour"),
+                        error_after=Time(count=1, period="day"),
+                    ),
+                    None,
+                    FreshnessThreshold(),
+                ],
+                None,
+            ),
+            (
+                [
+                    FreshnessThreshold(warn_after=Time(count=1, period="hour")),
+                    FreshnessThreshold(error_after=Time(count=1, period="day")),
+                ],
+                FreshnessThreshold(
+                    warn_after=Time(count=1, period="hour"),
+                    error_after=Time(count=1, period="day"),
+                ),
+            ),
+            (
+                [
+                    None,
+                    FreshnessThreshold(warn_after=Time(count=1, period="hour")),
+                    FreshnessThreshold(error_after=Time(count=1, period="day")),
+                ],
+                FreshnessThreshold(
+                    warn_after=Time(count=1, period="hour"),
+                    error_after=Time(count=1, period="day"),
+                ),
+            ),
+            (
+                [
+                    FreshnessThreshold(
+                        warn_after=Time(count=1, period="hour"),
+                        error_after=Time(count=1, period="day"),
+                    ),
+                    FreshnessThreshold(error_after=Time(count=48, period="hour")),
+                ],
+                FreshnessThreshold(
+                    warn_after=Time(count=1, period="hour"),
+                    error_after=Time(count=48, period="hour"),
+                ),
+            ),
+        ],
+    )
+    def test_merge_source_freshness(
+        self,
+        thresholds: List[Optional[FreshnessThreshold]],
+        expected_result: Optional[FreshnessThreshold],
+    ):
+        result = merge_source_freshness(*thresholds)
+        assert result == expected_result