Merge branch 'ticket_3034'

11 files changed, 258 insertions, 68 deletions

diff --git a/lib/sqlalchemy/dialects/firebird/base.py b/lib/sqlalchemy/dialects/firebird/base.py
index 21db57b68..a3867457e 100644
--- a/lib/sqlalchemy/dialects/firebird/base.py
+++ b/lib/sqlalchemy/dialects/firebird/base.py
@@ -77,7 +77,6 @@ from sqlalchemy.sql import expression
 from sqlalchemy.engine import base, default, reflection
 from sqlalchemy.sql import compiler
 
-
 from sqlalchemy.types import (BIGINT, BLOB, DATE, FLOAT, INTEGER, NUMERIC,
                               SMALLINT, TEXT, TIME, TIMESTAMP, Integer)
 
@@ -300,10 +299,10 @@ class FBCompiler(sql.compiler.SQLCompiler):
         """
 
         result = ""
-        if select._limit:
-            result += "FIRST %s " % self.process(sql.literal(select._limit))
-        if select._offset:
-            result += "SKIP %s " % self.process(sql.literal(select._offset))
+        if select._limit_clause:
+            result += "FIRST %s " % self.process(select._limit_clause)
+        if select._offset_clause:
+            result += "SKIP %s " % self.process(select._offset_clause)
         if select._distinct:
             result += "DISTINCT "
         return result
diff --git a/lib/sqlalchemy/dialects/mssql/base.py b/lib/sqlalchemy/dialects/mssql/base.py
index 3fcc95f46..59cbb80bb 100644
--- a/lib/sqlalchemy/dialects/mssql/base.py
+++ b/lib/sqlalchemy/dialects/mssql/base.py
@@ -740,18 +740,22 @@ class MSSQLCompiler(compiler.SQLCompiler):
                                         self.process(binary.right, **kw))
 
     def get_select_precolumns(self, select):
-        """ MS-SQL puts TOP, its version of LIMIT, here """
-        if select._distinct or select._limit is not None:
-            s = select._distinct and "DISTINCT " or ""
+        """ MS-SQL puts TOP, it's version of LIMIT here """
 
+        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 select._limit is not None:
-                if not select._offset:
-                    s += "TOP %d " % select._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
@@ -768,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/dialects/mysql/base.py b/lib/sqlalchemy/dialects/mysql/base.py
index 85cbd6e22..e8a15fdbc 100644
--- a/lib/sqlalchemy/dialects/mysql/base.py
+++ b/lib/sqlalchemy/dialects/mysql/base.py
@@ -1568,15 +1568,15 @@ class MySQLCompiler(compiler.SQLCompiler):
         # The latter is more readable for offsets but we're stuck with the
         # former until we can refine dialects by server revision.
 
-        limit, offset = select._limit, select._offset
+        limit_clause, offset_clause = select._limit_clause, select._offset_clause
 
-        if (limit, offset) == (None, None):
+        if (limit_clause, offset_clause) == (None, None):
             return ''
-        elif offset is not None:
+        elif offset_clause is not None:
             # As suggested by the MySQL docs, need to apply an
             # artificial limit if one wasn't provided
             # http://dev.mysql.com/doc/refman/5.0/en/select.html
-            if limit is None:
+            if limit_clause is None:
                 # hardwire the upper limit.  Currently
                 # needed by OurSQL with Python 3
                 # (https://bugs.launchpad.net/oursql/+bug/686232),
@@ -1584,15 +1584,15 @@ class MySQLCompiler(compiler.SQLCompiler):
                 # bound as part of MySQL's "syntax" for OFFSET with
                 # no LIMIT
                 return ' \n LIMIT %s, %s' % (
-                                self.process(sql.literal(offset)),
+                                self.process(offset_clause),
                                 "18446744073709551615")
             else:
                 return ' \n LIMIT %s, %s' % (
-                                self.process(sql.literal(offset)),
-                                self.process(sql.literal(limit)))
+                                self.process(offset_clause),
+                                self.process(limit_clause))
         else:
             # No offset provided, so just use the limit
