Convert execution to move through Session

This patch replaces the ORM execution flow with a
single pathway through Session.execute() for all queries,
including Core and ORM.

Currently included is full support for ORM Query,
Query.from_statement(), select(), as well as the
baked query and horizontal shard systems.  Initial
changes have also been made to the dogpile caching
example, which like baked query makes use of a
new ORM-specific execution hook that replaces the
use of both QueryEvents.before_compile() as well
as Query._execute_and_instances() as the central
ORM interception hooks.

select() and Query() constructs alike can be passed to
Session.execute() where they will return ORM
results in a Results object.   This API is currently
used internally by Query.   Full support for
Session.execute()->results to behave in a fully
2.0 fashion will be in later changesets.

bulk update/delete with ORM support will also
be delivered via the update() and delete()
constructs, however these have not yet been adapted
to the new system and may follow in a subsequent
update.

Performance is also beginning to lag as of this
commit and some previous ones.   It is hoped that
a few central functions such as the coercions
functions can be rewritten in C to re-gain
performance.  Additionally, query caching
is now available and some subsequent patches
will attempt to cache more of the per-execution
work from the ORM layer, e.g. column getters
and adapters.

This patch also contains initial "turn on" of the
caching system enginewide via the query_cache_size
parameter to create_engine(). Still defaulting at
zero for "no caching".   The caching system still
needs adjustments in order to gain adequate performance.

Change-Id: I047a7ebb26aa85dc01f6789fac2bff561dcd555d

2 files changed, 204 insertions, 78 deletions

diff --git a/test/ext/test_baked.py b/test/ext/test_baked.py
index 77e57aa36..ecb5e3919 100644
--- a/test/ext/test_baked.py
+++ b/test/ext/test_baked.py
@@ -289,6 +289,7 @@ class LikeQueryTest(BakedTest):
         # with multiple params, the **kwargs will be used
         bq += lambda q: q.filter(User.id == bindparam("anid"))
         eq_(bq(sess).params(uname="fred", anid=9).count(), 1)
+
         eq_(
             # wrong id, so 0 results:
             bq(sess).params(uname="fred", anid=8).count(),
@@ -388,7 +389,12 @@ class ResultPostCriteriaTest(BakedTest):
             def before_execute(
                 conn, clauseelement, multiparams, params, execution_options
             ):
-                assert "yes" in conn._execution_options
+                # execution options are kind of moving around a bit,
+                # test both places
+                assert (
+                    "yes" in clauseelement._execution_options
+                    or "yes" in execution_options
+                )
 
             bq = self.bakery(lambda s: s.query(User.id).order_by(User.id))
 
@@ -804,9 +810,7 @@ class ResultTest(BakedTest):
         Address = self.classes.Address
         Order = self.classes.Order
 
-        # Override the default bakery for one with a smaller size. This used to
-        # trigger a bug when unbaking subqueries.
-        self.bakery = baked.bakery(size=3)
+        self.bakery = baked.bakery()
         base_bq = self.bakery(lambda s: s.query(User))
 
         base_bq += lambda q: q.options(
@@ -840,6 +844,7 @@ class ResultTest(BakedTest):
             for cond1, cond2 in itertools.product(
                 *[(False, True) for j in range(2)]
             ):
+                print("HI----")
                 bq = base_bq._clone()
 
                 sess = Session()
@@ -903,7 +908,7 @@ class ResultTest(BakedTest):
             )
         ]
 
-        self.bakery = baked.bakery(size=3)
+        self.bakery = baked.bakery()
 
         bq = self.bakery(lambda s: s.query(User))
 
@@ -1288,33 +1293,72 @@ class LazyLoaderTest(testing.AssertsCompiledSQL, BakedTest):
 
     def _test_baked_lazy_loading_relationship_flag(self, flag):
         User, Address = self._o2m_fixture(bake_queries=flag)
+        from sqlalchemy import inspect
 
-        sess = Session()
-        u1 = sess.query(User).first()
-
-        from sqlalchemy.orm import Query
-
-        canary = mock.Mock()
+        address_mapper = inspect(Address)
+        sess = Session(testing.db)
+
+        # there's no event in the compile process either at the ORM
+        # or core level and it is not easy to patch.  the option object
+        # is the one thing that will get carried into the lazyload from the
+        # outside and invoked on a per-compile basis
+        mock_opt = mock.Mock(
+            _is_compile_state=True,
+            propagate_to_loaders=True,
+            _gen_cache_key=lambda *args: ("hi",),
+            _generate_path_cache_key=lambda path: ("hi",),
+        )
 
