fix(backend): ensure that chained limits respect prior limits

ibis/backends/base/sql/alchemy/query_builder.py

@@ -330,11 +330,9 @@ def _add_limit(self, fragment):
         if self.limit is None:
             return fragment
 
-        n, offset = self.limit['n'], self.limit['offset']
-        fragment = fragment.limit(n)
-        if offset is not None and offset != 0:
+        fragment = fragment.limit(self.limit.n)
+        if offset := self.limit.offset:
             fragment = fragment.offset(offset)
-
         return fragment
 
 

ibis/backends/base/sql/compiler/query_builder.py

@@ -9,7 +9,10 @@
 import ibis.expr.types as ir
 import ibis.util as util
 from ibis.backends.base.sql.compiler.base import DML, QueryAST, SetOp
-from ibis.backends.base.sql.compiler.select_builder import SelectBuilder
+from ibis.backends.base.sql.compiler.select_builder import (
+    SelectBuilder,
+    _LimitSpec,
+)
 from ibis.backends.base.sql.compiler.translator import (
     ExprTranslator,
     QueryContext,
@@ -446,9 +449,9 @@ def format_limit(self):
 
         buf = StringIO()
 
-        n, offset = self.limit['n'], self.limit['offset']
+        n = self.limit.n
         buf.write(f'LIMIT {n}')
-        if offset is not None and offset != 0:
+        if offset := self.limit.offset:
             buf.write(f' OFFSET {offset}')
 
         return buf.getvalue()
@@ -584,9 +587,9 @@ def to_ast_ensure_limit(cls, expr, limit=None, params=None):
                     else:
                         query_limit = limit
                     if query_limit:
-                        query.limit = {'n': query_limit, 'offset': 0}
+                        query.limit = _LimitSpec(query_limit, offset=0)
                 elif limit is not None and limit != 'default':
-                    query.limit = {'n': limit, 'offset': query.limit['offset']}
+                    query.limit = _LimitSpec(limit, query.limit.offset)
 
         return query_ast
 

ibis/backends/base/sql/compiler/select_builder.py

@@ -1,3 +1,7 @@
+from __future__ import annotations
+
+from typing import NamedTuple
+
 import toolz
 
 import ibis.common.exceptions as com
@@ -10,6 +14,11 @@
 )
 
 
+class _LimitSpec(NamedTuple):
+    n: int
+    offset: int
+
+
 class _CorrelatedRefCheck:
     def __init__(self, query, expr):
         self.query = query
@@ -438,11 +447,16 @@ def _collect_Limit(self, expr, toplevel=False):
             return
 
         op = expr.op()
+        n = op.n
+        offset = op.offset or 0
 
-        # Ignore "inner" limits, because they've been overrided by an exterior
-        # one
         if self.limit is None:
-            self.limit = {'n': op.n, 'offset': op.offset}
+            self.limit = _LimitSpec(n, offset)
+        else:
+            self.limit = _LimitSpec(
+                min(n, self.limit.n),
+                offset + self.limit.offset,
+            )
 
         self._collect(op.table, toplevel=toplevel)
 

ibis/backends/clickhouse/compiler.py

@@ -50,8 +50,8 @@ def format_limit(self):
 
         buf = StringIO()
 
-        n, offset = self.limit['n'], self.limit['offset']
-        if offset is not None and offset != 0:
+        n = self.limit.n
+        if offset := self.limit.offset:
             buf.write(f'LIMIT {offset}, {n}')
         else:
             buf.write(f'LIMIT {n}')

ibis/backends/dask/execution/generic.py

@@ -290,7 +290,7 @@ def execute_cast_series_date(op, data, type, **kwargs):
 @execute_node.register(ops.Limit, dd.DataFrame, integer_types, integer_types)
 def execute_limit_frame(op, data, nrows, offset, **kwargs):
     # NOTE: Dask Dataframes do not support iloc row based indexing
-    return data.loc[offset : offset + nrows]
+    return data.loc[offset : (offset + nrows) - 1]
 
 
 @execute_node.register(ops.Not, (dd.core.Scalar, dd.Series))

ibis/backends/tests/test_api.py

@@ -1,6 +1,8 @@
 import pytest
+from pytest import param
 
 import ibis
+import ibis.common.exceptions as exc
 
 
 def test_backend_name(backend):
@@ -79,3 +81,17 @@ def test_tables_accessor_tab_completion(con):
 
     keys = con.tables._ipython_key_completions_()
     assert 'functional_alltypes' in keys
+
+
+@pytest.mark.notimpl(["datafusion"], raises=exc.OperationNotDefinedError)
+@pytest.mark.parametrize(
+    "expr_fn",
+    [
+        param(lambda t: t.limit(5).limit(10), id="small_big"),
+        param(lambda t: t.limit(10).limit(5), id="big_small"),
+    ],
+)
+def test_limit_chain(alltypes, expr_fn):
+    expr = expr_fn(alltypes)
+    result = expr.execute()
+    assert len(result) == 5

ibis/tests/sql/test_select_sql.py

@@ -1335,7 +1335,7 @@ def test_multiple_limits(functional_alltypes):
     expr = t.limit(20).limit(10)
     stmt = get_query(expr)
 
-    assert stmt.limit['n'] == 10
+    assert stmt.limit.n == 10
 
 
 def test_self_join_filter_analysis_bug(filter_self_join_analysis_bug):