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

7 files changed, 141 insertions, 26 deletions

diff --git a/lib/sqlalchemy/orm/strategies.py b/lib/sqlalchemy/orm/strategies.py
index b11758090..822f7b96b 100644
--- a/lib/sqlalchemy/orm/strategies.py
+++ b/lib/sqlalchemy/orm/strategies.py
@@ -781,7 +781,7 @@ class LazyLoader(AbstractRelationshipLoader, util.MemoizedSlots):
             "'%s' is not available due to lazy='%s'" % (self, lazy)
         )
 
-    def _load_for_state(self, state, passive, loadopt=None):
+    def _load_for_state(self, state, passive, loadopt=None, extra_criteria=()):
 
         if not state.key and (
             (
@@ -872,7 +872,12 @@ class LazyLoader(AbstractRelationshipLoader, util.MemoizedSlots):
                 return attributes.PASSIVE_NO_RESULT
 
         return self._emit_lazyload(
-            session, state, primary_key_identity, passive, loadopt
+            session,
+            state,
+            primary_key_identity,
+            passive,
+            loadopt,
+            extra_criteria,
         )
 
     def _get_ident_for_use_get(self, session, state, passive):
@@ -899,7 +904,13 @@ class LazyLoader(AbstractRelationshipLoader, util.MemoizedSlots):
 
     @util.preload_module("sqlalchemy.orm.strategy_options")
     def _emit_lazyload(
-        self, session, state, primary_key_identity, passive, loadopt
+        self,
+        session,
+        state,
+        primary_key_identity,
+        passive,
+        loadopt,
+        extra_criteria,
     ):
         strategy_options = util.preloaded.orm_strategy_options
 
@@ -939,7 +950,6 @@ class LazyLoader(AbstractRelationshipLoader, util.MemoizedSlots):
         use_get = self.use_get
 
         if state.load_options or (loadopt and loadopt._extra_criteria):
-
             effective_path = state.load_path[self.parent_property]
 
             opts = list(state.load_options)
@@ -947,9 +957,7 @@ class LazyLoader(AbstractRelationshipLoader, util.MemoizedSlots):
             if loadopt and loadopt._extra_criteria:
                 use_get = False
                 opts += (
-                    orm_util.LoaderCriteriaOption(
-                        self.entity, sql.and_(*loadopt._extra_criteria)
-                    ),
+                    orm_util.LoaderCriteriaOption(self.entity, extra_criteria),
                 )
 
             stmt += lambda stmt: stmt.options(*opts)
@@ -1072,7 +1080,18 @@ class LazyLoader(AbstractRelationshipLoader, util.MemoizedSlots):
             # class-level lazyloader installed.
             set_lazy_callable = (
                 InstanceState._instance_level_callable_processor
-            )(mapper.class_manager, LoadLazyAttribute(key, self, loadopt), key)
+            )(
+                mapper.class_manager,
+                LoadLazyAttribute(
+                    key,
+                    self,
+                    loadopt,
+                    loadopt._generate_extra_criteria(context)
+                    if loadopt._extra_criteria
+                    else None,
+                ),
+                key,
+            )
 
             populators["new"].append((self.key, set_lazy_callable))
         elif context.populate_existing or mapper.always_refresh:
@@ -1092,12 +1111,42 @@ class LazyLoader(AbstractRelationshipLoader, util.MemoizedSlots):
 
 
 class LoadLazyAttribute(object):
-    """serializable loader object used by LazyLoader"""
+    """semi-serializable loader object used by LazyLoader
+
+    Historically, this object would be carried along with instances that
+    needed to run lazyloaders, so it had to be serializable to support
+    cached instances.
 
-    def __init__(self, key, initiating_strategy, loadopt):
+    this is no longer a general requirement, and the case where this object
+    is used is exactly the case where we can't really serialize easily,
+    which is when extra criteria in the loader option is present.
+
+    We can't reliably serialize that as it refers to mapped entities and
+    AliasedClass objects that are local to the current process, which would
+    need to be matched up on deserialize e.g. the sqlalchemy.ext.serializer
+    approach.
+
+    """
+
+    def __init__(self, key, initiating_strategy, loadopt, extra_criteria):
         self.key = key
         self.strategy_key = initiating_strategy.strategy_key
         self.loadopt = loadopt
+        self.extra_criteria = extra_criteria
+
+    def __getstate__(self):
+        if self.extra_criteria is not None:
+            util.warn(
+                "Can't reliably serialize a lazyload() option that "
+                "contains additional criteria; please use eager loading "
+                "for this case"
+            )
+        return {
+            "key": self.key,
+            "strategy_key": self.strategy_key,
+            "loadopt": self.loadopt,
+            "extra_criteria": (),
+        }
 
     def __call__(self, state, passive=attributes.PASSIVE_OFF):
         key = self.key
@@ -1105,7 +1154,12 @@ class LoadLazyAttribute(object):
         prop = instance_mapper._props[key]
         strategy = prop._strategies[self.strategy_key]
 
-        return strategy._load_for_state(state, passive, loadopt=self.loadopt)
+        return strategy._load_for_state(
+            state,
+            passive,
+            loadopt=self.loadopt,
+            extra_criteria=self.extra_criteria,
+        )
 
 
 class PostLoader(AbstractRelationshipLoader):
@@ -1416,6 +1470,7 @@ class SubqueryLoader(PostLoader):
 
     def _setup_options(
         self,
+        context,
         q,
         subq_path,
         rewritten_path,
@@ -1423,13 +1478,14 @@ class SubqueryLoader(PostLoader):
         effective_entity,
         loadopt,
     ):
-
         opts = orig_query._with_options
 
         if loadopt and loadopt._extra_criteria:
+
             opts += (
                 orm_util.LoaderCriteriaOption(
-                    self.entity, sql.and_(*loadopt._extra_criteria)
+                    self.entity,
+                    loadopt._generate_extra_criteria(context),
                 ),
             )
 
@@ -1641,7 +1697,13 @@ class SubqueryLoader(PostLoader):
         )
 
         q = self._setup_options(
-            q, subq_path, rewritten_path, orig_query, effective_entity, loadopt
+            context,
+            q,
+            subq_path,
+            rewritten_path,
+            orig_query,
+            effective_entity,
+            loadopt,
         )
         q = self._setup_outermost_orderby(q)
 
