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

22 files changed, 482 insertions, 215 deletions

diff --git a/lib/sqlalchemy/engine/base.py b/lib/sqlalchemy/engine/base.py
index 4ed3b9af7..34a4f04a9 100644
--- a/lib/sqlalchemy/engine/base.py
+++ b/lib/sqlalchemy/engine/base.py
@@ -1000,7 +1000,8 @@ class Connection(Connectable):
             tuple or scalar positional parameters.
 
         """
-        if isinstance(object_, util.string_types[0]):
+
+        if isinstance(object_, util.string_types):
             util.warn_deprecated_20(
                 "Passing a string to Connection.execute() is "
                 "deprecated and will be removed in version 2.0.  Use the "
@@ -1098,26 +1099,33 @@ class Connection(Connectable):
             keys = []
 
         dialect = self.dialect
+
         if "compiled_cache" in self._execution_options:
+            elem_cache_key, extracted_params = elem._generate_cache_key()
             key = (
                 dialect,
-                elem,
+                elem_cache_key,
                 tuple(sorted(keys)),
                 bool(self._schema_translate_map),
                 len(distilled_params) > 1,
             )
-            compiled_sql = self._execution_options["compiled_cache"].get(key)
+            cache = self._execution_options["compiled_cache"]
+            compiled_sql = cache.get(key)
+
             if compiled_sql is None:
                 compiled_sql = elem.compile(
                     dialect=dialect,
+                    cache_key=(elem_cache_key, extracted_params),
                     column_keys=keys,
                     inline=len(distilled_params) > 1,
                     schema_translate_map=self._schema_translate_map,
                     linting=self.dialect.compiler_linting
                     | compiler.WARN_LINTING,
                 )
-                self._execution_options["compiled_cache"][key] = compiled_sql
+                cache[key] = compiled_sql
+
         else:
+            extracted_params = None
             compiled_sql = elem.compile(
                 dialect=dialect,
                 column_keys=keys,
@@ -1133,6 +1141,8 @@ class Connection(Connectable):
             distilled_params,
             compiled_sql,
             distilled_params,
+            elem,
+            extracted_params,
         )
         if self._has_events or self.engine._has_events:
             self.dispatch.after_execute(self, elem, multiparams, params, ret)
@@ -1156,6 +1166,8 @@ class Connection(Connectable):
             parameters,
             compiled,
             parameters,
+            None,
+            None,
         )
         if self._has_events or self.engine._has_events:
             self.dispatch.after_execute(
diff --git a/lib/sqlalchemy/engine/default.py b/lib/sqlalchemy/engine/default.py
index c44f07538..af61be034 100644
--- a/lib/sqlalchemy/engine/default.py
+++ b/lib/sqlalchemy/engine/default.py
@@ -750,7 +750,14 @@ class DefaultExecutionContext(interfaces.ExecutionContext):
 
     @classmethod
     def _init_compiled(
-        cls, dialect, connection, dbapi_connection, compiled, parameters
+        cls,
+        dialect,
+        connection,
+        dbapi_connection,
+        compiled,
+        parameters,
+        invoked_statement,
+        extracted_parameters,
     ):
         """Initialize execution context for a Compiled construct."""
 
@@ -758,7 +765,8 @@ class DefaultExecutionContext(interfaces.ExecutionContext):
         self.root_connection = connection
         self._dbapi_connection = dbapi_connection
         self.dialect = connection.dialect
-
+        self.extracted_parameters = extracted_parameters
+        self.invoked_statement = invoked_statement
         self.compiled = compiled
 
         # this should be caught in the engine before
@@ -778,7 +786,6 @@ class DefaultExecutionContext(interfaces.ExecutionContext):
             compiled._textual_ordered_columns,
             compiled._loose_column_name_matching,
         )
-
         self.isinsert = compiled.isinsert
         self.isupdate = compiled.isupdate
         self.isdelete = compiled.isdelete
@@ -792,10 +799,18 @@ class DefaultExecutionContext(interfaces.ExecutionContext):
             )
 
         if not parameters:
-            self.compiled_parameters = [compiled.construct_params()]
+            self.compiled_parameters = [
+                compiled.construct_params(
+                    extracted_parameters=extracted_parameters
+                )
+            ]
         else:
             self.compiled_parameters = [
-                compiled.construct_params(m, _group_number=grp)
+                compiled.construct_params(
+                    m,
+                    _group_number=grp,
+                    extracted_parameters=extracted_parameters,
+                )
                 for grp, m in enumerate(parameters)
             ]
 
diff --git a/lib/sqlalchemy/engine/result.py b/lib/sqlalchemy/engine/result.py
index ac033a5ae..986edd617 100644
--- a/lib/sqlalchemy/engine/result.py
+++ b/lib/sqlalchemy/engine/result.py
@@ -114,6 +114,44 @@ class CursorResultMetaData(ResultMetaData):
         "keys",
     )
 
+    def _adapt_to_context(self, context):
+        """When using a cached result metadata against a new context,
+        we need to rewrite the _keymap so that it has the specific
+        Column objects in the new context inside of it.  this accommodates
+        for select() constructs that contain anonymized columns and
+        are cached.
+
+        """
+        if not context.compiled._result_columns:
+            return self
+
+        compiled_statement = context.compiled.statement
+        invoked_statement = context.invoked_statement
+
+        # same statement was invoked as the one we cached against,
+        # return self
+        if compiled_statement is invoked_statement:
+            return self
+
+        # make a copy and add the columns from the invoked statement
+        # to the result map.
+        md = self.__class__.__new__(self.__class__)
+
+        md._keymap = self._keymap.copy()
+
+        # match up new columns positionally to the result columns
+        for existing, new in zip(
+            context.compiled._result_columns,
+            invoked_statement._exported_columns_iterator(),
+        ):
+            md._keymap[new] = md._keymap[existing[RM_NAME]]
+
+        md.case_sensitive = self.case_sensitive
+        md.matched_on_name = self.matched_on_name
+        md._processors = self._processors
+        md.keys = self.keys
+        return md
+
     def __init__(self, parent, cursor_description):
         context = parent.context
         dialect = context.dialect
@@ -1107,7 +1145,9 @@ class BaseResult(object):
         if strat.cursor_description is not None:
             if self.context.compiled:
                 if self.context.compiled._cached_metadata:
-                    self._metadata = self.context.compiled._cached_metadata
+                    cached_md = self.context.compiled._cached_metadata
+                    self._metadata = cached_md._adapt_to_context(self.context)
+
                 else:
                     self._metadata = (
                         self.context.compiled._cached_metadata
diff --git a/lib/sqlalchemy/orm/instrumentation.py b/lib/sqlalchemy/orm/instrumentation.py
index 785f54085..432bff7d4 100644
--- a/lib/sqlalchemy/orm/instrumentation.py
+++ b/lib/sqlalchemy/orm/instrumentation.py
@@ -36,12 +36,10 @@ from . import exc
 from . import interfaces
 from . import state
 from .. import util
+from ..util import HasMemoized
 
 
-_memoized_key_collection = util.group_expirable_memoized_property()
-
-
-class ClassManager(dict):
+class ClassManager(HasMemoized, dict):
     """tracks state information at the class level."""
 
     MANAGER_ATTR = base.DEFAULT_MANAGER_ATTR
@@ -122,17 +120,17 @@ class ClassManager(dict):
     def is_mapped(self):
         return "mapper" in self.__dict__
 
-    @_memoized_key_collection
+    @HasMemoized.memoized_attribute
     def _all_key_set(self):
         return frozenset(self)
 
-    @_memoized_key_collection
+    @HasMemoized.memoized_attribute
     def _collection_impl_keys(self):
         return frozenset(
             [attr.key for attr in self.values() if attr.impl.collection]
         )
 
-    @_memoized_key_collection
+    @HasMemoized.memoized_attribute
     def _scalar_loader_impls(self):
         return frozenset(
             [
@@ -142,7 +140,7 @@ class ClassManager(dict):
             ]
         )
 
-    @_memoized_key_collection
+    @HasMemoized.memoized_attribute
     def _loader_impls(self):
         return frozenset([attr.impl for attr in self.values()])
 
@@ -261,7 +259,7 @@ class ClassManager(dict):
         else:
             self.local_attrs[key] = inst
             self.install_descriptor(key, inst)
-        _memoized_key_collection.expire_instance(self)
+        self._reset_memoizations()
         self[key] = inst
 
         for cls in self.class_.__subclasses__():
@@ -291,7 +289,7 @@ class ClassManager(dict):
         else:
             del self.local_attrs[key]
             self.uninstall_descriptor(key)
-        _memoized_key_collection.expire_instance(self)
+        self._reset_memoizations()
         del self[key]
         for cls in self.class_.__subclasses__():
             manager = manager_of_class(cls)
diff --git a/lib/sqlalchemy/orm/mapper.py b/lib/sqlalchemy/orm/mapper.py
index cd974190b..f4e20afdf 100644
--- a/lib/sqlalchemy/orm/mapper.py
+++ b/lib/sqlalchemy/orm/mapper.py
@@ -52,13 +52,12 @@ from ..sql import operators
 from ..sql import roles
 from ..sql import util as sql_util
 from ..sql import visitors
+from ..util import HasMemoized
 
 
 _mapper_registry = weakref.WeakKeyDictionary()
 _already_compiling = False
 
-_memoized_configured_property = util.group_expirable_memoized_property()
-
 
 # a constant returned by _get_attr_by_column to indicate
 # this mapper is not handling an attribute for a particular
@@ -1635,14 +1634,14 @@ class Mapper(sql_base.HasCacheKey, InspectionAttr):
 
     _validate_polymorphic_identity = None
 
-    @_memoized_configured_property
+    @HasMemoized.memoized_attribute
     def _version_id_prop(self):
         if self.version_id_col is not None:
             return self._columntoproperty[self.version_id_col]
         else:
             return None
 
-    @_memoized_configured_property
+    @HasMemoized.memoized_attribute
     def _acceptable_polymorphic_identities(self):
         identities = set()
 
@@ -1655,7 +1654,7 @@ class Mapper(sql_base.HasCacheKey, InspectionAttr):
 
         return identities
 
-    @_memoized_configured_property
+    @HasMemoized.memoized_attribute
     def _prop_set(self):
         return frozenset(self._props.values())
 
@@ -1708,7 +1707,7 @@ class Mapper(sql_base.HasCacheKey, InspectionAttr):
                     col = m.local_table.corresponding_column(prop.columns[0])
                     if col is not None:
                         for m2 in path:
-                            m2.persist_selectable._reset_exported()
+                            m2.persist_selectable._refresh_for_new_column(col)
                         col = self.persist_selectable.corresponding_column(
                             prop.columns[0]
                         )
@@ -1859,7 +1858,7 @@ class Mapper(sql_base.HasCacheKey, InspectionAttr):
                         # mapped table, this corresponds to adding a
                         # column after the fact to the local table.
                         # [ticket:1523]
-                        self.persist_selectable._reset_exported()
+                        self.persist_selectable._refresh_for_new_column(mc)
                     mc = self.persist_selectable.corresponding_column(c)
                     if mc is None:
                         raise sa_exc.ArgumentError(
@@ -1929,7 +1928,7 @@ class Mapper(sql_base.HasCacheKey, InspectionAttr):
 
     def _expire_memoizations(self):
         for mapper in self.iterate_to_root():
-            _memoized_configured_property.expire_instance(mapper)
+            mapper._reset_memoizations()
 
     @property
     def _log_desc(self):
@@ -2078,7 +2077,7 @@ class Mapper(sql_base.HasCacheKey, InspectionAttr):
 
         return from_obj
 
-    @_memoized_configured_property
+    @HasMemoized.memoized_attribute
     def _single_table_criterion(self):
         if self.single and self.inherits and self.polymorphic_on is not None:
             return self.polymorphic_on._annotate({"parentmapper": self}).in_(
@@ -2087,7 +2086,7 @@ class Mapper(sql_base.HasCacheKey, InspectionAttr):
         else:
             return None
 
-    @_memoized_configured_property
+    @HasMemoized.memoized_attribute
     def _with_polymorphic_mappers(self):
         if Mapper._new_mappers:
             configure_mappers()
@@ -2095,7 +2094,7 @@ class Mapper(sql_base.HasCacheKey, InspectionAttr):
             return []
         return self._mappers_from_spec(*self.with_polymorphic)
 
-    @_memoized_configured_property
+    @HasMemoized.memoized_attribute
     def _with_polymorphic_selectable(self):
         if not self.with_polymorphic:
             return self.persist_selectable
@@ -2114,7 +2113,7 @@ class Mapper(sql_base.HasCacheKey, InspectionAttr):
 
     """
 
