- more tests, including backend tests

- implement for SQL server, use window functions when simple limit/offset not available

4 files changed, 173 insertions, 48 deletions

diff --git a/lib/sqlalchemy/dialects/mssql/base.py b/lib/sqlalchemy/dialects/mssql/base.py
index 6a13d1dca..59cbb80bb 100644
--- a/lib/sqlalchemy/dialects/mssql/base.py
+++ b/lib/sqlalchemy/dialects/mssql/base.py
@@ -741,18 +741,21 @@ class MSSQLCompiler(compiler.SQLCompiler):
 
     def get_select_precolumns(self, select):
         """ MS-SQL puts TOP, it's version of LIMIT here """
-        limit = select._limit
-        if select._distinct or limit is not None:
-            s = select._distinct and "DISTINCT " or ""
 
+        s = ""
+        if select._distinct:
+            s += "DISTINCT "
+
+        if select._simple_int_limit and not select._offset:
             # ODBC drivers and possibly others
             # don't support bind params in the SELECT clause on SQL Server.
             # so have to use literal here.
-            if limit is not None:
-                if not select._offset:
-                    s += "TOP %d " % limit
+            s += "TOP %d " % select._limit
+
+        if s:
             return s
-        return compiler.SQLCompiler.get_select_precolumns(self, select)
+        else:
+            return compiler.SQLCompiler.get_select_precolumns(self, select)
 
     def get_from_hint_text(self, table, text):
         return text
@@ -769,28 +772,42 @@ class MSSQLCompiler(compiler.SQLCompiler):
         so tries to wrap it in a subquery with ``row_number()`` criterion.
 
         """
-        if select._offset and not getattr(select, '_mssql_visit', None):
+        if (
+                (
+                    not select._simple_int_limit and
+                    select._limit_clause is not None
+                ) or (
+                    select._offset_clause is not None and
+                    not select._simple_int_offset or select._offset
+                )
+            ) and not getattr(select, '_mssql_visit', None):
+
             # to use ROW_NUMBER(), an ORDER BY is required.
             if not select._order_by_clause.clauses:
                 raise exc.CompileError('MSSQL requires an order_by when '
-                                              'using an offset.')
+                                        'using an OFFSET or a non-simple '
+                                        'LIMIT clause')
 
-            _offset = select._offset
-            _limit = select._limit
             _order_by_clauses = select._order_by_clause.clauses
+            limit_clause = select._limit_clause
+            offset_clause = select._offset_clause
             select = select._generate()
             select._mssql_visit = True
             select = select.column(
-                 sql.func.ROW_NUMBER().over(order_by=_order_by_clauses)
-                     .label("mssql_rn")
-                                   ).order_by(None).alias()
+                    sql.func.ROW_NUMBER().over(order_by=_order_by_clauses)
+                    .label("mssql_rn")).order_by(None).alias()
 
             mssql_rn = sql.column('mssql_rn')
             limitselect = sql.select([c for c in select.c if
                                         c.key != 'mssql_rn'])
-            limitselect.append_whereclause(mssql_rn > _offset)
-            if _limit is not None:
-                limitselect.append_whereclause(mssql_rn <= (_limit + _offset))
+            if offset_clause is not None:
+                limitselect.append_whereclause(mssql_rn > offset_clause)
+                if limit_clause is not None:
+                    limitselect.append_whereclause(
+                            mssql_rn <= (limit_clause + offset_clause))
+            else:
+                limitselect.append_whereclause(
+                            mssql_rn <= (limit_clause))
             return self.process(limitselect, iswrapper=True, **kwargs)
         else:
             return compiler.SQLCompiler.visit_select(self, select, **kwargs)
diff --git a/lib/sqlalchemy/sql/selectable.py b/lib/sqlalchemy/sql/selectable.py
index 767eb086c..72e4a930d 100644
--- a/lib/sqlalchemy/sql/selectable.py
+++ b/lib/sqlalchemy/sql/selectable.py
@@ -48,38 +48,46 @@ def _interpret_as_select(element):
         element = element.select()
     return element
 
+class _OffsetLimitParam(BindParameter):
+    @property
+    def _limit_offset_value(self):
+        return self.effective_value
+
 def _offset_or_limit_clause(element, name=None, type_=None):
-    """
-    If the element is a custom clause of some sort, returns (None, element)
-    If the element is a BindParameter, return (element.effective_value, element)
-    Otherwise, assume element is an int and create a new bindparam and return (asint(element), BindParameter(...))
+    """Convert the given value to an "offset or limit" clause.
+
+    This handles incoming integers and converts to an expression; if
+    an expression is already given, it is passed through.
+
     """
     if element is None:
         return None
