Merge "Complete deprecation of from_self()"

18 files changed, 1962 insertions, 541 deletions

diff --git a/lib/sqlalchemy/ext/baked.py b/lib/sqlalchemy/ext/baked.py
index 1fad89286..288677387 100644
--- a/lib/sqlalchemy/ext/baked.py
+++ b/lib/sqlalchemy/ext/baked.py
@@ -434,7 +434,7 @@ class Result(object):
         """
 
         col = func.count(literal_column("*"))
-        bq = self.bq.with_criteria(lambda q: q.from_self(col))
+        bq = self.bq.with_criteria(lambda q: q._from_self(col))
         return bq.for_session(self.session).params(self._params).scalar()
 
     def scalar(self):
diff --git a/lib/sqlalchemy/orm/path_registry.py b/lib/sqlalchemy/orm/path_registry.py
index ac7a64c30..f6c03d007 100644
--- a/lib/sqlalchemy/orm/path_registry.py
+++ b/lib/sqlalchemy/orm/path_registry.py
@@ -356,7 +356,7 @@ class PropRegistry(PathRegistry):
             parent.path + self.prop._wildcard_token,
         )
         self._default_path_loader_key = self.prop._default_path_loader_key
-        self._loader_key = ("loader", self.path)
+        self._loader_key = ("loader", self.natural_path)
 
     def __str__(self):
         return " -> ".join(str(elem) for elem in self.path)
@@ -418,7 +418,15 @@ class AbstractEntityRegistry(PathRegistry):
                 self.natural_path = parent.natural_path + (
                     parent.natural_path[-1].entity,
                 )
+        # it seems to make sense that since these paths get mixed up
+        # with statements that are cached or not, we should make
+        # sure the natural path is cachable across different occurrences
+        # of equivalent AliasedClass objects.  however, so far this
+        # does not seem to be needed for whatever reason.
+        # elif not parent.path and self.is_aliased_class:
+        #     self.natural_path = (self.entity._generate_cache_key()[0], )
         else:
+            # self.natural_path = parent.natural_path + (entity, )
             self.natural_path = self.path
 
     @property
diff --git a/lib/sqlalchemy/orm/query.py b/lib/sqlalchemy/orm/query.py
index 9587fcd6c..c6e6f6466 100644
--- a/lib/sqlalchemy/orm/query.py
+++ b/lib/sqlalchemy/orm/query.py
@@ -170,6 +170,10 @@ class Query(
         self.session = session
         self._set_entities(entities)
 
+    def _set_propagate_attrs(self, values):
+        self._propagate_attrs = util.immutabledict(values)
+        return self
+
     def _set_entities(self, entities):
         self._raw_columns = [
             coercions.expect(
@@ -2526,7 +2530,7 @@ class Query(
 
         """
 
