Memoize load_path in all cases, run quick populators for path change

Adds a new variant to the "isnew" state within entity loading
for isnew=False, but the load path is new.  This is to address
the use case of an entity appearing in multiple places in
the row in a more generalized way than the fixes in [ticket:3431],
[ticket:3811] in that loading.py will be able to tell the
populator that this row is not "isnew" but is a "new" path
for the entity.   For the moment, the new information is only
being applied to the use of "quick" populators so that
simple column loads can take place on top of a deferred loader
from elsewhere in the row.

As part of this change, state.load_path() will now always
be populated with the "path" that was in effect when this state
was originally loaded, which for multi-path loads of the
same entity is still non-deterministic.  Ideally there'd be some
kind of "here's all the paths that loaded this state and how"
type of data structure though it's not clear if that could be
done while maintaining performance.

Fixes: #3822
Change-Id: Ib915365353dfcca09e15c24001a8581113b97d5e

6 files changed, 137 insertions, 15 deletions

diff --git a/doc/build/changelog/changelog_11.rst b/doc/build/changelog/changelog_11.rst
index 127227ded..45d98aa22 100644
--- a/doc/build/changelog/changelog_11.rst
+++ b/doc/build/changelog/changelog_11.rst
@@ -40,6 +40,17 @@
         in [ticket:1495], where the rule would fail if the parent object
         had some other lazyloader-bound query options associated with it.
 
+    .. change::
+        :tags: bug, orm
+        :tickets: 3822
+
+        Fixed self-referential entity, deferred column loading issue in a
+        similar style as that of [ticket:3431], [ticket:3811] where an entity
+        is present in multiple positions within the row due to self-referential
+        eager loading; when the deferred loader only applies to one of the
+        paths, the "present" column loader will now override the deferred non-
+        load for that entity regardless of row ordering.
+
 .. changelog::
     :version: 1.1.1
     :released: October 7, 2016
diff --git a/lib/sqlalchemy/ext/baked.py b/lib/sqlalchemy/ext/baked.py
index 2f658edf3..a6191f5cb 100644
--- a/lib/sqlalchemy/ext/baked.py
+++ b/lib/sqlalchemy/ext/baked.py
@@ -441,14 +441,12 @@ class BakedLazyLoader(strategies.LazyLoader):
         if pending or passive & attributes.NO_AUTOFLUSH:
             q.add_criteria(lambda q: q.autoflush(False))
 
-        if state.load_path:
+        if state.load_options:
             q.spoil()
+            args = state.load_path[self.parent_property]
             q.add_criteria(
                 lambda q:
-                q._with_current_path(state.load_path[self.parent_property]))
-
-        if state.load_options:
-            q.spoil()
+                q._with_current_path(args), args)
             q.add_criteria(
                 lambda q: q._conditional_options(*state.load_options))
 
diff --git a/lib/sqlalchemy/orm/loading.py b/lib/sqlalchemy/orm/loading.py
index d457f3c63..5ee8f6abe 100644
--- a/lib/sqlalchemy/orm/loading.py
+++ b/lib/sqlalchemy/orm/loading.py
@@ -330,9 +330,8 @@ def _instance_processor(
                 context, path, mapper, result, adapter, populators)
 
     propagate_options = context.propagate_options
-    if propagate_options:
-        load_path = context.query._current_path + path \
-            if context.query._current_path.path else path
+    load_path = context.query._current_path + path \
+        if context.query._current_path.path else path
 
     session_identity_map = context.session.identity_map
 