-    if hasattr(element, '__clause_element__'):
+    elif hasattr(element, '__clause_element__'):
         return element.__clause_element__()
-    if isinstance(element, Visitable):
+    elif isinstance(element, Visitable):
         return element
+    else:
+        value = util.asint(element)
+        return _OffsetLimitParam(name, value, type_=type_, unique=True)
 
-    value = util.asint(element)
-    return BindParameter(name, value, type_=type_, unique=True)
+def _offset_or_limit_clause_asint(clause, attrname):
+    """Convert the "offset or limit" clause of a select construct to an
+    integer.
+
+    This is only possible if the value is stored as a simple bound parameter.
+    Otherwise, a compilation error is raised.
 
-def _offset_or_limit_clause_asint(clause):
-    """
-    Get the integer value of an offset or limit clause, for database engines that
-    require it to be a plain integer instead of a BindParameter or other custom
-    clause.
-
-    If the clause is None, returns None.
-    If the clause is not a BindParameter, throws an exception.
-    If the clause is a BindParameter but its value is not set yet or not an int, throws an exception.
-    Otherwise, returns the integer in the clause.
     """
     if clause is None:
         return None
-    if not isinstance(clause, BindParameter):
-        raise Exception("Limit is not a simple integer")
-    return util.asint(clause.effective_value)
+    try:
+        value = clause._limit_offset_value
+    except AttributeError:
+        raise exc.CompileError(
+                        "This SELECT structure does not use a simple "
+                        "integer value for %s" % attrname)
+    else:
+        return util.asint(value)
 
 def subquery(alias, *args, **kwargs):
     """Return an :class:`.Alias` object derived
@@ -1676,21 +1684,37 @@ class GenerativeSelect(SelectBase):
 
     @property
     def _limit(self):
+        """Get an integer value for the limit.  This should only be used
+        by code that cannot support a limit as a BindParameter or
+        other custom clause as it will throw an exception if the limit
+        isn't currently set to an integer.
+
         """
-        Get an integer value for the limit.  This should only be used by code that
-        cannot support a limit as a BindParameter or other custom clause as it will
-        throw an exception if the limit isn't currently set to an integer.
+        return _offset_or_limit_clause_asint(self._limit_clause, "limit")
+
+    @property
+    def _simple_int_limit(self):
+        """True if the LIMIT clause is a simple integer, False
+        if it is not present or is a SQL expression.
         """
-        return _offset_or_limit_clause_asint(self._limit_clause)
+        return isinstance(self._limit_clause, _OffsetLimitParam)
 
     @property
-    def _offset(self):
+    def _simple_int_offset(self):
+        """True if the OFFSET clause is a simple integer, False
+        if it is not present or is a SQL expression.
         """
-        Get an integer value for the offset.  This should only be used by code that
-        cannot support an offset as a BindParameter or other custom clause as it will
-        throw an exception if the offset isn't currently set to an integer.
+        return isinstance(self._offset_clause, _OffsetLimitParam)
+
+    @property
+    def _offset(self):
+        """Get an integer value for the offset.  This should only be used
+        by code that cannot support an offset as a BindParameter or
+        other custom clause as it will throw an exception if the
+        offset isn't currently set to an integer.
+
         """
-        return _offset_or_limit_clause_asint(self._offset_clause)
+        return _offset_or_limit_clause_asint(self._offset_clause, "offset")
 
     @_generative
     def limit(self, limit):
