Unify Query and select() , move all processing to compile phase

Convert Query to do virtually all compile state computation
in the _compile_context() phase, and organize it all
such that a plain select() construct may also be used as the
source of information in order to generate ORM query state.
This makes it such that Query is not needed except for
its additional methods like from_self() which are all to
be deprecated.

The construction of ORM state will occur beyond the
caching boundary when the new execution model is integrated.

future select() gains a working join() and filter_by() method.
as we continue to rebase and merge each commit in the steps,
callcounts continue to bump around.  will have to look at
the final result when it's all in.

References: #5159
References: #4705
References: #4639
References: #4871
References: #5010

Change-Id: I19e05b3424b07114cce6c439b05198ac47f7ac10

4 files changed, 372 insertions, 15 deletions

diff --git a/test/sql/test_compare.py b/test/sql/test_compare.py
index 8cc7b7fb6..247332d8c 100644
--- a/test/sql/test_compare.py
+++ b/test/sql/test_compare.py
@@ -28,6 +28,7 @@ from sqlalchemy import util
 from sqlalchemy import values
 from sqlalchemy.dialects import mysql
 from sqlalchemy.dialects import postgresql
+from sqlalchemy.future import select as future_select
 from sqlalchemy.schema import Sequence
 from sqlalchemy.sql import bindparam
 from sqlalchemy.sql import ColumnElement
@@ -91,6 +92,18 @@ table_c = Table("c", meta, Column("x", Integer), Column("y", Integer))
 table_d = Table("d", meta, Column("y", Integer), Column("z", Integer))
 
 
+def opt1(ctx):
+    pass
+
+
+def opt2(ctx):
+    pass
+
+
+def opt3(ctx):
+    pass
+
+
 class MyEntity(HasCacheKey):
     def __init__(self, name, element):
         self.name = name
