Fix label referencing in SQL Server OFFSET logic

Fixed bug where by ROW_NUMBER OVER clause applied for OFFSET
selects in SQL Server would inappropriately substitute a plain column
from the local statement that overlaps with a label name used by
the ORDER BY criteria of the statement.

Change-Id: Ic2500c886cbfc83a1ad5a2681783f008b9f23838
Fixes: #3711

4 files changed, 49 insertions, 1 deletions

diff --git a/doc/build/changelog/changelog_10.rst b/doc/build/changelog/changelog_10.rst
index c51040dd5..e508e7b81 100644
--- a/doc/build/changelog/changelog_10.rst
+++ b/doc/build/changelog/changelog_10.rst
@@ -19,6 +19,15 @@
     :version: 1.0.13
 
     .. change::
+        :tags: bug, mssql
+        :tickets: 3711
+
+        Fixed bug where by ROW_NUMBER OVER clause applied for OFFSET
+        selects in SQL Server would inappropriately substitute a plain column
+        from the local statement that overlaps with a label name used by
+        the ORDER BY criteria of the statement.
+
+    .. change::
         :tags: bug, orm
         :tickets: 3710
 
diff --git a/lib/sqlalchemy/dialects/mssql/base.py b/lib/sqlalchemy/dialects/mssql/base.py
index 051efa719..966700420 100644
--- a/lib/sqlalchemy/dialects/mssql/base.py
+++ b/lib/sqlalchemy/dialects/mssql/base.py
@@ -1155,7 +1155,11 @@ class MSSQLCompiler(compiler.SQLCompiler):
                                        'using an OFFSET or a non-simple '
                                        'LIMIT clause')
 
-            _order_by_clauses = select._order_by_clause.clauses
+            _order_by_clauses = [
+                sql_util.unwrap_label_reference(elem)
+                for elem in select._order_by_clause.clauses
+            ]
+
             limit_clause = select._limit_clause
             offset_clause = select._offset_clause
             kwargs['select_wraps_for'] = select
diff --git a/lib/sqlalchemy/sql/util.py b/lib/sqlalchemy/sql/util.py
index 5f180646c..24c6f5441 100644
--- a/lib/sqlalchemy/sql/util.py
+++ b/lib/sqlalchemy/sql/util.py
@@ -176,6 +176,16 @@ def unwrap_order_by(clause):
     return result
 
 
+def unwrap_label_reference(element):
+    def replace(elem):
+        if isinstance(elem, (_label_reference, _textual_label_reference)):
+            return elem.element
+
+    return visitors.replacement_traverse(
+        element, {}, replace
+    )
+
+
 def expand_column_list_from_order_by(collist, order_by):
     """Given the columns clause and ORDER BY of a selectable,
     return a list of column expressions that can be added to the collist
diff --git a/test/dialect/mssql/test_compiler.py b/test/dialect/mssql/test_compiler.py
index b59ca4fd1..599820492 100644
--- a/test/dialect/mssql/test_compiler.py
+++ b/test/dialect/mssql/test_compiler.py
@@ -571,6 +571,31 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
         assert t1.c.x in set(c._create_result_map()['x'][1])
         assert t1.c.y in set(c._create_result_map()['y'][1])
 
+    def test_offset_dont_misapply_labelreference(self):
+        m = MetaData()
+
+        t = Table('t', m, Column('x', Integer))
+
+        expr1 = func.foo(t.c.x).label('x')
+        expr2 = func.foo(t.c.x).label('y')
+
+        stmt1 = select([expr1]).order_by(expr1.desc()).offset(1)
+        stmt2 = select([expr2]).order_by(expr2.desc()).offset(1)
+
+        self.assert_compile(
+            stmt1,
+            "SELECT anon_1.x FROM (SELECT foo(t.x) AS x, "
+            "ROW_NUMBER() OVER (ORDER BY foo(t.x) DESC) AS mssql_rn FROM t) "
+            "AS anon_1 WHERE mssql_rn > :param_1"
+        )
+
+        self.assert_compile(
+            stmt2,
+            "SELECT anon_1.y FROM (SELECT foo(t.x) AS y, "
+            "ROW_NUMBER() OVER (ORDER BY foo(t.x) DESC) AS mssql_rn FROM t) "
+            "AS anon_1 WHERE mssql_rn > :param_1"
+        )
+
     def test_limit_zero_offset_using_window(self):
         t = table('t', column('x', Integer), column('y', Integer))