feat: configure distributed models with is_distributed

CHANGELOG.md

@@ -2,6 +2,8 @@
 
 #### Improvements
 * `dbt-core` is now capped at `<2.0`, and `dbt-common` and `agate` are declared as direct dependencies, matching dbt-labs adapters ([#776](https://github.com/ClickHouse/dbt-clickhouse/pull/776)).
+* Configure distributed models with `is_distributed=True` on a `table` or `incremental` model, while `materialized='distributed_table'` and `'distributed_incremental'` keep working but are deprecated and log a warning. Contracted incremental models then need `on_schema_change='append_new_columns'` or `'fail'`, as dbt requires for any `incremental` model ([#777](https://github.com/ClickHouse/dbt-clickhouse/pull/777)).
+* Support the `microbatch` incremental strategy on distributed incremental models (`materialized='incremental'` with `is_distributed=True`) ([#777](https://github.com/ClickHouse/dbt-clickhouse/pull/777)).
 
 #### Repository maintenance
 * The shard-local table name of distributed materializations is now resolved in one place, `clickhouse_local_relation`, instead of being rebuilt inline in each materialization and incremental strategy ([#753](https://github.com/ClickHouse/dbt-clickhouse/pull/753)).

README.md

@@ -43,8 +43,7 @@ pip install dbt-clickhouse
 - [x] Snapshots
 - [x] Most dbt-utils macros (now included in dbt-core)
 - [x] Ephemeral materialization
-- [x] Distributed table materialization (experimental)
-- [x] Distributed incremental materialization (experimental)
+- [x] Distributed table and incremental models via `is_distributed=True` (experimental)
 - [x] Contracts
 - [x] ClickHouse-specific column configurations (Codec, TTL...)
 - [x] ClickHouse-specific table settings (indexes, projections...)

dbt/adapters/clickhouse/impl.py

@@ -554,10 +554,28 @@ def _build_settings_str(self, settings: Dict[str, Any]) -> str:
                 res.append(f"{key}={settings[key]}")
         return '' if len(res) == 0 else 'SETTINGS ' + ', '.join(res) + '\n'
 
+    @available
+    def is_distributed(self, model_config) -> bool:
+        """`is_distributed=True` on a table or incremental model, or the deprecated `materialized='distributed_*'`."""
+        materialized = model_config.get('materialized') or ''
+        if materialized not in ('table', 'incremental'):
+            return materialized.startswith('distributed_')
+        # Quoted YAML values and rendered env vars arrive as strings
+        value = model_config.get('is_distributed')
+        if value is None or isinstance(value, bool):
+            return bool(value)
+        if isinstance(value, str) and value.strip().lower() in ('true', 'false'):
+            return value.strip().lower() == 'true'
+        raise DbtRuntimeError(f"Invalid value for is_distributed: {value!r}. Use true or false.")
+
     @available
     def get_model_settings(self, model, engine='MergeTree'):
         settings = model['config'].get('settings', {})
         materialization_type = model['config'].get('materialized')
+        if materialization_type in ('table', 'incremental') and self.is_distributed(
+            model['config']
+        ):
+            materialization_type = f'distributed_{materialization_type}'
         conn = self.connections.get_if_exists()
         conn.handle.update_model_settings(settings, materialization_type)
         settings = self.filter_settings_by_engine(settings, engine)

dbt/include/clickhouse/macros/materializations/clone.sql

@@ -6,7 +6,7 @@
         - Distributed tables -> not supported, fall back to a view.
         - Materialized views -> only the target table is cloned; now MV is crated/attached to the clon.
     #}
-    {%- if config.get('materialized', '').startswith('distributed_') -%}
+    {%- if adapter.is_distributed(model['config']) -%}
         {{ return(False) }}
     {%- endif -%}
     {%- set engine = config.get('engine', default='MergeTree()') -%}

dbt/include/clickhouse/macros/materializations/distributed_table.sql

@@ -1,4 +1,15 @@
 {% materialization distributed_table, adapter='clickhouse' %}
+  {% do clickhouse__warn_deprecated_distributed_config('table') %}
+  {{ return(clickhouse__distributed_table()) }}
+{%- endmaterialization %}
+
+{% macro clickhouse__warn_deprecated_distributed_config(materialized) %}
+  {% do exceptions.warn(
+      model.unique_id ~ ": configuring a distributed model with materialized='distributed_" ~ materialized
+      ~ "' is deprecated. Configure it with materialized='" ~ materialized ~ "' and is_distributed=True instead.") %}
+{% endmacro %}
+
+{% macro clickhouse__distributed_table() %}
   {% set insert_distributed_sync = run_query("SELECT value FROM system.settings WHERE name = 'insert_distributed_sync'")[0][0] %}
   {% if insert_distributed_sync != '1' %}
      {% do exceptions.raise_compiler_error('To use distributed materialization setting insert_distributed_sync should be set to 1') %}
@@ -77,8 +88,7 @@
   {{ drop_relation_if_exists(backup_relation) }}
   {{ run_hooks(post_hooks, inside_transaction=False) }}
   {{ return({'relations': [target_relation]}) }}
-
-{% endmaterialization %}
+{%- endmacro %}
 
 {% macro create_distributed_table(relation, local_relation) %}
     {%- set cluster = adapter.get_clickhouse_cluster_name() -%}

dbt/include/clickhouse/macros/materializations/incremental/distributed_incremental.sql

@@ -1,4 +1,9 @@
 {% materialization distributed_incremental, adapter='clickhouse' %}
+  {% do clickhouse__warn_deprecated_distributed_config('incremental') %}
+  {{ return(clickhouse__distributed_incremental()) }}
+{%- endmaterialization %}
+
+{% macro clickhouse__distributed_incremental() %}
   {% set insert_distributed_sync = run_query("SELECT value FROM system.settings WHERE name = 'insert_distributed_sync'")[0][0] %}
   {% if insert_distributed_sync != '1' %}
      {% do exceptions.raise_compiler_error('To use distributed materialization setting insert_distributed_sync should be set to 1') %}
@@ -102,6 +107,8 @@
       {% set need_swap = true %}
     {% elif incremental_strategy == 'delete_insert' %}
       {% do clickhouse__incremental_delete_insert(existing_relation, unique_key, incremental_predicates, True) %}
+    {% elif incremental_strategy == 'microbatch' %}
+      {% do clickhouse__incremental_microbatch(existing_relation, unique_key, incremental_predicates, True) %}
     {% elif incremental_strategy == 'insert_overwrite' %}
       {% do clickhouse__incremental_insert_overwrite(existing_relation, partition_by, True) %}
     {% elif incremental_strategy == 'append' %}
@@ -157,6 +164,6 @@
 
   {{ run_hooks(post_hooks, inside_transaction=False) }}
 
-  {{ return({'relations': [target_relation]}) }}
+  {{ return({'relations': [target_relation, target_relation_local]}) }}
 
-{%- endmaterialization %}
+{%- endmacro %}

dbt/include/clickhouse/macros/materializations/incremental/incremental.sql

@@ -1,4 +1,7 @@
 {% materialization incremental, adapter='clickhouse' %}
+  {%- if adapter.is_distributed(model['config']) -%}
+    {{ return(clickhouse__distributed_incremental()) }}
+  {%- endif -%}
 
   {%- set existing_relation = load_cached_relation(this) -%}
   {%- set target_relation = this.incorporate(type='table') -%}
@@ -73,13 +76,7 @@
     {% elif incremental_strategy == 'delete_insert' %}
       {% do clickhouse__incremental_delete_insert(existing_relation, unique_key, incremental_predicates) %}
     {% elif incremental_strategy == 'microbatch' %}
-      {%- if config.get("__dbt_internal_microbatch_event_time_start") -%}
-        {% do incremental_predicates.append(config.get("event_time") ~ " >= toDateTime('" ~ config.get("__dbt_internal_microbatch_event_time_start").strftime("%Y-%m-%d %H:%M:%S") ~ "')") %}
-      {%- endif -%}
-      {%- if model.config.__dbt_internal_microbatch_event_time_end -%}
-        {% do incremental_predicates.append(config.get("event_time") ~ " < toDateTime('" ~ config.get("__dbt_internal_microbatch_event_time_end").strftime("%Y-%m-%d %H:%M:%S") ~ "')") %}
-      {%- endif -%}
-      {% do clickhouse__incremental_delete_insert(existing_relation, unique_key, incremental_predicates) %}
+      {% do clickhouse__incremental_microbatch(existing_relation, unique_key, incremental_predicates) %}
     {% elif incremental_strategy == 'append' %}
       {% call statement('main') %}
         {{ clickhouse__insert_into(target_relation, sql, has_contract) }}
@@ -247,6 +244,18 @@
     {{ drop_relation_if_exists(distributed_new_data_relation) }}
 {% endmacro %}
 
+{% macro clickhouse__incremental_microbatch(existing_relation, unique_key, incremental_predicates, is_distributed=False) %}
+    {%- set event_time_start = config.get("__dbt_internal_microbatch_event_time_start") -%}
+    {%- set event_time_end = config.get("__dbt_internal_microbatch_event_time_end") -%}
+    {%- if event_time_start -%}
+      {% do incremental_predicates.append(config.get("event_time") ~ " >= toDateTime('" ~ event_time_start.strftime("%Y-%m-%d %H:%M:%S") ~ "')") %}
+    {%- endif -%}
+    {%- if event_time_end -%}
+      {% do incremental_predicates.append(config.get("event_time") ~ " < toDateTime('" ~ event_time_end.strftime("%Y-%m-%d %H:%M:%S") ~ "')") %}
+    {%- endif -%}
+    {% do clickhouse__incremental_delete_insert(existing_relation, unique_key, incremental_predicates, is_distributed) %}
+{% endmacro %}
+
 {% macro clickhouse__incremental_insert_overwrite(existing_relation, partition_by, is_distributed=False) %}
     {% set new_data_relation = existing_relation.incorporate(path={"identifier": existing_relation.identifier
        + '__dbt_new_data_' + invocation_id.replace('-', '_')}) %}

dbt/include/clickhouse/macros/materializations/table.sql

@@ -1,4 +1,7 @@
 {% materialization table, adapter='clickhouse' %}
+  {%- if adapter.is_distributed(model['config']) -%}
+    {{ return(clickhouse__distributed_table()) }}
+  {%- endif -%}
 
   {%- set existing_relation = load_cached_relation(this) -%}
   {%- set target_relation = this.incorporate(type='table') -%}

docs/dbt-core-v2-fusion-and-platform.mdx

@@ -78,7 +78,7 @@ The tables below compare the features available in `dbt-clickhouse` (v1) with th
 | `snapshot` | Type 2 slowly changing dimensions. | <Badge color="green">Supported</Badge> | |
 | `seed` | Load CSVs as tables. | <Badge color="green">Supported</Badge> | Numeric column types are inferred as `Int64`/`Float64` instead of `Int32`/`Float32`; use `column_types` to set them explicitly. |
 | `ephemeral` | Inline CTE, no object created. | <Badge color="green">Supported</Badge> | |
-| `distributed_table`, `distributed_incremental` | Sharded tables behind a Distributed engine. | <Badge color="green">Supported</Badge> | Self-managed clusters with `cluster` set in the profile, as in v1. |
+| `is_distributed` on `table` and `incremental` | Sharded tables behind a Distributed engine. | <Badge color="green">Supported</Badge> | Self-managed clusters with `cluster` set in the profile, as in v1. |
 
 #### Model configuration {#parity-model-configuration}
 

docs/index.mdx

@@ -79,8 +79,7 @@ The following are [experimental features](/reference/settings/beta-and-experimen
 | Type                                    | Supported?                                    | Details                                                                                                                                                                                 |
 |-----------------------------------------|-----------------------------------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
 | Materialized View materialization       | Yes. Creation with explicit target is in Beta | Creates a [materialized view](/reference/statements/create/view#materialized-view).                                                                                                 |
-| Distributed table materialization       | Yes, Experimental                             | Creates a [distributed table](/reference/engines/table-engines/special/distributed).                                                                                                              |
-| Distributed incremental materialization | Yes, Experimental                             | Incremental model based on the same idea as distributed table. Note that not all strategies are supported, visit [their docs section](/integrations/connectors/data-ingestion/etl-tools/dbt/materializations#materialization-distributed-incremental) for more info. |
+| Distributed models                      | Yes, Experimental                             | `table` and `incremental` models with `is_distributed=True` create a [distributed table](/reference/engines/table-engines/special/distributed). Not all incremental strategies are supported, see [their docs section](/integrations/connectors/data-ingestion/etl-tools/dbt/materializations#distributed-models). |
 | Dictionary materialization              | Yes, Experimental                             | Creates a [dictionary](/reference/engines/table-engines/special/dictionary).                                                                                                                      |
 
 ## Setup of dbt and the ClickHouse adapter {#setup-of-dbt-and-the-clickhouse-adapter}

docs/materializations.mdx

@@ -173,7 +173,7 @@ You can add [data skipping indexes](/concepts/features/performance/skip-indexes/
 
 ### Projections {#projections}
 
-You can add [projections](/concepts/features/projections/projections) to `table` and `distributed_table` materializations using the `projections` configuration. Each projection entry requires either a `query` or an `index` key (not both).
+You can add [projections](/concepts/features/projections/projections) to `table` materializations, including [distributed](#distributed-models) ones, using the `projections` configuration. Each projection entry requires either a `query` or an `index` key (not both).
 
 **Note**: For distributed tables, the projection is applied to the `_local` tables, not to the distributed proxy table.
 **Note**: Specifying both `query` and `index` on the same projection entry raises a compile-time error.
@@ -426,27 +426,40 @@ select 1
 
 See the [dictionary tests](https://github.com/ClickHouse/dbt-clickhouse/blob/main/tests/integration/adapter/dictionary/test_dictionary.py) for more examples, including range and direct dictionaries.
 
-## Materialization: distributed_table (experimental) {#materialization-distributed-table}
+## Distributed models (experimental) {#distributed-models}
 
-Distributed table created with following steps:
+Set `is_distributed=True` on a `table` or `incremental` model to build it as a [Distributed](/reference/engines/table-engines/special/distributed) table over local tables on every shard.
+Distributed models require the `cluster` profile setting.
+The local table names follow the `local_suffix` and `local_db_prefix` profile settings.
+Other materializations ignore `is_distributed`, so it can be set for a whole folder in `dbt_project.yml`.
+
+Configuring a distributed model with `materialized='distributed_table'` or `materialized='distributed_incremental'` is deprecated.
+That configuration keeps working and builds the same model, but logs a deprecation warning.
+When moving an incremental model with an enforced contract to `is_distributed=True`, set `on_schema_change` as dbt requires for [incremental models with contracts](https://docs.getdbt.com/reference/resource-configs/contract#incremental-models-and-on_schema_change).
+
+dbt-clickhouse queries automatically include the setting `insert_distributed_sync = 1` so that downstream incremental operations execute correctly.
+This could cause some distributed table inserts to run more slowly than expected.
+
+| Option       | Description                                                                                                                                                    | Default if any |
+|--------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------|----------------|
+| sharding_key | Sharding key determines the destination server when inserting into distributed engine table. The sharding key can be random or as an output of a hash function | `rand()`       |
+
+### Distributed table {#materialization-distributed-table}
+
+A distributed table is built with the following steps:
 
 1. Creates temp view with sql query to get right structure
 2. Create empty local tables based on view
 3. Create distributed table based on local tables.
 4. Data inserts into distributed table, so it is distributed across shards without duplicating.
 
-Notes:
-- dbt-clickhouse queries now automatically include the setting `insert_distributed_sync = 1` in order to ensure that
-  downstream incremental
-  materialization operations execute correctly. This could cause some distributed table inserts to run more slowly than
-  expected.
-
-### Distributed table model example {#distributed-table-model-example}
+#### Distributed table model example {#distributed-table-model-example}
 
 ```sql
 {{
     config(
-        materialized='distributed_table',
+        materialized='table',
+        is_distributed=True,
         order_by='id, created_at',
         sharding_key='cityHash64(id)',
         engine='ReplacingMergeTree'
@@ -457,7 +470,7 @@ select id, created_at, item
 from {{ source('db', 'table') }}
 ```
 
-### Generated migrations {#distributed-table-generated-migrations}
+#### Generated migrations {#distributed-table-generated-migrations}
 
 ```sql
 CREATE TABLE db.table_local on cluster cluster (
@@ -476,31 +489,26 @@ CREATE TABLE db.table on cluster cluster (
     ENGINE = Distributed ('cluster', 'db', 'table_local', cityHash64(id));
 ```
 
-### Configurations {#distributed-table-configurations}
-Configurations that are specific for this materialization type are listed below:
-
-| Option                 | Description                                                                                                                                                                                                                                                                                                          | Default if any |
-| ---------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -------------- |
-| sharding_key           | Sharding key determines the destination server when inserting into distributed engine table.  The sharding key can be random or as an output of a hash function                                                                                                                                                      | `rand()`)      |
-
-## materialization: distributed_incremental (experimental) {#materialization-distributed-incremental}
+### Distributed incremental {#materialization-distributed-incremental}
 
 Incremental model based on the same idea as distributed table, the main difficulty is to process all incremental
 strategies correctly.
 
 1. _The Append Strategy_ just insert data into distributed table.
 2. _The Delete+Insert_ Strategy creates distributed temp table to work with all data on every shard.
 3. _The Default (Legacy) Strategy_ creates distributed temp and intermediate tables for the same reason.
+4. _The Microbatch Strategy_ runs the Delete+Insert strategy per batch.
 
 Only shard tables are replacing, because distributed table does not keep data.
 The distributed table reloads only when the full_refresh mode is enabled or the table structure may have changed.
 
-### Distributed incremental model example {#distributed-incremental-model-example}
+#### Distributed incremental model example {#distributed-incremental-model-example}
 
 ```sql
 {{
     config(
-        materialized='distributed_incremental',
+        materialized='incremental',
+        is_distributed=True,
         engine='MergeTree',
         incremental_strategy='append',
         unique_key='id,created_at'
@@ -511,7 +519,7 @@ select id, created_at, item
 from {{ source('db', 'table') }}
 ```
 
-### Generated migrations {#distributed-incremental-generated-migrations}
+#### Generated migrations {#distributed-incremental-generated-migrations}
 
 ```sql
 CREATE TABLE db.table_local on cluster cluster (