dremio support in anomaly tests - WIP

Author: haritamar

macros/edr/data_monitoring/anomaly_detection/get_anomaly_scores_query.sql

@@ -28,6 +28,7 @@
         {%- set bucket_seasonality_expr = elementary.const_as_text('no_seasonality') %}
     {%- endif %}
     {%- set detection_end = elementary.get_detection_end(test_configuration.detection_delay) %}
+    {%- set detection_end_expr = elementary.edr_cast_as_timestamp(elementary.edr_datetime_to_sql(detection_end)) %}
     {%- set min_bucket_start_expr = elementary.get_trunc_min_bucket_start_expr(detection_end, metric_properties, test_configuration.days_back) %}
 
     {# For timestamped tests, this will be the bucket start, and for non-timestamped tests it will be the
@@ -39,9 +40,9 @@
             with buckets as (
                 select edr_bucket_start, edr_bucket_end
                 from ({{ elementary.complete_buckets_cte(metric_properties, min_bucket_start_expr,
-                                                         elementary.edr_quote(detection_end)) }}) results
-                where edr_bucket_start >= {{ elementary.edr_cast_as_timestamp(min_bucket_start_expr) }}
-                  and edr_bucket_end <= {{ elementary.edr_cast_as_timestamp(elementary.edr_quote(detection_end)) }}
+                                                         detection_end_expr) }}) results
+                where edr_bucket_start >= {{ min_bucket_start_expr }}
+                  and edr_bucket_end <= {{ detection_end_expr }}
             ),
         {% else %}
             with

macros/edr/data_monitoring/data_monitors_configuration/get_buckets_configuration.sql

@@ -1,10 +1,10 @@
 {% macro get_detection_end(detection_delay) %}
     {% if not detection_delay %}
-        {% do return(elementary.get_run_started_at()) %}
+        {% do return(elementary.get_run_started_at().replace(microsecond=0)) %}
     {% endif %}
 
     {%- set kwargs = {detection_delay.period+'s': detection_delay.count} %}
-    {%- set detection_end = elementary.get_run_started_at() - modules.datetime.timedelta(**kwargs) %}
+    {%- set detection_end = elementary.get_run_started_at().replace(microsecond=0) - modules.datetime.timedelta(**kwargs) %}
     {% do return(detection_end) %}
 {% endmacro %}
 
@@ -105,8 +105,8 @@
         {%- set buckets = elementary.agate_to_dicts(elementary.run_query(incremental_bucket_times_query))[0] %}
     {% endif %}
     {%- if buckets %}
-        {%- set min_bucket_start = elementary.edr_quote(buckets.get('min_bucket_start')) %}
-        {%- set max_bucket_end = elementary.edr_quote(buckets.get('max_bucket_end')) %}
+        {%- set min_bucket_start = elementary.edr_datetime_to_sql(buckets.get('min_bucket_start')) %}
+        {%- set max_bucket_end = elementary.edr_datetime_to_sql(buckets.get('max_bucket_end')) %}
         {{ return([min_bucket_start, max_bucket_end]) }}
     {%- else %}
         {{ exceptions.raise_compiler_error("Failed to calc test buckets min and max") }}

macros/edr/system/system_utils/buckets_cte.sql

@@ -144,3 +144,21 @@
     {{ return(complete_buckets_cte) }}
 {% endmacro %}
 
+{% macro dremio__complete_buckets_cte(time_bucket, bucket_end_expr, min_bucket_start_expr, max_bucket_end_expr) %}
+    {%- set complete_buckets_cte %}
+        with integers as (
+            select (row_number() over (order by t1.val, t2.val, t3.val, t4.val)) - 1 as num
+            from (values (1), (2), (3), (4), (5), (6), (7), (8), (9), (10)) t1(val)
+            cross join (values (1), (2), (3), (4), (5), (6), (7), (8), (9), (10)) t2(val)
+            cross join (values (1), (2), (3), (4), (5), (6), (7), (8), (9), (10)) t3(val)
+            cross join (values (1), (2), (3), (4), (5), (6), (7), (8), (9), (10)) t4(val)
+        )
+        select
+            {{ elementary.edr_timeadd(time_bucket.period, 'num * ' ~ time_bucket.count, min_bucket_start_expr) }} as edr_bucket_start,
+            {{ elementary.edr_timeadd(time_bucket.period, '(num + 1) * ' ~ time_bucket.count, min_bucket_start_expr) }} as edr_bucket_end
+        from integers
+        where {{ elementary.edr_timeadd(time_bucket.period, '(num + 1) * ' ~ time_bucket.count, min_bucket_start_expr) }} <= {{ max_bucket_end_expr }}
+    {%- endset %}
+    {{ return(complete_buckets_cte) }}
+{% endmacro %}
+  

