From 55ef572cfdbb65b227f3e8959906c778676643bb Mon Sep 17 00:00:00 2001 From: bobbyiliev Date: Wed, 22 Jul 2026 12:42:20 +0300 Subject: [PATCH 1/5] dbt: Add cluster AUTO SCALING STRATEGY (hydration burst) support --- .../manage/dbt/blue-green-deployments.md | 11 ++ misc/dbt-materialize/CHANGELOG.md | 15 ++ .../alter_cluster_auto_scaling_strategy.sql | 87 +++++++++ .../materialize/macros/ci/create_cluster.sql | 14 ++ .../materialize/macros/deploy/deploy_init.sql | 5 + .../macros/utils/auto_scaling_strategy.sql | 154 ++++++++++++++++ misc/dbt-materialize/tests/adapter/test_ci.py | 174 ++++++++++++++++++ .../tests/adapter/test_deploy.py | 46 +++++ 8 files changed, 506 insertions(+) create mode 100644 misc/dbt-materialize/dbt/include/materialize/macros/ci/alter_cluster_auto_scaling_strategy.sql create mode 100644 misc/dbt-materialize/dbt/include/materialize/macros/utils/auto_scaling_strategy.sql diff --git a/doc/user/content/manage/dbt/blue-green-deployments.md b/doc/user/content/manage/dbt/blue-green-deployments.md index d0fc655ba3c32..045f8ff9df09c 100644 --- a/doc/user/content/manage/dbt/blue-green-deployments.md +++ b/doc/user/content/manage/dbt/blue-green-deployments.md @@ -78,6 +78,17 @@ These environments are later swapped transparently. schema named `_dbt_deploy` using the same configuration as the current environment to swap with (including privileges). + If the production cluster has an [autoscaling strategy](/sql/create-cluster/#autoscaling) + configured, the deployment cluster inherits it, so the deployment + environment temporarily bursts to the configured hydration size while it + hydrates ahead of the cutover. To speed up blue/green deployments of a + cluster without a strategy, you can configure one using the + `alter_cluster_auto_scaling_strategy` operation before running `deploy_init`: + + ```bash + dbt run-operation alter_cluster_auto_scaling_strategy --args '{cluster_name: , auto_scaling_strategy: {on_hydration: {hydration_size: ""}}}' + ``` + 1. Run the dbt project containing the code changes against the new deployment environment. diff --git a/misc/dbt-materialize/CHANGELOG.md b/misc/dbt-materialize/CHANGELOG.md index ca2ed7cee124d..4c7a6cfae801b 100644 --- a/misc/dbt-materialize/CHANGELOG.md +++ b/misc/dbt-materialize/CHANGELOG.md @@ -2,6 +2,21 @@ ## 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 hydrates faster ahead of the cutover. + ## 1.9.10 - 2026-05-20 * Support unmanaged clusters in `deploy_init`. The deployment cluster diff --git a/misc/dbt-materialize/dbt/include/materialize/macros/ci/alter_cluster_auto_scaling_strategy.sql b/misc/dbt-materialize/dbt/include/materialize/macros/ci/alter_cluster_auto_scaling_strategy.sql new file mode 100644 index 0000000000000..bfa1c721fff1e --- /dev/null +++ b/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 %} diff --git a/misc/dbt-materialize/dbt/include/materialize/macros/ci/create_cluster.sql b/misc/dbt-materialize/dbt/include/materialize/macros/ci/create_cluster.sql index 2b9c2fd9cafd2..e7a5770b774e4 100644 --- a/misc/dbt-materialize/dbt/include/materialize/macros/ci/create_cluster.sql +++ b/misc/dbt-materialize/dbt/include/materialize/macros/ci/create_cluster.sql @@ -22,12 +22,18 @@ This macro creates a cluster with the specified properties. - replication_factor (int, optional): The replication factor for the cluster. Only applicable when schedule_type is 'manual'. - schedule_type (str, optional): The type of schedule for the cluster. Accepts 'manual' or 'on-refresh'. - refresh_hydration_time_estimate (str, optional): The estimated hydration time for the cluster. Only applicable when schedule_type is 'on-refresh'. + - 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'. - 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. 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. #} {% macro create_cluster( cluster_name, @@ -35,6 +41,7 @@ This macro creates a cluster with the specified properties. replication_factor=none, schedule_type=none, refresh_hydration_time_estimate=none, + auto_scaling_strategy=none, ignore_existing_objects=false, force_deploy_suffix=false ) %} @@ -48,6 +55,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 +77,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) }} diff --git a/misc/dbt-materialize/dbt/include/materialize/macros/deploy/deploy_init.sql b/misc/dbt-materialize/dbt/include/materialize/macros/deploy/deploy_init.sql index 90d426c7d5ff5..d2df2a12c5769 100644 --- a/misc/dbt-materialize/dbt/include/materialize/macros/deploy/deploy_init.sql +++ b/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 ) %} diff --git a/misc/dbt-materialize/dbt/include/materialize/macros/utils/auto_scaling_strategy.sql b/misc/dbt-materialize/dbt/include/materialize/macros/utils/auto_scaling_strategy.sql new file mode 100644 index 0000000000000..1f9c3f4e6f56f --- /dev/null +++ b/misc/dbt-materialize/dbt/include/materialize/macros/utils/auto_scaling_strategy.sql @@ -0,0 +1,154 @@ +-- 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. +#} +{% 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 %} diff --git a/misc/dbt-materialize/tests/adapter/test_ci.py b/misc/dbt-materialize/tests/adapter/test_ci.py index b480c52b84120..0c71da189f954 100644 --- a/misc/dbt-materialize/tests/adapter/test_ci.py +++ b/misc/dbt-materialize/tests/adapter/test_ci.py @@ -25,6 +25,9 @@ def cleanup(self, project): project.run_sql("DROP CLUSTER IF EXISTS test_cluster_dbt_deploy CASCADE") project.run_sql("DROP CLUSTER IF EXISTS test_manual_schedule CASCADE") project.run_sql("DROP CLUSTER IF EXISTS test_on_refresh_schedule CASCADE") + project.run_sql("DROP CLUSTER IF EXISTS test_autoscaling CASCADE") + project.run_sql("DROP CLUSTER IF EXISTS test_autoscaling_dbt_deploy CASCADE") + project.run_sql("DROP CLUSTER IF EXISTS test_autoscaling_unmanaged CASCADE") def test_create_and_drop_cluster(self, project): # Test creating a cluster @@ -280,6 +283,158 @@ def test_create_cluster_without_force_deploy_suffix(self, project): ] ) + def test_create_cluster_with_auto_scaling_strategy(self, project): + run_dbt( + [ + "run-operation", + "create_cluster", + "--args", + '{"cluster_name": "test_autoscaling", "size": "scale=1,workers=1", "auto_scaling_strategy": {"on_hydration": {"hydration_size": "scale=1,workers=2", "linger_duration": "15s"}}, "force_deploy_suffix": false}', + ] + ) + + strategy = get_cluster_auto_scaling_strategy(project, "test_autoscaling") + assert strategy is not None, "Autoscaling strategy was not configured" + assert strategy[0] == "scale=1,workers=2" + assert strategy[1] == 15 + + def test_create_cluster_with_auto_scaling_strategy_no_linger(self, project): + run_dbt( + [ + "run-operation", + "create_cluster", + "--args", + '{"cluster_name": "test_autoscaling", "size": "scale=1,workers=1", "auto_scaling_strategy": {"on_hydration": {"hydration_size": "scale=1,workers=2"}}, "force_deploy_suffix": false}', + ] + ) + + strategy = get_cluster_auto_scaling_strategy(project, "test_autoscaling") + assert strategy is not None, "Autoscaling strategy was not configured" + assert strategy[0] == "scale=1,workers=2" + + def test_create_cluster_auto_scaling_strategy_validation(self, project): + # Hydration size equal to the cluster size is a no-op burst + run_dbt( + [ + "run-operation", + "create_cluster", + "--args", + '{"cluster_name": "test_autoscaling", "size": "scale=1,workers=1", "auto_scaling_strategy": {"on_hydration": {"hydration_size": "scale=1,workers=1"}}}', + ], + expect_pass=False, + ) + + # Missing hydration_size + run_dbt( + [ + "run-operation", + "create_cluster", + "--args", + '{"cluster_name": "test_autoscaling", "size": "scale=1,workers=1", "auto_scaling_strategy": {"on_hydration": {"linger_duration": "15s"}}}', + ], + expect_pass=False, + ) + + # Unknown sub-policy + run_dbt( + [ + "run-operation", + "create_cluster", + "--args", + '{"cluster_name": "test_autoscaling", "size": "scale=1,workers=1", "auto_scaling_strategy": {"on_demand": {"hydration_size": "scale=1,workers=2"}}}', + ], + expect_pass=False, + ) + + # Cannot be combined with a non-manual schedule + run_dbt( + [ + "run-operation", + "create_cluster", + "--args", + '{"cluster_name": "test_autoscaling", "size": "scale=1,workers=1", "schedule_type": "on-refresh", "auto_scaling_strategy": {"on_hydration": {"hydration_size": "scale=1,workers=2"}}}', + ], + expect_pass=False, + ) + + def test_alter_cluster_auto_scaling_strategy(self, project): + project.run_sql("CREATE CLUSTER test_autoscaling SIZE = 'scale=1,workers=1'") + + # Set a strategy on an existing cluster + run_dbt( + [ + "run-operation", + "alter_cluster_auto_scaling_strategy", + "--args", + '{"cluster_name": "test_autoscaling", "auto_scaling_strategy": {"on_hydration": {"hydration_size": "scale=1,workers=2", "linger_duration": "15s"}}}', + ] + ) + + strategy = get_cluster_auto_scaling_strategy(project, "test_autoscaling") + assert strategy is not None, "Autoscaling strategy was not configured" + assert strategy[0] == "scale=1,workers=2" + assert strategy[1] == 15 + + # Remove the strategy via RESET + run_dbt( + [ + "run-operation", + "alter_cluster_auto_scaling_strategy", + "--args", + '{"cluster_name": "test_autoscaling"}', + ] + ) + + strategy = get_cluster_auto_scaling_strategy(project, "test_autoscaling") + assert strategy is None, "Autoscaling strategy was not removed" + + # Disable via an empty strategy (AUTO SCALING STRATEGY = ()) + run_dbt( + [ + "run-operation", + "alter_cluster_auto_scaling_strategy", + "--args", + '{"cluster_name": "test_autoscaling", "auto_scaling_strategy": {"on_hydration": {"hydration_size": "scale=1,workers=2"}}}', + ] + ) + run_dbt( + [ + "run-operation", + "alter_cluster_auto_scaling_strategy", + "--args", + '{"cluster_name": "test_autoscaling", "auto_scaling_strategy": {}}', + ] + ) + + strategy = get_cluster_auto_scaling_strategy(project, "test_autoscaling") + assert strategy is None, "Autoscaling strategy was not disabled" + + def test_alter_cluster_auto_scaling_strategy_errors(self, project): + # Nonexistent cluster + run_dbt( + [ + "run-operation", + "alter_cluster_auto_scaling_strategy", + "--args", + '{"cluster_name": "test_autoscaling", "auto_scaling_strategy": {"on_hydration": {"hydration_size": "scale=1,workers=2"}}}', + ], + expect_pass=False, + ) + + # Unmanaged cluster + project.run_sql( + "CREATE CLUSTER test_autoscaling_unmanaged REPLICAS (r1 (SIZE 'scale=1,workers=1'))" + ) + run_dbt( + [ + "run-operation", + "alter_cluster_auto_scaling_strategy", + "--args", + '{"cluster_name": "test_autoscaling_unmanaged", "auto_scaling_strategy": {"on_hydration": {"hydration_size": "scale=1,workers=2"}}}', + ], + expect_pass=False, + ) + def test_create_cluster_without_size_and_name(self, project): # Test creating a cluster without providing a size parameter run_dbt( @@ -315,6 +470,25 @@ def test_create_cluster_without_size_and_name(self, project): ) +def get_cluster_auto_scaling_strategy(project, cluster_name): + """Return (hydration_size, linger_secs) for the cluster's configured + autoscaling strategy, or None when no strategy is configured.""" + query = f""" + 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 = '{cluster_name}' + """ + result = project.run_sql(query, fetch="one") + # A row with a null hydration size can linger while a burst from a + # just-removed policy drains. + if result is None or result[0] is None: + return None + return result + + def get_cluster_properties(project, cluster_name): query = f""" SELECT diff --git a/misc/dbt-materialize/tests/adapter/test_deploy.py b/misc/dbt-materialize/tests/adapter/test_deploy.py index 6b874fbeb2169..7ee763600f1b6 100644 --- a/misc/dbt-materialize/tests/adapter/test_deploy.py +++ b/misc/dbt-materialize/tests/adapter/test_deploy.py @@ -581,6 +581,52 @@ def test_dbt_deploy_missing_deployment_schema(self, project): run_dbt(["run-operation", "deploy_promote"], expect_pass=False) + def test_dbt_deploy_init_inherits_auto_scaling_strategy(self, project): + try: + project.run_sql( + "CREATE CLUSTER prod (" + "SIZE = 'scale=1,workers=1', " + "AUTO SCALING STRATEGY = (ON HYDRATION (" + "HYDRATION SIZE = 'scale=1,workers=2', LINGER DURATION = '15s')))" + ) + except psycopg2.Error as e: + pytest.skip(f"local Materialize image rejects AUTO SCALING STRATEGY: {e}") + project.run_sql("CREATE SCHEMA prod") + project.run_sql("CREATE SCHEMA staging") + + run_dbt(["run-operation", "deploy_init"]) + + strategy = project.run_sql( + """ + SELECT + s.strategy->'on_hydration'->>'hydration_size', + (s.strategy->'on_hydration'->'linger_duration'->>'secs')::bigint + FROM mz_internal.mz_cluster_auto_scaling_strategies s + JOIN mz_clusters c ON c.id = s.cluster_id + WHERE c.name = 'prod_dbt_deploy' + """, + fetch="one", + ) + assert strategy == ("scale=1,workers=2", 15) + + def test_dbt_deploy_init_without_auto_scaling_strategy(self, project): + project.run_sql("CREATE CLUSTER prod SIZE = 'scale=1,workers=1'") + project.run_sql("CREATE SCHEMA prod") + project.run_sql("CREATE SCHEMA staging") + + run_dbt(["run-operation", "deploy_init"]) + + strategy_count = project.run_sql( + """ + SELECT count(*) + FROM mz_internal.mz_cluster_auto_scaling_strategies s + JOIN mz_clusters c ON c.id = s.cluster_id + WHERE c.name = 'prod_dbt_deploy' + """, + fetch="one", + ) + assert strategy_count == (0,) + def test_dbt_deploy_init_unmanaged_empty(self, project): project.run_sql("CREATE CLUSTER prod REPLICAS ()") project.run_sql("CREATE SCHEMA prod") From eeafd6f1ca5498a3ee3def172b871b64c9fa4f44 Mon Sep 17 00:00:00 2001 From: bobbyiliev Date: Wed, 22 Jul 2026 12:52:18 +0300 Subject: [PATCH 2/5] dbt: Extend auto scaling strategy test coverage and note catalog read-back lag --- .../macros/utils/auto_scaling_strategy.sql | 5 + misc/dbt-materialize/tests/adapter/test_ci.py | 29 +++++ .../tests/adapter/test_deploy.py | 106 +++++++++++++++++- 3 files changed, 135 insertions(+), 5 deletions(-) diff --git a/misc/dbt-materialize/dbt/include/materialize/macros/utils/auto_scaling_strategy.sql b/misc/dbt-materialize/dbt/include/materialize/macros/utils/auto_scaling_strategy.sql index 1f9c3f4e6f56f..749483d39bce8 100644 --- a/misc/dbt-materialize/dbt/include/materialize/macros/utils/auto_scaling_strategy.sql +++ b/misc/dbt-materialize/dbt/include/materialize/macros/utils/auto_scaling_strategy.sql @@ -115,6 +115,11 @@ 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 %} diff --git a/misc/dbt-materialize/tests/adapter/test_ci.py b/misc/dbt-materialize/tests/adapter/test_ci.py index 0c71da189f954..4e758e1fce264 100644 --- a/misc/dbt-materialize/tests/adapter/test_ci.py +++ b/misc/dbt-materialize/tests/adapter/test_ci.py @@ -435,6 +435,35 @@ def test_alter_cluster_auto_scaling_strategy_errors(self, project): expect_pass=False, ) + def test_alter_cluster_auto_scaling_strategy_validation(self, project): + # Hydration size equal to the cluster's current size, validated against + # the size fetched from the catalog + project.run_sql("CREATE CLUSTER test_autoscaling SIZE = 'scale=1,workers=1'") + run_dbt( + [ + "run-operation", + "alter_cluster_auto_scaling_strategy", + "--args", + '{"cluster_name": "test_autoscaling", "auto_scaling_strategy": {"on_hydration": {"hydration_size": "scale=1,workers=1"}}}', + ], + expect_pass=False, + ) + + # Cluster with a non-manual schedule, validated against the schedule + # fetched from the catalog + project.run_sql( + "CREATE CLUSTER test_on_refresh_schedule (SIZE = 'scale=1,workers=1', SCHEDULE = ON REFRESH (HYDRATION TIME ESTIMATE = '1 hour'))" + ) + run_dbt( + [ + "run-operation", + "alter_cluster_auto_scaling_strategy", + "--args", + '{"cluster_name": "test_on_refresh_schedule", "auto_scaling_strategy": {"on_hydration": {"hydration_size": "scale=1,workers=2"}}}', + ], + expect_pass=False, + ) + def test_create_cluster_without_size_and_name(self, project): # Test creating a cluster without providing a size parameter run_dbt( diff --git a/misc/dbt-materialize/tests/adapter/test_deploy.py b/misc/dbt-materialize/tests/adapter/test_deploy.py index 7ee763600f1b6..e37ef3a2e1fa5 100644 --- a/misc/dbt-materialize/tests/adapter/test_deploy.py +++ b/misc/dbt-materialize/tests/adapter/test_deploy.py @@ -31,6 +31,26 @@ ) +def get_deploy_cluster_auto_scaling_strategy(project): + """Return (hydration_size, linger_secs) for the deploy cluster's + configured autoscaling strategy, or None when no strategy is + configured.""" + result = project.run_sql( + """ + SELECT + s.strategy->'on_hydration'->>'hydration_size', + (s.strategy->'on_hydration'->'linger_duration'->>'secs')::bigint + FROM mz_internal.mz_cluster_auto_scaling_strategies s + JOIN mz_clusters c ON c.id = s.cluster_id + WHERE c.name = 'prod_dbt_deploy' + """, + fetch="one", + ) + if result is None or result[0] is None: + return None + return result + + class TestApplyGrantsAndPrivileges: @pytest.fixture(autouse=True) def cleanup(self, project): @@ -596,18 +616,94 @@ def test_dbt_deploy_init_inherits_auto_scaling_strategy(self, project): run_dbt(["run-operation", "deploy_init"]) - strategy = project.run_sql( + strategy = get_deploy_cluster_auto_scaling_strategy(project) + assert strategy == ("scale=1,workers=2", 15) + + def test_dbt_deploy_init_auto_scaling_strategy_idempotent(self, project): + try: + project.run_sql( + "CREATE CLUSTER prod (" + "SIZE = 'scale=1,workers=1', " + "AUTO SCALING STRATEGY = (ON HYDRATION (" + "HYDRATION SIZE = 'scale=1,workers=2')))" + ) + except psycopg2.Error as e: + pytest.skip(f"local Materialize image rejects AUTO SCALING STRATEGY: {e}") + project.run_sql("CREATE SCHEMA prod") + project.run_sql("CREATE SCHEMA staging") + + run_dbt(["run-operation", "deploy_init"]) + # The second run takes the existing-deploy-cluster branch, where the + # inherited strategy is validated but the cluster is not recreated. + run_dbt(["run-operation", "deploy_init"]) + + strategy = get_deploy_cluster_auto_scaling_strategy(project) + assert strategy == ("scale=1,workers=2", None) + + def test_dbt_deploy_promote_keeps_auto_scaling_strategy(self, project): + try: + project.run_sql( + "CREATE CLUSTER prod (" + "SIZE = 'scale=1,workers=1', " + "AUTO SCALING STRATEGY = (ON HYDRATION (" + "HYDRATION SIZE = 'scale=1,workers=2', LINGER DURATION = '15s')))" + ) + except psycopg2.Error as e: + pytest.skip(f"local Materialize image rejects AUTO SCALING STRATEGY: {e}") + project.run_sql("CREATE SCHEMA prod") + project.run_sql("CREATE SCHEMA staging") + + run_dbt(["run-operation", "deploy_init"]) + + green_cluster_id = project.run_sql( + "SELECT id FROM mz_clusters WHERE name = 'prod_dbt_deploy'", + fetch="one", + )[0] + + run_dbt(["run-operation", "deploy_promote"]) + + # The strategy travels with the swapped cluster: the promoted (formerly + # green) cluster is now production and still carries it. + promoted = project.run_sql( """ SELECT + c.id, s.strategy->'on_hydration'->>'hydration_size', (s.strategy->'on_hydration'->'linger_duration'->>'secs')::bigint - FROM mz_internal.mz_cluster_auto_scaling_strategies s - JOIN mz_clusters c ON c.id = s.cluster_id - WHERE c.name = 'prod_dbt_deploy' + FROM mz_clusters c + JOIN mz_internal.mz_cluster_auto_scaling_strategies s ON s.cluster_id = c.id + WHERE c.name = 'prod' """, fetch="one", ) - assert strategy == ("scale=1,workers=2", 15) + assert promoted == (green_cluster_id, "scale=1,workers=2", 15) + + run_dbt(["run-operation", "deploy_cleanup"]) + + result = project.run_sql( + "SELECT count(*) = 0 FROM mz_clusters WHERE name = 'prod_dbt_deploy'", + fetch="one", + ) + assert bool(result[0]) + + def test_dbt_deploy_init_auto_scaling_subsecond_linger(self, project): + try: + project.run_sql( + "CREATE CLUSTER prod (" + "SIZE = 'scale=1,workers=1', " + "AUTO SCALING STRATEGY = (ON HYDRATION (" + "HYDRATION SIZE = 'scale=1,workers=2', LINGER DURATION = '1500ms')))" + ) + except psycopg2.Error as e: + pytest.skip(f"local Materialize image rejects AUTO SCALING STRATEGY: {e}") + project.run_sql("CREATE SCHEMA prod") + project.run_sql("CREATE SCHEMA staging") + + run_dbt(["run-operation", "deploy_init"]) + + # The read-back keeps whole seconds only, so 1500ms is inherited as 1s. + strategy = get_deploy_cluster_auto_scaling_strategy(project) + assert strategy == ("scale=1,workers=2", 1) def test_dbt_deploy_init_without_auto_scaling_strategy(self, project): project.run_sql("CREATE CLUSTER prod SIZE = 'scale=1,workers=1'") From a129529504c0d9c79c10250f54afe11cd07ddcf4 Mon Sep 17 00:00:00 2001 From: bobbyiliev Date: Wed, 22 Jul 2026 12:57:28 +0300 Subject: [PATCH 3/5] dbt: Move blue/green autoscaling docs to a separate PR --- doc/user/content/manage/dbt/blue-green-deployments.md | 11 ----------- 1 file changed, 11 deletions(-) diff --git a/doc/user/content/manage/dbt/blue-green-deployments.md b/doc/user/content/manage/dbt/blue-green-deployments.md index 045f8ff9df09c..d0fc655ba3c32 100644 --- a/doc/user/content/manage/dbt/blue-green-deployments.md +++ b/doc/user/content/manage/dbt/blue-green-deployments.md @@ -78,17 +78,6 @@ These environments are later swapped transparently. schema named `_dbt_deploy` using the same configuration as the current environment to swap with (including privileges). - If the production cluster has an [autoscaling strategy](/sql/create-cluster/#autoscaling) - configured, the deployment cluster inherits it, so the deployment - environment temporarily bursts to the configured hydration size while it - hydrates ahead of the cutover. To speed up blue/green deployments of a - cluster without a strategy, you can configure one using the - `alter_cluster_auto_scaling_strategy` operation before running `deploy_init`: - - ```bash - dbt run-operation alter_cluster_auto_scaling_strategy --args '{cluster_name: , auto_scaling_strategy: {on_hydration: {hydration_size: ""}}}' - ``` - 1. Run the dbt project containing the code changes against the new deployment environment. From 374c0a570254314d643cbdc4890e6748de31f03a Mon Sep 17 00:00:00 2001 From: bobbyiliev Date: Wed, 22 Jul 2026 13:14:14 +0300 Subject: [PATCH 4/5] dbt: Keep new create_cluster argument last for positional-caller compatibility --- .../include/materialize/macros/ci/create_cluster.sql | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/misc/dbt-materialize/dbt/include/materialize/macros/ci/create_cluster.sql b/misc/dbt-materialize/dbt/include/materialize/macros/ci/create_cluster.sql index e7a5770b774e4..a07180360dac5 100644 --- a/misc/dbt-materialize/dbt/include/materialize/macros/ci/create_cluster.sql +++ b/misc/dbt-materialize/dbt/include/materialize/macros/ci/create_cluster.sql @@ -22,12 +22,12 @@ This macro creates a cluster with the specified properties. - replication_factor (int, optional): The replication factor for the cluster. Only applicable when schedule_type is 'manual'. - schedule_type (str, optional): The type of schedule for the cluster. Accepts 'manual' or 'on-refresh'. - 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'. - - 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. Incompatibilities: - replication_factor is only applicable when schedule_type is 'manual'. @@ -35,15 +35,17 @@ This macro creates a cluster with the specified properties. - 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, - auto_scaling_strategy=none, ignore_existing_objects=false, - force_deploy_suffix=false + force_deploy_suffix=false, + auto_scaling_strategy=none ) %} {# Input validation #} From 38d9859cf2d412f8f8446e7bc009420a4fa814d6 Mon Sep 17 00:00:00 2001 From: bobbyiliev Date: Wed, 22 Jul 2026 13:36:03 +0300 Subject: [PATCH 5/5] dbt: State the verified burst behavior in the changelog instead of the inferred speedup --- misc/dbt-materialize/CHANGELOG.md | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/misc/dbt-materialize/CHANGELOG.md b/misc/dbt-materialize/CHANGELOG.md index 4c7a6cfae801b..af07e9b4e5fe2 100644 --- a/misc/dbt-materialize/CHANGELOG.md +++ b/misc/dbt-materialize/CHANGELOG.md @@ -15,7 +15,8 @@ During blue/green deployments, `deploy_init` now copies the production cluster's autoscaling strategy to the deployment cluster, so the deployment - environment hydrates faster ahead of the cutover. + environment bursts to the configured hydration size while it hydrates ahead + of the cutover. ## 1.9.10 - 2026-05-20