-        # I would think Mock can do this but apparently
-        # it cannot (wrap / autospec don't work together)
-        real_compile_state = Query._compile_state
+        u1 = sess.query(User).options(mock_opt).first()
 
-        def _my_compile_state(*arg, **kw):
-            if arg[0].column_descriptions[0]["entity"] is Address:
-                canary()
-            return real_compile_state(*arg, **kw)
+        @event.listens_for(sess, "do_orm_execute")
+        def _my_compile_state(context):
+            if (
+                context.statement._raw_columns[0]._annotations["parententity"]
+                is address_mapper
+            ):
+                mock_opt.orm_execute()
 
-        with mock.patch.object(Query, "_compile_state", _my_compile_state):
-            u1.addresses
+        u1.addresses
 
-            sess.expire(u1)
-            u1.addresses
+        sess.expire(u1)
+        u1.addresses
 
         if flag:
-            eq_(canary.call_count, 1)
+            eq_(
+                mock_opt.mock_calls,
+                [
+                    mock.call.process_query(mock.ANY),
+                    mock.call.process_compile_state(mock.ANY),  # query.first()
+                    mock.call.process_query_conditionally(mock.ANY),
+                    mock.call.orm_execute(),  # lazyload addresses
+                    mock.call.process_compile_state(mock.ANY),  # emit lazyload
+                    mock.call.process_compile_state(
+                        mock.ANY
+                    ),  # load scalar attributes for user
+                    # lazyload addresses, no call to process_compile_state
+                    mock.call.orm_execute(),
+                ],
+            )
         else:
-            eq_(canary.call_count, 2)
+            eq_(
+                mock_opt.mock_calls,
+                [
+                    mock.call.process_query(mock.ANY),
+                    mock.call.process_compile_state(mock.ANY),  # query.first()
+                    mock.call.process_query_conditionally(mock.ANY),
+                    mock.call.orm_execute(),  # lazyload addresses
+                    mock.call.process_compile_state(mock.ANY),  # emit_lazyload
+                    mock.call.process_compile_state(
+                        mock.ANY
+                    ),  # load_scalar_attributes for user
+                    mock.call.process_query_conditionally(mock.ANY),
+                    mock.call.orm_execute(),  # lazyload addresses
+                    mock.call.process_compile_state(
+                        mock.ANY
+                    ),  # emit_lazyload, here the query was not cached
+                ],
+            )
 
     def test_baked_lazy_loading_option_o2m(self):
         User, Address = self._o2m_fixture()
@@ -1571,58 +1615,57 @@ class CustomIntegrationTest(testing.AssertsCompiledSQL, BakedTest):
         return User, Address
 
     def _query_fixture(self):
-        from sqlalchemy.orm.query import Query, _generative
+        from sqlalchemy.orm.query import Query
 
         class CachingQuery(Query):
             cache = {}
 
-            @_generative
             def set_cache_key(self, key):
-                self._cache_key = key
-
-            # in 1.4 / If1a23824ffb77d8d58cf2338cf35dd6b5963b17f ,
-            # we no longer override ``__iter__`` because we need the
-            # whole result object.  The FrozenResult is added for this
-            # use case.   A new session-level event will be added within
-            # the scope of ORM /execute() integration so that people
-            # don't have to subclass this anymore.
-
-            def _execute_and_instances(self, context, **kw):
-                super_ = super(CachingQuery, self)
-
-                if hasattr(self, "_cache_key"):
-                    return self.get_value(
-                        createfunc=lambda: super_._execute_and_instances(
-                            context, **kw
-                        )
-                    )
-                else:
-                    return super_._execute_and_instances(context, **kw)
-
-            def get_value(self, createfunc):
-                if self._cache_key in self.cache:
-                    return self.cache[self._cache_key]()
-                else:
-                    self.cache[
-                        self._cache_key
-                    ] = retval = createfunc().freeze()
-                    return retval()
+                return self.execution_options(_cache_key=key)
+
+            def set_cache_key_for_path(self, path, key):
+                return self.execution_options(**{"_cache_key_%s" % path: key})
+
+        def get_value(cache_key, cache, createfunc):
+            if cache_key in cache:
+                return cache[cache_key]()
+            else:
+                cache[cache_key] = retval = createfunc().freeze()
+                return retval()
+
+        s1 = Session(query_cls=CachingQuery)
+
+        @event.listens_for(s1, "do_orm_execute", retval=True)
+        def do_orm_execute(orm_context):
+            ckey = None
+            statement = orm_context.orm_query
+            for opt in orm_context.user_defined_options:
+                ckey = opt.get_cache_key(orm_context)
+                if ckey:
+                    break
+            else:
+                if "_cache_key" in statement._execution_options:
+                    ckey = statement._execution_options["_cache_key"]
+
+            if ckey is not None:
+                return get_value(
+                    ckey, CachingQuery.cache, orm_context.invoke_statement,
+                )
 
