text: import sqlalchemy as sa
  from sqlalchemy import and_
  from sqlalchemy import asc
  from sqlalchemy import desc
  from sqlalchemy import exc as sa_exc
  from sqlalchemy import exists
  from sqlalchemy import ForeignKey
  from sqlalchemy import func
  from sqlalchemy import Integer
  from sqlalchemy import literal_column
  from sqlalchemy import select
  from sqlalchemy import String
  from sqlalchemy import Table
  from sqlalchemy import testing
  from sqlalchemy import text
  from sqlalchemy import true
  from sqlalchemy import union
  from sqlalchemy import util
  from sqlalchemy.engine import default
  from sqlalchemy.orm import aliased
  from sqlalchemy.orm import backref
  from sqlalchemy.orm import clear_mappers
  from sqlalchemy.orm import column_property
  from sqlalchemy.orm import configure_mappers
  from sqlalchemy.orm import contains_eager
  from sqlalchemy.orm import create_session
  from sqlalchemy.orm import joinedload
  from sqlalchemy.orm import mapper
  from sqlalchemy.orm import relationship
  from sqlalchemy.orm import Session
  from sqlalchemy.orm.util import join
  from sqlalchemy.sql import column
  from sqlalchemy.sql import table
  from sqlalchemy.testing import assert_raises
  from sqlalchemy.testing import assert_raises_message
  from sqlalchemy.testing import AssertsCompiledSQL
  from sqlalchemy.testing import eq_
  from sqlalchemy.testing import fixtures
  from sqlalchemy.testing import is_
  from sqlalchemy.testing.schema import Column
  from test.orm import _fixtures
  
  
  class QueryTest(_fixtures.FixtureTest):
      run_setup_mappers = "once"
      run_inserts = "once"
      run_deletes = None
  
      @classmethod
      def setup_mappers(cls):
          (
              Node,
              composite_pk_table,
              users,
              Keyword,
              items,
              Dingaling,
              order_items,
              item_keywords,
              Item,
              User,
              dingalings,
              Address,
              keywords,
              CompositePk,
              nodes,
              Order,
              orders,
              addresses,
          ) = (
              cls.classes.Node,
              cls.tables.composite_pk_table,
              cls.tables.users,
              cls.classes.Keyword,
              cls.tables.items,
              cls.classes.Dingaling,
              cls.tables.order_items,
              cls.tables.item_keywords,
              cls.classes.Item,
              cls.classes.User,
              cls.tables.dingalings,
              cls.classes.Address,
              cls.tables.keywords,
              cls.classes.CompositePk,
              cls.tables.nodes,
              cls.classes.Order,
              cls.tables.orders,
              cls.tables.addresses,
          )
  
          mapper(
              User,
              users,
              properties={
                  "addresses": relationship(
                      Address, backref="user", order_by=addresses.c.id
                  ),
                  "orders": relationship(
                      Order, backref="user", order_by=orders.c.id
                  ),  # o2m, m2o
              },
          )
          mapper(
              Address,
              addresses,
              properties={
                  "dingaling": relationship(
                      Dingaling, uselist=False, backref="address"
                  )  # o2o
              },
          )
          mapper(Dingaling, dingalings)
          mapper(
              Order,
              orders,
              properties={
                  "items": relationship(
                      Item, secondary=order_items, order_by=items.c.id
                  ),  # m2m
                  "address": relationship(Address),  # m2o
              },
          )
          mapper(
              Item,
              items,
              properties={
                  "keywords": relationship(Keyword, secondary=item_keywords)
              },
          )  # m2m
          mapper(Keyword, keywords)
  
          mapper(
              Node,
              nodes,
              properties={
                  "children": relationship(
                      Node, backref=backref("parent", remote_side=[nodes.c.id])
                  )
              },
          )
  
          mapper(CompositePk, composite_pk_table)
  
          configure_mappers()
  
  
  class QueryCorrelatesLikeSelect(QueryTest, AssertsCompiledSQL):
      __dialect__ = "default"
  
      query_correlated = (
          "SELECT users.name AS users_name, "
          "(SELECT count(addresses.id) AS count_1 FROM addresses "
          "WHERE addresses.user_id = users.id) AS anon_1 FROM users"
      )
  
      query_not_correlated = (
          "SELECT users.name AS users_name, "
          "(SELECT count(addresses.id) AS count_1 FROM addresses, users "
          "WHERE addresses.user_id = users.id) AS anon_1 FROM users"
      )
  
      def test_scalar_subquery_select_auto_correlate(self):
          addresses, users = self.tables.addresses, self.tables.users
          query = (
              select(func.count(addresses.c.id))
              .where(addresses.c.user_id == users.c.id)
              .scalar_subquery()
          )
          query = select(users.c.name.label("users_name"), query)
          self.assert_compile(
              query, self.query_correlated, dialect=default.DefaultDialect()
          )
  
      def test_scalar_subquery_select_explicit_correlate(self):
          addresses, users = self.tables.addresses, self.tables.users
          query = (
              select(func.count(addresses.c.id))
              .where(addresses.c.user_id == users.c.id)
              .correlate(users)
              .scalar_subquery()
          )
          query = select(users.c.name.label("users_name"), query)
          self.assert_compile(
              query, self.query_correlated, dialect=default.DefaultDialect()
          )
  
      def test_scalar_subquery_select_correlate_off(self):
          addresses, users = self.tables.addresses, self.tables.users
          query = (
              select(func.count(addresses.c.id))
              .where(addresses.c.user_id == users.c.id)
              .correlate(None)
              .scalar_subquery()
          )
          query = select(users.c.name.label("users_name"), query)
          self.assert_compile(
              query, self.query_not_correlated, dialect=default.DefaultDialect()
          )
  
      def test_scalar_subquery_query_auto_correlate(self):
          sess = create_session()
          Address, User = self.classes.Address, self.classes.User
          query = (
              sess.query(func.count(Address.id))
              .filter(Address.user_id == User.id)
              .scalar_subquery()
          )
          query = sess.query(User.name, query)
          self.assert_compile(
              query, self.query_correlated, dialect=default.DefaultDialect()
          )
  
      def test_scalar_subquery_query_explicit_correlate(self):
          sess = create_session()
          Address, User = self.classes.Address, self.classes.User
          query = (
              sess.query(func.count(Address.id))
              .filter(Address.user_id == User.id)
              .correlate(self.tables.users)
              .scalar_subquery()
          )
          query = sess.query(User.name, query)
          self.assert_compile(
              query, self.query_correlated, dialect=default.DefaultDialect()
          )
  
      @testing.combinations(False, None)
      def test_scalar_subquery_query_correlate_off(self, value):
          sess = create_session()
          Address, User = self.classes.Address, self.classes.User
          query = (
              sess.query(func.count(Address.id))
              .filter(Address.user_id == User.id)
              .correlate(value)
              .scalar_subquery()
          )
          query = sess.query(User.name, query)
          self.assert_compile(
              query, self.query_not_correlated, dialect=default.DefaultDialect()
          )
  
      def test_correlate_to_union(self):
          User = self.classes.User
          sess = create_session()
  
          q = sess.query(User)
          q = sess.query(User).union(q)
          u_alias = aliased(User)
          raw_subq = exists().where(u_alias.id > User.id)
          orm_subq = sess.query(u_alias).filter(u_alias.id > User.id).exists()
  
          self.assert_compile(
              q.add_columns(raw_subq),
              "SELECT anon_1.users_id AS anon_1_users_id, "
              "anon_1.users_name AS anon_1_users_name, "
              "EXISTS (SELECT * FROM users AS users_1 "
              "WHERE users_1.id > anon_1.users_id) AS anon_2 "
              "FROM ("
              "SELECT users.id AS users_id, users.name AS users_name FROM users "
              "UNION SELECT users.id AS users_id, users.name AS users_name "
              "FROM users) AS anon_1",
          )
  
          # only difference is "1" vs. "*" (not sure why that is)
          self.assert_compile(
              q.add_columns(orm_subq),
              "SELECT anon_1.users_id AS anon_1_users_id, "
              "anon_1.users_name AS anon_1_users_name, "
              "EXISTS (SELECT 1 FROM users AS users_1 "
              "WHERE users_1.id > anon_1.users_id) AS anon_2 "
              "FROM ("
              "SELECT users.id AS users_id, users.name AS users_name FROM users "
              "UNION SELECT users.id AS users_id, users.name AS users_name "
              "FROM users) AS anon_1",
          )
  
      def test_correlate_to_union_newstyle(self):
          User = self.classes.User
  
          q = select(User).apply_labels()
  
          q = select(User).union(q).apply_labels().subquery()
  
          u_alias = aliased(User)
  
          raw_subq = exists().where(u_alias.id > q.c[0])
  
          self.assert_compile(
              select(q, raw_subq).apply_labels(),
              "SELECT anon_1.users_id AS anon_1_users_id, "
              "anon_1.users_name AS anon_1_users_name, "
              "EXISTS (SELECT * FROM users AS users_1 "
              "WHERE users_1.id > anon_1.users_id) AS anon_2 "
              "FROM ("
              "SELECT users.id AS users_id, users.name AS users_name FROM users "
              "UNION SELECT users.id AS users_id, users.name AS users_name "
              "FROM users) AS anon_1",
          )
  
  
  class RawSelectTest(QueryTest, AssertsCompiledSQL):
      """compare a bunch of select() tests with the equivalent Query using
      straight table/columns.
  
      Results should be the same as Query should act as a select() pass-
      thru for ClauseElement entities.
  
      """
  
      __dialect__ = "default"
  
      def test_select(self):
          addresses, users = self.tables.addresses, self.tables.users
  
          sess = create_session()
  
          self.assert_compile(
              sess.query(users)
              .select_entity_from(users.select().subquery())
              .with_labels()
              .statement,
              "SELECT users.id AS users_id, users.name AS users_name "
              "FROM users, "
              "(SELECT users.id AS id, users.name AS name FROM users) AS anon_1",
          )
  
          self.assert_compile(
              sess.query(users, exists([1], from_obj=addresses))
              .with_labels()
              .statement,
              "SELECT users.id AS users_id, users.name AS users_name, EXISTS "
              "(SELECT 1 FROM addresses) AS anon_1 FROM users",
          )
  
          # a little tedious here, adding labels to work around Query's
          # auto-labelling.
          s = (
              sess.query(
                  addresses.c.id.label("id"),
                  addresses.c.email_address.label("email"),
              )
              .filter(addresses.c.user_id == users.c.id)
              .correlate(users)
              .statement.alias()
          )
  
          self.assert_compile(
              sess.query(users, s.c.email)
              .select_entity_from(users.join(s, s.c.id == users.c.id))
              .with_labels()
              .statement,
              "SELECT users.id AS users_id, users.name AS users_name, "
              "anon_1.email AS anon_1_email "
              "FROM users JOIN (SELECT addresses.id AS id, "
              "addresses.email_address AS email FROM addresses, users "
              "WHERE addresses.user_id = users.id) AS anon_1 "
              "ON anon_1.id = users.id",
          )
  
          x = func.lala(users.c.id).label("foo")
          self.assert_compile(
              sess.query(x).filter(x == 5).statement,
              "SELECT lala(users.id) AS foo FROM users WHERE "
              "lala(users.id) = :param_1",
          )
  
          self.assert_compile(
              sess.query(func.sum(x).label("bar")).statement,
              "SELECT sum(lala(users.id)) AS bar FROM users",
          )
  
  
  class FromSelfTest(QueryTest, AssertsCompiledSQL):
      __dialect__ = "default"
  
      def test_filter(self):
          User = self.classes.User
  
          eq_(
              [User(id=8), User(id=9)],
              create_session()
              .query(User)
              .filter(User.id.in_([8, 9]))
              .from_self()
              .all(),
          )
  
          eq_(
              [User(id=8), User(id=9)],
              create_session()
              .query(User)
              .order_by(User.id)
              .slice(1, 3)
              .from_self()
              .all(),
          )
          eq_(
              [User(id=8)],
              list(
                  create_session()
                  .query(User)
                  .filter(User.id.in_([8, 9]))
                  .from_self()
                  .order_by(User.id)[0:1]
              ),
          )
  
      def test_join(self):
          User, Address = self.classes.User, self.classes.Address
  
          eq_(
              [
                  (User(id=8), Address(id=2)),
                  (User(id=8), Address(id=3)),
                  (User(id=8), Address(id=4)),
                  (User(id=9), Address(id=5)),
              ],
              create_session()
              .query(User)
              .filter(User.id.in_([8, 9]))
              .from_self()
              .join("addresses")
              .add_entity(Address)
              .order_by(User.id, Address.id)
              .all(),
          )
  
      def test_group_by(self):
          Address = self.classes.Address
  
          eq_(
              create_session()
              .query(Address.user_id, func.count(Address.id).label("count"))
              .group_by(Address.user_id)
              .order_by(Address.user_id)
              .all(),
              [(7, 1), (8, 3), (9, 1)],
          )
  
          eq_(
              create_session()
              .query(Address.user_id, Address.id)
              .from_self(Address.user_id, func.count(Address.id))
              .group_by(Address.user_id)
              .order_by(Address.user_id)
              .all(),
              [(7, 1), (8, 3), (9, 1)],
          )
  
      def test_having(self):
          User = self.classes.User
  
          s = create_session()
  
          self.assert_compile(
              s.query(User.id).group_by(User.id).having(User.id > 5).from_self(),
              "SELECT anon_1.users_id AS anon_1_users_id FROM "
              "(SELECT users.id AS users_id FROM users GROUP "
              "BY users.id HAVING users.id > :id_1) AS anon_1",
          )
  
      def test_no_joinedload(self):
          """test that joinedloads are pushed outwards and not rendered in
          subqueries."""
  
          User = self.classes.User
  
          s = create_session()
  
          self.assert_compile(
              s.query(User)
              .options(joinedload(User.addresses))
              .from_self()
              .statement,
              "SELECT anon_1.users_id, anon_1.users_name, addresses_1.id, "
              "addresses_1.user_id, addresses_1.email_address FROM "
              "(SELECT users.id AS users_id, users.name AS "
              "users_name FROM users) AS anon_1 LEFT OUTER JOIN "
              "addresses AS addresses_1 ON anon_1.users_id = "
              "addresses_1.user_id ORDER BY addresses_1.id",
          )
  
      def test_aliases(self):
          """test that aliased objects are accessible externally to a from_self()
          call."""
  
          User, Address = self.classes.User, self.classes.Address
  
          s = create_session()
  
          ualias = aliased(User)
          eq_(
              s.query(User, ualias)
              .filter(User.id > ualias.id)
              .from_self(User.name, ualias.name)
              .order_by(User.name, ualias.name)
              .all(),
              [
                  ("chuck", "ed"),
                  ("chuck", "fred"),
                  ("chuck", "jack"),
                  ("ed", "jack"),
                  ("fred", "ed"),
                  ("fred", "jack"),
              ],
          )
  
          eq_(
              s.query(User, ualias)
              .filter(User.id > ualias.id)
              .from_self(User.name, ualias.name)
              .filter(ualias.name == "ed")
              .order_by(User.name, ualias.name)
              .all(),
              [("chuck", "ed"), ("fred", "ed")],
          )
  
          eq_(
              s.query(User, ualias)
              .filter(User.id > ualias.id)
              .from_self(ualias.name, Address.email_address)
              .join(ualias.addresses)
              .order_by(ualias.name, Address.email_address)
              .all(),
              [
                  ("ed", "fred@fred.com"),
                  ("jack", "ed@bettyboop.com"),
                  ("jack", "ed@lala.com"),
                  ("jack", "ed@wood.com"),
                  ("jack", "fred@fred.com"),
              ],
          )
  
      def test_multiple_entities(self):
          User, Address = self.classes.User, self.classes.Address
  
          sess = create_session()
  
          eq_(
              sess.query(User, Address)
              .filter(User.id == Address.user_id)
              .filter(Address.id.in_([2, 5]))
              .from_self()
              .all(),
              [(User(id=8), Address(id=2)), (User(id=9), Address(id=5))],
          )
  
          eq_(
              sess.query(User, Address)
              .filter(User.id == Address.user_id)
              .filter(Address.id.in_([2, 5]))
              .from_self()
              .options(joinedload("addresses"))
              .first(),
              (
                  User(id=8, addresses=[Address(), Address(), Address()]),
                  Address(id=2),
              ),
          )
  
      def test_multiple_with_column_entities_oldstyle(self):
          # relies upon _orm_only_from_obj_alias setting
          User = self.classes.User
  
          sess = create_session()
  
          eq_(
              sess.query(User.id)
              .from_self()
              .add_columns(func.count().label("foo"))
              .group_by(User.id)
              .order_by(User.id)
              .from_self()
              .all(),
              [(7, 1), (8, 1), (9, 1), (10, 1)],
          )
  
      def test_multiple_with_column_entities_newstyle(self):
          User = self.classes.User
  
          sess = create_session()
  
          q1 = sess.query(User.id)
  
          subq1 = aliased(User, q1.subquery())
  
          q2 = sess.query(subq1.id).add_columns(func.count().label("foo"))
          q2 = q2.group_by(subq1.id).order_by(subq1.id).subquery()
  
          q3 = sess.query(q2)
          eq_(
              q3.all(), [(7, 1), (8, 1), (9, 1), (10, 1)],
          )
  
          q3 = select(q2)
          eq_(sess.execute(q3).fetchall(), [(7, 1), (8, 1), (9, 1), (10, 1)])
  
  
  class ColumnAccessTest(QueryTest, AssertsCompiledSQL):
      """test access of columns after _from_selectable has been applied"""
  
      __dialect__ = "default"
  
      def test_from_self(self):
          User = self.classes.User
          sess = create_session()
  
          q = sess.query(User).from_self()
          self.assert_compile(
              q.filter(User.name == "ed"),
              "SELECT anon_1.users_id AS anon_1_users_id, anon_1.users_name AS "
              "anon_1_users_name FROM (SELECT users.id AS users_id, users.name "
              "AS users_name FROM users) AS anon_1 WHERE anon_1.users_name = "
              ":name_1",
          )
  
      def test_from_self_twice(self):
          User = self.classes.User
          sess = create_session()
  
          q = sess.query(User).from_self(User.id, User.name).from_self()
          self.assert_compile(
              q.filter(User.name == "ed"),
              "SELECT anon_1.anon_2_users_id AS anon_1_anon_2_users_id, "
              "anon_1.anon_2_users_name AS anon_1_anon_2_users_name FROM "
              "(SELECT anon_2.users_id AS anon_2_users_id, anon_2.users_name "
              "AS anon_2_users_name FROM (SELECT users.id AS users_id, "
              "users.name AS users_name FROM users) AS anon_2) AS anon_1 "
              "WHERE anon_1.anon_2_users_name = :name_1",
          )
  
      def test_select_entity_from(self):
          User = self.classes.User
          sess = create_session()
  
          q = sess.query(User)
          q = sess.query(User).select_entity_from(q.statement.subquery())
          self.assert_compile(
              q.filter(User.name == "ed"),
              "SELECT anon_1.id AS anon_1_id, anon_1.name AS anon_1_name "
              "FROM (SELECT users.id AS id, users.name AS name FROM "
              "users) AS anon_1 WHERE anon_1.name = :name_1",
          )
  
      def test_select_entity_from_no_entities(self):
          User = self.classes.User
          sess = create_session()
  
          assert_raises_message(
              sa.exc.ArgumentError,
              r"A selectable \(FromClause\) instance is "
              "expected when the base alias is being set",
              sess.query(User).select_entity_from(User)._compile_context,
          )
  
      def test_select_from_no_aliasing(self):
          User = self.classes.User
          sess = create_session()
  
          q = sess.query(User)
          q = sess.query(User).select_from(q.statement.subquery())
          self.assert_compile(
              q.filter(User.name == "ed"),
              "SELECT users.id AS users_id, users.name AS users_name "
              "FROM users, (SELECT users.id AS id, users.name AS name FROM "
              "users) AS anon_1 WHERE users.name = :name_1",
          )
  
      def test_anonymous_expression_oldstyle(self):
          # relies upon _orm_only_from_obj_alias setting
          from sqlalchemy.sql import column
  
          sess = create_session()
          c1, c2 = column("c1"), column("c2")
          q1 = sess.query(c1, c2).filter(c1 == "dog")
          q2 = sess.query(c1, c2).filter(c1 == "cat")
          q3 = q1.union(q2)
          self.assert_compile(
              q3.order_by(c1),
              "SELECT anon_1.c1 AS anon_1_c1, anon_1.c2 "
              "AS anon_1_c2 FROM (SELECT c1, c2 WHERE "
              "c1 = :c1_1 UNION SELECT c1, c2 "
              "WHERE c1 = :c1_2) AS anon_1 ORDER BY anon_1.c1",
          )
  
      def test_anonymous_expression_newstyle(self):
          from sqlalchemy.sql import column
  
          c1, c2 = column("c1"), column("c2")
          q1 = select(c1, c2).where(c1 == "dog")
          q2 = select(c1, c2).where(c1 == "cat")
          subq = q1.union(q2).subquery()
          q3 = select(subq).apply_labels()
  
          self.assert_compile(
              q3.order_by(subq.c.c1),
              "SELECT anon_1.c1 AS anon_1_c1, anon_1.c2 "
              "AS anon_1_c2 FROM (SELECT c1, c2 WHERE "
              "c1 = :c1_1 UNION SELECT c1, c2 "
              "WHERE c1 = :c1_2) AS anon_1 ORDER BY anon_1.c1",
          )
  
      def test_table_anonymous_expression_from_self_twice_oldstyle(self):
          # relies upon _orm_only_from_obj_alias setting
          from sqlalchemy.sql import column
  
          sess = create_session()
          t1 = table("t1", column("c1"), column("c2"))
          q1 = sess.query(t1.c.c1, t1.c.c2).filter(t1.c.c1 == "dog")
          q1 = q1.from_self().from_self()
          self.assert_compile(
              q1.order_by(t1.c.c1),
              "SELECT anon_1.anon_2_t1_c1 "
              "AS anon_1_anon_2_t1_c1, anon_1.anon_2_t1_c2 "
              "AS anon_1_anon_2_t1_c2 "
              "FROM (SELECT anon_2.t1_c1 AS anon_2_t1_c1, "
              "anon_2.t1_c2 AS anon_2_t1_c2 FROM (SELECT t1.c1 AS t1_c1, t1.c2 "
              "AS t1_c2 FROM t1 WHERE t1.c1 = :c1_1) AS anon_2) AS anon_1 "
              "ORDER BY anon_1.anon_2_t1_c1",
          )
  
      def test_table_anonymous_expression_from_self_twice_newstyle(self):
          from sqlalchemy.sql import column
  
          t1 = table("t1", column("c1"), column("c2"))
          stmt = select(t1.c.c1, t1.c.c2).where(t1.c.c1 == "dog").apply_labels()
  
          subq1 = stmt.subquery("anon_2").select().apply_labels()
  
          subq2 = subq1.subquery("anon_1")
  
          q1 = select(subq2).apply_labels()
  
          self.assert_compile(
              # as in test_anonymous_expression_from_self_twice_newstyle_wlabels,
              # apply_labels() means the subquery cols have long names.  however,
              # here we illustrate if they did use apply_labels(), but they also
              # named the subqueries explicitly as one would certainly do if they
              # were using apply_labels(), we can get at that column based on how
              # it is aliased, no different than plain SQL.
              q1.order_by(subq2.c.anon_2_t1_c1),
              "SELECT anon_1.anon_2_t1_c1 "
              "AS anon_1_anon_2_t1_c1, anon_1.anon_2_t1_c2 "
              "AS anon_1_anon_2_t1_c2 "
              "FROM (SELECT anon_2.t1_c1 AS anon_2_t1_c1, "
              "anon_2.t1_c2 AS anon_2_t1_c2 FROM (SELECT t1.c1 AS t1_c1, t1.c2 "
              "AS t1_c2 FROM t1 WHERE t1.c1 = :c1_1) AS anon_2) AS anon_1 "
              "ORDER BY anon_1.anon_2_t1_c1",
          )
  
      def test_anonymous_expression_from_self_twice_oldstyle(self):
          # relies upon _orm_only_from_obj_alias setting
          from sqlalchemy.sql import column
  
          sess = create_session()
          c1, c2 = column("c1"), column("c2")
          q1 = sess.query(c1, c2).filter(c1 == "dog")
          q1 = q1.from_self().from_self()
          self.assert_compile(
              q1.order_by(c1),
              "SELECT anon_1.anon_2_c1 AS anon_1_anon_2_c1, anon_1.anon_2_c2 AS "
              "anon_1_anon_2_c2 FROM (SELECT anon_2.c1 AS anon_2_c1, anon_2.c2 "
              "AS anon_2_c2 "
              "FROM (SELECT c1, c2 WHERE c1 = :c1_1) AS "
              "anon_2) AS anon_1 ORDER BY anon_1.anon_2_c1",
          )
  
      def test_anonymous_expression_from_self_twice_newstyle_wlabels(self):
          from sqlalchemy.sql import column
  
          c1, c2 = column("c1"), column("c2")
          subq = select(c1, c2).where(c1 == "dog").subquery()
  
          subq2 = select(subq).apply_labels().subquery()
  
          stmt = select(subq2).apply_labels()
  
          self.assert_compile(
              # because of the apply labels we don't have simple keys on
              # subq2.c
              stmt.order_by(subq2.c.corresponding_column(c1)),
              "SELECT anon_1.anon_2_c1 AS anon_1_anon_2_c1, anon_1.anon_2_c2 AS "
              "anon_1_anon_2_c2 FROM (SELECT anon_2.c1 AS anon_2_c1, anon_2.c2 "
              "AS anon_2_c2 "
              "FROM (SELECT c1, c2 WHERE c1 = :c1_1) AS "
              "anon_2) AS anon_1 ORDER BY anon_1.anon_2_c1",
          )
  
      def test_anonymous_expression_from_self_twice_newstyle_wolabels(self):
          from sqlalchemy.sql import column
  
          c1, c2 = column("c1"), column("c2")
          subq = select(c1, c2).where(c1 == "dog").subquery()
  
          subq2 = select(subq).subquery()
  
          stmt = select(subq2)
  
          self.assert_compile(
              # without labels we can access .c1 but the statement will not
              # have the same labeling applied (which does not matter)
              stmt.order_by(subq2.c.c1),
              "SELECT anon_1.c1, anon_1.c2 FROM "
              "(SELECT anon_2.c1 AS c1, anon_2.c2 AS c2 "
              "FROM (SELECT c1, c2 WHERE c1 = :c1_1) AS "
              "anon_2) AS anon_1 ORDER BY anon_1.c1",
          )
  
      def test_anonymous_labeled_expression_oldstyle(self):
          # relies upon _orm_only_from_obj_alias setting
          sess = create_session()
          c1, c2 = column("c1"), column("c2")
          q1 = sess.query(c1.label("foo"), c2.label("bar")).filter(c1 == "dog")
          q2 = sess.query(c1.label("foo"), c2.label("bar")).filter(c1 == "cat")
          q3 = q1.union(q2)
          self.assert_compile(
              q3.order_by(c1),
              "SELECT anon_1.foo AS anon_1_foo, anon_1.bar AS anon_1_bar FROM "
              "(SELECT c1 AS foo, c2 AS bar WHERE c1 = :c1_1 UNION SELECT "
              "c1 AS foo, c2 AS bar "
              "WHERE c1 = :c1_2) AS anon_1 ORDER BY anon_1.foo",
          )
  
      def test_anonymous_labeled_expression_newstyle(self):
          c1, c2 = column("c1"), column("c2")
          q1 = select(c1.label("foo"), c2.label("bar")).where(c1 == "dog")
          q2 = select(c1.label("foo"), c2.label("bar")).where(c1 == "cat")
          subq = union(q1, q2).subquery()
          q3 = select(subq).apply_labels()
          self.assert_compile(
              q3.order_by(subq.c.foo),
              "SELECT anon_1.foo AS anon_1_foo, anon_1.bar AS anon_1_bar FROM "
              "(SELECT c1 AS foo, c2 AS bar WHERE c1 = :c1_1 UNION SELECT "
              "c1 AS foo, c2 AS bar "
              "WHERE c1 = :c1_2) AS anon_1 ORDER BY anon_1.foo",
          )
  
      def test_anonymous_expression_plus_flag_aliased_join(self):
          """test that the 'dont alias non-ORM' rule remains for other
          kinds of aliasing when _from_selectable() is used."""
  
          User = self.classes.User
          Address = self.classes.Address
          addresses = self.tables.addresses
  
          sess = create_session()
          q1 = sess.query(User.id).filter(User.id > 5)
          q1 = q1.from_self()
  
          q1 = q1.join(User.addresses, aliased=True).order_by(
              User.id, Address.id, addresses.c.id
          )
  
          self.assert_compile(
              q1,
              "SELECT anon_1.users_id AS anon_1_users_id "
              "FROM (SELECT users.id AS users_id FROM users "
              "WHERE users.id > :id_1) AS anon_1 JOIN addresses AS addresses_1 "
              "ON anon_1.users_id = addresses_1.user_id "
              "ORDER BY anon_1.users_id, addresses_1.id, addresses.id",
          )
  
      def test_anonymous_expression_plus_explicit_aliased_join(self):
          """test that the 'dont alias non-ORM' rule remains for other
          kinds of aliasing when _from_selectable() is used."""
  
          User = self.classes.User
          Address = self.classes.Address
          addresses = self.tables.addresses
  
          sess = create_session()
          q1 = sess.query(User.id).filter(User.id > 5)
          q1 = q1.from_self()
  
          aa = aliased(Address)
          q1 = q1.join(aa, User.addresses).order_by(
              User.id, aa.id, addresses.c.id
          )
          self.assert_compile(
              q1,
              "SELECT anon_1.users_id AS anon_1_users_id "
              "FROM (SELECT users.id AS users_id FROM users "
              "WHERE users.id > :id_1) AS anon_1 JOIN addresses AS addresses_1 "
              "ON anon_1.users_id = addresses_1.user_id "
              "ORDER BY anon_1.users_id, addresses_1.id, addresses.id",
          )
  
  
  class AddEntityEquivalenceTest(fixtures.MappedTest, AssertsCompiledSQL):
      run_setup_mappers = "once"
  
      @classmethod
      def define_tables(cls, metadata):
          Table(
              "a",
              metadata,
              Column(
                  "id", Integer, primary_key=True, test_needs_autoincrement=True
              ),
              Column("name", String(50)),
              Column("type", String(20)),
              Column("bid", Integer, ForeignKey("b.id")),
          )
  
          Table(
              "b",
              metadata,
              Column(
                  "id", Integer, primary_key=True, test_needs_autoincrement=True
              ),
              Column("name", String(50)),
              Column("type", String(20)),
          )
  
          Table(
              "c",
              metadata,
              Column("id", Integer, ForeignKey("b.id"), primary_key=True),
              Column("age", Integer),
          )
  
          Table(
              "d",
              metadata,
              Column("id", Integer, ForeignKey("a.id"), primary_key=True),
              Column("dede", Integer),
          )
  
      @classmethod
      def setup_classes(cls):
          a, c, b, d = (cls.tables.a, cls.tables.c, cls.tables.b, cls.tables.d)
  
          class A(cls.Comparable):
              pass
  
          class B(cls.Comparable):
              pass
  
          class C(B):
              pass
  
          class D(A):
              pass
  
          mapper(
              A,
              a,
              polymorphic_identity="a",
              polymorphic_on=a.c.type,
              with_polymorphic=("*", None),
              properties={
                  "link": relationship(B, uselist=False, backref="back")
              },
          )
          mapper(
              B,
              b,
              polymorphic_identity="b",
              polymorphic_on=b.c.type,
              with_polymorphic=("*", None),
          )
          mapper(C, c, inherits=B, polymorphic_identity="c")
          mapper(D, d, inherits=A, polymorphic_identity="d")
  
      @classmethod
      def insert_data(cls, connection):
          A, C, B = (cls.classes.A, cls.classes.C, cls.classes.B)
  
          sess = create_session(connection)
          sess.add_all(
              [
                  B(name="b1"),
                  A(name="a1", link=C(name="c1", age=3)),
                  C(name="c2", age=6),
                  A(name="a2"),
              ]
          )
          sess.flush()
  
      def test_add_entity_equivalence(self):
          A, C, B = (self.classes.A, self.classes.C, self.classes.B)
  
          sess = create_session()
  
          for q in [
              sess.query(A, B).join(A.link),
              sess.query(A).join(A.link).add_entity(B),
          ]:
              eq_(
                  q.all(),
                  [
                      (
                          A(bid=2, id=1, name="a1", type="a"),
                          C(age=3, id=2, name="c1", type="c"),
                      )
                  ],
              )
  
          for q in [
              sess.query(B, A).join(B.back),
              sess.query(B).join(B.back).add_entity(A),
              sess.query(B).add_entity(A).join(B.back),
          ]:
              eq_(
                  q.all(),
                  [
                      (
                          C(age=3, id=2, name="c1", type="c"),
                          A(bid=2, id=1, name="a1", type="a"),
                      )
                  ],
              )
  
  
  class InstancesTest(QueryTest, AssertsCompiledSQL):
      def test_from_alias_two_needs_nothing(self):
          User, addresses, users = (
              self.classes.User,
              self.tables.addresses,
              self.tables.users,
          )
  
          query = (
              users.select(users.c.id == 7)
              .union(users.select(users.c.id > 7))
              .alias("ulist")
              .outerjoin(addresses)
              .select(
                  use_labels=True, order_by=[text("ulist.id"), addresses.c.id]
              )
          )
          sess = create_session()
          q = sess.query(User)
  
          def go():
              result = (
                  q.options(contains_eager("addresses"))
                  .from_statement(query)
                  .all()
              )
              assert self.static.user_address_result == result
  
          self.assert_sql_count(testing.db, go, 1)
  
      def test_from_alias_two(self):
          User, addresses, users = (
              self.classes.User,
              self.tables.addresses,
              self.tables.users,
          )
  
          query = (
              users.select(users.c.id == 7)
              .union(users.select(users.c.id > 7))
              .alias("ulist")
              .outerjoin(addresses)
              .select(
                  use_labels=True, order_by=[text("ulist.id"), addresses.c.id]
              )
          )
          sess = create_session()
          q = sess.query(User)
  
          def go():
              ulist_alias = aliased(User, alias=query.alias("ulist"))
              result = (
                  q.options(contains_eager("addresses"))
                  .select_entity_from(ulist_alias)
                  .all()
              )
              assert self.static.user_address_result == result
  
          self.assert_sql_count(testing.db, go, 1)
  
      def test_from_alias_three(self):
          User, addresses, users = (
              self.classes.User,
              self.tables.addresses,
              self.tables.users,
          )
  
          query = (
              users.select(users.c.id == 7)
              .union(users.select(users.c.id > 7))
              .alias("ulist")
              .outerjoin(addresses)
              .select(
                  use_labels=True, order_by=[text("ulist.id"), addresses.c.id]
              )
          )
          sess = create_session()
  
          # better way.  use select_entity_from()
          def go():
              result = (
                  sess.query(User)
                  .select_entity_from(query.subquery())
                  .options(contains_eager("addresses"))
                  .all()
              )
              assert self.static.user_address_result == result
  
          self.assert_sql_count(testing.db, go, 1)
  
      def test_from_alias_four(self):
          User, addresses, users = (
              self.classes.User,
              self.tables.addresses,
              self.tables.users,
          )
  
          sess = create_session()
  
          # same thing, but alias addresses, so that the adapter
          # generated by select_entity_from() is wrapped within
          # the adapter created by contains_eager()
          adalias = addresses.alias()
          query = (
              users.select(users.c.id == 7)
              .union(users.select(users.c.id > 7))
              .alias("ulist")
              .outerjoin(adalias)
              .select(use_labels=True, order_by=[text("ulist.id"), adalias.c.id])
          )
  
          def go():
              result = (
                  sess.query(User)
                  .select_entity_from(query.subquery())
                  .options(contains_eager("addresses", alias=adalias))
                  .all()
              )
              assert self.static.user_address_result == result
  
          self.assert_sql_count(testing.db, go, 1)
  
      def test_contains_eager_one(self):
          addresses, User = (self.tables.addresses, self.classes.User)
  
          sess = create_session()
  
          # test that contains_eager suppresses the normal outer join rendering
          q = (
              sess.query(User)
              .outerjoin(User.addresses)
              .options(contains_eager(User.addresses))
              .order_by(User.id, addresses.c.id)
          )
          self.assert_compile(
              q.with_labels().statement,
              "SELECT addresses.id AS addresses_id, "
              "addresses.user_id AS addresses_user_id, "
              "addresses.email_address AS "
              "addresses_email_address, users.id AS "
              "users_id, users.name AS users_name FROM "
              "users LEFT OUTER JOIN addresses ON "
              "users.id = addresses.user_id ORDER BY "
              "users.id, addresses.id",
              dialect=default.DefaultDialect(),
          )
  
          def go():
              assert self.static.user_address_result == q.all()
  
          self.assert_sql_count(testing.db, go, 1)
  
      def test_contains_eager_two(self):
          users, addresses, User = (
              self.tables.users,
              self.tables.addresses,
              self.classes.User,
          )
  
          sess = create_session()
  
          adalias = addresses.alias()
          q = (
              sess.query(User)
              .select_entity_from(users.outerjoin(adalias))
              .options(contains_eager(User.addresses, alias=adalias))
              .order_by(User.id, adalias.c.id)
          )
  
          def go():
              eq_(self.static.user_address_result, q.all())
  
          self.assert_sql_count(testing.db, go, 1)
  
      def test_contains_eager_four(self):
          users, addresses, User = (
              self.tables.users,
              self.tables.addresses,
              self.classes.User,
          )
  
          sess = create_session()
  
          selectquery = users.outerjoin(addresses).select(
              users.c.id < 10,
              use_labels=True,
              order_by=[users.c.id, addresses.c.id],
          )
  
          q = sess.query(User)
  
          def go():
              result = (
                  q.options(contains_eager("addresses"))
                  .from_statement(selectquery)
                  .all()
              )
              assert self.static.user_address_result[0:3] == result
  
          self.assert_sql_count(testing.db, go, 1)
  
      def test_contains_eager_four_future(self):
          users, addresses, User = (
              self.tables.users,
              self.tables.addresses,
              self.classes.User,
          )
  
          sess = create_session(testing.db, future=True)
  
          selectquery = users.outerjoin(addresses).select(
              users.c.id < 10,
              use_labels=True,
              order_by=[users.c.id, addresses.c.id],
          )
  
          q = select(User)
  
          def go():
              result = (
                  sess.execute(
                      q.options(contains_eager("addresses")).from_statement(
                          selectquery
                      )
                  )
                  .scalars()
                  .unique()
                  .all()
              )
              assert self.static.user_address_result[0:3] == result
  
          self.assert_sql_count(testing.db, go, 1)
  
      def test_contains_eager_aliased(self):
          User, Address = self.classes.User, self.classes.Address
  
          sess = create_session()
          q = sess.query(User)
  
          # Aliased object
          adalias = aliased(Address)
  
          def go():
              result = (
                  q.options(contains_eager("addresses", alias=adalias))
                  .outerjoin(adalias, User.addresses)
                  .order_by(User.id, adalias.id)
              )
              assert self.static.user_address_result == result.all()
  
          self.assert_sql_count(testing.db, go, 1)
  
      def test_contains_eager_multi_alias(self):
          orders, items, users, order_items, User = (
              self.tables.orders,
              self.tables.items,
              self.tables.users,
              self.tables.order_items,
              self.classes.User,
          )
  
          sess = create_session()
          q = sess.query(User)
  
          oalias = orders.alias("o1")
          ialias = items.alias("i1")
          query = (
              users.outerjoin(oalias)
              .outerjoin(order_items)
              .outerjoin(ialias)
              .select(use_labels=True)
              .order_by(users.c.id, oalias.c.id, ialias.c.id)
          )
  
          # test using Alias with more than one level deep
  
          # new way:
          # from sqlalchemy.orm.strategy_options import Load
          # opt = Load(User).contains_eager('orders', alias=oalias).
          #     contains_eager('items', alias=ialias)
  
          def go():
              result = list(
                  q.options(
                      contains_eager("orders", alias=oalias),
                      contains_eager("orders.items", alias=ialias),
                  ).from_statement(query)
              )
              assert self.static.user_order_result == result
  
          self.assert_sql_count(testing.db, go, 1)
  
      def test_contains_eager_multi_aliased(self):
          Item, User, Order = (
              self.classes.Item,
              self.classes.User,
              self.classes.Order,
          )
  
          sess = create_session()
          q = sess.query(User)
  
          # test using Aliased with more than one level deep
          oalias = aliased(Order)
          ialias = aliased(Item)
  
          def go():
              result = (
                  q.options(
                      contains_eager(User.orders, alias=oalias),
                      contains_eager(User.orders, Order.items, alias=ialias),
                  )
                  .outerjoin(oalias, User.orders)
                  .outerjoin(ialias, oalias.items)
                  .order_by(User.id, oalias.id, ialias.id)
              )
              assert self.static.user_order_result == result.all()
  
          self.assert_sql_count(testing.db, go, 1)
  
      def test_contains_eager_multi_aliased_of_type(self):
          # test newer style that does not use the alias parameter
          Item, User, Order = (
              self.classes.Item,
              self.classes.User,
              self.classes.Order,
          )
  
          sess = create_session()
          q = sess.query(User)
  
          # test using Aliased with more than one level deep
          oalias = aliased(Order)
          ialias = aliased(Item)
  
          def go():
              result = (
                  q.options(
                      contains_eager(User.orders.of_type(oalias)).contains_eager(
                          oalias.items.of_type(ialias)
                      )
                  )
                  .outerjoin(User.orders.of_type(oalias))
                  .outerjoin(oalias.items.of_type(ialias))
                  .order_by(User.id, oalias.id, ialias.id)
              )
              assert self.static.user_order_result == result.all()
  
          self.assert_sql_count(testing.db, go, 1)
  
      def test_contains_eager_chaining(self):
          """test that contains_eager() 'chains' by default."""
  
          Dingaling, User, Address = (
              self.classes.Dingaling,
              self.classes.User,
              self.classes.Address,
          )
  
          sess = create_session()
          q = (
              sess.query(User)
              .join(User.addresses)
              .join(Address.dingaling)
              .options(contains_eager(User.addresses, Address.dingaling))
          )
  
          def go():
              eq_(
                  q.all(),
                  # note we only load the Address records that
                  # have a Dingaling here due to using the inner
                  # join for the eager load
                  [
                      User(
                          name="ed",
                          addresses=[
                              Address(
                                  email_address="ed@wood.com",
                                  dingaling=Dingaling(data="ding 1/2"),
                              )
                          ],
                      ),
                      User(
                          name="fred",
                          addresses=[
                              Address(
                                  email_address="fred@fred.com",
                                  dingaling=Dingaling(data="ding 2/5"),
                              )
                          ],
                      ),
                  ],
              )
  
          self.assert_sql_count(testing.db, go, 1)
  
      def test_contains_eager_chaining_aliased_endpoint(self):
          """test that contains_eager() 'chains' by default and supports
          an alias at the end."""
  
          Dingaling, User, Address = (
              self.classes.Dingaling,
              self.classes.User,
              self.classes.Address,
          )
  
          sess = create_session()
          da = aliased(Dingaling, name="foob")
          q = (
              sess.query(User)
              .join(User.addresses)
              .join(da, Address.dingaling)
              .options(
                  contains_eager(User.addresses, Address.dingaling, alias=da)
              )
          )
  
          def go():
              eq_(
                  q.all(),
                  # note we only load the Address records that
                  # have a Dingaling here due to using the inner
                  # join for the eager load
                  [
                      User(
                          name="ed",
                          addresses=[
                              Address(
                                  email_address="ed@wood.com",
                                  dingaling=Dingaling(data="ding 1/2"),
                              )
                          ],
                      ),
                      User(
                          name="fred",
                          addresses=[
                              Address(
                                  email_address="fred@fred.com",
                                  dingaling=Dingaling(data="ding 2/5"),
                              )
                          ],
                      ),
                  ],
              )
  
          self.assert_sql_count(testing.db, go, 1)
  
      def test_mixed_eager_contains_with_limit(self):
          Order, User, Address = (
              self.classes.Order,
              self.classes.User,
              self.classes.Address,
          )
  
          sess = create_session()
  
          q = sess.query(User)
  
          def go():
              # outerjoin to User.orders, offset 1/limit 2 so we get user
              # 7 + second two orders. then joinedload the addresses.
              # User + Order columns go into the subquery, address left
              # outer joins to the subquery, joinedloader for User.orders
              # applies context.adapter to result rows.  This was
              # [ticket:1180].
  
              result = (
                  q.outerjoin(User.orders)
                  .options(
                      joinedload(User.addresses), contains_eager(User.orders)
                  )
                  .order_by(User.id, Order.id)
                  .offset(1)
                  .limit(2)
                  .all()
              )
              eq_(
                  result,
                  [
                      User(
                          id=7,
                          addresses=[
                              Address(
                                  email_address="jack@bean.com", user_id=7, id=1
                              )
                          ],
                          name="jack",
                          orders=[
                              Order(
                                  address_id=1,
                                  user_id=7,
                                  description="order 3",
                                  isopen=1,
                                  id=3,
                              ),
                              Order(
                                  address_id=None,
                                  user_id=7,
                                  description="order 5",
                                  isopen=0,
                                  id=5,
                              ),
                          ],
                      )
                  ],
              )
  
          self.assert_sql_count(testing.db, go, 1)
          sess.expunge_all()
  
          def go():
  
              # same as above, except Order is aliased, so two adapters
              # are applied by the eager loader
  
              oalias = aliased(Order)
              result = (
                  q.outerjoin(oalias, User.orders)
                  .options(
                      joinedload(User.addresses),
                      contains_eager(User.orders, alias=oalias),
                  )
                  .order_by(User.id, oalias.id)
                  .offset(1)
                  .limit(2)
                  .all()
              )
              eq_(
                  result,
                  [
                      User(
                          id=7,
                          addresses=[
                              Address(
                                  email_address="jack@bean.com", user_id=7, id=1
                              )
                          ],
                          name="jack",
                          orders=[
                              Order(
                                  address_id=1,
                                  user_id=7,
                                  description="order 3",
                                  isopen=1,
                                  id=3,
                              ),
                              Order(
                                  address_id=None,
                                  user_id=7,
                                  description="order 5",
                                  isopen=0,
                                  id=5,
                              ),
                          ],
                      )
                  ],
              )
  
          self.assert_sql_count(testing.db, go, 1)
  
  
  class MixedEntitiesTest(QueryTest, AssertsCompiledSQL):
      __dialect__ = "default"
  
      def test_alias_naming(self):
          User = self.classes.User
  
          sess = create_session()
  
          ua = aliased(User, name="foobar")
          q = sess.query(ua)
          self.assert_compile(
              q,
              "SELECT foobar.id AS foobar_id, "
              "foobar.name AS foobar_name FROM users AS foobar",
          )
  
      def test_correlated_subquery(self):
          """test that a subquery constructed from ORM attributes doesn't leak
          out those entities to the outermost query."""
  
          Address, users, User = (
              self.classes.Address,
              self.tables.users,
              self.classes.User,
          )
  
          sess = create_session()
  
          subq = (
              select(func.count())
              .where(User.id == Address.user_id)
              .correlate(users)
              .label("count")
          )
  
          # we don't want Address to be outside of the subquery here
          eq_(
              list(sess.query(User, subq)[0:3]),
              [
                  (User(id=7, name="jack"), 1),
                  (User(id=8, name="ed"), 3),
                  (User(id=9, name="fred"), 1),
              ],
          )
  
          # same thing without the correlate, as it should
          # not be needed
          subq = (
              select(func.count())
              .where(User.id == Address.user_id)
              .label("count")
          )
  
          # we don't want Address to be outside of the subquery here
          eq_(
              list(sess.query(User, subq)[0:3]),
              [
                  (User(id=7, name="jack"), 1),
                  (User(id=8, name="ed"), 3),
                  (User(id=9, name="fred"), 1),
              ],
          )
  
      def test_column_queries_one(self):
          User = self.classes.User
  
          sess = create_session()
  
          eq_(
              sess.query(User.name).all(),
              [("jack",), ("ed",), ("fred",), ("chuck",)],
          )
  
      def test_column_queries_two(self):
          users, User = (
              self.tables.users,
              self.classes.User,
          )
  
          sess = create_session()
          sel = users.select(User.id.in_([7, 8])).alias()
          q = sess.query(User.name)
          q2 = q.select_entity_from(sel).all()
          eq_(list(q2), [("jack",), ("ed",)])
  
      def test_column_queries_three(self):
          Address, User = (
              self.classes.Address,
              self.classes.User,
          )
  
          sess = create_session()
          eq_(
              sess.query(User.name, Address.email_address)
              .filter(User.id == Address.user_id)
              .all(),
              [
                  ("jack", "jack@bean.com"),
                  ("ed", "ed@wood.com"),
                  ("ed", "ed@bettyboop.com"),
                  ("ed", "ed@lala.com"),
                  ("fred", "fred@fred.com"),
              ],
          )
  
      def test_column_queries_four(self):
          Address, User = (
              self.classes.Address,
              self.classes.User,
          )
  
          sess = create_session()
          eq_(
              sess.query(User.name, func.count(Address.email_address))
              .outerjoin(User.addresses)
              .group_by(User.id, User.name)
              .order_by(User.id)
              .all(),
              [("jack", 1), ("ed", 3), ("fred", 1), ("chuck", 0)],
          )
  
      def test_column_queries_five(self):
          Address, User = (
              self.classes.Address,
              self.classes.User,
          )
  
          sess = create_session()
          eq_(
              sess.query(User, func.count(Address.email_address))
              .outerjoin(User.addresses)
              .group_by(User)
              .order_by(User.id)
              .all(),
              [
                  (User(name="jack", id=7), 1),
                  (User(name="ed", id=8), 3),
                  (User(name="fred", id=9), 1),
                  (User(name="chuck", id=10), 0),
              ],
          )
  
      def test_column_queries_six(self):
          Address, User = (
              self.classes.Address,
              self.classes.User,
          )
  
          sess = create_session()
          eq_(
              sess.query(func.count(Address.email_address), User)
              .outerjoin(User.addresses)
              .group_by(User)
              .order_by(User.id)
              .all(),
              [
                  (1, User(name="jack", id=7)),
                  (3, User(name="ed", id=8)),
                  (1, User(name="fred", id=9)),
                  (0, User(name="chuck", id=10)),
              ],
          )
  
      def test_column_queries_seven(self):
          Address, User = (
              self.classes.Address,
              self.classes.User,
          )
  
          sess = create_session()
          adalias = aliased(Address)
          eq_(
              sess.query(User, func.count(adalias.email_address))
              .outerjoin(adalias, "addresses")
              .group_by(User)
              .order_by(User.id)
              .all(),
              [
                  (User(name="jack", id=7), 1),
                  (User(name="ed", id=8), 3),
                  (User(name="fred", id=9), 1),
                  (User(name="chuck", id=10), 0),
              ],
          )
  
      def test_column_queries_eight(self):
          Address, User = (
              self.classes.Address,
              self.classes.User,
          )
  
          sess = create_session()
          adalias = aliased(Address)
          eq_(
              sess.query(func.count(adalias.email_address), User)
              .outerjoin(adalias, User.addresses)
              .group_by(User)
              .order_by(User.id)
              .all(),
              [
                  (1, User(name="jack", id=7)),
                  (3, User(name="ed", id=8)),
                  (1, User(name="fred", id=9)),
                  (0, User(name="chuck", id=10)),
              ],
          )
  
      def test_column_queries_nine(self):
          Address, User = (
              self.classes.Address,
              self.classes.User,
          )
  
          sess = create_session()
  
          adalias = aliased(Address)
          # select from aliasing + explicit aliasing
          eq_(
              sess.query(User, adalias.email_address, adalias.id)
              .outerjoin(adalias, User.addresses)
              .from_self(User, adalias.email_address)
              .order_by(User.id, adalias.id)
              .all(),
              [
                  (User(name="jack", id=7), "jack@bean.com"),
                  (User(name="ed", id=8), "ed@wood.com"),
                  (User(name="ed", id=8), "ed@bettyboop.com"),
                  (User(name="ed", id=8), "ed@lala.com"),
                  (User(name="fred", id=9), "fred@fred.com"),
                  (User(name="chuck", id=10), None),
              ],
          )
  
      def test_column_queries_ten(self):
          Address, User = (
              self.classes.Address,
              self.classes.User,
          )
  
          sess = create_session()
  
          # anon + select from aliasing
          aa = aliased(Address)
          eq_(
              sess.query(User)
              .join(aa, User.addresses)
              .filter(aa.email_address.like("%ed%"))
              .from_self()
              .all(),
              [User(name="ed", id=8), User(name="fred", id=9)],
          )
  
      def test_column_queries_eleven(self):
          Address, User = (
              self.classes.Address,
              self.classes.User,
          )
  
          sess = create_session()
  
          adalias = aliased(Address)
          # test eager aliasing, with/without select_entity_from aliasing
          for q in [
              sess.query(User, adalias.email_address)
              .outerjoin(adalias, User.addresses)
              .options(joinedload(User.addresses))
              .order_by(User.id, adalias.id)
              .limit(10),
              sess.query(User, adalias.email_address, adalias.id)
              .outerjoin(adalias, User.addresses)
              .from_self(User, adalias.email_address)
              .options(joinedload(User.addresses))
              .order_by(User.id, adalias.id)
              .limit(10),
          ]:
              eq_(
                  q.all(),
                  [
                      (
                          User(
                              addresses=[
                                  Address(
                                      user_id=7,
                                      email_address="jack@bean.com",
                                      id=1,
                                  )
                              ],
                              name="jack",
                              id=7,
                          ),
                          "jack@bean.com",
                      ),
                      (
                          User(
                              addresses=[
                                  Address(
                                      user_id=8,
                                      email_address="ed@wood.com",
                                      id=2,
                                  ),
                                  Address(
                                      user_id=8,
                                      email_address="ed@bettyboop.com",
                                      id=3,
                                  ),
                                  Address(
                                      user_id=8,
                                      email_address="ed@lala.com",
                                      id=4,
                                  ),
                              ],
                              name="ed",
                              id=8,
                          ),
                          "ed@wood.com",
                      ),
                      (
                          User(
                              addresses=[
                                  Address(
                                      user_id=8,
                                      email_address="ed@wood.com",
                                      id=2,
                                  ),
                                  Address(
                                      user_id=8,
                                      email_address="ed@bettyboop.com",
                                      id=3,
                                  ),
                                  Address(
                                      user_id=8,
                                      email_address="ed@lala.com",
                                      id=4,
                                  ),
                              ],
                              name="ed",
                              id=8,
                          ),
                          "ed@bettyboop.com",
                      ),
                      (
                          User(
                              addresses=[
                                  Address(
                                      user_id=8,
                                      email_address="ed@wood.com",
                                      id=2,
                                  ),
                                  Address(
                                      user_id=8,
                                      email_address="ed@bettyboop.com",
                                      id=3,
                                  ),
                                  Address(
                                      user_id=8,
                                      email_address="ed@lala.com",
                                      id=4,
                                  ),
                              ],
                              name="ed",
                              id=8,
                          ),
                          "ed@lala.com",
                      ),
                      (
                          User(
                              addresses=[
                                  Address(
                                      user_id=9,
                                      email_address="fred@fred.com",
                                      id=5,
                                  )
                              ],
                              name="fred",
                              id=9,
                          ),
                          "fred@fred.com",
                      ),
                      (User(addresses=[], name="chuck", id=10), None),
                  ],
              )
  
      def test_column_from_limited_joinedload(self):
          User = self.classes.User
  
          sess = create_session()
  
          def go():
              results = (
                  sess.query(User)
                  .limit(1)
                  .options(joinedload("addresses"))
                  .add_columns(User.name)
                  .all()
              )
              eq_(results, [(User(name="jack"), "jack")])
  
          self.assert_sql_count(testing.db, go, 1)
  
      @testing.fails_on("firebird", "unknown")
      def test_self_referential(self):
          Order = self.classes.Order
  
          sess = create_session()
          oalias = aliased(Order)
  
          for q in [
              sess.query(Order, oalias)
              .filter(Order.user_id == oalias.user_id)
              .filter(Order.user_id == 7)
              .filter(Order.id > oalias.id)
              .order_by(Order.id, oalias.id),
              sess.query(Order, oalias)
              .filter(Order.id > oalias.id)
              .from_self()
              .filter(Order.user_id == oalias.user_id)
              .filter(Order.user_id == 7)
              .order_by(Order.id, oalias.id),
              # same thing, but reversed.
              sess.query(oalias, Order)
              .filter(Order.id < oalias.id)
              .from_self()
              .filter(oalias.user_id == Order.user_id)
              .filter(oalias.user_id == 7)
              .order_by(oalias.id, Order.id),
              # here we go....two layers of aliasing
              sess.query(Order, oalias)
              .filter(Order.user_id == oalias.user_id)
              .filter(Order.user_id == 7)
              .filter(Order.id > oalias.id)
              .from_self()
              .order_by(Order.id, oalias.id)
              .limit(10)
              .options(joinedload(Order.items)),
              # gratuitous four layers
              sess.query(Order, oalias)
              .filter(Order.user_id == oalias.user_id)
              .filter(Order.user_id == 7)
              .filter(Order.id > oalias.id)
              .from_self()
              .from_self()
              .from_self()
              .order_by(Order.id, oalias.id)
              .limit(10)
              .options(joinedload(Order.items)),
          ]:
  
              eq_(
                  q.all(),
                  [
                      (
                          Order(
                              address_id=1,
                              description="order 3",
                              isopen=1,
                              user_id=7,
                              id=3,
                          ),
                          Order(
                              address_id=1,
                              description="order 1",
                              isopen=0,
                              user_id=7,
                              id=1,
                          ),
                      ),
                      (
                          Order(
                              address_id=None,
                              description="order 5",
                              isopen=0,
                              user_id=7,
                              id=5,
                          ),
                          Order(
                              address_id=1,
                              description="order 1",
                              isopen=0,
                              user_id=7,
                              id=1,
                          ),
                      ),
                      (
                          Order(
                              address_id=None,
                              description="order 5",
                              isopen=0,
                              user_id=7,
                              id=5,
                          ),
                          Order(
                              address_id=1,
                              description="order 3",
                              isopen=1,
                              user_id=7,
                              id=3,
                          ),
                      ),
                  ],
              )
  
      def test_from_self_internal_literals_oldstyle(self):
          # relies upon _orm_only_from_obj_alias setting
          Order = self.classes.Order
  
          sess = create_session()
  
          # ensure column expressions are taken from inside the subquery, not
          # restated at the top
          q = (
              sess.query(
                  Order.id, Order.description, literal_column("'q'").label("foo")
              )
              .filter(Order.description == "order 3")
              .from_self()
          )
          self.assert_compile(
              q,
              "SELECT anon_1.orders_id AS "
              "anon_1_orders_id, "
              "anon_1.orders_description AS anon_1_orders_description, "
              "anon_1.foo AS anon_1_foo FROM (SELECT "
              "orders.id AS orders_id, "
              "orders.description AS orders_description, "
              "'q' AS foo FROM orders WHERE "
              "orders.description = :description_1) AS "
              "anon_1",
          )
          eq_(q.all(), [(3, "order 3", "q")])
  
      def test_from_self_internal_literals_newstyle(self):
          Order = self.classes.Order
  
          stmt = select(
              Order.id, Order.description, literal_column("'q'").label("foo")
          ).where(Order.description == "order 3")
  
          subq = aliased(Order, stmt.apply_labels().subquery())
  
          stmt = select(subq).apply_labels()
          self.assert_compile(
              stmt,
              "SELECT anon_1.orders_id AS "
              "anon_1_orders_id, "
              "anon_1.orders_description AS anon_1_orders_description "
              "FROM (SELECT "
              "orders.id AS orders_id, "
              "orders.description AS orders_description, "
              "'q' AS foo FROM orders WHERE "
              "orders.description = :description_1) AS "
              "anon_1",
          )
  
      def test_multi_mappers(self):
          Address, addresses, users, User = (
              self.classes.Address,
              self.tables.addresses,
              self.tables.users,
              self.classes.User,
          )
  
          test_session = create_session()
  
          (user7, user8, user9, user10) = test_session.query(User).all()
          (
              address1,
              address2,
              address3,
              address4,
              address5,
          ) = test_session.query(Address).all()
  
          expected = [
              (user7, address1),
              (user8, address2),
              (user8, address3),
              (user8, address4),
              (user9, address5),
              (user10, None),
          ]
  
          sess = create_session(testing.db, future=True)
  
          selectquery = users.outerjoin(addresses).select(
              use_labels=True, order_by=[users.c.id, addresses.c.id]
          )
  
          result = sess.execute(
              select(User, Address).from_statement(selectquery)
          )
          eq_(
              list(result), expected,
          )
          sess.expunge_all()
  
          for address_entity in (Address, aliased(Address)):
              q = (
                  sess.query(User)
                  .add_entity(address_entity)
                  .outerjoin(address_entity, "addresses")
                  .order_by(User.id, address_entity.id)
              )
              eq_(q.all(), expected)
              sess.expunge_all()
  
              q = sess.query(User).add_entity(address_entity)
              q = q.join(address_entity, "addresses")
              q = q.filter_by(email_address="ed@bettyboop.com")
              eq_(q.all(), [(user8, address3)])
              sess.expunge_all()
  
              q = (
                  sess.query(User, address_entity)
                  .join(address_entity, "addresses")
                  .filter_by(email_address="ed@bettyboop.com")
              )
              eq_(q.all(), [(user8, address3)])
              sess.expunge_all()
  
              q = (
                  sess.query(User, address_entity)
                  .join(address_entity, "addresses")
                  .options(joinedload("addresses"))
                  .filter_by(email_address="ed@bettyboop.com")
              )
              eq_(list(util.OrderedSet(q.all())), [(user8, address3)])
              sess.expunge_all()
  
      def test_aliased_multi_mappers(self):
          User, addresses, users, Address = (
              self.classes.User,
              self.tables.addresses,
              self.tables.users,
              self.classes.Address,
          )
  
          sess = create_session()
  
          (user7, user8, user9, user10) = sess.query(User).all()
          (address1, address2, address3, address4, address5) = sess.query(
              Address
          ).all()
  
          expected = [
              (user7, address1),
              (user8, address2),
              (user8, address3),
              (user8, address4),
              (user9, address5),
              (user10, None),
          ]
  
          q = sess.query(User)
          adalias = addresses.alias("adalias")
          q = q.add_entity(Address, alias=adalias).select_entity_from(
              users.outerjoin(adalias)
          )
          result = q.order_by(User.id, adalias.c.id).all()
          assert result == expected
  
          sess.expunge_all()
  
          q = sess.query(User).add_entity(Address, alias=adalias)
          result = (
              q.select_entity_from(users.outerjoin(adalias))
              .filter(adalias.c.email_address == "ed@bettyboop.com")
              .all()
          )
          assert result == [(user8, address3)]
  
      def test_with_entities(self):
          User, Address = self.classes.User, self.classes.Address
  
          sess = create_session()
  
          q = sess.query(User).filter(User.id == 7).order_by(User.name)
  
          self.assert_compile(
              q.with_entities(User.id, Address).filter(
                  Address.user_id == User.id
              ),
              "SELECT users.id AS users_id, addresses.id "
              "AS addresses_id, addresses.user_id AS "
              "addresses_user_id, addresses.email_address"
              " AS addresses_email_address FROM users, "
              "addresses WHERE users.id = :id_1 AND "
              "addresses.user_id = users.id ORDER BY "
              "users.name",
          )
  
      def test_multi_columns(self):
          users, User = self.tables.users, self.classes.User
  
          sess = create_session()
  
          expected = [(u, u.name) for u in sess.query(User).all()]
  
          for add_col in (User.name, users.c.name):
              assert sess.query(User).add_columns(add_col).all() == expected
              sess.expunge_all()
  
          assert_raises(
              sa_exc.ArgumentError, sess.query(User).add_columns, object()
          )
  
      def test_add_multi_columns(self):
          """test that add_column accepts a FROM clause."""
  
          users, User = self.tables.users, self.classes.User
  
          sess = create_session()
  
          eq_(
              sess.query(User.id).add_columns(users).all(),
              [(7, 7, "jack"), (8, 8, "ed"), (9, 9, "fred"), (10, 10, "chuck")],
          )
  
      def test_multi_columns_2(self):
          """test aliased/nonalised joins with the usage of add_columns()"""
  
          User, Address, addresses, users = (
              self.classes.User,
              self.classes.Address,
              self.tables.addresses,
              self.tables.users,
          )
  
          sess = create_session()
  
          (user7, user8, user9, user10) = sess.query(User).all()
          expected = [(user7, 1), (user8, 3), (user9, 1), (user10, 0)]
  
          q = sess.query(User)
          q = (
              q.group_by(users)
              .order_by(User.id)
              .outerjoin("addresses")
              .add_columns(func.count(Address.id).label("count"))
          )
          eq_(q.all(), expected)
          sess.expunge_all()
  
          adalias = aliased(Address)
          q = sess.query(User)
          q = (
              q.group_by(users)
              .order_by(User.id)
              .outerjoin(adalias, "addresses")
              .add_columns(func.count(adalias.id).label("count"))
          )
          eq_(q.all(), expected)
          sess.expunge_all()
  
          # TODO: figure out why group_by(users) doesn't work here
          count = func.count(addresses.c.id).label("count")
          s = (
              select(users, count)
              .select_from(users.outerjoin(addresses))
              .group_by(*[c for c in users.c])
              .order_by(User.id)
          )
          q = sess.query(User)
          result = (
              q.add_columns(s.selected_columns.count).from_statement(s).all()
          )
          assert result == expected
  
      def test_multi_columns_3(self):
          User = self.classes.User
          users = self.tables.users
  
          sess = create_session()
  
          q = sess.query(User.id, User.name)
          stmt = select(users).order_by(users.c.id)
          q = q.from_statement(stmt)
  
          eq_(q.all(), [(7, "jack"), (8, "ed"), (9, "fred"), (10, "chuck")])
  
      def test_raw_columns(self):
          addresses, users, User = (
              self.tables.addresses,
              self.tables.users,
              self.classes.User,
          )
  
          sess = create_session()
          (user7, user8, user9, user10) = sess.query(User).all()
          expected = [
              (user7, 1, "Name:jack"),
              (user8, 3, "Name:ed"),
              (user9, 1, "Name:fred"),
              (user10, 0, "Name:chuck"),
          ]
  
          adalias = addresses.alias()
          q = (
              create_session()
              .query(User)
              .add_columns(func.count(adalias.c.id), ("Name:" + users.c.name))
              .outerjoin(adalias, "addresses")
              .group_by(users)
              .order_by(users.c.id)
          )
  
          assert q.all() == expected
  
          # test with a straight statement
          s = (
              select(
                  users,
                  func.count(addresses.c.id).label("count"),
                  ("Name:" + users.c.name).label("concat"),
              )
              .select_from(users.outerjoin(addresses))
              .group_by(*[c for c in users.c])
              .order_by(users.c.id)
          )
          q = create_session().query(User)
          result = (
              q.add_columns(s.selected_columns.count, s.selected_columns.concat)
              .from_statement(s)
              .all()
          )
          assert result == expected
  
          sess.expunge_all()
  
          # test with select_entity_from()
          q = (
              create_session()
              .query(User)
              .add_columns(func.count(addresses.c.id), ("Name:" + users.c.name))
              .select_entity_from(users.outerjoin(addresses))
              .group_by(users)
              .order_by(users.c.id)
          )
  
          assert q.all() == expected
          sess.expunge_all()
  
          q = (
              create_session()
              .query(User)
              .add_columns(func.count(addresses.c.id), ("Name:" + users.c.name))
              .outerjoin("addresses")
              .group_by(users)
              .order_by(users.c.id)
          )
  
          assert q.all() == expected
          sess.expunge_all()
  
          q = (
              create_session()
              .query(User)
              .add_columns(func.count(adalias.c.id), ("Name:" + users.c.name))
              .outerjoin(adalias, "addresses")
              .group_by(users)
              .order_by(users.c.id)
          )
  
          assert q.all() == expected
          sess.expunge_all()
  
      def test_expression_selectable_matches_mzero(self):
          User, Address = self.classes.User, self.classes.Address
  
          ua = aliased(User)
          aa = aliased(Address)
          s = create_session()
          for crit, j, exp in [
              (
                  User.id + Address.id,
                  User.addresses,
                  "SELECT users.id + addresses.id AS anon_1 "
                  "FROM users JOIN addresses ON users.id = "
                  "addresses.user_id",
              ),
              (
                  User.id + Address.id,
                  Address.user,
                  "SELECT users.id + addresses.id AS anon_1 "
                  "FROM addresses JOIN users ON users.id = "
                  "addresses.user_id",
              ),
              (
                  Address.id + User.id,
                  User.addresses,
                  "SELECT addresses.id + users.id AS anon_1 "
                  "FROM users JOIN addresses ON users.id = "
                  "addresses.user_id",
              ),
              (
                  User.id + aa.id,
                  (aa, User.addresses),
                  "SELECT users.id + addresses_1.id AS anon_1 "
                  "FROM users JOIN addresses AS addresses_1 "
                  "ON users.id = addresses_1.user_id",
              ),
          ]:
              q = s.query(crit)
              mzero = q._compile_state()._entity_zero()
              is_(mzero, q._compile_state()._entities[0].entity_zero)
              q = q.join(j)
              self.assert_compile(q, exp)
  
          for crit, j, exp in [
              (
                  ua.id + Address.id,
                  ua.addresses,
                  "SELECT users_1.id + addresses.id AS anon_1 "
                  "FROM users AS users_1 JOIN addresses "
                  "ON users_1.id = addresses.user_id",
              ),
              (
                  ua.id + aa.id,
                  (aa, ua.addresses),
                  "SELECT users_1.id + addresses_1.id AS anon_1 "
                  "FROM users AS users_1 JOIN addresses AS "
                  "addresses_1 ON users_1.id = addresses_1.user_id",
              ),
              (
                  ua.id + aa.id,
                  (ua, aa.user),
                  "SELECT users_1.id + addresses_1.id AS anon_1 "
                  "FROM addresses AS addresses_1 JOIN "
                  "users AS users_1 "
                  "ON users_1.id = addresses_1.user_id",
              ),
          ]:
              q = s.query(crit)
              mzero = q._compile_state()._entity_zero()
              is_(mzero, q._compile_state()._entities[0].entity_zero)
              q = q.join(j)
              self.assert_compile(q, exp)
  
      def test_aliased_adapt_on_names(self):
          User, Address = self.classes.User, self.classes.Address
  
          sess = Session()
          agg_address = sess.query(
              Address.id,
              func.sum(func.length(Address.email_address)).label(
                  "email_address"
              ),
          ).group_by(Address.user_id)
  
          ag1 = aliased(Address, agg_address.subquery())
          ag2 = aliased(Address, agg_address.subquery(), adapt_on_names=True)
  
          # first, without adapt on names, 'email_address' isn't matched up - we
          # get the raw "address" element in the SELECT
          self.assert_compile(
              sess.query(User, ag1.email_address)
              .join(ag1, User.addresses)
              .filter(ag1.email_address > 5),
              "SELECT users.id "
              "AS users_id, users.name AS users_name, addresses.email_address "
              "AS addresses_email_address FROM addresses, users JOIN "
              "(SELECT addresses.id AS id, sum(length(addresses.email_address)) "
              "AS email_address FROM addresses GROUP BY addresses.user_id) AS "
              "anon_1 ON users.id = addresses.user_id "
              "WHERE addresses.email_address > :email_address_1",
          )
  
          # second, 'email_address' matches up to the aggregate, and we get a
          # smooth JOIN from users->subquery and that's it
          self.assert_compile(
              sess.query(User, ag2.email_address)
              .join(ag2, User.addresses)
              .filter(ag2.email_address > 5),
              "SELECT users.id AS users_id, users.name AS users_name, "
              "anon_1.email_address AS anon_1_email_address FROM users "
              "JOIN ("
              "SELECT addresses.id AS id, sum(length(addresses.email_address)) "
              "AS email_address FROM addresses GROUP BY addresses.user_id) AS "
              "anon_1 ON users.id = addresses.user_id "
              "WHERE anon_1.email_address > :email_address_1",
          )
  
  
  class SelectFromTest(QueryTest, AssertsCompiledSQL):
      run_setup_mappers = None
      __dialect__ = "default"
  
      def test_replace_with_select(self):
          users, Address, addresses, User = (
              self.tables.users,
              self.classes.Address,
              self.tables.addresses,
              self.classes.User,
          )
  
          mapper(User, users, properties={"addresses": relationship(Address)})
          mapper(Address, addresses)
  
          sel = users.select(users.c.id.in_([7, 8])).alias()
          sess = create_session()
  
          eq_(
              sess.query(User).select_entity_from(sel).all(),
              [User(id=7), User(id=8)],
          )
  
          eq_(
              sess.query(User)
              .select_entity_from(sel)
              .filter(User.id == 8)
              .all(),
              [User(id=8)],
          )
  
          eq_(
              sess.query(User)
              .select_entity_from(sel)
              .order_by(desc(User.name))
              .all(),
              [User(name="jack", id=7), User(name="ed", id=8)],
          )
  
          eq_(
              sess.query(User)
              .select_entity_from(sel)
              .order_by(asc(User.name))
              .all(),
              [User(name="ed", id=8), User(name="jack", id=7)],
          )
  
          eq_(
              sess.query(User)
              .select_entity_from(sel)
              .options(joinedload("addresses"))
              .first(),
              User(name="jack", addresses=[Address(id=1)]),
          )
  
      def test_select_from_aliased_one(self):
          User, users = self.classes.User, self.tables.users
  
          mapper(User, users)
  
          sess = create_session()
  
          not_users = table("users", column("id"), column("name"))
          ua = aliased(User, select(not_users).alias(), adapt_on_names=True)
  
          q = sess.query(User.name).select_entity_from(ua).order_by(User.name)
          self.assert_compile(
              q,
              "SELECT anon_1.name AS anon_1_name FROM (SELECT users.id AS id, "
              "users.name AS name FROM users) AS anon_1 ORDER BY anon_1.name",
          )
          eq_(q.all(), [("chuck",), ("ed",), ("fred",), ("jack",)])
  
      def test_select_from_aliased_two(self):
          User, users = self.classes.User, self.tables.users
  
          mapper(User, users)
  
          sess = create_session()
  
          ua = aliased(User)
  
          q = sess.query(User.name).select_entity_from(ua).order_by(User.name)
          self.assert_compile(
              q,
              "SELECT users_1.name AS users_1_name FROM users AS users_1 "
              "ORDER BY users_1.name",
          )
          eq_(q.all(), [("chuck",), ("ed",), ("fred",), ("jack",)])
  
      def test_select_from_core_alias_one(self):
          User, users = self.classes.User, self.tables.users
  
          mapper(User, users)
  
          sess = create_session()
  
          ua = users.alias()
  
          q = sess.query(User.name).select_entity_from(ua).order_by(User.name)
          self.assert_compile(
              q,
              "SELECT users_1.name AS users_1_name FROM users AS users_1 "
              "ORDER BY users_1.name",
          )
          eq_(q.all(), [("chuck",), ("ed",), ("fred",), ("jack",)])
  
      def test_differentiate_self_external(self):
          """test some different combinations of joining a table to a subquery of
          itself."""
  
          users, User = self.tables.users, self.classes.User
  
          mapper(User, users)
  
          sess = create_session()
  
          sel = sess.query(User).filter(User.id.in_([7, 8])).subquery()
          ualias = aliased(User)
  
          self.assert_compile(
              sess.query(User).join(sel, User.id > sel.c.id),
              "SELECT users.id AS users_id, users.name AS users_name FROM "
              "users JOIN (SELECT users.id AS id, users.name AS name FROM users "
              "WHERE users.id IN ([POSTCOMPILE_id_1])) "
              "AS anon_1 ON users.id > anon_1.id",
          )
  
          self.assert_compile(
              sess.query(ualias)
              .select_entity_from(sel)
              .filter(ualias.id > sel.c.id),
              "SELECT users_1.id AS users_1_id, users_1.name AS users_1_name "
              "FROM users AS users_1, ("
              "SELECT users.id AS id, users.name AS name FROM users "
              "WHERE users.id IN ([POSTCOMPILE_id_1])) AS anon_1 "
              "WHERE users_1.id > anon_1.id",
              check_post_param={"id_1": [7, 8]},
          )
  
          self.assert_compile(
              sess.query(ualias)
              .select_entity_from(sel)
              .join(ualias, ualias.id > sel.c.id),
              "SELECT users_1.id AS users_1_id, users_1.name AS users_1_name "
              "FROM (SELECT users.id AS id, users.name AS name "
              "FROM users WHERE users.id IN ([POSTCOMPILE_id_1])) AS anon_1 "
              "JOIN users AS users_1 ON users_1.id > anon_1.id",
              check_post_param={"id_1": [7, 8]},
          )
  
          self.assert_compile(
              sess.query(ualias)
              .select_entity_from(sel)
              .join(ualias, ualias.id > User.id),
              "SELECT users_1.id AS users_1_id, users_1.name AS users_1_name "
              "FROM (SELECT users.id AS id, users.name AS name FROM "
              "users WHERE users.id IN ([POSTCOMPILE_id_1])) AS anon_1 "
              "JOIN users AS users_1 ON users_1.id > anon_1.id",
              check_post_param={"id_1": [7, 8]},
          )
  
          salias = aliased(User, sel)
          self.assert_compile(
              sess.query(salias).join(ualias, ualias.id > salias.id),
              "SELECT anon_1.id AS anon_1_id, anon_1.name AS anon_1_name FROM "
              "(SELECT users.id AS id, users.name AS name "
              "FROM users WHERE users.id IN ([POSTCOMPILE_id_1])) AS anon_1 "
              "JOIN users AS users_1 ON users_1.id > anon_1.id",
              check_post_param={"id_1": [7, 8]},
          )
  
          self.assert_compile(
              sess.query(ualias).select_entity_from(
                  join(sel, ualias, ualias.id > sel.c.id)
              ),
              "SELECT users_1.id AS users_1_id, users_1.name AS users_1_name "
              "FROM "
              "(SELECT users.id AS id, users.name AS name "
              "FROM users WHERE users.id "
              "IN ([POSTCOMPILE_id_1])) AS anon_1 "
              "JOIN users AS users_1 ON users_1.id > anon_1.id",
              check_post_param={"id_1": [7, 8]},
          )
  
      def test_aliased_class_vs_nonaliased(self):
          User, users = self.classes.User, self.tables.users
          mapper(User, users)
  
          ua = aliased(User)
  
          sess = create_session()
          self.assert_compile(
              sess.query(User).select_from(ua).join(User, ua.name > User.name),
              "SELECT users.id AS users_id, users.name AS users_name "
              "FROM users AS users_1 JOIN users ON users_1.name > users.name",
          )
  
          self.assert_compile(
              sess.query(User.name)
              .select_from(ua)
              .join(User, ua.name > User.name),
              "SELECT users.name AS users_name FROM users AS users_1 "
              "JOIN users ON users_1.name > users.name",
          )
  
          self.assert_compile(
              sess.query(ua.name)
              .select_from(ua)
              .join(User, ua.name > User.name),
              "SELECT users_1.name AS users_1_name FROM users AS users_1 "
              "JOIN users ON users_1.name > users.name",
          )
  
          self.assert_compile(
              sess.query(ua).select_from(User).join(ua, ua.name > User.name),
              "SELECT users_1.id AS users_1_id, users_1.name AS users_1_name "
              "FROM users JOIN users AS users_1 ON users_1.name > users.name",
          )
  
          self.assert_compile(
              sess.query(ua).select_from(User).join(ua, User.name > ua.name),
              "SELECT users_1.id AS users_1_id, users_1.name AS users_1_name "
              "FROM users JOIN users AS users_1 ON users.name > users_1.name",
          )
  
          # this is tested in many other places here, just adding it
          # here for comparison
          self.assert_compile(
              sess.query(User.name).select_entity_from(
                  users.select().where(users.c.id > 5).subquery()
              ),
              "SELECT anon_1.name AS anon_1_name FROM (SELECT users.id AS id, "
              "users.name AS name FROM users WHERE users.id > :id_1) AS anon_1",
          )
  
      def test_join_no_order_by(self):
          User, users = self.classes.User, self.tables.users
  
          mapper(User, users)
  
          sel = users.select(users.c.id.in_([7, 8]))
          sess = create_session()
  
          eq_(
              sess.query(User).select_entity_from(sel.subquery()).all(),
              [User(name="jack", id=7), User(name="ed", id=8)],
          )
  
      def test_join_relname_from_selected_from(self):
          User, Address = self.classes.User, self.classes.Address
          users, addresses = self.tables.users, self.tables.addresses
          mapper(
              User,
              users,
              properties={
                  "addresses": relationship(
                      mapper(Address, addresses), backref="user"
                  )
              },
          )
  
          sess = create_session()
  
          self.assert_compile(
              sess.query(User).select_from(Address).join("user"),
              "SELECT users.id AS users_id, users.name AS users_name "
              "FROM addresses JOIN users ON users.id = addresses.user_id",
          )
  
      def test_filter_by_selected_from(self):
          User, Address = self.classes.User, self.classes.Address
          users, addresses = self.tables.users, self.tables.addresses
          mapper(
              User,
              users,
              properties={"addresses": relationship(mapper(Address, addresses))},
          )
  
          sess = create_session()
  
          self.assert_compile(
              sess.query(User)
              .select_from(Address)
              .filter_by(email_address="ed")
              .join(User),
              "SELECT users.id AS users_id, users.name AS users_name "
              "FROM addresses JOIN users ON users.id = addresses.user_id "
              "WHERE addresses.email_address = :email_address_1",
          )
  
      def test_join_ent_selected_from(self):
          User, Address = self.classes.User, self.classes.Address
          users, addresses = self.tables.users, self.tables.addresses
          mapper(
              User,
              users,
              properties={"addresses": relationship(mapper(Address, addresses))},
          )
  
          sess = create_session()
  
          self.assert_compile(
              sess.query(User).select_from(Address).join(User),
              "SELECT users.id AS users_id, users.name AS users_name "
              "FROM addresses JOIN users ON users.id = addresses.user_id",
          )
  
      def test_join(self):
          users, Address, addresses, User = (
              self.tables.users,
              self.classes.Address,
              self.tables.addresses,
              self.classes.User,
          )
  
          mapper(User, users, properties={"addresses": relationship(Address)})
          mapper(Address, addresses)
  
          sel = users.select(users.c.id.in_([7, 8]))
          sess = create_session()
  
          eq_(
              sess.query(User)
              .select_entity_from(sel.subquery())
              .join("addresses")
              .add_entity(Address)
              .order_by(User.id)
              .order_by(Address.id)
              .all(),
              [
                  (
                      User(name="jack", id=7),
                      Address(user_id=7, email_address="jack@bean.com", id=1),
                  ),
                  (
                      User(name="ed", id=8),
                      Address(user_id=8, email_address="ed@wood.com", id=2),
                  ),
                  (
                      User(name="ed", id=8),
                      Address(user_id=8, email_address="ed@bettyboop.com", id=3),
                  ),
                  (
                      User(name="ed", id=8),
                      Address(user_id=8, email_address="ed@lala.com", id=4),
                  ),
              ],
          )
  
          adalias = aliased(Address)
          eq_(
              sess.query(User)
              .select_entity_from(sel.subquery())
              .join(adalias, "addresses")
              .add_entity(adalias)
              .order_by(User.id)
              .order_by(adalias.id)
              .all(),
              [
                  (
                      User(name="jack", id=7),
                      Address(user_id=7, email_address="jack@bean.com", id=1),
                  ),
                  (
                      User(name="ed", id=8),
                      Address(user_id=8, email_address="ed@wood.com", id=2),
                  ),
                  (
                      User(name="ed", id=8),
                      Address(user_id=8, email_address="ed@bettyboop.com", id=3),
                  ),
                  (
                      User(name="ed", id=8),
                      Address(user_id=8, email_address="ed@lala.com", id=4),
                  ),
              ],
          )
  
      def test_more_joins(self):
          (
              users,
              Keyword,
              orders,
              items,
              order_items,
              Order,
              Item,
              User,
              keywords,
              item_keywords,
          ) = (
              self.tables.users,
              self.classes.Keyword,
              self.tables.orders,
              self.tables.items,
              self.tables.order_items,
              self.classes.Order,
              self.classes.Item,
              self.classes.User,
              self.tables.keywords,
              self.tables.item_keywords,
          )
  
          mapper(
              User,
              users,
              properties={"orders": relationship(Order, backref="user")},
          )  # o2m, m2o
          mapper(
              Order,
              orders,
              properties={
                  "items": relationship(
                      Item, secondary=order_items, order_by=items.c.id
                  )
              },
          )  # m2m
  
          mapper(
              Item,
              items,
              properties={
                  "keywords": relationship(
                      Keyword, secondary=item_keywords, order_by=keywords.c.id
                  )
              },
          )  # m2m
          mapper(Keyword, keywords)
  
          sess = create_session()
          sel = users.select(users.c.id.in_([7, 8]))
  
          eq_(
              sess.query(User)
              .select_entity_from(sel.subquery())
              .join(User.orders, Order.items, Item.keywords)
              .filter(Keyword.name.in_(["red", "big", "round"]))
              .all(),
              [User(name="jack", id=7)],
          )
  
      def test_very_nested_joins_with_joinedload(self):
          (
              users,
              Keyword,
              orders,
              items,
              order_items,
              Order,
              Item,
              User,
              keywords,
              item_keywords,
          ) = (
              self.tables.users,
              self.classes.Keyword,
              self.tables.orders,
              self.tables.items,
              self.tables.order_items,
              self.classes.Order,
              self.classes.Item,
              self.classes.User,
              self.tables.keywords,
              self.tables.item_keywords,
          )
  
          mapper(
              User,
              users,
              properties={"orders": relationship(Order, backref="user")},
          )  # o2m, m2o
          mapper(
              Order,
              orders,
              properties={
                  "items": relationship(
                      Item, secondary=order_items, order_by=items.c.id
                  )
              },
          )  # m2m
          mapper(
              Item,
              items,
              properties={
                  "keywords": relationship(
                      Keyword, secondary=item_keywords, order_by=keywords.c.id
                  )
              },
          )  # m2m
          mapper(Keyword, keywords)
  
          sess = create_session()
  
          sel = users.select(users.c.id.in_([7, 8]))
  
          def go():
              eq_(
                  sess.query(User)
                  .select_entity_from(sel.subquery())
                  .options(
                      joinedload("orders")
                      .joinedload("items")
                      .joinedload("keywords")
                  )
                  .join(User.orders, Order.items, Item.keywords)
                  .filter(Keyword.name.in_(["red", "big", "round"]))
                  .all(),
                  [
                      User(
                          name="jack",
                          orders=[
                              Order(
                                  description="order 1",
                                  items=[
                                      Item(
                                          description="item 1",
                                          keywords=[
                                              Keyword(name="red"),
                                              Keyword(name="big"),
                                              Keyword(name="round"),
                                          ],
                                      ),
                                      Item(
                                          description="item 2",
                                          keywords=[
                                              Keyword(name="red", id=2),
                                              Keyword(name="small", id=5),
                                              Keyword(name="square"),
                                          ],
                                      ),
                                      Item(
                                          description="item 3",
                                          keywords=[
                                              Keyword(name="green", id=3),
                                              Keyword(name="big", id=4),
                                              Keyword(name="round", id=6),
                                          ],
                                      ),
                                  ],
                              ),
                              Order(
                                  description="order 3",
                                  items=[
                                      Item(
                                          description="item 3",
                                          keywords=[
                                              Keyword(name="green", id=3),
                                              Keyword(name="big", id=4),
                                              Keyword(name="round", id=6),
                                          ],
                                      ),
                                      Item(
                                          description="item 4", keywords=[], id=4
                                      ),
                                      Item(
                                          description="item 5", keywords=[], id=5
                                      ),
                                  ],
                              ),
                              Order(
                                  description="order 5",
                                  items=[
                                      Item(description="item 5", keywords=[])
                                  ],
                              ),
                          ],
                      )
                  ],
              )
  
          self.assert_sql_count(testing.db, go, 1)
  
          sess.expunge_all()
          sel2 = orders.select(orders.c.id.in_([1, 2, 3]))
          eq_(
              sess.query(Order)
              .select_entity_from(sel2.subquery())
              .join(Order.items)
              .join(Item.keywords)
              .filter(Keyword.name == "red")
              .order_by(Order.id)
              .all(),
              [
                  Order(description="order 1", id=1),
                  Order(description="order 2", id=2),
              ],
          )
  
      def test_replace_with_eager(self):
          users, Address, addresses, User = (
              self.tables.users,
              self.classes.Address,
              self.tables.addresses,
              self.classes.User,
          )
  
          mapper(
              User,
              users,
              properties={
                  "addresses": relationship(Address, order_by=addresses.c.id)
              },
          )
          mapper(Address, addresses)
  
          sel = users.select(users.c.id.in_([7, 8]))
          sess = create_session()
  
          def go():
              eq_(
                  sess.query(User)
                  .options(joinedload("addresses"))
                  .select_entity_from(sel.subquery())
                  .order_by(User.id)
                  .all(),
                  [
                      User(id=7, addresses=[Address(id=1)]),
                      User(
                          id=8,
                          addresses=[
                              Address(id=2),
                              Address(id=3),
                              Address(id=4),
                          ],
                      ),
                  ],
              )
  
          self.assert_sql_count(testing.db, go, 1)
          sess.expunge_all()
  
          def go():
              eq_(
                  sess.query(User)
                  .options(joinedload("addresses"))
                  .select_entity_from(sel.subquery())
                  .filter(User.id == 8)
                  .order_by(User.id)
                  .all(),
                  [
                      User(
                          id=8,
                          addresses=[
                              Address(id=2),
                              Address(id=3),
                              Address(id=4),
                          ],
                      )
                  ],
              )
  
          self.assert_sql_count(testing.db, go, 1)
          sess.expunge_all()
  
          def go():
              eq_(
                  sess.query(User)
                  .options(joinedload("addresses"))
                  .select_entity_from(sel.subquery())
                  .order_by(User.id)[1],
                  User(
                      id=8,
                      addresses=[Address(id=2), Address(id=3), Address(id=4)],
                  ),
              )
  
          self.assert_sql_count(testing.db, go, 1)
  
  
  class CustomJoinTest(QueryTest):
      run_setup_mappers = None
  
      def test_double_same_mappers_flag_alias(self):
          """test aliasing of joins with a custom join condition"""
  
          (
              addresses,
              items,
              order_items,
              orders,
              Item,
              User,
              Address,
              Order,
              users,
          ) = (
              self.tables.addresses,
              self.tables.items,
              self.tables.order_items,
              self.tables.orders,
              self.classes.Item,
              self.classes.User,
              self.classes.Address,
              self.classes.Order,
              self.tables.users,
          )
  
          mapper(Address, addresses)
          mapper(
              Order,
              orders,
              properties={
                  "items": relationship(
                      Item,
                      secondary=order_items,
                      lazy="select",
                      order_by=items.c.id,
                  )
              },
          )
          mapper(Item, items)
          mapper(
              User,
              users,
              properties=dict(
                  addresses=relationship(Address, lazy="select"),
                  open_orders=relationship(
                      Order,
                      primaryjoin=and_(
                          orders.c.isopen == 1, users.c.id == orders.c.user_id
                      ),
                      lazy="select",
                      viewonly=True,
                  ),
                  closed_orders=relationship(
                      Order,
                      primaryjoin=and_(
                          orders.c.isopen == 0, users.c.id == orders.c.user_id
                      ),
                      lazy="select",
                      viewonly=True,
                  ),
              ),
          )
          q = create_session().query(User)
  
          eq_(
              q.join("open_orders", "items", aliased=True)
              .filter(Item.id == 4)
              .join("closed_orders", "items", aliased=True)
              .filter(Item.id == 3)
              .all(),
              [User(id=7)],
          )
  
      def test_double_same_mappers_explicit_alias(self):
          """test aliasing of joins with a custom join condition"""
  
          (
              addresses,
              items,
              order_items,
              orders,
              Item,
              User,
              Address,
              Order,
              users,
          ) = (
              self.tables.addresses,
              self.tables.items,
              self.tables.order_items,
              self.tables.orders,
              self.classes.Item,
              self.classes.User,
              self.classes.Address,
              self.classes.Order,
              self.tables.users,
          )
  
          mapper(Address, addresses)
          mapper(
              Order,
              orders,
              properties={
                  "items": relationship(
                      Item,
                      secondary=order_items,
                      lazy="select",
                      order_by=items.c.id,
                  )
              },
          )
          mapper(Item, items)
          mapper(
              User,
              users,
              properties=dict(
                  addresses=relationship(Address, lazy="select"),
                  open_orders=relationship(
                      Order,
                      primaryjoin=and_(
                          orders.c.isopen == 1, users.c.id == orders.c.user_id
                      ),
                      lazy="select",
                      viewonly=True,
                  ),
                  closed_orders=relationship(
                      Order,
                      primaryjoin=and_(
                          orders.c.isopen == 0, users.c.id == orders.c.user_id
                      ),
                      lazy="select",
                      viewonly=True,
                  ),
              ),
          )
          q = create_session().query(User)
  
          oo = aliased(Order)
          co = aliased(Order)
          oi = aliased(Item)
          ci = aliased(Item)
  
          # converted from aliased=True.  This is kind of the worst case
          # kind of query when we don't have aliased=True.  two different
          # styles are illustrated here, but the important point is that
          # the filter() is not doing any trickery, you need to pass it the
          # aliased entity explicitly.
          eq_(
              q.join(oo, User.open_orders)
              .join(oi, oo.items)
              .filter(oi.id == 4)
              .join(User.closed_orders.of_type(co))
              .join(co.items.of_type(ci))
              .filter(ci.id == 3)
              .all(),
              [User(id=7)],
          )
  
  
  class ExternalColumnsTest(QueryTest):
      """test mappers with SQL-expressions added as column properties."""
  
      run_setup_mappers = None
  
      def test_external_columns_bad(self):
          users, User = self.tables.users, self.classes.User
  
          assert_raises_message(
              sa_exc.ArgumentError,
              "not represented in the mapper's table",
              mapper,
              User,
              users,
              properties={"concat": (users.c.id * 2)},
          )
          clear_mappers()
  
      def test_external_columns(self):
          """test querying mappings that reference external columns or
          selectables."""
  
          users, Address, addresses, User = (
              self.tables.users,
              self.classes.Address,
              self.tables.addresses,
              self.classes.User,
          )
  
          mapper(
              User,
              users,
              properties={
                  "concat": column_property((users.c.id * 2)),
                  "count": column_property(
                      select(func.count(addresses.c.id))
                      .where(users.c.id == addresses.c.user_id,)
                      .correlate(users)
                      .scalar_subquery()
                  ),
              },
          )
  
          mapper(Address, addresses, properties={"user": relationship(User,)})
  
          sess = create_session()
  
          sess.query(Address).options(joinedload("user")).all()
  
          eq_(
              sess.query(User).all(),
              [
                  User(id=7, concat=14, count=1),
                  User(id=8, concat=16, count=3),
                  User(id=9, concat=18, count=1),
                  User(id=10, concat=20, count=0),
              ],
          )
  
          address_result = [
              Address(id=1, user=User(id=7, concat=14, count=1)),
              Address(id=2, user=User(id=8, concat=16, count=3)),
              Address(id=3, user=User(id=8, concat=16, count=3)),
              Address(id=4, user=User(id=8, concat=16, count=3)),
              Address(id=5, user=User(id=9, concat=18, count=1)),
          ]
          # TODO: ISSUE: BUG:  cached metadata is confusing the user.id
          # column here with the anon_1 for some reason, when we
          # use compiled cache.  this bug may even be present in
          # regular master / 1.3.  right now the caching of result
          # metadata is disabled.
          eq_(sess.query(Address).all(), address_result)
  
          # run the eager version twice to test caching of aliased clauses
          for x in range(2):
              sess.expunge_all()
  
              def go():
                  eq_(
                      sess.query(Address)
                      .options(joinedload("user"))
                      .order_by(Address.id)
                      .all(),
                      address_result,
                  )
  
              self.assert_sql_count(testing.db, go, 1)
  
          ualias = aliased(User)
          eq_(
              sess.query(Address, ualias).join(ualias, "user").all(),
              [(address, address.user) for address in address_result],
          )
  
          ualias2 = aliased(User)
          eq_(
              sess.query(Address, ualias.count)
              .join(ualias, "user")
              .join(ualias2, "user")
              .order_by(Address.id)
              .all(),
              [
                  (Address(id=1), 1),
                  (Address(id=2), 3),
                  (Address(id=3), 3),
                  (Address(id=4), 3),
                  (Address(id=5), 1),
              ],
          )
  
          eq_(
              sess.query(Address, ualias.concat, ualias.count)
              .join(ualias, "user")
              .join(ualias2, "user")
              .order_by(Address.id)
              .all(),
              [
                  (Address(id=1), 14, 1),
                  (Address(id=2), 16, 3),
                  (Address(id=3), 16, 3),
                  (Address(id=4), 16, 3),
                  (Address(id=5), 18, 1),
              ],
          )
  
          ua = aliased(User)
          eq_(
              sess.query(Address, ua.concat, ua.count)
              .select_entity_from(join(Address, ua, "user"))
              .options(joinedload(Address.user))
              .order_by(Address.id)
              .all(),
              [
                  (Address(id=1, user=User(id=7, concat=14, count=1)), 14, 1),
                  (Address(id=2, user=User(id=8, concat=16, count=3)), 16, 3),
                  (Address(id=3, user=User(id=8, concat=16, count=3)), 16, 3),
                  (Address(id=4, user=User(id=8, concat=16, count=3)), 16, 3),
                  (Address(id=5, user=User(id=9, concat=18, count=1)), 18, 1),
              ],
          )
  
          eq_(
              list(
                  sess.query(Address)
                  .join("user")
                  .with_entities(Address.id, User.id, User.concat, User.count)
              ),
              [
                  (1, 7, 14, 1),
                  (2, 8, 16, 3),
                  (3, 8, 16, 3),
                  (4, 8, 16, 3),
                  (5, 9, 18, 1),
              ],
          )
  
          eq_(
              list(
                  sess.query(Address, ua)
                  .select_entity_from(join(Address, ua, "user"))
                  .with_entities(Address.id, ua.id, ua.concat, ua.count)
              ),
              [
                  (1, 7, 14, 1),
                  (2, 8, 16, 3),
                  (3, 8, 16, 3),
                  (4, 8, 16, 3),
                  (5, 9, 18, 1),
              ],
          )
  
      def test_external_columns_joinedload(self):
          users, orders, User, Address, Order, addresses = (
              self.tables.users,
              self.tables.orders,
              self.classes.User,
              self.classes.Address,
              self.classes.Order,
              self.tables.addresses,
          )
  
          # in this test, we have a subquery on User that accesses "addresses",
          # underneath an joinedload for "addresses".  So the "addresses" alias
          # adapter needs to *not* hit the "addresses" table within the "user"
          # subquery, but "user" still needs to be adapted. therefore the long
          # standing practice of eager adapters being "chained" has been removed
          # since its unnecessary and breaks this exact condition.
          mapper(
              User,
              users,
              properties={
                  "addresses": relationship(
                      Address, backref="user", order_by=addresses.c.id
                  ),
                  "concat": column_property((users.c.id * 2)),
                  "count": column_property(
                      select(func.count(addresses.c.id))
                      .where(users.c.id == addresses.c.user_id,)
                      .correlate(users)
                      .scalar_subquery()
                  ),
              },
          )
          mapper(Address, addresses)
          mapper(
              Order, orders, properties={"address": relationship(Address)}
          )  # m2o
  
          sess = create_session()
  
          def go():
              o1 = (
                  sess.query(Order)
                  .options(joinedload("address").joinedload("user"))
                  .get(1)
              )
              eq_(o1.address.user.count, 1)
  
          self.assert_sql_count(testing.db, go, 1)
  
          sess = create_session()
  
          def go():
              o1 = (
                  sess.query(Order)
                  .options(joinedload("address").joinedload("user"))
                  .first()
              )
              eq_(o1.address.user.count, 1)
  
          self.assert_sql_count(testing.db, go, 1)
  
      def test_external_columns_compound(self):
          # see [ticket:2167] for background
          users, Address, addresses, User = (
              self.tables.users,
              self.classes.Address,
              self.tables.addresses,
              self.classes.User,
          )
  
          mapper(
              User,
              users,
              properties={"fullname": column_property(users.c.name.label("x"))},
          )
  
          mapper(
              Address,
              addresses,
              properties={
                  "username": column_property(
                      select(User.fullname)
                      .where(User.id == addresses.c.user_id)
                      .label("y")
                  )
              },
          )
          sess = create_session()
          a1 = sess.query(Address).first()
          eq_(a1.username, "jack")
  
          sess = create_session()
          a1 = sess.query(Address).from_self().first()
          eq_(a1.username, "jack")
  
  
  class TestOverlyEagerEquivalentCols(fixtures.MappedTest):
      @classmethod
      def define_tables(cls, metadata):
          Table(
              "base",
              metadata,
              Column(
                  "id", Integer, primary_key=True, test_needs_autoincrement=True
              ),
              Column("data", String(50)),
          )
  
          Table(
              "sub1",
              metadata,
              Column("id", Integer, ForeignKey("base.id"), primary_key=True),
              Column("data", String(50)),
          )
  
          Table(
              "sub2",
              metadata,
              Column(
                  "id",
                  Integer,
                  ForeignKey("base.id"),
                  ForeignKey("sub1.id"),
                  primary_key=True,
              ),
              Column("data", String(50)),
          )
  
      def test_equivs(self):
          base, sub2, sub1 = (
              self.tables.base,
              self.tables.sub2,
              self.tables.sub1,
          )
  
          class Base(fixtures.ComparableEntity):
              pass
  
          class Sub1(fixtures.ComparableEntity):
              pass
  
          class Sub2(fixtures.ComparableEntity):
              pass
  
          mapper(
              Base,
              base,
              properties={
                  "sub1": relationship(Sub1),
                  "sub2": relationship(Sub2),
              },
          )
  
          mapper(Sub1, sub1)
          mapper(Sub2, sub2)
          sess = create_session()
  
          s11 = Sub1(data="s11")
          s12 = Sub1(data="s12")
          b1 = Base(data="b1", sub1=[s11], sub2=[])
          b2 = Base(data="b1", sub1=[s12], sub2=[])
          sess.add(b1)
          sess.add(b2)
          sess.flush()
  
  
  class LabelCollideTest(fixtures.MappedTest):
      """Test handling for a label collision.  This collision
      is handled by core, see ticket:2702 as well as
      test/sql/test_selectable->WithLabelsTest.  here we want
      to make sure the end result is as we expect.
  
      """
  
      @classmethod
      def define_tables(cls, metadata):
          Table(
              "foo",
              metadata,
              Column("id", Integer, primary_key=True),
              Column("bar_id", Integer),
          )
          Table("foo_bar", metadata, Column("id", Integer, primary_key=True))
  
      @classmethod
      def setup_classes(cls):
          class Foo(cls.Basic):
              pass
  
          class Bar(cls.Basic):
              pass
  
      @classmethod
      def setup_mappers(cls):
          mapper(cls.classes.Foo, cls.tables.foo)
          mapper(cls.classes.Bar, cls.tables.foo_bar)
  
      @classmethod
      def insert_data(cls, connection):
          s = Session(connection)
          s.add_all([cls.classes.Foo(id=1, bar_id=2), cls.classes.Bar(id=3)])
          s.commit()
  
      def test_overlap_plain(self):
          s = Session()
          row = (
              s.query(self.classes.Foo, self.classes.Bar)
              .join(self.classes.Bar, true())
              .all()[0]
          )
  
          def go():
              eq_(row.Foo.id, 1)
              eq_(row.Foo.bar_id, 2)
              eq_(row.Bar.id, 3)
  
          # all three columns are loaded independently without
          # overlap, no additional SQL to load all attributes
          self.assert_sql_count(testing.db, go, 0)
  
      def test_overlap_subquery(self):
          s = Session()
          row = (
              s.query(self.classes.Foo, self.classes.Bar)
              .join(self.classes.Bar, true())
              .from_self()
              .all()[0]
          )
  
          def go():
              eq_(row.Foo.id, 1)
              eq_(row.Foo.bar_id, 2)
              eq_(row.Bar.id, 3)
  
          # all three columns are loaded independently without
          # overlap, no additional SQL to load all attributes
          self.assert_sql_count(testing.db, go, 0)
  