Try to measure new style caching in the ORM, take two

Supercedes: If78fbb557c6f2cae637799c3fec2cbc5ac248aaf

Trying to see if by making the cache key memoized, we
still can have the older "identity" form of caching
which is the cheapest of all, at the same time as the
newer "cache key each time" version that is not nearly
as cheap; but still much cheaper than no caching at all.

Also needed is a per-execution update of _keymap when
we invoke from a cached select, so that Column objects
that are anonymous or otherwise adapted will match up.
this is analogous to the adaption of bound parameters
from the cache key.

Adds test coverage for the keymap / construct_params()
 changes related to caching.  Also hones performance
to a large extent for statement construction and
cache key generation.

Also includes a new memoized attribute
approach that vastly simplifies the previous approach
of "group_expirable_memoized_property" and finally
integrates cleanly with _clone(), _generate(), etc.
no more hardcoding of attributes is needed, as well
as that most _reset_memoization() calls are no longer
needed as the reset is inherent in a _generate() call;
this also has dramatic performance improvements.

Change-Id: I95c560ffcbfa30b26644999412fb6a385125f663

14 files changed, 305 insertions, 132 deletions

diff --git a/lib/sqlalchemy/sql/annotation.py b/lib/sqlalchemy/sql/annotation.py
index 7984dc7ea..d895e730c 100644
--- a/lib/sqlalchemy/sql/annotation.py
+++ b/lib/sqlalchemy/sql/annotation.py
@@ -13,6 +13,7 @@ associations.
 
 from . import operators
 from .base import HasCacheKey
+from .traversals import anon_map
 from .visitors import InternalTraversal
 from .. import util
 
@@ -20,12 +21,13 @@ from .. import util
 class SupportsAnnotations(object):
     @util.memoized_property
     def _annotations_cache_key(self):
