Iterate options per path for baked cache key

Fixed an issue that was both a performance regression in 1.2 as well as an
incorrect result regarding the "baked" lazy loader, involving the
generation of cache keys from the original :class:`.Query` object's loader
options.  If the loader options were built up in a "branched" style using
common base elements for multiple options, the same options would be
rendered into the cache key repeatedly, causing both a performance issue as
well as generating the wrong cache key.  This is fixed, along with a
performance improvement when such "branched" options are applied via
:meth:`.Query.options` to prevent the same option objects from being
applied repeatedly.

Change-Id: I955fe2f50186abd8e753ad490fd3eb8f017e26f9
Fixes: #4270

2 files changed, 68 insertions, 55 deletions

diff --git a/lib/sqlalchemy/orm/query.py b/lib/sqlalchemy/orm/query.py
index 56e42a702..7bfffa04b 100644
--- a/lib/sqlalchemy/orm/query.py
+++ b/lib/sqlalchemy/orm/query.py
@@ -1411,6 +1411,8 @@ class Query(object):
         # most MapperOptions write to the '_attributes' dictionary,
         # so copy that as well
         self._attributes = self._attributes.copy()
+        if '_unbound_load_dedupes' not in self._attributes:
+            self._attributes['_unbound_load_dedupes'] = set()
         opts = tuple(util.flatten_iterator(args))
         self._with_options = self._with_options + opts
         if conditional:
diff --git a/lib/sqlalchemy/orm/strategy_options.py b/lib/sqlalchemy/orm/strategy_options.py
index 43f571146..f54020fb7 100644
--- a/lib/sqlalchemy/orm/strategy_options.py
+++ b/lib/sqlalchemy/orm/strategy_options.py
@@ -83,50 +83,54 @@ class Load(Generative, MapperOption):
             if key != "loader":
                 continue
 
-            endpoint = obj._of_type or obj.path.path[-1]
-            chopped = self._chop_path(loader_path, path)
-
-            if (
-                # means loader_path and path are unrelated,
-                # this does not need to be part of a cache key
-                chopped is None
-            ) or (
-                # means no additional path with loader_path + path
-                # and the endpoint isn't using of_type so isn't modified into
-                # an alias or other unsafe entity
-                not chopped and not obj._of_type
-            ):
-                continue
-
-            serialized_path = []
-
-            for token in chopped:
-                if isinstance(token, util.string_types):
-                    serialized_path.append(token)
-                elif token.is_aliased_class:
-                    return False
-                elif token.is_property:
-                    serialized_path.append(token.key)
-                else:
-                    assert token.is_mapper
-                    serialized_path.append(token.class_)
-
-            if not serialized_path or endpoint != serialized_path[-1]:
-                if endpoint.is_mapper:
-                    serialized_path.append(endpoint.class_)
-                elif endpoint.is_aliased_class:
-                    return False
-
-            serialized.append(
-                (
-                    tuple(serialized_path) +
-                    (obj.strategy or ()) +
-                    (tuple([
-                        (key, obj.local_opts[key])
-                        for key in sorted(obj.local_opts)
-                    ]) if obj.local_opts else ())
+            for local_elem, obj_elem in zip(self.path.path, loader_path):
+                if local_elem is not obj_elem:
+                    break
+            else:
+                endpoint = obj._of_type or obj.path.path[-1]
+                chopped = self._chop_path(loader_path, path)
+
+                if (
+                    # means loader_path and path are unrelated,
+                    # this does not need to be part of a cache key
+                    chopped is None
+                ) or (
+                    # means no additional path with loader_path + path
+                    # and the endpoint isn't using of_type so isn't modified
+                    # into an alias or other unsafe entity
+                    not chopped and not obj._of_type
+                ):
+                    continue
+
+                serialized_path = []
+
+                for token in chopped:
+                    if isinstance(token, util.string_types):
+                        serialized_path.append(token)
+                    elif token.is_aliased_class:
+                        return False
+                    elif token.is_property:
+                        serialized_path.append(token.key)
+                    else:
+                        assert token.is_mapper
+                        serialized_path.append(token.class_)
+
+                if not serialized_path or endpoint != serialized_path[-1]:
+                    if endpoint.is_mapper:
+                        serialized_path.append(endpoint.class_)
+                    elif endpoint.is_aliased_class:
+                        return False
+
+                serialized.append(
+                    (
+                        tuple(serialized_path) +
+                        (obj.strategy or ()) +
+                        (tuple([
+                            (key, obj.local_opts[key])
+                            for key in sorted(obj.local_opts)
+                        ]) if obj.local_opts else ())
+                    )
                 )
-            )
         if not serialized:
             return None
         else:
@@ -407,15 +411,19 @@ class _UnboundLoad(Load):
     def _generate_cache_key(self, path):
         serialized = ()
         for val in self._to_bind:
-            opt = val._bind_loader(
-                [path.path[0]],
-                None, None, False)
-            if opt:
-                c_key = opt._generate_cache_key(path)
-                if c_key is False:
-                    return False
-                elif c_key:
-                    serialized += c_key
+            for local_elem, val_elem in zip(self.path, val.path):
+                if local_elem is not val_elem:
+                    break
+            else:
+                opt = val._bind_loader(
+                    [path.path[0]],
+                    None, None, False)
+                if opt:
+                    c_key = opt._generate_cache_key(path)
+                    if c_key is False:
+                        return False
+                    elif c_key:
+                        serialized += c_key
         if not serialized:
             return None
         else:
@@ -462,10 +470,13 @@ class _UnboundLoad(Load):
         self.__dict__ = state
 
     def _process(self, query, raiseerr):
+        dedupes = query._attributes['_unbound_load_dedupes']
         for val in self._to_bind:
-            val._bind_loader(
-                [ent.entity_zero for ent in query._mapper_entities],
-                query._current_path, query._attributes, raiseerr)
+            if val not in dedupes:
+                dedupes.add(val)
+                val._bind_loader(
+                    [ent.entity_zero for ent in query._mapper_entities],
+                    query._current_path, query._attributes, raiseerr)
 
     @classmethod
     def _from_keys(cls, meth, keys, chained, kw):