# orm/context.py
# Copyright (C) 2005-2024 the SQLAlchemy authors and contributors
# <see AUTHORS file>
#
# This module is part of SQLAlchemy and is released under
# the MIT License: https://www.opensource.org/licenses/mit-license.php
# mypy: ignore-errors

from __future__ import annotations

import itertools
from typing import Any
from typing import cast
from typing import Dict
from typing import Iterable
from typing import List
from typing import Optional
from typing import Set
from typing import Tuple
from typing import Type
from typing import TYPE_CHECKING
from typing import TypeVar
from typing import Union

from . import attributes
from . import interfaces
from . import loading
from .base import _is_aliased_class
from .interfaces import ORMColumnDescription
from .interfaces import ORMColumnsClauseRole
from .path_registry import PathRegistry
from .util import _entity_corresponds_to
from .util import _ORMJoin
from .util import _TraceAdaptRole
from .util import AliasedClass
from .util import Bundle
from .util import ORMAdapter
from .util import ORMStatementAdapter
from .. import exc as sa_exc
from .. import future
from .. import inspect
from .. import sql
from .. import util
from ..sql import coercions
from ..sql import expression
from ..sql import roles
from ..sql import util as sql_util
from ..sql import visitors
from ..sql._typing import _TP
from ..sql._typing import is_dml
from ..sql._typing import is_insert_update
from ..sql._typing import is_select_base
from ..sql.base import _select_iterables
from ..sql.base import CacheableOptions
from ..sql.base import CompileState
from ..sql.base import Executable
from ..sql.base import Generative
from ..sql.base import Options
from ..sql.dml import UpdateBase
from ..sql.elements import GroupedElement
from ..sql.elements import TextClause
from ..sql.selectable import CompoundSelectState
from ..sql.selectable import LABEL_STYLE_DISAMBIGUATE_ONLY
from ..sql.selectable import LABEL_STYLE_NONE
from ..sql.selectable import LABEL_STYLE_TABLENAME_PLUS_COL
from ..sql.selectable import Select
from ..sql.selectable import SelectLabelStyle
from ..sql.selectable import SelectState
from ..sql.selectable import TypedReturnsRows
from ..sql.visitors import InternalTraversal

if TYPE_CHECKING:
    from ._typing import _InternalEntityType
    from ._typing import OrmExecuteOptionsParameter
    from .loading import PostLoad
    from .mapper import Mapper
    from .query import Query
    from .session import _BindArguments
    from .session import Session
    from ..engine import Result
    from ..engine.interfaces import _CoreSingleExecuteParams
    from ..sql._typing import _ColumnsClauseArgument
    from ..sql.compiler import SQLCompiler
    from ..sql.dml import _DMLTableElement
    from ..sql.elements import ColumnElement
    from ..sql.selectable import _JoinTargetElement
    from ..sql.selectable import _LabelConventionCallable
    from ..sql.selectable import _SetupJoinsElement
    from ..sql.selectable import ExecutableReturnsRows
    from ..sql.selectable import SelectBase
    from ..sql.type_api import TypeEngine

_T = TypeVar("_T", bound=Any)
_path_registry = PathRegistry.root

_EMPTY_DICT = util.immutabledict()


LABEL_STYLE_LEGACY_ORM = SelectLabelStyle.LABEL_STYLE_LEGACY_ORM


class QueryContext:
    __slots__ = (
        "top_level_context",
        "compile_state",
        "query",
        "params",
        "load_options",
        "bind_arguments",
        "execution_options",
        "session",
        "autoflush",
        "populate_existing",
        "invoke_all_eagers",
        "version_check",
        "refresh_state",
        "create_eager_joins",
        "propagated_loader_options",
        "attributes",
        "runid",
        "partials",
        "post_load_paths",
        "identity_token",
        "yield_per",
        "loaders_require_buffering",
        "loaders_require_uniquing",
    )

    runid: int
    post_load_paths: Dict[PathRegistry, PostLoad]
    compile_state: ORMCompileState

    class default_load_options(Options):
        _only_return_tuples = False
        _populate_existing = False
        _version_check = False
        _invoke_all_eagers = True
        _autoflush = True
        _identity_token = None
        _yield_per = None
        _refresh_state = None
        _lazy_loaded_from = None
        _legacy_uniquing = False
        _sa_top_level_orm_context = None
        _is_user_refresh = False

    def __init__(
        self,
        compile_state: CompileState,
        statement: Union[Select[Any], FromStatement[Any]],
        params: _CoreSingleExecuteParams,
        session: Session,
        load_options: Union[
            Type[QueryContext.default_load_options],
            QueryContext.default_load_options,
        ],
        execution_options: Optional[OrmExecuteOptionsParameter] = None,
        bind_arguments: Optional[_BindArguments] = None,
    ):
        self.load_options = load_options
        self.execution_options = execution_options or _EMPTY_DICT
        self.bind_arguments = bind_arguments or _EMPTY_DICT
        self.compile_state = compile_state
        self.query = statement
        self.session = session
        self.loaders_require_buffering = False
        self.loaders_require_uniquing = False
        self.params = params
        self.top_level_context = load_options._sa_top_level_orm_context

        cached_options = compile_state.select_statement._with_options
        uncached_options = statement._with_options

        # see issue #7447 , #8399 for some background
        # propagated loader options will be present on loaded InstanceState
        # objects under state.load_options and are typically used by
        # LazyLoader to apply options to the SELECT statement it emits.
        # For compile state options (i.e. loader strategy options), these
        # need to line up with the ".load_path" attribute which in
        # loader.py is pulled from context.compile_state.current_path.
        # so, this means these options have to be the ones from the
        # *cached* statement that's travelling with compile_state, not the
        # *current* statement which won't match up for an ad-hoc
        # AliasedClass
        self.propagated_loader_options = tuple(
            opt._adapt_cached_option_to_uncached_option(self, uncached_opt)
            for opt, uncached_opt in zip(cached_options, uncached_options)
            if opt.propagate_to_loaders
        )

        self.attributes = dict(compile_state.attributes)

        self.autoflush = load_options._autoflush
        self.populate_existing = load_options._populate_existing
        self.invoke_all_eagers = load_options._invoke_all_eagers
        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._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}
)


class AbstractORMCompileState(CompileState):
    is_dml_returning = False

    def _init_global_attributes(
        self, statement, compiler, *, toplevel, process_criteria_for_toplevel
    ):
        self.attributes = {}

        if compiler is None:
            # this is the legacy / testing only ORM _compile_state() use case.
            # there is no need to apply criteria options for this.
            self.global_attributes = ga = {}
            assert toplevel
            return
        else:
            self.global_attributes = ga = compiler._global_attributes

        if toplevel:
            ga["toplevel_orm"] = True

            if process_criteria_for_toplevel:
                for opt in statement._with_options:
                    if opt._is_criteria_option:
                        opt.process_compile_state(self)

            return
        elif ga.get("toplevel_orm", False):
            return

        stack_0 = compiler.stack[0]

        try:
            toplevel_stmt = stack_0["selectable"]
        except KeyError:
            pass
        else:
            for opt in toplevel_stmt._with_options:
                if opt._is_compile_state and opt._is_criteria_option:
                    opt.process_compile_state(self)

        ga["toplevel_orm"] = True

    @classmethod
    def create_for_statement(
        cls,
        statement: Union[Select, FromStatement],
        compiler: Optional[SQLCompiler],
        **kw: Any,
    ) -> AbstractORMCompileState:
        """Create a context for a statement given a :class:`.Compiler`.

        This method is always invoked in the context of SQLCompiler.process().

        For a Select object, this would be invoked from
        SQLCompiler.visit_select(). For the special FromStatement object used
        by Query to indicate "Query.from_statement()", this is called by
        FromStatement._compiler_dispatch() that would be called by
        SQLCompiler.process().
        """
        return super().create_for_statement(statement, compiler, **kw)

    @classmethod
    def orm_pre_session_exec(
        cls,
        session,
        statement,
        params,
        execution_options,
        bind_arguments,
        is_pre_event,
    ):
        raise NotImplementedError()

    @classmethod
    def orm_execute_statement(
        cls,
        session,
        statement,
        params,
        execution_options,
        bind_arguments,
        conn,
    ) -> Result:
        result = conn.execute(
            statement, params or {}, execution_options=execution_options
        )
        return cls.orm_setup_cursor_result(
            session,
            statement,
            params,
            execution_options,
            bind_arguments,
            result,
        )

    @classmethod
    def orm_setup_cursor_result(
        cls,
        session,
        statement,
        params,
        execution_options,
        bind_arguments,
        result,
    ):
        raise NotImplementedError()


class AutoflushOnlyORMCompileState(AbstractORMCompileState):
    """ORM compile state that is a passthrough, except for autoflush."""

    @classmethod
    def orm_pre_session_exec(
        cls,
        session,
        statement,
        params,
        execution_options,
        bind_arguments,
        is_pre_event,
    ):
        # consume result-level load_options.  These may have been set up
        # in an ORMExecuteState hook
        (
            load_options,
            execution_options,
        ) = QueryContext.default_load_options.from_execution_options(
            "_sa_orm_load_options",
            {
                "autoflush",
            },
            execution_options,
            statement._execution_options,
        )

        if not is_pre_event and load_options._autoflush:
            session._autoflush()

        return statement, execution_options

    @classmethod
    def orm_setup_cursor_result(
        cls,
        session,
        statement,
        params,
        execution_options,
        bind_arguments,
        result,
    ):
        return result