-            return ' \n LIMIT %s' % (self.process(sql.literal(limit)),)
+            return ' \n LIMIT %s' % (self.process(limit_clause),)
 
     def update_limit_clause(self, update_stmt):
         limit = update_stmt.kwargs.get('%s_limit' % self.dialect.name, None)
diff --git a/lib/sqlalchemy/dialects/oracle/base.py b/lib/sqlalchemy/dialects/oracle/base.py
index 5dcc93f79..b0f32f491 100644
--- a/lib/sqlalchemy/dialects/oracle/base.py
+++ b/lib/sqlalchemy/dialects/oracle/base.py
@@ -652,24 +652,26 @@ class OracleCompiler(compiler.SQLCompiler):
 
                 # Wrap the middle select and add the hint
                 limitselect = sql.select([c for c in select.c])
-                if select._limit and self.dialect.optimize_limits:
-                    limitselect = limitselect.prefix_with("/*+ FIRST_ROWS(%d) */" % select._limit)
+                limit = select._limit
+                if limit and self.dialect.optimize_limits:
+                    limitselect = limitselect.prefix_with("/*+ FIRST_ROWS(%d) */" % limit)
 
                 limitselect._oracle_visit = True
                 limitselect._is_wrapper = True
 
                 # If needed, add the limiting clause
-                if select._limit is not None:
-                    max_row = select._limit
-                    if select._offset is not None:
-                        max_row += select._offset
+                offset = select._offset
+                if limit is not None:
+                    max_row = limit
+                    if offset is not None:
+                        max_row += offset
                     if not self.dialect.use_binds_for_limits:
                         max_row = sql.literal_column("%d" % max_row)
                     limitselect.append_whereclause(
                             sql.literal_column("ROWNUM") <= max_row)
 
                 # If needed, add the ora_rn, and wrap again with offset.
-                if select._offset is None:
+                if offset is None:
                     limitselect._for_update_arg = select._for_update_arg
                     select = limitselect
                 else:
@@ -683,7 +685,7 @@ class OracleCompiler(compiler.SQLCompiler):
                     offsetselect._oracle_visit = True
                     offsetselect._is_wrapper = True
 
