Parallelize test execution using xdist

.github/workflows/steps_local_testing.yml

@@ -55,17 +55,21 @@ jobs:
       - name: Install requirements
         run: pip3 install . -r dev_requirements.txt
 
+      # -n 4 --dist loadscope: run test classes concurrently on 4 xdist workers.
+      # loadscope is required — the project/schema fixtures are class-scoped.
+      # The per-test timeout is higher than in sequential runs because tests slow
+      # down under the extra load on the shared runner.
       - name: Run HTTP tests
         env:
           DBT_CH_TEST_CLUSTER: test_shard
         run: |
           PYTHONPATH="${PYTHONPATH}:dbt"
-          pytest tests --timeout=120
+          pytest tests -n 5 --dist loadscope --timeout=300
 
       - name: Run Native tests
         env:
           DBT_CH_TEST_PORT: 9000
           DBT_CH_TEST_CLUSTER: test_shard
         run: |
           PYTHONPATH="${PYTHONPATH}:dbt"
-          pytest tests --timeout=120
+          pytest tests -n 5 --dist loadscope --timeout=300

.github/workflows/test_cloud.yml

@@ -33,12 +33,14 @@ jobs:
       - name: Install requirements
         run: pip3 install . -r dev_requirements.txt
 
+      # -n 1 --dist loadscope: run test classes concurrently on 3 xdist workers
+      # (loadscope is required — the project/schema fixtures are class-scoped).
       - name: Run HTTP tests
         env:
           DBT_CH_TEST_PORT: 8443
-        run: pytest tests
+        run: pytest tests -n 1 --dist loadscope --timeout=300
 
       - name: Run Native tests
         env:
           DBT_CH_TEST_PORT: 9440
-        run: pytest tests
+        run: pytest tests -n 1 --dist loadscope --timeout=300

CONTRIBUTING.md

