Merge "create new approach for deeply nested post loader options" into main

5 files changed, 393 insertions, 72 deletions

diff --git a/lib/sqlalchemy/orm/context.py b/lib/sqlalchemy/orm/context.py
index 8676f828e..a468244e9 100644
--- a/lib/sqlalchemy/orm/context.py
+++ b/lib/sqlalchemy/orm/context.py
@@ -97,6 +97,7 @@ LABEL_STYLE_LEGACY_ORM = SelectLabelStyle.LABEL_STYLE_LEGACY_ORM
 
 class QueryContext:
     __slots__ = (
+        "top_level_context",
         "compile_state",
         "query",
         "params",
@@ -136,6 +137,7 @@ class QueryContext:
         _refresh_state = None
         _lazy_loaded_from = None
         _legacy_uniquing = False
+        _sa_top_level_orm_context = None
 
     def __init__(
         self,
@@ -159,6 +161,7 @@ class QueryContext:
         self.loaders_require_buffering = False
         self.loaders_require_uniquing = False
         self.params = params
+        self.top_level_context = load_options._sa_top_level_orm_context
 
         self.propagated_loader_options = tuple(
             # issue 7447.
@@ -194,6 +197,9 @@ class QueryContext:
         self.yield_per = load_options._yield_per
         self.identity_token = load_options._refresh_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}
@@ -327,11 +333,15 @@ class ORMCompileState(CompileState):
             execution_options,
         ) = QueryContext.default_load_options.from_execution_options(
             "_sa_orm_load_options",
-            {"populate_existing", "autoflush", "yield_per"},
+            {
+                "populate_existing",
+                "autoflush",
+                "yield_per",
+                "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()
@@ -357,6 +367,21 @@ class ORMCompileState(CompileState):
                 }
             )
 
+        if (
+            getattr(statement._compile_options, "_current_path", None)
+            and len(statement._compile_options._current_path) > 10
+            and execution_options.get("compiled_cache", True) is not None
+        ):
+            util.warn(
+                "Loader depth for query is excessively deep; caching will "
+                "be disabled for additional loaders.  Consider using the "
+                "recursion_depth feature for deeply nested recursive eager "
+                "loaders."
+            )
+            execution_options = execution_options.union(
+                {"compiled_cache": None}
+            )
+
         bind_arguments["clause"] = statement
 
         # new in 1.4 - the coercions system is leveraged to allow the
diff --git a/lib/sqlalchemy/orm/loading.py b/lib/sqlalchemy/orm/loading.py
index 1a5ea5fe6..5d78a5580 100644
--- a/lib/sqlalchemy/orm/loading.py
+++ b/lib/sqlalchemy/orm/loading.py
@@ -89,7 +89,13 @@ def instances(cursor: CursorResult[Any], context: QueryContext) -> Result[Any]:
     """
 
     context.runid = _new_runid()
-    context.post_load_paths = {}
+
+    if context.top_level_context:
+        is_top_level = False
+        context.post_load_paths = context.top_level_context.post_load_paths
+    else:
+        is_top_level = True
+        context.post_load_paths = {}
 
     compile_state = context.compile_state
     filtered = compile_state._has_mapper_entities
@@ -190,8 +196,28 @@ def instances(cursor: CursorResult[Any], context: QueryContext) -> Result[Any]:
                     tuple([proc(row) for proc in process]) for row in fetch
                 ]
 
-            for path, post_load in context.post_load_paths.items():
-                post_load.invoke(context, path)
+            # if we are the originating load from a query, meaning we
+            # aren't being called as a result of a nested "post load",
+            # iterate through all the collected post loaders and fire them
+            # off.  Previously this used to work recursively, however that
+            # prevented deeply nested structures from being loadable
+            if is_top_level:
+                if yield_per:
+                    # if using yield per, memoize the state of the
+                    # collection so that it can be restored
+                    top_level_post_loads = list(
+                        context.post_load_paths.items()
+                    )
+
+                while context.post_load_paths:
+                    post_loads = list(context.post_load_paths.items())
+                    context.post_load_paths.clear()
+                    for path, post_load in post_loads:
+                        post_load.invoke(context, path)
+
+                if yield_per:
+                    context.post_load_paths.clear()
+                    context.post_load_paths.update(top_level_post_loads)
 
             yield rows
 
@@ -747,7 +773,6 @@ def _instance_processor(
             "quick": [],
             "deferred": [],
             "expire": [],
-            "delayed": [],
             "existing": [],
             "eager": [],
         }
@@ -1180,8 +1205,7 @@ def _populate_full(
 
         for key, populator in populators["new"]:
             populator(state, dict_, row)
-        for key, populator in populators["delayed"]:
-            populator(state, dict_, row)
+
     elif load_path != state.load_path:
         # new load path, e.g. object is present in more than one
         # column position in a series of rows
@@ -1233,9 +1257,7 @@ def _populate_partial(
         for key, populator in populators["new"]:
             if key in to_load:
                 populator(state, dict_, row)
-        for key, populator in populators["delayed"]:
-            if key in to_load:
-                populator(state, dict_, row)
+
     for key, populator in populators["eager"]:
         if key not in unloaded:
             populator(state, dict_, row)
@@ -1371,14 +1393,23 @@ class PostLoad:
         if not self.states:
             return
         path = path_registry.PathRegistry.coerce(path)
-        for token, limit_to_mapper, loader, arg, kw in self.loaders.values():
+        for (
+            effective_context,
+            token,
+            limit_to_mapper,
+            loader,
+            arg,
+            kw,
+        ) in self.loaders.values():
             states = [
                 (state, overwrite)
                 for state, overwrite in self.states.items()
                 if state.manager.mapper.isa(limit_to_mapper)
             ]
             if states:
-                loader(context, path, states, self.load_keys, *arg, **kw)
+                loader(
+                    effective_context, path, states, self.load_keys, *arg, **kw
+                )
         self.states.clear()
 
     @classmethod
@@ -1403,7 +1434,14 @@ class PostLoad:
             pl = context.post_load_paths[path.path]
         else:
             pl = context.post_load_paths[path.path] = PostLoad()
-        pl.loaders[token] = (token, limit_to_mapper, loader_callable, arg, kw)
+        pl.loaders[token] = (
+            context,
+            token,
+            limit_to_mapper,
+            loader_callable,
+            arg,
+            kw,
+        )
 
 
 def load_scalar_attributes(mapper, state, attribute_names, passive):
diff --git a/lib/sqlalchemy/orm/path_registry.py b/lib/sqlalchemy/orm/path_registry.py
index 36c14a672..8a51ded5f 100644
--- a/lib/sqlalchemy/orm/path_registry.py
+++ b/lib/sqlalchemy/orm/path_registry.py
@@ -393,6 +393,9 @@ class RootRegistry(CreatesToken):
                     f"invalid argument for RootRegistry.__getitem__: {entity}"
                 )
 
+    def _truncate_recursive(self) -> RootRegistry:
+        return self
+
     if not TYPE_CHECKING:
         __getitem__ = _getitem
 
@@ -584,6 +587,17 @@ class PropRegistry(PathRegistry):
         self._default_path_loader_key = self.prop._default_path_loader_key
         self._loader_key = ("loader", self.natural_path)
 
+    def _truncate_recursive(self) -> PropRegistry:
+        earliest = None
+        for i, token in enumerate(reversed(self.path[:-1])):
+            if token is self.prop:
+                earliest = i
+
+        if earliest is None:
+            return self
+        else:
+            return self.coerce(self.path[0 : -(earliest + 1)])  # type: ignore
+
     @property
     def entity_path(self) -> AbstractEntityRegistry:
         assert self.entity is not None
@@ -663,6 +677,9 @@ class AbstractEntityRegistry(CreatesToken):
             # self.natural_path = parent.natural_path + (entity, )
             self.natural_path = self.path
 
+    def _truncate_recursive(self) -> AbstractEntityRegistry:
+        return self.parent._truncate_recursive()[self.entity]
+
     @property
     def root_entity(self) -> _InternalEntityType[Any]:
         return cast("_InternalEntityType[Any]", self.path[0])
diff --git a/lib/sqlalchemy/orm/strategies.py b/lib/sqlalchemy/orm/strategies.py
index c4c0fb180..db9dcffdc 100644
--- a/lib/sqlalchemy/orm/strategies.py
+++ b/lib/sqlalchemy/orm/strategies.py
@@ -43,6 +43,7 @@ from .interfaces import LoaderStrategy
 from .interfaces import StrategizedProperty
 from .session import _state_session
 from .state import InstanceState
+from .strategy_options import Load
 from .util import _none_set
 from .util import AliasedClass
 from .. import event
@@ -830,7 +831,16 @@ class LazyLoader(
             "'%s' is not available due to lazy='%s'" % (self, lazy)
         )
 
-    def _load_for_state(self, state, passive, loadopt=None, extra_criteria=()):
+    def _load_for_state(
+        self,
+        state,
+        passive,
+        loadopt=None,
+        extra_criteria=(),
+        extra_options=(),
+        alternate_effective_path=None,
+        execution_options=util.EMPTY_DICT,
+    ):
         if not state.key and (
             (
                 not self.parent_property.load_on_pending
@@ -929,6 +939,9 @@ class LazyLoader(
             passive,
             loadopt,
             extra_criteria,
+            extra_options,
+            alternate_effective_path,
+            execution_options,
         )
 
     def _get_ident_for_use_get(self, session, state, passive):
@@ -955,6 +968,9 @@ class LazyLoader(
         passive,
         loadopt,
         extra_criteria,
+        extra_options,
+        alternate_effective_path,
+        execution_options,
     ):
         strategy_options = util.preloaded.orm_strategy_options
 
@@ -986,7 +1002,10 @@ class LazyLoader(
         use_get = self.use_get
 
         if state.load_options or (loadopt and loadopt._extra_criteria):
-            effective_path = state.load_path[self.parent_property]
+            if alternate_effective_path is None:
+                effective_path = state.load_path[self.parent_property]
+            else:
+                effective_path = alternate_effective_path[self.parent_property]
 
             opts = state.load_options
 
@@ -997,10 +1016,16 @@ class LazyLoader(
                 )
 
             stmt._with_options = opts
-        else:
+        elif alternate_effective_path is None:
             # this path is used if there are not already any options
             # in the query, but an event may want to add them
             effective_path = state.mapper._path_registry[self.parent_property]
+        else:
+            # added by immediateloader
+            effective_path = alternate_effective_path[self.parent_property]
+
+        if extra_options:
+            stmt._with_options += extra_options
 
         stmt._compile_options += {"_current_path": effective_path}
 
@@ -1009,7 +1034,11 @@ class LazyLoader(
                 self._invoke_raise_load(state, passive, "raise_on_sql")
 
             return loading.load_on_pk_identity(
-                session, stmt, primary_key_identity, load_options=load_options
+                session,
+                stmt,
+                primary_key_identity,
+                load_options=load_options,
+                execution_options=execution_options,
             )
 
         if self._order_by:
@@ -1036,9 +1065,18 @@ class LazyLoader(
 
         lazy_clause, params = self._generate_lazy_clause(state, passive)
 
-        execution_options = {
-            "_sa_orm_load_options": load_options,
-        }
+        if execution_options:
+
+            execution_options = util.EMPTY_DICT.merge_with(
+                execution_options,
+                {
+                    "_sa_orm_load_options": load_options,
+                },
+            )
+        else:
+            execution_options = {
+                "_sa_orm_load_options": load_options,
+            }
 
         if (
             self.key in state.dict
@@ -1191,15 +1229,54 @@ class PostLoader(AbstractRelationshipLoader):
 
     __slots__ = ()
 
-    def _check_recursive_postload(self, context, path, join_depth=None):
+    def _setup_for_recursion(self, context, path, loadopt, join_depth=None):
+
         effective_path = (
             context.compile_state.current_path or orm_util.PathRegistry.root
         ) + path
 
+        top_level_context = context._get_top_level_context()
+        execution_options = util.immutabledict(
+            {"sa_top_level_orm_context": top_level_context}
+        )
+
+        if loadopt:
+            recursion_depth = loadopt.local_opts.get("recursion_depth", None)
+            unlimited_recursion = recursion_depth == -1
+        else:
+            recursion_depth = None
+            unlimited_recursion = False
+
+        if recursion_depth is not None:
+            if not self.parent_property._is_self_referential:
+                raise sa_exc.InvalidRequestError(
+                    f"recursion_depth option on relationship "
+                    f"{self.parent_property} not valid for "
+                    "non-self-referential relationship"
+                )
+            recursion_depth = context.execution_options.get(
+                f"_recursion_depth_{id(self)}", recursion_depth
+            )
+
+            if not unlimited_recursion and recursion_depth < 0:
+                return (
+                    effective_path,
+                    False,
+                    execution_options,
+                    recursion_depth,
+                )
+
+            if not unlimited_recursion:
+                execution_options = execution_options.union(
+                    {
+                        f"_recursion_depth_{id(self)}": recursion_depth - 1,
+                    }
+                )
+
         if loading.PostLoad.path_exists(
             context, effective_path, self.parent_property
         ):
-            return True
+            return effective_path, False, execution_options, recursion_depth
 
         path_w_prop = path[self.parent_property]
         effective_path_w_prop = effective_path[self.parent_property]
@@ -1207,11 +1284,21 @@ class PostLoader(AbstractRelationshipLoader):
         if not path_w_prop.contains(context.attributes, "loader"):
             if join_depth:
                 if effective_path_w_prop.length / 2 > join_depth:
-                    return True
+                    return (
+                        effective_path,
+                        False,
+                        execution_options,
+                        recursion_depth,
+                    )
             elif effective_path_w_prop.contains_mapper(self.mapper):
-                return True
+                return (
+                    effective_path,
+                    False,
+                    execution_options,
+                    recursion_depth,
+                )
 
-        return False
+        return effective_path, True, execution_options, recursion_depth
 
     def _immediateload_create_row_processor(
         self,
@@ -1258,10 +1345,14 @@ class ImmediateLoader(PostLoader):
         adapter,
         populators,
     ):
-        def load_immediate(state, dict_, row):
-            state.get_impl(self.key).get(state, dict_, flags)
 
-        if self._check_recursive_postload(context, path):
+        (
+            effective_path,
+            run_loader,
+            execution_options,
+            recursion_depth,
+        ) = self._setup_for_recursion(context, path, loadopt)
+        if not run_loader:
             # this will not emit SQL and will only emit for a many-to-one
             # "use get" load.   the "_RELATED" part means it may return
             # instance even if its expired, since this is a mutually-recursive
@@ -1270,7 +1361,57 @@ class ImmediateLoader(PostLoader):
         else:
             flags = attributes.PASSIVE_OFF | PassiveFlag.NO_RAISE
 
-        populators["delayed"].append((self.key, load_immediate))
+        loading.PostLoad.callable_for_path(
+            context,
+            effective_path,
+            self.parent,
+            self.parent_property,
+            self._load_for_path,
+            loadopt,
+            flags,
+            recursion_depth,
+            execution_options,
+        )
+
+    def _load_for_path(
+        self,
+        context,
+        path,
+        states,
+        load_only,
+        loadopt,
+        flags,
+        recursion_depth,
+        execution_options,
+    ):
+
+        if recursion_depth:
+            new_opt = Load(loadopt.path.entity)
+            new_opt.context = (
+                loadopt,
+                loadopt._recurse(),
+            )
+            alternate_effective_path = path._truncate_recursive()
+            extra_options = (new_opt,)
+        else:
+            new_opt = None
+            alternate_effective_path = path
+            extra_options = ()
+
+        key = self.key
+        lazyloader = self.parent_property._get_strategy((("lazy", "select"),))
+        for state, overwrite in states:
+            dict_ = state.dict
+
+            if overwrite or key not in dict_:
+                value = lazyloader._load_for_state(
+                    state,
+                    flags,
+                    extra_options=extra_options,
+                    alternate_effective_path=alternate_effective_path,
+                    execution_options=execution_options,
+                )
+                state.get_impl(key).set_committed_value(state, dict_, value)
 
 
 @log.class_logger
@@ -1677,24 +1818,6 @@ class SubqueryLoader(PostLoader):
         subq_path = subq_path + path
         rewritten_path = rewritten_path + path
 
-        # if not via query option, check for
-        # a cycle
-        # TODO: why is this here???  this is now handled
-        # by the _check_recursive_postload call
-        if not path.contains(compile_state.attributes, "loader"):
-            if self.join_depth:
-                if (
-                    (
-                        compile_state.current_path.length
-                        if compile_state.current_path
-                        else 0
-                    )
-                    + path.length
-                ) / 2 > self.join_depth:
-                    return
-            elif subq_path.contains_mapper(self.mapper):
-                return
-
         # use the current query being invoked, not the compile state
         # one.  this is so that we get the current parameters.  however,
         # it means we can't use the existing compile state, we have to make
@@ -1814,11 +1937,14 @@ class SubqueryLoader(PostLoader):
                 adapter,
                 populators,
             )
-        # the subqueryloader does a similar check in setup_query() unlike
-        # the other post loaders, however we have this here for consistency
-        elif self._check_recursive_postload(context, path, self.join_depth):
+
+        _, run_loader, _, _ = self._setup_for_recursion(
+            context, path, loadopt, self.join_depth
+        )
+        if not run_loader:
             return
-        elif not isinstance(context.compile_state, ORMSelectCompileState):
+
+        if not isinstance(context.compile_state, ORMSelectCompileState):
             # issue 7505 - subqueryload() in 1.3 and previous would silently
             # degrade for from_statement() without warning. this behavior
             # is restored here
@@ -2787,7 +2913,16 @@ class SelectInLoader(PostLoader, util.MemoizedSlots):
                 adapter,
                 populators,
             )
-        elif self._check_recursive_postload(context, path, self.join_depth):
+
+        (
+            effective_path,
+            run_loader,
+            execution_options,
+            recursion_depth,
+        ) = self._setup_for_recursion(
+            context, path, loadopt, join_depth=self.join_depth
+        )
+        if not run_loader:
             return
 
         if not self.parent.class_manager[self.key].impl.supports_population:
@@ -2806,9 +2941,7 @@ class SelectInLoader(PostLoader, util.MemoizedSlots):
         elif not orm_util._entity_isa(path[-1], self.parent):
             return
 
-        selectin_path = (
-            context.compile_state.current_path or orm_util.PathRegistry.root
-        ) + path
+        selectin_path = effective_path
 
         path_w_prop = path[self.parent_property]
 
@@ -2830,10 +2963,20 @@ class SelectInLoader(PostLoader, util.MemoizedSlots):
             self._load_for_path,
             effective_entity,
             loadopt,
+            recursion_depth,
+            execution_options,
         )
 
     def _load_for_path(
-        self, context, path, states, load_only, effective_entity, loadopt
+        self,
+        context,
+        path,
+        states,
+        load_only,
+        effective_entity,
+        loadopt,
+        recursion_depth,
+        execution_options,
     ):
         if load_only and self.key not in load_only:
             return
@@ -3003,9 +3146,13 @@ class SelectInLoader(PostLoader, util.MemoizedSlots):
                 ),
             )
 
+        if recursion_depth is not None:
+            effective_path = effective_path._truncate_recursive()
+
         q = q.options(*new_options)._update_compile_options(
             {"_current_path": effective_path}
         )
+
         if user_defined_options:
             q = q.options(*user_defined_options)
 
@@ -3034,12 +3181,27 @@ class SelectInLoader(PostLoader, util.MemoizedSlots):
 
         if query_info.load_only_child:
             self._load_via_child(
-                our_states, none_states, query_info, q, context
+                our_states,
+                none_states,
+                query_info,
+                q,
+                context,
+                execution_options,
             )
         else:
-            self._load_via_parent(our_states, query_info, q, context)
+            self._load_via_parent(
+                our_states, query_info, q, context, execution_options
+            )
 
-    def _load_via_child(self, our_states, none_states, query_info, q, context):
+    def _load_via_child(
+        self,
+        our_states,
+        none_states,
+        query_info,
+        q,
+        context,
+        execution_options,
+    ):
         uselist = self.uselist
 
         # this sort is really for the benefit of the unit tests
@@ -3057,6 +3219,7 @@ class SelectInLoader(PostLoader, util.MemoizedSlots):
                             for key in chunk
                         ]
                     },
+                    execution_options=execution_options,
                 ).unique()
             }
 
@@ -3085,7 +3248,9 @@ class SelectInLoader(PostLoader, util.MemoizedSlots):
             # collection will be populated
             state.get_impl(self.key).set_committed_value(state, dict_, None)
 
-    def _load_via_parent(self, our_states, query_info, q, context):
+    def _load_via_parent(
+        self, our_states, query_info, q, context, execution_options
+    ):
         uselist = self.uselist
         _empty_result = () if uselist else None
 
@@ -3101,7 +3266,9 @@ class SelectInLoader(PostLoader, util.MemoizedSlots):
             data = collections.defaultdict(list)
             for k, v in itertools.groupby(
                 context.session.execute(
-                    q, params={"primary_keys": primary_keys}
+                    q,
+                    params={"primary_keys": primary_keys},
+                    execution_options=execution_options,
                 ).unique(),
                 lambda x: x[0],
             ):
diff --git a/lib/sqlalchemy/orm/strategy_options.py b/lib/sqlalchemy/orm/strategy_options.py
index 593e2abd2..aa51eca16 100644
--- a/lib/sqlalchemy/orm/strategy_options.py
+++ b/lib/sqlalchemy/orm/strategy_options.py
@@ -343,7 +343,9 @@ class _AbstractLoad(traversals.GenerativeOnTraversal, LoaderOption):
         return self._set_relationship_strategy(attr, {"lazy": "subquery"})
 
     def selectinload(
-        self: Self_AbstractLoad, attr: _AttrType
+        self: Self_AbstractLoad,
+        attr: _AttrType,
+        recursion_depth: Optional[int] = None,
     ) -> Self_AbstractLoad:
         """Indicate that the given attribute should be loaded using
         SELECT IN eager loading.
@@ -365,7 +367,22 @@ class _AbstractLoad(traversals.GenerativeOnTraversal, LoaderOption):
             query(Order).options(
                 lazyload(Order.items).selectinload(Item.keywords))
 
-        .. versionadded:: 1.2
+        :param recursion_depth: optional int; when set to a positive integer
+         in conjunction with a self-referential relationship,
+         indicates "selectin" loading will continue that many levels deep
+         automatically until no items are found.
+
+         .. note:: The :paramref:`_orm.selectinload.recursion_depth` option
+            currently supports only self-referential relationships.  There
+            is not yet an option to automatically traverse recursive structures
+            with more than one relationship involved.
+
+         .. warning:: This parameter is new and experimental and should be
+            treated as "alpha" status
+
+         .. versionadded:: 2.0 added
+            :paramref:`_orm.selectinload.recursion_depth`
+
 
         .. seealso::
 
@@ -374,7 +391,11 @@ class _AbstractLoad(traversals.GenerativeOnTraversal, LoaderOption):
             :ref:`selectin_eager_loading`
 
         """
-        return self._set_relationship_strategy(attr, {"lazy": "selectin"})
+        return self._set_relationship_strategy(
+            attr,
+            {"lazy": "selectin"},
+            opts={"recursion_depth": recursion_depth},
+        )
 
     def lazyload(
         self: Self_AbstractLoad, attr: _AttrType
@@ -395,7 +416,9 @@ class _AbstractLoad(traversals.GenerativeOnTraversal, LoaderOption):
         return self._set_relationship_strategy(attr, {"lazy": "select"})
 
     def immediateload(
-        self: Self_AbstractLoad, attr: _AttrType
+        self: Self_AbstractLoad,
+        attr: _AttrType,
+        recursion_depth: Optional[int] = None,
     ) -> Self_AbstractLoad:
         """Indicate that the given attribute should be loaded using
         an immediate load with a per-attribute SELECT statement.
@@ -410,6 +433,23 @@ class _AbstractLoad(traversals.GenerativeOnTraversal, LoaderOption):
         This function is part of the :class:`_orm.Load` interface and supports
         both method-chained and standalone operation.
 
+        :param recursion_depth: optional int; when set to a positive integer
+         in conjunction with a self-referential relationship,
+         indicates "selectin" loading will continue that many levels deep
+         automatically until no items are found.
+
+         .. note:: The :paramref:`_orm.immediateload.recursion_depth` option
+            currently supports only self-referential relationships.  There
+            is not yet an option to automatically traverse recursive structures
+            with more than one relationship involved.
+
+         .. warning:: This parameter is new and experimental and should be
+            treated as "alpha" status
+
+         .. versionadded:: 2.0 added
+            :paramref:`_orm.immediateload.recursion_depth`
+
+
         .. seealso::
 
             :ref:`loading_toplevel`
@@ -417,7 +457,11 @@ class _AbstractLoad(traversals.GenerativeOnTraversal, LoaderOption):
             :ref:`selectin_eager_loading`
 
         """
-        loader = self._set_relationship_strategy(attr, {"lazy": "immediate"})
+        loader = self._set_relationship_strategy(
+            attr,
+            {"lazy": "immediate"},
+            opts={"recursion_depth": recursion_depth},
+        )
         return loader
 
     def noload(self: Self_AbstractLoad, attr: _AttrType) -> Self_AbstractLoad:
@@ -1256,6 +1300,15 @@ class Load(_AbstractLoad):
                     if wildcard_key is _RELATIONSHIP_TOKEN:
                         self.path = load_element.path
                     self.context += (load_element,)
+
+                    # this seems to be effective for selectinloader,
+                    # giving the extra match to one more level deep.
+                    # but does not work for immediateloader, which still
+                    # must add additional options at load time
+                    if load_element.local_opts.get("recursion_depth", False):
+                        r1 = load_element._recurse()
+                        self.context += (r1,)
+
         return self
 
     def __getstate__(self):
@@ -1524,6 +1577,11 @@ class _LoadElement(
         self._shallow_copy_to(s)
         return s
 
+    def _update_opts(self, **kw: Any) -> _LoadElement:
+        new = self._clone()
+        new.local_opts = new.local_opts.union(kw)
+        return new
+
     def __getstate__(self) -> Dict[str, Any]:
         d = self._shallow_to_dict()
         d["path"] = self.path.serialize()
@@ -1690,7 +1748,15 @@ class _LoadElement(
     def __init__(self) -> None:
         raise NotImplementedError()
 
-    def _prepend_path_from(self, parent):
+    def _recurse(self) -> _LoadElement:
+        cloned = self._clone()
+        cloned.path = PathRegistry.coerce(self.path[:] + self.path[-2:])
+
+        return cloned
+
+    def _prepend_path_from(
+        self, parent: Union[Load, _LoadElement]
+    ) -> _LoadElement:
         """adjust the path of this :class:`._LoadElement` to be
         a subpath of that of the given parent :class:`_orm.Load` object's
         path.
@@ -2337,8 +2403,12 @@ def subqueryload(*keys: _AttrType) -> _AbstractLoad:
 
 
 @loader_unbound_fn
-def selectinload(*keys: _AttrType) -> _AbstractLoad:
-    return _generate_from_keys(Load.selectinload, keys, False, {})
+def selectinload(
+    *keys: _AttrType, recursion_depth: Optional[int] = None
+) -> _AbstractLoad:
+    return _generate_from_keys(
+        Load.selectinload, keys, False, {"recursion_depth": recursion_depth}
+    )
 
 
 @loader_unbound_fn
@@ -2347,8 +2417,12 @@ def lazyload(*keys: _AttrType) -> _AbstractLoad:
 
 
 @loader_unbound_fn
-def immediateload(*keys: _AttrType) -> _AbstractLoad:
-    return _generate_from_keys(Load.immediateload, keys, False, {})
+def immediateload(
+    *keys: _AttrType, recursion_depth: Optional[int] = None
+) -> _AbstractLoad:
+    return _generate_from_keys(
+        Load.immediateload, keys, False, {"recursion_depth": recursion_depth}
+    )
 
 
 @loader_unbound_fn