class ORMCompileState(AbstractORMCompileState):
    class default_compile_options(CacheableOptions):
        _cache_key_traversal = [
            ("_use_legacy_query_style", InternalTraversal.dp_boolean),
            ("_for_statement", InternalTraversal.dp_boolean),
            ("_bake_ok", InternalTraversal.dp_boolean),
            ("_current_path", InternalTraversal.dp_has_cache_key),
            ("_enable_single_crit", InternalTraversal.dp_boolean),
            ("_enable_eagerloads", InternalTraversal.dp_boolean),
            ("_only_load_props", InternalTraversal.dp_plain_obj),
            ("_set_base_alias", InternalTraversal.dp_boolean),
            ("_for_refresh_state", InternalTraversal.dp_boolean),
            ("_render_for_subquery", InternalTraversal.dp_boolean),
            ("_is_star", InternalTraversal.dp_boolean),
        ]

        # set to True by default from Query._statement_20(), to indicate
        # the rendered query should look like a legacy ORM query.  right
        # now this basically indicates we should use tablename_columnname
        # style labels.    Generally indicates the statement originated
        # from a Query object.
        _use_legacy_query_style = False

        # set *only* when we are coming from the Query.statement
        # accessor, or a Query-level equivalent such as
        # query.subquery().  this supersedes "toplevel".
        _for_statement = False

        _bake_ok = True
        _current_path = _path_registry
        _enable_single_crit = True
        _enable_eagerloads = True
        _only_load_props = None
        _set_base_alias = False
        _for_refresh_state = False
        _render_for_subquery = False
        _is_star = False

    attributes: Dict[Any, Any]
    global_attributes: Dict[Any, Any]

    statement: Union[Select[Any], FromStatement[Any]]
    select_statement: Union[Select[Any], FromStatement[Any]]
    _entities: List[_QueryEntity]
    _polymorphic_adapters: Dict[_InternalEntityType, ORMAdapter]
    compile_options: Union[
        Type[default_compile_options], default_compile_options
    ]
    _primary_entity: Optional[_QueryEntity]
    use_legacy_query_style: bool
    _label_convention: _LabelConventionCallable
    primary_columns: List[ColumnElement[Any]]
    secondary_columns: List[ColumnElement[Any]]
    dedupe_columns: Set[ColumnElement[Any]]
    create_eager_joins: List[
        # TODO: this structure is set up by JoinedLoader
        Tuple[Any, ...]
    ]
    current_path: PathRegistry = _path_registry
    _has_mapper_entities = False

    def __init__(self, *arg, **kw):
        raise NotImplementedError()

    if TYPE_CHECKING:

        @classmethod
        def create_for_statement(
            cls,
            statement: Union[Select, FromStatement],
            compiler: Optional[SQLCompiler],
            **kw: Any,
        ) -> ORMCompileState: ...

    def _append_dedupe_col_collection(self, obj, col_collection):
        dedupe = self.dedupe_columns
        if obj not in dedupe:
            dedupe.add(obj)
            col_collection.append(obj)

    @classmethod
    def _column_naming_convention(
        cls, label_style: SelectLabelStyle, legacy: bool
    ) -> _LabelConventionCallable:
        if legacy:

            def name(col, col_name=None):
                if col_name:
                    return col_name
                else:
                    return getattr(col, "key")

            return name
        else:
            return SelectState._column_naming_convention(label_style)

    @classmethod
    def get_column_descriptions(cls, statement):
        return _column_descriptions(statement)

    @classmethod
    def orm_pre_session_exec(
        cls,
        session,
        statement,
        params,
        execution_options,
        bind_arguments,
        is_pre_event,
    ):
        # consume result-level load_options.  These may have been set up
        # in an ORMExecuteState hook
        (
            load_options,
            execution_options,
        ) = QueryContext.default_load_options.from_execution_options(
            "_sa_orm_load_options",
            {
                "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.

        if "sa_top_level_orm_context" in execution_options:
            ctx = execution_options["sa_top_level_orm_context"]
            execution_options = ctx.query._execution_options.merge_with(
                ctx.execution_options, execution_options
            )

        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}
            )

        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
        ):
            execution_options: util.immutabledict[str, Any] = (
                execution_options.union(
                    {
                        "compiled_cache": None,
                        "_cache_disable_reason": "excess depth for "
                        "ORM loader options",
                    }
                )
            )

        bind_arguments["clause"] = statement

        # new in 1.4 - the coercions system is leveraged to allow the
        # "subject" mapper of a statement be propagated to the top
        # as the statement is built.   "subject" mapper is the generally
        # standard object used as an identifier for multi-database schemes.

        # we are here based on the fact that _propagate_attrs contains
        # "compile_state_plugin": "orm".   The "plugin_subject"
        # needs to be present as well.

        try:
            plugin_subject = statement._propagate_attrs["plugin_subject"]
        except KeyError:
            assert False, "statement had 'orm' plugin but no plugin_subject"
        else:
            if plugin_subject:
                bind_arguments["mapper"] = plugin_subject.mapper

        if not is_pre_event and load_options._autoflush:
            session._autoflush()

        return statement, execution_options

    @classmethod
    def orm_setup_cursor_result(
        cls,
        session,
        statement,
        params,
        execution_options,
        bind_arguments,
        result,
    ):
        execution_context = result.context
        compile_state = execution_context.compiled.compile_state

        # cover edge case where ORM entities used in legacy select
        # were passed to session.execute:
        # session.execute(legacy_select([User.id, User.name]))
        # see test_query->test_legacy_tuple_old_select

        load_options = execution_options.get(
            "_sa_orm_load_options", QueryContext.default_load_options
        )

        if compile_state.compile_options._is_star:
            return result

        querycontext = QueryContext(
            compile_state,
            statement,
            params,
            session,
            load_options,
            execution_options,
            bind_arguments,
        )
        return loading.instances(result, querycontext)

    @property
    def _lead_mapper_entities(self):
        """return all _MapperEntity objects in the lead entities collection.

        Does **not** include entities that have been replaced by
        with_entities(), with_only_columns()

        """
        return [
            ent for ent in self._entities if isinstance(ent, _MapperEntity)
        ]

    def _create_with_polymorphic_adapter(self, ext_info, selectable):
        """given MapperEntity or ORMColumnEntity, setup polymorphic loading
        if called for by the Mapper.

        As of #8168 in 2.0.0rc1, polymorphic adapters, which greatly increase
        the complexity of the query creation process, are not used at all
        except in the quasi-legacy cases of with_polymorphic referring to an
        alias and/or subquery. This would apply to concrete polymorphic
        loading, and joined inheritance where a subquery is
        passed to with_polymorphic (which is completely unnecessary in modern
        use).

        """
        if (
            not ext_info.is_aliased_class
            and ext_info.mapper.persist_selectable
            not in self._polymorphic_adapters
        ):
            for mp in ext_info.mapper.iterate_to_root():
                self._mapper_loads_polymorphically_with(
                    mp,
                    ORMAdapter(
                        _TraceAdaptRole.WITH_POLYMORPHIC_ADAPTER,
                        mp,
                        equivalents=mp._equivalent_columns,
                        selectable=selectable,
                    ),
                )

    def _mapper_loads_polymorphically_with(self, mapper, adapter):
        for m2 in mapper._with_polymorphic_mappers or [mapper]:
            self._polymorphic_adapters[m2] = adapter

            for m in m2.iterate_to_root():
                self._polymorphic_adapters[m.local_table] = adapter

    @classmethod
    def _create_entities_collection(cls, query, legacy):
        raise NotImplementedError(
            "this method only works for ORMSelectCompileState"
        )


class DMLReturningColFilter:
    """an adapter used for the DML RETURNING case.

    Has a subset of the interface used by
    :class:`.ORMAdapter` and is used for :class:`._QueryEntity`
    instances to set up their columns as used in RETURNING for a
    DML statement.

    """

    __slots__ = ("mapper", "columns", "__weakref__")

    def __init__(self, target_mapper, immediate_dml_mapper):
        if (
            immediate_dml_mapper is not None
            and target_mapper.local_table
            is not immediate_dml_mapper.local_table
        ):
            # joined inh, or in theory other kinds of multi-table mappings
            self.mapper = immediate_dml_mapper
        else:
            # single inh, normal mappings, etc.
            self.mapper = target_mapper
        self.columns = self.columns = util.WeakPopulateDict(
            self.adapt_check_present  # type: ignore
        )

    def __call__(self, col, as_filter):
        for cc in sql_util._find_columns(col):
            c2 = self.adapt_check_present(cc)
            if c2 is not None:
                return col
        else:
            return None

    def adapt_check_present(self, col):
        mapper = self.mapper
        prop = mapper._columntoproperty.get(col, None)
        if prop is None:
            return None
        return mapper.local_table.c.corresponding_column(col)


@sql.base.CompileState.plugin_for("orm", "orm_from_statement")
class ORMFromStatementCompileState(ORMCompileState):
    _from_obj_alias = None
    _has_mapper_entities = False

    statement_container: FromStatement
    requested_statement: Union[SelectBase, TextClause, UpdateBase]
    dml_table: Optional[_DMLTableElement] = None

    _has_orm_entities = False
    multi_row_eager_loaders = False
    eager_adding_joins = False
    compound_eager_adapter = None

    extra_criteria_entities = _EMPTY_DICT
    eager_joins = _EMPTY_DICT

    @classmethod
    def create_for_statement(
        cls,
        statement_container: Union[Select, FromStatement],
        compiler: Optional[SQLCompiler],
        **kw: Any,
    ) -> ORMFromStatementCompileState:
        assert isinstance(statement_container, FromStatement)

        if compiler is not None and compiler.stack:
            raise sa_exc.CompileError(
                "The ORM FromStatement construct only supports being "
                "invoked as the topmost statement, as it is only intended to "
                "define how result rows should be returned."
            )

        self = cls.__new__(cls)
        self._primary_entity = None

        self.use_legacy_query_style = (
            statement_container._compile_options._use_legacy_query_style
        )
        self.statement_container = self.select_statement = statement_container
        self.requested_statement = statement = statement_container.element

        if statement.is_dml:
            self.dml_table = statement.table
            self.is_dml_returning = True

        self._entities = []
        self._polymorphic_adapters = {}

        self.compile_options = statement_container._compile_options

        if (
            self.use_legacy_query_style
            and isinstance(statement, expression.SelectBase)
            and not statement._is_textual
            and not statement.is_dml
            and statement._label_style is LABEL_STYLE_NONE
        ):
            self.statement = statement.set_label_style(
                LABEL_STYLE_TABLENAME_PLUS_COL
            )
        else:
            self.statement = statement

        self._label_convention = self._column_naming_convention(
            (
                statement._label_style
                if not statement._is_textual and not statement.is_dml
                else LABEL_STYLE_NONE
            ),
            self.use_legacy_query_style,
        )

        _QueryEntity.to_compile_state(
            self,
            statement_container._raw_columns,
            self._entities,
            is_current_entities=True,
        )

        self.current_path = statement_container._compile_options._current_path

        self._init_global_attributes(
            statement_container,
            compiler,
            process_criteria_for_toplevel=False,
            toplevel=True,
        )

        if statement_container._with_options:
            for opt in statement_container._with_options:
                if opt._is_compile_state:
                    opt.process_compile_state(self)

        if statement_container._with_context_options:
            for fn, key in statement_container._with_context_options:
                fn(self)

        self.primary_columns = []
        self.secondary_columns = []
        self.dedupe_columns = set()
        self.create_eager_joins = []
        self._fallback_from_clauses = []

        self.order_by = None

        if isinstance(self.statement, expression.TextClause):
            # TextClause has no "column" objects at all.  for this case,
            # we generate columns from our _QueryEntity objects, then
            # flip on all the "please match no matter what" parameters.
            self.extra_criteria_entities = {}

            for entity in self._entities:
                entity.setup_compile_state(self)

            compiler._ordered_columns = compiler._textual_ordered_columns = (
                False
            )

            # enable looser result column matching.  this is shown to be
            # needed by test_query.py::TextTest
            compiler._loose_column_name_matching = True

            for c in self.primary_columns:
                compiler.process(
                    c,
                    within_columns_clause=True,
                    add_to_result_map=compiler._add_to_result_map,
                )
        else:
            # for everyone else, Select, Insert, Update, TextualSelect, they
            # have column objects already.  After much
            # experimentation here, the best approach seems to be, use
            # those columns completely, don't interfere with the compiler
            # at all; just in ORM land, use an adapter to convert from
            # our ORM columns to whatever columns are in the statement,
            # before we look in the result row. Adapt on names
            # to accept cases such as issue #9217, however also allow
            # this to be overridden for cases such as #9273.
            self._from_obj_alias = ORMStatementAdapter(
                _TraceAdaptRole.ADAPT_FROM_STATEMENT,
                self.statement,
                adapt_on_names=statement_container._adapt_on_names,
            )

        return self

    def _adapt_col_list(self, cols, current_adapter):
        return cols

    def _get_current_adapter(self):
        return None

    def setup_dml_returning_compile_state(self, dml_mapper):
        """used by BulkORMInsert (and Update / Delete?) to set up a handler
        for RETURNING to return ORM objects and expressions

        """
        target_mapper = self.statement._propagate_attrs.get(
            "plugin_subject", None
        )
        adapter = DMLReturningColFilter(target_mapper, dml_mapper)

        if self.compile_options._is_star and (len(self._entities) != 1):
            raise sa_exc.CompileError(
                "Can't generate ORM query that includes multiple expressions "
                "at the same time as '*'; query for '*' alone if present"
            )

        for entity in self._entities:
            entity.setup_dml_returning_compile_state(self, adapter)


