This is an automated email from the ASF dual-hosted git repository.
kaxil pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 24a809533ba Fix lost transaction in SnowflakeHook when
`session_parameters` is set (#75425)
24a809533ba is described below
commit 24a809533ba882726b7e845d35cf5c0e3eebb37e
Author: Andrey Karazhev <[email protected]>
AuthorDate: Thu Oct 22 23:57:04 2026 +0300
Fix lost transaction in SnowflakeHook when `session_parameters` is set
(#75425)
snowflake-connector-python merges the server's session parameters,
AUTOCOMMIT
among them, into the session_parameters dict it is given. The hook handed it
its own dict, or the one cached from the connection extra, so run() then
found
AUTOCOMMIT there, skipped ALTER SESSION SET autocommit=False and the final
COMMIT, and each statement of run(autocommit=False) was committed on its
own.
The task still succeeded and nothing was logged.
---
.../airflow/providers/snowflake/hooks/snowflake.py | 6 +++-
.../tests/unit/snowflake/hooks/test_snowflake.py | 35 ++++++++++++++++++++++
2 files changed, 40 insertions(+), 1 deletion(-)
diff --git
a/providers/snowflake/src/airflow/providers/snowflake/hooks/snowflake.py
b/providers/snowflake/src/airflow/providers/snowflake/hooks/snowflake.py
index 52d2f03d541..3931f2cf4a2 100644
--- a/providers/snowflake/src/airflow/providers/snowflake/hooks/snowflake.py
+++ b/providers/snowflake/src/airflow/providers/snowflake/hooks/snowflake.py
@@ -488,7 +488,11 @@ class SnowflakeHook(DbApiHook):
This is used in ``get_uri()`` and ``get_conn()``.
"""
static_config = self._get_static_conn_params
- conn_config = dict(static_config)
+ conn_config: dict[str, Any] = dict(static_config)
+ # The connector merges the server's session parameters, AUTOCOMMIT
among them, into the dict it is
+ # given; a copy keeps them out of the dicts that
_has_autocommit_session_parameter() reads.
+ if isinstance(conn_config.get("session_parameters"), dict):
+ conn_config["session_parameters"] =
dict(conn_config["session_parameters"])
if conn_config.get("authenticator") == "oauth":
azure_conn_id = conn_config.get("azure_conn_id")
diff --git a/providers/snowflake/tests/unit/snowflake/hooks/test_snowflake.py
b/providers/snowflake/tests/unit/snowflake/hooks/test_snowflake.py
index 897a7f8a224..2fdc6a030b7 100644
--- a/providers/snowflake/tests/unit/snowflake/hooks/test_snowflake.py
+++ b/providers/snowflake/tests/unit/snowflake/hooks/test_snowflake.py
@@ -32,6 +32,7 @@ from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from requests.exceptions import ConnectionError, HTTPError
+from snowflake.connector.cursor import SnowflakeCursor
from airflow.models import Connection
from airflow.providers.common.compat.sdk import
AirflowOptionalProviderFeatureException, timezone
@@ -1335,6 +1336,40 @@ class TestPytestSnowflakeHook:
conn.autocommit.assert_called_once_with(False)
+ @pytest.mark.parametrize(
+ ("hook_kwargs", "extra"),
+ [
+ pytest.param({"session_parameters": {"QUERY_TAG": "airflow"}}, {},
id="hook"),
+ pytest.param({}, {"session_parameters": {"QUERY_TAG": "airflow"}},
id="connection"),
+ ],
+ )
+ @mock.patch("snowflake.connector.SnowflakeConnection", autospec=True)
+ def test_run_keeps_transaction_when_connect_merges_session_parameters(
+ self, mock_connection_class, hook_kwargs, extra
+ ):
+ """The connector merges the server's AUTOCOMMIT into the
session_parameters dict it receives."""
+ conn = mock_connection_class.return_value
+ cur = mock.MagicMock(spec=SnowflakeCursor, rowcount=0)
+ conn.cursor.return_value = cur
+ type(cur).sfqid = mock.PropertyMock(return_value="qid")
+
+ def connect(**kwargs):
+ kwargs["session_parameters"].update({"AUTOCOMMIT": True})
+ return conn
+
+ mock_connection_class.side_effect = connect
+ connection_kwargs = deepcopy(BASE_CONNECTION_KWARGS)
+ connection_kwargs["extra"].update(extra)
+ with mock.patch.dict(
+ "os.environ",
+
AIRFLOW_CONN_SNOWFLAKE_DEFAULT=Connection(**connection_kwargs).get_uri(),
+ ):
+ hook = SnowflakeHook(**hook_kwargs)
+ hook.run("DELETE FROM t; INSERT INTO t VALUES (1);",
autocommit=False)
+
+ conn.autocommit.assert_called_once_with(False)
+ conn.commit.assert_called_once_with()
+
def test_get_openlineage_default_schema_with_no_schema_set(self):
connection_kwargs = {
**BASE_CONNECTION_KWARGS,
