reorganize pre_session_exec around do_orm_execute

Allow do_orm_execute() events to both receive the complete
state of bind_argments, load_options, update_delete_options
as they do already, but also allow them to *change* all those
things via new execution options.   Options like autoflush,
populate_existing etc. can now be updated within a
do_orm_execute() hook and those changes will take effect
all the way through.

Took a few tries to get something that covers every case here,
in particular horizontal sharding which is consuming those
options as well as using context.invoke(), without excess
complexity.  The good news seems to be that a simple
reorg and replacing the "reentrant" boolean with
"is this before do_orm_execute is invoked" was all that was
needed.

As part of this we add a new "identity_token" option allowing
this option to be controlled from do_orm_execute() as well
as from the outside.

WIP

Fixes: #7837
Change-Id: I087728215edec8d1b1712322ab389e3f52ff76ba

6 files changed, 216 insertions, 157 deletions

diff --git a/lib/sqlalchemy/ext/horizontal_shard.py b/lib/sqlalchemy/ext/horizontal_shard.py
index 69767ad6c..fd53c6046 100644
--- a/lib/sqlalchemy/ext/horizontal_shard.py
+++ b/lib/sqlalchemy/ext/horizontal_shard.py
@@ -13,11 +13,14 @@ distribute queries and persistence operations across multiple databases.
 For a usage example, see the :ref:`examples_sharding` example included in
 the source distribution.
 
-.. legacy:: The horizontal sharding API is not fully updated for the
-   SQLAlchemy 2.0 API, and still relies in part on the
-   legacy :class:`.Query` architecture, in particular as part of the
-   signature for the :paramref:`.ShardedSession.id_chooser` parameter.
-   This may change in a future release.
+.. deepalchemy:: The horizontal sharding extension is an advanced feature,
+   involving a complex statement -> database interaction as well as
+   use of semi-public APIs for non-trivial cases.   Simpler approaches to
+   refering to multiple database "shards", most commonly using a distinct
+   :class:`_orm.Session` per "shard", should always be considered first
+   before using this more complex and less-production-tested system.
+
+
 
 """
 from __future__ import annotations
@@ -38,8 +41,11 @@ from .. import exc
 from .. import inspect
 from .. import util
 from ..orm import PassiveFlag
+from ..orm._typing import OrmExecuteOptionsParameter
 from ..orm.mapper import Mapper
 from ..orm.query import Query
+from ..orm.session import _BindArguments
+from ..orm.session import _PKIdentityArgument
 from ..orm.session import Session
 from ..util.typing import Protocol
 
@@ -80,6 +86,20 @@ class ShardChooser(Protocol):
         ...
 
 
+class IdentityChooser(Protocol):
+    def __call__(
+        self,
+        mapper: Mapper[_T],
+        primary_key: _PKIdentityArgument,
+        *,
+        lazy_loaded_from: Optional[InstanceState[Any]],
+        execution_options: OrmExecuteOptionsParameter,
+        bind_arguments: _BindArguments,
+        **kw: Any,
+    ) -> Any:
+        ...
+
+
 class ShardedQuery(Query[_T]):
     """Query class used with :class:`.ShardedSession`.
 
@@ -94,8 +114,7 @@ class ShardedQuery(Query[_T]):
         super().__init__(*args, **kwargs)
         assert isinstance(self.session, ShardedSession)
 
-        self.id_chooser = self.session.id_chooser
-        self.query_chooser = self.session.query_chooser
+        self.identity_chooser = self.session.identity_chooser
         self.execute_chooser = self.session.execute_chooser
         self._shard_id = None
 
@@ -119,19 +138,22 @@ class ShardedQuery(Query[_T]):
 
 class ShardedSession(Session):
     shard_chooser: ShardChooser
-    id_chooser: Callable[[Query[Any], Iterable[Any]], Iterable[Any]]
+    identity_chooser: IdentityChooser
     execute_chooser: Callable[[ORMExecuteState], Iterable[Any]]
 
     def __init__(
         self,
         shard_chooser: ShardChooser,
-        id_chooser: Callable[[Query[_T], Iterable[_T]], Iterable[Any]],
+        identity_chooser: Optional[IdentityChooser] = None,
         execute_chooser: Optional[
             Callable[[ORMExecuteState], Iterable[Any]]
         ] = None,
         shards: Optional[Dict[str, Any]] = None,
         query_cls: Type[Query[_T]] = ShardedQuery,
         *,
+        id_chooser: Optional[
+            Callable[[Query[_T], Iterable[_T]], Iterable[Any]]
+        ] = None,
         query_chooser: Optional[Callable[[Executable], Iterable[Any]]] = None,
         **kwargs: Any,
     ) -> None:
@@ -171,12 +193,41 @@ class ShardedSession(Session):
             self, "do_orm_execute", execute_and_instances, retval=True
         )
         self.shard_chooser = shard_chooser
-        self.id_chooser = id_chooser
+
+        if id_chooser:
+            _id_chooser = id_chooser
+            util.warn_deprecated(
+                "The ``id_chooser`` parameter is deprecated; "
+                "please use ``identity_chooser``.",
+                "2.0",
+            )
+
+            def _legacy_identity_chooser(
+                mapper: Mapper[_T],
+                primary_key: _PKIdentityArgument,
+                *,
+                lazy_loaded_from: Optional[InstanceState[Any]],
+                execution_options: OrmExecuteOptionsParameter,
+                bind_arguments: _BindArguments,
+                **kw: Any,
+            ) -> Any:
+                q = self.query(mapper)
+                if lazy_loaded_from:
+                    q = q._set_lazyload_from(lazy_loaded_from)
+                return _id_chooser(q, primary_key)
+
+            self.identity_chooser = _legacy_identity_chooser
+        elif identity_chooser:
+            self.identity_chooser = identity_chooser
+        else:
+            raise exc.ArgumentError(
+                "identity_chooser or id_chooser is required"
+            )
 
         if query_chooser:
             _query_chooser = query_chooser
             util.warn_deprecated(
-                "The ``query_choser`` parameter is deprecated; "
+                "The ``query_chooser`` parameter is deprecated; "
                 "please use ``execute_chooser``.",
                 "1.4",
             )
@@ -199,7 +250,6 @@ class ShardedSession(Session):
                 "execute_chooser or query_chooser is required"
             )
         self.execute_chooser = execute_chooser
-        self.query_chooser = query_chooser
         self.__shards: Dict[_ShardKey, _SessionBind] = {}
         if shards is not None:
             for k in shards:
@@ -212,6 +262,8 @@ class ShardedSession(Session):
         identity_token: Optional[Any] = None,
         passive: PassiveFlag = PassiveFlag.PASSIVE_OFF,
         lazy_loaded_from: Optional[InstanceState[Any]] = None,