class FromStatement(GroupedElement, Generative, TypedReturnsRows[_TP]):
    """Core construct that represents a load of ORM objects from various
    :class:`.ReturnsRows` and other classes including:

    :class:`.Select`, :class:`.TextClause`, :class:`.TextualSelect`,
    :class:`.CompoundSelect`, :class`.Insert`, :class:`.Update`,
    and in theory, :class:`.Delete`.

    """

    __visit_name__ = "orm_from_statement"

    _compile_options = ORMFromStatementCompileState.default_compile_options

    _compile_state_factory = ORMFromStatementCompileState.create_for_statement

    _for_update_arg = None

    element: Union[ExecutableReturnsRows, TextClause]

    _adapt_on_names: bool

    _traverse_internals = [
        ("_raw_columns", InternalTraversal.dp_clauseelement_list),
        ("element", InternalTraversal.dp_clauseelement),
    ] + Executable._executable_traverse_internals

    _cache_key_traversal = _traverse_internals + [
        ("_compile_options", InternalTraversal.dp_has_cache_key)
    ]

    is_from_statement = True

    def __init__(
        self,
        entities: Iterable[_ColumnsClauseArgument[Any]],
        element: Union[ExecutableReturnsRows, TextClause],
        _adapt_on_names: bool = True,
    ):
        self._raw_columns = [
            coercions.expect(
                roles.ColumnsClauseRole,
                ent,
                apply_propagate_attrs=self,
                post_inspect=True,
            )
            for ent in util.to_list(entities)
        ]
        self.element = element
        self.is_dml = element.is_dml
        self.is_select = element.is_select
        self.is_delete = element.is_delete
        self.is_insert = element.is_insert
        self.is_update = element.is_update
        self._label_style = (
            element._label_style if is_select_base(element) else None
        )
        self._adapt_on_names = _adapt_on_names

    def _compiler_dispatch(self, compiler, **kw):
        """provide a fixed _compiler_dispatch method.

        This is roughly similar to using the sqlalchemy.ext.compiler
        ``@compiles`` extension.

        """

        compile_state = self._compile_state_factory(self, compiler, **kw)

        toplevel = not compiler.stack

        if toplevel:
            compiler.compile_state = compile_state

        return compiler.process(compile_state.statement, **kw)

    @property
    def column_descriptions(self):
        """Return a :term:`plugin-enabled` 'column descriptions' structure
        referring to the columns which are SELECTed by this statement.

        See the section :ref:`queryguide_inspection` for an overview
        of this feature.

        .. seealso::

            :ref:`queryguide_inspection` - ORM background

        """
        meth = cast(
            ORMSelectCompileState, SelectState.get_plugin_class(self)
        ).get_column_descriptions
        return meth(self)

    def _ensure_disambiguated_names(self):
        return self

    def get_children(self, **kw):
        yield from itertools.chain.from_iterable(
            element._from_objects for element in self._raw_columns
        )
        yield from super().get_children(**kw)

    @property
    def _all_selected_columns(self):
        return self.element._all_selected_columns

    @property
    def _return_defaults(self):
        return self.element._return_defaults if is_dml(self.element) else None

    @property
    def _returning(self):
        return self.element._returning if is_dml(self.element) else None

    @property
    def _inline(self):
        return self.element._inline if is_insert_update(self.element) else None


@sql.base.CompileState.plugin_for("orm", "compound_select")
class CompoundSelectCompileState(
    AutoflushOnlyORMCompileState, CompoundSelectState
):
    pass


