Feat: pass session parameters for Snowflake (#2912)

Author: alexischicoinedatadeveloper

docs/integrations/engines/snowflake.md

@@ -24,6 +24,7 @@ pip install "sqlmesh[snowflake]"
 | `private_key`            | The optional private key to use for authentication. Key can be Base64-encoded DER format (representing the key bytes), a plain-text PEM format, or bytes (Python config only). | string |    N     |
 | `private_key_path`       | The optional path to the private key to use for authentication. This would be used instead of `private_key`.                                                                   | string |    N     |
 | `private_key_passphrase` | The optional passphrase to use to decrypt `private_key` (if in PEM format) or `private_key_path`. Keys can be created without encryption so only provide this if needed.       | string |    N     |
+| `session_parameters`     | The optional session parameters to set for the connection.                                                                                                                     | dict   |    N     |
 
 
 #### Lowercase object names

sqlmesh/core/config/connection.py

@@ -333,6 +333,7 @@ class SnowflakeConnectionConfig(ConnectionConfig):
         private_key_passphrase: The optional passphrase to use to decrypt `private_key` or `private_key_path`. Keys can be created without encryption so only provide this if needed.
         register_comments: Whether or not to register model comments with the SQL engine.
         pre_ping: Whether or not to pre-ping the connection before starting a new transaction to ensure it is still alive.
+        session_parameters: The optional session parameters to set for the connection.
     """
 
     account: str
@@ -353,6 +354,8 @@ class SnowflakeConnectionConfig(ConnectionConfig):
     register_comments: bool = True
     pre_ping: bool = False
 
+    session_parameters: t.Optional[dict] = None
+
     type_: Literal["snowflake"] = Field(alias="type", default="snowflake")
 
     _concurrent_tasks_validator = concurrent_tasks_validator
@@ -471,6 +474,7 @@ def _connection_kwargs_keys(self) -> t.Set[str]:
             "authenticator",
             "token",
             "private_key",
+            "session_parameters",
         }
 
     @property

tests/core/test_connection_config.py

@@ -183,6 +183,17 @@ def test_snowflake(make_config, snowflake_key_passphrase_bytes, snowflake_oauth_
             authenticator="oauth",
         )
 
+    # Can pass in session parameters
+    config = make_config(
+        type="snowflake",
+        account="test",
+        user="test",
+        password="test",
+        authenticator="externalbrowser",
+        session_parameters={"query_tag": "test", "timezone": "UTC"},
+    )
+    assert isinstance(config, SnowflakeConnectionConfig)
+
 
 @pytest.mark.parametrize(
     "key_path_fixture, key_pem_fixture, key_b64_fixture, key_bytes_fixture, key_passphrase",