Merge pull request #806 from elementary-data/ele-4224

Ele 4224

Author: michael-myaskovsky

.github/workflows/test-all-warehouses.yml

@@ -48,6 +48,7 @@ jobs:
             databricks_catalog,
             athena,
             trino,
+            clickhouse,
           ]
         include:
           # If we're not running on a specific dbt version, then always add postgres on 1.7.0

.github/workflows/test-warehouse.yml

@@ -16,6 +16,7 @@ on:
           - spark
           - athena
           - trino
+          - clickhouse
       elementary-ref:
         type: string
         required: false
@@ -82,6 +83,11 @@ jobs:
         working-directory: ${{ env.TESTS_DIR }}
         run: docker compose -f docker-compose-trino.yml up -d
 
+      - name: Start Clickhouse
+        if: inputs.warehouse-type == 'clickhouse'
+        working-directory: ${{ env.TESTS_DIR }}
+        run: docker compose up -d clickhouse
+
       - name: Setup Python
         uses: actions/setup-python@v4
         with:

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/dbt_project/macros/get_anomaly_config.sql

@@ -1,4 +1,8 @@
 {% macro get_anomaly_config(model_config, config) %}
+    {{ return(adapter.dispatch('get_anomaly_config', 'elementary')(model_config, config)) }}
+{% endmacro %}
+
+{% macro default__get_anomaly_config(model_config, config) %}
   {% set mock_model = {
     "alias": "mock_model",
     "config": {
@@ -19,4 +23,27 @@
     }
   }) %}
   {% do return(elementary.get_anomalies_test_configuration(api.Relation.create("db", "schema", "mock_model"), **config)[0]) %}
+{% endmacro %}
+
+{% macro clickhouse__get_anomaly_config(model_config, config) %}
+  {% set mock_model = {
+    "alias": "mock_model",
+    "config": {
+      "elementary": model_config
+    }
+  } %}
+  {# trick elementary into thinking this is the running model #}
+  {% do context.update({
+    "model": {
+      "depends_on": {
+        "nodes": ["id"]
+      }
+    },
+    "graph": {
+      "nodes": {
+        "id": mock_model
+      }
+    }
+  }) %}
+  {% do return(elementary.get_anomalies_test_configuration(api.Relation.create("schema", "schema", "mock_model"), **config)[0]) %}
 {% endmacro %}

integration_tests/docker-compose.yml

@@ -12,6 +12,23 @@ services:
     volumes:
       - postgres:/var/lib/postgresql/data
 
+  clickhouse:
+    image: clickhouse/clickhouse-server:latest
+    container_name: clickhouse
+    ports:
+      - "8123:8123"
+      - "9000:9000"
+    volumes:
+      - ./clickhouse-data:/var/lib/clickhouse
+    environment:
+      CLICKHOUSE_DB: default
+      CLICKHOUSE_USER: default
+      CLICKHOUSE_PASSWORD: "default"
+    ulimits:
+      nofile:
+        soft: 262144
+        hard: 262144
+
   pgadmin:
     image: dpage/pgadmin4
     ports:

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
 

integration_tests/tests/test_all_columns_anomalies.py

@@ -1,6 +1,7 @@
 from datetime import datetime, timedelta
 from typing import Any, Dict, List
 
+import pytest
 from data_generator import DATE_FORMAT, generate_dates
 from dbt_project import DbtProject
 
@@ -28,6 +29,8 @@ def test_anomalyless_all_columns_anomalies(test_id: str, dbt_project: DbtProject
     assert all([res["status"] == "pass" for res in test_results])
 
 
+# Anomalies currently not supported on ClickHouse
+@pytest.mark.skip_targets(["clickhouse"])
 def test_anomalous_all_columns_anomalies(test_id: str, dbt_project: DbtProject):
     utc_today = datetime.utcnow().date()
     test_date, *training_dates = generate_dates(base_date=utc_today - timedelta(1))
@@ -52,6 +55,8 @@ def test_anomalous_all_columns_anomalies(test_id: str, dbt_project: DbtProject):
     assert col_to_status == {"superhero": "fail", TIMESTAMP_COLUMN: "pass"}
 
 
+# Anomalies currently not supported on ClickHouse
+@pytest.mark.skip_targets(["clickhouse"])
 def test_all_columns_anomalies_with_where_expression(
     test_id: str, dbt_project: DbtProject
 ):

integration_tests/tests/test_anomalies_backfill_logic.py

@@ -2,6 +2,7 @@
 from datetime import datetime, time, timedelta
 
 import dateutil.parser
+import pytest
 from data_generator import DATE_FORMAT, generate_dates
 from dbt_project import DbtProject
 
@@ -106,6 +107,8 @@ def test_full_backfill_for_non_incremental_model(dbt_project: DbtProject, test_i
     }
 
 
+# Anomalies currently not supported on ClickHouse
+@pytest.mark.skip_targets(["clickhouse"])
 def test_partial_backfill_for_incremental_models(dbt_project: DbtProject, test_id: str):
     utc_today = datetime.utcnow().date()
     data_dates = generate_dates(base_date=utc_today - timedelta(1))
@@ -152,6 +155,8 @@ def test_partial_backfill_for_incremental_models(dbt_project: DbtProject, test_i
     }
 
 
+# Anomalies currently not supported on ClickHouse
+@pytest.mark.skip_targets(["clickhouse"])
 def test_longer_backfill_in_case_of_a_gap(dbt_project: DbtProject, test_id: str):
     date_gap_size = 5
     utc_today = datetime.utcnow().date()
@@ -204,6 +209,8 @@ def test_longer_backfill_in_case_of_a_gap(dbt_project: DbtProject, test_id: str)
     }
 
 
+# Anomalies currently not supported on ClickHouse
+@pytest.mark.skip_targets(["clickhouse"])
 def test_full_backfill_if_metric_not_updated_for_a_long_time(
     dbt_project: DbtProject, test_id: str
 ):
@@ -263,6 +270,8 @@ def test_full_backfill_if_metric_not_updated_for_a_long_time(
     }
 
 
+# Anomalies currently not supported on ClickHouse
+@pytest.mark.skip_targets(["clickhouse"])
 def test_backfill_when_metric_doesnt_exist_back_enough(
     dbt_project: DbtProject, test_id: str
 ):
@@ -307,6 +316,8 @@ def test_backfill_when_metric_doesnt_exist_back_enough(
     }
 
 
+# Anomalies currently not supported on ClickHouse
+@pytest.mark.skip_targets(["clickhouse"])
 def test_backfill_with_middle_buckets_gap(dbt_project: DbtProject, test_id: str):
     utc_today = datetime.utcnow().date()
     data_start = utc_today - timedelta(21)
@@ -375,6 +386,8 @@ def test_backfill_with_middle_buckets_gap(dbt_project: DbtProject, test_id: str)
     }
 
 