@@ -426,12 +425,15 @@ def _instance_processor(
             # full population routines.  Objects here are either
             # just created, or we are doing a populate_existing
 
-            if isnew and propagate_options:
+            # be conservative about setting load_path when populate_existing
+            # is in effect; want to maintain options from the original
+            # load.  see test_expire->test_refresh_maintains_deferred_options
+            if isnew and (propagate_options or not populate_existing):
                 state.load_options = propagate_options
                 state.load_path = load_path
 
             _populate_full(
-                context, row, state, dict_, isnew,
+                context, row, state, dict_, isnew, load_path,
                 loaded_instance, populate_existing, populators)
 
             if isnew:
@@ -463,7 +465,7 @@ def _instance_processor(
                 # and add to the "context.partials" collection.
 
                 to_load = _populate_partial(
-                    context, row, state, dict_, isnew,
+                    context, row, state, dict_, isnew, load_path,
                     unloaded, populators)
 
                 if isnew:
@@ -486,7 +488,7 @@ def _instance_processor(
 
 
 def _populate_full(
-        context, row, state, dict_, isnew,
+        context, row, state, dict_, isnew, load_path,
         loaded_instance, populate_existing, populators):
     if isnew:
         # first time we are seeing a row with this identity.
@@ -507,15 +509,37 @@ def _populate_full(
             populator(state, dict_, row)
         for key, populator in populators["delayed"]:
             populator(state, dict_, row)
+    elif load_path != state.load_path:
+        # new load path, e.g. object is present in more than one
+        # column position in a series of rows
+        state.load_path = load_path
+
+        # if we have data, and the data isn't in the dict, OK, let's put
+        # it in.
+        for key, getter in populators["quick"]:
+            if key not in dict_:
+                dict_[key] = getter(row)
+
+        # otherwise treat like an "already seen" row
+        for key, populator in populators["existing"]:
+            populator(state, dict_, row)
+            # TODO:  allow "existing" populator to know this is
+            # a new path for the state:
+            # populator(state, dict_, row, new_path=True)
+
     else:
-        # have already seen rows with this identity.
+        # have already seen rows with this identity in this same path.
         for key, populator in populators["existing"]:
             populator(state, dict_, row)
 
+            # TODO: same path
+            # populator(state, dict_, row, new_path=False)
+
 
 def _populate_partial(
-        context, row, state, dict_, isnew,
+        context, row, state, dict_, isnew, load_path,
         unloaded, populators):
+
     if not isnew:
         to_load = context.partials[state]
         for key, populator in populators["existing"]:
diff --git a/test/orm/test_deferred.py b/test/orm/test_deferred.py
index 957e1d419..1ae6cbd89 100644
--- a/test/orm/test_deferred.py
+++ b/test/orm/test_deferred.py
@@ -680,6 +680,54 @@ class DeferredOptionsTest(AssertsCompiledSQL, _fixtures.FixtureTest):
         )
 
 
+class SelfReferentialMultiPathTest(testing.fixtures.DeclarativeMappedTest):
+    """test for [ticket:3822]"""
+
+    @classmethod
+    def setup_classes(cls):
+        Base = cls.DeclarativeBasic
+
+        class Node(Base):
+            __tablename__ = 'node'
+
+            id = sa.Column(sa.Integer, primary_key=True)
+            parent_id = sa.Column(sa.ForeignKey('node.id'))
+            parent = relationship('Node', remote_side=[id])
+            name = sa.Column(sa.String(10))
+
+    @classmethod
+    def insert_data(cls):
+        Node = cls.classes.Node
+
+        session = Session()
+        session.add_all([
+            Node(id=1, name='name'),
+            Node(id=2, parent_id=1, name='name'),
+            Node(id=3, parent_id=1, name='name')
+        ])
+        session.commit()
+
+    def test_present_overrides_deferred(self):
+        Node = self.classes.Node
+
+        session = Session()
+
+        q = session.query(Node).options(
+            joinedload(Node.parent).load_only(Node.id, Node.parent_id)
+        )
+
+        # Node #1 will appear first as Node.parent and have
+        # deferred applied to Node.name.  it will then appear
+        # as Node in the last row and "name" should be populated.
+        nodes = q.order_by(Node.id.desc()).all()
+
+        def go():
+            for node in nodes:
+                eq_(node.name, 'name')
+
+        self.assert_sql_count(testing.db, go, 0)
+
+
 class InheritanceTest(_Polymorphic):
     __dialect__ = 'default'
 
diff --git a/test/orm/test_expire.py b/test/orm/test_expire.py
index 158957ebb..5399bff80 100644
--- a/test/orm/test_expire.py
+++ b/test/orm/test_expire.py
@@ -1435,6 +1435,47 @@ class RefreshTest(_fixtures.FixtureTest):
         s.expire(u)
         assert len(u.addresses) == 3
 
+    def test_refresh_maintains_deferred_options(self):
+        # testing a behavior that may have changed with
+        # [ticket:3822]
+        User, Address, Dingaling = self.classes(
+            "User", "Address", "Dingaling")
+        users, addresses, dingalings = self.tables(
+            "users", "addresses", "dingalings")
+
+        mapper(User, users, properties={
+            'addresses': relationship(Address)
+        })
+
+        mapper(Address, addresses, properties={
+            'dingalings': relationship(Dingaling)
+        })
+
+        mapper(Dingaling, dingalings)
+
+        s = create_session()
+        q = s.query(User).filter_by(name='fred').options(
+            sa.orm.lazyload('addresses').joinedload("dingalings"))
+
+        u1 = q.one()
+
+        # "addresses" is not present on u1, but when u1.addresses
+        # lazy loads, it should also joinedload dingalings.  This is
+        # present in state.load_options and state.load_path.   The
+        # refresh operation should not reset these attributes.
+        s.refresh(u1)
+
+        def go():
+            eq_(
+                u1.addresses,
+                [Address(
+                    email_address='fred@fred.com',
+                    dingalings=[Dingaling(data="ding 2/5")]
+                )]
+            )
+
+        self.assert_sql_count(testing.db, go, 1)
+
     def test_refresh2(self):
         """test a hang condition that was occurring on expire/refresh"""
 
diff --git a/test/orm/test_merge.py b/test/orm/test_merge.py
index 8a3419ecc..2676ebc4a 100644
--- a/test/orm/test_merge.py
+++ b/test/orm/test_merge.py
@@ -1124,7 +1124,7 @@ class MergeTest(_fixtures.FixtureTest):
 
         for u in s1_users:
             ustate = attributes.instance_state(u)
-            eq_(ustate.load_path, ())
+            eq_(ustate.load_path.path, (umapper, ))
             eq_(ustate.load_options, set())
 
         for u in s2_users: