Back compat for previous return type of collect_freshness

.changes/unreleased/Fixes-20230506-173315.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: Back-compat for previous return type of 'collect_freshness' macro
+time: 2023-05-06T17:33:15.104953+02:00
+custom:
+  Author: jtcohen6
+  Issue: "7489"

core/dbt/adapters/base/impl.py

@@ -71,7 +71,7 @@
 from dbt.adapters.base import Column as BaseColumn
 from dbt.adapters.base import Credentials
 from dbt.adapters.cache import RelationsCache, _make_ref_key_dict
-
+from dbt import deprecations
 
 GET_CATALOG_MACRO_NAME = "get_catalog"
 FRESHNESS_MACRO_NAME = "collect_freshness"
@@ -1104,7 +1104,7 @@ def calculate_freshness(
         loaded_at_field: str,
         filter: Optional[str],
         manifest: Optional[Manifest] = None,
-    ) -> Tuple[AdapterResponse, Dict[str, Any]]:
+    ) -> Tuple[Optional[AdapterResponse], Dict[str, Any]]:
         """Calculate the freshness of sources in dbt, and return it"""
         kwargs: Dict[str, Any] = {
             "source": source,
@@ -1113,8 +1113,18 @@ def calculate_freshness(
         }
 
         # run the macro
+        # in older versions of dbt-core, the 'collect_freshness' macro returned the table of results directly
+        # starting in v1.5, by default, we return both the table and the adapter response (metadata about the query)
+        result: Union[
+            AttrDict, agate.Table  # current: contains AdapterResponse + agate.Table  # previous
+        ]
         result = self.execute_macro(FRESHNESS_MACRO_NAME, kwargs=kwargs, manifest=manifest)
-        adapter_response, table = result.response, result.table  # type: ignore[attr-defined]
+        if isinstance(result, agate.Table):
+            deprecations.warn("collect-freshness-return-signature")
+            adapter_response = None
+            table = result
+        else:
+            adapter_response, table = result.response, result.table  # type: ignore[attr-defined]
         # now we have a 1-row table of the maximum `loaded_at_field` value and
         # the current time according to the db.
         if len(table) != 1 or len(table[0]) != 2:

core/dbt/deprecations.py

@@ -91,6 +91,11 @@ class ConfigTargetPathDeprecation(DBTDeprecation):
     _event = "ConfigTargetPathDeprecation"
 
 
+class CollectFreshnessReturnSignature(DBTDeprecation):
+    _name = "collect-freshness-return-signature"
+    _event = "CollectFreshnessReturnSignature"
+
+
 def renamed_env_var(old_name: str, new_name: str):
     class EnvironmentVariableRenamed(DBTDeprecation):
         _name = f"environment-variable-renamed:{old_name}"
@@ -128,6 +133,7 @@ def warn(name, *args, **kwargs):
     ExposureNameDeprecation(),
     ConfigLogPathDeprecation(),
     ConfigTargetPathDeprecation(),
+    CollectFreshnessReturnSignature(),
 ]
 
 deprecations: Dict[str, DBTDeprecation] = {d.name: d for d in deprecations_list}

core/dbt/events/types.proto

@@ -385,6 +385,15 @@ message ConfigTargetPathDeprecationMsg {
     ConfigTargetPathDeprecation data = 2;
 }
 
+// D012
+message CollectFreshnessReturnSignature {
+}
+
+message CollectFreshnessReturnSignatureMsg {
+    EventInfo info = 1;
+    CollectFreshnessReturnSignature data = 2;
+}
+
 // E - DB Adapter
 
 // E001

core/dbt/events/types.py

@@ -407,6 +407,19 @@ def message(self):
         return line_wrap_message(warning_tag(f"Deprecated functionality\n\n{description}"))
 
 
+class CollectFreshnessReturnSignature(WarnLevel):
+    def code(self):
+        return "D012"
+
+    def message(self):
+        description = (
+            "The 'collect_freshness' macro signature has changed to return the full "
+            "query result, rather than just a table of values. See the v1.5 migration guide "
+            "for details on how to update your custom macro: https://docs.getdbt.com/guides/migration/versions/upgrading-to-v1.5"
+        )
+        return line_wrap_message(warning_tag(f"Deprecated functionality\n\n{description}"))
+
+
 # =======================================================
 # E - DB Adapter
 # =======================================================

core/dbt/task/freshness.py

@@ -117,14 +117,18 @@ def execute(self, compiled_node, manifest):
 
         status = compiled_node.freshness.status(freshness["age"])
 
+        # adapter_response was not returned in previous versions, so this will be {}
+        if adapter_response:
+            adapter_response = adapter_response.to_dict(omit_none=True)
+
         return SourceFreshnessResult(
             node=compiled_node,
             status=status,
             thread_id=threading.current_thread().name,
             timing=[],
             execution_time=0,
             message=None,
-            adapter_response=adapter_response.to_dict(omit_none=True),
+            adapter_response=adapter_response or {},
             failures=None,
             **freshness,
         )

tests/functional/sources/fixtures.py

@@ -439,3 +439,20 @@
             enabled: True and False
       - name: other_test_table
 """
+
+
+collect_freshness_macro_override_previous_return_signature = """
+{% macro collect_freshness(source, loaded_at_field, filter) %}
+  {{ log("This one is mine", info = true) }}
+  {% call statement('collect_freshness', fetch_result=True, auto_begin=False) -%}
+    select
+      max({{ loaded_at_field }}) as max_loaded_at,
+      {{ current_timestamp() }} as snapshotted_at
+    from {{ source }}
+    {% if filter %}
+    where {{ filter }}
+    {% endif %}
+  {% endcall %}
+  {{ return(load_result('collect_freshness').table) }}
+{% endmacro %}
+"""

tests/functional/sources/test_source_freshness.py

@@ -10,8 +10,10 @@
     error_models_model_sql,
     filtered_models_schema_yml,
     override_freshness_models_schema_yml,
+    collect_freshness_macro_override_previous_return_signature,
 )
 from dbt.tests.util import AnyStringWith, AnyFloat
+from dbt import deprecations
 
 
 class SuccessfulSourceFreshnessTest(BaseSourcesTest):
@@ -353,3 +355,23 @@ def test_override_source_freshness(self, project):
             "filter": None,
         }
         assert result_source_d["criteria"] == expected
+
+
+class TestSourceFreshnessMacroOverride(SuccessfulSourceFreshnessTest):
+    @pytest.fixture(scope="class")
+    def macros(self):
+        return {
+            "collect_freshness.sql": collect_freshness_macro_override_previous_return_signature
+        }
+
+    def test_source_freshness(self, project):
+        # ensure that the deprecation warning is raised
+        deprecations.reset_deprecations()
+        assert deprecations.active_deprecations == set()
+        self.run_dbt_with_vars(
+            project,
+            ["source", "freshness"],
+            expect_pass=False,
+        )
+        expected = {"collect-freshness-return-signature"}
+        assert expected == deprecations.active_deprecations

tests/unit/test_events.py

@@ -133,6 +133,7 @@ def test_event_codes(self):
     types.EnvironmentVariableRenamed(old_name="", new_name=""),
     types.ConfigLogPathDeprecation(deprecated_path=""),
     types.ConfigTargetPathDeprecation(deprecated_path=""),
+    types.CollectFreshnessReturnSignature(),
     # E - DB Adapter ======================
     types.AdapterEventDebug(),
     types.AdapterEventInfo(),