Turn on caching everywhere, add logging

A variety of caching issues found by running
all tests with statement caching turned on.

The cache system now has a more conservative approach where
any subclass of a SQL element will by default invalidate
the cache key unless it adds the flag inherit_cache=True
at the class level, or if it implements its own caching.

Add working caching to a few elements that were
omitted previously; fix some caching implementations
to suit lesser used edge cases such as json casts
and array slices.

Refine the way BaseCursorResult and CursorMetaData
interact with caching; to suit cases like Alembic
modifying table structures, don't cache the
cursor metadata if it were created against a
cursor.description using non-positional matching,
e.g. "select *".   if a table re-ordered its columns
or added/removed, now that data is obsolete.

Additionally we have to adapt the cursor metadata
_keymap regardless of if we just processed
cursor.description, because if we ran against
a cached SQLCompiler we won't have the right
columns in _keymap.

Other refinements to how and when we do this
adaption as some weird cases
were exposed in the Postgresql dialect,
a text() construct that names just one column that
is not actually in the statement.   Fixed that
also as it looks like a cut-and-paste artifact
that doesn't actually affect anything.

Various issues with re-use of compiled result maps
and cursor metadata in conjunction with tables being
changed, such as change in order of columns.

mappers can be cleared but the class remains, meaning
a mapper has to use itself as the cache key not the class.

lots of bound parameter / literal issues, due to Alembic
creating a straight subclass of bindparam that renders
inline directly.   While we can update Alembic to not
do this, we have to assume other people might be doing
this, so bindparam() implements the inherit_cache=True
logic as well that was a bit involved.

turn on cache stats in logging.

Includes a fix to subqueryloader which moves all setup to
the create_row_processor() phase and elminates any storage
within the compiled context.   This includes some changes
to create_row_processor() signature and a revising of the
technique used to determine if the loader can participate
in polymorphic queries, which is also applied to
selectinloading.

DML update.values() and ordered_values() now coerces the
keys as we have tests that pass an arbitrary class here
which only includes __clause_element__(), so the
key can't be cached unless it is coerced.  this in turn
changed how composite attributes support bulk update
to use the standard approach of ClauseElement with
annotations that are parsed in the ORM context.

memory profiling successfully caught that the Session
from Query was getting passed into _statement_20()
so that was a big win for that test suite.

Apparently Compiler had .execute() and .scalar() methods
stuck on it, these date back to version 0.4 and there
was a single test in the PostgreSQL dialect tests
that exercised it for no apparent reason.   Removed
these methods as well as the concept of a Compiler
holding onto a "bind".

Fixes: #5386

Change-Id: I990b43aab96b42665af1b2187ad6020bee778784

4 files changed, 300 insertions, 8 deletions

diff --git a/test/sql/test_compare.py b/test/sql/test_compare.py
index 3f74bdbcc..6aaafc716 100644
--- a/test/sql/test_compare.py
+++ b/test/sql/test_compare.py
@@ -13,6 +13,7 @@ from sqlalchemy import exists
 from sqlalchemy import extract
 from sqlalchemy import Float
 from sqlalchemy import Integer
+from sqlalchemy import literal_column
 from sqlalchemy import MetaData
 from sqlalchemy import or_
 from sqlalchemy import select
@@ -43,6 +44,7 @@ from sqlalchemy.sql.base import HasCacheKey
 from sqlalchemy.sql.elements import _label_reference
 from sqlalchemy.sql.elements import _textual_label_reference
 from sqlalchemy.sql.elements import Annotated
+from sqlalchemy.sql.elements import BindParameter
 from sqlalchemy.sql.elements import ClauseElement
 from sqlalchemy.sql.elements import ClauseList
 from sqlalchemy.sql.elements import CollationClause
@@ -68,6 +70,8 @@ from sqlalchemy.testing import is_not_
 from sqlalchemy.testing import is_true
 from sqlalchemy.testing import ne_
 from sqlalchemy.testing.util import random_choices