diff --git a/lib/sqlalchemy/testing/requirements.py b/lib/sqlalchemy/testing/requirements.py
index 04e8ad272..e1669e952 100644
--- a/lib/sqlalchemy/testing/requirements.py
+++ b/lib/sqlalchemy/testing/requirements.py
@@ -98,6 +98,12 @@ class SuiteRequirements(Requirements):
         return exclusions.open()
 
     @property
+    def bound_limit_offset(self):
+        """target database can render LIMIT and/or OFFSET using a bound parameter"""
+
+        return exclusions.open()
+
+    @property
     def boolean_col_expressions(self):
         """Target database must support boolean expressions as columns"""
 
diff --git a/lib/sqlalchemy/testing/suite/test_select.py b/lib/sqlalchemy/testing/suite/test_select.py
index 2ccff61ea..3461b1e94 100644
--- a/lib/sqlalchemy/testing/suite/test_select.py
+++ b/lib/sqlalchemy/testing/suite/test_select.py
@@ -2,7 +2,8 @@ from .. import fixtures, config
 from ..assertions import eq_
 
 from sqlalchemy import util
-from sqlalchemy import Integer, String, select, func
+from sqlalchemy import Integer, String, select, func, bindparam
+from sqlalchemy import testing
 
 from ..schema import Table, Column
 
@@ -84,3 +85,80 @@ class OrderByLabelTest(fixtures.TablesTest):
             select([lx]).order_by(lx.desc()),
             [(7, ), (5, ), (3, )]
         )
+
+class LimitOffsetTest(fixtures.TablesTest):
+    __backend__ = True
+
+    @classmethod
+    def define_tables(cls, metadata):
+        Table("some_table", metadata,
+            Column('id', Integer, primary_key=True),
+            Column('x', Integer),
+            Column('y', Integer))
+
+    @classmethod
+    def insert_data(cls):
+        config.db.execute(
+            cls.tables.some_table.insert(),
+            [
+                {"id": 1, "x": 1, "y": 2},
+                {"id": 2, "x": 2, "y": 3},
+                {"id": 3, "x": 3, "y": 4},
+                {"id": 4, "x": 4, "y": 5},
+            ]
+        )
+
+    def _assert_result(self, select, result, params=()):
+        eq_(
+            config.db.execute(select, params).fetchall(),
+            result
+        )
+
+    def test_simple_limit(self):
+        table = self.tables.some_table
+        self._assert_result(
+            select([table]).order_by(table.c.id).limit(2),
+            [(1, 1, 2), (2, 2, 3)]
+        )
+
+    def test_simple_offset(self):
+        table = self.tables.some_table
+        self._assert_result(
+            select([table]).order_by(table.c.id).offset(2),
+            [(3, 3, 4), (4, 4, 5)]
+        )
+
+    def test_simple_limit_offset(self):
+        table = self.tables.some_table
+        self._assert_result(
+            select([table]).order_by(table.c.id).limit(2).offset(1),
+            [(2, 2, 3), (3, 3, 4)]
+        )
+
+    @testing.requires.bound_limit_offset
+    def test_bound_limit(self):
+        table = self.tables.some_table
+        self._assert_result(
+            select([table]).order_by(table.c.id).limit(bindparam('l')),
+            [(1, 1, 2), (2, 2, 3)],
+            params={"l": 2}
+        )
+
+    @testing.requires.bound_limit_offset
+    def test_bound_offset(self):
+        table = self.tables.some_table
+        self._assert_result(
+            select([table]).order_by(table.c.id).offset(bindparam('o')),
+            [(3, 3, 4), (4, 4, 5)],
+            params={"o": 2}
+        )
+
+    @testing.requires.bound_limit_offset
+    def test_bound_limit_offset(self):
+        table = self.tables.some_table
+        self._assert_result(
+            select([table]).order_by(table.c.id).\
+                limit(bindparam("l")).offset(bindparam("o")),
+            [(2, 2, 3), (3, 3, 4)],
+            params={"l": 2, "o": 1}
+        )