Fix: [SNOW-2193494] Use name part of FQN for DBT temporary stage creation

Author: sfc-gh-jwilkowski

RELEASE-NOTES.md

@@ -21,6 +21,7 @@
 ## New additions
 
 ## Fixes and improvements
+* Fixed DBT deploy command to properly handle fully qualified names
 
 
 # v3.10.0

src/snowflake/cli/_plugins/dbt/manager.py

@@ -44,7 +44,7 @@ def exists(name: FQN) -> bool:
 
     def deploy(
         self,
-        name: FQN,
+        fqn: FQN,
         path: SecurePath,
         profiles_path: SecurePath,
         force: bool,
@@ -66,7 +66,7 @@ def deploy(
 
         with cli_console.phase("Creating temporary stage"):
             stage_manager = StageManager()
-            stage_fqn = FQN.from_string(f"dbt_{name}_stage").using_context()
+            stage_fqn = FQN.from_string(f"dbt_{fqn.name}_stage").using_context()
             stage_name = stage_manager.get_standard_stage_prefix(stage_fqn)
             stage_manager.create(stage_fqn, temporary=True)
 
@@ -86,11 +86,11 @@ def deploy(
 
         with cli_console.phase("Creating DBT project"):
             if force is True:
-                query = f"CREATE OR REPLACE DBT PROJECT {name}"
-            elif self.exists(name=name):
-                query = f"ALTER DBT PROJECT {name} ADD VERSION"
+                query = f"CREATE OR REPLACE DBT PROJECT {fqn}"
+            elif self.exists(name=fqn):
+                query = f"ALTER DBT PROJECT {fqn} ADD VERSION"
             else:
-                query = f"CREATE DBT PROJECT {name}"
+                query = f"CREATE DBT PROJECT {fqn}"
             query += f"\nFROM {stage_name}"
             return self.execute_query(query)
 
@@ -174,7 +174,7 @@ def _prepare_profiles_file(profiles_path: Path, tmp_path: Path):
             yaml.safe_dump(yaml.safe_load(sfd), tfd)
 
     def execute(
-        self, dbt_command: str, name: str, run_async: bool, *dbt_cli_args
+        self, dbt_command: str, name: FQN, run_async: bool, *dbt_cli_args
     ) -> SnowflakeCursor:
         if dbt_cli_args:
             dbt_command = " ".join([dbt_command, *dbt_cli_args]).strip()

tests/dbt/test_dbt_commands.py

@@ -213,6 +213,38 @@ def test_dbt_deploy_with_custom_profiles_dir(
         assert result.exit_code == 0, result.output
         mock_put_recursive.assert_called_once()
 
+    @mock.patch("snowflake.cli._plugins.dbt.manager.StageManager.put_recursive")
+    @mock.patch("snowflake.cli._plugins.dbt.manager.StageManager.create")
+    def test_deploys_project_with_fqn_uses_name_only_for_stage(
+        self,
+        mock_create,
+        mock_put_recursive,
+        mock_connect,
+        runner,
+        dbt_project_path,
+        mock_exists,
+    ):
+
+        result = runner.invoke(
+            [
+                "dbt",
+                "deploy",
+                "MockDatabase.MockSchema.test_dbt_project",
+                f"--source={dbt_project_path}",
+            ]
+        )
+
+        assert result.exit_code == 0, result.output
+        assert (
+            mock_connect.mocked_ctx.get_query()
+            == """CREATE DBT PROJECT MockDatabase.MockSchema.test_dbt_project
+FROM @MockDatabase.MockSchema.dbt_test_dbt_project_stage"""
+        )
+        # Verify stage creation uses only the name part of the FQN
+        stage_fqn = FQN.from_string(f"dbt_test_dbt_project_stage").using_context()
+        mock_create.assert_called_once_with(stage_fqn, temporary=True)
+        mock_put_recursive.assert_called_once()
+
     def test_raises_when_dbt_project_yml_is_not_available(
         self, dbt_project_path, mock_connect, runner
     ):
@@ -347,6 +379,16 @@ class TestDBTExecute:
                 "EXECUTE DBT PROJECT pipeline_name args='compile'",
                 id="with-cli-flag",
             ),
+            pytest.param(
+                [
+                    "dbt",
+                    "execute",
+                    "database.schema.pipeline_name",
+                    "run",
+                ],
+                "EXECUTE DBT PROJECT database.schema.pipeline_name args='run'",
+                id="with-fqn",
+            ),
         ],
     )
     def test_dbt_execute(self, mock_connect, mock_cursor, runner, args, expected_query):

tests_integration/test_dbt.py

@@ -18,6 +18,7 @@
 import pytest
 import yaml
 
+from snowflake.cli.api.identifiers import FQN
 from snowflake.cli._plugins.dbt.constants import PROFILES_FILENAME
 
 
@@ -97,6 +98,51 @@ def test_deploy_and_execute(
         assert result.json[0]["COUNT"] == 1, result.json[0]
 
 
+@pytest.mark.integration
+@pytest.mark.qa_only
+def test_deploy_and_execute_with_full_fqn(
+    runner,
+    snowflake_session,
+    test_database,
+    project_directory,
+    snapshot,
+):
+    with project_directory("dbt_project") as root_dir:
+        # Given a local dbt project
+        ts = int(datetime.datetime.now().timestamp())
+        name = f"dbt_project_{ts}"
+        fqn = FQN.from_string(
+            f"{snowflake_session.database}.{snowflake_session.schema}.{name}"
+        )
+
+        _setup_dbt_profile(root_dir, snowflake_session, include_password=False)
+        result = runner.invoke_with_connection_json(["dbt", "deploy", str(fqn)])
+        assert result.exit_code == 0, result.output
+
+        # call `run` on dbt object
+        result = runner.invoke_passthrough_with_connection(
+            args=[
+                "dbt",
+                "execute",
+            ],
+            passthrough_args=[str(fqn), "run"],
+        )
+
+        # a successful execution should produce data in my_second_dbt_model and
+        assert result.exit_code == 0, result.output
+        assert "Done. PASS=2 WARN=0 ERROR=0 SKIP=0 TOTAL=2" in result.output
+
+        result = runner.invoke_with_connection_json(
+            [
+                "sql",
+                "-q",
+                f"select count(*) as COUNT from {fqn.database}.{fqn.schema}.my_second_dbt_model;",
+            ]
+        )
+        assert len(result.json) == 1, result.json
+        assert result.json[0]["COUNT"] == 1, result.json[0]
+
+
 @pytest.mark.integration
 @pytest.mark.qa_only
 def test_dbt_deploy_options(