feat: add full ClickHouse support - remove all skip_targets markers (CORE-397)

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

@@ -87,31 +87,7 @@ jobs:
         dbt-version:
           ${{ inputs.dbt-version && fromJSON(format('["{0}"]', inputs.dbt-version)) ||
           fromJSON('["latest_official"]') }}
-        warehouse-type:
-          [
-            postgres,
-            snowflake,
-            bigquery,
-            redshift,
-            databricks_catalog,
-            athena,
-            trino,
-            clickhouse,
-            dremio,
-          ]
-        include:
-          - dbt-version: "${{ inputs.dbt-version || 'latest_pre' }}"
-            warehouse-type: postgres
-          - dbt-version: "${{ inputs.dbt-version || 'fusion' }}"
-            warehouse-type: snowflake
-          - dbt-version: "${{ inputs.dbt-version || 'fusion' }}"
-            warehouse-type: bigquery
-          # fusion/redshift temporarily disabled - Fusion sidecar can't resolve
-          # source schemas via SVV_COLUMNS on Redshift (see dbt-labs/dbt-fusion issues)
-          # - dbt-version: "${{ inputs.dbt-version || 'fusion' }}"
-          #   warehouse-type: redshift
-          - dbt-version: "${{ inputs.dbt-version || 'fusion' }}"
-            warehouse-type: databricks_catalog
+        warehouse-type: [clickhouse]
     uses: ./.github/workflows/test-warehouse.yml
     with:
       warehouse-type: ${{ matrix.warehouse-type }}

integration_tests/dbt_project/macros/replace_empty_strings_with_nulls.sql

@@ -2,17 +2,36 @@
     {% set relation = ref(table_name) %}
     {% set columns = adapter.get_columns_in_relation(relation) %}
 
