Merge "Convert remaining ORM APIs to support 2.0 style"

18 files changed, 1230 insertions, 836 deletions

diff --git a/lib/sqlalchemy/engine/base.py b/lib/sqlalchemy/engine/base.py
index 6bc9588ad..2d672099b 100644
--- a/lib/sqlalchemy/engine/base.py
+++ b/lib/sqlalchemy/engine/base.py
@@ -759,6 +759,40 @@ class Connection(Connectable):
 
         return self._transaction is not None and self._transaction.is_active
 
+    def in_nested_transaction(self):
+        """Return True if a transaction is in progress."""
+        if self.__branch_from is not None:
+            return self.__branch_from.in_nested_transaction()
+
+        return (
+            self._nested_transaction is not None
+            and self._nested_transaction.is_active
+        )
+
+    def get_transaction(self):
+        """Return the current root transaction in progress, if any.
+
+        .. versionadded:: 1.4
+
+        """
+
+        if self.__branch_from is not None:
+            return self.__branch_from.get_transaction()
+
+        return self._transaction
+
+    def get_nested_transaction(self):
+        """Return the current nested transaction in progress, if any.
+
+        .. versionadded:: 1.4
+
+        """
+        if self.__branch_from is not None:
+
+            return self.__branch_from.get_nested_transaction()
+
+        return self._nested_transaction
+
     def _begin_impl(self, transaction):
         assert not self.__branch_from
 
diff --git a/lib/sqlalchemy/engine/result.py b/lib/sqlalchemy/engine/result.py
index 7df17cf22..db546380e 100644
--- a/lib/sqlalchemy/engine/result.py
+++ b/lib/sqlalchemy/engine/result.py
@@ -282,7 +282,8 @@ class Result(InPlaceGenerative):
        updated usage model and calling facade for SQLAlchemy Core and
        SQLAlchemy ORM.   In Core, it forms the basis of the
        :class:`.CursorResult` object which replaces the previous
-       :class:`.ResultProxy` interface.
+       :class:`.ResultProxy` interface.   When using the ORM, a higher level
+       object called :class:`.ChunkedIteratorResult` is normally used.
 
     """
 
diff --git a/lib/sqlalchemy/ext/baked.py b/lib/sqlalchemy/ext/baked.py
index fc6623609..e642a83d5 100644
--- a/lib/sqlalchemy/ext/baked.py
+++ b/lib/sqlalchemy/ext/baked.py
@@ -234,7 +234,7 @@ class BakedQuery(object):
         # used by the Connection, which in itself is more expensive to
         # generate than what BakedQuery was able to provide in 1.3 and prior
 
-        if statement.compile_options._bake_ok:
+        if statement._compile_options._bake_ok:
             self._bakery[self._effective_key(session)] = (
                 query,
                 statement,
diff --git a/lib/sqlalchemy/orm/__init__.py b/lib/sqlalchemy/orm/__init__.py
index fabb095a2..32ec60322 100644
--- a/lib/sqlalchemy/orm/__init__.py
+++ b/lib/sqlalchemy/orm/__init__.py
@@ -90,8 +90,13 @@ def create_session(bind=None, **kwargs):
     create_session().
 
     """
+
+    if kwargs.get("future", False):
+        kwargs.setdefault("autocommit", False)
+    else:
+        kwargs.setdefault("autocommit", True)
+
     kwargs.setdefault("autoflush", False)
-    kwargs.setdefault("autocommit", True)
     kwargs.setdefault("expire_on_commit", False)
     return Session(bind=bind, **kwargs)
 
@@ -267,12 +272,15 @@ contains_alias = public_factory(AliasOption, ".orm.contains_alias")
 
 def __go(lcls):
     global __all__
+    global AppenderQuery
     from .. import util as sa_util  # noqa
     from . import dynamic  # noqa
     from . import events  # noqa
     from . import loading  # noqa
     import inspect as _inspect
 
