dbt: Add cluster AUTO SCALING STRATEGY (hydration burst) support DEX-67

misc/dbt-materialize/CHANGELOG.md

@@ -2,6 +2,22 @@
 
 ## Unreleased
 
+* Add support for the [`AUTO SCALING STRATEGY`](https://materialize.com/docs/sql/create-cluster/#autoscaling)
+  cluster option (hydration burst), which lets a managed cluster temporarily
+  burst to a larger size while it has un-hydrated objects. The `create_cluster`
+  operation accepts a new `auto_scaling_strategy` argument, and the new
+  `alter_cluster_auto_scaling_strategy` operation sets or removes the strategy
+  on an existing cluster:
+
+  ```bash
+  dbt run-operation alter_cluster_auto_scaling_strategy --args '{cluster_name: quickstart, auto_scaling_strategy: {on_hydration: {hydration_size: "800cc", linger_duration: "15s"}}}'
+  ```
+
+  During blue/green deployments, `deploy_init` now copies the production
+  cluster's autoscaling strategy to the deployment cluster, so the deployment
+  environment bursts to the configured hydration size while it hydrates ahead
+  of the cutover.
+
 ## 1.9.10 - 2026-05-20
 
 * Support unmanaged clusters in `deploy_init`. The deployment cluster

misc/dbt-materialize/dbt/include/materialize/macros/ci/alter_cluster_auto_scaling_strategy.sql

@@ -0,0 +1,87 @@
+-- Copyright Materialize, Inc. and contributors. All rights reserved.
+--
+-- Licensed under the Apache License, Version 2.0 (the "License");
+-- you may not use this file except in compliance with the License.
+-- You may obtain a copy of the License in the LICENSE file at the
+-- root of this repository, or online at
+--
+--     http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing, software
+-- distributed under the License is distributed on an "AS IS" BASIS,
+-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+-- See the License for the specific language governing permissions and
+-- limitations under the License.
+
+{#
+This macro sets or removes the autoscaling strategy of an existing managed
+cluster.
+
+  ## Arguments
+  - cluster_name (str): The name of the cluster. This parameter is required.
+  - auto_scaling_strategy (dict, optional): The autoscaling strategy to set,
+    e.g. {'on_hydration': {'hydration_size': '800cc', 'linger_duration': '15s'}}.
+    When omitted, the strategy is removed via RESET (AUTO SCALING STRATEGY).
+    An empty mapping ({}) disables autoscaling via the equivalent
+    AUTO SCALING STRATEGY = () form.
+
+  ## Usage
+  dbt run-operation alter_cluster_auto_scaling_strategy --args '{cluster_name: quickstart, auto_scaling_strategy: {on_hydration: {hydration_size: "800cc", linger_duration: "15s"}}}'
+  dbt run-operation alter_cluster_auto_scaling_strategy --args '{cluster_name: quickstart}'
+
+  Incompatibilities:
+  - Only managed clusters support an autoscaling strategy.
+  - The strategy cannot be combined with a schedule other than 'manual', and
+    its hydration_size must differ from the cluster's size.
+#}
+{% macro alter_cluster_auto_scaling_strategy(cluster_name, auto_scaling_strategy=none) %}
+
+    {% if not cluster_name %}
+        {{ exceptions.raise_compiler_error("cluster_name must be provided") }}
+    {% endif %}
+
+    {% if not cluster_exists(cluster_name) %}
+        {{ exceptions.raise_compiler_error("Cluster " ~ cluster_name ~ " does not exist") }}
+    {% endif %}
+
+    {% set cluster_configuration %}
+        SELECT
+            c.managed,
+            c.size,
+            cs.type AS schedule_type
+        FROM mz_clusters c
+        LEFT JOIN mz_internal.mz_cluster_schedules cs ON cs.cluster_id = c.id
+        WHERE c.name = {{ dbt.string_literal(cluster_name) }}
+    {% endset %}
+
+    {% set cluster_config_results = run_query(cluster_configuration) %}
+
+    {% if execute %}
+        {% set results = cluster_config_results.rows[0] %}
+        {% set managed = results[0] %}
+        {% set size = results[1] %}
+        {% set schedule_type = results[2] %}
+
+        {% if not managed %}
+            {{ exceptions.raise_compiler_error(
+                "Cluster " ~ cluster_name ~ " is unmanaged. AUTO SCALING STRATEGY "
+                ~ "is only supported on managed clusters."
+            ) }}
+        {% endif %}
+
+        {% if auto_scaling_strategy is none %}
+            {{ log("Removing autoscaling strategy from cluster " ~ cluster_name, info=True) }}
+            {% set alter_cluster_ddl %}
+                ALTER CLUSTER {{ adapter.quote(cluster_name) }} RESET (AUTO SCALING STRATEGY)
+            {% endset %}
+        {% else %}
+            {% do internal_validate_auto_scaling_strategy(auto_scaling_strategy, size, schedule_type) %}
+            {{ log("Setting autoscaling strategy on cluster " ~ cluster_name, info=True) }}
+            {% set alter_cluster_ddl %}
+                ALTER CLUSTER {{ adapter.quote(cluster_name) }} SET ({{ internal_render_auto_scaling_strategy(auto_scaling_strategy) }})
+            {% endset %}
+        {% endif %}
+
+        {{ run_query(alter_cluster_ddl) }}
+    {% endif %}
+{% endmacro %}

misc/dbt-materialize/dbt/include/materialize/macros/ci/create_cluster.sql

@@ -24,19 +24,28 @@ This macro creates a cluster with the specified properties.
   - refresh_hydration_time_estimate (str, optional): The estimated hydration time for the cluster. Only applicable when schedule_type is 'on-refresh'.
   - ignore_existing_objects (bool, optional): Whether to ignore existing objects in the cluster. Defaults to false.
   - force_deploy_suffix (bool, optional): Whether to forcefully add a deploy suffix to the cluster name. Defaults to false.
+  - auto_scaling_strategy (dict, optional): An autoscaling strategy for the cluster,
+    e.g. {'on_hydration': {'hydration_size': '800cc', 'linger_duration': '15s'}}.
+    The cluster temporarily bursts to the hydration size while it has un-hydrated
+    objects. Only applicable when schedule_type is 'manual'.
 
   Incompatibilities:
   - replication_factor is only applicable when schedule_type is 'manual'.
   - refresh_hydration_time_estimate is only applicable when schedule_type is 'on-refresh'.
+  - auto_scaling_strategy is only applicable when schedule_type is 'manual', and
+    its hydration_size must differ from size.
 #}
+{# NOTE: new optional arguments must be appended at the end of the signature,
+   since callers may invoke this macro with positional arguments. #}
 {% macro create_cluster(
     cluster_name,
     size,
     replication_factor=none,
     schedule_type=none,
     refresh_hydration_time_estimate=none,
     ignore_existing_objects=false,
-    force_deploy_suffix=false
+    force_deploy_suffix=false,
+    auto_scaling_strategy=none
 ) %}
 
     {# Input validation #}
@@ -48,6 +57,10 @@ This macro creates a cluster with the specified properties.
         {{ exceptions.raise_compiler_error("size must be provided") }}
     {% endif %}
 
+    {% if auto_scaling_strategy is not none %}
+        {% do internal_validate_auto_scaling_strategy(auto_scaling_strategy, size, schedule_type) %}
+    {% endif %}
+
     {% set deploy_cluster = adapter.generate_final_cluster_name(cluster_name, force_deploy_suffix) %}
 
     {% if cluster_exists(deploy_cluster) %}
@@ -66,6 +79,9 @@ This macro creates a cluster with the specified properties.
                         , SCHEDULE = ON REFRESH
                     {% endif %}
                 {% endif %}
+                {% if auto_scaling_strategy is not none %}
+                    , {{ internal_render_auto_scaling_strategy(auto_scaling_strategy) }}
+                {% endif %}
             )
         {% endset %}
         {{ run_query(create_cluster_ddl) }}

misc/dbt-materialize/dbt/include/materialize/macros/deploy/deploy_init.sql

@@ -133,12 +133,17 @@
         {% set refresh_hydration_time_estimate = results[4] %}
 
         {% if managed %}
+            {# The deployment cluster inherits the production cluster's
+               autoscaling strategy, so it bursts to the hydration size while
+               the deployment environment hydrates ahead of the cutover. #}
+            {% set auto_scaling_strategy = internal_get_cluster_auto_scaling_strategy(origin_cluster) %}
             {% set deploy_cluster = create_cluster(
                 cluster_name=cluster,
                 size=size,
                 replication_factor=replication_factor,
                 schedule_type=schedule_type,
                 refresh_hydration_time_estimate=refresh_hydration_time_estimate,
+                auto_scaling_strategy=auto_scaling_strategy,
                 ignore_existing_objects=ignore_existing_objects,
                 force_deploy_suffix=True
             ) %}

misc/dbt-materialize/dbt/include/materialize/macros/utils/auto_scaling_strategy.sql

@@ -0,0 +1,159 @@
+-- Copyright Materialize, Inc. and contributors. All rights reserved.
+--
+-- Licensed under the Apache License, Version 2.0 (the "License");
+-- you may not use this file except in compliance with the License.
+-- You may obtain a copy of the License in the LICENSE file at the
+-- root of this repository, or online at
+--
+--     http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing, software
+-- distributed under the License is distributed on an "AS IS" BASIS,
+-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+-- See the License for the specific language governing permissions and
+-- limitations under the License.
+
+{#
+Helpers for the `AUTO SCALING STRATEGY` cluster option (hydration burst),
+which lets a managed cluster temporarily burst to a larger size while it has
+un-hydrated objects, then return to its steady size once hydration completes.
+
+The adapter models the strategy as a nested mapping:
+
+  auto_scaling_strategy:
+    on_hydration:
+      hydration_size: '800cc'   # required; must differ from the cluster SIZE
+      linger_duration: '15s'    # optional; default '0s'
+
+An empty mapping ({}) renders as `AUTO SCALING STRATEGY = ()`, which disables
+autoscaling.
+#}
+
+{#
+Validates the shape of an `auto_scaling_strategy` mapping and its
+compatibility with the cluster's SIZE and schedule. Raises a compiler error
+on invalid input. `size` and `schedule_type` are optional and only checked
+when provided.
+#}
+{% macro internal_validate_auto_scaling_strategy(auto_scaling_strategy, size=none, schedule_type=none) %}
+    {% if auto_scaling_strategy is not mapping %}
+        {{ exceptions.raise_compiler_error(
+            "auto_scaling_strategy must be a mapping, e.g. "
+            ~ "{'on_hydration': {'hydration_size': '800cc', 'linger_duration': '15s'}}. "
+            ~ "Got: " ~ auto_scaling_strategy
+        ) }}
+    {% endif %}
+
+    {# AUTO SCALING STRATEGY is only supported on clusters with the default
+       MANUAL schedule, matching the server-side restriction. #}
+    {% if schedule_type is not none and schedule_type != 'manual' %}
+        {{ exceptions.raise_compiler_error(
+            "auto_scaling_strategy cannot be combined with a schedule_type "
+            ~ "other than 'manual'. Got schedule_type: " ~ schedule_type
+        ) }}
+    {% endif %}
+
+    {% for key in auto_scaling_strategy %}
+        {% if key != 'on_hydration' %}
+            {{ exceptions.raise_compiler_error(
+                "Unknown auto_scaling_strategy key '" ~ key ~ "'. "
+                ~ "The only supported sub-policy is 'on_hydration'."
+            ) }}
+        {% endif %}
+    {% endfor %}
+
+    {% if auto_scaling_strategy %}
+        {% set on_hydration = auto_scaling_strategy.get('on_hydration') %}
+        {% if on_hydration is not mapping or not on_hydration.get('hydration_size') %}
+            {{ exceptions.raise_compiler_error(
+                "auto_scaling_strategy.on_hydration must be a mapping with a "
+                ~ "'hydration_size' key, e.g. {'hydration_size': '800cc', "
+                ~ "'linger_duration': '15s'}. Got: " ~ on_hydration
+            ) }}
+        {% endif %}
+        {% for key in on_hydration %}
+            {% if key not in ['hydration_size', 'linger_duration'] %}
+                {{ exceptions.raise_compiler_error(
+                    "Unknown auto_scaling_strategy.on_hydration key '" ~ key ~ "'. "
+                    ~ "Supported keys: hydration_size, linger_duration."
+                ) }}
+            {% endif %}
+        {% endfor %}
+        {% if size is not none and on_hydration.get('hydration_size') == size %}
+            {{ exceptions.raise_compiler_error(
+                "auto_scaling_strategy.on_hydration.hydration_size must differ "
+                ~ "from the cluster size '" ~ size ~ "'. A burst to the same "
+                ~ "size is a no-op."
+            ) }}
+        {% endif %}
+    {% endif %}
+{% endmacro %}
+
+{#
+Renders an `auto_scaling_strategy` mapping as the corresponding
+`AUTO SCALING STRATEGY = (...)` clause. An empty mapping renders the
+disabling form `AUTO SCALING STRATEGY = ()`.
+#}
+{% macro internal_render_auto_scaling_strategy(auto_scaling_strategy) %}
+    {% if not auto_scaling_strategy %}
+        {{ return("AUTO SCALING STRATEGY = ()") }}
+    {% endif %}
+    {% set on_hydration = auto_scaling_strategy.get('on_hydration') %}
+    {% set parts = ["HYDRATION SIZE = " ~ dbt.string_literal(on_hydration.get('hydration_size'))] %}
+    {% if on_hydration.get('linger_duration') is not none %}
+        {% do parts.append("LINGER DURATION = " ~ dbt.string_literal(on_hydration.get('linger_duration'))) %}
+    {% endif %}
+    {{ return("AUTO SCALING STRATEGY = (ON HYDRATION (" ~ parts | join(", ") ~ "))") }}
+{% endmacro %}
+
+{#
+Reads the configured autoscaling strategy of a cluster from
+mz_internal.mz_cluster_auto_scaling_strategies and returns it as an
+`auto_scaling_strategy` mapping, or none when the cluster has no strategy
+configured (or the server predates the feature).
+
+NOTE: the catalog stores the linger duration as a {secs, nanos} object. The
+read-back keeps whole seconds only, so sub-second linger components are
+dropped when the strategy is copied to another cluster.
+
+NOTE: mz_cluster_auto_scaling_strategies is a builtin materialized view
+maintained on mz_catalog_server, so it reflects DDL with a small delay,
+unlike mz_clusters. A strategy configured immediately before this read may
+not be visible yet.
+#}
+{% macro internal_get_cluster_auto_scaling_strategy(cluster_name) %}
+    {% set catalog_relation_exists %}
+        SELECT count(*) > 0
+        FROM mz_objects o
+        JOIN mz_schemas s ON o.schema_id = s.id
+        WHERE s.name = 'mz_internal'
+            AND o.name = 'mz_cluster_auto_scaling_strategies'
+    {% endset %}
+
+    {% set exists_result = run_query(catalog_relation_exists) %}
+    {% if not execute or not exists_result.rows[0][0] %}
+        {{ return(none) }}
+    {% endif %}
+
+    {% set strategy_query %}
+        SELECT
+            s.strategy->'on_hydration'->>'hydration_size' AS hydration_size,
+            (s.strategy->'on_hydration'->'linger_duration'->>'secs')::bigint AS linger_secs
+        FROM mz_internal.mz_cluster_auto_scaling_strategies s
+        JOIN mz_clusters c ON c.id = s.cluster_id
+        WHERE c.name = {{ dbt.string_literal(cluster_name) }}
+    {% endset %}
+
+    {% set strategy_result = run_query(strategy_query) %}
+    {# A row can exist with a null strategy while a burst from a just-removed
+       policy is still draining, so also require a hydration size. #}
+    {% if strategy_result.rows and strategy_result.rows[0][0] is not none %}
+        {% set on_hydration = {'hydration_size': strategy_result.rows[0][0]} %}
+        {% if strategy_result.rows[0][1] is not none %}
+            {% do on_hydration.update({'linger_duration': strategy_result.rows[0][1] ~ 's'}) %}
+        {% endif %}
+        {{ return({'on_hydration': on_hydration}) }}
+    {% endif %}
+
+    {{ return(none) }}
+{% endmacro %}