tests almost work

Author: Michael Myaskovsky

integration_tests/dbt_project/macros/clear_env.sql

@@ -1,11 +1,20 @@
 {% macro clear_env() %}
-    {% do elementary_tests.edr_drop_schema(elementary.target_database(), generate_schema_name()) %}
     {% set database_name, schema_name = elementary.get_package_database_and_schema('elementary') %}
     {% do elementary_tests.edr_drop_schema(database_name, schema_name) %}
+    {% do elementary_tests.edr_drop_schema(elementary.target_database(), generate_schema_name()) %}
 {% endmacro %}
 
 {% macro edr_drop_schema(database_name, schema_name) %}
+    {% do return(adapter.dispatch('edr_drop_schema', 'elementary_tests')(database_name, schema_name)) %}
+{% endmacro %}
+
+{% macro default__edr_drop_schema(database_name, schema_name) %}
     {% set schema_relation = api.Relation.create(database=database_name, schema=schema_name) %}
     {% do dbt.drop_schema(schema_relation) %}
     {% do adapter.commit() %}
 {% endmacro %}
+
+{% macro clickhouse__edr_drop_schema(database_name, schema_name) %}
+    {% do run_query("DROP DATABASE IF EXISTS " ~ schema_name) %}
+    {% do adapter.commit() %}
+{% endmacro %}

integration_tests/tests/dbt_project.py

@@ -151,7 +151,6 @@ def test(
     ) -> Union[Dict[str, Any], List[Dict[str, Any]]]:
         if columns and test_column:
             raise ValueError("You can't specify both 'columns' and 'test_column'.")
-
         test_vars = test_vars or {}
         test_vars["elementary_enabled"] = elementary_enabled
 

macros/edr/data_monitoring/anomaly_detection/get_anomaly_scores_query.sql

@@ -223,7 +223,6 @@
         )
 
         select * from anomaly_scores
-
     {% endset %}
     {{ return(anomaly_scores_query) }}
 {% endmacro %}

macros/edr/data_monitoring/data_monitors_configuration/get_buckets_configuration.sql

@@ -14,7 +14,7 @@
     {{ return(trunc_min_bucket_start_expr) }}
 {% endmacro %}
 
