Fix error when SnowflakeHook take empty list in sql param (#23767)

airflow/providers/snowflake/hooks/snowflake.py

@@ -308,14 +308,18 @@ def run(
         """
         self.query_ids = []
 
+        if isinstance(sql, str):
+            split_statements_tuple = split_statements(StringIO(sql))
+            sql = [sql_string for sql_string, _ in split_statements_tuple if sql_string]
+
+        if sql:
+            self.log.debug("Executing %d statements against Snowflake DB", len(sql))
+        else:
+            raise ValueError("List of SQL statements is empty")
+
         with closing(self.get_conn()) as conn:
             self.set_autocommit(conn, autocommit)
 
-            if isinstance(sql, str):
-                split_statements_tuple = split_statements(StringIO(sql))
-                sql = [sql_string for sql_string, _ in split_statements_tuple if sql_string]
-
-            self.log.debug("Executing %d statements against Snowflake DB", len(sql))
             # SnowflakeCursor does not extend ContextManager, so we have to ignore mypy error here
             with closing(conn.cursor(DictCursor)) as cur:  # type: ignore[type-var]
 

tests/providers/snowflake/hooks/test_snowflake.py

@@ -515,3 +515,11 @@ def test_connection_failure(self, mock_run):
             assert status is False
             assert msg == 'Connection Errors'
             mock_run.assert_called_once_with(sql='select 1')
+
+    def test_empty_sql_parameter(self):
+        hook = SnowflakeHook()
+
+        for empty_statement in ([], '', '\n'):
+            with pytest.raises(ValueError) as err:
+                hook.run(sql=empty_statement)
+            assert err.value.args[0] == "List of SQL statements is empty"