Implementation of metadata-based freshness

.changes/unreleased/Features-20231008-195410.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Add support for checking table-last-modified by metadata
+time: 2023-10-08T19:54:10.503476-04:00
+custom:
+  Author: peterallenwebb
+  Issue: "785"

dbt/adapters/snowflake/impl.py

@@ -3,7 +3,13 @@
 
 import agate
 
-from dbt.adapters.base.impl import AdapterConfig, ConstraintSupport  # type: ignore
+from dbt.adapters.base.impl import (
+    AdapterConfig,
+    Capability,
+    CapabilitySupport,
+    ConstraintSupport,
+    Support,
+)  # type: ignore
 from dbt.adapters.base.meta import available
 from dbt.adapters.sql import SQLAdapter  # type: ignore
 from dbt.adapters.sql.impl import (
@@ -49,6 +55,17 @@ class SnowflakeAdapter(SQLAdapter):
         ConstraintType.foreign_key: ConstraintSupport.NOT_ENFORCED,
     }
 
+    _capabilities: Dict[Capability, CapabilitySupport] = {
+        Capability.TableLastModifiedMetadata: CapabilitySupport(
+            capability=Capability.TableLastModifiedMetadata,
+            support=Support.Full,
+        ),
+        Capability.SchemaMetadataByRelations: CapabilitySupport(
+            capability=Capability.SchemaMetadataByRelations,
+            support=Support.NotImplemented,
+        ),
+    }
+
     @classmethod
     def date_function(cls):
         return "CURRENT_TIMESTAMP()"
@@ -259,3 +276,15 @@ def valid_incremental_strategies(self):
     def debug_query(self):
         """Override for DebugTask method"""
         self.execute("select 1 as id")
+
+    def capabilities(self) -> List[CapabilitySupport]:
+        return [c for c in self._capabilities.values()]
+
+    def capability_support(self, capability: Capability) -> CapabilitySupport:
+        if capability in self._capabilities:
+            return self._capabilities[capability]
+        else:
+            return CapabilitySupport(
+                capability=capability,
+                support=Support.Unknown,
+            )

dbt/include/snowflake/macros/metadata.sql

@@ -0,0 +1,19 @@
+{% macro snowflake__get_relation_last_modified(information_schema, relations) -%}
+
+  {%- call statement('last_modified', fetch_result=True) -%}
+        select table_schema as schema,
+               table_name as identifier,
+               last_altered as last_modified,
+               {{ current_timestamp() }} as snapshotted_at
+        from {{ information_schema }}.tables
+        where (
+          {%- for relation in relations -%}
+            (upper(table_schema) = upper('{{ relation.schema }}') and
+             upper(table_name) = upper('{{ relation.identifier }}')){%- if not loop.last %} or {% endif -%}
+          {%- endfor -%}
+        )
+  {%- endcall -%}
+
+  {{ return(load_result('last_modified')) }}
+
+{% endmacro %}

dev-requirements.txt

@@ -1,6 +1,6 @@
 # install latest changes in dbt-core
 # TODO: how to automate switching from develop to version branches?
-git+https://github.com/dbt-labs/dbt-core.git#egg=dbt-core&subdirectory=core
+git+https://github.com/dbt-labs/dbt-core.git@paw%2Fmetadata-freshness#egg=dbt-core&subdirectory=core
 git+https://github.com/dbt-labs/dbt-core.git#egg=dbt-tests-adapter&subdirectory=tests/adapter
 
 # if version 1.x or greater -> pin to major version

tests/functional/adapter/test_get_last_relation_modified.py

@@ -0,0 +1,46 @@
+import os
+import pytest
+
+from dbt.cli.main import dbtRunner
+
+
+freshness_via_metadata_schema_yml = """version: 2
+sources:
+  - name: test_source
+    freshness:
+      warn_after: {count: 10, period: hour}
+      error_after: {count: 1, period: day}
+    schema: "{{ env_var('DBT_TEST_SCHEMA_NAME_VARIABLE') }}"
+    tables:
+      - name: test_table
+"""
+
+
+class TestGetLastRelationModified:
+    @pytest.fixture(scope="class", autouse=True)
+    def set_env_vars(self, project):
+        os.environ["DBT_TEST_SCHEMA_NAME_VARIABLE"] = project.test_schema
+        yield
+        del os.environ["DBT_TEST_SCHEMA_NAME_VARIABLE"]
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {"schema.yml": freshness_via_metadata_schema_yml}
+
+    def test_get_last_relation_modified(self, project, set_env_vars):
+        project.run_sql(
+            f"create table {project.test_schema}.test_table (id integer autoincrement, name varchar(100) not null);"
+        )
+
+        warning_or_error = False
+
+        def probe(e):
+            nonlocal warning_or_error
+            if e.info.level in ["warning", "error"]:
+                warning_or_error = True
+
+        runner = dbtRunner(callbacks=[probe])
+        runner.invoke(["source", "freshness"])
+
+        # The 'source freshness' command should succeed without warnings or errors.
+        assert not warning_or_error