WIP: Fix for #8626 compliation of deprecated VALUES() clause instead of newer alias...

diff --git a/lib/sqlalchemy/dialects/mysql/base.py b/lib/sqlalchemy/dialects/mysql/base.py
index 50e0ec07eaee43cfabf37d1b95dca4f86d6ac905..93c13477467eb4788aa70a60876bef37808edf55 100644 (file)
--- a/lib/sqlalchemy/dialects/mysql/base.py
+++ b/lib/sqlalchemy/dialects/mysql/base.py
@@ -1079,7 +1079,7 @@ from ...util import topological
 SET_RE = re.compile(
     r"\s*SET\s+(?:(?:GLOBAL|SESSION)\s+)?\w", re.I | re.UNICODE
 )
-
+ON_DUP_ALIAS_NAME = "_new"
 
 # old names
 MSTime = TIME
@@ -1315,10 +1315,10 @@ class MySQLCompiler(compiler.SQLCompiler):
         else:
             cols = statement.table.c
 
+        alias_clause = ""
         clauses = []
         # traverses through all table columns to preserve table column order
         for column in (col for col in cols if col.key in on_duplicate.update):
-
             val = on_duplicate.update[column.key]
 
             if coercions._is_literal(val):
@@ -1327,6 +1327,7 @@ class MySQLCompiler(compiler.SQLCompiler):
             else:
 
                 def replace(obj):
+                    nonlocal alias
                     if (
                         isinstance(obj, elements.BindParameter)
                         and obj.type._isnull
@@ -1335,16 +1336,17 @@ class MySQLCompiler(compiler.SQLCompiler):
                         obj.type = column.type
                         return obj
                     elif (
-                        isinstance(obj, elements.ColumnClause)
-                        and obj.table is on_duplicate.inserted_alias
+                      isinstance(obj, elements.ColumnClause)
+                      and obj.table is on_duplicate.inserted_alias
                     ):
-                        obj = literal_column(
-                            "VALUES(" + self.preparer.quote(obj.name) + ")"
+                        if not alias_clause:
+                          alias = f"AS {ON_DUP_ALIAS_NAME} "
+                        return literal_column(
+                          f"{ON_DUP_ALIAS_NAME}.{self.preparer.quote(obj.name)}"
                         )
-                        return obj
                     else:
-                        # element is not replaced
-                        return None
+                      # element is not replaced
+                      return None
 
                 val = visitors.replacement_traverse(val, {}, replace)
                 value_text = self.process(val.self_group(), use_schema=False)
@@ -1362,8 +1364,7 @@ class MySQLCompiler(compiler.SQLCompiler):
                     (", ".join("'%s'" % c for c in non_matching)),
                 )
             )
-
-        return "ON DUPLICATE KEY UPDATE " + ", ".join(clauses)
+        return alias_clause + "ON DUPLICATE KEY UPDATE " + ", ".join(clauses)
 
     def visit_concat_op_expression_clauselist(
         self, clauselist, operator, **kw

diff --git a/test/dialect/mysql/test_compiler.py b/test/dialect/mysql/test_compiler.py
index 414f73ad7650f44d15e91f8939072de78c2afa3c..9e03672eba16748b6ac425a6f99fbc53f46e59a4 100644 (file)
--- a/test/dialect/mysql/test_compiler.py
+++ b/test/dialect/mysql/test_compiler.py
@@ -1109,7 +1109,9 @@ class InsertOnDuplicateTest(fixtures.TestBase, AssertsCompiledSQL):
         )
         expected_sql = (
             "INSERT INTO foos (id, bar) VALUES (%s, %s), (%s, %s) "
-            "ON DUPLICATE KEY UPDATE bar = VALUES(bar), baz = VALUES(baz)"
+            f"AS {mysql.ON_DUP_ALIAS_NAME} ON DUPLICATE KEY UPDATE "
+            f"bar = {mysql.ON_DUP_ALIAS_NAME}.bar, "
+            f"baz = {mysql.ON_DUP_ALIAS_NAME}.baz"
         )
         self.assert_compile(stmt, expected_sql)
 
@@ -1144,9 +1146,10 @@ class InsertOnDuplicateTest(fixtures.TestBase, AssertsCompiledSQL):
             baz=stmt.inserted.baz + "some literal" + stmt.inserted.bar,
         )
         expected_sql = (
-            "INSERT INTO foos (id, bar) VALUES (%s, %s), (%s, %s) ON "
-            "DUPLICATE KEY UPDATE bar = coalesce(VALUES(bar)), "
-            "baz = (concat(VALUES(baz), %s, VALUES(bar)))"
+            "INSERT INTO foos (id, bar) VALUES (%s, %s), (%s, %s) "
+            f"AS {mysql.ON_DUP_ALIAS_NAME} ON DUPLICATE KEY UPDATE bar = "
+            f"coalesce({mysql.ON_DUP_ALIAS_NAME}.bar), "
+            f"baz = (concat({mysql.ON_DUP_ALIAS_NAME}.baz, %s, {mysql.ON_DUP_ALIAS_NAME}.bar))"
         )
         self.assert_compile(
             stmt,
@@ -1160,7 +1163,6 @@ class InsertOnDuplicateTest(fixtures.TestBase, AssertsCompiledSQL):
             },
         )
 
-
 class RegexpCommon(testing.AssertsCompiledSQL):
     def setup_test(self):
         self.table = table(