create new approach for deeply nested post loader options

Added very experimental feature to the :func:`_orm.selectinload` and
:func:`_orm.immediateload` loader options called
:paramref:`_orm.selectinload.recursion_depth` /
:paramref:`_orm.immediateload.recursion_depth` , which allows a single
loader option to automatically recurse into self-referential relationships.
Is set to an integer indicating depth, and may also be set to -1 to
indicate to continue loading until no more levels deep are found.
Major internal changes to :func:`_orm.selectinload` and
:func:`_orm.immediateload` allow this feature to work while continuing
to make correct use of the compilation cache, as well as not using
arbitrary recursion, so any level of depth is supported (though would
emit that many queries).  This may be useful for
self-referential structures that must be loaded fully eagerly, such as when
using asyncio.

A warning is also emitted when loader options are connected together with
arbitrary lengths (that is, without using the new ``recursion_depth``
option) when excessive recursion depth is detected in related object
loading. This operation continues to use huge amounts of memory and
performs extremely poorly; the cache is disabled when this condition is
detected to protect the cache from being flooded with arbitrary statements.

Fixes: #8126
Change-Id: I9f162e0a09c1ed327dd19498aac193f649333a01

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