- Fixed bug in new :class:`.TextAsFrom` construct where :class:`.Column`-

oriented row lookups were not matching up to the ad-hoc :class:`.ColumnClause`
objects that :class:`.TextAsFrom` generates, thereby making it not
usable as a target in :meth:`.Query.from_statement`.  Also fixed
:meth:`.Query.from_statement` mechanics to not mistake a :class:`.TextAsFrom`
for a :class:`.Select` construct.  This bug is also an 0.9 regression
as the :meth:`.Text.columns` method is called to accommodate the
:paramref:`.text.typemap` argument. [ticket:2932]

6 files changed, 108 insertions, 3 deletions

diff --git a/doc/build/changelog/changelog_09.rst b/doc/build/changelog/changelog_09.rst
index a218e0f54..573548cfa 100644
--- a/doc/build/changelog/changelog_09.rst
+++ b/doc/build/changelog/changelog_09.rst
@@ -15,6 +15,19 @@
     :version: 0.9.2
 
     .. change::
+        :tags: bug, orm
+        :tickets: 2932
+
+        Fixed bug in new :class:`.TextAsFrom` construct where :class:`.Column`-
+        oriented row lookups were not matching up to the ad-hoc :class:`.ColumnClause`
+        objects that :class:`.TextAsFrom` generates, thereby making it not
+        usable as a target in :meth:`.Query.from_statement`.  Also fixed
+        :meth:`.Query.from_statement` mechanics to not mistake a :class:`.TextAsFrom`
+        for a :class:`.Select` construct.  This bug is also an 0.9 regression
+        as the :meth:`.Text.columns` method is called to accommodate the
+        :paramref:`.text.typemap` argument.
+
+    .. change::
         :tags: feature, sql
         :tickets: 2923
 
diff --git a/lib/sqlalchemy/orm/query.py b/lib/sqlalchemy/orm/query.py
index 10d635b8d..949c7eff9 100644
--- a/lib/sqlalchemy/orm/query.py
+++ b/lib/sqlalchemy/orm/query.py
@@ -3471,6 +3471,7 @@ class QueryContext(object):
 
         if query._statement is not None:
             if isinstance(query._statement, expression.SelectBase) and \
+                                not query._statement._textual and \
                                 not query._statement.use_labels:
                 self.statement = query._statement.apply_labels()
             else:
diff --git a/lib/sqlalchemy/sql/elements.py b/lib/sqlalchemy/sql/elements.py
index 975cf2d79..cca03851f 100644
--- a/lib/sqlalchemy/sql/elements.py
+++ b/lib/sqlalchemy/sql/elements.py
@@ -2920,9 +2920,12 @@ class ColumnClause(Immutable, ColumnElement):
 
     def _compare_name_for_result(self, other):
         if self.is_literal or \
-            self.table is None or \
+            self.table is None or self.table._textual or \
             not hasattr(other, 'proxy_set') or (
-            isinstance(other, ColumnClause) and other.is_literal
+            isinstance(other, ColumnClause) and
+                (other.is_literal or
+                        other.table is None or
+                        other.table._textual)
         ):
             return super(ColumnClause, self).\
                     _compare_name_for_result(other)
diff --git a/lib/sqlalchemy/sql/selectable.py b/lib/sqlalchemy/sql/selectable.py
index 01d617259..bda3d655e 100644
--- a/lib/sqlalchemy/sql/selectable.py
+++ b/lib/sqlalchemy/sql/selectable.py
@@ -137,6 +137,10 @@ class FromClause(Selectable):
     named_with_column = False
     _hide_froms = []
 
+    _textual = False
+    """a marker that allows us to easily distinguish a :class:`.TextAsFrom`
+    or similar object from other kinds of :class:`.FromClause` objects."""
+
     schema = None
     """Define the 'schema' attribute for this :class:`.FromClause`.
 
@@ -3046,6 +3050,8 @@ class TextAsFrom(SelectBase):
     """
     __visit_name__ = "text_as_from"
 
