Ele 2204 bug timestamp column as sql expression doesnt work

integration_tests/tests/test_column_anomalies.py

@@ -107,3 +107,26 @@ def test_column_anomalies_with_where_parameter(test_id: str, dbt_project: DbtPro
         test_vars={"force_metrics_backfill": True},
     )
     assert test_result["status"] == "fail"
+
+
+def test_column_anomalies_with_timestamp_as_sql_expression(
+    test_id: str, dbt_project: DbtProject
+):
+    utc_today = datetime.utcnow().date()
+    data: List[Dict[str, Any]] = [
+        {
+            TIMESTAMP_COLUMN: cur_date.strftime(DATE_FORMAT),
+            "superhero": superhero,
+        }
+        for cur_date in generate_dates(base_date=utc_today - timedelta(1))
+        for superhero in ["Superman", "Batman"]
+    ]
+    test_args = {
+        "timestamp_column": "case when updated_at is not null then updated_at else updated_at end",
+        "column_anomalies": ["null_count"],
+    }
+
+    test_result = dbt_project.test(
+        test_id, DBT_TEST_NAME, test_args, data=data, test_column="superhero"
+    )
+    assert test_result["status"] == "pass"

integration_tests/tests/test_dimension_anomalies.py

@@ -23,6 +23,26 @@ def test_anomalyless_dimension_anomalies(test_id: str, dbt_project: DbtProject):
     assert test_result["status"] == "pass"
 
 
+def test_dimension_anomalies_with_timestamp_as_sql_expression(
+    test_id: str, dbt_project: DbtProject
+):
+    utc_today = datetime.utcnow().date()
+    data: List[Dict[str, Any]] = [
+        {
+            TIMESTAMP_COLUMN: cur_date.strftime(DATE_FORMAT),
+            "superhero": superhero,
+        }
+        for cur_date in generate_dates(base_date=utc_today - timedelta(1))
+        for superhero in ["Superman", "Spiderman"]
+    ]
+    test_args = {
+        "timestamp_column": "case when updated_at is not null then updated_at else updated_at end",
+        "dimensions": ["superhero"],
+    }
+    test_result = dbt_project.test(test_id, DBT_TEST_NAME, test_args, data=data)
+    assert test_result["status"] == "pass"
+
+
 def test_anomalous_dimension_anomalies(test_id: str, dbt_project: DbtProject):
     utc_today = datetime.utcnow().date()
     test_date, *training_dates = generate_dates(base_date=utc_today - timedelta(1))

integration_tests/tests/test_volume_anomalies.py

