Disable row level locking for Mariadb and MySQL <8
diff --git a/airflow/jobs/scheduler_job.py b/airflow/jobs/scheduler_job.py
index 60475c48a5853..747b6cba90e6d 100644
--- a/airflow/jobs/scheduler_job.py
+++ b/airflow/jobs/scheduler_job.py
@@ -812,7 +812,9 @@ def _change_state_for_tis_without_dagrun(
         # We need to do this for mysql as well because it can cause deadlocks
         # as discussed in https://issues.apache.org/jira/browse/AIRFLOW-2516
         if self.using_sqlite or self.using_mysql:
-            tis_to_change: List[TI] = with_row_locks(query, of=TI, **skip_locked(session=session)).all()
+            tis_to_change: List[TI] = with_row_locks(
+                query, of=TI, session=session, **skip_locked(session=session)
+            ).all()
             for ti in tis_to_change:
                 ti.set_state(new_state, session=session)
                 tis_changed += 1
@@ -922,6 +924,7 @@ def _executable_task_instances_to_queued(self, max_tis: int, session: Session =
         task_instances_to_examine: List[TI] = with_row_locks(
             query,
             of=TI,
+            session=session,
             **skip_locked(session=session),
         ).all()
         # TODO[HA]: This was wrong before anyway, as it only looked at a sub-set of dags, not everything.
@@ -1159,7 +1162,7 @@ def _change_state_for_tasks_failed_to_execute(self, session: Session = None):
             for dag_id, task_id, execution_date, try_number in self.executor.queued_tasks.keys()
         ]
         ti_query = session.query(TI).filter(or_(*filter_for_ti_state_change))
-        tis_to_set_to_scheduled: List[TI] = with_row_locks(ti_query).all()
+        tis_to_set_to_scheduled: List[TI] = with_row_locks(ti_query, session=session).all()
         if not tis_to_set_to_scheduled:
             return
 
@@ -1828,7 +1831,9 @@ def adopt_or_reset_orphaned_tasks(self, session: Session = None):
         )
 
         # Lock these rows, so that another scheduler can't try and adopt these too
-        tis_to_reset_or_adopt = with_row_locks(query, of=TI, **skip_locked(session=session)).all()
+        tis_to_reset_or_adopt = with_row_locks(
+            query, of=TI, session=session, **skip_locked(session=session)
+        ).all()
         to_reset = self.executor.try_adopt_task_instances(tis_to_reset_or_adopt)
 
         reset_tis_message = []
diff --git a/airflow/models/dag.py b/airflow/models/dag.py
index 4a0f5f88d40a7..15332f3345e7f 100644
--- a/airflow/models/dag.py
+++ b/airflow/models/dag.py
@@ -1823,7 +1823,7 @@ def bulk_write_to_db(cls, dags: Collection["DAG"], session=None):
             .options(joinedload(DagModel.tags, innerjoin=False))
             .filter(DagModel.dag_id.in_(dag_ids))
         )
-        orm_dags = with_row_locks(query, of=DagModel).all()
+        orm_dags = with_row_locks(query, of=DagModel, session=session).all()
 
         existing_dag_ids = {orm_dag.dag_id for orm_dag in orm_dags}
         missing_dag_ids = dag_ids.difference(existing_dag_ids)
@@ -2246,7 +2246,7 @@ def dags_needing_dagruns(cls, session: Session):
             .limit(cls.NUM_DAGS_PER_DAGRUN_QUERY)
         )
 