+from sqlalchemy.types import ARRAY
+from sqlalchemy.types import JSON
 from sqlalchemy.util import class_hierarchy
 
 meta = MetaData()
@@ -188,6 +192,10 @@ class CoreFixtures(object):
             column("q").like("somstr", escape="X"),
         ),
         lambda: (
+            column("q", ARRAY(Integer))[3] == 5,
+            column("q", ARRAY(Integer))[3:5] == 5,
+        ),
+        lambda: (
             table_a.c.a,
             table_a.c.a._annotate({"orm": True}),
             table_a.c.a._annotate({"orm": True})._annotate({"bar": False}),
@@ -234,6 +242,15 @@ class CoreFixtures(object):
             cast(column("p"), Integer),
         ),
         lambda: (
+            column("x", JSON)["key1"],
+            column("x", JSON)["key1"].as_boolean(),
+            column("x", JSON)["key1"].as_float(),
+            column("x", JSON)["key1"].as_integer(),
+            column("x", JSON)["key1"].as_string(),
+            column("y", JSON)["key1"].as_integer(),
+            column("y", JSON)["key1"].as_string(),
+        ),
+        lambda: (
             bindparam("x"),
             bindparam("y"),
             bindparam("x", type_=Integer),
@@ -334,6 +351,11 @@ class CoreFixtures(object):
             select([table_a.c.a]),
             select([table_a.c.a, table_a.c.b]),
             select([table_a.c.b, table_a.c.a]),
+            select([table_a.c.b, table_a.c.a]).limit(5),
+            select([table_a.c.b, table_a.c.a]).limit(5).offset(10),
+            select([table_a.c.b, table_a.c.a])
+            .limit(literal_column("foobar"))
+            .offset(10),
             select([table_a.c.b, table_a.c.a]).apply_labels(),
             select([table_a.c.a]).where(table_a.c.b == 5),
             select([table_a.c.a])
@@ -827,18 +849,21 @@ class CacheKeyFixture(object):
                     ne_(a_key.key, b_key.key)
 
             # ClauseElement-specific test to ensure the cache key
-            # collected all the bound parameters
+            # collected all the bound parameters that aren't marked
+            # as "literal execute"
             if isinstance(case_a[a], ClauseElement) and isinstance(
                 case_b[b], ClauseElement
             ):
                 assert_a_params = []
                 assert_b_params = []
-                visitors.traverse(
-                    case_a[a], {}, {"bindparam": assert_a_params.append}
-                )
-                visitors.traverse(
-                    case_b[b], {}, {"bindparam": assert_b_params.append}
-                )
+
+                for elem in visitors.iterate(case_a[a]):
+                    if elem.__visit_name__ == "bindparam":
+                        assert_a_params.append(elem)
+
+                for elem in visitors.iterate(case_b[b]):
+                    if elem.__visit_name__ == "bindparam":
+                        assert_b_params.append(elem)
 
                 # note we're asserting the order of the params as well as
                 # if there are dupes or not.  ordering has to be
@@ -907,6 +932,39 @@ class CacheKeyTest(CacheKeyFixture, CoreFixtures, fixtures.TestBase):
             for fixture in fixtures_:
                 self._run_cache_key_fixture(fixture, compare_values)
 
+    def test_literal_binds(self):
+        def fixture():
+            return (
+                bindparam(None, value="x", literal_execute=True),
+                bindparam(None, value="y", literal_execute=True),
+            )
+
+        self._run_cache_key_fixture(
+            fixture, True,
+        )
+
+    def test_bindparam_subclass_nocache(self):
+        # does not implement inherit_cache
+        class _literal_bindparam(BindParameter):
+            pass
+
+        l1 = _literal_bindparam(None, value="x1")
+        is_(l1._generate_cache_key(), None)
+
+    def test_bindparam_subclass_ok_cache(self):
+        # implements inherit_cache
+        class _literal_bindparam(BindParameter):
+            inherit_cache = True
+
+        def fixture():
+            return (
+                _literal_bindparam(None, value="x1"),
+                _literal_bindparam(None, value="x2"),
+                _literal_bindparam(None),
+            )
+
+        self._run_cache_key_fixture(fixture, True)
+
     def test_cache_key_unknown_traverse(self):
         class Foobar1(ClauseElement):
             _traverse_internals = [
@@ -1264,6 +1322,30 @@ class CompareClausesTest(fixtures.TestBase):
 
         is_false(l1.compare(l2))
 
+    def test_cache_key_limit_offset_values(self):
+        s1 = select([column("q")]).limit(10)
+        s2 = select([column("q")]).limit(25)
+        s3 = select([column("q")]).limit(25).offset(5)
+        s4 = select([column("q")]).limit(25).offset(18)
+        s5 = select([column("q")]).limit(7).offset(12)
+        s6 = select([column("q")]).limit(literal_column("q")).offset(12)
+
+        for should_eq_left, should_eq_right in [(s1, s2), (s3, s4), (s3, s5)]:
+            eq_(
+                should_eq_left._generate_cache_key().key,
+                should_eq_right._generate_cache_key().key,
+            )
+
+        for shouldnt_eq_left, shouldnt_eq_right in [
+            (s1, s3),
+            (s5, s6),
+            (s2, s3),
+        ]:
+            ne_(
+                shouldnt_eq_left._generate_cache_key().key,
+                shouldnt_eq_right._generate_cache_key().key,
+            )
+
     def test_compare_labels(self):
         is_true(column("q").label(None).compare(column("q").label(None)))
 
diff --git a/test/sql/test_compiler.py b/test/sql/test_compiler.py
index 9881d1247..c12543f82 100644
--- a/test/sql/test_compiler.py
+++ b/test/sql/test_compiler.py
@@ -75,6 +75,7 @@ from sqlalchemy.sql import elements
 from sqlalchemy.sql import label
 from sqlalchemy.sql import operators
 from sqlalchemy.sql import table
+from sqlalchemy.sql import util as sql_util
 from sqlalchemy.sql.elements import BooleanClauseList
 from sqlalchemy.sql.expression import ClauseList
 from sqlalchemy.sql.expression import HasPrefixes
@@ -85,7 +86,9 @@ from sqlalchemy.testing import eq_
 from sqlalchemy.testing import eq_ignore_whitespace
 from sqlalchemy.testing import fixtures
 from sqlalchemy.testing import is_
+from sqlalchemy.testing import is_true
 from sqlalchemy.testing import mock
+from sqlalchemy.testing import ne_
 from sqlalchemy.util import u
 
 table1 = table(
@@ -3493,6 +3496,140 @@ class BindParameterTest(AssertsCompiledSQL, fixtures.TestBase):
             extracted_parameters=s1_cache_key[1],
         )
 
+    def test_construct_params_w_bind_clones_post(self):
+        """test that a BindParameter that has been cloned after the cache
+        key was generated still matches up when construct_params()
+        is called with an extracted parameter collection.
+
+        This case occurs now with the ORM as the ORM construction will
+        frequently run clause adaptation on elements of the statement within
+        compilation, after the cache key has been generated.  this adaptation
+        hits BindParameter objects which will change their key as they
+        will usually have unqique=True.   So the construct_params() process
+        when it links its internal bind_names to the cache key binds,
+        must do this badsed on bindparam._identifying_key, which does not
+        change across clones, rather than .key which usually will.
+
+        """
+
+        stmt = select([table1.c.myid]).where(table1.c.myid == 5)
+
+        # get the original bindparam.
+        original_bind = stmt._where_criteria[0].right
+
+        # it's anonymous so unique=True
+        is_true(original_bind.unique)
+
+        # cache key against hte original param
+        cache_key = stmt._generate_cache_key()
+
+        # now adapt the statement
+        stmt_adapted = sql_util.ClauseAdapter(table1).traverse(stmt)
+
+        # new bind parameter has a different key but same
+        # identifying key
+        new_bind = stmt_adapted._where_criteria[0].right
+        eq_(original_bind._identifying_key, new_bind._identifying_key)
+        ne_(original_bind.key, new_bind.key)
+
+        # compile the adapted statement but set the cache key to the one
+        # generated from the unadapted statement.  this will look like
+        # when the ORM runs clause adaption inside of visit_select, after
+        # the cache key is generated but before the compiler is given the
+        # core select statement to actually render.
+        compiled = stmt_adapted.compile(cache_key=cache_key)
+
+        # params set up as 5
+        eq_(compiled.construct_params(params={},), {"myid_1": 5})
+
+        # also works w the original cache key
+        eq_(
+            compiled.construct_params(
+                params={}, extracted_parameters=cache_key[1]
+            ),
+            {"myid_1": 5},
+        )
+
+        # now make a totally new statement with the same cache key
+        new_stmt = select([table1.c.myid]).where(table1.c.myid == 10)
+        new_cache_key = new_stmt._generate_cache_key()
+
+        # cache keys match
+        eq_(cache_key.key, new_cache_key.key)
+
+        # ensure we get "10" from construct params.   if it matched
+        # based on .key and not ._identifying_key, it would not see that
+        # the bind parameter is part of the cache key.
+        eq_(
+            compiled.construct_params(
+                params={}, extracted_parameters=new_cache_key[1]
+            ),
+            {"myid_1": 10},
+        )
+
+    def test_construct_params_w_bind_clones_pre(self):
+        """test that a BindParameter that has been cloned before the cache
+        key was generated, and was doubled up just to make sure it has to
+        be unique, still matches up when construct_params()
+        is called with an extracted parameter collection.
+
+        other ORM feaures like optimized_compare() end up doing something
+        like this, such as if there are multiple "has()" or "any()" which would
+        have cloned the join condition and changed the values of bound
+        parameters.
+
+        """
+
+        stmt = select([table1.c.myid]).where(table1.c.myid == 5)
+
+        original_bind = stmt._where_criteria[0].right
+        # it's anonymous so unique=True
+        is_true(original_bind.unique)
+
+        b1 = original_bind._clone()
+        b1.value = 10
+        b2 = original_bind._clone()
+        b2.value = 12
+
+        # make a new statement that uses the clones as distinct
+        # parameters
+        modified_stmt = select([table1.c.myid]).where(
+            or_(table1.c.myid == b1, table1.c.myid == b2)
+        )
+
+        cache_key = modified_stmt._generate_cache_key()
+        compiled = modified_stmt.compile(cache_key=cache_key)
+
+        eq_(
+            compiled.construct_params(params={}), {"myid_1": 10, "myid_2": 12},
+        )
+
+        # make a new statement doing the same thing and make sure
+        # the binds match up correctly
+        new_stmt = select([table1.c.myid]).where(table1.c.myid == 8)
+
+        new_original_bind = new_stmt._where_criteria[0].right
+        new_b1 = new_original_bind._clone()
+        new_b1.value = 20
+        new_b2 = new_original_bind._clone()
+        new_b2.value = 18
+        modified_new_stmt = select([table1.c.myid]).where(
+            or_(table1.c.myid == new_b1, table1.c.myid == new_b2)
+        )
+
+        new_cache_key = modified_new_stmt._generate_cache_key()
+
+        # cache keys match
+        eq_(cache_key.key, new_cache_key.key)
+
+        # ensure we get both values
+        eq_(
+            compiled.construct_params(
+                params={}, extracted_parameters=new_cache_key[1]
+            ),
+            {"myid_1": 20, "myid_2": 18},
+        )
+
     def test_tuple_expanding_in_no_values(self):
         expr = tuple_(table1.c.myid, table1.c.name).in_(
             [(1, "foo"), (5, "bar")]
diff --git a/test/sql/test_operators.py b/test/sql/test_operators.py
index af01d9a3d..4c5c51b4a 100644
--- a/test/sql/test_operators.py
+++ b/test/sql/test_operators.py
@@ -113,11 +113,17 @@ class DefaultColumnComparatorTest(fixtures.TestBase):
     def test_operate(self, operator, right):
         left = column("left")
 
+        if operators.is_comparison(operator):
+            type_ = sqltypes.BOOLEANTYPE
+        else:
+            type_ = sqltypes.NULLTYPE
+
         assert left.comparator.operate(operator, right).compare(
             BinaryExpression(
                 coercions.expect(roles.WhereHavingRole, left),
                 coercions.expect(roles.WhereHavingRole, right),
                 operator,
+                type_=type_,
             )
         )
 
@@ -129,6 +135,7 @@ class DefaultColumnComparatorTest(fixtures.TestBase):
                 coercions.expect(roles.WhereHavingRole, right),
                 operator,
                 modifiers=modifiers,
+                type_=type_,
             )
         )
 
@@ -167,6 +174,7 @@ class DefaultColumnComparatorTest(fixtures.TestBase):
                     "left", value=[1, 2, 3], unique=True, expanding=True
                 ),
                 operators.in_op,
