Add s3_stage_vpce_dns_name session param (#1162)

Co-authored-by: Craig Squire <[email protected]>

Author: csquire

dbt-snowflake/.changes/unreleased/Features-20250618-084607.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Add 's3_stage_vpce_dns_name' session parameter
+time: 2025-06-18T08:46:07.352953-05:00
+custom:
+    Author: csquire
+    Issue: "1161"

dbt-snowflake/src/dbt/adapters/snowflake/connections.py

@@ -113,6 +113,7 @@ class SnowflakeCredentials(Credentials):
     insecure_mode: Optional[bool] = False
     # this needs to default to `None` so that we can tell if the user set it; see `__post_init__()`
     reuse_connections: Optional[bool] = None
+    s3_stage_vpce_dns_name: Optional[str] = None
 
     def __post_init__(self):
         if self.authenticator != "oauth" and (self.oauth_client_secret or self.oauth_client_id):
@@ -180,6 +181,7 @@ def _connection_keys(self):
             "retry_all",
             "insecure_mode",
             "reuse_connections",
+            "s3_stage_vpce_dns_name",
         )
 
     def auth_args(self):
@@ -370,6 +372,9 @@ def connect():
 
             if creds.query_tag:
                 session_parameters.update({"QUERY_TAG": creds.query_tag})
+
+            if creds.s3_stage_vpce_dns_name:
+                session_parameters.update({"S3_STAGE_VPCE_DNS_NAME": creds.s3_stage_vpce_dns_name})
             handle = None
 
             # In replay mode, we won't connect to a real database at all, while

dbt-snowflake/tests/functional/s3_stage_vpce_dns_name/test_s3_stage_vpce_dns_name.py

@@ -0,0 +1,41 @@
+import pytest
+from dbt.tests.util import run_dbt
+
+
+models__test_model_sql = """
+{{ config(materialized = 'table') }}
+select 1 as id
+"""
+
+
+class TestS3StageVpceDnsName:
+    """
+    Test that the s3_stage_vpce_dns_name session parameter is properly set
+    when specified in the profile configuration.
+    """
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "test_model.sql": models__test_model_sql,
+        }
+
+    @pytest.fixture(scope="class")
+    def profiles_config_update(self, prefix, dbt_profile_target):
+        outputs = {"default": dbt_profile_target}
+        outputs["default"][
+            "s3_stage_vpce_dns_name"
+        ] = f"vpce-{prefix}-xxxxxxxx.s3.us-east-1.vpce.amazonaws.com"
+
+    def test_s3_stage_vpce_dns_name_set(self, project, prefix):
+        """Test that the s3_stage_vpce_dns_name session parameter is properly set."""
+        # Run dbt to create the model
+        run_dbt(["run"])
+
+        # Check that the session parameter was set correctly
+        result = project.run_sql(
+            "show parameters like 'S3_STAGE_VPCE_DNS_NAME' in session", fetch="one"
+        )
+        assert result is not None
+        expected_value = f"vpce-{prefix}-xxxxxxxx.s3.us-east-1.vpce.amazonaws.com"
+        assert result[1] == expected_value, f"Expected {expected_value}, got {result[1]}"