-        self._limit_clause, self._offset_clause = orm_util._make_slice(
+        self._limit_clause, self._offset_clause = sql_util._make_slice(
             self._limit_clause, self._offset_clause, start, stop
         )
 
@@ -2537,7 +2541,7 @@ class Query(
         ``Query``.
 
         """
-        self._limit_clause = orm_util._offset_or_limit_clause(limit)
+        self._limit_clause = sql_util._offset_or_limit_clause(limit)
 
     @_generative
     @_assertions(_no_statement_condition)
@@ -2546,7 +2550,7 @@ class Query(
         ``Query``.
 
         """
-        self._offset_clause = orm_util._offset_or_limit_clause(offset)
+        self._offset_clause = sql_util._offset_or_limit_clause(offset)
 
     @_generative
     @_assertions(_no_statement_condition)
diff --git a/lib/sqlalchemy/orm/strategies.py b/lib/sqlalchemy/orm/strategies.py
index b9826ac87..fbf153dc5 100644
--- a/lib/sqlalchemy/orm/strategies.py
+++ b/lib/sqlalchemy/orm/strategies.py
@@ -1137,7 +1137,8 @@ class SubqueryLoader(PostLoader):
             (("lazy", "select"),)
         ).init_class_attribute(mapper)
 
-    def _get_leftmost(self, subq_path):
+    def _get_leftmost(self, subq_path, current_compile_state, is_root):
+        given_subq_path = subq_path
         subq_path = subq_path.path
         subq_mapper = orm_util._class_to_mapper(subq_path[0])
 
@@ -1150,16 +1151,29 @@ class SubqueryLoader(PostLoader):
         else:
             leftmost_mapper, leftmost_prop = subq_mapper, subq_path[1]
 
+        if is_root:
+            # the subq_path is also coming from cached state, so when we start
+            # building up this path, it has to also be converted to be in terms
+            # of the current state. this is for the specific case of the entity
+            # is an AliasedClass against a subquery that's not otherwise going
+            # to adapt
+            new_subq_path = current_compile_state._entities[
+                0
+            ].entity_zero._path_registry[leftmost_prop]
+        else:
+            new_subq_path = given_subq_path
+
         leftmost_cols = leftmost_prop.local_columns
 
         leftmost_attr = [
             getattr(
-                subq_path[0].entity, leftmost_mapper._columntoproperty[c].key
+                new_subq_path.path[0].entity,
+                leftmost_mapper._columntoproperty[c].key,
             )
             for c in leftmost_cols
         ]
 
-        return leftmost_mapper, leftmost_attr, leftmost_prop
+        return leftmost_mapper, leftmost_attr, leftmost_prop, new_subq_path
 
     def _generate_from_original_query(
         self,
@@ -1361,10 +1375,12 @@ class SubqueryLoader(PostLoader):
 
         return q
 
-    def _setup_options(self, q, subq_path, orig_query, effective_entity):
+    def _setup_options(
+        self, q, subq_path, rewritten_path, orig_query, effective_entity
+    ):
         # propagate loader options etc. to the new query.
         # these will fire relative to subq_path.
-        q = q._with_current_path(subq_path)
+        q = q._with_current_path(rewritten_path)
         q = q.options(*orig_query._with_options)
 
         return q
@@ -1462,11 +1478,13 @@ class SubqueryLoader(PostLoader):
         else:
             effective_entity = self.entity
 
-        subq_path = context.query._execution_options.get(
-            ("subquery_path", None), orm_util.PathRegistry.root
+        subq_path, rewritten_path = context.query._execution_options.get(
+            ("subquery_paths", None),
+            (orm_util.PathRegistry.root, orm_util.PathRegistry.root),
         )
-
+        is_root = subq_path is orm_util.PathRegistry.root
         subq_path = subq_path + path
+        rewritten_path = rewritten_path + path
 
         # if not via query option, check for
         # a cycle
@@ -1484,12 +1502,6 @@ class SubqueryLoader(PostLoader):
             elif subq_path.contains_mapper(self.mapper):
                 return
 
-        (
-            leftmost_mapper,
-            leftmost_attr,
-            leftmost_relationship,
-        ) = self._get_leftmost(subq_path)
-
         # use the current query being invoked, not the compile state
         # one.  this is so that we get the current parameters.  however,
         # it means we can't use the existing compile state, we have to make
@@ -1525,6 +1537,13 @@ class SubqueryLoader(PostLoader):
             orig_query
         )
 
+        (
+            leftmost_mapper,
+            leftmost_attr,
+            leftmost_relationship,
+            rewritten_path,
+        ) = self._get_leftmost(rewritten_path, orig_compile_state, is_root)
+
         # generate a new Query from the original, then
         # produce a subquery from it.
         left_alias = self._generate_from_original_query(
@@ -1547,7 +1566,7 @@ class SubqueryLoader(PostLoader):
         q._execution_options = q._execution_options.union(
             {
                 ("orig_query", SubqueryLoader): orig_query,
-                ("subquery_path", None): subq_path,
+                ("subquery_paths", None): (subq_path, rewritten_path),
             }
         )
 
@@ -1561,7 +1580,9 @@ class SubqueryLoader(PostLoader):
             q, to_join, left_alias, parent_alias, effective_entity
         )
 
-        q = self._setup_options(q, subq_path, orig_query, effective_entity)
+        q = self._setup_options(
+            q, subq_path, rewritten_path, orig_query, effective_entity
+        )
         q = self._setup_outermost_orderby(q)
 
         return q
diff --git a/lib/sqlalchemy/orm/util.py b/lib/sqlalchemy/orm/util.py
index 2bfba16b5..27b14d95b 100644
--- a/lib/sqlalchemy/orm/util.py
+++ b/lib/sqlalchemy/orm/util.py
@@ -1811,76 +1811,6 @@ def randomize_unitofwork():
     ) = session.set = mapper.set = dependency.set = RandomSet
 
 
-def _offset_or_limit_clause(element, name=None, type_=None):
-    """Convert the given value to an "offset or limit" clause.
-
-    This handles incoming integers and converts to an expression; if
-    an expression is already given, it is passed through.
-
-    """
-    return coercions.expect(
-        roles.LimitOffsetRole, element, name=name, type_=type_
-    )
-
-
-def _offset_or_limit_clause_asint_if_possible(clause):
-    """Return the offset or limit clause as a simple integer if possible,
-    else return the clause.
-
-    """
-    if clause is None:
-        return None
-    if hasattr(clause, "_limit_offset_value"):
-        value = clause._limit_offset_value
-        return util.asint(value)
-    else:
-        return clause
-
-
-def _make_slice(limit_clause, offset_clause, start, stop):
-    """Compute LIMIT/OFFSET in terms of slice start/end
-    """
-
-    # for calculated limit/offset, try to do the addition of
-    # values to offset in Python, however if a SQL clause is present
-    # then the addition has to be on the SQL side.
-    if start is not None and stop is not None:
-        offset_clause = _offset_or_limit_clause_asint_if_possible(
-            offset_clause
-        )
-        if offset_clause is None:
-            offset_clause = 0
-
-        if start != 0:
-            offset_clause = offset_clause + start
-
-        if offset_clause == 0:
-            offset_clause = None
-        else:
-            offset_clause = _offset_or_limit_clause(offset_clause)
-
-        limit_clause = _offset_or_limit_clause(stop - start)
-
-    elif start is None and stop is not None:
-        limit_clause = _offset_or_limit_clause(stop)
-    elif start is not None and stop is None:
-        offset_clause = _offset_or_limit_clause_asint_if_possible(
-            offset_clause
-        )
-        if offset_clause is None:
-            offset_clause = 0
-
-        if start != 0:
-            offset_clause = offset_clause + start
-
-        if offset_clause == 0:
-            offset_clause = None
-        else:
-            offset_clause = _offset_or_limit_clause(offset_clause)
-
-    return limit_clause, offset_clause
-
-
 def _getitem(iterable_query, item):
     """calculate __getitem__ in terms of an iterable query object
     that also has a slice() method.
diff --git a/lib/sqlalchemy/sql/coercions.py b/lib/sqlalchemy/sql/coercions.py
index b3a38f802..b8525925b 100644
--- a/lib/sqlalchemy/sql/coercions.py
+++ b/lib/sqlalchemy/sql/coercions.py
@@ -226,14 +226,19 @@ class RoleImpl(object):
         code=None,
         err=None,
     ):
+        if resolved is not None and resolved is not element:
+            got = "%r object resolved from %r object" % (resolved, element)
+        else:
+            got = repr(element)
+
         if argname:
-            msg = "%s expected for argument %r; got %r." % (
+            msg = "%s expected for argument %r; got %s." % (
                 self.name,
                 argname,
-                element,
+                got,
             )
         else:
-            msg = "%s expected, got %r." % (self.name, element)
+            msg = "%s expected, got %s." % (self.name, got)
 
         if advice:
             msg += " " + advice
@@ -369,7 +374,7 @@ class _SelectIsNotFrom(object):
             advice = (
                 "To create a "
                 "FROM clause from a %s object, use the .subquery() method."
-                % (element.__class__,)
+                % (resolved.__class__ if resolved is not None else element,)
             )
             code = "89ve"
         else:
diff --git a/lib/sqlalchemy/sql/selectable.py b/lib/sqlalchemy/sql/selectable.py
index 6e0ac1fac..c78b1ec57 100644
--- a/lib/sqlalchemy/sql/selectable.py
+++ b/lib/sqlalchemy/sql/selectable.py
@@ -3028,6 +3028,45 @@ class GenerativeSelect(DeprecatedSelectBaseGenerations, SelectBase):
         self._offset_clause = self._offset_or_limit_clause(offset)
 
     @_generative
+    @util.preload_module("sqlalchemy.sql.util")
+    def slice(self, start, stop):
+        """Apply LIMIT / OFFSET to this statement based on a slice.
+
+        The start and stop indices behave like the argument to Python's
+        built-in :func:`range` function. This method provides an
+        alternative to using ``LIMIT``/``OFFSET`` to get a slice of the
+        query.
+
+        For example, ::
+
+            stmt = select(User).order_by(User).id.slice(1, 3)
+
+        renders as
+
+        .. sourcecode:: sql
+
+           SELECT users.id AS users_id,
+                  users.name AS users_name
+           FROM users ORDER BY users.id
+           LIMIT ? OFFSET ?
+           (2, 1)
+
+        .. versionadded:: 1.4  Added the :meth:`_sql.GenerativeSelect.slice`
+           method generalized from the ORM.
+
+        .. seealso::
+
+           :meth:`_sql.GenerativeSelect.limit`
+
+           :meth:`_sql.GenerativeSelect.offset`
+
+        """
+        sql_util = util.preloaded.sql_util
+        self._limit_clause, self._offset_clause = sql_util._make_slice(
+            self._limit_clause, self._offset_clause, start, stop
+        )
+
+    @_generative
     def order_by(self, *clauses):
         r"""Return a new selectable with the given list of ORDER BY
         criterion applied.
diff --git a/lib/sqlalchemy/sql/util.py b/lib/sqlalchemy/sql/util.py
index b3ead718a..264976cc8 100644
--- a/lib/sqlalchemy/sql/util.py
+++ b/lib/sqlalchemy/sql/util.py
@@ -12,7 +12,9 @@
 from collections import deque
 from itertools import chain
 
+from . import coercions
 from . import operators
+from . import roles
 from . import visitors
 from .annotation import _deep_annotate  # noqa
 from .annotation import _deep_deannotate  # noqa
@@ -980,3 +982,73 @@ class ColumnAdapter(ClauseAdapter):
     def __setstate__(self, state):
         self.__dict__.update(state)
         self.columns = util.WeakPopulateDict(self._locate_col)
+
+
+def _offset_or_limit_clause(element, name=None, type_=None):
+    """Convert the given value to an "offset or limit" clause.
+
+    This handles incoming integers and converts to an expression; if
+    an expression is already given, it is passed through.
+
+    """
+    return coercions.expect(
+        roles.LimitOffsetRole, element, name=name, type_=type_
+    )
+
+
+def _offset_or_limit_clause_asint_if_possible(clause):
+    """Return the offset or limit clause as a simple integer if possible,
+    else return the clause.
+
+    """
+    if clause is None:
+        return None
+    if hasattr(clause, "_limit_offset_value"):
+        value = clause._limit_offset_value
+        return util.asint(value)
+    else:
+        return clause
+
+
+def _make_slice(limit_clause, offset_clause, start, stop):
+    """Compute LIMIT/OFFSET in terms of slice start/end
+    """
+
+    # for calculated limit/offset, try to do the addition of
+    # values to offset in Python, however if a SQL clause is present
+    # then the addition has to be on the SQL side.
+    if start is not None and stop is not None:
+        offset_clause = _offset_or_limit_clause_asint_if_possible(
+            offset_clause
+        )
+        if offset_clause is None:
+            offset_clause = 0
+
+        if start != 0:
+            offset_clause = offset_clause + start
+
+        if offset_clause == 0:
+            offset_clause = None
+        else:
+            offset_clause = _offset_or_limit_clause(offset_clause)
+
+        limit_clause = _offset_or_limit_clause(stop - start)
+
+    elif start is None and stop is not None:
+        limit_clause = _offset_or_limit_clause(stop)
+    elif start is not None and stop is None:
+        offset_clause = _offset_or_limit_clause_asint_if_possible(
+            offset_clause
+        )
+        if offset_clause is None:
+            offset_clause = 0
+
+        if start != 0:
+            offset_clause = offset_clause + start
+
+        if offset_clause == 0:
+            offset_clause = None
+        else:
+            offset_clause = _offset_or_limit_clause(offset_clause)
+
+    return limit_clause, offset_clause
diff --git a/lib/sqlalchemy/testing/warnings.py b/lib/sqlalchemy/testing/warnings.py
index de5db6467..d97447ec8 100644
--- a/lib/sqlalchemy/testing/warnings.py
+++ b/lib/sqlalchemy/testing/warnings.py
@@ -100,7 +100,7 @@ def setup_filters():
         # ORM Query
         #
         r"The Query\.get\(\) function",
-        r"The Query\.from_self\(\) function",
+        # r"The Query\.from_self\(\) function",
         #
         # ORM Session
         #
diff --git a/test/dialect/postgresql/test_compiler.py b/test/dialect/postgresql/test_compiler.py
index 2dd64d9bc..64e945124 100644
--- a/test/dialect/postgresql/test_compiler.py
+++ b/test/dialect/postgresql/test_compiler.py
@@ -2243,12 +2243,15 @@ class DistinctOnTest(fixtures.TestBase, AssertsCompiledSQL):
 
         mapper(Foo, self.table)
         sess = Session()
+        subq = sess.query(Foo).subquery()
+
+        f1 = aliased(Foo, subq)
         self.assert_compile(
-            sess.query(Foo).from_self().distinct(Foo.a, Foo.b),
-            "SELECT DISTINCT ON (anon_1.t_a, anon_1.t_b) anon_1.t_id "
-            "AS anon_1_t_id, anon_1.t_a AS anon_1_t_a, anon_1.t_b "
-            "AS anon_1_t_b FROM (SELECT t.id AS t_id, t.a AS t_a, "
-            "t.b AS t_b FROM t) AS anon_1",
+            sess.query(f1).distinct(f1.a, f1.b),
+            "SELECT DISTINCT ON (anon_1.a, anon_1.b) anon_1.id "
+            "AS anon_1_id, anon_1.a AS anon_1_a, anon_1.b "
+            "AS anon_1_b FROM (SELECT t.id AS id, t.a AS a, "
+            "t.b AS b FROM t) AS anon_1",
         )
 
     def test_query_distinct_on_aliased(self):
diff --git a/test/ext/test_baked.py b/test/ext/test_baked.py
index 4f8159979..6aa584017 100644
--- a/test/ext/test_baked.py
+++ b/test/ext/test_baked.py
@@ -636,7 +636,7 @@ class ResultTest(BakedTest):
 
         bq = self.bakery(lambda s: s.query(User.id, User.name))
 
-        bq += lambda q: q.from_self().with_entities(func.count(User.id))
+        bq += lambda q: q._from_self().with_entities(func.count(User.id))
 
         for i in range(3):
             session = Session(autocommit=True)
@@ -671,7 +671,7 @@ class ResultTest(BakedTest):
                     bq += lambda q: q.filter(User.name == "jack")
 
                 if cond4:
-                    bq += lambda q: q.from_self().with_entities(
+                    bq += lambda q: q._from_self().with_entities(
                         func.count(User.id)
                     )
                 sess = Session(autocommit=True)
diff --git a/test/orm/test_deprecations.py b/test/orm/test_deprecations.py
index 48a8b99a6..532d0aa10 100644
--- a/test/orm/test_deprecations.py
+++ b/test/orm/test_deprecations.py
@@ -1,21 +1,25 @@
 import sqlalchemy as sa
 from sqlalchemy import and_
 from sqlalchemy import cast
+from sqlalchemy import column
 from sqlalchemy import desc
 from sqlalchemy import event
 from sqlalchemy import exc as sa_exc
+from sqlalchemy import ForeignKey
 from sqlalchemy import func
 from sqlalchemy import Integer
 from sqlalchemy import literal_column
 from sqlalchemy import or_
 from sqlalchemy import select
 from sqlalchemy import String
+from sqlalchemy import table
 from sqlalchemy import testing
 from sqlalchemy import text
 from sqlalchemy import true
 from sqlalchemy.orm import aliased
 from sqlalchemy.orm import as_declarative
 from sqlalchemy.orm import attributes
+from sqlalchemy.orm import backref
 from sqlalchemy.orm import collections
 from sqlalchemy.orm import column_property
 from sqlalchemy.orm import configure_mappers
@@ -34,6 +38,7 @@ from sqlalchemy.orm import mapper
 from sqlalchemy.orm import relation
 from sqlalchemy.orm import relationship
 from sqlalchemy.orm import Session
+from sqlalchemy.orm import subqueryload
 from sqlalchemy.orm import synonym
 from sqlalchemy.orm import undefer
 from sqlalchemy.orm import with_polymorphic
@@ -49,6 +54,8 @@ from sqlalchemy.testing import fixtures
 from sqlalchemy.testing import is_
 from sqlalchemy.testing import is_true
 from sqlalchemy.testing import mock
+from sqlalchemy.testing.assertsql import CompiledSQL
+from sqlalchemy.testing.fixtures import ComparableEntity
 from sqlalchemy.testing.mock import call
 from sqlalchemy.testing.mock import Mock
 from sqlalchemy.testing.schema import Column
@@ -518,6 +525,1216 @@ class DeprecatedQueryTest(_fixtures.FixtureTest, AssertsCompiledSQL):
             )
 
 
+class SelfRefFromSelfTest(fixtures.MappedTest, AssertsCompiledSQL):
+    run_setup_mappers = "once"
+    run_inserts = "once"
+    run_deletes = None
+    __dialect__ = "default"
+
+    @classmethod
+    def define_tables(cls, metadata):
+        Table(
+            "nodes",
+            metadata,
+            Column(
+                "id", Integer, primary_key=True, test_needs_autoincrement=True
+            ),
+            Column("parent_id", Integer, ForeignKey("nodes.id")),
+            Column("data", String(30)),
+        )
+
+    @classmethod
+    def setup_classes(cls):
+        class Node(cls.Comparable):
+            def append(self, node):
+                self.children.append(node)
+
+    @classmethod
+    def setup_mappers(cls):
+        Node, nodes = cls.classes.Node, cls.tables.nodes
+
+        mapper(
+            Node,
+            nodes,
+            properties={
+                "children": relationship(
+                    Node,
+                    lazy="select",
+                    join_depth=3,
+                    backref=backref("parent", remote_side=[nodes.c.id]),
+                )
+            },
+        )
+
+    @classmethod
+    def insert_data(cls, connection):
+        Node = cls.classes.Node
+
+        sess = create_session(connection)
+        n1 = Node(data="n1")
+        n1.append(Node(data="n11"))
+        n1.append(Node(data="n12"))
+        n1.append(Node(data="n13"))
+        n1.children[1].append(Node(data="n121"))
+        n1.children[1].append(Node(data="n122"))
+        n1.children[1].append(Node(data="n123"))
+        sess.add(n1)
+        sess.flush()
+        sess.close()
+
+    def test_from_self_inside_excludes_outside(self):
+        """test the propagation of aliased() from inside to outside
+        on a from_self()..
+        """
+
+        Node = self.classes.Node
+
+        sess = create_session()
+
+        n1 = aliased(Node)
+
+        # n1 is not inside the from_self(), so all cols must be maintained
+        # on the outside
+        with self._from_self_deprecated():
+            self.assert_compile(
+                sess.query(Node)
+                .filter(Node.data == "n122")
+                .from_self(n1, Node.id),
+                "SELECT nodes_1.id AS nodes_1_id, "
+                "nodes_1.parent_id AS nodes_1_parent_id, "
+                "nodes_1.data AS nodes_1_data, anon_1.nodes_id "
+                "AS anon_1_nodes_id "
+                "FROM nodes AS nodes_1, (SELECT nodes.id AS nodes_id, "
+                "nodes.parent_id AS nodes_parent_id, "
+                "nodes.data AS nodes_data FROM "
+                "nodes WHERE nodes.data = :data_1) AS anon_1",
+                use_default_dialect=True,
+            )
+
+        parent = aliased(Node)
+        grandparent = aliased(Node)
+        with self._from_self_deprecated():
+            q = (
+                sess.query(Node, parent, grandparent)
+                .join(parent, Node.parent)
+                .join(grandparent, parent.parent)
+                .filter(Node.data == "n122")
+                .filter(parent.data == "n12")
+                .filter(grandparent.data == "n1")
+                .from_self()
+                .limit(1)
+            )
+
+        # parent, grandparent *are* inside the from_self(), so they
+        # should get aliased to the outside.
+        self.assert_compile(
+            q,
+            "SELECT anon_1.nodes_id AS anon_1_nodes_id, "
+            "anon_1.nodes_parent_id AS anon_1_nodes_parent_id, "
+            "anon_1.nodes_data AS anon_1_nodes_data, "
+            "anon_1.nodes_1_id AS anon_1_nodes_1_id, "
+            "anon_1.nodes_1_parent_id AS anon_1_nodes_1_parent_id, "
+            "anon_1.nodes_1_data AS anon_1_nodes_1_data, "
+            "anon_1.nodes_2_id AS anon_1_nodes_2_id, "
+            "anon_1.nodes_2_parent_id AS anon_1_nodes_2_parent_id, "
+            "anon_1.nodes_2_data AS anon_1_nodes_2_data "
+            "FROM (SELECT nodes.id AS nodes_id, nodes.parent_id "
+            "AS nodes_parent_id, nodes.data AS nodes_data, "
+            "nodes_1.id AS nodes_1_id, "
+            "nodes_1.parent_id AS nodes_1_parent_id, "
+            "nodes_1.data AS nodes_1_data, nodes_2.id AS nodes_2_id, "
+            "nodes_2.parent_id AS nodes_2_parent_id, nodes_2.data AS "
+            "nodes_2_data FROM nodes JOIN nodes AS nodes_1 ON "
+            "nodes_1.id = nodes.parent_id JOIN nodes AS nodes_2 "
+            "ON nodes_2.id = nodes_1.parent_id "
+            "WHERE nodes.data = :data_1 AND nodes_1.data = :data_2 AND "
+            "nodes_2.data = :data_3) AS anon_1 LIMIT :param_1",
+            {"param_1": 1},
+            use_default_dialect=True,
+        )
+
+    def test_multiple_explicit_entities_two(self):
+        Node = self.classes.Node
+
+        sess = create_session()
+
+        parent = aliased(Node)
+        grandparent = aliased(Node)
+        with self._from_self_deprecated():
+            eq_(
+                sess.query(Node, parent, grandparent)
+                .join(parent, Node.parent)
+                .join(grandparent, parent.parent)
+                .filter(Node.data == "n122")
+                .filter(parent.data == "n12")
+                .filter(grandparent.data == "n1")
+                .from_self()
+                .first(),
+                (Node(data="n122"), Node(data="n12"), Node(data="n1")),
+            )
+
+    def test_multiple_explicit_entities_three(self):
+        Node = self.classes.Node
+
+        sess = create_session()
+
+        parent = aliased(Node)
+        grandparent = aliased(Node)
+        # same, change order around
+        with self._from_self_deprecated():
+            eq_(
+                sess.query(parent, grandparent, Node)
+                .join(parent, Node.parent)
+                .join(grandparent, parent.parent)
+                .filter(Node.data == "n122")
+                .filter(parent.data == "n12")
+                .filter(grandparent.data == "n1")
+                .from_self()
+                .first(),
+                (Node(data="n12"), Node(data="n1"), Node(data="n122")),
+            )
+
+    def test_multiple_explicit_entities_five(self):
+        Node = self.classes.Node
+
+        sess = create_session()
+
+        parent = aliased(Node)
+        grandparent = aliased(Node)
+        with self._from_self_deprecated():
+            eq_(
+                sess.query(Node, parent, grandparent)
+                .join(parent, Node.parent)
+                .join(grandparent, parent.parent)
+                .filter(Node.data == "n122")
+                .filter(parent.data == "n12")
+                .filter(grandparent.data == "n1")
+                .from_self()
+                .options(joinedload(Node.children))
+                .first(),
+                (Node(data="n122"), Node(data="n12"), Node(data="n1")),
+            )
+
+    def _from_self_deprecated(self):
+        return testing.expect_deprecated_20(
+            r"The Query.from_self\(\) function/method"
+        )
+
+
+class FromSelfTest(QueryTest, AssertsCompiledSQL):
+    __dialect__ = "default"
+
+    def _from_self_deprecated(self):
+        return testing.expect_deprecated_20(
+            r"The Query.from_self\(\) function/method"
+        )
+
+    def test_illegal_operations(self):
+
+        User = self.classes.User
+
+        s = Session()
+
+        with self._from_self_deprecated():
+            q = s.query(User).from_self()
+        assert_raises_message(
+            sa.exc.InvalidRequestError,
+            r"Can't call Query.update\(\) or Query.delete\(\)",
+            q.update,
+            {},
+        )
+
+        assert_raises_message(
+            sa.exc.InvalidRequestError,
+            r"Can't call Query.update\(\) or Query.delete\(\)",
+            q.delete,
+            {},
+        )
+
+    def test_columns_augmented_distinct_on(self):
+        User, Address = self.classes.User, self.classes.Address
+
+        sess = create_session()
+
+        with self._from_self_deprecated():
+            q = (
+                sess.query(
+                    User.id,
+                    User.name.label("foo"),
+                    Address.id,
+                    Address.email_address,
+                )
+                .distinct(Address.email_address)
+                .order_by(User.id, User.name, Address.email_address)
+                .from_self(User.id, User.name.label("foo"), Address.id)
+            )
+
+        # Address.email_address is added because of DISTINCT,
+        # however User.id, User.name are not b.c. they're already there,
+        # even though User.name is labeled
+        self.assert_compile(
+            q,
+            "SELECT anon_1.users_id AS anon_1_users_id, anon_1.foo AS foo, "
+            "anon_1.addresses_id AS anon_1_addresses_id "
+            "FROM ("
+            "SELECT DISTINCT ON (addresses.email_address) "
+            "users.id AS users_id, users.name AS foo, "
+            "addresses.id AS addresses_id, addresses.email_address AS "
+            "addresses_email_address FROM users, addresses ORDER BY "
+            "users.id, users.name, addresses.email_address"
+            ") AS anon_1",
+            dialect="postgresql",
+        )
+
+    def test_columns_augmented_roundtrip_one_from_self(self):
+        """Test workaround for legacy style DISTINCT on extra column.
+
+        See #5134
+
+        """
+        User, Address = self.classes.User, self.classes.Address
+
+        sess = create_session()
+        with self._from_self_deprecated():
+            q = (
+                sess.query(User, Address.email_address)
+                .join("addresses")
+                .distinct()
+                .from_self(User)
+                .order_by(desc(Address.email_address))
+            )
+
+        eq_([User(id=7), User(id=9), User(id=8)], q.all())
+
+    def test_columns_augmented_roundtrip_three_from_self(self):
+        """Test workaround for legacy style DISTINCT on extra column.
+
+        See #5134
+
+        """
+
+        User, Address = self.classes.User, self.classes.Address
+
+        sess = create_session()
+
+        with self._from_self_deprecated():
+            q = (
+                sess.query(
+                    User.id,
+                    User.name.label("foo"),
+                    Address.id,
+                    Address.email_address,
+                )
+                .join(Address, true())
+                .filter(User.name == "jack")
+                .filter(User.id + Address.user_id > 0)
+                .distinct()
+                .from_self(User.id, User.name.label("foo"), Address.id)
+                .order_by(User.id, User.name, Address.email_address)
+            )
+
+        eq_(
+            q.all(),
+            [
+                (7, "jack", 3),
+                (7, "jack", 4),
+                (7, "jack", 2),
+                (7, "jack", 5),
+                (7, "jack", 1),
+            ],
+        )
+        for row in q:
+            eq_(row._mapping.keys(), ["id", "foo", "id"])
+
+    def test_clause_onclause(self):
+        Order, User = (
+            self.classes.Order,
+            self.classes.User,
+        )
+
+        sess = create_session()
+        # explicit onclause with from_self(), means
+        # the onclause must be aliased against the query's custom
+        # FROM object
+        with self._from_self_deprecated():
+            eq_(
+                sess.query(User)
+                .order_by(User.id)
+                .offset(2)
+                .from_self()
+                .join(Order, User.id == Order.user_id)
+                .all(),
+                [User(name="fred")],
+            )
+
+    def test_from_self_resets_joinpaths(self):
+        """test a join from from_self() doesn't confuse joins inside the subquery
+        with the outside.
+        """
+
+        Item, Keyword = self.classes.Item, self.classes.Keyword
+
+        sess = create_session()
+
+        with self._from_self_deprecated():
+            self.assert_compile(
+                sess.query(Item)
+                .join(Item.keywords)
+                .from_self(Keyword)
+                .join(Item.keywords),
+                "SELECT keywords.id AS keywords_id, "
+                "keywords.name AS keywords_name "
+                "FROM (SELECT items.id AS items_id, "
+                "items.description AS items_description "
+                "FROM items JOIN item_keywords AS item_keywords_1 "
+                "ON items.id = "
+                "item_keywords_1.item_id JOIN keywords "
+                "ON keywords.id = item_keywords_1.keyword_id) "
+                "AS anon_1 JOIN item_keywords AS item_keywords_2 ON "
+                "anon_1.items_id = item_keywords_2.item_id "
+                "JOIN keywords ON "
+                "keywords.id = item_keywords_2.keyword_id",
+                use_default_dialect=True,
+            )
+
+    def test_single_prop_9(self):
+        User = self.classes.User
+
+        sess = create_session()
+        with self._from_self_deprecated():
+            self.assert_compile(
+                sess.query(User)
+                .filter(User.name == "ed")
+                .from_self()
+                .join(User.orders),
+                "SELECT anon_1.users_id AS anon_1_users_id, "
+                "anon_1.users_name AS anon_1_users_name "
+                "FROM (SELECT users.id AS users_id, users.name AS users_name "
+                "FROM users "
+                "WHERE users.name = :name_1) AS anon_1 JOIN orders "
+                "ON anon_1.users_id = orders.user_id",
+            )
+
+    def test_anonymous_expression_from_self_twice_oldstyle(self):
+        # relies upon _orm_only_from_obj_alias setting
+
+        sess = create_session()
+        c1, c2 = column("c1"), column("c2")
+        q1 = sess.query(c1, c2).filter(c1 == "dog")
+        with self._from_self_deprecated():
+            q1 = q1.from_self().from_self()
+        self.assert_compile(
+            q1.order_by(c1),
+            "SELECT anon_1.anon_2_c1 AS anon_1_anon_2_c1, anon_1.anon_2_c2 AS "
+            "anon_1_anon_2_c2 FROM (SELECT anon_2.c1 AS anon_2_c1, anon_2.c2 "
+            "AS anon_2_c2 "
+            "FROM (SELECT c1, c2 WHERE c1 = :c1_1) AS "
+            "anon_2) AS anon_1 ORDER BY anon_1.anon_2_c1",
+        )
+
+    def test_anonymous_expression_plus_flag_aliased_join(self):
+        """test that the 'dont alias non-ORM' rule remains for other
+        kinds of aliasing when _from_selectable() is used."""
+
+        User = self.classes.User
+        Address = self.classes.Address
+        addresses = self.tables.addresses
+
+        sess = create_session()
+        q1 = sess.query(User.id).filter(User.id > 5)
+        with self._from_self_deprecated():
+            q1 = q1.from_self()
+
+        q1 = q1.join(User.addresses, aliased=True).order_by(
+            User.id, Address.id, addresses.c.id
+        )
+
+        self.assert_compile(
+            q1,
+            "SELECT anon_1.users_id AS anon_1_users_id "
+            "FROM (SELECT users.id AS users_id FROM users "
+            "WHERE users.id > :id_1) AS anon_1 JOIN addresses AS addresses_1 "
+            "ON anon_1.users_id = addresses_1.user_id "
+            "ORDER BY anon_1.users_id, addresses_1.id, addresses.id",
+        )
+
+    def test_anonymous_expression_plus_explicit_aliased_join(self):
+        """test that the 'dont alias non-ORM' rule remains for other
+        kinds of aliasing when _from_selectable() is used."""
+
+        User = self.classes.User
+        Address = self.classes.Address
+        addresses = self.tables.addresses
+
+        sess = create_session()
+        q1 = sess.query(User.id).filter(User.id > 5)
+        with self._from_self_deprecated():
+            q1 = q1.from_self()
+
+        aa = aliased(Address)
+        q1 = q1.join(aa, User.addresses).order_by(
+            User.id, aa.id, addresses.c.id
+        )
+        self.assert_compile(
+            q1,
+            "SELECT anon_1.users_id AS anon_1_users_id "
+            "FROM (SELECT users.id AS users_id FROM users "
+            "WHERE users.id > :id_1) AS anon_1 JOIN addresses AS addresses_1 "
+            "ON anon_1.users_id = addresses_1.user_id "
+            "ORDER BY anon_1.users_id, addresses_1.id, addresses.id",
+        )
+
+    def test_table_anonymous_expression_from_self_twice_oldstyle(self):
+        # relies upon _orm_only_from_obj_alias setting
+        from sqlalchemy.sql import column
+
+        sess = create_session()
+        t1 = table("t1", column("c1"), column("c2"))
+        q1 = sess.query(t1.c.c1, t1.c.c2).filter(t1.c.c1 == "dog")
+        with self._from_self_deprecated():
+            q1 = q1.from_self().from_self()
+        self.assert_compile(
+            q1.order_by(t1.c.c1),
+            "SELECT anon_1.anon_2_t1_c1 "
+            "AS anon_1_anon_2_t1_c1, anon_1.anon_2_t1_c2 "
+            "AS anon_1_anon_2_t1_c2 "
+            "FROM (SELECT anon_2.t1_c1 AS anon_2_t1_c1, "
+            "anon_2.t1_c2 AS anon_2_t1_c2 FROM (SELECT t1.c1 AS t1_c1, t1.c2 "
+            "AS t1_c2 FROM t1 WHERE t1.c1 = :c1_1) AS anon_2) AS anon_1 "
+            "ORDER BY anon_1.anon_2_t1_c1",
+        )
+
+    def test_self_referential(self):
+        Order = self.classes.Order
+
+        sess = create_session()
+        oalias = aliased(Order)
+
+        with self._from_self_deprecated():
+            for q in [
+                sess.query(Order, oalias)
+                .filter(Order.user_id == oalias.user_id)
+                .filter(Order.user_id == 7)
+                .filter(Order.id > oalias.id)
+                .order_by(Order.id, oalias.id),
+                sess.query(Order, oalias)
+                .filter(Order.id > oalias.id)
+                .from_self()
+                .filter(Order.user_id == oalias.user_id)
+                .filter(Order.user_id == 7)
+                .order_by(Order.id, oalias.id),
+                # same thing, but reversed.
+                sess.query(oalias, Order)
+                .filter(Order.id < oalias.id)
+                .from_self()
+                .filter(oalias.user_id == Order.user_id)
+                .filter(oalias.user_id == 7)
+                .order_by(oalias.id, Order.id),
+                # here we go....two layers of aliasing
+                sess.query(Order, oalias)
+                .filter(Order.user_id == oalias.user_id)
+                .filter(Order.user_id == 7)
+                .filter(Order.id > oalias.id)
+                .from_self()
+                .order_by(Order.id, oalias.id)
+                .limit(10)
+                .options(joinedload(Order.items)),
+                # gratuitous four layers
+                sess.query(Order, oalias)
+                .filter(Order.user_id == oalias.user_id)
+                .filter(Order.user_id == 7)
+                .filter(Order.id > oalias.id)
+                .from_self()
+                .from_self()
+                .from_self()
+                .order_by(Order.id, oalias.id)
+                .limit(10)
+                .options(joinedload(Order.items)),
+            ]:
+
+                eq_(
+                    q.all(),
+                    [
+                        (
+                            Order(
+                                address_id=1,
+                                description="order 3",
+                                isopen=1,
+                                user_id=7,
+                                id=3,
+                            ),
+                            Order(
+                                address_id=1,
+                                description="order 1",
+                                isopen=0,
+                                user_id=7,
+                                id=1,
+                            ),
+                        ),
+                        (
+                            Order(
+                                address_id=None,
+                                description="order 5",
+                                isopen=0,
+                                user_id=7,
+                                id=5,
+                            ),
+                            Order(
+                                address_id=1,
+                                description="order 1",
+                                isopen=0,
+                                user_id=7,
+                                id=1,
+                            ),
+                        ),
+                        (
+                            Order(
+                                address_id=None,
+                                description="order 5",
+                                isopen=0,
+                                user_id=7,
+                                id=5,
+                            ),
+                            Order(
+                                address_id=1,
+                                description="order 3",
+                                isopen=1,
+                                user_id=7,
+                                id=3,
+                            ),
+                        ),
+                    ],
+                )
+
+    def test_from_self_internal_literals_oldstyle(self):
+        # relies upon _orm_only_from_obj_alias setting
+        Order = self.classes.Order
+
+        sess = create_session()
+
+        # ensure column expressions are taken from inside the subquery, not
+        # restated at the top
+        with self._from_self_deprecated():
+            q = (
+                sess.query(
+                    Order.id,
+                    Order.description,
+                    literal_column("'q'").label("foo"),
+                )
+                .filter(Order.description == "order 3")
+                .from_self()
+            )
+        self.assert_compile(
+            q,
+            "SELECT anon_1.orders_id AS "
+            "anon_1_orders_id, "
+            "anon_1.orders_description AS anon_1_orders_description, "
+            "anon_1.foo AS anon_1_foo FROM (SELECT "
+            "orders.id AS orders_id, "
+            "orders.description AS orders_description, "
+            "'q' AS foo FROM orders WHERE "
+            "orders.description = :description_1) AS "
+            "anon_1",
+        )
+        eq_(q.all(), [(3, "order 3", "q")])
+
+    def test_column_access_from_self(self):
+        User = self.classes.User
+        sess = create_session()
+
+        with self._from_self_deprecated():
+            q = sess.query(User).from_self()
+        self.assert_compile(
+            q.filter(User.name == "ed"),
+            "SELECT anon_1.users_id AS anon_1_users_id, anon_1.users_name AS "
+            "anon_1_users_name FROM (SELECT users.id AS users_id, users.name "
+            "AS users_name FROM users) AS anon_1 WHERE anon_1.users_name = "
+            ":name_1",
+        )
+
+    def test_column_access_from_self_twice(self):
+        User = self.classes.User
+        sess = create_session()
+
+        with self._from_self_deprecated():
+            q = sess.query(User).from_self(User.id, User.name).from_self()
+        self.assert_compile(
+            q.filter(User.name == "ed"),
+            "SELECT anon_1.anon_2_users_id AS anon_1_anon_2_users_id, "
+            "anon_1.anon_2_users_name AS anon_1_anon_2_users_name FROM "
+            "(SELECT anon_2.users_id AS anon_2_users_id, anon_2.users_name "
+            "AS anon_2_users_name FROM (SELECT users.id AS users_id, "
+            "users.name AS users_name FROM users) AS anon_2) AS anon_1 "
+            "WHERE anon_1.anon_2_users_name = :name_1",
+        )
+
+    def test_column_queries_nine(self):
+        Address, User = (
+            self.classes.Address,
+            self.classes.User,
+        )
+
+        sess = create_session()
+
+        adalias = aliased(Address)
+        # select from aliasing + explicit aliasing
+        with self._from_self_deprecated():
+            eq_(
+                sess.query(User, adalias.email_address, adalias.id)
+                .outerjoin(adalias, User.addresses)
+                .from_self(User, adalias.email_address)
+                .order_by(User.id, adalias.id)
+                .all(),
+                [
+                    (User(name="jack", id=7), "jack@bean.com"),
+                    (User(name="ed", id=8), "ed@wood.com"),
+                    (User(name="ed", id=8), "ed@bettyboop.com"),
+                    (User(name="ed", id=8), "ed@lala.com"),
+                    (User(name="fred", id=9), "fred@fred.com"),
+                    (User(name="chuck", id=10), None),
+                ],
+            )
+
+    def test_column_queries_ten(self):
+        Address, User = (
+            self.classes.Address,
+            self.classes.User,
+        )
+
+        sess = create_session()
+
+        # anon + select from aliasing
+        aa = aliased(Address)
+        with self._from_self_deprecated():
+            eq_(
+                sess.query(User)
+                .join(aa, User.addresses)
+                .filter(aa.email_address.like("%ed%"))
+                .from_self()
+                .all(),
+                [User(name="ed", id=8), User(name="fred", id=9)],
+            )
+
+    def test_column_queries_eleven(self):
+        Address, User = (
+            self.classes.Address,
+            self.classes.User,
+        )
+
+        sess = create_session()
+
+        adalias = aliased(Address)
+        # test eager aliasing, with/without select_entity_from aliasing
+        with self._from_self_deprecated():
+            for q in [
+                sess.query(User, adalias.email_address)
+                .outerjoin(adalias, User.addresses)
+                .options(joinedload(User.addresses))
+                .order_by(User.id, adalias.id)
+                .limit(10),
+                sess.query(User, adalias.email_address, adalias.id)
+                .outerjoin(adalias, User.addresses)
+                .from_self(User, adalias.email_address)
+                .options(joinedload(User.addresses))
+                .order_by(User.id, adalias.id)
+                .limit(10),
+            ]:
+                eq_(
+                    q.all(),
+                    [
+                        (
+                            User(
+                                addresses=[
+                                    Address(
+                                        user_id=7,
+                                        email_address="jack@bean.com",
+                                        id=1,
+                                    )
+                                ],
+                                name="jack",
+                                id=7,
+                            ),
+                            "jack@bean.com",
+                        ),
+                        (
+                            User(
+                                addresses=[
+                                    Address(
+                                        user_id=8,
+                                        email_address="ed@wood.com",
+                                        id=2,
+                                    ),
+                                    Address(
+                                        user_id=8,
+                                        email_address="ed@bettyboop.com",
+                                        id=3,
+                                    ),
+                                    Address(
+                                        user_id=8,
+                                        email_address="ed@lala.com",
+                                        id=4,
+                                    ),
+                                ],
+                                name="ed",
+                                id=8,
+                            ),
+                            "ed@wood.com",
+                        ),
+                        (
+                            User(
+                                addresses=[
+                                    Address(
+                                        user_id=8,
+                                        email_address="ed@wood.com",
+                                        id=2,
+                                    ),
+                                    Address(
+                                        user_id=8,
+                                        email_address="ed@bettyboop.com",
+                                        id=3,
+                                    ),
+                                    Address(
+                                        user_id=8,
+                                        email_address="ed@lala.com",
+                                        id=4,
+                                    ),
+                                ],
+                                name="ed",
+                                id=8,
+                            ),
+                            "ed@bettyboop.com",
+                        ),
+                        (
+                            User(
+                                addresses=[
+                                    Address(
+                                        user_id=8,
+                                        email_address="ed@wood.com",
+                                        id=2,
+                                    ),
+                                    Address(
+                                        user_id=8,
+                                        email_address="ed@bettyboop.com",
+                                        id=3,
+                                    ),
+                                    Address(
+                                        user_id=8,
+                                        email_address="ed@lala.com",
+                                        id=4,
+                                    ),
+                                ],
+                                name="ed",
+                                id=8,
+                            ),
+                            "ed@lala.com",
+                        ),
+                        (
+                            User(
+                                addresses=[
+                                    Address(
+                                        user_id=9,
+                                        email_address="fred@fred.com",
+                                        id=5,
+                                    )
+                                ],
+                                name="fred",
+                                id=9,
+                            ),
+                            "fred@fred.com",
+                        ),
+                        (User(addresses=[], name="chuck", id=10), None),
+                    ],
+                )
+
+    def test_filter(self):
+        User = self.classes.User
+
+        with self._from_self_deprecated():
+            eq_(
+                [User(id=8), User(id=9)],
+                create_session()
+                .query(User)
+                .filter(User.id.in_([8, 9]))
+                .from_self()
+                .all(),
+            )
+
+        with self._from_self_deprecated():
+            eq_(
+                [User(id=8), User(id=9)],
+                create_session()
+                .query(User)
+                .order_by(User.id)
+                .slice(1, 3)
+                .from_self()
+                .all(),
+            )
+
+        with self._from_self_deprecated():
+            eq_(
+                [User(id=8)],
+                list(
+                    create_session()
+                    .query(User)
+                    .filter(User.id.in_([8, 9]))
+                    .from_self()
+                    .order_by(User.id)[0:1]
+                ),
+            )
+
+    def test_join(self):
+        User, Address = self.classes.User, self.classes.Address
+
+        with self._from_self_deprecated():
+            eq_(
+                [
+                    (User(id=8), Address(id=2)),
+                    (User(id=8), Address(id=3)),
+                    (User(id=8), Address(id=4)),
+                    (User(id=9), Address(id=5)),
+                ],
+                create_session()
+                .query(User)
+                .filter(User.id.in_([8, 9]))
+                .from_self()
+                .join("addresses")
+                .add_entity(Address)
+                .order_by(User.id, Address.id)
+                .all(),
+            )
+
+    def test_group_by(self):
+        Address = self.classes.Address
+
+        eq_(
+            create_session()
+            .query(Address.user_id, func.count(Address.id).label("count"))
+            .group_by(Address.user_id)
+            .order_by(Address.user_id)
+            .all(),
+            [(7, 1), (8, 3), (9, 1)],
+        )
+
+        with self._from_self_deprecated():
+            eq_(
+                create_session()
+                .query(Address.user_id, Address.id)
+                .from_self(Address.user_id, func.count(Address.id))
+                .group_by(Address.user_id)
+                .order_by(Address.user_id)
+                .all(),
+                [(7, 1), (8, 3), (9, 1)],
+            )
+
+    def test_having(self):
+        User = self.classes.User
+
+        s = create_session()
+
+        with self._from_self_deprecated():
+            self.assert_compile(
+                s.query(User.id)
+                .group_by(User.id)
+                .having(User.id > 5)
+                .from_self(),
+                "SELECT anon_1.users_id AS anon_1_users_id FROM "
+                "(SELECT users.id AS users_id FROM users GROUP "
+                "BY users.id HAVING users.id > :id_1) AS anon_1",
+            )
+
+    def test_no_joinedload(self):
+        """test that joinedloads are pushed outwards and not rendered in
+        subqueries."""
+
+        User = self.classes.User
+
+        s = create_session()
+
+        with self._from_self_deprecated():
+            q = s.query(User).options(joinedload(User.addresses)).from_self()
+
+        self.assert_compile(
+            q.statement,
+            "SELECT anon_1.users_id, anon_1.users_name, addresses_1.id, "
+            "addresses_1.user_id, addresses_1.email_address FROM "
+            "(SELECT users.id AS users_id, users.name AS "
+            "users_name FROM users) AS anon_1 LEFT OUTER JOIN "
+            "addresses AS addresses_1 ON anon_1.users_id = "
+            "addresses_1.user_id ORDER BY addresses_1.id",
+        )
+
+    def test_aliases(self):
+        """test that aliased objects are accessible externally to a from_self()
+        call."""
+
+        User, Address = self.classes.User, self.classes.Address
+
+        s = create_session()
+
+        ualias = aliased(User)
+
+        with self._from_self_deprecated():
+            eq_(
+                s.query(User, ualias)
+                .filter(User.id > ualias.id)
+                .from_self(User.name, ualias.name)
+                .order_by(User.name, ualias.name)
+                .all(),
+                [
+                    ("chuck", "ed"),
+                    ("chuck", "fred"),
+                    ("chuck", "jack"),
+                    ("ed", "jack"),
+                    ("fred", "ed"),
+                    ("fred", "jack"),
+                ],
+            )
+
+        with self._from_self_deprecated():
+            eq_(
+                s.query(User, ualias)
+                .filter(User.id > ualias.id)
+                .from_self(User.name, ualias.name)
+                .filter(ualias.name == "ed")
+                .order_by(User.name, ualias.name)
+                .all(),
+                [("chuck", "ed"), ("fred", "ed")],
+            )
+
+        with self._from_self_deprecated():
+            eq_(
+                s.query(User, ualias)
+                .filter(User.id > ualias.id)
+                .from_self(ualias.name, Address.email_address)
+                .join(ualias.addresses)
+                .order_by(ualias.name, Address.email_address)
+                .all(),
+                [
+                    ("ed", "fred@fred.com"),
+                    ("jack", "ed@bettyboop.com"),
+                    ("jack", "ed@lala.com"),
+                    ("jack", "ed@wood.com"),
+                    ("jack", "fred@fred.com"),
+                ],
+            )
+
+    def test_multiple_entities(self):
+        User, Address = self.classes.User, self.classes.Address
+
+        sess = create_session()
+
+        with self._from_self_deprecated():
+            eq_(
+                sess.query(User, Address)
+                .filter(User.id == Address.user_id)
+                .filter(Address.id.in_([2, 5]))
+                .from_self()
+                .all(),
+                [(User(id=8), Address(id=2)), (User(id=9), Address(id=5))],
+            )
+
+        with self._from_self_deprecated():
+            eq_(
+                sess.query(User, Address)
+                .filter(User.id == Address.user_id)
+                .filter(Address.id.in_([2, 5]))
+                .from_self()
+                .options(joinedload("addresses"))
+                .first(),
+                (
+                    User(id=8, addresses=[Address(), Address(), Address()]),
+                    Address(id=2),
+                ),
+            )
+
+    def test_multiple_with_column_entities_oldstyle(self):
+        # relies upon _orm_only_from_obj_alias setting
+        User = self.classes.User
+
+        sess = create_session()
+
+        with self._from_self_deprecated():
+            eq_(
+                sess.query(User.id)
+                .from_self()
+                .add_columns(func.count().label("foo"))
+                .group_by(User.id)
+                .order_by(User.id)
+                .from_self()
+                .all(),
+                [(7, 1), (8, 1), (9, 1), (10, 1)],
+            )
+
+
+class SubqRelationsFromSelfTest(fixtures.DeclarativeMappedTest):
+    def _from_self_deprecated(self):
+        return testing.expect_deprecated_20(
+            r"The Query.from_self\(\) function/method"
+        )
+
+    @classmethod
+    def setup_classes(cls):
+        Base = cls.DeclarativeBasic
+
+        class A(Base, ComparableEntity):
+            __tablename__ = "a"
+
+            id = Column(Integer, primary_key=True)
+            cs = relationship("C", order_by="C.id")
+
+        class B(Base, ComparableEntity):
+            __tablename__ = "b"
+            id = Column(Integer, primary_key=True)
+            a_id = Column(ForeignKey("a.id"))
+            a = relationship("A")
+            ds = relationship("D", order_by="D.id")
+
+        class C(Base, ComparableEntity):
+            __tablename__ = "c"
+            id = Column(Integer, primary_key=True)
+            a_id = Column(ForeignKey("a.id"))
+
+        class D(Base, ComparableEntity):
+            __tablename__ = "d"
+            id = Column(Integer, primary_key=True)
+            b_id = Column(ForeignKey("b.id"))
+
+    @classmethod
+    def insert_data(cls, connection):
+        A, B, C, D = cls.classes("A", "B", "C", "D")
+
+        s = Session(connection)
+
+        as_ = [A(id=i, cs=[C(), C()],) for i in range(1, 5)]
+
+        s.add_all(
+            [
+                B(a=as_[0], ds=[D()]),
+                B(a=as_[1], ds=[D()]),
+                B(a=as_[2]),
+                B(a=as_[3]),
+            ]
+        )
+
+        s.commit()
+
+    def test_subq_w_from_self_one(self):
+        A, B, C = self.classes("A", "B", "C")
+
+        s = Session()
+
+        cache = {}
+
+        for i in range(3):
+            with self._from_self_deprecated():
+                q = (
+                    s.query(B)
+                    .execution_options(compiled_cache=cache)
+                    .join(B.a)
+                    .filter(B.id < 4)
+                    .filter(A.id > 1)
+                    .from_self()
+                    .options(subqueryload(B.a).subqueryload(A.cs))
+                    .from_self()
+                )
+
+            def go():
+                results = q.all()
+                eq_(
+                    results,
+                    [
+                        B(
+                            a=A(cs=[C(a_id=2, id=3), C(a_id=2, id=4)], id=2),
+                            a_id=2,
+                            id=2,
+                        ),
+                        B(
+                            a=A(cs=[C(a_id=3, id=5), C(a_id=3, id=6)], id=3),
+                            a_id=3,
+                            id=3,
+                        ),
+                    ],
+                )
+
+            self.assert_sql_execution(
+                testing.db,
+                go,
+                CompiledSQL(
+                    "SELECT anon_1.anon_2_b_id AS anon_1_anon_2_b_id, "
+                    "anon_1.anon_2_b_a_id AS anon_1_anon_2_b_a_id FROM "
+                    "(SELECT anon_2.b_id AS anon_2_b_id, anon_2.b_a_id "
+                    "AS anon_2_b_a_id FROM (SELECT b.id AS b_id, b.a_id "
+                    "AS b_a_id FROM b JOIN a ON a.id = b.a_id "
+                    "WHERE b.id < :id_1 AND a.id > :id_2) AS anon_2) AS anon_1"
+                ),
+                CompiledSQL(
+                    "SELECT a.id AS a_id, anon_1.anon_2_anon_3_b_a_id AS "
+                    "anon_1_anon_2_anon_3_b_a_id FROM (SELECT DISTINCT "
+                    "anon_2.anon_3_b_a_id AS anon_2_anon_3_b_a_id FROM "
+                    "(SELECT anon_3.b_id AS anon_3_b_id, anon_3.b_a_id "
+                    "AS anon_3_b_a_id FROM (SELECT b.id AS b_id, b.a_id "
+                    "AS b_a_id FROM b JOIN a ON a.id = b.a_id "
+                    "WHERE b.id < :id_1 AND a.id > :id_2) AS anon_3) "
+                    "AS anon_2) AS anon_1 JOIN a "
+                    "ON a.id = anon_1.anon_2_anon_3_b_a_id"
+                ),
+                CompiledSQL(
+                    "SELECT c.id AS c_id, c.a_id AS c_a_id, a_1.id "
+                    "AS a_1_id FROM (SELECT DISTINCT anon_2.anon_3_b_a_id AS "
+                    "anon_2_anon_3_b_a_id FROM "
+                    "(SELECT anon_3.b_id AS anon_3_b_id, anon_3.b_a_id "
+                    "AS anon_3_b_a_id FROM (SELECT b.id AS b_id, b.a_id "
+                    "AS b_a_id FROM b JOIN a ON a.id = b.a_id "
+                    "WHERE b.id < :id_1 AND a.id > :id_2) AS anon_3) "
+                    "AS anon_2) AS anon_1 JOIN a AS a_1 ON a_1.id = "
+                    "anon_1.anon_2_anon_3_b_a_id JOIN c ON a_1.id = c.a_id "
+                    "ORDER BY c.id"
+                ),
+            )
+
+            s.close()
+
+    def test_subq_w_from_self_two(self):
+
+        A, B, C = self.classes("A", "B", "C")
+
+        s = Session()
+        cache = {}
+
+        for i in range(3):
+
+            def go():
+                with self._from_self_deprecated():
+                    q = (
+                        s.query(B)
+                        .execution_options(compiled_cache=cache)
+                        .join(B.a)
+                        .from_self()
+                    )
+                q = q.options(subqueryload(B.ds))
+
+                q.all()
+
+            self.assert_sql_execution(
+                testing.db,
+                go,
+                CompiledSQL(
+                    "SELECT anon_1.b_id AS anon_1_b_id, anon_1.b_a_id AS "
+                    "anon_1_b_a_id FROM (SELECT b.id AS b_id, b.a_id "
+                    "AS b_a_id FROM b JOIN a ON a.id = b.a_id) AS anon_1"
+                ),
+                CompiledSQL(
+                    "SELECT d.id AS d_id, d.b_id AS d_b_id, "
+                    "anon_1.anon_2_b_id AS anon_1_anon_2_b_id "
+                    "FROM (SELECT anon_2.b_id AS anon_2_b_id FROM "
+                    "(SELECT b.id AS b_id, b.a_id AS b_a_id FROM b "
+                    "JOIN a ON a.id = b.a_id) AS anon_2) AS anon_1 "
+                    "JOIN d ON anon_1.anon_2_b_id = d.b_id ORDER BY d.id"
+                ),
+            )
+            s.close()
+
+
 class SessionTest(fixtures.RemovesEvents, _LocalFixture):
     def test_subtransactions_deprecated(self):
         s1 = Session(testing.db)
diff --git a/test/orm/test_froms.py b/test/orm/test_froms.py
index d29ed65e8..79ed3cad4 100644
--- a/test/orm/test_froms.py
+++ b/test/orm/test_froms.py
@@ -370,44 +370,45 @@ class RawSelectTest(QueryTest, AssertsCompiledSQL):
         )
 
 