+        execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
+        bind_arguments: Optional[_BindArguments] = None,
         **kw: Any,
     ) -> Union[Optional[_O], LoaderCallableStatus]:
         """override the default :meth:`.Session._identity_lookup` method so
@@ -233,10 +285,13 @@ class ShardedSession(Session):
 
             return obj
         else:
-            q = self.query(mapper)
-            if lazy_loaded_from:
-                q = q._set_lazyload_from(lazy_loaded_from)
-            for shard_id in self.id_chooser(q, primary_key_identity):
+            for shard_id in self.identity_chooser(
+                mapper,
+                primary_key_identity,
+                lazy_loaded_from=lazy_loaded_from,
+                execution_options=execution_options,
+                bind_arguments=dict(bind_arguments) if bind_arguments else {},
+            ):
                 obj2 = super()._identity_lookup(
                     mapper,
                     primary_key_identity,
@@ -325,11 +380,6 @@ class ShardedSession(Session):
 def execute_and_instances(
     orm_context: ORMExecuteState,
 ) -> Union[Result[_T], IteratorResult[_TP]]:
-    update_options: Union[
-        None,
-        BulkUDCompileState.default_update_options,
-        Type[BulkUDCompileState.default_update_options],
-    ]
     active_options: Union[
         None,
         QueryContext.default_load_options,
@@ -337,58 +387,30 @@ def execute_and_instances(
         BulkUDCompileState.default_update_options,
         Type[BulkUDCompileState.default_update_options],
     ]
-    load_options: Union[
-        None,
-        QueryContext.default_load_options,
-        Type[QueryContext.default_load_options],
-    ]
 
     if orm_context.is_select:
-        load_options = active_options = orm_context.load_options
-        update_options = None
+        active_options = orm_context.load_options
 
     elif orm_context.is_update or orm_context.is_delete:
-        load_options = None
-        update_options = active_options = orm_context.update_delete_options
+        active_options = orm_context.update_delete_options
     else:
-        load_options = update_options = active_options = None
+        active_options = None
 
     session = orm_context.session
     assert isinstance(session, ShardedSession)
 
     def iter_for_shard(
         shard_id: str,
-        load_options: Union[
-            None,
-            QueryContext.default_load_options,
-            Type[QueryContext.default_load_options],
-        ],
-        update_options: Union[
-            None,
-            BulkUDCompileState.default_update_options,
-            Type[BulkUDCompileState.default_update_options],
-        ],
     ) -> Union[Result[_T], IteratorResult[_TP]]:
-        execution_options = dict(orm_context.local_execution_options)
 
         bind_arguments = dict(orm_context.bind_arguments)
         bind_arguments["shard_id"] = shard_id
 
-        if orm_context.is_select:
-            assert load_options is not None
-            load_options += {"_refresh_identity_token": shard_id}
-            execution_options["_sa_orm_load_options"] = load_options
-        elif orm_context.is_update or orm_context.is_delete:
-            assert update_options is not None
-            update_options += {"_refresh_identity_token": shard_id}
-            execution_options["_sa_orm_update_options"] = update_options
-
-        return orm_context.invoke_statement(
-            bind_arguments=bind_arguments, execution_options=execution_options
-        )
+        orm_context.update_execution_options(identity_token=shard_id)
+        return orm_context.invoke_statement(bind_arguments=bind_arguments)
 
-    if active_options and active_options._refresh_identity_token is not None:
-        shard_id = active_options._refresh_identity_token
+    if active_options and active_options._identity_token is not None:
+        shard_id = active_options._identity_token
     elif "_sa_shard_id" in orm_context.execution_options:
         shard_id = orm_context.execution_options["_sa_shard_id"]
     elif "shard_id" in orm_context.bind_arguments:
@@ -397,10 +419,10 @@ def execute_and_instances(
         shard_id = None
 
     if shard_id is not None:
-        return iter_for_shard(shard_id, load_options, update_options)
+        return iter_for_shard(shard_id)
     else:
         partial = []
         for shard_id in session.execute_chooser(orm_context):
-            result_ = iter_for_shard(shard_id, load_options, update_options)
+            result_ = iter_for_shard(shard_id)
             partial.append(result_)
         return partial[0].merge(*partial[1:])
diff --git a/lib/sqlalchemy/orm/bulk_persistence.py b/lib/sqlalchemy/orm/bulk_persistence.py
index 181dbd4a2..805bfdc65 100644
--- a/lib/sqlalchemy/orm/bulk_persistence.py
+++ b/lib/sqlalchemy/orm/bulk_persistence.py
@@ -555,7 +555,7 @@ class BulkUDCompileState(ORMDMLState):
         _resolved_values = EMPTY_DICT
         _eval_condition = None
         _matched_rows = None
-        _refresh_identity_token = None
+        _identity_token = None
 
     @classmethod
     def can_use_returning(
@@ -577,10 +577,8 @@ class BulkUDCompileState(ORMDMLState):
         params,
         execution_options,
         bind_arguments,
-        is_reentrant_invoke,
+        is_pre_event,
     ):
-        if is_reentrant_invoke:
-            return statement, execution_options
 
         (
             update_options,
@@ -590,6 +588,7 @@ class BulkUDCompileState(ORMDMLState):
             {
                 "synchronize_session",
                 "autoflush",
+                "identity_token",
                 "is_delete_using",
                 "is_update_from",
                 "dml_strategy",
@@ -637,55 +636,56 @@ class BulkUDCompileState(ORMDMLState):
                     "for 'bulk' ORM updates (i.e. multiple parameter sets)"
                 )
 
-        if update_options._autoflush:
-            session._autoflush()
-
-        if update_options._dml_strategy == "orm":
+        if not is_pre_event:
+            if update_options._autoflush:
+                session._autoflush()
 
-            if update_options._synchronize_session == "auto":
-                update_options = cls._do_pre_synchronize_auto(
-                    session,
-                    statement,
-                    params,
-                    execution_options,
-                    bind_arguments,
-                    update_options,
-                )
-            elif update_options._synchronize_session == "evaluate":
-                update_options = cls._do_pre_synchronize_evaluate(
-                    session,
-                    statement,
-                    params,
-                    execution_options,
-                    bind_arguments,
-                    update_options,
-                )
-            elif update_options._synchronize_session == "fetch":
-                update_options = cls._do_pre_synchronize_fetch(
-                    session,
-                    statement,
-                    params,
-                    execution_options,
-                    bind_arguments,
-                    update_options,
-                )
-        elif update_options._dml_strategy == "bulk":
-            if update_options._synchronize_session == "auto":
-                update_options += {"_synchronize_session": "evaluate"}
+            if update_options._dml_strategy == "orm":
 
-        # indicators from the "pre exec" step that are then
-        # added to the DML statement, which will also be part of the cache
-        # key.  The compile level create_for_statement() method will then
-        # consume these at compiler time.
-        statement = statement._annotate(
-            {
-                "synchronize_session": update_options._synchronize_session,
-                "is_delete_using": update_options._is_delete_using,
-                "is_update_from": update_options._is_update_from,
-                "dml_strategy": update_options._dml_strategy,
-                "can_use_returning": update_options._can_use_returning,
-            }
-        )
+                if update_options._synchronize_session == "auto":
+                    update_options = cls._do_pre_synchronize_auto(
+                        session,
+                        statement,
+                        params,
+                        execution_options,
+                        bind_arguments,
+                        update_options,
+                    )
+                elif update_options._synchronize_session == "evaluate":
+                    update_options = cls._do_pre_synchronize_evaluate(
+                        session,
+                        statement,
+                        params,
+                        execution_options,
+                        bind_arguments,
+                        update_options,
+                    )
+                elif update_options._synchronize_session == "fetch":
+                    update_options = cls._do_pre_synchronize_fetch(
+                        session,
+                        statement,
+                        params,
+                        execution_options,
+                        bind_arguments,
+                        update_options,
+                    )
+            elif update_options._dml_strategy == "bulk":
+                if update_options._synchronize_session == "auto":
+                    update_options += {"_synchronize_session": "evaluate"}
+
+            # indicators from the "pre exec" step that are then
+            # added to the DML statement, which will also be part of the cache
+            # key.  The compile level create_for_statement() method will then
+            # consume these at compiler time.
+            statement = statement._annotate(
+                {
+                    "synchronize_session": update_options._synchronize_session,
+                    "is_delete_using": update_options._is_delete_using,
+                    "is_update_from": update_options._is_update_from,
+                    "dml_strategy": update_options._dml_strategy,
+                    "can_use_returning": update_options._can_use_returning,
+                }
+            )
 
         return (
             statement,
@@ -836,7 +836,7 @@ class BulkUDCompileState(ORMDMLState):
             if state.mapper.isa(mapper) and not state.expired
         ]
 
-        identity_token = update_options._refresh_identity_token
+        identity_token = update_options._identity_token
         if identity_token is not None:
             raw_data = [
                 (obj, state, dict_)
@@ -1091,7 +1091,7 @@ class BulkORMInsert(ORMDMLState, InsertDMLState):
         params,
         execution_options,
         bind_arguments,
-        is_reentrant_invoke,
+        is_pre_event,
     ):
 
         (
@@ -1143,7 +1143,7 @@ class BulkORMInsert(ORMDMLState, InsertDMLState):
                     context._orm_load_exec_options
                 )
 
-        if insert_options._autoflush:
+        if not is_pre_event and insert_options._autoflush:
             session._autoflush()
 
         statement = statement._annotate(
@@ -1577,7 +1577,7 @@ class BulkORMUpdate(BulkUDCompileState, UpdateDMLState):
         for param in params:
             identity_key = mapper.identity_key_from_primary_key(
                 (param[key] for key in pk_keys),
-                update_options._refresh_identity_token,
+                update_options._identity_token,
             )
             state = identity_map.fast_get_state(identity_key)
             if not state:
@@ -1635,7 +1635,7 @@ class BulkORMUpdate(BulkUDCompileState, UpdateDMLState):
             )
 
             matched_rows = [
-                tuple(row) + (update_options._refresh_identity_token,)
+                tuple(row) + (update_options._identity_token,)
                 for row in pk_rows
             ]
         else:
@@ -1651,8 +1651,8 @@ class BulkORMUpdate(BulkUDCompileState, UpdateDMLState):
                 for primary_key, identity_token in [
                     (row[0:-1], row[-1]) for row in matched_rows
                 ]
-                if update_options._refresh_identity_token is None
-                or identity_token == update_options._refresh_identity_token
+                if update_options._identity_token is None
+                or identity_token == update_options._identity_token
             ]
             if identity_key in session.identity_map
         ]
@@ -1912,7 +1912,7 @@ class BulkORMDelete(BulkUDCompileState, DeleteDMLState):
             )
 
             matched_rows = [
-                tuple(row) + (update_options._refresh_identity_token,)
+                tuple(row) + (update_options._identity_token,)
                 for row in pk_rows
             ]
         else:
diff --git a/lib/sqlalchemy/orm/context.py b/lib/sqlalchemy/orm/context.py
index 3bd8b02a7..b3478b83e 100644
--- a/lib/sqlalchemy/orm/context.py
+++ b/lib/sqlalchemy/orm/context.py
@@ -135,7 +135,7 @@ class QueryContext:
         _version_check = False
         _invoke_all_eagers = True
         _autoflush = True
-        _refresh_identity_token = None
+        _identity_token = None
         _yield_per = None
         _refresh_state = None
         _lazy_loaded_from = None
@@ -194,14 +194,14 @@ class QueryContext:
         self.version_check = load_options._version_check
         self.refresh_state = load_options._refresh_state
         self.yield_per = load_options._yield_per
-        self.identity_token = load_options._refresh_identity_token
+        self.identity_token = load_options._identity_token
 
     def _get_top_level_context(self) -> QueryContext:
         return self.top_level_context or self
 
 
 _orm_load_exec_options = util.immutabledict(
-    {"_result_disable_adapt_to_context": True, "future_result": True}
+    {"_result_disable_adapt_to_context": True}
 )
 
 
@@ -235,7 +235,7 @@ class AbstractORMCompileState(CompileState):
         params,
         execution_options,
         bind_arguments,
-        is_reentrant_invoke,
+        is_pre_event,
     ):
         raise NotImplementedError()
 
@@ -384,11 +384,11 @@ class ORMCompileState(AbstractORMCompileState):
         params,
         execution_options,
         bind_arguments,
-        is_reentrant_invoke,
+        is_pre_event,
     ):
-        if is_reentrant_invoke:
-            return statement, execution_options
 
+        # consume result-level load_options.  These may have been set up
+        # in an ORMExecuteState hook
         (
             load_options,
             execution_options,
@@ -398,26 +398,24 @@ class ORMCompileState(AbstractORMCompileState):
                 "populate_existing",
                 "autoflush",
                 "yield_per",
+                "identity_token",
                 "sa_top_level_orm_context",
             },
             execution_options,
             statement._execution_options,
         )
+
         # default execution options for ORM results:
         # 1. _result_disable_adapt_to_context=True
         #    this will disable the ResultSetMetadata._adapt_to_context()
         #    step which we don't need, as we have result processors cached
         #    against the original SELECT statement before caching.
-        # 2. future_result=True.  The ORM should **never** resolve columns
-        #    in a result set based on names, only on Column objects that
-        #    are correctly adapted to the context.   W the legacy result
-        #    it will still attempt name-based resolution and also emit a
-        #    warning.
         if not execution_options:
             execution_options = _orm_load_exec_options
         else:
             execution_options = execution_options.union(_orm_load_exec_options)
 
+        # would have been placed here by legacy Query only
         if load_options._yield_per:
             execution_options = execution_options.union(
                 {"yield_per": load_options._yield_per}
@@ -457,7 +455,7 @@ class ORMCompileState(AbstractORMCompileState):
             if plugin_subject:
                 bind_arguments["mapper"] = plugin_subject.mapper
 
-        if load_options._autoflush:
+        if not is_pre_event and load_options._autoflush:
             session._autoflush()
 
         return statement, execution_options
@@ -483,6 +481,7 @@ class ORMCompileState(AbstractORMCompileState):
         load_options = execution_options.get(
             "_sa_orm_load_options", QueryContext.default_load_options
         )
+
         if compile_state.compile_options._is_star:
             return result
 
@@ -3119,6 +3118,6 @@ class _IdentityTokenEntity(_ORMColumnEntity):
 
     def row_processor(self, context, result):
         def getter(row):
-            return context.load_options._refresh_identity_token
+            return context.load_options._identity_token
 
         return getter, self._label_name, self._extra_entities
diff --git a/lib/sqlalchemy/orm/loading.py b/lib/sqlalchemy/orm/loading.py
index 6e7695f86..f331cd63b 100644
--- a/lib/sqlalchemy/orm/loading.py
+++ b/lib/sqlalchemy/orm/loading.py
@@ -701,7 +701,7 @@ def _set_get_options(
     if only_load_props:
         compile_options["_only_load_props"] = frozenset(only_load_props)
     if identity_token:
-        load_options["_refresh_identity_token"] = identity_token
+        load_options["_identity_token"] = identity_token
 
     if load_options:
         load_opt += load_options
diff --git a/lib/sqlalchemy/orm/query.py b/lib/sqlalchemy/orm/query.py
index 01db08eb4..d2bd930ff 100644
--- a/lib/sqlalchemy/orm/query.py
+++ b/lib/sqlalchemy/orm/query.py
@@ -470,7 +470,7 @@ class Query(
         if only_load_props:
             compile_options["_only_load_props"] = frozenset(only_load_props)
         if identity_token:
-            load_options["_refresh_identity_token"] = identity_token
+            load_options["_identity_token"] = identity_token
 
         if load_options:
             self.load_options += load_options
diff --git a/lib/sqlalchemy/orm/session.py b/lib/sqlalchemy/orm/session.py
index bf3df0599..8b5f7c88a 100644
--- a/lib/sqlalchemy/orm/session.py
+++ b/lib/sqlalchemy/orm/session.py
@@ -267,6 +267,7 @@ class ORMExecuteState(util.MemoizedSlots):
         "execution_options",
         "local_execution_options",
         "bind_arguments",
+        "identity_token",
         "_compile_state_cls",
         "_starting_event_idx",
         "_events_todo",
@@ -579,9 +580,8 @@ class ORMExecuteState(util.MemoizedSlots):
     def _is_crud(self) -> bool:
         return isinstance(self.statement, (dml.Update, dml.Delete))
 
-    def update_execution_options(self, **opts: _ExecuteOptions) -> None:
+    def update_execution_options(self, **opts: Any) -> None:
         """Update the local execution options with new values."""
-        # TODO: no coverage
         self.local_execution_options = self.local_execution_options.union(opts)
 
     def _orm_compile_options(
@@ -1912,27 +1912,10 @@ class Session(_SessionClassMethods, EventTarget):
                 )
         else:
             compile_state_cls = None
+            bind_arguments.setdefault("clause", statement)
 
         execution_options = util.coerce_to_immutabledict(execution_options)
 
-        if compile_state_cls is not None:
-            (
-                statement,
-                execution_options,
-            ) = compile_state_cls.orm_pre_session_exec(
-                self,
-                statement,
-                params,
-                execution_options,
-                bind_arguments,
-                _parent_execute_state is not None,
-            )
-        else:
-            bind_arguments.setdefault("clause", statement)
-            execution_options = execution_options.union(
-                {"future_result": True}
-            )
-
         if _parent_execute_state:
             events_todo = _parent_execute_state._remaining_events()
         else:
@@ -1941,6 +1924,25 @@ class Session(_SessionClassMethods, EventTarget):
                 events_todo = list(events_todo) + [_add_event]
 
         if events_todo:
+            if compile_state_cls is not None:
+                # for event handlers, do the orm_pre_session_exec
+                # pass ahead of the event handlers, so that things like
+                # .load_options, .update_delete_options etc. are populated.
+                # is_pre_event=True allows the hook to hold off on things
+                # it doesn't want to do twice, including autoflush as well
+                # as "pre fetch" for DML, etc.
+                (
+                    statement,
+                    execution_options,
+                ) = compile_state_cls.orm_pre_session_exec(
+                    self,
+                    statement,
+                    params,
+                    execution_options,
+                    bind_arguments,
+                    True,
+                )
+
             orm_exec_state = ORMExecuteState(
                 self,
                 statement,
@@ -1962,6 +1964,24 @@ class Session(_SessionClassMethods, EventTarget):
             statement = orm_exec_state.statement
             execution_options = orm_exec_state.local_execution_options
 
+        if compile_state_cls is not None:
+            # now run orm_pre_session_exec() "for real".   if there were
+            # event hooks, this will re-run the steps that interpret
+            # new execution_options into load_options / update_delete_options,
+            # which we assume the event hook might have updated.
+            # autoflush will also be invoked in this step if enabled.
+            (
+                statement,
+                execution_options,
+            ) = compile_state_cls.orm_pre_session_exec(
+                self,
+                statement,
+                params,
+                execution_options,
+                bind_arguments,
+                False,
+            )
+
         bind = self.get_bind(**bind_arguments)
 
         conn = self._connection_for_bind(bind)
@@ -2379,6 +2399,7 @@ class Session(_SessionClassMethods, EventTarget):
         bind: Optional[_SessionBind] = None,
         _sa_skip_events: Optional[bool] = None,
         _sa_skip_for_implicit_returning: bool = False,
+        **kw: Any,
     ) -> Union[Engine, Connection]:
         """Return a "bind" to which this :class:`.Session` is bound.
 