@sql.base.CompileState.plugin_for("orm", "select")
class ORMSelectCompileState(ORMCompileState, SelectState):
    _already_joined_edges = ()

    _memoized_entities = _EMPTY_DICT

    _from_obj_alias = None
    _has_mapper_entities = False

    _has_orm_entities = False
    multi_row_eager_loaders = False
    eager_adding_joins = False
    compound_eager_adapter = None

    correlate = None
    correlate_except = None
    _where_criteria = ()
    _having_criteria = ()

    @classmethod
    def create_for_statement(
        cls,
        statement: Union[Select, FromStatement],
        compiler: Optional[SQLCompiler],
        **kw: Any,
    ) -> ORMSelectCompileState:
        """compiler hook, we arrive here from compiler.visit_select() only."""

        self = cls.__new__(cls)

        if compiler is not None:
            toplevel = not compiler.stack
        else:
            toplevel = True

        select_statement = statement

        # if we are a select() that was never a legacy Query, we won't
        # have ORM level compile options.
        statement._compile_options = cls.default_compile_options.safe_merge(
            statement._compile_options
        )

        if select_statement._execution_options:
            # execution options should not impact the compilation of a
            # query, and at the moment subqueryloader is putting some things
            # in here that we explicitly don't want stuck in a cache.
            self.select_statement = select_statement._clone()
            self.select_statement._execution_options = util.immutabledict()
        else:
            self.select_statement = select_statement

        # indicates this select() came from Query.statement
        self.for_statement = select_statement._compile_options._for_statement

        # generally if we are from Query or directly from a select()
        self.use_legacy_query_style = (
            select_statement._compile_options._use_legacy_query_style
        )

        self._entities = []
        self._primary_entity = None
        self._polymorphic_adapters = {}

        self.compile_options = select_statement._compile_options

        if not toplevel:
            # for subqueries, turn off eagerloads and set
            # "render_for_subquery".
            self.compile_options += {
                "_enable_eagerloads": False,
                "_render_for_subquery": True,
            }

        # determine label style.   we can make different decisions here.
        # at the moment, trying to see if we can always use DISAMBIGUATE_ONLY
        # rather than LABEL_STYLE_NONE, and if we can use disambiguate style
        # for new style ORM selects too.
        if (
            self.use_legacy_query_style
            and self.select_statement._label_style is LABEL_STYLE_LEGACY_ORM
        ):
            if not self.for_statement:
                self.label_style = LABEL_STYLE_TABLENAME_PLUS_COL
            else:
                self.label_style = LABEL_STYLE_DISAMBIGUATE_ONLY
        else:
            self.label_style = self.select_statement._label_style

        if select_statement._memoized_select_entities:
            self._memoized_entities = {
                memoized_entities: _QueryEntity.to_compile_state(
                    self,
                    memoized_entities._raw_columns,
                    [],
                    is_current_entities=False,
                )
                for memoized_entities in (
                    select_statement._memoized_select_entities
                )
            }

        # label_convention is stateful and will yield deduping keys if it
        # sees the same key twice.  therefore it's important that it is not
        # invoked for the above "memoized" entities that aren't actually
        # in the columns clause
        self._label_convention = self._column_naming_convention(
            statement._label_style, self.use_legacy_query_style
        )

        _QueryEntity.to_compile_state(
            self,
            select_statement._raw_columns,
            self._entities,
            is_current_entities=True,
        )

        self.current_path = select_statement._compile_options._current_path

        self.eager_order_by = ()

        self._init_global_attributes(
            select_statement,
            compiler,
            toplevel=toplevel,
            process_criteria_for_toplevel=False,
        )

        if toplevel and (
            select_statement._with_options
            or select_statement._memoized_select_entities
        ):
            for (
                memoized_entities
            ) in select_statement._memoized_select_entities:
                for opt in memoized_entities._with_options:
                    if opt._is_compile_state:
                        opt.process_compile_state_replaced_entities(
                            self,
                            [
                                ent
                                for ent in self._memoized_entities[
                                    memoized_entities
                                ]
                                if isinstance(ent, _MapperEntity)
                            ],
                        )

            for opt in self.select_statement._with_options:
                if opt._is_compile_state:
                    opt.process_compile_state(self)

        # uncomment to print out the context.attributes structure
        # after it's been set up above
        # self._dump_option_struct()

        if select_statement._with_context_options:
            for fn, key in select_statement._with_context_options:
                fn(self)

        self.primary_columns = []
        self.secondary_columns = []
        self.dedupe_columns = set()
        self.eager_joins = {}
        self.extra_criteria_entities = {}
        self.create_eager_joins = []
        self._fallback_from_clauses = []

        # normalize the FROM clauses early by themselves, as this makes
        # it an easier job when we need to assemble a JOIN onto these,
        # for select.join() as well as joinedload().   As of 1.4 there are now
        # potentially more complex sets of FROM objects here as the use
        # of lambda statements for lazyload, load_on_pk etc. uses more
        # cloning of the select() construct.  See #6495
        self.from_clauses = self._normalize_froms(
            info.selectable for info in select_statement._from_obj
        )

        # this is a fairly arbitrary break into a second method,
        # so it might be nicer to break up create_for_statement()
        # and _setup_for_generate into three or four logical sections
        self._setup_for_generate()

        SelectState.__init__(self, self.statement, compiler, **kw)
        return self

    def _dump_option_struct(self):
        print("\n---------------------------------------------------\n")
        print(f"current path: {self.current_path}")
        for key in self.attributes:
            if isinstance(key, tuple) and key[0] == "loader":
                print(f"\nLoader:           {PathRegistry.coerce(key[1])}")
                print(f"    {self.attributes[key]}")
                print(f"    {self.attributes[key].__dict__}")
            elif isinstance(key, tuple) and key[0] == "path_with_polymorphic":
                print(f"\nWith Polymorphic: {PathRegistry.coerce(key[1])}")
                print(f"    {self.attributes[key]}")

    def _setup_for_generate(self):
        query = self.select_statement

        self.statement = None
        self._join_entities = ()

        if self.compile_options._set_base_alias:
            # legacy Query only
            self._set_select_from_alias()

        for memoized_entities in query._memoized_select_entities:
            if memoized_entities._setup_joins:
                self._join(
                    memoized_entities._setup_joins,
                    self._memoized_entities[memoized_entities],
                )

        if query._setup_joins:
            self._join(query._setup_joins, self._entities)

        current_adapter = self._get_current_adapter()

        if query._where_criteria:
            self._where_criteria = query._where_criteria

            if current_adapter:
                self._where_criteria = tuple(
                    current_adapter(crit, True)
                    for crit in self._where_criteria
                )

        # TODO: some complexity with order_by here was due to mapper.order_by.
        # now that this is removed we can hopefully make order_by /
        # group_by act identically to how they are in Core select.
        self.order_by = (
            self._adapt_col_list(query._order_by_clauses, current_adapter)
            if current_adapter and query._order_by_clauses not in (None, False)
            else query._order_by_clauses
        )

        if query._having_criteria:
            self._having_criteria = tuple(
                current_adapter(crit, True) if current_adapter else crit
                for crit in query._having_criteria
            )

        self.group_by = (
            self._adapt_col_list(
                util.flatten_iterator(query._group_by_clauses), current_adapter
            )
            if current_adapter and query._group_by_clauses not in (None, False)
            else query._group_by_clauses or None
        )

        if self.eager_order_by:
            adapter = self.from_clauses[0]._target_adapter
            self.eager_order_by = adapter.copy_and_process(self.eager_order_by)

        if query._distinct_on:
            self.distinct_on = self._adapt_col_list(
                query._distinct_on, current_adapter
            )
        else:
            self.distinct_on = ()

        self.distinct = query._distinct

        if query._correlate:
            # ORM mapped entities that are mapped to joins can be passed
            # to .correlate, so here they are broken into their component
            # tables.
            self.correlate = tuple(
                util.flatten_iterator(
                    sql_util.surface_selectables(s) if s is not None else None
                    for s in query._correlate
                )
            )
        elif query._correlate_except is not None:
            self.correlate_except = tuple(
                util.flatten_iterator(
                    sql_util.surface_selectables(s) if s is not None else None
                    for s in query._correlate_except
                )
            )
        elif not query._auto_correlate:
            self.correlate = (None,)

        # PART II

        self._for_update_arg = query._for_update_arg

        if self.compile_options._is_star and (len(self._entities) != 1):
            raise sa_exc.CompileError(
                "Can't generate ORM query that includes multiple expressions "
                "at the same time as '*'; query for '*' alone if present"
            )
        for entity in self._entities:
            entity.setup_compile_state(self)

        for rec in self.create_eager_joins:
            strategy = rec[0]
            strategy(self, *rec[1:])

        # else "load from discrete FROMs" mode,
        # i.e. when each _MappedEntity has its own FROM

        if self.compile_options._enable_single_crit:
            self._adjust_for_extra_criteria()

        if not self.primary_columns:
            if self.compile_options._only_load_props:
                assert False, "no columns were included in _only_load_props"

            raise sa_exc.InvalidRequestError(
                "Query contains no columns with which to SELECT from."
            )

        if not self.from_clauses:
            self.from_clauses = list(self._fallback_from_clauses)

        if self.order_by is False:
            self.order_by = None

        if (
            self.multi_row_eager_loaders
            and self.eager_adding_joins
            and self._should_nest_selectable
        ):
            self.statement = self._compound_eager_statement()
        else:
            self.statement = self._simple_statement()

        if self.for_statement:
            ezero = self._mapper_zero()
            if ezero is not None:
                # TODO: this goes away once we get rid of the deep entity
                # thing
                self.statement = self.statement._annotate(
                    {"deepentity": ezero}
                )

    @classmethod
    def _create_entities_collection(cls, query, legacy):
        """Creates a partial ORMSelectCompileState that includes
        the full collection of _MapperEntity and other _QueryEntity objects.

        Supports a few remaining use cases that are pre-compilation
        but still need to gather some of the column  / adaption information.

        """
        self = cls.__new__(cls)

        self._entities = []
        self._primary_entity = None
        self._polymorphic_adapters = {}

        self._label_convention = self._column_naming_convention(
            query._label_style, legacy
        )

        # entities will also set up polymorphic adapters for mappers
        # that have with_polymorphic configured
        _QueryEntity.to_compile_state(
            self, query._raw_columns, self._entities, is_current_entities=True
        )
        return self

    @classmethod
    def determine_last_joined_entity(cls, statement):
        setup_joins = statement._setup_joins

        return _determine_last_joined_entity(setup_joins, None)

    @classmethod
    def all_selected_columns(cls, statement):
        for element in statement._raw_columns:
            if (
                element.is_selectable
                and "entity_namespace" in element._annotations
            ):
                ens = element._annotations["entity_namespace"]
                if not ens.is_mapper and not ens.is_aliased_class:
                    yield from _select_iterables([element])
                else:
                    yield from _select_iterables(ens._all_column_expressions)
            else:
                yield from _select_iterables([element])

    @classmethod
    def get_columns_clause_froms(cls, statement):
        return cls._normalize_froms(
            itertools.chain.from_iterable(
                (
                    element._from_objects
                    if "parententity" not in element._annotations
                    else [
                        element._annotations[
                            "parententity"
                        ].__clause_element__()
                    ]
                )
                for element in statement._raw_columns
            )
        )

    @classmethod
    def from_statement(cls, statement, from_statement):
        from_statement = coercions.expect(
            roles.ReturnsRowsRole,
            from_statement,
            apply_propagate_attrs=statement,
        )

        stmt = FromStatement(statement._raw_columns, from_statement)

        stmt.__dict__.update(
            _with_options=statement._with_options,
            _with_context_options=statement._with_context_options,
            _execution_options=statement._execution_options,
            _propagate_attrs=statement._propagate_attrs,
        )
        return stmt

    def _set_select_from_alias(self):
        """used only for legacy Query cases"""

        query = self.select_statement  # query

        assert self.compile_options._set_base_alias
        assert len(query._from_obj) == 1

        adapter = self._get_select_from_alias_from_obj(query._from_obj[0])
        if adapter:
            self.compile_options += {"_enable_single_crit": False}
            self._from_obj_alias = adapter

    def _get_select_from_alias_from_obj(self, from_obj):
        """used only for legacy Query cases"""

        info = from_obj

        if "parententity" in info._annotations:
            info = info._annotations["parententity"]

        if hasattr(info, "mapper"):
            if not info.is_aliased_class:
                raise sa_exc.ArgumentError(
                    "A selectable (FromClause) instance is "
                    "expected when the base alias is being set."
                )
            else:
                return info._adapter

        elif isinstance(info.selectable, sql.selectable.AliasedReturnsRows):
            equivs = self._all_equivs()
            assert info is info.selectable
            return ORMStatementAdapter(
                _TraceAdaptRole.LEGACY_SELECT_FROM_ALIAS,
                info.selectable,
                equivalents=equivs,
            )
        else:
            return None

    def _mapper_zero(self):
        """return the Mapper associated with the first QueryEntity."""
        return self._entities[0].mapper

    def _entity_zero(self):
        """Return the 'entity' (mapper or AliasedClass) associated
        with the first QueryEntity, or alternatively the 'select from'
        entity if specified."""

        for ent in self.from_clauses:
            if "parententity" in ent._annotations:
                return ent._annotations["parententity"]
        for qent in self._entities:
            if qent.entity_zero:
                return qent.entity_zero

        return None

    def _only_full_mapper_zero(self, methname):
        if self._entities != [self._primary_entity]:
            raise sa_exc.InvalidRequestError(
                "%s() can only be used against "
                "a single mapped class." % methname
            )
        return self._primary_entity.entity_zero

    def _only_entity_zero(self, rationale=None):
        if len(self._entities) > 1:
            raise sa_exc.InvalidRequestError(
                rationale
                or "This operation requires a Query "
                "against a single mapper."
            )
        return self._entity_zero()

    def _all_equivs(self):
        equivs = {}

        for memoized_entities in self._memoized_entities.values():
            for ent in [
                ent
                for ent in memoized_entities
                if isinstance(ent, _MapperEntity)
            ]:
                equivs.update(ent.mapper._equivalent_columns)

        for ent in [
            ent for ent in self._entities if isinstance(ent, _MapperEntity)
        ]:
            equivs.update(ent.mapper._equivalent_columns)
        return equivs

    def _compound_eager_statement(self):
        # for eager joins present and LIMIT/OFFSET/DISTINCT,
        # wrap the query inside a select,
        # then append eager joins onto that

        if self.order_by:
            # the default coercion for ORDER BY is now the OrderByRole,
            # which adds an additional post coercion to ByOfRole in that
            # elements are converted into label references.  For the
            # eager load / subquery wrapping case, we need to un-coerce
            # the original expressions outside of the label references
            # in order to have them render.
            unwrapped_order_by = [
                (
                    elem.element
                    if isinstance(elem, sql.elements._label_reference)
                    else elem
                )
                for elem in self.order_by
            ]

            order_by_col_expr = sql_util.expand_column_list_from_order_by(
                self.primary_columns, unwrapped_order_by
            )
        else:
            order_by_col_expr = []
            unwrapped_order_by = None

        # put FOR UPDATE on the inner query, where MySQL will honor it,
        # as well as if it has an OF so PostgreSQL can use it.
        inner = self._select_statement(
            self.primary_columns
            + [c for c in order_by_col_expr if c not in self.dedupe_columns],
            self.from_clauses,
            self._where_criteria,
            self._having_criteria,
            self.label_style,
            self.order_by,
            for_update=self._for_update_arg,
            hints=self.select_statement._hints,
            statement_hints=self.select_statement._statement_hints,
            correlate=self.correlate,
            correlate_except=self.correlate_except,
            **self._select_args,
        )

        inner = inner.alias()

        equivs = self._all_equivs()

        self.compound_eager_adapter = ORMStatementAdapter(
            _TraceAdaptRole.COMPOUND_EAGER_STATEMENT, inner, equivalents=equivs
        )

        statement = future.select(
            *([inner] + self.secondary_columns)  # use_labels=self.labels
        )
        statement._label_style = self.label_style

        # Oracle however does not allow FOR UPDATE on the subquery,
        # and the Oracle dialect ignores it, plus for PostgreSQL, MySQL
        # we expect that all elements of the row are locked, so also put it
        # on the outside (except in the case of PG when OF is used)
        if (
            self._for_update_arg is not None
            and self._for_update_arg.of is None
        ):
            statement._for_update_arg = self._for_update_arg

        from_clause = inner
        for eager_join in self.eager_joins.values():
            # EagerLoader places a 'stop_on' attribute on the join,
            # giving us a marker as to where the "splice point" of
            # the join should be
            from_clause = sql_util.splice_joins(
                from_clause, eager_join, eager_join.stop_on
            )

        statement.select_from.non_generative(statement, from_clause)

        if unwrapped_order_by:
            statement.order_by.non_generative(
                statement,
                *self.compound_eager_adapter.copy_and_process(
                    unwrapped_order_by
                ),
            )

        statement.order_by.non_generative(statement, *self.eager_order_by)
        return statement

    def _simple_statement(self):
        statement = self._select_statement(
            self.primary_columns + self.secondary_columns,
            tuple(self.from_clauses) + tuple(self.eager_joins.values()),
            self._where_criteria,
            self._having_criteria,
            self.label_style,
            self.order_by,
            for_update=self._for_update_arg,
            hints=self.select_statement._hints,
            statement_hints=self.select_statement._statement_hints,
            correlate=self.correlate,
            correlate_except=self.correlate_except,
            **self._select_args,
        )

        if self.eager_order_by:
            statement.order_by.non_generative(statement, *self.eager_order_by)
        return statement

    def _select_statement(
        self,
        raw_columns,
        from_obj,
        where_criteria,
        having_criteria,
        label_style,
        order_by,
        for_update,
        hints,
        statement_hints,
        correlate,
        correlate_except,
        limit_clause,
        offset_clause,
        fetch_clause,
        fetch_clause_options,
        distinct,
        distinct_on,
        prefixes,
        suffixes,
        group_by,
        independent_ctes,
        independent_ctes_opts,
    ):
        statement = Select._create_raw_select(
            _raw_columns=raw_columns,
            _from_obj=from_obj,
            _label_style=label_style,
        )

        if where_criteria:
            statement._where_criteria = where_criteria
        if having_criteria:
            statement._having_criteria = having_criteria

        if order_by:
            statement._order_by_clauses += tuple(order_by)

        if distinct_on:
            statement.distinct.non_generative(statement, *distinct_on)
        elif distinct:
            statement.distinct.non_generative(statement)

        if group_by:
            statement._group_by_clauses += tuple(group_by)

        statement._limit_clause = limit_clause
        statement._offset_clause = offset_clause
        statement._fetch_clause = fetch_clause
        statement._fetch_clause_options = fetch_clause_options
        statement._independent_ctes = independent_ctes
        statement._independent_ctes_opts = independent_ctes_opts

        if prefixes:
            statement._prefixes = prefixes

        if suffixes:
            statement._suffixes = suffixes

        statement._for_update_arg = for_update

        if hints:
            statement._hints = hints
        if statement_hints:
            statement._statement_hints = statement_hints

        if correlate:
            statement.correlate.non_generative(statement, *correlate)

        if correlate_except is not None:
            statement.correlate_except.non_generative(
                statement, *correlate_except
            )

        return statement

    def _adapt_polymorphic_element(self, element):
        if "parententity" in element._annotations:
            search = element._annotations["parententity"]
            alias = self._polymorphic_adapters.get(search, None)
            if alias:
                return alias.adapt_clause(element)

        if isinstance(element, expression.FromClause):
            search = element
        elif hasattr(element, "table"):
            search = element.table
        else:
            return None

        alias = self._polymorphic_adapters.get(search, None)
        if alias:
            return alias.adapt_clause(element)

    def _adapt_col_list(self, cols, current_adapter):
        if current_adapter:
            return [current_adapter(o, True) for o in cols]
        else:
            return cols

    def _get_current_adapter(self):
        adapters = []

        if self._from_obj_alias:
            # used for legacy going forward for query set_ops, e.g.
            # union(), union_all(), etc.
            # 1.4 and previously, also used for from_self(),
            # select_entity_from()
            #
            # for the "from obj" alias, apply extra rule to the
            # 'ORM only' check, if this query were generated from a
            # subquery of itself, i.e. _from_selectable(), apply adaption
            # to all SQL constructs.
            adapters.append(
                (
                    True,
                    self._from_obj_alias.replace,
                )
            )

        # this was *hopefully* the only adapter we were going to need
        # going forward...however, we unfortunately need _from_obj_alias
        # for query.union(), which we can't drop
        if self._polymorphic_adapters:
            adapters.append((False, self._adapt_polymorphic_element))

        if not adapters:
            return None

        def _adapt_clause(clause, as_filter):
            # do we adapt all expression elements or only those
            # tagged as 'ORM' constructs ?

            def replace(elem):
                is_orm_adapt = (
                    "_orm_adapt" in elem._annotations
                    or "parententity" in elem._annotations
                )
                for always_adapt, adapter in adapters:
                    if is_orm_adapt or always_adapt:
                        e = adapter(elem)
                        if e is not None:
                            return e

            return visitors.replacement_traverse(clause, {}, replace)

        return _adapt_clause

    def _join(self, args, entities_collection):
        for right, onclause, from_, flags in args:
            isouter = flags["isouter"]
            full = flags["full"]

            right = inspect(right)
            if onclause is not None:
                onclause = inspect(onclause)

            if isinstance(right, interfaces.PropComparator):
                if onclause is not None:
                    raise sa_exc.InvalidRequestError(
                        "No 'on clause' argument may be passed when joining "
                        "to a relationship path as a target"
                    )

                onclause = right
                right = None
            elif "parententity" in right._annotations:
                right = right._annotations["parententity"]

            if onclause is None:
                if not right.is_selectable and not hasattr(right, "mapper"):
                    raise sa_exc.ArgumentError(
                        "Expected mapped entity or "
                        "selectable/table as join target"
                    )

            of_type = None

            if isinstance(onclause, interfaces.PropComparator):
                # descriptor/property given (or determined); this tells us
                # explicitly what the expected "left" side of the join is.

                of_type = getattr(onclause, "_of_type", None)

                if right is None:
                    if of_type:
                        right = of_type
                    else:
                        right = onclause.property

                        try:
                            right = right.entity
                        except AttributeError as err:
                            raise sa_exc.ArgumentError(
                                "Join target %s does not refer to a "
                                "mapped entity" % right
                            ) from err

                left = onclause._parententity

                prop = onclause.property
                if not isinstance(onclause, attributes.QueryableAttribute):
                    onclause = prop

                # check for this path already present.  don't render in that
                # case.
                if (left, right, prop.key) in self._already_joined_edges:
                    continue

                if from_ is not None:
                    if (
                        from_ is not left
                        and from_._annotations.get("parententity", None)
                        is not left
                    ):
                        raise sa_exc.InvalidRequestError(
                            "explicit from clause %s does not match left side "
                            "of relationship attribute %s"
                            % (
                                from_._annotations.get("parententity", from_),
                                onclause,
                            )
                        )
            elif from_ is not None:
                prop = None
                left = from_
            else:
                # no descriptor/property given; we will need to figure out
                # what the effective "left" side is
                prop = left = None

            # figure out the final "left" and "right" sides and create an
            # ORMJoin to add to our _from_obj tuple
            self._join_left_to_right(
                entities_collection,
                left,
                right,
                onclause,
                prop,
                isouter,
                full,
            )

    def _join_left_to_right(
        self,
        entities_collection,
        left,
        right,
        onclause,
        prop,
        outerjoin,
        full,
    ):
        """given raw "left", "right", "onclause" parameters consumed from
        a particular key within _join(), add a real ORMJoin object to
        our _from_obj list (or augment an existing one)

        """

        if left is None:
            # left not given (e.g. no relationship object/name specified)
            # figure out the best "left" side based on our existing froms /
            # entities
            assert prop is None
            (
                left,
                replace_from_obj_index,
                use_entity_index,
            ) = self._join_determine_implicit_left_side(
                entities_collection, left, right, onclause
            )
        else:
            # left is given via a relationship/name, or as explicit left side.
            # Determine where in our
            # "froms" list it should be spliced/appended as well as what
            # existing entity it corresponds to.
            (
                replace_from_obj_index,
                use_entity_index,
            ) = self._join_place_explicit_left_side(entities_collection, left)

        if left is right:
            raise sa_exc.InvalidRequestError(
                "Can't construct a join from %s to %s, they "
                "are the same entity" % (left, right)
            )

        # the right side as given often needs to be adapted.  additionally
        # a lot of things can be wrong with it.  handle all that and
        # get back the new effective "right" side
        r_info, right, onclause = self._join_check_and_adapt_right_side(
            left, right, onclause, prop
        )

        if not r_info.is_selectable:
            extra_criteria = self._get_extra_criteria(r_info)
        else:
            extra_criteria = ()

        if replace_from_obj_index is not None:
            # splice into an existing element in the
            # self._from_obj list
            left_clause = self.from_clauses[replace_from_obj_index]

            self.from_clauses = (
                self.from_clauses[:replace_from_obj_index]
                + [
                    _ORMJoin(
                        left_clause,
                        right,
                        onclause,
                        isouter=outerjoin,
                        full=full,
                        _extra_criteria=extra_criteria,
                    )
                ]
                + self.from_clauses[replace_from_obj_index + 1 :]
            )
        else:
            # add a new element to the self._from_obj list
            if use_entity_index is not None:
                # make use of _MapperEntity selectable, which is usually
                # entity_zero.selectable, but if with_polymorphic() were used
                # might be distinct
                assert isinstance(
                    entities_collection[use_entity_index], _MapperEntity
                )
                left_clause = entities_collection[use_entity_index].selectable
            else:
                left_clause = left

            self.from_clauses = self.from_clauses + [
                _ORMJoin(
                    left_clause,
                    r_info,
                    onclause,
                    isouter=outerjoin,
                    full=full,
                    _extra_criteria=extra_criteria,
                )
            ]

    def _join_determine_implicit_left_side(
        self, entities_collection, left, right, onclause
    ):
        """When join conditions don't express the left side explicitly,
        determine if an existing FROM or entity in this query
        can serve as the left hand side.

        """

        # when we are here, it means join() was called without an ORM-
        # specific way of telling us what the "left" side is, e.g.:
        #
        # join(RightEntity)
        #
        # or
        #
        # join(RightEntity, RightEntity.foo == LeftEntity.bar)
        #

        r_info = inspect(right)

        replace_from_obj_index = use_entity_index = None

        if self.from_clauses:
            # we have a list of FROMs already.  So by definition this
            # join has to connect to one of those FROMs.

            indexes = sql_util.find_left_clause_to_join_from(
                self.from_clauses, r_info.selectable, onclause
            )

            if len(indexes) == 1:
                replace_from_obj_index = indexes[0]
                left = self.from_clauses[replace_from_obj_index]
            elif len(indexes) > 1:
                raise sa_exc.InvalidRequestError(
                    "Can't determine which FROM clause to join "
                    "from, there are multiple FROMS which can "
                    "join to this entity. Please use the .select_from() "
                    "method to establish an explicit left side, as well as "
                    "providing an explicit ON clause if not present already "
                    "to help resolve the ambiguity."
                )
            else:
                raise sa_exc.InvalidRequestError(
                    "Don't know how to join to %r. "
                    "Please use the .select_from() "
                    "method to establish an explicit left side, as well as "
                    "providing an explicit ON clause if not present already "
                    "to help resolve the ambiguity." % (right,)
                )

        elif entities_collection:
            # we have no explicit FROMs, so the implicit left has to
            # come from our list of entities.

            potential = {}
            for entity_index, ent in enumerate(entities_collection):
                entity = ent.entity_zero_or_selectable
                if entity is None:
                    continue
                ent_info = inspect(entity)
                if ent_info is r_info:  # left and right are the same, skip
                    continue

                # by using a dictionary with the selectables as keys this
                # de-duplicates those selectables as occurs when the query is
                # against a series of columns from the same selectable
                if isinstance(ent, _MapperEntity):
                    potential[ent.selectable] = (entity_index, entity)
                else:
                    potential[ent_info.selectable] = (None, entity)

            all_clauses = list(potential.keys())
            indexes = sql_util.find_left_clause_to_join_from(
                all_clauses, r_info.selectable, onclause
            )

            if len(indexes) == 1:
                use_entity_index, left = potential[all_clauses[indexes[0]]]
            elif len(indexes) > 1:
                raise sa_exc.InvalidRequestError(
                    "Can't determine which FROM clause to join "
                    "from, there are multiple FROMS which can "
                    "join to this entity. Please use the .select_from() "
                    "method to establish an explicit left side, as well as "
                    "providing an explicit ON clause if not present already "
                    "to help resolve the ambiguity."
                )
            else:
                raise sa_exc.InvalidRequestError(
                    "Don't know how to join to %r. "
                    "Please use the .select_from() "
                    "method to establish an explicit left side, as well as "
                    "providing an explicit ON clause if not present already "
                    "to help resolve the ambiguity." % (right,)
                )
        else:
            raise sa_exc.InvalidRequestError(
                "No entities to join from; please use "
                "select_from() to establish the left "
                "entity/selectable of this join"
            )

        return left, replace_from_obj_index, use_entity_index

    def _join_place_explicit_left_side(self, entities_collection, left):
        """When join conditions express a left side explicitly, determine
        where in our existing list of FROM clauses we should join towards,
        or if we need to make a new join, and if so is it from one of our
        existing entities.

        """

        # when we are here, it means join() was called with an indicator
        # as to an exact left side, which means a path to a
        # Relationship was given, e.g.:
        #
        # join(RightEntity, LeftEntity.right)
        #
        # or
        #
        # join(LeftEntity.right)
        #
        # as well as string forms:
        #
        # join(RightEntity, "right")
        #
        # etc.
        #

        replace_from_obj_index = use_entity_index = None

        l_info = inspect(left)
        if self.from_clauses:
            indexes = sql_util.find_left_clause_that_matches_given(
                self.from_clauses, l_info.selectable
            )

            if len(indexes) > 1:
                raise sa_exc.InvalidRequestError(
                    "Can't identify which entity in which to assign the "
                    "left side of this join.   Please use a more specific "
                    "ON clause."
                )

            # have an index, means the left side is already present in
            # an existing FROM in the self._from_obj tuple
            if indexes:
                replace_from_obj_index = indexes[0]

            # no index, means we need to add a new element to the
            # self._from_obj tuple

        # no from element present, so we will have to add to the
        # self._from_obj tuple.  Determine if this left side matches up
        # with existing mapper entities, in which case we want to apply the
        # aliasing / adaptation rules present on that entity if any
        if (
            replace_from_obj_index is None
            and entities_collection
            and hasattr(l_info, "mapper")
        ):
            for idx, ent in enumerate(entities_collection):
                # TODO: should we be checking for multiple mapper entities
                # matching?
                if isinstance(ent, _MapperEntity) and ent.corresponds_to(left):
                    use_entity_index = idx
                    break

        return replace_from_obj_index, use_entity_index

    def _join_check_and_adapt_right_side(self, left, right, onclause, prop):
        """transform the "right" side of the join as well as the onclause
        according to polymorphic mapping translations, aliasing on the query
        or on the join, special cases where the right and left side have
        overlapping tables.

        """

        l_info = inspect(left)
        r_info = inspect(right)

        overlap = False

        right_mapper = getattr(r_info, "mapper", None)
        # if the target is a joined inheritance mapping,
        # be more liberal about auto-aliasing.
        if right_mapper and (
            right_mapper.with_polymorphic
            or isinstance(right_mapper.persist_selectable, expression.Join)
        ):
            for from_obj in self.from_clauses or [l_info.selectable]:
                if sql_util.selectables_overlap(
                    l_info.selectable, from_obj
                ) and sql_util.selectables_overlap(
                    from_obj, r_info.selectable
                ):
                    overlap = True
                    break

        if overlap and l_info.selectable is r_info.selectable:
            raise sa_exc.InvalidRequestError(
                "Can't join table/selectable '%s' to itself"
                % l_info.selectable
            )

        right_mapper, right_selectable, right_is_aliased = (
            getattr(r_info, "mapper", None),
            r_info.selectable,
            getattr(r_info, "is_aliased_class", False),
        )

        if (
            right_mapper
            and prop
            and not right_mapper.common_parent(prop.mapper)
        ):
            raise sa_exc.InvalidRequestError(
                "Join target %s does not correspond to "
                "the right side of join condition %s" % (right, onclause)
            )

        # _join_entities is used as a hint for single-table inheritance
        # purposes at the moment
        if hasattr(r_info, "mapper"):
            self._join_entities += (r_info,)

        need_adapter = False

        # test for joining to an unmapped selectable as the target
        if r_info.is_clause_element:
            if prop:
                right_mapper = prop.mapper

            if right_selectable._is_lateral:
                # orm_only is disabled to suit the case where we have to
                # adapt an explicit correlate(Entity) - the select() loses
                # the ORM-ness in this case right now, ideally it would not
                current_adapter = self._get_current_adapter()
                if current_adapter is not None:
                    # TODO: we had orm_only=False here before, removing
                    # it didn't break things.   if we identify the rationale,
                    # may need to apply "_orm_only" annotation here.
                    right = current_adapter(right, True)

            elif prop:
                # joining to selectable with a mapper property given
                # as the ON clause

                if not right_selectable.is_derived_from(
                    right_mapper.persist_selectable
                ):
                    raise sa_exc.InvalidRequestError(
                        "Selectable '%s' is not derived from '%s'"
                        % (
                            right_selectable.description,
                            right_mapper.persist_selectable.description,
                        )
                    )

                # if the destination selectable is a plain select(),
                # turn it into an alias().
                if isinstance(right_selectable, expression.SelectBase):
                    right_selectable = coercions.expect(
                        roles.FromClauseRole, right_selectable
                    )
                    need_adapter = True

                # make the right hand side target into an ORM entity
                right = AliasedClass(right_mapper, right_selectable)

                util.warn_deprecated(
                    "An alias is being generated automatically against "
                    "joined entity %s for raw clauseelement, which is "
                    "deprecated and will be removed in a later release. "
                    "Use the aliased() "
                    "construct explicitly, see the linked example."
                    % right_mapper,
                    "1.4",
                    code="xaj1",
                )

        # test for overlap:
        # orm/inheritance/relationships.py
        # SelfReferentialM2MTest
        aliased_entity = right_mapper and not right_is_aliased and overlap

        if not need_adapter and aliased_entity:
            # there are a few places in the ORM that automatic aliasing
            # is still desirable, and can't be automatic with a Core
            # only approach.  For illustrations of "overlaps" see
            # test/orm/inheritance/test_relationships.py.  There are also
            # general overlap cases with many-to-many tables where automatic
            # aliasing is desirable.
            right = AliasedClass(right, flat=True)
            need_adapter = True

            util.warn(
                "An alias is being generated automatically against "
                "joined entity %s due to overlapping tables.  This is a "
                "legacy pattern which may be "
                "deprecated in a later release.  Use the "
                "aliased(<entity>, flat=True) "
                "construct explicitly, see the linked example." % right_mapper,
                code="xaj2",
            )

        if need_adapter:
            # if need_adapter is True, we are in a deprecated case and
            # a warning has been emitted.
            assert right_mapper

            adapter = ORMAdapter(
                _TraceAdaptRole.DEPRECATED_JOIN_ADAPT_RIGHT_SIDE,
                inspect(right),
                equivalents=right_mapper._equivalent_columns,
            )

            # if an alias() on the right side was generated,
            # which is intended to wrap a the right side in a subquery,
            # ensure that columns retrieved from this target in the result
            # set are also adapted.
            self._mapper_loads_polymorphically_with(right_mapper, adapter)
        elif (
            not r_info.is_clause_element
            and not right_is_aliased
            and right_mapper._has_aliased_polymorphic_fromclause
        ):
            # for the case where the target mapper has a with_polymorphic
            # set up, ensure an adapter is set up for criteria that works
            # against this mapper.  Previously, this logic used to
            # use the "create_aliases or aliased_entity" case to generate
            # an aliased() object, but this creates an alias that isn't
            # strictly necessary.
            # see test/orm/test_core_compilation.py
            # ::RelNaturalAliasedJoinsTest::test_straight
            # and similar
            self._mapper_loads_polymorphically_with(
                right_mapper,
                ORMAdapter(
                    _TraceAdaptRole.WITH_POLYMORPHIC_ADAPTER_RIGHT_JOIN,
                    right_mapper,
                    selectable=right_mapper.selectable,
                    equivalents=right_mapper._equivalent_columns,
                ),
            )
        # if the onclause is a ClauseElement, adapt it with any
        # adapters that are in place right now
        if isinstance(onclause, expression.ClauseElement):
            current_adapter = self._get_current_adapter()
            if current_adapter:
                onclause = current_adapter(onclause, True)

        # if joining on a MapperProperty path,
        # track the path to prevent redundant joins
        if prop:
            self._already_joined_edges += ((left, right, prop.key),)

        return inspect(right), right, onclause

    @property
    def _select_args(self):
        return {
            "limit_clause": self.select_statement._limit_clause,
            "offset_clause": self.select_statement._offset_clause,
            "distinct": self.distinct,
            "distinct_on": self.distinct_on,
            "prefixes": self.select_statement._prefixes,
            "suffixes": self.select_statement._suffixes,
            "group_by": self.group_by or None,
            "fetch_clause": self.select_statement._fetch_clause,
            "fetch_clause_options": (
                self.select_statement._fetch_clause_options
            ),
            "independent_ctes": self.select_statement._independent_ctes,
            "independent_ctes_opts": (
                self.select_statement._independent_ctes_opts
            ),
        }

    @property
    def _should_nest_selectable(self):
        kwargs = self._select_args
        return (
            kwargs.get("limit_clause") is not None
            or kwargs.get("offset_clause") is not None
            or kwargs.get("distinct", False)
            or kwargs.get("distinct_on", ())
            or kwargs.get("group_by", False)
        )

    def _get_extra_criteria(self, ext_info):
        if (
            "additional_entity_criteria",
            ext_info.mapper,
        ) in self.global_attributes:
            return tuple(
                ae._resolve_where_criteria(ext_info)
                for ae in self.global_attributes[
                    ("additional_entity_criteria", ext_info.mapper)
                ]
                if (ae.include_aliases or ae.entity is ext_info)
                and ae._should_include(self)
            )
        else:
            return ()

    def _adjust_for_extra_criteria(self):
        """Apply extra criteria filtering.

        For all distinct single-table-inheritance mappers represented in
        the columns clause of this query, as well as the "select from entity",
        add criterion to the WHERE
        clause of the given QueryContext such that only the appropriate
        subtypes are selected from the total results.

        Additionally, add WHERE criteria originating from LoaderCriteriaOptions
        associated with the global context.

        """

        for fromclause in self.from_clauses:
            ext_info = fromclause._annotations.get("parententity", None)

            if (
                ext_info
                and (
                    ext_info.mapper._single_table_criterion is not None
                    or ("additional_entity_criteria", ext_info.mapper)
                    in self.global_attributes
                )
                and ext_info not in self.extra_criteria_entities
            ):
                self.extra_criteria_entities[ext_info] = (
                    ext_info,
                    ext_info._adapter if ext_info.is_aliased_class else None,
                )

        search = set(self.extra_criteria_entities.values())

        for ext_info, adapter in search:
            if ext_info in self._join_entities:
                continue

            single_crit = ext_info.mapper._single_table_criterion

            if self.compile_options._for_refresh_state:
                additional_entity_criteria = []
            else:
                additional_entity_criteria = self._get_extra_criteria(ext_info)

            if single_crit is not None:
                additional_entity_criteria += (single_crit,)

            current_adapter = self._get_current_adapter()
            for crit in additional_entity_criteria:
                if adapter:
                    crit = adapter.traverse(crit)

                if current_adapter:
                    crit = sql_util._deep_annotate(crit, {"_orm_adapt": True})
                    crit = current_adapter(crit, False)
                self._where_criteria += (crit,)


