Don't apply no-traverse to query.statement

Fixed long-standing issue in :class:`.Query` where a scalar subquery such
as produced by :meth:`.Query.exists`, :meth:`.Query.as_scalar` and other
derivations from :attr:`.Query.statement` would not correctly be adapted
when used in a new :class:`.Query` that required entity adaptation, such as
when the query were turned into a union, or a from_self(), etc. The change
removes the "no adaptation" annotation from the :func:`.select` object
produced by the :attr:`.Query.statement` accessor.

Change-Id: I554e0e909ac6ee785ec3b3b14aaec9d235aa28cf
Fixes: #4304

4 files changed, 54 insertions, 14 deletions

diff --git a/doc/build/changelog/unreleased_13/4304.rst b/doc/build/changelog/unreleased_13/4304.rst
new file mode 100644
index 000000000..128d9be64
--- /dev/null
+++ b/doc/build/changelog/unreleased_13/4304.rst
@@ -0,0 +1,11 @@
+.. change::
+    :tags: bug, orm
+    :tickets: 4304
+
+    Fixed long-standing issue in :class:`.Query` where a scalar subquery such
+    as produced by :meth:`.Query.exists`, :meth:`.Query.as_scalar` and other
+    derivations from :attr:`.Query.statement` would not correctly be adapted
+    when used in a new :class:`.Query` that required entity adaptation, such as
+    when the query were turned into a union, or a from_self(), etc. The change
+    removes the "no adaptation" annotation from the :func:`.select` object
+    produced by the :attr:`.Query.statement` accessor.
diff --git a/lib/sqlalchemy/orm/query.py b/lib/sqlalchemy/orm/query.py
index 272fed3e2..627a4e01c 100644
--- a/lib/sqlalchemy/orm/query.py
+++ b/lib/sqlalchemy/orm/query.py
@@ -512,9 +512,7 @@ class Query(object):
         if self._params:
             stmt = stmt.params(self._params)
 
-        # TODO: there's no tests covering effects of
-        # the annotation not being there
-        return stmt._annotate({'no_replacement_traverse': True})
+        return stmt
 
     def subquery(self, name=None, with_labels=False, reduce_columns=False):
         """return the full SELECT statement represented by
diff --git a/test/orm/inheritance/test_polymorphic_rel.py b/test/orm/inheritance/test_polymorphic_rel.py
index e5234d254..d46448355 100644
--- a/test/orm/inheritance/test_polymorphic_rel.py
+++ b/test/orm/inheritance/test_polymorphic_rel.py
@@ -1286,10 +1286,10 @@ class _PolymorphicTestBase(object):
     def test_correlation_one(self):
         sess = create_session()
 
-        # unfortunately this pattern can't yet work for PolymorphicAliased
-        # and PolymorphicUnions, because the subquery does not compile
-        # out including the polymorphic selectable; only if Person is in
-        # the query() list does that happen.
+        # this for a long time did not work with PolymorphicAliased and
+        # PolymorphicUnions, which was due to the no_replacement_traverse
+        # annotation added to query.statement which then went into as_scalar().
+        # this is removed as of :ticket:`4304` so now works.
         eq_(sess.query(Person.name)
                 .filter(
                     sess.query(Company.name).
@@ -1472,17 +1472,12 @@ class PolymorphicPolymorphicTest(
 
 
 class PolymorphicUnionsTest(_PolymorphicTestBase, _PolymorphicUnions):
-
-    @testing.fails()
-    def test_correlation_one(self):
-        super(PolymorphicUnionsTest, self).test_correlation_one()
+    pass
 
 
 class PolymorphicAliasedJoinsTest(
         _PolymorphicTestBase, _PolymorphicAliasedJoins):
-    @testing.fails()
-    def test_correlation_one(self):
-        super(PolymorphicAliasedJoinsTest, self).test_correlation_one()
+    pass
 
 
 class PolymorphicJoinsTest(_PolymorphicTestBase, _PolymorphicJoins):
diff --git a/test/orm/test_froms.py b/test/orm/test_froms.py
index 45656f3fc..27924c413 100644
--- a/test/orm/test_froms.py
+++ b/test/orm/test_froms.py
@@ -76,6 +76,7 @@ class QueryTest(_fixtures.FixtureTest):
 
 
 class QueryCorrelatesLikeSelect(QueryTest, AssertsCompiledSQL):
+    __dialect__ = "default"
 
     query_correlated = "SELECT users.name AS users_name, " \
         "(SELECT count(addresses.id) AS count_1 FROM addresses " \
@@ -141,6 +142,41 @@ class QueryCorrelatesLikeSelect(QueryTest, AssertsCompiledSQL):
         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_column(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_column(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"
+        )
+
 
 class RawSelectTest(QueryTest, AssertsCompiledSQL):
     """compare a bunch of select() tests with the equivalent Query using