-{# This macro can't be used without truncating to full buckets #}
+{# This macro cant be used without truncating to full buckets #}
 {% macro get_backfill_bucket_start(detection_end, backfill_days) %}
     {% do return((detection_end - modules.datetime.timedelta(backfill_days)).strftime("%Y-%m-%d 00:00:00")) %}
 {% endmacro %}
@@ -38,7 +38,6 @@
     {%- else %}
         {%- set data_monitoring_metrics_relation = elementary.get_elementary_relation('data_monitoring_metrics') %}
     {%- endif %}
-
     {%- set regular_bucket_times_query %}
         with bucket_times as (
             select
@@ -100,11 +99,11 @@
 
     {# We assume we should also cosider sources as incremental #}
     {% if force_metrics_backfill or not (elementary.is_incremental_model(elementary.get_model_graph_node(model_relation), source_included=true) or unit_test) %}
+ 
         {%- set buckets = elementary.agate_to_dicts(elementary.run_query(regular_bucket_times_query))[0] %}
     {%- else %}
         {%- 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')) %}

macros/edr/data_monitoring/monitors_query/column_monitoring_query.sql

@@ -13,7 +13,6 @@
         {% do metric_name_to_type.update({metric.name: metric.type}) %}
     {% endfor %}
 
-
     with monitored_table as (
         select * from {{ monitored_table }}
             {% if metric_properties.where_expression %} where {{ metric_properties.where_expression }} {% endif %}
@@ -26,7 +25,6 @@
              where edr_bucket_start >= {{ elementary.edr_cast_as_timestamp(min_bucket_start) }}
                and edr_bucket_end <= {{ elementary.edr_cast_as_timestamp(max_bucket_end) }}
          ),
-
          filtered_monitored_table as (
             select {{ column_obj.quoted }},
                    {%- if dimensions -%} {{ elementary.select_dimensions_columns(dimensions, "dimension") }}, {%- endif -%}
@@ -96,39 +94,10 @@
     ),
 
     column_metrics_unpivot as (
-
-        {%- if column_metrics %}
-            {% for metric_name, metric_type in metric_name_to_type.items() %}
-                select
-                    {{ elementary.const_as_string(column_obj.name) }} as edr_column_name,
-                    bucket_start,
-                    bucket_end,
-                    {% if timestamp_column %}
-                        {{ elementary.timediff("hour", "bucket_start", "bucket_end") }} as bucket_duration_hours,
-                    {% else %}
-                        {{ elementary.null_int() }} as bucket_duration_hours,
-                    {% endif %}
-                    {% if dimensions | length > 0 %}
-                      {{ elementary.const_as_string(elementary.join_list(dimensions, separator='; ')) }} as dimension,
-                      {{ elementary.list_concat_with_separator(prefixed_dimensions, separator='; ') }} as dimension_value,
-                    {% else %}
-                      {{ elementary.null_string() }} as dimension,
-                      {{ elementary.null_string() }} as dimension_value,
-                    {% endif %}
-                    {{ elementary.edr_cast_as_float(metric_type) }} as metric_value,
-                    {{ elementary.edr_cast_as_string(elementary.edr_quote(metric_name)) }} as metric_name,
-                    {{ elementary.edr_cast_as_string(elementary.edr_quote(metric_type)) }} as metric_type
-                from column_metrics where {{ metric_type }} is not null
-                {% if not loop.last %} union all {% endif %}
-            {%- endfor %}
-        {%- else %}
-            {{ elementary.empty_table([('edr_column_name','string'),('bucket_start','timestamp'),('bucket_end','timestamp'),('bucket_duration_hours','int'),('dimension','string'),('dimension_value','string'),('metric_name','string'),('metric_type','string'),('metric_value','float')]) }}
-        {%- endif %}
-
+        {{ elementary.get_column_metrics_unpivot_query(column_metrics, column_obj, metric_name_to_type, timestamp_column, dimensions) }}
     ),
 
     metrics_final as (
-
         select
             {{ elementary.edr_cast_as_string(full_table_name_str) }} as full_table_name,
             edr_column_name as column_name,
@@ -143,7 +112,6 @@
             dimension_value,
             {{elementary.dict_to_quoted_json(metric_properties) }} as metric_properties
         from column_metrics_unpivot
-
     )
 
     select
@@ -174,6 +142,70 @@
 
 {% endmacro %}
 
+{% macro get_column_metrics_unpivot_query(column_metrics, column_obj, metric_name_to_type, timestamp_column, dimensions) %}
+    {{ return(adapter.dispatch('get_column_metrics_unpivot_query', 'elementary')(column_metrics, column_obj, metric_name_to_type, timestamp_column, dimensions)) }}
+{% endmacro %}
+
+{% macro default__get_column_metrics_unpivot_query(column_metrics, column_obj, metric_name_to_type, timestamp_column, dimensions) %}
+    {%- if column_metrics %}
+        {% for metric_name, metric_type in metric_name_to_type.items() %}
+            select
+                {{ elementary.const_as_string(column_obj.name) }} as edr_column_name,
+                bucket_start,
+                bucket_end,
+                {% if timestamp_column %}
+                    {{ elementary.timediff("hour", "bucket_start", "bucket_end") }} as bucket_duration_hours,
+                {% else %}
+                    {{ elementary.null_int() }} as bucket_duration_hours,
+                {% endif %}
+                {% if dimensions | length > 0 %}
+                  {{ elementary.const_as_string(elementary.join_list(dimensions, separator='; ')) }} as dimension,
+                  {{ elementary.list_concat_with_separator(prefixed_dimensions, separator='; ') }} as dimension_value,
+                {% else %}
+                  {{ elementary.null_string() }} as dimension,
+                  {{ elementary.null_string() }} as dimension_value,
+                {% endif %}
+                {{ elementary.edr_cast_as_float(metric_type) }} as metric_value,
+                {{ elementary.edr_cast_as_string(elementary.edr_quote(metric_name)) }} as metric_name,
+                {{ elementary.edr_cast_as_string(elementary.edr_quote(metric_type)) }} as metric_type
+            from column_metrics where {{ metric_type }} is not null
+            {% if not loop.last %} union all {% endif %}
+        {%- endfor %}
+    {%- else %}
+        {{ elementary.empty_table([('edr_column_name','string'),('bucket_start','timestamp'),('bucket_end','timestamp'),('bucket_duration_hours','int'),('dimension','string'),('dimension_value','string'),('metric_name','string'),('metric_type','string'),('metric_value','float')]) }}
+    {%- endif %}
+{% endmacro %}
+
+{% macro clickhouse__get_column_metrics_unpivot_query(column_metrics, column_obj, metric_name_to_type, timestamp_column, dimensions) %}
+    {%- if column_metrics %}
+        {% for metric_name, metric_type in metric_name_to_type.items() %}
+            select
+                {{ elementary.const_as_string(column_obj.name) }} as edr_column_name,
+                bucket_start,
+                bucket_end,
+                {% if timestamp_column %}
+                    {{ elementary.timediff("hour", "bucket_start", "bucket_end") }} as bucket_duration_hours,
+                {% else %}
+                    {{ elementary.null_int() }} as bucket_duration_hours,
+                {% endif %}
+                {% if dimensions | length > 0 %}
+                  {{ elementary.const_as_string(elementary.join_list(dimensions, separator='; ')) }} as dimension,
+                  {{ elementary.list_concat_with_separator(prefixed_dimensions, separator='; ') }} as dimension_value,
+                {% else %}
+                  CAST(NULL AS Nullable(String)) as dimension,
+                  CAST(NULL AS Nullable(String)) as dimension_value,
+                {% endif %}
+                {{ elementary.edr_cast_as_float(metric_type) }} as metric_value,
+                cast({{ elementary.edr_quote(metric_name) }} as Nullable({{ elementary.edr_type_string() }})) as metric_name, 
+                cast({{ elementary.edr_quote(metric_type) }} as Nullable({{ elementary.edr_type_string() }})) as metric_type
+            from column_metrics where {{ metric_type }} is not null
+            {% if not loop.last %} union all {% endif %}
+        {%- endfor %}
+    {%- else %}
+        {{ elementary.empty_table([('edr_column_name','string'),('bucket_start','timestamp'),('bucket_end','timestamp'),('bucket_duration_hours','int'),('dimension','string'),('dimension_value','string'),('metric_name','string'),('metric_type','string'),('metric_value','float')]) }}
+    {%- endif %}
+{% endmacro %}
+
 {% macro select_dimensions_columns(dimension_columns, as_prefix="") %}
   {% set select_statements %}
     {%- for column in dimension_columns -%}

macros/edr/system/system_utils/buckets_cte.sql

@@ -11,6 +11,16 @@
     {{ return('') }}
 {% endmacro %}
 
+{% macro clickhouse__complete_buckets_cte(time_bucket, bucket_end_expr, min_bucket_start_expr, max_bucket_end_expr) %}
+    with numbers as (
+        select arrayJoin(range(0, toUInt32({{ elementary.edr_datediff(min_bucket_start_expr, max_bucket_end_expr, time_bucket.period) }}))) as n
+    )
+    select
+        {{ elementary.edr_timeadd(time_bucket.period, 'n', min_bucket_start_expr) }} as edr_bucket_start,
+        {{ elementary.edr_timeadd(time_bucket.period, 'n + 1', min_bucket_start_expr) }} as edr_bucket_end
+    from numbers
+    where {{ elementary.edr_timeadd(time_bucket.period, 'n + 1', min_bucket_start_expr) }} <= {{ max_bucket_end_expr }}
+{% endmacro %}
 
 {% macro spark__complete_buckets_cte(time_bucket, bucket_end_expr, min_bucket_start_expr, max_bucket_end_expr) %}
     {%- set complete_buckets_cte %}
@@ -133,3 +143,4 @@
     {%- endset %}
     {{ return(complete_buckets_cte) }}
 {% endmacro %}
+

macros/edr/tests/test_all_columns_anomalies.sql

@@ -5,7 +5,6 @@
         {% if not model_relation %}
             {{ exceptions.raise_compiler_error("Unsupported model: " ~ model ~ " (this might happen if you override 'ref' or 'source')") }}
         {% endif %}
-
         {%- if elementary.is_ephemeral_model(model_relation) %}
             {{ exceptions.raise_compiler_error("The test is not supported for ephemeral models, model name: {}".format(model_relation.identifier)) }}
         {%- endif %}
@@ -66,6 +65,7 @@
                                                                                                 column_name=column_name,
                                                                                                 metric_properties=metric_properties) %}
                     {%- endif %}
+
                     {{ elementary.debug_log('min_bucket_start - ' ~ min_bucket_start) }}
                     {{ elementary.test_log('start', full_table_name, column_name) }}
 
@@ -89,7 +89,6 @@
                                                                            metric_names=all_columns_monitors,
                                                                            columns_only=true,
                                                                            metric_properties=metric_properties) %}
-
         {% set anomaly_scores_test_table_relation = elementary.create_elementary_test_table(database_name, tests_schema_name, test_table_name, 'anomaly_scores', anomaly_scores_query) %}
 
         {{- elementary.test_log('end', full_table_name, 'all columns') }}

macros/edr/tests/test_utils/clean_elementary_test_tables.sql

@@ -35,6 +35,10 @@
     {% do return(elementary.get_transactionless_clean_elementary_test_tables_queries(test_table_relations)) %}
 {% endmacro %}
 
+{% macro clickhouse__get_clean_elementary_test_tables_queries(test_table_relations) %}
+    {% do return(elementary.get_transactionless_clean_elementary_test_tables_queries(test_table_relations)) %}
+{% endmacro %}
+
 {% macro athena__get_clean_elementary_test_tables_queries(test_table_relations) %}
     {% set queries = [] %}
     {% for test_relation in test_table_relations %}

macros/edr/tests/test_utils/create_elementary_test_table.sql

@@ -14,7 +14,7 @@
                                                                    identifier=temp_table_name,
                                                                    type='table') -%}
 
-        {# Create the table if it doesn't exist #}
+        {# Create the table if it doesnt exist #}
         {%- do elementary.create_or_replace(false, temp_table_relation, sql_query) %}
 
         {# Cache the test table for easy access later #}

macros/utils/cross_db_utils/datediff.sql

@@ -12,6 +12,12 @@
     {{ return(macro(first_date, second_date, date_part)) }}
 {% endmacro %}
 
+{% macro clickhouse__edr_datediff(first_date, second_date, date_part) %}
+    {%- set first_date_expr = elementary.edr_cast_as_timestamp(first_date) if first_date is string else first_date -%}
+    {%- set second_date_expr = elementary.edr_cast_as_timestamp(second_date) if second_date is string else second_date -%}
+    coalesce(dateDiff('{{ date_part }}', {{ first_date_expr }}, {{ second_date_expr }}), 0)::Nullable(Int32)
+{% endmacro %}
+
 {% macro bigquery__edr_datediff(first_date, second_date, date_part) %}
     {%- if date_part | lower in ['second', 'minute', 'hour', 'day'] %}
         timestamp_diff({{ second_date }}, {{ first_date }}, {{ date_part }})