-        return Session(query_cls=CachingQuery)
+        return s1
 
     def _option_fixture(self):
-        from sqlalchemy.orm.interfaces import MapperOption
+        from sqlalchemy.orm.interfaces import UserDefinedOption
 
-        class RelationshipCache(MapperOption):
+        class RelationshipCache(UserDefinedOption):
 
             propagate_to_loaders = True
 
-            def process_query_conditionally(self, query):
-                if query._current_path:
-                    query._cache_key = "user7_addresses"
-
-            def _generate_path_cache_key(self, path):
-                return None
+            def get_cache_key(self, orm_context):
+                if orm_context.loader_strategy_path:
+                    return "user7_addresses"
+                else:
+                    return None
 
         return RelationshipCache()
 
@@ -1641,6 +1684,21 @@ class CustomIntegrationTest(testing.AssertsCompiledSQL, BakedTest):
 
         eq_(q.all(), [User(id=7, addresses=[Address(id=1)])])
 
+    def test_non_baked_tuples(self):
+        User, Address = self._o2m_fixture()
+
+        sess = self._query_fixture()
+        q = sess._query_cls
+        eq_(q.cache, {})
+
+        q = sess.query(User).filter(User.id == 7).set_cache_key("user7")
+
+        eq_(sess.execute(q).all(), [(User(id=7, addresses=[Address(id=1)]),)])
+
+        eq_(list(q.cache), ["user7"])
+
+        eq_(sess.execute(q).all(), [(User(id=7, addresses=[Address(id=1)]),)])
+
     def test_use_w_baked(self):
         User, Address = self._o2m_fixture()
 
diff --git a/test/ext/test_horizontal_shard.py b/test/ext/test_horizontal_shard.py
index 77b716b0a..eb9c5147a 100644
--- a/test/ext/test_horizontal_shard.py
+++ b/test/ext/test_horizontal_shard.py
@@ -15,6 +15,7 @@ from sqlalchemy import Table
 from sqlalchemy import testing
 from sqlalchemy import util
 from sqlalchemy.ext.horizontal_shard import ShardedSession
+from sqlalchemy.future import select as future_select
 from sqlalchemy.orm import clear_mappers
 from sqlalchemy.orm import create_session
 from sqlalchemy.orm import deferred
@@ -27,11 +28,11 @@ from sqlalchemy.pool import SingletonThreadPool
 from sqlalchemy.sql import operators
 from sqlalchemy.testing import eq_
 from sqlalchemy.testing import fixtures
+from sqlalchemy.testing import is_
 from sqlalchemy.testing import provision
 from sqlalchemy.testing.engines import testing_engine
 from sqlalchemy.testing.engines import testing_reaper
 
-
 # TODO: ShardTest can be turned into a base for further subclasses
 
 
@@ -190,11 +191,45 @@ class ShardTest(object):
         sess.close()
         return sess
 
-    def test_roundtrip(self):
+    def test_get(self):
         sess = self._fixture_data()
