Convert execution to move through Session

This patch replaces the ORM execution flow with a
single pathway through Session.execute() for all queries,
including Core and ORM.

Currently included is full support for ORM Query,
Query.from_statement(), select(), as well as the
baked query and horizontal shard systems.  Initial
changes have also been made to the dogpile caching
example, which like baked query makes use of a
new ORM-specific execution hook that replaces the
use of both QueryEvents.before_compile() as well
as Query._execute_and_instances() as the central
ORM interception hooks.

select() and Query() constructs alike can be passed to
Session.execute() where they will return ORM
results in a Results object.   This API is currently
used internally by Query.   Full support for
Session.execute()->results to behave in a fully
2.0 fashion will be in later changesets.

bulk update/delete with ORM support will also
be delivered via the update() and delete()
constructs, however these have not yet been adapted
to the new system and may follow in a subsequent
update.

Performance is also beginning to lag as of this
commit and some previous ones.   It is hoped that
a few central functions such as the coercions
functions can be rewritten in C to re-gain
performance.  Additionally, query caching
is now available and some subsequent patches
will attempt to cache more of the per-execution
work from the ORM layer, e.g. column getters
and adapters.

This patch also contains initial "turn on" of the
caching system enginewide via the query_cache_size
parameter to create_engine(). Still defaulting at
zero for "no caching".   The caching system still
needs adjustments in order to gain adequate performance.

Change-Id: I047a7ebb26aa85dc01f6789fac2bff561dcd555d

12 files changed, 408 insertions, 215 deletions

diff --git a/lib/sqlalchemy/sql/annotation.py b/lib/sqlalchemy/sql/annotation.py
index 71d05f38f..08ed121d3 100644
--- a/lib/sqlalchemy/sql/annotation.py
+++ b/lib/sqlalchemy/sql/annotation.py
@@ -17,8 +17,12 @@ from .traversals import anon_map
 from .visitors import InternalTraversal
 from .. import util
 
+EMPTY_ANNOTATIONS = util.immutabledict()
+
 
 class SupportsAnnotations(object):
+    _annotations = EMPTY_ANNOTATIONS
+
     @util.memoized_property
     def _annotations_cache_key(self):
         anon_map_ = anon_map()
@@ -40,7 +44,6 @@ class SupportsAnnotations(object):
 
 
 class SupportsCloneAnnotations(SupportsAnnotations):