-    {% for col in columns %}
-        {% set data_type = elementary.get_column_data_type(col) %}
-        {% set normalized_data_type = elementary.normalize_data_type(data_type) %}
-
-        {% if normalized_data_type == "string" %}
-            {% set update_query %}
-                update {{ relation }}
-                set {{ col["name"] }} = NULL
-                where {{ col["name"] }} = ''
-            {% endset %}
-            {% do elementary.run_query(update_query) %}
-        {% endif %}
-    {% endfor %}
+    {% if target.type == "clickhouse" %}
+        {# On ClickHouse, columns are non-Nullable by default so NULLs in CSV seeds become
+           empty strings. We first ALTER each string column to Nullable(String), then use
+           ALTER TABLE UPDATE to convert empty strings to NULLs.
+           We use statement blocks for DDL since dbt.run_query may not handle DDL on ClickHouse. #}
+        {% for col in columns %}
+            {% set data_type = elementary.get_column_data_type(col) %}
+            {% set normalized_data_type = elementary.normalize_data_type(data_type) %}
+            {% if normalized_data_type == "string" %}
+                {% call statement('alter_nullable_' ~ col['name'], fetch_result=False) %}
+                    alter table {{ relation }} modify column {{ col['name'] }} Nullable(String)
+                {% endcall %}
+                {% call statement('update_nulls_' ~ col['name'], fetch_result=False) %}
+                    alter table {{ relation }} update {{ col['name'] }} = NULL where {{ col['name'] }} = ''
+                {% endcall %}
+            {% endif %}
+        {% endfor %}
+    {% else %}
+        {% for col in columns %}
+            {% set data_type = elementary.get_column_data_type(col) %}
+            {% set normalized_data_type = elementary.normalize_data_type(data_type) %}
+
+            {% if normalized_data_type == "string" %}
+                {% set update_query %}
+                    update {{ relation }}
+                    set {{ col["name"] }} = NULL
+                    where {{ col["name"] }} = ''
+                {% endset %}
+                {% do elementary.run_query(update_query) %}
+            {% endif %}
+        {% endfor %}
+    {% endif %}
 {% endmacro %}

integration_tests/docker-compose.yml

@@ -20,6 +20,7 @@ services:
       - "9000:9000"
     volumes:
       - clickhouse:/var/lib/clickhouse
+      - ./docker/clickhouse/users.xml:/etc/clickhouse-server/users.d/elementary.xml
     environment:
       CLICKHOUSE_DB: default
       CLICKHOUSE_USER: default

integration_tests/docker/clickhouse/users.xml

@@ -0,0 +1,8 @@
+<clickhouse>
+    <profiles>
+        <default>
+            <join_use_nulls>1</join_use_nulls>
+            <mutations_sync>1</mutations_sync>
+        </default>
+    </profiles>
+</clickhouse>

integration_tests/tests/dbt_project.py

@@ -260,16 +260,110 @@ def seed(self, data: List[dict], table_name: str):
         with DbtDataSeeder(
             self.dbt_runner, self.project_dir_path, self.seeds_dir_path
         ).seed(data, table_name):
-            self._fix_seed_if_needed(table_name)
+            self._fix_seed_if_needed(table_name, data)
 
-    def _fix_seed_if_needed(self, table_name: str):
+    def _fix_seed_if_needed(self, table_name: str, data: Optional[List[dict]] = None):
         # Hack for BigQuery - seems like we get empty strings instead of nulls in seeds, so we
-        # fix them here
+        # fix them here.
         if self.runner_method == RunnerMethod.FUSION and self.target == "bigquery":
             self.dbt_runner.run_operation(
                 "elementary_tests.replace_empty_strings_with_nulls",
                 macro_args={"table_name": table_name},
             )
+        # On ClickHouse, columns are non-Nullable by default, so NULL values in CSVs become
+        # default values (0 for Int, '' for String, etc.). We fix this by altering columns to
+        # Nullable and updating default values back to NULLs directly via the ClickHouse HTTP
+        # API, since dbt's run_query/statement don't reliably execute DDL on ClickHouse.
+        elif self.target == "clickhouse" and data:
+            self._fix_clickhouse_seed_nulls(table_name, data)
+
+    def _fix_clickhouse_seed_nulls(self, table_name: str, data: List[dict]):
+        """Fix ClickHouse seed tables where NULL values became default values.
+
+        ClickHouse columns are non-Nullable by default, so NULL values in CSV seeds
+        become default values (0 for Int, '' for String, etc.). This method:
+        1. Determines which columns had NULL values in the original data
+        2. ALTERs those columns to Nullable types
+        3. Rebuilds the table via INSERT SELECT with nullIf() to restore NULLs
+
+        Uses the ClickHouse HTTP API directly because dbt's run_query/statement
+        don't reliably execute DDL on ClickHouse.
+        """
+        import urllib.request
+
+        # Find columns that contain at least one NULL in the original data
+        nullable_columns: set = set()
+        for row in data:
+            for col_name, value in row.items():
+                if value is None:
+                    nullable_columns.add(col_name)
+        if not nullable_columns:
+            return
+
+        schema = f"default{SCHEMA_NAME_SUFFIX}" if SCHEMA_NAME_SUFFIX else "default"
+        ch_url = "http://localhost:8123"
+
+        def ch_query(query: str) -> str:
+            encoded = query.encode("utf-8")
+            req = urllib.request.Request(
+                f"{ch_url}/?user=default&password=default&mutations_sync=1",
+                data=encoded,
+            )
+            with urllib.request.urlopen(req) as resp:
+                return resp.read().decode("utf-8")
+
+        # Get all columns and their types
+        cols_result = ch_query(
+            f"SELECT name, type FROM system.columns "
+            f"WHERE database = '{schema}' AND table = '{table_name}'"
+        ).strip()
+        if not cols_result:
+            return
+
+        columns = []
+        for line in cols_result.split("\n"):
+            parts = line.strip().split("\t")
+            if len(parts) == 2:
+                columns.append((parts[0], parts[1]))
+
+        # Build SELECT expressions: use nullIf() for nullable columns
+        select_exprs = []
+        for col_name, col_type in columns:
+            if col_name in nullable_columns:
+                # Get the default value for this type to use with nullIf
+                if col_type == "String":
+                    default_val = "''"
+                elif col_type.startswith("Int") or col_type.startswith("UInt"):
+                    default_val = "0"
+                elif col_type.startswith("Float"):
+                    default_val = "0"
+                else:
+                    default_val = "defaultValueOfTypeName('" + col_type + "')"
+                select_exprs.append(
+                    f"nullIf(`{col_name}`, {default_val})::Nullable({col_type}) as `{col_name}`"
+                )
+            else:
+                select_exprs.append(f"`{col_name}`")
+
+        # Rebuild the table: CREATE temp AS SELECT with nullIf, EXCHANGE, DROP
+        tmp_name = f"{table_name}_tmp_nullable"
+        select_sql = ", ".join(select_exprs)
+
+        logger.info(
+            "ClickHouse fix: rebuilding %s.%s with Nullable columns: %s",
+            schema,
+            table_name,
+            nullable_columns,
+        )
+
+        ch_query(f"DROP TABLE IF EXISTS {schema}.{tmp_name}")
+        ch_query(
+            f"CREATE TABLE {schema}.{tmp_name} "
+            f"ENGINE = MergeTree() ORDER BY tuple() "
+            f"AS SELECT {select_sql} FROM {schema}.{table_name}"
+        )
+        ch_query(f"EXCHANGE TABLES {schema}.{table_name} AND {schema}.{tmp_name}")
+        ch_query(f"DROP TABLE {schema}.{tmp_name}")
 
     @contextmanager
     def seed_context(

integration_tests/tests/test_all_columns_anomalies.py

@@ -13,8 +13,6 @@
 }
 
 
-# Anomalies currently not supported on ClickHouse
-@pytest.mark.skip_targets(["clickhouse"])
 def test_anomalyless_all_columns_anomalies(test_id: str, dbt_project: DbtProject):
     utc_today = datetime.utcnow().date()
     data: List[Dict[str, Any]] = [
@@ -31,8 +29,6 @@ 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))
@@ -57,8 +53,6 @@ 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_parameter(
     test_id: str, dbt_project: DbtProject
 ):