def _column_descriptions(
    query_or_select_stmt: Union[Query, Select, FromStatement],
    compile_state: Optional[ORMSelectCompileState] = None,
    legacy: bool = False,
) -> List[ORMColumnDescription]:
    if compile_state is None:
        compile_state = ORMSelectCompileState._create_entities_collection(
            query_or_select_stmt, legacy=legacy
        )
    ctx = compile_state
    d = [
        {
            "name": ent._label_name,
            "type": ent.type,
            "aliased": getattr(insp_ent, "is_aliased_class", False),
            "expr": ent.expr,
            "entity": (
                getattr(insp_ent, "entity", None)
                if ent.entity_zero is not None
                and not insp_ent.is_clause_element
                else None
            ),
        }
        for ent, insp_ent in [
            (_ent, _ent.entity_zero) for _ent in ctx._entities
        ]
    ]
    return d


def _legacy_filter_by_entity_zero(
    query_or_augmented_select: Union[Query[Any], Select[Any]]
) -> Optional[_InternalEntityType[Any]]:
    self = query_or_augmented_select
    if self._setup_joins:
        _last_joined_entity = self._last_joined_entity
        if _last_joined_entity is not None:
            return _last_joined_entity

    if self._from_obj and "parententity" in self._from_obj[0]._annotations:
        return self._from_obj[0]._annotations["parententity"]

    return _entity_from_pre_ent_zero(self)


