From 9d80276e07d64076e7e46285be074b57f9758c16 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E5=86=AF=E5=9F=BA=E9=AD=81?= <1412414664@qq.com>
Date: Wed, 24 Jun 2026 15:05:34 +0800
Subject: [PATCH] fix snowflake schema creation race

---
 dlt/destinations/impl/snowflake/sql_client.py |  3 +++
 .../snowflake/test_snowflake_table_builder.py | 26 +++++++++++++++++++
 2 files changed, 29 insertions(+)

diff --git a/dlt/destinations/impl/snowflake/sql_client.py b/dlt/destinations/impl/snowflake/sql_client.py
index 22b917a368..03d0872269 100644
--- a/dlt/destinations/impl/snowflake/sql_client.py
+++ b/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.
diff --git a/tests/load/snowflake/test_snowflake_table_builder.py b/tests/load/snowflake/test_snowflake_table_builder.py
index 84e3db924b..fce797be24 100644
--- a/tests/load/snowflake/test_snowflake_table_builder.py
+++ b/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