[Backport 1.8.latest] Warn if timestamp updated_at field uses incompatible timestamp (#10352)

.changes/unreleased/Features-20240621-141635.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Warning message for snapshot timestamp data types
+time: 2024-06-21T14:16:35.717637-04:00
+custom:
+  Author: gshank
+  Issue: "10234"

core/dbt/context/exceptions_jinja.py

@@ -2,7 +2,7 @@
 from typing import NoReturn
 
 from dbt_common.events.functions import warn_or_error
-from dbt.events.types import JinjaLogWarning
+from dbt.events.types import JinjaLogWarning, SnapshotTimestampWarning
 
 from dbt_common.exceptions import (
     DbtRuntimeError,
@@ -117,6 +117,17 @@ def raise_fail_fast_error(msg, node=None) -> NoReturn:
     raise FailFastError(msg, node=node)
 
 
+def warn_snapshot_timestamp_data_types(
+    snapshot_time_data_type: str, updated_at_data_type: str
+) -> None:
+    warn_or_error(
+        SnapshotTimestampWarning(
+            snapshot_time_data_type=snapshot_time_data_type,
+            updated_at_data_type=updated_at_data_type,
+        )
+    )
+
+
 # Update this when a new function should be added to the
 # dbt context's `exceptions` key!
 CONTEXT_EXPORTS = {
@@ -142,6 +153,7 @@ def raise_fail_fast_error(msg, node=None) -> NoReturn:
         raise_contract_error,
         column_type_missing,
         raise_fail_fast_error,
+        warn_snapshot_timestamp_data_types,
     ]
 }
 

core/dbt/events/core_types.proto

@@ -1610,6 +1610,17 @@ message CompiledNodeMsg {
     CompiledNode data = 2;
 }
 
+// Q043
+message SnapshotTimestampWarning {
+    string snapshot_time_data_type = 1;
+    string updated_at_data_type = 2;
+}
+
+message SnapshotTimestampWarningMsg {
+    CoreEventInfo info = 1;
+    SnapshotTimestampWarning data = 2;
+}
+
 // W - Node testing
 
 // Skipped W001

core/dbt/events/types.py

@@ -1607,6 +1607,18 @@ def message(self) -> str:
                 return f"Compiled node '{self.node_name}' is:\n{self.compiled}"
 
 
+class SnapshotTimestampWarning(WarnLevel):
+    def code(self) -> str:
+        return "Q043"
+
+    def message(self) -> str:
+        return (
+            f"Data type of snapshot table timestamp columns ({self.snapshot_time_data_type}) "
+            f"doesn't match derived column 'updated_at' ({self.updated_at_data_type}). "
+            "Please update snapshot config 'updated_at'."
+        )
+
+
 # =======================================================
 # W - Node testing
 # =======================================================

tests/functional/simple_snapshot/test_snapshot_timestamps.py

@@ -0,0 +1,72 @@
+import pytest
+
+from dbt.tests.util import run_dbt, run_dbt_and_capture
+
+create_source_sql = """
+create table {database}.{schema}.source_users (
+    id INTEGER,
+    first_name VARCHAR(50),
+    last_name VARCHAR(50),
+    email VARCHAR(50),
+    gender VARCHAR(50),
+    ip_address VARCHAR(20),
+    updated_time TIMESTAMP WITH TIME ZONE
+);
+insert into {database}.{schema}.source_users (id, first_name, last_name, email, gender, ip_address, updated_time) values
+(1, 'Judith', 'Kennedy', '(not provided)', 'Female', '54.60.24.128', '2015-12-24 12:19:28'),
+(2, 'Arthur', 'Kelly', '(not provided)', 'Male', '62.56.24.215', '2015-10-28 16:22:15'),
+(3, 'Rachel', 'Moreno', '[email protected]', 'Female', '31.222.249.23', '2016-04-05 02:05:30');
+"""
+
+model_users_sql = """
+select * from {{ source('test_source', 'source_users') }}
+"""
+
+snapshot_sql = """
+{% snapshot users_snapshot %}
+
+select * from {{ ref('users') }}
+
+{% endsnapshot %}
+"""
+
+source_schema_yml = """
+sources:
+  - name: test_source
+    loader: custom
+    schema: "{{ target.schema }}"
+    tables:
+      - name: source_users
+        loaded_at_field: updated_time
+"""
+
+snapshot_schema_yml = """
+snapshots:
+  - name: users_snapshot
+    config:
+      target_schema: "{{ target.schema }}"
+      strategy: timestamp
+      unique_key: id
+      updated_at: updated_time
+"""
+
+
+class TestSnapshotConfig:
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "users.sql": model_users_sql,
+            "source_schema.yml": source_schema_yml,
+            "snapshot_schema.yml": snapshot_schema_yml,
+        }
+
+    @pytest.fixture(scope="class")
+    def snapshots(self):
+        return {"snapshot.sql": snapshot_sql}
+
+    def test_timestamp_snapshot(self, project):
+        project.run_sql(create_source_sql)
+        run_dbt(["run"])
+        results, log_output = run_dbt_and_capture(["snapshot"])
+        assert len(results) == 1
+        assert "Please update snapshot config" in log_output

tests/unit/test_events.py

@@ -416,6 +416,9 @@ def test_event_codes(self):
     core_types.CompiledNode(
         node_name="", compiled="", is_inline=True, unique_id="model.test.my_model"
     ),
+    core_types.SnapshotTimestampWarning(
+        snapshot_time_data_type="DATETIME", updated_at_data_type="DATETIMEZ"
+    ),
     # W - Node testing ======================
     core_types.CatchableExceptionOnRun(exc=""),
     core_types.InternalErrorOnRun(build_path="", exc=""),