def _entity_from_pre_ent_zero(
    query_or_augmented_select: Union[Query[Any], Select[Any]]
) -> Optional[_InternalEntityType[Any]]:
    self = query_or_augmented_select
    if not self._raw_columns:
        return None

    ent = self._raw_columns[0]

    if "parententity" in ent._annotations:
        return ent._annotations["parententity"]
    elif isinstance(ent, ORMColumnsClauseRole):
        return ent.entity
    elif "bundle" in ent._annotations:
        return ent._annotations["bundle"]
    else:
        return ent


def _determine_last_joined_entity(
    setup_joins: Tuple[_SetupJoinsElement, ...],
    entity_zero: Optional[_InternalEntityType[Any]] = None,
) -> Optional[Union[_InternalEntityType[Any], _JoinTargetElement]]:
    if not setup_joins:
        return None

    (target, onclause, from_, flags) = setup_joins[-1]

    if isinstance(
        target,
        attributes.QueryableAttribute,
    ):
        return target.entity
    else:
        return target


class _QueryEntity:
    """represent an entity column returned within a Query result."""

    __slots__ = ()

    supports_single_entity: bool

    _non_hashable_value = False
    _null_column_type = False
    use_id_for_hash = False

    _label_name: Optional[str]
    type: Union[Type[Any], TypeEngine[Any]]
    expr: Union[_InternalEntityType, ColumnElement[Any]]
    entity_zero: Optional[_InternalEntityType]

    def setup_compile_state(self, compile_state: ORMCompileState) -> None:
        raise NotImplementedError()

    def setup_dml_returning_compile_state(
        self,
        compile_state: ORMCompileState,
        adapter: DMLReturningColFilter,
    ) -> None:
        raise NotImplementedError()

    def row_processor(self, context, result):
        raise NotImplementedError()

    @classmethod
    def to_compile_state(
        cls, compile_state, entities, entities_collection, is_current_entities
    ):
        for idx, entity in enumerate(entities):
            if entity._is_lambda_element:
                if entity._is_sequence:
                    cls.to_compile_state(
                        compile_state,
                        entity._resolved,
                        entities_collection,
                        is_current_entities,
                    )
                    continue
                else:
                    entity = entity._resolved

            if entity.is_clause_element:
                if entity.is_selectable:
                    if "parententity" in entity._annotations:
                        _MapperEntity(
                            compile_state,
                            entity,
                            entities_collection,
                            is_current_entities,
                        )
                    else:
                        _ColumnEntity._for_columns(
                            compile_state,
                            entity._select_iterable,
                            entities_collection,
                            idx,
                            is_current_entities,
                        )
                else:
                    if entity._annotations.get("bundle", False):
                        _BundleEntity(
                            compile_state,
                            entity,
                            entities_collection,
                            is_current_entities,
                        )
                    elif entity._is_clause_list:
                        # this is legacy only - test_composites.py
                        # test_query_cols_legacy
                        _ColumnEntity._for_columns(
                            compile_state,
                            entity._select_iterable,
                            entities_collection,
                            idx,
                            is_current_entities,
                        )
                    else:
                        _ColumnEntity._for_columns(
                            compile_state,
                            [entity],
                            entities_collection,
                            idx,
                            is_current_entities,
                        )
            elif entity.is_bundle:
                _BundleEntity(compile_state, entity, entities_collection)

        return entities_collection


