introduce deferred lambdas

The coercions system allows us to add in lambdas as arguments
to Core and ORM elements without changing them at all.   By allowing
the lambda to produce a deterministic cache key where we can also
cheat and yank out literal parameters means we can move towards
having 90% of "baked" functionality in a clearer way right in
Core / ORM.

As a second step, we can have whole statements inside the lambda,
and can then add generation with __add__(), so then we have
100% of "baked" functionality with full support of ad-hoc
literal values.

Adds some more short_selects tests for the moment for comparison.

Other tweaks inside cache key generation as we're trying to
approach a certain level of performance such that we can
remove the use of "baked" from the loader strategies.

As we have not yet closed #4639, however the caching feature
has been fully integrated as of
b0cfa7379cf8513a821a3dbe3028c4965d9f85bd, we will also
add complete caching documentation here and close that issue
as well.

Closes: #4639
Fixes: #5380
Change-Id: If91f61527236fd4d7ae3cad1f24c38be921c90ba

1 files changed, 438 insertions, 0 deletions

diff --git a/test/orm/test_lambdas.py b/test/orm/test_lambdas.py
new file mode 100644
index 000000000..407f70094
--- /dev/null
+++ b/test/orm/test_lambdas.py
@@ -0,0 +1,438 @@
+import random
+
+from sqlalchemy import exc
+from sqlalchemy import ForeignKey
+from sqlalchemy import Integer
+from sqlalchemy import lambda_stmt
+from sqlalchemy import String
+from sqlalchemy import testing
+from sqlalchemy import update
+from sqlalchemy.future import select
+from sqlalchemy.orm import mapper
+from sqlalchemy.orm import relationship
+from sqlalchemy.orm import selectinload
+from sqlalchemy.orm import Session
+from sqlalchemy.orm import subqueryload
+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.schema import Column
+from sqlalchemy.testing.schema import Table
+from .inheritance import _poly_fixtures
+from .test_query import QueryTest
+
+
+class LambdaTest(QueryTest, AssertsCompiledSQL):
+    __dialect__ = "default"
+
+    # we want to test the lambda expiration logic so use backend
+    # to exercise that
+
+    __backend__ = True
+    run_setup_mappers = None
+
+    @testing.fixture
+    def plain_fixture(self):
+        users, Address, addresses, User = (
+            self.tables.users,
+            self.classes.Address,
+            self.tables.addresses,
+            self.classes.User,
+        )
+
+        mapper(
+            User,
+            users,
+            properties={
+                "addresses": relationship(Address, back_populates="user")
+            },
+        )
+
+        mapper(
+            Address,
+            addresses,
+            properties={
+                "user": relationship(User, back_populates="addresses")
+            },
+        )
+
+        return User, Address
+
+    def test_user_cols_single_lambda(self, plain_fixture):
+        User, Address = plain_fixture
+
+        q = select(lambda: (User.id, User.name)).select_from(lambda: User)
+
+        self.assert_compile(q, "SELECT users.id, users.name FROM users")
+
+    def test_user_cols_single_lambda_query(self, plain_fixture):
+        User, Address = plain_fixture
+
+        s = Session()
+        q = s.query(lambda: (User.id, User.name)).select_from(lambda: User)
+
+        self.assert_compile(
+            q,
+            "SELECT users.id AS users_id, users.name AS users_name FROM users",
+        )
+
+    def test_multiple_entities_single_lambda(self, plain_fixture):
+        User, Address = plain_fixture
+
+        q = select(lambda: (User, Address)).join(lambda: User.addresses)
+
+        self.assert_compile(
+            q,
+            "SELECT users.id, users.name, addresses.id AS id_1, "
+            "addresses.user_id, addresses.email_address "
+            "FROM users JOIN addresses ON users.id = addresses.user_id",
+        )
+
+    def test_cols_round_trip(self, plain_fixture):
+        User, Address = plain_fixture
+
+        s = Session()
+
+        # note this does a traversal + _clone of the InstrumentedAttribute
+        # for the first time ever
+        def query(names):
+            stmt = lambda_stmt(
+                lambda: select(User.name, Address.email_address)
+                .where(User.name.in_(names))
+                .join(User.addresses)
+            ) + (lambda s: s.order_by(User.id, Address.id))
+
+            return s.execute(stmt)
+
+        def go1():
+            r1 = query(["ed"])
+            eq_(
+                r1.all(),
+                [
+                    ("ed", "ed@wood.com"),
+                    ("ed", "ed@bettyboop.com"),
+                    ("ed", "ed@lala.com"),
+                ],
+            )
+
+        def go2():
+            r1 = query(["ed", "fred"])
+            eq_(
+                r1.all(),
+                [
+                    ("ed", "ed@wood.com"),
+                    ("ed", "ed@bettyboop.com"),
+                    ("ed", "ed@lala.com"),
+                    ("fred", "fred@fred.com"),
+                ],
+            )
+
+        for i in range(5):
+            fn = random.choice([go1, go2])
+            fn()
+
+    def test_entity_round_trip(self, plain_fixture):
+        User, Address = plain_fixture
+
+        s = Session()
+
+        def query(names):
+            stmt = lambda_stmt(
+                lambda: select(User)
+                .where(User.name.in_(names))
+                .options(selectinload(User.addresses))
+            ) + (lambda s: s.order_by(User.id))
+
+            return s.execute(stmt)
+
+        def go1():
+            r1 = query(["ed"])
+            eq_(
+                r1.scalars().all(),
+                [User(name="ed", addresses=[Address(), Address(), Address()])],
+            )
+
+        def go2():
+            r1 = query(["ed", "fred"])
+            eq_(
+                r1.scalars().all(),
+                [
+                    User(
+                        name="ed", addresses=[Address(), Address(), Address()]
+                    ),
+                    User(name="fred", addresses=[Address()]),
+                ],
+            )
+
+        for i in range(5):
+            fn = random.choice([go1, go2])
+            self.assert_sql_count(testing.db, fn, 2)
+
+    def test_lambdas_rejected_in_options(self, plain_fixture):
+        User, Address = plain_fixture
+
+        assert_raises_message(
+            exc.ArgumentError,
+            "Cacheable Core or ORM object expected, got",
+            select(lambda: User).options,
+            lambda: subqueryload(User.addresses),
+        )
+
+    def test_subqueryload_internal_lambda(self, plain_fixture):
+        User, Address = plain_fixture
+
+        s = Session()
+
+        def query(names):
+            stmt = (
+                select(lambda: User)
+                .where(lambda: User.name.in_(names))
+                .options(subqueryload(User.addresses))
+                .order_by(lambda: User.id)
+            )
+
+            return s.execute(stmt)
+
+        def go1():
+            r1 = query(["ed"])
+            eq_(
+                r1.scalars().all(),
+                [User(name="ed", addresses=[Address(), Address(), Address()])],
+            )
+
+        def go2():
+            r1 = query(["ed", "fred"])
+            eq_(
+                r1.scalars().all(),
+                [
+                    User(
+                        name="ed", addresses=[Address(), Address(), Address()]
+                    ),
+                    User(name="fred", addresses=[Address()]),
+                ],
+            )
+
+        for i in range(5):
+            fn = random.choice([go1, go2])
+            self.assert_sql_count(testing.db, fn, 2)
+
+    def test_subqueryload_external_lambda_caveats(self, plain_fixture):
+        User, Address = plain_fixture
+
+        s = Session()
+
+        def query(names):
+            stmt = lambda_stmt(
+                lambda: select(User)
+                .where(User.name.in_(names))
+                .options(subqueryload(User.addresses))
+            ) + (lambda s: s.order_by(User.id))
+
+            return s.execute(stmt)
+
+        def go1():
+            r1 = query(["ed"])
+            eq_(
+                r1.scalars().all(),
+                [User(name="ed", addresses=[Address(), Address(), Address()])],
+            )
+
+        def go2():
+            r1 = query(["ed", "fred"])
+            eq_(
+                r1.scalars().all(),
+                [
+                    User(
+                        name="ed", addresses=[Address(), Address(), Address()]
+                    ),
+                    User(name="fred", addresses=[Address()]),
+                ],
+            )
+
+        for i in range(5):
+            fn = random.choice([go1, go2])
+            with testing.expect_warnings(
+                'subqueryloader for "User.addresses" must invoke lambda '
+                r"callable at .*LambdaElement\(<code object <lambda> "
+                r".*test_lambdas.py.* in order to produce a new query, "
+                r"decreasing the efficiency of caching"
+            ):
+                self.assert_sql_count(testing.db, fn, 2)
+
+    def test_does_filter_aliasing_work(self, plain_fixture):
+        User, Address = plain_fixture
+
+        s = Session()
+
+        # aliased=True is to be deprecated, other filter lambdas
+        # that go into effect include polymorphic filtering.
+        q = (
+            s.query(lambda: User)
+            .join(lambda: User.addresses, aliased=True)
+            .filter(lambda: Address.email_address == "foo")
+        )
+        self.assert_compile(
+            q,
+            "SELECT users.id AS users_id, users.name AS users_name "
+            "FROM users JOIN addresses AS addresses_1 "
+            "ON users.id = addresses_1.user_id "
+            "WHERE addresses_1.email_address = :email_address_1",
+        )
+
+    @testing.combinations(
+        lambda s, User, Address: s.query(lambda: User).join(lambda: Address),
+        lambda s, User, Address: s.query(lambda: User).join(
+            lambda: User.addresses
+        ),
+        lambda s, User, Address: s.query(lambda: User).join(
+            lambda: Address, lambda: User.addresses
+        ),
+        lambda s, User, Address: s.query(lambda: User).join(
+            Address, lambda: User.addresses
+        ),
+        lambda s, User, Address: s.query(lambda: User).join(
+            lambda: Address, User.addresses
+        ),
+        lambda User, Address: select(lambda: User)
+        .join(lambda: Address)
+        .apply_labels(),
+        lambda User, Address: select(lambda: User)
+        .join(lambda: User.addresses)
+        .apply_labels(),
+        lambda User, Address: select(lambda: User)
+        .join(lambda: Address, lambda: User.addresses)
+        .apply_labels(),
+        lambda User, Address: select(lambda: User)
+        .join(Address, lambda: User.addresses)
+        .apply_labels(),
+        lambda User, Address: select(lambda: User)
+        .join(lambda: Address, User.addresses)
+        .apply_labels(),
+        argnames="test_case",
+    )
+    def test_join_entity_arg(self, plain_fixture, test_case):
+        User, Address = plain_fixture
+
+        s = Session()
+
+        stmt = testing.resolve_lambda(test_case, **locals())
+        self.assert_compile(
+            stmt,
+            "SELECT users.id AS users_id, users.name AS users_name "
+            "FROM users JOIN addresses ON users.id = addresses.user_id",
+        )
+
+
+class PolymorphicTest(_poly_fixtures._Polymorphic):
+    run_setup_mappers = "once"
+    __dialect__ = "default"
+
+    def test_join_second_prop_lambda(self):
+        Company = self.classes.Company
+        Manager = self.classes.Manager
+
+        s = Session()
+
+        q = s.query(Company).join(lambda: Manager, lambda: Company.employees)
+
+        self.assert_compile(
+            q,
+            "SELECT companies.company_id AS companies_company_id, "
+            "companies.name AS companies_name FROM companies "
+            "JOIN (people JOIN managers ON people.person_id = "
+            "managers.person_id) ON companies.company_id = people.company_id",
+        )
+
+
+class UpdateDeleteTest(fixtures.MappedTest):
+    __backend__ = True
+
+    run_setup_mappers = "once"
+
+    @classmethod
+    def define_tables(cls, metadata):
+        Table(
+            "users",
+            metadata,
+            Column(
+                "id", Integer, primary_key=True, test_needs_autoincrement=True
+            ),
+            Column("name", String(32)),
+            Column("age_int", Integer),
+        )
+        Table(
+            "addresses",
+            metadata,
+            Column("id", Integer, primary_key=True),
+            Column("user_id", ForeignKey("users.id")),
+        )
+
+    @classmethod
+    def setup_classes(cls):
+        class User(cls.Comparable):
+            pass
+
+        class Address(cls.Comparable):
+            pass
+
+    @classmethod
+    def insert_data(cls, connection):
+        users = cls.tables.users
+
+        connection.execute(
+            users.insert(),
+            [
+                dict(id=1, name="john", age_int=25),
+                dict(id=2, name="jack", age_int=47),
+                dict(id=3, name="jill", age_int=29),
+                dict(id=4, name="jane", age_int=37),
+            ],
+        )
+
+    @classmethod
+    def setup_mappers(cls):
+        User = cls.classes.User
+        users = cls.tables.users
+
+        Address = cls.classes.Address
+        addresses = cls.tables.addresses
+
+        mapper(
+            User,
+            users,
+            properties={
+                "age": users.c.age_int,
+                "addresses": relationship(Address),
+            },
+        )
+        mapper(Address, addresses)
+
+    def test_update(self):
+        User, Address = self.classes("User", "Address")
+
+        s = Session()
+
+        def go(ids, values):
+            stmt = lambda_stmt(lambda: update(User).where(User.id.in_(ids)))
+            s.execute(
+                stmt,
+                values,
+                # note this currently just unrolls the lambda on the statement.
+                # so lambda caching for updates is not actually that useful
+                # unless synchronize_session is turned off.
+                # evaluate is similar just doesn't work for IN yet.
+                execution_options={"synchronize_session": "fetch"},
+            )
+
+        go([1, 2], {"name": "jack2"})
+        eq_(
+            s.execute(select(User.id, User.name).order_by(User.id)).all(),
+            [(1, "jack2"), (2, "jack2"), (3, "jill"), (4, "jane")],
+        )
+
+        go([3], {"name": "jane2"})
+        eq_(
+            s.execute(select(User.id, User.name).order_by(User.id)).all(),
+            [(1, "jack2"), (2, "jack2"), (3, "jane2"), (4, "jane")],
+        )