-    _annotations = util.immutabledict()
 
     _clone_annotations_traverse_internals = [
         ("_annotations", InternalTraversal.dp_annotations_key)
@@ -113,12 +116,9 @@ class SupportsWrappingAnnotations(SupportsAnnotations):
 
         """
         if clone:
-            # clone is used when we are also copying
-            # the expression for a deep deannotation
-            return self._clone()
+            s = self._clone()
+            return s
         else:
-            # if no clone, since we have no annotations we return
-            # self
             return self
 
 
@@ -163,12 +163,11 @@ class Annotated(object):
         self.__dict__.pop("_annotations_cache_key", None)
         self.__dict__.pop("_generate_cache_key", None)
         self.__element = element
-        self._annotations = values
+        self._annotations = util.immutabledict(values)
         self._hash = hash(element)
 
     def _annotate(self, values):
-        _values = self._annotations.copy()
-        _values.update(values)
+        _values = self._annotations.union(values)
         return self._with_annotations(_values)
 
     def _with_annotations(self, values):
@@ -183,10 +182,15 @@ class Annotated(object):
         if values is None:
             return self.__element
         else:
-            _values = self._annotations.copy()
-            for v in values:
-                _values.pop(v, None)
-            return self._with_annotations(_values)
+            return self._with_annotations(
+                util.immutabledict(
+                    {
+                        key: value
+                        for key, value in self._annotations.items()
+                        if key not in values
+                    }
+                )
+            )
 
     def _compiler_dispatch(self, visitor, **kw):
         return self.__element.__class__._compiler_dispatch(self, visitor, **kw)
diff --git a/lib/sqlalchemy/sql/base.py b/lib/sqlalchemy/sql/base.py
index 04cc34480..bb606a4d6 100644
--- a/lib/sqlalchemy/sql/base.py
+++ b/lib/sqlalchemy/sql/base.py
@@ -439,46 +439,53 @@ class CompileState(object):
     plugins = {}
 
     @classmethod
-    def _create(cls, statement, compiler, **kw):
+    def create_for_statement(cls, statement, compiler, **kw):
         # factory construction.
 
-        if statement._compile_state_plugin is not None:
-            constructor = cls.plugins.get(
-                (
-                    statement._compile_state_plugin,
-                    statement.__visit_name__,
-                    None,
-                ),
-                cls,
+        if statement._propagate_attrs:
+            plugin_name = statement._propagate_attrs.get(
+                "compile_state_plugin", "default"
             )
         else:
-            constructor = cls
+            plugin_name = "default"
+
+        klass = cls.plugins[(plugin_name, statement.__visit_name__)]
 
-        return constructor(statement, compiler, **kw)
+        if klass is cls:
+            return cls(statement, compiler, **kw)
+        else:
+            return klass.create_for_statement(statement, compiler, **kw)
 
     def __init__(self, statement, compiler, **kw):
         self.statement = statement
 
     @classmethod
-    def get_plugin_classmethod(cls, statement, name):
-        if statement._compile_state_plugin is not None:
-            fn = cls.plugins.get(
-                (
-                    statement._compile_state_plugin,
-                    statement.__visit_name__,
-                    name,
-                ),
-                None,
-            )
-            if fn is not None:
-                return fn
-        return getattr(cls, name)
+    def get_plugin_class(cls, statement):
+        plugin_name = statement._propagate_attrs.get(
+            "compile_state_plugin", "default"
+        )
+        try:
+            return cls.plugins[(plugin_name, statement.__visit_name__)]
+        except KeyError:
+            return None
 
     @classmethod
-    def plugin_for(cls, plugin_name, visit_name, method_name=None):
-        def decorate(fn):
-            cls.plugins[(plugin_name, visit_name, method_name)] = fn
-            return fn
+    def _get_plugin_compile_state_cls(cls, statement, plugin_name):
+        statement_plugin_name = statement._propagate_attrs.get(
+            "compile_state_plugin", "default"
+        )
+        if statement_plugin_name != plugin_name:
+            return None
+        try:
+            return cls.plugins[(plugin_name, statement.__visit_name__)]
+        except KeyError:
+            return None
+
+    @classmethod
+    def plugin_for(cls, plugin_name, visit_name):
+        def decorate(cls_to_decorate):
+            cls.plugins[(plugin_name, visit_name)] = cls_to_decorate
+            return cls_to_decorate
 
         return decorate
 
@@ -508,12 +515,12 @@ class InPlaceGenerative(HasMemoized):
 class HasCompileState(Generative):
     """A class that has a :class:`.CompileState` associated with it."""
 
-    _compile_state_factory = CompileState._create
-
     _compile_state_plugin = None
 
     _attributes = util.immutabledict()
 
+    _compile_state_factory = CompileState.create_for_statement
+
 
 class _MetaOptions(type):
     """metaclass for the Options class."""
@@ -549,6 +556,16 @@ class Options(util.with_metaclass(_MetaOptions)):
     def add_to_element(self, name, value):
         return self + {name: getattr(self, name) + value}
 
+    @hybridmethod
+    def _state_dict(self):
+        return self.__dict__
+
+    _state_dict_const = util.immutabledict()
+
+    @_state_dict.classlevel
+    def _state_dict(cls):
+        return cls._state_dict_const
+
 
 class CacheableOptions(Options, HasCacheKey):
     @hybridmethod
@@ -590,6 +607,9 @@ class Executable(Generative):
     def _disable_caching(self):
         self._cache_enable = HasCacheKey()
 
+    def _get_plugin_compile_state_cls(self, plugin_name):
+        return CompileState._get_plugin_compile_state_cls(self, plugin_name)
+
     @_generative
     def options(self, *options):
         """Apply options to this statement.
diff --git a/lib/sqlalchemy/sql/coercions.py b/lib/sqlalchemy/sql/coercions.py
index 2fc63b82f..d8ef0222a 100644
--- a/lib/sqlalchemy/sql/coercions.py
+++ b/lib/sqlalchemy/sql/coercions.py
@@ -50,19 +50,26 @@ def _document_text_coercion(paramname, meth_rst, param_rst):
     )
 
 
-def expect(role, element, **kw):
+def expect(role, element, apply_propagate_attrs=None, **kw):
     # major case is that we are given a ClauseElement already, skip more
     # elaborate logic up front if possible
     impl = _impl_lookup[role]
 
     if not isinstance(
         element,
-        (elements.ClauseElement, schema.SchemaItem, schema.FetchedValue,),
+        (elements.ClauseElement, schema.SchemaItem, schema.FetchedValue),
     ):
         resolved = impl._resolve_for_clause_element(element, **kw)
     else:
         resolved = element
 
+    if (
+        apply_propagate_attrs is not None
+        and not apply_propagate_attrs._propagate_attrs
+        and resolved._propagate_attrs
+    ):
+        apply_propagate_attrs._propagate_attrs = resolved._propagate_attrs
+
     if impl._role_class in resolved.__class__.__mro__:
         if impl._post_coercion:
             resolved = impl._post_coercion(resolved, **kw)
@@ -106,32 +113,32 @@ class RoleImpl(object):
         self.name = role_class._role_name
         self._use_inspection = issubclass(role_class, roles.UsesInspection)
 
-    def _resolve_for_clause_element(
-        self, element, argname=None, apply_plugins=None, **kw
-    ):
+    def _resolve_for_clause_element(self, element, argname=None, **kw):
         original_element = element
 
         is_clause_element = False
+
         while hasattr(element, "__clause_element__"):
             is_clause_element = True
             if not getattr(element, "is_clause_element", False):
                 element = element.__clause_element__()
             else:
-                break
-
-        should_apply_plugins = (
-            apply_plugins is not None
-            and apply_plugins._compile_state_plugin is None
-        )
+                return element
+
+        if not is_clause_element:
+            if self._use_inspection:
+                insp = inspection.inspect(element, raiseerr=False)
+                if insp is not None:
+                    insp._post_inspect
+                    try:
+                        element = insp.__clause_element__()
+                    except AttributeError:
+                        self._raise_for_expected(original_element, argname)
+                    else:
+                        return element
 
-        if is_clause_element:
-            if (
-                should_apply_plugins
-                and "compile_state_plugin" in element._annotations
-            ):
-                apply_plugins._compile_state_plugin = element._annotations[
-                    "compile_state_plugin"
-                ]
+            return self._literal_coercion(element, argname=argname, **kw)
+        else:
             return element
 
         if self._use_inspection:
@@ -142,14 +149,6 @@ class RoleImpl(object):
                     element = insp.__clause_element__()
                 except AttributeError:
                     self._raise_for_expected(original_element, argname)
-                else:
-                    if (
-                        should_apply_plugins
-                        and "compile_state_plugin" in element._annotations
-                    ):
-                        plugin = element._annotations["compile_state_plugin"]
-                        apply_plugins._compile_state_plugin = plugin
-                    return element
 
         return self._literal_coercion(element, argname=argname, **kw)
 
@@ -649,8 +648,8 @@ class SelectStatementImpl(_NoTextCoercion, RoleImpl):
             self._raise_for_expected(original_element, argname, resolved)
 
 
-class HasCTEImpl(ReturnsRowsImpl, roles.HasCTERole):
-    pass
+class HasCTEImpl(ReturnsRowsImpl):
+    __slots__ = ()
 
 
 class JoinTargetImpl(RoleImpl):
diff --git a/lib/sqlalchemy/sql/compiler.py b/lib/sqlalchemy/sql/compiler.py
index 9a7646743..8eae0ab7d 100644
--- a/lib/sqlalchemy/sql/compiler.py
+++ b/lib/sqlalchemy/sql/compiler.py
@@ -28,6 +28,7 @@ import contextlib
 import itertools
 import operator
 import re
+import time
 
 from . import base
 from . import coercions
@@ -380,6 +381,54 @@ class Compiled(object):
     sub-elements of the statement can modify these.
     """
 
+    compile_state = None
+    """Optional :class:`.CompileState` object that maintains additional
+    state used by the compiler.
+
+    Major executable objects such as :class:`_expression.Insert`,
+    :class:`_expression.Update`, :class:`_expression.Delete`,
+    :class:`_expression.Select` will generate this
+    state when compiled in order to calculate additional information about the
+    object.   For the top level object that is to be executed, the state can be
+    stored here where it can also have applicability towards result set
+    processing.
+
+    .. versionadded:: 1.4
+
+    """
+
+    _rewrites_selected_columns = False
+    """if True, indicates the compile_state object rewrites an incoming
+    ReturnsRows (like a Select) so that the columns we compile against in the
+    result set are not what were expressed on the outside.   this is a hint to
+    the execution context to not link the statement.selected_columns to the
+    columns mapped in the result object.
+
+    That is, when this flag is False::
+
+        stmt = some_statement()
+
+        result = conn.execute(stmt)
+        row = result.first()
+
+        # selected_columns are in a 1-1 relationship with the
+        # columns in the result, and are targetable in mapping
+        for col in stmt.selected_columns:
+            assert col in row._mapping
+
+    When True::
+
+        # selected columns are not what are in the rows.  the context
+        # rewrote the statement for some other set of selected_columns.
+        for col in stmt.selected_columns:
+            assert col not in row._mapping
+
+
+    """
+
+    cache_key = None
+    _gen_time = None
+
     def __init__(
         self,
         dialect,
@@ -433,6 +482,7 @@ class Compiled(object):
                 self.string = self.preparer._render_schema_translates(
                     self.string, schema_translate_map
                 )
+        self._gen_time = time.time()
 
     def _execute_on_connection(
         self, connection, multiparams, params, execution_options
@@ -637,28 +687,6 @@ class SQLCompiler(Compiled):
 
     insert_prefetch = update_prefetch = ()
 
-    compile_state = None
-    """Optional :class:`.CompileState` object that maintains additional
-    state used by the compiler.
-
-    Major executable objects such as :class:`_expression.Insert`,
-    :class:`_expression.Update`, :class:`_expression.Delete`,
-    :class:`_expression.Select` will generate this
-    state when compiled in order to calculate additional information about the
-    object.   For the top level object that is to be executed, the state can be
-    stored here where it can also have applicability towards result set
-    processing.
-
-    .. versionadded:: 1.4
-
-    """
-
-    compile_state_factories = util.immutabledict()
-    """Dictionary of alternate :class:`.CompileState` factories for given
-    classes, identified by their visit_name.
-
-    """
-
     def __init__(
         self,
         dialect,
@@ -667,7 +695,6 @@ class SQLCompiler(Compiled):
         column_keys=None,
         inline=False,
         linting=NO_LINTING,
-        compile_state_factories=None,
         **kwargs
     ):
         """Construct a new :class:`.SQLCompiler` object.
@@ -734,9 +761,6 @@ class SQLCompiler(Compiled):
         # dialect.label_length or dialect.max_identifier_length
         self.truncated_names = {}
 
-        if compile_state_factories:
-            self.compile_state_factories = compile_state_factories
-
         Compiled.__init__(self, dialect, statement, **kwargs)
 
         if (
@@ -1542,7 +1566,7 @@ class SQLCompiler(Compiled):
 
         compile_state = cs._compile_state_factory(cs, self, **kwargs)
 
-        if toplevel:
+        if toplevel and not self.compile_state:
             self.compile_state = compile_state
 
         entry = self._default_stack_entry if toplevel else self.stack[-1]
@@ -2541,6 +2565,13 @@ class SQLCompiler(Compiled):
             )
         return froms
 
+    translate_select_structure = None
+    """if none None, should be a callable which accepts (select_stmt, **kw)
+    and returns a select object.   this is used for structural changes
+    mostly to accommodate for LIMIT/OFFSET schemes
+
+    """
+
     def visit_select(
         self,
         select_stmt,
@@ -2552,7 +2583,17 @@ class SQLCompiler(Compiled):
         from_linter=None,
         **kwargs
     ):