+                type_=sqltypes.BOOLEANTYPE,
             )
         )
         self._loop_test(operators.in_op, [1, 2, 3])
@@ -180,6 +188,7 @@ class DefaultColumnComparatorTest(fixtures.TestBase):
                     "left", value=[1, 2, 3], unique=True, expanding=True
                 ),
                 operators.notin_op,
+                type_=sqltypes.BOOLEANTYPE,
             )
         )
         self._loop_test(operators.notin_op, [1, 2, 3])
diff --git a/test/sql/test_update.py b/test/sql/test_update.py
index 22c4b1743..664862dcb 100644
--- a/test/sql/test_update.py
+++ b/test/sql/test_update.py
@@ -111,10 +111,46 @@ class UpdateTest(_UpdateFromTestBase, fixtures.TablesTest, AssertsCompiledSQL):
     def test_update_literal_binds(self):
         table1 = self.tables.mytable
 
+        stmt = (
+            table1.update().values(name="jack").where(table1.c.name == "jill")
+        )
+
+        self.assert_compile(
+            stmt,
+            "UPDATE mytable SET name='jack' WHERE mytable.name = 'jill'",
+            literal_binds=True,
+        )
+
+    def test_update_custom_key_thing(self):
         table1 = self.tables.mytable
 
+        class Thing(object):
+            def __clause_element__(self):
+                return table1.c.name
+
         stmt = (
-            table1.update().values(name="jack").where(table1.c.name == "jill")
+            table1.update()
+            .values({Thing(): "jack"})
+            .where(table1.c.name == "jill")
+        )
+
+        self.assert_compile(
+            stmt,
+            "UPDATE mytable SET name='jack' WHERE mytable.name = 'jill'",
+            literal_binds=True,
+        )
+
+    def test_update_ordered_custom_key_thing(self):
+        table1 = self.tables.mytable
+
+        class Thing(object):
+            def __clause_element__(self):
+                return table1.c.name
+
+        stmt = (
+            table1.update()
+            .ordered_values((Thing(), "jack"))
+            .where(table1.c.name == "jill")
         )
 
         self.assert_compile(
@@ -123,6 +159,34 @@ class UpdateTest(_UpdateFromTestBase, fixtures.TablesTest, AssertsCompiledSQL):
             literal_binds=True,
         )
 
+    def test_update_broken_custom_key_thing(self):
+        table1 = self.tables.mytable
+
+        class Thing(object):
+            def __clause_element__(self):
+                return 5
+
+        assert_raises_message(
+            exc.ArgumentError,
+            "SET/VALUES column expression or string key expected, got .*Thing",
+            table1.update().values,
+            {Thing(): "jack"},
+        )
+
+    def test_update_ordered_broken_custom_key_thing(self):
+        table1 = self.tables.mytable
+
+        class Thing(object):
+            def __clause_element__(self):
+                return 5
+
+        assert_raises_message(
+            exc.ArgumentError,
+            "SET/VALUES column expression or string key expected, got .*Thing",
+            table1.update().ordered_values,
+            (Thing(), "jack"),
+        )
+
     def test_correlated_update_one(self):
         table1 = self.tables.mytable