-class FromSelfTest(QueryTest, AssertsCompiledSQL):
+class EntityFromSubqueryTest(QueryTest, AssertsCompiledSQL):
+    # formerly FromSelfTest
+
     __dialect__ = "default"
 
     def test_filter(self):
         User = self.classes.User
 
+        subq = select(User).filter(User.id.in_([8, 9])).subquery()
+        q = create_session().query(aliased(User, subq))
         eq_(
-            [User(id=8), User(id=9)],
-            create_session()
-            .query(User)
-            .filter(User.id.in_([8, 9]))
-            .from_self()
-            .all(),
+            [User(id=8), User(id=9)], q.all(),
         )
 
+        subq = select(User).order_by(User.id).slice(1, 3).subquery()
+        q = create_session().query(aliased(User, subq))
+        eq_([User(id=8), User(id=9)], q.all())
+
+        subq = select(User).filter(User.id.in_([8, 9])).subquery()
+        u = aliased(User, subq)
+        q = create_session().query(u).order_by(u.id)
         eq_(
-            [User(id=8), User(id=9)],
-            create_session()
-            .query(User)
-            .order_by(User.id)
-            .slice(1, 3)
-            .from_self()
-            .all(),
-        )
-        eq_(
-            [User(id=8)],
-            list(
-                create_session()
-                .query(User)
-                .filter(User.id.in_([8, 9]))
-                .from_self()
-                .order_by(User.id)[0:1]
-            ),
+            [User(id=8)], list(q[0:1]),
         )
 
     def test_join(self):
         User, Address = self.classes.User, self.classes.Address
 