class _MapperEntity(_QueryEntity):
    """mapper/class/AliasedClass entity"""

    __slots__ = (
        "expr",
        "mapper",
        "entity_zero",
        "is_aliased_class",
        "path",
        "_extra_entities",
        "_label_name",
        "_with_polymorphic_mappers",
        "selectable",
        "_polymorphic_discriminator",
    )

    expr: _InternalEntityType
    mapper: Mapper[Any]
    entity_zero: _InternalEntityType
    is_aliased_class: bool
    path: PathRegistry
    _label_name: str

    def __init__(
        self, compile_state, entity, entities_collection, is_current_entities
    ):
        entities_collection.append(self)
        if is_current_entities:
            if compile_state._primary_entity is None:
                compile_state._primary_entity = self
            compile_state._has_mapper_entities = True
            compile_state._has_orm_entities = True

        entity = entity._annotations["parententity"]
        entity._post_inspect
        ext_info = self.entity_zero = entity
        entity = ext_info.entity

        self.expr = entity
        self.mapper = mapper = ext_info.mapper

        self._extra_entities = (self.expr,)

        if ext_info.is_aliased_class:
            self._label_name = ext_info.name
        else:
            self._label_name = mapper.class_.__name__

        self.is_aliased_class = ext_info.is_aliased_class
        self.path = ext_info._path_registry

        self.selectable = ext_info.selectable
        self._with_polymorphic_mappers = ext_info.with_polymorphic_mappers
        self._polymorphic_discriminator = ext_info.polymorphic_on

        if mapper._should_select_with_poly_adapter:
            compile_state._create_with_polymorphic_adapter(
                ext_info, self.selectable
            )

    supports_single_entity = True

    _non_hashable_value = True
    use_id_for_hash = True

    @property
    def type(self):
        return self.mapper.class_

    @property
    def entity_zero_or_selectable(self):
        return self.entity_zero

    def corresponds_to(self, entity):
        return _entity_corresponds_to(self.entity_zero, entity)

    def _get_entity_clauses(self, compile_state):
        adapter = None

        if not self.is_aliased_class:
            if compile_state._polymorphic_adapters:
                adapter = compile_state._polymorphic_adapters.get(
                    self.mapper, None
                )
        else:
            adapter = self.entity_zero._adapter

        if adapter:
            if compile_state._from_obj_alias:
                ret = adapter.wrap(compile_state._from_obj_alias)
            else:
                ret = adapter
        else:
            ret = compile_state._from_obj_alias

        return ret

    def row_processor(self, context, result):
        compile_state = context.compile_state
        adapter = self._get_entity_clauses(compile_state)

        if compile_state.compound_eager_adapter and adapter:
            adapter = adapter.wrap(compile_state.compound_eager_adapter)
        elif not adapter:
            adapter = compile_state.compound_eager_adapter

        if compile_state._primary_entity is self:
            only_load_props = compile_state.compile_options._only_load_props
            refresh_state = context.refresh_state
        else:
            only_load_props = refresh_state = None

        _instance = loading._instance_processor(
            self,
            self.mapper,
            context,
            result,
            self.path,
            adapter,
            only_load_props=only_load_props,
            refresh_state=refresh_state,
            polymorphic_discriminator=self._polymorphic_discriminator,
        )

        return _instance, self._label_name, self._extra_entities

    def setup_dml_returning_compile_state(
        self,
        compile_state: ORMCompileState,
        adapter: DMLReturningColFilter,
    ) -> None:
        loading._setup_entity_query(
            compile_state,
            self.mapper,
            self,
            self.path,
            adapter,
            compile_state.primary_columns,
            with_polymorphic=self._with_polymorphic_mappers,
            only_load_props=compile_state.compile_options._only_load_props,
            polymorphic_discriminator=self._polymorphic_discriminator,
        )

    def setup_compile_state(self, compile_state):
        adapter = self._get_entity_clauses(compile_state)

        single_table_crit = self.mapper._single_table_criterion
        if (
            single_table_crit is not None
            or ("additional_entity_criteria", self.mapper)
            in compile_state.global_attributes
        ):
            ext_info = self.entity_zero
            compile_state.extra_criteria_entities[ext_info] = (
                ext_info,
                ext_info._adapter if ext_info.is_aliased_class else None,
            )

        loading._setup_entity_query(
            compile_state,
            self.mapper,
            self,
            self.path,
            adapter,
            compile_state.primary_columns,
            with_polymorphic=self._with_polymorphic_mappers,
            only_load_props=compile_state.compile_options._only_load_props,
            polymorphic_discriminator=self._polymorphic_discriminator,
        )
        compile_state._fallback_from_clauses.append(self.selectable)


class _BundleEntity(_QueryEntity):
    _extra_entities = ()

    __slots__ = (
        "bundle",
        "expr",
        "type",
        "_label_name",
        "_entities",
        "supports_single_entity",
    )

    _entities: List[_QueryEntity]
    bundle: Bundle
    type: Type[Any]
    _label_name: str
    supports_single_entity: bool
    expr: Bundle

    def __init__(
        self,
        compile_state,
        expr,
        entities_collection,
        is_current_entities,
        setup_entities=True,
        parent_bundle=None,
    ):
        compile_state._has_orm_entities = True

        expr = expr._annotations["bundle"]
        if parent_bundle:
            parent_bundle._entities.append(self)
        else:
            entities_collection.append(self)

        if isinstance(
            expr, (attributes.QueryableAttribute, interfaces.PropComparator)
        ):
            bundle = expr.__clause_element__()
        else:
            bundle = expr

        self.bundle = self.expr = bundle
        self.type = type(bundle)
        self._label_name = bundle.name
        self._entities = []

        if setup_entities:
            for expr in bundle.exprs:
                if "bundle" in expr._annotations:
                    _BundleEntity(
                        compile_state,
                        expr,
                        entities_collection,
                        is_current_entities,
                        parent_bundle=self,
                    )
                elif isinstance(expr, Bundle):
                    _BundleEntity(
                        compile_state,
                        expr,
                        entities_collection,
                        is_current_entities,
                        parent_bundle=self,
                    )
                else:
                    _ORMColumnEntity._for_columns(
                        compile_state,
                        [expr],
                        entities_collection,
                        None,
                        is_current_entities,
                        parent_bundle=self,
                    )

        self.supports_single_entity = self.bundle.single_entity

    @property
    def mapper(self):
        ezero = self.entity_zero
        if ezero is not None:
            return ezero.mapper
        else:
            return None

    @property
    def entity_zero(self):
        for ent in self._entities:
            ezero = ent.entity_zero
            if ezero is not None:
                return ezero
        else:
            return None

    def corresponds_to(self, entity):
        # TODO: we might be able to implement this but for now
        # we are working around it
        return False

    @property
    def entity_zero_or_selectable(self):
        for ent in self._entities:
            ezero = ent.entity_zero_or_selectable
            if ezero is not None:
                return ezero
        else:
            return None

    def setup_compile_state(self, compile_state):
        for ent in self._entities:
            ent.setup_compile_state(compile_state)

    def setup_dml_returning_compile_state(
        self,
        compile_state: ORMCompileState,
        adapter: DMLReturningColFilter,
    ) -> None:
        return self.setup_compile_state(compile_state)

    def row_processor(self, context, result):
        procs, labels, extra = zip(
            *[ent.row_processor(context, result) for ent in self._entities]
        )

        proc = self.bundle.create_row_processor(context.query, procs, labels)

        return proc, self._label_name, self._extra_entities


class _ColumnEntity(_QueryEntity):
    __slots__ = (
        "_fetch_column",
        "_row_processor",
        "raw_column_index",
        "translate_raw_column",
    )

    @classmethod
    def _for_columns(
        cls,
        compile_state,
        columns,
        entities_collection,
        raw_column_index,
        is_current_entities,
        parent_bundle=None,
    ):
        for column in columns:
            annotations = column._annotations
            if "parententity" in annotations:
                _entity = annotations["parententity"]
            else:
                _entity = sql_util.extract_first_column_annotation(
                    column, "parententity"
                )

            if _entity:
                if "identity_token" in column._annotations:
                    _IdentityTokenEntity(
                        compile_state,
                        column,
                        entities_collection,
                        _entity,
                        raw_column_index,
                        is_current_entities,
                        parent_bundle=parent_bundle,
                    )
                else:
                    _ORMColumnEntity(
                        compile_state,
                        column,
                        entities_collection,
                        _entity,
                        raw_column_index,
                        is_current_entities,
                        parent_bundle=parent_bundle,
                    )
            else:
                _RawColumnEntity(
                    compile_state,
                    column,
                    entities_collection,
                    raw_column_index,
                    is_current_entities,
                    parent_bundle=parent_bundle,
                )

    @property
    def type(self):
        return self.column.type

    @property
    def _non_hashable_value(self):
        return not self.column.type.hashable

    @property
    def _null_column_type(self):
        return self.column.type._isnull

    def row_processor(self, context, result):
        compile_state = context.compile_state

        # the resulting callable is entirely cacheable so just return
        # it if we already made one
        if self._row_processor is not None:
            getter, label_name, extra_entities = self._row_processor
            if self.translate_raw_column:
                extra_entities += (
                    context.query._raw_columns[self.raw_column_index],
                )

            return getter, label_name, extra_entities

        # retrieve the column that would have been set up in
        # setup_compile_state, to avoid doing redundant work
        if self._fetch_column is not None:
            column = self._fetch_column
        else:
            # fetch_column will be None when we are doing a from_statement
            # and setup_compile_state may not have been called.
            column = self.column

            # previously, the RawColumnEntity didn't look for from_obj_alias
            # however I can't think of a case where we would be here and
            # we'd want to ignore it if this is the from_statement use case.
            # it's not really a use case to have raw columns + from_statement
            if compile_state._from_obj_alias:
                column = compile_state._from_obj_alias.columns[column]

            if column._annotations:
                # annotated columns perform more slowly in compiler and
                # result due to the __eq__() method, so use deannotated
                column = column._deannotate()

        if compile_state.compound_eager_adapter:
            column = compile_state.compound_eager_adapter.columns[column]

        getter = result._getter(column)
        ret = getter, self._label_name, self._extra_entities
        self._row_processor = ret

        if self.translate_raw_column:
            extra_entities = self._extra_entities + (
                context.query._raw_columns[self.raw_column_index],
            )
            return getter, self._label_name, extra_entities
        else:
            return ret


class _RawColumnEntity(_ColumnEntity):
    entity_zero = None
    mapper = None
    supports_single_entity = False

    __slots__ = (
        "expr",
        "column",
        "_label_name",
        "entity_zero_or_selectable",
        "_extra_entities",
    )

    def __init__(
        self,
        compile_state,
        column,
        entities_collection,
        raw_column_index,
        is_current_entities,
        parent_bundle=None,
    ):
        self.expr = column
        self.raw_column_index = raw_column_index
        self.translate_raw_column = raw_column_index is not None

        if column._is_star:
            compile_state.compile_options += {"_is_star": True}

        if not is_current_entities or column._is_text_clause:
            self._label_name = None
        else:
            if parent_bundle:
                self._label_name = column._proxy_key
            else:
                self._label_name = compile_state._label_convention(column)

        if parent_bundle:
            parent_bundle._entities.append(self)
        else:
            entities_collection.append(self)

        self.column = column
        self.entity_zero_or_selectable = (
            self.column._from_objects[0] if self.column._from_objects else None
        )
        self._extra_entities = (self.expr, self.column)
        self._fetch_column = self._row_processor = None

    def corresponds_to(self, entity):
        return False

    def setup_dml_returning_compile_state(
        self,
        compile_state: ORMCompileState,
        adapter: DMLReturningColFilter,
    ) -> None:
        return self.setup_compile_state(compile_state)

    def setup_compile_state(self, compile_state):
        current_adapter = compile_state._get_current_adapter()
        if current_adapter:
            column = current_adapter(self.column, False)
            if column is None:
                return
        else:
            column = self.column

        if column._annotations:
            # annotated columns perform more slowly in compiler and
            # result due to the __eq__() method, so use deannotated
            column = column._deannotate()

        compile_state.dedupe_columns.add(column)
        compile_state.primary_columns.append(column)
        self._fetch_column = column


class _ORMColumnEntity(_ColumnEntity):
    """Column/expression based entity."""

    supports_single_entity = False

    __slots__ = (
        "expr",
        "mapper",
        "column",
        "_label_name",
        "entity_zero_or_selectable",
        "entity_zero",
        "_extra_entities",
    )

    def __init__(
        self,
        compile_state,
        column,
        entities_collection,
        parententity,
        raw_column_index,
        is_current_entities,
        parent_bundle=None,
    ):
        annotations = column._annotations

        _entity = parententity

        # an AliasedClass won't have proxy_key in the annotations for
        # a column if it was acquired using the class' adapter directly,
        # such as using AliasedInsp._adapt_element().  this occurs
        # within internal loaders.

        orm_key = annotations.get("proxy_key", None)
        proxy_owner = annotations.get("proxy_owner", _entity)
        if orm_key:
            self.expr = getattr(proxy_owner.entity, orm_key)
            self.translate_raw_column = False
        else:
            # if orm_key is not present, that means this is an ad-hoc
            # SQL ColumnElement, like a CASE() or other expression.
            # include this column position from the invoked statement
            # in the ORM-level ResultSetMetaData on each execute, so that
            # it can be targeted by identity after caching
            self.expr = column
            self.translate_raw_column = raw_column_index is not None

        self.raw_column_index = raw_column_index

        if is_current_entities:
            if parent_bundle:
                self._label_name = orm_key if orm_key else column._proxy_key
            else:
                self._label_name = compile_state._label_convention(
                    column, col_name=orm_key
                )
        else:
            self._label_name = None

        _entity._post_inspect
        self.entity_zero = self.entity_zero_or_selectable = ezero = _entity
        self.mapper = mapper = _entity.mapper

        if parent_bundle:
            parent_bundle._entities.append(self)
        else:
            entities_collection.append(self)

        compile_state._has_orm_entities = True

        self.column = column

        self._fetch_column = self._row_processor = None

        self._extra_entities = (self.expr, self.column)

        if mapper._should_select_with_poly_adapter:
            compile_state._create_with_polymorphic_adapter(
                ezero, ezero.selectable
            )

    def corresponds_to(self, entity):
        if _is_aliased_class(entity):
            # TODO: polymorphic subclasses ?
            return entity is self.entity_zero
        else:
            return not _is_aliased_class(
                self.entity_zero
            ) and entity.common_parent(self.entity_zero)

    def setup_dml_returning_compile_state(
        self,
        compile_state: ORMCompileState,
        adapter: DMLReturningColFilter,
    ) -> None:
        self._fetch_column = self.column
        column = adapter(self.column, False)
        if column is not None:
            compile_state.dedupe_columns.add(column)
            compile_state.primary_columns.append(column)

    def setup_compile_state(self, compile_state):
        current_adapter = compile_state._get_current_adapter()
        if current_adapter:
            column = current_adapter(self.column, False)
            if column is None:
                assert compile_state.is_dml_returning
                self._fetch_column = self.column
                return
        else:
            column = self.column

        ezero = self.entity_zero

        single_table_crit = self.mapper._single_table_criterion
        if (
            single_table_crit is not None
            or ("additional_entity_criteria", self.mapper)
            in compile_state.global_attributes
        ):
            compile_state.extra_criteria_entities[ezero] = (
                ezero,
                ezero._adapter if ezero.is_aliased_class else None,
            )

        if column._annotations and not column._expression_label:
            # annotated columns perform more slowly in compiler and
            # result due to the __eq__() method, so use deannotated
            column = column._deannotate()

        # use entity_zero as the from if we have it. this is necessary
        # for polymorphic scenarios where our FROM is based on ORM entity,
        # not the FROM of the column.  but also, don't use it if our column
        # doesn't actually have any FROMs that line up, such as when its
        # a scalar subquery.
        if set(self.column._from_objects).intersection(
            ezero.selectable._from_objects
        ):
            compile_state._fallback_from_clauses.append(ezero.selectable)

        compile_state.dedupe_columns.add(column)
        compile_state.primary_columns.append(column)
        self._fetch_column = column


class _IdentityTokenEntity(_ORMColumnEntity):
    translate_raw_column = False

    def setup_compile_state(self, compile_state):
        pass

    def row_processor(self, context, result):
        def getter(row):
            return context.load_options._identity_token

        return getter, self._label_name, self._extra_entities