yrossi-tf1 opened a new issue, #75419:
URL: https://github.com/apache/airflow/issues/75419
### Under which category would you file this issue?
Providers
### Apache Airflow version
3.2.2
### What happened and how to reproduce it?
**Provider:** apache-airflow-providers-snowflake 6.15.0 (bug confirmed on
6.12.2, 6.13.0, 6.14.0, 6.16.0 and 6.19.0; absent in 6.12.1),
apache-airflow-providers-common-sql 1.35.0, snowflake-connector-python 3.18.1
(connector behaviour identical in 3.15.0). Airflow 3.2.2,
`apache/airflow:3.2.2-python3.12` image, Docker Compose, Debian 12; also
reproduced on macOS.
### Issue description
Since `apache-airflow-providers-snowflake` 6.12.2 (#65040, "Respect
AUTOCOMMIT in session_parameters"), `SnowflakeHook.run(sql, autocommit=False)`
no longer executes `ALTER SESSION SET autocommit=False` nor the final
`conn.commit()` as soon as the hook (or the connection Extra) carries **any**
`session_parameters` dict, even one that does not contain `AUTOCOMMIT`, for
example `{"QUERY_TAG": "..."}`.
Every statement of a `SQLExecuteQueryOperator(autocommit=False)` task is
then auto-committed by Snowflake (its default), so multi-statement SQL such as
`DELETE ...; INSERT ...;` is no longer atomic. Nothing is logged: the task
succeeds and the data is written, only the transaction is gone. Snowflake
`QUERY_HISTORY` for the session shows the user statements only, with no
`ALTER_SESSION` and no `COMMIT`. `autocommit=True` is unaffected because
`run()` short-circuits on `autocommit` before the check.
### Root cause
`hooks/snowflake.py`, 6.15.0 line numbers, code still identical in 6.19.0:
1. `_build_base_conn_config()` puts `self.session_parameters` **by
reference** into the connect kwargs (l.484), and `get_conn()` calls
`snowflake.connector.connect(**conn_config)` (l.762).
2. The connector keeps that very dict object as `self._session_parameters`
and, during login, merges the session parameters returned by the server into
it: `session_parameters.update({p["name"]: p["value"] ...})` in
`snowflake/connector/auth/_auth.py` (l.477-481 in connector 3.18.1), then
`_update_parameters()`. The server always returns `AUTOCOMMIT` (true by
default), `TIMEZONE`, `CLIENT_SESSION_KEEP_ALIVE`, etc.
3. `run()` evaluates `_has_autocommit_session_parameter()` **after**
`get_conn()` (l.964 then l.968). `self.session_parameters` now contains
`AUTOCOMMIT: True`, so the check returns True, `set_autocommit()` is skipped
and `autocommit_mode` is forced to True, which also skips `conn.commit()`
(l.1000).
The same happens when `session_parameters` comes from the connection Extra
only (no hook kwarg): the dict cached in `_get_static_conn_params` is mutated
the same way. When no `session_parameters` is set anywhere, the connector
creates its own dict and the hook behaves correctly.
### Steps to reproduce
With a real Snowflake connection:
```python
from airflow.providers.snowflake.hooks.snowflake import SnowflakeHook
from airflow.providers.common.sql.hooks.handlers import fetch_all_handler
hook = SnowflakeHook(snowflake_conn_id="snowflake_default",
session_parameters={"QUERY_TAG": "repro"})
print(hook.session_parameters) # {'QUERY_TAG': 'repro'}
rows = hook.run(
"SELECT 1;\nSELECT QUERY_TYPE, QUERY_TEXT FROM
TABLE(INFORMATION_SCHEMA.QUERY_HISTORY_BY_SESSION()) ORDER BY START_TIME;",
autocommit=False, handler=fetch_all_handler, split_statements=True,
return_last=True,
)
print(hook.session_parameters) # {'QUERY_TAG': 'repro',
'AUTOCOMMIT': True, 'TIMEZONE': ..., ...} <- mutated by connect()
print(hook._has_autocommit_session_parameter()) # True
print(rows) # only the two SELECTs: no
ALTER_SESSION, no COMMIT
```
With 6.12.1 the same script prints `ALTER SESSION SET autocommit=False`, the
two SELECTs and `COMMIT`.
Offline reproduction (no Snowflake account; real
`SnowflakeConnection`/`SnowflakeCursor` classes, only the login is emulated the
way `_auth.py` does it):
```python
import json
from unittest import mock
from airflow.sdk import Connection
from airflow.providers.snowflake.hooks.snowflake import SnowflakeHook
from snowflake.connector.connection import SnowflakeConnection
from snowflake.connector.cursor import SnowflakeCursor
conn = Connection(conn_id="c", conn_type="snowflake", login="u",
password="p", extra=json.dumps({"account": "a"}))
sent = []
def fake_open(self): # what the real login does with the dict it received
self.converter = mock.MagicMock()
self._session_parameters.update({"AUTOCOMMIT": True, "TIMEZONE": "UTC"})
self._update_parameters(self._session_parameters)
self._rest = mock.MagicMock()
def fake_execute(self, command, *a, **k):
sent.append(command.strip()[:40]); self._sfqid = "q"; self._rowcount =
1; self._description = None; return self
with mock.patch.object(SnowflakeHook, "get_connection", classmethod(lambda
cls, cid: conn)), \
mock.patch.object(SnowflakeConnection,
"_SnowflakeConnection__open_connection", fake_open), \
mock.patch.object(SnowflakeConnection, "close", lambda self: None), \
mock.patch.object(SnowflakeCursor, "execute", fake_execute):
SnowflakeHook(snowflake_conn_id="c", session_parameters={"QUERY_TAG":
"t"}).run(
"DELETE FROM T;\nINSERT INTO T SELECT 1;", autocommit=False,
split_statements=True)
print(sent)
# 6.12.1 : ['ALTER SESSION SET autocommit=False', 'DELETE FROM T;', 'INSERT
INTO T SELECT 1;', 'COMMIT']
# 6.12.2+: ['DELETE FROM T;', 'INSERT INTO T SELECT 1;']
```
Note: mocking `snowflake.connector.connect` itself (as the provider unit
tests do) hides the bug, because a `MagicMock` does not mutate the dict.
### What you think should happen instead
`autocommit=False` should still emit `ALTER SESSION SET autocommit=False`
and a final `COMMIT` when the user did not put `AUTOCOMMIT` in
`session_parameters`.
Possible fixes: pass a copy of `session_parameters` to `connect()`
(`conn_config["session_parameters"] = dict(...)` in `_get_conn_params()`), or
evaluate `_has_autocommit_session_parameter()` before `get_conn()` in `run()`.
A regression test should emulate the connector's in-place merge rather than
mock `connect`.
Workarounds: pin `apache-airflow-providers-snowflake==6.12.1`, or subclass
the hook to copy the dict before `connect()`.
I am not planning to submit a PR for this.
### What you think should happen instead?
`autocommit=False` should still emit `ALTER SESSION SET autocommit=False`
and a final `COMMIT` when the user did not put `AUTOCOMMIT` in
`session_parameters` (behaviour of 6.12.1 and earlier). A `session_parameters`
dict containing only `QUERY_TAG` must not disable transactions.
Possible fixes: pass a copy of `session_parameters` to `connect()`
(`conn_config["session_parameters"] = dict(...)` in `_get_conn_params()`), or
evaluate `_has_autocommit_session_parameter()` before `get_conn()` in `run()`.
A regression test should emulate the connector's in-place merge of server
session parameters rather than mock `snowflake.connector.connect`.
### Operating System
Debian GNU/Linux 12 (bookworm), apache/airflow:3.2.2-python3.12 image; also
reproduced on macOS 26
### Deployment
Docker-Compose
### Apache Airflow Provider(s)
snowflake
### Versions of Apache Airflow Providers
apache-airflow-providers-snowflake==6.15.0 (bug confirmed on 6.12.2,
6.13.0, 6.14.0, 6.16.0 and 6.19.0; absent in 6.12.1)
apache-airflow-providers-common-sql==1.35.0
snowflake-connector-python==3.18.1 (connector behaviour identical in 3.15.0)
### Official Helm Chart version
Not Applicable
### Kubernetes Version
Not Applicable
### Helm Chart configuration
Not Applicable
### Docker Image customizations
Official `apache/airflow:3.2.2-python3.12` image with the providers above
installed from a `requirements.txt` into the image's venv (`uv pip install
--python /home/airflow/.local/bin/python -r requirements.txt`). Not relevant to
the bug: it also reproduces in a plain virtualenv on macOS with
`apache-airflow==3.2.2`, `apache-airflow-providers-snowflake==6.15.0` (or
6.19.0), `apache-airflow-providers-common-sql==1.35.0`,
`snowflake-connector-python==3.18.1`.
### Anything else?
Happens on every task, deterministically, for any `SQLExecuteQueryOperator`
or `SnowflakeHook.run()` call with `autocommit=False` and a
`session_parameters` dict (hook kwarg `hook_params={"session_parameters":
{...}}` or connection Extra). Setting a `QUERY_TAG` through
`session_parameters` is a common pattern, so the impact is wide and silent:
tasks succeed, only atomicity is lost.
Related: #30236 (the issue fixed by #65040).
### Are you willing to submit PR?
- [ ] Yes I am willing to submit a PR!
### Code of Conduct
- [x] I agree to follow this project's [Code of
Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
