Ensure BakedQuery is cloned before we add options to it

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.

Change-Id: Iaceecb50557f78484d09e55b3029a0483dfe873f
Fixes: #4286

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()])])]
+            )