+    from .dynamic import AppenderQuery
+
     __all__ = sorted(
         name
         for name, obj in lcls.items()
diff --git a/lib/sqlalchemy/orm/context.py b/lib/sqlalchemy/orm/context.py
index d5f001db1..55a6b4cd2 100644
--- a/lib/sqlalchemy/orm/context.py
+++ b/lib/sqlalchemy/orm/context.py
@@ -188,7 +188,7 @@ class ORMCompileState(CompileState):
         raise NotImplementedError()
 
     @classmethod
-    def get_column_descriptions(self, statement):
+    def get_column_descriptions(cls, statement):
         return _column_descriptions(statement)
 
     @classmethod
@@ -204,8 +204,14 @@ class ORMCompileState(CompileState):
         if is_reentrant_invoke:
             return statement, execution_options
 
-        load_options = execution_options.get(
-            "_sa_orm_load_options", QueryContext.default_load_options
+        (
+            load_options,
+            execution_options,
+        ) = QueryContext.default_load_options.from_execution_options(
+            "_sa_orm_load_options",
+            {"populate_existing", "autoflush", "yield_per"},
+            execution_options,
+            statement._execution_options,
         )
 
         bind_arguments["clause"] = statement
@@ -246,6 +252,7 @@ class ORMCompileState(CompileState):
         load_options = execution_options.get(
             "_sa_orm_load_options", QueryContext.default_load_options
         )
+
         querycontext = QueryContext(
             compile_state,
             statement,
@@ -304,7 +311,7 @@ class ORMFromStatementCompileState(ORMCompileState):
         self._primary_entity = None
 
         self.use_legacy_query_style = (
-            statement_container.compile_options._use_legacy_query_style
+            statement_container._compile_options._use_legacy_query_style
         )
         self.statement_container = self.select_statement = statement_container
         self.requested_statement = statement = statement_container.element
@@ -315,9 +322,9 @@ class ORMFromStatementCompileState(ORMCompileState):
 
         _QueryEntity.to_compile_state(self, statement_container._raw_columns)
 
-        self.compile_options = statement_container.compile_options
+        self.compile_options = statement_container._compile_options
 
-        self.current_path = statement_container.compile_options._current_path
+        self.current_path = statement_container._compile_options._current_path
 
         if toplevel and statement_container._with_options:
             self.attributes = {"_unbound_load_dedupes": set()}
@@ -416,8 +423,8 @@ class ORMSelectCompileState(ORMCompileState, SelectState):
 
         # if we are a select() that was never a legacy Query, we won't
         # have ORM level compile options.
-        statement.compile_options = cls.default_compile_options.safe_merge(
-            statement.compile_options
+        statement._compile_options = cls.default_compile_options.safe_merge(
+            statement._compile_options
         )
 
         self = cls.__new__(cls)
@@ -434,20 +441,20 @@ class ORMSelectCompileState(ORMCompileState, SelectState):
         # indicates this select() came from Query.statement
         self.for_statement = (
             for_statement
-        ) = select_statement.compile_options._for_statement
+        ) = select_statement._compile_options._for_statement
 
         if not for_statement and not toplevel:
             # for subqueries, turn off eagerloads.
             # if "for_statement" mode is set, Query.subquery()
             # would have set this flag to False already if that's what's
             # desired
-            select_statement.compile_options += {
+            select_statement._compile_options += {
                 "_enable_eagerloads": False,
             }
 
         # generally if we are from Query or directly from a select()
         self.use_legacy_query_style = (
-            select_statement.compile_options._use_legacy_query_style
+            select_statement._compile_options._use_legacy_query_style
         )
 
         self._entities = []
@@ -457,15 +464,15 @@ class ORMSelectCompileState(ORMCompileState, SelectState):
         self._no_yield_pers = set()
 
         # legacy: only for query.with_polymorphic()
-        if select_statement.compile_options._with_polymorphic_adapt_map:
+        if select_statement._compile_options._with_polymorphic_adapt_map:
             self._with_polymorphic_adapt_map = dict(
-                select_statement.compile_options._with_polymorphic_adapt_map
+                select_statement._compile_options._with_polymorphic_adapt_map
             )
             self._setup_with_polymorphics()
 
         _QueryEntity.to_compile_state(self, select_statement._raw_columns)
 
-        self.compile_options = select_statement.compile_options
+        self.compile_options = select_statement._compile_options
 
         # determine label style.   we can make different decisions here.
         # at the moment, trying to see if we can always use DISAMBIGUATE_ONLY
@@ -479,7 +486,7 @@ class ORMSelectCompileState(ORMCompileState, SelectState):
         else:
             self.label_style = self.select_statement._label_style
 
-        self.current_path = select_statement.compile_options._current_path
+        self.current_path = select_statement._compile_options._current_path
 
         self.eager_order_by = ()
 
@@ -668,7 +675,7 @@ class ORMSelectCompileState(ORMCompileState, SelectState):
         self._polymorphic_adapters = {}
 
         compile_options = cls.default_compile_options.safe_merge(
-            query.compile_options
+            query._compile_options
         )
         # legacy: only for query.with_polymorphic()
         if compile_options._with_polymorphic_adapt_map:
@@ -711,6 +718,26 @@ class ORMSelectCompileState(ORMCompileState, SelectState):
                 for elem in _select_iterables([element]):
                     yield elem
 
+    @classmethod
+    @util.preload_module("sqlalchemy.orm.query")
+    def from_statement(cls, statement, from_statement):
+        query = util.preloaded.orm_query
+
+        from_statement = coercions.expect(
+            roles.SelectStatementRole,
+            from_statement,
+            apply_propagate_attrs=statement,
+        )
+
+        stmt = query.FromStatement(statement._raw_columns, from_statement)
+        stmt.__dict__.update(
+            _with_options=statement._with_options,
+            _with_context_options=statement._with_context_options,
+            _execution_options=statement._execution_options,
+            _propagate_attrs=statement._propagate_attrs,
+        )
+        return stmt
+
     def _setup_with_polymorphics(self):
         # legacy: only for query.with_polymorphic()
         for ext_info, wp in self._with_polymorphic_adapt_map.items():
diff --git a/lib/sqlalchemy/orm/dynamic.py b/lib/sqlalchemy/orm/dynamic.py
index d15127563..7832152a2 100644
--- a/lib/sqlalchemy/orm/dynamic.py
+++ b/lib/sqlalchemy/orm/dynamic.py
@@ -23,7 +23,12 @@ from . import util as orm_util
 from .query import Query
 from .. import exc
 from .. import log
+from .. import sql
 from .. import util
+from ..engine import result as _result
+from ..sql import selectable
+from ..sql.base import _generative
+from ..sql.base import Generative
 
 
 @log.class_logger
@@ -74,7 +79,6 @@ class DynamicAttributeImpl(attributes.AttributeImpl):
         dispatch,
         target_mapper,
         order_by,
-        query_class=None,
         **kw
     ):
         super(DynamicAttributeImpl, self).__init__(
@@ -82,12 +86,7 @@ class DynamicAttributeImpl(attributes.AttributeImpl):
         )
         self.target_mapper = target_mapper
         self.order_by = order_by
-        if not query_class:
-            self.query_class = AppenderQuery
-        elif AppenderMixin in query_class.mro():
-            self.query_class = query_class
-        else:
-            self.query_class = mixin_user_query(query_class)
+        self.query_class = AppenderQuery
 
     def get(self, state, dict_, passive=attributes.PASSIVE_OFF):
         if not passive & attributes.SQL_OK:
@@ -259,15 +258,26 @@ class DynamicAttributeImpl(attributes.AttributeImpl):
         self.remove(state, dict_, value, initiator, passive=passive)
 
 
-class AppenderMixin(object):
-    query_class = None
+class AppenderQuery(Generative):
+    """A dynamic query that supports basic collection storage operations."""
 
     def __init__(self, attr, state):
-        super(AppenderMixin, self).__init__(attr.target_mapper, None)
+
+        # this can be select() except for aliased=True flag on join()
+        # and corresponding behaviors on select().
+        self._is_core = False
+        self._statement = Query([attr.target_mapper], None)
+
+        # self._is_core = True
+        # self._statement = sql.select(attr.target_mapper)._set_label_style(
+        #    selectable.LABEL_STYLE_TABLENAME_PLUS_COL
+        # )
+
+        self._autoflush = True
         self.instance = instance = state.obj()
         self.attr = attr
 
-        mapper = object_mapper(instance)
+        self.mapper = mapper = object_mapper(instance)
         prop = mapper._props[self.attr.key]
 
         if prop.secondary is not None:
@@ -277,29 +287,154 @@ class AppenderMixin(object):
             # is in the FROM.  So we purposely put the mapper selectable
             # in _from_obj[0] to ensure a user-defined join() later on
             # doesn't fail, and secondary is then in _from_obj[1].
-            self._from_obj = (prop.mapper.selectable, prop.secondary)
+            self._statement = self._statement.select_from(
+                prop.mapper.selectable, prop.secondary
+            )
 
-        self._where_criteria += (
+        self._statement = self._statement.where(
             prop._with_parent(instance, alias_secondary=False),
         )
 
         if self.attr.order_by:
 
-            if (
-                self._order_by_clauses is False
-                or self._order_by_clauses is None
-            ):
-                self._order_by_clauses = tuple(self.attr.order_by)
-            else:
-                self._order_by_clauses = self._order_by_clauses + tuple(
-                    self.attr.order_by
-                )
+            self._statement = self._statement.order_by(*self.attr.order_by)
+
+    @_generative
+    def autoflush(self, setting):
+        """Set autoflush to a specific setting.
+
+        Note that a Session with autoflush=False will
+        not autoflush, even if this flag is set to True at the
+        Query level.  Therefore this flag is usually used only
+        to disable autoflush for a specific Query.
+
+        """
+        self._autoflush = setting
+
+    @property
+    def statement(self):
+        """Return the Core statement represented by this
+        :class:`.AppenderQuery`.
+
+        """
+        if self._is_core:
+            return self._statement._set_label_style(
+                selectable.LABEL_STYLE_DISAMBIGUATE_ONLY
+            )
+
+        else:
+            return self._statement.statement
+
+    def filter(self, *criteria):
+        """A synonym for the :meth:`_orm.AppenderQuery.where` method."""
+
+        return self.where(*criteria)
+
+    @_generative
+    def where(self, *criteria):
+        r"""Apply the given WHERE criterion, using SQL expressions.
+
+        Equivalent to :meth:`.Select.where`.
+
+        """
+        self._statement = self._statement.where(*criteria)
+
+    @_generative
+    def order_by(self, *criteria):
+        r"""Apply the given ORDER BY criterion, using SQL expressions.
+
+        Equivalent to :meth:`.Select.order_by`.
+
+        """
+        self._statement = self._statement.order_by(*criteria)
+
+    @_generative
+    def filter_by(self, **kwargs):
+        r"""Apply the given filtering criterion using keyword expressions.
+
+        Equivalent to :meth:`.Select.filter_by`.
+
+        """
+        self._statement = self._statement.filter_by(**kwargs)
+
+    @_generative
+    def join(self, target, *props, **kwargs):
+        r"""Create a SQL JOIN against this
+        object's criterion.
+
+        Equivalent to :meth:`.Select.join`.
+        """
+
+        self._statement = self._statement.join(target, *props, **kwargs)
+
+    @_generative
+    def outerjoin(self, target, *props, **kwargs):
+        r"""Create a SQL LEFT OUTER JOIN against this
+        object's criterion.
+
+        Equivalent to :meth:`.Select.outerjoin`.
+
+        """
+
+        self._statement = self._statement.outerjoin(target, *props, **kwargs)
+
+    def scalar(self):
+        """Return the first element of the first result or None
+        if no rows present.  If multiple rows are returned,
+        raises MultipleResultsFound.
+
+        Equivalent to :meth:`_query.Query.scalar`.
+
+        .. versionadded:: 1.1.6
+
+        """
+        return self._iter().scalar()
+
+    def first(self):
+        """Return the first row.
+
+        Equivalent to :meth:`_query.Query.first`.
+
+        """
+
+        # replicates limit(1) behavior
+        if self._statement is not None:
+            return self._iter().first()
+        else:
+            return self.limit(1)._iter().first()
+
+    def one(self):
+        """Return exactly one result or raise an exception.
+
+        Equivalent to :meth:`_query.Query.one`.
+
+        """
+        return self._iter().one()
+
+    def one_or_none(self):
+        """Return one or zero results, or raise an exception for multiple
+        rows.
+
+        Equivalent to :meth:`_query.Query.one_or_none`.
+
+        .. versionadded:: 1.0.9
+
+        """
+        return self._iter().one_or_none()
+
+    def all(self):
+        """Return all rows.
+
+        Equivalent to :meth:`_query.Query.all`.
+
+        """
+        return self._iter().all()
 
     def session(self):
         sess = object_session(self.instance)
         if (
             sess is not None
-            and self.autoflush
+            and self._autoflush
             and sess.autoflush
             and self.instance in sess
         ):
@@ -311,17 +446,60 @@ class AppenderMixin(object):
 
     session = property(session, lambda s, x: None)
 
-    def __iter__(self):
+    def _execute(self, sess=None):
+        # note we're returning an entirely new Query class instance
+        # here without any assignment capabilities; the class of this
+        # query is determined by the session.
+        instance = self.instance
+        if sess is None:
+            sess = object_session(instance)
+            if sess is None:
+                raise orm_exc.DetachedInstanceError(
+                    "Parent instance %s is not bound to a Session, and no "
+                    "contextual session is established; lazy load operation "
+                    "of attribute '%s' cannot proceed"
+                    % (orm_util.instance_str(instance), self.attr.key)
+                )
+
+        result = sess.execute(self._statement, future=True)
+        result = result.scalars()
+
+        if result._attributes.get("filtered", False):
+            result = result.unique()
+
+        return result
+
+    def _iter(self):
         sess = self.session
         if sess is None:
-            return iter(
-                self.attr._get_collection_history(
-                    attributes.instance_state(self.instance),
-                    attributes.PASSIVE_NO_INITIALIZE,
-                ).added_items
-            )
+            instance = self.instance
+            state = attributes.instance_state(instance)
+
+            if state.detached:
+                raise orm_exc.DetachedInstanceError(
+                    "Parent instance %s is not bound to a Session, and no "
+                    "contextual session is established; lazy load operation "
+                    "of attribute '%s' cannot proceed"
+                    % (orm_util.instance_str(instance), self.attr.key)
+                )
+            else:
+                iterator = (
+                    (item,)
+                    for item in self.attr._get_collection_history(
+                        state, attributes.PASSIVE_NO_INITIALIZE,
+                    ).added_items
+                )
+
+                row_metadata = _result.SimpleResultMetaData(
+                    (self.mapper.class_.__name__,), [], _unique_filters=[id],
+                )
+
+                return _result.IteratorResult(row_metadata, iterator).scalars()
         else:
-            return iter(self._generate(sess))
+            return self._execute(sess)
+
+    def __iter__(self):
+        return iter(self._iter())
 
     def __getitem__(self, index):
         sess = self.session
@@ -331,9 +509,32 @@ class AppenderMixin(object):
                 attributes.PASSIVE_NO_INITIALIZE,
             ).indexed(index)
         else:
-            return self._generate(sess).__getitem__(index)
+            return orm_util._getitem(self, index)
+
+    def slice(self, start, stop):
+        """Computes the "slice" represented by
+        the given indices and apply as LIMIT/OFFSET.
+
+
+        """
+        limit_clause, offset_clause = orm_util._make_slice(
+            self._statement._limit_clause,
+            self._statement._offset_clause,
+            start,
+            stop,
+        )
+        self._statement = self._statement.limit(limit_clause).offset(
+            offset_clause
+        )
 
     def count(self):
+        """return the 'count'.
+
+        Equivalent to :meth:`_query.Query.count`.
+
+
+        """
+
         sess = self.session
         if sess is None:
             return len(
@@ -343,33 +544,10 @@ class AppenderMixin(object):
                 ).added_items
             )
         else:
-            return self._generate(sess).count()
-
-    def _generate(self, sess=None):
-        # note we're returning an entirely new Query class instance
-        # here without any assignment capabilities; the class of this
-        # query is determined by the session.
-        instance = self.instance
-        if sess is None:
-            sess = object_session(instance)
-            if sess is None:
-                raise orm_exc.DetachedInstanceError(
-                    "Parent instance %s is not bound to a Session, and no "
-                    "contextual session is established; lazy load operation "
-                    "of attribute '%s' cannot proceed"
-                    % (orm_util.instance_str(instance), self.attr.key)
-                )
-
-        if self.query_class:
-            query = self.query_class(self.attr.target_mapper, session=sess)
-        else:
-            query = sess.query(self.attr.target_mapper)
-
-        query._where_criteria = self._where_criteria
-        query._from_obj = self._from_obj
-        query._order_by_clauses = self._order_by_clauses
+            col = sql.func.count(sql.literal_column("*"))
 
-        return query
+            stmt = sql.select(col).select_from(self._statement.subquery())
+            return self.session.execute(stmt).scalar()
 
     def extend(self, iterator):
         for item in iterator:
@@ -397,16 +575,6 @@ class AppenderMixin(object):
         )
 
 
-class AppenderQuery(AppenderMixin, Query):
-    """A dynamic query that supports basic collection storage operations."""
-
-
-def mixin_user_query(cls):
-    """Return a new class with AppenderQuery functionality layered over."""
-    name = "Appender" + cls.__name__
-    return type(name, (AppenderMixin, cls), {"query_class": cls})
-
-
 class CollectionHistory(object):
     """Overrides AttributeHistory to receive append/remove events directly."""
 
diff --git a/lib/sqlalchemy/orm/loading.py b/lib/sqlalchemy/orm/loading.py
index 55c2b79f5..8d1ae2e69 100644
--- a/lib/sqlalchemy/orm/loading.py
+++ b/lib/sqlalchemy/orm/loading.py
@@ -346,7 +346,7 @@ def load_on_pk_identity(
         load_options = QueryContext.default_load_options
 
     compile_options = ORMCompileState.default_compile_options.safe_merge(
-        q.compile_options
+        q._compile_options
     )
 
     if primary_key_identity is not None:
@@ -411,7 +411,7 @@ def load_on_pk_identity(
     # TODO: most of the compile_options that are not legacy only involve this
     # function, so try to see if handling of them can mostly be local to here
 
-    q.compile_options, load_options = _set_get_options(
+    q._compile_options, load_options = _set_get_options(
         compile_options,
         load_options,
         populate_existing=bool(refresh_state),
diff --git a/lib/sqlalchemy/orm/persistence.py b/lib/sqlalchemy/orm/persistence.py
index 1b2779c00..a78af92b9 100644
--- a/lib/sqlalchemy/orm/persistence.py
+++ b/lib/sqlalchemy/orm/persistence.py
@@ -1762,24 +1762,23 @@ class BulkUDCompileState(CompileState):
         if is_reentrant_invoke:
             return statement, execution_options
 
-        sync = execution_options.get("synchronize_session", None)
-        if sync is None:
-            sync = statement._execution_options.get(
-                "synchronize_session", None
-            )
-
-        update_options = execution_options.get(
+        (
+            update_options,
+            execution_options,
+        ) = BulkUDCompileState.default_update_options.from_execution_options(
             "_sa_orm_update_options",
-            BulkUDCompileState.default_update_options,
+            {"synchronize_session"},
+            execution_options,
+            statement._execution_options,
         )
 
+        sync = update_options._synchronize_session
         if sync is not None:
             if sync not in ("evaluate", "fetch", False):
                 raise sa_exc.ArgumentError(
                     "Valid strategies for session synchronization "
                     "are 'evaluate', 'fetch', False"
                 )
-            update_options += {"_synchronize_session": sync}
 
         bind_arguments["clause"] = statement
         try:
diff --git a/lib/sqlalchemy/orm/query.py b/lib/sqlalchemy/orm/query.py
index acc76094b..7bf69f99f 100644
--- a/lib/sqlalchemy/orm/query.py
+++ b/lib/sqlalchemy/orm/query.py
@@ -22,10 +22,10 @@ import itertools
 import operator
 import types
 
-from . import attributes
 from . import exc as orm_exc
 from . import interfaces
 from . import loading
+from . import util as orm_util
 from .base import _assertions
 from .context import _column_descriptions
 from .context import _legacy_determine_last_joined_entity
@@ -121,7 +121,7 @@ class Query(
     _legacy_setup_joins = ()
     _label_style = LABEL_STYLE_NONE
 
-    compile_options = ORMCompileState.default_compile_options
+    _compile_options = ORMCompileState.default_compile_options
 
     load_options = QueryContext.default_load_options
 
@@ -215,7 +215,7 @@ class Query(
             for elem in obj
         ]
 
-        self.compile_options += {"_set_base_alias": set_base_alias}
+        self._compile_options += {"_set_base_alias": set_base_alias}
         self._from_obj = tuple(fa)
 
     @_generative
@@ -254,7 +254,7 @@ class Query(
 
         self._from_obj = self._legacy_setup_joins = ()
         if self._statement is not None:
-            self.compile_options += {"_statement": None}
+            self._compile_options += {"_statement": None}
         self._where_criteria = ()
         self._distinct = False
 
@@ -320,7 +320,7 @@ class Query(
         if load_options:
             self.load_options += load_options
         if compile_options:
-            self.compile_options += compile_options
+            self._compile_options += compile_options
 
         return self
 
@@ -357,8 +357,8 @@ class Query(
         # passed into the execute process and wont generate its own cache
         # key; this will all occur in terms of the ORM-enabled Select.
         if (
-            not self.compile_options._set_base_alias
-            and not self.compile_options._with_polymorphic_adapt_map
+            not self._compile_options._set_base_alias
+            and not self._compile_options._with_polymorphic_adapt_map
         ):
             # if we don't have legacy top level aliasing features in use
             # then convert to a future select() directly
@@ -400,9 +400,9 @@ class Query(
                 if new_query is not None and new_query is not self:
                     self = new_query
                     if not fn._bake_ok:
-                        self.compile_options += {"_bake_ok": False}
+                        self._compile_options += {"_bake_ok": False}
 
-        compile_options = self.compile_options
+        compile_options = self._compile_options
         compile_options += {
             "_for_statement": for_statement,
             "_use_legacy_query_style": use_legacy_query_style,
@@ -413,21 +413,21 @@ class Query(
             stmt.__dict__.update(
                 _with_options=self._with_options,
                 _with_context_options=self._with_context_options,
-                compile_options=compile_options,
+                _compile_options=compile_options,
                 _execution_options=self._execution_options,
+                _propagate_attrs=self._propagate_attrs,
             )
-            stmt._propagate_attrs = self._propagate_attrs
         else:
             # Query / select() internal attributes are 99% cross-compatible
             stmt = Select.__new__(Select)
             stmt.__dict__.update(self.__dict__)
             stmt.__dict__.update(
                 _label_style=self._label_style,
-                compile_options=compile_options,
+                _compile_options=compile_options,
+                _propagate_attrs=self._propagate_attrs,
             )
             stmt.__dict__.pop("session", None)
 
-        stmt._propagate_attrs = self._propagate_attrs
         return stmt
 
     def subquery(
@@ -629,7 +629,7 @@ class Query(
         selectable, or when using :meth:`_query.Query.yield_per`.
 
         """
-        self.compile_options += {"_enable_eagerloads": value}
+        self._compile_options += {"_enable_eagerloads": value}
 
     @_generative
     def with_labels(self):
@@ -710,7 +710,7 @@ class Query(
         query intended for the deferred load.
 
         """
-        self.compile_options += {"_current_path": path}
+        self._compile_options += {"_current_path": path}
 
     # TODO: removed in 2.0
     @_generative
@@ -744,7 +744,7 @@ class Query(
             polymorphic_on=polymorphic_on,
         )
 
-        self.compile_options = self.compile_options.add_to_element(
+        self._compile_options = self._compile_options.add_to_element(
             "_with_polymorphic_adapt_map", ((entity, inspect(wp)),)
         )
 
@@ -818,6 +818,10 @@ class Query(
             {"stream_results": True, "max_row_buffer": count}
         )
 
+    @util.deprecated_20(
+        ":meth:`_orm.Query.get`",
+        alternative="The method is now available as :meth:`_orm.Session.get`",
+    )
     def get(self, ident):
         """Return an instance based on the given primary key identifier,
         or ``None`` if not found.
@@ -858,14 +862,6 @@ class Query(
         however, and will be used if the object is not
         yet locally present.
 
-        A lazy-loading, many-to-one attribute configured
-        by :func:`_orm.relationship`, using a simple
-        foreign-key-to-primary-key criterion, will also use an
-        operation equivalent to :meth:`_query.Query.get` in order to retrieve
-        the target value from the local identity map
-        before querying the database.  See :doc:`/orm/loading_relationships`
-        for further details on relationship loading.
-
         :param ident: A scalar, tuple, or dictionary representing the
          primary key.  For a composite (e.g. multiple column) primary key,
          a tuple or dictionary should be passed.
@@ -905,80 +901,22 @@ class Query(
 
         """
         self._no_criterion_assertion("get", order_by=False, distinct=False)
+
+        # we still implement _get_impl() so that baked query can override
+        # it
         return self._get_impl(ident, loading.load_on_pk_identity)
 
     def _get_impl(self, primary_key_identity, db_load_fn, identity_token=None):
-
-        # convert composite types to individual args
-        if hasattr(primary_key_identity, "__composite_values__"):
-            primary_key_identity = primary_key_identity.__composite_values__()
-
         mapper = self._only_full_mapper_zero("get")
-
-        is_dict = isinstance(primary_key_identity, dict)
-        if not is_dict:
-            primary_key_identity = util.to_list(
-                primary_key_identity, default=(None,)
-            )
-
-        if len(primary_key_identity) != len(mapper.primary_key):
-            raise sa_exc.InvalidRequestError(
-                "Incorrect number of values in identifier to formulate "
-                "primary key for query.get(); primary key columns are %s"
-                % ",".join("'%s'" % c for c in mapper.primary_key)
-            )
-
-        if is_dict:
-            try:
-                primary_key_identity = list(
-                    primary_key_identity[prop.key]
-                    for prop in mapper._identity_key_props
-                )
-
-            except KeyError as err:
-                util.raise_(
-                    sa_exc.InvalidRequestError(
-                        "Incorrect names of values in identifier to formulate "
-                        "primary key for query.get(); primary key attribute "
-                        "names are %s"
-                        % ",".join(
-                            "'%s'" % prop.key
-                            for prop in mapper._identity_key_props
-                        )
-                    ),
-                    replace_context=err,
-                )
-
-        if (
-            not self.load_options._populate_existing
-            and not mapper.always_refresh
-            and self._for_update_arg is None
-        ):
-
-            instance = self.session._identity_lookup(
-                mapper, primary_key_identity, identity_token=identity_token
-            )
-
-            if instance is not None:
-                self._get_existing_condition()
-                # reject calls for id in identity map but class
-                # mismatch.
-                if not issubclass(instance.__class__, mapper.class_):
-                    return None
-                return instance
-            elif instance is attributes.PASSIVE_CLASS_MISMATCH:
-                return None
-
-        # apply_labels() not strictly necessary, however this will ensure that
-        # tablename_colname style is used which at the moment is asserted
-        # in a lot of unit tests :)
-
-        statement = self._statement_20().apply_labels()
-        return db_load_fn(
-            self.session,
-            statement,
+        return self.session._get_impl(
+            mapper,
             primary_key_identity,
-            load_options=self.load_options,
+            db_load_fn,
+            populate_existing=self.load_options._populate_existing,
+            with_for_update=self._for_update_arg,
+            options=self._with_options,
+            identity_token=identity_token,
+            execution_options=self._execution_options,
         )
 
     @property
@@ -1000,7 +938,7 @@ class Query(
 
     @property
     def _current_path(self):
-        return self.compile_options._current_path
+        return self._compile_options._current_path
 
     @_generative
     def correlate(self, *fromclauses):
@@ -1375,7 +1313,7 @@ class Query(
 
     @_generative
     def _set_enable_single_crit(self, val):
-        self.compile_options += {"_enable_single_crit": val}
+        self._compile_options += {"_enable_single_crit": val}
 
     @_generative
     def _from_selectable(self, fromclause, set_entity_from=True):
@@ -1394,7 +1332,7 @@ class Query(
         ):
             self.__dict__.pop(attr, None)
         self._set_select_from([fromclause], set_entity_from)
-        self.compile_options += {
+        self._compile_options += {
             "_enable_single_crit": False,
             "_statement": None,
         }
@@ -1404,7 +1342,7 @@ class Query(
         # legacy.  see test/orm/test_froms.py for various
         # "oldstyle" tests that rely on this and the correspoinding
         # "newtyle" that do not.
-        self.compile_options += {"_orm_only_from_obj_alias": False}
+        self._compile_options += {"_orm_only_from_obj_alias": False}
 
     @util.deprecated(
         "1.4",
@@ -1517,7 +1455,7 @@ class Query(
         """
 
         opts = tuple(util.flatten_iterator(args))
-        if self.compile_options._current_path:
+        if self._compile_options._current_path:
             for opt in opts:
                 if opt._is_legacy_option:
                     opt.process_query_conditionally(self)
@@ -1641,6 +1579,14 @@ class Query(
         params = self.load_options._params.union(kwargs)
         self.load_options += {"_params": params}
 
+    def where(self, *criterion):
+        """A synonym for :meth:`.Query.filter`.
+
+        .. versionadded:: 1.4
+
+        """
+        return self.filter(*criterion)
+
     @_generative
     @_assertions(_no_statement_condition, _no_limit_offset)
     def filter(self, *criterion):
@@ -2204,6 +2150,7 @@ class Query(
             SQLAlchemy versions was the primary ORM-level joining interface.
 
         """
+
         aliased, from_joinpoint, isouter, full = (
             kwargs.pop("aliased", False),
             kwargs.pop("from_joinpoint", False),
@@ -2496,36 +2443,10 @@ class Query(
         """
 
         self._set_select_from([from_obj], True)
-        self.compile_options += {"_enable_single_crit": False}
+        self._compile_options += {"_enable_single_crit": False}
 
     def __getitem__(self, item):
-        if isinstance(item, slice):
-            start, stop, step = util.decode_slice(item)
-
-            if (
-                isinstance(stop, int)
-                and isinstance(start, int)
-                and stop - start <= 0
-            ):
-                return []
-
-            # perhaps we should execute a count() here so that we
-            # can still use LIMIT/OFFSET ?
-            elif (isinstance(start, int) and start < 0) or (
-                isinstance(stop, int) and stop < 0
-            ):
-                return list(self)[item]
-
-            res = self.slice(start, stop)
-            if step is not None:
-                return list(res)[None : None : item.step]
-            else:
-                return list(res)
-        else:
-            if item == -1:
-                return list(self)[-1]
-            else:
-                return list(self[item : item + 1])[0]
+        return orm_util._getitem(self, item)
 
     @_generative
     @_assertions(_no_statement_condition)
@@ -2559,46 +2480,10 @@ class Query(
            :meth:`_query.Query.offset`
 
         """
-        # for calculated limit/offset, try to do the addition of
-        # values to offset in Python, howver 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 = self._offset_or_limit_clause_asint_if_possible(
-                self._offset_clause
-            )
-            if offset_clause is None:
-                offset_clause = 0
 
-            if start != 0:
-                offset_clause = offset_clause + start
-
-            if offset_clause == 0:
-                self._offset_clause = None
-            else:
-                self._offset_clause = self._offset_or_limit_clause(
-                    offset_clause
-                )
-
-            self._limit_clause = self._offset_or_limit_clause(stop - start)
-
-        elif start is None and stop is not None:
-            self._limit_clause = self._offset_or_limit_clause(stop)
-        elif start is not None and stop is None:
-            offset_clause = self._offset_or_limit_clause_asint_if_possible(
-                self._offset_clause
-            )
-            if offset_clause is None:
-                offset_clause = 0
-
-            if start != 0:
-                offset_clause = offset_clause + start
-
-            if offset_clause == 0:
-                self._offset_clause = None
-            else:
-                self._offset_clause = self._offset_or_limit_clause(
-                    offset_clause
-                )
+        self._limit_clause, self._offset_clause = orm_util._make_slice(
+            self._limit_clause, self._offset_clause, start, stop
+        )
 
     @_generative
     @_assertions(_no_statement_condition)
@@ -2607,7 +2492,7 @@ class Query(
         ``Query``.
 
         """
-        self._limit_clause = self._offset_or_limit_clause(limit)
+        self._limit_clause = orm_util._offset_or_limit_clause(limit)
 
     @_generative
     @_assertions(_no_statement_condition)
@@ -2616,31 +2501,7 @@ class Query(
         ``Query``.
 
         """
-        self._offset_clause = self._offset_or_limit_clause(offset)
-
-    def _offset_or_limit_clause(self, 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(self, 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
+        self._offset_clause = orm_util._offset_or_limit_clause(offset)
 
     @_generative
     @_assertions(_no_statement_condition)
@@ -2723,7 +2584,7 @@ class Query(
             roles.SelectStatementRole, statement, apply_propagate_attrs=self
         )
         self._statement = statement
-        self.compile_options += {"_statement": statement}
+        self._compile_options += {"_statement": statement}
 
     def first(self):
         """Return the first result of this ``Query`` or
@@ -3088,110 +2949,22 @@ class Query(
             sess.query(User).filter(User.age == 25).\
                 delete(synchronize_session='evaluate')
 
-        .. warning:: The :meth:`_query.Query.delete`
-           method is a "bulk" operation,
-           which bypasses ORM unit-of-work automation in favor of greater
-           performance.  **Please read all caveats and warnings below.**
-
-        :param synchronize_session: chooses the strategy for the removal of
-            matched objects from the session. Valid values are:
-
-            ``False`` - don't synchronize the session. This option is the most
-            efficient and is reliable once the session is expired, which
-            typically occurs after a commit(), or explicitly using
-            expire_all(). Before the expiration, objects may still remain in
-            the session which were in fact deleted which can lead to confusing
-            results if they are accessed via get() or already loaded
-            collections.
-
-            ``'fetch'`` - performs a select query before the delete to find
-            objects that are matched by the delete query and need to be
-            removed from the session. Matched objects are removed from the
-            session.
+        .. warning::
 
-            ``'evaluate'`` - Evaluate the query's criteria in Python straight
-            on the objects in the session. If evaluation of the criteria isn't
-            implemented, an error is raised.
+            See the section :ref:`bulk_update_delete` for important caveats
+            and warnings, including limitations when using bulk UPDATE
+            and DELETE with mapper inheritance configurations.
 
-            The expression evaluator currently doesn't account for differing
-            string collations between the database and Python.
+        :param synchronize_session: chooses the strategy to update the
+         attributes on objects in the session.   See the section
+         :ref:`bulk_update_delete` for a discussion of these strategies.
 
         :return: the count of rows matched as returned by the database's
           "row count" feature.
 
-        .. warning:: **Additional Caveats for bulk query deletes**
-
-            * This method does **not work for joined
-              inheritance mappings**, since the **multiple table
-              deletes are not supported by SQL** as well as that the
-              **join condition of an inheritance mapper is not
-              automatically rendered**.  Care must be taken in any
-              multiple-table delete to first accommodate via some other means
-              how the related table will be deleted, as well as to
-              explicitly include the joining
-              condition between those tables, even in mappings where
-              this is normally automatic. E.g. if a class ``Engineer``
-              subclasses ``Employee``, a DELETE against the ``Employee``
-              table would look like::
-
-                    session.query(Engineer).\
-                        filter(Engineer.id == Employee.id).\
-                        filter(Employee.name == 'dilbert').\
-                        delete()
-
-              However the above SQL will not delete from the Engineer table,
-              unless an ON DELETE CASCADE rule is established in the database
-              to handle it.
-
-              Short story, **do not use this method for joined inheritance
-              mappings unless you have taken the additional steps to make
-              this feasible**.
-
-            * The polymorphic identity WHERE criteria is **not** included
-              for single- or
-              joined- table updates - this must be added **manually** even
-              for single table inheritance.
-
-            * The method does **not** offer in-Python cascading of
-              relationships - it is assumed that ON DELETE CASCADE/SET
-              NULL/etc. is configured for any foreign key references
-              which require it, otherwise the database may emit an
-              integrity violation if foreign key references are being
-              enforced.
-
-              After the DELETE, dependent objects in the
-              :class:`.Session` which were impacted by an ON DELETE
-              may not contain the current state, or may have been
-              deleted. This issue is resolved once the
-              :class:`.Session` is expired, which normally occurs upon
-              :meth:`.Session.commit` or can be forced by using
-              :meth:`.Session.expire_all`.  Accessing an expired
-              object whose row has been deleted will invoke a SELECT
-              to locate the row; when the row is not found, an
-              :class:`~sqlalchemy.orm.exc.ObjectDeletedError` is
-              raised.
-
-            * The ``'fetch'`` strategy results in an additional
-              SELECT statement emitted and will significantly reduce
-              performance.
-
-            * The ``'evaluate'`` strategy performs a scan of
-              all matching objects within the :class:`.Session`; if the
-              contents of the :class:`.Session` are expired, such as
-              via a proceeding :meth:`.Session.commit` call, **this will
-              result in SELECT queries emitted for every matching object**.
-
-            * The :meth:`.MapperEvents.before_delete` and
-              :meth:`.MapperEvents.after_delete`
-              events **are not invoked** from this method.  Instead, the
-              :meth:`.SessionEvents.after_bulk_delete` method is provided to
-              act upon a mass DELETE of entity rows.
-
         .. seealso::
 
-            :meth:`_query.Query.update`
-
-            :ref:`inserts_and_updates` - Core SQL tutorial
+            :ref:`bulk_update_delete`
 
         """
 
@@ -3231,12 +3004,11 @@ class Query(
             sess.query(User).filter(User.age == 25).\
                 update({"age": User.age - 10}, synchronize_session='evaluate')
 
+        .. warning::
 
-        .. warning:: The :meth:`_query.Query.update`
-           method is a "bulk" operation,
-           which bypasses ORM unit-of-work automation in favor of greater
-           performance.  **Please read all caveats and warnings below.**
-
+            See the section :ref:`bulk_update_delete` for important caveats
+            and warnings, including limitations when using bulk UPDATE
+            and DELETE with mapper inheritance configurations.
 
         :param values: a dictionary with attributes names, or alternatively
          mapped attributes or SQL expressions, as keys, and literal
@@ -3248,31 +3020,9 @@ class Query(
          flag is passed to the :paramref:`.Query.update.update_args` dictionary
          as well.
 
-          .. versionchanged:: 1.0.0 - string names in the values dictionary
-             are now resolved against the mapped entity; previously, these
-             strings were passed as literal column names with no mapper-level
-             translation.
-
         :param synchronize_session: chooses the strategy to update the
-         attributes on objects in the session. Valid values are:
-
-            ``False`` - don't synchronize the session. This option is the most
-            efficient and is reliable once the session is expired, which
-            typically occurs after a commit(), or explicitly using
-            expire_all(). Before the expiration, updated objects may still
-            remain in the session with stale values on their attributes, which
-            can lead to confusing results.
-
-            ``'fetch'`` - performs a select query before the update to find
-            objects that are matched by the update query. The updated
-            attributes are expired on matched objects.
-
-            ``'evaluate'`` - Evaluate the Query's criteria in Python straight
-            on the objects in the session. If evaluation of the criteria isn't
-            implemented, an exception is raised.
-
-            The expression evaluator currently doesn't account for differing
-            string collations between the database and Python.
+         attributes on objects in the session.   See the section
+         :ref:`bulk_update_delete` for a discussion of these strategies.
 
         :param update_args: Optional dictionary, if present will be passed
          to the underlying :func:`_expression.update`
@@ -3281,70 +3031,14 @@ class Query(
          as ``mysql_limit``, as well as other special arguments such as
          :paramref:`~sqlalchemy.sql.expression.update.preserve_parameter_order`.
 
-         .. versionadded:: 1.0.0
-
         :return: the count of rows matched as returned by the database's
          "row count" feature.
 
-        .. warning:: **Additional Caveats for bulk query updates**
-
-            * The method does **not** offer in-Python cascading of
-              relationships - it is assumed that ON UPDATE CASCADE is
-              configured for any foreign key references which require
-              it, otherwise the database may emit an integrity
-              violation if foreign key references are being enforced.
-
-              After the UPDATE, dependent objects in the
-              :class:`.Session` which were impacted by an ON UPDATE
-              CASCADE may not contain the current state; this issue is
-              resolved once the :class:`.Session` is expired, which
-              normally occurs upon :meth:`.Session.commit` or can be
-              forced by using :meth:`.Session.expire_all`.
-
-            * The ``'fetch'`` strategy results in an additional
-              SELECT statement emitted and will significantly reduce
-              performance.
-
-            * The ``'evaluate'`` strategy performs a scan of
-              all matching objects within the :class:`.Session`; if the
-              contents of the :class:`.Session` are expired, such as
-              via a proceeding :meth:`.Session.commit` call, **this will
-              result in SELECT queries emitted for every matching object**.
-
-            * The method supports multiple table updates, as detailed
-              in :ref:`multi_table_updates`, and this behavior does
-              extend to support updates of joined-inheritance and
-              other multiple table mappings.  However, the **join
-              condition of an inheritance mapper is not
-              automatically rendered**. Care must be taken in any
-              multiple-table update to explicitly include the joining
-              condition between those tables, even in mappings where
-              this is normally automatic. E.g. if a class ``Engineer``
-              subclasses ``Employee``, an UPDATE of the ``Engineer``
-              local table using criteria against the ``Employee``
-              local table might look like::
-
-                    session.query(Engineer).\
-                        filter(Engineer.id == Employee.id).\
-                        filter(Employee.name == 'dilbert').\
-                        update({"engineer_type": "programmer"})
-
-            * The polymorphic identity WHERE criteria is **not** included
-              for single- or
-              joined- table updates - this must be added **manually**, even
-              for single table inheritance.
-
-            * The :meth:`.MapperEvents.before_update` and
-              :meth:`.MapperEvents.after_update`
-              events **are not invoked from this method**.  Instead, the
-              :meth:`.SessionEvents.after_bulk_update` method is provided to
-              act upon a mass UPDATE of entity rows.
 
         .. seealso::
 
-            :meth:`_query.Query.delete`
+            :ref:`bulk_update_delete`
 
-            :ref:`inserts_and_updates` - Core SQL tutorial
 
         """
 
@@ -3390,7 +3084,7 @@ class Query(
         """
 
         stmt = self._statement_20(for_statement=for_statement, **kw)
-        assert for_statement == stmt.compile_options._for_statement
+        assert for_statement == stmt._compile_options._for_statement
 
         # this chooses between ORMFromStatementCompileState and
         # ORMSelectCompileState.  We could also base this on
@@ -3422,7 +3116,7 @@ class FromStatement(SelectStatementGrouping, Executable):
 
     __visit_name__ = "orm_from_statement"
 
-    compile_options = ORMFromStatementCompileState.default_compile_options
+    _compile_options = ORMFromStatementCompileState.default_compile_options
 
     _compile_state_factory = ORMFromStatementCompileState.create_for_statement
 
diff --git a/lib/sqlalchemy/orm/session.py b/lib/sqlalchemy/orm/session.py
index f4f7374e4..3d2f26e0d 100644
--- a/lib/sqlalchemy/orm/session.py
+++ b/lib/sqlalchemy/orm/session.py
@@ -30,6 +30,7 @@ from .unitofwork import UOWTransaction
 from .. import engine
 from .. import exc as sa_exc
 from .. import future
+from .. import sql
 from .. import util
 from ..inspection import inspect
 from ..sql import coercions
@@ -288,7 +289,7 @@ class ORMExecuteState(util.MemoizedSlots):
         return self.statement._execution_options.union(self._execution_options)
 
     def _orm_compile_options(self):
-        opts = self.statement.compile_options
+        opts = self.statement._compile_options
         if isinstance(opts, context.ORMCompileState.default_compile_options):
             return opts
         else:
@@ -367,134 +368,53 @@ class ORMExecuteState(util.MemoizedSlots):
 class SessionTransaction(object):
     """A :class:`.Session`-level transaction.
 
-    :class:`.SessionTransaction` is a mostly behind-the-scenes object
-    not normally referenced directly by application code.   It coordinates
-    among multiple :class:`_engine.Connection` objects, maintaining a database
-    transaction for each one individually, committing or rolling them
-    back all at once.   It also provides optional two-phase commit behavior
-    which can augment this coordination operation.
-
-    The :attr:`.Session.transaction` attribute of :class:`.Session`
-    refers to the current :class:`.SessionTransaction` object in use, if any.
-    The :attr:`.SessionTransaction.parent` attribute refers to the parent
-    :class:`.SessionTransaction` in the stack of :class:`.SessionTransaction`
-    objects.  If this attribute is ``None``, then this is the top of the stack.
-    If non-``None``, then this :class:`.SessionTransaction` refers either
-    to a so-called "subtransaction" or a "nested" transaction.  A
-    "subtransaction" is a scoping concept that demarcates an inner portion
-    of the outermost "real" transaction.  A nested transaction, which
-    is indicated when the :attr:`.SessionTransaction.nested`
-    attribute is also True, indicates that this :class:`.SessionTransaction`
-    corresponds to a SAVEPOINT.
-
-    **Life Cycle**
-
-    A :class:`.SessionTransaction` is associated with a :class:`.Session` in
-    its default mode of ``autocommit=False`` whenever the "autobegin" process
-    takes place, associated with no database connections.  As the
-    :class:`.Session` is called upon to emit SQL on behalf of various
-    :class:`_engine.Engine` or :class:`_engine.Connection` objects,
-    a corresponding
-    :class:`_engine.Connection` and associated :class:`.Transaction`
-    is added to a
-    collection within the :class:`.SessionTransaction` object, becoming one of
-    the connection/transaction pairs maintained by the
-    :class:`.SessionTransaction`.  The start of a :class:`.SessionTransaction`
-    can be tracked using the :meth:`.SessionEvents.after_transaction_create`
-    event.
-
-    The lifespan of the :class:`.SessionTransaction` ends when the
-    :meth:`.Session.commit`, :meth:`.Session.rollback` or
-    :meth:`.Session.close` methods are called.  At this point, the
-    :class:`.SessionTransaction` removes its association with its parent
-    :class:`.Session`.   A :class:`.Session` that is in ``autocommit=False``
-    mode will create a new :class:`.SessionTransaction` to replace it when the
-    next "autobegin" event occurs, whereas a :class:`.Session` that's in
-    ``autocommit=True`` mode will remain without a :class:`.SessionTransaction`
-    until the :meth:`.Session.begin` method is called.  The end of a
-    :class:`.SessionTransaction` can be tracked using the
-    :meth:`.SessionEvents.after_transaction_end` event.
-
-    .. versionchanged:: 1.4 the :class:`.SessionTransaction` is not created
-       immediately within a :class:`.Session` when constructed or when the
-       previous transaction is removed, it instead is created when the
-       :class:`.Session` is next used.
-
-    **Nesting and Subtransactions**
-
-    Another detail of :class:`.SessionTransaction` behavior is that it is
-    capable of "nesting".  This means that the :meth:`.Session.begin` method
-    can be called while an existing :class:`.SessionTransaction` is already
-    present, producing a new :class:`.SessionTransaction` that temporarily
-    replaces the parent :class:`.SessionTransaction`.   When a
-    :class:`.SessionTransaction` is produced as nested, it assigns itself to
-    the :attr:`.Session.transaction` attribute, and it additionally will assign
-    the previous :class:`.SessionTransaction` to its :attr:`.Session.parent`
-    attribute.  The behavior is effectively a
-    stack, where :attr:`.Session.transaction` refers to the current head of
-    the stack, and the :attr:`.SessionTransaction.parent` attribute allows
-    traversal up the stack until :attr:`.SessionTransaction.parent` is
-    ``None``, indicating the top of the stack.
-
-    When the scope of :class:`.SessionTransaction` is ended via
-    :meth:`.Session.commit` or :meth:`.Session.rollback`, it restores its
-    parent :class:`.SessionTransaction` back onto the
-    :attr:`.Session.transaction` attribute.
-
-    The purpose of this stack is to allow nesting of
-    :meth:`.Session.rollback` or :meth:`.Session.commit` calls in context
-    with various flavors of :meth:`.Session.begin`. This nesting behavior
-    applies to when :meth:`.Session.begin_nested` is used to emit a
-    SAVEPOINT transaction, and is also used to produce a so-called
-    "subtransaction" which allows a block of code to use a
-    begin/rollback/commit sequence regardless of whether or not its enclosing
-    code block has begun a transaction.  The :meth:`.flush` method, whether
-    called explicitly or via autoflush, is the primary consumer of the
-    "subtransaction" feature, in that it wishes to guarantee that it works
-    within in a transaction block regardless of whether or not the
-    :class:`.Session` is in transactional mode when the method is called.
-
-    Note that the flush process that occurs within the "autoflush" feature
-    as well as when the :meth:`.Session.flush` method is used **always**
-    creates a :class:`.SessionTransaction` object.   This object is normally
-    a subtransaction, unless the :class:`.Session` is in autocommit mode
-    and no transaction exists at all, in which case it's the outermost
-    transaction.   Any event-handling logic or other inspection logic
-    needs to take into account whether a :class:`.SessionTransaction`
-    is the outermost transaction, a subtransaction, or a "nested" / SAVEPOINT
-    transaction.
+    :class:`.SessionTransaction` is produced from the
+    :meth:`_orm.Session.begin`
+    and :meth:`_orm.Session.begin_nested` methods.   It's largely an internal
+    object that in modern use provides a context manager for session
+    transactions.
 
-    .. seealso::
+    Documentation on interacting with :class:`_orm.SessionTransaction` is
+    at: :ref:`unitofwork_transaction`.
 
-        :meth:`.Session.rollback`
 
-        :meth:`.Session.commit`
+    .. versionchanged:: 1.4  The scoping and API methods to work with the
+       :class:`_orm.SessionTransaction` object directly have been simplified.
+
+    .. seealso::
+
+        :ref:`unitofwork_transaction`
 
         :meth:`.Session.begin`
 
         :meth:`.Session.begin_nested`
 
-        :attr:`.Session.is_active`
+        :meth:`.Session.rollback`
+
+        :meth:`.Session.commit`
 
-        :meth:`.SessionEvents.after_transaction_create`
+        :meth:`.Session.in_transaction`
 
-        :meth:`.SessionEvents.after_transaction_end`
+        :meth:`.Session.in_nested_transaction`
 
-        :meth:`.SessionEvents.after_commit`
+        :meth:`.Session.get_transaction`
 
-        :meth:`.SessionEvents.after_rollback`
+        :meth:`.Session.get_nested_transaction`
 
-        :meth:`.SessionEvents.after_soft_rollback`
 
     """
 
     _rollback_exception = None
 
-    def __init__(self, session, parent=None, nested=False, autobegin=False):
+    def __init__(
+        self, session, parent=None, nested=False, autobegin=False,
+    ):
         self.session = session
         self._connections = {}
         self._parent = parent
         self.nested = nested
+        if nested:
+            self._previous_nested_transaction = session._nested_transaction
         self._state = ACTIVE
         if not parent and nested:
             raise sa_exc.InvalidRequestError(
@@ -688,6 +608,8 @@ class SessionTransaction(object):
             return self._connections[bind][0]
 
         local_connect = False
+        should_commit = True
+
         if self._parent:
             conn = self._parent._connection_for_bind(bind, execution_options)
             if not self.nested:
@@ -712,11 +634,16 @@ class SessionTransaction(object):
                 transaction = conn.begin_twophase()
             elif self.nested:
                 transaction = conn.begin_nested()
-            else:
-                if conn._is_future and conn.in_transaction():
-                    transaction = conn._transaction
+            elif conn.in_transaction():
+                # if given a future connection already in a transaction, don't
+                # commit that transaction unless it is a savepoint
+                if conn.in_nested_transaction():
+                    transaction = conn.get_nested_transaction()
                 else:
-                    transaction = conn.begin()
+                    transaction = conn.get_transaction()
+                    should_commit = False
+            else:
+                transaction = conn.begin()
         except:
             # connection will not not be associated with this Session;
             # close it immediately so that it isn't closed under GC
@@ -729,7 +656,7 @@ class SessionTransaction(object):
             self._connections[conn] = self._connections[conn.engine] = (
                 conn,
                 transaction,
-                not bind_is_connection or not conn._is_future,
+                should_commit,
                 not bind_is_connection,
             )
             self.session.dispatch.after_begin(self.session, self, conn)
@@ -748,7 +675,7 @@ class SessionTransaction(object):
         if self._parent is None or self.nested:
             self.session.dispatch.before_commit(self.session)
 
-        stx = self.session.transaction
+        stx = self.session._transaction
         if stx is not self:
             for subtransaction in stx._iterate_self_and_parents(upto=self):
                 subtransaction.commit()
@@ -775,7 +702,7 @@ class SessionTransaction(object):
 
         self._state = PREPARED
 
-    def commit(self):
+    def commit(self, _to_root=False):
         self._assert_active(prepared_ok=True)
         if self._state is not PREPARED:
             self._prepare_impl()
@@ -793,12 +720,16 @@ class SessionTransaction(object):
             self._remove_snapshot()
 
         self.close()
+
+        if _to_root and self._parent:
+            return self._parent.commit(_to_root=True)
+
         return self._parent
 
-    def rollback(self, _capture_exception=False):
+    def rollback(self, _capture_exception=False, _to_root=False):
         self._assert_active(prepared_ok=True, rollback_ok=True)
 
-        stx = self.session.transaction
+        stx = self.session._transaction
         if stx is not self:
             for subtransaction in stx._iterate_self_and_parents(upto=self):
                 subtransaction.close()
@@ -849,20 +780,28 @@ class SessionTransaction(object):
 
         sess.dispatch.after_soft_rollback(sess, self)
 
+        if _to_root and self._parent:
+            return self._parent.rollback(_to_root=True)
         return self._parent
 
     def close(self, invalidate=False):
+        if self.nested:
+            self.session._nested_transaction = (
+                self._previous_nested_transaction
+            )
+
         self.session._transaction = self._parent
+
         if self._parent is None:
             for connection, transaction, should_commit, autoclose in set(
                 self._connections.values()
             ):
                 if invalidate:
                     connection.invalidate()
+                if should_commit and transaction.is_active:
+                    transaction.close()
                 if autoclose:
                     connection.close()
-                else:
-                    transaction.close()
 
         self._state = CLOSED
         self.session.dispatch.after_transaction_end(self.session, self)
@@ -924,6 +863,15 @@ class Session(_SessionClassMethods):
         "scalar",
     )
 
+    @util.deprecated_params(
+        autocommit=(
+            "2.0",
+            "The :paramref:`.Session.autocommit` parameter is deprecated "
+            "and will be removed in SQLAlchemy version 2.0.  Please use the "
+            ":paramref:`.Session.autobegin` parameter set to False to support "
+            "explicit use of the :meth:`.Session.begin` method.",
+        ),
+    )
     def __init__(
         self,
         bind=None,
@@ -1071,8 +1019,6 @@ class Session(_SessionClassMethods):
            :class:`.Session` dictionary will be local to that
            :class:`.Session`.
 
-           .. versionadded:: 0.9.0
-
         :param query_cls:  Class which should be used to create new Query
           objects, as returned by the :meth:`~.Session.query` method.
           Defaults to :class:`_query.Query`.
@@ -1096,13 +1042,23 @@ class Session(_SessionClassMethods):
         self._flushing = False
         self._warn_on_events = False
         self._transaction = None
+        self._nested_transaction = None
         self.future = future
         self.hash_key = _new_sessionid()
         self.autoflush = autoflush
-        self.autocommit = autocommit
         self.expire_on_commit = expire_on_commit
         self.enable_baked_queries = enable_baked_queries
 
+        if autocommit:
+            if future:
+                raise sa_exc.ArgumentError(
+                    "Cannot use autocommit mode with future=True.  "
+                    "use the autobegin flag."
+                )
+            self.autocommit = True
+        else:
+            self.autocommit = False
+
         self.twophase = twophase
         self._query_cls = query_cls if query_cls else query.Query
         if info:
@@ -1116,21 +1072,77 @@ class Session(_SessionClassMethods):
 
     connection_callable = None
 
+    def __enter__(self):
+        return self
+
+    def __exit__(self, type_, value, traceback):
+        self.close()
+
     @property
+    @util.deprecated_20(
+        "The :attr:`_orm.Session.transaction` accessor is deprecated and "
+        "will be removed in SQLAlchemy version 2.0.  "
+        "For context manager use, use :meth:`_orm.Session.begin`.  To access "
+        "the current root transaction, use "
+        ":meth:`_orm.Session.get_transaction()"
+    )
     def transaction(self):
         """The current active or inactive :class:`.SessionTransaction`.
 
-        If this session is in "autobegin" mode and the transaction was not
-        begun, this accessor will implicitly begin the transaction.
+        May be None if no transaction has begun yet.
 
         .. versionchanged:: 1.4  the :attr:`.Session.transaction` attribute
-           is now a read-only descriptor that will automatically start a
-           transaction in "autobegin" mode if one is not present.
+           is now a read-only descriptor that also may return None if no
+           transaction has begun yet.
+
 
         """
-        self._autobegin()
+        if not self.future:
+            self._autobegin()
         return self._transaction
 
+    def in_transaction(self):
+        """Return True if this :class:`_orm.Session` has begun a transaction.
+
+        .. versionadded:: 1.4
+
+        .. seealso::
+
+            :attr:`_orm.Session.is_active`
+
+
+        """
+        return self._transaction is not None
+
+    def in_nested_transaction(self):
+        """Return True if this :class:`_orm.Session` has begun a nested
+        transaction, e.g. SAVEPOINT.
+
+        .. versionadded:: 1.4
+
+        """
+        return self._nested_transaction is not None
+
+    def get_transaction(self):
+        """Return the current root transaction in progress, if any.
+
+        .. versionadded:: 1.4
+
+        """
+        trans = self._transaction
+        while trans is not None and trans._parent is not None:
+            trans = trans._parent
+        return trans
+
+    def get_nested_transaction(self):
+        """Return the current nested transaction in progress, if any.
+
+        .. versionadded:: 1.4
+
+        """
+
+        return self._nested_transaction
+
     @util.memoized_property
     def info(self):
         """A user-modifiable dictionary.
@@ -1141,8 +1153,6 @@ class Session(_SessionClassMethods):
         here is always local to this :class:`.Session` and can be modified
         independently of all other :class:`.Session` objects.
 
-        .. versionadded:: 0.9.0
-
         """
         return {}
 
@@ -1153,7 +1163,17 @@ class Session(_SessionClassMethods):
 
         return False
 
-    def begin(self, subtransactions=False, nested=False):
+    @util.deprecated_params(
+        subtransactions=(
+            "2.0",
+            "The :paramref:`_orm.Session.begin.subtransactions` flag is "
+            "deprecated and "
+            "will be removed in SQLAlchemy version 2.0.  The "
+            ":attr:`_orm.Session.transaction` flag may "
+            "be checked for None before invoking :meth:`_orm.Session.begin`.",
+        )
+    )
+    def begin(self, subtransactions=False, nested=False, _subtrans=False):
         """Begin a transaction on this :class:`.Session`.
 
         .. warning::
@@ -1206,17 +1226,24 @@ class Session(_SessionClassMethods):
 
         """
 
+        if subtransactions and self.future:
+            raise NotImplementedError(
+                "subtransactions are not implemented in future "
+                "Session objects."
+            )
         if self._autobegin():
             if not subtransactions and not nested:
-                return
+                return self._transaction
 
         if self._transaction is not None:
-            if subtransactions or nested:
-                self._transaction = self._transaction._begin(nested=nested)
+            if subtransactions or _subtrans or nested:
+                trans = self._transaction._begin(nested=nested)
+                self._transaction = trans
+                if nested:
+                    self._nested_transaction = trans
             else:
                 raise sa_exc.InvalidRequestError(
-                    "A transaction is already begun.  Use "
-                    "subtransactions=True to allow subtransactions."
+                    "A transaction is already begun on this Session."
                 )
         else:
             self._transaction = SessionTransaction(self, nested=nested)
@@ -1265,7 +1292,7 @@ class Session(_SessionClassMethods):
         if self._transaction is None:
             pass
         else:
-            self._transaction.rollback()
+            self._transaction.rollback(_to_root=self.future)
 
     def commit(self):
         """Flush pending changes and commit the current transaction.
@@ -1299,7 +1326,7 @@ class Session(_SessionClassMethods):
             if not self._autobegin():
                 raise sa_exc.InvalidRequestError("No transaction is begun.")
 
-        self._transaction.commit()
+        self._transaction.commit(_to_root=self.future)
 
     def prepare(self):
         """Prepare the current transaction in progress for two phase commit.
@@ -1371,8 +1398,6 @@ class Session(_SessionClassMethods):
          present within the :class:`.Session`, a warning is emitted and
          the arguments are ignored.
 
-         .. versionadded:: 0.9.9
-
          .. seealso::
 
             :ref:`session_transaction_isolation`
@@ -1402,6 +1427,7 @@ class Session(_SessionClassMethods):
             )
 
         assert self._transaction is None
+        assert self.autocommit
         conn = engine.connect(**kw)
         if execution_options:
             conn = conn.execution_options(**execution_options)
@@ -1663,12 +1689,19 @@ class Session(_SessionClassMethods):
 
         This is a variant of :meth:`.Session.close` that will additionally
         ensure that the :meth:`_engine.Connection.invalidate`
-        method will be called
-        on all :class:`_engine.Connection` objects.  This can be called when
-        the database is known to be in a state where the connections are
-        no longer safe to be used.
+        method will be called on each :class:`_engine.Connection` object
+        that is currently in use for a transaction (typically there is only
+        one connection unless the :class:`_orm.Session` is used with
+        multiple engines).
 
-        E.g.::
+        This can be called when the database is known to be in a state where
+        the connections are no longer safe to be used.
+
+        Below illustrates a scenario when using `gevent
+        <http://www.gevent.org/>`_, which can produce ``Timeout`` exceptions
+        that may mean the underlying connection should be discarded::
+
+            import gevent
 
             try:
                 sess = Session()
@@ -1681,13 +1714,8 @@ class Session(_SessionClassMethods):
                 sess.rollback()
                 raise
 
-        This clears all items and ends any transaction in progress.
-
-        If this session were created with ``autocommit=False``, a new
-        transaction is immediately begun.  Note that this new transaction does
-        not use any connection resources until they are first needed.
-
-        .. versionadded:: 0.9.9
+        The method additionally does everything that :meth:`_orm.Session.close`
+        does, including that all ORM objects are expunged.
 
         """
         self._close_impl(invalidate=True)
@@ -2118,13 +2146,7 @@ class Session(_SessionClassMethods):
                 "A blank dictionary is ambiguous."
             )
 
-        if with_for_update is not None:
-            if with_for_update is True:
-                with_for_update = query.ForUpdateArg()
-            elif with_for_update:
-                with_for_update = query.ForUpdateArg(**with_for_update)
-            else:
-                with_for_update = None
+        with_for_update = query.ForUpdateArg._from_argument(with_for_update)
 
         stmt = future.select(object_mapper(instance))
         if (
@@ -2482,6 +2504,200 @@ class Session(_SessionClassMethods):
             for o, m, st_, dct_ in cascade_states:
                 self._delete_impl(st_, o, False)
 
+    def get(
+        self,
+        entity,
+        ident,
+        options=None,
+        populate_existing=False,
+        with_for_update=None,
+        identity_token=None,
+    ):
+        """Return an instance based on the given primary key identifier,
+        or ``None`` if not found.
+
+        E.g.::
+
+            my_user = session.get(User, 5)
+
+            some_object = session.get(VersionedFoo, (5, 10))
+
+            some_object = session.get(
+                VersionedFoo,
+                {"id": 5, "version_id": 10}
+            )
+
+        .. versionadded:: 1.4 Added :meth:`_orm.Session.get`, which is moved
+           from the now deprecated :meth:`_orm.Query.get` method.
+
+        :meth:`_orm.Session.get` is special in that it provides direct
+        access to the identity map of the :class:`.Session`.
+        If the given primary key identifier is present
+        in the local identity map, the object is returned
+        directly from this collection and no SQL is emitted,
+        unless the object has been marked fully expired.
+        If not present,
+        a SELECT is performed in order to locate the object.
+
+        :meth:`_orm.Session.get` also will perform a check if
+        the object is present in the identity map and
+        marked as expired - a SELECT
+        is emitted to refresh the object as well as to
+        ensure that the row is still present.
+        If not, :class:`~sqlalchemy.orm.exc.ObjectDeletedError` is raised.
+
+        :param entity: a mapped class or :class:`.Mapper` indicating the
+         type of entity to be loaded.
+
+        :param ident: A scalar, tuple, or dictionary representing the
+         primary key.  For a composite (e.g. multiple column) primary key,
+         a tuple or dictionary should be passed.
+
+         For a single-column primary key, the scalar calling form is typically
+         the most expedient.  If the primary key of a row is the value "5",
+         the call looks like::
+
+            my_object = session.get(SomeClass, 5)
+
+         The tuple form contains primary key values typically in
+         the order in which they correspond to the mapped
+         :class:`_schema.Table`
+         object's primary key columns, or if the
+         :paramref:`_orm.Mapper.primary_key` configuration parameter were
+         used, in
+         the order used for that parameter. For example, if the primary key
+         of a row is represented by the integer
+         digits "5, 10" the call would look like::
+
+             my_object = session.get(SomeClass, (5, 10))
+
+         The dictionary form should include as keys the mapped attribute names
+         corresponding to each element of the primary key.  If the mapped class
+         has the attributes ``id``, ``version_id`` as the attributes which
+         store the object's primary key value, the call would look like::
+
+            my_object = session.get(SomeClass, {"id": 5, "version_id": 10})
+
+        :param options: optional sequence of loader options which will be
+         applied to the query, if one is emitted.
+
+        :param populate_existing: causes the method to unconditionally emit
+         a SQL query and refresh the object with the newly loaded data,
+         regardless of whether or not the object is already present.
+
+        :param with_for_update: optional boolean ``True`` indicating FOR UPDATE
+          should be used, or may be a dictionary containing flags to
+          indicate a more specific set of FOR UPDATE flags for the SELECT;
+          flags should match the parameters of
+          :meth:`_query.Query.with_for_update`.
+          Supersedes the :paramref:`.Session.refresh.lockmode` parameter.
+
+        :return: The object instance, or ``None``.
+
+        """
+        return self._get_impl(
+            entity,
+            ident,
+            loading.load_on_pk_identity,
+            options,
+            populate_existing=populate_existing,
+            with_for_update=with_for_update,
+            identity_token=identity_token,
+        )
+
+    def _get_impl(
+        self,
+        entity,
+        primary_key_identity,
+        db_load_fn,
+        options=None,
+        populate_existing=False,
+        with_for_update=None,
+        identity_token=None,
+        execution_options=None,
+    ):
+
+        # convert composite types to individual args
+        if hasattr(primary_key_identity, "__composite_values__"):
+            primary_key_identity = primary_key_identity.__composite_values__()
+
+        mapper = inspect(entity)
+
+        is_dict = isinstance(primary_key_identity, dict)
+        if not is_dict:
+            primary_key_identity = util.to_list(
+                primary_key_identity, default=(None,)
+            )
+
+        if len(primary_key_identity) != len(mapper.primary_key):
+            raise sa_exc.InvalidRequestError(
+                "Incorrect number of values in identifier to formulate "
+                "primary key for query.get(); primary key columns are %s"
+                % ",".join("'%s'" % c for c in mapper.primary_key)
+            )
+
+        if is_dict:
+            try:
+                primary_key_identity = list(
+                    primary_key_identity[prop.key]
+                    for prop in mapper._identity_key_props
+                )
+
+            except KeyError as err:
+                util.raise_(
+                    sa_exc.InvalidRequestError(
+                        "Incorrect names of values in identifier to formulate "
+                        "primary key for query.get(); primary key attribute "
+                        "names are %s"
+                        % ",".join(
+                            "'%s'" % prop.key
+                            for prop in mapper._identity_key_props
+                        )
+                    ),
+                    replace_context=err,
+                )
+
+        if (
+            not populate_existing
+            and not mapper.always_refresh
+            and with_for_update is None
+        ):
+
+            instance = self._identity_lookup(
+                mapper, primary_key_identity, identity_token=identity_token
+            )
+
+            if instance is not None:
+                # reject calls for id in identity map but class
+                # mismatch.
+                if not issubclass(instance.__class__, mapper.class_):
+                    return None
+                return instance
+            elif instance is attributes.PASSIVE_CLASS_MISMATCH:
+                return None
+
+        # apply_labels() not strictly necessary, however this will ensure that
+        # tablename_colname style is used which at the moment is asserted
+        # in a lot of unit tests :)
+
+        load_options = context.QueryContext.default_load_options
+
+        if populate_existing:
+            load_options += {"_populate_existing": populate_existing}
+        statement = sql.select(mapper).apply_labels()
+        if with_for_update is not None:
+            statement._for_update_arg = query.ForUpdateArg._from_argument(
+                with_for_update
+            )
+
+        if options:
+            statement = statement.options(*options)
+        if execution_options:
+            statement = statement.execution_options(**execution_options)
+        return db_load_fn(
+            self, statement, primary_key_identity, load_options=load_options,
+        )
+
     def merge(self, instance, load=True):
         """Copy the state of a given instance into a corresponding instance
         within this :class:`.Session`.
@@ -2629,7 +2845,7 @@ class Session(_SessionClassMethods):
                 new_instance = True
 
             elif key_is_persistent:
-                merged = self.query(mapper.class_).get(key[1])
+                merged = self.get(mapper.class_, key[1], identity_token=key[2])
 
         if merged is None:
             merged = mapper.class_manager.new_instance()
@@ -3021,9 +3237,7 @@ class Session(_SessionClassMethods):
         if not flush_context.has_work:
             return
 
-        flush_context.transaction = transaction = self.begin(
-            subtransactions=True
-        )
+        flush_context.transaction = transaction = self.begin(_subtrans=True)
         try:
             self._warn_on_events = True
             try:
@@ -3338,7 +3552,7 @@ class Session(_SessionClassMethods):
         mapper = _class_to_mapper(mapper)
         self._flushing = True
 
-        transaction = self.begin(subtransactions=True)
+        transaction = self.begin(_subtrans=True)
         try:
             if isupdate:
                 persistence._bulk_update(
@@ -3441,62 +3655,38 @@ class Session(_SessionClassMethods):
 
     @property
     def is_active(self):
-        """True if this :class:`.Session` is in "transaction mode" and
-        is not in "partial rollback" state.
-
-        The :class:`.Session` in its default mode of ``autocommit=False``
-        is essentially always in "transaction mode", in that a
-        :class:`.SessionTransaction` is associated with it as soon as
-        it is instantiated.  This :class:`.SessionTransaction` is immediately
-        replaced with a new one as soon as it is ended, due to a rollback,
-        commit, or close operation.
-
-        "Transaction mode" does *not* indicate whether
-        or not actual database connection resources are in use;  the
-        :class:`.SessionTransaction` object coordinates among zero or more
-        actual database transactions, and starts out with none, accumulating
-        individual DBAPI connections as different data sources are used
-        within its scope.   The best way to track when a particular
-        :class:`.Session` has actually begun to use DBAPI resources is to
-        implement a listener using the :meth:`.SessionEvents.after_begin`
-        method, which will deliver both the :class:`.Session` as well as the
-        target :class:`_engine.Connection` to a user-defined event listener.
-
-        The "partial rollback" state refers to when an "inner" transaction,
-        typically used during a flush, encounters an error and emits a
-        rollback of the DBAPI connection.  At this point, the
-        :class:`.Session` is in "partial rollback" and awaits for the user to
-        call :meth:`.Session.rollback`, in order to close out the
-        transaction stack.  It is in this "partial rollback" period that the
-        :attr:`.is_active` flag returns False.  After the call to
-        :meth:`.Session.rollback`, the :class:`.SessionTransaction` is
-        replaced with a new one and :attr:`.is_active` returns ``True`` again.
-
-        When a :class:`.Session` is used in ``autocommit=True`` mode, the
-        :class:`.SessionTransaction` is only instantiated within the scope
-        of a flush call, or when :meth:`.Session.begin` is called.  So
-        :attr:`.is_active` will always be ``False`` outside of a flush or
-        :meth:`.Session.begin` block in this mode, and will be ``True``
-        within the :meth:`.Session.begin` block as long as it doesn't enter
-        "partial rollback" state.
-
-        From all the above, it follows that the only purpose to this flag is
-        for application frameworks that wish to detect if a "rollback" is
-        necessary within a generic error handling routine, for
-        :class:`.Session` objects that would otherwise be in
-        "partial rollback" mode.  In a typical integration case, this is also
-        not necessary as it is standard practice to emit
-        :meth:`.Session.rollback` unconditionally within the outermost
-        exception catch.
-
-        To track the transactional state of a :class:`.Session` fully,
-        use event listeners, primarily the :meth:`.SessionEvents.after_begin`,
-        :meth:`.SessionEvents.after_commit`,
-        :meth:`.SessionEvents.after_rollback` and related events.
+        """True if this :class:`.Session` not in "partial rollback" state.
+
+        .. versionchanged:: 1.4 The :class:`_orm.Session` no longer begins
+           a new transaction immediately, so this attribute will be False
+           when the :class:`_orm.Session` is first instantiated.
+
+        "partial rollback" state typically indicates that the flush process
+        of the :class:`_orm.Session` has failed, and that the
+        :meth:`_orm.Session.rollback` method must be emitted in order to
+        fully roll back the transaction.
+
+        If this :class:`_orm.Session` is not in a transaction at all, the
+        :class:`_orm.Session` will autobegin when it is first used, so in this
+        case :attr:`_orm.Session.is_active` will return True.
+
+        Otherwise, if this :class:`_orm.Session` is within a transaction,
+        and that transaction has not been rolled back internally, the
+        :attr:`_orm.Session.is_active` will also return True.
+
+        .. seealso::
+
+            :ref:`faq_session_rollback`
+
+            :meth:`_orm.Session.in_transaction`
 
         """
-        self._autobegin()
-        return self._transaction and self._transaction.is_active
+        if self.autocommit:
+            return (
+                self._transaction is not None and self._transaction.is_active
+            )
+        else:
+            return self._transaction is None or self._transaction.is_active
 
     identity_map = None
     """A mapping of object identities to objects themselves.
@@ -3576,36 +3766,84 @@ class sessionmaker(_SessionClassMethods):
 
     e.g.::
 
-        # global scope
-        Session = sessionmaker(autoflush=False)
+        from sqlalchemy import create_engine
+        from sqlalchemy.orm import sessionmaker
 
-        # later, in a local scope, create and use a session:
-        sess = Session()
+        # an Engine, which the Session will use for connection
+        # resources
+        engine = create_engine('postgresql://scott:tiger@localhost/')
 
-    Any keyword arguments sent to the constructor itself will override the
-    "configured" keywords::
+        Session = sessionmaker(engine)
 
-        Session = sessionmaker()
+        with Session() as session:
+            session.add(some_object)
+            session.add(some_other_object)
+            session.commit()
+
+    Context manager use is optional; otherwise, the returned
+    :class:`_orm.Session` object may be closed explicitly via the
+    :meth:`_orm.Session.close` method.   Using a
+    ``try:/finally:`` block is optional, however will ensure that the close
+    takes place even if there are database errors::
+
+        session = Session()
+        try:
+            session.add(some_object)
+            session.add(some_other_object)
+            session.commit()
+        finally:
+            session.close()
+
+    :class:`.sessionmaker` acts as a factory for :class:`_orm.Session`
+    objects in the same way as an :class:`_engine.Engine` acts as a factory
+    for :class:`_engine.Connection` objects.  In this way it also includes
+    a :meth:`_orm.sessionmaker.begin` method, that provides a context
+    manager which both begins and commits a transaction, as well as closes
+    out the :class:`_orm.Session` when complete, rolling back the transaction
+    if any errors occur::
+
+        Session = sessionmaker(engine)
+
+        wih Session.begin() as session:
+            session.add(some_object)
+            session.add(some_other_object)
+        # commits transaction, closes session
+
+    .. versionadded:: 1.4
+
+    When calling upon :class:`_orm.sessionmaker` to construct a
+    :class:`_orm.Session`, keyword arguments may also be passed to the
+    method; these arguments will override that of the globally configured
+    parameters.  Below we use a :class:`_orm.sessionmaker` bound to a certain
+    :class:`_engine.Engine` to produce a :class:`_orm.Session` that is instead
+    bound to a specific :class:`_engine.Connection` procured from that engine::
+
+        Session = sessionmaker(engine)
 
         # bind an individual session to a connection
-        sess = Session(bind=connection)
 
-    The class also includes a method :meth:`.configure`, which can
-    be used to specify additional keyword arguments to the factory, which
-    will take effect for subsequent :class:`.Session` objects generated.
-    This is usually used to associate one or more :class:`_engine.Engine`
-    objects
-    with an existing :class:`.sessionmaker` factory before it is first
-    used::
+        with engine.connect() as connection:
+            with Session(bind=connection) as session:
+                # work with session
+
+    The class also includes a method :meth:`_orm.sessionmaker.configure`, which
+    can be used to specify additional keyword arguments to the factory, which
+    will take effect for subsequent :class:`.Session` objects generated. This
+    is usually used to associate one or more :class:`_engine.Engine` objects
+    with an existing
+    :class:`.sessionmaker` factory before it is first used::
 
-        # application starts
+        # application starts, sessionmaker does not have
+        # an engine bound yet
         Session = sessionmaker()
 
-        # ... later
+        # ... later, when an engine URL is read from a configuration
+        # file or other events allow the engine to be created
         engine = create_engine('sqlite:///foo.db')
         Session.configure(bind=engine)
 
         sess = Session()
+        # work with session
 
     .. seealso::
 
@@ -3646,8 +3884,6 @@ class sessionmaker(_SessionClassMethods):
          replaced, when the ``info`` parameter is specified to the specific
          :class:`.Session` construction operation.
 
-         .. versionadded:: 0.9.0
-
         :param \**kw: all other keyword arguments are passed to the
          constructor of newly created :class:`.Session` objects.
 
@@ -3663,6 +3899,29 @@ class sessionmaker(_SessionClassMethods):
         # events can be associated with it specifically.
         self.class_ = type(class_.__name__, (class_,), {})
 
+    @util.contextmanager
+    def begin(self):
+        """Produce a context manager that both provides a new
+        :class:`_orm.Session` as well as a transaction that commits.
+
+
+        e.g.::
+
+            Session = sessionmaker(some_engine)
+
+            with Session.begin() as session:
+                session.add(some_object)
+
+            # commits transaction, closes session
+
+        .. versionadded:: 1.4
+
+
+        """
+        with self() as session:
+            with session.begin():
+                yield session
+
     def __call__(self, **local_kw):
         """Produce a new :class:`.Session` object using the configuration
         established in this :class:`.sessionmaker`.
@@ -3806,8 +4065,6 @@ def make_transient_to_detached(instance):
     call to :meth:`.Session.merge` in that a given persistent state
     can be manufactured without any SQL calls.
 
-    .. versionadded:: 0.9.5
-
     .. seealso::
 
         :func:`.make_transient`
diff --git a/lib/sqlalchemy/orm/strategies.py b/lib/sqlalchemy/orm/strategies.py
index 53cc99ccd..db82f0b74 100644
--- a/lib/sqlalchemy/orm/strategies.py
+++ b/lib/sqlalchemy/orm/strategies.py
@@ -677,7 +677,7 @@ class LazyLoader(AbstractRelationshipLoader, util.MemoizedSlots):
                         self._equated_columns[c] = self._equated_columns[col]
 
             self.logger.info(
-                "%s will use query.get() to " "optimize instance loads", self
+                "%s will use Session.get() to " "optimize instance loads", self
             )
 
     def init_class_attribute(self, mapper):
diff --git a/lib/sqlalchemy/orm/util.py b/lib/sqlalchemy/orm/util.py
index e04c54497..68ffa2393 100644
--- a/lib/sqlalchemy/orm/util.py
+++ b/lib/sqlalchemy/orm/util.py
@@ -1621,3 +1621,108 @@ def randomize_unitofwork():
     topological.set = (
         unitofwork.set
     ) = 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.
+
+    """
+
+    if isinstance(item, slice):
+        start, stop, step = util.decode_slice(item)
+
+        if (
+            isinstance(stop, int)
+            and isinstance(start, int)
+            and stop - start <= 0
+        ):
+            return []
+
+        # perhaps we should execute a count() here so that we
+        # can still use LIMIT/OFFSET ?
+        elif (isinstance(start, int) and start < 0) or (
+            isinstance(stop, int) and stop < 0
+        ):
+            return list(iterable_query)[item]
+
+        res = iterable_query.slice(start, stop)
+        if step is not None:
+            return list(res)[None : None : item.step]
+        else:
+            return list(res)
+    else:
+        if item == -1:
+            return list(iterable_query)[-1]
+        else:
+            return list(iterable_query[item : item + 1])[0]
diff --git a/lib/sqlalchemy/sql/base.py b/lib/sqlalchemy/sql/base.py
index 4bc6d8280..36a8151d3 100644
--- a/lib/sqlalchemy/sql/base.py
+++ b/lib/sqlalchemy/sql/base.py
@@ -571,6 +571,26 @@ class Options(util.with_metaclass(_MetaOptions)):
         o1.__dict__.update(other)
         return o1
 
+    def __eq__(self, other):
+        # TODO: very inefficient.  This is used only in test suites
+        # right now.
+        for a, b in util.zip_longest(self._cache_attrs, other._cache_attrs):
+            if getattr(self, a) != getattr(other, b):
+                return False
+        return True
+
+    def __repr__(self):
+        # TODO: fairly inefficient, used only in debugging right now.
+
+        return "%s(%s)" % (
+            self.__class__.__name__,
+            ", ".join(
+                "%s=%r" % (k, self.__dict__[k])
+                for k in self._cache_attrs
+                if k in self.__dict__
+            ),
+        )
+
     @hybridmethod
     def add_to_element(self, name, value):
         return self + {name: getattr(self, name) + value}
@@ -610,6 +630,60 @@ class Options(util.with_metaclass(_MetaOptions)):
             )
         return cls + d
 
+    @classmethod
+    def from_execution_options(
+        cls, key, attrs, exec_options, statement_exec_options
+    ):
+        """"process Options argument in terms of execution options.
+
+
+        e.g.::
+
+            (
+                load_options,
+                execution_options,
+            ) = QueryContext.default_load_options.from_execution_options(
+                "_sa_orm_load_options",
+                {
+                    "populate_existing",
+                    "autoflush",
+                    "yield_per"
+                },
+                execution_options,
+                statement._execution_options,
+            )
+
+        get back the Options and refresh "_sa_orm_load_options" in the
+        exec options dict w/ the Options as well
+
+        """
+
+        # common case is that no options we are looking for are
+        # in either dictionary, so cancel for that first
+        check_argnames = attrs.intersection(
+            set(exec_options).union(statement_exec_options)
+        )
+
+        existing_options = exec_options.get(key, cls)
+
+        if check_argnames:
+            result = {}
+            for argname in check_argnames:
+                local = "_" + argname
+                if argname in exec_options:
+                    result[local] = exec_options[argname]
+                elif argname in statement_exec_options:
+                    result[local] = statement_exec_options[argname]
+
+            new_options = existing_options + result
+            exec_options = util.immutabledict().merge_with(
+                exec_options, {key: new_options}
+            )
+            return new_options, exec_options
+
+        else:
+            return existing_options, exec_options
+
 
 class CacheableOptions(Options, HasCacheKey):
     @hybridmethod
diff --git a/lib/sqlalchemy/sql/coercions.py b/lib/sqlalchemy/sql/coercions.py
index be412c770..588c485ae 100644
--- a/lib/sqlalchemy/sql/coercions.py
+++ b/lib/sqlalchemy/sql/coercions.py
@@ -52,6 +52,18 @@ def _document_text_coercion(paramname, meth_rst, param_rst):
     )
 
 
+def _expression_collection_was_a_list(attrname, fnname, args):
+    if args and isinstance(args[0], (list, set)) and len(args) == 1:
+        util.warn_deprecated_20(
+            'The "%s" argument to %s() is now passed as a series of '
+            "positional "
+            "elements, rather than as a list. " % (attrname, fnname)
+        )
+        return args[0]
+    else:
+        return args
+
+
 def expect(role, element, apply_propagate_attrs=None, argname=None, **kw):
     if (
         role.allows_lambda
diff --git a/lib/sqlalchemy/sql/elements.py b/lib/sqlalchemy/sql/elements.py
index 6ce505412..c7e5aabcc 100644
--- a/lib/sqlalchemy/sql/elements.py
+++ b/lib/sqlalchemy/sql/elements.py
@@ -2573,10 +2573,8 @@ class Case(ColumnElement):
         stmt = select([users_table]).\
                     where(
                         case(
-                            [
-                                (users_table.c.name == 'wendy', 'W'),
-                                (users_table.c.name == 'jack', 'J')
-                            ],
+                            (users_table.c.name == 'wendy', 'W'),
+                            (users_table.c.name == 'jack', 'J'),
                             else_='E'
                         )
                     )
@@ -2597,7 +2595,10 @@ class Case(ColumnElement):
         ("else_", InternalTraversal.dp_clauseelement),
     ]
 
-    def __init__(self, whens, value=None, else_=None):
+    # TODO: for Py2k removal, this will be:
+    # def __init__(self, *whens, value=None, else_=None):
+
+    def __init__(self, *whens, **kw):
         r"""Produce a ``CASE`` expression.
 
         The ``CASE`` construct in SQL is a conditional object that
@@ -2612,10 +2613,8 @@ class Case(ColumnElement):
             stmt = select([users_table]).\
                         where(
                             case(
-                                [
-                                    (users_table.c.name == 'wendy', 'W'),
-                                    (users_table.c.name == 'jack', 'J')
-                                ],
+                                (users_table.c.name == 'wendy', 'W'),
+                                (users_table.c.name == 'jack', 'J'),
                                 else_='E'
                             )
                         )
@@ -2660,16 +2659,14 @@ class Case(ColumnElement):
             from sqlalchemy import case, literal_column
 
             case(
-                [
-                    (
-                        orderline.c.qty > 100,
-                        literal_column("'greaterthan100'")
-                    ),
-                    (
-                        orderline.c.qty > 10,
-                        literal_column("'greaterthan10'")
-                    )
-                ],
+                (
+                    orderline.c.qty > 100,
+                    literal_column("'greaterthan100'")
+                ),
+                (
+                    orderline.c.qty > 10,
+                    literal_column("'greaterthan10'")
+                ),
                 else_=literal_column("'lessthan10'")
             )
 
@@ -2683,19 +2680,23 @@ class Case(ColumnElement):
                 ELSE 'lessthan10'
             END
 
-        :param whens: The criteria to be compared against,
+        :param \*whens: The criteria to be compared against,
          :paramref:`.case.whens` accepts two different forms, based on
          whether or not :paramref:`.case.value` is used.
 
+         .. versionchanged:: 1.4 the :func:`_sql.case`
+            function now accepts the series of WHEN conditions positionally;
+            passing the expressions within a list is deprecated.
+
          In the first form, it accepts a list of 2-tuples; each 2-tuple
          consists of ``(<sql expression>, <value>)``, where the SQL
          expression is a boolean expression and "value" is a resulting value,
          e.g.::
 
-            case([
+            case(
                 (users_table.c.name == 'wendy', 'W'),
                 (users_table.c.name == 'jack', 'J')
-            ])
+            )
 
          In the second form, it accepts a Python dictionary of comparison
          values mapped to a resulting value; this form requires
@@ -2720,11 +2721,23 @@ class Case(ColumnElement):
 
         """
 
+        if "whens" in kw:
+            util.warn_deprecated_20(
+                'The "whens" argument to case() is now passed as a series of '
+                "positional "
+                "elements, rather than as a list. "
+            )
+            whens = kw.pop("whens")
+        else:
+            whens = coercions._expression_collection_was_a_list(
+                "whens", "case", whens
+            )
         try:
             whens = util.dictlike_iteritems(whens)
         except TypeError:
             pass
 
+        value = kw.pop("value", None)
         if value is not None:
             whenlist = [
                 (
@@ -2760,11 +2773,16 @@ class Case(ColumnElement):
 
         self.type = type_
         self.whens = whenlist
+
+        else_ = kw.pop("else_", None)
         if else_ is not None:
             self.else_ = coercions.expect(roles.ExpressionElementRole, else_)
         else:
             self.else_ = None
 
+        if kw:
+            raise TypeError("unknown arguments: %s" % (", ".join(sorted(kw))))
+
     @property
     def _from_objects(self):
         return list(
diff --git a/lib/sqlalchemy/sql/operators.py b/lib/sqlalchemy/sql/operators.py
index 85db88345..2d369cdf8 100644
--- a/lib/sqlalchemy/sql/operators.py
+++ b/lib/sqlalchemy/sql/operators.py
@@ -169,9 +169,6 @@ class Operators(object):
           :class:`.Boolean`, and those that do not will be of the same
           type as the left-hand operand.
 
-          .. versionadded:: 1.2.0b3 - added the
-             :paramref:`.Operators.op.return_type` argument.
-
         .. seealso::
 
             :ref:`types_operators`
@@ -194,8 +191,6 @@ class Operators(object):
         :paramref:`.Operators.op.is_comparison`
         flag with True.
 
-        .. versionadded:: 1.2.0b3
-
         .. seealso::
 
             :meth:`.Operators.op`
@@ -723,15 +718,6 @@ class ColumnOperators(Operators):
 
           With the value of ``:param`` as ``"foo/%bar"``.
 
-          .. versionadded:: 1.2
-
-          .. versionchanged:: 1.2.0 The
-            :paramref:`.ColumnOperators.startswith.autoescape` parameter is
-             now a simple boolean rather than a character; the escape
-             character itself is also escaped, and defaults to a forwards
-             slash, which itself can be customized using the
-             :paramref:`.ColumnOperators.startswith.escape` parameter.
-
         :param escape: a character which when given will render with the
           ``ESCAPE`` keyword to establish that character as the escape
           character.  This character can then be placed preceding occurrences
@@ -811,15 +797,6 @@ class ColumnOperators(Operators):
 
           With the value of ``:param`` as ``"foo/%bar"``.
 
-          .. versionadded:: 1.2
-
-          .. versionchanged:: 1.2.0 The
-            :paramref:`.ColumnOperators.endswith.autoescape` parameter is
-             now a simple boolean rather than a character; the escape
-             character itself is also escaped, and defaults to a forwards
-             slash, which itself can be customized using the
-             :paramref:`.ColumnOperators.endswith.escape` parameter.
-
         :param escape: a character which when given will render with the
           ``ESCAPE`` keyword to establish that character as the escape
           character.  This character can then be placed preceding occurrences
@@ -899,15 +876,6 @@ class ColumnOperators(Operators):
 
           With the value of ``:param`` as ``"foo/%bar"``.
 
-          .. versionadded:: 1.2
-
-          .. versionchanged:: 1.2.0 The
-            :paramref:`.ColumnOperators.contains.autoescape` parameter is
-             now a simple boolean rather than a character; the escape
-             character itself is also escaped, and defaults to a forwards
-             slash, which itself can be customized using the
-             :paramref:`.ColumnOperators.contains.escape` parameter.
-
         :param escape: a character which when given will render with the
           ``ESCAPE`` keyword to establish that character as the escape
           character.  This character can then be placed preceding occurrences
diff --git a/lib/sqlalchemy/sql/selectable.py b/lib/sqlalchemy/sql/selectable.py
index 12fcc00c3..1155c273b 100644
--- a/lib/sqlalchemy/sql/selectable.py
+++ b/lib/sqlalchemy/sql/selectable.py
@@ -2225,6 +2225,17 @@ class ForUpdateArg(ClauseElement):
         ("skip_locked", InternalTraversal.dp_boolean),
     ]
 
+    @classmethod
+    def _from_argument(cls, with_for_update):
+        if isinstance(with_for_update, ForUpdateArg):
+            return with_for_update
+        elif with_for_update in (None, False):
+            return None
+        elif with_for_update is True:
+            return ForUpdateArg()
+        else:
+            return ForUpdateArg(**with_for_update)
+
     def __eq__(self, other):
         return (
             isinstance(other, ForUpdateArg)
@@ -2699,6 +2710,12 @@ class SelectStatementGrouping(GroupedElement, SelectBase):
 
 
 class DeprecatedSelectBaseGenerations(object):
+    """A collection of methods available on :class:`_sql.Select` and
+    :class:`_sql.CompoundSelect`, these are all **deprecated** methods as they
+    modify the object in-place.
+
+    """
+
     @util.deprecated(
         "1.4",
         "The :meth:`_expression.GenerativeSelect.append_order_by` "
@@ -2740,9 +2757,6 @@ class DeprecatedSelectBaseGenerations(object):
         as it
         provides standard :term:`method chaining`.
 
-        .. seealso::
-
-            :meth:`_expression.GenerativeSelect.group_by`
 
         """
         self.group_by.non_generative(self, *clauses)
@@ -3353,6 +3367,12 @@ class CompoundSelect(HasCompileState, GenerativeSelect):
 
 
 class DeprecatedSelectGenerations(object):
+    """A collection of methods available on :class:`_sql.Select`, these
+    are all **deprecated** methods as they modify the :class:`_sql.Select`
+    object in -place.
+
+    """
+
     @util.deprecated(
         "1.4",
         "The :meth:`_expression.Select.append_correlation` "
@@ -3377,7 +3397,7 @@ class DeprecatedSelectGenerations(object):
         "1.4",
         "The :meth:`_expression.Select.append_column` method is deprecated "
         "and will be removed in a future release.  Use the generative "
-        "method :meth:`_expression.Select.column`.",
+        "method :meth:`_expression.Select.add_columns`.",
     )
     def append_column(self, column):
         """Append the given column expression to the columns clause of this
@@ -3388,14 +3408,10 @@ class DeprecatedSelectGenerations(object):
             my_select.append_column(some_table.c.new_column)
 
         This is an **in-place** mutation method; the
-        :meth:`_expression.Select.column` method is preferred,
+        :meth:`_expression.Select.add_columns` method is preferred,
         as it provides standard
         :term:`method chaining`.
 
-        See the documentation for :meth:`_expression.Select.with_only_columns`
-        for guidelines on adding /replacing the columns of a
-        :class:`_expression.Select` object.
-
         """
         self.add_columns.non_generative(self, column)
 
@@ -3501,6 +3517,21 @@ class SelectState(util.MemoizedSlots, CompileState):
 
         self.columns_plus_names = statement._generate_columns_plus_names(True)
 
+    @classmethod
+    def _plugin_not_implemented(cls):
+        raise NotImplementedError(
+            "The default SELECT construct without plugins does not "
+            "implement this method."
+        )
+
+    @classmethod
+    def get_column_descriptions(cls, statement):
+        cls._plugin_not_implemented()
+
+    @classmethod
+    def from_statement(cls, statement, from_statement):
+        cls._plugin_not_implemented()
+
     def _get_froms(self, statement):
         seen = set()
         froms = []
@@ -3805,6 +3836,15 @@ class Select(
 ):
     """Represents a ``SELECT`` statement.
 
+    The :class:`_sql.Select` object is normally constructed using the
+    :func:`_sql.select` function.  See that function for details.
+
+    .. seealso::
+
+        :func:`_sql.select`
+
+        :ref:`coretutorial_selecting` - in the Core tutorial
+
     """
 
     __visit_name__ = "select"
@@ -3821,7 +3861,7 @@ class Select(
     _from_obj = ()
     _auto_correlate = True
 
-    compile_options = SelectState.default_select_compile_options
+    _compile_options = SelectState.default_select_compile_options
 
     _traverse_internals = (
         [
@@ -3851,7 +3891,7 @@ class Select(
     )
 
     _cache_key_traversal = _traverse_internals + [
-        ("compile_options", InternalTraversal.dp_has_cache_key)
+        ("_compile_options", InternalTraversal.dp_has_cache_key)
     ]
 
     @classmethod
@@ -4274,12 +4314,35 @@ class Select(
     @property
     def column_descriptions(self):
         """Return a 'column descriptions' structure which may be
-        plugin-specific.
+        :term:`plugin-specific`.
 
         """
         meth = SelectState.get_plugin_class(self).get_column_descriptions
         return meth(self)
 
+    def from_statement(self, statement):
+        """Apply the columns which this :class:`.Select` would select
+        onto another statement.
+
+        This operation is :term:`plugin-specific` and will raise a not
+        supported exception if this :class:`_sql.Select` does not select from
+        plugin-enabled entities.
+
+
+        The statement is typically either a :func:`_expression.text` or
+        :func:`_expression.select` construct, and should return the set of
+        columns appropriate to the entities represented by this
+        :class:`.Select`.
+
+        .. seealso::
+
+            :ref:`orm_tutorial_literal_sql` - usage examples in the
+            ORM tutorial
+
+        """
+        meth = SelectState.get_plugin_class(self).from_statement
+        return meth(self, statement)
+
     @_generative
     def join(self, target, onclause=None, isouter=False, full=False):
         r"""Create a SQL JOIN against this :class:`_expresson.Select`
@@ -4550,7 +4613,7 @@ class Select(
         )
 
     @_generative
-    def with_only_columns(self, columns):
+    def with_only_columns(self, *columns):
         r"""Return a new :func:`_expression.select` construct with its columns
         clause replaced with the given columns.
 
@@ -4558,65 +4621,26 @@ class Select(
         :func:`_expression.select` had been called with the given columns
         clause.   I.e. a statement::
 
-            s = select([table1.c.a, table1.c.b])
-            s = s.with_only_columns([table1.c.b])
+            s = select(table1.c.a, table1.c.b)
+            s = s.with_only_columns(table1.c.b)
 
         should be exactly equivalent to::
 
-            s = select([table1.c.b])
-
-        This means that FROM clauses which are only derived
-        from the column list will be discarded if the new column
-        list no longer contains that FROM::
-
-            >>> table1 = table('t1', column('a'), column('b'))
-            >>> table2 = table('t2', column('a'), column('b'))
-            >>> s1 = select([table1.c.a, table2.c.b])
-            >>> print(s1)
-            SELECT t1.a, t2.b FROM t1, t2
-            >>> s2 = s1.with_only_columns([table2.c.b])
-            >>> print(s2)
-            SELECT t2.b FROM t1
-
-        The preferred way to maintain a specific FROM clause
-        in the construct, assuming it won't be represented anywhere
-        else (i.e. not in the WHERE clause, etc.) is to set it using
-        :meth:`_expression.Select.select_from`::
-
-            >>> s1 = select([table1.c.a, table2.c.b]).\
-            ...         select_from(table1.join(table2,
-            ...                 table1.c.a==table2.c.a))
-            >>> s2 = s1.with_only_columns([table2.c.b])
-            >>> print(s2)
-            SELECT t2.b FROM t1 JOIN t2 ON t1.a=t2.a
-
-        Care should also be taken to use the correct set of column objects
-        passed to :meth:`_expression.Select.with_only_columns`.
-        Since the method is
-        essentially equivalent to calling the :func:`_expression.select`
-        construct in the first place with the given columns, the columns passed
-        to :meth:`_expression.Select.with_only_columns`
-        should usually be a subset of
-        those which were passed to the :func:`_expression.select`
-        construct, not those which are available from the ``.c`` collection of
-        that :func:`_expression.select`.  That is::
-
-            s = select([table1.c.a, table1.c.b]).select_from(table1)
-            s = s.with_only_columns([table1.c.b])
-
-        and **not**::
-
-            # usually incorrect
-            s = s.with_only_columns([s.c.b])
-
-        The latter would produce the SQL::
-
-            SELECT b
-            FROM (SELECT t1.a AS a, t1.b AS b
-            FROM t1), t1
-
-        Since the :func:`_expression.select` construct is essentially
-        being asked to select both from ``table1`` as well as itself.
+            s = select(table1.c.b)
+
+        Note that this will also dynamically alter the FROM clause of the
+        statement if it is not explicitly stated.  To maintain the FROM
+        clause, ensure the :meth:`_sql.Select.select_from` method is
+        used appropriately::
+
+            s = select(table1.c.a, table2.c.b)
+            s = s.select_from(table2.c.b).with_only_columns(table1.c.a)
+
+        :param \*columns: column expressions to be used.
+
+         .. versionchanged:: 1.4 the :meth:`_sql.Select.with_only_columns`
+            method accepts the list of column expressions positionally;
+            passing the expressions as a list is deprecateed.
 
         """
 
@@ -4626,7 +4650,9 @@ class Select(
         self._assert_no_memoizations()
 
         rc = []
-        for c in columns:
+        for c in coercions._expression_collection_was_a_list(
+            "columns", "Select.with_only_columns", columns
+        ):
             c = coercions.expect(roles.ColumnsClauseRole, c,)
             # TODO: why are we doing this here?
             if isinstance(c, ScalarSelect):
diff --git a/lib/sqlalchemy/testing/assertions.py b/lib/sqlalchemy/testing/assertions.py
index 1ce59431e..ecc6a4ab8 100644
--- a/lib/sqlalchemy/testing/assertions.py
+++ b/lib/sqlalchemy/testing/assertions.py
@@ -404,6 +404,9 @@ class AssertsCompiledSQL(object):
 
         from sqlalchemy import orm
 
+        if isinstance(clause, orm.dynamic.AppenderQuery):
+            clause = clause._statement
+
         if isinstance(clause, orm.Query):
             compile_state = clause._compile_state()
             compile_state.statement._label_style = (