+        stmt = select(User).filter(User.id.in_([8, 9])).subquery()
+
+        u = aliased(User, stmt)
+
+        q = (
+            create_session()
+            .query(u)
+            .join(u.addresses)
+            .add_entity(Address)
+            .order_by(u.id, Address.id)
+        )
         eq_(
             [
                 (User(id=8), Address(id=2)),
@@ -415,36 +416,55 @@ class FromSelfTest(QueryTest, AssertsCompiledSQL):
                 (User(id=8), Address(id=4)),
                 (User(id=9), Address(id=5)),
             ],
-            create_session()
-            .query(User)
-            .filter(User.id.in_([8, 9]))
-            .from_self()
-            .join("addresses")
-            .add_entity(Address)
-            .order_by(User.id, Address.id)
-            .all(),
+            q.all(),
         )
 
     def test_group_by(self):
         Address = self.classes.Address
 
-        eq_(
-            create_session()
-            .query(Address.user_id, func.count(Address.id).label("count"))
+        subq = (
+            select(Address.user_id, func.count(Address.id).label("count"))
             .group_by(Address.user_id)
             .order_by(Address.user_id)
-            .all(),
-            [(7, 1), (8, 3), (9, 1)],
+            .subquery()
         )
-
+        # there's no reason to do aliased(Address) in this case but we're just
+        # testing
+        aq = aliased(Address, subq)
+        q = create_session().query(aq.user_id, subq.c.count)
         eq_(
+            q.all(), [(7, 1), (8, 3), (9, 1)],
+        )
+
+        subq = select(Address.user_id, Address.id)
+        aq = aliased(Address, subq)
+
+        q = (
             create_session()
-            .query(Address.user_id, Address.id)
-            .from_self(Address.user_id, func.count(Address.id))
-            .group_by(Address.user_id)
-            .order_by(Address.user_id)
-            .all(),
-            [(7, 1), (8, 3), (9, 1)],
+            .query(aq.user_id, func.count(aq.id))
+            .group_by(aq.user_id)
+            .order_by(aq.user_id)
+        )
+
+        eq_(
+            q.all(), [(7, 1), (8, 3), (9, 1)],
+        )
+
+    def test_error_w_aliased_against_select(self):
+        User = self.classes.User
+
+        s = create_session()
+
+        stmt = select(User.id)
+
+        assert_raises_message(
+            sa_exc.ArgumentError,
+            "Column expression or FROM clause expected, got "
+            "<sqlalchemy.sql.selectable.Select .*> object resolved from "
+            "<AliasedClass .* User> object. To create a FROM clause from "
+            "a <class 'sqlalchemy.sql.selectable.Select'> object",
+            s.query,
+            aliased(User, stmt),
         )
 
     def test_having(self):
@@ -452,26 +472,49 @@ class FromSelfTest(QueryTest, AssertsCompiledSQL):
 
         s = create_session()
 
+        stmt = (
+            select(User.id)
+            .group_by(User.id)
+            .having(User.id > 5)
+            .apply_labels()
+            .subquery()
+        )
+
+        q = s.query(aliased(User, stmt))
         self.assert_compile(
-            s.query(User.id).group_by(User.id).having(User.id > 5).from_self(),
+            q,
             "SELECT anon_1.users_id AS anon_1_users_id FROM "
             "(SELECT users.id AS users_id FROM users GROUP "
             "BY users.id HAVING users.id > :id_1) AS anon_1",
         )
 
     def test_no_joinedload(self):
-        """test that joinedloads are pushed outwards and not rendered in
-        subqueries."""
 
         User = self.classes.User
 
         s = create_session()
 
-        self.assert_compile(
-            s.query(User)
+        subq = (
+            select(User)
             .options(joinedload(User.addresses))
-            .from_self()
-            .statement,
+            .apply_labels()
+            .subquery()
+        )
+
+        uq = aliased(User, subq)
+        q = s.query(uq)
+
+        # in 2.0 style, joinedload in the subquery is just ignored
+        self.assert_compile(
+            q.statement,
+            "SELECT anon_1.users_id, anon_1.users_name FROM (SELECT "
+            "users.id AS users_id, users.name AS users_name FROM users) "
+            "AS anon_1",
+        )
+
+        # needs to be on the outside
+        self.assert_compile(
+            q.options(joinedload(uq.addresses)).statement,
             "SELECT anon_1.users_id, anon_1.users_name, addresses_1.id, "
             "addresses_1.user_id, addresses_1.email_address FROM "
             "(SELECT users.id AS users_id, users.name AS "
@@ -489,12 +532,16 @@ class FromSelfTest(QueryTest, AssertsCompiledSQL):
         s = create_session()
 
         ualias = aliased(User)
+
+        subq = select(User, ualias).filter(User.id > ualias.id).subquery()
+
+        uq1 = aliased(User, subq)
+        uq2 = aliased(ualias, subq)
+
+        q = s.query(uq1.name, uq2.name).order_by(uq1.name, uq2.name)
+
         eq_(
-            s.query(User, ualias)
-            .filter(User.id > ualias.id)
-            .from_self(User.name, ualias.name)
-            .order_by(User.name, ualias.name)
-            .all(),
+            q.all(),
             [
                 ("chuck", "ed"),
                 ("chuck", "fred"),
@@ -505,23 +552,24 @@ class FromSelfTest(QueryTest, AssertsCompiledSQL):
             ],
         )
 
+        q = (
+            s.query(uq1.name, uq2.name)
+            .filter(uq2.name == "ed")
+            .order_by(uq1.name, uq2.name)
+        )
+
         eq_(
-            s.query(User, ualias)
-            .filter(User.id > ualias.id)
-            .from_self(User.name, ualias.name)
-            .filter(ualias.name == "ed")
-            .order_by(User.name, ualias.name)
-            .all(),
-            [("chuck", "ed"), ("fred", "ed")],
+            q.all(), [("chuck", "ed"), ("fred", "ed")],
+        )
+
+        q = (
+            s.query(uq2.name, Address.email_address)
+            .join(uq2.addresses)
+            .order_by(uq2.name, Address.email_address)
         )
 
         eq_(
-            s.query(User, ualias)
-            .filter(User.id > ualias.id)
-            .from_self(ualias.name, Address.email_address)
-            .join(ualias.addresses)
-            .order_by(ualias.name, Address.email_address)
-            .all(),
+            q.all(),
             [
                 ("ed", "fred@fred.com"),
                 ("jack", "ed@bettyboop.com"),
@@ -536,22 +584,23 @@ class FromSelfTest(QueryTest, AssertsCompiledSQL):
 
         sess = create_session()
 
-        eq_(
-            sess.query(User, Address)
+        subq = (
+            select(User, Address)
             .filter(User.id == Address.user_id)
             .filter(Address.id.in_([2, 5]))
-            .from_self()
-            .all(),
+            .subquery()
+        )
+
+        uq = aliased(User, subq)
+        aq = aliased(Address, subq)
+
+        eq_(
+            sess.query(uq, aq).all(),
             [(User(id=8), Address(id=2)), (User(id=9), Address(id=5))],
         )
 
         eq_(
-            sess.query(User, Address)
-            .filter(User.id == Address.user_id)
-            .filter(Address.id.in_([2, 5]))
-            .from_self()
-            .options(joinedload("addresses"))
-            .first(),
+            sess.query(uq, aq).options(joinedload(uq.addresses)).first(),
             (
                 User(id=8, addresses=[Address(), Address(), Address()]),
                 Address(id=2),
@@ -559,19 +608,26 @@ class FromSelfTest(QueryTest, AssertsCompiledSQL):
         )
 
     def test_multiple_with_column_entities_oldstyle(self):
-        # relies upon _orm_only_from_obj_alias setting
+        # this is now very awkward and not very useful
         User = self.classes.User
 
+        subq = select(User.id).subquery()
+
+        uq = aliased(User, subq)
+
+        subq2 = (
+            select(uq.id)
+            .add_columns(func.count().label("foo"))
+            .group_by(uq.id)
+            .order_by(uq.id)
+            .subquery()
+        )
+
+        uq2 = aliased(User, subq2)
         sess = create_session()
 
         eq_(
-            sess.query(User.id)
-            .from_self()
-            .add_columns(func.count().label("foo"))
-            .group_by(User.id)
-            .order_by(User.id)
-            .from_self()
-            .all(),
+            sess.query(uq2.id, subq2.c.foo).all(),
             [(7, 1), (8, 1), (9, 1), (10, 1)],
         )
 
@@ -601,34 +657,6 @@ class ColumnAccessTest(QueryTest, AssertsCompiledSQL):
 
     __dialect__ = "default"
 
-    def test_from_self(self):
-        User = self.classes.User
-        sess = create_session()
-
-        q = sess.query(User).from_self()
-        self.assert_compile(
-            q.filter(User.name == "ed"),
-            "SELECT anon_1.users_id AS anon_1_users_id, anon_1.users_name AS "
-            "anon_1_users_name FROM (SELECT users.id AS users_id, users.name "
-            "AS users_name FROM users) AS anon_1 WHERE anon_1.users_name = "
-            ":name_1",
-        )
-
-    def test_from_self_twice(self):
-        User = self.classes.User
-        sess = create_session()
-
-        q = sess.query(User).from_self(User.id, User.name).from_self()
-        self.assert_compile(
-            q.filter(User.name == "ed"),
-            "SELECT anon_1.anon_2_users_id AS anon_1_anon_2_users_id, "
-            "anon_1.anon_2_users_name AS anon_1_anon_2_users_name FROM "
-            "(SELECT anon_2.users_id AS anon_2_users_id, anon_2.users_name "
-            "AS anon_2_users_name FROM (SELECT users.id AS users_id, "
-            "users.name AS users_name FROM users) AS anon_2) AS anon_1 "
-            "WHERE anon_1.anon_2_users_name = :name_1",
-        )
-
     def test_select_entity_from(self):
         User = self.classes.User
         sess = create_session()
@@ -700,25 +728,6 @@ class ColumnAccessTest(QueryTest, AssertsCompiledSQL):
             "WHERE c1 = :c1_2) AS anon_1 ORDER BY anon_1.c1",
         )
 
-    def test_table_anonymous_expression_from_self_twice_oldstyle(self):
-        # relies upon _orm_only_from_obj_alias setting
-        from sqlalchemy.sql import column
-
-        sess = create_session()
-        t1 = table("t1", column("c1"), column("c2"))
-        q1 = sess.query(t1.c.c1, t1.c.c2).filter(t1.c.c1 == "dog")
-        q1 = q1.from_self().from_self()
-        self.assert_compile(
-            q1.order_by(t1.c.c1),
-            "SELECT anon_1.anon_2_t1_c1 "
-            "AS anon_1_anon_2_t1_c1, anon_1.anon_2_t1_c2 "
-            "AS anon_1_anon_2_t1_c2 "
-            "FROM (SELECT anon_2.t1_c1 AS anon_2_t1_c1, "
-            "anon_2.t1_c2 AS anon_2_t1_c2 FROM (SELECT t1.c1 AS t1_c1, t1.c2 "
-            "AS t1_c2 FROM t1 WHERE t1.c1 = :c1_1) AS anon_2) AS anon_1 "
-            "ORDER BY anon_1.anon_2_t1_c1",
-        )
-
     def test_table_anonymous_expression_from_self_twice_newstyle(self):
         from sqlalchemy.sql import column
 
@@ -748,23 +757,6 @@ class ColumnAccessTest(QueryTest, AssertsCompiledSQL):
             "ORDER BY anon_1.anon_2_t1_c1",
         )
 
-    def test_anonymous_expression_from_self_twice_oldstyle(self):
-        # relies upon _orm_only_from_obj_alias setting
-        from sqlalchemy.sql import column
-
-        sess = create_session()
-        c1, c2 = column("c1"), column("c2")
-        q1 = sess.query(c1, c2).filter(c1 == "dog")
-        q1 = q1.from_self().from_self()
-        self.assert_compile(
-            q1.order_by(c1),
-            "SELECT anon_1.anon_2_c1 AS anon_1_anon_2_c1, anon_1.anon_2_c2 AS "
-            "anon_1_anon_2_c2 FROM (SELECT anon_2.c1 AS anon_2_c1, anon_2.c2 "
-            "AS anon_2_c2 "
-            "FROM (SELECT c1, c2 WHERE c1 = :c1_1) AS "
-            "anon_2) AS anon_1 ORDER BY anon_1.anon_2_c1",
-        )
-
     def test_anonymous_expression_from_self_twice_newstyle_wlabels(self):
         from sqlalchemy.sql import column
 
@@ -835,9 +827,7 @@ class ColumnAccessTest(QueryTest, AssertsCompiledSQL):
             "WHERE c1 = :c1_2) AS anon_1 ORDER BY anon_1.foo",
         )
 
-    def test_anonymous_expression_plus_flag_aliased_join(self):
-        """test that the 'dont alias non-ORM' rule remains for other
-        kinds of aliasing when _from_selectable() is used."""
+    def test_anonymous_expression_plus_flag_aliased_join_newstyle(self):
 
         User = self.classes.User
         Address = self.classes.Address
@@ -845,10 +835,14 @@ class ColumnAccessTest(QueryTest, AssertsCompiledSQL):
 
         sess = create_session()
         q1 = sess.query(User.id).filter(User.id > 5)
-        q1 = q1.from_self()
 
-        q1 = q1.join(User.addresses, aliased=True).order_by(
-            User.id, Address.id, addresses.c.id
+        uq = aliased(User, q1.apply_labels().subquery())
+
+        aa = aliased(Address)
+        q1 = (
+            sess.query(uq.id)
+            .join(uq.addresses.of_type(aa))
+            .order_by(uq.id, aa.id, addresses.c.id)
         )
 
         self.assert_compile(
@@ -860,7 +854,7 @@ class ColumnAccessTest(QueryTest, AssertsCompiledSQL):
             "ORDER BY anon_1.users_id, addresses_1.id, addresses.id",
         )
 
-    def test_anonymous_expression_plus_explicit_aliased_join(self):
+    def test_anonymous_expression_plus_explicit_aliased_join_newstyle(self):
         """test that the 'dont alias non-ORM' rule remains for other
         kinds of aliasing when _from_selectable() is used."""
 
@@ -869,12 +863,16 @@ class ColumnAccessTest(QueryTest, AssertsCompiledSQL):
         addresses = self.tables.addresses
 
         sess = create_session()
-        q1 = sess.query(User.id).filter(User.id > 5)
-        q1 = q1.from_self()
+        q1 = sess.query(User.id).filter(User.id > 5).apply_labels().subquery()
+
+        uq = aliased(User, q1)
 
         aa = aliased(Address)
-        q1 = q1.join(aa, User.addresses).order_by(
-            User.id, aa.id, addresses.c.id
+
+        q1 = (
+            sess.query(uq.id)
+            .join(aa, uq.addresses)
+            .order_by(uq.id, aa.id, addresses.c.id)
         )
         self.assert_compile(
             q1,
@@ -1790,13 +1788,19 @@ class MixedEntitiesTest(QueryTest, AssertsCompiledSQL):
         sess = create_session()
 
         adalias = aliased(Address)
-        # select from aliasing + explicit aliasing
-        eq_(
+
+        subq = (
             sess.query(User, adalias.email_address, adalias.id)
             .outerjoin(adalias, User.addresses)
-            .from_self(User, adalias.email_address)
-            .order_by(User.id, adalias.id)
-            .all(),
+            .subquery()
+        )
+        ua = aliased(User, subq)
+        aa = aliased(adalias, subq)
+
+        q = sess.query(ua, aa.email_address).order_by(ua.id, aa.id)
+        # select from aliasing + explicit aliasing
+        eq_(
+            q.all(),
             [
                 (User(name="jack", id=7), "jack@bean.com"),
                 (User(name="ed", id=8), "ed@wood.com"),
@@ -1817,12 +1821,17 @@ class MixedEntitiesTest(QueryTest, AssertsCompiledSQL):
 
         # anon + select from aliasing
         aa = aliased(Address)
-        eq_(
+
+        subq = (
             sess.query(User)
             .join(aa, User.addresses)
             .filter(aa.email_address.like("%ed%"))
-            .from_self()
-            .all(),
+            .subquery()
+        )
+        ua = aliased(User, subq)
+
+        eq_(
+            sess.query(ua).all(),
             [User(name="ed", id=8), User(name="fred", id=9)],
         )
 
@@ -1835,20 +1844,32 @@ class MixedEntitiesTest(QueryTest, AssertsCompiledSQL):
         sess = create_session()
 
         adalias = aliased(Address)
-        # test eager aliasing, with/without select_entity_from aliasing
-        for q in [
+
+        q1 = (
             sess.query(User, adalias.email_address)
             .outerjoin(adalias, User.addresses)
             .options(joinedload(User.addresses))
             .order_by(User.id, adalias.id)
-            .limit(10),
+            .limit(10)
+        )
+
+        subq = (
             sess.query(User, adalias.email_address, adalias.id)
             .outerjoin(adalias, User.addresses)
-            .from_self(User, adalias.email_address)
-            .options(joinedload(User.addresses))
-            .order_by(User.id, adalias.id)
-            .limit(10),
-        ]:
+            .subquery()
+        )
+        ua = aliased(User, subq)
+        aa = aliased(adalias, subq)
+
+        q2 = (
+            sess.query(ua, aa.email_address)
+            .options(joinedload(ua.addresses))
+            .order_by(ua.id, aa.id)
+            .limit(10)
+        )
+
+        # test eager aliasing, with/without select_entity_from aliasing
+        for q in [q1, q2]:
             eq_(
                 q.all(),
                 [
@@ -1973,52 +1994,78 @@ class MixedEntitiesTest(QueryTest, AssertsCompiledSQL):
 
         self.assert_sql_count(testing.db, go, 1)
 
-    @testing.fails_on("firebird", "unknown")
-    def test_self_referential(self):
+    def test_self_referential_from_self(self):
         Order = self.classes.Order
 
         sess = create_session()
         oalias = aliased(Order)
 
-        for q in [
+        q1 = (
             sess.query(Order, oalias)
             .filter(Order.user_id == oalias.user_id)
             .filter(Order.user_id == 7)
             .filter(Order.id > oalias.id)
-            .order_by(Order.id, oalias.id),
-            sess.query(Order, oalias)
-            .filter(Order.id > oalias.id)
-            .from_self()
-            .filter(Order.user_id == oalias.user_id)
-            .filter(Order.user_id == 7)
-            .order_by(Order.id, oalias.id),
-            # same thing, but reversed.
-            sess.query(oalias, Order)
-            .filter(Order.id < oalias.id)
-            .from_self()
-            .filter(oalias.user_id == Order.user_id)
-            .filter(oalias.user_id == 7)
-            .order_by(oalias.id, Order.id),
-            # here we go....two layers of aliasing
-            sess.query(Order, oalias)
-            .filter(Order.user_id == oalias.user_id)
-            .filter(Order.user_id == 7)
-            .filter(Order.id > oalias.id)
-            .from_self()
             .order_by(Order.id, oalias.id)
-            .limit(10)
-            .options(joinedload(Order.items)),
-            # gratuitous four layers
+        )
+
+        subq = (
+            sess.query(Order, oalias).filter(Order.id > oalias.id).subquery()
+        )
+        oa, oaa = aliased(Order, subq), aliased(oalias, subq)
+        q2 = (
+            sess.query(oa, oaa)
+            .filter(oa.user_id == oaa.user_id)
+            .filter(oa.user_id == 7)
+            .order_by(oa.id, oaa.id)
+        )
+
+        # same thing, but reversed.
+        subq = (
+            sess.query(oalias, Order).filter(Order.id < oalias.id).subquery()
+        )
+        oa, oaa = aliased(Order, subq), aliased(oalias, subq)
+        q3 = (
+            sess.query(oaa, oa)
+            .filter(oaa.user_id == oa.user_id)
+            .filter(oaa.user_id == 7)
+            .order_by(oaa.id, oa.id)
+        )
+
+        subq = (
             sess.query(Order, oalias)
             .filter(Order.user_id == oalias.user_id)
             .filter(Order.user_id == 7)
             .filter(Order.id > oalias.id)
-            .from_self()
-            .from_self()
-            .from_self()
-            .order_by(Order.id, oalias.id)
+            .subquery()
+        )
+        oa, oaa = aliased(Order, subq), aliased(oalias, subq)
+
+        # here we go....two layers of aliasing (due to joinedload w/ limit)
+        q4 = (
+            sess.query(oa, oaa)
+            .order_by(oa.id, oaa.id)
+            .limit(10)
+            .options(joinedload(oa.items))
+        )
+
+        # gratuitous four layers
+        subq4 = subq
+        for i in range(4):
+            oa, oaa = aliased(Order, subq4), aliased(oaa, subq4)
+            subq4 = sess.query(oa, oaa).subquery()
+        oa, oaa = aliased(Order, subq4), aliased(oaa, subq4)
+        q5 = (
+            sess.query(oa, oaa)
+            .order_by(oa.id, oaa.id)
             .limit(10)
-            .options(joinedload(Order.items)),
+            .options(joinedload(oa.items))
+        )
+        for q in [
+            q1,
+            q2,
+            q3,
+            q4,
+            q5,
         ]:
 
             eq_(
@@ -2075,35 +2122,6 @@ class MixedEntitiesTest(QueryTest, AssertsCompiledSQL):
                 ],
             )
 
-    def test_from_self_internal_literals_oldstyle(self):
-        # relies upon _orm_only_from_obj_alias setting
-        Order = self.classes.Order
-
-        sess = create_session()
-
-        # ensure column expressions are taken from inside the subquery, not
-        # restated at the top
-        q = (
-            sess.query(
-                Order.id, Order.description, literal_column("'q'").label("foo")
-            )
-            .filter(Order.description == "order 3")
-            .from_self()
-        )
-        self.assert_compile(
-            q,
-            "SELECT anon_1.orders_id AS "
-            "anon_1_orders_id, "
-            "anon_1.orders_description AS anon_1_orders_description, "
-            "anon_1.foo AS anon_1_foo FROM (SELECT "
-            "orders.id AS orders_id, "
-            "orders.description AS orders_description, "
-            "'q' AS foo FROM orders WHERE "
-            "orders.description = :description_1) AS "
-            "anon_1",
-        )
-        eq_(q.all(), [(3, "order 3", "q")])
-
     def test_from_self_internal_literals_newstyle(self):
         Order = self.classes.Order
 
@@ -3651,7 +3669,9 @@ class ExternalColumnsTest(QueryTest):
         eq_(a1.username, "jack")
 
         sess = create_session()
-        a1 = sess.query(Address).from_self().first()
+        subq = sess.query(Address).subquery()
+        aa = aliased(Address, subq)
+        a1 = sess.query(aa).first()
         eq_(a1.username, "jack")
 
 
@@ -3781,13 +3801,18 @@ class LabelCollideTest(fixtures.MappedTest):
 
     def test_overlap_subquery(self):
         s = Session()
-        row = (
+
+        subq = (
             s.query(self.classes.Foo, self.classes.Bar)
             .join(self.classes.Bar, true())
-            .from_self()
-            .all()[0]
+            .subquery()
         )
 
+        fa = aliased(self.classes.Foo, subq, name="Foo")
+        ba = aliased(self.classes.Bar, subq, name="Bar")
+
+        row = s.query(fa, ba).all()[0]
+
         def go():
             eq_(row.Foo.id, 1)
             eq_(row.Foo.bar_id, 2)
diff --git a/test/orm/test_joins.py b/test/orm/test_joins.py
index 8225214f6..02742da8f 100644
--- a/test/orm/test_joins.py
+++ b/test/orm/test_joins.py
@@ -702,11 +702,18 @@ class JoinTest(QueryTest, AssertsCompiledSQL):
         User = self.classes.User
 
         sess = create_session()
-        self.assert_compile(
+
+        subq = (
             sess.query(User)
             .filter(User.name == "ed")
-            .from_self()
-            .join(User.orders),
+            .apply_labels()
+            .subquery()
+        )
+
+        ua = aliased(User, subq)
+
+        self.assert_compile(
+            sess.query(ua).join(ua.orders),
             "SELECT anon_1.users_id AS anon_1_users_id, "
             "anon_1.users_name AS anon_1_users_name "
             "FROM (SELECT users.id AS users_id, users.name AS users_name "
@@ -1661,13 +1668,10 @@ class JoinTest(QueryTest, AssertsCompiledSQL):
         # explicit onclause with from_self(), means
         # the onclause must be aliased against the query's custom
         # FROM object
+        subq = sess.query(User).order_by(User.id).offset(2).subquery()
+        ua = aliased(User, subq)
         eq_(
-            sess.query(User)
-            .order_by(User.id)
-            .offset(2)
-            .from_self()
-            .join(Order, User.id == Order.user_id)
-            .all(),
+            sess.query(ua).join(Order, ua.id == Order.user_id).all(),
             [User(name="fred")],
         )
 
@@ -2303,34 +2307,6 @@ class JoinTest(QueryTest, AssertsCompiledSQL):
             use_default_dialect=True,
         )
 
-    def test_from_self_resets_joinpaths(self):
-        """test a join from from_self() doesn't confuse joins inside the subquery
-        with the outside.
-        """
-
-        Item, Keyword = self.classes.Item, self.classes.Keyword
-
-        sess = create_session()
-
-        self.assert_compile(
-            sess.query(Item)
-            .join(Item.keywords)
-            .from_self(Keyword)
-            .join(Item.keywords),
-            "SELECT keywords.id AS keywords_id, "
-            "keywords.name AS keywords_name "
-            "FROM (SELECT items.id AS items_id, "
-            "items.description AS items_description "
-            "FROM items JOIN item_keywords AS item_keywords_1 ON items.id = "
-            "item_keywords_1.item_id JOIN keywords "
-            "ON keywords.id = item_keywords_1.keyword_id) "
-            "AS anon_1 JOIN item_keywords AS item_keywords_2 ON "
-            "anon_1.items_id = item_keywords_2.item_id "
-            "JOIN keywords ON "
-            "keywords.id = item_keywords_2.keyword_id",
-            use_default_dialect=True,
-        )
-
 
 class JoinFromSelectableTest(fixtures.MappedTest, AssertsCompiledSQL):
     __dialect__ = "default"
@@ -3246,10 +3222,18 @@ class SelfReferentialTest(fixtures.MappedTest, AssertsCompiledSQL):
 
         # n1 is not inside the from_self(), so all cols must be maintained
         # on the outside
-        self.assert_compile(
+
+        subq = (
             sess.query(Node)
             .filter(Node.data == "n122")
-            .from_self(n1, Node.id),
+            .apply_labels()
+            .subquery()
+        )
+
+        na = aliased(Node, subq)
+
+        self.assert_compile(
+            sess.query(n1, na.id),
             "SELECT nodes_1.id AS nodes_1_id, "
             "nodes_1.parent_id AS nodes_1_parent_id, "
             "nodes_1.data AS nodes_1_data, anon_1.nodes_id AS anon_1_nodes_id "
@@ -3262,16 +3246,21 @@ class SelfReferentialTest(fixtures.MappedTest, AssertsCompiledSQL):
 
         parent = aliased(Node)
         grandparent = aliased(Node)
-        q = (
+        subq = (
             sess.query(Node, parent, grandparent)
             .join(parent, Node.parent)
             .join(grandparent, parent.parent)
             .filter(Node.data == "n122")
             .filter(parent.data == "n12")
             .filter(grandparent.data == "n1")
-            .from_self()
-            .limit(1)
+            .apply_labels()
+            .subquery()
         )
+        na = aliased(Node, subq)
+        pa = aliased(parent, subq)
+        ga = aliased(grandparent, subq)
+
+        q = sess.query(na, pa, ga).limit(1)
 
         # parent, grandparent *are* inside the from_self(), so they
         # should get aliased to the outside.
@@ -3517,15 +3506,23 @@ class SelfReferentialTest(fixtures.MappedTest, AssertsCompiledSQL):
 
         parent = aliased(Node)
         grandparent = aliased(Node)
-        eq_(
+
+        subq = (
             sess.query(Node, parent, grandparent)
             .join(parent, Node.parent)
             .join(grandparent, parent.parent)
             .filter(Node.data == "n122")
             .filter(parent.data == "n12")
             .filter(grandparent.data == "n1")
-            .from_self()
-            .first(),
+            .subquery()
+        )
+
+        na = aliased(Node, subq)
+        pa = aliased(parent, subq)
+        ga = aliased(grandparent, subq)
+
+        eq_(
+            sess.query(na, pa, ga).first(),
             (Node(data="n122"), Node(data="n12"), Node(data="n1")),
         )
 
@@ -3537,15 +3534,22 @@ class SelfReferentialTest(fixtures.MappedTest, AssertsCompiledSQL):
         parent = aliased(Node)
         grandparent = aliased(Node)
         # same, change order around
-        eq_(
+        subq = (
             sess.query(parent, grandparent, Node)
             .join(parent, Node.parent)
             .join(grandparent, parent.parent)
             .filter(Node.data == "n122")
             .filter(parent.data == "n12")
             .filter(grandparent.data == "n1")
-            .from_self()
-            .first(),
+            .subquery()
+        )
+
+        na = aliased(Node, subq)
+        pa = aliased(parent, subq)
+        ga = aliased(grandparent, subq)
+
+        eq_(
+            sess.query(pa, ga, na).first(),
             (Node(data="n12"), Node(data="n1"), Node(data="n122")),
         )
 
@@ -3575,16 +3579,23 @@ class SelfReferentialTest(fixtures.MappedTest, AssertsCompiledSQL):
 
         parent = aliased(Node)
         grandparent = aliased(Node)
-        eq_(
+
+        subq = (
             sess.query(Node, parent, grandparent)
             .join(parent, Node.parent)
             .join(grandparent, parent.parent)
             .filter(Node.data == "n122")
             .filter(parent.data == "n12")
             .filter(grandparent.data == "n1")
-            .from_self()
-            .options(joinedload(Node.children))
-            .first(),
+            .subquery()
+        )
+
+        na = aliased(Node, subq)
+        pa = aliased(parent, subq)
+        ga = aliased(grandparent, subq)
+
+        eq_(
+            sess.query(na, pa, ga).options(joinedload(na.children)).first(),
             (Node(data="n122"), Node(data="n12"), Node(data="n1")),
         )
 
diff --git a/test/orm/test_query.py b/test/orm/test_query.py
index a0171295b..4dbaa9166 100644
--- a/test/orm/test_query.py
+++ b/test/orm/test_query.py
@@ -983,8 +983,9 @@ class InvalidGenerationsTest(QueryTest, AssertsCompiledSQL):
 
         q.enable_assertions(False).select_from(users)
 
-        # this is fine, however
-        q.from_self()
+        with testing.expect_deprecated("The Query.from_self"):
+            # this is fine, however
+            q.from_self()
 
     def test_invalid_select_from(self):
         User = self.classes.User
@@ -3917,22 +3918,24 @@ class DistinctTest(QueryTest, AssertsCompiledSQL):
             "FROM users ORDER BY label DESC",
         )
 
-    def test_columns_augmented_roundtrip_one_from_self(self):
+    def test_columns_augmented_roundtrip_one_from_subq(self):
         """Test workaround for legacy style DISTINCT on extra column.
 
         See #5134
 
         """
         User, Address = self.classes.User, self.classes.Address
-
         sess = create_session()
-        q = (
+
+        subq = (
             sess.query(User, Address.email_address)
             .join("addresses")
             .distinct()
-            .from_self(User)
-            .order_by(desc(Address.email_address))
+            .subquery()
         )
+        ua = aliased(User, subq)
+        aa = aliased(Address, subq)
+        q = sess.query(ua).order_by(desc(aa.email_address))
 
         eq_([User(id=7), User(id=9), User(id=8)], q.all())
 
@@ -3999,7 +4002,7 @@ class DistinctTest(QueryTest, AssertsCompiledSQL):
 
         sess = create_session()
 
-        q = (
+        subq = (
             sess.query(
                 User.id,
                 User.name.label("foo"),
@@ -4010,8 +4013,13 @@ class DistinctTest(QueryTest, AssertsCompiledSQL):
             .filter(User.name == "jack")
             .filter(User.id + Address.user_id > 0)
             .distinct()
-            .from_self(User.id, User.name.label("foo"), Address.id)
-            .order_by(User.id, User.name, Address.email_address)
+            .subquery()
+        )
+
+        ua, aa = aliased(User, subq), aliased(Address, subq)
+
+        q = sess.query(ua.id, ua.name.label("foo"), aa.id).order_by(
+            ua.id, ua.name, aa.email_address
         )
 
         eq_(
@@ -4079,7 +4087,7 @@ class DistinctTest(QueryTest, AssertsCompiledSQL):
 
         sess = create_session()
 
-        q = (
+        subq = (
             sess.query(
                 User.id,
                 User.name.label("foo"),
@@ -4088,9 +4096,14 @@ class DistinctTest(QueryTest, AssertsCompiledSQL):
             )
             .distinct()
             .order_by(User.id, User.name, Address.email_address)
-            .from_self(User.id, User.name.label("foo"), Address.id)
+            .apply_labels()
+            .subquery()
         )
 
+        ua, aa = aliased(User, subq), aliased(Address, subq)
+
+        q = sess.query(ua.id, ua.name.label("foo"), aa.id)
+
         # Address.email_address is added because of DISTINCT,
         # however User.id, User.name are not b.c. they're already there,
         # even though User.name is labeled
@@ -4227,7 +4240,7 @@ class DistinctTest(QueryTest, AssertsCompiledSQL):
 
         sess = create_session()
 
-        q = (
+        subq = (
             sess.query(
                 User.id,
                 User.name.label("foo"),
@@ -4236,9 +4249,14 @@ class DistinctTest(QueryTest, AssertsCompiledSQL):
             )
             .distinct(Address.email_address)
             .order_by(User.id, User.name, Address.email_address)
-            .from_self(User.id, User.name.label("foo"), Address.id)
+            .apply_labels()
+            .subquery()
         )
 
+        ua = aliased(User, subq)
+        aa = aliased(Address, subq)
+        q = sess.query(ua.id, ua.name.label("foo"), aa.id)
+
         # Address.email_address is added because of DISTINCT,
         # however User.id, User.name are not b.c. they're already there,
         # even though User.name is labeled
diff --git a/test/orm/test_selectin_relations.py b/test/orm/test_selectin_relations.py
index e704e1624..5da1e1a4b 100644
--- a/test/orm/test_selectin_relations.py
+++ b/test/orm/test_selectin_relations.py
@@ -88,7 +88,7 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL):
 
         self.assert_sql_count(testing.db, go, 2)
 
-    def test_from_aliased(self):
+    def user_dingaling_fixture(self):
         users, Dingaling, User, dingalings, Address, addresses = (
             self.tables.users,
             self.classes.Dingaling,
@@ -113,60 +113,92 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL):
                 "addresses": relationship(Address, order_by=Address.id)
             },
         )
-        sess = create_session()
+        return User, Dingaling, Address
 
-        u = aliased(User)
+    def test_from_aliased_w_cache_one(self):
+        User, Dingaling, Address = self.user_dingaling_fixture()
 
-        q = sess.query(u).options(selectinload(u.addresses))
+        for i in range(3):
 
-        def go():
-            eq_(
-                [
-                    User(
-                        id=7,
-                        addresses=[
-                            Address(id=1, email_address="jack@bean.com")
-                        ],
-                    )
-                ],
-                q.filter(u.id == 7).all(),
-            )
+            def go():
 
-        self.assert_sql_count(testing.db, go, 2)
+                sess = create_session()
 
-        def go():
-            eq_(self.static.user_address_result, q.order_by(u.id).all())
+                u = aliased(User)
 
-        self.assert_sql_count(testing.db, go, 2)
+                q = sess.query(u).options(selectinload(u.addresses))
 
-        q = sess.query(u).options(
-            selectinload(u.addresses).selectinload(Address.dingalings)
-        )
+                eq_(
+                    [
+                        User(
+                            id=7,
+                            addresses=[
+                                Address(id=1, email_address="jack@bean.com")
+                            ],
+                        )
+                    ],
+                    q.filter(u.id == 7).all(),
+                )
 
-        def go():
-            eq_(
-                [
-                    User(
-                        id=8,
-                        addresses=[
-                            Address(
-                                id=2,
-                                email_address="ed@wood.com",
-                                dingalings=[Dingaling()],
-                            ),
-                            Address(id=3, email_address="ed@bettyboop.com"),
-                            Address(id=4, email_address="ed@lala.com"),
-                        ],
-                    ),
-                    User(
-                        id=9,
-                        addresses=[Address(id=5, dingalings=[Dingaling()])],
-                    ),
-                ],
-                q.filter(u.id.in_([8, 9])).all(),
-            )
+            self.assert_sql_count(testing.db, go, 2)
 
-        self.assert_sql_count(testing.db, go, 3)
+    def test_from_aliased_w_cache_two(self):
+        User, Dingaling, Address = self.user_dingaling_fixture()
+
+        for i in range(3):
+
+            def go():
+                sess = create_session()
+
+                u = aliased(User)
+
+                q = sess.query(u).options(selectinload(u.addresses))
+
+                eq_(self.static.user_address_result, q.order_by(u.id).all())
+
+            self.assert_sql_count(testing.db, go, 2)
+
+    def test_from_aliased_w_cache_three(self):
+
+        User, Dingaling, Address = self.user_dingaling_fixture()
+
+        for i in range(3):
+
+            def go():
+                sess = create_session()
+
+                u = aliased(User)
+
+                q = sess.query(u).options(
+                    selectinload(u.addresses).selectinload(Address.dingalings)
+                )
+                eq_(
+                    [
+                        User(
+                            id=8,
+                            addresses=[
+                                Address(
+                                    id=2,
+                                    email_address="ed@wood.com",
+                                    dingalings=[Dingaling()],
+                                ),
+                                Address(
+                                    id=3, email_address="ed@bettyboop.com"
+                                ),
+                                Address(id=4, email_address="ed@lala.com"),
+                            ],
+                        ),
+                        User(
+                            id=9,
+                            addresses=[
+                                Address(id=5, dingalings=[Dingaling()])
+                            ],
+                        ),
+                    ],
+                    q.filter(u.id.in_([8, 9])).all(),
+                )
+
+            self.assert_sql_count(testing.db, go, 3)
 
     def test_from_get(self):
         users, Address, addresses, User = (
diff --git a/test/orm/test_subquery_relations.py b/test/orm/test_subquery_relations.py
index 453092164..8ea79151b 100644
--- a/test/orm/test_subquery_relations.py
+++ b/test/orm/test_subquery_relations.py
@@ -124,7 +124,7 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL):
         eq_(len(u1.addresses), 1)
         eq_(len(u2.addresses), 3)
 
-    def test_from_aliased(self):
+    def user_dingaling_fixture(self):
         users, Dingaling, User, dingalings, Address, addresses = (
             self.tables.users,
             self.classes.Dingaling,
@@ -149,60 +149,87 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL):
                 "addresses": relationship(Address, order_by=Address.id)
             },
         )
-        sess = create_session()
+        return User, Dingaling, Address
 
-        u = aliased(User)
+    def test_from_aliased_w_cache_one(self):
+        User, Dingaling, Address = self.user_dingaling_fixture()
 
-        q = sess.query(u).options(subqueryload(u.addresses))
+        for i in range(3):
+            sess = create_session()
 
-        def go():
-            eq_(
-                [
-                    User(
-                        id=7,
-                        addresses=[
-                            Address(id=1, email_address="jack@bean.com")
-                        ],
-                    )
-                ],
-                q.filter(u.id == 7).all(),
-            )
+            u = aliased(User)
 
-        self.assert_sql_count(testing.db, go, 2)
+            q = sess.query(u).options(subqueryload(u.addresses))
 
-        def go():
-            eq_(self.static.user_address_result, q.order_by(u.id).all())
+            def go():
+                eq_(
+                    [
+                        User(
+                            id=7,
+                            addresses=[
+                                Address(id=1, email_address="jack@bean.com")
+                            ],
+                        )
+                    ],
+                    q.filter(u.id == 7).all(),
+                )
 
-        self.assert_sql_count(testing.db, go, 2)
+            self.assert_sql_count(testing.db, go, 2)
 
-        q = sess.query(u).options(
-            subqueryload(u.addresses).subqueryload(Address.dingalings)
-        )
+    def test_from_aliased_w_cache_two(self):
+        User, Dingaling, Address = self.user_dingaling_fixture()
 
-        def go():
-            eq_(
-                [
-                    User(
-                        id=8,
-                        addresses=[
-                            Address(
-                                id=2,
-                                email_address="ed@wood.com",
-                                dingalings=[Dingaling()],
-                            ),
-                            Address(id=3, email_address="ed@bettyboop.com"),
-                            Address(id=4, email_address="ed@lala.com"),
-                        ],
-                    ),
-                    User(
-                        id=9,
-                        addresses=[Address(id=5, dingalings=[Dingaling()])],
-                    ),
-                ],
-                q.filter(u.id.in_([8, 9])).all(),
+        for i in range(3):
+            sess = create_session()
+
+            u = aliased(User)
+
+            q = sess.query(u).options(subqueryload(u.addresses))
+
+            def go():
+                eq_(self.static.user_address_result, q.order_by(u.id).all())
+
+            self.assert_sql_count(testing.db, go, 2)
+
+    def test_from_aliased_w_cache_three(self):
+        User, Dingaling, Address = self.user_dingaling_fixture()
+
+        for i in range(3):
+            sess = create_session()
+
+            u = aliased(User)
+            q = sess.query(u).options(
+                subqueryload(u.addresses).subqueryload(Address.dingalings)
             )
 
-        self.assert_sql_count(testing.db, go, 3)
+            def go():
+                eq_(
+                    [
+                        User(
+                            id=8,
+                            addresses=[
+                                Address(
+                                    id=2,
+                                    email_address="ed@wood.com",
+                                    dingalings=[Dingaling()],
+                                ),
+                                Address(
+                                    id=3, email_address="ed@bettyboop.com"
+                                ),
+                                Address(id=4, email_address="ed@lala.com"),
+                            ],
+                        ),
+                        User(
+                            id=9,
+                            addresses=[
+                                Address(id=5, dingalings=[Dingaling()])
+                            ],
+                        ),
+                    ],
+                    q.filter(u.id.in_([8, 9])).all(),
+                )
+
+            self.assert_sql_count(testing.db, go, 3)
 
     def test_from_get(self):
         users, Address, addresses, User = (
@@ -3223,7 +3250,7 @@ class TestExistingRowPopulation(fixtures.DeclarativeMappedTest):
         is_true("c2_m2o" in a1.b.__dict__)
 
 
-class FromSelfTest(fixtures.DeclarativeMappedTest):
+class FromSubqTest(fixtures.DeclarativeMappedTest):
     """because subqueryloader relies upon the .subquery() method, this means
     if the original Query has a from_self() present, it needs to create
     .subquery() in terms of the Query class as a from_self() selectable
@@ -3295,15 +3322,25 @@ class FromSelfTest(fixtures.DeclarativeMappedTest):
         cache = {}
 
         for i in range(3):
-            q = (
+
+            subq = (
                 s.query(B)
-                .execution_options(compiled_cache=cache)
                 .join(B.a)
                 .filter(B.id < 4)
                 .filter(A.id > 1)
-                .from_self()
-                .options(subqueryload(B.a).subqueryload(A.cs))
-                .from_self()
+                .subquery()
+            )
+
+            bb = aliased(B, subq)
+
+            subq2 = s.query(bb).subquery()
+
+            bb2 = aliased(bb, subq2)
+
+            q = (
+                s.query(bb2)
+                .execution_options(compiled_cache=cache)
+                .options(subqueryload(bb2.a).subqueryload(A.cs))
             )
 
             def go():
@@ -3328,34 +3365,34 @@ class FromSelfTest(fixtures.DeclarativeMappedTest):
                 testing.db,
                 go,
                 CompiledSQL(
-                    "SELECT anon_1.anon_2_b_id AS anon_1_anon_2_b_id, "
-                    "anon_1.anon_2_b_a_id AS anon_1_anon_2_b_a_id FROM "
-                    "(SELECT anon_2.b_id AS anon_2_b_id, anon_2.b_a_id "
-                    "AS anon_2_b_a_id FROM (SELECT b.id AS b_id, b.a_id "
-                    "AS b_a_id FROM b JOIN a ON a.id = b.a_id "
+                    "SELECT anon_1.id AS anon_1_id, "
+                    "anon_1.a_id AS anon_1_a_id FROM "
+                    "(SELECT anon_2.id AS id, anon_2.a_id "
+                    "AS a_id FROM (SELECT b.id AS id, b.a_id "
+                    "AS a_id FROM b JOIN a ON a.id = b.a_id "
                     "WHERE b.id < :id_1 AND a.id > :id_2) AS anon_2) AS anon_1"
                 ),
                 CompiledSQL(
-                    "SELECT a.id AS a_id, anon_1.anon_2_anon_3_b_a_id AS "
-                    "anon_1_anon_2_anon_3_b_a_id FROM (SELECT DISTINCT "
-                    "anon_2.anon_3_b_a_id AS anon_2_anon_3_b_a_id FROM "
-                    "(SELECT anon_3.b_id AS anon_3_b_id, anon_3.b_a_id "
-                    "AS anon_3_b_a_id FROM (SELECT b.id AS b_id, b.a_id "
-                    "AS b_a_id FROM b JOIN a ON a.id = b.a_id "
+                    "SELECT a.id AS a_id, anon_1.anon_2_a_id AS "
+                    "anon_1_anon_2_a_id FROM (SELECT DISTINCT "
+                    "anon_2.a_id AS anon_2_a_id FROM "
+                    "(SELECT anon_3.id AS id, anon_3.a_id "
+                    "AS a_id FROM (SELECT b.id AS id, b.a_id "
+                    "AS a_id FROM b JOIN a ON a.id = b.a_id "
                     "WHERE b.id < :id_1 AND a.id > :id_2) AS anon_3) "
                     "AS anon_2) AS anon_1 JOIN a "
-                    "ON a.id = anon_1.anon_2_anon_3_b_a_id"
+                    "ON a.id = anon_1.anon_2_a_id"
                 ),
                 CompiledSQL(
                     "SELECT c.id AS c_id, c.a_id AS c_a_id, a_1.id "
-                    "AS a_1_id FROM (SELECT DISTINCT anon_2.anon_3_b_a_id AS "
-                    "anon_2_anon_3_b_a_id FROM "
-                    "(SELECT anon_3.b_id AS anon_3_b_id, anon_3.b_a_id "
-                    "AS anon_3_b_a_id FROM (SELECT b.id AS b_id, b.a_id "
-                    "AS b_a_id FROM b JOIN a ON a.id = b.a_id "
+                    "AS a_1_id FROM (SELECT DISTINCT anon_2.a_id AS "
+                    "anon_2_a_id FROM "
+                    "(SELECT anon_3.id AS id, anon_3.a_id "
+                    "AS a_id FROM (SELECT b.id AS id, b.a_id "
+                    "AS a_id FROM b JOIN a ON a.id = b.a_id "
                     "WHERE b.id < :id_1 AND a.id > :id_2) AS anon_3) "
                     "AS anon_2) AS anon_1 JOIN a AS a_1 ON a_1.id = "
-                    "anon_1.anon_2_anon_3_b_a_id JOIN c ON a_1.id = c.a_id "
+                    "anon_1.anon_2_a_id JOIN c ON a_1.id = c.a_id "
                     "ORDER BY c.id"
                 ),
             )
@@ -3372,13 +3409,16 @@ class FromSelfTest(fixtures.DeclarativeMappedTest):
         for i in range(3):
 
             def go():
+
+                subq = s.query(B).join(B.a).subquery()
+
+                bq = aliased(B, subq)
+
                 q = (
-                    s.query(B)
+                    s.query(bq)
                     .execution_options(compiled_cache=cache)
-                    .join(B.a)
-                    .from_self()
+                    .options(subqueryload(bq.ds))
                 )
-                q = q.options(subqueryload(B.ds))
 
                 q.all()
 
@@ -3386,17 +3426,17 @@ class FromSelfTest(fixtures.DeclarativeMappedTest):
                 testing.db,
                 go,
                 CompiledSQL(
-                    "SELECT anon_1.b_id AS anon_1_b_id, anon_1.b_a_id AS "
-                    "anon_1_b_a_id FROM (SELECT b.id AS b_id, b.a_id "
-                    "AS b_a_id FROM b JOIN a ON a.id = b.a_id) AS anon_1"
+                    "SELECT anon_1.id AS anon_1_id, anon_1.a_id AS "
+                    "anon_1_a_id FROM (SELECT b.id AS id, b.a_id "
+                    "AS a_id FROM b JOIN a ON a.id = b.a_id) AS anon_1"
                 ),
                 CompiledSQL(
                     "SELECT d.id AS d_id, d.b_id AS d_b_id, "
-                    "anon_1.anon_2_b_id AS anon_1_anon_2_b_id "
-                    "FROM (SELECT anon_2.b_id AS anon_2_b_id FROM "
-                    "(SELECT b.id AS b_id, b.a_id AS b_a_id FROM b "
+                    "anon_1.anon_2_id AS anon_1_anon_2_id "
+                    "FROM (SELECT anon_2.id AS anon_2_id FROM "
+                    "(SELECT b.id AS id, b.a_id AS a_id FROM b "
                     "JOIN a ON a.id = b.a_id) AS anon_2) AS anon_1 "
-                    "JOIN d ON anon_1.anon_2_b_id = d.b_id ORDER BY d.id"
+                    "JOIN d ON anon_1.anon_2_id = d.b_id ORDER BY d.id"
                 ),
             )
             s.close()
diff --git a/test/orm/test_update_delete.py b/test/orm/test_update_delete.py
index aec5d0553..ac290257d 100644
--- a/test/orm/test_update_delete.py
+++ b/test/orm/test_update_delete.py
@@ -131,10 +131,6 @@ class UpdateDeleteTest(fixtures.MappedTest):
                 s.query(User).select_from(Address),
                 r"join\(\), outerjoin\(\), select_from\(\), or from_self\(\)",
             ),
-            (
-                s.query(User).from_self(),
-                r"join\(\), outerjoin\(\), select_from\(\), or from_self\(\)",
-            ),
         ):
             assert_raises_message(
                 exc.InvalidRequestError,