Add deprecation warning for source_freshness_run_project_hooks

core/dbt/deprecations.py

@@ -128,6 +128,11 @@ class ResourceNamesWithSpacesDeprecation(DBTDeprecation):
     _event = "ResourceNamesWithSpacesDeprecation"
 
 
+class SourceFreshnessProjectHooksNotRun(DBTDeprecation):
+    _name = "source-freshness-project-hooks"
+    _event = "SourceFreshnessProjectHooksNotRun"
+
+
 def renamed_env_var(old_name: str, new_name: str):
     class EnvironmentVariableRenamed(DBTDeprecation):
         _name = f"environment-variable-renamed:{old_name}"
@@ -169,6 +174,7 @@ def warn(name, *args, **kwargs):
     ProjectFlagsMovedDeprecation(),
     PackageMaterializationOverrideDeprecation(),
     ResourceNamesWithSpacesDeprecation(),
+    SourceFreshnessProjectHooksNotRun(),
 ]
 
 deprecations: Dict[str, DBTDeprecation] = {d.name: d for d in deprecations_list}

core/dbt/events/core_types.proto

@@ -437,6 +437,14 @@ message PackageMaterializationOverrideDeprecationMsg {
     PackageMaterializationOverrideDeprecation data = 2;
 }
 
+// D017
+message SourceFreshnessProjectHooksNotRun {}
+
+message SourceFreshnessProjectHooksNotRunMsg {
+    CoreEventInfo info = 1;
+    SourceFreshnessProjectHooksNotRun data = 2;
+}
+
 // I065
 message DeprecatedModel {
     string model_name = 1;

core/dbt/events/types.py

@@ -456,6 +456,16 @@ def message(self) -> str:
         return line_wrap_message(warning_tag(description))
 
 
+class SourceFreshnessProjectHooksNotRun(WarnLevel):
+    def code(self) -> str:
+        return "D017"
+
+    def message(self) -> str:
+        description = "In a future version of dbt, the `source freshness` command will start running `on-run-start` and `on-run-end` hooks by default. Please refer to https://docs.getdbt.com/reference/global-configs/legacy-behaviors#source_freshness_run_project_hooks for detailed documentation and suggested workarounds."
+
+        return line_wrap_message(warning_tag(description))
+
+
 # =======================================================
 # I - Project parsing
 # =======================================================

core/dbt/task/freshness.py

@@ -18,6 +18,7 @@
 from dbt_common.exceptions import DbtRuntimeError, DbtInternalError
 from dbt_common.events.functions import fire_event
 from dbt_common.events.types import Note
+from dbt import deprecations
 from dbt.events.types import (
     FreshnessCheckComplete,
     LogStartLine,
@@ -240,9 +241,12 @@ def task_end_messages(self, results):
         fire_event(FreshnessCheckComplete())
 
     def get_hooks_by_type(self, hook_type: RunHookType) -> List[HookNode]:
+        hooks = super().get_hooks_by_type(hook_type)
         if self.args.source_freshness_run_project_hooks:
             return super().get_hooks_by_type(hook_type)
         else:
+            if hooks:
+                deprecations.warn("source-freshness-project-hooks")
             return []
 
     def populate_metadata_freshness_cache(self, adapter, selected_uids: AbstractSet[str]) -> None:

tests/functional/sources/test_source_freshness.py

@@ -8,6 +8,7 @@
 from dbt.artifacts.schemas.freshness import FreshnessResult
 from dbt.artifacts.schemas.results import FreshnessStatus
 from dbt.cli.main import dbtRunner
+from dbt import deprecations
 from tests.functional.sources.common_source_setup import BaseSourcesTest
 from tests.functional.sources.fixtures import (
     error_models_schema_yml,
@@ -414,6 +415,38 @@ def test_metadata_freshness_unsupported_error_when_run(self, project):
         assert "Could not compute freshness for source test_table" in freshness_result.message
 
 
+class TestSourceFreshnessProjectHooksNotRun(SuccessfulSourceFreshnessTest):
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {
+            "config-version": 2,
+            "on-run-start": ["{{ log('on-run-start hooks called') }}"],
+            "on-run-end": ["{{ log('on-run-end hooks called') }}"],
+            "flags": {
+                "source_freshness_run_project_hooks": False,
+            },
+        }
+
+    def test_hooks_do_run_for_source_freshness(
+        self,
+        project,
+    ):
+        deprecations.reset_deprecations()
+        assert deprecations.active_deprecations == set()
+        _, log_output = self.run_dbt_and_capture_with_vars(
+            project,
+            [
+                "source",
+                "freshness",
+            ],
+            expect_pass=False,
+        )
+        assert "on-run-start hooks called" not in log_output
+        assert "on-run-end hooks called" not in log_output
+        expected = {"source-freshness-project-hooks"}
+        assert expected == deprecations.active_deprecations
+
+
 class TestHooksInSourceFreshness(SuccessfulSourceFreshnessTest):
     @pytest.fixture(scope="class")
     def project_config_update(self):