Adapt loader_criteria params for current query

Fixed critical issue in the new :meth:`_orm.PropComparator.and_` feature
where loader strategies that emit secondary SELECT statements such as
:func:`_orm.selectinload` and :func:`_orm.lazyload` would fail to
accommodate for bound parameters in the user-defined criteria in terms of
the current statement being executed, as opposed to the cached statement,
causing stale bound values to be used.

This also adds a warning for the case where an object that uses
:func:`_orm.lazyload` in conjunction with :meth:`_orm.PropComparator.and_`
is attempted to be serialized; the loader criteria cannot reliably
be serialized and deserialized and eager loading should be used for this
case.

Fixes: #6139
Change-Id: I5a638bbecb7b583db2d3c0b76469f5a25c13dd3b

5 files changed, 23 insertions, 11 deletions

diff --git a/lib/sqlalchemy/sql/annotation.py b/lib/sqlalchemy/sql/annotation.py
index 8e5cdf148..2436c9c3f 100644
--- a/lib/sqlalchemy/sql/annotation.py
+++ b/lib/sqlalchemy/sql/annotation.py
@@ -262,7 +262,7 @@ def _deep_annotate(element, annotations, exclude=None):
             and hasattr(elem, "proxy_set")
             and elem.proxy_set.intersection(exclude)
         ):
-            newelem = elem._clone()
+            newelem = elem._clone(**kw)
         elif annotations != elem._annotations:
             newelem = elem._annotate(annotations)
         else:
diff --git a/lib/sqlalchemy/sql/base.py b/lib/sqlalchemy/sql/base.py
index 726800717..ac9d66970 100644
--- a/lib/sqlalchemy/sql/base.py
+++ b/lib/sqlalchemy/sql/base.py
@@ -46,7 +46,7 @@ class Immutable(object):
     def params(self, *optionaldict, **kwargs):
         raise NotImplementedError("Immutable objects do not support copying")
 
-    def _clone(self):
+    def _clone(self, **kw):
         return self
 
     def _copy_internals(self, **kw):
@@ -128,7 +128,7 @@ def _exclusive_against(*names, **kw):
 
 
 def _clone(element, **kw):
-    return element._clone()
+    return element._clone(**kw)
 
 
 def _expand_cloned(elements):
@@ -747,7 +747,7 @@ class ExecutableOption(HasCopyInternals, HasCacheKey):
 
     __visit_name__ = "executable_option"
 
-    def _clone(self):
+    def _clone(self, **kw):
         """Create a shallow copy of this ExecutableOption."""
         c = self.__class__.__new__(self.__class__)
         c.__dict__ = dict(self.__dict__)
diff --git a/lib/sqlalchemy/sql/elements.py b/lib/sqlalchemy/sql/elements.py
index 74e8dceff..b64427d51 100644
--- a/lib/sqlalchemy/sql/elements.py
+++ b/lib/sqlalchemy/sql/elements.py
@@ -235,7 +235,7 @@ class ClauseElement(
         self._propagate_attrs = util.immutabledict(values)
         return self
 
-    def _clone(self):
+    def _clone(self, **kw):
         """Create a shallow copy of this ClauseElement.
 
         This method may be used by a generative API.  Its also used as
@@ -360,7 +360,9 @@ class ClauseElement(
             if unique:
                 bind._convert_to_unique()
 
-        return cloned_traverse(self, {}, {"bindparam": visit_bindparam})
+        return cloned_traverse(
+            self, {"maintain_key": True}, {"bindparam": visit_bindparam}
+        )
 
     def compare(self, other, **kw):
         r"""Compare this :class:`_expression.ClauseElement` to
@@ -1432,8 +1434,8 @@ class BindParameter(roles.InElementRole, ColumnElement):
         c.type = type_
         return c
 
-    def _clone(self, maintain_key=False):
-        c = ClauseElement._clone(self)
+    def _clone(self, maintain_key=False, **kw):
+        c = ClauseElement._clone(self, **kw)
         if not maintain_key and self.unique:
             c.key = _anonymous_label.safe_construct(
                 id(c), c._orig_key or "param"
diff --git a/lib/sqlalchemy/sql/traversals.py b/lib/sqlalchemy/sql/traversals.py
index 3849749de..f2099f191 100644
--- a/lib/sqlalchemy/sql/traversals.py
+++ b/lib/sqlalchemy/sql/traversals.py
@@ -378,6 +378,14 @@ class CacheKey(namedtuple("CacheKey", ["key", "bindparams"])):
         _anon_map = prefix_anon_map()
         return {b.key % _anon_map: b.effective_value for b in self.bindparams}
 
+    def _apply_params_to_element(self, original_cache_key, target_element):
+        translate = {
+            k.key: v.value
+            for k, v in zip(original_cache_key.bindparams, self.bindparams)
+        }
+
+        return target_element.params(translate)
+
 
 def _clone(element, **kw):
     return element._clone()
diff --git a/lib/sqlalchemy/sql/visitors.py b/lib/sqlalchemy/sql/visitors.py
index 8e113849e..5d60774aa 100644
--- a/lib/sqlalchemy/sql/visitors.py
+++ b/lib/sqlalchemy/sql/visitors.py
@@ -731,7 +731,7 @@ def cloned_traverse(obj, opts, visitors):
                         cloned[id(elem)] = newelem
                         return newelem
 
-                cloned[id(elem)] = newelem = elem._clone()
+                cloned[id(elem)] = newelem = elem._clone(**kw)
                 newelem._copy_internals(clone=clone, **kw)
                 meth = visitors.get(newelem.__visit_name__, None)
                 if meth:
@@ -739,7 +739,9 @@ def cloned_traverse(obj, opts, visitors):
             return cloned[id(elem)]
 
     if obj is not None:
-        obj = clone(obj, deferred_copy_internals=deferred_copy_internals)
+        obj = clone(
+            obj, deferred_copy_internals=deferred_copy_internals, **opts
+        )
     clone = None  # remove gc cycles
     return obj
 
@@ -797,7 +799,7 @@ def replacement_traverse(obj, opts, replace):
                             cloned[id_elem] = newelem
                             return newelem
 
-                    cloned[id_elem] = newelem = elem._clone()
+                    cloned[id_elem] = newelem = elem._clone(**kw)
                     newelem._copy_internals(clone=clone, **kw)
                 return cloned[id_elem]