dbt-adapter: normalize cluster name generation in deploy operation (#31086)

<!--
Describe the contents of the PR briefly but completely.

If you write detailed commit messages, it is acceptable to copy/paste
them
here, or write "see commit messages for details." If there is only one
commit
in the PR, GitHub will have already added its commit message above.
-->

### Motivation

Updating the dbt blue-green deployment actions so that the
`adapter.generate_final_cluster_name` is used consistently when checking
source clusters to respect customer-defined cluster naming logic using
`generate_cluster_name` in their projects.

This allows us to use a correct cluster validation when users override
the `generate_cluster_name` macro.

<!--
Which of the following best describes the motivation behind this PR?

  * This PR fixes a recognized bug.

    [Ensure issue is linked somewhere.]

  * This PR adds a known-desirable feature.

    [Ensure issue is linked somewhere.]

  * This PR fixes a previously unreported bug.

    [Describe the bug in detail, as if you were filing a bug report.]

  * This PR adds a feature that has not yet been specified.

[Write a brief specification for the feature, including justification
for its inclusion in Materialize, as if you were writing the original
     feature specification.]

   * This PR refactors existing code.

[Describe what was wrong with the existing code, if it is not obvious.]
-->

### Tips for reviewer

<!--
Leave some tips for your reviewer, like:

    * The diff is much smaller if viewed with whitespace hidden.
    * [Some function/module/file] deserves extra attention.
* [Some function/module/file] is pure code movement and only needs a
skim.

Delete this section if no tips.
-->

### Checklist

- [ ] This PR has adequate test coverage / QA involvement has been duly
considered. ([trigger-ci for additional test/nightly
runs](https://trigger-ci.dev.materialize.com/))
- [ ] This PR has an associated up-to-date [design
doc](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/design/README.md),
is a design doc
([template](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/design/00000000_template.md)),
or is sufficiently small to not require a design.
  <!-- Reference the design in the description. -->
- [ ] If this PR evolves [an existing `$T ⇔ Proto$T`
mapping](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/command-and-response-binary-encoding.md)
(possibly in a backwards-incompatible way), then it is tagged with a
`T-proto` label.
- [ ] If this PR will require changes to cloud orchestration or tests,
there is a companion cloud PR to account for those changes that is
tagged with the release-blocker label
([example](MaterializeInc/cloud#5021)).
<!-- Ask in #team-cloud on Slack if you need help preparing the cloud
PR. -->
- [ ] If this PR includes major [user-facing behavior
changes](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/guide-changes.md#what-changes-require-a-release-note),
I have pinged the relevant PM to schedule a changelog post.

Author: bobbyiliev

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

@@ -44,12 +44,13 @@
 {% endfor %}
 
 {% for cluster in clusters %}
-    {% if not cluster_exists(cluster) %}
-        {{ exceptions.raise_compiler_error("Production cluster " ~ cluster ~ " does not exist") }}
+    {% set origin_cluster = adapter.generate_final_cluster_name(cluster, force_deploy_suffix=False) %}
+    {% if not cluster_exists(origin_cluster) %}
+        {{ exceptions.raise_compiler_error("Production cluster " ~ origin_cluster ~ " does not exist") }}
     {% endif %}
-    {% if cluster_contains_sinks(cluster) %}
+    {% if cluster_contains_sinks(origin_cluster) %}
         {{ exceptions.raise_compiler_error("""
-        Production cluster " ~ cluster ~ " contains sinks.
+        Production cluster " ~ origin_cluster ~ " contains sinks.
         Blue/green deployments require sinks to be in a dedicated cluster.
         """) }}
     {% endif %}
@@ -107,6 +108,7 @@
 {% endfor %}
 
 {% for cluster in clusters %}
+    {% set origin_cluster = adapter.generate_final_cluster_name(cluster, force_deploy_suffix=False) %}
     {% set cluster_configuration %}
         SELECT
             c.managed,
@@ -118,7 +120,7 @@
             cs.refresh_hydration_time_estimate
         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) }}
+        WHERE c.name = {{ dbt.string_literal(origin_cluster) }}
     {% endset %}
 
     {% set cluster_config_results = run_query(cluster_configuration) %}
@@ -133,7 +135,7 @@
         {% set refresh_hydration_time_estimate = results[6] %}
 
         {% if not managed %}
-            {{ exceptions.raise_compiler_error("Production cluster " ~ cluster ~ " is not managed") }}
+            {{ exceptions.raise_compiler_error("Production cluster " ~ origin_cluster ~ " is not managed") }}
         {% endif %}
 
         {% set deploy_cluster = create_cluster(

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

@@ -89,7 +89,8 @@ WITH clusters_under_deployment AS (
     WHERE name IN (
     {% if clusters|length > 0 %}
         {% for cluster in clusters %}
-            {{ dbt.string_literal(cluster) }}{% if not loop.last %},{% endif %}
+            {% set origin_cluster = adapter.generate_final_cluster_name(cluster, force_deploy_suffix=False) %}
+            {{ dbt.string_literal(origin_cluster) }}{% if not loop.last %},{% endif %}
         {% endfor %}
     {% else %}
         NULL

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

@@ -67,8 +67,9 @@
 
 {% for cluster in clusters %}
     {% set deploy_cluster = adapter.generate_final_cluster_name(cluster, force_deploy_suffix=True) %}
-    {% if not cluster_exists(cluster) %}
-        {{ exceptions.raise_compiler_error("Production cluster " ~ cluster ~ " does not exist") }}
+    {% set origin_cluster = adapter.generate_final_cluster_name(cluster, force_deploy_suffix=False) %}
+    {% if not cluster_exists(origin_cluster) %}
+        {{ exceptions.raise_compiler_error("Production cluster " ~ origin_cluster ~ " does not exist") }}
     {% endif %}
     {% if not cluster_exists(deploy_cluster) %}
         {{ exceptions.raise_compiler_error("Deployment cluster " ~ deploy_cluster ~ " does not exist") }}
@@ -91,8 +92,9 @@
 
     {% for cluster in clusters %}
         {% set deploy_cluster = adapter.generate_final_cluster_name(cluster, force_deploy_suffix=True) %}
-        {{ log("Swapping clusters " ~ adapter.generate_final_cluster_name(cluster) ~ " and " ~ deploy_cluster, info=True) }}
-        ALTER CLUSTER {{ adapter.quote(cluster) }} SWAP WITH {{ adapter.quote(deploy_cluster) }};
+        {% set origin_cluster = adapter.generate_final_cluster_name(cluster, force_deploy_suffix=False) %}
+        {{ log("Swapping clusters " ~ origin_cluster ~ " and " ~ deploy_cluster, info=True) }}
+        ALTER CLUSTER {{ adapter.quote(origin_cluster) }} SWAP WITH {{ adapter.quote(deploy_cluster) }};
     {% endfor %}
 
     COMMIT;
@@ -108,8 +110,9 @@
 
     {% for cluster in clusters %}
         {% set deploy_cluster = adapter.generate_final_cluster_name(cluster, force_deploy_suffix=True) %}
+        {% set origin_cluster = adapter.generate_final_cluster_name(cluster, force_deploy_suffix=False) %}
         {{ log("DRY RUN: Swapping clusters " ~ adapter.generate_final_cluster_name(cluster) ~ " and " ~ deploy_cluster, info=True) }}
-        {{ log("DRY RUN: ALTER CLUSTER " ~ adapter.quote(cluster) ~ " SWAP WITH " ~ adapter.quote(deploy_cluster), info=True) }}
+        {{ log("DRY RUN: ALTER CLUSTER " ~ adapter.quote(origin_cluster) ~ " SWAP WITH " ~ adapter.quote(deploy_cluster), info=True) }}
     {% endfor %}
     {{ log("Dry run completed. The statements above were **not** executed against Materialize.", info=True) }}
 {% endif %}