- wip, many more checks and changes than in the original patch

7 files changed, 74 insertions, 33 deletions

diff --git a/lib/sqlalchemy/engine/default.py b/lib/sqlalchemy/engine/default.py
index 6c42af8b1..3e5f339b1 100644
--- a/lib/sqlalchemy/engine/default.py
+++ b/lib/sqlalchemy/engine/default.py
@@ -544,7 +544,8 @@ class DefaultExecutionContext(interfaces.ExecutionContext):
             connection._execution_options)
 
         self.result_column_struct = (
-            compiled._result_columns, compiled._ordered_columns)
+            compiled._result_columns, compiled._ordered_columns,
+            compiled._textual_ordered_columns)
 
         self.unicode_statement = util.text_type(compiled)
         if not dialect.supports_unicode_statements:
diff --git a/lib/sqlalchemy/engine/result.py b/lib/sqlalchemy/engine/result.py
index 7d1425c28..4dc5354bb 100644
--- a/lib/sqlalchemy/engine/result.py
+++ b/lib/sqlalchemy/engine/result.py
@@ -85,7 +85,7 @@ except ImportError:
             if index is None:
                 raise exc.InvalidRequestError(
                     "Ambiguous column name '%s' in result set! "
-                    "try 'use_labels' option on select statement." % key)
+                    "try 'use_labels' option on select statement." % obj)
             if processor is not None:
                 return processor(self._row[index])
             else:
@@ -194,14 +194,18 @@ class ResultMetaData(object):
         self.case_sensitive = case_sensitive = dialect.case_sensitive
 
         if context.result_column_struct:
-            result_columns, cols_are_ordered = context.result_column_struct
+            result_columns, cols_are_ordered, textual_ordered = \
+                context.result_column_struct
             num_ctx_cols = len(result_columns)
+            num_metadata_cols = len(metadata)
         else:
+            textual_ordered = False
             num_ctx_cols = None
 
         if num_ctx_cols and \
                 cols_are_ordered and \
-                num_ctx_cols == len(metadata):
+                not textual_ordered and \
+                num_ctx_cols == num_metadata_cols:
             # case 1 - SQL expression statement, number of columns
             # in result matches number of cols in compiled.  This is the
             # vast majority case for SQL expression constructs.  In this
@@ -223,6 +227,7 @@ class ResultMetaData(object):
             self.keys = [
                 elem[0] for elem in result_columns
             ]
+
         else:
             # case 2 - raw string, or number of columns in result does
             # not match number of cols in compiled.  The raw string case
@@ -235,13 +240,16 @@ class ResultMetaData(object):
             # In all these cases we fall back to the "named" approach
             # that SQLAlchemy has used up through 0.9.
 
-            if num_ctx_cols:
+            if num_ctx_cols and not textual_ordered:
                 result_map = self._create_result_map(
                     result_columns, case_sensitive)
 
             raw = []
             self.keys = []
             untranslated = None
+            if textual_ordered:
+                seen = set()
+
             for idx, rec in enumerate(metadata):
                 colname = rec[0]
                 coltype = rec[1]
@@ -259,7 +267,20 @@ class ResultMetaData(object):
                 if not case_sensitive:
                     colname = colname.lower()
 
-                if num_ctx_cols:
+                if textual_ordered:
+                    if idx < num_ctx_cols:
+                        ctx_rec = result_columns[idx]
+                        obj = ctx_rec[2]
+                        mapped_type = ctx_rec[3]
+                        if obj[0] in seen:
+                            raise exc.InvalidRequestError(
+                                "Duplicate column expression requested "
+                                "in textual SQL: %r" % obj[0])
+                        seen.add(obj[0])
+                    else:
+                        mapped_type = typemap.get(coltype, sqltypes.NULLTYPE)
+                        obj = None
+                elif num_ctx_cols:
                     try:
                         ctx_rec = result_map[colname]
                     except KeyError:
@@ -295,6 +316,17 @@ class ResultMetaData(object):
                 for elem in raw
             ])
 
+            if textual_ordered:
+                if num_ctx_cols > len(metadata):
+                    util.warn(
+                        "Number of columns in textual SQL (%d) is "
+                        "smaller than number of columns requested (%d)" % (
+                            num_ctx_cols, len(metadata)
+                        ))
+
+                # TODO: check for dupes
+                pass
+
             # if by-primary-string dictionary smaller (or bigger?!) than
             # number of columns, assume we have dupes, rewrite
             # dupe records with "None" for index which results in
@@ -304,7 +336,7 @@ class ResultMetaData(object):
                 for rec in raw:
                     key = rec[1]
                     if key in seen:
-                        by_key[key] = (None, by_key[key][1], None)
+                        by_key[key] = (None, key, None)
                     seen.add(key)
 
             # update keymap with secondary "object"-based keys
@@ -428,7 +460,7 @@ class ResultMetaData(object):
         if index is None:
             raise exc.InvalidRequestError(
                 "Ambiguous column name '%s' in result set! "
-                "try 'use_labels' option on select statement." % key)
+                "try 'use_labels' option on select statement." % obj)
 
         return operator.itemgetter(index)
 