macros/utils/cross_db_utils/current_timestamp.sql

@@ -69,3 +69,12 @@
 {% macro trino__edr_current_timestamp_in_utc() -%}
     cast(current_timestamp at time zone 'UTC' as timestamp(6))
 {%- endmacro -%}
+
+{% macro dremio__edr_current_timestamp() -%}
+    CURRENT_TIMESTAMP()
+{%- endmacro -%}
+
+{% macro dremio__edr_current_timestamp_in_utc() -%}
+    {# Dremio timezones are always in UTC #}
+    CURRENT_TIMESTAMP()
+{%- endmacro -%}

macros/utils/cross_db_utils/datediff.sql

@@ -156,3 +156,18 @@
     {% endif %}
     {{ return(macro(elementary.edr_cast_as_timestamp(first_date), elementary.edr_cast_as_timestamp(second_date), date_part)) }}
 {% endmacro %}
+
+{% macro dremio__edr_datediff(first_date, second_date, date_part) %}
+    {% set macro = dbt.datediff or dbt_utils.datediff %}
+    {% if not macro %}
+        {{ exceptions.raise_compiler_error("Did not find a `datediff` macro.") }}
+    {% endif %}
+
+    {% set sql = macro(elementary.edr_cast_as_timestamp(first_date), elementary.edr_cast_as_timestamp(second_date), date_part) %}
+
+    {# Hack - dbt-dremio implements this macro as a select statement (which seems to be necessary), but in order 
+       for it to really work we wrap it in parentheses and remove ; if it is there #}
+    {% set sql = '(' ~ sql.strip().replace(';', '') ~ ')' %}
+
+    {% do return(sql) %}
+{% endmacro %}

macros/utils/cross_db_utils/datetime_to_sql.sql

@@ -0,0 +1,14 @@
+{% macro edr_datetime_to_sql(dt) %}
+  {% do return(adapter.dispatch("edr_datetime_to_sql", "elementary")(dt)) %}
+{% endmacro %}
+
+{% macro default__edr_datetime_to_sql(dt) %}
+  {% do return(elementary.edr_quote(dt)) %}
+{% endmacro %}
+
+{% macro dremio__edr_datetime_to_sql(dt) %}
+  {% if dt is string %}
+    {% set dt = modules.datetime.datetime.fromisoformat(dt) %}
+  {% endif %}
+  {% do return(elementary.edr_quote(dt.strftime(elementary.get_time_format()))) %}
+{% endmacro %}

macros/utils/cross_db_utils/timeadd.sql

@@ -33,3 +33,7 @@
 {% macro trino__edr_timeadd(date_part, number, timestamp_expression) %}
     date_add('{{ date_part }}', {{ elementary.edr_cast_as_int(number) }}, {{ elementary.edr_cast_as_timestamp(timestamp_expression) }})
 {% endmacro %}
+
+{% macro dremio__edr_timeadd(date_part, number, timestamp_expression) %}
+    timestampadd({{ date_part }}, {{ elementary.edr_cast_as_int(number) }}, {{ elementary.edr_cast_as_timestamp(timestamp_expression) }})
+{% endmacro %}

macros/utils/data_types/cast_column.sql

@@ -29,6 +29,10 @@
     )
 {%- endmacro -%}
 
+{%- macro dremio__edr_cast_as_timestamp(timestamp_field) -%}
+    cast({{ timestamp_field }} as {{ elementary.edr_type_timestamp() }})
+{%- endmacro -%}
+
 {%- macro edr_cast_as_float(column) -%}
     cast({{ column }} as {{ elementary.edr_type_float() }})
 {%- endmacro -%}
@@ -85,6 +89,10 @@
     )
 {%- endmacro -%}
 
+{%- macro dremio__edr_cast_as_date(timestamp_field) -%}
+    cast({{ timestamp_field }} as {{ elementary.edr_type_date() }})
+{%- endmacro -%}
+
 
 {%- macro const_as_text(string) -%}
     {{ return(adapter.dispatch('const_as_text', 'elementary')(string)) }}

macros/utils/data_types/data_type.sql

@@ -156,3 +156,7 @@
 {% macro trino__edr_type_timestamp() %}
     timestamp(6)
 {% endmacro %}
+
+{% macro dremio__edr_type_timestamp() %}
+    timestamp
+{% endmacro %}