Baked query needs to spoil fully on uncachable option

Fixed issue in the area of where loader options such as selectinload()
interact with the baked query system, such that the caching of a query is
not supposed to occur if the loader options themselves have elements such
as with_polymorphic() objects in them that currently are not
cache-compatible.  The baked loader could sometimes not fully invalidate
itself in these some of these scenarios leading to missed eager loads.

Fixes: #5303
Change-Id: Iecf847204a619694d89297f83b63b613ef9767de

4 files changed, 123 insertions, 37 deletions

diff --git a/doc/build/changelog/unreleased_13/5303.rst b/doc/build/changelog/unreleased_13/5303.rst
new file mode 100644
index 000000000..38358a0fe
--- /dev/null
+++ b/doc/build/changelog/unreleased_13/5303.rst
@@ -0,0 +1,11 @@
+.. change::
+    :tags: bug, orm
+    :tickets: 5303
+
+    Fixed issue in the area of where loader options such as selectinload()
+    interact with the baked query system, such that the caching of a query is
+    not supposed to occur if the loader options themselves have elements such
+    as with_polymorphic() objects in them that currently are not
+    cache-compatible.  The baked loader could sometimes not fully invalidate
+    itself in these some of these scenarios leading to missed eager loads.
+
diff --git a/lib/sqlalchemy/ext/baked.py b/lib/sqlalchemy/ext/baked.py
index ca07be784..a9c79d6bd 100644
--- a/lib/sqlalchemy/ext/baked.py
+++ b/lib/sqlalchemy/ext/baked.py
@@ -199,12 +199,12 @@ class BakedQuery(object):
         if cache_path.path[0].is_aliased_class:
             # paths that are against an AliasedClass are unsafe to cache
             # with since the AliasedClass is an ad-hoc object.
-            self.spoil()
+            self.spoil(full=True)
         else:
             for opt in options:
                 cache_key = opt._generate_path_cache_key(cache_path)
                 if cache_key is False:
-                    self.spoil()
+                    self.spoil(full=True)
                 elif cache_key is not None:
                     key += cache_key
 
diff --git a/test/ext/test_baked.py b/test/ext/test_baked.py
index 47c2c57b1..d36a646dd 100644
--- a/test/ext/test_baked.py
+++ b/test/ext/test_baked.py
@@ -1201,7 +1201,7 @@ class LazyLoaderTest(testing.AssertsCompiledSQL, BakedTest):
                 ad.dingalings
         l2 = len(lru)
         eq_(l1, 0)
-        eq_(l2, 1)
+        eq_(l2, 0)
 
     def test_unsafe_bound_option_cancels_bake(self):
         User, Address, Dingaling = self._o2m_twolevel_fixture(lazy="joined")
@@ -1232,7 +1232,7 @@ class LazyLoaderTest(testing.AssertsCompiledSQL, BakedTest):
                 ad.dingalings
         l2 = len(lru)
         eq_(l1, 0)
-        eq_(l2, 1)
+        eq_(l2, 0)
 
     def test_safe_unbound_option_allows_bake(self):
         User, Address, Dingaling = self._o2m_twolevel_fixture(lazy="joined")