diff --git a/lib/sqlalchemy/sql/compiler.py b/lib/sqlalchemy/sql/compiler.py
index c4e73a1e3..c5f87cc33 100644
--- a/lib/sqlalchemy/sql/compiler.py
+++ b/lib/sqlalchemy/sql/compiler.py
@@ -345,6 +345,18 @@ class SQLCompiler(Compiled):
     driver/DB enforces this
     """
 
+    _textual_ordered_columns = False
+    """tell the result object that the column names as rendered are important,
+    but they are also "ordered" vs. what is in the compiled object here.
+    """
+
+    _ordered_columns = True
+    """
+    if False, means we can't be sure the list of entries
+    in _result_columns is actually the rendered order.  Usually
+    True unless using an unordered TextAsFrom.
+    """
+
     def __init__(self, dialect, statement, column_keys=None,
                  inline=False, **kwargs):
         """Construct a new :class:`.SQLCompiler` object.
@@ -386,11 +398,6 @@ class SQLCompiler(Compiled):
         # column targeting
         self._result_columns = []
 
-        # if False, means we can't be sure the list of entries
-        # in _result_columns is actually the rendered order.   This
-        # gets flipped when we use TextAsFrom, for example.
-        self._ordered_columns = True
-
         # true if the paramstyle is positional
         self.positional = dialect.positional
         if self.positional:
@@ -733,7 +740,8 @@ class SQLCompiler(Compiled):
             ) or entry.get('need_result_map_for_nested', False)
 
         if populate_result_map:
-            self._ordered_columns = False
+            self._ordered_columns = \
+                self._textual_ordered_columns = taf.positional
             for c in taf.column_args:
                 self.process(c, within_columns_clause=True,
                              add_to_result_map=self._add_to_result_map)
@@ -1326,7 +1334,7 @@ class SQLCompiler(Compiled):
             add_to_result_map = lambda keyname, name, objects, type_: \
                 self._add_to_result_map(
                     keyname, name,
-                    objects + (column,), type_)
+                    (column,) + objects, type_)
         else:
             col_expr = column
             if populate_result_map:
diff --git a/lib/sqlalchemy/sql/elements.py b/lib/sqlalchemy/sql/elements.py
index 774e42609..58e8d78ec 100644
--- a/lib/sqlalchemy/sql/elements.py
+++ b/lib/sqlalchemy/sql/elements.py
@@ -1529,13 +1529,19 @@ class TextClause(Executable, ClauseElement):
 
         """
 
-        input_cols = [
+        positional_input_cols = [
             ColumnClause(col.key, types.pop(col.key))
             if col.key in types
             else col
             for col in cols
-        ] + [ColumnClause(key, type_) for key, type_ in types.items()]
-        return selectable.TextAsFrom(self, input_cols)
+        ]
+        keyed_input_cols = [
+            ColumnClause(key, type_) for key, type_ in types.items()]
+
+        return selectable.TextAsFrom(
+            self,
+            positional_input_cols + keyed_input_cols,
+            positional=bool(positional_input_cols) and not keyed_input_cols)
 
     @property
     def type(self):
diff --git a/lib/sqlalchemy/sql/selectable.py b/lib/sqlalchemy/sql/selectable.py
index 73341053d..1955fc934 100644
--- a/lib/sqlalchemy/sql/selectable.py
+++ b/lib/sqlalchemy/sql/selectable.py
@@ -3420,9 +3420,10 @@ class TextAsFrom(SelectBase):
 
     _textual = True
 
-    def __init__(self, text, columns):
+    def __init__(self, text, columns, positional=False):
         self.element = text
         self.column_args = columns
+        self.positional = positional
 
     @property
     def _bind(self):
diff --git a/test/sql/test_resultset.py b/test/sql/test_resultset.py
index 8461996ea..39ecad0d5 100644
--- a/test/sql/test_resultset.py
+++ b/test/sql/test_resultset.py
@@ -664,19 +664,11 @@ class ResultProxyTest(fixtures.TablesTest):
             lambda: row[ua.c.user_id]
         )
 
-        # Unfortunately, this fails -
-        # we'd like
-        # "Could not locate column in row"
-        # to be raised here, but the check for
-        # "common column" in _compare_name_for_result()
-        # has other requirements to be more liberal.
-        # Ultimately the
-        # expression system would need a way to determine
-        # if given two columns in a "proxy" relationship, if they
-        # refer to a different parent table
+        # this now works as of 1.1 issue #3501;
+        # previously this was stuck on "ambiguous column name"
         assert_raises_message(
             exc.InvalidRequestError,
-            "Ambiguous column name",
+            "Could not locate column in row",
             lambda: row[u2.c.user_id]
         )
 
diff --git a/test/sql/test_type_expressions.py b/test/sql/test_type_expressions.py
index 574edfe9e..0ef3a3e16 100644
--- a/test/sql/test_type_expressions.py
+++ b/test/sql/test_type_expressions.py
@@ -59,13 +59,14 @@ class SelectTest(_ExprFixture, fixtures.TestBase, AssertsCompiledSQL):
         # the lower() function goes into the result_map, we don't really
         # need this but it's fine
         self.assert_compile(
-            compiled._create_result_map()['test_table_y'][1][2],
+            compiled._create_result_map()['test_table_y'][1][3],
             "lower(test_table.y)"
         )
         # then the original column gets put in there as well.
-        # it's not important that it's the last value.
+        # as of 1.1 it's important that it is first as this is
+        # taken as significant by the result processor.
         self.assert_compile(
-            compiled._create_result_map()['test_table_y'][1][-1],
+            compiled._create_result_map()['test_table_y'][1][0],
             "test_table.y"
         )