@@ -2832,10 +2894,12 @@ class SelectInLoader(PostLoader, util.MemoizedSlots):
         effective_path = path[self.parent_property]
 
         options = orig_query._with_options
+
         if loadopt and loadopt._extra_criteria:
             options += (
                 orm_util.LoaderCriteriaOption(
-                    effective_entity, sql.and_(*loadopt._extra_criteria)
+                    effective_entity,
+                    loadopt._generate_extra_criteria(context),
                 ),
             )
 
diff --git a/lib/sqlalchemy/orm/strategy_options.py b/lib/sqlalchemy/orm/strategy_options.py
index ba4e5c466..8fa79bfdb 100644
--- a/lib/sqlalchemy/orm/strategy_options.py
+++ b/lib/sqlalchemy/orm/strategy_options.py
@@ -25,12 +25,18 @@ from .util import _orm_full_deannotate
 from .. import exc as sa_exc
 from .. import inspect
 from .. import util
+from ..sql import and_
 from ..sql import coercions
 from ..sql import roles
 from ..sql import visitors
 from ..sql.base import _generative
 from ..sql.base import Generative
 
+if util.TYPE_CHECKING:
+    from .context import QueryContext
+    from typing import Sequence
+    from ..sql.elements import ColumnElement
+
 
 class Load(Generative, LoaderOption):
     """Represents loader options which modify the state of a
@@ -108,6 +114,31 @@ class Load(Generative, LoaderOption):
         load._extra_criteria = ()
         return load
 
+    def _generate_extra_criteria(self, context):
+        # type: (QueryContext) -> Sequence[ColumnElement]
+        """Apply the current bound parameters in a QueryContext to the
+        "extra_criteria" stored with this Load object.
+
+        Load objects are typically pulled from the cached version of
+        the statement from a QueryContext.  The statement currently being
+        executed will have new values (and keys) for bound parameters in the
+        extra criteria which need to be applied by loader strategies when
+        they handle this criteria for a result set.
+
+        """
+
+        assert (
+            self._extra_criteria
+        ), "this should only be called if _extra_criteria is present"
+
+        orig_query = context.compile_state.select_statement
+        current_query = context.query
+
+        k1 = orig_query._generate_cache_key()
+        k2 = current_query._generate_cache_key()
+
+        return k2._apply_params_to_element(k1, and_(*self._extra_criteria))
+
     @property
     def _context_cache_key(self):
         serialized = []
@@ -488,6 +519,10 @@ class Load(Generative, LoaderOption):
 
     def __getstate__(self):
         d = self.__dict__.copy()
+
+        # can't pickle this right now; warning is raised by strategies
+        d["_extra_criteria"] = ()
+
         if d["context"] is not None:
             d["context"] = PathRegistry.serialize_context_dict(
                 d["context"], ("loader",)
@@ -623,6 +658,10 @@ class _UnboundLoad(Load):
 
     def __getstate__(self):
         d = self.__dict__.copy()
+
+        # can't pickle this right now; warning is raised by strategies
+        d["_extra_criteria"] = ()
+
         d["path"] = self._serialize_path(self.path, filter_aliased_class=True)
         return d
 
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]