+# Anomalies currently not supported on ClickHouse
+@pytest.mark.skip_targets(["clickhouse"])
 def test_bucket_size_not_aligned_with_days(dbt_project: DbtProject, test_id: str):
     """
     In this test we choose a bucket size that is not aligned with one day - specifically 7 hours.

integration_tests/tests/test_anomalies_ranges.py

@@ -2,6 +2,7 @@
 from datetime import datetime, timedelta
 from typing import Any, Dict, List
 
+import pytest
 from data_generator import DATE_FORMAT, generate_dates
 from dbt_project import DbtProject
 
@@ -33,6 +34,8 @@ def get_latest_anomaly_test_points(dbt_project: DbtProject, test_id: str):
     return [json.loads(result["result_row"]) for result in results]
 
 
+# Anomalies currently not supported on ClickHouse
+@pytest.mark.skip_targets(["clickhouse"])
 def test_anomaly_ranges_are_valid(test_id: str, dbt_project: DbtProject):
     utc_today = datetime.utcnow().date()
     test_date, *training_dates = generate_dates(base_date=utc_today - timedelta(1))
@@ -66,6 +69,8 @@ def test_anomaly_ranges_are_valid(test_id: str, dbt_project: DbtProject):
     assert all([row["min_value"] == row["max_value"] for row in anomaly_test_points])
 
 
+# Anomalies currently not supported on ClickHouse
+@pytest.mark.skip_targets(["clickhouse"])
 def test_anomaly_ranges_are_valid_with_seasonality(
     test_id: str, dbt_project: DbtProject
 ):

integration_tests/tests/test_anomaly_exclude_metrics.py

@@ -1,6 +1,7 @@
 from datetime import datetime, timedelta
 from typing import Any, Dict, List
 
+import pytest
 from data_generator import DATE_FORMAT, generate_dates
 from dbt_project import DbtProject
 from parametrization import Parametrization
@@ -24,6 +25,8 @@
     time_bucket={"period": "hour", "count": 6},
     dates_step=timedelta(hours=6),
 )
+# Anomalies currently not supported on ClickHouse
+@pytest.mark.skip_targets(["clickhouse"])
 def test_exclude_specific_dates(
     test_id: str, dbt_project: DbtProject, time_bucket: dict, dates_step: timedelta
 ):
@@ -123,6 +126,8 @@ def test_exclude_specific_timestamps(test_id: str, dbt_project: DbtProject):
     assert test_result["status"] == "fail"
 
 
+# Anomalies currently not supported on ClickHouse
+@pytest.mark.skip_targets(["clickhouse"])
 def test_exclude_date_range(test_id: str, dbt_project: DbtProject):
     utc_today = datetime.utcnow().date()
     test_date, *training_dates = generate_dates(base_date=utc_today - timedelta(1))
@@ -158,6 +163,8 @@ def test_exclude_date_range(test_id: str, dbt_project: DbtProject):
     assert test_result["status"] == "fail"
 
 
+# Anomalies currently not supported on ClickHouse
+@pytest.mark.skip_targets(["clickhouse"])
 def test_exclude_by_metric_value(test_id: str, dbt_project: DbtProject):
     utc_today = datetime.utcnow().date()
     test_date, *training_dates = generate_dates(base_date=utc_today - timedelta(1))