- reworked the way the "select_wraps_for" expression is

handled within visit_select(); this attribute was added in the
1.0 series to accommodate the subquery wrapping behavior of
SQL Server and Oracle while also working with positional
column targeting and no longer relying upon "key fallback"
in order to target columns in such a statement.  The IBM DB2
third-party dialect also has this use case, but its implementation
is using regular expressions to rewrite the textual SELECT only
and does not make use of a "wrapped" select at this time.
The logic no longer attempts to reconcile proxy set collections as
this was not deterministic, and instead assumes that the select()
and the wrapper select() match their columns postionally,
at least for the column positions they have in common,
so it is now very simple and safe.  fixes #3657.
- as a side effect of #3657 it was also revealed that the
strategy of calling upon a ResultProxy._getter was not
correctly calling into NoSuchColumnError when an expected
column was not present, and instead returned None up to
loading.instances() to produce NoneType failures; added
a raiseerr argument to _getter() which is called when we
aren't expecting None, fixes #3658.

4 files changed, 75 insertions, 3 deletions

diff --git a/test/dialect/mssql/test_compiler.py b/test/dialect/mssql/test_compiler.py
index d91c79db2..b59ca4fd1 100644
--- a/test/dialect/mssql/test_compiler.py
+++ b/test/dialect/mssql/test_compiler.py
@@ -1,5 +1,5 @@
 # -*- encoding: utf-8
-from sqlalchemy.testing import eq_
+from sqlalchemy.testing import eq_, is_
 from sqlalchemy import schema
 from sqlalchemy.sql import table, column
 from sqlalchemy.databases import mssql
@@ -521,6 +521,30 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
         assert t.c.x in set(c._create_result_map()['x'][1])
         assert t.c.y in set(c._create_result_map()['y'][1])
 
+    def test_limit_offset_w_ambiguous_cols(self):
+        t = table('t', column('x', Integer), column('y', Integer))
+
+        cols = [t.c.x, t.c.x.label('q'), t.c.x.label('p'), t.c.y]
+        s = select(cols).where(t.c.x == 5).order_by(t.c.y).limit(10).offset(20)
+
+        self.assert_compile(
+            s,
+            "SELECT anon_1.x, anon_1.q, anon_1.p, anon_1.y "
+            "FROM (SELECT t.x AS x, t.x AS q, t.x AS p, t.y AS y, "
+            "ROW_NUMBER() OVER (ORDER BY t.y) AS mssql_rn "
+            "FROM t "
+            "WHERE t.x = :x_1) AS anon_1 "
+            "WHERE mssql_rn > :param_1 AND mssql_rn <= :param_2 + :param_1",
+            checkparams={'param_1': 20, 'param_2': 10, 'x_1': 5}
+        )
+        c = s.compile(dialect=mssql.MSDialect())
+        eq_(len(c._result_columns), 4)
+
+        result_map = c._create_result_map()
+
+        for col in cols:
+            is_(result_map[col.key][1][0], col)
+
     def test_limit_offset_with_correlated_order_by(self):
         t1 = table('t1', column('x', Integer), column('y', Integer))
         t2 = table('t2', column('x', Integer), column('y', Integer))
diff --git a/test/orm/test_loading.py b/test/orm/test_loading.py
index f86477ec2..6f3f6a016 100644
--- a/test/orm/test_loading.py
+++ b/test/orm/test_loading.py
@@ -1,8 +1,11 @@
 from . import _fixtures
 from sqlalchemy.orm import loading, Session, aliased
-from sqlalchemy.testing.assertions import eq_, assert_raises
+from sqlalchemy.testing.assertions import eq_, \
+    assert_raises, assert_raises_message
 from sqlalchemy.util import KeyedTuple
 from sqlalchemy.testing import mock
+from sqlalchemy import select
+from sqlalchemy import exc
 # class GetFromIdentityTest(_fixtures.FixtureTest):
 # class LoadOnIdentTest(_fixtures.FixtureTest):
 # class InstanceProcessorTest(_fixture.FixtureTest):
@@ -34,6 +37,19 @@ class InstancesTest(_fixtures.FixtureTest):
         )
         assert cursor.close.called, "Cursor wasn't closed"
 
+    def test_row_proc_not_created(self):
+        User = self.classes.User
+        s = Session()
+
+        q = s.query(User.id, User.name)
+        stmt = select([User.id])
+
+        assert_raises_message(
+            exc.NoSuchColumnError,
+            "Could not locate column in row for column 'users.name'",
+            q.from_statement(stmt).all
+        )
+
 
 class MergeResultTest(_fixtures.FixtureTest):
     run_setup_mappers = 'once'
diff --git a/test/sql/test_compiler.py b/test/sql/test_compiler.py
index f11a4e61a..1c5dc2340 100644
--- a/test/sql/test_compiler.py
+++ b/test/sql/test_compiler.py
@@ -3855,3 +3855,26 @@ class ResultMapTest(fixtures.TestBase):
                     (table1.c.description, 'description', 'description'),
                     table1.c.description.type)}
         )
+
+    def test_select_wraps_for_translate_ambiguity(self):
+        # test for issue #3657
+        t = table('a', column('x'), column('y'), column('z'))
+
+        l1, l2, l3 = t.c.z.label('a'), t.c.x.label('b'), t.c.x.label('c')
+        orig = [t.c.x, t.c.y, l1, l2, l3]
+        stmt = select(orig)
+        wrapped = stmt._generate()
+        wrapped = wrapped.column(
+            func.ROW_NUMBER().over(order_by=t.c.z)).alias()
+
+        wrapped_again = select([c for c in wrapped.c])
+
+        compiled = wrapped_again.compile(
+            compile_kwargs={'select_wraps_for': stmt})
+
+        proxied = [obj[0] for (k, n, obj, type_) in compiled._result_columns]
+        for orig_obj, proxied_obj in zip(
+            orig,
+            proxied
+        ):
+            is_(orig_obj, proxied_obj)
diff --git a/test/sql/test_resultset.py b/test/sql/test_resultset.py
index bd2b8c0ae..aaeb82fa4 100644
--- a/test/sql/test_resultset.py
+++ b/test/sql/test_resultset.py
@@ -204,7 +204,8 @@ class ResultProxyTest(fixtures.TablesTest):
                     lambda: result[0][addresses.c.address_id])
 
     def test_column_error_printing(self):
-        row = testing.db.execute(select([1])).first()
+        result = testing.db.execute(select([1]))
+        row = result.first()
 
         class unprintable(object):
 
@@ -222,6 +223,14 @@ class ResultProxyTest(fixtures.TablesTest):
             assert_raises_message(
                 exc.NoSuchColumnError,
                 msg % repl,
+                result._getter, accessor
+            )
+
+            is_(result._getter(accessor, False), None)
+
+            assert_raises_message(
+                exc.NoSuchColumnError,
+                msg % repl,
                 lambda: row[accessor]
             )