Corrected mssql schema named subqueries from not properly aliasing the columns. Fixes #973.

2 files changed, 7 insertions, 3 deletions

diff --git a/lib/sqlalchemy/databases/mssql.py b/lib/sqlalchemy/databases/mssql.py
index a313b98bc..d806492c6 100644
--- a/lib/sqlalchemy/databases/mssql.py
+++ b/lib/sqlalchemy/databases/mssql.py
@@ -1001,7 +1001,8 @@ class MSSQLCompiler(compiler.DefaultCompiler):
         return super(MSSQLCompiler, self).visit_alias(alias, **kwargs)
 
     def visit_column(self, column, result_map=None, **kwargs):
-        if column.table is not None and not self.isupdate and not self.isdelete:
+        if column.table is not None and \
+            (not self.isupdate and not self.isdelete) or self.is_subquery():
             # translate for schema-qualified table aliases
             t = self._schema_aliased_table(column.table)
             if t is not None:
@@ -1009,9 +1010,9 @@ class MSSQLCompiler(compiler.DefaultCompiler):
 
                 if result_map is not None:
                     result_map[column.name.lower()] = (column.name, (column, ), column.type)
-                    
+
                 return super(MSSQLCompiler, self).visit_column(converted, result_map=None, **kwargs)
-                
+
         return super(MSSQLCompiler, self).visit_column(column, result_map=result_map, **kwargs)
 
     def visit_binary(self, binary, **kwargs):
diff --git a/test/dialect/mssql.py b/test/dialect/mssql.py
index 1927de3f3..5a4b38f6e 100755
--- a/test/dialect/mssql.py
+++ b/test/dialect/mssql.py
@@ -83,6 +83,9 @@ class CompileTest(TestBase, AssertsCompiledSQL):
         tbl = Table('test', metadata, Column('id', Integer, primary_key=True), schema='paj')
         self.assert_compile(tbl.delete(tbl.c.id == 1), "DELETE FROM paj.test WHERE paj.test.id = :id_1")
 
+        s = select([tbl.c.id]).where(tbl.c.id==1)
+        self.assert_compile(tbl.delete().where(tbl.c.id==(s)), "DELETE FROM paj.test WHERE paj.test.id IN (SELECT test_1.id FROM paj.test AS test_1 WHERE test_1.id = :id_1)")
+
     def test_union(self):
         t1 = table('t1',
             column('col1'),