Complete deprecation of from_self()

For most from_self() tests, move them into
test/orm/test_deprecated.py and replace the existing
test with one that uses aliased() plus a subquery.
This then revealed a few more issues.

Related items:

* Added slice() method to GenerativeSelect, to match that
  of orm.Query and to make possible migration of one of the
  from_self() tests.  moved the utility functions used for this
  from orm/util into sql/util.

* repairs a caching issue related to subqueryload
  where information being derived from the cached path info
  was mixing up with query information based on the per-query
  state, specifically an AliasedClass that is per query.

* for the above issue, it seemed like path_registry maybe
  had to change so that it represents AliasedClass objects
  as their cache key rather than on identity, but it wasn't
  needed.  still seems like it would be more correct.

* enhances the error message raised by coercions for a case
  such as when an AliasedClass holds onto a select() object
  and not a subquery(); will name the original and resolved
  object for clarity  (although how is AliasedClass able to
  accept a Select() object in the first place?)

* Added _set_propagate_attrs() to Query so that again if
  it's passed to AliasedClass, it doesn't raise an error
  during coercion, but again maybe that should also be
  rejected up front

Fixes: #5368
Change-Id: I5912aa611d899acc87a75eb5ee9f95990592f210

9 files changed, 175 insertions, 96 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
         #