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

5 files changed, 105 insertions, 67 deletions

diff --git a/lib/sqlalchemy/ext/hybrid.py b/lib/sqlalchemy/ext/hybrid.py
index 298d957f6..eab3f2b73 100644
--- a/lib/sqlalchemy/ext/hybrid.py
+++ b/lib/sqlalchemy/ext/hybrid.py
@@ -805,7 +805,6 @@ things it can be used for.
 from .. import util
 from ..orm import attributes
 from ..orm import interfaces
-from ..sql import elements
 
 HYBRID_METHOD = util.symbol("HYBRID_METHOD")
 """Symbol indicating an :class:`InspectionAttr` that's
@@ -1183,9 +1182,6 @@ class ExprComparator(Comparator):
         return self.hybrid.info
 
     def _bulk_update_tuples(self, value):
-        if isinstance(value, elements.BindParameter):
-            value = value.value
-
         if isinstance(self.expression, attributes.QueryableAttribute):
             return self.expression._bulk_update_tuples(value)
         elif self.hybrid.update_expr is not None:
diff --git a/lib/sqlalchemy/orm/persistence.py b/lib/sqlalchemy/orm/persistence.py
index fd484b52b..3d20cfdea 100644
--- a/lib/sqlalchemy/orm/persistence.py
+++ b/lib/sqlalchemy/orm/persistence.py
@@ -23,6 +23,7 @@ from . import evaluator
 from . import exc as orm_exc
 from . import loading
 from . import sync
+from .base import NO_VALUE
 from .base import state_str
 from .. import exc as sa_exc
 from .. import future
@@ -34,6 +35,7 @@ from ..sql import expression
 from ..sql import operators
 from ..sql import roles
 from ..sql import select
+from ..sql import sqltypes
 from ..sql.base import _entity_namespace_key
 from ..sql.base import CompileState
 from ..sql.base import Options
@@ -2002,31 +2004,12 @@ class BulkUDCompileState(CompileState):
         if statement._multi_values:
             return []
         elif statement._ordered_values:
-            iterator = statement._ordered_values
+            return list(statement._ordered_values)
         elif statement._values:
-            iterator = statement._values.items()
+            return list(statement._values.items())
         else:
             return []
 
-        values = []
-        if iterator:
-            for k, v in iterator:
-                if mapper:
-                    if isinstance(k, util.string_types):
-                        desc = _entity_namespace_key(mapper, k)
-                        values.extend(desc._bulk_update_tuples(v))
-                    elif "entity_namespace" in k._annotations:
-                        k_anno = k._annotations
-                        attr = _entity_namespace_key(
-                            k_anno["entity_namespace"], k_anno["proxy_key"]
-                        )
-                        values.extend(attr._bulk_update_tuples(v))
-                    else:
-                        values.append((k, v))
-                else:
-                    values.append((k, v))
-        return values
-
     @classmethod
     def _resolved_keys_as_propnames(cls, mapper, resolved_values):
         values = []
@@ -2191,6 +2174,68 @@ class BulkORMUpdate(UpdateDMLState, BulkUDCompileState):
         return self
 
     @classmethod
+    def _get_crud_kv_pairs(cls, statement, kv_iterator):
+        plugin_subject = statement._propagate_attrs["plugin_subject"]
+
+        if plugin_subject:
+            mapper = plugin_subject.mapper
+        else:
+            mapper = None
+
+        values = []
+        core_get_crud_kv_pairs = UpdateDMLState._get_crud_kv_pairs
+
+        for k, v in kv_iterator:
+            if mapper:
+                k = coercions.expect(roles.DMLColumnRole, k)
+
+                if isinstance(k, util.string_types):
+                    desc = _entity_namespace_key(mapper, k, default=NO_VALUE)
+                    if desc is NO_VALUE:
+                        values.append(
+                            (
+                                k,
+                                coercions.expect(
+                                    roles.ExpressionElementRole,
+                                    v,
+                                    type_=sqltypes.NullType(),
+                                    is_crud=True,
+                                ),
+                            )
+                        )
+                    else:
+                        values.extend(
+                            core_get_crud_kv_pairs(
+                                statement, desc._bulk_update_tuples(v)
+                            )
+                        )
+                elif "entity_namespace" in k._annotations:
+                    k_anno = k._annotations
+                    attr = _entity_namespace_key(
+                        k_anno["entity_namespace"], k_anno["proxy_key"]
+                    )
+                    values.extend(
+                        core_get_crud_kv_pairs(
+                            statement, attr._bulk_update_tuples(v)
+                        )
+                    )
+                else:
+                    values.append(
+                        (
+                            k,
+                            coercions.expect(
+                                roles.ExpressionElementRole,
+                                v,
+                                type_=sqltypes.NullType(),
+                                is_crud=True,
+                            ),
+                        )
+                    )
+            else:
+                values.extend(core_get_crud_kv_pairs(statement, [(k, v)]))
+        return values
+
+    @classmethod
     def _do_post_synchronize_evaluate(cls, session, result, update_options):
 
         states = set()
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: