feat: added snowflake warehouse vars config

.changes/unreleased/Features-20250103-164753.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Add support for overriding Snowflake warehouse via dbt vars
+time: 2025-01-03T16:47:53.137762Z
+custom:
+    Author: christopher-turnbull
+    Issue: "1280"

dbt/adapters/snowflake/impl.py

@@ -139,9 +139,21 @@ def _use_warehouse(self, warehouse: str):
 
     def pre_model_hook(self, config: Mapping[str, Any]) -> Optional[str]:
         default_warehouse = self.config.credentials.warehouse
-        warehouse = config.get("snowflake_warehouse", default_warehouse)
+
+        # Get warehouse from vars or config
+        warehouse = None
+
+        # Check command line vars first (stored in config.cli_vars)
+        if hasattr(self.config, 'cli_vars'):
+            warehouse = self.config.cli_vars.get('snowflake_warehouse')
+
+        # Then check config
+        if not warehouse:
+            warehouse = config.get("snowflake_warehouse", default_warehouse)
+
         if warehouse == default_warehouse or warehouse is None:
             return None
+
         previous = self._get_warehouse()
         self._use_warehouse(warehouse)
         return previous

tests/functional/warehouse_test/test_warehouses.py

@@ -133,3 +133,79 @@ def project_config_update(self):
     def test_snowflake_warehouse_valid(self, project):
         result = run_dbt(["run", "--models", "valid_warehouse"])
         assert "DBT_TESTING" in result[0].node.config.get("snowflake_warehouse")
+
+
+class TestWarehouseVarsOverride:
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "test_model.sql": """
+                {{ config(materialized='table') }}
+                select 1 as id
+            """,
+            "schema.yml": """
+version: 2
+models:
+  - name: test_model
+    tests:
+      - unique:
+          column_name: id
+"""
+        }
+
+    def test_snowflake_vars_override_ok_run(self, project):
+        # Test dbt run with warehouse in vars
+        result = run_dbt([
+            "run",
+            "--vars",
+            "{snowflake_warehouse: DBT_TESTING_ALT}",
+            "--models",
+            "test_model"
+        ])
+        assert len(result) == 1
+        assert result[0].status == "success"
+
+    def test_snowflake_vars_override_ok_test(self, project):
+        # First run the model
+        run_dbt([
+            "run",
+            "--vars",
+            "{snowflake_warehouse: DBT_TESTING_ALT}",
+            "--models",
+            "test_model"
+        ])
+
+        # Then test it
+        result = run_dbt([
+            "test",
+            "--vars",
+            "{snowflake_warehouse: DBT_TESTING_ALT}",
+            "--models",
+            "test_model"
+        ])
+        assert len(result) == 1
+        assert result[0].status == "pass"
+
+    def test_snowflake_vars_override_invalid(self, project):
+        # Test with non-existent warehouse
+        result = run_dbt([
+            "run",
+            "--vars",
+            "{snowflake_warehouse: DOES_NOT_EXIST}",
+            "--models",
+            "test_model"
+        ], expect_pass=False)
+        assert "Object does not exist, or operation cannot be performed" in result[0].message
+
+    def test_snowflake_vars_override_precedence(self, project):
+        # Test that vars override project config
+        result = run_dbt([
+            "run",
+            "--vars",
+            "{snowflake_warehouse: DBT_TESTING_ALT}",
+            "--models",
+            "test_model"
+        ])
+        assert len(result) == 1
+        assert result[0].status == "success"
+        # Could add additional verification that DBT_TESTING_ALT was used instead of DBT_TESTING