Merge "Ensure BakedQuery is cloned before we add options to it"

4 files changed, 186 insertions, 5 deletions

diff --git a/doc/build/changelog/unreleased_12/4286.rst b/doc/build/changelog/unreleased_12/4286.rst
new file mode 100644
index 000000000..ff3cc81ed
--- /dev/null
+++ b/doc/build/changelog/unreleased_12/4286.rst
@@ -0,0 +1,8 @@
+.. change::
+    :tags: bug, orm
+    :tickets: 4286
+
+    Fixed bug in new polymorphic selectin loading where the BakedQuery used
+    internally would be mutated by the given loader options, which would both
+    inappropriately mutate the subclass query as well as carry over the effect
+    to subsequent queries.
diff --git a/lib/sqlalchemy/ext/baked.py b/lib/sqlalchemy/ext/baked.py
index 79457e86e..addec90da 100644
--- a/lib/sqlalchemy/ext/baked.py
+++ b/lib/sqlalchemy/ext/baked.py
@@ -154,6 +154,13 @@ class BakedQuery(object):
         self._spoiled = True
         return self
 
+    def _with_lazyload_options(self, options, effective_path, cache_path=None):
+        """Cloning version of _add_lazyload_options.
+        """
+        q = self._clone()
+        q._add_lazyload_options(options, effective_path, cache_path=cache_path)
+        return q
+
     def _add_lazyload_options(self, options, effective_path, cache_path=None):
         """Used by per-state lazy loaders to add options to the
         "lazy load" query from a parent query.
diff --git a/lib/sqlalchemy/orm/loading.py b/lib/sqlalchemy/orm/loading.py
index a169845d4..0a6f8023a 100644
--- a/lib/sqlalchemy/orm/loading.py
+++ b/lib/sqlalchemy/orm/loading.py
@@ -580,17 +580,17 @@ def _load_subclass_via_in(context, path, entity):
     def do_load(context, path, states, load_only, effective_entity):
         orig_query = context.query
 
-        q._add_lazyload_options(
+        q2 = q._with_lazyload_options(
             (enable_opt, ) + orig_query._with_options + (disable_opt, ),
             path.parent, cache_path=path
         )
 
         if orig_query._populate_existing:
-            q.add_criteria(
+            q2.add_criteria(
                 lambda q: q.populate_existing()
             )
 
-        q(context.session).params(
+        q2(context.session).params(
             primary_keys=[
                 state.key[1][0] if zero_idx else state.key[1]
                 for state, load_attrs in states
diff --git a/test/orm/inheritance/test_poly_loading.py b/test/orm/inheritance/test_poly_loading.py
index b07b498a6..b724e8dba 100644
--- a/test/orm/inheritance/test_poly_loading.py
+++ b/test/orm/inheritance/test_poly_loading.py
@@ -1,6 +1,6 @@
 from sqlalchemy import String, Integer, Column, ForeignKey
-from sqlalchemy.orm import relationship, Session, \
-    selectin_polymorphic, selectinload, with_polymorphic
+from sqlalchemy.orm import relationship, Session, joinedload, \
+    selectin_polymorphic, selectinload, with_polymorphic, backref
 from sqlalchemy.testing import fixtures
 from sqlalchemy import testing
 from sqlalchemy.testing import eq_
@@ -497,3 +497,169 @@ class TestGeometries(GeometryFixtureBase):
         # the poly loader won't locate a state limited to the "a1" mapper,
         # needs to test that it has states
         sess.query(a)._with_invoke_all_eagers(False).all()
+
+
+class LoaderOptionsTest(
+        fixtures.DeclarativeMappedTest, testing.AssertsExecutionResults):
+    @classmethod
+    def setup_classes(cls):
+        Base = cls.DeclarativeBasic
+
+        class Parent(fixtures.ComparableEntity, Base):
+            __tablename__ = 'parent'
+            id = Column(Integer, primary_key=True)
+
+        class Child(fixtures.ComparableEntity, Base):
+            __tablename__ = 'child'
+            id = Column(Integer, primary_key=True)
+            parent_id = Column(Integer, ForeignKey('parent.id'))
+            parent = relationship('Parent', backref=backref('children'))
+
+            type = Column(String(50), nullable=False)
+            __mapper_args__ = {
+                'polymorphic_on': type,
+            }
+
+        class ChildSubclass1(Child):
+            __tablename__ = 'child_subclass1'
+            id = Column(Integer, ForeignKey('child.id'), primary_key=True)
+            __mapper_args__ = {
+                'polymorphic_identity': 'subclass1',
+                'polymorphic_load': 'selectin'
+            }
+
+        class Other(fixtures.ComparableEntity, Base):
+            __tablename__ = 'other'
+
+            id = Column(Integer, primary_key=True)
+            child_subclass_id = Column(Integer,
+                                       ForeignKey('child_subclass1.id'))
+            child_subclass = relationship('ChildSubclass1',
+                                          backref=backref('others'))
+
+    @classmethod
+    def insert_data(cls):
+        Parent, ChildSubclass1, Other = cls.classes(
+            "Parent", "ChildSubclass1", "Other")
+        session = Session()
+
+        parent = Parent(id=1)
+        subclass1 = ChildSubclass1(id=1, parent=parent)
+        other = Other(id=1, child_subclass=subclass1)
+        session.add_all([parent, subclass1, other])
+        session.commit()
+
+    def test_options_dont_pollute_baked(self):
+        self._test_options_dont_pollute(True)
+
+    def test_options_dont_pollute_unbaked(self):
+        self._test_options_dont_pollute(False)
+
+    def _test_options_dont_pollute(self, enable_baked):
+        Parent, ChildSubclass1, Other = self.classes(
+            "Parent", "ChildSubclass1", "Other")
+        session = Session(enable_baked_queries=enable_baked)
+
+        def no_opt():
+            q = session.query(Parent).options(
+                joinedload(Parent.children.of_type(ChildSubclass1)))
+
+            return self.assert_sql_execution(
+                testing.db,
+                q.all,
+                CompiledSQL(
+                    "SELECT parent.id AS parent_id, "
+                    "anon_1.child_id AS anon_1_child_id, "
+                    "anon_1.child_parent_id AS anon_1_child_parent_id, "
+                    "anon_1.child_type AS anon_1_child_type, "
+                    "anon_1.child_subclass1_id AS anon_1_child_subclass1_id "
+                    "FROM parent "
+                    "LEFT OUTER JOIN (SELECT child.id AS child_id, "
+                    "child.parent_id AS child_parent_id, "
+                    "child.type AS child_type, "
+                    "child_subclass1.id AS child_subclass1_id "
+                    "FROM child "
+                    "LEFT OUTER JOIN child_subclass1 "
+                    "ON child.id = child_subclass1.id) AS anon_1 "
+                    "ON parent.id = anon_1.child_parent_id",
+                    {}
+                ),
+                CompiledSQL(
+                    "SELECT child_subclass1.id AS child_subclass1_id, "
+                    "child.id AS child_id, "
+                    "child.parent_id AS child_parent_id, "
+                    "child.type AS child_type "
+                    "FROM child JOIN child_subclass1 "
+                    "ON child.id = child_subclass1.id "
+                    "WHERE child.id IN ([EXPANDING_primary_keys]) "
+                    "ORDER BY child.id",
+                    [{'primary_keys': [1]}]
+                ),
+            )
+
+        result = no_opt()
+        with self.assert_statement_count(testing.db, 1):
+            eq_(
+                result,
+                [Parent(children=[ChildSubclass1(others=[Other()])])]
+            )
+
+        session.expunge_all()
+
+        q = session.query(Parent).options(
+            joinedload(Parent.children.of_type(ChildSubclass1))
+            .joinedload(ChildSubclass1.others)
+        )
+
+        result = self.assert_sql_execution(
+            testing.db,
+            q.all,
+            CompiledSQL(
+                "SELECT parent.id AS parent_id, "
+                "anon_1.child_id AS anon_1_child_id, "
+                "anon_1.child_parent_id AS anon_1_child_parent_id, "
+                "anon_1.child_type AS anon_1_child_type, "
+                "anon_1.child_subclass1_id AS anon_1_child_subclass1_id, "
+                "other_1.id AS other_1_id, "
+                "other_1.child_subclass_id AS other_1_child_subclass_id "
+                "FROM parent LEFT OUTER JOIN "
+                "(SELECT child.id AS child_id, "
+                "child.parent_id AS child_parent_id, "
+                "child.type AS child_type, "
+                "child_subclass1.id AS child_subclass1_id "
+                "FROM child LEFT OUTER JOIN child_subclass1 "
+                "ON child.id = child_subclass1.id) AS anon_1 "
+                "ON parent.id = anon_1.child_parent_id "
+                "LEFT OUTER JOIN other AS other_1 "
+                "ON anon_1.child_subclass1_id = other_1.child_subclass_id",
+                {}
+            ),
+            CompiledSQL(
+                "SELECT child_subclass1.id AS child_subclass1_id, "
+                "child.id AS child_id, child.parent_id AS child_parent_id, "
+                "child.type AS child_type, other_1.id AS other_1_id, "
+                "other_1.child_subclass_id AS other_1_child_subclass_id "
+                "FROM child JOIN child_subclass1 "
+                "ON child.id = child_subclass1.id "
+                "LEFT OUTER JOIN other AS other_1 "
+                "ON child_subclass1.id = other_1.child_subclass_id "
+                "WHERE child.id IN ([EXPANDING_primary_keys]) "
+                "ORDER BY child.id",
+                [{'primary_keys': [1]}]
+            )
+        )
+
+        with self.assert_statement_count(testing.db, 0):
+            eq_(
+                result,
+                [Parent(children=[ChildSubclass1(others=[Other()])])]
+            )
+
+        session.expunge_all()
+
+        result = no_opt()
+        with self.assert_statement_count(testing.db, 1):
+            eq_(
+                result,
+                [Parent(children=[ChildSubclass1(others=[Other()])])]
+            )