@@ -326,6 +339,28 @@ class CoreFixtures(object):
             .correlate_except(table_b),
         ),
         lambda: (
+            future_select(table_a.c.a),
+            future_select(table_a.c.a).join(
+                table_b, table_a.c.a == table_b.c.a
+            ),
+            future_select(table_a.c.a).join_from(
+                table_a, table_b, table_a.c.a == table_b.c.a
+            ),
+            future_select(table_a.c.a).join_from(table_a, table_b),
+            future_select(table_a.c.a).join_from(table_c, table_b),
+            future_select(table_a.c.a)
+            .join(table_b, table_a.c.a == table_b.c.a)
+            .join(table_c, table_b.c.b == table_c.c.x),
+            future_select(table_a.c.a).join(table_b),
+            future_select(table_a.c.a).join(table_c),
+            future_select(table_a.c.a).join(
+                table_b, table_a.c.a == table_b.c.b
+            ),
+            future_select(table_a.c.a).join(
+                table_c, table_a.c.a == table_c.c.x
+            ),
+        ),
+        lambda: (
             select([table_a.c.a]).cte(),
             select([table_a.c.a]).cte(recursive=True),
             select([table_a.c.a]).cte(name="some_cte", recursive=True),
@@ -610,6 +645,22 @@ class CoreFixtures(object):
 
     fixtures.append(_complex_fixtures)
 
+    def _statements_w_context_options_fixtures():
+
+        return [
+            select([table_a])._add_context_option(opt1, True),
+            select([table_a])._add_context_option(opt1, 5),
+            select([table_a])
+            ._add_context_option(opt1, True)
+            ._add_context_option(opt2, True),
+            select([table_a])
+            ._add_context_option(opt1, True)
+            ._add_context_option(opt2, 5),
+            select([table_a])._add_context_option(opt3, True),
+        ]
+
+    fixtures.append(_statements_w_context_options_fixtures)
+
 
 class CacheKeyFixture(object):
     def _run_cache_key_fixture(self, fixture, compare_values):
@@ -986,30 +1037,34 @@ class CompareAndCopyTest(CoreFixtures, fixtures.TestBase):
 
 
 class CompareClausesTest(fixtures.TestBase):
-    def test_compare_metadata_tables(self):
-        # metadata Table objects cache on their own identity, not their
-        # structure.   This is mainly to reduce the size of cache keys
-        # as well as reduce computational overhead, as Table objects have
-        # very large internal state and they are also generally global
-        # objects.
+    def test_compare_metadata_tables_annotations_one(self):
+        # test that cache keys from annotated version of tables refresh
+        # properly
 
         t1 = Table("a", MetaData(), Column("q", Integer), Column("p", Integer))
         t2 = Table("a", MetaData(), Column("q", Integer), Column("p", Integer))
 
         ne_(t1._generate_cache_key(), t2._generate_cache_key())
 
-        eq_(t1._generate_cache_key().key, (t1, "_annotations", ()))
+        eq_(t1._generate_cache_key().key, (t1,))
+
+        t2 = t1._annotate({"foo": "bar"})
+        eq_(
+            t2._generate_cache_key().key,
+            (t1, "_annotations", (("foo", "bar"),)),
+        )
+        eq_(
+            t2._annotate({"bat": "bar"})._generate_cache_key().key,
+            (t1, "_annotations", (("bat", "bar"), ("foo", "bar"))),
+        )
 
-    def test_compare_metadata_tables_annotations(self):
-        # metadata Table objects cache on their own identity, not their
-        # structure.   This is mainly to reduce the size of cache keys
-        # as well as reduce computational overhead, as Table objects have
-        # very large internal state and they are also generally global
-        # objects.
+    def test_compare_metadata_tables_annotations_two(self):
 
         t1 = Table("a", MetaData(), Column("q", Integer), Column("p", Integer))
         t2 = Table("a", MetaData(), Column("q", Integer), Column("p", Integer))
 
+        eq_(t2._generate_cache_key().key, (t2,))
+
         t1 = t1._annotate({"orm": True})
         t2 = t2._annotate({"orm": True})
 
diff --git a/test/sql/test_compiler.py b/test/sql/test_compiler.py
index b3ae7e12d..c580e972d 100644
--- a/test/sql/test_compiler.py
+++ b/test/sql/test_compiler.py
@@ -29,6 +29,7 @@ from sqlalchemy import exc
 from sqlalchemy import except_
 from sqlalchemy import exists
 from sqlalchemy import Float
+from sqlalchemy import ForeignKey
 from sqlalchemy import func
 from sqlalchemy import Integer
 from sqlalchemy import intersect
@@ -85,7 +86,6 @@ from sqlalchemy.testing import fixtures
 from sqlalchemy.testing import is_
 from sqlalchemy.util import u
 
-
 table1 = table(
     "mytable",
     column("myid", Integer),
@@ -123,6 +123,13 @@ table5 = Table(
     schema="dbo.remote_owner",
 )
 
+parent = Table("parent", metadata, Column("id", Integer, primary_key=True))
+child = Table(
+    "child",
+    metadata,
+    Column("id", Integer, primary_key=True),
+    Column("parent_id", ForeignKey("parent.id")),
+)
 users = table(
     "users", column("user_id"), column("user_name"), column("password")
 )
@@ -5218,7 +5225,14 @@ class ResultMapTest(fixtures.TestBase):
         eq_(len(stmt.subquery().c), 7)
 
         # will render 7 as well
-        eq_(len(stmt._compile_state_factory(stmt, None).columns_plus_names), 7)
+        eq_(
+            len(
+                stmt._compile_state_factory(
+                    stmt, stmt.compile()
+                ).columns_plus_names
+            ),
+            7,
+        )
 
         wrapped = stmt._generate()
         wrapped = wrapped.add_columns(
diff --git a/test/sql/test_external_traversal.py b/test/sql/test_external_traversal.py
index 37fb752fe..fb2501667 100644
--- a/test/sql/test_external_traversal.py
+++ b/test/sql/test_external_traversal.py
@@ -17,6 +17,7 @@ from sqlalchemy import testing
 from sqlalchemy import text
 from sqlalchemy import tuple_
 from sqlalchemy import union
+from sqlalchemy.future import select as future_select
 from sqlalchemy.sql import ClauseElement
 from sqlalchemy.sql import column
 from sqlalchemy.sql import operators
@@ -754,6 +755,59 @@ class ClauseTest(fixtures.TestBase, AssertsCompiledSQL):
             ":col1_1) AS anon_1",
         )
 
+    def test_this_thing_using_setup_joins_one(self):
+        s = (
+            future_select(t1)
+            .join_from(t1, t2, t1.c.col1 == t2.c.col2)
+            .subquery()
+        )
+        s2 = future_select(s.c.col1).join_from(t3, s, t3.c.col2 == s.c.col1)
+
+        self.assert_compile(
+            s2,
+            "SELECT anon_1.col1 FROM table3 JOIN (SELECT table1.col1 AS "
+            "col1, table1.col2 AS col2, table1.col3 AS col3 FROM table1 "
+            "JOIN table2 ON table1.col1 = table2.col2) AS anon_1 "
+            "ON table3.col2 = anon_1.col1",
+        )
+        t1a = t1.alias()
+        s2 = sql_util.ClauseAdapter(t1a).traverse(s2)
+        self.assert_compile(
+            s2,
+            "SELECT anon_1.col1 FROM table3 JOIN (SELECT table1_1.col1 AS "
+            "col1, table1_1.col2 AS col2, table1_1.col3 AS col3 "
+            "FROM table1 AS table1_1 JOIN table2 ON table1_1.col1 = "
+            "table2.col2) AS anon_1 ON table3.col2 = anon_1.col1",
+        )
+
+    def test_this_thing_using_setup_joins_two(self):
+        s = (
+            future_select(t1.c.col1)
+            .join(t2, t1.c.col1 == t2.c.col2)
+            .subquery()
+        )
+        s2 = future_select(s.c.col1)
+
+        self.assert_compile(
+            s2,
+            "SELECT anon_1.col1 FROM (SELECT table1.col1 AS col1 "
+            "FROM table1 JOIN table2 ON table1.col1 = table2.col2) AS anon_1",
+        )
+
+        t1alias = t1.alias("t1alias")
+        j = t1.join(t1alias, t1.c.col1 == t1alias.c.col2)
+
+        vis = sql_util.ClauseAdapter(j)
+
+        s2 = vis.traverse(s2)
+        self.assert_compile(
+            s2,
+            "SELECT anon_1.col1 FROM (SELECT table1.col1 AS col1 "
+            "FROM table1 JOIN table1 AS t1alias "
+            "ON table1.col1 = t1alias.col2 "
+            "JOIN table2 ON table1.col1 = table2.col2) AS anon_1",
+        )
+
     def test_select_fromtwice_one(self):
         t1a = t1.alias()
 
@@ -803,6 +857,77 @@ class ClauseTest(fixtures.TestBase, AssertsCompiledSQL):
             "AS anon_1 WHERE table1.col1 = anon_1.col1)",
         )
 
+    def test_select_setup_joins_adapt_element_one(self):
+        s = future_select(t1).join(t2, t1.c.col1 == t2.c.col2)
+
+        t1a = t1.alias()
+
+        s2 = sql_util.ClauseAdapter(t1a).traverse(s)
+
+        self.assert_compile(
+            s,
+            "SELECT table1.col1, table1.col2, table1.col3 "
+            "FROM table1 JOIN table2 ON table1.col1 = table2.col2",
+        )
+        self.assert_compile(
+            s2,
+            "SELECT table1_1.col1, table1_1.col2, table1_1.col3 "
+            "FROM table1 AS table1_1 JOIN table2 "
+            "ON table1_1.col1 = table2.col2",
+        )
+
+    def test_select_setup_joins_adapt_element_two(self):
+        s = future_select(literal_column("1")).join_from(
+            t1, t2, t1.c.col1 == t2.c.col2
+        )
+
+        t1a = t1.alias()
+
+        s2 = sql_util.ClauseAdapter(t1a).traverse(s)
+
+        self.assert_compile(
+            s, "SELECT 1 FROM table1 JOIN table2 ON table1.col1 = table2.col2"
+        )
+        self.assert_compile(
+            s2,
+            "SELECT 1 FROM table1 AS table1_1 "
+            "JOIN table2 ON table1_1.col1 = table2.col2",
+        )
+
+    def test_select_setup_joins_adapt_element_three(self):
+        s = future_select(literal_column("1")).join_from(
+            t1, t2, t1.c.col1 == t2.c.col2
+        )
+
+        t2a = t2.alias()
+
+        s2 = sql_util.ClauseAdapter(t2a).traverse(s)
+
+        self.assert_compile(
+            s, "SELECT 1 FROM table1 JOIN table2 ON table1.col1 = table2.col2"
+        )
+        self.assert_compile(
+            s2,
+            "SELECT 1 FROM table1 "
+            "JOIN table2 AS table2_1 ON table1.col1 = table2_1.col2",
+        )
+
+    def test_select_setup_joins_straight_clone(self):
+        s = future_select(t1).join(t2, t1.c.col1 == t2.c.col2)
+
+        s2 = CloningVisitor().traverse(s)
+
+        self.assert_compile(
+            s,
+            "SELECT table1.col1, table1.col2, table1.col3 "
+            "FROM table1 JOIN table2 ON table1.col1 = table2.col2",
+        )
+        self.assert_compile(
+            s2,
+            "SELECT table1.col1, table1.col2, table1.col3 "
+            "FROM table1 JOIN table2 ON table1.col1 = table2.col2",
+        )
+
 
 class ColumnAdapterTest(fixtures.TestBase, AssertsCompiledSQL):
     __dialect__ = "default"
diff --git a/test/sql/test_select.py b/test/sql/test_select.py
new file mode 100644
index 000000000..7bac921a1
--- /dev/null
+++ b/test/sql/test_select.py
@@ -0,0 +1,163 @@
+from sqlalchemy import Column
+from sqlalchemy import exc
+from sqlalchemy import ForeignKey
+from sqlalchemy import Integer
+from sqlalchemy import MetaData
+from sqlalchemy import String
+from sqlalchemy import Table
+from sqlalchemy.future import select as future_select
+from sqlalchemy.sql import column
+from sqlalchemy.sql import table
+from sqlalchemy.testing import assert_raises_message
+from sqlalchemy.testing import AssertsCompiledSQL
+from sqlalchemy.testing import fixtures
+
+table1 = table(
+    "mytable",
+    column("myid", Integer),
+    column("name", String),
+    column("description", String),
+)
+
+table2 = table(
+    "myothertable", column("otherid", Integer), column("othername", String)
+)
+
+metadata = MetaData()
+
+
+parent = Table(
+    "parent",
+    metadata,
+    Column("id", Integer, primary_key=True),
+    Column("data", String(50)),
+)
+child = Table(
+    "child",
+    metadata,
+    Column("id", Integer, primary_key=True),
+    Column("parent_id", ForeignKey("parent.id")),
+    Column("data", String(50)),
+)
+
+
+class FutureSelectTest(fixtures.TestBase, AssertsCompiledSQL):
+    __dialect__ = "default"
+
+    def test_join_nofrom_implicit_left_side_explicit_onclause(self):
+        stmt = future_select(table1).join(
+            table2, table1.c.myid == table2.c.otherid
+        )
+
+        self.assert_compile(
+            stmt,
+            "SELECT mytable.myid, mytable.name, mytable.description "
+            "FROM mytable JOIN myothertable "
+            "ON mytable.myid = myothertable.otherid",
+        )
+
+    def test_join_nofrom_explicit_left_side_explicit_onclause(self):
+        stmt = future_select(table1).join_from(
+            table1, table2, table1.c.myid == table2.c.otherid
+        )
+
+        self.assert_compile(
+            stmt,
+            "SELECT mytable.myid, mytable.name, mytable.description "
+            "FROM mytable JOIN myothertable "
+            "ON mytable.myid = myothertable.otherid",
+        )
+
+    def test_join_nofrom_implicit_left_side_implicit_onclause(self):
+        stmt = future_select(parent).join(child)
+
+        self.assert_compile(
+            stmt,
+            "SELECT parent.id, parent.data FROM parent JOIN child "
+            "ON parent.id = child.parent_id",
+        )
+
+    def test_join_nofrom_explicit_left_side_implicit_onclause(self):
+        stmt = future_select(parent).join_from(parent, child)
+
+        self.assert_compile(
+            stmt,
+            "SELECT parent.id, parent.data FROM parent JOIN child "
+            "ON parent.id = child.parent_id",
+        )
+
+    def test_join_froms_implicit_left_side_explicit_onclause(self):
+        stmt = (
+            future_select(table1)
+            .select_from(table1)
+            .join(table2, table1.c.myid == table2.c.otherid)
+        )
+
+        self.assert_compile(
+            stmt,
+            "SELECT mytable.myid, mytable.name, mytable.description "
+            "FROM mytable JOIN myothertable "
+            "ON mytable.myid = myothertable.otherid",
+        )
+
+    def test_join_froms_explicit_left_side_explicit_onclause(self):
+        stmt = (
+            future_select(table1)
+            .select_from(table1)
+            .join_from(table1, table2, table1.c.myid == table2.c.otherid)
+        )
+
+        self.assert_compile(
+            stmt,
+            "SELECT mytable.myid, mytable.name, mytable.description "
+            "FROM mytable JOIN myothertable "
+            "ON mytable.myid = myothertable.otherid",
+        )
+
+    def test_join_froms_implicit_left_side_implicit_onclause(self):
+        stmt = future_select(parent).select_from(parent).join(child)
+
+        self.assert_compile(
+            stmt,
+            "SELECT parent.id, parent.data FROM parent JOIN child "
+            "ON parent.id = child.parent_id",
+        )
+
+    def test_join_froms_explicit_left_side_implicit_onclause(self):
+        stmt = (
+            future_select(parent).select_from(parent).join_from(parent, child)
+        )
+
+        self.assert_compile(
+            stmt,
+            "SELECT parent.id, parent.data FROM parent JOIN child "
+            "ON parent.id = child.parent_id",
+        )
+
+    def test_joins_w_filter_by(self):
+        stmt = (
+            future_select(parent)
+            .filter_by(data="p1")
+            .join(child)
+            .filter_by(data="c1")
+            .join_from(table1, table2, table1.c.myid == table2.c.otherid)
+            .filter_by(otherid=5)
+        )
+
+        self.assert_compile(
+            stmt,
+            "SELECT parent.id, parent.data FROM parent JOIN child "
+            "ON parent.id = child.parent_id, mytable JOIN myothertable "
+            "ON mytable.myid = myothertable.otherid "
+            "WHERE parent.data = :data_1 AND child.data = :data_2 "
+            "AND myothertable.otherid = :otherid_1",
+            checkparams={"data_1": "p1", "data_2": "c1", "otherid_1": 5},
+        )
+
+    def test_filter_by_no_property(self):
+        assert_raises_message(
+            exc.InvalidRequestError,
+            'Entity namespace for "mytable" has no property "foo"',
+            future_select(table1).filter_by,
+            foo="bar",
+        )