+        anon_map_ = anon_map()
         return (
             "_annotations",
             tuple(
                 (
                     key,
-                    value._gen_cache_key(None, [])
+                    value._gen_cache_key(anon_map_, [])
                     if isinstance(value, HasCacheKey)
                     else value,
                 )
@@ -38,7 +40,7 @@ class SupportsCloneAnnotations(SupportsAnnotations):
     _annotations = util.immutabledict()
 
     _clone_annotations_traverse_internals = [
-        ("_annotations_cache_key", InternalTraversal.dp_plain_obj)
+        ("_annotations", InternalTraversal.dp_annotations_key)
     ]
 
     def _annotate(self, values):
@@ -133,6 +135,8 @@ class Annotated(object):
 
     """
 
+    _is_column_operators = False
+
     def __new__(cls, *args):
         if not args:
             # clone constructor
@@ -200,7 +204,7 @@ class Annotated(object):
         return self._hash
 
     def __eq__(self, other):
-        if isinstance(self.__element, operators.ColumnOperators):
+        if self._is_column_operators:
             return self.__element.__class__.__eq__(self, other)
         else:
             return hash(other) == hash(self)
@@ -208,7 +212,9 @@ class Annotated(object):
 
 # hard-generate Annotated subclasses.  this technique
 # is used instead of on-the-fly types (i.e. type.__new__())
-# so that the resulting objects are pickleable.
+# so that the resulting objects are pickleable; additionally, other
+# decisions can be made up front about the type of object being annotated
+# just once per class rather than per-instance.
 annotated_classes = {}
 
 
@@ -310,8 +316,11 @@ def _new_annotation_type(cls, base_cls):
 
     if "_traverse_internals" in cls.__dict__:
         anno_cls._traverse_internals = list(cls._traverse_internals) + [
-            ("_annotations_cache_key", InternalTraversal.dp_plain_obj)
+            ("_annotations", InternalTraversal.dp_annotations_key)
         ]
+
+    anno_cls._is_column_operators = issubclass(cls, operators.ColumnOperators)
+
     return anno_cls
 
 
diff --git a/lib/sqlalchemy/sql/base.py b/lib/sqlalchemy/sql/base.py
index 974ca6ddb..eea4003f2 100644
--- a/lib/sqlalchemy/sql/base.py
+++ b/lib/sqlalchemy/sql/base.py
@@ -19,6 +19,7 @@ from .visitors import ClauseVisitor
 from .visitors import InternalTraversal
 from .. import exc
 from .. import util
+from ..util import HasMemoized
 
 if util.TYPE_CHECKING:
     from types import ModuleType
@@ -58,18 +59,6 @@ class SingletonConstant(Immutable):
         cls._singleton = obj
 
 
-class HasMemoized(object):
-    def _reset_memoizations(self):
-        self._memoized_property.expire_instance(self)
-
-    def _reset_exported(self):
-        self._memoized_property.expire_instance(self)
-
-    def _copy_internals(self, **kw):
-        super(HasMemoized, self)._copy_internals(**kw)
-        self._reset_memoizations()
-
-
 def _from_objects(*elements):
     return itertools.chain.from_iterable(
         [element._from_objects for element in elements]
@@ -461,13 +450,14 @@ class CompileState(object):
         self.statement = statement
 
 
-class Generative(object):
+class Generative(HasMemoized):
     """Provide a method-chaining pattern in conjunction with the
     @_generative decorator."""
 
     def _generate(self):
+        skip = self._memoized_keys
         s = self.__class__.__new__(self.__class__)
-        s.__dict__ = self.__dict__.copy()
+        s.__dict__ = {k: v for k, v in self.__dict__.items() if k not in skip}
         return s
 
 
diff --git a/lib/sqlalchemy/sql/coercions.py b/lib/sqlalchemy/sql/coercions.py
index 679d9c6e9..e605b486b 100644
--- a/lib/sqlalchemy/sql/coercions.py
+++ b/lib/sqlalchemy/sql/coercions.py
@@ -320,11 +320,7 @@ class BinaryElementImpl(
             self._raise_for_expected(element, err=err)
 
     def _post_coercion(self, resolved, expr, **kw):
-        if (
-            isinstance(resolved, (elements.Grouping, elements.BindParameter))
-            and resolved.type._isnull
-            and not expr.type._isnull
-        ):
+        if resolved.type._isnull and not expr.type._isnull:
             resolved = resolved._with_binary_element_type(expr.type)
         return resolved
 
diff --git a/lib/sqlalchemy/sql/compiler.py b/lib/sqlalchemy/sql/compiler.py
index 87ae5232e..799fca2f5 100644
--- a/lib/sqlalchemy/sql/compiler.py
+++ b/lib/sqlalchemy/sql/compiler.py
@@ -470,7 +470,7 @@ class Compiled(object):
 
         return self.string or ""
 
-    def construct_params(self, params=None):
+    def construct_params(self, params=None, extracted_parameters=None):
         """Return the bind params for this compiled object.
 
         :param params: a dict of string/object pairs whose values will
@@ -664,6 +664,7 @@ class SQLCompiler(Compiled):
         self,
         dialect,
         statement,
+        cache_key=None,
         column_keys=None,
         inline=False,
         linting=NO_LINTING,
@@ -687,6 +688,8 @@ class SQLCompiler(Compiled):
         """
         self.column_keys = column_keys
 
+        self.cache_key = cache_key
+
         # compile INSERT/UPDATE defaults/sequences inlined (no pre-
         # execute)
         self.inline = inline or getattr(statement, "_inline", False)
@@ -818,9 +821,38 @@ class SQLCompiler(Compiled):
     def sql_compiler(self):
         return self
 
-    def construct_params(self, params=None, _group_number=None, _check=True):
+    def construct_params(
+        self,
+        params=None,
+        _group_number=None,
+        _check=True,
+        extracted_parameters=None,
+    ):
         """return a dictionary of bind parameter keys and values"""
 
+        if extracted_parameters:
+            # related the bound parameters collected in the original cache key
+            # to those collected in the incoming cache key.  They will not have
+            # matching names but they will line up positionally in the same
+            # way.   The parameters present in self.bind_names may be clones of
+            # these original cache key params in the case of DML but the .key
+            # will be guaranteed to match.
+            try:
+                orig_extracted = self.cache_key[1]
+            except TypeError as err:
+                util.raise_(
+                    exc.CompileError(
+                        "This compiled object has no original cache key; "
+                        "can't pass extracted_parameters to construct_params"
+                    ),
+                    replace_context=err,
+                )
+            resolved_extracted = dict(
+                zip([b.key for b in orig_extracted], extracted_parameters)
+            )
+        else:
+            resolved_extracted = None
+
         if params:
             pd = {}
             for bindparam in self.bind_names:
@@ -844,11 +876,18 @@ class SQLCompiler(Compiled):
                             % bindparam.key,
                             code="cd3x",
                         )
-
-                elif bindparam.callable:
-                    pd[name] = bindparam.effective_value
                 else:
-                    pd[name] = bindparam.value
+                    if resolved_extracted:
+                        value_param = resolved_extracted.get(
+                            bindparam.key, bindparam
+                        )
+                    else:
+                        value_param = bindparam
+
+                    if bindparam.callable:
+                        pd[name] = value_param.effective_value
+                    else:
+                        pd[name] = value_param.value
             return pd
         else:
             pd = {}
@@ -868,10 +907,19 @@ class SQLCompiler(Compiled):
                             code="cd3x",
                         )
 
+                if resolved_extracted:
+                    value_param = resolved_extracted.get(
+                        bindparam.key, bindparam
+                    )
+                else:
+                    value_param = bindparam
+
                 if bindparam.callable:
-                    pd[self.bind_names[bindparam]] = bindparam.effective_value
+                    pd[
+                        self.bind_names[bindparam]
+                    ] = value_param.effective_value
                 else:
-                    pd[self.bind_names[bindparam]] = bindparam.value
+                    pd[self.bind_names[bindparam]] = value_param.value
             return pd
 
     @property
@@ -2144,7 +2192,9 @@ class SQLCompiler(Compiled):
                         assert False
                     recur_cols = [
                         c
-                        for c in util.unique_list(col_source.inner_columns)
+                        for c in util.unique_list(
+                            col_source._exported_columns_iterator()
+                        )
                         if c is not None
                     ]
 
@@ -3375,7 +3425,7 @@ class DDLCompiler(Compiled):
     def type_compiler(self):
         return self.dialect.type_compiler
 
-    def construct_params(self, params=None):
+    def construct_params(self, params=None, extracted_parameters=None):
         return None
 
     def visit_ddl(self, ddl, **kwargs):
diff --git a/lib/sqlalchemy/sql/crud.py b/lib/sqlalchemy/sql/crud.py
index 2827a5817..114dbec9e 100644
--- a/lib/sqlalchemy/sql/crud.py
+++ b/lib/sqlalchemy/sql/crud.py
@@ -16,7 +16,6 @@ from . import coercions
 from . import dml
 from . import elements
 from . import roles
-from .elements import ClauseElement
 from .. import exc
 from .. import util
 
@@ -198,11 +197,8 @@ def _handle_values_anonymous_param(compiler, col, value, name, **kw):
     if value.type._isnull:
         # either unique parameter, or other bound parameters that were
         # passed in directly
-        # clone using base ClauseElement to retain unique key
-        value = ClauseElement._clone(value)
-
         # set type to that of the column unconditionally
-        value.type = col.type
+        value = value._with_binary_element_type(col.type)
 
     return value._compiler_dispatch(compiler, **kw)
 
diff --git a/lib/sqlalchemy/sql/dml.py b/lib/sqlalchemy/sql/dml.py
index 5c75e068f..cbcf54d1c 100644
--- a/lib/sqlalchemy/sql/dml.py
+++ b/lib/sqlalchemy/sql/dml.py
@@ -14,6 +14,7 @@ from . import coercions
 from . import roles
 from .base import _from_objects
 from .base import _generative
+from .base import ColumnCollection
 from .base import CompileState
 from .base import DialectKWArgs
 from .base import Executable
@@ -364,6 +365,28 @@ class UpdateBase(
         """
         self._returning = cols
 
+    def _exported_columns_iterator(self):
+        """Return the RETURNING columns as a sequence for this statement.
+
+        .. versionadded:: 1.4
+
+        """
+
+        return self._returning or ()
+
+    @property
+    def exported_columns(self):
+        """Return the RETURNING columns as a column collection for this
+        statement.
+
+        .. versionadded:: 1.4
+
+        """
+        # TODO: no coverage here
+        return ColumnCollection(
+            (c.key, c) for c in self._exported_columns_iterator()
+        ).as_immutable()
+
     @_generative
     def with_hint(self, text, selectable=None, dialect_name="*"):
         """Add a table hint for a single table to this
diff --git a/lib/sqlalchemy/sql/elements.py b/lib/sqlalchemy/sql/elements.py
index 2b994c513..57d41b06f 100644
--- a/lib/sqlalchemy/sql/elements.py
+++ b/lib/sqlalchemy/sql/elements.py
@@ -175,7 +175,7 @@ def not_(clause):
 
 @inspection._self_inspects
 class ClauseElement(
-    roles.SQLRole, SupportsWrappingAnnotations, HasCacheKey, Traversible
+    roles.SQLRole, SupportsWrappingAnnotations, HasCacheKey, Traversible,
 ):
     """Base class for elements of a programmatically constructed SQL
     expression.
@@ -215,10 +215,9 @@ class ClauseElement(
         the _copy_internals() method.
 
         """
+        skip = self._memoized_keys
         c = self.__class__.__new__(self.__class__)
-        c.__dict__ = self.__dict__.copy()
-        ClauseElement._cloned_set._reset(c)
-        ColumnElement.comparator._reset(c)
+        c.__dict__ = {k: v for k, v in self.__dict__.items() if k not in skip}
 
         # this is a marker that helps to "equate" clauses to each other
         # when a Select returns its list of FROM clauses.  the cloning
@@ -250,7 +249,7 @@ class ClauseElement(
         """
         return self.__class__
 
-    @util.memoized_property
+    @HasMemoized.memoized_attribute
     def _cloned_set(self):
         """Return the set consisting all cloned ancestors of this
         ClauseElement.
@@ -276,6 +275,7 @@ class ClauseElement(
     def __getstate__(self):
         d = self.__dict__.copy()
         d.pop("_is_clone_of", None)
+        d.pop("_generate_cache_key", None)
         return d
 
     def _execute_on_connection(self, connection, multiparams, params):
@@ -740,15 +740,7 @@ class ColumnElement(
     def type(self):
         return type_api.NULLTYPE
 
-    def _with_binary_element_type(self, type_):
-        cloned = self._clone()
-        cloned._copy_internals(
-            clone=lambda element: element._with_binary_element_type(type_)
-        )
-        cloned.type = type_
-        return cloned
-
-    @util.memoized_property
+    @HasMemoized.memoized_attribute
     def comparator(self):
         try:
             comparator_factory = self.type.comparator_factory
@@ -1022,6 +1014,7 @@ class BindParameter(roles.InElementRole, ColumnElement):
     _is_crud = False
     _expanding_in_types = ()
     _is_bind_parameter = True
+    _key_is_anon = False
 
     def __init__(
         self,
@@ -1273,9 +1266,6 @@ class BindParameter(roles.InElementRole, ColumnElement):
 
 
         """
-        if isinstance(key, ColumnClause):
-            type_ = key.type
-            key = key.key
         if required is NO_ARG:
             required = value is NO_ARG and callable_ is None
         if value is NO_ARG:
@@ -1297,8 +1287,12 @@ class BindParameter(roles.InElementRole, ColumnElement):
                     else "param",
                 )
             )
+            self._key_is_anon = True
+        elif key:
+            self.key = key
         else:
-            self.key = key or _anonymous_label("%%(%d param)s" % id(self))
+            self.key = _anonymous_label("%%(%d param)s" % id(self))
+            self._key_is_anon = True
 
         # identifying key that won't change across
         # clones, used to identify the bind's logical
@@ -1366,6 +1360,11 @@ class BindParameter(roles.InElementRole, ColumnElement):
         else:
             return self.value
 
+    def _with_binary_element_type(self, type_):
+        c = ClauseElement._clone(self)
+        c.type = type_
+        return c
+
     def _clone(self):
         c = ClauseElement._clone(self)
         if self.unique:
@@ -1390,7 +1389,7 @@ class BindParameter(roles.InElementRole, ColumnElement):
             id_,
             self.__class__,
             self.type._static_cache_key,
-            traversals._resolve_name_for_compare(self, self.key, anon_map),
+            self.key % anon_map if self._key_is_anon else self.key,
         )
 
     def _convert_to_unique(self):