-        tokyo = sess.query(WeatherLocation).filter_by(city="Tokyo").one()
-        tokyo.city  # reload 'city' attribute on tokyo
-        sess.expire_all()
+        tokyo = sess.query(WeatherLocation).get(1)
+        eq_(tokyo.city, "Tokyo")
+
+        newyork = sess.query(WeatherLocation).get(2)
+        eq_(newyork.city, "New York")
+
+        t2 = sess.query(WeatherLocation).get(1)
+        is_(t2, tokyo)
+
+    def test_get_explicit_shard(self):
+        sess = self._fixture_data()
+        tokyo = sess.query(WeatherLocation).set_shard("europe").get(1)
+        is_(tokyo, None)
+
+        newyork = sess.query(WeatherLocation).set_shard("north_america").get(2)
+        eq_(newyork.city, "New York")
+
+        # now it found it
+        t2 = sess.query(WeatherLocation).get(1)
+        eq_(t2.city, "Tokyo")
+
+    def test_query_explicit_shard_via_bind_opts(self):
+        sess = self._fixture_data()
+
+        stmt = future_select(WeatherLocation).filter(WeatherLocation.id == 1)
+
+        tokyo = (
+            sess.execute(stmt, bind_arguments={"shard_id": "asia"})
+            .scalars()
+            .first()
+        )
+
+        eq_(tokyo.city, "Tokyo")
+
+    def test_plain_db_lookup(self):
+        self._fixture_data()
+        # not sure what this is testing except the fixture data itself
         eq_(
             db2.execute(weather_locations.select()).fetchall(),
             [(1, "Asia", "Tokyo")],
@@ -206,12 +241,45 @@ class ShardTest(object):
                 (3, "North America", "Toronto"),
             ],
         )
+
+    def test_plain_core_lookup_w_shard(self):
+        sess = self._fixture_data()
         eq_(
             sess.execute(
                 weather_locations.select(), shard_id="asia"
             ).fetchall(),
             [(1, "Asia", "Tokyo")],
         )
+
+    def test_roundtrip_future(self):
+        sess = self._fixture_data()
+
+        tokyo = (
+            sess.execute(
+                future_select(WeatherLocation).filter_by(city="Tokyo")
+            )
+            .scalars()
+            .one()
+        )
+        eq_(tokyo.city, "Tokyo")
+
+        asia_and_europe = sess.execute(
+            future_select(WeatherLocation).filter(
+                WeatherLocation.continent.in_(["Europe", "Asia"])
+            )
+        ).scalars()
+        eq_(
+            {c.city for c in asia_and_europe}, {"Tokyo", "London", "Dublin"},
+        )
+
+    def test_roundtrip(self):
+        sess = self._fixture_data()
+        tokyo = sess.query(WeatherLocation).filter_by(city="Tokyo").one()
+
+        eq_(tokyo.city, "Tokyo")
+        tokyo.city  # reload 'city' attribute on tokyo
+        sess.expire_all()
+
         t = sess.query(WeatherLocation).get(tokyo.id)
         eq_(t.city, tokyo.city)
         eq_(t.reports[0].temperature, 80.0)
@@ -219,26 +287,23 @@ class ShardTest(object):
             WeatherLocation.continent == "North America"
         )
         eq_(
-            set([c.city for c in north_american_cities]),
-            set(["New York", "Toronto"]),
+            {c.city for c in north_american_cities}, {"New York", "Toronto"},
         )
         asia_and_europe = sess.query(WeatherLocation).filter(
             WeatherLocation.continent.in_(["Europe", "Asia"])
         )
         eq_(
-            set([c.city for c in asia_and_europe]),
-            set(["Tokyo", "London", "Dublin"]),
+            {c.city for c in asia_and_europe}, {"Tokyo", "London", "Dublin"},
         )
 
         # inspect the shard token stored with each instance
         eq_(
-            set(inspect(c).key[2] for c in asia_and_europe),
-            set(["europe", "asia"]),
+            {inspect(c).key[2] for c in asia_and_europe}, {"europe", "asia"},
         )
 
         eq_(
-            set(inspect(c).identity_token for c in asia_and_europe),
-            set(["europe", "asia"]),
+            {inspect(c).identity_token for c in asia_and_europe},
+            {"europe", "asia"},
         )
 
         newyork = sess.query(WeatherLocation).filter_by(city="New York").one()
@@ -324,7 +389,7 @@ class ShardTest(object):
         canary = []
 
         def load(instance, ctx):
-            canary.append(ctx.attributes["shard_id"])
+            canary.append(ctx.bind_arguments["shard_id"])
 
         event.listen(WeatherLocation, "load", load)
         sess = self._fixture_data()
@@ -571,6 +636,9 @@ class RefreshDeferExpireTest(fixtures.DeclarativeMappedTest):
         s.commit()
 
     def _session_fixture(self, **kw):
+        # the "fake" key here is to ensure that neither id_chooser
+        # nor query_chooser are actually used, only shard_chooser
+        # should be used.
 
         return ShardedSession(
             shards={"main": testing.db},