-                    offset_value = select._offset
+                    offset_value = offset
                     if not self.dialect.use_binds_for_limits:
                         offset_value = sql.literal_column("%d" % offset_value)
                     offsetselect.append_whereclause(
diff --git a/lib/sqlalchemy/dialects/postgresql/base.py b/lib/sqlalchemy/dialects/postgresql/base.py
index bcbf0b12c..e2accfedb 100644
--- a/lib/sqlalchemy/dialects/postgresql/base.py
+++ b/lib/sqlalchemy/dialects/postgresql/base.py
@@ -1161,12 +1161,12 @@ class PGCompiler(compiler.SQLCompiler):
 
     def limit_clause(self, select):
         text = ""
-        if select._limit is not None:
-            text += " \n LIMIT " + self.process(sql.literal(select._limit))
-        if select._offset is not None:
-            if select._limit is None:
+        if select._limit_clause is not None:
+            text += " \n LIMIT " + self.process(select._limit_clause)
+        if select._offset_clause is not None:
+            if select._limit_clause is None:
                 text += " \n LIMIT ALL"
-            text += " OFFSET " + self.process(sql.literal(select._offset))
+            text += " OFFSET " + self.process(select._offset_clause)
         return text
 
     def format_from_hint_text(self, sqltext, table, hint, iscrud):
diff --git a/lib/sqlalchemy/dialects/sqlite/base.py b/lib/sqlalchemy/dialects/sqlite/base.py
index 7687493b2..de8b01e6f 100644
--- a/lib/sqlalchemy/dialects/sqlite/base.py
+++ b/lib/sqlalchemy/dialects/sqlite/base.py
@@ -519,12 +519,12 @@ class SQLiteCompiler(compiler.SQLCompiler):
 
     def limit_clause(self, select):
         text = ""
-        if select._limit is not None:
-            text += "\n LIMIT " + self.process(sql.literal(select._limit))
-        if select._offset is not None:
-            if select._limit is None:
+        if select._limit_clause is not None:
+            text += "\n LIMIT " + self.process(select._limit_clause)
+        if select._offset_clause is not None:
+            if select._limit_clause is None:
                 text += "\n LIMIT " + self.process(sql.literal(-1))
-            text += " OFFSET " + self.process(sql.literal(select._offset))
+            text += " OFFSET " + self.process(select._offset_clause)
         else:
             text += " OFFSET " + self.process(sql.literal(0))
         return text
diff --git a/lib/sqlalchemy/dialects/sybase/base.py b/lib/sqlalchemy/dialects/sybase/base.py
index 501270778..3e61b5ba6 100644
--- a/lib/sqlalchemy/dialects/sybase/base.py
+++ b/lib/sqlalchemy/dialects/sybase/base.py
@@ -325,18 +325,20 @@ class SybaseSQLCompiler(compiler.SQLCompiler):
         s = select._distinct and "DISTINCT " or ""
         # TODO: don't think Sybase supports
         # bind params for FIRST / TOP
-        if select._limit:
+        limit = select._limit
+        if limit:
             #if select._limit == 1:
                 #s += "FIRST "
             #else:
                 #s += "TOP %s " % (select._limit,)
-            s += "TOP %s " % (select._limit,)
-        if select._offset:
-            if not select._limit:
+            s += "TOP %s " % (limit,)
+        offset = select._offset
+        if offset:
+            if not limit:
                 # FIXME: sybase doesn't allow an offset without a limit
                 # so use a huge value for TOP here
                 s += "TOP 1000000 "
-            s += "START AT %s " % (select._offset + 1,)
+            s += "START AT %s " % (offset + 1,)
         return s
 
     def get_from_hint_text(self, table, text):
diff --git a/lib/sqlalchemy/sql/compiler.py b/lib/sqlalchemy/sql/compiler.py
index cd01ea5e5..65d0169f4 100644
--- a/lib/sqlalchemy/sql/compiler.py
+++ b/lib/sqlalchemy/sql/compiler.py
@@ -773,7 +773,7 @@ class SQLCompiler(Compiled):
             text += " GROUP BY " + group_by
 
         text += self.order_by_clause(cs, **kwargs)
-        text += (cs._limit is not None or cs._offset is not None) and \
+        text += (cs._limit_clause is not None or cs._offset_clause is not None) and \
                         self.limit_clause(cs) or ""
 
         if self.ctes and \
@@ -1557,7 +1557,7 @@ class SQLCompiler(Compiled):
             text += self.order_by_clause(select,
                             order_by_select=order_by_select, **kwargs)
 
-        if select._limit is not None or select._offset is not None:
+        if select._limit_clause is not None or select._offset_clause is not None:
             text += self.limit_clause(select)
 
         if select._for_update_arg is not None:
@@ -1625,12 +1625,12 @@ class SQLCompiler(Compiled):
 
     def limit_clause(self, select):
         text = ""
-        if select._limit is not None:
-            text += "\n LIMIT " + self.process(elements.literal(select._limit))
-        if select._offset is not None:
-            if select._limit is None:
+        if select._limit_clause is not None:
+            text += "\n LIMIT " + self.process(select._limit_clause)
+        if select._offset_clause is not None:
+            if select._limit_clause is None:
                 text += "\n LIMIT -1"
-            text += " OFFSET " + self.process(elements.literal(select._offset))
+            text += " OFFSET " + self.process(select._offset_clause)
         return text
 
     def visit_table(self, table, asfrom=False, iscrud=False, ashint=False,
diff --git a/lib/sqlalchemy/sql/selectable.py b/lib/sqlalchemy/sql/selectable.py
index a13186097..72e4a930d 100644
--- a/lib/sqlalchemy/sql/selectable.py
+++ b/lib/sqlalchemy/sql/selectable.py
@@ -10,7 +10,7 @@ SQL tables and derived rowsets.
 """
 
 from .elements import ClauseElement, TextClause, ClauseList, \
-        and_, Grouping, UnaryExpression, literal_column
+        and_, Grouping, UnaryExpression, literal_column, BindParameter
 from .elements import _clone, \
         _literal_as_text, _interpret_as_column_or_from, _expand_cloned,\
         _select_iterables, _anonymous_label, _clause_element_as_expr,\
@@ -28,6 +28,8 @@ import operator
 import collections
 from .annotation import Annotated
 import itertools
+from sqlalchemy.sql.visitors import Visitable
+
 
 def _interpret_as_from(element):
     insp = inspection.inspect(element, raiseerr=False)
@@ -46,6 +48,47 @@ 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):
+    """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
+    elif hasattr(element, '__clause_element__'):
+        return element.__clause_element__()
+    elif isinstance(element, Visitable):
+        return element
+    else:
+        value = util.asint(element)
+        return _OffsetLimitParam(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.
+
+    """
+    if clause is None:
+        return None
+    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
     from a :class:`.Select`.
@@ -1536,8 +1579,8 @@ class GenerativeSelect(SelectBase):
     """
     _order_by_clause = ClauseList()
     _group_by_clause = ClauseList()
-    _limit = None
-    _offset = None
+    _limit_clause = None
+    _offset_clause = None
     _for_update_arg = None
 
     def __init__(self,
@@ -1562,9 +1605,9 @@ class GenerativeSelect(SelectBase):
                 self._execution_options.union(
                   {'autocommit': autocommit})
         if limit is not None:
-            self._limit = util.asint(limit)
+            self._limit_clause = _offset_or_limit_clause(limit)
         if offset is not None:
-            self._offset = util.asint(offset)
+            self._offset_clause = _offset_or_limit_clause(offset)
         self._bind = bind
 
         if order_by is not None:
@@ -1639,19 +1682,53 @@ class GenerativeSelect(SelectBase):
         """
         self.use_labels = True
 
+    @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.
+
+        """
+        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 isinstance(self._limit_clause, _OffsetLimitParam)
+
+    @property
+    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.
+        """
+        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, "offset")
+
     @_generative
     def limit(self, limit):
         """return a new selectable with the given LIMIT criterion
         applied."""
 
-        self._limit = util.asint(limit)
+        self._limit_clause = _offset_or_limit_clause(limit)
 
     @_generative
     def offset(self, offset):
         """return a new selectable with the given OFFSET criterion
         applied."""
 
-        self._offset = util.asint(offset)
+        self._offset_clause = _offset_or_limit_clause(offset)
 
     @_generative
     def order_by(self, *clauses):
@@ -1712,6 +1789,12 @@ class GenerativeSelect(SelectBase):
             self._group_by_clause = ClauseList(*clauses)
 
 
+    def _copy_internals(self, clone=_clone, **kw):
+        if self._limit_clause is not None:
+            self._limit_clause = clone(self._limit_clause, **kw)
+        if self._offset_clause is not None:
+            self._offset_clause = clone(self._offset_clause, **kw)
+
 class CompoundSelect(GenerativeSelect):
     """Forms the basis of ``UNION``, ``UNION ALL``, and other
         SELECT-based set operations.
@@ -1930,6 +2013,7 @@ class CompoundSelect(GenerativeSelect):
                 "addition of columns to underlying selectables")
 
     def _copy_internals(self, clone=_clone, **kw):
+        super(CompoundSelect, self)._copy_internals(clone, **kw)
         self._reset_exported()
         self.selects = [clone(s, **kw) for s in self.selects]
         if hasattr(self, '_col_map'):
@@ -2380,6 +2464,7 @@ class Select(HasPrefixes, GenerativeSelect):
         return False
 
     def _copy_internals(self, clone=_clone, **kw):
+        super(Select, self)._copy_internals(clone, **kw)
 
         # Select() object has been cloned and probably adapted by the
         # given clone function.  Apply the cloning function to internal
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}
+        )