diff --git a/changelogs/fragments/0_mysql_query_session_vars.yml b/changelogs/fragments/0_mysql_query_session_vars.yml
new file mode 100644
index 0000000..b6e2749
--- /dev/null
+++ b/changelogs/fragments/0_mysql_query_session_vars.yml
@@ -0,0 +1,3 @@
+---
+minor_changes:
+  - '`mysql_query` - add new `session_vars` argument, similar to ansible-collections/community.mysql#489.'
diff --git a/plugins/modules/mysql_query.py b/plugins/modules/mysql_query.py
index f7c900a..bc83133 100644
--- a/plugins/modules/mysql_query.py
+++ b/plugins/modules/mysql_query.py
@@ -51,6 +51,12 @@ options:
     - Where passed queries run in a single transaction (C(yes)) or commit them one-by-one (C(no)).
     type: bool
     default: false
+  session_vars:
+     description:
+     - "Dictionary of session variables in form of C(variable: value) to set at the beginning of module execution."
+     - Cannot be used to set global variables, use the M(community.mysql.mysql_variables) module instead.
+     type: dict
+     version_added: '3.16.0'
 attributes:
   check_mode:
     support: none
@@ -136,6 +142,7 @@ from ansible_collections.community.mysql.plugins.module_utils.mysql import (
     mysql_common_argument_spec,
     mysql_driver,
     mysql_driver_fail_msg,
+    set_session_vars,
 )
 from ansible.module_utils._text import to_native
 
@@ -176,6 +183,7 @@ def main():
         positional_args=dict(type='list', elements='raw'),
         named_args=dict(type='dict'),
         single_transaction=dict(type='bool', default=False),
+        session_vars=dict(type='dict'),
     )
 
     module = AnsibleModule(
@@ -195,6 +203,7 @@ def main():
     check_hostname = module.params['check_hostname']
     config_file = module.params['config_file']
     query = module.params["query"]
+    session_vars = module.params["session_vars"]
 
     if not isinstance(query, (str, list)):
         module.fail_json(msg="the query option value must be a string or list, passed %s" % type(query))
@@ -238,6 +247,9 @@ def main():
 
     max_keyword_len = len(max(DML_QUERY_KEYWORDS + DDL_QUERY_KEYWORDS, key=len))
 
+    if session_vars:
+        set_session_vars(module, cursor, session_vars)
+
     # Execute query:
     query_result = []
     executed_queries = []
diff --git a/tests/integration/targets/test_mysql_query/tasks/main.yml b/tests/integration/targets/test_mysql_query/tasks/main.yml
index ffb54e2..e2683cd 100644
--- a/tests/integration/targets/test_mysql_query/tasks/main.yml
+++ b/tests/integration/targets/test_mysql_query/tasks/main.yml
@@ -7,3 +7,5 @@
 - import_tasks: mysql_query_initial.yml
 
 - include_tasks: issue-28.yml
+
+- include_tasks: session_vars.yml
diff --git a/tests/integration/targets/test_mysql_query/tasks/session_vars.yml b/tests/integration/targets/test_mysql_query/tasks/session_vars.yml
new file mode 100644
index 0000000..b8d2f36
--- /dev/null
+++ b/tests/integration/targets/test_mysql_query/tasks/session_vars.yml
@@ -0,0 +1,29 @@
+---
+- vars:
+    mysql_parameters: &mysql_params
+      login_user: '{{ mysql_user }}'
+      login_password: '{{ mysql_password }}'
+      login_host: '{{ mysql_host }}'
+      login_port: '{{ mysql_primary_port }}'
+
+  block:
+
+  - name: Select sql_log_bin without session vars
+    mysql_query:
+      <<: *mysql_params
+      query: SELECT @@session.sql_log_bin AS sql_log_bin
+    register: result_without_vars
+
+  - name: Select sql_log_bin with session vars
+    mysql_query:
+      <<: *mysql_params
+      query: SELECT @@session.sql_log_bin AS sql_log_bin
+      session_vars:
+        sql_log_bin: 0
+    register: result_with_vars
+
+  - name: Assert sql_log_bin is set
+    ansible.builtin.assert:
+      that:
+        - 'result_without_vars.query_result[0][0].sql_log_bin == 1'
+        - 'result_with_vars.query_result[0][0].sql_log_bin == 0'