Draft 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 %}