@@ -2790,7 +2789,7 @@ class Cast(WrapsColumnExpression, ColumnElement):
         return self.clause
 
 
-class TypeCoerce(HasMemoized, WrapsColumnExpression, ColumnElement):
+class TypeCoerce(WrapsColumnExpression, ColumnElement):
     """Represent a Python-side type-coercion wrapper.
 
     :class:`.TypeCoerce` supplies the :func:`.expression.type_coerce`
@@ -2815,8 +2814,6 @@ class TypeCoerce(HasMemoized, WrapsColumnExpression, ColumnElement):
         ("type", InternalTraversal.dp_type),
     ]
 
-    _memoized_property = util.group_expirable_memoized_property()
-
     def __init__(self, expression, type_):
         r"""Associate a SQL expression with a particular type, without rendering
         ``CAST``.
@@ -2889,7 +2886,7 @@ class TypeCoerce(HasMemoized, WrapsColumnExpression, ColumnElement):
     def _from_objects(self):
         return self.clause._from_objects
 
-    @_memoized_property
+    @HasMemoized.memoized_attribute
     def typed_expression(self):
         if isinstance(self.clause, BindParameter):
             bp = self.clause._clone()
@@ -3435,7 +3432,7 @@ class BinaryExpression(ColumnElement):
         # refer to BinaryExpression directly and pass strings
         if isinstance(operator, util.string_types):
             operator = operators.custom_op(operator)
-        self._orig = (hash(left), hash(right))
+        self._orig = (left.__hash__(), right.__hash__())
         self.left = left.self_group(against=operator)
         self.right = right.self_group(against=operator)
         self.operator = operator
@@ -3450,7 +3447,7 @@ class BinaryExpression(ColumnElement):
 
     def __bool__(self):
         if self.operator in (operator.eq, operator.ne):
-            return self.operator(self._orig[0], self._orig[1])
+            return self.operator(*self._orig)
         else:
             raise TypeError("Boolean value of this clause is not defined")
 
@@ -3546,6 +3543,9 @@ class Grouping(GroupedElement, ColumnElement):
         self.element = element
         self.type = getattr(element, "type", type_api.NULLTYPE)
 
+    def _with_binary_element_type(self, type_):
+        return Grouping(self.element._with_binary_element_type(type_))
+
     @util.memoized_property
     def _is_implicitly_boolean(self):
         return self.element._is_implicitly_boolean
@@ -4015,7 +4015,7 @@ class FunctionFilter(ColumnElement):
         )
 
 
-class Label(HasMemoized, roles.LabeledColumnExprRole, ColumnElement):
+class Label(roles.LabeledColumnExprRole, ColumnElement):
     """Represents a column label (AS).
 
     Represent a label, as typically applied to any column-level
@@ -4031,8 +4031,6 @@ class Label(HasMemoized, roles.LabeledColumnExprRole, ColumnElement):
         ("_element", InternalTraversal.dp_clauseelement),
     ]
 
-    _memoized_property = util.group_expirable_memoized_property()
-
     def __init__(self, name, element, type_=None):
         """Return a :class:`Label` object for the
         given :class:`.ColumnElement`.
@@ -4075,7 +4073,7 @@ class Label(HasMemoized, roles.LabeledColumnExprRole, ColumnElement):
     def _is_implicitly_boolean(self):
         return self.element._is_implicitly_boolean
 
-    @_memoized_property
+    @HasMemoized.memoized_attribute
     def _allow_label_resolve(self):
         return self.element._allow_label_resolve
 
@@ -4089,7 +4087,7 @@ class Label(HasMemoized, roles.LabeledColumnExprRole, ColumnElement):
             self._type or getattr(self._element, "type", None)
         )
 
-    @_memoized_property
+    @HasMemoized.memoized_attribute
     def element(self):
         return self._element.self_group(against=operators.as_)
 
@@ -4116,7 +4114,6 @@ class Label(HasMemoized, roles.LabeledColumnExprRole, ColumnElement):
         return self.element.foreign_keys
 
     def _copy_internals(self, clone=_clone, anonymize_labels=False, **kw):