+        assert select_wraps_for is None, (
+            "SQLAlchemy 1.4 requires use of "
+            "the translate_select_structure hook for structural "
+            "translations of SELECT objects"
+        )
 
+        # initial setup of SELECT.  the compile_state_factory may now
+        # be creating a totally different SELECT from the one that was
+        # passed in.  for ORM use this will convert from an ORM-state
+        # SELECT to a regular "Core" SELECT.  other composed operations
+        # such as computation of joins will be performed.
         compile_state = select_stmt._compile_state_factory(
             select_stmt, self, **kwargs
         )
@@ -2560,9 +2601,29 @@ class SQLCompiler(Compiled):
 
         toplevel = not self.stack
 
-        if toplevel:
+        if toplevel and not self.compile_state:
             self.compile_state = compile_state
 
+        # translate step for Oracle, SQL Server which often need to
+        # restructure the SELECT to allow for LIMIT/OFFSET and possibly
+        # other conditions
+        if self.translate_select_structure:
+            new_select_stmt = self.translate_select_structure(
+                select_stmt, asfrom=asfrom, **kwargs
+            )
+
+            # if SELECT was restructured, maintain a link to the originals
+            # and assemble a new compile state
+            if new_select_stmt is not select_stmt:
+                compile_state_wraps_for = compile_state
+                select_wraps_for = select_stmt
+                select_stmt = new_select_stmt
+
+                compile_state = select_stmt._compile_state_factory(
+                    select_stmt, self, **kwargs
+                )
+                select_stmt = compile_state.statement
+
         entry = self._default_stack_entry if toplevel else self.stack[-1]
 
         populate_result_map = need_column_expressions = (
@@ -2624,13 +2685,9 @@ class SQLCompiler(Compiled):
         ]
 
         if populate_result_map and select_wraps_for is not None:
-            # if this select is a compiler-generated wrapper,
+            # if this select was generated from translate_select,
             # rewrite the targeted columns in the result map
 