@@ -2653,6 +2674,8 @@ class Session(_SessionClassMethods, EventTarget):
         identity_token: Any = None,
         passive: PassiveFlag = PassiveFlag.PASSIVE_OFF,
         lazy_loaded_from: Optional[InstanceState[Any]] = None,
+        execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
+        bind_arguments: Optional[_BindArguments] = None,
     ) -> Union[Optional[_O], LoaderCallableStatus]:
         """Locate an object in the identity map.
 
@@ -3262,6 +3285,7 @@ class Session(_SessionClassMethods, EventTarget):
         with_for_update: Optional[ForUpdateArg] = None,
         identity_token: Optional[Any] = None,
         execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
+        bind_arguments: Optional[_BindArguments] = None,
     ) -> Optional[_O]:
         """Return an instance based on the given primary key identifier,
         or ``None`` if not found.
@@ -3355,6 +3379,13 @@ class Session(_SessionClassMethods, EventTarget):
             :ref:`orm_queryguide_execution_options` - ORM-specific execution
             options
 
+        :param bind_arguments: dictionary of additional arguments to determine
+         the bind.  May include "mapper", "bind", or other custom arguments.
+         Contents of this dictionary are passed to the
+         :meth:`.Session.get_bind` method.
+
+         .. versionadded: 2.0.0b5
+
         :return: The object instance, or ``None``.
 
         """
@@ -3367,6 +3398,7 @@ class Session(_SessionClassMethods, EventTarget):
             with_for_update=with_for_update,
             identity_token=identity_token,
             execution_options=execution_options,
+            bind_arguments=bind_arguments,
         )
 
     def _get_impl(
@@ -3379,7 +3411,8 @@ class Session(_SessionClassMethods, EventTarget):
         populate_existing: bool = False,
         with_for_update: Optional[ForUpdateArg] = None,
         identity_token: Optional[Any] = None,
-        execution_options: Optional[OrmExecuteOptionsParameter] = None,
+        execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
+        bind_arguments: Optional[_BindArguments] = None,
     ) -> Optional[_O]:
 
         # convert composite types to individual args
@@ -3453,7 +3486,11 @@ class Session(_SessionClassMethods, EventTarget):
         ):
 
             instance = self._identity_lookup(
-                mapper, primary_key_identity, identity_token=identity_token
+                mapper,
+                primary_key_identity,
+                identity_token=identity_token,
+                execution_options=execution_options,
+                bind_arguments=bind_arguments,
             )
 
             if instance is not None:
@@ -3484,13 +3521,14 @@ class Session(_SessionClassMethods, EventTarget):
 
         if options:
             statement = statement.options(*options)
-        if execution_options:
-            statement = statement.execution_options(**execution_options)
         return db_load_fn(
             self,
             statement,
             primary_key_identity,
             load_options=load_options,
+            identity_token=identity_token,
+            execution_options=execution_options,
+            bind_arguments=bind_arguments,
         )
 
     def merge(