fix: make Snowflake schema creation idempotent

dlt/destinations/impl/snowflake/sql_client.py

@@ -117,6 +117,9 @@ def rollback_transaction(self) -> None:
     def native_connection(self) -> "snowflake_lib.SnowflakeConnection":
         return self._conn
 
+    def create_dataset(self) -> None:
+        self.execute_sql("CREATE SCHEMA IF NOT EXISTS %s" % self.fully_qualified_dataset_name())
+
     def drop_tables(self, *tables: str) -> None:
         # Tables are drop with `IF EXISTS`, but snowflake raises when the schema doesn't exist.
         # Multi statement exec is safe and the error can be ignored since all tables are in the same schema.

tests/load/snowflake/test_snowflake_table_builder.py

@@ -21,6 +21,7 @@
     SnowflakeClientConfiguration,
     SnowflakeCredentials,
 )
+from dlt.destinations.impl.snowflake.sql_client import SnowflakeSqlClient
 
 from dlt.common.destination.typing import PreparedTableSchema
 from dlt.common.schema.typing import TColumnSchema
@@ -57,6 +58,31 @@ def create_client(schema: Schema, use_decfloat: bool = False) -> SnowflakeClient
     )
 
 
+def test_create_dataset_uses_idempotent_schema_sql(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    credentials = SnowflakeCredentials()
+    credentials.database = "test_db"
+    sql_client = SnowflakeSqlClient(
+        "test_dataset",
+        "test_dataset_staging",
+        credentials,
+        snowflake().capabilities(),
+    )
+    executed_sql: list[str] = []
+
+    def capture_sql(sql: str, *args: object, **kwargs: object) -> None:
+        executed_sql.append(sql)
+
+    monkeypatch.setattr(sql_client, "execute_sql", capture_sql)
+
+    sql_client.create_dataset()
+
+    assert executed_sql == [
+        "CREATE SCHEMA IF NOT EXISTS %s" % sql_client.fully_qualified_dataset_name()
+    ]
+
+
 def test_create_table(snowflake_client: SnowflakeClient) -> None:
     # make sure we are in case insensitive mode
     assert snowflake_client.capabilities.generates_case_sensitive_identifiers() is False