diff --git a/test/orm/test_selectin_relations.py b/test/orm/test_selectin_relations.py
index d89fb4949..8bac71c1b 100644
--- a/test/orm/test_selectin_relations.py
+++ b/test/orm/test_selectin_relations.py
@@ -2850,40 +2850,40 @@ class SelfRefInheritanceAliasedTest(
     def test_twolevel_selectin_w_polymorphic(self):
         Foo, Bar = self.classes("Foo", "Bar")
 
-        r = with_polymorphic(Foo, "*", aliased=True)
-        attr1 = Foo.foo.of_type(r)
-        attr2 = r.foo
+        for count in range(3):
+            r = with_polymorphic(Foo, "*", aliased=True)
+            attr1 = Foo.foo.of_type(r)
+            attr2 = r.foo
 
-        s = Session()
-        q = (
-            s.query(Foo)
-            .filter(Foo.id == 2)
-            .options(selectinload(attr1).selectinload(attr2))
-        )
-        results = self.assert_sql_execution(
-            testing.db,
-            q.all,
-            CompiledSQL(
-                "SELECT foo.id AS foo_id_1, foo.type AS foo_type, "
-                "foo.foo_id AS foo_foo_id FROM foo WHERE foo.id = :id_1",
-                [{"id_1": 2}],
-            ),
-            CompiledSQL(
-                "SELECT foo_1.id AS foo_1_id, "
-                "foo_1.type AS foo_1_type, foo_1.foo_id AS foo_1_foo_id "
-                "FROM foo AS foo_1 "
-                "WHERE foo_1.id IN ([POSTCOMPILE_primary_keys])",
-                {"primary_keys": [3]},
-            ),
-            CompiledSQL(
-                "SELECT foo_1.id AS foo_1_id, "
-                "foo_1.type AS foo_1_type, foo_1.foo_id AS foo_1_foo_id "
-                "FROM foo AS foo_1 "
-                "WHERE foo_1.id IN ([POSTCOMPILE_primary_keys])",
-                {"primary_keys": [1]},
-            ),
-        )
-        eq_(results, [Bar(id=2, foo=Foo(id=3, foo=Bar(id=1)))])
+            s = Session()
+            q = (
+                s.query(Foo)
+                .filter(Foo.id == 2)
+                .options(selectinload(attr1).selectinload(attr2))
+            )
+            results = self.assert_sql_execution(
+                testing.db,
+                q.all,
+                CompiledSQL(
+                    "SELECT foo.id AS foo_id_1, foo.type AS foo_type, "
+                    "foo.foo_id AS foo_foo_id FROM foo WHERE foo.id = :id_1",
+                    [{"id_1": 2}],
+                ),
+                CompiledSQL(
+                    "SELECT foo_1.id AS foo_1_id, "
+                    "foo_1.type AS foo_1_type, foo_1.foo_id AS foo_1_foo_id "
+                    "FROM foo AS foo_1 "
+                    "WHERE foo_1.id IN ([POSTCOMPILE_primary_keys])",
+                    {"primary_keys": [3]},
+                ),
+                CompiledSQL(
+                    "SELECT foo.id AS foo_id_1, foo.type AS foo_type, "
+                    "foo.foo_id AS foo_foo_id FROM foo "
+                    "WHERE foo.id IN ([POSTCOMPILE_primary_keys])",
+                    {"primary_keys": [1]},
+                ),
+            )
+            eq_(results, [Bar(id=2, foo=Foo(id=3, foo=Bar(id=1)))])
 
 
 class TestExistingRowPopulation(fixtures.DeclarativeMappedTest):
@@ -3421,3 +3421,78 @@ class SameNamePolymorphicTest(fixtures.DeclarativeMappedTest):
                 ),
             ),
         )
+
+
+class TestBakedCancelsCorrectly(fixtures.DeclarativeMappedTest):
+    # test issue #5303
+
+    @classmethod
+    def setup_classes(cls):
+        Base = cls.DeclarativeBasic
+
+        class User(Base):
+            __tablename__ = "users"
+
+            id = Column(Integer, primary_key=True)
+
+        class Foo(Base):
+            __tablename__ = "foos"
+            __mapper_args__ = {"polymorphic_on": "type"}
+
+            id = Column(Integer, primary_key=True)
+            type = Column(String(50), nullable=False)
+
+        class SubFoo(Foo):
+            __tablename__ = "foos_sub"
+            __mapper_args__ = {"polymorphic_identity": "USER"}
+
+            id = Column(Integer, ForeignKey("foos.id"), primary_key=True)
+            user_id = Column(Integer, ForeignKey("users.id"))
+            user = relationship("User")
+
+        class Bar(Base):
+            __tablename__ = "bars"
+
+            id = Column(Integer, primary_key=True)
+            foo_id = Column(Integer, ForeignKey("foos.id"))
+            foo = relationship("Foo", cascade="all", uselist=False)
+
+    @classmethod
+    def insert_data(cls, connection):
+        User, Bar, SubFoo = cls.classes("User", "Bar", "SubFoo")
+
+        session = Session(connection)
+
+        user = User()
+        sub_foo = SubFoo(user=user)
+        sub_sub_bar = Bar(foo=sub_foo)
+        session.add_all([user, sub_foo, sub_sub_bar])
+        session.commit()
+
+    def test_option_accepted_each_time(self):
+        Foo, User, Bar, SubFoo = self.classes("Foo", "User", "Bar", "SubFoo")
+
+        def go():
+            # in this test, the loader options cancel caching because
+            # the with_polymorphic() can't be cached, and this actually
+            # fails because it won't match up to the with_polymorphic
+            # used in the query if the query is in fact cached.  however
+            # the cache spoil did not use full=True which kept the lead
+            # entities around.
+
+            sess = Session()
+            foo_polymorphic = with_polymorphic(Foo, [SubFoo], aliased=True)
+
+            credit_adjustment_load = selectinload(
+                Bar.foo.of_type(foo_polymorphic)
+            )
+            user_load = credit_adjustment_load.joinedload(
+                foo_polymorphic.SubFoo.user
+            )
+            query = sess.query(Bar).options(user_load)
+            ledger_entry = query.first()
+            ledger_entry.foo.user
+
+        self.assert_sql_count(testing.db, go, 2)
+        self.assert_sql_count(testing.db, go, 2)
+        self.assert_sql_count(testing.db, go, 2)