@@ -94,6 +94,27 @@ Add a user-facing entry to the unreleased section at the top of [CHANGELOG.md](C
 This adapter passes all of dbt basic tests as presented in dbt's [official docs](https://docs.getdbt.com/docs/contributing/testing-a-new-adapter#testing-your-adapter).
 Use `pytest tests` to run tests.
 
+There are two ways to run the integration tests:
+
+**Sequential (default):**  
+```shell
+# If you add DBT_CH_TEST_USE_DOCKER=True, as an environment variable or in the test.env file, pytest will automatically start and stop the ClickHouse cluster from tests/integration/docker-compose.yml for you.
+pytest tests
+```
+**Parallel:** 
+This method is incompatible with `DBT_CH_TEST_USE_DOCKER=True`. You need to start and stop the Docker container yourself. You can use the tests docker compose file to start the ClickHouse cluster:
+
+```shell
+docker compose -f tests/integration/docker-compose.yml up -d
+```
+
+And then:
+```shell
+pytest tests -n 10 --dist loadscope
+```
+
+#### Test configuration
+
 You can customize the test environment via environment variables. We recommend doing so with the pytest `pytest-dotenv` plugin combined with root level `test.env`
 configuration file (this file should not be checked into git).  The following environment variables are recognized:
 
@@ -111,7 +132,7 @@ configuration file (this file should not be checked into git).  The following en
 
 ### Example Configurations
 
-Local development using the docker compose file included (cluster start/stop will be managed by the tests)
+Local development using the docker compose file included (cluster start/stop will be managed by the tests; sequential runs only, see above)
 
 ```env
 DBT_CH_TEST_USE_DOCKER=True

dbt/adapters/clickhouse/dbclient.py

@@ -11,7 +11,7 @@
 )
 from dbt.adapters.clickhouse.logger import logger
 from dbt.adapters.clickhouse.query import quote_identifier
-from dbt.adapters.clickhouse.util import engine_can_atomic_exchange
+from dbt.adapters.clickhouse.util import engine_can_atomic_exchange, retry_on_database_error
 from dbt.adapters.exceptions import FailedToConnectError
 from dbt_common.exceptions import DbtConfigError, DbtDatabaseError
 
@@ -130,16 +130,38 @@ def __init__(self, credentials: ClickHouseCredentials):
     def query(self, sql: str, **kwargs):
         pass
 
+    def command(self, sql: str, retries: int = 1, **kwargs):
+        """Run a ClickHouse statement. Pass `retries` > 1 to retry transient
+        failures — only safe for idempotent statements."""
+        return retry_on_database_error(
+            lambda: self._command(sql, **kwargs), 'ClickHouse command', retries
+        )
+
+    def get_ch_setting(self, setting_name, retries: int = 3):
+        """Return (value, readonly) for a server setting, retrying transient
+        failures; (None, 0) when the setting could not be read."""
+        try:
+            return retry_on_database_error(
+                lambda: self._get_ch_setting(setting_name),
+                f'Reading ClickHouse setting {setting_name}',
+                retries,
+            )
+        except DbtDatabaseError as ex:
+            logger.warning(
+                f'Failed to read ClickHouse setting {setting_name} after {retries} attempts: {ex}'
+            )
+            return (None, 0)
+
     @abstractmethod
-    def command(self, sql: str, **kwargs):
+    def _command(self, sql: str, **kwargs):
         pass
 
     @abstractmethod
     def columns_in_query(self, sql: str, **kwargs):
         pass
 
     @abstractmethod
-    def get_ch_setting(self, setting_name):
+    def _get_ch_setting(self, setting_name):
         pass
 
     def database_dropped(self, database: str):
@@ -179,12 +201,25 @@ def _check_lightweight_deletes(self, requested: bool):
         # checked (and enabled when the user has permission to change it).
         global _nd_mutation_probe
         with _nd_mutation_lock:
-            if _nd_mutation_probe is None:
-                _nd_mutation_probe = self.get_ch_setting(ND_MUTATION_SETTING)
-            nd_mutations, nd_mutations_read_only = _nd_mutation_probe
+            probe = _nd_mutation_probe
+            if probe is None:
+                probe = self.get_ch_setting(ND_MUTATION_SETTING)
+                if probe[0] is not None:
+                    _nd_mutation_probe = probe
+                else:
+                    # Failed probe: leave the cache empty so the next connection
+                    # probes again instead of poisoning the whole process.
+                    logger.warning(
+                        f'Could not read the {ND_MUTATION_SETTING} setting from the server; '
+                        f'lightweight deletes are disabled for this connection only'
+                    )
+            nd_mutations, nd_mutations_read_only = probe
         if nd_mutations is None:
             if requested:
-                logger.warning(nd_mutations_not_enabled_warning)
+                logger.warning(
+                    f'The {ND_MUTATION_SETTING} setting could not be determined, so lightweight '
+                    f'deletes and the delete+insert incremental strategy are disabled'
+                )
             return False, False
         nd_mutations = int(nd_mutations) > 0
         if not nd_mutations:
@@ -194,12 +229,19 @@ def _check_lightweight_deletes(self, requested: bool):
                     raise DbtConfigError(nd_mutations_not_enabled_error)
                 logger.warning(nd_mutations_not_enabled_warning)
             else:
+                # A failure here silently downgrades the connection to the legacy
+                # incremental strategy, so command() retries transient errors and
+                # the final error is logged for diagnosis.
                 try:
-                    self.command(f'SET {ND_MUTATION_SETTING} = 1')
+                    self.command(f'SET {ND_MUTATION_SETTING} = 1', retries=3)
                     self._conn_settings[ND_MUTATION_SETTING] = '1'
                     nd_mutations = True
-                except DbtDatabaseError:
-                    logger.warning(nd_mutations_not_enabled_warning)
+                except DbtDatabaseError as ex:
+                    logger.warning(
+                        f'Could not enable {ND_MUTATION_SETTING} for this connection, so '
+                        f'lightweight deletes and the delete+insert incremental strategy '
+                        f'are disabled: {ex}'
+                    )
         if nd_mutations:
             return True, requested
         return False, False

dbt/adapters/clickhouse/httpclient.py

@@ -28,7 +28,7 @@ def query(self, sql, **kwargs):
             err_msg = hide_stack_trace(ex)
             raise DbtDatabaseError(err_msg) from ex
 
-    def command(self, sql, **kwargs):
+    def _command(self, sql, **kwargs):
         try:
             self._inject_query_id(kwargs)
             return self._client.command(sql, **kwargs)
@@ -52,7 +52,7 @@ def columns_in_query(self, sql: str, **kwargs) -> List[ClickHouseColumn]:
             err_msg = hide_stack_trace(ex)
             raise DbtDatabaseError(err_msg) from ex
 
-    def get_ch_setting(self, setting_name):
+    def _get_ch_setting(self, setting_name):
         setting = self._client.server_settings.get(setting_name)
         return (setting.value, setting.readonly) if setting else (None, 0)
 

dbt/adapters/clickhouse/nativeclient.py

@@ -7,7 +7,6 @@
 from dbt.adapters.clickhouse import ClickHouseColumn, ClickHouseCredentials
 from dbt.adapters.clickhouse.__version__ import version as dbt_clickhouse_version
 from dbt.adapters.clickhouse.dbclient import ChClientWrapper, ChRetryableException
-from dbt.adapters.clickhouse.logger import logger
 from dbt.adapters.clickhouse.util import hide_stack_trace
 from dbt_common.exceptions import DbtDatabaseError
 
@@ -25,7 +24,7 @@ def query(self, sql, **kwargs):
             err_msg = hide_stack_trace(ex)
             raise DbtDatabaseError(err_msg) from ex
 
-    def command(self, sql, **kwargs):
+    def _command(self, sql, **kwargs):
         try:
             result = self._client.execute(sql, **kwargs)
             if len(result) and len(result[0]):
@@ -46,14 +45,14 @@ def columns_in_query(self, sql: str, **kwargs) -> List[ClickHouseColumn]:
             err_msg = hide_stack_trace(ex)
             raise DbtDatabaseError(err_msg) from ex
 
-    def get_ch_setting(self, setting_name):
+    def _get_ch_setting(self, setting_name):
         try:
             result = self._client.execute(
                 f"SELECT value, readonly FROM system.settings WHERE name = '{setting_name}'"
             )
         except clickhouse_driver.errors.Error as ex:
-            logger.warn('Unexpected error retrieving ClickHouse server setting', ex)
-            return None
+            err_msg = hide_stack_trace(ex)
+            raise DbtDatabaseError(err_msg) from ex
         return (result[0][0], result[0][1]) if result else (None, 0)
 
     def close(self):

dbt/adapters/clickhouse/util.py

@@ -1,6 +1,30 @@
 import os
+import time
+from typing import Callable, TypeVar
 
-from dbt_common.exceptions import DbtRuntimeError
+from dbt.adapters.clickhouse.logger import logger
+from dbt_common.exceptions import DbtDatabaseError, DbtRuntimeError
+
+T = TypeVar('T')
+
+
+def retry_on_database_error(
+    fn: Callable[[], T], description: str, attempts: int, delay: float = 0.5
+) -> T:
+    """Call `fn`, retrying up to `attempts` total tries when it raises
+    DbtDatabaseError, sleeping `delay` seconds between tries. Retried failures
+    are logged with `description`; the exception from the final attempt
+    propagates to the caller. Only use for idempotent operations."""
+    attempt = 0
+    while True:
+        attempt += 1
+        try:
+            return fn()
+        except DbtDatabaseError as ex:
+            if attempt >= attempts:
+                raise
+            logger.warning(f'{description} failed (attempt {attempt}/{attempts}), retrying: {ex}')
+            time.sleep(delay)
 
 
 def compare_versions(v1: str, v2: str) -> int:

dev_requirements.txt

@@ -3,6 +3,7 @@ dbt-tests-adapter>=1.10,<2.0
 pytest>=7.2.0
 pytest-dotenv==0.5.2
 pytest-timeout==2.4.0
+pytest-xdist==3.8.0
 ruff==0.16.0
 mypy==1.19.1
 yamllint==1.37.1

pytest.ini

@@ -5,6 +5,10 @@ filterwarnings =
 env_files =
     test.env
 pythonpath = .
+# name per convention
 testpaths =
-    tests/integration  # name per convention
-addopts = --doctest-modules
+    tests/integration
+# --ignore-glob: --doctest-modules would otherwise import every conftest.py as a
+# plain module named `conftest`, which collides once two directories have one.
+# Conftest loading for fixtures is a separate mechanism and is not affected.
+addopts = --doctest-modules --ignore-glob=**/conftest.py

tests/integration/adapter/clickhouse/test_clickhouse_errors.py

@@ -2,7 +2,7 @@
 from dbt.tests.util import run_dbt
 
 oom_table_sql = """
-SELECT a FROM system.numbers_mt GROUP BY repeat(toString(number), 100000) as a
+select sum(number) as a from system.numbers limit 1000 SETTINGS max_memory_usage = 1
 """
 
 schema_yaml = """

tests/integration/adapter/dbt_clone/conftest.py

@@ -0,0 +1,9 @@
+import pytest
+
+
+# The upstream dbt_clone fixtures grant to "{{ env_var('DBT_TEST_USER_1') }}" at
+# parse time; requesting ch_test_users here creates per-class users and exports
+# the DBT_TEST_USER_1..3 variables before any test in this package runs dbt.
+@pytest.fixture(scope="class", autouse=True)
+def ensure_ch_test_users(ch_test_users):
+    pass

tests/integration/adapter/grants/conftest.py

@@ -0,0 +1,12 @@
+import pytest
+
+
+# The upstream grants base classes read DBT_TEST_USER_1..3 from the environment:
+# BaseGrants.get_test_users at class setup, and the model/schema fixtures via
+# env_var() at parse time. Requesting ch_test_users here creates per-class users
+# and exports those variables first — conftest-level autouse fixtures are ordered
+# before same-scope autouse fixtures defined on the test classes, so this resolves
+# ahead of BaseGrants.get_test_users.
+@pytest.fixture(scope="class", autouse=True)
+def ensure_ch_test_users(ch_test_users):
+    pass