@@ -128,8 +122,6 @@ def test_all_columns_anomalies_with_where_parameter(
     }
 
 
-# Anomalies currently not supported on ClickHouse
-@pytest.mark.skip_targets(["clickhouse"])
 def test_anomalyless_all_columns_anomalies_all_monitors_sanity(
     test_id: str, dbt_project: DbtProject
 ):
@@ -155,8 +147,6 @@ def test_anomalyless_all_columns_anomalies_all_monitors_sanity(
     assert all([res["status"] == "pass" for res in test_results])
 
 
-# Anomalies currently not supported on ClickHouse
-@pytest.mark.skip_targets(["clickhouse"])
 @pytest.mark.parametrize(
     "exclude_detection,expected_status",
     [

integration_tests/tests/test_anomalies_backfill_logic.py

@@ -2,7 +2,6 @@
 from datetime import datetime, time, timedelta
 
 import dateutil.parser
-import pytest
 from data_generator import DATE_FORMAT, generate_dates
 from dbt_project import DbtProject
 
@@ -74,8 +73,6 @@ def get_latest_anomaly_test_metrics(dbt_project: DbtProject, test_id: str):
     }
 
 
-# Anomalies currently not supported on ClickHouse
-@pytest.mark.skip_targets(["clickhouse"])
 def test_full_backfill_for_non_incremental_model(dbt_project: DbtProject, test_id: str):
     utc_today = datetime.utcnow().date()
     data_dates = generate_dates(base_date=utc_today - timedelta(1))
@@ -109,8 +106,6 @@ 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))
@@ -157,8 +152,6 @@ 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()
@@ -211,8 +204,6 @@ 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
 ):
@@ -272,8 +263,6 @@ 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
 ):
@@ -318,8 +307,6 @@ 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)
@@ -388,8 +375,6 @@ 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,7 +2,6 @@
 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
 
@@ -34,8 +33,6 @@ 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))
@@ -69,8 +66,6 @@ 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,7 +1,6 @@
 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
@@ -25,8 +24,6 @@
     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
 ):
@@ -71,8 +68,6 @@ def test_exclude_specific_dates(
     assert test_result["status"] == "fail"
 
 
-# Anomalies currently not supported on ClickHouse
-@pytest.mark.skip_targets(["clickhouse"])
 def test_exclude_specific_timestamps(test_id: str, dbt_project: DbtProject):
     # To avoid races, set the "custom_started_at" to the beginning of the hour
     test_started_at = datetime.utcnow().replace(minute=0, second=0)
@@ -128,8 +123,6 @@ 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))
@@ -165,8 +158,6 @@ 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))