@@ -21,6 +21,21 @@ def test_anomalyless_table_volume_anomalies(test_id: str, dbt_project: DbtProjec
     assert test_result["status"] == "pass"
 
 
+def test_table_volume_anomalies_with_timestamp_as_sql_expression(
+    test_id: str, dbt_project: DbtProject
+):
+    utc_today = datetime.utcnow().date()
+    data = [
+        {TIMESTAMP_COLUMN: cur_date.strftime(DATE_FORMAT)}
+        for cur_date in generate_dates(base_date=utc_today)
+    ]
+    test_args = {
+        "timestamp_column": "case when updated_at is not null then updated_at else updated_at end"
+    }
+    test_result = dbt_project.test(test_id, DBT_TEST_NAME, test_args, data=data)
+    assert test_result["status"] == "pass"
+
+
 def test_full_drop_table_volume_anomalies(test_id: str, dbt_project: DbtProject):
     utc_today = datetime.utcnow().date()
     data = [

macros/edr/data_monitoring/monitors_query/table_monitoring_query.sql

@@ -73,9 +73,9 @@
 {% macro get_timestamp_table_query(monitored_table, metric_properties, timestamp_column, table_monitors, min_bucket_start, max_bucket_end, full_table_name_str) %}
     with partially_time_filtered_monitored_table as (
         select
-            {{ elementary.edr_cast_as_timestamp(timestamp_column) }} as {{ timestamp_column }}
+            {{ elementary.edr_cast_as_timestamp(timestamp_column) }} as monitored_table_timestamp_column
             {%- if metric_properties.timestamp_column and metric_properties.event_timestamp_column %}
-            , {{ elementary.edr_cast_as_timestamp(metric_properties.event_timestamp_column) }} as {{ metric_properties.event_timestamp_column }}
+            , {{ elementary.edr_cast_as_timestamp(metric_properties.event_timestamp_column) }} as monitored_table_event_timestamp_column
             {%- endif %}
         from {{ monitored_table }}
         -- Freshness metric calculated differences between consecutive buckets, thus the first diff
@@ -88,7 +88,7 @@
         select
             *
         from partially_time_filtered_monitored_table
-        where {{ timestamp_column }} >= {{ elementary.edr_cast_as_timestamp(min_bucket_start) }}
+        where monitored_table_timestamp_column >= {{ elementary.edr_cast_as_timestamp(min_bucket_start) }}
     ),
     buckets as (
         select edr_bucket_start, edr_bucket_end
@@ -99,13 +99,13 @@
 
     time_filtered_monitored_table as (
         select *,
-               {{ elementary.get_start_bucket_in_data(timestamp_column, min_bucket_start, metric_properties.time_bucket) }} as start_bucket_in_data
+               {{ elementary.get_start_bucket_in_data('monitored_table_timestamp_column', min_bucket_start, metric_properties.time_bucket) }} as start_bucket_in_data
         from monitored_table
         where
-            {{ timestamp_column }} >= (select min(edr_bucket_start) from buckets)
-            and {{ timestamp_column }} < (select max(edr_bucket_end) from buckets)
+            monitored_table_timestamp_column >= (select min(edr_bucket_start) from buckets)
+            and monitored_table_timestamp_column < (select max(edr_bucket_end) from buckets)
             {# To avoid adding buckets before the table first timestamp #}
-            and {{ timestamp_column }} >= (select min({{ timestamp_column }}) from monitored_table)
+            and monitored_table_timestamp_column >= (select min(monitored_table_timestamp_column) from monitored_table)
     ),
 
     metrics as (
@@ -205,14 +205,9 @@
 {% endmacro %}
 
 {% macro freshness_metric_query(metric_properties) %}
-    {%- set freshness_column = metric_properties.freshness_column %}
-    {%- if not freshness_column %}
-        {%- set freshness_column = metric_properties.timestamp_column %}
-    {%- endif %}
-
     -- get ordered consecutive update timestamps in the source data
     with unique_timestamps as (
-        select distinct {{ elementary.edr_cast_as_timestamp(freshness_column) }} as timestamp_val
+        select distinct monitored_table_timestamp_column as timestamp_val
         from partially_time_filtered_monitored_table
         order by 1
     ),
@@ -278,15 +273,13 @@
 {% endmacro %}
 
 {% macro event_freshness_metric_query(metric_properties) %}
-{% set event_timestamp_column = metric_properties.event_timestamp_column %}
-{% set update_timestamp_column = metric_properties.timestamp_column %}
     select
         edr_bucket_start,
         edr_bucket_end,
         {{ elementary.const_as_string('event_freshness') }} as metric_name,
-        {{ elementary.edr_cast_as_string('max({})'.format(event_timestamp_column)) }} as source_value,
+        {{ elementary.edr_cast_as_string('max({})'.format('monitored_table_event_timestamp_column')) }} as source_value,
         {{ 'coalesce(max({}), {})'.format(
-                elementary.timediff('second', elementary.edr_cast_as_timestamp(event_timestamp_column), elementary.edr_cast_as_timestamp(update_timestamp_column)),
+                elementary.timediff('second', elementary.edr_cast_as_timestamp('monitored_table_event_timestamp_column'), elementary.edr_cast_as_timestamp('monitored_table_timestamp_column')),
                 elementary.timediff('second', 'edr_bucket_start', 'edr_bucket_end')
             ) }} as metric_value
     from buckets left join time_filtered_monitored_table on (edr_bucket_start = start_bucket_in_data)