+    _textual = True
+
     def __init__(self, text, columns):
         self.element = text
         self.column_args = columns
diff --git a/test/orm/test_query.py b/test/orm/test_query.py
index 64b8cfdc4..07cb24cb6 100644
--- a/test/orm/test_query.py
+++ b/test/orm/test_query.py
@@ -1,7 +1,7 @@
 from sqlalchemy.sql import operators
 from sqlalchemy import MetaData, null, exists, text, union, literal, \
     literal_column, func, between, Unicode, desc, and_, bindparam, \
-    select, distinct, or_, collate, insert
+    select, distinct, or_, collate, insert, Integer, String
 from sqlalchemy import inspect
 from sqlalchemy import exc as sa_exc, util
 from sqlalchemy.sql import compiler, table, column
@@ -2080,6 +2080,43 @@ class TextTest(QueryTest):
             [User(id=7), User(id=8), User(id=9), User(id=10)]
         )
 
+    def test_via_textasfrom_from_statement(self):
+        User = self.classes.User
+        s = create_session()
+
+        eq_(
+            s.query(User).from_statement(
+                text("select * from users order by id").\
+                        columns(id=Integer, name=String)
+            ).all(),
+            [User(id=7), User(id=8), User(id=9), User(id=10)]
+        )
+
+    def test_via_textasfrom_select_from(self):
+        User = self.classes.User
+        s = create_session()
+
+        eq_(
+            s.query(User).select_from(
+                text("select * from users order by id").\
+                        columns(id=Integer, name=String)
+            ).all(),
+            [User(id=7), User(id=8), User(id=9), User(id=10)]
+        )
+
+    def test_via_textasfrom_use_mapped_columns(self):
+        User = self.classes.User
+        s = create_session()
+
+        eq_(
+            s.query(User).select_from(
+                text("select * from users order by id").\
+                        columns(User.id, User.name)
+            ).all(),
+            [User(id=7), User(id=8), User(id=9), User(id=10)]
+        )
+
+
 class ParentTest(QueryTest, AssertsCompiledSQL):
     __dialect__ = 'default'
 
diff --git a/test/sql/test_query.py b/test/sql/test_query.py
index 26e2fcf34..6e2227650 100644
--- a/test/sql/test_query.py
+++ b/test/sql/test_query.py
@@ -1787,6 +1787,51 @@ class KeyTargetingTest(fixtures.TablesTest):
         assert bar.c.content_type not in row
         assert sql.column('content_type') not in row
 
+    def test_columnclause_schema_column_one(self):
+        keyed2 = self.tables.keyed2
+
+        # this is addressed by [ticket:2932]
+        # ColumnClause._compare_name_for_result allows the
+        # columns which the statement is against to be lightweight
+        # cols, which results in a more liberal comparison scheme
+        a, b = sql.column('a'), sql.column('b')
+        stmt = select([a, b]).select_from("keyed2")
+        row = testing.db.execute(stmt).first()
+
+        assert keyed2.c.a in row
+        assert keyed2.c.b in row
+        assert a in row
+        assert b in row
+
+    def test_columnclause_schema_column_two(self):
+        keyed2 = self.tables.keyed2
+
+        a, b = sql.column('a'), sql.column('b')
+        stmt = select([keyed2.c.a, keyed2.c.b])
+        row = testing.db.execute(stmt).first()
+
+        assert keyed2.c.a in row
+        assert keyed2.c.b in row
+        assert a in row
+        assert b in row
+
+    def test_columnclause_schema_column_three(self):
+        keyed2 = self.tables.keyed2
+
+        # this is also addressed by [ticket:2932]
+
+        a, b = sql.column('a'), sql.column('b')
+        stmt = text("select a, b from keyed2").columns(a=CHAR, b=CHAR)
+        row = testing.db.execute(stmt).first()
+
+        assert keyed2.c.a in row
+        assert keyed2.c.b in row
+        assert a in row
+        assert b in row
+        assert stmt.c.a in row
+        assert stmt.c.b in row
+
+
 
 class LimitTest(fixtures.TestBase):