From 914dd23c4c1b15f6d80b8017cf886895d0f72b18 Mon Sep 17 00:00:00 2001
From: harjeevan maan <harjeevanm@gmail.com>
Date: Thu, 3 Oct 2024 21:33:48 -0400
Subject: [PATCH] - Snowflake does not currently support variable binding in
 multi-statement SQL requests. (#42719)

- added warning for multi-statement query bindings in Snowflake hook
test: added unit test for multi-statement bindings warning

Related: #42033
---
 .../snowflake/hooks/snowflake_sql_api.py      | 10 ++++--
 .../snowflake/hooks/test_snowflake_sql_api.py | 35 +++++++++++++++++++
 2 files changed, 43 insertions(+), 2 deletions(-)

diff --git a/airflow/providers/snowflake/hooks/snowflake_sql_api.py b/airflow/providers/snowflake/hooks/snowflake_sql_api.py
index 7b7e22c228ff..5143f573c263 100644
--- a/airflow/providers/snowflake/hooks/snowflake_sql_api.py
+++ b/airflow/providers/snowflake/hooks/snowflake_sql_api.py
@@ -153,8 +153,14 @@ def execute_query(
         url = f"{self.account_identifier}.snowflakecomputing.com/api/v2/statements"
         params: dict[str, Any] | None = {"requestId": str(req_id), "async": True, "pageSize": 10}
         headers = self.get_headers()
-        if bindings is None:
-            bindings = {}
+        sql_is_multi_stmt = ";" in sql.strip()
+        if not isinstance(bindings, dict) and bindings is not None:
+            raise AirflowException("Bindings should be a dictionary or None.")
+        if bindings and sql_is_multi_stmt:
+            self.log.warning(
+                "Bindings are not supported for multi-statement queries. Bindings will be ignored."
+            )
+        bindings = bindings or {}
         data = {
             "statement": sql,
             "resultSetMetaData": {"format": "json"},
diff --git a/tests/providers/snowflake/hooks/test_snowflake_sql_api.py b/tests/providers/snowflake/hooks/test_snowflake_sql_api.py
index 2a7389f30559..df3f06db2f30 100644
--- a/tests/providers/snowflake/hooks/test_snowflake_sql_api.py
+++ b/tests/providers/snowflake/hooks/test_snowflake_sql_api.py
@@ -234,6 +234,41 @@ def test_execute_query_exception_without_statement_handle(
             hook.execute_query(sql, statement_count)
         assert exception_info
 
+    @pytest.mark.parametrize(
+        "sql,statement_count,bindings",
+        [
+            (SQL_MULTIPLE_STMTS, 4, {"1": {"type": "FIXED", "value": "123"}}),
+        ],
+    )
+    @mock.patch("airflow.providers.snowflake.hooks.snowflake_sql_api.requests")
+    @mock.patch(
+        "airflow.providers.snowflake.hooks.snowflake_sql_api.SnowflakeSqlApiHook._get_conn_params",
+        new_callable=PropertyMock,
+    )
+    @mock.patch("airflow.providers.snowflake.hooks.snowflake_sql_api.SnowflakeSqlApiHook.get_headers")
+    def test_execute_query_bindings_warning(
+        self,
+        mock_get_headers,
+        mock_conn_params,
+        mock_requests,
+        sql,
+        statement_count,
+        bindings,
+    ):
+        """Test execute_query method logs warning when bindings are provided for multi-statement queries"""
+        mock_conn_params.return_value = CONN_PARAMS
+        mock_get_headers.return_value = HEADERS
+        mock_requests.post.return_value = create_successful_response_mock(
+            {"statementHandles": ["uuid", "uuid1"]}
+        )
+
+        hook = SnowflakeSqlApiHook(snowflake_conn_id="mock_conn_id")
+        with mock.patch.object(hook.log, "warning") as mock_log_warning:
+            hook.execute_query(sql, statement_count, bindings=bindings)
+            mock_log_warning.assert_called_once_with(
+                "Bindings are not supported for multi-statement queries. Bindings will be ignored."
+            )
+
     @pytest.mark.parametrize(
         "query_ids",
         [