-        self._reset_memoizations()
         self._element = clone(self._element, **kw)
         if anonymize_labels:
             self.name = self._resolve_label = _anonymous_label(
@@ -4194,8 +4191,6 @@ class ColumnClause(
 
     _is_multiparam_column = False
 
-    _memoized_property = util.group_expirable_memoized_property()
-
     def __init__(self, text, type_=None, is_literal=False, _selectable=None):
         """Produce a :class:`.ColumnClause` object.
 
@@ -4312,7 +4307,7 @@ class ColumnClause(
         else:
             return []
 
-    @_memoized_property
+    @HasMemoized.memoized_attribute
     def _from_objects(self):
         t = self.table
         if t is not None:
@@ -4327,18 +4322,18 @@ class ColumnClause(
         else:
             return self.name.encode("ascii", "backslashreplace")
 
-    @_memoized_property
+    @HasMemoized.memoized_attribute
     def _key_label(self):
         if self.key != self.name:
             return self._gen_label(self.key)
         else:
             return self._label
 
-    @_memoized_property
+    @HasMemoized.memoized_attribute
     def _label(self):
         return self._gen_label(self.name)
 
-    @_memoized_property
+    @HasMemoized.memoized_attribute
     def _render_label_in_columns_clause(self):
         return self.table is not None
 
@@ -4599,14 +4594,14 @@ def _corresponding_column_or_error(fromclause, column, require_embedded=False):
 class AnnotatedColumnElement(Annotated):
     def __init__(self, element, values):
         Annotated.__init__(self, element, values)
-        ColumnElement.comparator._reset(self)
+        self.__dict__.pop("comparator", None)
         for attr in ("name", "key", "table"):
             if self.__dict__.get(attr, False) is None:
                 self.__dict__.pop(attr)
 
     def _with_annotations(self, values):
         clone = super(AnnotatedColumnElement, self)._with_annotations(values)
-        ColumnElement.comparator._reset(clone)
+        clone.__dict__.pop("comparator", None)
         return clone
 
     @util.memoized_property
diff --git a/lib/sqlalchemy/sql/functions.py b/lib/sqlalchemy/sql/functions.py
index 6004f6b51..7973871f3 100644
--- a/lib/sqlalchemy/sql/functions.py
+++ b/lib/sqlalchemy/sql/functions.py
@@ -17,6 +17,7 @@ from . import sqltypes
 from . import util as sqlutil
 from .base import ColumnCollection
 from .base import Executable
+from .base import HasMemoized
 from .elements import _type_from_args
 from .elements import BinaryExpression
 from .elements import BindParameter
@@ -85,8 +86,6 @@ class FunctionElement(Executable, ColumnElement, FromClause):
 
     _has_args = False
 
-    _memoized_property = FromClause._memoized_property
-
     def __init__(self, *clauses, **kwargs):
         r"""Construct a :class:`.FunctionElement`.
 
@@ -141,7 +140,7 @@ class FunctionElement(Executable, ColumnElement, FromClause):
         col = self.label(None)
         return ColumnCollection(columns=[(col.key, col)])
 
-    @_memoized_property
+    @HasMemoized.memoized_attribute
     def clauses(self):
         """Return the underlying :class:`.ClauseList` which contains
         the arguments for this :class:`.FunctionElement`.
diff --git a/lib/sqlalchemy/sql/schema.py b/lib/sqlalchemy/sql/schema.py
index 02c14d751..5c6b1f3c6 100644
--- a/lib/sqlalchemy/sql/schema.py
+++ b/lib/sqlalchemy/sql/schema.py
@@ -1412,7 +1412,7 @@ class Column(DialectKWArgs, SchemaItem, ColumnClause):
                 "assign a non-blank .name before adding to a Table."
             )
 
-        Column._memoized_property.expire_instance(self)
+        self._reset_memoizations()
 
         if self.key is None:
             self.key = self.name
diff --git a/lib/sqlalchemy/sql/selectable.py b/lib/sqlalchemy/sql/selectable.py
index 4eab60801..e39d61fdb 100644
--- a/lib/sqlalchemy/sql/selectable.py
+++ b/lib/sqlalchemy/sql/selectable.py
@@ -106,31 +106,33 @@ class ReturnsRows(roles.ReturnsRowsRole, ClauseElement):
     def selectable(self):
         raise NotImplementedError()
 
+    def _exported_columns_iterator(self):
+        """An iterator of column objects that represents the "exported"
+        columns of this :class:`.ReturnsRows`.
 
-class Selectable(ReturnsRows):
-    """mark a class as being selectable.
-
-    """
+        This is the same set of columns as are returned by
+        :meth:`.ReturnsRows.exported_columns` except they are returned
+        as a simple iterator or sequence, rather than as a
+        :class:`.ColumnCollection` namespace.
 
-    __visit_name__ = "selectable"
-
-    is_selectable = True
+        Subclasses should re-implement this method to bypass the interim
+        creation of the :class:`.ColumnCollection` if appropriate.
 
-    @property
-    def selectable(self):
-        return self
+        """
+        return iter(self.exported_columns)
 
     @property
     def exported_columns(self):
         """A :class:`.ColumnCollection` that represents the "exported"
-        columns of this :class:`.Selectable`.
+        columns of this :class:`.ReturnsRows`.
 
         The "exported" columns represent the collection of
         :class:`.ColumnElement` expressions that are rendered by this SQL
-        construct.   There are two primary varieties which are the
+        construct.   There are primary varieties which are the
         "FROM clause columns" of a FROM clause, such as a table, join,
-        or subquery, and the "SELECTed columns", which are the columns in
-        the "columns clause" of a SELECT statement.
+        or subquery, the "SELECTed columns", which are the columns in
+        the "columns clause" of a SELECT statement, and the RETURNING
+        columns in a DML statement..
 
         .. versionadded:: 1.4
 
@@ -143,6 +145,20 @@ class Selectable(ReturnsRows):
 
         raise NotImplementedError()
 
+
+class Selectable(ReturnsRows):
+    """mark a class as being selectable.
+
+    """
+
+    __visit_name__ = "selectable"
+
+    is_selectable = True
+
+    @property
+    def selectable(self):
+        return self
+
     def _refresh_for_new_column(self, column):
         raise NotImplementedError()
 
@@ -312,7 +328,7 @@ class HasSuffixes(object):
         )
 
 
-class FromClause(HasMemoized, roles.AnonymizedFromClauseRole, Selectable):
+class FromClause(roles.AnonymizedFromClauseRole, Selectable):
     """Represent an element that can be used within the ``FROM``
     clause of a ``SELECT`` statement.
 
@@ -350,8 +366,6 @@ class FromClause(HasMemoized, roles.AnonymizedFromClauseRole, Selectable):
 
     _use_schema_map = False
 
-    _memoized_property = util.group_expirable_memoized_property(["_columns"])
-
     @util.deprecated(
         "1.1",
         message="The :meth:`.FromClause.count` method is deprecated, "
@@ -571,7 +585,7 @@ class FromClause(HasMemoized, roles.AnonymizedFromClauseRole, Selectable):
         """
         return self.columns
 
-    @_memoized_property
+    @util.memoized_property
     def columns(self):
         """A named-based collection of :class:`.ColumnElement` objects
         maintained by this :class:`.FromClause`.
@@ -589,7 +603,7 @@ class FromClause(HasMemoized, roles.AnonymizedFromClauseRole, Selectable):
             self._populate_column_collection()
         return self._columns.as_immutable()
 
-    @_memoized_property
+    @util.memoized_property
     def primary_key(self):
         """Return the collection of Column objects which comprise the
         primary key of this FromClause."""
@@ -598,7 +612,7 @@ class FromClause(HasMemoized, roles.AnonymizedFromClauseRole, Selectable):
         self._populate_column_collection()
         return self.primary_key
 
-    @_memoized_property
+    @util.memoized_property
     def foreign_keys(self):
         """Return the collection of ForeignKey objects which this
         FromClause references."""
@@ -607,6 +621,23 @@ class FromClause(HasMemoized, roles.AnonymizedFromClauseRole, Selectable):
         self._populate_column_collection()
         return self.foreign_keys
 
+    def _reset_column_collection(self):
+        """Reset the attributes linked to the FromClause.c attribute.
+
+        This collection is separate from all the other memoized things
+        as it has shown to be sensitive to being cleared out in situations
+        where enclosing code, typically in a replacement traversal scenario,
+        has already established strong relationships
+        with the exported columns.
+
+        The collection is cleared for the case where a table is having a
+        column added to it as well as within a Join during copy internals.
+
+        """
+
+        for key in ["_columns", "columns", "primary_key", "foreign_keys"]:
+            self.__dict__.pop(key, None)
+
     c = property(
         attrgetter("columns"),
         doc="An alias for the :attr:`.columns` attribute.",
@@ -659,7 +690,7 @@ class FromClause(HasMemoized, roles.AnonymizedFromClauseRole, Selectable):
         derivations.
 
         """
-        self._reset_exported()
+        self._reset_column_collection()
 
 
 class Join(FromClause):
@@ -1239,7 +1270,7 @@ class AliasedReturnsRows(NoInit, FromClause):
         # same object.  don't reset exported .c. collections and other
         # memoized details if nothing changed
         if element is not self.element:
-            self._reset_exported()
+            self._reset_column_collection()
             self.element = element
 
     @property
@@ -2141,7 +2172,6 @@ class SelectBase(
     roles.DMLSelectRole,
     roles.CompoundElementRole,
     roles.InElementRole,
-    HasMemoized,
     HasCTE,
     Executable,
     SupportsCloneAnnotations,
@@ -2158,8 +2188,6 @@ class SelectBase(
 
     _is_select_statement = True
 
-    _memoized_property = util.group_expirable_memoized_property()
-
     def _generate_fromclause_column_proxies(self, fromclause):
         # type: (FromClause) -> None
         raise NotImplementedError()
@@ -2254,7 +2282,7 @@ class SelectBase(
     def outerjoin(self, *arg, **kw):
         return self._implicit_subquery.outerjoin(*arg, **kw)
 
-    @_memoized_property
+    @HasMemoized.memoized_attribute
     def _implicit_subquery(self):
         return self.subquery()
 
@@ -2315,15 +2343,6 @@ class SelectBase(
         """
         return Lateral._factory(self, name)
 
-    def _generate(self):
-        """Override the default _generate() method to also clear out
-        exported collections."""
-
-        s = self.__class__.__new__(self.__class__)
-        s.__dict__ = self.__dict__.copy()
-        s._reset_memoizations()
-        return s
-
     @property
     def _from_objects(self):
         return [self]
@@ -2431,6 +2450,9 @@ class SelectStatementGrouping(GroupedElement, SelectBase):
     def _generate_proxy_for_new_column(self, column, subquery):
         return self.element._generate_proxy_for_new_column(subquery)
 
+    def _exported_columns_iterator(self):
+        return self.element._exported_columns_iterator()
+
     @property
     def selected_columns(self):
         """A :class:`.ColumnCollection` representing the columns that
@@ -3046,6 +3068,9 @@ class CompoundSelect(HasCompileState, GenerativeSelect):
         for select in self.selects:
             select._refresh_for_new_column(column)
 
+    def _exported_columns_iterator(self):
+        return self.selects[0]._exported_columns_iterator()
+
     @property
     def selected_columns(self):
         """A :class:`.ColumnCollection` representing the columns that
@@ -3339,8 +3364,6 @@ class Select(
     _from_obj = ()
     _auto_correlate = True
 
-    _memoized_property = SelectBase._memoized_property
-
     _traverse_internals = (
         [
             ("_from_obj", InternalTraversal.dp_clauseelement_list),
@@ -3400,8 +3423,7 @@ class Select(
 
         self = cls.__new__(cls)
         self._raw_columns = [
-            coercions.expect(roles.ColumnsClauseRole, ent)
-            for ent in util.to_list(entities)
+            coercions.expect(roles.ColumnsClauseRole, ent) for ent in entities
         ]
 
         GenerativeSelect.__init__(self)
@@ -3739,8 +3761,12 @@ class Select(
         """an iterator of all ColumnElement expressions which would
         be rendered into the columns clause of the resulting SELECT statement.
 
+        This method is legacy as of 1.4 and is superseded by the
+        :attr:`.Select.exported_columns` collection.
+
         """
-        return _select_iterables(self._raw_columns)
+
+        return self._exported_columns_iterator()
 
     def is_derived_from(self, fromclause):
         if self in fromclause._cloned_set:
@@ -3786,7 +3812,10 @@ class Select(
             clone=clone, omit_attrs=("_from_obj",), **kw
         )
 
-        self._reset_memoizations()
+        # memoizations should be cleared here as of
+        # I95c560ffcbfa30b26644999412fb6a385125f663 , asserting this
+        # is the case for now.
+        self._assert_no_memoizations()
 
     def get_children(self, **kwargs):
         return list(set(self._iterate_from_elements())) + super(
@@ -3809,7 +3838,10 @@ class Select(
             :class:`.Select` object.
 
         """
-        self._reset_memoizations()
+        # memoizations should be cleared here as of
+        # I95c560ffcbfa30b26644999412fb6a385125f663 , asserting this
+        # is the case for now.
+        self._assert_no_memoizations()
 
         self._raw_columns = self._raw_columns + [
             coercions.expect(roles.ColumnsClauseRole, column,)
@@ -3861,7 +3893,7 @@ class Select(
         """
         return self.with_only_columns(
             util.preloaded.sql_util.reduce_columns(
-                self.inner_columns,
+                self._exported_columns_iterator(),
                 only_synonyms=only_synonyms,
                 *(self._where_criteria + self._from_obj)
             )
@@ -3935,7 +3967,12 @@ class Select(
         being asked to select both from ``table1`` as well as itself.
 
         """
-        self._reset_memoizations()
+
+        # memoizations should be cleared here as of
+        # I95c560ffcbfa30b26644999412fb6a385125f663 , asserting this
+        # is the case for now.
+        self._assert_no_memoizations()
+
         rc = []
         for c in columns:
             c = coercions.expect(roles.ColumnsClauseRole, c,)
@@ -4112,7 +4149,7 @@ class Select(
                 coercions.expect(roles.FromClauseRole, f) for f in fromclauses
             )
 
-    @_memoized_property
+    @HasMemoized.memoized_attribute
     def selected_columns(self):
         """A :class:`.ColumnCollection` representing the columns that
         this SELECT statement or similar construct returns in its result set.
@@ -4167,6 +4204,9 @@ class Select(
 
         return ColumnCollection(collection).as_immutable()
 
+    def _exported_columns_iterator(self):
+        return _select_iterables(self._raw_columns)
+
     def _ensure_disambiguated_names(self):
         if self._label_style is LABEL_STYLE_NONE:
             self = self._set_label_style(LABEL_STYLE_DISAMBIGUATE_ONLY)
@@ -4558,7 +4598,7 @@ class TextualSelect(SelectBase):
         ]
         self.positional = positional
 
-    @SelectBase._memoized_property
+    @HasMemoized.memoized_attribute
     def selected_columns(self):
         """A :class:`.ColumnCollection` representing the columns that
         this SELECT statement or similar construct returns in its result set.
diff --git a/lib/sqlalchemy/sql/traversals.py b/lib/sqlalchemy/sql/traversals.py
index 1fcc2d023..9ac6cda97 100644
--- a/lib/sqlalchemy/sql/traversals.py
+++ b/lib/sqlalchemy/sql/traversals.py
@@ -7,6 +7,7 @@ from .visitors import ExtendedInternalTraversal
 from .visitors import InternalTraversal
 from .. import util
 from ..inspection import inspect
+from ..util import HasMemoized
 
 SKIP_TRAVERSE = util.symbol("skip_traverse")
 COMPARE_FAILED = False
@@ -26,7 +27,7 @@ def compare(obj1, obj2, **kw):
     return strategy.compare(obj1, obj2, **kw)
 
 
-class HasCacheKey(object):
+class HasCacheKey(HasMemoized):
     _cache_key_traversal = NO_CACHE
 
     __slots__ = ()
@@ -105,6 +106,14 @@ class HasCacheKey(object):
                         attrname,
                         obj._gen_cache_key(anon_map, bindparams),
                     )
+                elif meth is InternalTraversal.dp_annotations_key:
+                    # obj is here is the _annotations dict.   however,
+                    # we want to use the memoized cache key version of it.
+                    # for Columns, this should be long lived.   For select()
+                    # statements, not so much, but they usually won't have
+                    # annotations.
+                    if obj:
+                        result += self._annotations_cache_key
                 elif meth is InternalTraversal.dp_clauseelement_list:
                     if obj:
                         result += (
@@ -130,6 +139,7 @@ class HasCacheKey(object):
 
         return result
 
+    @HasMemoized.memoized_instancemethod
     def _generate_cache_key(self):
         """return a cache key.
 
@@ -161,6 +171,7 @@ class HasCacheKey(object):
         will return None, indicating no cache key is available.
 
         """
+
         bindparams = []
 
         _anon_map = anon_map()
@@ -178,6 +189,36 @@ class CacheKey(namedtuple("CacheKey", ["key", "bindparams"])):
     def __eq__(self, other):
         return self.key == other.key
 
+    def __str__(self):
+        stack = [self.key]
+
+        output = []
+        sentinel = object()
+        indent = -1
+        while stack:
+            elem = stack.pop(0)
+            if elem is sentinel:
+                output.append((" " * (indent * 2)) + "),")
+                indent -= 1
+            elif isinstance(elem, tuple):
+                if not elem:
+                    output.append((" " * ((indent + 1) * 2)) + "()")
+                else:
+                    indent += 1
+                    stack = list(elem) + [sentinel] + stack
+                    output.append((" " * (indent * 2)) + "(")
+            else:
+                if isinstance(elem, HasCacheKey):
+                    repr_ = "<%s object at %s>" % (
+                        type(elem).__name__,
+                        hex(id(elem)),
+                    )
+                else:
+                    repr_ = repr(elem)
+                output.append((" " * (indent * 2)) + "  " + repr_ + ", ")
+
+        return "CacheKey(key=%s)" % ("\n".join(output),)
+
 
 def _clone(element, **kw):
     return element._clone()
@@ -189,6 +230,8 @@ class _CacheKey(ExtendedInternalTraversal):
 
     visit_has_cache_key = visit_clauseelement = CALL_GEN_CACHE_KEY
     visit_clauseelement_list = InternalTraversal.dp_clauseelement_list
+    visit_annotations_key = InternalTraversal.dp_annotations_key
+
     visit_string = (
         visit_boolean
     ) = visit_operator = visit_plain_obj = CACHE_IN_PLACE
@@ -690,8 +733,8 @@ class TraversalComparatorStrategy(InternalTraversal, util.MemoizedSlots):
                 fillvalue=(None, None),
             ):
                 if not compare_annotations and (
-                    (left_attrname == "_annotations_cache_key")
-                    or (right_attrname == "_annotations_cache_key")
+                    (left_attrname == "_annotations")
+                    or (right_attrname == "_annotations")
                 ):
                     continue
 
@@ -827,6 +870,17 @@ class TraversalComparatorStrategy(InternalTraversal, util.MemoizedSlots):
     ):
         return left == right
 
+    def visit_annotations_key(
+        self, left_parent, left, right_parent, right, **kw
+    ):
+        if left and right:
+            return (
+                left_parent._annotations_cache_key
+                == right_parent._annotations_cache_key
+            )
+        else:
+            return left == right
+
     def visit_plain_obj(self, left_parent, left, right_parent, right, **kw):
         return left == right
 
diff --git a/lib/sqlalchemy/sql/type_api.py b/lib/sqlalchemy/sql/type_api.py
index 38189ec9d..e3929fac7 100644
--- a/lib/sqlalchemy/sql/type_api.py
+++ b/lib/sqlalchemy/sql/type_api.py
@@ -1414,6 +1414,17 @@ class Variant(TypeDecorator):
         self.impl = base
         self.mapping = mapping
 
+    @util.memoized_property
+    def _static_cache_key(self):
+        # TODO: needs tests in test/sql/test_compare.py
+        return (self.__class__,) + (
+            self.impl._static_cache_key,
+            tuple(
+                (key, self.mapping[key]._static_cache_key)
+                for key in sorted(self.mapping)
+            ),
+        )
+
     def coerce_compared_value(self, operator, value):
         result = self.impl.coerce_compared_value(operator, value)
         if result is self.impl:
diff --git a/lib/sqlalchemy/sql/util.py b/lib/sqlalchemy/sql/util.py
index 8d185ce7d..fae68da98 100644
--- a/lib/sqlalchemy/sql/util.py
+++ b/lib/sqlalchemy/sql/util.py
@@ -584,7 +584,6 @@ def splice_joins(left, right, stop_on=None):
         (right, prevright) = stack.pop()
         if isinstance(right, Join) and right is not stop_on:
             right = right._clone()
-            right._reset_exported()
             right.onclause = adapter.traverse(right.onclause)
             stack.append((right.left, right))
         else:
diff --git a/lib/sqlalchemy/sql/visitors.py b/lib/sqlalchemy/sql/visitors.py
index 4c1aab62f..5504bf3d8 100644
--- a/lib/sqlalchemy/sql/visitors.py
+++ b/lib/sqlalchemy/sql/visitors.py
@@ -331,6 +331,17 @@ class InternalTraversal(util.with_metaclass(_InternalTraversalType, object)):
 
     """
 
+    dp_annotations_key = symbol("AK")
+    """Visit the _annotations_cache_key element.
+
+    This is a dictionary of additional information about a ClauseElement
+    that modifies its role.  It should be included when comparing or caching
+    objects, however generating this key is relatively expensive.   Visitors
+    should check the "_annotations" dict for non-None first before creating
+    this key.
+
+    """
+
     dp_plain_obj = symbol("PO")
     """Visit a plain python object.