process bulk_update_tuples before cache key or compilation

Fixed regression where the use of a :class:`_orm.hybrid_property` attribute
or a mapped :func:`_orm.composite` attribute as a key passed to the
:meth:`_dml.Update.values` method for an ORM-enabled :class:`_dml.Update`
statement, as well as when using it via the legacy
:meth:`_orm.Query.update` method, would be processed for incoming
ORM/hybrid/composite values within the compilation stage of the UPDATE
statement, which meant that in those cases where caching occurred,
subsequent invocations of the same statement would no longer receive the
correct values. This would include not only hybrids that use the
:meth:`_orm.hybrid_property.update_expression` method, but any use of a
plain hybrid attribute as well. For composites, the issue instead caused a
non-repeatable cache key to be generated, which would break caching and
could fill up the statement cache with repeated statements.

The :class:`_dml.Update` construct now handles the processing of key/value
pairs passed to :meth:`_dml.Update.values` and
:meth:`_dml.Update.ordered_values` up front when the construct is first
generated, before the cache key has been generated so that the key/value
pairs are processed each time, and so that the cache key is generated
against the individual column/value pairs that will ultimately be
used in the statement.

Fixes: #7209
Change-Id: I08f248d1d60ea9690b014c21439b775d951fb9e5

3 files changed, 39 insertions, 42 deletions

diff --git a/lib/sqlalchemy/sql/base.py b/lib/sqlalchemy/sql/base.py
index b235f5132..aba80222a 100644
--- a/lib/sqlalchemy/sql/base.py
+++ b/lib/sqlalchemy/sql/base.py
@@ -515,12 +515,20 @@ class CompileState(object):
     @classmethod
     def get_plugin_class(cls, statement):
         plugin_name = statement._propagate_attrs.get(
-            "compile_state_plugin", "default"
+            "compile_state_plugin", None
         )
+
+        if plugin_name:
+            key = (plugin_name, statement._effective_plugin_target)
+            if key in cls.plugins:
+                return cls.plugins[key]
+
+        # there's no case where we call upon get_plugin_class() and want
+        # to get None back, there should always be a default.  return that
+        # if there was no plugin-specific class  (e.g. Insert with "orm"
+        # plugin)
         try:
-            return cls.plugins[
-                (plugin_name, statement._effective_plugin_target)
-            ]
+            return cls.plugins[("default", statement._effective_plugin_target)]
         except KeyError:
             return None
 
@@ -1665,7 +1673,7 @@ def _entity_namespace(entity):
             raise
 
 
-def _entity_namespace_key(entity, key):
+def _entity_namespace_key(entity, key, default=NO_ARG):
     """Return an entry from an entity_namespace.
 
 
@@ -1676,7 +1684,10 @@ def _entity_namespace_key(entity, key):
 
     try:
         ns = _entity_namespace(entity)
-        return getattr(ns, key)
+        if default is not NO_ARG:
+            return getattr(ns, key, default)
+        else:
+            return getattr(ns, key)
     except AttributeError as err:
         util.raise_(
             exc.InvalidRequestError(
diff --git a/lib/sqlalchemy/sql/dml.py b/lib/sqlalchemy/sql/dml.py
index 158cb40f2..ebff0df88 100644
--- a/lib/sqlalchemy/sql/dml.py
+++ b/lib/sqlalchemy/sql/dml.py
@@ -52,6 +52,21 @@ class DMLState(CompileState):
     def dml_table(self):
         return self.statement.table
 
+    @classmethod
+    def _get_crud_kv_pairs(cls, statement, kv_iterator):
+        return [
+            (
+                coercions.expect(roles.DMLColumnRole, k),
+                coercions.expect(
+                    roles.ExpressionElementRole,
+                    v,
+                    type_=NullType(),
+                    is_crud=True,
+                ),
+            )
+            for k, v in kv_iterator
+        ]
+
     def _make_extra_froms(self, statement):
         froms = []
 
@@ -674,30 +689,12 @@ class ValuesBase(UpdateBase):
         # crud.py now intercepts bound parameters with unique=True from here
         # and ensures they get the "crud"-style name when rendered.
 
+        kv_generator = DMLState.get_plugin_class(self)._get_crud_kv_pairs
+
         if self._preserve_parameter_order:
-            arg = [
-                (
-                    coercions.expect(roles.DMLColumnRole, k),
-                    coercions.expect(
-                        roles.ExpressionElementRole,
-                        v,
-                        type_=NullType(),
-                        is_crud=True,
-                    ),
-                )
-                for k, v in arg
-            ]
-            self._ordered_values = arg
+            self._ordered_values = kv_generator(self, arg)
         else:
-            arg = {
-                coercions.expect(roles.DMLColumnRole, k): coercions.expect(
-                    roles.ExpressionElementRole,
-                    v,
-                    type_=NullType(),
-                    is_crud=True,
-                )
-                for k, v in arg.items()
-            }
+            arg = {k: v for k, v in kv_generator(self, arg.items())}
             if self._values:
                 self._values = self._values.union(arg)
             else:
@@ -1319,19 +1316,9 @@ class Update(DMLWhereBase, ValuesBase):
             raise exc.ArgumentError(
                 "This statement already has ordered values present"
             )
-        arg = [
-            (
-                coercions.expect(roles.DMLColumnRole, k),
-                coercions.expect(
-                    roles.ExpressionElementRole,
-                    v,
-                    type_=NullType(),
-                    is_crud=True,
-                ),
-            )
-            for k, v in args
-        ]
-        self._ordered_values = arg
+
+        kv_generator = DMLState.get_plugin_class(self)._get_crud_kv_pairs
+        self._ordered_values = kv_generator(self, args)
 
     @_generative
     def inline(self):
diff --git a/lib/sqlalchemy/sql/elements.py b/lib/sqlalchemy/sql/elements.py
index 3699f872b..ae105428c 100644
--- a/lib/sqlalchemy/sql/elements.py
+++ b/lib/sqlalchemy/sql/elements.py
@@ -1493,7 +1493,6 @@ class BindParameter(roles.InElementRole, ColumnElement):
                 :ref:`change_4808`.
 
         """
-
         if required is NO_ARG:
             required = value is NO_ARG and callable_ is None
         if value is NO_ARG: