Merge "Implement relationship to AliasedClass; deprecate non primary mappers"

17 files changed, 1245 insertions, 397 deletions

diff --git a/test/aaa_profiling/test_memusage.py b/test/aaa_profiling/test_memusage.py
index 0915244f7..36f28c063 100644
--- a/test/aaa_profiling/test_memusage.py
+++ b/test/aaa_profiling/test_memusage.py
@@ -321,8 +321,6 @@ class MemUsageWBackendTest(EnsureZeroed):
         )
         m2 = mapper(B, table2)
 
-        m3 = mapper(A, table1, non_primary=True)
-
         @profile_memory()
         def go():
             sess = create_session()
@@ -354,7 +352,7 @@ class MemUsageWBackendTest(EnsureZeroed):
         go()
 
         metadata.drop_all()
-        del m1, m2, m3
+        del m1, m2
         assert_no_mappers()
 
     def test_sessionmaker(self):
@@ -415,8 +413,6 @@ class MemUsageWBackendTest(EnsureZeroed):
         )
         m2 = mapper(B, table2, _compiled_cache_size=50)
 
-        m3 = mapper(A, table1, non_primary=True)
-
         @profile_memory()
         def go():
             engine = engines.testing_engine(
@@ -458,7 +454,7 @@ class MemUsageWBackendTest(EnsureZeroed):
         go()
 
         metadata.drop_all()
-        del m1, m2, m3
+        del m1, m2
         assert_no_mappers()
 
     @testing.emits_warning("Compiled statement cache for.*")
@@ -642,8 +638,6 @@ class MemUsageWBackendTest(EnsureZeroed):
             )
             mapper(B, table2)
 
-            mapper(A, table1, non_primary=True)
-
             sess = create_session()
             a1 = A(col2="a1")
             a2 = A(col2="a2")
diff --git a/test/ext/test_deprecations.py b/test/ext/test_deprecations.py
new file mode 100644
index 000000000..099393cf7
--- /dev/null
+++ b/test/ext/test_deprecations.py
@@ -0,0 +1,37 @@
+from sqlalchemy import testing
+from sqlalchemy.orm import mapper
+from .test_mutable import Foo
+from .test_mutable import (
+    MutableAssociationScalarPickleTest as _MutableAssociationScalarPickleTest,
+)
+from .test_mutable import (
+    MutableWithScalarJSONTest as _MutableWithScalarJSONTest,
+)
+
+
+class MutableIncludeNonPrimaryTest(_MutableWithScalarJSONTest):
+    @classmethod
+    def setup_mappers(cls):
+        foo = cls.tables.foo
+
+        mapper(Foo, foo)
+        with testing.expect_deprecated(
+            "The mapper.non_primary parameter is deprecated"
+        ):
+            mapper(
+                Foo, foo, non_primary=True, properties={"foo_bar": foo.c.data}
+            )
+
+
+class MutableAssocIncludeNonPrimaryTest(_MutableAssociationScalarPickleTest):
+    @classmethod
+    def setup_mappers(cls):
+        foo = cls.tables.foo
+
+        mapper(Foo, foo)
+        with testing.expect_deprecated(
+            "The mapper.non_primary parameter is deprecated"
+        ):
+            mapper(
+                Foo, foo, non_primary=True, properties={"foo_bar": foo.c.data}
+            )
diff --git a/test/ext/test_mutable.py b/test/ext/test_mutable.py
index 9b8f761e9..eae764d40 100644
--- a/test/ext/test_mutable.py
+++ b/test/ext/test_mutable.py
@@ -824,15 +824,6 @@ class MutableWithScalarJSONTest(_MutableDictTestBase, fixtures.MappedTest):
         self._test_non_mutable()
 
 
-class MutableIncludeNonPrimaryTest(MutableWithScalarJSONTest):
-    @classmethod
-    def setup_mappers(cls):
-        foo = cls.tables.foo
-
-        mapper(Foo, foo)
-        mapper(Foo, foo, non_primary=True, properties={"foo_bar": foo.c.data})
-
-
 class MutableColumnCopyJSONTest(_MutableDictTestBase, fixtures.MappedTest):
     @classmethod
     def define_tables(cls, metadata):
@@ -1013,15 +1004,6 @@ class MutableAssociationScalarPickleTest(
         )
 
 
-class MutableAssocIncludeNonPrimaryTest(MutableAssociationScalarPickleTest):
-    @classmethod
-    def setup_mappers(cls):
-        foo = cls.tables.foo
-
-        mapper(Foo, foo)
-        mapper(Foo, foo, non_primary=True, properties={"foo_bar": foo.c.data})
-
-
 class MutableAssociationScalarJSONTest(
     _MutableDictTestBase, fixtures.MappedTest
 ):
diff --git a/test/orm/test_ac_relationships.py b/test/orm/test_ac_relationships.py
new file mode 100644
index 000000000..d5f9b013d
--- /dev/null
+++ b/test/orm/test_ac_relationships.py
@@ -0,0 +1,292 @@
+from sqlalchemy import and_
+from sqlalchemy import Column
+from sqlalchemy import ForeignKey
+from sqlalchemy import func
+from sqlalchemy import Integer
+from sqlalchemy import join
+from sqlalchemy import select
+from sqlalchemy import testing
+from sqlalchemy.orm import aliased
+from sqlalchemy.orm import joinedload
+from sqlalchemy.orm import noload
+from sqlalchemy.orm import relationship
+from sqlalchemy.orm import selectinload
+from sqlalchemy.orm import Session
+from sqlalchemy.testing import eq_
+from sqlalchemy.testing import fixtures
+from sqlalchemy.testing.assertsql import CompiledSQL
+from sqlalchemy.testing.fixtures import ComparableEntity
+
+
+class PartitionByFixture(fixtures.DeclarativeMappedTest):
+    @classmethod
+    def setup_classes(cls):
+        Base = cls.DeclarativeBasic
+
+        class A(Base):
+            __tablename__ = "a"
+
+            id = Column(Integer, primary_key=True)
+
+        class B(Base):
+            __tablename__ = "b"
+            id = Column(Integer, primary_key=True)
+            a_id = Column(ForeignKey("a.id"))
+            cs = relationship("C")
+
+        class C(Base):
+            __tablename__ = "c"
+            id = Column(Integer, primary_key=True)
+            b_id = Column(ForeignKey("b.id"))
+
+        partition = select(
+            [
+                B,
+                func.row_number()
+                .over(order_by=B.id, partition_by=B.a_id)
+                .label("index"),
+            ]
+        ).alias()
+
+        partitioned_b = aliased(B, alias=partition)
+
+        A.partitioned_bs = relationship(
+            partitioned_b,
+            primaryjoin=and_(
+                partitioned_b.a_id == A.id, partition.c.index < 10
+            ),
+        )
+
+    @classmethod
+    def insert_data(cls):
+        A, B, C = cls.classes("A", "B", "C")
+
+        s = Session(testing.db)
+        s.add_all([A(id=i) for i in range(1, 4)])
+        s.flush()
+        s.add_all(
+            [
+                B(a_id=i, cs=[C(), C()])
+                for i in range(1, 4)
+                for j in range(1, 21)
+            ]
+        )
+        s.commit()
+
+
+class AliasedClassRelationshipTest(
+    PartitionByFixture, testing.AssertsCompiledSQL
+):
+    # TODO: maybe make this more  backend agnostic
+    __requires__ = ("window_functions",)
+    __dialect__ = "default"
+
+    def test_lazyload(self):
+        A, B, C = self.classes("A", "B", "C")
+
+        s = Session(testing.db)
+
+        def go():
+            for a1 in s.query(A):  # 1 query
+                eq_(len(a1.partitioned_bs), 9)  # 3 queries
+                for b in a1.partitioned_bs:
+                    eq_(len(b.cs), 2)  # 9 * 3 = 27 queries
+
+        self.assert_sql_count(testing.db, go, 31)
+
+    def test_join_one(self):
+        A, B, C = self.classes("A", "B", "C")
+
+        s = Session(testing.db)
+
+        q = s.query(A).join(A.partitioned_bs)
+        self.assert_compile(
+            q,
+            "SELECT a.id AS a_id FROM a JOIN "
+            "(SELECT b.id AS id, b.a_id AS a_id, row_number() "
+            "OVER (PARTITION BY b.a_id ORDER BY b.id) "
+            "AS index FROM b) AS anon_1 "
+            "ON anon_1.a_id = a.id AND anon_1.index < :index_1",
+        )
+
+    def test_join_two(self):
+        A, B, C = self.classes("A", "B", "C")
+
+        s = Session(testing.db)
+
+        q = s.query(A, A.partitioned_bs.entity).join(A.partitioned_bs)
+        self.assert_compile(
+            q,
+            "SELECT a.id AS a_id, anon_1.id AS anon_1_id, "
+            "anon_1.a_id AS anon_1_a_id "
+            "FROM a JOIN "
+            "(SELECT b.id AS id, b.a_id AS a_id, row_number() "
+            "OVER (PARTITION BY b.a_id ORDER BY b.id) "
+            "AS index FROM b) AS anon_1 "
+            "ON anon_1.a_id = a.id AND anon_1.index < :index_1",
+        )
+
+    def test_selectinload_w_noload_after(self):
+        A, B, C = self.classes("A", "B", "C")
+
+        s = Session(testing.db)
+
+        def go():
+            for a1 in s.query(A).options(
+                noload("*"), selectinload(A.partitioned_bs)
+            ):
+                for b in a1.partitioned_bs:
+                    eq_(b.cs, [])
+
+        self.assert_sql_count(testing.db, go, 2)
+
+    def test_selectinload_w_joinedload_after(self):
+        A, B, C = self.classes("A", "B", "C")
+
+        s = Session(testing.db)
+
+        def go():
+            for a1 in s.query(A).options(
+                selectinload(A.partitioned_bs).joinedload("cs")
+            ):
+                for b in a1.partitioned_bs:
+                    eq_(len(b.cs), 2)
+
+        self.assert_sql_count(testing.db, go, 2)
+
+
+class AltSelectableTest(
+    fixtures.DeclarativeMappedTest, testing.AssertsCompiledSQL
+):
+    __dialect__ = "default"
+
+    @classmethod
+    def setup_classes(cls):
+        Base = cls.DeclarativeBasic
+
+        class A(ComparableEntity, Base):
+            __tablename__ = "a"
+
+            id = Column(Integer, primary_key=True)
+            b_id = Column(ForeignKey("b.id"))
+
+        class B(ComparableEntity, Base):
+            __tablename__ = "b"
+
+            id = Column(Integer, primary_key=True)
+
+        class C(ComparableEntity, Base):
+            __tablename__ = "c"
+
+            id = Column(Integer, primary_key=True)
+            a_id = Column(ForeignKey("a.id"))
+
+        class D(ComparableEntity, Base):
+            __tablename__ = "d"
+
+            id = Column(Integer, primary_key=True)
+            c_id = Column(ForeignKey("c.id"))
+            b_id = Column(ForeignKey("b.id"))
+
+        # 1. set up the join() as a variable, so we can refer
+        # to it in the mapping multiple times.
+        j = join(B, D, D.b_id == B.id).join(C, C.id == D.c_id)
+
+        # 2. Create an AliasedClass to B
+        B_viacd = aliased(B, j, flat=True)
+
+        A.b = relationship(B_viacd, primaryjoin=A.b_id == j.c.b_id)
+
+    @classmethod
+    def insert_data(cls):
+        A, B, C, D = cls.classes("A", "B", "C", "D")
+        sess = Session()
+
+        for obj in [
+            B(id=1),
+            A(id=1, b_id=1),
+            C(id=1, a_id=1),
+            D(id=1, c_id=1, b_id=1),
+        ]:
+            sess.add(obj)
+            sess.flush()
+        sess.commit()
+
+    def test_lazyload(self):
+        A, B = self.classes("A", "B")
+
+        sess = Session()
+        a1 = sess.query(A).first()
+
+        with self.sql_execution_asserter() as asserter:
+            # note this is many-to-one.  use_get is unconditionally turned
+            # off for relationship to aliased class for now.
+            eq_(a1.b, B(id=1))
+
+        asserter.assert_(
+            CompiledSQL(
+                "SELECT b.id AS b_id FROM b JOIN d ON d.b_id = b.id "
+                "JOIN c ON c.id = d.c_id WHERE :param_1 = b.id",
+                [{"param_1": 1}],
+            )
+        )
+
+    def test_joinedload(self):
+        A, B = self.classes("A", "B")
+
+        sess = Session()
+
+        with self.sql_execution_asserter() as asserter:
+            # note this is many-to-one.  use_get is unconditionally turned
+            # off for relationship to aliased class for now.
+            a1 = sess.query(A).options(joinedload(A.b)).first()
+            eq_(a1.b, B(id=1))
+
+        asserter.assert_(
+            CompiledSQL(
+                "SELECT a.id AS a_id, a.b_id AS a_b_id, b_1.id AS b_1_id "
+                "FROM a LEFT OUTER JOIN (b AS b_1 "
+                "JOIN d AS d_1 ON d_1.b_id = b_1.id "
+                "JOIN c AS c_1 ON c_1.id = d_1.c_id) ON a.b_id = b_1.id "
+                "LIMIT :param_1",
+                [{"param_1": 1}],
+            )
+        )
+
+    def test_selectinload(self):
+        A, B = self.classes("A", "B")
+
+        sess = Session()
+
+        with self.sql_execution_asserter() as asserter:
+            # note this is many-to-one.  use_get is unconditionally turned
+            # off for relationship to aliased class for now.
+            a1 = sess.query(A).options(selectinload(A.b)).first()
+            eq_(a1.b, B(id=1))
+
+        asserter.assert_(
+            CompiledSQL(
+                "SELECT a.id AS a_id, a.b_id AS a_b_id "
+                "FROM a LIMIT :param_1",
+                [{"param_1": 1}],
+            ),
+            CompiledSQL(
+                "SELECT a_1.id AS a_1_id, b.id AS b_id FROM a AS a_1 "
+                "JOIN (b JOIN d ON d.b_id = b.id JOIN c ON c.id = d.c_id) "
+                "ON a_1.b_id = b.id WHERE a_1.id "
+                "IN ([EXPANDING_primary_keys]) ORDER BY a_1.id",
+                [{"primary_keys": [1]}],
+            ),
+        )
+
+    def test_join(self):
+        A, B = self.classes("A", "B")
+
+        sess = Session()
+
+        self.assert_compile(
+            sess.query(A).join(A.b),
+            "SELECT a.id AS a_id, a.b_id AS a_b_id "
+            "FROM a JOIN (b JOIN d ON d.b_id = b.id "
+            "JOIN c ON c.id = d.c_id) ON a.b_id = b.id",
+        )
diff --git a/test/orm/test_deprecations.py b/test/orm/test_deprecations.py
index 04dff0252..7ce05345b 100644
--- a/test/orm/test_deprecations.py
+++ b/test/orm/test_deprecations.py
@@ -2162,3 +2162,369 @@ class InstrumentationTest(fixtures.ORMTest):
         l2 = Collection()
         f1.attr = l2
         eq_(canary, [adapter_1, f1.attr._sa_adapter, None])
+
+
+class NonPrimaryRelationshipLoaderTest(_fixtures.FixtureTest):
+    run_inserts = "once"
+    run_deletes = None
+
+    def test_selectload(self):
+        """tests lazy loading with two relationships simultaneously,
+        from the same table, using aliases.  """
+
+        users, orders, User, Address, Order, addresses = (
+            self.tables.users,
+            self.tables.orders,
+            self.classes.User,
+            self.classes.Address,
+            self.classes.Order,
+            self.tables.addresses,
+        )
+
+        openorders = sa.alias(orders, "openorders")
+        closedorders = sa.alias(orders, "closedorders")
+
+        mapper(Address, addresses)
+
+        mapper(Order, orders)
+
+        with testing.expect_deprecated(
+            "The mapper.non_primary parameter is deprecated"
+        ):
+            open_mapper = mapper(Order, openorders, non_primary=True)
+            closed_mapper = mapper(Order, closedorders, non_primary=True)
+        mapper(
+            User,
+            users,
+            properties=dict(
+                addresses=relationship(Address, lazy=True),
+                open_orders=relationship(
+                    open_mapper,
+                    primaryjoin=sa.and_(
+                        openorders.c.isopen == 1,
+                        users.c.id == openorders.c.user_id,
+                    ),
+                    lazy="select",
+                ),
+                closed_orders=relationship(
+                    closed_mapper,
+                    primaryjoin=sa.and_(
+                        closedorders.c.isopen == 0,
+                        users.c.id == closedorders.c.user_id,
+                    ),
+                    lazy="select",
+                ),
+            ),
+        )
+
+        self._run_double_test(10)
+
+    def test_joinedload(self):
+        """Eager loading with two relationships simultaneously,
+            from the same table, using aliases."""
+
+        users, orders, User, Address, Order, addresses = (
+            self.tables.users,
+            self.tables.orders,
+            self.classes.User,
+            self.classes.Address,
+            self.classes.Order,
+            self.tables.addresses,
+        )
+
+        openorders = sa.alias(orders, "openorders")
+        closedorders = sa.alias(orders, "closedorders")
+
+        mapper(Address, addresses)
+        mapper(Order, orders)
+
+        with testing.expect_deprecated(
+            "The mapper.non_primary parameter is deprecated"
+        ):
+            open_mapper = mapper(Order, openorders, non_primary=True)
+            closed_mapper = mapper(Order, closedorders, non_primary=True)
+
+        mapper(
+            User,
+            users,
+            properties=dict(
+                addresses=relationship(
+                    Address, lazy="joined", order_by=addresses.c.id
+                ),
+                open_orders=relationship(
+                    open_mapper,
+                    primaryjoin=sa.and_(
+                        openorders.c.isopen == 1,
+                        users.c.id == openorders.c.user_id,
+                    ),
+                    lazy="joined",
+                    order_by=openorders.c.id,
+                ),
+                closed_orders=relationship(
+                    closed_mapper,
+                    primaryjoin=sa.and_(
+                        closedorders.c.isopen == 0,
+                        users.c.id == closedorders.c.user_id,
+                    ),
+                    lazy="joined",
+                    order_by=closedorders.c.id,
+                ),
+            ),
+        )
+        self._run_double_test(1)
+
+    def test_selectin(self):
+
+        users, orders, User, Address, Order, addresses = (
+            self.tables.users,
+            self.tables.orders,
+            self.classes.User,
+            self.classes.Address,
+            self.classes.Order,
+            self.tables.addresses,
+        )
+
+        openorders = sa.alias(orders, "openorders")
+        closedorders = sa.alias(orders, "closedorders")
+
+        mapper(Address, addresses)
+        mapper(Order, orders)
+
+        with testing.expect_deprecated(
+            "The mapper.non_primary parameter is deprecated"
+        ):
+            open_mapper = mapper(Order, openorders, non_primary=True)
+            closed_mapper = mapper(Order, closedorders, non_primary=True)
+
+        mapper(
+            User,
+            users,
+            properties=dict(
+                addresses=relationship(
+                    Address, lazy="selectin", order_by=addresses.c.id
+                ),
+                open_orders=relationship(
+                    open_mapper,
+                    primaryjoin=sa.and_(
+                        openorders.c.isopen == 1,
+                        users.c.id == openorders.c.user_id,
+                    ),
+                    lazy="selectin",
+                    order_by=openorders.c.id,
+                ),
+                closed_orders=relationship(
+                    closed_mapper,
+                    primaryjoin=sa.and_(
+                        closedorders.c.isopen == 0,
+                        users.c.id == closedorders.c.user_id,
+                    ),
+                    lazy="selectin",
+                    order_by=closedorders.c.id,
+                ),
+            ),
+        )
+
+        self._run_double_test(4)
+
+    def test_subqueryload(self):
+
+        users, orders, User, Address, Order, addresses = (
+            self.tables.users,
+            self.tables.orders,
+            self.classes.User,
+            self.classes.Address,
+            self.classes.Order,
+            self.tables.addresses,
+        )
+
+        openorders = sa.alias(orders, "openorders")
+        closedorders = sa.alias(orders, "closedorders")
+
+        mapper(Address, addresses)
+        mapper(Order, orders)
+
+        with testing.expect_deprecated(
+            "The mapper.non_primary parameter is deprecated"
+        ):
+            open_mapper = mapper(Order, openorders, non_primary=True)
+            closed_mapper = mapper(Order, closedorders, non_primary=True)
+
+        mapper(
+            User,
+            users,
+            properties=dict(
+                addresses=relationship(
+                    Address, lazy="subquery", order_by=addresses.c.id
+                ),
+                open_orders=relationship(
+                    open_mapper,
+                    primaryjoin=sa.and_(
+                        openorders.c.isopen == 1,
+                        users.c.id == openorders.c.user_id,
+                    ),
+                    lazy="subquery",
+                    order_by=openorders.c.id,
+                ),
+                closed_orders=relationship(
+                    closed_mapper,
+                    primaryjoin=sa.and_(
+                        closedorders.c.isopen == 0,
+                        users.c.id == closedorders.c.user_id,
+                    ),
+                    lazy="subquery",
+                    order_by=closedorders.c.id,
+                ),
+            ),
+        )
+
+        self._run_double_test(4)
+
+    def _run_double_test(self, count):
+        User, Address, Order, Item = self.classes(
+            "User", "Address", "Order", "Item"
+        )
+        q = create_session().query(User).order_by(User.id)
+
+        def go():
+            eq_(
+                [
+                    User(
+                        id=7,
+                        addresses=[Address(id=1)],
+                        open_orders=[Order(id=3)],
+                        closed_orders=[Order(id=1), Order(id=5)],
+                    ),
+                    User(
+                        id=8,
+                        addresses=[
+                            Address(id=2),
+                            Address(id=3),
+                            Address(id=4),
+                        ],
+                        open_orders=[],
+                        closed_orders=[],
+                    ),
+                    User(
+                        id=9,
+                        addresses=[Address(id=5)],
+                        open_orders=[Order(id=4)],
+                        closed_orders=[Order(id=2)],
+                    ),
+                    User(id=10),
+                ],
+                q.all(),
+            )
+
+        self.assert_sql_count(testing.db, go, count)
+
+        sess = create_session()
+        user = sess.query(User).get(7)
+
+        closed_mapper = User.closed_orders.entity
+        open_mapper = User.open_orders.entity
+        eq_(
+            [Order(id=1), Order(id=5)],
+            create_session()
+            .query(closed_mapper)
+            .with_parent(user, property="closed_orders")
+            .all(),
+        )
+        eq_(
+            [Order(id=3)],
+            create_session()
+            .query(open_mapper)
+            .with_parent(user, property="open_orders")
+            .all(),
+        )
+
+
+class NonPrimaryMapperTest(_fixtures.FixtureTest, AssertsCompiledSQL):
+    __dialect__ = "default"
+
+    def test_non_primary_identity_class(self):
+        User = self.classes.User
+        users, addresses = self.tables.users, self.tables.addresses
+
+        class AddressUser(User):
+            pass
+
+        m1 = mapper(User, users, polymorphic_identity="user")
+        m2 = mapper(
+            AddressUser,
+            addresses,
+            inherits=User,
+            polymorphic_identity="address",
+            properties={"address_id": addresses.c.id},
+        )
+        with testing.expect_deprecated(
+            "The mapper.non_primary parameter is deprecated"
+        ):
+            m3 = mapper(AddressUser, addresses, non_primary=True)
+        assert m3._identity_class is m2._identity_class
+        eq_(
+            m2.identity_key_from_instance(AddressUser()),
+            m3.identity_key_from_instance(AddressUser()),
+        )
+
+    def test_illegal_non_primary(self):
+        users, Address, addresses, User = (
+            self.tables.users,
+            self.classes.Address,
+            self.tables.addresses,
+            self.classes.User,
+        )
+
+        mapper(User, users)
+        mapper(Address, addresses)
+        with testing.expect_deprecated(
+            "The mapper.non_primary parameter is deprecated"
+        ):
+            mapper(
+                User,
+                users,
+                non_primary=True,
+                properties={"addresses": relationship(Address)},
+            )
+        assert_raises_message(
+            sa.exc.ArgumentError,
+            "Attempting to assign a new relationship 'addresses' "
+            "to a non-primary mapper on class 'User'",
+            configure_mappers,
+        )
+
+    def test_illegal_non_primary_2(self):
+        User, users = self.classes.User, self.tables.users
+
+        with testing.expect_deprecated(
+            "The mapper.non_primary parameter is deprecated"
+        ):
+            assert_raises_message(
+                sa.exc.InvalidRequestError,
+                "Configure a primary mapper first",
+                mapper,
+                User,
+                users,
+                non_primary=True,
+            )
+
+    def test_illegal_non_primary_3(self):
+        users, addresses = self.tables.users, self.tables.addresses
+
+        class Base(object):
+            pass
+
+        class Sub(Base):
+            pass
+
+        mapper(Base, users)
+        with testing.expect_deprecated(
+            "The mapper.non_primary parameter is deprecated"
+        ):
+            assert_raises_message(
+                sa.exc.InvalidRequestError,
+                "Configure a primary mapper first",
+                mapper,
+                Sub,
+                addresses,
+                non_primary=True,
+            )
diff --git a/test/orm/test_eager_relations.py b/test/orm/test_eager_relations.py
index fd272b181..71010f02b 100644
--- a/test/orm/test_eager_relations.py
+++ b/test/orm/test_eager_relations.py
@@ -718,27 +718,49 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL):
             sess.query(User).order_by(User.id).all(),
         )
 
-    def test_double(self):
+    def test_double_w_ac(self):
         """Eager loading with two relationships simultaneously,
             from the same table, using aliases."""
 
-        users, orders, User, Address, Order, addresses = (
+        (
+            users,
+            orders,
+            User,
+            Address,
+            Order,
+            addresses,
+            Item,
+            items,
+            order_items,
+        ) = (
             self.tables.users,
             self.tables.orders,
             self.classes.User,
             self.classes.Address,
             self.classes.Order,
             self.tables.addresses,
+            self.classes.Item,
+            self.tables.items,
+            self.tables.order_items,
         )
 
-        openorders = sa.alias(orders, "openorders")
-        closedorders = sa.alias(orders, "closedorders")
-
         mapper(Address, addresses)
-        mapper(Order, orders)
+        mapper(
+            Order,
+            orders,
+            properties={
+                "items": relationship(
+                    Item,
+                    secondary=order_items,
+                    lazy="joined",
+                    order_by=items.c.id,
+                )
+            },
+        )
+        mapper(Item, items)
 
-        open_mapper = mapper(Order, openorders, non_primary=True)
-        closed_mapper = mapper(Order, closedorders, non_primary=True)
+        open_mapper = aliased(Order, orders)
+        closed_mapper = aliased(Order, orders)
 
         mapper(
             User,
@@ -750,57 +772,91 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL):
                 open_orders=relationship(
                     open_mapper,
                     primaryjoin=sa.and_(
-                        openorders.c.isopen == 1,
-                        users.c.id == openorders.c.user_id,
+                        open_mapper.isopen == 1,
+                        users.c.id == open_mapper.user_id,
                     ),
                     lazy="joined",
-                    order_by=openorders.c.id,
+                    order_by=open_mapper.id,
                 ),
                 closed_orders=relationship(
                     closed_mapper,
                     primaryjoin=sa.and_(
-                        closedorders.c.isopen == 0,
-                        users.c.id == closedorders.c.user_id,
+                        closed_mapper.isopen == 0,
+                        users.c.id == closed_mapper.user_id,
                     ),
                     lazy="joined",
-                    order_by=closedorders.c.id,
+                    order_by=closed_mapper.id,
                 ),
             ),
         )
 
-        q = create_session().query(User).order_by(User.id)
+        self._run_double_test()
 
-        def go():
-            eq_(
-                [
-                    User(
-                        id=7,
-                        addresses=[Address(id=1)],
-                        open_orders=[Order(id=3)],
-                        closed_orders=[Order(id=1), Order(id=5)],
-                    ),
-                    User(
-                        id=8,
-                        addresses=[
-                            Address(id=2),
-                            Address(id=3),
-                            Address(id=4),
-                        ],
-                        open_orders=[],
-                        closed_orders=[],
-                    ),
-                    User(
-                        id=9,
-                        addresses=[Address(id=5)],
-                        open_orders=[Order(id=4)],
-                        closed_orders=[Order(id=2)],
-                    ),
-                    User(id=10),
-                ],
-                q.all(),
-            )
+    def test_double_w_ac_against_subquery(self):
+        """Eager loading with two relationships simultaneously,
+            from the same table, using aliases."""
 
-        self.assert_sql_count(testing.db, go, 1)
+        (
+            users,
+            orders,
+            User,
+            Address,
+            Order,
+            addresses,
+            Item,
+            items,
+            order_items,
+        ) = (
+            self.tables.users,
+            self.tables.orders,
+            self.classes.User,
+            self.classes.Address,
+            self.classes.Order,
+            self.tables.addresses,
+            self.classes.Item,
+            self.tables.items,
+            self.tables.order_items,
+        )
+
+        mapper(Address, addresses)
+        mapper(
+            Order,
+            orders,
+            properties={
+                "items": relationship(
+                    Item,
+                    secondary=order_items,
+                    lazy="joined",
+                    order_by=items.c.id,
+                )
+            },
+        )
+        mapper(Item, items)
+
+        open_mapper = aliased(
+            Order, select([orders]).where(orders.c.isopen == 1).alias()
+        )
+        closed_mapper = aliased(
+            Order, select([orders]).where(orders.c.isopen == 0).alias()
+        )
+
+        mapper(
+            User,
+            users,
+            properties=dict(
+                addresses=relationship(
+                    Address, lazy="joined", order_by=addresses.c.id
+                ),
+                open_orders=relationship(
+                    open_mapper, lazy="joined", order_by=open_mapper.id
+                ),
+                closed_orders=relationship(
+                    closed_mapper, lazy="joined", order_by=closed_mapper.id
+                ),
+            ),
+        )
+
+        self._run_double_test()
 
     def test_double_same_mappers(self):
         """Eager loading with two relationships simultaneously,
@@ -867,26 +923,30 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL):
                 ),
             ),
         )
+        self._run_double_test()
+
+    def _run_double_test(self, no_items=False):
+        User, Address, Order, Item = self.classes(
+            "User", "Address", "Order", "Item"
+        )
         q = create_session().query(User).order_by(User.id)
 
+        def items(*ids):
+            if no_items:
+                return {}
+            else:
+                return {"items": [Item(id=id_) for id_ in ids]}
+
         def go():
             eq_(
                 [
                     User(
                         id=7,
                         addresses=[Address(id=1)],
-                        open_orders=[
-                            Order(
-                                id=3,
-                                items=[Item(id=3), Item(id=4), Item(id=5)],
-                            )
-                        ],
+                        open_orders=[Order(id=3, **items(3, 4, 5))],
                         closed_orders=[
-                            Order(
-                                id=1,
-                                items=[Item(id=1), Item(id=2), Item(id=3)],
-                            ),
-                            Order(id=5, items=[Item(id=5)]),
+                            Order(id=1, **items(1, 2, 3)),
+                            Order(id=5, **items(5)),
                         ],
                     ),
                     User(
@@ -902,15 +962,8 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL):
                     User(
                         id=9,
                         addresses=[Address(id=5)],
-                        open_orders=[
-                            Order(id=4, items=[Item(id=1), Item(id=5)])
-                        ],
-                        closed_orders=[
-                            Order(
-                                id=2,
-                                items=[Item(id=1), Item(id=2), Item(id=3)],
-                            )
-                        ],
+                        open_orders=[Order(id=4, **items(1, 5))],
+                        closed_orders=[Order(id=2, **items(1, 2, 3))],
                     ),
                     User(id=10),
                 ],
@@ -1739,9 +1792,7 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL):
                     Order, backref="user", lazy="joined", order_by=orders.c.id
                 ),
                 "max_order": relationship(
-                    mapper(Order, max_orders, non_primary=True),
-                    lazy="joined",
-                    uselist=False,
+                    aliased(Order, max_orders), lazy="joined", uselist=False
                 ),
             },
         )
@@ -3021,7 +3072,7 @@ class InnerJoinSplicingTest(fixtures.MappedTest, testing.AssertsCompiledSQL):
         )
         self._assert_result(q)
 
-    def test_splice_onto_np_mapper(self):
+    def test_splice_onto_ac(self):
         A = self.classes.A
         B = self.classes.B
         C1 = self.classes.C1
@@ -3032,20 +3083,7 @@ class InnerJoinSplicingTest(fixtures.MappedTest, testing.AssertsCompiledSQL):
 
         weird_selectable = b_table.outerjoin(c1_table)
 
-        b_np = mapper(
-            B,
-            weird_selectable,
-            non_primary=True,
-            properties=odict(
-                [
-                    # note we need to make this fixed with lazy=False until
-                    # [ticket:3348] is resolved
-                    ("c1s", relationship(C1, lazy=False, innerjoin=True)),
-                    ("c_id", c1_table.c.id),
-                    ("b_value", b_table.c.value),
-                ]
-            ),
-        )
+        b_np = aliased(B, weird_selectable, flat=True)
 
         a_mapper = inspect(A)
         a_mapper.add_property("bs_np", relationship(b_np))
@@ -3055,14 +3093,10 @@ class InnerJoinSplicingTest(fixtures.MappedTest, testing.AssertsCompiledSQL):
         q = s.query(A).options(joinedload("bs_np", innerjoin=False))
         self.assert_compile(
             q,
-            "SELECT a.id AS a_id, c1_1.id AS c1_1_id, c1_1.b_id AS c1_1_b_id, "
-            "c1_1.value AS c1_1_value, c1_2.id AS c1_2_id, "
-            "b_1.value AS b_1_value, b_1.id AS b_1_id, "
-            "b_1.a_id AS b_1_a_id, c1_2.b_id AS c1_2_b_id, "
-            "c1_2.value AS c1_2_value "
-            "FROM a LEFT OUTER JOIN "
-            "(b AS b_1 LEFT OUTER JOIN c1 AS c1_2 ON b_1.id = c1_2.b_id "
-            "JOIN c1 AS c1_1 ON b_1.id = c1_1.b_id) ON a.id = b_1.a_id",
+            "SELECT a.id AS a_id, b_1.id AS b_1_id, b_1.a_id AS b_1_a_id, "
+            "b_1.value AS b_1_value FROM a LEFT OUTER JOIN "
+            "(b AS b_1 LEFT OUTER JOIN c1 AS c1_1 ON b_1.id = c1_1.b_id) "
+            "ON a.id = b_1.a_id",
         )
 
 
diff --git a/test/orm/test_froms.py b/test/orm/test_froms.py
index c5e1d1485..1cccfff26 100644
--- a/test/orm/test_froms.py
+++ b/test/orm/test_froms.py
@@ -2399,7 +2399,7 @@ class MixedEntitiesTest(QueryTest, AssertsCompiledSQL):
         ]:
             q = s.query(crit)
             mzero = q._entity_zero()
-            is_(mzero.mapped_table, q._query_entity_zero().selectable)
+            is_(mzero.persist_selectable, q._query_entity_zero().selectable)
             q = q.join(j)
             self.assert_compile(q, exp)
 
diff --git a/test/orm/test_inspect.py b/test/orm/test_inspect.py
index ca127580d..368199b96 100644
--- a/test/orm/test_inspect.py
+++ b/test/orm/test_inspect.py
@@ -48,11 +48,11 @@ class TestORMInspection(_fixtures.FixtureTest):
         insp = inspect(User)
         is_(insp.local_table, user_table)
 
-    def test_mapped_table(self):
+    def test_persist_selectable(self):
         User = self.classes.User
         user_table = self.tables.users
         insp = inspect(User)
-        is_(insp.mapped_table, user_table)
+        is_(insp.persist_selectable, user_table)
 
     def test_mapper_selectable(self):
         User = self.classes.User
diff --git a/test/orm/test_joins.py b/test/orm/test_joins.py
index 046d5c3e2..bb9747cf2 100644
--- a/test/orm/test_joins.py
+++ b/test/orm/test_joins.py
@@ -2912,7 +2912,7 @@ class JoinToNonPolyAliasesTest(fixtures.MappedTest, AssertsCompiledSQL):
         mapper(Child, child)
 
         derived = select([child]).alias()
-        npc = mapper(Child, derived, non_primary=True)
+        npc = aliased(Child, derived)
         cls.npc = npc
         cls.derived = derived
         mp.add_property("npc", relationship(npc))
diff --git a/test/orm/test_lazy_relations.py b/test/orm/test_lazy_relations.py
index 4d9e460f9..78680701e 100644
--- a/test/orm/test_lazy_relations.py
+++ b/test/orm/test_lazy_relations.py
@@ -10,9 +10,11 @@ from sqlalchemy import ForeignKey
 from sqlalchemy import ForeignKeyConstraint
 from sqlalchemy import Integer
 from sqlalchemy import orm
+from sqlalchemy import select
 from sqlalchemy import SmallInteger
 from sqlalchemy import String
 from sqlalchemy import testing
+from sqlalchemy.orm import aliased
 from sqlalchemy.orm import attributes
 from sqlalchemy.orm import configure_mappers
 from sqlalchemy.orm import create_session
@@ -530,28 +532,107 @@ class LazyTest(_fixtures.FixtureTest):
             list(q),
         )
 
-    def test_double(self):
-        """tests lazy loading with two relationships simultaneously,
-        from the same table, using aliases.  """
+    def test_double_w_ac_against_subquery(self):
 
-        users, orders, User, Address, Order, addresses = (
+        (
+            users,
+            orders,
+            User,
+            Address,
+            Order,
+            addresses,
+            Item,
+            items,
+            order_items,
+        ) = (
             self.tables.users,
             self.tables.orders,
             self.classes.User,
             self.classes.Address,
             self.classes.Order,
             self.tables.addresses,
+            self.classes.Item,
+            self.tables.items,
+            self.tables.order_items,
+        )
+
+        mapper(Address, addresses)
+
+        mapper(
+            Order,
+            orders,
+            properties={
+                "items": relationship(
+                    Item,
+                    secondary=order_items,
+                    lazy="select",
+                    order_by=items.c.id,
+                )
+            },
+        )
+        mapper(Item, items)
+
+        open_mapper = aliased(
+            Order, select([orders]).where(orders.c.isopen == 1).alias()
+        )
+        closed_mapper = aliased(
+            Order, select([orders]).where(orders.c.isopen == 0).alias()
+        )
+
+        mapper(
+            User,
+            users,
+            properties=dict(
+                addresses=relationship(Address, lazy=True),
+                open_orders=relationship(open_mapper, lazy="select"),
+                closed_orders=relationship(closed_mapper, lazy="select"),
+            ),
         )
 
-        openorders = sa.alias(orders, "openorders")
-        closedorders = sa.alias(orders, "closedorders")
+        self._run_double_test()
+
+    def test_double_w_ac(self):
+
+        (
+            users,
+            orders,
+            User,
+            Address,
+            Order,
+            addresses,
+            Item,
+            items,
+            order_items,
+        ) = (
+            self.tables.users,
+            self.tables.orders,
+            self.classes.User,
+            self.classes.Address,
+            self.classes.Order,
+            self.tables.addresses,
+            self.classes.Item,
+            self.tables.items,
+            self.tables.order_items,
+        )
 
         mapper(Address, addresses)
 
-        mapper(Order, orders)
+        mapper(
+            Order,
+            orders,
+            properties={
+                "items": relationship(
+                    Item,
+                    secondary=order_items,
+                    lazy="select",
+                    order_by=items.c.id,
+                )
+            },
+        )
+        mapper(Item, items)
 
-        open_mapper = mapper(Order, openorders, non_primary=True)
-        closed_mapper = mapper(Order, closedorders, non_primary=True)
+        open_mapper = aliased(Order, orders)
+        closed_mapper = aliased(Order, orders)
         mapper(
             User,
             users,
@@ -560,47 +641,79 @@ class LazyTest(_fixtures.FixtureTest):
                 open_orders=relationship(
                     open_mapper,
                     primaryjoin=sa.and_(
-                        openorders.c.isopen == 1,
-                        users.c.id == openorders.c.user_id,
+                        open_mapper.isopen == 1,
+                        users.c.id == open_mapper.user_id,
                     ),
                     lazy="select",
                 ),
                 closed_orders=relationship(
                     closed_mapper,
                     primaryjoin=sa.and_(
-                        closedorders.c.isopen == 0,
-                        users.c.id == closedorders.c.user_id,
+                        closed_mapper.isopen == 0,
+                        users.c.id == closed_mapper.user_id,
                     ),
                     lazy="select",
                 ),
             ),
         )
-        q = create_session().query(User)
 
-        assert [
-            User(
-                id=7,
-                addresses=[Address(id=1)],
-                open_orders=[Order(id=3)],
-                closed_orders=[Order(id=1), Order(id=5)],
-            ),
-            User(
-                id=8,
-                addresses=[Address(id=2), Address(id=3), Address(id=4)],
-                open_orders=[],
-                closed_orders=[],
-            ),
-            User(
-                id=9,
-                addresses=[Address(id=5)],
-                open_orders=[Order(id=4)],
-                closed_orders=[Order(id=2)],
-            ),
-            User(id=10),
-        ] == q.all()
+        self._run_double_test()
+
+    def _run_double_test(self, no_items=False):
+        User, Address, Order, Item = self.classes(
+            "User", "Address", "Order", "Item"
+        )
+        q = create_session().query(User).order_by(User.id)
+
+        def items(*ids):
+            if no_items:
+                return {}
+            else:
+                return {"items": [Item(id=id_) for id_ in ids]}
+
+        def go():
+            eq_(
+                [
+                    User(
+                        id=7,
+                        addresses=[Address(id=1)],
+                        open_orders=[Order(id=3, **items(3, 4, 5))],
+                        closed_orders=[
+                            Order(id=1, **items(1, 2, 3)),
+                            Order(id=5, **items(5)),
+                        ],
+                    ),
+                    User(
+                        id=8,
+                        addresses=[
+                            Address(id=2),
+                            Address(id=3),
+                            Address(id=4),
+                        ],
+                        open_orders=[],
+                        closed_orders=[],
+                    ),
+                    User(
+                        id=9,
+                        addresses=[Address(id=5)],
+                        open_orders=[Order(id=4, **items(1, 5))],
+                        closed_orders=[Order(id=2, **items(1, 2, 3))],
+                    ),
+                    User(id=10),
+                ],
+                q.all(),
+            )
+
+        if no_items:
+            self.assert_sql_count(testing.db, go, 10)
+        else:
+            self.assert_sql_count(testing.db, go, 15)
 
         sess = create_session()
         user = sess.query(User).get(7)
+
+        closed_mapper = User.closed_orders.entity
+        open_mapper = User.open_orders.entity
         eq_(
             [Order(id=1), Order(id=5)],
             create_session()
diff --git a/test/orm/test_mapper.py b/test/orm/test_mapper.py
index 4e42e4f9a..06929c6bc 100644
--- a/test/orm/test_mapper.py
+++ b/test/orm/test_mapper.py
@@ -853,28 +853,6 @@ class MapperTest(_fixtures.FixtureTest, AssertsCompiledSQL):
         assert n1.children[0] is n1._children[0] is n2
         eq_(str(Node.parent == n2), ":param_1 = nodes.parent_id")
 
-    def test_non_primary_identity_class(self):
-        User = self.classes.User
-        users, addresses = self.tables.users, self.tables.addresses
-
-        class AddressUser(User):
-            pass
-
-        m1 = mapper(User, users, polymorphic_identity="user")
-        m2 = mapper(
-            AddressUser,
-            addresses,
-            inherits=User,
-            polymorphic_identity="address",
-            properties={"address_id": addresses.c.id},
-        )
-        m3 = mapper(AddressUser, addresses, non_primary=True)
-        assert m3._identity_class is m2._identity_class
-        eq_(
-            m2.identity_key_from_instance(AddressUser()),
-            m3.identity_key_from_instance(AddressUser()),
-        )
-
     def test_reassign_polymorphic_identity_warns(self):
         User = self.classes.User
         users = self.tables.users
@@ -898,60 +876,6 @@ class MapperTest(_fixtures.FixtureTest, AssertsCompiledSQL):
             polymorphic_identity="user",
         )
 
-    def test_illegal_non_primary(self):
-        users, Address, addresses, User = (
-            self.tables.users,
-            self.classes.Address,
-            self.tables.addresses,
-            self.classes.User,
-        )
-
-        mapper(User, users)
-        mapper(Address, addresses)
-        mapper(
-            User,
-            users,
-            non_primary=True,
-            properties={"addresses": relationship(Address)},
-        )
-        assert_raises_message(
-            sa.exc.ArgumentError,
-            "Attempting to assign a new relationship 'addresses' "
-            "to a non-primary mapper on class 'User'",
-            configure_mappers,
-        )
-
-    def test_illegal_non_primary_2(self):
-        User, users = self.classes.User, self.tables.users
-
-        assert_raises_message(
-            sa.exc.InvalidRequestError,
-            "Configure a primary mapper first",
-            mapper,
-            User,
-            users,
-            non_primary=True,
-        )
-
-    def test_illegal_non_primary_3(self):
-        users, addresses = self.tables.users, self.tables.addresses
-
-        class Base(object):
-            pass
-
-        class Sub(Base):
-            pass
-
-        mapper(Base, users)
-        assert_raises_message(
-            sa.exc.InvalidRequestError,
-            "Configure a primary mapper first",
-            mapper,
-            Sub,
-            addresses,
-            non_primary=True,
-        )
-
     def test_prop_filters(self):
         t = Table(
             "person",
diff --git a/test/orm/test_rel_fn.py b/test/orm/test_rel_fn.py
index 8fd521ae8..5e6ac53fe 100644
--- a/test/orm/test_rel_fn.py
+++ b/test/orm/test_rel_fn.py
@@ -1120,46 +1120,53 @@ class AdaptedJoinTest(_JoinFixtures, fixtures.TestBase, AssertsCompiledSQL):
 
     def test_join_targets_o2m_selfref(self):
         joincond = self._join_fixture_o2m_selfref()
-        left = select([joincond.parent_selectable]).alias("pj")
+        left = select([joincond.parent_persist_selectable]).alias("pj")
         pj, sj, sec, adapter, ds = joincond.join_targets(
-            left, joincond.child_selectable, True
+            left, joincond.child_persist_selectable, True
         )
         self.assert_compile(pj, "pj.id = selfref.sid")
+        self.assert_compile(pj, "pj.id = selfref.sid")
 
-        right = select([joincond.child_selectable]).alias("pj")
+        right = select([joincond.child_persist_selectable]).alias("pj")
         pj, sj, sec, adapter, ds = joincond.join_targets(
-            joincond.parent_selectable, right, True
+            joincond.parent_persist_selectable, right, True
         )
         self.assert_compile(pj, "selfref.id = pj.sid")
+        self.assert_compile(pj, "selfref.id = pj.sid")
 
     def test_join_targets_o2m_plain(self):
         joincond = self._join_fixture_o2m()
         pj, sj, sec, adapter, ds = joincond.join_targets(
-            joincond.parent_selectable, joincond.child_selectable, False
+            joincond.parent_persist_selectable,
+            joincond.child_persist_selectable,
+            False,
         )
         self.assert_compile(pj, "lft.id = rgt.lid")
+        self.assert_compile(pj, "lft.id = rgt.lid")
 
     def test_join_targets_o2m_left_aliased(self):
         joincond = self._join_fixture_o2m()
-        left = select([joincond.parent_selectable]).alias("pj")
+        left = select([joincond.parent_persist_selectable]).alias("pj")
         pj, sj, sec, adapter, ds = joincond.join_targets(
-            left, joincond.child_selectable, True
+            left, joincond.child_persist_selectable, True
         )
         self.assert_compile(pj, "pj.id = rgt.lid")
+        self.assert_compile(pj, "pj.id = rgt.lid")
 
     def test_join_targets_o2m_right_aliased(self):
         joincond = self._join_fixture_o2m()
-        right = select([joincond.child_selectable]).alias("pj")
+        right = select([joincond.child_persist_selectable]).alias("pj")
         pj, sj, sec, adapter, ds = joincond.join_targets(
-            joincond.parent_selectable, right, True
+            joincond.parent_persist_selectable, right, True
         )
         self.assert_compile(pj, "lft.id = pj.lid")
+        self.assert_compile(pj, "lft.id = pj.lid")
 
     def test_join_targets_o2m_composite_selfref(self):
         joincond = self._join_fixture_o2m_composite_selfref()
-        right = select([joincond.child_selectable]).alias("pj")
+        right = select([joincond.child_persist_selectable]).alias("pj")
         pj, sj, sec, adapter, ds = joincond.join_targets(
-            joincond.parent_selectable, right, True
+            joincond.parent_persist_selectable, right, True
         )
         self.assert_compile(
             pj,
@@ -1169,9 +1176,9 @@ class AdaptedJoinTest(_JoinFixtures, fixtures.TestBase, AssertsCompiledSQL):
 
     def test_join_targets_m2o_composite_selfref(self):
         joincond = self._join_fixture_m2o_composite_selfref()
-        right = select([joincond.child_selectable]).alias("pj")
+        right = select([joincond.child_persist_selectable]).alias("pj")
         pj, sj, sec, adapter, ds = joincond.join_targets(
-            joincond.parent_selectable, right, True
+            joincond.parent_persist_selectable, right, True
         )
         self.assert_compile(
             pj,
diff --git a/test/orm/test_selectable.py b/test/orm/test_selectable.py
index 1fc2f3f16..35d32a444 100644
--- a/test/orm/test_selectable.py
+++ b/test/orm/test_selectable.py
@@ -74,7 +74,7 @@ class SelectableNoFromsTest(fixtures.MappedTest, AssertsCompiledSQL):
         Subset, common = self.classes.Subset, self.tables.common
 
         subset_select = select([common.c.id, common.c.data]).alias()
-        subset_mapper = mapper(Subset, subset_select)
+        mapper(Subset, subset_select)
         sess = Session(bind=testing.db)
         sess.add(Subset(data=1))
         sess.flush()
@@ -84,7 +84,7 @@ class SelectableNoFromsTest(fixtures.MappedTest, AssertsCompiledSQL):
         eq_(sess.query(Subset).filter(Subset.data == 1).one(), Subset(data=1))
         eq_(sess.query(Subset).filter(Subset.data != 1).first(), None)
 
-        subset_select = sa.orm.class_mapper(Subset).mapped_table
+        subset_select = sa.orm.class_mapper(Subset).persist_selectable
         eq_(
             sess.query(Subset).filter(subset_select.c.data == 1).one(),
             Subset(data=1),
diff --git a/test/orm/test_selectin_relations.py b/test/orm/test_selectin_relations.py
index b891835d1..e95fbb050 100644
--- a/test/orm/test_selectin_relations.py
+++ b/test/orm/test_selectin_relations.py
@@ -2,6 +2,7 @@ import sqlalchemy as sa
 from sqlalchemy import bindparam
 from sqlalchemy import ForeignKey
 from sqlalchemy import Integer
+from sqlalchemy import select
 from sqlalchemy import String
 from sqlalchemy import testing
 from sqlalchemy.orm import aliased
@@ -776,27 +777,111 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL):
             sess.query(User).order_by(User.id).all(),
         )
 
-    def test_double(self):
-        """Eager loading with two relationships simultaneously,
-            from the same table, using aliases."""
+    def test_double_w_ac_against_subquery(self):
 
-        users, orders, User, Address, Order, addresses = (
+        (
+            users,
+            orders,
+            User,
+            Address,
+            Order,
+            addresses,
+            Item,
+            items,
+            order_items,
+        ) = (
             self.tables.users,
             self.tables.orders,
             self.classes.User,
             self.classes.Address,
             self.classes.Order,
             self.tables.addresses,
+            self.classes.Item,
+            self.tables.items,
+            self.tables.order_items,
+        )
+
+        mapper(Address, addresses)
+        mapper(
+            Order,
+            orders,
+            properties={
+                "items": relationship(
+                    Item,
+                    secondary=order_items,
+                    lazy="selectin",
+                    order_by=items.c.id,
+                )
+            },
+        )
+        mapper(Item, items)
+
+        open_mapper = aliased(
+            Order, select([orders]).where(orders.c.isopen == 1).alias()
         )
+        closed_mapper = aliased(
+            Order, select([orders]).where(orders.c.isopen == 0).alias()
+        )
+
+        mapper(
+            User,
+            users,
+            properties=dict(
+                addresses=relationship(
+                    Address, lazy="selectin", order_by=addresses.c.id
+                ),
+                open_orders=relationship(
+                    open_mapper, lazy="selectin", order_by=open_mapper.id
+                ),
+                closed_orders=relationship(
+                    closed_mapper, lazy="selectin", order_by=closed_mapper.id
+                ),
+            ),
+        )
+
+        self._run_double_test()
 
-        openorders = sa.alias(orders, "openorders")
-        closedorders = sa.alias(orders, "closedorders")
+    def test_double_w_ac(self):
+
+        (
+            users,
+            orders,
+            User,
+            Address,
+            Order,
+            addresses,
+            Item,
+            items,
+            order_items,
+        ) = (
+            self.tables.users,
+            self.tables.orders,
+            self.classes.User,
+            self.classes.Address,
+            self.classes.Order,
+            self.tables.addresses,
+            self.classes.Item,
+            self.tables.items,
+            self.tables.order_items,
+        )
 
         mapper(Address, addresses)
-        mapper(Order, orders)
+        mapper(
+            Order,
+            orders,
+            properties={
+                "items": relationship(
+                    Item,
+                    secondary=order_items,
+                    lazy="selectin",
+                    order_by=items.c.id,
+                )
+            },
+        )
+        mapper(Item, items)
 
-        open_mapper = mapper(Order, openorders, non_primary=True)
-        closed_mapper = mapper(Order, closedorders, non_primary=True)
+        open_mapper = aliased(Order, orders)
+        closed_mapper = aliased(Order, orders)
 
         mapper(
             User,
@@ -808,57 +893,25 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL):
                 open_orders=relationship(
                     open_mapper,
                     primaryjoin=sa.and_(
-                        openorders.c.isopen == 1,
-                        users.c.id == openorders.c.user_id,
+                        open_mapper.isopen == 1,
+                        users.c.id == open_mapper.user_id,
                     ),
                     lazy="selectin",
-                    order_by=openorders.c.id,
+                    order_by=open_mapper.id,
                 ),
                 closed_orders=relationship(
                     closed_mapper,
                     primaryjoin=sa.and_(
-                        closedorders.c.isopen == 0,
-                        users.c.id == closedorders.c.user_id,
+                        closed_mapper.isopen == 0,
+                        users.c.id == closed_mapper.user_id,
                     ),
                     lazy="selectin",
-                    order_by=closedorders.c.id,
+                    order_by=closed_mapper.id,
                 ),
             ),
         )
 
-        q = create_session().query(User).order_by(User.id)
-
-        def go():
-            eq_(
-                [
-                    User(
-                        id=7,
-                        addresses=[Address(id=1)],
-                        open_orders=[Order(id=3)],
-                        closed_orders=[Order(id=1), Order(id=5)],
-                    ),
-                    User(
-                        id=8,
-                        addresses=[
-                            Address(id=2),
-                            Address(id=3),
-                            Address(id=4),
-                        ],
-                        open_orders=[],
-                        closed_orders=[],
-                    ),
-                    User(
-                        id=9,
-                        addresses=[Address(id=5)],
-                        open_orders=[Order(id=4)],
-                        closed_orders=[Order(id=2)],
-                    ),
-                    User(id=10),
-                ],
-                q.all(),
-            )
-
-        self.assert_sql_count(testing.db, go, 4)
+        self._run_double_test()
 
     def test_double_same_mappers(self):
         """Eager loading with two relationships simultaneously,
@@ -925,26 +978,31 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL):
                 ),
             ),
         )
+
+        self._run_double_test()
+
+    def _run_double_test(self, no_items=False):
+        User, Address, Order, Item = self.classes(
+            "User", "Address", "Order", "Item"
+        )
         q = create_session().query(User).order_by(User.id)
 
+        def items(*ids):
+            if no_items:
+                return {}
+            else:
+                return {"items": [Item(id=id_) for id_ in ids]}
+
         def go():
             eq_(
                 [
                     User(
                         id=7,
                         addresses=[Address(id=1)],
-                        open_orders=[
-                            Order(
-                                id=3,
-                                items=[Item(id=3), Item(id=4), Item(id=5)],
-                            )
-                        ],
+                        open_orders=[Order(id=3, **items(3, 4, 5))],
                         closed_orders=[
-                            Order(
-                                id=1,
-                                items=[Item(id=1), Item(id=2), Item(id=3)],
-                            ),
-                            Order(id=5, items=[Item(id=5)]),
+                            Order(id=1, **items(1, 2, 3)),
+                            Order(id=5, **items(5)),
                         ],
                     ),
                     User(
@@ -960,22 +1018,18 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL):
                     User(
                         id=9,
                         addresses=[Address(id=5)],
-                        open_orders=[
-                            Order(id=4, items=[Item(id=1), Item(id=5)])
-                        ],
-                        closed_orders=[
-                            Order(
-                                id=2,
-                                items=[Item(id=1), Item(id=2), Item(id=3)],
-                            )
-                        ],
+                        open_orders=[Order(id=4, **items(1, 5))],
+                        closed_orders=[Order(id=2, **items(1, 2, 3))],
                     ),
                     User(id=10),
                 ],
                 q.all(),
             )
 
