Don't include SelectBase when searching for surface column elements

Fixed bug where correlated select used against single-table inheritance
entity would fail to render correctly in the outer query, due to adjustment
for single inheritance discriminator criteria inappropriately re-applying
the criteria to the outer query.

Change-Id: I38df21f1392af1843e10119682fa0635d346e2a8
Fixes: #4103

4 files changed, 50 insertions, 4 deletions

diff --git a/doc/build/changelog/unreleased_11/4103.rst b/doc/build/changelog/unreleased_11/4103.rst
new file mode 100644
index 000000000..b667dd10b
--- /dev/null
+++ b/doc/build/changelog/unreleased_11/4103.rst
@@ -0,0 +1,9 @@
+.. change::
+    :tags: bug, orm
+    :tickets: 4103
+    :versions: 1.2.0b3
+
+    Fixed bug where correlated select used against single-table inheritance
+    entity would fail to render correctly in the outer query, due to adjustment
+    for single inheritance discriminator criteria inappropriately re-applying
+    the criteria to the outer query.
\ No newline at end of file
diff --git a/lib/sqlalchemy/orm/query.py b/lib/sqlalchemy/orm/query.py
index 19a7b07c1..b3bb5302f 100644
--- a/lib/sqlalchemy/orm/query.py
+++ b/lib/sqlalchemy/orm/query.py
@@ -4031,7 +4031,8 @@ class _ColumnEntity(_QueryEntity):
             self._from_entities = set(self.entities)
         else:
             all_elements = [
-                elem for elem in sql_util.surface_column_elements(column)
+                elem for elem in sql_util.surface_column_elements(
+                    column, include_scalar_selects=False)
                 if 'parententity' in elem._annotations
             ]
 
diff --git a/lib/sqlalchemy/sql/util.py b/lib/sqlalchemy/sql/util.py
index 281d5f6a3..0c122949b 100644
--- a/lib/sqlalchemy/sql/util.py
+++ b/lib/sqlalchemy/sql/util.py
@@ -18,7 +18,7 @@ from collections import deque
 from .elements import BindParameter, ColumnClause, ColumnElement, \
     Null, UnaryExpression, literal_column, Label, _label_reference, \
     _textual_label_reference
-from .selectable import ScalarSelect, Join, FromClause, FromGrouping
+from .selectable import SelectBase, ScalarSelect, Join, FromClause, FromGrouping
 from .schema import Column
 
 join_condition = util.langhelpers.public_factory(
@@ -235,17 +235,21 @@ def surface_selectables(clause):
             stack.append(elem.element)
 
 
-def surface_column_elements(clause):
+def surface_column_elements(clause, include_scalar_selects=True):
     """traverse and yield only outer-exposed column elements, such as would
     be addressable in the WHERE clause of a SELECT if this element were
     in the columns clause."""
 
+    filter_ = (FromGrouping, )
+    if not include_scalar_selects:
+        filter_ += (SelectBase, )
+
     stack = deque([clause])
     while stack:
         elem = stack.popleft()
         yield elem
         for sub in elem.get_children():
-            if isinstance(sub, FromGrouping):
+            if isinstance(sub, filter_):
                 continue
             stack.append(sub)
 
diff --git a/test/orm/inheritance/test_single.py b/test/orm/inheritance/test_single.py
index 0c7fc6f8d..0ad6dcdee 100644
--- a/test/orm/inheritance/test_single.py
+++ b/test/orm/inheritance/test_single.py
@@ -769,6 +769,38 @@ class RelationshipToSingleTest(
             "AND employees_1.type IN (:type_1)"
         )
 
+    def test_correlated_column_select(self):
+        Company, Employee, Engineer = (self.classes.Company,
+                                       self.classes.Employee,
+                                       self.classes.Engineer)
+        companies, employees = self.tables.companies, self.tables.employees
+
+        mapper(Company, companies)
+        mapper(
+            Employee, employees,
+            polymorphic_on=employees.c.type,
+            properties={
+                'company': relationship(Company)
+            }
+        )
+        mapper(Engineer, inherits=Employee, polymorphic_identity='engineer')
+
+        sess = create_session()
+        engineer_count = sess.query(func.count(Engineer.employee_id)) \
+            .select_from(Engineer) \
+            .filter(Engineer.company_id == Company.company_id) \
+            .correlate(Company) \
+            .as_scalar()
+
+        self.assert_compile(
+            sess.query(Company.company_id, engineer_count),
+            "SELECT companies.company_id AS companies_company_id, "
+            "(SELECT count(employees.employee_id) AS count_1 "
+            "FROM employees WHERE employees.company_id = "
+            "companies.company_id AND employees.type IN (:type_1)) AS anon_1 "
+            "FROM companies"
+        )
+
     def test_no_aliasing_from_overlap(self):
         # test [ticket:3233]