Do not use of and skip_locked with mariadb

diff --git a/lib/sqlalchemy/dialects/mysql/base.py b/lib/sqlalchemy/dialects/mysql/base.py
index a814de1b6d290aa758202dbc6c468b6d493b88e2..fa49135d54c15dc1c1cc5613392847d9ae43fc61 100644 (file)
--- a/lib/sqlalchemy/dialects/mysql/base.py
+++ b/lib/sqlalchemy/dialects/mysql/base.py
@@ -1499,7 +1499,7 @@ class MySQLCompiler(compiler.SQLCompiler):
         else:
             tmp = " FOR UPDATE"
 
-        if select._for_update_arg.of:
+        if select._for_update_arg.of and self.dialect._is_mysql:
 
             tables = util.OrderedSet()
             for c in select._for_update_arg.of:
@@ -1512,7 +1512,8 @@ class MySQLCompiler(compiler.SQLCompiler):
 
         if select._for_update_arg.nowait:
             tmp += " NOWAIT"
-        if select._for_update_arg.skip_locked:
+
+        if select._for_update_arg.skip_locked and self.dialect._is_mysql:
             tmp += " SKIP LOCKED"
 
         return tmp

diff --git a/test/dialect/mysql/test_compiler.py b/test/dialect/mysql/test_compiler.py
index 636f9d52979fc9a3567fea3a0b352e1cd2e4b223..2a712e6817ab896eabf434405834120a938cbed7 100644 (file)
--- a/test/dialect/mysql/test_compiler.py
+++ b/test/dialect/mysql/test_compiler.py
@@ -353,6 +353,32 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
         expr = literal("x", type_=String) + literal("y", type_=String)
         self.assert_compile(expr, "concat('x', 'y')", literal_binds=True)
 
+    def test_mariadb_for_update(self):
+        dialect = mysql.dialect()
+        dialect.server_version_info = (10, 1, 1, "MariaDB")
+
+        table1 = table(
+            "mytable", column("myid"), column("name"), column("description")
+        )
+
+        self.assert_compile(
+            table1.select(table1.c.myid == 7).with_for_update(of=table1),
+            "SELECT mytable.myid, mytable.name, mytable.description "
+            "FROM mytable WHERE mytable.myid = %s "
+            "FOR UPDATE",
+            dialect=dialect,
+        )
+
+        self.assert_compile(
+            table1.select(table1.c.myid == 7).with_for_update(
+                skip_locked=True
+            ),
+            "SELECT mytable.myid, mytable.name, mytable.description "
+            "FROM mytable WHERE mytable.myid = %s "
+            "FOR UPDATE",
+            dialect=dialect,
+        )
+
     def test_for_update(self):
         table1 = table(
             "mytable", column("myid"), column("name"), column("description")