SNOW-1184290: Skip CREATE TABLE IF NOT EXISTS when auto_create_table

DESCRIPTION.md

@@ -16,6 +16,7 @@ Source code is also available at: https://github.com/snowflakedb/snowflake-conne
   - Added support for AWS outbound JWT token attestation for Workload Identity Federation (WIF). This can be enabled by setting the `SNOWFLAKE_ENABLE_AWS_WIF_OUTBOUND_TOKEN` environment variable to `true`. Note: This environment variable will be removed in a future release.
   - Removed dynamic class deserialization from the OCSP response validation cache to prevent arbitrary code execution via crafted cache files (SNOW-2439940). The `SNOWFLAKE_ENABLE_CUSTOM_REVOCATION_ERRORS` environment variable is now a no-op.
   - Updated SPCS token injection to gate on `SNOWFLAKE_RUNNING_INSIDE_SPCS` environment variable, trim whitespace, and remove configurable token path.
+  - Fixed a bug where `write_pandas()` with `auto_create_table=False` and `overwrite=True` would execute `CREATE TABLE IF NOT EXISTS`, which required unnecessary `OWNERSHIP` privilege on the table. Now only `TRUNCATE TABLE` is executed in this case. Note: users who relied on the table being implicitly created despite `auto_create_table=False` should set `auto_create_table=True` instead.
 
 - v4.4.0(March 25,2026)
   - Bump the lower boundary of cryptography to 46.0.5 due to CVE-2026-26007.

src/snowflake/connector/aio/_pandas_tools.py

@@ -443,7 +443,7 @@ async def drop_object(name: str, object_type: str) -> None:
             quote_identifiers,
         )
 
-        if auto_create_table or overwrite:
+        if auto_create_table:
             iceberg = "ICEBERG " if iceberg_config else ""
             iceberg_config_statement = _iceberg_config_statement_helper(
                 iceberg_config or {}

src/snowflake/connector/pandas_tools.py

@@ -528,7 +528,7 @@ def drop_object(name: str, object_type: str) -> None:
             quote_identifiers,
         )
 
