- more tests, including backend tests

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

7 files changed, 251 insertions, 61 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}
+        )
diff --git a/test/dialect/mssql/test_compiler.py b/test/dialect/mssql/test_compiler.py
index f12ab0330..3de8ea5c9 100644
--- a/test/dialect/mssql/test_compiler.py
+++ b/test/dialect/mssql/test_compiler.py
@@ -430,8 +430,8 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
                 "SELECT anon_1.x, anon_1.y FROM (SELECT t.x AS x, 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 > :mssql_rn_1",
-                checkparams={'mssql_rn_1': 20, 'x_1': 5}
+                "anon_1 WHERE mssql_rn > :param_1",
+                checkparams={'param_1': 20, 'x_1': 5}
             )
 
     def test_limit_offset_using_window(self):
@@ -446,8 +446,8 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
             "ROW_NUMBER() OVER (ORDER BY t.y) AS mssql_rn "
             "FROM t "
             "WHERE t.x = :x_1) AS anon_1 "
-            "WHERE mssql_rn > :mssql_rn_1 AND mssql_rn <= :mssql_rn_2",
-            checkparams={'mssql_rn_1': 20, 'mssql_rn_2': 30, 'x_1': 5}
+            "WHERE mssql_rn > :param_1 AND mssql_rn <= :param_2 + :param_1",
+            checkparams={'param_1': 20, 'param_2': 10, 'x_1': 5}
         )
 
     def test_limit_offset_with_correlated_order_by(self):
@@ -467,8 +467,8 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
             ") AS mssql_rn "
             "FROM t1 "
             "WHERE t1.x = :x_1) AS anon_1 "
-            "WHERE mssql_rn > :mssql_rn_1 AND mssql_rn <= :mssql_rn_2",
-            checkparams={'mssql_rn_1': 20, 'mssql_rn_2': 30, 'x_1': 5}
+            "WHERE mssql_rn > :param_1 AND mssql_rn <= :param_2 + :param_1",
+            checkparams={'param_1': 20, 'param_2': 10, 'x_1': 5}
         )
 
     def test_limit_zero_offset_using_window(self):
diff --git a/test/orm/test_query.py b/test/orm/test_query.py
index b4418042e..cc1f50023 100644
--- a/test/orm/test_query.py
+++ b/test/orm/test_query.py
@@ -1279,15 +1279,27 @@ class FilterTest(QueryTest, AssertsCompiledSQL):
         """Does a query allow bindparam for the limit?"""
         User = self.classes.User
         sess = create_session()
-        users = []
+        q1 = sess.query(self.classes.User).\
+                    order_by(self.classes.User.id).limit(bindparam('n'))
 
-        q1 = sess.query(self.classes.User).order_by(self.classes.User.id).limit(bindparam('n'))
-        for n in xrange(1,4):
-            users[:] = q1.params(n=n).all()
-            assert len(users) == n
+        for n in range(1, 4):
+            result = q1.params(n=n).all()
+            eq_(len(result), n)
 
-        assert [User(id=8), User(id=9)] == sess.query(User).order_by(User.id).limit(bindparam('limit')).offset(bindparam('offset')).params(limit=2, offset=1).all()
-        assert [User(id=8), User(id=9)] == list(sess.query(User).params(a=1, b=3).order_by(User.id)[bindparam('a'):bindparam('b')])
+        eq_(
+            sess.query(User).order_by(User.id).
+                    limit(bindparam('limit')).
+                    offset(bindparam('offset')).
+                    params(limit=2, offset=1).all()
+            [User(id=8), User(id=9)]
+        )
+        eq_(
+            list(
+                sess.query(User).params(a=1, b=3).
+                    order_by(User.id)[bindparam('a'):bindparam('b')]
+                )
+            [User(id=8), User(id=9)]
+        )
 
     @testing.requires.boolean_col_expressions
     def test_exists(self):
diff --git a/test/sql/test_compiler.py b/test/sql/test_compiler.py
index 917c7d89d..301cf149c 100644
--- a/test/sql/test_compiler.py
+++ b/test/sql/test_compiler.py
@@ -167,6 +167,57 @@ class SelectTest(fixtures.TestBase, AssertsCompiledSQL):
         assert_raises(ValueError, select, offset="foo")
         assert_raises(ValueError, select, limit="foo")
 
+    def test_limit_offset_no_int_coercion_one(self):
+        exp1 = literal_column("Q")
+        exp2 = literal_column("Y")
+        self.assert_compile(
+            select([1]).limit(exp1).offset(exp2),
+            "SELECT 1 LIMIT Q OFFSET Y"
+        )
+
+        self.assert_compile(
+            select([1]).limit(bindparam('x')).offset(bindparam('y')),
+            "SELECT 1 LIMIT :x OFFSET :y"
+        )
+
+    def test_limit_offset_no_int_coercion_two(self):
+        exp1 = literal_column("Q")
+        exp2 = literal_column("Y")
+        sel = select([1]).limit(exp1).offset(exp2)
+
+        assert_raises_message(
+            exc.CompileError,
+            "This SELECT structure does not use a simple integer "
+            "value for limit",
+            getattr, sel, "_limit"
+        )
+
+        assert_raises_message(
+            exc.CompileError,
+            "This SELECT structure does not use a simple integer "
+            "value for offset",
+            getattr, sel, "_offset"
+        )
+
+    def test_limit_offset_no_int_coercion_three(self):
+        exp1 = bindparam("Q")
+        exp2 = bindparam("Y")
+        sel = select([1]).limit(exp1).offset(exp2)
+
+        assert_raises_message(
+            exc.CompileError,
+            "This SELECT structure does not use a simple integer "
+            "value for limit",
+            getattr, sel, "_limit"
+        )
+
+        assert_raises_message(
+            exc.CompileError,
+            "This SELECT structure does not use a simple integer "
+            "value for offset",
+            getattr, sel, "_offset"
+        )
+
     def test_limit_offset(self):
         for lim, offset, exp, params in [
             (5, 10, "LIMIT :param_1 OFFSET :param_2",
@@ -182,6 +233,8 @@ class SelectTest(fixtures.TestBase, AssertsCompiledSQL):
                 checkparams=params
             )
 
+
+
     def test_select_precol_compile_ordering(self):
         s1 = select([column('x')]).select_from('a').limit(5).as_scalar()
         s2 = select([s1]).limit(10)