-        self.assert_sql_count(testing.db, go, 6)
+        if no_items:
+            self.assert_sql_count(testing.db, go, 4)
+        else:
+            self.assert_sql_count(testing.db, go, 6)
 
     def test_limit(self):
         """Limit operations combined with lazy-load relationships."""
@@ -1119,9 +1173,7 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL):
                     order_by=orders.c.id,
                 ),
                 "max_order": relationship(
-                    mapper(Order, max_orders, non_primary=True),
-                    lazy="selectin",
-                    uselist=False,
+                    aliased(Order, max_orders), lazy="selectin", uselist=False
                 ),
             },
         )
diff --git a/test/orm/test_subquery_relations.py b/test/orm/test_subquery_relations.py
index b4be6debe..117ab5be4 100644
--- a/test/orm/test_subquery_relations.py
+++ b/test/orm/test_subquery_relations.py
@@ -3,6 +3,7 @@ from sqlalchemy import bindparam
 from sqlalchemy import ForeignKey
 from sqlalchemy import inspect
 from sqlalchemy import Integer
+from sqlalchemy import select
 from sqlalchemy import String
 from sqlalchemy import testing
 from sqlalchemy.orm import aliased
@@ -796,27 +797,111 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL):
             "1 FROM users",
         )
 
-    def test_double(self):
-        """Eager loading with two relationships simultaneously,
-            from the same table, using aliases."""
+    def test_double_w_ac_against_subquery(self):
 