-    @_memoized_configured_property
+    @HasMemoized.memoized_attribute
     def _insert_cols_evaluating_none(self):
         return dict(
             (
@@ -2126,7 +2125,7 @@ class Mapper(sql_base.HasCacheKey, InspectionAttr):
             for table, columns in self._cols_by_table.items()
         )
 
-    @_memoized_configured_property
+    @HasMemoized.memoized_attribute
     def _insert_cols_as_none(self):
         return dict(
             (
@@ -2143,7 +2142,7 @@ class Mapper(sql_base.HasCacheKey, InspectionAttr):
             for table, columns in self._cols_by_table.items()
         )
 
-    @_memoized_configured_property
+    @HasMemoized.memoized_attribute
     def _propkey_to_col(self):
         return dict(
             (
@@ -2155,14 +2154,14 @@ class Mapper(sql_base.HasCacheKey, InspectionAttr):
             for table, columns in self._cols_by_table.items()
         )
 
-    @_memoized_configured_property
+    @HasMemoized.memoized_attribute
     def _pk_keys_by_table(self):
         return dict(
             (table, frozenset([col.key for col in pks]))
             for table, pks in self._pks_by_table.items()
         )
 
-    @_memoized_configured_property
+    @HasMemoized.memoized_attribute
     def _pk_attr_keys_by_table(self):
         return dict(
             (
@@ -2172,7 +2171,7 @@ class Mapper(sql_base.HasCacheKey, InspectionAttr):
             for table, pks in self._pks_by_table.items()
         )
 
-    @_memoized_configured_property
+    @HasMemoized.memoized_attribute
     def _server_default_cols(self):
         return dict(
             (
@@ -2188,7 +2187,7 @@ class Mapper(sql_base.HasCacheKey, InspectionAttr):
             for table, columns in self._cols_by_table.items()
         )
 
-    @_memoized_configured_property
+    @HasMemoized.memoized_attribute
     def _server_default_plus_onupdate_propkeys(self):
         result = set()
 
@@ -2202,7 +2201,7 @@ class Mapper(sql_base.HasCacheKey, InspectionAttr):
 
         return result
 
-    @_memoized_configured_property
+    @HasMemoized.memoized_attribute
     def _server_onupdate_default_cols(self):
         return dict(
             (
@@ -2258,7 +2257,7 @@ class Mapper(sql_base.HasCacheKey, InspectionAttr):
         else:
             return mappers, self._selectable_from_mappers(mappers, innerjoin)
 
-    @_memoized_configured_property
+    @HasMemoized.memoized_attribute
     def _polymorphic_properties(self):
         return list(
             self._iterate_polymorphic_properties(
@@ -2294,7 +2293,7 @@ class Mapper(sql_base.HasCacheKey, InspectionAttr):
                     continue
                 yield c
 
-    @_memoized_configured_property
+    @HasMemoized.memoized_attribute
     def attrs(self):
         """A namespace of all :class:`.MapperProperty` objects
         associated this mapper.
@@ -2332,7 +2331,7 @@ class Mapper(sql_base.HasCacheKey, InspectionAttr):
             configure_mappers()
         return util.ImmutableProperties(self._props)
 
-    @_memoized_configured_property
+    @HasMemoized.memoized_attribute
     def all_orm_descriptors(self):
         """A namespace of all :class:`.InspectionAttr` attributes associated
         with the mapped class.
@@ -2379,7 +2378,7 @@ class Mapper(sql_base.HasCacheKey, InspectionAttr):
             dict(self.class_manager._all_sqla_attributes())
         )
 
-    @_memoized_configured_property
+    @HasMemoized.memoized_attribute
     @util.preload_module("sqlalchemy.orm.descriptor_props")
     def synonyms(self):
         """Return a namespace of all :class:`.SynonymProperty`
@@ -2395,7 +2394,7 @@ class Mapper(sql_base.HasCacheKey, InspectionAttr):
 
         return self._filter_properties(descriptor_props.SynonymProperty)
 
-    @_memoized_configured_property
+    @HasMemoized.memoized_attribute
     def column_attrs(self):
         """Return a namespace of all :class:`.ColumnProperty`
         properties maintained by this :class:`.Mapper`.
@@ -2409,7 +2408,7 @@ class Mapper(sql_base.HasCacheKey, InspectionAttr):
         return self._filter_properties(properties.ColumnProperty)
 
     @util.preload_module("sqlalchemy.orm.relationships")
-    @_memoized_configured_property
+    @HasMemoized.memoized_attribute
     def relationships(self):
         """A namespace of all :class:`.RelationshipProperty` properties
         maintained by this :class:`.Mapper`.
@@ -2436,7 +2435,7 @@ class Mapper(sql_base.HasCacheKey, InspectionAttr):
             util.preloaded.orm_relationships.RelationshipProperty
         )
 
-    @_memoized_configured_property
+    @HasMemoized.memoized_attribute
     @util.preload_module("sqlalchemy.orm.descriptor_props")
     def composites(self):
         """Return a namespace of all :class:`.CompositeProperty`
@@ -2461,7 +2460,7 @@ class Mapper(sql_base.HasCacheKey, InspectionAttr):
             )
         )
 
-    @_memoized_configured_property
+    @HasMemoized.memoized_attribute
     def _get_clause(self):
         """create a "get clause" based on the primary key.  this is used
         by query.get() and many-to-one lazyloads to load this item
@@ -2477,7 +2476,7 @@ class Mapper(sql_base.HasCacheKey, InspectionAttr):
             util.column_dict(params),
         )
 
-    @_memoized_configured_property
+    @HasMemoized.memoized_attribute
     def _equivalent_columns(self):
         """Create a map of all equivalent columns, based on
         the determination of column pairs that are equated to
@@ -2610,7 +2609,7 @@ class Mapper(sql_base.HasCacheKey, InspectionAttr):
             yield m
             m = m.inherits
 
-    @_memoized_configured_property
+    @HasMemoized.memoized_attribute
     def self_and_descendants(self):
         """The collection including this mapper and all descendant mappers.
 
@@ -2737,7 +2736,7 @@ class Mapper(sql_base.HasCacheKey, InspectionAttr):
         )
         return identity_key[1]
 
-    @_memoized_configured_property
+    @HasMemoized.memoized_attribute
     def _persistent_sortkey_fn(self):
         key_fns = [col.type.sort_key_function for col in self.primary_key]
 
@@ -2756,25 +2755,25 @@ class Mapper(sql_base.HasCacheKey, InspectionAttr):
 
         return key
 
-    @_memoized_configured_property
+    @HasMemoized.memoized_attribute
     def _identity_key_props(self):
         return [self._columntoproperty[col] for col in self.primary_key]
 
-    @_memoized_configured_property
+    @HasMemoized.memoized_attribute
     def _all_pk_props(self):
         collection = set()
         for table in self.tables:
             collection.update(self._pks_by_table[table])
         return collection
 
-    @_memoized_configured_property
+    @HasMemoized.memoized_attribute
     def _should_undefer_in_wildcard(self):
         cols = set(self.primary_key)
         if self.polymorphic_on is not None:
             cols.add(self.polymorphic_on)
         return cols
 
-    @_memoized_configured_property
+    @HasMemoized.memoized_attribute
     def _primary_key_propkeys(self):
         return {prop.key for prop in self._all_pk_props}
 
@@ -2993,7 +2992,7 @@ class Mapper(sql_base.HasCacheKey, InspectionAttr):
 
         return q, enable_opt, disable_opt
 
-    @_memoized_configured_property
+    @HasMemoized.memoized_attribute
     def _subclass_load_via_in_mapper(self):
         return self._subclass_load_via_in(self)
 
@@ -3074,11 +3073,11 @@ class Mapper(sql_base.HasCacheKey, InspectionAttr):
                     )
                 )
 
-    @_memoized_configured_property
+    @HasMemoized.memoized_attribute
     def _compiled_cache(self):
         return util.LRUCache(self._compiled_cache_size)
 
-    @_memoized_configured_property
+    @HasMemoized.memoized_attribute
     def _sorted_tables(self):
         table_to_mapper = {}
 
diff --git a/lib/sqlalchemy/orm/relationships.py b/lib/sqlalchemy/orm/relationships.py
index 227543485..8b7a4b549 100644
--- a/lib/sqlalchemy/orm/relationships.py
+++ b/lib/sqlalchemy/orm/relationships.py
@@ -1412,13 +1412,14 @@ class RelationshipProperty(StrategizedProperty):
             if self.property.direction == MANYTOONE:
                 state = attributes.instance_state(other)
 
-                def state_bindparam(x, state, col):
+                def state_bindparam(local_col, state, remote_col):
                     dict_ = state.dict
                     return sql.bindparam(
-                        x,
+                        local_col.key,
+                        type_=local_col.type,
                         unique=True,
                         callable_=self.property._get_attr_w_warn_on_none(
-                            self.property.mapper, state, dict_, col
+                            self.property.mapper, state, dict_, remote_col
                         ),
                     )
 
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.
 
diff --git a/lib/sqlalchemy/util/__init__.py b/lib/sqlalchemy/util/__init__.py
index c0e290cb0..695985a91 100644
--- a/lib/sqlalchemy/util/__init__.py
+++ b/lib/sqlalchemy/util/__init__.py
@@ -119,7 +119,7 @@ from .langhelpers import get_callable_argspec  # noqa
 from .langhelpers import get_cls_kwargs  # noqa
 from .langhelpers import get_func_kwargs  # noqa
 from .langhelpers import getargspec_init  # noqa
-from .langhelpers import group_expirable_memoized_property  # noqa
+from .langhelpers import HasMemoized  # noqa
 from .langhelpers import hybridmethod  # noqa
 from .langhelpers import hybridproperty  # noqa
 from .langhelpers import iterate_attributes  # noqa
@@ -134,8 +134,8 @@ from .langhelpers import NoneType  # noqa
 from .langhelpers import only_once  # noqa
 from .langhelpers import PluginLoader  # noqa
 from .langhelpers import portable_instancemethod  # noqa
-from .langhelpers import preloaded  # noqa
 from .langhelpers import preload_module  # noqa
+from .langhelpers import preloaded  # noqa
 from .langhelpers import quoted_token_parser  # noqa
 from .langhelpers import safe_reraise  # noqa
 from .langhelpers import set_creation_order  # noqa
diff --git a/lib/sqlalchemy/util/langhelpers.py b/lib/sqlalchemy/util/langhelpers.py
index 474463981..7e9906028 100644
--- a/lib/sqlalchemy/util/langhelpers.py
+++ b/lib/sqlalchemy/util/langhelpers.py
@@ -924,27 +924,56 @@ def memoized_instancemethod(fn):
     return update_wrapper(oneshot, fn)
 
 
-class group_expirable_memoized_property(object):
-    """A family of @memoized_properties that can be expired in tandem."""
-
-    def __init__(self, attributes=()):
-        self.attributes = []
-        if attributes:
-            self.attributes.extend(attributes)
-
-    def expire_instance(self, instance):
-        """Expire all memoized properties for *instance*."""
-        stash = instance.__dict__
-        for attribute in self.attributes:
-            stash.pop(attribute, None)
-
-    def __call__(self, fn):
-        self.attributes.append(fn.__name__)
-        return memoized_property(fn)
-
-    def method(self, fn):
-        self.attributes.append(fn.__name__)
-        return memoized_instancemethod(fn)
+class HasMemoized(object):
+    """A class that maintains the names of memoized elements in a
+    collection for easy cache clearing, generative, etc.
+
+    """
+
+    _memoized_keys = frozenset()
+
+    def _reset_memoizations(self):
+        for elem in self._memoized_keys:
+            self.__dict__.pop(elem, None)
+
+    def _assert_no_memoizations(self):
+        for elem in self._memoized_keys:
+            assert elem not in self.__dict__
+
+    class memoized_attribute(object):
+        """A read-only @property that is only evaluated once."""
+
+        def __init__(self, fget, doc=None):
+            self.fget = fget
+            self.__doc__ = doc or fget.__doc__
+            self.__name__ = fget.__name__
+
+        def __get__(self, obj, cls):
+            if obj is None:
+                return self
+            obj.__dict__[self.__name__] = result = self.fget(obj)
+            obj._memoized_keys |= {self.__name__}
+            return result
+
+    @classmethod
+    def memoized_instancemethod(cls, fn):
+        """Decorate a method memoize its return value.
+
+        """
+
+        def oneshot(self, *args, **kw):
+            result = fn(self, *args, **kw)
+
+            def memo(*a, **kw):
+                return result
+
+            memo.__name__ = fn.__name__
+            memo.__doc__ = fn.__doc__
+            self.__dict__[fn.__name__] = memo
+            self._memoized_keys |= {fn.__name__}
+            return result
+
+        return update_wrapper(oneshot, fn)
 
 
 class MemoizedSlots(object):