-            compile_state_wraps_for = select_wraps_for._compile_state_factory(
-                select_wraps_for, self, **kwargs
-            )
-
             translate = dict(
                 zip(
                     [
@@ -3013,7 +3070,8 @@ class SQLCompiler(Compiled):
 
         if toplevel:
             self.isinsert = True
-            self.compile_state = compile_state
+            if not self.compile_state:
+                self.compile_state = compile_state
 
         self.stack.append(
             {
diff --git a/lib/sqlalchemy/sql/dml.py b/lib/sqlalchemy/sql/dml.py
index 3dc4e917c..467a764d6 100644
--- a/lib/sqlalchemy/sql/dml.py
+++ b/lib/sqlalchemy/sql/dml.py
@@ -39,54 +39,8 @@ class DMLState(CompileState):
     isdelete = False
     isinsert = False
 
-    @classmethod
-    def _create_insert(cls, statement, compiler, **kw):
-        return DMLState(statement, compiler, isinsert=True, **kw)
-
-    @classmethod
-    def _create_update(cls, statement, compiler, **kw):
-        return DMLState(statement, compiler, isupdate=True, **kw)
-
-    @classmethod
-    def _create_delete(cls, statement, compiler, **kw):
-        return DMLState(statement, compiler, isdelete=True, **kw)
-
-    def __init__(
-        self,
-        statement,
-        compiler,
-        isinsert=False,
-        isupdate=False,
-        isdelete=False,
-        **kw
-    ):
-        self.statement = statement
-
-        if isupdate:
-            self.isupdate = True
-            self._preserve_parameter_order = (
-                statement._preserve_parameter_order
-            )
-            if statement._ordered_values is not None:
-                self._process_ordered_values(statement)
-            elif statement._values is not None:
-                self._process_values(statement)
-            elif statement._multi_values:
-                self._process_multi_values(statement)
-            self._extra_froms = self._make_extra_froms(statement)
-        elif isinsert:
-            self.isinsert = True
-            if statement._select_names:
-                self._process_select_values(statement)
-            if statement._values is not None:
-                self._process_values(statement)
-            if statement._multi_values:
-                self._process_multi_values(statement)
-        elif isdelete:
-            self.isdelete = True
-            self._extra_froms = self._make_extra_froms(statement)
-        else:
-            assert False, "one of isinsert, isupdate, or isdelete must be set"
+    def __init__(self, statement, compiler, **kw):
+        raise NotImplementedError()
 
     def _make_extra_froms(self, statement):
         froms = []
@@ -174,6 +128,51 @@ class DMLState(CompileState):
         )
 
 
+@CompileState.plugin_for("default", "insert")
+class InsertDMLState(DMLState):
+    isinsert = True
+
+    def __init__(self, statement, compiler, **kw):
+        self.statement = statement
+
+        self.isinsert = True
+        if statement._select_names:
+            self._process_select_values(statement)
+        if statement._values is not None:
+            self._process_values(statement)
+        if statement._multi_values:
+            self._process_multi_values(statement)
+
+
+@CompileState.plugin_for("default", "update")
+class UpdateDMLState(DMLState):
+    isupdate = True
+
+    def __init__(self, statement, compiler, **kw):
+        self.statement = statement
+
+        self.isupdate = True
+        self._preserve_parameter_order = statement._preserve_parameter_order
+        if statement._ordered_values is not None:
+            self._process_ordered_values(statement)
+        elif statement._values is not None:
+            self._process_values(statement)
+        elif statement._multi_values:
+            self._process_multi_values(statement)
+        self._extra_froms = self._make_extra_froms(statement)
+
+
+@CompileState.plugin_for("default", "delete")
+class DeleteDMLState(DMLState):
+    isdelete = True
+
+    def __init__(self, statement, compiler, **kw):
+        self.statement = statement
+
+        self.isdelete = True
+        self._extra_froms = self._make_extra_froms(statement)
+
+
 class UpdateBase(
     roles.DMLRole,
     HasCTE,
@@ -754,8 +753,6 @@ class Insert(ValuesBase):
 
     _supports_multi_parameters = True
 
-    _compile_state_factory = DMLState._create_insert
-
     select = None
     include_insert_from_select_defaults = False
 
@@ -964,8 +961,6 @@ class Update(DMLWhereBase, ValuesBase):
 
     __visit_name__ = "update"
 
-    _compile_state_factory = DMLState._create_update
-
     _traverse_internals = (
         [
             ("table", InternalTraversal.dp_clauseelement),
@@ -1210,8 +1205,6 @@ class Delete(DMLWhereBase, UpdateBase):
 
     __visit_name__ = "delete"
 
-    _compile_state_factory = DMLState._create_delete
-
     _traverse_internals = (
         [
             ("table", InternalTraversal.dp_clauseelement),
diff --git a/lib/sqlalchemy/sql/elements.py b/lib/sqlalchemy/sql/elements.py
index c1bc9edbc..287e53724 100644
--- a/lib/sqlalchemy/sql/elements.py
+++ b/lib/sqlalchemy/sql/elements.py
@@ -191,7 +191,12 @@ class ClauseElement(
 
     __visit_name__ = "clause"
 
-    _annotations = {}
+    _propagate_attrs = util.immutabledict()
+    """like annotations, however these propagate outwards liberally
+    as SQL constructs are built, and are set up at construction time.
+
+    """
+
     supports_execution = False
     _from_objects = []
     bind = None
@@ -215,6 +220,16 @@ class ClauseElement(
 
     _cache_key_traversal = None
 
+    def _set_propagate_attrs(self, values):
+        # usually, self._propagate_attrs is empty here.  one case where it's
+        # not is a subquery against ORM select, that is then pulled as a
+        # property of an aliased class.   should all be good
+
+        # assert not self._propagate_attrs
+
+        self._propagate_attrs = util.immutabledict(values)
+        return self
+
     def _clone(self):
         """Create a shallow copy of this ClauseElement.
 
@@ -870,6 +885,7 @@ class ColumnElement(
             type_=getattr(self, "type", None),
             _selectable=selectable,
         )
+        co._propagate_attrs = selectable._propagate_attrs
         co._proxies = [self]
         if selectable._is_clone_of is not None:
             co._is_clone_of = selectable._is_clone_of.columns.get(key)
@@ -1495,6 +1511,8 @@ class TextClause(
 
     _render_label_in_columns_clause = False
 
+    _hide_froms = ()
+
     def __and__(self, other):
         # support use in select.where(), query.filter()
         return and_(self, other)
@@ -1509,10 +1527,6 @@ class TextClause(
 
     _allow_label_resolve = False
 
-    @property
-    def _hide_froms(self):
-        return []
-
     def __init__(self, text, bind=None):
         self._bind = bind
         self._bindparams = {}
@@ -2093,14 +2107,16 @@ class ClauseList(
         )
         if self.group_contents:
             self.clauses = [
-                coercions.expect(text_converter_role, clause).self_group(
-                    against=self.operator
-                )
+                coercions.expect(
+                    text_converter_role, clause, apply_propagate_attrs=self
+                ).self_group(against=self.operator)
                 for clause in clauses
             ]
         else:
             self.clauses = [
-                coercions.expect(text_converter_role, clause)
+                coercions.expect(
+                    text_converter_role, clause, apply_propagate_attrs=self
+                )
                 for clause in clauses
             ]
         self._is_implicitly_boolean = operators.is_boolean(self.operator)
@@ -2641,7 +2657,9 @@ class Case(ColumnElement):
             whenlist = [
                 (
                     coercions.expect(
-                        roles.ExpressionElementRole, c
+                        roles.ExpressionElementRole,
+                        c,
+                        apply_propagate_attrs=self,
                     ).self_group(),
                     coercions.expect(roles.ExpressionElementRole, r),
                 )
@@ -2650,7 +2668,9 @@ class Case(ColumnElement):
         else:
             whenlist = [
                 (
-                    coercions.expect(roles.ColumnArgumentRole, c).self_group(),
+                    coercions.expect(
+                        roles.ColumnArgumentRole, c, apply_propagate_attrs=self
+                    ).self_group(),
                     coercions.expect(roles.ExpressionElementRole, r),
                 )
                 for (c, r) in whens
@@ -2805,7 +2825,10 @@ class Cast(WrapsColumnExpression, ColumnElement):
         """
         self.type = type_api.to_instance(type_)
         self.clause = coercions.expect(
-            roles.ExpressionElementRole, expression, type_=self.type
+            roles.ExpressionElementRole,
+            expression,
+            type_=self.type,
+            apply_propagate_attrs=self,
         )
         self.typeclause = TypeClause(self.type)
 
@@ -2906,7 +2929,10 @@ class TypeCoerce(WrapsColumnExpression, ColumnElement):
         """
         self.type = type_api.to_instance(type_)
         self.clause = coercions.expect(
-            roles.ExpressionElementRole, expression, type_=self.type
+            roles.ExpressionElementRole,
+            expression,
+            type_=self.type,
+            apply_propagate_attrs=self,
         )
 
     @property
@@ -3031,6 +3057,7 @@ class UnaryExpression(ColumnElement):
     ):
         self.operator = operator
         self.modifier = modifier
+        self._propagate_attrs = element._propagate_attrs
         self.element = element.self_group(
             against=self.operator or self.modifier
         )
@@ -3474,6 +3501,7 @@ class BinaryExpression(ColumnElement):
         if isinstance(operator, util.string_types):
             operator = operators.custom_op(operator)
         self._orig = (left.__hash__(), right.__hash__())
+        self._propagate_attrs = left._propagate_attrs or right._propagate_attrs
         self.left = left.self_group(against=operator)
         self.right = right.self_group(against=operator)
         self.operator = operator
@@ -4159,6 +4187,7 @@ class Label(roles.LabeledColumnExprRole, ColumnElement):
             name=name if name else self.name,
             disallow_is_literal=True,
         )
+        e._propagate_attrs = selectable._propagate_attrs
         e._proxies.append(self)
         if self._type is not None:
             e.type = self._type
@@ -4340,16 +4369,10 @@ class ColumnClause(
             return other.proxy_set.intersection(self.proxy_set)
 
     def get_children(self, column_tables=False, **kw):
-        if column_tables and self.table is not None:
-            # TODO: this is only used by ORM query deep_entity_zero.
-            # this is being removed in a later release so remove
-            # column_tables also at that time.
-            return [self.table]
-        else:
-            # override base get_children() to not return the Table
-            # or selectable that is parent to this column.  Traversals
-            # expect the columns of tables and subqueries to be leaf nodes.
-            return []
+        # override base get_children() to not return the Table
+        # or selectable that is parent to this column.  Traversals
+        # expect the columns of tables and subqueries to be leaf nodes.
+        return []
 
     @HasMemoized.memoized_attribute
     def _from_objects(self):
@@ -4474,6 +4497,7 @@ class ColumnClause(
             _selectable=selectable,
             is_literal=is_literal,
         )
+        c._propagate_attrs = selectable._propagate_attrs
         if name is None:
             c.key = self.key
         c._proxies = [self]
diff --git a/lib/sqlalchemy/sql/functions.py b/lib/sqlalchemy/sql/functions.py
index cedb76f55..6b1172eba 100644
--- a/lib/sqlalchemy/sql/functions.py
+++ b/lib/sqlalchemy/sql/functions.py
@@ -107,6 +107,7 @@ class FunctionElement(Executable, ColumnElement, FromClause):
                 roles.ExpressionElementRole,
                 c,
                 name=getattr(self, "name", None),
+                apply_propagate_attrs=self,
             )
             for c in clauses
         ]
@@ -749,7 +750,10 @@ class GenericFunction(util.with_metaclass(_GenericMeta, Function)):
         if parsed_args is None:
             parsed_args = [
                 coercions.expect(
-                    roles.ExpressionElementRole, c, name=self.name
+                    roles.ExpressionElementRole,
+                    c,
+                    name=self.name,
+                    apply_propagate_attrs=self,
                 )
                 for c in args
             ]
@@ -813,7 +817,12 @@ class ReturnTypeFromArgs(GenericFunction):
 
     def __init__(self, *args, **kwargs):
         args = [
-            coercions.expect(roles.ExpressionElementRole, c, name=self.name)
+            coercions.expect(
+                roles.ExpressionElementRole,
+                c,
+                name=self.name,
+                apply_propagate_attrs=self,
+            )
             for c in args
         ]
         kwargs.setdefault("type_", _type_from_args(args))
@@ -944,7 +953,12 @@ class array_agg(GenericFunction):
     type = sqltypes.ARRAY
 
     def __init__(self, *args, **kwargs):
-        args = [coercions.expect(roles.ExpressionElementRole, c) for c in args]
+        args = [
+            coercions.expect(
+                roles.ExpressionElementRole, c, apply_propagate_attrs=self
+            )
+            for c in args
+        ]
 
         default_array_type = kwargs.pop("_default_array_type", sqltypes.ARRAY)
         if "type_" not in kwargs:
diff --git a/lib/sqlalchemy/sql/roles.py b/lib/sqlalchemy/sql/roles.py
index b861f721b..d0f4fef60 100644
--- a/lib/sqlalchemy/sql/roles.py
+++ b/lib/sqlalchemy/sql/roles.py
@@ -142,12 +142,20 @@ class AnonymizedFromClauseRole(StrictFromClauseRole):
 
 
 class CoerceTextStatementRole(SQLRole):
-    _role_name = "Executable SQL, text() construct, or string statement"
+    _role_name = "Executable SQL or text() construct"
+
+
+#    _executable_statement = None
 
 
 class StatementRole(CoerceTextStatementRole):
     _role_name = "Executable SQL or text() construct"
 
+    _is_future = False
+
+    def _get_plugin_compile_state_cls(self, name):
+        return None
+
 
 class ReturnsRowsRole(StatementRole):
     _role_name = (
diff --git a/lib/sqlalchemy/sql/schema.py b/lib/sqlalchemy/sql/schema.py
index 65f8bd81c..263f579de 100644
--- a/lib/sqlalchemy/sql/schema.py
+++ b/lib/sqlalchemy/sql/schema.py
@@ -1632,6 +1632,7 @@ class Column(DialectKWArgs, SchemaItem, ColumnClause):
             )
 
         c.table = selectable
+        c._propagate_attrs = selectable._propagate_attrs
         if selectable._is_clone_of is not None:
             c._is_clone_of = selectable._is_clone_of.columns.get(c.key)
         if self.primary_key:
diff --git a/lib/sqlalchemy/sql/selectable.py b/lib/sqlalchemy/sql/selectable.py
index 6a552c18c..008959aec 100644
--- a/lib/sqlalchemy/sql/selectable.py
+++ b/lib/sqlalchemy/sql/selectable.py
@@ -1342,7 +1342,9 @@ class AliasedReturnsRows(NoInit, FromClause):
         raise NotImplementedError()
 
     def _init(self, selectable, name=None):
-        self.element = selectable
+        self.element = coercions.expect(
+            roles.ReturnsRowsRole, selectable, apply_propagate_attrs=self
+        )
         self.supports_execution = selectable.supports_execution
         if self.supports_execution:
             self._execution_options = selectable._execution_options
@@ -3026,6 +3028,7 @@ class GenerativeSelect(DeprecatedSelectBaseGenerations, SelectBase):
             )
 
 
+@CompileState.plugin_for("default", "compound_select")
 class CompoundSelectState(CompileState):
     @util.memoized_property
     def _label_resolve_dict(self):
@@ -3058,7 +3061,6 @@ class CompoundSelect(HasCompileState, GenerativeSelect):
     """
 
     __visit_name__ = "compound_select"
-    _compile_state_factory = CompoundSelectState._create
 
     _traverse_internals = [
         ("selects", InternalTraversal.dp_clauseelement_list),
@@ -3425,6 +3427,7 @@ class DeprecatedSelectGenerations(object):
         self.select_from.non_generative(self, fromclause)
 
 
+@CompileState.plugin_for("default", "select")
 class SelectState(CompileState):
     class default_select_compile_options(CacheableOptions):
         _cache_key_traversal = []
@@ -3462,7 +3465,7 @@ class SelectState(CompileState):
                 )
             if not seen.intersection(item._cloned_set):
                 froms.append(item)
-            seen.update(item._cloned_set)
+                seen.update(item._cloned_set)
 
         return froms
 
@@ -3714,12 +3717,29 @@ class SelectState(CompileState):
         return replace_from_obj_index
 
 
+class _SelectFromElements(object):
+    def _iterate_from_elements(self):
+        # note this does not include elements
+        # in _setup_joins or _legacy_setup_joins
+
+        return itertools.chain(
+            itertools.chain.from_iterable(
+                [element._from_objects for element in self._raw_columns]
+            ),
+            itertools.chain.from_iterable(
+                [element._from_objects for element in self._where_criteria]
+            ),
+            self._from_obj,
+        )
+
+
 class Select(
     HasPrefixes,
     HasSuffixes,
     HasHints,
     HasCompileState,
     DeprecatedSelectGenerations,
+    _SelectFromElements,
     GenerativeSelect,
 ):
     """Represents a ``SELECT`` statement.
@@ -3728,7 +3748,6 @@ class Select(
 
     __visit_name__ = "select"
 
-    _compile_state_factory = SelectState._create
     _is_future = False
     _setup_joins = ()
     _legacy_setup_joins = ()
@@ -4047,7 +4066,7 @@ class Select(
         if cols_present:
             self._raw_columns = [
                 coercions.expect(
-                    roles.ColumnsClauseRole, c, apply_plugins=self
+                    roles.ColumnsClauseRole, c, apply_propagate_attrs=self
                 )
                 for c in columns
             ]
@@ -4073,17 +4092,6 @@ class Select(
         cols = list(elem._select_iterable)
         return cols[0].type
 
-    def _iterate_from_elements(self):
-        return itertools.chain(
-            itertools.chain.from_iterable(
-                [element._from_objects for element in self._raw_columns]
-            ),
-            itertools.chain.from_iterable(
-                [element._from_objects for element in self._where_criteria]
-            ),
-            self._from_obj,
-        )
-
     @property
     def froms(self):
         """Return the displayed list of FromClause elements."""
@@ -4192,14 +4200,16 @@ class Select(
 
         self._raw_columns = self._raw_columns + [
             coercions.expect(
-                roles.ColumnsClauseRole, column, apply_plugins=self
+                roles.ColumnsClauseRole, column, apply_propagate_attrs=self
             )
             for column in columns
         ]
 
     def _set_entities(self, entities):
         self._raw_columns = [
-            coercions.expect(roles.ColumnsClauseRole, ent, apply_plugins=self)
+            coercions.expect(
+                roles.ColumnsClauseRole, ent, apply_propagate_attrs=self
+            )
             for ent in util.to_list(entities)
         ]
 
@@ -4342,14 +4352,24 @@ class Select(
         self._raw_columns = rc
 
     @property
-    def _whereclause(self):
-        """Legacy, return the WHERE clause as a """
-        """:class:`_expression.BooleanClauseList`"""
+    def whereclause(self):
+        """Return the completed WHERE clause for this :class:`.Select`
+        statement.
+
+        This assembles the current collection of WHERE criteria
+        into a single :class:`_expression.BooleanClauseList` construct.
+
+
+        .. versionadded:: 1.4
+
+        """
 
         return BooleanClauseList._construct_for_whereclause(
             self._where_criteria
         )
 
+    _whereclause = whereclause
+
     @_generative
     def where(self, whereclause):
         """return a new select() construct with the given expression added to
@@ -4430,7 +4450,7 @@ class Select(
 
         self._from_obj += tuple(
             coercions.expect(
-                roles.FromClauseRole, fromclause, apply_plugins=self
+                roles.FromClauseRole, fromclause, apply_propagate_attrs=self
             )
             for fromclause in froms
         )
diff --git a/lib/sqlalchemy/sql/traversals.py b/lib/sqlalchemy/sql/traversals.py
index a308feb7c..482248ada 100644
--- a/lib/sqlalchemy/sql/traversals.py
+++ b/lib/sqlalchemy/sql/traversals.py
@@ -179,7 +179,7 @@ class HasCacheKey(object):
         if NO_CACHE in _anon_map:
             return None
         else:
-            return CacheKey(key, bindparams)
+            return CacheKey(key, bindparams, self)
 
     @classmethod
     def _generate_cache_key_for_object(cls, obj):
@@ -190,7 +190,7 @@ class HasCacheKey(object):
         if NO_CACHE in _anon_map:
             return None
         else:
-            return CacheKey(key, bindparams)
+            return CacheKey(key, bindparams, obj)
 
 
 class MemoizedHasCacheKey(HasCacheKey, HasMemoized):
@@ -199,9 +199,42 @@ class MemoizedHasCacheKey(HasCacheKey, HasMemoized):
         return HasCacheKey._generate_cache_key(self)
 
 
-class CacheKey(namedtuple("CacheKey", ["key", "bindparams"])):
+class CacheKey(namedtuple("CacheKey", ["key", "bindparams", "statement"])):
     def __hash__(self):
-        return hash(self.key)
+        """CacheKey itself is not hashable - hash the .key portion"""
+
+        return None
+
+    def to_offline_string(self, statement_cache, parameters):
+        """generate an "offline string" form of this :class:`.CacheKey`
+
+        The "offline string" is basically the string SQL for the
+        statement plus a repr of the bound parameter values in series.
+        Whereas the :class:`.CacheKey` object is dependent on in-memory
+        identities in order to work as a cache key, the "offline" version
+        is suitable for a cache that will work for other processes as well.
+
+        The given "statement_cache" is a dictionary-like object where the
+        string form of the statement itself will be cached.  this dictionary
+        should be in a longer lived scope in order to reduce the time spent
+        stringifying statements.
+
+
+        """
+        if self.key not in statement_cache:
+            statement_cache[self.key] = sql_str = str(self.statement)
+        else:
+            sql_str = statement_cache[self.key]
+
+        return repr(
+            (
+                sql_str,
+                tuple(
+                    parameters.get(bindparam.key, bindparam.value)
+                    for bindparam in self.bindparams
+                ),
+            )
+        )
 
     def __eq__(self, other):
         return self.key == other.key
@@ -411,7 +444,6 @@ class _CacheKey(ExtendedInternalTraversal):
     def visit_setup_join_tuple(
         self, attrname, obj, parent, anon_map, bindparams
     ):
-        # TODO: look at attrname for "legacy_join" and use different structure
         return tuple(
             (
                 target._gen_cache_key(anon_map, bindparams),
@@ -596,7 +628,6 @@ class _CopyInternals(InternalTraversal):
     def visit_setup_join_tuple(
         self, attrname, parent, element, clone=_clone, **kw
     ):
-        # TODO: look at attrname for "legacy_join" and use different structure
         return tuple(
             (
                 clone(target, **kw) if target is not None else None,
@@ -668,6 +699,15 @@ class _CopyInternals(InternalTraversal):
 _copy_internals = _CopyInternals()
 
 
+def _flatten_clauseelement(element):
+    while hasattr(element, "__clause_element__") and not getattr(
+        element, "is_clause_element", False
+    ):
+        element = element.__clause_element__()
+
+    return element
+
+
 class _GetChildren(InternalTraversal):
     """Generate a _children_traversal internal traversal dispatch for classes
     with a _traverse_internals collection."""
@@ -696,6 +736,17 @@ class _GetChildren(InternalTraversal):
     def visit_clauseelement_unordered_set(self, element, **kw):
         return element
 
+    def visit_setup_join_tuple(self, element, **kw):
+        for (target, onclause, from_, flags) in element:
+            if from_ is not None:
+                yield from_
+
+            if not isinstance(target, str):
+                yield _flatten_clauseelement(target)
+
+    #            if onclause is not None and not isinstance(onclause, str):
+    #                yield _flatten_clauseelement(onclause)
+
     def visit_dml_ordered_values(self, element, **kw):
         for k, v in element:
             if hasattr(k, "__clause_element__"):
diff --git a/lib/sqlalchemy/sql/visitors.py b/lib/sqlalchemy/sql/visitors.py
index 030fd2fde..683f545dd 100644
--- a/lib/sqlalchemy/sql/visitors.py
+++ b/lib/sqlalchemy/sql/visitors.py
@@ -591,6 +591,7 @@ def iterate(obj, opts=util.immutabledict()):
     """
     yield obj
     children = obj.get_children(**opts)
+
     if not children:
         return