-        users, orders, User, Address, Order, addresses = (
+        (
+            users,
+            orders,
+            User,
+            Address,
+            Order,
+            addresses,
+            Item,
+            items,
+            order_items,
+        ) = (
             self.tables.users,
             self.tables.orders,
             self.classes.User,
             self.classes.Address,
             self.classes.Order,
             self.tables.addresses,
+            self.classes.Item,
+            self.tables.items,
+            self.tables.order_items,
+        )
+
+        mapper(Address, addresses)
+        mapper(
+            Order,
+            orders,
+            properties={
+                "items": relationship(
+                    Item,
+                    secondary=order_items,
+                    lazy="subquery",
+                    order_by=items.c.id,
+                )
+            },
+        )
+        mapper(Item, items)
+
+        open_mapper = aliased(
+            Order, select([orders]).where(orders.c.isopen == 1).alias()
         )
+        closed_mapper = aliased(
+            Order, select([orders]).where(orders.c.isopen == 0).alias()
+        )
+
+        mapper(
+            User,
+            users,
+            properties=dict(
+                addresses=relationship(
+                    Address, lazy="subquery", order_by=addresses.c.id
+                ),
+                open_orders=relationship(
+                    open_mapper, lazy="subquery", order_by=open_mapper.id
+                ),
+                closed_orders=relationship(
+                    closed_mapper, lazy="subquery", order_by=closed_mapper.id
+                ),
+            ),
+        )
+
+        self._run_double_test()
 
-        openorders = sa.alias(orders, "openorders")
-        closedorders = sa.alias(orders, "closedorders")
+    def test_double_w_ac(self):
+
+        (
+            users,
+            orders,
+            User,
+            Address,
+            Order,
+            addresses,
+            Item,
+            items,
+            order_items,
+        ) = (
+            self.tables.users,
+            self.tables.orders,
+            self.classes.User,
+            self.classes.Address,
+            self.classes.Order,
+            self.tables.addresses,
+            self.classes.Item,
+            self.tables.items,
+            self.tables.order_items,
+        )
 
         mapper(Address, addresses)
-        mapper(Order, orders)
+        mapper(
+            Order,
+            orders,
+            properties={
+                "items": relationship(
+                    Item,
+                    secondary=order_items,
+                    lazy="subquery",
+                    order_by=items.c.id,
+                )
+            },
+        )
+        mapper(Item, items)
 
-        open_mapper = mapper(Order, openorders, non_primary=True)
-        closed_mapper = mapper(Order, closedorders, non_primary=True)
+        open_mapper = aliased(Order, orders)
+        closed_mapper = aliased(Order, orders)
 
         mapper(
             User,
@@ -828,57 +913,25 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL):
                 open_orders=relationship(
                     open_mapper,
                     primaryjoin=sa.and_(
-                        openorders.c.isopen == 1,
-                        users.c.id == openorders.c.user_id,
+                        open_mapper.isopen == 1,
+                        users.c.id == open_mapper.user_id,
                     ),
                     lazy="subquery",
-                    order_by=openorders.c.id,
+                    order_by=open_mapper.id,
                 ),
                 closed_orders=relationship(
                     closed_mapper,
                     primaryjoin=sa.and_(
-                        closedorders.c.isopen == 0,
-                        users.c.id == closedorders.c.user_id,
+                        closed_mapper.isopen == 0,
+                        users.c.id == closed_mapper.user_id,
                     ),
                     lazy="subquery",
-                    order_by=closedorders.c.id,
+                    order_by=closed_mapper.id,
                 ),
             ),
         )
 
-        q = create_session().query(User).order_by(User.id)
-
-        def go():
-            eq_(
-                [
-                    User(
-                        id=7,
-                        addresses=[Address(id=1)],
-                        open_orders=[Order(id=3)],
-                        closed_orders=[Order(id=1), Order(id=5)],
-                    ),
-                    User(
-                        id=8,
-                        addresses=[
-                            Address(id=2),
-                            Address(id=3),
-                            Address(id=4),
-                        ],
-                        open_orders=[],
-                        closed_orders=[],
-                    ),
-                    User(
-                        id=9,
-                        addresses=[Address(id=5)],
-                        open_orders=[Order(id=4)],
-                        closed_orders=[Order(id=2)],
-                    ),
-                    User(id=10),
-                ],
-                q.all(),
-            )
-
-        self.assert_sql_count(testing.db, go, 4)
+        self._run_double_test()
 
     def test_double_same_mappers(self):
         """Eager loading with two relationships simultaneously,
@@ -945,26 +998,30 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL):
                 ),
             ),
         )
+        self._run_double_test()
+
+    def _run_double_test(self, no_items=False):
+        User, Address, Order, Item = self.classes(
+            "User", "Address", "Order", "Item"
+        )
         q = create_session().query(User).order_by(User.id)
 
+        def items(*ids):
+            if no_items:
+                return {}
+            else:
+                return {"items": [Item(id=id_) for id_ in ids]}
+
         def go():
             eq_(
                 [
                     User(
                         id=7,
                         addresses=[Address(id=1)],
-                        open_orders=[
-                            Order(
-                                id=3,
-                                items=[Item(id=3), Item(id=4), Item(id=5)],
-                            )
-                        ],
+                        open_orders=[Order(id=3, **items(3, 4, 5))],
                         closed_orders=[
-                            Order(
-                                id=1,
-                                items=[Item(id=1), Item(id=2), Item(id=3)],
-                            ),
-                            Order(id=5, items=[Item(id=5)]),
+                            Order(id=1, **items(1, 2, 3)),
+                            Order(id=5, **items(5)),
                         ],
                     ),
                     User(
@@ -980,22 +1037,18 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL):
                     User(
                         id=9,
                         addresses=[Address(id=5)],
-                        open_orders=[
-                            Order(id=4, items=[Item(id=1), Item(id=5)])
-                        ],
-                        closed_orders=[
-                            Order(
-                                id=2,
-                                items=[Item(id=1), Item(id=2), Item(id=3)],
-                            )
-                        ],
+                        open_orders=[Order(id=4, **items(1, 5))],
+                        closed_orders=[Order(id=2, **items(1, 2, 3))],
                     ),
                     User(id=10),
                 ],
                 q.all(),
             )
 
-        self.assert_sql_count(testing.db, go, 6)
+        if no_items:
+            self.assert_sql_count(testing.db, go, 4)
+        else:
+            self.assert_sql_count(testing.db, go, 6)
 
     def test_limit(self):
         """Limit operations combined with lazy-load relationships."""
@@ -1139,9 +1192,7 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL):
                     order_by=orders.c.id,
                 ),
                 "max_order": relationship(
-                    mapper(Order, max_orders, non_primary=True),
-                    lazy="subquery",
-                    uselist=False,
+                    aliased(Order, max_orders), lazy="subquery", uselist=False
                 ),
             },
         )
diff --git a/test/orm/test_unitofwork.py b/test/orm/test_unitofwork.py
index f7a735e21..6326f5f1a 100644
--- a/test/orm/test_unitofwork.py
+++ b/test/orm/test_unitofwork.py
@@ -2660,17 +2660,13 @@ class ManyToManyTest(_fixtures.FixtureTest):
 
         mapper(Keyword, keywords)
 
-        # note that we are breaking a rule here, making a second
-        # mapper(Keyword, keywords) the reorganization of mapper construction
-        # affected this, but was fixed again
-
         mapper(
             IKAssociation,
             item_keywords,
             primary_key=[item_keywords.c.item_id, item_keywords.c.keyword_id],
             properties=dict(
                 keyword=relationship(
-                    mapper(Keyword, keywords, non_primary=True),
+                    Keyword,
                     lazy="joined",
                     uselist=False,
                     # note here is a valid place where
diff --git a/test/requirements.py b/test/requirements.py
index c265bb3c9..b9d61ed56 100644
--- a/test/requirements.py
+++ b/test/requirements.py
@@ -596,7 +596,7 @@ class DefaultRequirements(SuiteRequirements):
     @property
     def window_functions(self):
         return only_if(
-            ["postgresql>=8.4", "mssql", "oracle"],
+            ["postgresql>=8.4", "mssql", "oracle", "sqlite>=3.25.0"],
             "Backend does not support window functions",
         )