-        if auto_create_table or overwrite:
+        if auto_create_table:
             iceberg = "ICEBERG " if iceberg_config else ""
             iceberg_config_statement = _iceberg_config_statement_helper(
                 iceberg_config or {}

test/unit/aio/test_pandas_tools_async.py

@@ -0,0 +1,143 @@
+from __future__ import annotations
+
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from snowflake.connector.options import pandas
+
+# Fake COPY INTO result row: (file, status, rows_parsed, rows_loaded, ...)
+_COPY_RESULT = [("file0.txt", "LOADED", 1, 1, 0, 0, None, None, None, None)]
+# Fake infer_schema result
+_INFER_SCHEMA_RESULT = [("name", "VARCHAR"), ("points", "NUMBER")]
+
+
+@pytest.fixture
+def mock_connection():
+    conn = MagicMock()
+    conn._session_parameters = {}
+    return conn
+
+
+@pytest.fixture
+def mock_cursor(mock_connection):
+    cursor = AsyncMock()
+    mock_connection.cursor.return_value = cursor
+
+    async def _execute_side_effect(sql, *args, **kwargs):
+        result = AsyncMock()
+        if "infer_schema" in sql:
+            result.fetchall.return_value = _INFER_SCHEMA_RESULT
+        elif "COPY INTO" in sql:
+            result.fetchall.return_value = _COPY_RESULT
+        return result
+
+    cursor.execute = AsyncMock(side_effect=_execute_side_effect)
+    return cursor
+
+
+def _get_executed_sqls(mock_cursor):
+    """Extract the SQL strings from all cursor.execute calls."""
+    return [c.args[0] for c in mock_cursor.execute.call_args_list]
+
+
+class TestWritePandasAsyncOverwriteWithoutAutoCreate:
+    """Tests for SNOW-1184290: async write_pandas() with auto_create_table=False and
+    overwrite=True should NOT execute CREATE TABLE IF NOT EXISTS."""
+
+    @pytest.mark.asyncio
+    @patch(
+        "snowflake.connector.aio._pandas_tools._create_temp_stage",
+        new_callable=AsyncMock,
+        return_value="tmp_stage",
+    )
+    @patch(
+        "snowflake.connector.aio._pandas_tools._create_temp_file_format",
+        new_callable=AsyncMock,
+        return_value="tmp_fmt",
+    )
+    async def test_overwrite_without_auto_create_does_not_create_table(
+        self, mock_file_format, mock_stage, mock_connection, mock_cursor
+    ):
+        from snowflake.connector.aio._pandas_tools import write_pandas
+
+        df = pandas.DataFrame([("Mark", 10)], columns=["name", "points"])
+
+        await write_pandas(
+            mock_connection,
+            df,
+            "test_table",
+            auto_create_table=False,
+            overwrite=True,
+        )
+
+        executed_sqls = _get_executed_sqls(mock_cursor)
+        assert not any(
+            "CREATE" in sql and "TABLE IF NOT EXISTS" in sql for sql in executed_sqls
+        ), "Should not CREATE TABLE when auto_create_table=False"
+        assert any(
+            "TRUNCATE" in sql for sql in executed_sqls
+        ), "Expected TRUNCATE TABLE when overwrite=True and auto_create_table=False"
+
+    @pytest.mark.asyncio
+    @patch(
+        "snowflake.connector.aio._pandas_tools._create_temp_stage",
+        new_callable=AsyncMock,
+        return_value="tmp_stage",
+    )
+    @patch(
+        "snowflake.connector.aio._pandas_tools._create_temp_file_format",
+        new_callable=AsyncMock,
+        return_value="tmp_fmt",
+    )
+    async def test_overwrite_with_auto_create_does_create_table(
+        self, mock_file_format, mock_stage, mock_connection, mock_cursor
+    ):
+        from snowflake.connector.aio._pandas_tools import write_pandas
+
+        df = pandas.DataFrame([("Mark", 10)], columns=["name", "points"])
+
+        await write_pandas(
+            mock_connection,
+            df,
+            "test_table",
+            auto_create_table=True,
+            overwrite=True,
+        )
+
+        executed_sqls = _get_executed_sqls(mock_cursor)
+        assert any(
+            "CREATE" in sql and "TABLE IF NOT EXISTS" in sql for sql in executed_sqls
+        ), "Expected CREATE TABLE when auto_create_table=True"
+        assert not any(
+            "TRUNCATE" in sql for sql in executed_sqls
+        ), "Should not TRUNCATE when auto_create_table=True (uses drop+rename instead)"
+
+    @pytest.mark.asyncio
+    @patch(
+        "snowflake.connector.aio._pandas_tools._create_temp_stage",
+        new_callable=AsyncMock,
+        return_value="tmp_stage",
+    )
+    async def test_no_overwrite_no_auto_create_no_create_table(
+        self, mock_stage, mock_connection, mock_cursor
+    ):
+        from snowflake.connector.aio._pandas_tools import write_pandas
+
+        df = pandas.DataFrame([("Mark", 10)], columns=["name", "points"])
+
+        await write_pandas(
+            mock_connection,
+            df,
+            "test_table",
+            auto_create_table=False,
+            overwrite=False,
+        )
+
+        executed_sqls = _get_executed_sqls(mock_cursor)
+        assert not any(
+            "CREATE" in sql and "TABLE IF NOT EXISTS" in sql for sql in executed_sqls
+        ), "Should not CREATE TABLE when auto_create_table=False"
+        assert not any(
+            "TRUNCATE" in sql for sql in executed_sqls
+        ), "Should not TRUNCATE when overwrite=False"

test/unit/test_pandas_tools.py

@@ -0,0 +1,134 @@
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+pandas = pytest.importorskip("pandas")
+
+# Fake COPY INTO result row: (file, status, rows_parsed, rows_loaded, ...)
+_COPY_RESULT = [("file0.txt", "LOADED", 1, 1, 0, 0, None, None, None, None)]
+# Fake infer_schema result
+_INFER_SCHEMA_RESULT = [("name", "VARCHAR"), ("points", "NUMBER")]
+
+
+@pytest.fixture
+def mock_connection():
+    conn = MagicMock()
+    conn._session_parameters = {}
+    return conn
+
+
+@pytest.fixture
+def mock_cursor(mock_connection):
+    cursor = MagicMock()
+    mock_connection.cursor.return_value = cursor
+
+    def _execute_side_effect(sql, *args, **kwargs):
+        result = MagicMock()
+        if "infer_schema" in sql:
+            result.fetchall.return_value = _INFER_SCHEMA_RESULT
+        elif "COPY INTO" in sql:
+            result.fetchall.return_value = _COPY_RESULT
+        return result
+
+    cursor.execute.side_effect = _execute_side_effect
+    return cursor
+
+
+def _get_executed_sqls(mock_cursor):
+    """Extract the SQL strings from all cursor.execute calls."""
+    return [c.args[0] for c in mock_cursor.execute.call_args_list]
+
+
+@pytest.mark.pandas
+@pytest.mark.unit
+class TestWritePandasOverwriteWithoutAutoCreate:
+    """Tests for SNOW-1184290: write_pandas() with auto_create_table=False and
+    overwrite=True should NOT execute CREATE TABLE IF NOT EXISTS."""
+
+    @patch(
+        "snowflake.connector.pandas_tools._create_temp_stage", return_value="tmp_stage"
+    )
+    @patch(
+        "snowflake.connector.pandas_tools._create_temp_file_format",
+        return_value="tmp_fmt",
+    )
+    def test_overwrite_without_auto_create_does_not_create_table(
+        self, mock_file_format, mock_stage, mock_connection, mock_cursor
+    ):
+        from snowflake.connector.pandas_tools import write_pandas
+
+        df = pandas.DataFrame([("Mark", 10)], columns=["name", "points"])
+
+        write_pandas(
+            mock_connection,
+            df,
+            "test_table",
+            auto_create_table=False,
+            overwrite=True,
+        )
+
+        executed_sqls = _get_executed_sqls(mock_cursor)
+        assert not any(
+            "CREATE" in sql and "TABLE IF NOT EXISTS" in sql for sql in executed_sqls
+        ), "Should not CREATE TABLE when auto_create_table=False"
+        assert any(
+            "TRUNCATE" in sql for sql in executed_sqls
+        ), "Expected TRUNCATE TABLE when overwrite=True and auto_create_table=False"
+
+    @patch(
+        "snowflake.connector.pandas_tools._create_temp_stage", return_value="tmp_stage"
+    )
+    @patch(
+        "snowflake.connector.pandas_tools._create_temp_file_format",
+        return_value="tmp_fmt",
+    )
+    def test_overwrite_with_auto_create_does_create_table(
+        self, mock_file_format, mock_stage, mock_connection, mock_cursor
+    ):
+        from snowflake.connector.pandas_tools import write_pandas
+
+        df = pandas.DataFrame([("Mark", 10)], columns=["name", "points"])
+
+        write_pandas(
+            mock_connection,
+            df,
+            "test_table",
+            auto_create_table=True,
+            overwrite=True,
+        )
+
+        executed_sqls = _get_executed_sqls(mock_cursor)
+        assert any(
+            "CREATE" in sql and "TABLE IF NOT EXISTS" in sql for sql in executed_sqls
+        ), "Expected CREATE TABLE when auto_create_table=True"
+        assert not any(
+            "TRUNCATE" in sql for sql in executed_sqls
+        ), "Should not TRUNCATE when auto_create_table=True (uses drop+rename instead)"
+
+    @patch(
+        "snowflake.connector.pandas_tools._create_temp_stage", return_value="tmp_stage"
+    )
+    def test_no_overwrite_no_auto_create_no_create_table(
+        self, mock_stage, mock_connection, mock_cursor
+    ):
+        from snowflake.connector.pandas_tools import write_pandas
+
+        df = pandas.DataFrame([("Mark", 10)], columns=["name", "points"])
+
+        write_pandas(
+            mock_connection,
+            df,
+            "test_table",
+            auto_create_table=False,
+            overwrite=False,
+        )
+
+        executed_sqls = _get_executed_sqls(mock_cursor)
+        assert not any(
+            "CREATE" in sql and "TABLE IF NOT EXISTS" in sql for sql in executed_sqls
+        ), "Should not CREATE TABLE when auto_create_table=False"
+        assert not any(
+            "TRUNCATE" in sql for sql in executed_sqls
+        ), "Should not TRUNCATE when overwrite=False"