-        return with_row_locks(query, of=cls, **skip_locked(session=session))
+        return with_row_locks(query, of=cls, session=session, **skip_locked(session=session))
 
     def calculate_dagrun_date_fields(
         self, dag: DAG, most_recent_dag_run: Optional[pendulum.DateTime], active_runs_of_dag: int
diff --git a/airflow/models/dagrun.py b/airflow/models/dagrun.py
index c71a52f11e440..f7c9026805f8e 100644
--- a/airflow/models/dagrun.py
+++ b/airflow/models/dagrun.py
@@ -224,7 +224,9 @@ def next_dagruns_to_examine(
         if not settings.ALLOW_FUTURE_EXEC_DATES:
             query = query.filter(DagRun.execution_date <= func.now())
 
-        return with_row_locks(query.limit(max_number), of=cls, **skip_locked(session=session))
+        return with_row_locks(
+            query.limit(max_number), of=cls, session=session, **skip_locked(session=session)
+        )
 
     @staticmethod
     @provide_session
diff --git a/airflow/models/pool.py b/airflow/models/pool.py
index 67cdee52098d2..61d48c32b380d 100644
--- a/airflow/models/pool.py
+++ b/airflow/models/pool.py
@@ -102,7 +102,7 @@ def slots_stats(
         query = session.query(Pool.pool, Pool.slots)
 
         if lock_rows:
-            query = with_row_locks(query, **nowait(session))
+            query = with_row_locks(query, session=session, **nowait(session))
 
         pool_rows: Iterable[Tuple[str, int]] = query.all()
         for (pool_name, total_slots) in pool_rows:
diff --git a/airflow/models/taskinstance.py b/airflow/models/taskinstance.py
index 7e4b895a7bacd..6fdd0000ce5b0 100644
--- a/airflow/models/taskinstance.py
+++ b/airflow/models/taskinstance.py
@@ -1187,7 +1187,8 @@ def _run_mini_scheduler_on_child_tasks(self, session=None) -> None:
                     session.query(DagRun).filter_by(
                         dag_id=self.dag_id,
                         execution_date=self.execution_date,
-                    )
+                    ),
+                    session=session,
                 ).one()
 
                 # Get a partial dag with just the specific tasks we want to
diff --git a/airflow/utils/sqlalchemy.py b/airflow/utils/sqlalchemy.py
index 5e0a07cf98da5..d60f57d5ba54b 100644
--- a/airflow/utils/sqlalchemy.py
+++ b/airflow/utils/sqlalchemy.py
@@ -188,15 +188,19 @@ def nulls_first(col, session: Session) -> Dict[str, Any]:
 USE_ROW_LEVEL_LOCKING: bool = conf.getboolean('scheduler', 'use_row_level_locking', fallback=True)
 
 
-def with_row_locks(query, **kwargs):
+def with_row_locks(query, session: Session, **kwargs):
     """
     Apply with_for_update to an SQLAlchemy query, if row level locking is in use.
 
     :param query: An SQLAlchemy Query object
+    :param session: ORM Session
     :param kwargs: Extra kwargs to pass to with_for_update (of, nowait, skip_locked, etc)
     :return: updated query
     """
-    if USE_ROW_LEVEL_LOCKING:
+    dialect = session.bind.dialect
+
+    # Don't use row level locks if the MySQL dialect (Mariadb & MySQL < 8) does not support it.
+    if USE_ROW_LEVEL_LOCKING and (dialect.name != "mysql" or dialect.supports_for_update_of):
         return query.with_for_update(**kwargs)
     else:
         return query
diff --git a/tests/utils/test_sqlalchemy.py b/tests/utils/test_sqlalchemy.py
index 816f147bba7e8..250bd37756c48 100644
--- a/tests/utils/test_sqlalchemy.py
+++ b/tests/utils/test_sqlalchemy.py
@@ -27,7 +27,7 @@
 from airflow import settings
 from airflow.models import DAG
 from airflow.settings import Session
-from airflow.utils.sqlalchemy import nowait, prohibit_commit, skip_locked
+from airflow.utils.sqlalchemy import nowait, prohibit_commit, skip_locked, with_row_locks
 from airflow.utils.state import State
 from airflow.utils.timezone import utcnow
 
@@ -161,6 +161,33 @@ def test_nowait(self, dialect, supports_for_update_of, expected_return_value):
         session.bind.dialect.supports_for_update_of = supports_for_update_of
         assert nowait(session=session) == expected_return_value
 
+    @parameterized.expand(
+        [
+            ("postgresql", True, True, True),
+            ("postgresql", True, False, False),
+            ("mysql", False, True, False),
+            ("mysql", False, False, False),
+            ("mysql", True, True, True),
+            ("mysql", True, False, False),
+            ("sqlite", False, True, True),
+        ]
+    )
+    def test_with_row_locks(
+        self, dialect, supports_for_update_of, use_row_level_lock_conf, expected_use_row_level_lock
+    ):
+        query = mock.Mock()
+        session = mock.Mock()
+        session.bind.dialect.name = dialect
+        session.bind.dialect.supports_for_update_of = supports_for_update_of
+        with mock.patch("airflow.utils.sqlalchemy.USE_ROW_LEVEL_LOCKING", use_row_level_lock_conf):
+            returned_value = with_row_locks(query=query, session=session, nowait=True)
+
+        if expected_use_row_level_lock:
+            query.with_for_update.assert_called_once_with(nowait=True)
+        else:
+            assert returned_value == query
+            query.with_for_update.assert_not_called()
+
     def test_prohibit_commit(self):
         with prohibit_commit(self.session) as guard:
             self.session.execute('SELECT 1')

closes https://github.com/apache/airflow/issues/11899 
closes https://github.com/apache/airflow/issues/13668

This PR disable row-level locking for MySQL variants that do not support skip_locked and no_wait -- **MySQL < 8** and **MariaDB**

<!--
Thank you for contributing! Please make sure that your code changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.

Feel free to ping committers for the review!

In case of existing issue, reference it using one of the following:

closes: #ISSUE
related: #ISSUE

How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->

---
**^ Add meaningful description above**

Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/master/CONTRIBUTING.rst#pull-request-guidelines)** for more information.
In case of fundamental code change, Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvements+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in [UPDATING.md](https://github.com/apache/airflow/blob/master/UPDATING.md).
